GH-50936: [C++][Integration] Replace RapidJSON with simdjson (#50937)

### Rationale for this change

This PR continues the simdjson migration by replacing the RapidJSON usage in the C++ JSON integration implementation with **simdjson's DOM API** and Arrow's existing `JsonWriter`.

### Changes

- Replace RapidJSON DOM parsing with simdjson's DOM API.
- Update the integration JSON reader and internal parsing helpers to use `simdjson::dom` types.
- Replace RapidJSON JSON serialization with Arrow's existing `JsonWriter`.
- Handle optional JSON array members without creating an invalid simdjson array.
- Remove the unused RapidJSON dependencies from the CMake and Meson integration test targets.
- Update remaining RapidJSON-specific implementation references.

Fixes: #50936
* GitHub Issue: #50936

Authored-by: Aaditya Srinivasan <aadityasri03@gmail.com>
Signed-off-by: Antoine Pitrou <antoine@python.org>

Author: Reranko05

cpp/src/arrow/integration/CMakeLists.txt

@@ -21,16 +21,12 @@ arrow_install_all_headers("arrow/integration")
 # - an executable that can be called to answer integration test requests
 # - a self-(unit)test for the C++ side of integration testing
 if(ARROW_BUILD_TESTS)
-  add_arrow_test(json_integration_test
-                 EXTRA_LINK_LIBS
-                 RapidJSON
-                 simdjson::simdjson
+  add_arrow_test(json_integration_test EXTRA_LINK_LIBS simdjson::simdjson
                  ${GFLAGS_LIBRARIES})
   add_dependencies(arrow-integration arrow-json-integration-test)
 elseif(ARROW_BUILD_INTEGRATION)
   add_executable(arrow-json-integration-test json_integration_test.cc)
   target_link_libraries(arrow-json-integration-test
-                        RapidJSON
                         simdjson::simdjson
                         ${ARROW_TEST_LINK_LIBS}
                         ${GFLAGS_LIBRARIES}

cpp/src/arrow/integration/json_integration.cc

@@ -21,7 +21,9 @@
 #include <cstdint>
 #include <memory>
 #include <string>
+#include <string_view>
 #include <utility>
+#include <vector>
 
 #include "arrow/buffer.h"
 #include "arrow/integration/json_internal.h"
@@ -33,6 +35,7 @@
 #include "arrow/status.h"
 #include "arrow/type.h"
 #include "arrow/util/logging_internal.h"
+#include "arrow/util/simdjson_internal.h"
 
 using arrow::ipc::DictionaryFieldMapper;
 using arrow::ipc::DictionaryMemo;
@@ -125,44 +128,46 @@ Status IntegrationJsonWriter::WriteRecordBatch(const RecordBatch& batch) {
 class IntegrationJsonReader::Impl {
  public:
   Impl(MemoryPool* pool, const std::shared_ptr<Buffer>& data)
-      : pool_(pool), data_(data), record_batches_(nullptr) {}
+      : pool_(pool), data_(data) {}
 
   Status ParseAndReadSchema() {
-    doc_.Parse(reinterpret_cast<const rj::Document::Ch*>(data_->data()),
-               static_cast<size_t>(data_->size()));
-    if (doc_.HasParseError()) {
-      return Status::IOError("JSON parsing failed");
-    }
+    ARROW_ASSIGN_OR_RAISE(doc_,
+                          internal::ResolveSimdjsonResult(
+                              parser_.parse(reinterpret_cast<const char*>(data_->data()),
+                                            static_cast<size_t>(data_->size())),
+                              "Failed to parse JSON"));
 
     ARROW_ASSIGN_OR_RAISE(schema_, json::ReadSchema(doc_, pool_, &dictionary_memo_));
 
-    auto it = std::as_const(doc_).FindMember("batches");
-    RETURN_NOT_ARRAY("batches", it, doc_);
-    record_batches_ = &it->value;
+    ARROW_ASSIGN_OR_RAISE(auto batches,
+                          internal::ResolveSimdjsonResult(doc_["batches"].get_array(),
+                                                          "Failed to get batches"));
+
+    batches.get_values(record_batches_);
 
     return Status::OK();
   }
 
   Result<std::shared_ptr<RecordBatch>> ReadRecordBatch(int i) {
-    if (i < 0 || i >= static_cast<int>(record_batches_->GetArray().Size())) {
+    if (i < 0 || i >= static_cast<int>(record_batches_.size())) {
       return Status::IndexError("record batch index ", i, " out of bounds");
     }
-    return json::ReadRecordBatch(record_batches_->GetArray()[i], schema_,
-                                 &dictionary_memo_, pool_);
+
+    return json::ReadRecordBatch(record_batches_[i], schema_, &dictionary_memo_, pool_);
   }
 
   std::shared_ptr<Schema> schema() const { return schema_; }
 
-  int num_record_batches() const {
-    return static_cast<int>(record_batches_->GetArray().Size());
-  }
+  int num_record_batches() const { return static_cast<int>(record_batches_.size()); }
 
  private:
   MemoryPool* pool_;
   std::shared_ptr<Buffer> data_;
-  rj::Document doc_;
 
-  const rj::Value* record_batches_;
+  simdjson::dom::parser parser_;
+  JsonValue doc_;
+  std::vector<JsonValue> record_batches_;
+
   std::shared_ptr<Schema> schema_;
   DictionaryMemo dictionary_memo_;
 };

cpp/src/arrow/integration/json_integration.h

@@ -55,7 +55,7 @@ class ARROW_EXPORT IntegrationJsonWriter {
  private:
   explicit IntegrationJsonWriter(const std::shared_ptr<Schema>& schema);
 
-  // Hide RapidJSON details from public API
+  // Hide JSON implementation details from public API
   class Impl;
   std::unique_ptr<Impl> impl_;
 };
@@ -106,7 +106,7 @@ class ARROW_EXPORT IntegrationJsonReader {
  private:
   IntegrationJsonReader(MemoryPool* pool, std::shared_ptr<Buffer> data);
 
-  // Hide RapidJSON details from public API
+  // Hide JSON implementation details from public API
   class Impl;
   std::unique_ptr<Impl> impl_;
 };

cpp/src/arrow/integration/json_integration_test.cc

@@ -49,6 +49,7 @@
 #include "arrow/type.h"
 #include "arrow/type_fwd.h"
 #include "arrow/util/io_util.h"
+#include "arrow/util/simdjson_internal.h"
 
 DEFINE_string(arrow, "", "Arrow file name");
 DEFINE_string(json, "", "JSON file name");
@@ -734,10 +735,10 @@ void TestSchemaRoundTrip(const std::shared_ptr<Schema>& schema) {
 
   ASSERT_OK_AND_ASSIGN(std::string_view json_schema, writer.GetString());
 
-  rj::Document d;
-  // Pass explicit size to avoid ASAN issues with
-  // SIMD loads in RapidJson.
-  d.Parse(json_schema.data(), json_schema.size());
+  simdjson::dom::parser parser;
+  ASSERT_OK_AND_ASSIGN(auto d, internal::ResolveSimdjsonResult(
+                                   parser.parse(json_schema.data(), json_schema.size()),
+                                   "Failed to parse JSON"));
 
   DictionaryMemo in_memo;
   ASSERT_OK_AND_ASSIGN(auto result_schema,
@@ -754,14 +755,11 @@ void TestArrayRoundTrip(const Array& array) {
 
   ASSERT_OK_AND_ASSIGN(std::string_view array_as_json, writer.GetString());
 
-  rj::Document d;
-  // Pass explicit size to avoid ASAN issues with
-  // SIMD loads in RapidJson.
-  d.Parse(array_as_json.data(), array_as_json.size());
-  if (d.HasParseError()) {
-    FAIL() << "JSON parsing failed";
-  }
-
+  simdjson::dom::parser parser;
+  ASSERT_OK_AND_ASSIGN(auto d,
+                       internal::ResolveSimdjsonResult(
+                           parser.parse(array_as_json.data(), array_as_json.size()),
+                           "Failed to parse JSON"));
   ASSERT_OK_AND_ASSIGN(
       auto result_array,
       json::ReadArray(default_memory_pool(), d, ::arrow::field(name, array.type())));
@@ -1111,10 +1109,10 @@ TEST(TestJsonFileReadWrite, JsonExample6) {
 }
 
 static void AssertInvalidBinaryViewJson(const std::string& json_array) {
-  rj::Document d;
-  // Pass explicit size to avoid ASAN issues with SIMD loads in RapidJson.
-  d.Parse(json_array.data(), json_array.size());
-  ASSERT_FALSE(d.HasParseError());
+  simdjson::dom::parser parser;
+  ASSERT_OK_AND_ASSIGN(auto d, internal::ResolveSimdjsonResult(
+                                   parser.parse(json_array.data(), json_array.size()),
+                                   "Failed to parse JSON"));
 
   ASSERT_RAISES(Invalid,
                 json::ReadArray(default_memory_pool(), d, field("f", binary_view())));