Trait SinkDriver

Source

pub trait SinkDriver: Send + Sync {
Show 25 methods    // Required methods
    fn connector_name(&self) -> &'static str;
    fn write<'life0, 'life1, 'life2, 'async_trait>(
        &'life0 self,
        calls: &'life1 CallSlot,
        rows: &'life2 [Value],
        policy: RowPolicy,
    ) -> Pin<Box<dyn Future<Output = Result<PageWrite, FaucetError>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait,
             'life2: 'async_trait;

    // Provided methods
    fn config_schema(&self) -> Value { ... }
    fn dataset_uri(&self) -> String { ... }
    fn call_context(&self) -> CallContext { ... }
    fn check<'life0, 'life1, 'life2, 'async_trait>(
        &'life0 self,
        calls: &'life1 CallSlot,
        ctx: &'life2 CheckContext,
    ) -> Pin<Box<dyn Future<Output = Result<CheckReport, FaucetError>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait,
             'life2: 'async_trait { ... }
    fn flush<'life0, 'life1, 'async_trait>(
        &'life0 self,
        calls: &'life1 CallSlot,
    ) -> Pin<Box<dyn Future<Output = Result<(), FaucetError>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait { ... }
    fn write_modes(&self) -> &'static [WriteMode] { ... }
    fn dedups_by_key(&self) -> bool { ... }
    fn write_is_replay_safe(&self) -> bool { ... }
    fn batch_atomicity(&self) -> BatchAtomicity { ... }
    fn staged_load(&self) -> bool { ... }
    fn capabilities(&self, derived: Capabilities) -> Capabilities { ... }
    fn sink_guarantee(&self, derived: SinkGuarantee) -> SinkGuarantee { ... }
    fn staging(&self) -> Option<&dyn StagingDriver> { ... }
    fn commit_tokens(&self) -> Option<&dyn CommitTokenDriver> { ... }
    fn cleanup(&self) -> Option<&dyn CleanupDriver> { ... }
    fn schema(&self) -> Option<&dyn SchemaDriver> { ... }
    fn rollback(&self) -> Option<&dyn RollbackDriver> { ... }
    fn columnar(&self) -> Option<&dyn ColumnarDriver> { ... }
    fn readback_source(&self) -> Option<(String, Value)> { ... }
    fn complete_run<'life0, 'life1, 'async_trait>(
        &'life0 self,
        calls: &'life1 CallSlot,
    ) -> Pin<Box<dyn Future<Output = Result<(), FaucetError>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait { ... }
    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 { ... }
    fn set_call_context(&self, ctx: CallContext) { ... }
    fn set_roundtrip_recorder(&self, recorder: Arc<RoundtripRecorder>) { ... }
}

Expand description

A destination’s I/O primitives (RFC 0014). Wrap it in a DriverSink to get a Sink.

Every async primitive receives the run’s CallSlot; make each backend request through it (CallSlot::call / CallSlot::call_once) so the run’s deadline, cancellation and round-trip metering apply. A driver never stores a context itself.

The optional groups are how a driver declares a lifecycle: return one and DriverSink advertises the capability and drives it; leave the default None and the capability is off. There is no flag to set that could disagree with the code.

use faucet_core::driver::{DriverSink, PageWrite, RowPolicy, SinkDriver};
use faucet_core::{CallSlot, FaucetError, Sink, Value, async_trait};
use std::sync::Mutex;

/// Appends each page to a vector.
#[derive(Default)]
struct Memory(Mutex<Vec<Value>>);

#[async_trait]
impl SinkDriver for Memory {
    fn connector_name(&self) -> &'static str {
        "memory"
    }
    async fn write(
        &self,
        _calls: &CallSlot,
        rows: &[Value],
        _policy: RowPolicy,
    ) -> Result<PageWrite, FaucetError> {
        self.0.lock().unwrap().extend_from_slice(rows);
        Ok(PageWrite::accepted(rows.len()))
    }
}

let sink = DriverSink::new(Memory::default());
assert_eq!(sink.write_batch(&[faucet_core::json!({"id": 1})]).await?, 1);
assert!(!sink.supports_idempotent_writes()); // no CommitTokenDriver

Required Methods§

Source

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

The connector’s name, the connector metric label.

Source

fn write<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, calls: &'life1 CallSlot, rows: &'life2 [Value], policy: RowPolicy, ) -> Pin<Box<dyn Future<Output = Result<PageWrite, FaucetError>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Write one page. Under RowPolicy::PerRow, report the rows that failed on their own in PageWrite::failed; an Err means no row of the call was written.

Provided Methods§

Source

fn config_schema(&self) -> Value

The JSON Schema of the connector’s config.

Source

fn dataset_uri(&self) -> String

The dataset this sink writes, for lineage (credential-free).

Source

fn call_context(&self) -> CallContext

The connector’s own call defaults (timeout, retry). The run’s context is layered over it.

Source

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

A non-mutating preflight probe.

Source

fn flush<'life0, 'life1, 'async_trait>( &'life0 self, calls: &'life1 CallSlot, ) -> Pin<Box<dyn Future<Output = Result<(), FaucetError>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Make buffered writes durable.

Source

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

The write modes this instance applies. Must include WriteMode::Overwrite exactly when staging is Some (DriverSink::check_declarations).

Source

fn dedups_by_key(&self) -> bool

Whether this instance is configured to merge rows by key (write_mode: upsert|delete with a key).

Source

fn write_is_replay_safe(&self) -> bool

Whether replaying a page after an ambiguous failure converges.

Source

fn batch_atomicity(&self) -> BatchAtomicity

What a failed write leaves behind.

Source

fn staged_load(&self) -> bool

Whether the destination bulk-loads through an object-store stage.

Source

fn capabilities(&self, derived: Capabilities) -> Capabilities

Adjust the capabilities DriverSink derived, for a fact no primitive expresses (how unknown fields are treated, scoped overwrite). Default: unchanged.

Source