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 4a3c7a344 fix(c/driver/postgresql): encode integer values for numeric
COPY targets (#4499)
4a3c7a344 is described below
commit 4a3c7a3449804251eb0a712fc1a76daf41e67efe
Author: 복준수 <[email protected]>
AuthorDate: Tue Jul 14 10:04:36 2026 +0900
fix(c/driver/postgresql): encode integer values for numeric COPY targets
(#4499)
### Summary
When ingesting into an existing PostgreSQL table, the COPY writer
selected binary encoders from the Arrow source schema only. Appending an
Arrow integer column into an existing `numeric` column therefore sent
integer binary COPY bytes that PostgreSQL interpreted as the binary
`numeric` layout, producing incorrect values like `0.00` for small
integers and other numeric decoding errors for larger values.
This updates append/create_append ingestion to resolve the target table
column types and pass them into COPY writer selection. For PostgreSQL
`numeric` targets, signed and unsigned Arrow integer arrays are now
encoded directly in PostgreSQL's binary numeric format. Unsupported
source types such as Arrow `double` now fail explicitly instead of
sending malformed binary COPY data.
Prerequisite for #2177.
---------
Co-authored-by: BOKJUNSOO <[email protected]>
---
c/driver/postgresql/bind_stream.h | 6 +-
.../postgresql/copy/postgres_copy_writer_test.cc | 2 +-
c/driver/postgresql/copy/writer.h | 138 ++++++++++++-
c/driver/postgresql/postgresql_test.cc | 220 +++++++++++++++++++++
c/driver/postgresql/statement.cc | 78 +++++++-
c/driver/postgresql/statement.h | 7 +-
6 files changed, 432 insertions(+), 19 deletions(-)
diff --git a/c/driver/postgresql/bind_stream.h
b/c/driver/postgresql/bind_stream.h
index 4e73477d4..5144a8920 100644
--- a/c/driver/postgresql/bind_stream.h
+++ b/c/driver/postgresql/bind_stream.h
@@ -164,7 +164,7 @@ struct BindStream {
UNWRAP_NANOARROW(
na_error, Internal,
MakeCopyFieldWriter(bind_schema->children[i],
array_view->children[i],
- type_resolver, &writer, &na_error));
+ type_resolver, nullptr, &writer, &na_error));
param_types[i] = type.oid();
param_formats[i] = kPgBinaryFormat;
@@ -317,13 +317,13 @@ struct BindStream {
}
Status ExecuteCopy(PGconn* pg_conn, const PostgresTypeResolver&
type_resolver,
- int64_t* rows_affected) {
+ const PostgresType* target_types, int64_t* rows_affected)
{
if (rows_affected) *rows_affected = 0;
PostgresCopyStreamWriter writer;
UNWRAP_ERRNO(Internal, writer.Init(&bind_schema.value));
UNWRAP_NANOARROW(na_error, Internal,
- writer.InitFieldWriters(type_resolver, &na_error));
+ writer.InitFieldWriters(type_resolver, target_types,
&na_error));
UNWRAP_NANOARROW(na_error, Internal, writer.WriteHeader(&na_error));
diff --git a/c/driver/postgresql/copy/postgres_copy_writer_test.cc
b/c/driver/postgresql/copy/postgres_copy_writer_test.cc
index 68ba54ed3..8cb71789b 100644
--- a/c/driver/postgresql/copy/postgres_copy_writer_test.cc
+++ b/c/driver/postgresql/copy/postgres_copy_writer_test.cc
@@ -41,7 +41,7 @@ class PostgresCopyStreamWriteTester {
const PostgresTypeResolver& type_resolver,
struct ArrowError* error = nullptr) {
NANOARROW_RETURN_NOT_OK(writer_.Init(schema));
- NANOARROW_RETURN_NOT_OK(writer_.InitFieldWriters(type_resolver, error));
+ NANOARROW_RETURN_NOT_OK(writer_.InitFieldWriters(type_resolver, nullptr,
error));
NANOARROW_RETURN_NOT_OK(writer_.SetArray(array));
return NANOARROW_OK;
}
diff --git a/c/driver/postgresql/copy/writer.h
b/c/driver/postgresql/copy/writer.h
index b2a11942f..0853983ff 100644
--- a/c/driver/postgresql/copy/writer.h
+++ b/c/driver/postgresql/copy/writer.h
@@ -180,6 +180,86 @@ class PostgresCopyNetworkEndianFieldWriter : public
PostgresCopyFieldWriter {
}
};
+class PostgresCopyIntegerNumericFieldWriter : public PostgresCopyFieldWriter {
+ public:
+ explicit PostgresCopyIntegerNumericFieldWriter(bool is_signed)
+ : is_signed_(is_signed) {}
+
+ ArrowErrorCode Write(ArrowBuffer* buffer, int64_t index, ArrowError* error)
override {
+ uint64_t value;
+ int16_t sign = kNumericPos;
+ if (is_signed_) {
+ const int64_t raw = ArrowArrayViewGetIntUnsafe(array_view_, index);
+ if (raw < 0) {
+ sign = kNumericNeg;
+ value = static_cast<uint64_t>(-(raw + 1)) + 1;
+ } else {
+ value = static_cast<uint64_t>(raw);
+ }
+ } else {
+ value = ArrowArrayViewGetUIntUnsafe(array_view_, index);
+ }
+
+ if (value == 0) {
+ return WriteZero(buffer, error);
+ }
+
+ constexpr int kMaxUInt64PgDigits = 5;
+ int16_t reversed_digits[kMaxUInt64PgDigits];
+ int n_digits = 0;
+ while (value != 0) {
+ reversed_digits[n_digits++] = static_cast<int16_t>(value % 10000);
+ value /= 10000;
+ }
+
+ int first_digit = 0;
+ while (first_digit < n_digits && reversed_digits[first_digit] == 0) {
+ first_digit++;
+ }
+
+ const int16_t ndigits = static_cast<int16_t>(n_digits - first_digit);
+ const int16_t weight = static_cast<int16_t>(n_digits - 1);
+ constexpr int16_t dscale = 0;
+ const int32_t field_size_bytes = sizeof(ndigits) + sizeof(weight) +
sizeof(sign) +
+ sizeof(dscale) + ndigits *
sizeof(int16_t);
+
+ NANOARROW_RETURN_NOT_OK(WriteChecked<int32_t>(buffer, field_size_bytes,
error));
+ NANOARROW_RETURN_NOT_OK(WriteChecked<int16_t>(buffer, ndigits, error));
+ NANOARROW_RETURN_NOT_OK(WriteChecked<int16_t>(buffer, weight, error));
+ NANOARROW_RETURN_NOT_OK(WriteChecked<int16_t>(buffer, sign, error));
+ NANOARROW_RETURN_NOT_OK(WriteChecked<int16_t>(buffer, dscale, error));
+
+ const size_t pg_digit_bytes = sizeof(int16_t) * ndigits;
+ NANOARROW_RETURN_NOT_OK(ArrowBufferReserve(buffer, pg_digit_bytes));
+ for (int i = n_digits - 1; i >= first_digit; i--) {
+ WriteUnsafe<int16_t>(buffer, reversed_digits[i]);
+ }
+
+ return ADBC_STATUS_OK;
+ }
+
+ private:
+ static ArrowErrorCode WriteZero(ArrowBuffer* buffer, ArrowError* error) {
+ constexpr int16_t ndigits = 0;
+ constexpr int16_t weight = 0;
+ constexpr int16_t sign = kNumericPos;
+ constexpr int16_t dscale = 0;
+ constexpr int32_t field_size_bytes =
+ sizeof(ndigits) + sizeof(weight) + sizeof(sign) + sizeof(dscale);
+
+ NANOARROW_RETURN_NOT_OK(WriteChecked<int32_t>(buffer, field_size_bytes,
error));
+ NANOARROW_RETURN_NOT_OK(WriteChecked<int16_t>(buffer, ndigits, error));
+ NANOARROW_RETURN_NOT_OK(WriteChecked<int16_t>(buffer, weight, error));
+ NANOARROW_RETURN_NOT_OK(WriteChecked<int16_t>(buffer, sign, error));
+ NANOARROW_RETURN_NOT_OK(WriteChecked<int16_t>(buffer, dscale, error));
+ return ADBC_STATUS_OK;
+ }
+
+ static constexpr int16_t kNumericPos = 0x0000;
+ static constexpr int16_t kNumericNeg = 0x4000;
+ const bool is_signed_;
+};
+
class PostgresCopyFloatFieldWriter : public PostgresCopyFieldWriter {
public:
ArrowErrorCode Write(ArrowBuffer* buffer, int64_t index, ArrowError* error)
override {
@@ -883,11 +963,42 @@ class PostgresCopyTimestampFieldWriter : public
PostgresCopyFieldWriter {
static inline ArrowErrorCode MakeCopyFieldWriter(
struct ArrowSchema* schema, struct ArrowArrayView* array_view,
- const PostgresTypeResolver& type_resolver,
+ const PostgresTypeResolver& type_resolver, const PostgresType* target_type,
std::unique_ptr<PostgresCopyFieldWriter>* out, ArrowError* error) {
struct ArrowSchemaView schema_view;
NANOARROW_RETURN_NOT_OK(ArrowSchemaViewInit(&schema_view, schema, error));
+ if (target_type != nullptr && target_type->type_id() ==
PostgresTypeId::kNumeric) {
+ switch (schema_view.type) {
+ case NANOARROW_TYPE_INT8:
+ case NANOARROW_TYPE_INT16:
+ case NANOARROW_TYPE_INT32:
+ case NANOARROW_TYPE_INT64: {
+ using T = PostgresCopyIntegerNumericFieldWriter;
+ *out = T::Create<T>(array_view, /*is_signed=*/true);
+ return NANOARROW_OK;
+ }
+ case NANOARROW_TYPE_UINT8:
+ case NANOARROW_TYPE_UINT16:
+ case NANOARROW_TYPE_UINT32:
+ case NANOARROW_TYPE_UINT64: {
+ using T = PostgresCopyIntegerNumericFieldWriter;
+ *out = T::Create<T>(array_view, /*is_signed=*/false);
+ return NANOARROW_OK;
+ }
+ case NANOARROW_TYPE_NA:
+ case NANOARROW_TYPE_DECIMAL128:
+ case NANOARROW_TYPE_DECIMAL256:
+ break;
+ default:
+ ArrowErrorSet(error,
+ "COPY Writer from Arrow type '%s' to PostgreSQL numeric
is not "
+ "implemented",
+ ArrowTypeString(schema_view.type));
+ return ENOTSUP;
+ }
+ }
+
switch (schema_view.type) {
case NANOARROW_TYPE_NA:
*out =
PostgresCopyFieldWriter::Create<PostgresCopyNullFieldWriter>(array_view);
@@ -1050,13 +1161,20 @@ static inline ArrowErrorCode MakeCopyFieldWriter(
NANOARROW_RETURN_NOT_OK(
ArrowSchemaViewInit(&child_schema_view, schema->children[0], error));
PostgresType child_type;
- NANOARROW_RETURN_NOT_OK(PostgresType::FromSchema(type_resolver,
schema->children[0],
- &child_type, error));
+ const PostgresType* target_child_type = nullptr;
+ if (target_type != nullptr && target_type->type_id() ==
PostgresTypeId::kArray &&
+ target_type->n_children() == 1) {
+ target_child_type = &target_type->child(0);
+ child_type = *target_child_type;
+ } else {
+ NANOARROW_RETURN_NOT_OK(PostgresType::FromSchema(
+ type_resolver, schema->children[0], &child_type, error));
+ }
std::unique_ptr<PostgresCopyFieldWriter> child_writer;
- NANOARROW_RETURN_NOT_OK(MakeCopyFieldWriter(schema->children[0],
- array_view->children[0],
type_resolver,
- &child_writer, error));
+ NANOARROW_RETURN_NOT_OK(
+ MakeCopyFieldWriter(schema->children[0], array_view->children[0],
type_resolver,
+ target_child_type, &child_writer, error));
if (schema_view.type == NANOARROW_TYPE_FIXED_SIZE_LIST) {
using T = PostgresCopyListFieldWriter<true>;
@@ -1114,16 +1232,20 @@ class PostgresCopyStreamWriter {
}
ArrowErrorCode InitFieldWriters(const PostgresTypeResolver& type_resolver,
- ArrowError* error) {
+ const PostgresType* target_types,
ArrowError* error) {
if (schema_->release == nullptr) {
return EINVAL;
}
for (int64_t i = 0; i < schema_->n_children; i++) {
std::unique_ptr<PostgresCopyFieldWriter> child_writer;
+ const PostgresType* target_type = nullptr;
+ if (target_types != nullptr && i < target_types->n_children()) {
+ target_type = &target_types->child(i);
+ }
NANOARROW_RETURN_NOT_OK(MakeCopyFieldWriter(schema_->children[i],
array_view_->children[i],
type_resolver,
- &child_writer, error));
+ target_type, &child_writer,
error));
root_writer_->AppendChild(std::move(child_writer));
}
diff --git a/c/driver/postgresql/postgresql_test.cc
b/c/driver/postgresql/postgresql_test.cc
index e81f47ac4..20f19264a 100644
--- a/c/driver/postgresql/postgresql_test.cc
+++ b/c/driver/postgresql/postgresql_test.cc
@@ -1347,6 +1347,226 @@ TEST_F(PostgresStatementTest, SqlIngestTemporaryTable) {
}
}
+TEST_F(PostgresStatementTest, SqlIngestAppendIntegerIntoNumeric) {
+ ASSERT_THAT(quirks()->DropTable(&connection, "numeric_ingest", &error),
+ IsOkStatus(&error));
+ ASSERT_THAT(AdbcStatementNew(&connection, &statement, &error),
IsOkStatus(&error));
+
+ ASSERT_THAT(
+ AdbcStatementSetSqlQuery(
+ &statement, "CREATE TABLE numeric_ingest (amount NUMERIC(20, 2))",
&error),
+ IsOkStatus(&error));
+ ASSERT_THAT(AdbcStatementExecuteQuery(&statement, nullptr, nullptr, &error),
+ IsOkStatus(&error));
+
+ adbc_validation::Handle<struct ArrowSchema> schema;
+ adbc_validation::Handle<struct ArrowArray> batch;
+
+ ArrowSchemaInit(&schema.value);
+ ASSERT_THAT(ArrowSchemaSetTypeStruct(&schema.value, 1),
adbc_validation::IsOkErrno());
+ ASSERT_THAT(ArrowSchemaSetType(schema->children[0], NANOARROW_TYPE_INT64),
+ adbc_validation::IsOkErrno());
+ ASSERT_THAT(ArrowSchemaSetName(schema->children[0], "amount"),
+ adbc_validation::IsOkErrno());
+
+ ASSERT_THAT((adbc_validation::MakeBatch<int64_t>(
+ &schema.value, &batch.value, static_cast<struct
ArrowError*>(nullptr),
+ {1, 42, 9999, -7, std::nullopt})),
+ adbc_validation::IsOkErrno());
+
+ ASSERT_THAT(AdbcStatementSetOption(&statement,
ADBC_INGEST_OPTION_TARGET_TABLE,
+ "numeric_ingest", &error),
+ IsOkStatus(&error));
+ ASSERT_THAT(AdbcStatementSetOption(&statement, ADBC_INGEST_OPTION_MODE,
+ ADBC_INGEST_OPTION_MODE_APPEND, &error),
+ IsOkStatus(&error));
+ ASSERT_THAT(AdbcStatementBind(&statement, &batch.value, &schema.value,
&error),
+ IsOkStatus(&error));
+ ASSERT_THAT(AdbcStatementExecuteQuery(&statement, nullptr, nullptr, &error),
+ IsOkStatus(&error));
+
+ ASSERT_THAT(AdbcStatementSetSqlQuery(&statement,
+ "SELECT amount::text FROM
numeric_ingest "
+ "ORDER BY numeric_ingest.amount NULLS
FIRST",
+ &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],
+ {std::nullopt, "-7.00", "1.00", "42.00", "9999.00"}));
+}
+
+TEST_F(PostgresStatementTest, SqlIngestAppendIntegerListIntoNumericArray) {
+ ASSERT_THAT(quirks()->DropTable(&connection, "numeric_array_ingest", &error),
+ IsOkStatus(&error));
+ ASSERT_THAT(AdbcStatementNew(&connection, &statement, &error),
IsOkStatus(&error));
+
+ ASSERT_THAT(AdbcStatementSetSqlQuery(
+ &statement,
+ "CREATE TABLE numeric_array_ingest (amounts NUMERIC(20,
2)[])", &error),
+ IsOkStatus(&error));
+ ASSERT_THAT(AdbcStatementExecuteQuery(&statement, nullptr, nullptr, &error),
+ IsOkStatus(&error));
+
+ adbc_validation::Handle<struct ArrowSchema> schema;
+ adbc_validation::Handle<struct ArrowArray> batch;
+ struct ArrowError na_error;
+
+ ASSERT_EQ(adbc_validation::MakeSchema(
+ &schema.value,
+ {adbc_validation::SchemaField::Nested("amounts",
NANOARROW_TYPE_LIST,
+ {{"item",
NANOARROW_TYPE_INT64}})}),
+ ADBC_STATUS_OK);
+ ASSERT_EQ(
+ adbc_validation::MakeBatch<std::vector<int64_t>>(
+ &schema.value, &batch.value, &na_error,
+ {std::vector<int64_t>{1, 42}, std::vector<int64_t>{-7, 9999},
std::nullopt}),
+ ADBC_STATUS_OK);
+
+ ASSERT_THAT(AdbcStatementSetOption(&statement,
ADBC_INGEST_OPTION_TARGET_TABLE,
+ "numeric_array_ingest", &error),
+ IsOkStatus(&error));
+ ASSERT_THAT(AdbcStatementSetOption(&statement, ADBC_INGEST_OPTION_MODE,
+ ADBC_INGEST_OPTION_MODE_APPEND, &error),
+ IsOkStatus(&error));
+ ASSERT_THAT(AdbcStatementBind(&statement, &batch.value, &schema.value,
&error),
+ IsOkStatus(&error));
+ ASSERT_THAT(AdbcStatementExecuteQuery(&statement, nullptr, nullptr, &error),
+ IsOkStatus(&error));
+
+ ASSERT_THAT(AdbcStatementSetSqlQuery(
+ &statement,
+ "SELECT amounts::text FROM numeric_array_ingest 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], {"{1.00,42.00}", "{-7.00,9999.00}",
std::nullopt}));
+}
+
+TEST_F(PostgresStatementTest, SqlIngestAppendIntegerBoundsIntoNumeric) {
+ ASSERT_THAT(quirks()->DropTable(&connection, "numeric_ingest", &error),
+ IsOkStatus(&error));
+ ASSERT_THAT(AdbcStatementNew(&connection, &statement, &error),
IsOkStatus(&error));
+
+ ASSERT_THAT(AdbcStatementSetSqlQuery(&statement,
+ "CREATE TABLE numeric_ingest ("
+ "signed_amount NUMERIC(30, 0), "
+ "unsigned_amount NUMERIC(30, 0))",
+ &error),
+ IsOkStatus(&error));
+ ASSERT_THAT(AdbcStatementExecuteQuery(&statement, nullptr, nullptr, &error),
+ IsOkStatus(&error));
+
+ adbc_validation::Handle<struct ArrowSchema> schema;
+ adbc_validation::Handle<struct ArrowArray> batch;
+
+ ArrowSchemaInit(&schema.value);
+ ASSERT_THAT(ArrowSchemaSetTypeStruct(&schema.value, 2),
adbc_validation::IsOkErrno());
+ ASSERT_THAT(ArrowSchemaSetType(schema->children[0], NANOARROW_TYPE_INT64),
+ adbc_validation::IsOkErrno());
+ ASSERT_THAT(ArrowSchemaSetName(schema->children[0], "signed_amount"),
+ adbc_validation::IsOkErrno());
+ ASSERT_THAT(ArrowSchemaSetType(schema->children[1], NANOARROW_TYPE_UINT64),
+ adbc_validation::IsOkErrno());
+ ASSERT_THAT(ArrowSchemaSetName(schema->children[1], "unsigned_amount"),
+ adbc_validation::IsOkErrno());
+
+ ASSERT_THAT((adbc_validation::MakeBatch<int64_t, uint64_t>(
+ &schema.value, &batch.value, static_cast<struct
ArrowError*>(nullptr),
+ {std::numeric_limits<int64_t>::min(), -10000, 0, 10000,
+ std::numeric_limits<int64_t>::max()},
+ {0, 10000, std::numeric_limits<uint64_t>::max(), 42, 1})),
+ adbc_validation::IsOkErrno());
+
+ ASSERT_THAT(AdbcStatementSetOption(&statement,
ADBC_INGEST_OPTION_TARGET_TABLE,
+ "numeric_ingest", &error),
+ IsOkStatus(&error));
+ ASSERT_THAT(AdbcStatementSetOption(&statement, ADBC_INGEST_OPTION_MODE,
+ ADBC_INGEST_OPTION_MODE_APPEND, &error),
+ IsOkStatus(&error));
+ ASSERT_THAT(AdbcStatementBind(&statement, &batch.value, &schema.value,
&error),
+ IsOkStatus(&error));
+ ASSERT_THAT(AdbcStatementExecuteQuery(&statement, nullptr, nullptr, &error),
+ IsOkStatus(&error));
+
+ ASSERT_THAT(
+ AdbcStatementSetSqlQuery(&statement,
+ "SELECT signed_amount::text,
unsigned_amount::text "
+ "FROM numeric_ingest 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],
+ {"-9223372036854775808", "-10000", "0", "10000",
"9223372036854775807"}));
+ ASSERT_NO_FATAL_FAILURE(adbc_validation::CompareArray<std::string>(
+ reader.array_view->children[1], {"0", "10000", "18446744073709551615",
"42", "1"}));
+}
+
+TEST_F(PostgresStatementTest, SqlIngestAppendFloatIntoNumericNotImplemented) {
+ ASSERT_THAT(quirks()->DropTable(&connection, "numeric_ingest", &error),
+ IsOkStatus(&error));
+ ASSERT_THAT(AdbcStatementNew(&connection, &statement, &error),
IsOkStatus(&error));
+
+ ASSERT_THAT(
+ AdbcStatementSetSqlQuery(
+ &statement, "CREATE TABLE numeric_ingest (amount NUMERIC(20, 2))",
&error),
+ IsOkStatus(&error));
+ ASSERT_THAT(AdbcStatementExecuteQuery(&statement, nullptr, nullptr, &error),
+ IsOkStatus(&error));
+
+ adbc_validation::Handle<struct ArrowSchema> schema;
+ adbc_validation::Handle<struct ArrowArray> batch;
+
+ ArrowSchemaInit(&schema.value);
+ ASSERT_THAT(ArrowSchemaSetTypeStruct(&schema.value, 1),
adbc_validation::IsOkErrno());
+ ASSERT_THAT(ArrowSchemaSetType(schema->children[0], NANOARROW_TYPE_DOUBLE),
+ adbc_validation::IsOkErrno());
+ ASSERT_THAT(ArrowSchemaSetName(schema->children[0], "amount"),
+ adbc_validation::IsOkErrno());
+
+ ASSERT_THAT(
+ (adbc_validation::MakeBatch<double>(
+ &schema.value, &batch.value, static_cast<struct
ArrowError*>(nullptr), {1.25})),
+ adbc_validation::IsOkErrno());
+
+ ASSERT_THAT(AdbcStatementSetOption(&statement,
ADBC_INGEST_OPTION_TARGET_TABLE,
+ "numeric_ingest", &error),
+ IsOkStatus(&error));
+ ASSERT_THAT(AdbcStatementSetOption(&statement, ADBC_INGEST_OPTION_MODE,
+ ADBC_INGEST_OPTION_MODE_APPEND, &error),
+ IsOkStatus(&error));
+ ASSERT_THAT(AdbcStatementBind(&statement, &batch.value, &schema.value,
&error),
+ IsOkStatus(&error));
+ ASSERT_THAT(AdbcStatementExecuteQuery(&statement, nullptr, nullptr, &error),
+ IsStatus(ADBC_STATUS_INTERNAL, &error));
+ ASSERT_THAT(error.message,
+ ::testing::HasSubstr(
+ "COPY Writer from Arrow type 'double' to PostgreSQL numeric
is not "
+ "implemented"));
+}
+
TEST_F(PostgresStatementTest, SqlIngestTimestampOverflow) {
ASSERT_THAT(AdbcStatementNew(&connection, &statement, &error),
IsOkStatus(&error));
diff --git a/c/driver/postgresql/statement.cc b/c/driver/postgresql/statement.cc
index b0c70dda6..041ab4452 100644
--- a/c/driver/postgresql/statement.cc
+++ b/c/driver/postgresql/statement.cc
@@ -367,8 +367,11 @@ AdbcStatusCode PostgresStatement::CreateBulkTable(const
std::string& current_sch
const struct ArrowSchema&
source_schema,
std::string* escaped_table,
std::string*
escaped_field_list,
+ PostgresType*
copy_target_types,
+ bool* has_copy_target_types,
struct AdbcError* error) {
PGconn* conn = connection_->conn();
+ *has_copy_target_types = false;
if (!ingest_.db_schema.empty() && ingest_.temporary) {
InternalAdbcSetError(error, "[libpq] Cannot set both %s and %s",
@@ -452,6 +455,8 @@ AdbcStatusCode PostgresStatement::CreateBulkTable(const
std::string& current_sch
}
create += *escaped_table;
create += " (";
+ std::vector<std::string> source_field_names;
+ source_field_names.reserve(source_schema.n_children);
for (int64_t i = 0; i < source_schema.n_children; i++) {
if (i > 0) {
@@ -468,6 +473,7 @@ AdbcStatusCode PostgresStatement::CreateBulkTable(const
std::string& current_sch
}
create += escaped;
*escaped_field_list += escaped;
+ source_field_names.emplace_back(unescaped);
PQfreemem(escaped);
PostgresType pg_type;
@@ -480,6 +486,12 @@ AdbcStatusCode PostgresStatement::CreateBulkTable(const
std::string& current_sch
}
if (ingest_.mode == IngestMode::kAppend) {
+ AdbcStatusCode status = ResolveCopyTargetTypes(*escaped_table,
source_field_names,
+ copy_target_types, error);
+ if (status != ADBC_STATUS_OK) {
+ return status;
+ }
+ *has_copy_target_types = true;
return ADBC_STATUS_OK;
}
@@ -497,6 +509,57 @@ AdbcStatusCode PostgresStatement::CreateBulkTable(const
std::string& current_sch
return code;
}
PQclear(result);
+ if (ingest_.mode == IngestMode::kCreateAppend) {
+ AdbcStatusCode status = ResolveCopyTargetTypes(*escaped_table,
source_field_names,
+ copy_target_types, error);
+ if (status != ADBC_STATUS_OK) {
+ return status;
+ }
+ *has_copy_target_types = true;
+ }
+ return ADBC_STATUS_OK;
+}
+
+AdbcStatusCode PostgresStatement::ResolveCopyTargetTypes(
+ const std::string& escaped_table, const std::vector<std::string>&
source_field_names,
+ PostgresType* copy_target_types, struct AdbcError* error) {
+ PqResultHelper result_helper{connection_->conn(),
+ "SELECT attr.attname, attr.atttypid "
+ "FROM pg_catalog.pg_attribute AS attr "
+ "WHERE attr.attrelid = $1::regclass::oid "
+ "AND attr.attnum > 0 AND NOT
attr.attisdropped"};
+ RAISE_STATUS(error, result_helper.Execute({escaped_table}));
+
+ std::vector<std::pair<std::string, uint32_t>> table_fields;
+ table_fields.reserve(result_helper.NumRows());
+ for (auto row : result_helper) {
+ table_fields.emplace_back(
+ row[0].data, static_cast<uint32_t>(std::strtol(row[1].data,
/*str_end=*/nullptr,
+ /*base=*/10)));
+ }
+
+ *copy_target_types = PostgresType(PostgresTypeId::kRecord);
+ for (const auto& field_name : source_field_names) {
+ auto it = std::find_if(table_fields.begin(), table_fields.end(),
+ [&](const auto& item) { return item.first ==
field_name; });
+ if (it == table_fields.end()) {
+ InternalAdbcSetError(error,
+ "[libpq] Column \"%s\" does not exist in target
table %s",
+ field_name.c_str(), escaped_table.c_str());
+ return ADBC_STATUS_INVALID_ARGUMENT;
+ }
+
+ PostgresType pg_type;
+ if (type_resolver_->FindWithDefault(it->second, &pg_type) != NANOARROW_OK)
{
+ InternalAdbcSetError(error,
+ "[libpq] Could not resolve PostgreSQL type oid %"
PRIu32
+ " for column \"%s\"",
+ it->second, field_name.c_str());
+ return ADBC_STATUS_NOT_IMPLEMENTED;
+ }
+ copy_target_types->AppendChild(field_name, pg_type);
+ }
+
return ADBC_STATUS_OK;
}
@@ -674,11 +737,13 @@ AdbcStatusCode PostgresStatement::ExecuteIngest(struct
ArrowArrayStream* stream,
std::memset(&bind_, 0, sizeof(bind_));
std::string escaped_table;
std::string escaped_field_list;
+ PostgresType copy_target_types;
+ bool has_copy_target_types = false;
RAISE_STATUS(error, bind_stream.Begin([&]() -> Status {
struct AdbcError tmp_error = ADBC_ERROR_INIT;
- AdbcStatusCode status_code =
- CreateBulkTable(current_schema, bind_stream.bind_schema.value,
&escaped_table,
- &escaped_field_list, &tmp_error);
+ AdbcStatusCode status_code = CreateBulkTable(
+ current_schema, bind_stream.bind_schema.value, &escaped_table,
+ &escaped_field_list, ©_target_types, &has_copy_target_types,
&tmp_error);
return Status::FromAdbc(status_code, tmp_error);
}));
@@ -697,9 +762,10 @@ AdbcStatusCode PostgresStatement::ExecuteIngest(struct
ArrowArrayStream* stream,
}
PQclear(result);
- RAISE_STATUS(error,
- bind_stream.ExecuteCopy(connection_->conn(),
*connection_->type_resolver(),
- rows_affected));
+ RAISE_STATUS(
+ error, bind_stream.ExecuteCopy(connection_->conn(),
*connection_->type_resolver(),
+ has_copy_target_types ?
©_target_types : nullptr,
+ rows_affected));
return ADBC_STATUS_OK;
}
diff --git a/c/driver/postgresql/statement.h b/c/driver/postgresql/statement.h
index a2c3f5e88..937d218d0 100644
--- a/c/driver/postgresql/statement.h
+++ b/c/driver/postgresql/statement.h
@@ -140,7 +140,12 @@ class PostgresStatement {
const struct ArrowSchema& source_schema,
std::string* escaped_table,
std::string* escaped_field_list,
- struct AdbcError* error);
+ PostgresType* copy_target_types,
+ bool* has_copy_target_types, struct
AdbcError* error);
+ AdbcStatusCode ResolveCopyTargetTypes(
+ const std::string& escaped_table,
+ const std::vector<std::string>& source_field_names, PostgresType*
copy_target_types,
+ struct AdbcError* error);
AdbcStatusCode ExecuteIngest(struct ArrowArrayStream* stream, int64_t*
rows_affected,
struct AdbcError* error);
AdbcStatusCode ExecuteBind(struct ArrowArrayStream* stream, int64_t*
rows_affected,
