Struct DriverSink 

pub struct DriverSink<D> { /* private fields */ }

A SinkDriver as a Sink (RFC 0014).

It plugs a driver into everything that takes a sink — the pipeline, topology, the governance decorators, the conformance suite — and owns the rules a driver would otherwise re-implement:

• Capabilities follow the primitives. Overwrite, exactly-once, cleanup, schema evolution, rollback and the columnar path are on exactly when the driver returns the matching group.
• One write, three paths. The plain and reported writes call SinkDriver::write with RowPolicy::WholePage and fail on the first failed row; the per-row write calls it with RowPolicy::PerRow. Each result is checked against the rows offered, so a write that loses rows fails with FaucetError::RowsUnaccounted.
• The call context is the run’s. Sink::set_call_context installs into the slot every primitive receives.

Implementations

impl<D: SinkDriver> DriverSink<D>

pub fn new(driver: D) -> Self

The sink over driver.

Panics

In debug builds, when the driver’s declarations disagree with its primitives (check_declarations) — a driver that lists overwrite without a staging driver, or the reverse, is a programming error every connector test then catches.

pub fn driver(&self) -> &D

The driver.

pub fn driver_mut(&mut self) -> &mut D

The driver, mutably (for a builder method of the connector).

pub fn calls(&self) -> &CallSlot

The call slot the primitives receive.

pub fn into_driver(self) -> D

Unwrap the driver.

pub fn check_declarations(&self) -> Result<(), FaucetError>

Whether the driver’s declarations agree with its primitives: the overwrite mode is declared exactly when a staging driver exists, an active staging driver is only reachable through that mode, and keyed dedup comes with a keyed write mode.

Trait Implementations

impl<D: SinkDriver> Sink for DriverSink<D>

fn write_batch<'life0, 'life1, 'async_trait>( &'life0 self, records: &'life1 [Value], ) -> Pin<Box<dyn Future<Output = Result<usize, FaucetError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Write a batch of records to the destination.

fn write_batch_reported<'life0, 'life1, 'async_trait>( &'life0 self, records: &'life1 [Value], ) -> Pin<Box<dyn Future<Output = Result<WriteReport, FaucetError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Write a batch and report what happened to every row offered (#879).

fn write_batch_partial<'life0, 'life1, 'async_trait>( &'life0 self, records: &'life1 [Value], ) -> Pin<Box<dyn Future<Output = Result<Vec<RowOutcome>, FaucetError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Write a batch and report per-row outcomes.

fn flush<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<(), FaucetError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Flush any buffered data to the destination.

fn batch_atomicity(&self) -> BatchAtomicity

