This is an automated email from the ASF dual-hosted git repository.
lidavidm pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-adbc.git
The following commit(s) were added to refs/heads/main by this push:
new a79b6741e fix(c/driver/postgresql): encode Arrow JSON for JSONB COPY
(#4505)
a79b6741e is described below
commit a79b6741e6ff8dadef547fa0f92732a81af28a5c
Author: 복준수 <[email protected]>
AuthorDate: Wed Jul 15 07:34:28 2026 +0900
fix(c/driver/postgresql): encode Arrow JSON for JSONB COPY (#4505)
### Summary
PostgreSQL binary COPY represents a `jsonb` field as a version
byte followed by the JSON text. The PostgreSQL driver currently
sends the physical string storage of `extension<arrow.json>`
unchanged, so PostgreSQL interprets the first JSON byte as the
version (for example, `{` becomes version 123) and rejects the
row. This adds a target-aware JSONB COPY field writer.
Closes #3293
Co-authored-by: BOKJUNSOO <[email protected]>
---
c/driver/postgresql/copy/writer.h | 39 ++++++++++++++++++++++++++++++++++
c/driver/postgresql/postgresql_test.cc | 20 ++++++++++++-----
2 files changed, 54 insertions(+), 5 deletions(-)
diff --git a/c/driver/postgresql/copy/writer.h
b/c/driver/postgresql/copy/writer.h
index 0853983ff..88db6023b 100644
--- a/c/driver/postgresql/copy/writer.h
+++ b/c/driver/postgresql/copy/writer.h
@@ -818,6 +818,26 @@ class PostgresCopyBinaryFieldWriter : public
PostgresCopyFieldWriter {
}
};
+class PostgresCopyJsonbFieldWriter : public PostgresCopyFieldWriter {
+ public:
+ ArrowErrorCode Write(ArrowBuffer* buffer, int64_t index, ArrowError* error)
override {
+ struct ArrowBufferView buffer_view =
ArrowArrayViewGetBytesUnsafe(array_view_, index);
+ if (buffer_view.size_bytes >= (std::numeric_limits<int32_t>::max)()) {
+ ArrowErrorSet(error, "[libpq] JSON value at row %" PRId64 " is too
large", index);
+ return EOVERFLOW;
+ }
+
+ const int32_t field_size_bytes =
static_cast<int32_t>(buffer_view.size_bytes) + 1;
+ constexpr int8_t kJsonbVersion = 1;
+ NANOARROW_RETURN_NOT_OK(WriteChecked<int32_t>(buffer, field_size_bytes,
error));
+ NANOARROW_RETURN_NOT_OK(WriteChecked<int8_t>(buffer, kJsonbVersion,
error));
+ NANOARROW_RETURN_NOT_OK(
+ ArrowBufferAppend(buffer, buffer_view.data.as_uint8,
buffer_view.size_bytes));
+
+ return ADBC_STATUS_OK;
+ }
+};
+
class PostgresCopyBinaryDictFieldWriter : public PostgresCopyFieldWriter {
public:
ArrowErrorCode Write(ArrowBuffer* buffer, int64_t index, ArrowError* error)
override {
@@ -968,6 +988,25 @@ static inline ArrowErrorCode MakeCopyFieldWriter(
struct ArrowSchemaView schema_view;
NANOARROW_RETURN_NOT_OK(ArrowSchemaViewInit(&schema_view, schema, error));
+ const bool is_arrow_json =
+ schema_view.extension_name.data != nullptr &&
+ std::string_view(schema_view.extension_name.data,
+ schema_view.extension_name.size_bytes) == "arrow.json";
+ if (target_type != nullptr && target_type->type_id() ==
PostgresTypeId::kJsonb &&
+ is_arrow_json) {
+ switch (schema_view.type) {
+ case NANOARROW_TYPE_STRING:
+ case NANOARROW_TYPE_LARGE_STRING:
+ case NANOARROW_TYPE_STRING_VIEW: {
+ using T = PostgresCopyJsonbFieldWriter;
+ *out = T::Create<T>(array_view);
+ return NANOARROW_OK;
+ }
+ default:
+ break;
+ }
+ }
+
if (target_type != nullptr && target_type->type_id() ==
PostgresTypeId::kNumeric) {
switch (schema_view.type) {
case NANOARROW_TYPE_INT8:
diff --git a/c/driver/postgresql/postgresql_test.cc
b/c/driver/postgresql/postgresql_test.cc
index 20f19264a..3bc4db35e 100644
--- a/c/driver/postgresql/postgresql_test.cc
+++ b/c/driver/postgresql/postgresql_test.cc
@@ -1802,12 +1802,22 @@ TEST_F(PostgresStatementTest, SqlIngestJsonb) {
IsOkStatus(&error));
ASSERT_THAT(AdbcStatementBind(&statement, &batch.value, &schema.value,
&error),
IsOkStatus(&error));
- // TODO(https://github.com/apache/arrow-adbc/issues/3293): we need a
- // different extension type for JSONB so the driver can know to generate
- // the appropriate COPY representation
- // (JSON-representation-version-prefixed JSON string).
ASSERT_THAT(AdbcStatementExecuteQuery(&statement, nullptr, nullptr,
&error),
- IsStatus(ADBC_STATUS_INVALID_ARGUMENT, &error));
+ IsOkStatus(&error));
+
+ ASSERT_THAT(AdbcStatementSetSqlQuery(
+ &statement, "SELECT j::text FROM jsontable ORDER BY ctid",
&error),
+ IsOkStatus(&error));
+
+ adbc_validation::StreamReader reader;
+ ASSERT_THAT(AdbcStatementExecuteQuery(&statement, &reader.stream.value,
+ &reader.rows_affected, &error),
+ IsOkStatus(&error));
+ ASSERT_NO_FATAL_FAILURE(reader.GetSchema());
+ ASSERT_NO_FATAL_FAILURE(reader.Next());
+ ASSERT_NE(nullptr, reader.array->release);
+ ASSERT_NO_FATAL_FAILURE(adbc_validation::CompareArray<std::string>(
+ reader.array_view->children[0], {R"({"a": 1, "b": [1, 2, 3]})",
std::nullopt}));
}
}
