Trait PipelineCallback 

pub trait PipelineCallback:
    Send
    + 'static
    + Send {
    // Required methods
    fn execute_cycle(
        &mut self,
        source_batches: &FxHashMap<Arc<str>, Vec<RecordBatch>>,
        watermark: i64,
    ) -> impl Future<Output = Result<FxHashMap<Arc<str>, Vec<RecordBatch>>, String>> + Send;
    fn push_to_streams(&self, results: &FxHashMap<Arc<str>, Vec<RecordBatch>>);
    fn write_to_sinks(
        &mut self,
        results: &FxHashMap<Arc<str>, Vec<RecordBatch>>,
    ) -> impl Future<Output = ()> + Send;
    fn extract_watermark(&mut self, source_name: &str, batch: &RecordBatch);
    fn filter_late_rows(
        &self,
        source_name: &str,
        batch: &RecordBatch,
    ) -> Option<RecordBatch>;
    fn current_watermark(&self) -> i64;
    fn maybe_checkpoint(
        &mut self,
        force: bool,
        source_offsets: FxHashMap<String, SourceCheckpoint>,
    ) -> impl Future<Output = Option<u64>> + Send;
    fn checkpoint_with_barrier(
        &mut self,
        source_checkpoints: FxHashMap<String, SourceCheckpoint>,
    ) -> impl Future<Output = Option<u64>> + Send;
    fn record_cycle(&self, events_ingested: u64, batches: u64, elapsed_ns: u64);
    fn poll_tables(&mut self) -> impl Future<Output = ()> + Send;
    fn apply_control(&mut self, msg: ControlMsg);

    // Provided methods
    fn update_mv_stores(&self, results: &FxHashMap<Arc<str>, Vec<RecordBatch>>) { ... }
    fn is_backpressured(&self) -> bool { ... }
    fn has_deferred_input(&self) -> bool { ... }
}

Callback trait for the coordinator to interact with the rest of the DB. Trait exists for test seam; production impl is ConnectorPipelineCallback.

Required Methods

fn execute_cycle( &mut self, source_batches: &FxHashMap<Arc<str>, Vec<RecordBatch>>, watermark: i64, ) -> impl Future<Output = Result<FxHashMap<Arc<str>, Vec<RecordBatch>>, String>> + Send

Called with accumulated source batches to execute a SQL cycle.

fn push_to_streams(&self, results: &FxHashMap<Arc<str>, Vec<RecordBatch>>)

Called with results to push to stream subscriptions.

fn write_to_sinks( &mut self, results: &FxHashMap<Arc<str>, Vec<RecordBatch>>, ) -> impl Future<Output = ()> + Send

Called with results to write to sinks.

fn extract_watermark(&mut self, source_name: &str, batch: &RecordBatch)

Extract watermark from a batch for a given source.

fn filter_late_rows( &self, source_name: &str, batch: &RecordBatch, ) -> Option<RecordBatch>

Filter late rows from a batch. Returns the filtered batch (or None if all late).

fn current_watermark(&self) -> i64

Get the current pipeline watermark.

fn maybe_checkpoint( &mut self, force: bool, source_offsets: FxHashMap<String, SourceCheckpoint>, ) -> impl Future<Output = Option<u64>> + Send

Perform a periodic (timer-based) checkpoint. At-least-once semantics. For exactly-once, use checkpoint_with_barrier.

fn checkpoint_with_barrier( &mut self, source_checkpoints: FxHashMap<String, SourceCheckpoint>, ) -> impl Future<Output = Option<u64>> + Send

Called when all sources have aligned on a barrier.

fn record_cycle(&self, events_ingested: u64, batches: u64, elapsed_ns: u64)

Record cycle metrics.

fn poll_tables(&mut self) -> impl Future<Output = ()> + Send

Poll table sources for incremental CDC changes.

fn apply_control(&mut self, msg: ControlMsg)

Apply a DDL control message (add/drop stream) to the running pipeline.

Provided Methods

fn update_mv_stores(&self, results: &FxHashMap<Arc<str>, Vec<RecordBatch>>)

Update materialized view stores with cycle results.

fn is_backpressured(&self) -> bool

Returns true when internal buffers are near capacity.

fn has_deferred_input(&self) -> bool

Returns true when deferred operators have pending input to drain.

Dyn Compatibility

This trait is not dyn compatible.

In older versions of Rust, dyn compatibility was called "object safety", so this trait is not object safe.

Implementors