GH-35460: Migrate integration RapidJSON writer to JsonWriter

Author: Reranko05

ci/docker/ubuntu-24.04-cpp.dockerfile

@@ -221,4 +221,5 @@ ENV absl_SOURCE=BUNDLED \
     PARQUET_BUILD_EXECUTABLES=ON \
     PATH=/usr/lib/ccache/:$PATH \
     PYTHON=python3 \
+    simdjson_SOURCE=BUNDLED \
     xsimd_SOURCE=BUNDLED

cpp/src/arrow/CMakeLists.txt

@@ -733,7 +733,7 @@ if(ARROW_BUILD_INTEGRATION OR ARROW_BUILD_TESTS)
   arrow_add_object_library(ARROW_INTEGRATION integration/json_integration.cc
                            integration/json_internal.cc)
   foreach(ARROW_INTEGRATION_TARGET ${ARROW_INTEGRATION_TARGETS})
-    target_link_libraries(${ARROW_INTEGRATION_TARGET} PRIVATE RapidJSON)
+    target_link_libraries(${ARROW_INTEGRATION_TARGET} PRIVATE RapidJSON simdjson::simdjson)
   endforeach()
 else()
   set(ARROW_INTEGRATION_TARGET_SHARED)

cpp/src/arrow/integration/CMakeLists.txt

@@ -21,12 +21,17 @@ 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 ${GFLAGS_LIBRARIES})
+  add_arrow_test(json_integration_test
+                 EXTRA_LINK_LIBS
+                 RapidJSON
+                 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}
                         ${ARROW_GTEST_GTEST})

cpp/src/arrow/integration/json_integration.cc

@@ -27,6 +27,7 @@
 #include "arrow/integration/json_internal.h"
 #include "arrow/io/file.h"
 #include "arrow/ipc/dictionary.h"
+#include "arrow/json/json_writer.h"
 #include "arrow/record_batch.h"
 #include "arrow/result.h"
 #include "arrow/status.h"
@@ -36,6 +37,8 @@
 using arrow::ipc::DictionaryFieldMapper;
 using arrow::ipc::DictionaryMemo;
 
