diff --git a/cpp/cmake_modules/ThirdpartyToolchain.cmake b/cpp/cmake_modules/ThirdpartyToolchain.cmake index 56f781bb7cf6..b65a13c8a0f6 100644 --- a/cpp/cmake_modules/ThirdpartyToolchain.cmake +++ b/cpp/cmake_modules/ThirdpartyToolchain.cmake @@ -2833,6 +2833,7 @@ function(build_simdjson) URL_HASH "SHA256=${ARROW_SIMDJSON_BUILD_SHA256_CHECKSUM}") prepare_fetchcontent() + set(SIMDJSON_ENABLE_THREADS ${ARROW_ENABLE_THREADING}) fetchcontent_makeavailable(simdjson) diff --git a/cpp/src/arrow/json/parser.cc b/cpp/src/arrow/json/parser.cc index 53f856d8012e..b83590997309 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,13 +45,30 @@ using internal::checked_cast; namespace json { -namespace rj = arrow::rapidjson; +namespace sj = simdjson::ondemand; template static Status ParseError(T&&... t) { return Status::Invalid("JSON parse error: ", std::forward(t)...); } +static std::string_view TrimTrailingWhitespace(std::string_view value) { + while (!value.empty() && (value.back() == ' ' || value.back() == '\t' || + value.back() == '\n' || value.back() == '\r')) { + value.remove_suffix(1); + } + return value; +} + +static bool IsWhitespaceOnly(std::string_view value) { + for (const auto c : value) { + if (c != ' ' && c != '\t' && c != '\n' && c != '\r') { + return false; + } + } + return true; +} + const std::string& Kind::Name(Kind::type kind) { static const std::string names[] = { "null", "boolean", "number", "string", "array", "object", "number_or_string", @@ -647,8 +661,7 @@ class RawBuilderSet { /// 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> { +class HandlerBase : public BlockParser { public: explicit HandlerBase(MemoryPool* pool) : BlockParser(pool), @@ -665,68 +678,34 @@ class HandlerBase : public BlockParser, /// 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_); } - 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,43 +741,199 @@ 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_); + template + Status DoParse(Handler& handler, const std::shared_ptr& json) { + RETURN_NOT_OK(ReserveScalarStorage(json->size())); + + const std::string_view input(reinterpret_cast(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::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(simdjson::SIMDJSON_PADDING)) { + const auto padded_json = simdjson::padded_string_view( + reinterpret_cast(json->data()), json->size(), json->capacity()); + return parse(padded_json); + } + + simdjson::padded_string padded_json(reinterpret_cast(json->data()), + json->size()); + return parse(padded_json); + } + + 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(); + } + + template + 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()); + + 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()); + + 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()); + return RawNumber(TrimTrailingWhitespace(value.raw_json_token())); + } + + case sj::json_type::array: + RETURN_NOT_OK(handler.template MaybePromoteFromNull()); + return ParseArray(handler, value); + + case sj::json_type::object: + RETURN_NOT_OK(handler.template MaybePromoteFromNull()); + return ParseObject(handler, value); + + default: + return ParseError("Invalid value"); } - return Status::Invalid("Row count overflowed int32_t"); } 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())); + 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 + 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, + arrow::internal::ResolveSimdjsonResult( + field.unescaped_key(), "Failed to get JSON object key")); + + auto field_value = field.value(); + + RETURN_NOT_OK(ParseObjectField(handler, key, field_value)); + } + + return EndObjectImpl(); + } + + template + Status ParseObjectField(Handler& handler, std::string_view key, sj::value value) { + bool duplicate_keys = false; + + if (SetFieldBuilder(key, &duplicate_keys)) { + return ParseValue(handler, value); + } + + if (duplicate_keys) { + return status_; + } + + return handler.HandleUnexpectedField(key, value); } /// \defgroup handlerbase-append-methods append non-nested values @@ -887,11 +1022,12 @@ class HandlerBase : public BlockParser, return Status::OK(); } - Status EndArrayImpl(rj::SizeType size) { + Status EndArrayImpl(size_t size) { EndNested(); // append to list_builder here auto list_builder = Cast(builder_); - return list_builder->Append(size); + DCHECK_LE(size, std::numeric_limits::max()); + return list_builder->Append(static_cast(size)); } /// helper method for StartArray and StartObject @@ -941,6 +1077,7 @@ class HandlerBase : public BlockParser, // top of this stack == field_index_ std::vector field_index_stack_; StringBuilder scalar_values_builder_; + sj::parser parser_; }; template @@ -955,19 +1092,8 @@ class Handler : public HandlerBase { 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; + Status HandleUnexpectedField(std::string_view, sj::value) { + return ParseError("unexpected field"); } }; @@ -980,96 +1106,7 @@ class Handler : public HandlerBase { 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(); + Status HandleUnexpectedField(std::string_view, sj::value) { return Status::OK(); } }; template <> @@ -1081,91 +1118,14 @@ class Handler : public HandlerBase { 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; - } + Status HandleUnexpectedField(std::string_view key, sj::value value) { 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(); - } + builder_ = BuilderPtr(Kind::kNull, leading_nulls, true); + field_index_ = struct_builder->AddField(key, builder_); - 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; + return ParseValue(*this, value); } }; diff --git a/cpp/src/arrow/json/reader_test.cc b/cpp/src/arrow/json/reader_test.cc index 7dcd30a0eb0b..ffe33c00111a 100644 --- a/cpp/src/arrow/json/reader_test.cc +++ b/cpp/src/arrow/json/reader_test.cc @@ -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); diff --git a/python/pyarrow/tests/test_json.py b/python/pyarrow/tests/test_json.py index ac2a027cfa24..8a8237cbc24c 100644 --- a/python/pyarrow/tests/test_json.py +++ b/python/pyarrow/tests/test_json.py @@ -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()