Struct RowChunkDriver 

pub struct RowChunkDriver<C> { /* private fields */ }

Long-lived SourceDriver over a RowChunkSource (CSV-like full-sync pattern).

Each poll loads one chunk, converts rows to upsert changes, and returns them as positioned events. run_source_runtime may poll the next chunk while earlier chunks are still transforming or sinking.

Implementations

impl<C> RowChunkDriver<C>

pub fn new(source: C) -> Self

Wrap a chunk source. next_index seeds Row::index / positions.

pub fn sunk_count(&self) -> u64

Rows successfully sunk (pre-transform input count via SourceDriver::note_sunk_events).

Trait Implementations

impl<C> SourceDriver for RowChunkDriver<C>

where C: RowChunkSource,

type Position = u64

Checkpoint / resume position type.

fn poll_work<'life0, 'async_trait>( &'life0 mut self, ) -> Pin<Box<dyn Future<Output = Result<Vec<PositionedEvent<Self::Position>>>> + Send + 'async_trait>>

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

Next work items (row and/or relation changes). May be empty on idle.

fn advance_watermark<'life0, 'async_trait>( &'life0 mut self, _position: Self::Position, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

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

Mark position sink-safe. May be in-memory only, broker offset commit, or both — durable store writes belong in persist_checkpoint unless policy is CheckpointPolicy::AdvanceOnly.

fn is_finished(&self) -> bool

Whether the driver will produce no more work (EOF).

fn checkpoint_policy(&self) -> CheckpointPolicy

Checkpoint persistence policy. Default: CheckpointPolicy::PersistAfterAdvance.

fn note_sunk_events(&mut self, count: u64)

Notify the driver that count input (pre-transform) events are accounted for before advance_watermark.

fn between_events<'life0, 'async_trait>( &'life0 mut self, ) -> Pin<Box<dyn Future<Output = Result<Vec<ControlSignal>>> + Send + 'async_trait>>

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

Polled between apply cycles; return control signals to handle before the next poll_work.

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

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

Handle ControlSignal::SchemaRefresh. Default: no-op.

fn on_adhoc_snapshot<'life0, 'life1, 'life2, 'async_trait>( &'life0 mut self, _tables: &'life1 [String], _apply: &'life2 dyn AdhocApply, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

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

Handle ControlSignal::AdHocSnapshot.

fn stop_reason(&self) -> Option<StopReason>

If Some, the runtime stops after draining in-flight apply work.

fn persist_checkpoint<'life0, 'async_trait>( &'life0 mut self, _position: Self::Position, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

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

Persist a sink-safe checkpoint (called after successful sink + advance_watermark according to CheckpointPolicy). Default: no-op.

fn read_progress_for_persist<'life0, 'async_trait>( &'life0 mut self, ) -> Pin<Box<dyn Future<Output = Result<Option<Self::Position>>> + Send + 'async_trait>>

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

Optional sink-safe position to persist when CheckpointPolicy::IntervalWhenDrained finds the apply window empty but nothing was advanced since the last persist (e.g. filtered-only binlog traffic that advanced the read cursor with no work items).