pitrou commented on code in PR #51038:
URL: https://github.com/apache/arrow/pull/51038#discussion_r4069466259
##########
cpp/src/arrow/json/parser_test.cc:
##########
@@ -321,5 +321,15 @@ TEST(BlockParser, AdHoc) {
R"([{"c":true, "d": "1991-02-03"}, {"c":false, "d":"2019-04-01"}])"});
}
+TEST(BlockParserWithSchema, ValidateIgnoredFields) {
+ auto options = ParseOptions::Defaults();
+ options.explicit_schema = schema({field("known", int64())});
+ options.unexpected_field_behavior = UnexpectedFieldBehavior::Ignore;
+
+ std::shared_ptr<Array> parsed;
+ ASSERT_RAISES(Invalid,
+ ParseFromString(options, R"({"known": 1, "ignored": [1,]})",
&parsed));
Review Comment:
This fails because of the extraneous comma, is this right? Can we test
something about the error message and/or add a comment?
##########
cpp/src/arrow/json/reader_test.cc:
##########
@@ -1032,5 +1030,21 @@ TEST_F(AsyncStreamingReaderTest,
StressSharedIoAndCpuExecutor) {
AssertBatchSequenceEquals(expected.batches, batches);
}
+TEST(ReaderTest, FailOnMalformedNumbers) {
+ auto read_options = ReadOptions::Defaults();
+ auto parse_options = ParseOptions::Defaults();
+ read_options.use_threads = false;
Review Comment:
Should test both with and without threads, no?
##########
cpp/src/arrow/util/simdjson_internal.cc:
##########
@@ -581,4 +581,23 @@ Status ValidateJsonDocument(simdjson::ondemand::parser&
parser,
return ConsumeJsonValue(value);
}
+// XXX We could try to SIMD-accelerate this routine but it's called only
+// once per chunk and also will presumably examine a minimal amount of bytes.
+int64_t ConsumeJsonWhitespace(std::string_view view, bool trailing) {
+ if (!trailing) {
+ const auto pos = view.find_first_not_of(" \t\r\n");
+ return static_cast<int64_t>(pos == std::string_view::npos ? view.size() :
pos);
+ }
+
+ int64_t count = 0;
Review Comment:
Why not use `view.find_last_not_of`?
##########
cpp/src/arrow/util/simdjson_internal.cc:
##########
@@ -581,4 +581,23 @@ Status ValidateJsonDocument(simdjson::ondemand::parser&
parser,
return ConsumeJsonValue(value);
}
+// XXX We could try to SIMD-accelerate this routine but it's called only
+// once per chunk and also will presumably examine a minimal amount of bytes.
Review Comment:
This comment was in the context of the chunker. Is that still true in the
parser?
##########
cpp/src/arrow/json/reader_test.cc:
##########
@@ -1032,5 +1030,21 @@ TEST_F(AsyncStreamingReaderTest,
StressSharedIoAndCpuExecutor) {
AssertBatchSequenceEquals(expected.batches, batches);
}
+TEST(ReaderTest, FailOnMalformedNumbers) {
+ auto read_options = ReadOptions::Defaults();
+ auto parse_options = ParseOptions::Defaults();
+ read_options.use_threads = false;
+
+ const std::vector<std::string> malformed = {
+ R"({"a": 01})",
+ R"({"a": 1.})",
+ };
+
+ for (const auto& json : malformed) {
+ auto result = ReadToTable(json, read_options, parse_options);
+ EXPECT_TRUE(result.status().IsInvalid()) << result.status().ToString();
Review Comment:
Why not test the error message as well?
##########
cpp/src/arrow/json/parser.cc:
##########
@@ -762,150 +743,276 @@ 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_);
+ Status DoParse(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());
+
+ const int64_t input_size = input.size();
+ if (internal::ConsumeJsonWhitespace(input, /*trailing=*/false) ==
input_size) {
+ 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(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);
}
- 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()));
+ simdjson::padded_string padded_json(reinterpret_cast<const
char*>(json->data()),
+ json->size());
+ return parse(padded_json);
Review Comment:
Can you comment that this makes a copy and open an issue to potentially
improve this?
##########
cpp/src/arrow/json/parser.cc:
##########
@@ -762,150 +743,276 @@ 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_);
+ Status DoParse(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());
+
+ const int64_t input_size = input.size();
+ if (internal::ConsumeJsonWhitespace(input, /*trailing=*/false) ==
input_size) {
+ 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(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);
}
- 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()));
+ simdjson::padded_string padded_json(reinterpret_cast<const
char*>(json->data()),
+ json->size());
+ return parse(padded_json);
}
- /// \defgroup handlerbase-append-methods append non-nested values
- ///
- /// @{
-
template <Kind::type kind>
- Status AppendScalar(BuilderPtr builder, std::string_view scalar) {
- if (ARROW_PREDICT_FALSE(builder.kind != kind)) {
- return IllegallyChangedTo(kind);
+ Status MaybePromoteFromNull() {
+ if (builder_.kind != Kind::kNull) {
+ return Status::OK();
}
- auto index = static_cast<int32_t>(scalar_values_builder_.length());
- auto value_length = static_cast<int32_t>(scalar.size());
- RETURN_NOT_OK(Cast<kind>(builder)->Append(index, value_length));
- RETURN_NOT_OK(scalar_values_builder_.Reserve(1));
- scalar_values_builder_.UnsafeAppend(scalar);
+
+ 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();
}
- /// @}
+ Status ParseValue(sj::value value) {
+ ARROW_ASSIGN_OR_RAISE(auto type, arrow::internal::ResolveSimdjsonResult(
+ value.type(), "Failed to determine
JSON type"));
- Status StartObjectImpl() {
+ 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(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(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(MaybePromoteFromNull<Kind::kNumber>());
+ auto raw_number = value.raw_json_token();
+ RETURN_NOT_OK(arrow::internal::ResolveSimdjsonResult(
+ value.get_number(), "Failed to parse JSON number"));
+ raw_number.remove_suffix(
+ internal::ConsumeJsonWhitespace(raw_number, /*trailing=*/true));
+ return RawNumber(raw_number);
+ }
+
+ case sj::json_type::array:
+ RETURN_NOT_OK(MaybePromoteFromNull<Kind::kArray>());
+ return ParseArray(value);
+
+ case sj::json_type::object:
+ RETURN_NOT_OK(MaybePromoteFromNull<Kind::kObject>());
+ return ParseObject(value);
+
+ default:
+ return ParseError("Invalid value");
+ }
+ }
+
+ Status ParseArray(sj::value value) {
+ constexpr auto kind = Kind::kArray;
+ if (ARROW_PREDICT_FALSE(builder_.kind != kind)) {
+ return IllegallyChangedTo(kind);
+ }
+
+ StartNested();
+
+ builder_ = Cast<kind>(builder_)->value_builder();
+
+ ARROW_ASSIGN_OR_RAISE(auto array, arrow::internal::ResolveSimdjsonResult(
+ value.get_array(), "Failed to get
JSON array"));
+
+ size_t size = 0;
Review Comment:
Why not a `int32_t` or `int64_t`?
##########
cpp/src/arrow/json/parser.cc:
##########
@@ -762,150 +743,276 @@ 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_);
+ Status DoParse(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());
+
+ const int64_t input_size = input.size();
+ if (internal::ConsumeJsonWhitespace(input, /*trailing=*/false) ==
input_size) {
+ 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(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);
}
- 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()));
+ simdjson::padded_string padded_json(reinterpret_cast<const
char*>(json->data()),
+ json->size());
+ return parse(padded_json);
}
- /// \defgroup handlerbase-append-methods append non-nested values
- ///
- /// @{
-
template <Kind::type kind>
- Status AppendScalar(BuilderPtr builder, std::string_view scalar) {
- if (ARROW_PREDICT_FALSE(builder.kind != kind)) {
- return IllegallyChangedTo(kind);
+ Status MaybePromoteFromNull() {
+ if (builder_.kind != Kind::kNull) {
+ return Status::OK();
}
- auto index = static_cast<int32_t>(scalar_values_builder_.length());
- auto value_length = static_cast<int32_t>(scalar.size());
- RETURN_NOT_OK(Cast<kind>(builder)->Append(index, value_length));
- RETURN_NOT_OK(scalar_values_builder_.Reserve(1));
- scalar_values_builder_.UnsafeAppend(scalar);
+
+ 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();
}
- /// @}
+ Status ParseValue(sj::value value) {
+ ARROW_ASSIGN_OR_RAISE(auto type, arrow::internal::ResolveSimdjsonResult(
+ value.type(), "Failed to determine
JSON type"));
- Status StartObjectImpl() {
+ 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(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(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(MaybePromoteFromNull<Kind::kNumber>());
+ auto raw_number = value.raw_json_token();
+ RETURN_NOT_OK(arrow::internal::ResolveSimdjsonResult(
+ value.get_number(), "Failed to parse JSON number"));
Review Comment:
Why call `value.get_number` since we're not using the return value?
##########
cpp/src/arrow/json/parser.cc:
##########
@@ -762,150 +743,276 @@ 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_);
+ Status DoParse(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());
+
+ const int64_t input_size = input.size();
+ if (internal::ConsumeJsonWhitespace(input, /*trailing=*/false) ==
input_size) {
+ 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(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);
}
- 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()));
+ simdjson::padded_string padded_json(reinterpret_cast<const
char*>(json->data()),
+ json->size());
+ return parse(padded_json);
}
- /// \defgroup handlerbase-append-methods append non-nested values
- ///
- /// @{
-
template <Kind::type kind>
- Status AppendScalar(BuilderPtr builder, std::string_view scalar) {
- if (ARROW_PREDICT_FALSE(builder.kind != kind)) {
- return IllegallyChangedTo(kind);
+ Status MaybePromoteFromNull() {
+ if (builder_.kind != Kind::kNull) {
+ return Status::OK();
}
- auto index = static_cast<int32_t>(scalar_values_builder_.length());
- auto value_length = static_cast<int32_t>(scalar.size());
- RETURN_NOT_OK(Cast<kind>(builder)->Append(index, value_length));
- RETURN_NOT_OK(scalar_values_builder_.Reserve(1));
- scalar_values_builder_.UnsafeAppend(scalar);
+
+ 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();
}
- /// @}
+ Status ParseValue(sj::value value) {
+ ARROW_ASSIGN_OR_RAISE(auto type, arrow::internal::ResolveSimdjsonResult(
+ value.type(), "Failed to determine
JSON type"));
- Status StartObjectImpl() {
+ 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(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(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(MaybePromoteFromNull<Kind::kNumber>());
+ auto raw_number = value.raw_json_token();
+ RETURN_NOT_OK(arrow::internal::ResolveSimdjsonResult(
+ value.get_number(), "Failed to parse JSON number"));
+ raw_number.remove_suffix(
+ internal::ConsumeJsonWhitespace(raw_number, /*trailing=*/true));
+ return RawNumber(raw_number);
+ }
+
+ case sj::json_type::array:
+ RETURN_NOT_OK(MaybePromoteFromNull<Kind::kArray>());
+ return ParseArray(value);
+
+ case sj::json_type::object:
+ RETURN_NOT_OK(MaybePromoteFromNull<Kind::kObject>());
+ return ParseObject(value);
+
+ default:
+ return ParseError("Invalid value");
+ }
+ }
+
+ Status ParseArray(sj::value value) {
+ constexpr auto kind = Kind::kArray;
+ if (ARROW_PREDICT_FALSE(builder_.kind != kind)) {
+ return IllegallyChangedTo(kind);
+ }
+
+ StartNested();
+
+ builder_ = Cast<kind>(builder_)->value_builder();
+
+ 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(element));
+ ++size;
+ }
+
+ EndNested();
+
+ auto list_builder = Cast<Kind::kArray>(builder_);
+ DCHECK_LE(size, std::numeric_limits<int32_t>::max());
+ return list_builder->Append(static_cast<int32_t>(size));
+ }
+
+ Status ParseObject(sj::value value) {
constexpr auto kind = Kind::kObject;
if (ARROW_PREDICT_FALSE(builder_.kind != kind)) {
return IllegallyChangedTo(kind);
}
+
auto struct_builder = Cast<kind>(builder_);
absent_fields_stack_.Push(struct_builder->num_fields(), true);
StartNested();
- return struct_builder->Append();
- }
+ RETURN_NOT_OK(struct_builder->Append());
- /// \brief helper for Key() functions
- ///
- /// sets the field builder with name key, or returns false if
- /// there is no field with that name
- bool SetFieldBuilder(std::string_view key, bool* duplicate_keys) {
- auto parent = Cast<Kind::kObject>(builder_stack_.back());
- field_index_ = parent->GetFieldIndex(key);
- if (ARROW_PREDICT_FALSE(field_index_ == -1)) {
- return false;
- }
- if (field_index_ < absent_fields_stack_.TopSize()) {
- *duplicate_keys = !absent_fields_stack_[field_index_];
- } else {
- // When field_index is beyond the range of absent_fields_stack_ we have
a duplicated
- // field that wasn't declared in schema or previous records.
- *duplicate_keys = true;
- }
- if (*duplicate_keys) {
- status_ = ParseError("Column(", Path(), ") was specified twice in row ",
num_rows_);
- return false;
+ 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,
+ arrow::internal::ResolveSimdjsonResult(
+ field.unescaped_key(), "Failed to get JSON
object key"));
+
+ RETURN_NOT_OK(ParseObjectField(key, field.value()));
}
- builder_ = parent->field_builder(field_index_);
- absent_fields_stack_[field_index_] = false;
- return true;
- }
- Status EndObjectImpl() {
auto parent = builder_stack_.back();
-
auto expected_count = absent_fields_stack_.TopSize();
+
Review Comment:
Why all the added empty lines?
##########
cpp/src/arrow/json/parser.cc:
##########
@@ -762,150 +743,276 @@ 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_);
+ Status DoParse(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());
+
+ const int64_t input_size = input.size();
+ if (internal::ConsumeJsonWhitespace(input, /*trailing=*/false) ==
input_size) {
+ 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(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);
}
- 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()));
+ simdjson::padded_string padded_json(reinterpret_cast<const
char*>(json->data()),
+ json->size());
+ return parse(padded_json);
}
- /// \defgroup handlerbase-append-methods append non-nested values
- ///
- /// @{
-
template <Kind::type kind>
- Status AppendScalar(BuilderPtr builder, std::string_view scalar) {
- if (ARROW_PREDICT_FALSE(builder.kind != kind)) {
- return IllegallyChangedTo(kind);
+ Status MaybePromoteFromNull() {
+ if (builder_.kind != Kind::kNull) {
+ return Status::OK();
}
- auto index = static_cast<int32_t>(scalar_values_builder_.length());
- auto value_length = static_cast<int32_t>(scalar.size());
- RETURN_NOT_OK(Cast<kind>(builder)->Append(index, value_length));
- RETURN_NOT_OK(scalar_values_builder_.Reserve(1));
- scalar_values_builder_.UnsafeAppend(scalar);
+
+ 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();
}
- /// @}
+ Status ParseValue(sj::value value) {
+ ARROW_ASSIGN_OR_RAISE(auto type, arrow::internal::ResolveSimdjsonResult(
+ value.type(), "Failed to determine
JSON type"));
- Status StartObjectImpl() {
+ 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(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(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(MaybePromoteFromNull<Kind::kNumber>());
+ auto raw_number = value.raw_json_token();
+ RETURN_NOT_OK(arrow::internal::ResolveSimdjsonResult(
+ value.get_number(), "Failed to parse JSON number"));
+ raw_number.remove_suffix(
+ internal::ConsumeJsonWhitespace(raw_number, /*trailing=*/true));
+ return RawNumber(raw_number);
+ }
+
+ case sj::json_type::array:
+ RETURN_NOT_OK(MaybePromoteFromNull<Kind::kArray>());
+ return ParseArray(value);
+
+ case sj::json_type::object:
+ RETURN_NOT_OK(MaybePromoteFromNull<Kind::kObject>());
+ return ParseObject(value);
+
+ default:
+ return ParseError("Invalid value");
+ }
+ }
+
+ Status ParseArray(sj::value value) {
+ constexpr auto kind = Kind::kArray;
+ if (ARROW_PREDICT_FALSE(builder_.kind != kind)) {
+ return IllegallyChangedTo(kind);
+ }
+
+ StartNested();
+
+ builder_ = Cast<kind>(builder_)->value_builder();
+
+ 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(element));
+ ++size;
+ }
+
+ EndNested();
+
+ auto list_builder = Cast<Kind::kArray>(builder_);
+ DCHECK_LE(size, std::numeric_limits<int32_t>::max());
+ return list_builder->Append(static_cast<int32_t>(size));
+ }
+
+ Status ParseObject(sj::value value) {
constexpr auto kind = Kind::kObject;
if (ARROW_PREDICT_FALSE(builder_.kind != kind)) {
return IllegallyChangedTo(kind);
}
+
auto struct_builder = Cast<kind>(builder_);
absent_fields_stack_.Push(struct_builder->num_fields(), true);
StartNested();
- return struct_builder->Append();
- }
+ RETURN_NOT_OK(struct_builder->Append());
- /// \brief helper for Key() functions
- ///
- /// sets the field builder with name key, or returns false if
- /// there is no field with that name
- bool SetFieldBuilder(std::string_view key, bool* duplicate_keys) {
- auto parent = Cast<Kind::kObject>(builder_stack_.back());
- field_index_ = parent->GetFieldIndex(key);
- if (ARROW_PREDICT_FALSE(field_index_ == -1)) {
- return false;
- }
- if (field_index_ < absent_fields_stack_.TopSize()) {
- *duplicate_keys = !absent_fields_stack_[field_index_];
- } else {
- // When field_index is beyond the range of absent_fields_stack_ we have
a duplicated
- // field that wasn't declared in schema or previous records.
- *duplicate_keys = true;
- }
- if (*duplicate_keys) {
- status_ = ParseError("Column(", Path(), ") was specified twice in row ",
num_rows_);
- return false;
+ 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,
+ arrow::internal::ResolveSimdjsonResult(
+ field.unescaped_key(), "Failed to get JSON
object key"));
+
+ RETURN_NOT_OK(ParseObjectField(key, field.value()));
}
- builder_ = parent->field_builder(field_index_);
- absent_fields_stack_[field_index_] = false;
- return true;
- }
- Status EndObjectImpl() {
auto parent = builder_stack_.back();
-
auto expected_count = absent_fields_stack_.TopSize();
+
for (int i = 0; i < expected_count; ++i) {
if (!absent_fields_stack_[i]) {
continue;
}
+
auto field_builder = Cast<Kind::kObject>(parent)->field_builder(i);
if (ARROW_PREDICT_FALSE(!field_builder.nullable)) {
return ParseError("a required field was absent");
}
+
RETURN_NOT_OK(builder_set_.AppendNull(parent, i, field_builder));
}
+
absent_fields_stack_.Pop();
EndNested();
return Status::OK();
}
- Status StartArrayImpl() {
- constexpr auto kind = Kind::kArray;
- if (ARROW_PREDICT_FALSE(builder_.kind != kind)) {
+ Status ParseObjectField(std::string_view key, sj::value value) {
+ bool duplicate_keys = false;
+
+ if (SetFieldBuilder(key, &duplicate_keys)) {
+ return ParseValue(value);
+ }
+
+ if (duplicate_keys) {
+ return status_;
+ }
+
+ return HandleUnexpectedField(key, value);
+ }
+
+ template <Kind::type kind>
+ Status AppendScalar(BuilderPtr builder, std::string_view scalar) {
+ if (ARROW_PREDICT_FALSE(builder.kind != kind)) {
return IllegallyChangedTo(kind);
}
- StartNested();
- // append to the list builder in EndArrayImpl
- builder_ = Cast<kind>(builder_)->value_builder();
+ auto index = static_cast<int32_t>(scalar_values_builder_.length());
+ auto value_length = static_cast<int32_t>(scalar.size());
+ RETURN_NOT_OK(Cast<kind>(builder)->Append(index, value_length));
+ RETURN_NOT_OK(scalar_values_builder_.Reserve(1));
+ scalar_values_builder_.UnsafeAppend(scalar);
return Status::OK();
}
- Status EndArrayImpl(rj::SizeType size) {
- EndNested();
- // append to list_builder here
- auto list_builder = Cast<Kind::kArray>(builder_);
- return list_builder->Append(size);
+ bool SetFieldBuilder(std::string_view key, bool* duplicate_keys) {
+ auto parent = Cast<Kind::kObject>(builder_stack_.back());
+ field_index_ = parent->GetFieldIndex(key);
+ if (ARROW_PREDICT_FALSE(field_index_ == -1)) {
+ return false;
+ }
+ if (field_index_ < absent_fields_stack_.TopSize()) {
+ *duplicate_keys = !absent_fields_stack_[field_index_];
+ } else {
+ *duplicate_keys = true;
Review Comment:
You also removed a previous comment here...
##########
cpp/src/arrow/json/parser.cc:
##########
@@ -736,13 +716,14 @@ class HandlerBase : public BlockParser,
return builder_set_.MakeBuilder(*type, 0, &builder_);
}
+ Status Parse(const std::shared_ptr<Buffer>& json) override { return
DoParse(json); }
Review Comment:
Why have both `Parse` and `DoParse`?
##########
cpp/src/arrow/json/parser.cc:
##########
@@ -644,90 +641,73 @@ class RawBuilderSet {
arenas_;
};
-/// Three implementations are provided for BlockParser, one for each
-/// UnexpectedFieldBehavior. However most of the logic is identical in each
-/// case, so the majority of the implementation is in this base class
-class HandlerBase : public BlockParser,
- public rj::BaseReaderHandler<rj::UTF8<>, HandlerBase> {
+/// Parser implementation for BlockParser.
+class ParseImpl : public BlockParser {
public:
- explicit HandlerBase(MemoryPool* pool)
+ explicit ParseImpl(MemoryPool* pool, UnexpectedFieldBehavior
unexpected_field_behavior)
: BlockParser(pool),
+ unexpected_field_behavior_(unexpected_field_behavior),
builder_set_(pool),
field_index_(-1),
scalar_values_builder_(pool) {}
- /// Retrieve a pointer to a builder from a BuilderPtr
template <Kind::type kind>
enable_if_t<kind != Kind::kNull, RawArrayBuilder<kind>*> Cast(BuilderPtr
builder) {
return builder_set_.Cast<kind>(builder);
}
- /// Accessor for a stored error Status
Status Error() { return status_; }
- /// \defgroup rapidjson-handler-interface functions expected by rj::Reader
- ///
- /// bool Key(const char* data, rj::SizeType size, ...) is omitted since
- /// the behavior varies greatly between UnexpectedFieldBehaviors
- ///
- /// @{
- bool Null() {
- status_ = builder_set_.AppendNull(builder_stack_.back(), field_index_,
builder_);
- return status_.ok();
+ Status Null() {
+ return builder_set_.AppendNull(builder_stack_.back(), field_index_,
builder_);
+ }
+
+ Status HandleUnexpectedField(std::string_view key, sj::value value) {
+ switch (unexpected_field_behavior_) {
+ case UnexpectedFieldBehavior::Error:
+ return ParseError("unexpected field");
+
+ case UnexpectedFieldBehavior::Ignore:
+ return internal::ConsumeJsonValue(value);
+
+ case UnexpectedFieldBehavior::InferType: {
+ auto struct_builder = Cast<Kind::kObject>(builder_stack_.back());
+ auto leading_nulls = static_cast<uint32_t>(struct_builder->length() -
1);
+
+ builder_ = BuilderPtr(Kind::kNull, leading_nulls, true);
Review Comment:
What is `true` for? Can we make the parameter explicit?
```suggestion
builder_ = BuilderPtr(Kind::kNull, leading_nulls, /*nullable*/ true);
```
##########
cpp/src/arrow/json/parser.cc:
##########
@@ -644,90 +641,73 @@ class RawBuilderSet {
arenas_;
};
-/// Three implementations are provided for BlockParser, one for each
-/// UnexpectedFieldBehavior. However most of the logic is identical in each
-/// case, so the majority of the implementation is in this base class
-class HandlerBase : public BlockParser,
- public rj::BaseReaderHandler<rj::UTF8<>, HandlerBase> {
+/// Parser implementation for BlockParser.
+class ParseImpl : public BlockParser {
public:
- explicit HandlerBase(MemoryPool* pool)
+ explicit ParseImpl(MemoryPool* pool, UnexpectedFieldBehavior
unexpected_field_behavior)
: BlockParser(pool),
+ unexpected_field_behavior_(unexpected_field_behavior),
builder_set_(pool),
field_index_(-1),
scalar_values_builder_(pool) {}
- /// Retrieve a pointer to a builder from a BuilderPtr
template <Kind::type kind>
enable_if_t<kind != Kind::kNull, RawArrayBuilder<kind>*> Cast(BuilderPtr
builder) {
return builder_set_.Cast<kind>(builder);
}
- /// Accessor for a stored error Status
Status Error() { return status_; }
Review Comment:
Is this still useful?
##########
cpp/src/arrow/json/parser.cc:
##########
@@ -918,7 +1025,6 @@ class HandlerBase : public BlockParser,
" to ", Kind::Name(illegally_changed_to), " in row ",
num_rows_);
}
- /// Reserve storage for scalars, these can occupy almost all of the JSON
buffer
Review Comment:
Why remove those comments?
##########
cpp/src/arrow/json/parser.cc:
##########
@@ -928,266 +1034,27 @@ class HandlerBase : public BlockParser,
return scalar_values_builder_.ReserveData(size - available_storage);
}
+ UnexpectedFieldBehavior unexpected_field_behavior_;
Status status_;
RawBuilderSet builder_set_;
BuilderPtr builder_;
- // top of this stack is the parent of builder_
Review Comment:
Why remove those comments? Please stop this habit as you've been doing that
on several PRs.
##########
cpp/src/arrow/json/parser.cc:
##########
@@ -762,150 +743,276 @@ 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_);
+ Status DoParse(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());
+
+ const int64_t input_size = input.size();
+ if (internal::ConsumeJsonWhitespace(input, /*trailing=*/false) ==
input_size) {
+ 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(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);
}
- 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()));
+ simdjson::padded_string padded_json(reinterpret_cast<const
char*>(json->data()),
+ json->size());
+ return parse(padded_json);
}
- /// \defgroup handlerbase-append-methods append non-nested values
- ///
- /// @{
-
template <Kind::type kind>
- Status AppendScalar(BuilderPtr builder, std::string_view scalar) {
- if (ARROW_PREDICT_FALSE(builder.kind != kind)) {
- return IllegallyChangedTo(kind);
+ Status MaybePromoteFromNull() {
+ if (builder_.kind != Kind::kNull) {
+ return Status::OK();
}
- auto index = static_cast<int32_t>(scalar_values_builder_.length());
- auto value_length = static_cast<int32_t>(scalar.size());
- RETURN_NOT_OK(Cast<kind>(builder)->Append(index, value_length));
- RETURN_NOT_OK(scalar_values_builder_.Reserve(1));
- scalar_values_builder_.UnsafeAppend(scalar);
+
+ 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();
}
- /// @}
+ Status ParseValue(sj::value value) {
+ ARROW_ASSIGN_OR_RAISE(auto type, arrow::internal::ResolveSimdjsonResult(
+ value.type(), "Failed to determine
JSON type"));
- Status StartObjectImpl() {
+ 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(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(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(MaybePromoteFromNull<Kind::kNumber>());
+ auto raw_number = value.raw_json_token();
+ RETURN_NOT_OK(arrow::internal::ResolveSimdjsonResult(
+ value.get_number(), "Failed to parse JSON number"));
+ raw_number.remove_suffix(
+ internal::ConsumeJsonWhitespace(raw_number, /*trailing=*/true));
+ return RawNumber(raw_number);
+ }
+
+ case sj::json_type::array:
+ RETURN_NOT_OK(MaybePromoteFromNull<Kind::kArray>());
+ return ParseArray(value);
+
+ case sj::json_type::object:
+ RETURN_NOT_OK(MaybePromoteFromNull<Kind::kObject>());
+ return ParseObject(value);
+
+ default:
+ return ParseError("Invalid value");
Review Comment:
Can this legitimately happen or is this some internal error?
--
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]
