Struct TransformingSource 

pub struct TransformingSource { /* private fields */ }

Source decorator that applies a fixed list of compiled stages to every record. Emits faucet_transform_* metrics per page via instrumented_apply_stages.

Example

use faucet_core::{RecordTransform, Source, TransformingSource};
use faucet_core::observability::Labels;
use faucet_core::stage::TransformStage;
use faucet_core::transform::KeyCaseMode;

let inner: Box<dyn Source> = build_inner();
let wrapped = TransformingSource::new(
    inner,
    vec![TransformStage::Map(RecordTransform::KeysCase { mode: KeyCaseMode::Snake })],
    Labels::for_named("rest"),
).unwrap();

Implementations

impl TransformingSource

pub fn new( inner: Box<dyn Source>, stages: Vec<TransformStage>, labels: Labels, ) -> Result<TransformingSource, FaucetError>

Compile stages and wrap inner. Returns FaucetError::Transform if any stage’s compilation fails (e.g. invalid regex in RenameKeys). The chain stays on the Value path (no columnar batch forms).

pub fn new_with_batches( inner: Box<dyn Source>, stages: Vec<TransformStage>, batch_fns: Vec<Option<Arc<dyn Fn(RecordBatch) -> Result<RecordBatch, FaucetError> + Sync + Send>>>, labels: Labels, ) -> Result<TransformingSource, FaucetError>

Available on crate feature arrow only.

Like new, but each stage may carry an Arrow RecordBatch form (batch_fns[i] parallels stages[i]), so the chain can run on the columnar fast path (#375) when the inner source and sink are Arrow-native and every stage supplies one. Used by the CLI for sql transforms.

Trait Implementations

impl Source for TransformingSource

fn supports_columnar(&self) -> bool

Available on crate feature arrow only.

Columnar only when the inner source is columnar and every stage has an Arrow batch form (today: the SQL transform). Any Value-only stage (Map / Filter / Explode / CdcUnwrap / Custom / plain PageFn) keeps the whole chain on the Value path (#375).

fn stream_batches<'a>( &'a self, ctx: &'a HashMap<String, Value>, batch_size: usize, ) -> Pin<Box<dyn Stream<Item = Result<ColumnarPage, FaucetError>> + Send + 'a>>

Available on crate feature arrow only.

Stream the inner source’s Arrow batches with every stage’s batch form applied in declared order — so parquet → sql → parquet runs Arrow end-to-end. Only reached when supports_columnar is true, i.e. every stage has a batch form.

fn fetch_with_context<'life0, 'life1, 'async_trait>( &'life0 self, ctx: &'life1 HashMap<String, Value>, ) -> Pin<Box<dyn Future<Output = Result<Vec<Value>, FaucetError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, TransformingSource: 'async_trait,

Primary fetch method. Receives context from a parent source’s records.

fn fetch_with_context_incremental<'life0, 'life1, 'async_trait>( &'life0 self, ctx: &'life1 HashMap<String, Value>, ) -> Pin<Box<dyn Future<Output = Result<(Vec<Value>, Option<Value>), FaucetError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, TransformingSource: 'async_trait,

Incremental fetch with parent context support.

fn stream_pages<'a>( &'a self, ctx: &'a HashMap<String, Value>, batch_size: usize, ) -> Pin<Box<dyn Stream<Item = Result<StreamPage, FaucetError>> + Send + 'a>>

Stream records page-by-page so the pipeline can write to the sink as pages arrive instead of buffering the full result set.

fn state_key(&self) -> Option<String>

Stable key under which this source’s incremental-replication bookmark should be persisted in a StateStore.

fn apply_start_bookmark<'life0, 'async_trait>( &'life0 self, bookmark: Value, ) -> Pin<Box<dyn Future<Output = Result<(), FaucetError>> + Send + 'async_trait>>

where 'life0: 'async_trait, TransformingSource: 'async_trait,

Apply a bookmark loaded from a StateStore as this run’s starting point.

fn supports_exactly_once(&self) -> bool

Whether this source deterministically replays the same page sequence from a given bookmark — the requirement for the atomic-watermark effectively-once path (a non-deterministic replay could cause the pipeline to skip a page whose contents differ from the one already committed). Default: false.

fn replay_guarantee(&self) -> ReplayGuarantee

The typed replay capability this source advertises — see ReplayGuarantee.

fn capture_resume_position<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Option<Value>, FaucetError>> + Send + 'async_trait>>

where 'life0: 'async_trait, TransformingSource: 'async_trait,

Capture the source’s current replication position without consuming any changes, ensuring any server-side resource (e.g. a logical replication slot) needed to later resume from that position exists.

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

Stable identifier used as the connector label on metrics and the connector attribute on spans. Defaults to the final segment of std::any::type_name::<Self>(), e.g. "RestSource". Built-in connectors override with a short, friendly snake_case name (e.g. "rest"). Must return a non-empty string; observability decorators fall back to "unknown" in release builds if it is empty (and debug_assert! in debug builds).

fn dataset_uri(&self) -> String

Logical dataset identity for lineage emission, following OpenLineage naming conventions (https://openlineage.io/docs/spec/naming).

fn fetch_all<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Vec<Value>, FaucetError>> + Send + 'async_trait>>

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

Convenience: fetch with no parent context.

fn fetch_all_incremental<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<(Vec<Value>, Option<Value>), FaucetError>> + Send + 'async_trait>>

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

Convenience: incremental fetch with no parent context.

fn config_schema(&self) -> Value

Return a JSON Schema describing the configuration this source accepts.

fn is_shardable(&self) -> bool

Whether this source can split its work into independent shards for clustered (Mode B) execution. Default: false (single whole-dataset shard). Sources with a natural partition (object-store prefixes, table primary-key ranges) override this to true and implement enumerate_shards + apply_shard.

fn enumerate_shards<'life0, 'async_trait>( &'life0 self, _target: usize, ) -> Pin<Box<dyn Future<Output = Result<Vec<ShardSpec>, FaucetError>> + Send + 'async_trait>>

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

Enumerate the shards this source splits into, aiming for roughly target of them (a hint — the source may return fewer, e.g. when the dataset is small, or one per natural partition regardless of target).

fn apply_shard<'life0, 'life1, 'async_trait>( &'life0 self, _shard: &'life1 ShardSpec, ) -> Pin<Box<dyn Future<Output = Result<(), FaucetError>> + Send + 'async_trait>>

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

Narrow this source instance to a single shard before streaming.

fn supports_discover(&self) -> bool

Whether this source can enumerate the datasets behind its connection via discover. Default: false. Sources backed by an introspectable catalog (database information_schema, MongoDB collections, Elasticsearch indices, object-store prefixes) override this to true.

fn discover<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Vec<DatasetDescriptor>, FaucetError>> + Send + 'async_trait>>

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

Enumerate the datasets living behind this source’s connection — one DatasetDescriptor per table / collection / index / prefix, each carrying a partial config override that selects it (used by faucet discover to scaffold one matrix row per dataset).

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

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

Run a fast, non-mutating preflight probe (used by faucet doctor).