diff --git a/c/include/arrow-adbc/adbc.h b/c/include/arrow-adbc/adbc.h index a461795ce0..91dc5ee2e5 100644 --- a/c/include/arrow-adbc/adbc.h +++ b/c/include/arrow-adbc/adbc.h @@ -1337,6 +1337,8 @@ typedef void (*AdbcWarningHandler)(const struct AdbcError* warning, void* user_d /// driver and the driver manager. /// @{ +struct AdbcSerializableHandle; + /// \brief An instance of an initialized database driver. /// /// This provides a common interface for vendor-specific driver @@ -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*, + const uint8_t*, size_t, size_t, + const uint8_t**, const size_t*, + int64_t*, struct AdbcError*); + AdbcStatusCode (*ConnectionAbortIngestPartitions)(struct AdbcConnection*, + const uint8_t*, size_t, size_t, + const uint8_t**, const size_t*, + struct AdbcError*); + /// @} }; @@ -2398,6 +2417,204 @@ AdbcStatusCode AdbcConnectionReadPartition(struct AdbcConnection* connection, /// @} +/// \defgroup adbc-connection-ingest-partition Partitioned Bulk Ingest +/// @{ + +/// \brief A driver-owned opaque byte buffer with a release callback. +/// +/// Used by partitioned bulk ingest for both handles (returned by +/// AdbcConnectionBeginIngestPartitions) and receipts (returned by +/// AdbcConnectionWriteIngestPartition). +/// +/// The bytes are opaque and serializable: the caller may copy +/// `bytes[0..length)` and ship that copy across processes or hosts. +/// +/// The struct itself is owned by the driver. Call `release` exactly +/// once to free it. +/// +/// \since ADBC API revision 1.2.0 +struct AdbcSerializableHandle { + /// \brief The length of `bytes`. + size_t length; + + /// \brief The serialized bytes (driver-owned). + const uint8_t* bytes; + + /// \brief Private driver state. + void* private_data; + + /// \brief Release the memory. Sets `release` to NULL. + void (*release)(struct AdbcSerializableHandle* 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 AdbcConnectionCompleteIngestPartitions +/// (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. +/// +/// The target table, mode, and optional catalog/schema are configured +/// via the ADBC_INGEST_OPTION_* connection options before calling +/// this function. The same option keys used for single-writer +/// statement-level ingest apply here at the connection level: +/// ADBC_INGEST_OPTION_TARGET_TABLE (required), +/// ADBC_INGEST_OPTION_MODE (required), +/// ADBC_INGEST_OPTION_TARGET_CATALOG (optional), +/// ADBC_INGEST_OPTION_TARGET_DB_SCHEMA (optional). +/// +/// 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] 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, or if required options are missing. +/// \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, struct ArrowSchema* schema, + struct AdbcSerializableHandle* 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 AdbcSerializableHandle* out_receipt, + struct AdbcError* error); + +/// \brief Complete 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 Complete 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 AdbcConnectionCompleteIngestPartitions( + 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. +/// +/// Best-effort: the driver may clean up any resources associated +/// with the handle. The handle is consumed. +/// +/// \since ADBC API revision 1.2.0 +/// \param[in] connection A connection. +/// \param[in] handle The handle from Begin. +/// \param[in] handle_len Length of handle. +/// \param[in] num_receipts Number of receipts, or 0. +/// \param[in] receipts Array of receipt byte-pointers, or NULL. +/// \param[in] receipt_lens Array of receipt lengths, or NULL. +/// \param[out] error Error details, if any. +ADBC_EXPORT +AdbcStatusCode AdbcConnectionAbortIngestPartitions(struct AdbcConnection* connection, + const uint8_t* handle, + size_t handle_len, size_t num_receipts, + const uint8_t** receipts, + const size_t* receipt_lens, + struct AdbcError* error); + +/// @} + /// \defgroup adbc-connection-transaction Transaction Semantics /// /// Connections start out in auto-commit mode by default (if diff --git a/docs/source/format/partitioned_bulk_ingest.rst b/docs/source/format/partitioned_bulk_ingest.rst new file mode 100644 index 0000000000..207a042750 --- /dev/null +++ b/docs/source/format/partitioned_bulk_ingest.rst @@ -0,0 +1,290 @@ +.. 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 + executors, each producing a partition of the output. Today each + executor opens its own ADBC connection and runs its own bulk + ingest, but the result is *not atomic*: there is no commit point at + which all N partitions become visible together. Workarounds + (per-job staging tables, ad-hoc swap SQL) are database-specific and + leak into application code. + +2. **Distributed-writer to table-format catalogs (Apache Iceberg, + Delta Lake).** These formats are *designed* for distributed + writes: many workers write data files in parallel, and a single + commit step writes a snapshot/manifest in the catalog or + transaction log. ADBC currently has no way to expose this shape. + A driver author who wants to write to Iceberg today has to pick + between (a) routing all writes through one process (defeats the + point) or (b) inventing a private API. + +The unifying observation is that both workloads need the same shape: +**coordinator decides what to ingest, workers write partitions in +parallel, coordinator commits or aborts atomically**. That is the +mirror image of partitioned read (``ExecutePartitions`` / +``ReadPartition``), which ADBC already supports. + +Goals +----- + +- Allow a coordinator to start an ingest, ship an opaque token to N + workers (possibly in different processes or hosts), have each + worker independently write a partition over its own connection, and + finally commit (or abort) atomically from the coordinator. +- Be implementable by both RDBMS drivers (via per-worker staging + tables) and table-format drivers (via per-worker data files + + catalog commit) without forcing either model on the other. +- Survive lost worker writes, dropped receipts, and coordinator + restarts without leaving silent data corruption. +- Keep the per-driver cost low: most of the ingest plumbing + (CREATE TABLE, COPY, schema mapping) is reused from existing + single-writer ingest. + +Non-goals +--------- + +- Schema evolution mid-ingest. Schema is fixed when ``Begin`` is + called; changing it requires starting a new ingest. +- Cross-driver atomicity (writing to two databases in one commit). +- Defining how a distributed engine (Spark, Flink) ships handles and + receipts between processes. That is the application's problem; + the API guarantees only that handles and receipts are opaque, + serializable byte strings. +- Idempotency on the coordinator side. If the coordinator + double-commits (calls ``Complete`` twice on the same handle) the + second call is undefined. + +Design overview +=============== + +Three new operations on ``AdbcConnection``, plus an ``Abort``: + +:: + + coordinator: Begin(schema) → handle + workers: Write(handle, stream) → receipt + Write(handle, stream) → receipt + ... + coordinator: Complete(handle, [receipt, receipt, …]) → rows_affected + (or) Abort(handle, [receipts...]) + +The handle and each receipt are **opaque, serializable byte +strings**. This is the same shape as the existing partitioned-read +side, where ``AdbcStatementExecutePartitions`` returns opaque +``AdbcPartitions`` byte strings that can be shipped to workers and +passed to ``AdbcConnectionReadPartition`` over a different +connection. + +API surface +----------- + +The target table, mode, and optional catalog/schema are set via the +existing ``ADBC_INGEST_OPTION_*`` connection options before calling +``Begin``. The same option keys used for single-writer +statement-level ingest apply here at the connection level. + +C declarations (see ``adbc.h`` for full doc comments): + +.. code-block:: c + + struct AdbcSerializableHandle { + size_t length; + const uint8_t* bytes; + void* private_data; + void (*release)(struct AdbcSerializableHandle*); + }; + + AdbcConnectionBeginIngestPartitions( + conn, schema, *out_handle, *error); + + AdbcConnectionWriteIngestPartition( + conn, handle_bytes, handle_len, *data_stream, + *out_receipt, *error); + + AdbcConnectionCompleteIngestPartitions( + conn, handle_bytes, handle_len, num_receipts, receipts, + receipt_lens, *rows_affected, *error); + + AdbcConnectionAbortIngestPartitions( + conn, handle_bytes, handle_len, num_receipts, receipts, + receipt_lens, *error); + +The asymmetry — outputs are driver-owned structs, inputs are raw +``bytes + len`` — is deliberate and matches the read side: the bytes +are the part the caller serializes for transport, while the structs +hold driver-owned memory that callers release locally. + +Driver-side semantics +--------------------- + +- **Begin** validates options, performs whatever setup the driver + requires for writes to proceed (e.g., creating the target table for + ``create``/``replace``/``create_append`` modes, reserving a + transaction snapshot, allocating an object-store prefix), and returns + a handle that encodes the state needed to scope subsequent writes. +- **Write** takes a handle and a stream, writes the partition into + driver-private staging (a per-write staging table, a per-write + object-store path), and returns a receipt encoding what was + written (staging name, file paths, row count, statistics, ...). + Each ``Write`` call must produce output that can be committed or + discarded *independently* — no shared state across concurrent + writes that would cause duplicate rows on retry. +- **Complete** atomically promotes the union of the supplied receipts + into the target. Atomic semantics are driver-specific: RDBMS + drivers swap staging into target in a transaction; table-format + drivers write a catalog or transaction-log entry referencing the + data files in the receipts. After successful commit the handle is + consumed. +- **Abort** discards all writes scoped to the handle. The driver + must clean up *every* write under the handle, not just the ones + named in the supplied receipts (see "Lost receipts" below). + +Cross-process flow +------------------ + +:: + + ┌──────────────┐ + │ coordinator │ Begin(...) ─→ handle + └──────┬───────┘ + │ copy handle.bytes; ship to workers + ▼ + ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ + │ worker 1 │ │ worker 2 │ … │ worker N │ + │ Write(...) →│ │ Write(...) →│ │ Write(...) →│ + │ receipt₁ │ │ receipt₂ │ │ receipt_N │ + └──────┬───────┘ └──────┬───────┘ └──────┬───────┘ + │ copy receipt.bytes; ship back │ + ▼ ▼ ▼ + ┌──────────────────────────────────────────────────────┐ + │ coordinator: Complete(handle, [r₁, r₂, ..., r_N]) │ + └──────────────────────────────────────────────────────┘ + +Workers may use *different* connections than the coordinator — the +handle is self-contained. + +Key design decisions +==================== + +The decisions below were the ones with non-obvious tradeoffs. + +1. Opaque handles and receipts +------------------------------ + +Driver-defined byte strings, no schema imposed by ADBC. This lets a +Postgres driver encode "staging table prefix + UUID" while an +Iceberg driver encodes "snapshot id + data file paths + column +stats" — without ADBC having to model both. The cost is that +applications cannot inspect handles or receipts. Worth it: the only +party that ever needs to interpret them is the driver. + +2. Schema is fixed at ``Begin``, not per-``Write`` +-------------------------------------------------- + +For ``create``/``replace``/``create_append`` modes, the driver +issues ``CREATE TABLE`` (or the catalog equivalent) at ``Begin`` +time, before any worker writes. Workers cannot race to "create on +first write" because they are on different machines. Iceberg/Delta +also need the schema pinned into the transaction snapshot at start. + +For ``append`` mode, the schema parameter is optional; if supplied +it is validated against the target so a thousand workers don't all +fail independently with the same schema-mismatch error. + +3. Driver-owned output structs (handle, receipt) +------------------------------------------------- + +An earlier draft used the ``GetOptionBytes`` two-phase sizing +pattern: caller passes a buffer + capacity, driver reports required +length, caller retries with a larger buffer. This is correct only +for *idempotent* operations. ``Begin`` and ``Write`` produce +irrecoverable side effects (``CREATE TABLE``, ``COPY``); a +buffer-too-small failure left the side effects in place but gave the +caller no handle/receipt to pass to ``Abort`` — an unrecoverable +orphan. + +The chosen pattern (driver-owned struct with a release callback) +mirrors ``AdbcPartitions`` on the read side, eliminates the orphan +window, and gives drivers a clean place to free internal state. + +4. ``Complete`` and ``Abort`` take raw bytes, not structs +------------------------------------------------------- + +Symmetric with ``AdbcConnectionReadPartition``, which takes the raw +bytes from a ``partitions[i]`` entry rather than the +``AdbcPartitions`` struct. Receipts that traveled across processes +arrive as raw bytes; forcing the caller to wrap them in +``AdbcSerializableHandle`` structs (with bogus ``release`` callbacks) +would be friction without benefit. + +5. Lost receipts are handled by handle-scoped sweep, not by receipts +-------------------------------------------------------------------- + +If a worker writes data but its receipt is lost in transit, the +coordinator's receipt list is incomplete. ``Complete`` will not +include the orphan (correct: only acknowledged writes are +committed). ``Abort``, however, must clean it up — and ``Abort`` +cannot rely on the supplied receipts alone, because the orphan +isn't in them. + +The handle therefore must encode enough scope (UUID prefix, +transaction id, object-store path) for the driver to enumerate +*everything* written under it. Receipts passed to ``Abort`` are an +optimization (fast-path deletion of known writes); the handle is the +authority for cleanup scope. Drivers that cannot enumerate from +the handle alone cannot correctly implement partitioned ingest. + +6. Coordinator may die without calling ``Complete`` or ``Abort`` +-------------------------------------------------------------- + +The handle is opaque to the driver outside of ``Write``, so the +driver has no built-in liveness signal. Recommended (not required) +behaviors: + +- Drivers may TTL or background-GC handle-scoped writes. +- Callers may persist the handle bytes and call ``Abort`` after + restart to recover. +- Iceberg/Delta drivers can rely on existing orphan-file cleanup + tooling. + +The spec does not mandate any of these; it documents the failure +mode and leaves the policy to drivers. + +Reference implementation +======================== + +A prototype lives in the PostgreSQL driver +(``c/driver/postgresql/ingest_partition.{h,cc}``). It uses +per-worker ``UNLOGGED`` staging tables of the form +``adbc_stg__``, a single ``BEGIN``/``COMMIT`` +wrapping ``INSERT INTO target SELECT cols FROM staging`` for each +receipt, and an ``Abort`` that scans +``information_schema.tables`` for the handle's prefix. Test +coverage is in ``c/driver/postgresql/partitioned_ingest_test.cc``. diff --git a/docs/source/index.rst b/docs/source/index.rst index 5d4b8966ff..5f01e11078 100644 --- a/docs/source/index.rst +++ b/docs/source/index.rst @@ -253,6 +253,7 @@ Why ADBC? :hidden: format/specification + format/partitioned_bulk_ingest format/versioning format/comparison format/how_manager diff --git a/go/adbc/drivermgr/arrow-adbc/adbc.h b/go/adbc/drivermgr/arrow-adbc/adbc.h index a461795ce0..91dc5ee2e5 100644 --- a/go/adbc/drivermgr/arrow-adbc/adbc.h +++ b/go/adbc/drivermgr/arrow-adbc/adbc.h @@ -1337,6 +1337,8 @@ typedef void (*AdbcWarningHandler)(const struct AdbcError* warning, void* user_d /// driver and the driver manager. /// @{ +struct AdbcSerializableHandle; + /// \brief An instance of an initialized database driver. /// /// This provides a common interface for vendor-specific driver @@ -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*, + const uint8_t*, size_t, size_t, + const uint8_t**, const size_t*, + int64_t*, struct AdbcError*); + AdbcStatusCode (*ConnectionAbortIngestPartitions)(struct AdbcConnection*, + const uint8_t*, size_t, size_t, + const uint8_t**, const size_t*, + struct AdbcError*); + /// @} }; @@ -2398,6 +2417,204 @@ AdbcStatusCode AdbcConnectionReadPartition(struct AdbcConnection* connection, /// @} +/// \defgroup adbc-connection-ingest-partition Partitioned Bulk Ingest +/// @{ + +/// \brief A driver-owned opaque byte buffer with a release callback. +/// +/// Used by partitioned bulk ingest for both handles (returned by +/// AdbcConnectionBeginIngestPartitions) and receipts (returned by +/// AdbcConnectionWriteIngestPartition). +/// +/// The bytes are opaque and serializable: the caller may copy +/// `bytes[0..length)` and ship that copy across processes or hosts. +/// +/// The struct itself is owned by the driver. Call `release` exactly +/// once to free it. +/// +/// \since ADBC API revision 1.2.0 +struct AdbcSerializableHandle { + /// \brief The length of `bytes`. + size_t length; + + /// \brief The serialized bytes (driver-owned). + const uint8_t* bytes; + + /// \brief Private driver state. + void* private_data; + + /// \brief Release the memory. Sets `release` to NULL. + void (*release)(struct AdbcSerializableHandle* 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 AdbcConnectionCompleteIngestPartitions +/// (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. +/// +/// The target table, mode, and optional catalog/schema are configured +/// via the ADBC_INGEST_OPTION_* connection options before calling +/// this function. The same option keys used for single-writer +/// statement-level ingest apply here at the connection level: +/// ADBC_INGEST_OPTION_TARGET_TABLE (required), +/// ADBC_INGEST_OPTION_MODE (required), +/// ADBC_INGEST_OPTION_TARGET_CATALOG (optional), +/// ADBC_INGEST_OPTION_TARGET_DB_SCHEMA (optional). +/// +/// 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] 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, or if required options are missing. +/// \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, struct ArrowSchema* schema, + struct AdbcSerializableHandle* 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 AdbcSerializableHandle* out_receipt, + struct AdbcError* error); + +/// \brief Complete 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 Complete 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 AdbcConnectionCompleteIngestPartitions( + 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. +/// +/// Best-effort: the driver may clean up any resources associated +/// with the handle. The handle is consumed. +/// +/// \since ADBC API revision 1.2.0 +/// \param[in] connection A connection. +/// \param[in] handle The handle from Begin. +/// \param[in] handle_len Length of handle. +/// \param[in] num_receipts Number of receipts, or 0. +/// \param[in] receipts Array of receipt byte-pointers, or NULL. +/// \param[in] receipt_lens Array of receipt lengths, or NULL. +/// \param[out] error Error details, if any. +ADBC_EXPORT +AdbcStatusCode AdbcConnectionAbortIngestPartitions(struct AdbcConnection* connection, + const uint8_t* handle, + size_t handle_len, size_t num_receipts, + const uint8_t** receipts, + const size_t* receipt_lens, + struct AdbcError* error); + +/// @} + /// \defgroup adbc-connection-transaction Transaction Semantics /// /// Connections start out in auto-commit mode by default (if