diff --git a/cpp/cmake_modules/ThirdpartyToolchain.cmake b/cpp/cmake_modules/ThirdpartyToolchain.cmake index 2b703a3e1ca3..90115f68244b 100644 --- a/cpp/cmake_modules/ThirdpartyToolchain.cmake +++ b/cpp/cmake_modules/ThirdpartyToolchain.cmake @@ -2837,6 +2837,9 @@ function(build_simdjson) prepare_fetchcontent() + # Keep simdjson's threading configuration consistent with Arrow's, + # which is required for Emscripten where Arrow threading is disabled. + set(SIMDJSON_ENABLE_THREADS ${ARROW_ENABLE_THREADING}) # simdjson enables precompiled headers unconditionally. # Recompiling simdjson.cpp against it produces differing artifacts # Disable precompiled headers to avoid reproducible build failures. diff --git a/cpp/src/arrow/json/chunker.cc b/cpp/src/arrow/json/chunker.cc index 4bff06e2ca5c..d073f9405408 100644 --- a/cpp/src/arrow/json/chunker.cc +++ b/cpp/src/arrow/json/chunker.cc @@ -32,16 +32,6 @@ namespace arrow { namespace json { namespace { -// 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 ConsumeWhitespace(std::string_view view) { - const auto ws_count = view.find_first_not_of(" \t\r\n"); - if (ws_count == std::string_view::npos) { - return view.size(); - } - return static_cast(ws_count); -} - // A BoundaryFinder implementation that assumes JSON objects can contain raw newlines, // and uses the structural indexes computed by simdjson to delimit them. class ParsingBoundaryFinder : public BoundaryFinder { @@ -151,7 +141,8 @@ class ParsingBoundaryFinder : public BoundaryFinder { if (consumed_length > 0) { // If we found at least one document, also consume its trailing whitespace // to avoid stray bytes at the end of the stream. - consumed_length += ConsumeWhitespace(input.substr(consumed_length)); + consumed_length += internal::ConsumeJsonWhitespace(input.substr(consumed_length), + /*trailing=*/false); } return consumed_length; } diff --git a/cpp/src/arrow/json/parser.cc b/cpp/src/arrow/json/parser.cc index 53f856d8012e..2ef339a19f74 100644 --- a/cpp/src/arrow/json/parser.cc +++ b/cpp/src/arrow/json/parser.cc @@ -27,10 +27,6 @@ #include #include -#include "arrow/json/rapidjson_defs.h" -#include "rapidjson/error/en.h" -#include "rapidjson/reader.h" - #include "arrow/array.h" #include "arrow/array/builder_binary.h" #include "arrow/buffer_builder.h" @@ -38,6 +34,7 @@ #include "arrow/util/bitset_stack_internal.h" #include "arrow/util/checked_cast.h" #include "arrow/util/logging_internal.h" +#include "arrow/util/simdjson_internal.h" #include "arrow/util/trie_internal.h" #include "arrow/visit_type_inline.h" @@ -48,7 +45,7 @@ using internal::checked_cast; namespace json { -namespace rj = arrow::rapidjson; +namespace sj = simdjson::ondemand; template static Status ParseError(T&&... t) { @@ -130,7 +127,7 @@ class RawArrayBuilder; /// \brief packed pointer to a RawArrayBuilder /// -/// RawArrayBuilders are stored in HandlerBase, +/// RawArrayBuilders are stored in ParseImpl, /// which allows storage of their indices (uint32_t) instead of a full pointer. /// BuilderPtr is also tagged with the json kind and nullable properties /// so those can be accessed before dereferencing the builder. @@ -644,14 +641,12 @@ 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, 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) {} @@ -662,71 +657,58 @@ class HandlerBase : public BlockParser, return builder_set_.Cast(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: { + // If an unexpected field is encountered, add a NullBuilder with leading nulls. + // The next value parsed will promote this field to its inferred type. + auto struct_builder = Cast(builder_stack_.back()); + auto leading_nulls = static_cast(struct_builder->length() - 1); + + builder_ = BuilderPtr(Kind::kNull, leading_nulls, /*nullable=*/true); + field_index_ = struct_builder->AddField(key, builder_); + + return ParseValue(value); + } + } + + return Status::OK(); } - bool Bool(bool value) { + Status Bool(bool value) { constexpr auto kind = Kind::kBoolean; if (ARROW_PREDICT_FALSE(builder_.kind != kind)) { - status_ = IllegallyChangedTo(kind); - return status_.ok(); + return IllegallyChangedTo(kind); } - status_ = Cast(builder_)->Append(value); - return status_.ok(); + return Cast(builder_)->Append(value); } - bool RawNumber(const char* data, rj::SizeType size, ...) { + Status RawNumber(std::string_view value) { if (builder_.kind == Kind::kNumberOrString) { - status_ = - AppendScalar(builder_, std::string_view(data, size)); + return AppendScalar(builder_, value); } else { - status_ = AppendScalar(builder_, std::string_view(data, size)); + return AppendScalar(builder_, value); } - return status_.ok(); } - bool String(const char* data, rj::SizeType size, ...) { + Status String(std::string_view value) { if (builder_.kind == Kind::kNumberOrString) { - status_ = - AppendScalar(builder_, std::string_view(data, size)); + return AppendScalar(builder_, value); } else { - status_ = AppendScalar(builder_, std::string_view(data, size)); + return AppendScalar(builder_, value); } - return status_.ok(); - } - - bool StartObject() { - status_ = StartObjectImpl(); - return status_.ok(); - } - - bool EndObject(...) { - status_ = EndObjectImpl(); - return status_.ok(); } - bool StartArray() { - status_ = StartArrayImpl(); - return status_.ok(); - } - - bool EndArray(rj::SizeType size) { - status_ = EndArrayImpl(size); - return status_.ok(); - } - /// @} - /// \brief Set up builders using an expected Schema Status Initialize(const std::shared_ptr& s) { auto type = struct_({}); @@ -762,102 +744,205 @@ class HandlerBase : public BlockParser, } protected: - template - 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::max(); ++num_rows_) { - auto ok = reader.Parse(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 Parse(const std::shared_ptr& json) override { + RETURN_NOT_OK(ReserveScalarStorage(json->size())); + + const std::string_view input(reinterpret_cast(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::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("JSON document was truncated"); } + + return Status::OK(); + }; + + if (json->capacity() - json->size() >= + static_cast(simdjson::SIMDJSON_PADDING)) { + const auto padded_json = simdjson::padded_string_view( + reinterpret_cast(json->data()), json->size(), json->capacity()); + return parse(padded_json); } - return Status::Invalid("Row count overflowed int32_t"); + + // TODO(GH-51463): Investigate allocating input buffers with enough capacity + // for SIMDJSON padding to avoid copying in this case. + simdjson::padded_string padded_json(reinterpret_cast(json->data()), + json->size()); + return parse(padded_json); } - template - Status DoParse(Handler& handler, const std::shared_ptr& json) { - RETURN_NOT_OK(ReserveScalarStorage(json->size())); - rj::MemoryStream ms(reinterpret_cast(json->data()), json->size()); - using InputStream = rj::EncodedInputStream, rj::MemoryStream>; - return DoParse(handler, InputStream(ms), static_cast(json->size())); + template + Status MaybePromoteFromNull() { + if (builder_.kind != Kind::kNull) { + return Status::OK(); + } + + auto parent = builder_stack_.back(); + + if (parent.kind == Kind::kArray) { + auto list_builder = Cast(parent); + DCHECK_EQ(list_builder->value_builder(), builder_); + + RETURN_NOT_OK(builder_set_.MakeBuilder(builder_.index, &builder_)); + + list_builder = Cast(parent); + list_builder->value_builder(builder_); + } else { + auto struct_builder = Cast(parent); + DCHECK_EQ(struct_builder->field_builder(field_index_), builder_); + + RETURN_NOT_OK(builder_set_.MakeBuilder(builder_.index, &builder_)); + + struct_builder = Cast(parent); + struct_builder->field_builder(field_index_, builder_); + } + + return Status::OK(); } - /// \defgroup handlerbase-append-methods append non-nested values - /// - /// @{ + Status ParseValue(sj::value value) { + ARROW_ASSIGN_OR_RAISE(auto type, arrow::internal::ResolveSimdjsonResult( + value.type(), "Failed to determine JSON type")); - template - Status AppendScalar(BuilderPtr builder, std::string_view scalar) { - if (ARROW_PREDICT_FALSE(builder.kind != kind)) { - return IllegallyChangedTo(kind); + 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()); + + 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()); + + 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()); + 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()); + return ParseArray(value); + + case sj::json_type::object: + RETURN_NOT_OK(MaybePromoteFromNull()); + return ParseObject(value); + + case sj::json_type::unknown: + return ParseError("Invalid JSON value"); } - auto index = static_cast(scalar_values_builder_.length()); - auto value_length = static_cast(scalar.size()); - RETURN_NOT_OK(Cast(builder)->Append(index, value_length)); - RETURN_NOT_OK(scalar_values_builder_.Reserve(1)); - scalar_values_builder_.UnsafeAppend(scalar); return Status::OK(); } - /// @} + Status ParseArray(sj::value value) { + constexpr auto kind = Kind::kArray; + if (ARROW_PREDICT_FALSE(builder_.kind != kind)) { + return IllegallyChangedTo(kind); + } + + StartNested(); + + builder_ = Cast(builder_)->value_builder(); + + ARROW_ASSIGN_OR_RAISE(auto array, arrow::internal::ResolveSimdjsonResult( + value.get_array(), "Failed to get JSON array")); + + int64_t size = 0; - Status StartObjectImpl() { + 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(builder_); + DCHECK_LE(size, std::numeric_limits::max()); + return list_builder->Append(static_cast(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(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(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(); @@ -876,25 +961,42 @@ class HandlerBase : public BlockParser, return Status::OK(); } - Status StartArrayImpl() { - constexpr auto kind = Kind::kArray; - if (ARROW_PREDICT_FALSE(builder_.kind != kind)) { - return IllegallyChangedTo(kind); + Status ParseObjectField(std::string_view key, sj::value value) { + auto parent = Cast(builder_stack_.back()); + field_index_ = parent->GetFieldIndex(key); + if (ARROW_PREDICT_FALSE(field_index_ == -1)) { + return HandleUnexpectedField(key, value); } - StartNested(); - // append to the list builder in EndArrayImpl - builder_ = Cast(builder_)->value_builder(); - return Status::OK(); + bool duplicate_keys; + 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) { + return ParseError("Column(", Path(), ") was specified twice in row ", num_rows_); + } + builder_ = parent->field_builder(field_index_); + absent_fields_stack_[field_index_] = false; + return ParseValue(value); } - Status EndArrayImpl(rj::SizeType size) { - EndNested(); - // append to list_builder here - auto list_builder = Cast(builder_); - return list_builder->Append(size); + template + Status AppendScalar(BuilderPtr builder, std::string_view scalar) { + if (ARROW_PREDICT_FALSE(builder.kind != kind)) { + return IllegallyChangedTo(kind); + } + auto index = static_cast(scalar_values_builder_.length()); + auto value_length = static_cast(scalar.size()); + RETURN_NOT_OK(Cast(builder)->Append(index, value_length)); + RETURN_NOT_OK(scalar_values_builder_.Reserve(1)); + scalar_values_builder_.UnsafeAppend(scalar); + return Status::OK(); } - /// helper method for StartArray and StartObject + /// helper method for ParseArray and ParseObject /// adds the current builder to a stack so its /// children can be visited and parsed. void StartNested() { @@ -928,7 +1030,7 @@ class HandlerBase : public BlockParser, return scalar_values_builder_.ReserveData(size - available_storage); } - Status status_; + UnexpectedFieldBehavior unexpected_field_behavior_; RawBuilderSet builder_set_; BuilderPtr builder_; // top of this stack is the parent of builder_ @@ -941,232 +1043,7 @@ class HandlerBase : public BlockParser, // top of this stack == field_index_ std::vector field_index_stack_; StringBuilder scalar_values_builder_; -}; - -template -class Handler; - -template <> -class Handler : public HandlerBase { - public: - using HandlerBase::HandlerBase; - - Status Parse(const std::shared_ptr& json) override { - return DoParse(*this, json); - } - - /// \ingroup rapidjson-handler-interface - /// - /// if an unexpected field is encountered, emit a parse error and bail - bool Key(const char* key, rj::SizeType len, ...) { - bool duplicate_keys = false; - if (ARROW_PREDICT_FALSE( - SetFieldBuilder(std::string_view(key, len), &duplicate_keys))) { - return true; - } - if (!duplicate_keys) { - status_ = ParseError("unexpected field"); - } - return false; - } -}; - -template <> -class Handler : public HandlerBase { - public: - using HandlerBase::HandlerBase; - - Status Parse(const std::shared_ptr& json) override { - return DoParse(*this, json); - } - - bool Null() { - if (Skipping()) { - return true; - } - return HandlerBase::Null(); - } - - bool Bool(bool value) { - if (Skipping()) { - return true; - } - return HandlerBase::Bool(value); - } - - bool RawNumber(const char* data, rj::SizeType size, ...) { - if (Skipping()) { - return true; - } - return HandlerBase::RawNumber(data, size); - } - - bool String(const char* data, rj::SizeType size, ...) { - if (Skipping()) { - return true; - } - return HandlerBase::String(data, size); - } - - bool StartObject() { - ++depth_; - if (Skipping()) { - return true; - } - return HandlerBase::StartObject(); - } - - /// \ingroup rapidjson-handler-interface - /// - /// if an unexpected field is encountered, skip until its value has been consumed - bool Key(const char* key, rj::SizeType len, ...) { - MaybeStopSkipping(); - if (Skipping()) { - return true; - } - bool duplicate_keys = false; - if (ARROW_PREDICT_TRUE( - SetFieldBuilder(std::string_view(key, len), &duplicate_keys))) { - return true; - } - if (ARROW_PREDICT_FALSE(duplicate_keys)) { - return false; - } - skip_depth_ = depth_; - return true; - } - - bool EndObject(...) { - MaybeStopSkipping(); - --depth_; - if (Skipping()) { - return true; - } - return HandlerBase::EndObject(); - } - - bool StartArray() { - if (Skipping()) { - return true; - } - return HandlerBase::StartArray(); - } - - bool EndArray(rj::SizeType size) { - if (Skipping()) { - return true; - } - return HandlerBase::EndArray(size); - } - - private: - bool Skipping() { return depth_ >= skip_depth_; } - - void MaybeStopSkipping() { - if (skip_depth_ == depth_) { - skip_depth_ = std::numeric_limits::max(); - } - } - - int depth_ = 0; - int skip_depth_ = std::numeric_limits::max(); -}; - -template <> -class Handler : public HandlerBase { - public: - using HandlerBase::HandlerBase; - - Status Parse(const std::shared_ptr& json) override { - return DoParse(*this, json); - } - - bool Bool(bool value) { - if (ARROW_PREDICT_FALSE(MaybePromoteFromNull())) { - return false; - } - return HandlerBase::Bool(value); - } - - bool RawNumber(const char* data, rj::SizeType size, ...) { - if (ARROW_PREDICT_FALSE(MaybePromoteFromNull())) { - return false; - } - return HandlerBase::RawNumber(data, size); - } - - bool String(const char* data, rj::SizeType size, ...) { - if (ARROW_PREDICT_FALSE(MaybePromoteFromNull())) { - return false; - } - return HandlerBase::String(data, size); - } - - bool StartObject() { - if (ARROW_PREDICT_FALSE(MaybePromoteFromNull())) { - return false; - } - return HandlerBase::StartObject(); - } - - /// \ingroup rapidjson-handler-interface - /// - /// If an unexpected field is encountered, add a new builder to - /// the current parent builder. It is added as a NullBuilder with - /// (parent.length - 1) leading nulls. The next value parsed - /// will probably trigger promotion of this field from null - bool Key(const char* key, rj::SizeType len, ...) { - bool duplicate_keys = false; - if (ARROW_PREDICT_TRUE( - SetFieldBuilder(std::string_view(key, len), &duplicate_keys))) { - return true; - } - if (ARROW_PREDICT_FALSE(duplicate_keys)) { - return false; - } - auto struct_builder = Cast(builder_stack_.back()); - auto leading_nulls = static_cast(struct_builder->length() - 1); - builder_ = BuilderPtr(Kind::kNull, leading_nulls, true); - field_index_ = struct_builder->AddField(std::string_view(key, len), builder_); - return true; - } - - bool StartArray() { - if (ARROW_PREDICT_FALSE(MaybePromoteFromNull())) { - return false; - } - return HandlerBase::StartArray(); - } - - private: - // return true if a terminal error was encountered - template - bool MaybePromoteFromNull() { - if (ARROW_PREDICT_TRUE(builder_.kind != Kind::kNull)) { - return false; - } - auto parent = builder_stack_.back(); - if (parent.kind == Kind::kArray) { - auto list_builder = Cast(parent); - DCHECK_EQ(list_builder->value_builder(), builder_); - status_ = builder_set_.MakeBuilder(builder_.index, &builder_); - if (ARROW_PREDICT_FALSE(!status_.ok())) { - return true; - } - list_builder = Cast(parent); - list_builder->value_builder(builder_); - } else { - auto struct_builder = Cast(parent); - DCHECK_EQ(struct_builder->field_builder(field_index_), builder_); - status_ = builder_set_.MakeBuilder(builder_.index, &builder_); - if (ARROW_PREDICT_FALSE(!status_.ok())) { - return true; - } - struct_builder = Cast(parent); - struct_builder->field_builder(field_index_, builder_); - } - return false; - } + sj::parser parser_; }; Status BlockParser::Make(MemoryPool* pool, const ParseOptions& options, @@ -1174,20 +1051,10 @@ Status BlockParser::Make(MemoryPool* pool, const ParseOptions& options, DCHECK(options.unexpected_field_behavior == UnexpectedFieldBehavior::InferType || options.explicit_schema != nullptr); - switch (options.unexpected_field_behavior) { - case UnexpectedFieldBehavior::Ignore: { - *out = std::make_unique>(pool); - break; - } - case UnexpectedFieldBehavior::Error: { - *out = std::make_unique>(pool); - break; - } - case UnexpectedFieldBehavior::InferType: - *out = std::make_unique>(pool); - break; - } - return static_cast(**out).Initialize(options.explicit_schema); + auto parser = std::make_unique(pool, options.unexpected_field_behavior); + RETURN_NOT_OK(parser->Initialize(options.explicit_schema)); + *out = std::move(parser); + return Status::OK(); } Status BlockParser::Make(const ParseOptions& options, std::unique_ptr* out) { diff --git a/cpp/src/arrow/json/parser_test.cc b/cpp/src/arrow/json/parser_test.cc index 1b107aa020fd..0b560043d442 100644 --- a/cpp/src/arrow/json/parser_test.cc +++ b/cpp/src/arrow/json/parser_test.cc @@ -235,6 +235,24 @@ TEST(BlockParserWithSchema, FailOnIncompleteJson) { ASSERT_RAISES(Invalid, ParseFromString(options, "{\"a\":0, \"b\"", &parsed)); } +TEST(BlockParserWithSchema, ValidateIgnoredFields) { + auto options = ParseOptions::Defaults(); + options.explicit_schema = schema({field("known", int64())}); + options.unexpected_field_behavior = UnexpectedFieldBehavior::Ignore; + + std::shared_ptr parsed; + // Ignored fields should still be validated for malformed JSON. + Status error = ParseFromString(options, R"({"known": 1, "ignored": [1,]})", &parsed); + ASSERT_RAISES(Invalid, error); + EXPECT_THAT(error.message(), testing::StartsWith("Invalid JSON value")); +} + +TEST(BlockParserWithSchema, NumberWithWhitespace) { + auto options = ParseOptions::Defaults(); + options.explicit_schema = schema({field("a", int64())}); + AssertParseColumns(options, R"({"a": 123 })", {field("a", utf8())}, {R"(["123"])"}); +} + TEST(BlockParser, Basics) { auto options = ParseOptions::Defaults(); options.unexpected_field_behavior = UnexpectedFieldBehavior::InferType; @@ -305,7 +323,7 @@ TEST(BlockParser, FailOnInvalidEOF) { auto status = ParseFromString(ParseOptions::Defaults(), "}", &parsed); ASSERT_RAISES(Invalid, status); EXPECT_THAT(status.message(), - ::testing::StartsWith("JSON parse error: The document is empty")); + ::testing::StartsWith("JSON parse error: JSON document was truncated")); } TEST(BlockParser, AdHoc) { diff --git a/cpp/src/arrow/json/reader_test.cc b/cpp/src/arrow/json/reader_test.cc index 549a49480579..bc533595614a 100644 --- a/cpp/src/arrow/json/reader_test.cc +++ b/cpp/src/arrow/json/reader_test.cc @@ -696,10 +696,10 @@ TEST_P(StreamingReaderTest, PropagateParsingErrors) { read_options_.block_size = 16; EXPECT_RAISES_WITH_MESSAGE_THAT( - Invalid, ::testing::StartsWith("Invalid: JSON parse error: Invalid value"), + Invalid, ::testing::StartsWith("Invalid: JSON parse error: Invalid JSON value"), MakeReader(bad_first_block)); EXPECT_RAISES_WITH_MESSAGE_THAT( - Invalid, ::testing::StartsWith("Invalid: JSON parse error: Invalid value"), + Invalid, ::testing::StartsWith("Invalid: JSON parse error: Invalid JSON value"), MakeReader(bad_first_block_after_empty)); std::shared_ptr batch; @@ -711,11 +711,9 @@ TEST_P(StreamingReaderTest, PropagateParsingErrors) { EXPECT_EQ(reader->bytes_processed(), 13); ASSERT_BATCHES_EQUAL(*RecordBatchFromJSON(test_schema, R"([{"n":10000}])"), *batch); - EXPECT_RAISES_WITH_MESSAGE_THAT( - Invalid, - ::testing::StartsWith( - "Invalid: JSON parse error: Missing a comma or '}' after an object member"), - reader->ReadNext(&batch)); + EXPECT_RAISES_WITH_MESSAGE_THAT(Invalid, + ::testing::StartsWith("Invalid: JSON parse error"), + reader->ReadNext(&batch)); EXPECT_EQ(reader->bytes_processed(), 13); AssertReadEnd(reader); EXPECT_EQ(reader->bytes_processed(), 13); @@ -1025,5 +1023,26 @@ TEST_F(AsyncStreamingReaderTest, StressSharedIoAndCpuExecutor) { AssertBatchSequenceEquals(expected.batches, batches); } +TEST(ReaderTest, FailOnMalformedNumbers) { + auto read_options = ReadOptions::Defaults(); + auto parse_options = ParseOptions::Defaults(); + + const std::vector malformed = { + R"({"a": 01})", + R"({"a": 1.})", + }; + + // Malformed numbers should be rejected + for (const bool use_threads : {false, true}) { + read_options.use_threads = use_threads; + + for (const auto& json : malformed) { + EXPECT_RAISES_WITH_MESSAGE_THAT( + Invalid, ::testing::StartsWith("Invalid: Failed to parse JSON number"), + ReadToTable(json, read_options, parse_options)); + } + } +} + } // namespace json } // namespace arrow diff --git a/cpp/src/arrow/util/simdjson_internal.cc b/cpp/src/arrow/util/simdjson_internal.cc index 146b48d9eb7f..d845eab0d473 100644 --- a/cpp/src/arrow/util/simdjson_internal.cc +++ b/cpp/src/arrow/util/simdjson_internal.cc @@ -581,4 +581,18 @@ Status ValidateJsonDocument(simdjson::ondemand::parser& parser, return ConsumeJsonValue(value); } +/// Returns the position of the first non-whitespace character when trailing is false, +/// or the number of trailing whitespace characters when trailing is true. +// XXX We could try to SIMD-accelerate this routine. +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(pos == std::string_view::npos ? view.size() : pos); + } + + const auto pos = view.find_last_not_of(" \t\r\n"); + return static_cast(pos == std::string_view::npos ? view.size() + : view.size() - pos - 1); +} + } // namespace arrow::internal diff --git a/cpp/src/arrow/util/simdjson_internal.h b/cpp/src/arrow/util/simdjson_internal.h index 5c52f7648c6a..d5816522bee0 100644 --- a/cpp/src/arrow/util/simdjson_internal.h +++ b/cpp/src/arrow/util/simdjson_internal.h @@ -266,7 +266,7 @@ Status VisitJsonValue(simdjson::ondemand::value value, ObjectFn&& object_fn, } case simdjson::ondemand::json_type::unknown: - return Status::Invalid("Unknown JSON type"); + return Status::Invalid("Invalid JSON value"); } return Status::Invalid("Unreachable"); @@ -350,4 +350,6 @@ ARROW_EXPORT Status ConsumeJsonValue(simdjson::ondemand::value value); ARROW_EXPORT Status ValidateJsonDocument(simdjson::ondemand::parser& parser, simdjson::padded_string& json); +ARROW_EXPORT int64_t ConsumeJsonWhitespace(std::string_view view, bool trailing); + } // namespace arrow::internal diff --git a/python/pyarrow/tests/test_json.py b/python/pyarrow/tests/test_json.py index ac2a027cfa24..e7ab574ead48 100644 --- a/python/pyarrow/tests/test_json.py +++ b/python/pyarrow/tests/test_json.py @@ -452,7 +452,7 @@ def test_bad_first_parse(self): read_options = ReadOptions() read_options.block_size = 16 with pytest.raises(pa.ArrowInvalid, - match="JSON parse error: Invalid value.*"): + match="JSON parse error: Invalid JSON value.*"): self.open_bytes(bad_first_block, read_options=read_options) def test_bad_middle_parse_after_empty(self): @@ -460,7 +460,7 @@ def test_bad_middle_parse_after_empty(self): read_options = ReadOptions() read_options.block_size = 16 with pytest.raises(pa.ArrowInvalid, - match="JSON parse error: Invalid value.*"): + match="JSON parse error: Invalid JSON value.*"): self.open_bytes(bad_first_block, read_options=read_options) def test_bad_middle_parse(self): @@ -476,8 +476,7 @@ def test_bad_middle_parse(self): } with pytest.raises( pa.ArrowInvalid, - match="JSON parse error:\ - Missing a comma or '}' after an object member*" + match="JSON parse error:" ): reader.read_next_batch()