Struct Pipeline 

pub struct Pipeline { /* private fields */ }

Ordered list of transform stages.

An empty pipeline is identity: is_identity is true and apply helpers return immediately without dispatching any stage.

Identity vs passthrough

Pushing a lone crate::pipeline::Passthrough via push_inplace does not make is_identity return true — the pipeline still has a stage and will dispatch into it. TOML config loading (crate::pipeline::Pipeline::from_config / crate::pipeline::parse_transforms_toml) collapses passthrough-only configs to an empty pipeline so the CLI identity path stays zero-dispatch.

Apply path

Shared apply loop: transform → ordered write → watermark. crate::pipeline::ApplyContext / crate::pipeline::SourceDriver gate on crate::pipeline::BatchTransformer::is_identity (implemented for Pipeline via is_identity) so the transform tokio::task::JoinSet can take an async no-op path (zero stage dispatch). That is not a write-path bypass: identity batches still go through the ordered sink step (including homogeneous Update coalesce to write_universal_*). Only an empty stage list is identity — not “stages happen to be no-ops.”

Schema-aware / FK transforms

Construct an InPlaceTransform with schema (e.g. FK → Thing links) and push_inplace it. Relation edges use the same pipeline via relation transform methods; full join-table→relation conversion may still live in a source crate.

External + relations

External stages exchange both row changes and relation changes over NDJSON (ExternalTransform::exchange_relation_changes / ExternalTransform::exchange_relations). There is no silent relation pass-through. Mixed change+relation batches may not filter/fan-out (length of each kind must be preserved); use homogeneous batches for length changes.

When a single External stage sees a mixed change+relation batch, the two wire exchanges use distinct batch_ids: changes keep the apply batch_id, relations use crate::pipeline::relation_wire_batch_id (high bit set) so workers and outstanding-id tracking never confuse the two sequential exchanges.

Implementations

impl Pipeline

pub fn from_config(cfg: &TransformsConfig) -> Result<Self>

Build a pipeline from validated config.

Empty config → identity (is_identity true). Command stages spawn child workers (persistent) or store argv (transient). Both modes resolve command[0] at config time so bad argv fails fast (transient would otherwise only fail on the first batch).

impl Pipeline

pub fn new() -> Self

Create an empty (identity) pipeline.

pub fn is_identity(&self) -> bool

Whether this pipeline has no stages (identity / no stage dispatch).

Only an empty stage list is identity. A pipeline that contains only crate::pipeline::Passthrough still returns false here — see type-level docs.

pub fn len(&self) -> usize

Number of stages (0 = identity).

pub fn is_empty(&self) -> bool

Whether there are no stages.

pub fn stages(&self) -> &[Stage]

Borrow the stage list.

pub fn push_inplace<T>(&mut self, transform: T)

where T: InPlaceTransform + 'static,

Append an in-place transform stage (library / embedder API).

Note: appending crate::pipeline::Passthrough alone does not yield an identity pipeline (is_identity stays false).

pub fn push_inplace_arc(&mut self, transform: Arc<dyn InPlaceTransform>)

Append a pre-boxed in-place stage.

pub fn push_external(&mut self, external: ExternalTransform)

Append an external (child-stdio) stage.

pub fn transform_rows_inplace(&self, rows: &mut [Row]) -> Result<()>

Transform owned rows in place (sync path — in-place stages only).

Empty pipeline: no-op with no stage dispatch. External stages are not supported here; use crate::pipeline::BatchTransformer::transform_rows (async).

pub fn transform_changes_inplace(&self, changes: &mut [Change]) -> Result<()>

Transform owned changes in place (sync path — in-place stages only).

Empty pipeline: no-op with no stage dispatch. External stages are not supported here; use crate::pipeline::BatchTransformer::transform_changes (async).

pub fn apply_rows(&self, rows: Vec<Row>) -> Result<Vec<Row>>

Consume an owned row batch, transform in place, and return it.

Preferred path for in-place-only pipelines: empty pipeline is a pure move with no transform dispatch.

pub fn apply_changes(&self, changes: Vec<Change>) -> Result<Vec<Change>>

Consume an owned change batch, transform in place, and return it.

pub fn transform_relation_changes_inplace( &self, changes: &mut [RelationChange], ) -> Result<()>

Transform owned relation changes in place (sync — in-place stages only).

pub fn transform_relations_inplace( &self, relations: &mut [Relation], ) -> Result<()>

Transform owned relations in place (sync — in-place stages only).

pub fn apply_relation_changes( &self, changes: Vec<RelationChange>, ) -> Result<Vec<RelationChange>>

Consume owned relation changes, transform in place, return them.

pub fn apply_relations(&self, relations: Vec<Relation>) -> Result<Vec<Relation>>

Consume owned relations, transform in place, return them.

Trait Implementations

impl BatchTransformer for Pipeline

fn is_identity(&self) -> bool

Whether transform dispatch can be skipped entirely (empty pipeline).

fn transform_changes<'life0, 'async_trait>( &'life0 self, batch_id: u64, changes: Vec<Change>, ) -> Pin<Box<dyn Future<Output = Result<Vec<Change>>> + Send + 'async_trait>>

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

Transform an owned change batch. batch_id is monotonic per apply run.

fn transform_rows<'life0, 'async_trait>( &'life0 self, batch_id: u64, rows: Vec<Row>, ) -> Pin<Box<dyn Future<Output = Result<Vec<Row>>> + Send + 'async_trait>>

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

Transform an owned row batch.

fn transform_relation_changes<'life0, 'async_trait>( &'life0 self, batch_id: u64, changes: Vec<RelationChange>, ) -> Pin<Box<dyn Future<Output = Result<Vec<RelationChange>>> + Send + 'async_trait>>

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

Transform an owned relation-change batch.

fn transform_relations<'life0, 'async_trait>( &'life0 self, batch_id: u64, relations: Vec<Relation>, ) -> Pin<Box<dyn Future<Output = Result<Vec<Relation>>> + Send + 'async_trait>>

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

Transform an owned relation (full-sync) batch.

fn transform_events<'life0, 'async_trait>( &'life0 self, batch_id: u64, events: Vec<ApplyEvent>, ) -> Pin<Box<dyn Future<Output = Result<Vec<ApplyEvent>>> + Send + 'async_trait>>

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

Transform a mixed apply-event batch, preserving order.

impl Clone for Pipeline

fn clone(&self) -> Pipeline

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for Pipeline

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl Default for Pipeline

fn default() -> Pipeline

Returns the “default value” for a type.