Struct Pipeline 

pub struct Pipeline<'a, So, Si>
where
    So: Source + ?Sized,
    Si: Sink + ?Sized,
{ /* private fields */ }

A pipeline that moves data from a Source to a Sink.

The pipeline is generic over the source and sink types — any combination of connectors works as long as they implement the respective traits.

Implementations

impl<'a, So, Si> Pipeline<'a, So, Si>

where So: Source + ?Sized, Si: Sink + ?Sized,

pub fn new(source: &'a So, sink: &'a Si) -> Pipeline<'a, So, Si>

Create a new pipeline from a source and a sink.

pub fn with_state_store( self, store: Arc<dyn StateStore>, ) -> Pipeline<'a, So, Si>

Attach a StateStore for persistent incremental-replication bookmarks.

When configured, run() will:

1. Read any previously stored bookmark at the source’s state_key and call apply_start_bookmark on the source so it can resume from that point.
2. Run the fetch + write as usual.
3. Persist the new bookmark only after the sink confirms the batch was written and flushed.

Sources that do not return a state_key are unaffected — the store is consulted only when the source opts in.

pub fn with_name(self, name: impl Into<String>) -> Pipeline<'a, So, Si>

Set the pipeline name used in spans and metric labels. Defaults to "unnamed" when unset.

pub fn with_row(self, row: impl Into<String>) -> Pipeline<'a, So, Si>

Set the matrix row id used in spans and metric labels. Defaults to "" (Prometheus treats empty labels as absent).

pub fn with_run_id(self, run_id: impl Into<String>) -> Pipeline<'a, So, Si>

Set an explicit run id (UUIDv7-shaped). When unset, Pipeline::run generates one. Used only as a tracing span attribute — never a metric label.

pub fn with_dlq(self, dlq: DlqConfig) -> Pipeline<'a, So, Si>

Attach a DLQ for per-row failure routing.

pub fn with_quality(self, quality: Arc<CompiledQuality>) -> Pipeline<'a, So, Si>

Available on crate feature quality only.

Attach a compiled quality spec. Checks run after transforms, before the sink, per page.

pub fn with_adaptive(self, cfg: AdaptiveBatchConfig) -> Pipeline<'a, So, Si>

Attach an adaptive batch-size controller (opt-in). When enabled, the pipeline reslices each source page into sub-batches whose size the controller tunes from observed sink latency + error rate.

pub fn with_cancel(self, cancel: CancellationToken) -> Pipeline<'a, So, Si>

Attach a cancellation token. When cancelled mid-run, the streaming loop stops at the next page boundary, flushes the sink(s) so buffered output (e.g. a Parquet footer) is durable, and returns the partial result instead of leaving the file unreadable (#146 H16).

pub async fn run(&self) -> Result<PipelineResult, FaucetError>

Run the pipeline in streaming mode.

1. Loads the stored bookmark and pushes it to the source (if a state store is configured and the source returns a state_key).
2. Drives Source::stream_pages with DEFAULT_BATCH_SIZE, writing each page to the sink as it arrives via Sink::write_batch.
3. Whenever a page carries Some(bookmark), flushes the sink and persists the bookmark to the state store before polling the next page. This makes per-page CDC checkpointing automatic.
4. Flushes the sink one final time after the stream completes.
5. Returns a PipelineResult with the total count and the last bookmark observed.