paleolimbot commented on code in PR #2157:
URL: https://github.com/apache/arrow-adbc/pull/2157#discussion_r1761303109
##########
c/driver/postgresql/bind_stream.h:
##########
@@ -317,170 +237,40 @@ struct BindStream {
AdbcStatusCode BindAndExecuteCurrentRow(PGconn* pg_conn, PGresult**
result_out,
int result_format, AdbcError* error)
{
- int64_t row = current_row;
+ param_buffer->size_bytes = 0;
+ int64_t last_offset = 0;
for (int64_t col = 0; col < array_view->n_children; col++) {
- if (ArrowArrayViewIsNull(array_view->children[col], row)) {
- param_values[col] = nullptr;
- continue;
+ if (!ArrowArrayViewIsNull(array_view->children[col], current_row)) {
+ // Note that this Write() call currently writes the (int32_t) byte
size of the
+ // field in addition to the serialized value.
+ CHECK_NA_DETAIL(
+ INTERNAL,
+ bind_field_writers[col]->Write(¶m_buffer.value, current_row,
&na_error),
+ &na_error, error);
} else {
- param_values[col] = param_values_buffer.data() +
param_values_offsets[col];
+ CHECK_NA(INTERNAL, ArrowBufferAppendInt32(¶m_buffer.value, 0),
error);
}
- switch (bind_schema_fields[col].type) {
- case ArrowType::NANOARROW_TYPE_BOOL: {
- const int8_t val =
-
ArrowBitGet(array_view->children[col]->buffer_views[1].data.as_uint8, row);
- std::memcpy(param_values[col], &val, sizeof(int8_t));
- break;
- }
-
- case ArrowType::NANOARROW_TYPE_INT8: {
- const int16_t val =
- array_view->children[col]->buffer_views[1].data.as_int8[row];
- const uint16_t value = ToNetworkInt16(val);
- std::memcpy(param_values[col], &value, sizeof(int16_t));
- break;
- }
- case ArrowType::NANOARROW_TYPE_INT16: {
- const uint16_t value = ToNetworkInt16(
- array_view->children[col]->buffer_views[1].data.as_int16[row]);
- std::memcpy(param_values[col], &value, sizeof(int16_t));
- break;
- }
- case ArrowType::NANOARROW_TYPE_INT32: {
- const uint32_t value = ToNetworkInt32(
- array_view->children[col]->buffer_views[1].data.as_int32[row]);
- std::memcpy(param_values[col], &value, sizeof(int32_t));
- break;
- }
- case ArrowType::NANOARROW_TYPE_INT64: {
- const int64_t value = ToNetworkInt64(
- array_view->children[col]->buffer_views[1].data.as_int64[row]);
- std::memcpy(param_values[col], &value, sizeof(int64_t));
- break;
- }
- case ArrowType::NANOARROW_TYPE_FLOAT: {
- const uint32_t value = ToNetworkFloat4(
- array_view->children[col]->buffer_views[1].data.as_float[row]);
- std::memcpy(param_values[col], &value, sizeof(uint32_t));
- break;
- }
- case ArrowType::NANOARROW_TYPE_DOUBLE: {
- const uint64_t value = ToNetworkFloat8(
- array_view->children[col]->buffer_views[1].data.as_double[row]);
- std::memcpy(param_values[col], &value, sizeof(uint64_t));
- break;
- }
- case ArrowType::NANOARROW_TYPE_STRING:
- case ArrowType::NANOARROW_TYPE_LARGE_STRING:
- case ArrowType::NANOARROW_TYPE_BINARY: {
- const ArrowBufferView view =
- ArrowArrayViewGetBytesUnsafe(array_view->children[col], row);
- // TODO: overflow check?
- param_lengths[col] = static_cast<int>(view.size_bytes);
- param_values[col] = const_cast<char*>(view.data.as_char);
- break;
- }
- case ArrowType::NANOARROW_TYPE_DATE32: {
- // 2000-01-01
- constexpr int32_t kPostgresDateEpoch = 10957;
- const int32_t raw_value =
- array_view->children[col]->buffer_views[1].data.as_int32[row];
- if (raw_value < INT32_MIN + kPostgresDateEpoch) {
- SetError(error, "[libpq] Field #%" PRId64 "%s%s%s%" PRId64 "%s",
col + 1,
- "('", bind_schema->children[col]->name, "') Row #", row +
1,
- "has value which exceeds postgres date limits");
- return ADBC_STATUS_INVALID_ARGUMENT;
- }
-
- const uint32_t value = ToNetworkInt32(raw_value -
kPostgresDateEpoch);
- std::memcpy(param_values[col], &value, sizeof(int32_t));
- break;
- }
- case ArrowType::NANOARROW_TYPE_DURATION:
- case ArrowType::NANOARROW_TYPE_TIMESTAMP: {
- int64_t val =
array_view->children[col]->buffer_views[1].data.as_int64[row];
-
- bool overflow_safe = true;
-
- auto unit = bind_schema_fields[col].time_unit;
-
- switch (unit) {
- case NANOARROW_TIME_UNIT_SECOND:
- overflow_safe =
- val <= kMaxSafeSecondsToMicros && val >=
kMinSafeSecondsToMicros;
- if (overflow_safe) {
- val *= 1000000;
- }
-
- break;
- case NANOARROW_TIME_UNIT_MILLI:
- overflow_safe =
- val <= kMaxSafeMillisToMicros && val >=
kMinSafeMillisToMicros;
- if (overflow_safe) {
- val *= 1000;
- }
- break;
- case NANOARROW_TIME_UNIT_MICRO:
- break;
- case NANOARROW_TIME_UNIT_NANO:
- val /= 1000;
- break;
- }
-
- if (!overflow_safe) {
- SetError(error,
- "[libpq] Field #%" PRId64 " ('%s') Row #%" PRId64
- " has value '%" PRIi64 "' which exceeds PostgreSQL
timestamp limits",
- col + 1, bind_schema->children[col]->name, row + 1,
-
array_view->children[col]->buffer_views[1].data.as_int64[row]);
- return ADBC_STATUS_INVALID_ARGUMENT;
- }
-
- if (val < (std::numeric_limits<int64_t>::min)() +
kPostgresTimestampEpoch) {
- SetError(error,
- "[libpq] Field #%" PRId64 " ('%s') Row #%" PRId64
- " has value '%" PRIi64 "' which would underflow",
- col + 1, bind_schema->children[col]->name, row + 1,
-
array_view->children[col]->buffer_views[1].data.as_int64[row]);
- return ADBC_STATUS_INVALID_ARGUMENT;
- }
-
- if (bind_schema_fields[col].type ==
ArrowType::NANOARROW_TYPE_TIMESTAMP) {
- const uint64_t value = ToNetworkInt64(val -
kPostgresTimestampEpoch);
- std::memcpy(param_values[col], &value, sizeof(int64_t));
- } else if (bind_schema_fields[col].type ==
ArrowType::NANOARROW_TYPE_DURATION) {
- // postgres stores an interval as a 64 bit offset in microsecond
- // resolution alongside a 32 bit day and 32 bit month
- // for now we just send 0 for the day / month values
- const uint64_t value = ToNetworkInt64(val);
- std::memcpy(param_values[col], &value, sizeof(int64_t));
- std::memset(param_values[col] + sizeof(int64_t), 0,
sizeof(int64_t));
- }
- break;
- }
- case ArrowType::NANOARROW_TYPE_INTERVAL_MONTH_DAY_NANO: {
- struct ArrowInterval interval;
- ArrowIntervalInit(&interval, NANOARROW_TYPE_INTERVAL_MONTH_DAY_NANO);
- ArrowArrayViewGetIntervalUnsafe(array_view->children[col], row,
&interval);
-
- const uint32_t months = ToNetworkInt32(interval.months);
- const uint32_t days = ToNetworkInt32(interval.days);
- const uint64_t ms = ToNetworkInt64(interval.ns / 1000);
-
- std::memcpy(param_values[col], &ms, sizeof(uint64_t));
- std::memcpy(param_values[col] + sizeof(uint64_t), &days,
sizeof(uint32_t));
- std::memcpy(param_values[col] + sizeof(uint64_t) + sizeof(uint32_t),
&months,
- sizeof(uint32_t));
- break;
- }
- default:
- SetError(error, "%s%" PRId64 "%s%s%s%s", "[libpq] Field #", col + 1,
" ('",
- bind_schema->children[col]->name,
- "') has unsupported type for ingestion ",
- ArrowTypeString(bind_schema_fields[col].type));
- return ADBC_STATUS_NOT_IMPLEMENTED;
+
+ int64_t param_length = param_buffer->size_bytes - last_offset -
sizeof(int32_t);
+ if (param_length > INT32_MAX) {
Review Comment:
I can't get `std::numeric_limits<int>::max()` to work here even with
`#define NOMINMAX`. It works in the Windows CI but *not* in the Python package
build :shrug:
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
