Trait Sink 

pub trait Sink: Send + Sync {
    // Required method
    fn write_batch<'life0, 'life1, 'async_trait>(
        &'life0 self,
        records: &'life1 [Value],
    ) -> Pin<Box<dyn Future<Output = Result<usize, FaucetError>> + Send + 'async_trait>>
       where 'life0: 'async_trait,
             'life1: 'async_trait,
             Self: 'async_trait;

    // Provided methods
    fn flush<'life0, 'async_trait>(
        &'life0 self,
    ) -> Pin<Box<dyn Future<Output = Result<(), FaucetError>> + Send + 'async_trait>>
       where 'life0: 'async_trait,
             Self: 'async_trait { ... }
    fn write_batch_partial<'life0, 'life1, 'async_trait>(
        &'life0 self,
        records: &'life1 [Value],
    ) -> Pin<Box<dyn Future<Output = Result<Vec<Result<(), FaucetError>>, FaucetError>> + Send + 'async_trait>>
       where 'life0: 'async_trait,
             'life1: 'async_trait,
             Self: 'async_trait { ... }
    fn supports_idempotent_writes(&self) -> bool { ... }
    fn sink_guarantee(&self) -> SinkGuarantee { ... }
    fn dedups_by_key(&self) -> bool { ... }
    fn supported_write_modes(&self) -> &'static [WriteMode] { ... }
    fn current_schema<'life0, 'async_trait>(
        &'life0 self,
    ) -> Pin<Box<dyn Future<Output = Result<Option<Value>, FaucetError>> + Send + 'async_trait>>
       where 'life0: 'async_trait,
             Self: 'async_trait { ... }
    fn supports_schema_evolution(&self) -> bool { ... }
    fn evolve_schema<'life0, 'life1, 'async_trait>(
        &'life0 self,
        evolution: &'life1 SchemaEvolution,
    ) -> Pin<Box<dyn Future<Output = Result<(), FaucetError>> + Send + 'async_trait>>
       where 'life0: 'async_trait,
             'life1: 'async_trait,
             Self: 'async_trait { ... }
    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 'life0: 'async_trait,
             'life1: 'async_trait,
             'life2: 'async_trait,
             'life3: 'async_trait,
             Self: 'async_trait { ... }
    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 'life0: 'async_trait,
             'life1: 'async_trait,
             Self: 'async_trait { ... }
    fn config_schema(&self) -> Value { ... }
    fn connector_name(&self) -> &'static str { ... }
    fn dataset_uri(&self) -> String { ... }
    fn check<'life0, 'life1, 'async_trait>(
        &'life0 self,
        _ctx: &'life1 CheckContext,
    ) -> Pin<Box<dyn Future<Output = Result<CheckReport, FaucetError>> + Send + 'async_trait>>
       where 'life0: 'async_trait,
             'life1: 'async_trait,
             Self: 'async_trait { ... }
}

A sink writes records to an external system.

Required Methods

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

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

Write a batch of records to the destination.

Returns the number of records successfully written.

Provided Methods

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

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

Flush any buffered data to the destination.

The default implementation is a no-op (suitable for sinks that write immediately in write_batch).

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

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

Write a batch and report per-row outcomes.

Sinks whose underlying API exposes per-row results (BigQuery insertAll, Elasticsearch _bulk) override this. The default implementation delegates to Self::write_batch and maps a single success onto a uniform all-Ok(()) vector. An outer failure is bubbled up unchanged so the pipeline’s DLQ router can apply its on_batch_error policy at a single decision point.

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).

Only return true from a sink that genuinely commits both atomically — see write_batch_idempotent. The pipeline rejects DeliveryMode::ExactlyOnce against a sink that returns false.

fn sink_guarantee(&self) -> SinkGuarantee

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

The default derives from the two back-compat primitives: supports_idempotent_writes → AtomicWatermark, else an upsert-capable supported_write_modes → KeyedUpsert, else AtLeastOnce. Existing connectors that override only the primitives automatically advertise the right capability here.

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.

Distinct from sink_guarantee (capability): this reflects the live config. Sinks that flatten a WriteSpec into their config override it as self.config.write.dedups_by_key(). The pipeline consults it to derive the keyed-upsert effectively-once mechanism at run time.

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 'life0: 'async_trait, Self: '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.

Used by the schema-drift policy to diff each page’s shape against the real destination. Default: Ok(None) (drift handling is inert).

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 'life0: 'async_trait, 'life1: 'async_trait, Self: '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.

Default: a typed “unsupported” error. Override only when the backend supports in-place additive DDL (and return true from supports_schema_evolution).

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 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait, Self: 'async_trait,

Write records AND durably record token for scope, atomically.

scope namespaces the watermark (the pipeline passes the per-row state key, e.g. "{name}::{row_id}"). token is a monotonic, fixed-width string (see format_token).

The default is not idempotent: it ignores the token and delegates to write_batch. Override only when the commit is genuinely atomic (and return true from supports_idempotent_writes).

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 'life0: 'async_trait, 'life1: 'async_trait, Self: 'async_trait,

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

fn config_schema(&self) -> Value

Return a JSON Schema describing the configuration this sink accepts.

The schema is auto-generated from the config struct using schemars. Callers can inspect it to discover required fields, types, defaults, and descriptions before constructing the sink.

The default returns an empty object schema.

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 dataset_uri(&self) -> String

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

The default returns "<connector_name>://unknown". Built-in connectors override with a credential-free URI derived from their config. Strip any credentials with redact_uri_credentials. Informational metadata only — never used for I/O.

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

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

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

Unlike sources, a sink has no non-mutating “first page” equivalent (write_batch mutates the destination), so the default returns CheckReport::not_implemented. Built-in sinks override this with a connect / auth / metadata probe.

The probe MUST be idempotent and side-effect-free — no inserts, no residual rows or objects — and must never put credentials or connection strings in a probe reason/hint.

Dyn Compatibility

This trait is dyn compatible.

In older versions of Rust, dyn compatibility was called "object safety".

Implementors

impl<'a, S> Sink for InstrumentedSink<'a, S>

where S: Sink + ?Sized,