pitrou commented on code in PR #51038:
URL: https://github.com/apache/arrow/pull/51038#discussion_r4062437677
##########
cpp/src/arrow/json/parser.cc:
##########
@@ -48,13 +45,30 @@ using internal::checked_cast;
namespace json {
-namespace rj = arrow::rapidjson;
+namespace sj = simdjson::ondemand;
template <typename... T>
static Status ParseError(T&&... t) {
return Status::Invalid("JSON parse error: ", std::forward<T>(t)...);
}
+static std::string_view TrimTrailingWhitespace(std::string_view value) {
Review Comment:
Can we factor this out together with similar code in the JSON chunker? We
don't want to maintain multiple lists of allowed whitespace characters.
##########
cpp/src/arrow/json/parser.cc:
##########
@@ -762,43 +741,199 @@ class HandlerBase : public BlockParser,
}
protected:
- template <typename Handler, typename Stream>
- Status DoParse(Handler& handler, Stream&& json, size_t json_size) {
- constexpr auto parse_flags = rj::kParseIterativeFlag |
rj::kParseNanAndInfFlag |
- rj::kParseStopWhenDoneFlag |
- rj::kParseNumbersAsStringsFlag;
-
- rj::Reader reader;
- // ensure that the loop can exit when the block too large.
- for (; num_rows_ < std::numeric_limits<int32_t>::max(); ++num_rows_) {
- auto ok = reader.Parse<parse_flags>(json, handler);
- switch (ok.Code()) {
- case rj::kParseErrorNone:
- // parse the next object
- continue;
- case rj::kParseErrorDocumentEmpty:
- if (json.Tell() < json_size) {
- return ParseError(rj::GetParseError_En(ok.Code()));
- }
- // parsed all objects, finish
- return Status::OK();
- case rj::kParseErrorTermination:
- // handler emitted an error
- return handler.Error();
- default:
- // rj emitted an error
- return ParseError(rj::GetParseError_En(ok.Code()), " in row ",
num_rows_);
+ template <typename Handler>
+ Status DoParse(Handler& handler, const std::shared_ptr<Buffer>& json) {
+ RETURN_NOT_OK(ReserveScalarStorage(json->size()));
+
+ const std::string_view input(reinterpret_cast<const char*>(json->data()),
+ json->size());
+
+ if (IsWhitespaceOnly(input)) {
+ return Status::OK();
+ }
+
+ auto parse = [&](const auto& input) -> Status {
+ ARROW_ASSIGN_OR_RAISE(auto stream,
arrow::internal::ResolveSimdjsonResult(
+ parser_.iterate_many(input),
+ "Failed to create JSON document
stream"));
+
+ for (auto document_result : stream) {
+ ARROW_ASSIGN_OR_RAISE(
+ auto document,
+ arrow::internal::ResolveSimdjsonResult(
+ document_result, "Failed to iterate JSON document stream"));
+
+ if (num_rows_ == std::numeric_limits<int32_t>::max()) {
+ return Status::Invalid("Row count overflowed int32_t");
+ }
+
+ ARROW_ASSIGN_OR_RAISE(
+ auto value,
+ arrow::internal::ResolveSimdjsonResult(
+ document.get_value(), "JSON parse error: Failed to get JSON
value"));
+
+ RETURN_NOT_OK(ParseValue(handler, value));
+
+ ++num_rows_;
+ }
+
+ if (stream.truncated_bytes() != 0) {
+ return ParseError("The document is empty");
+ }
+
+ return Status::OK();
+ };
+
+ if (json->capacity() - json->size() >=
+ static_cast<int64_t>(simdjson::SIMDJSON_PADDING)) {
+ const auto padded_json = simdjson::padded_string_view(
+ reinterpret_cast<const char*>(json->data()), json->size(),
json->capacity());
+ return parse(padded_json);
+ }
+
+ simdjson::padded_string padded_json(reinterpret_cast<const
char*>(json->data()),
+ json->size());
+ return parse(padded_json);
+ }
+
+ template <Kind::type kind>
+ Status MaybePromoteFromNull() {
+ if (builder_.kind != Kind::kNull) {
+ return Status::OK();
+ }
+
+ auto parent = builder_stack_.back();
+
+ if (parent.kind == Kind::kArray) {
+ auto list_builder = Cast<Kind::kArray>(parent);
+ DCHECK_EQ(list_builder->value_builder(), builder_);
+
+ RETURN_NOT_OK(builder_set_.MakeBuilder<kind>(builder_.index, &builder_));
+
+ list_builder = Cast<Kind::kArray>(parent);
+ list_builder->value_builder(builder_);
+ } else {
+ auto struct_builder = Cast<Kind::kObject>(parent);
+ DCHECK_EQ(struct_builder->field_builder(field_index_), builder_);
+
+ RETURN_NOT_OK(builder_set_.MakeBuilder<kind>(builder_.index, &builder_));
+
+ struct_builder = Cast<Kind::kObject>(parent);
+ struct_builder->field_builder(field_index_, builder_);
+ }
+
+ return Status::OK();
+ }
+
+ template <typename Handler>
+ Status ParseValue(Handler& handler, sj::value value) {
+ ARROW_ASSIGN_OR_RAISE(auto type, arrow::internal::ResolveSimdjsonResult(
+ value.type(), "Failed to determine
JSON type"));
+
+ switch (type) {
+ case sj::json_type::null: {
+ ARROW_ASSIGN_OR_RAISE([[maybe_unused]] auto is_null,
+ arrow::internal::ResolveSimdjsonResult(
+ value.is_null(), "Failed to validate JSON
null"));
+ return Null();
+ }
+
+ case sj::json_type::boolean: {
+ RETURN_NOT_OK(handler.template MaybePromoteFromNull<Kind::kBoolean>());
+
+ ARROW_ASSIGN_OR_RAISE(auto boolean,
+ arrow::internal::ResolveSimdjsonResult(
+ value.get_bool(), "Failed to get JSON
boolean"));
+ return Bool(boolean);
+ }
+
+ case sj::json_type::string: {
+ RETURN_NOT_OK(handler.template MaybePromoteFromNull<Kind::kString>());
+
+ ARROW_ASSIGN_OR_RAISE(auto string,
+ arrow::internal::ResolveSimdjsonResult(
+ value.get_string(), "Failed to get JSON
string"));
+ return String(string);
}
+
+ case sj::json_type::number: {
+ RETURN_NOT_OK(handler.template MaybePromoteFromNull<Kind::kNumber>());
+ return RawNumber(TrimTrailingWhitespace(value.raw_json_token()));
+ }
+
+ case sj::json_type::array:
+ RETURN_NOT_OK(handler.template MaybePromoteFromNull<Kind::kArray>());
+ return ParseArray(handler, value);
+
+ case sj::json_type::object:
+ RETURN_NOT_OK(handler.template MaybePromoteFromNull<Kind::kObject>());
+ return ParseObject(handler, value);
+
+ default:
+ return ParseError("Invalid value");
}
- return Status::Invalid("Row count overflowed int32_t");
}
template <typename Handler>
- Status DoParse(Handler& handler, const std::shared_ptr<Buffer>& json) {
- RETURN_NOT_OK(ReserveScalarStorage(json->size()));
- rj::MemoryStream ms(reinterpret_cast<const char*>(json->data()),
json->size());
- using InputStream = rj::EncodedInputStream<rj::UTF8<>, rj::MemoryStream>;
- return DoParse(handler, InputStream(ms),
static_cast<size_t>(json->size()));
+ Status ParseArray(Handler& handler, sj::value value) {
+ RETURN_NOT_OK(StartArrayImpl());
+
+ ARROW_ASSIGN_OR_RAISE(auto array, arrow::internal::ResolveSimdjsonResult(
+ value.get_array(), "Failed to get
JSON array"));
+
+ size_t size = 0;
+
+ for (auto element_result : array) {
+ ARROW_ASSIGN_OR_RAISE(auto element,
+ arrow::internal::ResolveSimdjsonResult(
+ element_result, "Failed to iterate JSON
array"));
+
+ RETURN_NOT_OK(ParseValue(handler, element));
+ ++size;
+ }
+
+ return EndArrayImpl(size);
+ }
+
+ template <typename Handler>
+ Status ParseObject(Handler& handler, sj::value value) {
+ RETURN_NOT_OK(StartObjectImpl());
Review Comment:
Similarly, `StartObjectImpl` and friends exist because RapidJSON would call
a handler method call `StartObject`. This is not really necessary anymore
because simdjson doesn't call anything from us.
##########
cpp/src/arrow/json/parser.cc:
##########
@@ -762,43 +741,199 @@ class HandlerBase : public BlockParser,
}
protected:
- template <typename Handler, typename Stream>
- Status DoParse(Handler& handler, Stream&& json, size_t json_size) {
- constexpr auto parse_flags = rj::kParseIterativeFlag |
rj::kParseNanAndInfFlag |
- rj::kParseStopWhenDoneFlag |
- rj::kParseNumbersAsStringsFlag;
-
- rj::Reader reader;
- // ensure that the loop can exit when the block too large.
- for (; num_rows_ < std::numeric_limits<int32_t>::max(); ++num_rows_) {
- auto ok = reader.Parse<parse_flags>(json, handler);
- switch (ok.Code()) {
- case rj::kParseErrorNone:
- // parse the next object
- continue;
- case rj::kParseErrorDocumentEmpty:
- if (json.Tell() < json_size) {
- return ParseError(rj::GetParseError_En(ok.Code()));
- }
- // parsed all objects, finish
- return Status::OK();
- case rj::kParseErrorTermination:
- // handler emitted an error
- return handler.Error();
- default:
- // rj emitted an error
- return ParseError(rj::GetParseError_En(ok.Code()), " in row ",
num_rows_);
+ template <typename Handler>
+ Status DoParse(Handler& handler, const std::shared_ptr<Buffer>& json) {
Review Comment:
The `Handler` pattern was mandated by RapidJSON's parser architecture, but
we probably don't need it anymore. We can probably simplify the code in this
file by removing this layer of indirection?
##########
cpp/src/arrow/json/parser.cc:
##########
@@ -762,43 +741,199 @@ class HandlerBase : public BlockParser,
}
protected:
- template <typename Handler, typename Stream>
- Status DoParse(Handler& handler, Stream&& json, size_t json_size) {
- constexpr auto parse_flags = rj::kParseIterativeFlag |
rj::kParseNanAndInfFlag |
- rj::kParseStopWhenDoneFlag |
- rj::kParseNumbersAsStringsFlag;
-
- rj::Reader reader;
- // ensure that the loop can exit when the block too large.
- for (; num_rows_ < std::numeric_limits<int32_t>::max(); ++num_rows_) {
- auto ok = reader.Parse<parse_flags>(json, handler);
- switch (ok.Code()) {
- case rj::kParseErrorNone:
- // parse the next object
- continue;
- case rj::kParseErrorDocumentEmpty:
- if (json.Tell() < json_size) {
- return ParseError(rj::GetParseError_En(ok.Code()));
- }
- // parsed all objects, finish
- return Status::OK();
- case rj::kParseErrorTermination:
- // handler emitted an error
- return handler.Error();
- default:
- // rj emitted an error
- return ParseError(rj::GetParseError_En(ok.Code()), " in row ",
num_rows_);
+ template <typename Handler>
+ Status DoParse(Handler& handler, const std::shared_ptr<Buffer>& json) {
+ RETURN_NOT_OK(ReserveScalarStorage(json->size()));
+
+ const std::string_view input(reinterpret_cast<const char*>(json->data()),
+ json->size());
+
+ if (IsWhitespaceOnly(input)) {
Review Comment:
Is this required? Would `parse` simply be able to parse the empty stream?
##########
cpp/cmake_modules/ThirdpartyToolchain.cmake:
##########
@@ -2833,11 +2833,7 @@ function(build_simdjson)
URL_HASH
"SHA256=${ARROW_SIMDJSON_BUILD_SHA256_CHECKSUM}")
prepare_fetchcontent()
-
- # simdjson enables precompiled headers unconditionally.
- # Recompiling simdjson.cpp against it produces differing artifacts
- # Disable precompiled headers to avoid reproducible build failures.
- set(CMAKE_DISABLE_PRECOMPILE_HEADERS ON)
+ set(SIMDJSON_ENABLE_THREADS ${ARROW_ENABLE_THREADING})
Review Comment:
Add a comment why we're enabling this?
##########
cpp/src/arrow/json/parser.cc:
##########
@@ -762,43 +741,199 @@ class HandlerBase : public BlockParser,
}
protected:
- template <typename Handler, typename Stream>
- Status DoParse(Handler& handler, Stream&& json, size_t json_size) {
- constexpr auto parse_flags = rj::kParseIterativeFlag |
rj::kParseNanAndInfFlag |
- rj::kParseStopWhenDoneFlag |
- rj::kParseNumbersAsStringsFlag;
-
- rj::Reader reader;
- // ensure that the loop can exit when the block too large.
- for (; num_rows_ < std::numeric_limits<int32_t>::max(); ++num_rows_) {
- auto ok = reader.Parse<parse_flags>(json, handler);
- switch (ok.Code()) {
- case rj::kParseErrorNone:
- // parse the next object
- continue;
- case rj::kParseErrorDocumentEmpty:
- if (json.Tell() < json_size) {
- return ParseError(rj::GetParseError_En(ok.Code()));
- }
- // parsed all objects, finish
- return Status::OK();
- case rj::kParseErrorTermination:
- // handler emitted an error
- return handler.Error();
- default:
- // rj emitted an error
- return ParseError(rj::GetParseError_En(ok.Code()), " in row ",
num_rows_);
+ template <typename Handler>
+ Status DoParse(Handler& handler, const std::shared_ptr<Buffer>& json) {
+ RETURN_NOT_OK(ReserveScalarStorage(json->size()));
+
+ const std::string_view input(reinterpret_cast<const char*>(json->data()),
+ json->size());
+
+ if (IsWhitespaceOnly(input)) {
+ return Status::OK();
+ }
+
+ auto parse = [&](const auto& input) -> Status {
+ ARROW_ASSIGN_OR_RAISE(auto stream,
arrow::internal::ResolveSimdjsonResult(
+ parser_.iterate_many(input),
+ "Failed to create JSON document
stream"));
+
+ for (auto document_result : stream) {
+ ARROW_ASSIGN_OR_RAISE(
+ auto document,
+ arrow::internal::ResolveSimdjsonResult(
+ document_result, "Failed to iterate JSON document stream"));
+
+ if (num_rows_ == std::numeric_limits<int32_t>::max()) {
+ return Status::Invalid("Row count overflowed int32_t");
+ }
+
+ ARROW_ASSIGN_OR_RAISE(
+ auto value,
+ arrow::internal::ResolveSimdjsonResult(
+ document.get_value(), "JSON parse error: Failed to get JSON
value"));
+
+ RETURN_NOT_OK(ParseValue(handler, value));
+
+ ++num_rows_;
+ }
+
+ if (stream.truncated_bytes() != 0) {
+ return ParseError("The document is empty");
+ }
+
+ return Status::OK();
+ };
+
+ if (json->capacity() - json->size() >=
+ static_cast<int64_t>(simdjson::SIMDJSON_PADDING)) {
+ const auto padded_json = simdjson::padded_string_view(
+ reinterpret_cast<const char*>(json->data()), json->size(),
json->capacity());
+ return parse(padded_json);
+ }
+
+ simdjson::padded_string padded_json(reinterpret_cast<const
char*>(json->data()),
+ json->size());
+ return parse(padded_json);
+ }
+
+ template <Kind::type kind>
+ Status MaybePromoteFromNull() {
+ if (builder_.kind != Kind::kNull) {
+ return Status::OK();
+ }
+
+ auto parent = builder_stack_.back();
+
+ if (parent.kind == Kind::kArray) {
+ auto list_builder = Cast<Kind::kArray>(parent);
+ DCHECK_EQ(list_builder->value_builder(), builder_);
+
+ RETURN_NOT_OK(builder_set_.MakeBuilder<kind>(builder_.index, &builder_));
+
+ list_builder = Cast<Kind::kArray>(parent);
+ list_builder->value_builder(builder_);
+ } else {
+ auto struct_builder = Cast<Kind::kObject>(parent);
+ DCHECK_EQ(struct_builder->field_builder(field_index_), builder_);
+
+ RETURN_NOT_OK(builder_set_.MakeBuilder<kind>(builder_.index, &builder_));
+
+ struct_builder = Cast<Kind::kObject>(parent);
+ struct_builder->field_builder(field_index_, builder_);
+ }
+
+ return Status::OK();
+ }
+
+ template <typename Handler>
+ Status ParseValue(Handler& handler, sj::value value) {
+ ARROW_ASSIGN_OR_RAISE(auto type, arrow::internal::ResolveSimdjsonResult(
+ value.type(), "Failed to determine
JSON type"));
+
+ switch (type) {
+ case sj::json_type::null: {
+ ARROW_ASSIGN_OR_RAISE([[maybe_unused]] auto is_null,
+ arrow::internal::ResolveSimdjsonResult(
+ value.is_null(), "Failed to validate JSON
null"));
+ return Null();
+ }
+
+ case sj::json_type::boolean: {
+ RETURN_NOT_OK(handler.template MaybePromoteFromNull<Kind::kBoolean>());
+
+ ARROW_ASSIGN_OR_RAISE(auto boolean,
+ arrow::internal::ResolveSimdjsonResult(
+ value.get_bool(), "Failed to get JSON
boolean"));
+ return Bool(boolean);
+ }
+
+ case sj::json_type::string: {
+ RETURN_NOT_OK(handler.template MaybePromoteFromNull<Kind::kString>());
+
+ ARROW_ASSIGN_OR_RAISE(auto string,
+ arrow::internal::ResolveSimdjsonResult(
+ value.get_string(), "Failed to get JSON
string"));
+ return String(string);
}
+
+ case sj::json_type::number: {
+ RETURN_NOT_OK(handler.template MaybePromoteFromNull<Kind::kNumber>());
+ return RawNumber(TrimTrailingWhitespace(value.raw_json_token()));
+ }
+
+ case sj::json_type::array:
+ RETURN_NOT_OK(handler.template MaybePromoteFromNull<Kind::kArray>());
+ return ParseArray(handler, value);
+
+ case sj::json_type::object:
+ RETURN_NOT_OK(handler.template MaybePromoteFromNull<Kind::kObject>());
+ return ParseObject(handler, value);
+
+ default:
+ return ParseError("Invalid value");
}
- return Status::Invalid("Row count overflowed int32_t");
}
template <typename Handler>
- Status DoParse(Handler& handler, const std::shared_ptr<Buffer>& json) {
- RETURN_NOT_OK(ReserveScalarStorage(json->size()));
- rj::MemoryStream ms(reinterpret_cast<const char*>(json->data()),
json->size());
- using InputStream = rj::EncodedInputStream<rj::UTF8<>, rj::MemoryStream>;
- return DoParse(handler, InputStream(ms),
static_cast<size_t>(json->size()));
+ Status ParseArray(Handler& handler, sj::value value) {
+ RETURN_NOT_OK(StartArrayImpl());
+
+ ARROW_ASSIGN_OR_RAISE(auto array, arrow::internal::ResolveSimdjsonResult(
+ value.get_array(), "Failed to get
JSON array"));
+
+ size_t size = 0;
+
+ for (auto element_result : array) {
+ ARROW_ASSIGN_OR_RAISE(auto element,
+ arrow::internal::ResolveSimdjsonResult(
+ element_result, "Failed to iterate JSON
array"));
+
+ RETURN_NOT_OK(ParseValue(handler, element));
+ ++size;
+ }
+
+ return EndArrayImpl(size);
+ }
+
+ template <typename Handler>
+ Status ParseObject(Handler& handler, sj::value value) {
+ RETURN_NOT_OK(StartObjectImpl());
+
+ ARROW_ASSIGN_OR_RAISE(
+ auto object, arrow::internal::ResolveSimdjsonResult(value.get_object(),
+ "Failed to get
JSON object"));
+
+ for (auto field_result : object) {
+ ARROW_ASSIGN_OR_RAISE(
+ auto field,
+ arrow::internal::ResolveSimdjsonResult(
+ field_result, "JSON parse error: Failed to iterate JSON
object"));
+
+ ARROW_ASSIGN_OR_RAISE(auto key,
Review Comment:
For the record, is `key` a `string` or a `string_view`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