What a failed batch write leaves behind (#737): nothing (Atomic), per-row outcomes with nothing committed on an outer Err (PerRow), or possibly some rows (BestEffort, the default).

fn supports_columnar(&self) -> bool

Available on crate feature arrow only.

Whether this sink can consume columnar (arrow::RecordBatch) writes via write_batch_columnar without first converting to Value. Default: false.

fn write_batch_columnar<'life0, 'life1, 'async_trait>( &'life0 self, batch: &'life1 RecordBatch, ) -> Pin<Box<dyn Future<Output = Result<usize, FaucetError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Available on crate feature arrow only.

Write a columnar RecordBatch to the destination, returning the number of rows written.

fn supports_idempotent_writes(&self) -> bool

Whether this sink can durably commit a page’s rows and a commit token in a single atomic transaction. Default: false (at-least-once only).

fn capabilities(&self) -> Capabilities

What this sink can do (RFC 0013, #876). The default derives every field from the per-capability methods (Capabilities::of_sink); override it to declare a capability no method expresses (how unknown fields are treated, scoped overwrite), keeping the connector’s ConnectorDescriptor in agreement. Decorators must forward this.

fn sink_guarantee(&self) -> SinkGuarantee

The strongest delivery guarantee this sink can uphold — see SinkGuarantee.

fn write_batch_is_replay_safe(&self) -> bool

Whether a plain write_batch is safe to replay — i.e. re-sending the same page after an ambiguous failure converges instead of duplicating rows. Default: false.

fn dedups_by_key(&self) -> bool

Whether this sink instance is configured to dedup by key — i.e. write_mode: upsert (or delete) with a non-empty key, so re-applying a record with the same key converges instead of duplicating. Default: false.

fn supported_write_modes(&self) -> &'static [WriteMode]

Write modes this sink can apply. Default: append-only. Sinks that implement key-based merge override this to include WriteMode::Upsert / WriteMode::Delete. The CLI rejects a configured mode that is not in this set at config-load time.

fn current_schema<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Option<Value>, FaucetError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

The sink’s live destination schema as an infer_schema-shaped object ({"type":"object","properties":{ <col>: <type-fragment>, … }}), or None for a schemaless sink or a target that does not exist yet.

fn supports_schema_evolution(&self) -> bool

Whether this sink can apply additive/widening DDL via evolve_schema. Default: false. The CLI rejects on_drift: evolve against a sink that returns false at config-load.

fn evolve_schema<'life0, 'life1, 'async_trait>( &'life0 self, evolution: &'life1 SchemaEvolution, ) -> Pin<Box<dyn Future<Output = Result<(), FaucetError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Apply an additive schema evolution (new columns, lossless widenings, nullability relaxations) to the destination. MUST be idempotent (ADD COLUMN IF NOT EXISTS semantics) so concurrent runs converge.

fn supports_cleanup(&self) -> bool

Whether this sink can delete a scoped set of rows for scoped cleanup (#478). Default false; the upsert-capable sinks override it.

fn supports_staged_load(&self) -> bool

Whether this sink can bulk-load via an object-store stage — write the page to S3/GCS/Azure, then pull it with the warehouse’s native load command (COPY / COPY INTO / s3() table function) (#528). Default false; warehouse sinks that honour a StagingSpec override it. Object-safe (no args, no generics) so Box<dyn Sink> is unaffected.

fn cleanup_scope<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, scope: &'life1 BTreeMap<String, Value>, seen: &'life2 SeenKeys, ) -> Pin<Box<dyn Future<Output = Result<u64, FaucetError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

👎Deprecated since 1.14.0:

implement faucet_core::driver::CleanupDriver on a family driver; see RFC 0014

Delete rows matching scope whose key is not in seen.

fn write_batch_idempotent<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, records: &'life1 [Value], scope: &'life2 str, token: &'life3 str, ) -> Pin<Box<dyn Future<Output = Result<usize, FaucetError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait,

👎Deprecated since 1.14.0:

implement faucet_core::driver::CommitTokenDriver on a family driver; see RFC 0014

Write records AND durably record token for scope, atomically.

fn write_batch_idempotent_reported<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, records: &'life1 [Value], scope: &'life2 str, token: &'life3 str, ) -> Pin<Box<dyn Future<Output = Result<WriteReport, FaucetError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait,

👎Deprecated since 1.14.0:

implement faucet_core::driver::CommitTokenDriver on a family driver; see RFC 0014

write_batch_idempotent with a WriteReport: the exactly-once path calls this, and a page whose report does not account for every row it offered fails with FaucetError::RowsUnaccounted before its bookmark is persisted (#879). Override it when the sink declines rows by design inside the atomic commit (a record without a key under a skip policy) and report them as skipped.

fn last_committed_token<'life0, 'life1, 'async_trait>( &'life0 self, scope: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Option<String>, FaucetError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

👎Deprecated since 1.14.0:

implement faucet_core::driver::CommitTokenDriver on a family driver; see RFC 0014

The last token durably committed for scope, or None if this sink has never committed under that scope. Default: None.

fn is_overwrite(&self) -> bool

Whether this sink instance is configured for full-destination replacement (WriteMode::Overwrite).

fn begin_overwrite<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<(), FaucetError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

👎Deprecated since 1.14.0:

implement faucet_core::driver::StagingDriver on a family driver; see RFC 0014

Prepare a staging target for an overwrite run.

fn commit_overwrite<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<(), FaucetError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

👎Deprecated since 1.14.0:

implement faucet_core::driver::StagingDriver on a family driver; see RFC 0014

Atomically replace the destination with the staged data.

fn abort_overwrite<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<(), FaucetError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

👎Deprecated since 1.14.0:

implement faucet_core::driver::StagingDriver on a family driver; see RFC 0014

Discard the staging target after a failed or cancelled overwrite run.

fn overwrite_staging_exists<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Option<bool>, FaucetError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

👎Deprecated since 1.14.0:

implement faucet_core::driver::StagingDriver on a family driver; see RFC 0014

Whether an overwrite staging object (the …__faucet_ovw table or collection a begin_overwrite creates) exists right now — a read-only probe used by faucet status --probe to report staging a crashed or aborted overwrite left behind.

fn supports_rollback(&self) -> bool

Whether this sink can undo a run it wrote (#706) through rollback_run. Default false.

fn rollback_run<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, run_id: &'life1 str, opts: &'life2 RollbackOptions, ) -> Pin<Box<dyn Future<Output = Result<RollbackOutcome, FaucetError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Undo everything run_id wrote to this destination, per opts.mode: delete the run’s rows (append), restore journaled before-images (upsert / delete), or swap the kept previous table back (overwrite). Must be all-or-nothing per destination where the backend allows it, and must leave the destination untouched when opts.dry_run is set or when conflicts block it (see RollbackOutcome).

fn forget_run<'life0, 'life1, 'async_trait>( &'life0 self, run_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<(), FaucetError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Drop whatever this sink kept to make run_id undoable (journal rows, a previous table) once the run is past the retention window. Default: no-op.

fn rewind_commit_token<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, scope: &'life1 str, token: Option<&'life2 str>, ) -> Pin<Box<dyn Future<Output = Result<(), FaucetError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Rewind the exactly-once watermark for scope to token (None clears it), so a rolled-back run’s pages are not skipped on the next run. Default: a typed “unsupported” error.

fn readback_source(&self) -> Option<(String, Value)>

A source (kind, config) that reads this destination back — what faucet verify (#701) compares the pipeline’s source against. Default None: the user names the destination reader in verify.destination. The config must not need a credential the sink config lacks.

fn complete_run<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<(), FaucetError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Finalize this run’s output after Pipeline::run finished successfully and uncancelled, on every transfer path (Value, columnar, native), after the terminal flush.

fn config_schema(&self) -> Value

Return a JSON Schema describing the configuration this sink accepts.

fn connector_name(&self) -> &'static str

Stable identifier used as the connector label on metrics and the connector attribute on spans. See Source::connector_name.

fn set_roundtrip_recorder(&self, recorder: Arc<RoundtripRecorder>)

Receive a pre-labelled handle for counting upstream round trips (#638). See Source::set_roundtrip_recorder; defaulted to a no-op.

fn set_call_context(&self, ctx: CallContext)

Receive the run’s CallContext (#880). See Source::set_call_context. A sink’s context never carries the run’s cancel token — a cancelled run still flushes its sinks — nor a retry policy: writes are retried page by page under ADR 0007. Defaulted to a no-op; a wrapping sink forwards it.

fn dataset_uri(&self) -> String

Logical dataset identity for lineage emission, following OpenLineage naming conventions (https://openlineage.io/docs/spec/naming).

fn local_outputs<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Vec<LocalOutput>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

The concrete local files this sink instance opened during the run.

fn check<'life0, 'life1, 'async_trait>( &'life0 self, ctx: &'life1 CheckContext, ) -> Pin<Box<dyn Future<Output = Result<CheckReport, FaucetError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Run a fast, non-mutating preflight probe (used by faucet doctor).

fn write_governed<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, page: &'life1 GovernedPage<'life2>, ) -> Pin<Box<dyn Future<Output = Result<WriteReport, FaucetError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Write a page the engine governed (RFC 0014): it went through masking, quality, contract and schema drift, so the records are exactly what may reach the destination. The engine’s plain write path calls this.

fn admit_page<'life0, 'life1, 'async_trait>( &'life0 self, records: &'life1 [Value], ) -> Pin<Box<dyn Future<Output = Result<(), FaucetError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Called with a whole page before the pipeline splits it into adaptive sub-batches. A sink enforcing a ceiling that must hold for the page as a unit (a run budget) refuses here, so no sub-batch of a refused page lands. Default: accept. Decorators must forward this.

fn native_load_capabilities(&self) -> Vec<NativeLoadCapability>

Native byte-passthrough load mechanisms this sink offers, each naming the wire format and the write modes it can honor (#633). Default: vec![] (no fast path; use write_batch).

fn load_native<'life0, 'life1, 'async_trait>( &'life0 self, batch: NativeBatch, scope: &'life1 str, ctx: NativeLoadContext, ) -> Pin<Box<dyn Future<Output = Result<usize, FaucetError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Bulk-load one native-format byte batch directly (#633), returning the number of rows written.