+using JsonWriter = arrow::json::JsonWriter;
+
 namespace arrow::internal::integration {
 
 // ----------------------------------------------------------------------
@@ -44,13 +47,10 @@ namespace arrow::internal::integration {
 class IntegrationJsonWriter::Impl {
  public:
   explicit Impl(const std::shared_ptr<Schema>& schema)
-      : schema_(schema), mapper_(*schema), first_batch_written_(false) {
-    writer_.reset(new RjWriter(string_buffer_));
-  }
-
+      : schema_(schema), mapper_(*schema), first_batch_written_(false) {}
   Status Start() {
-    writer_->StartObject();
-    RETURN_NOT_OK(json::WriteSchema(*schema_, mapper_, writer_.get()));
+    writer_.StartObject();
+    RETURN_NOT_OK(json::WriteSchema(*schema_, mapper_, &writer_));
     return Status::OK();
   }
 
@@ -59,26 +59,26 @@ class IntegrationJsonWriter::Impl {
 
     // Write dictionaries, if any
     if (!dictionaries.empty()) {
-      writer_->Key("dictionaries");
-      writer_->StartArray();
+      writer_.Key("dictionaries");
+      writer_.StartArray();
       for (const auto& entry : dictionaries) {
-        RETURN_NOT_OK(json::WriteDictionary(entry.first, entry.second, writer_.get()));
+        RETURN_NOT_OK(json::WriteDictionary(entry.first, entry.second, &writer_));
       }
-      writer_->EndArray();
+      writer_.EndArray();
     }
 
     // Record batches
-    writer_->Key("batches");
-    writer_->StartArray();
+    writer_.Key("batches");
+    writer_.StartArray();
     first_batch_written_ = true;
     return Status::OK();
   }
 
   Result<std::string> Finish() {
-    writer_->EndArray();  // Record batches
-    writer_->EndObject();
+    writer_.EndArray();  // Record batches
+    writer_.EndObject();
 
-    return string_buffer_.GetString();
+    return std::string(writer_.GetString());
   }
 
   Status WriteRecordBatch(const RecordBatch& batch) {
@@ -87,7 +87,7 @@ class IntegrationJsonWriter::Impl {
     if (!first_batch_written_) {
       RETURN_NOT_OK(FirstRecordBatch(batch));
     }
-    return json::WriteRecordBatch(batch, writer_.get());
+    return json::WriteRecordBatch(batch, &writer_);
   }
 
  private:
@@ -96,8 +96,7 @@ class IntegrationJsonWriter::Impl {
 
   bool first_batch_written_;
 
-  rj::StringBuffer string_buffer_;
-  std::unique_ptr<RjWriter> writer_;
+  JsonWriter writer_;
 };
 
 IntegrationJsonWriter::IntegrationJsonWriter(const std::shared_ptr<Schema>& schema) {

cpp/src/arrow/integration/json_integration_test.cc

@@ -38,6 +38,7 @@
 #include "arrow/ipc/reader.h"
 #include "arrow/ipc/test_common.h"
 #include "arrow/ipc/writer.h"
+#include "arrow/json/json_writer.h"
 #include "arrow/pretty_print.h"
 #include "arrow/status.h"
 #include "arrow/testing/builder.h"
@@ -723,16 +724,15 @@ static const char* json_example6 = R"example(
 )example";
 
 void TestSchemaRoundTrip(const std::shared_ptr<Schema>& schema) {
-  rj::StringBuffer sb;
-  rj::Writer<rj::StringBuffer> writer(sb);
+  arrow::json::JsonWriter writer;
 
   DictionaryFieldMapper mapper(*schema);
 
   writer.StartObject();
   ASSERT_OK(json::WriteSchema(*schema, mapper, &writer));
   writer.EndObject();
 
-  std::string json_schema = sb.GetString();
+  std::string json_schema(writer.GetString());
 
   rj::Document d;
   // Pass explicit size to avoid ASAN issues with
@@ -748,12 +748,11 @@ void TestSchemaRoundTrip(const std::shared_ptr<Schema>& schema) {
 void TestArrayRoundTrip(const Array& array) {
   static std::string name = "dummy";
 
-  rj::StringBuffer sb;
-  rj::Writer<rj::StringBuffer> writer(sb);
+  arrow::json::JsonWriter writer;
 
   ASSERT_OK(json::WriteArray(name, array, &writer));
 
-  std::string array_as_json = sb.GetString();
+  std::string array_as_json(writer.GetString());
 
   rj::Document d;
   // Pass explicit size to avoid ASAN issues with
@@ -768,7 +767,6 @@ void TestArrayRoundTrip(const Array& array) {
       json::ReadArray(default_memory_pool(), d, ::arrow::field(name, array.type())));
   ASSERT_OK(result_array->ValidateFull());
 
-  // std::cout << array_as_json << std::endl;
   CompareArraysDetailed(0, *result_array, array);
 }
 

cpp/src/arrow/integration/json_internal.cc

@@ -36,6 +36,7 @@
 #include "arrow/array/builder_time.h"
 #include "arrow/extension_type.h"
 #include "arrow/ipc/dictionary.h"
+#include "arrow/json/json_writer.h"
 #include "arrow/record_batch.h"
 #include "arrow/result.h"
 #include "arrow/scalar.h"
@@ -64,6 +65,8 @@ using arrow::ipc::DictionaryFieldMapper;
 using arrow::ipc::DictionaryMemo;
 using arrow::ipc::internal::FieldPosition;
 
+using JsonWriter = arrow::json::JsonWriter;
+
 namespace arrow::internal::integration::json {
 
 namespace {
@@ -118,7 +121,7 @@ Result<std::string_view> GetStringView(const rj::Value& str) {
 class SchemaWriter {
  public:
   explicit SchemaWriter(const Schema& schema, const DictionaryFieldMapper& mapper,
-                        RjWriter* writer)
+                        JsonWriter* writer)
       : schema_(schema), mapper_(mapper), writer_(writer) {}
 
   Status Write() {
@@ -460,7 +463,7 @@ class SchemaWriter {
  private:
   const Schema& schema_;
   const DictionaryFieldMapper& mapper_;
-  RjWriter* writer_;
+  JsonWriter* writer_;
 };
 
 Status SchemaWriter::VisitType(const DataType& type) {
@@ -469,7 +472,7 @@ Status SchemaWriter::VisitType(const DataType& type) {
 
 class ArrayWriter {
  public:
-  ArrayWriter(const std::string& name, const Array& array, RjWriter* writer)
+  ArrayWriter(const std::string& name, const Array& array, JsonWriter* writer)
       : name_(name), array_(array), writer_(writer) {}
 
   Status Write() { return VisitArray(name_, array_); }
@@ -493,7 +496,7 @@ class ArrayWriter {
   void WriteRawNumber(std::string_view v) {
     // Avoid RawNumber() as it misleadingly adds quotes
     // (see https://github.com/Tencent/rapidjson/pull/1155)
-    writer_->RawValue(v.data(), v.size(), rj::kNumberType);
+    writer_->RawValue(v);
   }
 
   template <typename ArrayType, typename TypeClass = typename ArrayType::TypeClass,
@@ -598,7 +601,7 @@ class ArrayWriter {
         const Decimal32 value(arr.GetValue(i));
         writer_->String(value.ToIntegerString());
       } else {
-        writer_->String(null_string, sizeof(null_string));
+        writer_->String(std::string_view(null_string));
       }
     }
   }
@@ -610,7 +613,7 @@ class ArrayWriter {
         const Decimal64 value(arr.GetValue(i));
         writer_->String(value.ToIntegerString());
       } else {
-        writer_->String(null_string, sizeof(null_string));
+        writer_->String(std::string_view(null_string));
       }
     }
   }
@@ -622,7 +625,7 @@ class ArrayWriter {
         const Decimal128 value(arr.GetValue(i));
         writer_->String(value.ToIntegerString());
       } else {
-        writer_->String(null_string, sizeof(null_string));
+        writer_->String(std::string_view(null_string));
       }
     }
   }
@@ -634,7 +637,7 @@ class ArrayWriter {
         const Decimal256 value(arr.GetValue(i));
         writer_->String(value.ToIntegerString());
       } else {
-        writer_->String(null_string, sizeof(null_string));
+        writer_->String(std::string_view(null_string));
       }
     }
   }
@@ -863,7 +866,7 @@ class ArrayWriter {
  private:
   const std::string& name_;
   const Array& array_;
-  RjWriter* writer_;
+  JsonWriter* writer_;
 };
 
 Result<TimeUnit::type> GetUnitFromString(const std::string& unit_str) {
@@ -2035,13 +2038,13 @@ Result<std::shared_ptr<RecordBatch>> ReadRecordBatch(
 }
 
 Status WriteSchema(const Schema& schema, const DictionaryFieldMapper& mapper,
-                   RjWriter* json_writer) {
+                   JsonWriter* json_writer) {
   SchemaWriter converter(schema, mapper, json_writer);
   return converter.Write();
 }
 
 Status WriteDictionary(int64_t id, const std::shared_ptr<Array>& dictionary,
-                       RjWriter* writer) {
+                       JsonWriter* writer) {
   writer->StartObject();
   writer->Key("id");
   writer->Int(static_cast<int32_t>(id));
@@ -2055,7 +2058,7 @@ Status WriteDictionary(int64_t id, const std::shared_ptr<Array>& dictionary,
   return Status::OK();
 }
 
-Status WriteRecordBatch(const RecordBatch& batch, RjWriter* writer) {
+Status WriteRecordBatch(const RecordBatch& batch, JsonWriter* writer) {
   writer->StartObject();
   writer->Key("count");
   writer->Int(static_cast<int32_t>(batch.num_rows()));
@@ -2076,7 +2079,7 @@ Status WriteRecordBatch(const RecordBatch& batch, RjWriter* writer) {
   return Status::OK();
 }
 
-Status WriteArray(const std::string& name, const Array& array, RjWriter* json_writer) {
+Status WriteArray(const std::string& name, const Array& array, JsonWriter* json_writer) {
   ArrayWriter converter(name, array, json_writer);
   return converter.Write();
 }

cpp/src/arrow/integration/json_internal.h

@@ -36,9 +36,14 @@
 #include "arrow/util/visibility.h"
 
 namespace rj = arrow::rapidjson;
-using RjWriter = rj::Writer<rj::StringBuffer>;
-using RjArray = rj::Value::ConstArray;
 using RjObject = rj::Value::ConstObject;
+using RjArray = rj::Value::ConstArray;
+
+namespace arrow {
+namespace json {
+class JsonWriter;
+}  // namespace json
+}  // namespace arrow
 
 #define RETURN_NOT_FOUND(TOK, NAME, PARENT)              \
   if (NAME == (PARENT).MemberEnd()) {                    \
@@ -80,17 +85,17 @@ namespace arrow::internal::integration::json {
 /// \brief Append integration test Schema format to rapidjson writer
 ARROW_EXPORT
 Status WriteSchema(const Schema& schema, const ipc::DictionaryFieldMapper& mapper,
-                   RjWriter* writer);
+                   arrow::json::JsonWriter*);
 
 ARROW_EXPORT
 Status WriteDictionary(int64_t id, const std::shared_ptr<Array>& dictionary,
-                       RjWriter* writer);
+                       arrow::json::JsonWriter*);
 
 ARROW_EXPORT
-Status WriteRecordBatch(const RecordBatch& batch, RjWriter* writer);
+Status WriteRecordBatch(const RecordBatch& batch, arrow::json::JsonWriter*);
 
 ARROW_EXPORT
-Status WriteArray(const std::string& name, const Array& array, RjWriter* writer);
+Status WriteArray(const std::string& name, const Array& array, arrow::json::JsonWriter*);
 
 ARROW_EXPORT
 Result<std::shared_ptr<Schema>> ReadSchema(const rj::Value& json_obj, MemoryPool* pool,