You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@arrow.apache.org by "tokoko (via GitHub)" <gi...@apache.org> on 2026/05/17 07:13:16 UTC
[PR] Partitioned ingest API [arrow-adbc]
tokoko opened a new pull request, #4317:
URL: https://github.com/apache/arrow-adbc/pull/4317
demo PR for a new partitioned ingest API.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(format): partitioned bulk ingest API [arrow-adbc]
Posted by "tokoko (via GitHub)" <gi...@apache.org>.
tokoko commented on code in PR #4317:
URL: https://github.com/apache/arrow-adbc/pull/4317#discussion_r4211485401
##########
c/include/arrow-adbc/adbc.h:
##########
@@ -1516,6 +1518,23 @@ struct ADBC_EXPORT AdbcDriver {
AdbcStatusCode (*StatementExecuteMulti)(struct AdbcStatement*,
struct AdbcMultiResultSet*, struct AdbcError*);
+ AdbcStatusCode (*ConnectionBeginIngestPartitions)(struct AdbcConnection*,
+ struct ArrowSchema*,
+ struct AdbcSerializableHandle*,
+ struct AdbcError*);
+ AdbcStatusCode (*ConnectionWriteIngestPartition)(struct AdbcConnection*, const uint8_t*,
+ size_t, struct ArrowArrayStream*,
+ struct AdbcSerializableHandle*,
+ struct AdbcError*);
+ AdbcStatusCode (*ConnectionCompleteIngestPartitions)(struct AdbcConnection*,
Review Comment:
@lidavidm I went with an out parameter as discussed here, but while mapping this to table formats I found that two states (retryable / terminal) aren't enough, so I ended up making a new outcome code rather than a boolean: `int* outcome` on `AdbcStatementCompleteIngestPartitions` with the possible values of:
- `ADBC_INGEST_COMPLETE_SUCCEEDED` – set iff the call returns `ADBC_STATUS_OK` (kinda redundant but a little cleaner)
- `ADBC_INGEST_COMPLETE_FAILED` – nothing was promoted and the ingest can't be completed; abort and start over
- `ADBC_INGEST_COMPLETE_RETRYABLE` – nothing was promoted; call Complete again with the same handle and receipts
- `ADBC_INGEST_COMPLETE_UNKNOWN` – the driver doesn't know whether the commit landed; caller must *not* abort (a handle-scoped cleanup could delete committed data) and should call Complete again. A driver that reports this has to be able to resolve it on the repeated call without promoting the writes twice.
The third state comes straight from Iceberg. A rough mapping of iceberg exceptions to adbc complete states:
- `CommitFailedException` maps to `RETRYABLE`. This is often internally retried by clients, but can surface if retries are exhausted.
- `ValidationException`, `NoSuchTableException`, `ForbiddenException` and others map to `FAILED`. Most of these are more likely to show up during Begin rather than Complete, but still a possibility.
- `CommitStateUnknownException` maps to `UNKNOWN` - basically driver has no way of knowing whether promote worked and calling abort is dangerous.
The key difference between `RETRYABLE` and `UNKNOWN` is that abort is not safe to be called for `UNKNOWN`. For example when `CommitStateUnknownException` is surfaced, Iceberg's Spark writer fails the job and skips file cleanup.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(format): partitioned bulk ingest API [arrow-adbc]
Posted by "lidavidm (via GitHub)" <gi...@apache.org>.
lidavidm commented on code in PR #4317:
URL: https://github.com/apache/arrow-adbc/pull/4317#discussion_r3432578733
##########
c/include/arrow-adbc/adbc.h:
##########
@@ -1516,6 +1518,23 @@ struct ADBC_EXPORT AdbcDriver {
AdbcStatusCode (*StatementExecuteMulti)(struct AdbcStatement*,
struct AdbcMultiResultSet*, struct AdbcError*);
+ AdbcStatusCode (*ConnectionBeginIngestPartitions)(struct AdbcConnection*,
+ struct ArrowSchema*,
+ struct AdbcSerializableHandle*,
+ struct AdbcError*);
+ AdbcStatusCode (*ConnectionWriteIngestPartition)(struct AdbcConnection*, const uint8_t*,
+ size_t, struct ArrowArrayStream*,
+ struct AdbcSerializableHandle*,
+ struct AdbcError*);
+ AdbcStatusCode (*ConnectionCompleteIngestPartitions)(struct AdbcConnection*,
Review Comment:
Maybe just an out parameter to indicate whether it was success-or-retryable? A new status code would be a big change
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(format): partitioned bulk ingest API [arrow-adbc]
Posted by "tokoko (via GitHub)" <gi...@apache.org>.
tokoko commented on code in PR #4317:
URL: https://github.com/apache/arrow-adbc/pull/4317#discussion_r3366958014
##########
c/include/arrow-adbc/adbc.h:
##########
@@ -979,6 +987,9 @@ struct AdbcPartitions {
/// driver and the driver manager.
/// @{
+struct AdbcIngestHandle;
+struct AdbcIngestReceipt;
Review Comment:
changed to `AdbcSerializableHandle`, thanks
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(format): partitioned bulk ingest API [arrow-adbc]
Posted by "CurtHagenlocher (via GitHub)" <gi...@apache.org>.
CurtHagenlocher commented on code in PR #4317:
URL: https://github.com/apache/arrow-adbc/pull/4317#discussion_r3414163346
##########
c/include/arrow-adbc/adbc.h:
##########
@@ -1135,6 +1146,30 @@ struct ADBC_EXPORT AdbcDriver {
struct AdbcError*);
/// @}
+
+ /// \defgroup adbc-1.2.0 ADBC API Revision 1.2.0
+ ///
+ /// Functions added in ADBC 1.2.0. For backwards compatibility,
+ /// these members must not be accessed unless the version passed to
+ /// the AdbcDriverInitFunc is greater than or equal to
+ /// ADBC_VERSION_1_2_0.
+ ///
+ /// @{
+
+ AdbcStatusCode (*ConnectionBeginIngestPartitions)(
+ struct AdbcConnection*, const char*, const char*, const char*, const char*,
+ struct ArrowSchema*, struct AdbcIngestHandle*, struct AdbcError*);
+ AdbcStatusCode (*ConnectionWriteIngestPartition)(
+ struct AdbcConnection*, const uint8_t*, size_t, struct ArrowArrayStream*,
+ struct AdbcIngestReceipt*, struct AdbcError*);
+ AdbcStatusCode (*ConnectionCommitIngestPartitions)(
Review Comment:
I like "complete".
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(format): partitioned bulk ingest API [arrow-adbc]
Posted by "tokoko (via GitHub)" <gi...@apache.org>.
tokoko commented on code in PR #4317:
URL: https://github.com/apache/arrow-adbc/pull/4317#discussion_r3366957525
##########
c/include/arrow-adbc/adbc.h:
##########
@@ -423,6 +423,14 @@ const struct AdbcError* AdbcErrorFromArrayStream(struct ArrowArrayStream* stream
/// \since ADBC API revision 1.1.0
#define ADBC_VERSION_1_1_0 1001000
+/// \brief ADBC revision 1.2.0.
+///
+/// When passed to an AdbcDriverInitFunc(), the driver parameter must
+/// point to an AdbcDriver.
+///
+/// \since ADBC API revision 1.2.0
+#define ADBC_VERSION_1_2_0 1002000
Review Comment:
done
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Partitioned ingest API [arrow-adbc]
Posted by "lidavidm (via GitHub)" <gi...@apache.org>.
lidavidm commented on code in PR #4317:
URL: https://github.com/apache/arrow-adbc/pull/4317#discussion_r3277644954
##########
c/include/arrow-adbc/adbc.h:
##########
@@ -1946,6 +1989,235 @@ AdbcStatusCode AdbcConnectionReadPartition(struct AdbcConnection* connection,
/// @}
+/// \defgroup adbc-connection-ingest-partition Partitioned Bulk Ingest
+/// @{
+
+/// \brief Driver-owned bytes returned by
+/// AdbcConnectionBeginIngestPartitions.
+///
+/// The bytes are opaque and serializable: the caller may copy
+/// `bytes[0..length)` and ship that copy to workers (other processes
+/// or hosts) which can pass it to AdbcConnectionWriteIngestPartition
+/// directly.
+///
+/// The struct itself is owned by the driver. Call `release` exactly
+/// once to free it. Releasing the handle does NOT roll back the
+/// ingest — call AdbcConnectionAbortIngestPartitions for that.
+///
+/// \since ADBC API revision 1.2.0
+struct AdbcIngestHandle {
+ /// \brief The length of `bytes`.
+ size_t length;
+
+ /// \brief The serialized handle bytes (driver-owned).
+ const uint8_t* bytes;
+
+ /// \brief Private driver state.
+ void* private_data;
+
+ /// \brief Release the handle's memory. Sets `release` to NULL.
+ void (*release)(struct AdbcIngestHandle* self);
+};
+
+/// \brief Driver-owned bytes returned by
+/// AdbcConnectionWriteIngestPartition.
+///
+/// Mirror of AdbcIngestHandle: opaque, serializable, single-use
+/// `release`. Releasing a receipt does NOT discard the underlying
+/// write; that happens at Commit (commit it) or Abort (drop it).
+///
+/// \since ADBC API revision 1.2.0
+struct AdbcIngestReceipt {
Review Comment:
It would be good to explain the receipt/handle explicitly here instead of leaving it implicit from the below definitions.
Additionally I think the comments could generally be cleaned up.
##########
c/include/arrow-adbc/adbc.h:
##########
@@ -1946,6 +1989,235 @@ AdbcStatusCode AdbcConnectionReadPartition(struct AdbcConnection* connection,
/// @}
+/// \defgroup adbc-connection-ingest-partition Partitioned Bulk Ingest
+/// @{
+
+/// \brief Driver-owned bytes returned by
+/// AdbcConnectionBeginIngestPartitions.
+///
+/// The bytes are opaque and serializable: the caller may copy
+/// `bytes[0..length)` and ship that copy to workers (other processes
+/// or hosts) which can pass it to AdbcConnectionWriteIngestPartition
+/// directly.
+///
+/// The struct itself is owned by the driver. Call `release` exactly
+/// once to free it. Releasing the handle does NOT roll back the
+/// ingest — call AdbcConnectionAbortIngestPartitions for that.
+///
+/// \since ADBC API revision 1.2.0
+struct AdbcIngestHandle {
+ /// \brief The length of `bytes`.
+ size_t length;
+
+ /// \brief The serialized handle bytes (driver-owned).
+ const uint8_t* bytes;
+
+ /// \brief Private driver state.
+ void* private_data;
+
+ /// \brief Release the handle's memory. Sets `release` to NULL.
+ void (*release)(struct AdbcIngestHandle* self);
+};
+
+/// \brief Driver-owned bytes returned by
+/// AdbcConnectionWriteIngestPartition.
+///
+/// Mirror of AdbcIngestHandle: opaque, serializable, single-use
+/// `release`. Releasing a receipt does NOT discard the underlying
+/// write; that happens at Commit (commit it) or Abort (drop it).
+///
+/// \since ADBC API revision 1.2.0
+struct AdbcIngestReceipt {
+ /// \brief The length of `bytes`.
+ size_t length;
+
+ /// \brief The serialized receipt bytes (driver-owned).
+ const uint8_t* bytes;
+
+ /// \brief Private driver state.
+ void* private_data;
+
+ /// \brief Release the receipt's memory. Sets `release` to NULL.
+ void (*release)(struct AdbcIngestReceipt* self);
+};
+/// @}
+
+/// \addtogroup adbc-connection-ingest-partition
+/// Some drivers can accept bulk writes from a distributed writer: a
+/// coordinator configures an ingest, many workers write partitions in
+/// parallel (possibly from different processes or hosts), and the
+/// coordinator commits or aborts atomically.
+///
+/// This mirrors the read-side partitioned execution model. The
+/// coordinator calls AdbcConnectionBeginIngestPartitions to obtain an
+/// opaque, serializable handle. The handle is shipped to workers by
+/// the caller (e.g. a Spark driver sending it to executors). Workers
+/// call AdbcConnectionWriteIngestPartition on their own connections —
+/// the connection does not have to be the same one that created the
+/// handle. Each write returns an opaque receipt. The coordinator
+/// collects receipts and calls AdbcConnectionCommitIngestPartitions
+/// (or AdbcConnectionAbortIngestPartitions on failure).
+///
+/// Handles and receipts are driver-defined opaque byte strings. They
+/// are safe to transmit between processes and to use concurrently
+/// from multiple connections.
+///
+/// Drivers are not required to support partitioned ingest.
+///
+/// \since ADBC API revision 1.2.0
+///
+/// @{
+
+/// \brief Begin a partitioned bulk ingest.
+///
+/// Uses the same semantics as the ADBC_INGEST_OPTION_* options on
+/// AdbcStatement. For ADBC_INGEST_OPTION_MODE_CREATE,
+/// ADBC_INGEST_OPTION_MODE_CREATE_APPEND, and
+/// ADBC_INGEST_OPTION_MODE_REPLACE, `schema` is required and the
+/// driver creates (or recreates) the target table at this call. For
+/// ADBC_INGEST_OPTION_MODE_APPEND, `schema` is optional; if provided,
+/// the driver validates it against the target and returns
+/// ADBC_STATUS_ALREADY_EXISTS on mismatch.
+///
+/// The returned handle is opaque, serializable, and usable from any
+/// connection that can open the same database. The caller releases
+/// it via `out_handle->release`; the bytes can be copied and shipped
+/// to workers before release.
+///
+/// \since ADBC API revision 1.2.0
+/// \param[in] connection The coordinator's connection.
+/// \param[in] target_catalog Catalog of the target table, or NULL.
+/// \param[in] target_db_schema Schema of the target table, or NULL.
+/// \param[in] target_table Name of the target table. Required.
+/// \param[in] mode One of ADBC_INGEST_OPTION_MODE_*. Required.
+/// \param[in] schema Arrow schema of the data to be written.
+/// Required for create/replace/create_append modes; optional for
+/// append.
+/// \param[out] out_handle Driver-owned handle. Must be released by
+/// the caller via `out_handle->release`.
+/// \param[out] error Error details, if any.
+/// \return ADBC_STATUS_INVALID_ARGUMENT if mode requires a schema
+/// but none was provided.
+/// \return ADBC_STATUS_ALREADY_EXISTS if append mode is requested
+/// and the target schema disagrees with the provided schema.
+/// \return ADBC_STATUS_NOT_IMPLEMENTED if the driver does not
+/// support partitioned ingest.
+ADBC_EXPORT
+AdbcStatusCode AdbcConnectionBeginIngestPartitions(
+ struct AdbcConnection* connection, const char* target_catalog,
+ const char* target_db_schema, const char* target_table, const char* mode,
+ struct ArrowSchema* schema, struct AdbcIngestHandle* out_handle,
+ struct AdbcError* error);
+
+/// \brief Write one partition of a partitioned bulk ingest.
+///
+/// Called by a worker, typically on a different connection than the
+/// one that created the handle. The driver reads the bound stream
+/// to completion, writes its contents to driver-specific staging
+/// (per-call: a unique staging table, unique object-store path, etc.
+/// — never shared across concurrent writes), and returns an opaque
+/// receipt.
+///
+/// The stream's schema should be compatible with the target table's
+/// schema. Drivers may validate this at any point during the write;
+/// on mismatch the call fails and produces no receipt. The exact
+/// validation mechanism is driver-specific (e.g., RDBMS drivers may
+/// rely on the staging table DDL to enforce compatibility).
+///
+/// On error of any kind, `out_receipt` is left with `release ==
+/// NULL` and the caller should retry the whole partition. Partial
+/// receipts are never produced. The driver may, however, leave
+/// partial server-side state (for example, a per-call staging
+/// table); the caller must still call `AbortIngestPartitions` for
+/// the handle (with no receipt for this failed write) to release
+/// any staging resources, or rely on driver housekeeping.
+///
+/// This call is safe to invoke concurrently from many connections
+/// using the same handle.
+///
+/// \since ADBC API revision 1.2.0
+/// \param[in] connection The worker's connection.
+/// \param[in] handle The handle bytes from Begin.
+/// \param[in] handle_len Length of handle.
+/// \param[in] data Arrow stream of partition data. The driver
+/// consumes the stream and releases it.
+/// \param[out] out_receipt Driver-owned receipt. Must be released
+/// by the caller via `out_receipt->release`.
+/// \param[out] error Error details, if any.
+ADBC_EXPORT
+AdbcStatusCode AdbcConnectionWriteIngestPartition(
+ struct AdbcConnection* connection, const uint8_t* handle, size_t handle_len,
+ struct ArrowArrayStream* data, struct AdbcIngestReceipt* out_receipt,
+ struct AdbcError* error);
+
+/// \brief Commit a partitioned bulk ingest.
+///
+/// Atomically promotes all writes named by `receipts` into the
+/// target table. Semantics of "atomic" are driver-specific: RDBMS
+/// drivers typically swap staging into the target in a transaction;
+/// table-format drivers (Iceberg, Delta) write a catalog or
+/// transaction-log entry referencing the data files in the
+/// receipts.
+///
+/// After Commit returns successfully, the handle is consumed and
+/// must not be used again.
+///
+/// Receipts from failed writes, or writes whose receipts were never
+/// observed by the coordinator, are not included in the commit.
+/// Their staging data is orphaned and is cleaned up as described in
+/// AdbcConnectionAbortIngestPartitions.
+///
+/// \since ADBC API revision 1.2.0
+/// \param[in] connection A connection — typically the coordinator's,
+/// but any connection that can open the same database works.
+/// \param[in] handle The handle from Begin.
+/// \param[in] handle_len Length of handle.
+/// \param[in] num_receipts Number of receipts in the batch.
+/// \param[in] receipts Array of receipt byte-pointers.
+/// \param[in] receipt_lens Array of receipt lengths.
+/// \param[out] rows_affected Number of rows committed, or -1 if
+/// unknown. Pass NULL if not wanted.
+/// \param[out] error Error details, if any.
+ADBC_EXPORT
+AdbcStatusCode AdbcConnectionCommitIngestPartitions(
+ struct AdbcConnection* connection, const uint8_t* handle, size_t handle_len,
+ size_t num_receipts, const uint8_t** receipts, const size_t* receipt_lens,
+ int64_t* rows_affected, struct AdbcError* error);
+
+/// \brief Abort a partitioned bulk ingest.
+///
+/// Discards all writes scoped to the handle and releases any
+/// driver-side resources. The handle is consumed.
+///
+/// Drivers must clean up every write scoped to the handle, including
+/// writes whose receipts were lost or never observed — not only
+/// those named in `receipts`. The handle is the authority for
+/// cleanup scope; `receipts`, when provided, are a hint that allows
+/// the driver to fast-path deletion of known writes.
+///
+/// Abort is best-effort. If cleanup is incomplete, the driver
+/// returns a warning status and orphaned storage may remain; it is
Review Comment:
What is a "warning status"?
##########
c/include/arrow-adbc/adbc.h:
##########
@@ -1946,6 +1989,235 @@ AdbcStatusCode AdbcConnectionReadPartition(struct AdbcConnection* connection,
/// @}
+/// \defgroup adbc-connection-ingest-partition Partitioned Bulk Ingest
+/// @{
+
+/// \brief Driver-owned bytes returned by
+/// AdbcConnectionBeginIngestPartitions.
+///
+/// The bytes are opaque and serializable: the caller may copy
+/// `bytes[0..length)` and ship that copy to workers (other processes
+/// or hosts) which can pass it to AdbcConnectionWriteIngestPartition
+/// directly.
+///
+/// The struct itself is owned by the driver. Call `release` exactly
+/// once to free it. Releasing the handle does NOT roll back the
+/// ingest — call AdbcConnectionAbortIngestPartitions for that.
+///
+/// \since ADBC API revision 1.2.0
+struct AdbcIngestHandle {
+ /// \brief The length of `bytes`.
+ size_t length;
+
+ /// \brief The serialized handle bytes (driver-owned).
+ const uint8_t* bytes;
+
+ /// \brief Private driver state.
+ void* private_data;
+
+ /// \brief Release the handle's memory. Sets `release` to NULL.
+ void (*release)(struct AdbcIngestHandle* self);
+};
+
+/// \brief Driver-owned bytes returned by
+/// AdbcConnectionWriteIngestPartition.
+///
+/// Mirror of AdbcIngestHandle: opaque, serializable, single-use
+/// `release`. Releasing a receipt does NOT discard the underlying
+/// write; that happens at Commit (commit it) or Abort (drop it).
+///
+/// \since ADBC API revision 1.2.0
+struct AdbcIngestReceipt {
+ /// \brief The length of `bytes`.
+ size_t length;
+
+ /// \brief The serialized receipt bytes (driver-owned).
+ const uint8_t* bytes;
+
+ /// \brief Private driver state.
+ void* private_data;
+
+ /// \brief Release the receipt's memory. Sets `release` to NULL.
+ void (*release)(struct AdbcIngestReceipt* self);
+};
+/// @}
+
+/// \addtogroup adbc-connection-ingest-partition
+/// Some drivers can accept bulk writes from a distributed writer: a
+/// coordinator configures an ingest, many workers write partitions in
+/// parallel (possibly from different processes or hosts), and the
+/// coordinator commits or aborts atomically.
+///
+/// This mirrors the read-side partitioned execution model. The
+/// coordinator calls AdbcConnectionBeginIngestPartitions to obtain an
+/// opaque, serializable handle. The handle is shipped to workers by
+/// the caller (e.g. a Spark driver sending it to executors). Workers
+/// call AdbcConnectionWriteIngestPartition on their own connections —
+/// the connection does not have to be the same one that created the
+/// handle. Each write returns an opaque receipt. The coordinator
+/// collects receipts and calls AdbcConnectionCommitIngestPartitions
+/// (or AdbcConnectionAbortIngestPartitions on failure).
+///
+/// Handles and receipts are driver-defined opaque byte strings. They
+/// are safe to transmit between processes and to use concurrently
+/// from multiple connections.
+///
+/// Drivers are not required to support partitioned ingest.
+///
+/// \since ADBC API revision 1.2.0
+///
+/// @{
+
+/// \brief Begin a partitioned bulk ingest.
+///
+/// Uses the same semantics as the ADBC_INGEST_OPTION_* options on
+/// AdbcStatement. For ADBC_INGEST_OPTION_MODE_CREATE,
+/// ADBC_INGEST_OPTION_MODE_CREATE_APPEND, and
+/// ADBC_INGEST_OPTION_MODE_REPLACE, `schema` is required and the
+/// driver creates (or recreates) the target table at this call. For
+/// ADBC_INGEST_OPTION_MODE_APPEND, `schema` is optional; if provided,
+/// the driver validates it against the target and returns
+/// ADBC_STATUS_ALREADY_EXISTS on mismatch.
+///
+/// The returned handle is opaque, serializable, and usable from any
+/// connection that can open the same database. The caller releases
+/// it via `out_handle->release`; the bytes can be copied and shipped
+/// to workers before release.
+///
+/// \since ADBC API revision 1.2.0
+/// \param[in] connection The coordinator's connection.
+/// \param[in] target_catalog Catalog of the target table, or NULL.
+/// \param[in] target_db_schema Schema of the target table, or NULL.
+/// \param[in] target_table Name of the target table. Required.
+/// \param[in] mode One of ADBC_INGEST_OPTION_MODE_*. Required.
+/// \param[in] schema Arrow schema of the data to be written.
+/// Required for create/replace/create_append modes; optional for
+/// append.
+/// \param[out] out_handle Driver-owned handle. Must be released by
+/// the caller via `out_handle->release`.
+/// \param[out] error Error details, if any.
+/// \return ADBC_STATUS_INVALID_ARGUMENT if mode requires a schema
+/// but none was provided.
+/// \return ADBC_STATUS_ALREADY_EXISTS if append mode is requested
+/// and the target schema disagrees with the provided schema.
+/// \return ADBC_STATUS_NOT_IMPLEMENTED if the driver does not
+/// support partitioned ingest.
+ADBC_EXPORT
+AdbcStatusCode AdbcConnectionBeginIngestPartitions(
+ struct AdbcConnection* connection, const char* target_catalog,
+ const char* target_db_schema, const char* target_table, const char* mode,
Review Comment:
I'd rather we use the existing options here instead of duplicating them into the signature
##########
c/include/arrow-adbc/adbc.h:
##########
@@ -1946,6 +1989,235 @@ AdbcStatusCode AdbcConnectionReadPartition(struct AdbcConnection* connection,
/// @}
+/// \defgroup adbc-connection-ingest-partition Partitioned Bulk Ingest
+/// @{
+
+/// \brief Driver-owned bytes returned by
+/// AdbcConnectionBeginIngestPartitions.
+///
+/// The bytes are opaque and serializable: the caller may copy
+/// `bytes[0..length)` and ship that copy to workers (other processes
+/// or hosts) which can pass it to AdbcConnectionWriteIngestPartition
+/// directly.
+///
+/// The struct itself is owned by the driver. Call `release` exactly
+/// once to free it. Releasing the handle does NOT roll back the
+/// ingest — call AdbcConnectionAbortIngestPartitions for that.
+///
+/// \since ADBC API revision 1.2.0
+struct AdbcIngestHandle {
+ /// \brief The length of `bytes`.
+ size_t length;
+
+ /// \brief The serialized handle bytes (driver-owned).
+ const uint8_t* bytes;
+
+ /// \brief Private driver state.
+ void* private_data;
+
+ /// \brief Release the handle's memory. Sets `release` to NULL.
+ void (*release)(struct AdbcIngestHandle* self);
+};
+
+/// \brief Driver-owned bytes returned by
+/// AdbcConnectionWriteIngestPartition.
+///
+/// Mirror of AdbcIngestHandle: opaque, serializable, single-use
+/// `release`. Releasing a receipt does NOT discard the underlying
+/// write; that happens at Commit (commit it) or Abort (drop it).
+///
+/// \since ADBC API revision 1.2.0
+struct AdbcIngestReceipt {
+ /// \brief The length of `bytes`.
+ size_t length;
+
+ /// \brief The serialized receipt bytes (driver-owned).
+ const uint8_t* bytes;
+
+ /// \brief Private driver state.
+ void* private_data;
+
+ /// \brief Release the receipt's memory. Sets `release` to NULL.
+ void (*release)(struct AdbcIngestReceipt* self);
+};
+/// @}
+
+/// \addtogroup adbc-connection-ingest-partition
+/// Some drivers can accept bulk writes from a distributed writer: a
+/// coordinator configures an ingest, many workers write partitions in
+/// parallel (possibly from different processes or hosts), and the
+/// coordinator commits or aborts atomically.
+///
+/// This mirrors the read-side partitioned execution model. The
+/// coordinator calls AdbcConnectionBeginIngestPartitions to obtain an
+/// opaque, serializable handle. The handle is shipped to workers by
+/// the caller (e.g. a Spark driver sending it to executors). Workers
+/// call AdbcConnectionWriteIngestPartition on their own connections —
+/// the connection does not have to be the same one that created the
+/// handle. Each write returns an opaque receipt. The coordinator
+/// collects receipts and calls AdbcConnectionCommitIngestPartitions
+/// (or AdbcConnectionAbortIngestPartitions on failure).
+///
+/// Handles and receipts are driver-defined opaque byte strings. They
+/// are safe to transmit between processes and to use concurrently
+/// from multiple connections.
+///
+/// Drivers are not required to support partitioned ingest.
+///
+/// \since ADBC API revision 1.2.0
+///
+/// @{
+
+/// \brief Begin a partitioned bulk ingest.
+///
+/// Uses the same semantics as the ADBC_INGEST_OPTION_* options on
+/// AdbcStatement. For ADBC_INGEST_OPTION_MODE_CREATE,
+/// ADBC_INGEST_OPTION_MODE_CREATE_APPEND, and
+/// ADBC_INGEST_OPTION_MODE_REPLACE, `schema` is required and the
+/// driver creates (or recreates) the target table at this call. For
+/// ADBC_INGEST_OPTION_MODE_APPEND, `schema` is optional; if provided,
+/// the driver validates it against the target and returns
+/// ADBC_STATUS_ALREADY_EXISTS on mismatch.
+///
+/// The returned handle is opaque, serializable, and usable from any
+/// connection that can open the same database. The caller releases
+/// it via `out_handle->release`; the bytes can be copied and shipped
+/// to workers before release.
+///
+/// \since ADBC API revision 1.2.0
+/// \param[in] connection The coordinator's connection.
+/// \param[in] target_catalog Catalog of the target table, or NULL.
+/// \param[in] target_db_schema Schema of the target table, or NULL.
+/// \param[in] target_table Name of the target table. Required.
+/// \param[in] mode One of ADBC_INGEST_OPTION_MODE_*. Required.
+/// \param[in] schema Arrow schema of the data to be written.
+/// Required for create/replace/create_append modes; optional for
+/// append.
+/// \param[out] out_handle Driver-owned handle. Must be released by
+/// the caller via `out_handle->release`.
+/// \param[out] error Error details, if any.
+/// \return ADBC_STATUS_INVALID_ARGUMENT if mode requires a schema
+/// but none was provided.
+/// \return ADBC_STATUS_ALREADY_EXISTS if append mode is requested
+/// and the target schema disagrees with the provided schema.
+/// \return ADBC_STATUS_NOT_IMPLEMENTED if the driver does not
+/// support partitioned ingest.
+ADBC_EXPORT
+AdbcStatusCode AdbcConnectionBeginIngestPartitions(
+ struct AdbcConnection* connection, const char* target_catalog,
+ const char* target_db_schema, const char* target_table, const char* mode,
+ struct ArrowSchema* schema, struct AdbcIngestHandle* out_handle,
+ struct AdbcError* error);
+
+/// \brief Write one partition of a partitioned bulk ingest.
+///
+/// Called by a worker, typically on a different connection than the
+/// one that created the handle. The driver reads the bound stream
+/// to completion, writes its contents to driver-specific staging
+/// (per-call: a unique staging table, unique object-store path, etc.
+/// — never shared across concurrent writes), and returns an opaque
+/// receipt.
+///
+/// The stream's schema should be compatible with the target table's
+/// schema. Drivers may validate this at any point during the write;
+/// on mismatch the call fails and produces no receipt. The exact
+/// validation mechanism is driver-specific (e.g., RDBMS drivers may
+/// rely on the staging table DDL to enforce compatibility).
+///
+/// On error of any kind, `out_receipt` is left with `release ==
+/// NULL` and the caller should retry the whole partition. Partial
+/// receipts are never produced. The driver may, however, leave
+/// partial server-side state (for example, a per-call staging
+/// table); the caller must still call `AbortIngestPartitions` for
+/// the handle (with no receipt for this failed write) to release
+/// any staging resources, or rely on driver housekeeping.
+///
+/// This call is safe to invoke concurrently from many connections
+/// using the same handle.
+///
+/// \since ADBC API revision 1.2.0
+/// \param[in] connection The worker's connection.
+/// \param[in] handle The handle bytes from Begin.
+/// \param[in] handle_len Length of handle.
+/// \param[in] data Arrow stream of partition data. The driver
+/// consumes the stream and releases it.
+/// \param[out] out_receipt Driver-owned receipt. Must be released
+/// by the caller via `out_receipt->release`.
+/// \param[out] error Error details, if any.
+ADBC_EXPORT
+AdbcStatusCode AdbcConnectionWriteIngestPartition(
+ struct AdbcConnection* connection, const uint8_t* handle, size_t handle_len,
+ struct ArrowArrayStream* data, struct AdbcIngestReceipt* out_receipt,
+ struct AdbcError* error);
+
+/// \brief Commit a partitioned bulk ingest.
+///
+/// Atomically promotes all writes named by `receipts` into the
+/// target table. Semantics of "atomic" are driver-specific: RDBMS
+/// drivers typically swap staging into the target in a transaction;
+/// table-format drivers (Iceberg, Delta) write a catalog or
+/// transaction-log entry referencing the data files in the
+/// receipts.
+///
+/// After Commit returns successfully, the handle is consumed and
+/// must not be used again.
+///
+/// Receipts from failed writes, or writes whose receipts were never
+/// observed by the coordinator, are not included in the commit.
+/// Their staging data is orphaned and is cleaned up as described in
+/// AdbcConnectionAbortIngestPartitions.
+///
+/// \since ADBC API revision 1.2.0
+/// \param[in] connection A connection — typically the coordinator's,
+/// but any connection that can open the same database works.
+/// \param[in] handle The handle from Begin.
+/// \param[in] handle_len Length of handle.
+/// \param[in] num_receipts Number of receipts in the batch.
+/// \param[in] receipts Array of receipt byte-pointers.
+/// \param[in] receipt_lens Array of receipt lengths.
+/// \param[out] rows_affected Number of rows committed, or -1 if
+/// unknown. Pass NULL if not wanted.
+/// \param[out] error Error details, if any.
+ADBC_EXPORT
+AdbcStatusCode AdbcConnectionCommitIngestPartitions(
+ struct AdbcConnection* connection, const uint8_t* handle, size_t handle_len,
+ size_t num_receipts, const uint8_t** receipts, const size_t* receipt_lens,
+ int64_t* rows_affected, struct AdbcError* error);
+
+/// \brief Abort a partitioned bulk ingest.
+///
+/// Discards all writes scoped to the handle and releases any
+/// driver-side resources. The handle is consumed.
+///
+/// Drivers must clean up every write scoped to the handle, including
+/// writes whose receipts were lost or never observed — not only
+/// those named in `receipts`. The handle is the authority for
+/// cleanup scope; `receipts`, when provided, are a hint that allows
+/// the driver to fast-path deletion of known writes.
+///
+/// Abort is best-effort. If cleanup is incomplete, the driver
+/// returns a warning status and orphaned storage may remain; it is
+/// the driver's responsibility to provide housekeeping (e.g. TTL,
+/// background GC, or documented manual cleanup). Callers may also
Review Comment:
This sounds more like the semantics of the backend system
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Partitioned ingest API [arrow-adbc]
Posted by "tokoko (via GitHub)" <gi...@apache.org>.
tokoko commented on code in PR #4317:
URL: https://github.com/apache/arrow-adbc/pull/4317#discussion_r3322056269
##########
c/include/arrow-adbc/adbc.h:
##########
@@ -1946,6 +1989,235 @@ AdbcStatusCode AdbcConnectionReadPartition(struct AdbcConnection* connection,
/// @}
+/// \defgroup adbc-connection-ingest-partition Partitioned Bulk Ingest
+/// @{
+
+/// \brief Driver-owned bytes returned by
+/// AdbcConnectionBeginIngestPartitions.
+///
+/// The bytes are opaque and serializable: the caller may copy
+/// `bytes[0..length)` and ship that copy to workers (other processes
+/// or hosts) which can pass it to AdbcConnectionWriteIngestPartition
+/// directly.
+///
+/// The struct itself is owned by the driver. Call `release` exactly
+/// once to free it. Releasing the handle does NOT roll back the
+/// ingest — call AdbcConnectionAbortIngestPartitions for that.
+///
+/// \since ADBC API revision 1.2.0
+struct AdbcIngestHandle {
+ /// \brief The length of `bytes`.
+ size_t length;
+
+ /// \brief The serialized handle bytes (driver-owned).
+ const uint8_t* bytes;
+
+ /// \brief Private driver state.
+ void* private_data;
+
+ /// \brief Release the handle's memory. Sets `release` to NULL.
+ void (*release)(struct AdbcIngestHandle* self);
+};
+
+/// \brief Driver-owned bytes returned by
+/// AdbcConnectionWriteIngestPartition.
+///
+/// Mirror of AdbcIngestHandle: opaque, serializable, single-use
+/// `release`. Releasing a receipt does NOT discard the underlying
+/// write; that happens at Commit (commit it) or Abort (drop it).
+///
+/// \since ADBC API revision 1.2.0
+struct AdbcIngestReceipt {
+ /// \brief The length of `bytes`.
+ size_t length;
+
+ /// \brief The serialized receipt bytes (driver-owned).
+ const uint8_t* bytes;
+
+ /// \brief Private driver state.
+ void* private_data;
+
+ /// \brief Release the receipt's memory. Sets `release` to NULL.
+ void (*release)(struct AdbcIngestReceipt* self);
+};
+/// @}
+
+/// \addtogroup adbc-connection-ingest-partition
+/// Some drivers can accept bulk writes from a distributed writer: a
+/// coordinator configures an ingest, many workers write partitions in
+/// parallel (possibly from different processes or hosts), and the
+/// coordinator commits or aborts atomically.
+///
+/// This mirrors the read-side partitioned execution model. The
+/// coordinator calls AdbcConnectionBeginIngestPartitions to obtain an
+/// opaque, serializable handle. The handle is shipped to workers by
+/// the caller (e.g. a Spark driver sending it to executors). Workers
+/// call AdbcConnectionWriteIngestPartition on their own connections —
+/// the connection does not have to be the same one that created the
+/// handle. Each write returns an opaque receipt. The coordinator
+/// collects receipts and calls AdbcConnectionCommitIngestPartitions
+/// (or AdbcConnectionAbortIngestPartitions on failure).
+///
+/// Handles and receipts are driver-defined opaque byte strings. They
+/// are safe to transmit between processes and to use concurrently
+/// from multiple connections.
+///
+/// Drivers are not required to support partitioned ingest.
+///
+/// \since ADBC API revision 1.2.0
+///
+/// @{
+
+/// \brief Begin a partitioned bulk ingest.
+///
+/// Uses the same semantics as the ADBC_INGEST_OPTION_* options on
+/// AdbcStatement. For ADBC_INGEST_OPTION_MODE_CREATE,
+/// ADBC_INGEST_OPTION_MODE_CREATE_APPEND, and
+/// ADBC_INGEST_OPTION_MODE_REPLACE, `schema` is required and the
+/// driver creates (or recreates) the target table at this call. For
+/// ADBC_INGEST_OPTION_MODE_APPEND, `schema` is optional; if provided,
+/// the driver validates it against the target and returns
+/// ADBC_STATUS_ALREADY_EXISTS on mismatch.
+///
+/// The returned handle is opaque, serializable, and usable from any
+/// connection that can open the same database. The caller releases
+/// it via `out_handle->release`; the bytes can be copied and shipped
+/// to workers before release.
+///
+/// \since ADBC API revision 1.2.0
+/// \param[in] connection The coordinator's connection.
+/// \param[in] target_catalog Catalog of the target table, or NULL.
+/// \param[in] target_db_schema Schema of the target table, or NULL.
+/// \param[in] target_table Name of the target table. Required.
+/// \param[in] mode One of ADBC_INGEST_OPTION_MODE_*. Required.
+/// \param[in] schema Arrow schema of the data to be written.
+/// Required for create/replace/create_append modes; optional for
+/// append.
+/// \param[out] out_handle Driver-owned handle. Must be released by
+/// the caller via `out_handle->release`.
+/// \param[out] error Error details, if any.
+/// \return ADBC_STATUS_INVALID_ARGUMENT if mode requires a schema
+/// but none was provided.
+/// \return ADBC_STATUS_ALREADY_EXISTS if append mode is requested
+/// and the target schema disagrees with the provided schema.
+/// \return ADBC_STATUS_NOT_IMPLEMENTED if the driver does not
+/// support partitioned ingest.
+ADBC_EXPORT
+AdbcStatusCode AdbcConnectionBeginIngestPartitions(
+ struct AdbcConnection* connection, const char* target_catalog,
+ const char* target_db_schema, const char* target_table, const char* mode,
Review Comment:
one caveat there is that existing options are statement-level in practice rather than connection-level. all these new functions are connection-level. are you also suggesting moving all (or some) to statement-level?
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Partitioned ingest API [arrow-adbc]
Posted by "CurtHagenlocher (via GitHub)" <gi...@apache.org>.
CurtHagenlocher commented on code in PR #4317:
URL: https://github.com/apache/arrow-adbc/pull/4317#discussion_r3283789864
##########
c/include/arrow-adbc/adbc.h:
##########
@@ -979,6 +987,9 @@ struct AdbcPartitions {
/// driver and the driver manager.
/// @{
+struct AdbcIngestHandle;
+struct AdbcIngestReceipt;
Review Comment:
The only role these seem to play is to have an API that allows the corresponding opaque buffer to be deallocated. We're also not getting any type safety from having two of them because both are just byte arrays post-serialization. Could we simplify and generalize a little and declare a single type for this e.g. `AdbcSerializableHandle`?
##########
c/include/arrow-adbc/adbc.h:
##########
@@ -1135,6 +1146,30 @@ struct ADBC_EXPORT AdbcDriver {
struct AdbcError*);
/// @}
+
+ /// \defgroup adbc-1.2.0 ADBC API Revision 1.2.0
+ ///
+ /// Functions added in ADBC 1.2.0. For backwards compatibility,
+ /// these members must not be accessed unless the version passed to
+ /// the AdbcDriverInitFunc is greater than or equal to
+ /// ADBC_VERSION_1_2_0.
+ ///
+ /// @{
+
+ AdbcStatusCode (*ConnectionBeginIngestPartitions)(
+ struct AdbcConnection*, const char*, const char*, const char*, const char*,
+ struct ArrowSchema*, struct AdbcIngestHandle*, struct AdbcError*);
+ AdbcStatusCode (*ConnectionWriteIngestPartition)(
+ struct AdbcConnection*, const uint8_t*, size_t, struct ArrowArrayStream*,
+ struct AdbcIngestReceipt*, struct AdbcError*);
+ AdbcStatusCode (*ConnectionCommitIngestPartitions)(
Review Comment:
Repeating my comment from the mailing list, is this use of `Commit` going to be misleading around its relationship with a database transactional commit?
##########
c/include/arrow-adbc/adbc.h:
##########
@@ -1946,6 +1989,235 @@ AdbcStatusCode AdbcConnectionReadPartition(struct AdbcConnection* connection,
/// @}
+/// \defgroup adbc-connection-ingest-partition Partitioned Bulk Ingest
+/// @{
+
+/// \brief Driver-owned bytes returned by
+/// AdbcConnectionBeginIngestPartitions.
+///
+/// The bytes are opaque and serializable: the caller may copy
+/// `bytes[0..length)` and ship that copy to workers (other processes
+/// or hosts) which can pass it to AdbcConnectionWriteIngestPartition
+/// directly.
+///
+/// The struct itself is owned by the driver. Call `release` exactly
+/// once to free it. Releasing the handle does NOT roll back the
+/// ingest — call AdbcConnectionAbortIngestPartitions for that.
+///
+/// \since ADBC API revision 1.2.0
+struct AdbcIngestHandle {
+ /// \brief The length of `bytes`.
+ size_t length;
+
+ /// \brief The serialized handle bytes (driver-owned).
+ const uint8_t* bytes;
+
+ /// \brief Private driver state.
+ void* private_data;
+
+ /// \brief Release the handle's memory. Sets `release` to NULL.
+ void (*release)(struct AdbcIngestHandle* self);
+};
+
+/// \brief Driver-owned bytes returned by
+/// AdbcConnectionWriteIngestPartition.
+///
+/// Mirror of AdbcIngestHandle: opaque, serializable, single-use
+/// `release`. Releasing a receipt does NOT discard the underlying
+/// write; that happens at Commit (commit it) or Abort (drop it).
+///
+/// \since ADBC API revision 1.2.0
+struct AdbcIngestReceipt {
+ /// \brief The length of `bytes`.
+ size_t length;
+
+ /// \brief The serialized receipt bytes (driver-owned).
+ const uint8_t* bytes;
+
+ /// \brief Private driver state.
+ void* private_data;
+
+ /// \brief Release the receipt's memory. Sets `release` to NULL.
+ void (*release)(struct AdbcIngestReceipt* self);
+};
+/// @}
+
+/// \addtogroup adbc-connection-ingest-partition
+/// Some drivers can accept bulk writes from a distributed writer: a
+/// coordinator configures an ingest, many workers write partitions in
+/// parallel (possibly from different processes or hosts), and the
+/// coordinator commits or aborts atomically.
+///
+/// This mirrors the read-side partitioned execution model. The
+/// coordinator calls AdbcConnectionBeginIngestPartitions to obtain an
+/// opaque, serializable handle. The handle is shipped to workers by
+/// the caller (e.g. a Spark driver sending it to executors). Workers
+/// call AdbcConnectionWriteIngestPartition on their own connections —
+/// the connection does not have to be the same one that created the
+/// handle. Each write returns an opaque receipt. The coordinator
+/// collects receipts and calls AdbcConnectionCommitIngestPartitions
+/// (or AdbcConnectionAbortIngestPartitions on failure).
+///
+/// Handles and receipts are driver-defined opaque byte strings. They
+/// are safe to transmit between processes and to use concurrently
+/// from multiple connections.
+///
+/// Drivers are not required to support partitioned ingest.
+///
+/// \since ADBC API revision 1.2.0
+///
+/// @{
+
+/// \brief Begin a partitioned bulk ingest.
+///
+/// Uses the same semantics as the ADBC_INGEST_OPTION_* options on
+/// AdbcStatement. For ADBC_INGEST_OPTION_MODE_CREATE,
+/// ADBC_INGEST_OPTION_MODE_CREATE_APPEND, and
+/// ADBC_INGEST_OPTION_MODE_REPLACE, `schema` is required and the
+/// driver creates (or recreates) the target table at this call. For
+/// ADBC_INGEST_OPTION_MODE_APPEND, `schema` is optional; if provided,
+/// the driver validates it against the target and returns
+/// ADBC_STATUS_ALREADY_EXISTS on mismatch.
+///
+/// The returned handle is opaque, serializable, and usable from any
+/// connection that can open the same database. The caller releases
+/// it via `out_handle->release`; the bytes can be copied and shipped
+/// to workers before release.
+///
+/// \since ADBC API revision 1.2.0
+/// \param[in] connection The coordinator's connection.
+/// \param[in] target_catalog Catalog of the target table, or NULL.
+/// \param[in] target_db_schema Schema of the target table, or NULL.
+/// \param[in] target_table Name of the target table. Required.
+/// \param[in] mode One of ADBC_INGEST_OPTION_MODE_*. Required.
+/// \param[in] schema Arrow schema of the data to be written.
+/// Required for create/replace/create_append modes; optional for
+/// append.
+/// \param[out] out_handle Driver-owned handle. Must be released by
+/// the caller via `out_handle->release`.
+/// \param[out] error Error details, if any.
+/// \return ADBC_STATUS_INVALID_ARGUMENT if mode requires a schema
+/// but none was provided.
+/// \return ADBC_STATUS_ALREADY_EXISTS if append mode is requested
+/// and the target schema disagrees with the provided schema.
+/// \return ADBC_STATUS_NOT_IMPLEMENTED if the driver does not
+/// support partitioned ingest.
+ADBC_EXPORT
+AdbcStatusCode AdbcConnectionBeginIngestPartitions(
+ struct AdbcConnection* connection, const char* target_catalog,
+ const char* target_db_schema, const char* target_table, const char* mode,
+ struct ArrowSchema* schema, struct AdbcIngestHandle* out_handle,
+ struct AdbcError* error);
+
+/// \brief Write one partition of a partitioned bulk ingest.
+///
+/// Called by a worker, typically on a different connection than the
+/// one that created the handle. The driver reads the bound stream
+/// to completion, writes its contents to driver-specific staging
+/// (per-call: a unique staging table, unique object-store path, etc.
+/// — never shared across concurrent writes), and returns an opaque
+/// receipt.
+///
+/// The stream's schema should be compatible with the target table's
+/// schema. Drivers may validate this at any point during the write;
+/// on mismatch the call fails and produces no receipt. The exact
+/// validation mechanism is driver-specific (e.g., RDBMS drivers may
+/// rely on the staging table DDL to enforce compatibility).
+///
+/// On error of any kind, `out_receipt` is left with `release ==
+/// NULL` and the caller should retry the whole partition. Partial
+/// receipts are never produced. The driver may, however, leave
+/// partial server-side state (for example, a per-call staging
+/// table); the caller must still call `AbortIngestPartitions` for
+/// the handle (with no receipt for this failed write) to release
+/// any staging resources, or rely on driver housekeeping.
+///
+/// This call is safe to invoke concurrently from many connections
+/// using the same handle.
+///
+/// \since ADBC API revision 1.2.0
+/// \param[in] connection The worker's connection.
+/// \param[in] handle The handle bytes from Begin.
+/// \param[in] handle_len Length of handle.
+/// \param[in] data Arrow stream of partition data. The driver
+/// consumes the stream and releases it.
+/// \param[out] out_receipt Driver-owned receipt. Must be released
+/// by the caller via `out_receipt->release`.
+/// \param[out] error Error details, if any.
+ADBC_EXPORT
+AdbcStatusCode AdbcConnectionWriteIngestPartition(
+ struct AdbcConnection* connection, const uint8_t* handle, size_t handle_len,
+ struct ArrowArrayStream* data, struct AdbcIngestReceipt* out_receipt,
+ struct AdbcError* error);
+
+/// \brief Commit a partitioned bulk ingest.
+///
+/// Atomically promotes all writes named by `receipts` into the
+/// target table. Semantics of "atomic" are driver-specific: RDBMS
+/// drivers typically swap staging into the target in a transaction;
+/// table-format drivers (Iceberg, Delta) write a catalog or
+/// transaction-log entry referencing the data files in the
+/// receipts.
+///
+/// After Commit returns successfully, the handle is consumed and
+/// must not be used again.
+///
+/// Receipts from failed writes, or writes whose receipts were never
+/// observed by the coordinator, are not included in the commit.
+/// Their staging data is orphaned and is cleaned up as described in
+/// AdbcConnectionAbortIngestPartitions.
+///
+/// \since ADBC API revision 1.2.0
+/// \param[in] connection A connection — typically the coordinator's,
+/// but any connection that can open the same database works.
+/// \param[in] handle The handle from Begin.
+/// \param[in] handle_len Length of handle.
+/// \param[in] num_receipts Number of receipts in the batch.
+/// \param[in] receipts Array of receipt byte-pointers.
+/// \param[in] receipt_lens Array of receipt lengths.
+/// \param[out] rows_affected Number of rows committed, or -1 if
+/// unknown. Pass NULL if not wanted.
+/// \param[out] error Error details, if any.
+ADBC_EXPORT
+AdbcStatusCode AdbcConnectionCommitIngestPartitions(
+ struct AdbcConnection* connection, const uint8_t* handle, size_t handle_len,
+ size_t num_receipts, const uint8_t** receipts, const size_t* receipt_lens,
+ int64_t* rows_affected, struct AdbcError* error);
+
+/// \brief Abort a partitioned bulk ingest.
+///
+/// Discards all writes scoped to the handle and releases any
+/// driver-side resources. The handle is consumed.
+///
+/// Drivers must clean up every write scoped to the handle, including
+/// writes whose receipts were lost or never observed — not only
+/// those named in `receipts`. The handle is the authority for
+/// cleanup scope; `receipts`, when provided, are a hint that allows
+/// the driver to fast-path deletion of known writes.
+///
+/// Abort is best-effort. If cleanup is incomplete, the driver
+/// returns a warning status and orphaned storage may remain; it is
+/// the driver's responsibility to provide housekeeping (e.g. TTL,
+/// background GC, or documented manual cleanup). Callers may also
+/// call Abort if the coordinator crashed and was restarted without
+/// the original receipts.
Review Comment:
Is it worth distinguishing the case of "`Abort` an unknown handle" from the case of "something went wrong during `Abort`" with a specific error code for the former?
##########
c/include/arrow-adbc/adbc.h:
##########
@@ -423,6 +423,14 @@ const struct AdbcError* AdbcErrorFromArrayStream(struct ArrowArrayStream* stream
/// \since ADBC API revision 1.1.0
#define ADBC_VERSION_1_1_0 1001000
+/// \brief ADBC revision 1.2.0.
+///
+/// When passed to an AdbcDriverInitFunc(), the driver parameter must
+/// point to an AdbcDriver.
+///
+/// \since ADBC API revision 1.2.0
+#define ADBC_VERSION_1_2_0 1002000
Review Comment:
Can you please resubmit this as a PR against the `spec-1.2.0` branch where this should already be defined?
##########
c/include/arrow-adbc/adbc.h:
##########
@@ -1946,6 +1989,235 @@ AdbcStatusCode AdbcConnectionReadPartition(struct AdbcConnection* connection,
/// @}
+/// \defgroup adbc-connection-ingest-partition Partitioned Bulk Ingest
+/// @{
+
+/// \brief Driver-owned bytes returned by
+/// AdbcConnectionBeginIngestPartitions.
+///
+/// The bytes are opaque and serializable: the caller may copy
+/// `bytes[0..length)` and ship that copy to workers (other processes
+/// or hosts) which can pass it to AdbcConnectionWriteIngestPartition
+/// directly.
+///
+/// The struct itself is owned by the driver. Call `release` exactly
+/// once to free it. Releasing the handle does NOT roll back the
+/// ingest — call AdbcConnectionAbortIngestPartitions for that.
+///
+/// \since ADBC API revision 1.2.0
+struct AdbcIngestHandle {
+ /// \brief The length of `bytes`.
+ size_t length;
+
+ /// \brief The serialized handle bytes (driver-owned).
+ const uint8_t* bytes;
+
+ /// \brief Private driver state.
+ void* private_data;
+
+ /// \brief Release the handle's memory. Sets `release` to NULL.
+ void (*release)(struct AdbcIngestHandle* self);
+};
+
+/// \brief Driver-owned bytes returned by
+/// AdbcConnectionWriteIngestPartition.
+///
+/// Mirror of AdbcIngestHandle: opaque, serializable, single-use
+/// `release`. Releasing a receipt does NOT discard the underlying
+/// write; that happens at Commit (commit it) or Abort (drop it).
+///
+/// \since ADBC API revision 1.2.0
+struct AdbcIngestReceipt {
+ /// \brief The length of `bytes`.
+ size_t length;
+
+ /// \brief The serialized receipt bytes (driver-owned).
+ const uint8_t* bytes;
+
+ /// \brief Private driver state.
+ void* private_data;
+
+ /// \brief Release the receipt's memory. Sets `release` to NULL.
+ void (*release)(struct AdbcIngestReceipt* self);
+};
+/// @}
+
+/// \addtogroup adbc-connection-ingest-partition
+/// Some drivers can accept bulk writes from a distributed writer: a
+/// coordinator configures an ingest, many workers write partitions in
+/// parallel (possibly from different processes or hosts), and the
+/// coordinator commits or aborts atomically.
+///
+/// This mirrors the read-side partitioned execution model. The
+/// coordinator calls AdbcConnectionBeginIngestPartitions to obtain an
+/// opaque, serializable handle. The handle is shipped to workers by
+/// the caller (e.g. a Spark driver sending it to executors). Workers
+/// call AdbcConnectionWriteIngestPartition on their own connections —
+/// the connection does not have to be the same one that created the
+/// handle. Each write returns an opaque receipt. The coordinator
+/// collects receipts and calls AdbcConnectionCommitIngestPartitions
+/// (or AdbcConnectionAbortIngestPartitions on failure).
+///
+/// Handles and receipts are driver-defined opaque byte strings. They
+/// are safe to transmit between processes and to use concurrently
+/// from multiple connections.
+///
+/// Drivers are not required to support partitioned ingest.
+///
+/// \since ADBC API revision 1.2.0
+///
+/// @{
+
+/// \brief Begin a partitioned bulk ingest.
+///
+/// Uses the same semantics as the ADBC_INGEST_OPTION_* options on
+/// AdbcStatement. For ADBC_INGEST_OPTION_MODE_CREATE,
+/// ADBC_INGEST_OPTION_MODE_CREATE_APPEND, and
+/// ADBC_INGEST_OPTION_MODE_REPLACE, `schema` is required and the
+/// driver creates (or recreates) the target table at this call. For
+/// ADBC_INGEST_OPTION_MODE_APPEND, `schema` is optional; if provided,
+/// the driver validates it against the target and returns
+/// ADBC_STATUS_ALREADY_EXISTS on mismatch.
+///
+/// The returned handle is opaque, serializable, and usable from any
+/// connection that can open the same database. The caller releases
+/// it via `out_handle->release`; the bytes can be copied and shipped
+/// to workers before release.
+///
+/// \since ADBC API revision 1.2.0
+/// \param[in] connection The coordinator's connection.
+/// \param[in] target_catalog Catalog of the target table, or NULL.
+/// \param[in] target_db_schema Schema of the target table, or NULL.
+/// \param[in] target_table Name of the target table. Required.
+/// \param[in] mode One of ADBC_INGEST_OPTION_MODE_*. Required.
+/// \param[in] schema Arrow schema of the data to be written.
+/// Required for create/replace/create_append modes; optional for
+/// append.
+/// \param[out] out_handle Driver-owned handle. Must be released by
+/// the caller via `out_handle->release`.
+/// \param[out] error Error details, if any.
+/// \return ADBC_STATUS_INVALID_ARGUMENT if mode requires a schema
+/// but none was provided.
+/// \return ADBC_STATUS_ALREADY_EXISTS if append mode is requested
+/// and the target schema disagrees with the provided schema.
+/// \return ADBC_STATUS_NOT_IMPLEMENTED if the driver does not
+/// support partitioned ingest.
+ADBC_EXPORT
+AdbcStatusCode AdbcConnectionBeginIngestPartitions(
+ struct AdbcConnection* connection, const char* target_catalog,
+ const char* target_db_schema, const char* target_table, const char* mode,
+ struct ArrowSchema* schema, struct AdbcIngestHandle* out_handle,
+ struct AdbcError* error);
+
+/// \brief Write one partition of a partitioned bulk ingest.
+///
+/// Called by a worker, typically on a different connection than the
+/// one that created the handle. The driver reads the bound stream
+/// to completion, writes its contents to driver-specific staging
+/// (per-call: a unique staging table, unique object-store path, etc.
+/// — never shared across concurrent writes), and returns an opaque
+/// receipt.
+///
+/// The stream's schema should be compatible with the target table's
+/// schema. Drivers may validate this at any point during the write;
+/// on mismatch the call fails and produces no receipt. The exact
+/// validation mechanism is driver-specific (e.g., RDBMS drivers may
+/// rely on the staging table DDL to enforce compatibility).
+///
+/// On error of any kind, `out_receipt` is left with `release ==
+/// NULL` and the caller should retry the whole partition. Partial
+/// receipts are never produced. The driver may, however, leave
+/// partial server-side state (for example, a per-call staging
+/// table); the caller must still call `AbortIngestPartitions` for
+/// the handle (with no receipt for this failed write) to release
+/// any staging resources, or rely on driver housekeeping.
+///
+/// This call is safe to invoke concurrently from many connections
+/// using the same handle.
+///
+/// \since ADBC API revision 1.2.0
+/// \param[in] connection The worker's connection.
+/// \param[in] handle The handle bytes from Begin.
+/// \param[in] handle_len Length of handle.
+/// \param[in] data Arrow stream of partition data. The driver
+/// consumes the stream and releases it.
+/// \param[out] out_receipt Driver-owned receipt. Must be released
+/// by the caller via `out_receipt->release`.
+/// \param[out] error Error details, if any.
+ADBC_EXPORT
+AdbcStatusCode AdbcConnectionWriteIngestPartition(
+ struct AdbcConnection* connection, const uint8_t* handle, size_t handle_len,
+ struct ArrowArrayStream* data, struct AdbcIngestReceipt* out_receipt,
+ struct AdbcError* error);
+
+/// \brief Commit a partitioned bulk ingest.
+///
+/// Atomically promotes all writes named by `receipts` into the
+/// target table. Semantics of "atomic" are driver-specific: RDBMS
+/// drivers typically swap staging into the target in a transaction;
+/// table-format drivers (Iceberg, Delta) write a catalog or
+/// transaction-log entry referencing the data files in the
+/// receipts.
+///
+/// After Commit returns successfully, the handle is consumed and
+/// must not be used again.
+///
+/// Receipts from failed writes, or writes whose receipts were never
+/// observed by the coordinator, are not included in the commit.
+/// Their staging data is orphaned and is cleaned up as described in
+/// AdbcConnectionAbortIngestPartitions.
+///
+/// \since ADBC API revision 1.2.0
+/// \param[in] connection A connection — typically the coordinator's,
+/// but any connection that can open the same database works.
+/// \param[in] handle The handle from Begin.
+/// \param[in] handle_len Length of handle.
+/// \param[in] num_receipts Number of receipts in the batch.
+/// \param[in] receipts Array of receipt byte-pointers.
+/// \param[in] receipt_lens Array of receipt lengths.
+/// \param[out] rows_affected Number of rows committed, or -1 if
+/// unknown. Pass NULL if not wanted.
+/// \param[out] error Error details, if any.
+ADBC_EXPORT
+AdbcStatusCode AdbcConnectionCommitIngestPartitions(
+ struct AdbcConnection* connection, const uint8_t* handle, size_t handle_len,
+ size_t num_receipts, const uint8_t** receipts, const size_t* receipt_lens,
+ int64_t* rows_affected, struct AdbcError* error);
+
+/// \brief Abort a partitioned bulk ingest.
+///
+/// Discards all writes scoped to the handle and releases any
+/// driver-side resources. The handle is consumed.
+///
+/// Drivers must clean up every write scoped to the handle, including
+/// writes whose receipts were lost or never observed — not only
+/// those named in `receipts`. The handle is the authority for
+/// cleanup scope; `receipts`, when provided, are a hint that allows
+/// the driver to fast-path deletion of known writes.
+///
+/// Abort is best-effort. If cleanup is incomplete, the driver
+/// returns a warning status and orphaned storage may remain; it is
Review Comment:
Perhaps this would leverage `ConnectionSetWarningHandler`?
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(format): partitioned bulk ingest API [arrow-adbc]
Posted by "lidavidm (via GitHub)" <gi...@apache.org>.
lidavidm commented on code in PR #4317:
URL: https://github.com/apache/arrow-adbc/pull/4317#discussion_r3412650690
##########
c/include/arrow-adbc/adbc.h:
##########
@@ -1946,6 +1989,235 @@ AdbcStatusCode AdbcConnectionReadPartition(struct AdbcConnection* connection,
/// @}
+/// \defgroup adbc-connection-ingest-partition Partitioned Bulk Ingest
+/// @{
+
+/// \brief Driver-owned bytes returned by
+/// AdbcConnectionBeginIngestPartitions.
+///
+/// The bytes are opaque and serializable: the caller may copy
+/// `bytes[0..length)` and ship that copy to workers (other processes
+/// or hosts) which can pass it to AdbcConnectionWriteIngestPartition
+/// directly.
+///
+/// The struct itself is owned by the driver. Call `release` exactly
+/// once to free it. Releasing the handle does NOT roll back the
+/// ingest — call AdbcConnectionAbortIngestPartitions for that.
+///
+/// \since ADBC API revision 1.2.0
+struct AdbcIngestHandle {
+ /// \brief The length of `bytes`.
+ size_t length;
+
+ /// \brief The serialized handle bytes (driver-owned).
+ const uint8_t* bytes;
+
+ /// \brief Private driver state.
+ void* private_data;
+
+ /// \brief Release the handle's memory. Sets `release` to NULL.
+ void (*release)(struct AdbcIngestHandle* self);
+};
+
+/// \brief Driver-owned bytes returned by
+/// AdbcConnectionWriteIngestPartition.
+///
+/// Mirror of AdbcIngestHandle: opaque, serializable, single-use
+/// `release`. Releasing a receipt does NOT discard the underlying
+/// write; that happens at Commit (commit it) or Abort (drop it).
+///
+/// \since ADBC API revision 1.2.0
+struct AdbcIngestReceipt {
+ /// \brief The length of `bytes`.
+ size_t length;
+
+ /// \brief The serialized receipt bytes (driver-owned).
+ const uint8_t* bytes;
+
+ /// \brief Private driver state.
+ void* private_data;
+
+ /// \brief Release the receipt's memory. Sets `release` to NULL.
+ void (*release)(struct AdbcIngestReceipt* self);
+};
+/// @}
+
+/// \addtogroup adbc-connection-ingest-partition
+/// Some drivers can accept bulk writes from a distributed writer: a
+/// coordinator configures an ingest, many workers write partitions in
+/// parallel (possibly from different processes or hosts), and the
+/// coordinator commits or aborts atomically.
+///
+/// This mirrors the read-side partitioned execution model. The
+/// coordinator calls AdbcConnectionBeginIngestPartitions to obtain an
+/// opaque, serializable handle. The handle is shipped to workers by
+/// the caller (e.g. a Spark driver sending it to executors). Workers
+/// call AdbcConnectionWriteIngestPartition on their own connections —
+/// the connection does not have to be the same one that created the
+/// handle. Each write returns an opaque receipt. The coordinator
+/// collects receipts and calls AdbcConnectionCommitIngestPartitions
+/// (or AdbcConnectionAbortIngestPartitions on failure).
+///
+/// Handles and receipts are driver-defined opaque byte strings. They
+/// are safe to transmit between processes and to use concurrently
+/// from multiple connections.
+///
+/// Drivers are not required to support partitioned ingest.
+///
+/// \since ADBC API revision 1.2.0
+///
+/// @{
+
+/// \brief Begin a partitioned bulk ingest.
+///
+/// Uses the same semantics as the ADBC_INGEST_OPTION_* options on
+/// AdbcStatement. For ADBC_INGEST_OPTION_MODE_CREATE,
+/// ADBC_INGEST_OPTION_MODE_CREATE_APPEND, and
+/// ADBC_INGEST_OPTION_MODE_REPLACE, `schema` is required and the
+/// driver creates (or recreates) the target table at this call. For
+/// ADBC_INGEST_OPTION_MODE_APPEND, `schema` is optional; if provided,
+/// the driver validates it against the target and returns
+/// ADBC_STATUS_ALREADY_EXISTS on mismatch.
+///
+/// The returned handle is opaque, serializable, and usable from any
+/// connection that can open the same database. The caller releases
+/// it via `out_handle->release`; the bytes can be copied and shipped
+/// to workers before release.
+///
+/// \since ADBC API revision 1.2.0
+/// \param[in] connection The coordinator's connection.
+/// \param[in] target_catalog Catalog of the target table, or NULL.
+/// \param[in] target_db_schema Schema of the target table, or NULL.
+/// \param[in] target_table Name of the target table. Required.
+/// \param[in] mode One of ADBC_INGEST_OPTION_MODE_*. Required.
+/// \param[in] schema Arrow schema of the data to be written.
+/// Required for create/replace/create_append modes; optional for
+/// append.
+/// \param[out] out_handle Driver-owned handle. Must be released by
+/// the caller via `out_handle->release`.
+/// \param[out] error Error details, if any.
+/// \return ADBC_STATUS_INVALID_ARGUMENT if mode requires a schema
+/// but none was provided.
+/// \return ADBC_STATUS_ALREADY_EXISTS if append mode is requested
+/// and the target schema disagrees with the provided schema.
+/// \return ADBC_STATUS_NOT_IMPLEMENTED if the driver does not
+/// support partitioned ingest.
+ADBC_EXPORT
+AdbcStatusCode AdbcConnectionBeginIngestPartitions(
+ struct AdbcConnection* connection, const char* target_catalog,
+ const char* target_db_schema, const char* target_table, const char* mode,
+ struct ArrowSchema* schema, struct AdbcIngestHandle* out_handle,
+ struct AdbcError* error);
+
+/// \brief Write one partition of a partitioned bulk ingest.
+///
+/// Called by a worker, typically on a different connection than the
+/// one that created the handle. The driver reads the bound stream
+/// to completion, writes its contents to driver-specific staging
+/// (per-call: a unique staging table, unique object-store path, etc.
+/// — never shared across concurrent writes), and returns an opaque
+/// receipt.
+///
+/// The stream's schema should be compatible with the target table's
+/// schema. Drivers may validate this at any point during the write;
+/// on mismatch the call fails and produces no receipt. The exact
+/// validation mechanism is driver-specific (e.g., RDBMS drivers may
+/// rely on the staging table DDL to enforce compatibility).
+///
+/// On error of any kind, `out_receipt` is left with `release ==
+/// NULL` and the caller should retry the whole partition. Partial
+/// receipts are never produced. The driver may, however, leave
+/// partial server-side state (for example, a per-call staging
+/// table); the caller must still call `AbortIngestPartitions` for
+/// the handle (with no receipt for this failed write) to release
+/// any staging resources, or rely on driver housekeeping.
+///
+/// This call is safe to invoke concurrently from many connections
+/// using the same handle.
+///
+/// \since ADBC API revision 1.2.0
+/// \param[in] connection The worker's connection.
+/// \param[in] handle The handle bytes from Begin.
+/// \param[in] handle_len Length of handle.
+/// \param[in] data Arrow stream of partition data. The driver
+/// consumes the stream and releases it.
+/// \param[out] out_receipt Driver-owned receipt. Must be released
+/// by the caller via `out_receipt->release`.
+/// \param[out] error Error details, if any.
+ADBC_EXPORT
+AdbcStatusCode AdbcConnectionWriteIngestPartition(
+ struct AdbcConnection* connection, const uint8_t* handle, size_t handle_len,
+ struct ArrowArrayStream* data, struct AdbcIngestReceipt* out_receipt,
+ struct AdbcError* error);
+
+/// \brief Commit a partitioned bulk ingest.
+///
+/// Atomically promotes all writes named by `receipts` into the
+/// target table. Semantics of "atomic" are driver-specific: RDBMS
+/// drivers typically swap staging into the target in a transaction;
+/// table-format drivers (Iceberg, Delta) write a catalog or
+/// transaction-log entry referencing the data files in the
+/// receipts.
+///
+/// After Commit returns successfully, the handle is consumed and
+/// must not be used again.
+///
+/// Receipts from failed writes, or writes whose receipts were never
+/// observed by the coordinator, are not included in the commit.
+/// Their staging data is orphaned and is cleaned up as described in
+/// AdbcConnectionAbortIngestPartitions.
+///
+/// \since ADBC API revision 1.2.0
+/// \param[in] connection A connection — typically the coordinator's,
+/// but any connection that can open the same database works.
+/// \param[in] handle The handle from Begin.
+/// \param[in] handle_len Length of handle.
+/// \param[in] num_receipts Number of receipts in the batch.
+/// \param[in] receipts Array of receipt byte-pointers.
+/// \param[in] receipt_lens Array of receipt lengths.
+/// \param[out] rows_affected Number of rows committed, or -1 if
+/// unknown. Pass NULL if not wanted.
+/// \param[out] error Error details, if any.
+ADBC_EXPORT
+AdbcStatusCode AdbcConnectionCommitIngestPartitions(
+ struct AdbcConnection* connection, const uint8_t* handle, size_t handle_len,
+ size_t num_receipts, const uint8_t** receipts, const size_t* receipt_lens,
+ int64_t* rows_affected, struct AdbcError* error);
+
+/// \brief Abort a partitioned bulk ingest.
+///
+/// Discards all writes scoped to the handle and releases any
+/// driver-side resources. The handle is consumed.
+///
+/// Drivers must clean up every write scoped to the handle, including
+/// writes whose receipts were lost or never observed — not only
+/// those named in `receipts`. The handle is the authority for
+/// cleanup scope; `receipts`, when provided, are a hint that allows
+/// the driver to fast-path deletion of known writes.
+///
+/// Abort is best-effort. If cleanup is incomplete, the driver
+/// returns a warning status and orphaned storage may remain; it is
Review Comment:
I guess the question is, in what sorts of scenarios would it fail? And would it be possible/sensible to retry? Should we try to provide some structured information on what wasn't cleaned up so the user/an operator can try to do it manually? (In which case maybe a warning handler isn't sufficient?)
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(format): partitioned bulk ingest API [arrow-adbc]
Posted by "tokoko (via GitHub)" <gi...@apache.org>.
tokoko commented on PR #4317:
URL: https://github.com/apache/arrow-adbc/pull/4317#issuecomment-4726518894
@lidavidm I took a look at the sdk. I think the answer depends on whether we're looking at some sort of a bounded batch scenario or pure continuous streaming.
**Batch**: The main problem here is that the API currently targets a scenario where writers are used to stage the data and then a coordinator "commits" during a Complete call. For iceberg/delta, that's a natural abstraction, for databases you can work around it by staging to a temp table and doing a hopefully inexpensive swap at the end. Having said that, there's no reason a driver can't implement it w/o a staging table dance and simply start writing to a target table in which case a Complete call is kind of a no-op, the data has already been materialized. maybe we should also have an option to allow a client to configure which mode they want the driver to work in? wdyt?
So, a ZeroBus driver would implement this the same way as any other driver, either stage writes somewhere (temp Delta path, staging topic, etc.), then atomically promote them on Complete (not sure how easy that is in dbx unity) or start writing to the target table directly and Complete becomes something of a no-op. Same staging-and-swap pattern as an RDBMS driver, just with different internals.
**Streaming/continuous**: The API doesn't prevent it — a driver could treat the handle as a long-lived session token. The main value the API provides would be the handle that's produced after a centralized setup (validate target, schema, permissions once). Complete semantics will probably be different in that case. you can either ignore it or treat it as some way to store acked receipts from writers, some sort of a state management. There's also a question whether that state needs to be exposed to the client as well and whether the exposed state should be opaque or something more concrete (for example, offsets). In short, I deliberately avoided going down the streaming rabbithole in this PR, but that can be changed of course.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Partitioned ingest API [arrow-adbc]
Posted by "lidavidm (via GitHub)" <gi...@apache.org>.
lidavidm commented on PR #4317:
URL: https://github.com/apache/arrow-adbc/pull/4317#issuecomment-4503281477
Do you want to target this against the spec-1.2.0 branch instead?
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Partitioned ingest API [arrow-adbc]
Posted by "tokoko (via GitHub)" <gi...@apache.org>.
tokoko commented on code in PR #4317:
URL: https://github.com/apache/arrow-adbc/pull/4317#discussion_r3322147949
##########
c/include/arrow-adbc/adbc.h:
##########
@@ -1946,6 +1989,235 @@ AdbcStatusCode AdbcConnectionReadPartition(struct AdbcConnection* connection,
/// @}
+/// \defgroup adbc-connection-ingest-partition Partitioned Bulk Ingest
+/// @{
+
+/// \brief Driver-owned bytes returned by
+/// AdbcConnectionBeginIngestPartitions.
+///
+/// The bytes are opaque and serializable: the caller may copy
+/// `bytes[0..length)` and ship that copy to workers (other processes
+/// or hosts) which can pass it to AdbcConnectionWriteIngestPartition
+/// directly.
+///
+/// The struct itself is owned by the driver. Call `release` exactly
+/// once to free it. Releasing the handle does NOT roll back the
+/// ingest — call AdbcConnectionAbortIngestPartitions for that.
+///
+/// \since ADBC API revision 1.2.0
+struct AdbcIngestHandle {
+ /// \brief The length of `bytes`.
+ size_t length;
+
+ /// \brief The serialized handle bytes (driver-owned).
+ const uint8_t* bytes;
+
+ /// \brief Private driver state.
+ void* private_data;
+
+ /// \brief Release the handle's memory. Sets `release` to NULL.
+ void (*release)(struct AdbcIngestHandle* self);
+};
+
+/// \brief Driver-owned bytes returned by
+/// AdbcConnectionWriteIngestPartition.
+///
+/// Mirror of AdbcIngestHandle: opaque, serializable, single-use
+/// `release`. Releasing a receipt does NOT discard the underlying
+/// write; that happens at Commit (commit it) or Abort (drop it).
+///
+/// \since ADBC API revision 1.2.0
+struct AdbcIngestReceipt {
+ /// \brief The length of `bytes`.
+ size_t length;
+
+ /// \brief The serialized receipt bytes (driver-owned).
+ const uint8_t* bytes;
+
+ /// \brief Private driver state.
+ void* private_data;
+
+ /// \brief Release the receipt's memory. Sets `release` to NULL.
+ void (*release)(struct AdbcIngestReceipt* self);
+};
+/// @}
+
+/// \addtogroup adbc-connection-ingest-partition
+/// Some drivers can accept bulk writes from a distributed writer: a
+/// coordinator configures an ingest, many workers write partitions in
+/// parallel (possibly from different processes or hosts), and the
+/// coordinator commits or aborts atomically.
+///
+/// This mirrors the read-side partitioned execution model. The
+/// coordinator calls AdbcConnectionBeginIngestPartitions to obtain an
+/// opaque, serializable handle. The handle is shipped to workers by
+/// the caller (e.g. a Spark driver sending it to executors). Workers
+/// call AdbcConnectionWriteIngestPartition on their own connections —
+/// the connection does not have to be the same one that created the
+/// handle. Each write returns an opaque receipt. The coordinator
+/// collects receipts and calls AdbcConnectionCommitIngestPartitions
+/// (or AdbcConnectionAbortIngestPartitions on failure).
+///
+/// Handles and receipts are driver-defined opaque byte strings. They
+/// are safe to transmit between processes and to use concurrently
+/// from multiple connections.
+///
+/// Drivers are not required to support partitioned ingest.
+///
+/// \since ADBC API revision 1.2.0
+///
+/// @{
+
+/// \brief Begin a partitioned bulk ingest.
+///
+/// Uses the same semantics as the ADBC_INGEST_OPTION_* options on
+/// AdbcStatement. For ADBC_INGEST_OPTION_MODE_CREATE,
+/// ADBC_INGEST_OPTION_MODE_CREATE_APPEND, and
+/// ADBC_INGEST_OPTION_MODE_REPLACE, `schema` is required and the
+/// driver creates (or recreates) the target table at this call. For
+/// ADBC_INGEST_OPTION_MODE_APPEND, `schema` is optional; if provided,
+/// the driver validates it against the target and returns
+/// ADBC_STATUS_ALREADY_EXISTS on mismatch.
+///
+/// The returned handle is opaque, serializable, and usable from any
+/// connection that can open the same database. The caller releases
+/// it via `out_handle->release`; the bytes can be copied and shipped
+/// to workers before release.
+///
+/// \since ADBC API revision 1.2.0
+/// \param[in] connection The coordinator's connection.
+/// \param[in] target_catalog Catalog of the target table, or NULL.
+/// \param[in] target_db_schema Schema of the target table, or NULL.
+/// \param[in] target_table Name of the target table. Required.
+/// \param[in] mode One of ADBC_INGEST_OPTION_MODE_*. Required.
+/// \param[in] schema Arrow schema of the data to be written.
+/// Required for create/replace/create_append modes; optional for
+/// append.
+/// \param[out] out_handle Driver-owned handle. Must be released by
+/// the caller via `out_handle->release`.
+/// \param[out] error Error details, if any.
+/// \return ADBC_STATUS_INVALID_ARGUMENT if mode requires a schema
+/// but none was provided.
+/// \return ADBC_STATUS_ALREADY_EXISTS if append mode is requested
+/// and the target schema disagrees with the provided schema.
+/// \return ADBC_STATUS_NOT_IMPLEMENTED if the driver does not
+/// support partitioned ingest.
+ADBC_EXPORT
+AdbcStatusCode AdbcConnectionBeginIngestPartitions(
+ struct AdbcConnection* connection, const char* target_catalog,
+ const char* target_db_schema, const char* target_table, const char* mode,
+ struct ArrowSchema* schema, struct AdbcIngestHandle* out_handle,
+ struct AdbcError* error);
+
+/// \brief Write one partition of a partitioned bulk ingest.
+///
+/// Called by a worker, typically on a different connection than the
+/// one that created the handle. The driver reads the bound stream
+/// to completion, writes its contents to driver-specific staging
+/// (per-call: a unique staging table, unique object-store path, etc.
+/// — never shared across concurrent writes), and returns an opaque
+/// receipt.
+///
+/// The stream's schema should be compatible with the target table's
+/// schema. Drivers may validate this at any point during the write;
+/// on mismatch the call fails and produces no receipt. The exact
+/// validation mechanism is driver-specific (e.g., RDBMS drivers may
+/// rely on the staging table DDL to enforce compatibility).
+///
+/// On error of any kind, `out_receipt` is left with `release ==
+/// NULL` and the caller should retry the whole partition. Partial
+/// receipts are never produced. The driver may, however, leave
+/// partial server-side state (for example, a per-call staging
+/// table); the caller must still call `AbortIngestPartitions` for
+/// the handle (with no receipt for this failed write) to release
+/// any staging resources, or rely on driver housekeeping.
+///
+/// This call is safe to invoke concurrently from many connections
+/// using the same handle.
+///
+/// \since ADBC API revision 1.2.0
+/// \param[in] connection The worker's connection.
+/// \param[in] handle The handle bytes from Begin.
+/// \param[in] handle_len Length of handle.
+/// \param[in] data Arrow stream of partition data. The driver
+/// consumes the stream and releases it.
+/// \param[out] out_receipt Driver-owned receipt. Must be released
+/// by the caller via `out_receipt->release`.
+/// \param[out] error Error details, if any.
+ADBC_EXPORT
+AdbcStatusCode AdbcConnectionWriteIngestPartition(
+ struct AdbcConnection* connection, const uint8_t* handle, size_t handle_len,
+ struct ArrowArrayStream* data, struct AdbcIngestReceipt* out_receipt,
+ struct AdbcError* error);
+
+/// \brief Commit a partitioned bulk ingest.
+///
+/// Atomically promotes all writes named by `receipts` into the
+/// target table. Semantics of "atomic" are driver-specific: RDBMS
+/// drivers typically swap staging into the target in a transaction;
+/// table-format drivers (Iceberg, Delta) write a catalog or
+/// transaction-log entry referencing the data files in the
+/// receipts.
+///
+/// After Commit returns successfully, the handle is consumed and
+/// must not be used again.
+///
+/// Receipts from failed writes, or writes whose receipts were never
+/// observed by the coordinator, are not included in the commit.
+/// Their staging data is orphaned and is cleaned up as described in
+/// AdbcConnectionAbortIngestPartitions.
+///
+/// \since ADBC API revision 1.2.0
+/// \param[in] connection A connection — typically the coordinator's,
+/// but any connection that can open the same database works.
+/// \param[in] handle The handle from Begin.
+/// \param[in] handle_len Length of handle.
+/// \param[in] num_receipts Number of receipts in the batch.
+/// \param[in] receipts Array of receipt byte-pointers.
+/// \param[in] receipt_lens Array of receipt lengths.
+/// \param[out] rows_affected Number of rows committed, or -1 if
+/// unknown. Pass NULL if not wanted.
+/// \param[out] error Error details, if any.
+ADBC_EXPORT
+AdbcStatusCode AdbcConnectionCommitIngestPartitions(
+ struct AdbcConnection* connection, const uint8_t* handle, size_t handle_len,
+ size_t num_receipts, const uint8_t** receipts, const size_t* receipt_lens,
+ int64_t* rows_affected, struct AdbcError* error);
+
+/// \brief Abort a partitioned bulk ingest.
+///
+/// Discards all writes scoped to the handle and releases any
+/// driver-side resources. The handle is consumed.
+///
+/// Drivers must clean up every write scoped to the handle, including
+/// writes whose receipts were lost or never observed — not only
+/// those named in `receipts`. The handle is the authority for
+/// cleanup scope; `receipts`, when provided, are a hint that allows
+/// the driver to fast-path deletion of known writes.
+///
+/// Abort is best-effort. If cleanup is incomplete, the driver
+/// returns a warning status and orphaned storage may remain; it is
Review Comment:
I meant some sort of an "okish" adbc status, but warning handler sounds like a more appropriate solution. unless you think it's not necessary at all.. just documenting it as best-effort could be enough as well.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(format): partitioned bulk ingest API [arrow-adbc]
Posted by "lidavidm (via GitHub)" <gi...@apache.org>.
lidavidm commented on code in PR #4317:
URL: https://github.com/apache/arrow-adbc/pull/4317#discussion_r3793401671
##########
c/include/arrow-adbc/adbc.h:
##########
@@ -1946,6 +1989,235 @@ AdbcStatusCode AdbcConnectionReadPartition(struct AdbcConnection* connection,
/// @}
+/// \defgroup adbc-connection-ingest-partition Partitioned Bulk Ingest
+/// @{
+
+/// \brief Driver-owned bytes returned by
+/// AdbcConnectionBeginIngestPartitions.
+///
+/// The bytes are opaque and serializable: the caller may copy
+/// `bytes[0..length)` and ship that copy to workers (other processes
+/// or hosts) which can pass it to AdbcConnectionWriteIngestPartition
+/// directly.
+///
+/// The struct itself is owned by the driver. Call `release` exactly
+/// once to free it. Releasing the handle does NOT roll back the
+/// ingest — call AdbcConnectionAbortIngestPartitions for that.
+///
+/// \since ADBC API revision 1.2.0
+struct AdbcIngestHandle {
+ /// \brief The length of `bytes`.
+ size_t length;
+
+ /// \brief The serialized handle bytes (driver-owned).
+ const uint8_t* bytes;
+
+ /// \brief Private driver state.
+ void* private_data;
+
+ /// \brief Release the handle's memory. Sets `release` to NULL.
+ void (*release)(struct AdbcIngestHandle* self);
+};
+
+/// \brief Driver-owned bytes returned by
+/// AdbcConnectionWriteIngestPartition.
+///
+/// Mirror of AdbcIngestHandle: opaque, serializable, single-use
+/// `release`. Releasing a receipt does NOT discard the underlying
+/// write; that happens at Commit (commit it) or Abort (drop it).
+///
+/// \since ADBC API revision 1.2.0
+struct AdbcIngestReceipt {
+ /// \brief The length of `bytes`.
+ size_t length;
+
+ /// \brief The serialized receipt bytes (driver-owned).
+ const uint8_t* bytes;
+
+ /// \brief Private driver state.
+ void* private_data;
+
+ /// \brief Release the receipt's memory. Sets `release` to NULL.
+ void (*release)(struct AdbcIngestReceipt* self);
+};
+/// @}
+
+/// \addtogroup adbc-connection-ingest-partition
+/// Some drivers can accept bulk writes from a distributed writer: a
+/// coordinator configures an ingest, many workers write partitions in
+/// parallel (possibly from different processes or hosts), and the
+/// coordinator commits or aborts atomically.
+///
+/// This mirrors the read-side partitioned execution model. The
+/// coordinator calls AdbcConnectionBeginIngestPartitions to obtain an
+/// opaque, serializable handle. The handle is shipped to workers by
+/// the caller (e.g. a Spark driver sending it to executors). Workers
+/// call AdbcConnectionWriteIngestPartition on their own connections —
+/// the connection does not have to be the same one that created the
+/// handle. Each write returns an opaque receipt. The coordinator
+/// collects receipts and calls AdbcConnectionCommitIngestPartitions
+/// (or AdbcConnectionAbortIngestPartitions on failure).
+///
+/// Handles and receipts are driver-defined opaque byte strings. They
+/// are safe to transmit between processes and to use concurrently
+/// from multiple connections.
+///
+/// Drivers are not required to support partitioned ingest.
+///
+/// \since ADBC API revision 1.2.0
+///
+/// @{
+
+/// \brief Begin a partitioned bulk ingest.
+///
+/// Uses the same semantics as the ADBC_INGEST_OPTION_* options on
+/// AdbcStatement. For ADBC_INGEST_OPTION_MODE_CREATE,
+/// ADBC_INGEST_OPTION_MODE_CREATE_APPEND, and
+/// ADBC_INGEST_OPTION_MODE_REPLACE, `schema` is required and the
+/// driver creates (or recreates) the target table at this call. For
+/// ADBC_INGEST_OPTION_MODE_APPEND, `schema` is optional; if provided,
+/// the driver validates it against the target and returns
+/// ADBC_STATUS_ALREADY_EXISTS on mismatch.
+///
+/// The returned handle is opaque, serializable, and usable from any
+/// connection that can open the same database. The caller releases
+/// it via `out_handle->release`; the bytes can be copied and shipped
+/// to workers before release.
+///
+/// \since ADBC API revision 1.2.0
+/// \param[in] connection The coordinator's connection.
+/// \param[in] target_catalog Catalog of the target table, or NULL.
+/// \param[in] target_db_schema Schema of the target table, or NULL.
+/// \param[in] target_table Name of the target table. Required.
+/// \param[in] mode One of ADBC_INGEST_OPTION_MODE_*. Required.
+/// \param[in] schema Arrow schema of the data to be written.
+/// Required for create/replace/create_append modes; optional for
+/// append.
+/// \param[out] out_handle Driver-owned handle. Must be released by
+/// the caller via `out_handle->release`.
+/// \param[out] error Error details, if any.
+/// \return ADBC_STATUS_INVALID_ARGUMENT if mode requires a schema
+/// but none was provided.
+/// \return ADBC_STATUS_ALREADY_EXISTS if append mode is requested
+/// and the target schema disagrees with the provided schema.
+/// \return ADBC_STATUS_NOT_IMPLEMENTED if the driver does not
+/// support partitioned ingest.
+ADBC_EXPORT
+AdbcStatusCode AdbcConnectionBeginIngestPartitions(
+ struct AdbcConnection* connection, const char* target_catalog,
+ const char* target_db_schema, const char* target_table, const char* mode,
Review Comment:
From https://github.com/apache/arrow-adbc/pull/4544#issuecomment-5290153012 I think we can keep iterating on this.
Maybe we place this at statement level (even if it's not really associated with a statement per se) so we can scope the options? Or we encode options flexibly (as suggested there)
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(format): partitioned bulk ingest API [arrow-adbc]
Posted by "tokoko (via GitHub)" <gi...@apache.org>.
tokoko commented on code in PR #4317:
URL: https://github.com/apache/arrow-adbc/pull/4317#discussion_r3416146288
##########
c/include/arrow-adbc/adbc.h:
##########
@@ -1946,6 +1989,235 @@ AdbcStatusCode AdbcConnectionReadPartition(struct AdbcConnection* connection,
/// @}
+/// \defgroup adbc-connection-ingest-partition Partitioned Bulk Ingest
+/// @{
+
+/// \brief Driver-owned bytes returned by
+/// AdbcConnectionBeginIngestPartitions.
+///
+/// The bytes are opaque and serializable: the caller may copy
+/// `bytes[0..length)` and ship that copy to workers (other processes
+/// or hosts) which can pass it to AdbcConnectionWriteIngestPartition
+/// directly.
+///
+/// The struct itself is owned by the driver. Call `release` exactly
+/// once to free it. Releasing the handle does NOT roll back the
+/// ingest — call AdbcConnectionAbortIngestPartitions for that.
+///
+/// \since ADBC API revision 1.2.0
+struct AdbcIngestHandle {
+ /// \brief The length of `bytes`.
+ size_t length;
+
+ /// \brief The serialized handle bytes (driver-owned).
+ const uint8_t* bytes;
+
+ /// \brief Private driver state.
+ void* private_data;
+
+ /// \brief Release the handle's memory. Sets `release` to NULL.
+ void (*release)(struct AdbcIngestHandle* self);
+};
+
+/// \brief Driver-owned bytes returned by
+/// AdbcConnectionWriteIngestPartition.
+///
+/// Mirror of AdbcIngestHandle: opaque, serializable, single-use
+/// `release`. Releasing a receipt does NOT discard the underlying
+/// write; that happens at Commit (commit it) or Abort (drop it).
+///
+/// \since ADBC API revision 1.2.0
+struct AdbcIngestReceipt {
+ /// \brief The length of `bytes`.
+ size_t length;
+
+ /// \brief The serialized receipt bytes (driver-owned).
+ const uint8_t* bytes;
+
+ /// \brief Private driver state.
+ void* private_data;
+
+ /// \brief Release the receipt's memory. Sets `release` to NULL.
+ void (*release)(struct AdbcIngestReceipt* self);
+};
+/// @}
+
+/// \addtogroup adbc-connection-ingest-partition
+/// Some drivers can accept bulk writes from a distributed writer: a
+/// coordinator configures an ingest, many workers write partitions in
+/// parallel (possibly from different processes or hosts), and the
+/// coordinator commits or aborts atomically.
+///
+/// This mirrors the read-side partitioned execution model. The
+/// coordinator calls AdbcConnectionBeginIngestPartitions to obtain an
+/// opaque, serializable handle. The handle is shipped to workers by
+/// the caller (e.g. a Spark driver sending it to executors). Workers
+/// call AdbcConnectionWriteIngestPartition on their own connections —
+/// the connection does not have to be the same one that created the
+/// handle. Each write returns an opaque receipt. The coordinator
+/// collects receipts and calls AdbcConnectionCommitIngestPartitions
+/// (or AdbcConnectionAbortIngestPartitions on failure).
+///
+/// Handles and receipts are driver-defined opaque byte strings. They
+/// are safe to transmit between processes and to use concurrently
+/// from multiple connections.
+///
+/// Drivers are not required to support partitioned ingest.
+///
+/// \since ADBC API revision 1.2.0
+///
+/// @{
+
+/// \brief Begin a partitioned bulk ingest.
+///
+/// Uses the same semantics as the ADBC_INGEST_OPTION_* options on
+/// AdbcStatement. For ADBC_INGEST_OPTION_MODE_CREATE,
+/// ADBC_INGEST_OPTION_MODE_CREATE_APPEND, and
+/// ADBC_INGEST_OPTION_MODE_REPLACE, `schema` is required and the
+/// driver creates (or recreates) the target table at this call. For
+/// ADBC_INGEST_OPTION_MODE_APPEND, `schema` is optional; if provided,
+/// the driver validates it against the target and returns
+/// ADBC_STATUS_ALREADY_EXISTS on mismatch.
+///
+/// The returned handle is opaque, serializable, and usable from any
+/// connection that can open the same database. The caller releases
+/// it via `out_handle->release`; the bytes can be copied and shipped
+/// to workers before release.
+///
+/// \since ADBC API revision 1.2.0
+/// \param[in] connection The coordinator's connection.
+/// \param[in] target_catalog Catalog of the target table, or NULL.
+/// \param[in] target_db_schema Schema of the target table, or NULL.
+/// \param[in] target_table Name of the target table. Required.
+/// \param[in] mode One of ADBC_INGEST_OPTION_MODE_*. Required.
+/// \param[in] schema Arrow schema of the data to be written.
+/// Required for create/replace/create_append modes; optional for
+/// append.
+/// \param[out] out_handle Driver-owned handle. Must be released by
+/// the caller via `out_handle->release`.
+/// \param[out] error Error details, if any.
+/// \return ADBC_STATUS_INVALID_ARGUMENT if mode requires a schema
+/// but none was provided.
+/// \return ADBC_STATUS_ALREADY_EXISTS if append mode is requested
+/// and the target schema disagrees with the provided schema.
+/// \return ADBC_STATUS_NOT_IMPLEMENTED if the driver does not
+/// support partitioned ingest.
+ADBC_EXPORT
+AdbcStatusCode AdbcConnectionBeginIngestPartitions(
+ struct AdbcConnection* connection, const char* target_catalog,
+ const char* target_db_schema, const char* target_table, const char* mode,
Review Comment:
sure, existing ones should be fine, I think. dropped `target_catalog`, `target_db_schema`, `target_table` and `mode` from signature.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(format): partitioned bulk ingest API [arrow-adbc]
Posted by "lidavidm (via GitHub)" <gi...@apache.org>.
lidavidm commented on PR #4317:
URL: https://github.com/apache/arrow-adbc/pull/4317#issuecomment-4723902602
How do you think this API would map to https://www.databricks.com/blog/ingesting-milky-way-petabyte-scale-zerobus-ingest?
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(format): partitioned bulk ingest API [arrow-adbc]
Posted by "tokoko (via GitHub)" <gi...@apache.org>.
tokoko commented on code in PR #4317:
URL: https://github.com/apache/arrow-adbc/pull/4317#discussion_r3431338012
##########
c/include/arrow-adbc/adbc.h:
##########
@@ -1516,6 +1518,23 @@ struct ADBC_EXPORT AdbcDriver {
AdbcStatusCode (*StatementExecuteMulti)(struct AdbcStatement*,
struct AdbcMultiResultSet*, struct AdbcError*);
+ AdbcStatusCode (*ConnectionBeginIngestPartitions)(struct AdbcConnection*,
+ struct ArrowSchema*,
+ struct AdbcSerializableHandle*,
+ struct AdbcError*);
+ AdbcStatusCode (*ConnectionWriteIngestPartition)(struct AdbcConnection*, const uint8_t*,
+ size_t, struct ArrowArrayStream*,
+ struct AdbcSerializableHandle*,
+ struct AdbcError*);
+ AdbcStatusCode (*ConnectionCompleteIngestPartitions)(struct AdbcConnection*,
Review Comment:
@lidavidm I think we need to distinguish between 3 possible groups of status codes of a Complete call:
- **success**: ADBC_STATUS_OK
- **retryable**: Complete failed, but a client should retry a complete call w/o restaging the data or regenerating receipts. The example is delta/iceberg concurrency conflict error.
- **terminal**: something else went wrong. start from scratch.
I couldn't really map retryable status to any existing status code. I'm thinking of adding either `ADBC_STATUS_CONFLICT` or `ADBC_STATUS_RETRY`. wdyt?
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(format): partitioned bulk ingest API [arrow-adbc]
Posted by "paleolimbot (via GitHub)" <gi...@apache.org>.
paleolimbot commented on code in PR #4317:
URL: https://github.com/apache/arrow-adbc/pull/4317#discussion_r4266272077
##########
docs/source/format/partitioned_bulk_ingest.rst:
##########
@@ -0,0 +1,402 @@
+.. Licensed to the Apache Software Foundation (ASF) under one
+.. or more contributor license agreements. See the NOTICE file
+.. distributed with this work for additional information
+.. regarding copyright ownership. The ASF licenses this file
+.. to you under the Apache License, Version 2.0 (the
+.. "License"); you may not use this file except in compliance
+.. with the License. You may obtain a copy of the License at
+..
+.. http://www.apache.org/licenses/LICENSE-2.0
+
+================================
+Proposal: Partitioned Bulk Ingest
+================================
+
+.. note::
+
+ Status: draft. Targets ADBC API revision 1.2.0.
+
+Motivation
+==========
+
+Today ADBC supports two ingest shapes:
+
+- **Single-writer bulk ingest** — one connection, one statement, one
+ ``ArrowArrayStream``, one transaction. Good for loading from a single
+ process; useless for distributed writers.
+- **Per-row binding** — slower, also single-connection.
+
+Two real workloads do not fit:
+
+1. **Distributed-writer to RDBMS.** A Spark/Flink/Beam job runs N
Review Comment:
Is it worth adding that a standard multithreaded / partitioned write on a single connection applies here too?
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(format): partitioned bulk ingest API [arrow-adbc]
Posted by "tokoko (via GitHub)" <gi...@apache.org>.
tokoko commented on PR #4317:
URL: https://github.com/apache/arrow-adbc/pull/4317#issuecomment-4606891030
some of the core cpp adbc manager codebase differs between main and spec-1.2.0. so I left only spec changes here. kept the impl in another [branch](https://github.com/tokoko/arrow-adbc/tree/partitioned-ingest-full) on my repo.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(format): partitioned bulk ingest API [arrow-adbc]
Posted by "lidavidm (via GitHub)" <gi...@apache.org>.
lidavidm commented on code in PR #4317:
URL: https://github.com/apache/arrow-adbc/pull/4317#discussion_r3435900511
##########
c/include/arrow-adbc/adbc.h:
##########
@@ -1516,6 +1518,23 @@ struct ADBC_EXPORT AdbcDriver {
AdbcStatusCode (*StatementExecuteMulti)(struct AdbcStatement*,
struct AdbcMultiResultSet*, struct AdbcError*);
+ AdbcStatusCode (*ConnectionBeginIngestPartitions)(struct AdbcConnection*,
+ struct ArrowSchema*,
+ struct AdbcSerializableHandle*,
+ struct AdbcError*);
+ AdbcStatusCode (*ConnectionWriteIngestPartition)(struct AdbcConnection*, const uint8_t*,
+ size_t, struct ArrowArrayStream*,
+ struct AdbcSerializableHandle*,
+ struct AdbcError*);
+ AdbcStatusCode (*ConnectionCompleteIngestPartitions)(struct AdbcConnection*,
Review Comment:
I hear you, but my worry is that a new error code could be in principle returned from any API function, which would break existing users.
If we say this code is only used for this particular API function, then I think it makes more sense for an out parameter than an error code (which has a non-local impact).
Also language bindings could use different strategies, e.g. Rust would return an `Result<PartitionedIngestStatus>` or something and not an out parameter, and deal with the messiness at the FFI layer. This is already the case for various APIs (e.g. Java treats bulk ingest itself differently).
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(format): partitioned bulk ingest API [arrow-adbc]
Posted by "lidavidm (via GitHub)" <gi...@apache.org>.
lidavidm commented on code in PR #4317:
URL: https://github.com/apache/arrow-adbc/pull/4317#discussion_r3412656876
##########
c/include/arrow-adbc/adbc.h:
##########
@@ -1946,6 +1989,235 @@ AdbcStatusCode AdbcConnectionReadPartition(struct AdbcConnection* connection,
/// @}
+/// \defgroup adbc-connection-ingest-partition Partitioned Bulk Ingest
+/// @{
+
+/// \brief Driver-owned bytes returned by
+/// AdbcConnectionBeginIngestPartitions.
+///
+/// The bytes are opaque and serializable: the caller may copy
+/// `bytes[0..length)` and ship that copy to workers (other processes
+/// or hosts) which can pass it to AdbcConnectionWriteIngestPartition
+/// directly.
+///
+/// The struct itself is owned by the driver. Call `release` exactly
+/// once to free it. Releasing the handle does NOT roll back the
+/// ingest — call AdbcConnectionAbortIngestPartitions for that.
+///
+/// \since ADBC API revision 1.2.0
+struct AdbcIngestHandle {
+ /// \brief The length of `bytes`.
+ size_t length;
+
+ /// \brief The serialized handle bytes (driver-owned).
+ const uint8_t* bytes;
+
+ /// \brief Private driver state.
+ void* private_data;
+
+ /// \brief Release the handle's memory. Sets `release` to NULL.
+ void (*release)(struct AdbcIngestHandle* self);
+};
+
+/// \brief Driver-owned bytes returned by
+/// AdbcConnectionWriteIngestPartition.
+///
+/// Mirror of AdbcIngestHandle: opaque, serializable, single-use
+/// `release`. Releasing a receipt does NOT discard the underlying
+/// write; that happens at Commit (commit it) or Abort (drop it).
+///
+/// \since ADBC API revision 1.2.0
+struct AdbcIngestReceipt {
+ /// \brief The length of `bytes`.
+ size_t length;
+
+ /// \brief The serialized receipt bytes (driver-owned).
+ const uint8_t* bytes;
+
+ /// \brief Private driver state.
+ void* private_data;
+
+ /// \brief Release the receipt's memory. Sets `release` to NULL.
+ void (*release)(struct AdbcIngestReceipt* self);
+};
+/// @}
+
+/// \addtogroup adbc-connection-ingest-partition
+/// Some drivers can accept bulk writes from a distributed writer: a
+/// coordinator configures an ingest, many workers write partitions in
+/// parallel (possibly from different processes or hosts), and the
+/// coordinator commits or aborts atomically.
+///
+/// This mirrors the read-side partitioned execution model. The
+/// coordinator calls AdbcConnectionBeginIngestPartitions to obtain an
+/// opaque, serializable handle. The handle is shipped to workers by
+/// the caller (e.g. a Spark driver sending it to executors). Workers
+/// call AdbcConnectionWriteIngestPartition on their own connections —
+/// the connection does not have to be the same one that created the
+/// handle. Each write returns an opaque receipt. The coordinator
+/// collects receipts and calls AdbcConnectionCommitIngestPartitions
+/// (or AdbcConnectionAbortIngestPartitions on failure).
+///
+/// Handles and receipts are driver-defined opaque byte strings. They
+/// are safe to transmit between processes and to use concurrently
+/// from multiple connections.
+///
+/// Drivers are not required to support partitioned ingest.
+///
+/// \since ADBC API revision 1.2.0
+///
+/// @{
+
+/// \brief Begin a partitioned bulk ingest.
+///
+/// Uses the same semantics as the ADBC_INGEST_OPTION_* options on
+/// AdbcStatement. For ADBC_INGEST_OPTION_MODE_CREATE,
+/// ADBC_INGEST_OPTION_MODE_CREATE_APPEND, and
+/// ADBC_INGEST_OPTION_MODE_REPLACE, `schema` is required and the
+/// driver creates (or recreates) the target table at this call. For
+/// ADBC_INGEST_OPTION_MODE_APPEND, `schema` is optional; if provided,
+/// the driver validates it against the target and returns
+/// ADBC_STATUS_ALREADY_EXISTS on mismatch.
+///
+/// The returned handle is opaque, serializable, and usable from any
+/// connection that can open the same database. The caller releases
+/// it via `out_handle->release`; the bytes can be copied and shipped
+/// to workers before release.
+///
+/// \since ADBC API revision 1.2.0
+/// \param[in] connection The coordinator's connection.
+/// \param[in] target_catalog Catalog of the target table, or NULL.
+/// \param[in] target_db_schema Schema of the target table, or NULL.
+/// \param[in] target_table Name of the target table. Required.
+/// \param[in] mode One of ADBC_INGEST_OPTION_MODE_*. Required.
+/// \param[in] schema Arrow schema of the data to be written.
+/// Required for create/replace/create_append modes; optional for
+/// append.
+/// \param[out] out_handle Driver-owned handle. Must be released by
+/// the caller via `out_handle->release`.
+/// \param[out] error Error details, if any.
+/// \return ADBC_STATUS_INVALID_ARGUMENT if mode requires a schema
+/// but none was provided.
+/// \return ADBC_STATUS_ALREADY_EXISTS if append mode is requested
+/// and the target schema disagrees with the provided schema.
+/// \return ADBC_STATUS_NOT_IMPLEMENTED if the driver does not
+/// support partitioned ingest.
+ADBC_EXPORT
+AdbcStatusCode AdbcConnectionBeginIngestPartitions(
+ struct AdbcConnection* connection, const char* target_catalog,
+ const char* target_db_schema, const char* target_table, const char* mode,
Review Comment:
Well, they wouldn't be "moved", just also accepted at the connection level for this operation. Or we could add new options to make the intent clearer. I just think it would be more consistent/flexible to keep them as options instead of baking them into the signature
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(format): partitioned bulk ingest API [arrow-adbc]
Posted by "tokoko (via GitHub)" <gi...@apache.org>.
tokoko commented on code in PR #4317:
URL: https://github.com/apache/arrow-adbc/pull/4317#discussion_r3416129977
##########
c/include/arrow-adbc/adbc.h:
##########
@@ -1135,6 +1146,30 @@ struct ADBC_EXPORT AdbcDriver {
struct AdbcError*);
/// @}
+
+ /// \defgroup adbc-1.2.0 ADBC API Revision 1.2.0
+ ///
+ /// Functions added in ADBC 1.2.0. For backwards compatibility,
+ /// these members must not be accessed unless the version passed to
+ /// the AdbcDriverInitFunc is greater than or equal to
+ /// ADBC_VERSION_1_2_0.
+ ///
+ /// @{
+
+ AdbcStatusCode (*ConnectionBeginIngestPartitions)(
+ struct AdbcConnection*, const char*, const char*, const char*, const char*,
+ struct ArrowSchema*, struct AdbcIngestHandle*, struct AdbcError*);
+ AdbcStatusCode (*ConnectionWriteIngestPartition)(
+ struct AdbcConnection*, const uint8_t*, size_t, struct ArrowArrayStream*,
+ struct AdbcIngestReceipt*, struct AdbcError*);
+ AdbcStatusCode (*ConnectionCommitIngestPartitions)(
Review Comment:
done
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Partitioned ingest API [arrow-adbc]
Posted by "lidavidm (via GitHub)" <gi...@apache.org>.
lidavidm commented on PR #4317:
URL: https://github.com/apache/arrow-adbc/pull/4317#issuecomment-4572076686
Impl is fine as well, but the branch seems to have picked up a lot of extra commits
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(format): partitioned bulk ingest API [arrow-adbc]
Posted by "lidavidm (via GitHub)" <gi...@apache.org>.
lidavidm commented on code in PR #4317:
URL: https://github.com/apache/arrow-adbc/pull/4317#discussion_r3795527336
##########
c/include/arrow-adbc/adbc.h:
##########
@@ -1946,6 +1989,235 @@ AdbcStatusCode AdbcConnectionReadPartition(struct AdbcConnection* connection,
/// @}
+/// \defgroup adbc-connection-ingest-partition Partitioned Bulk Ingest
+/// @{
+
+/// \brief Driver-owned bytes returned by
+/// AdbcConnectionBeginIngestPartitions.
+///
+/// The bytes are opaque and serializable: the caller may copy
+/// `bytes[0..length)` and ship that copy to workers (other processes
+/// or hosts) which can pass it to AdbcConnectionWriteIngestPartition
+/// directly.
+///
+/// The struct itself is owned by the driver. Call `release` exactly
+/// once to free it. Releasing the handle does NOT roll back the
+/// ingest — call AdbcConnectionAbortIngestPartitions for that.
+///
+/// \since ADBC API revision 1.2.0
+struct AdbcIngestHandle {
+ /// \brief The length of `bytes`.
+ size_t length;
+
+ /// \brief The serialized handle bytes (driver-owned).
+ const uint8_t* bytes;
+
+ /// \brief Private driver state.
+ void* private_data;
+
+ /// \brief Release the handle's memory. Sets `release` to NULL.
+ void (*release)(struct AdbcIngestHandle* self);
+};
+
+/// \brief Driver-owned bytes returned by
+/// AdbcConnectionWriteIngestPartition.
+///
+/// Mirror of AdbcIngestHandle: opaque, serializable, single-use
+/// `release`. Releasing a receipt does NOT discard the underlying
+/// write; that happens at Commit (commit it) or Abort (drop it).
+///
+/// \since ADBC API revision 1.2.0
+struct AdbcIngestReceipt {
+ /// \brief The length of `bytes`.
+ size_t length;
+
+ /// \brief The serialized receipt bytes (driver-owned).
+ const uint8_t* bytes;
+
+ /// \brief Private driver state.
+ void* private_data;
+
+ /// \brief Release the receipt's memory. Sets `release` to NULL.
+ void (*release)(struct AdbcIngestReceipt* self);
+};
+/// @}
+
+/// \addtogroup adbc-connection-ingest-partition
+/// Some drivers can accept bulk writes from a distributed writer: a
+/// coordinator configures an ingest, many workers write partitions in
+/// parallel (possibly from different processes or hosts), and the
+/// coordinator commits or aborts atomically.
+///
+/// This mirrors the read-side partitioned execution model. The
+/// coordinator calls AdbcConnectionBeginIngestPartitions to obtain an
+/// opaque, serializable handle. The handle is shipped to workers by
+/// the caller (e.g. a Spark driver sending it to executors). Workers
+/// call AdbcConnectionWriteIngestPartition on their own connections —
+/// the connection does not have to be the same one that created the
+/// handle. Each write returns an opaque receipt. The coordinator
+/// collects receipts and calls AdbcConnectionCommitIngestPartitions
+/// (or AdbcConnectionAbortIngestPartitions on failure).
+///
+/// Handles and receipts are driver-defined opaque byte strings. They
+/// are safe to transmit between processes and to use concurrently
+/// from multiple connections.
+///
+/// Drivers are not required to support partitioned ingest.
+///
+/// \since ADBC API revision 1.2.0
+///
+/// @{
+
+/// \brief Begin a partitioned bulk ingest.
+///
+/// Uses the same semantics as the ADBC_INGEST_OPTION_* options on
+/// AdbcStatement. For ADBC_INGEST_OPTION_MODE_CREATE,
+/// ADBC_INGEST_OPTION_MODE_CREATE_APPEND, and
+/// ADBC_INGEST_OPTION_MODE_REPLACE, `schema` is required and the
+/// driver creates (or recreates) the target table at this call. For
+/// ADBC_INGEST_OPTION_MODE_APPEND, `schema` is optional; if provided,
+/// the driver validates it against the target and returns
+/// ADBC_STATUS_ALREADY_EXISTS on mismatch.
+///
+/// The returned handle is opaque, serializable, and usable from any
+/// connection that can open the same database. The caller releases
+/// it via `out_handle->release`; the bytes can be copied and shipped
+/// to workers before release.
+///
+/// \since ADBC API revision 1.2.0
+/// \param[in] connection The coordinator's connection.
+/// \param[in] target_catalog Catalog of the target table, or NULL.
+/// \param[in] target_db_schema Schema of the target table, or NULL.
+/// \param[in] target_table Name of the target table. Required.
+/// \param[in] mode One of ADBC_INGEST_OPTION_MODE_*. Required.
+/// \param[in] schema Arrow schema of the data to be written.
+/// Required for create/replace/create_append modes; optional for
+/// append.
+/// \param[out] out_handle Driver-owned handle. Must be released by
+/// the caller via `out_handle->release`.
+/// \param[out] error Error details, if any.
+/// \return ADBC_STATUS_INVALID_ARGUMENT if mode requires a schema
+/// but none was provided.
+/// \return ADBC_STATUS_ALREADY_EXISTS if append mode is requested
+/// and the target schema disagrees with the provided schema.
+/// \return ADBC_STATUS_NOT_IMPLEMENTED if the driver does not
+/// support partitioned ingest.
+ADBC_EXPORT
+AdbcStatusCode AdbcConnectionBeginIngestPartitions(
+ struct AdbcConnection* connection, const char* target_catalog,
+ const char* target_db_schema, const char* target_table, const char* mode,
Review Comment:
Yeah, that's what I'm thinking. I admit it's not the most convenient API for a user and it does complicate the statement 'state machine' but it is more flexible/extensible/typesafe.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(format): partitioned bulk ingest API [arrow-adbc]
Posted by "tokoko (via GitHub)" <gi...@apache.org>.
tokoko commented on code in PR #4317:
URL: https://github.com/apache/arrow-adbc/pull/4317#discussion_r4210679330
##########
c/include/arrow-adbc/adbc.h:
##########
@@ -1946,6 +1989,235 @@ AdbcStatusCode AdbcConnectionReadPartition(struct AdbcConnection* connection,
/// @}
+/// \defgroup adbc-connection-ingest-partition Partitioned Bulk Ingest
+/// @{
+
+/// \brief Driver-owned bytes returned by
+/// AdbcConnectionBeginIngestPartitions.
+///
+/// The bytes are opaque and serializable: the caller may copy
+/// `bytes[0..length)` and ship that copy to workers (other processes
+/// or hosts) which can pass it to AdbcConnectionWriteIngestPartition
+/// directly.
+///
+/// The struct itself is owned by the driver. Call `release` exactly
+/// once to free it. Releasing the handle does NOT roll back the
+/// ingest — call AdbcConnectionAbortIngestPartitions for that.
+///
+/// \since ADBC API revision 1.2.0
+struct AdbcIngestHandle {
+ /// \brief The length of `bytes`.
+ size_t length;
+
+ /// \brief The serialized handle bytes (driver-owned).
+ const uint8_t* bytes;
+
+ /// \brief Private driver state.
+ void* private_data;
+
+ /// \brief Release the handle's memory. Sets `release` to NULL.
+ void (*release)(struct AdbcIngestHandle* self);
+};
+
+/// \brief Driver-owned bytes returned by
+/// AdbcConnectionWriteIngestPartition.
+///
+/// Mirror of AdbcIngestHandle: opaque, serializable, single-use
+/// `release`. Releasing a receipt does NOT discard the underlying
+/// write; that happens at Commit (commit it) or Abort (drop it).
+///
+/// \since ADBC API revision 1.2.0
+struct AdbcIngestReceipt {
+ /// \brief The length of `bytes`.
+ size_t length;
+
+ /// \brief The serialized receipt bytes (driver-owned).
+ const uint8_t* bytes;
+
+ /// \brief Private driver state.
+ void* private_data;
+
+ /// \brief Release the receipt's memory. Sets `release` to NULL.
+ void (*release)(struct AdbcIngestReceipt* self);
+};
+/// @}
+
+/// \addtogroup adbc-connection-ingest-partition
+/// Some drivers can accept bulk writes from a distributed writer: a
+/// coordinator configures an ingest, many workers write partitions in
+/// parallel (possibly from different processes or hosts), and the
+/// coordinator commits or aborts atomically.
+///
+/// This mirrors the read-side partitioned execution model. The
+/// coordinator calls AdbcConnectionBeginIngestPartitions to obtain an
+/// opaque, serializable handle. The handle is shipped to workers by
+/// the caller (e.g. a Spark driver sending it to executors). Workers
+/// call AdbcConnectionWriteIngestPartition on their own connections —
+/// the connection does not have to be the same one that created the
+/// handle. Each write returns an opaque receipt. The coordinator
+/// collects receipts and calls AdbcConnectionCommitIngestPartitions
+/// (or AdbcConnectionAbortIngestPartitions on failure).
+///
+/// Handles and receipts are driver-defined opaque byte strings. They
+/// are safe to transmit between processes and to use concurrently
+/// from multiple connections.
+///
+/// Drivers are not required to support partitioned ingest.
+///
+/// \since ADBC API revision 1.2.0
+///
+/// @{
+
+/// \brief Begin a partitioned bulk ingest.
+///
+/// Uses the same semantics as the ADBC_INGEST_OPTION_* options on
+/// AdbcStatement. For ADBC_INGEST_OPTION_MODE_CREATE,
+/// ADBC_INGEST_OPTION_MODE_CREATE_APPEND, and
+/// ADBC_INGEST_OPTION_MODE_REPLACE, `schema` is required and the
+/// driver creates (or recreates) the target table at this call. For
+/// ADBC_INGEST_OPTION_MODE_APPEND, `schema` is optional; if provided,
+/// the driver validates it against the target and returns
+/// ADBC_STATUS_ALREADY_EXISTS on mismatch.
+///
+/// The returned handle is opaque, serializable, and usable from any
+/// connection that can open the same database. The caller releases
+/// it via `out_handle->release`; the bytes can be copied and shipped
+/// to workers before release.
+///
+/// \since ADBC API revision 1.2.0
+/// \param[in] connection The coordinator's connection.
+/// \param[in] target_catalog Catalog of the target table, or NULL.
+/// \param[in] target_db_schema Schema of the target table, or NULL.
+/// \param[in] target_table Name of the target table. Required.
+/// \param[in] mode One of ADBC_INGEST_OPTION_MODE_*. Required.
+/// \param[in] schema Arrow schema of the data to be written.
+/// Required for create/replace/create_append modes; optional for
+/// append.
+/// \param[out] out_handle Driver-owned handle. Must be released by
+/// the caller via `out_handle->release`.
+/// \param[out] error Error details, if any.
+/// \return ADBC_STATUS_INVALID_ARGUMENT if mode requires a schema
+/// but none was provided.
+/// \return ADBC_STATUS_ALREADY_EXISTS if append mode is requested
+/// and the target schema disagrees with the provided schema.
+/// \return ADBC_STATUS_NOT_IMPLEMENTED if the driver does not
+/// support partitioned ingest.
+ADBC_EXPORT
+AdbcStatusCode AdbcConnectionBeginIngestPartitions(
+ struct AdbcConnection* connection, const char* target_catalog,
+ const char* target_db_schema, const char* target_table, const char* mode,
Review Comment:
better late than never... moved everything to statement-level.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(format): partitioned bulk ingest API [arrow-adbc]
Posted by "tokoko (via GitHub)" <gi...@apache.org>.
tokoko commented on PR #4317:
URL: https://github.com/apache/arrow-adbc/pull/4317#issuecomment-6045996096
`dotnet-format` pre-commit failure looks unrelated.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Partitioned ingest API [arrow-adbc]
Posted by "tokoko (via GitHub)" <gi...@apache.org>.
tokoko commented on code in PR #4317:
URL: https://github.com/apache/arrow-adbc/pull/4317#discussion_r3322120071
##########
c/include/arrow-adbc/adbc.h:
##########
@@ -1135,6 +1146,30 @@ struct ADBC_EXPORT AdbcDriver {
struct AdbcError*);
/// @}
+
+ /// \defgroup adbc-1.2.0 ADBC API Revision 1.2.0
+ ///
+ /// Functions added in ADBC 1.2.0. For backwards compatibility,
+ /// these members must not be accessed unless the version passed to
+ /// the AdbcDriverInitFunc is greater than or equal to
+ /// ADBC_VERSION_1_2_0.
+ ///
+ /// @{
+
+ AdbcStatusCode (*ConnectionBeginIngestPartitions)(
+ struct AdbcConnection*, const char*, const char*, const char*, const char*,
+ struct ArrowSchema*, struct AdbcIngestHandle*, struct AdbcError*);
+ AdbcStatusCode (*ConnectionWriteIngestPartition)(
+ struct AdbcConnection*, const uint8_t*, size_t, struct ArrowArrayStream*,
+ struct AdbcIngestReceipt*, struct AdbcError*);
+ AdbcStatusCode (*ConnectionCommitIngestPartitions)(
Review Comment:
wdyt about "complete" or maybe "apply"? I'm fine with changing to "complete" if you agree.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Partitioned ingest API [arrow-adbc]
Posted by "tokoko (via GitHub)" <gi...@apache.org>.
tokoko commented on PR #4317:
URL: https://github.com/apache/arrow-adbc/pull/4317#issuecomment-4570477100
@lidavidm should I leave only the spec changes in the PR against spec branch? or postgres impl as well?
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(format): partitioned bulk ingest API [arrow-adbc]
Posted by "tokoko (via GitHub)" <gi...@apache.org>.
tokoko commented on code in PR #4317:
URL: https://github.com/apache/arrow-adbc/pull/4317#discussion_r3433408549
##########
c/include/arrow-adbc/adbc.h:
##########
@@ -1516,6 +1518,23 @@ struct ADBC_EXPORT AdbcDriver {
AdbcStatusCode (*StatementExecuteMulti)(struct AdbcStatement*,
struct AdbcMultiResultSet*, struct AdbcError*);
+ AdbcStatusCode (*ConnectionBeginIngestPartitions)(struct AdbcConnection*,
+ struct ArrowSchema*,
+ struct AdbcSerializableHandle*,
+ struct AdbcError*);
+ AdbcStatusCode (*ConnectionWriteIngestPartition)(struct AdbcConnection*, const uint8_t*,
+ size_t, struct ArrowArrayStream*,
+ struct AdbcSerializableHandle*,
+ struct AdbcError*);
+ AdbcStatusCode (*ConnectionCompleteIngestPartitions)(struct AdbcConnection*,
Review Comment:
let me push back a little. out param feels clean for C intreface, but replicating the same for language apis will be a patchwork of different solutions, they would have to either go into exceptions or have additional output in the signature depending on the language. btw, what makes a new status code a big change, would something break? are there code paths that rely on exhaustive checks?
Another alternative is to put some additional metadata inside an Error message. That's probably abuse, but also could work.
##########
c/include/arrow-adbc/adbc.h:
##########
@@ -1516,6 +1518,23 @@ struct ADBC_EXPORT AdbcDriver {
AdbcStatusCode (*StatementExecuteMulti)(struct AdbcStatement*,
struct AdbcMultiResultSet*, struct AdbcError*);
+ AdbcStatusCode (*ConnectionBeginIngestPartitions)(struct AdbcConnection*,
+ struct ArrowSchema*,
+ struct AdbcSerializableHandle*,
+ struct AdbcError*);
+ AdbcStatusCode (*ConnectionWriteIngestPartition)(struct AdbcConnection*, const uint8_t*,
+ size_t, struct ArrowArrayStream*,
+ struct AdbcSerializableHandle*,
+ struct AdbcError*);
+ AdbcStatusCode (*ConnectionCompleteIngestPartitions)(struct AdbcConnection*,
Review Comment:
let me push back a little. out param feels clean for C interface, but replicating the same for language apis will be a patchwork of different solutions, they would have to either go into exceptions or have additional output in the signature depending on the language. btw, what makes a new status code a big change, would something break? are there code paths that rely on exhaustive checks?
Another alternative is to put some additional metadata inside an Error message. That's probably abuse, but also could work.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(format): partitioned bulk ingest API [arrow-adbc]
Posted by "tokoko (via GitHub)" <gi...@apache.org>.
tokoko commented on code in PR #4317:
URL: https://github.com/apache/arrow-adbc/pull/4317#discussion_r3795480966
##########
c/include/arrow-adbc/adbc.h:
##########
@@ -1946,6 +1989,235 @@ AdbcStatusCode AdbcConnectionReadPartition(struct AdbcConnection* connection,
/// @}
+/// \defgroup adbc-connection-ingest-partition Partitioned Bulk Ingest
+/// @{
+
+/// \brief Driver-owned bytes returned by
+/// AdbcConnectionBeginIngestPartitions.
+///
+/// The bytes are opaque and serializable: the caller may copy
+/// `bytes[0..length)` and ship that copy to workers (other processes
+/// or hosts) which can pass it to AdbcConnectionWriteIngestPartition
+/// directly.
+///
+/// The struct itself is owned by the driver. Call `release` exactly
+/// once to free it. Releasing the handle does NOT roll back the
+/// ingest — call AdbcConnectionAbortIngestPartitions for that.
+///
+/// \since ADBC API revision 1.2.0
+struct AdbcIngestHandle {
+ /// \brief The length of `bytes`.
+ size_t length;
+
+ /// \brief The serialized handle bytes (driver-owned).
+ const uint8_t* bytes;
+
+ /// \brief Private driver state.
+ void* private_data;
+
+ /// \brief Release the handle's memory. Sets `release` to NULL.
+ void (*release)(struct AdbcIngestHandle* self);
+};
+
+/// \brief Driver-owned bytes returned by
+/// AdbcConnectionWriteIngestPartition.
+///
+/// Mirror of AdbcIngestHandle: opaque, serializable, single-use
+/// `release`. Releasing a receipt does NOT discard the underlying
+/// write; that happens at Commit (commit it) or Abort (drop it).
+///
+/// \since ADBC API revision 1.2.0
+struct AdbcIngestReceipt {
+ /// \brief The length of `bytes`.
+ size_t length;
+
+ /// \brief The serialized receipt bytes (driver-owned).
+ const uint8_t* bytes;
+
+ /// \brief Private driver state.
+ void* private_data;
+
+ /// \brief Release the receipt's memory. Sets `release` to NULL.
+ void (*release)(struct AdbcIngestReceipt* self);
+};
+/// @}
+
+/// \addtogroup adbc-connection-ingest-partition
+/// Some drivers can accept bulk writes from a distributed writer: a
+/// coordinator configures an ingest, many workers write partitions in
+/// parallel (possibly from different processes or hosts), and the
+/// coordinator commits or aborts atomically.
+///
+/// This mirrors the read-side partitioned execution model. The
+/// coordinator calls AdbcConnectionBeginIngestPartitions to obtain an
+/// opaque, serializable handle. The handle is shipped to workers by
+/// the caller (e.g. a Spark driver sending it to executors). Workers
+/// call AdbcConnectionWriteIngestPartition on their own connections —
+/// the connection does not have to be the same one that created the
+/// handle. Each write returns an opaque receipt. The coordinator
+/// collects receipts and calls AdbcConnectionCommitIngestPartitions
+/// (or AdbcConnectionAbortIngestPartitions on failure).
+///
+/// Handles and receipts are driver-defined opaque byte strings. They
+/// are safe to transmit between processes and to use concurrently
+/// from multiple connections.
+///
+/// Drivers are not required to support partitioned ingest.
+///
+/// \since ADBC API revision 1.2.0
+///
+/// @{
+
+/// \brief Begin a partitioned bulk ingest.
+///
+/// Uses the same semantics as the ADBC_INGEST_OPTION_* options on
+/// AdbcStatement. For ADBC_INGEST_OPTION_MODE_CREATE,
+/// ADBC_INGEST_OPTION_MODE_CREATE_APPEND, and
+/// ADBC_INGEST_OPTION_MODE_REPLACE, `schema` is required and the
+/// driver creates (or recreates) the target table at this call. For
+/// ADBC_INGEST_OPTION_MODE_APPEND, `schema` is optional; if provided,
+/// the driver validates it against the target and returns
+/// ADBC_STATUS_ALREADY_EXISTS on mismatch.
+///
+/// The returned handle is opaque, serializable, and usable from any
+/// connection that can open the same database. The caller releases
+/// it via `out_handle->release`; the bytes can be copied and shipped
+/// to workers before release.
+///
+/// \since ADBC API revision 1.2.0
+/// \param[in] connection The coordinator's connection.
+/// \param[in] target_catalog Catalog of the target table, or NULL.
+/// \param[in] target_db_schema Schema of the target table, or NULL.
+/// \param[in] target_table Name of the target table. Required.
+/// \param[in] mode One of ADBC_INGEST_OPTION_MODE_*. Required.
+/// \param[in] schema Arrow schema of the data to be written.
+/// Required for create/replace/create_append modes; optional for
+/// append.
+/// \param[out] out_handle Driver-owned handle. Must be released by
+/// the caller via `out_handle->release`.
+/// \param[out] error Error details, if any.
+/// \return ADBC_STATUS_INVALID_ARGUMENT if mode requires a schema
+/// but none was provided.
+/// \return ADBC_STATUS_ALREADY_EXISTS if append mode is requested
+/// and the target schema disagrees with the provided schema.
+/// \return ADBC_STATUS_NOT_IMPLEMENTED if the driver does not
+/// support partitioned ingest.
+ADBC_EXPORT
+AdbcStatusCode AdbcConnectionBeginIngestPartitions(
+ struct AdbcConnection* connection, const char* target_catalog,
+ const char* target_db_schema, const char* target_table, const char* mode,
Review Comment:
just to check, you mean all new functions should be moved to statement-level, right? +1 from me as well, seems more consistent for both normal ingest and partitioned ingest to be on the same level even if statement doesn't add anything meaningful.
P.S. I can get back to this PR in a day or two.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org