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, on_collision: Default::default() })],
    Labels::for_named("rest"),
).unwrap();

Implementations

impl TransformingSource

pub fn new( inner: Box<dyn Source>, stages: Vec<TransformStage>, labels: Labels, ) -> Result<Self, 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<PageFnBatchBox>>, labels: Labels, ) -> Result<Self, 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.

pub fn with_row_accounts(self, accounts: Arc<RowAccounts>) -> Self

Count the rows entering and leaving the chain into accounts (#879), so a run summary can say how many a filter removed.

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 Self: 'async_trait, 'life0: 'async_trait, 'life1: '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 Self: 'async_trait, 'life0: 'async_trait, 'life1: '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 config_schema(&self) -> Value

Return a JSON Schema describing the configuration this source accepts.

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, ) -> Fut<'async_trait, Res<()>>

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

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

fn capture_resume_position<'life0, 'async_trait>( &'life0 self, ) -> Fut<'async_trait, Res<Option<Value>>>

where Self: 'async_trait, 'life0: '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 lag<'life0, 'async_trait>( &'life0 self, ) -> Fut<'async_trait, Res<Option<SourceLag>>>

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

How far this source is behind its head (#733) — unread WAL bytes, binlog distance, unconsumed messages, or the age of the oldest unread change.

fn state_schema(&self) -> u32

The version of this source’s bookmark shape (#736). Stored state carries it; bump it whenever the shape changes and teach migrate_state the step. Default 0. Decorators must forward this.

fn migrate_state(&self, from: u32, data: Value) -> Res<Value>

Bring a bookmark stored at shape version from up to state_schema (#736). Must be pure — the migrated value is only persisted by the next bookmark write, so a crash in between re-runs it. The default knows only its own version. Decorators must forward this.

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 consumes_destructively(&self) -> bool

Whether reading from this source removes data from its backend — a queue source that acks, deletes or settles messages as the pipeline moves past each page (SQS, Pub/Sub, JetStream, RabbitMQ). Such a source relies on the linear pipeline writing and flushing a page before it is polled again; anything that reads without durably writing (a preview, a dry run, a --limit run, a plan sample) or that polls ahead of its sinks (a topology graph) would lose those messages for good, so callers refuse it. Default false.

fn replay_guarantee(&self) -> ReplayGuarantee

The typed replay capability this source advertises — see ReplayGuarantee.

fn capabilities(&self) -> Capabilities

What this source can do (RFC 0013, #876). The default derives every field from the per-capability methods above (Capabilities::of_source); override it to declare a capability no method expresses (range digests, lag), keeping the connector’s ConnectorDescriptor in agreement. Decorators must forward this.

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, ) -> Fut<'async_trait, Res<Vec<ShardSpec>>>

where Self: 'async_trait, 'life0: '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, ) -> Fut<'async_trait, Res<()>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: '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, ) -> Fut<'async_trait, Res<Vec<DatasetDescriptor>>>

where Self: 'async_trait, 'life0: '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 record_table(&self, record: &Value) -> Option<String>

The dataset (table / collection) a change record belongs to, for a change stream that carries several tables (#731). A multi-table faucet mirror runs one stream and routes each record to its table’s pipeline by this name, which must match the name the paired bulk source’s discover reports (e.g. public.orders). None means the record belongs to no table (a DDL / control event) — and is the default, so a source that does not override this cannot be used for a multi-table mirror.

fn position_le(&self, a: &Value, b: &Value) -> Option<bool>

Order two of this source’s bookmarks (#731): Some(true) when stream position a is at or before b (every change up to a is also covered by b), Some(false) when it is not, and None when the source cannot tell (the default, or two positions it cannot relate). A multi-table mirror uses it to resume one shared stream from the earliest table and skip, per table, pages that table has already committed.

fn record_covered(&self, record: &Value, position: &Value) -> Option<bool>

Whether the stream had already delivered record’s change at stream position position (#854): Some(true) when a run resumed from position would not deliver it again, Some(false) when it would, and None when the source cannot tell (the default). A multi-table mirror resumes one shared stream from the earliest table, so a replayed page can straddle another table’s committed position; this per-record test drops the records that table already applied, which a per-page position_le check cannot when the replay cuts its pages at different points than the run that committed them.

fn position_min(&self, positions: &[Value]) -> Option<Value>

The earliest stream position every one of positions can resume from (#731): the shared change stream restarts there and each table skips what it has already applied. The default picks the position that position_le orders at or before all the others, and None when there is none (an empty slice, or positions this source cannot order). Sources whose positions are only partially ordered (one cursor per capture instance, say) override it with a component-wise minimum.

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 set_roundtrip_recorder(&self, recorder: Arc<RoundtripRecorder>)

Receive a pre-labelled handle for counting upstream round trips (#638) — calls this connector makes to its own backend.

fn set_call_context(&self, ctx: CallContext)

Receive the run’s CallContext (#880): the cancel token, the round-trip recorder and any deadline every backend call should honour. Called once before streaming.

fn set_run_clock(&self, now: DateTime<Utc>)

Receive the run clock (#769): the instant ${now.*} renders from — faucet run --clock, a schedule tick, a backfill unit’s start — so a source that bounds reads by “now” (REST window slicing) reproduces the run as of that instant rather than the wall clock.

fn set_cancel(&self, cancel: CancellationToken) -> bool

Receive the run’s cancellation token (Pipeline::with_cancel), installed once before streaming.

fn dataset_uri(&self) -> String

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

fn check<'life0, 'life1, 'async_trait>( &'life0 self, ctx: &'life1 CheckContext, ) -> Fut<'async_trait, Res<CheckReport>>

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

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

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

where Self: 'async_trait, 'life0: '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 Self: 'async_trait, 'life0: 'async_trait,

Convenience: incremental fetch with no parent context.

fn native_output_formats(&self) -> &'static [NativeFormat]

Wire formats this source can emit as raw bytes for the native byte-passthrough fast path (#633), in preference order (first = best). Default: &[] (no native fast path).

fn stream_native<'a>( &'a self, context: &'a HashMap<String, Value>, format: NativeFormat, batch_size: usize, ) -> Pin<Box<dyn Stream<Item = Result<NativeBatch, FaucetError>> + Send + 'a>>

Stream the source natively as byte batches in format (#633).

fn range_digest<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, range: &'life1 KeyRange, key: &'life2 str, columns: &'life3 [String], ) -> Pin<Box<dyn Future<Output = Result<Option<ServerDigest>, FaucetError>> + Send + 'async_trait>>

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

Compute a ServerDigest of the rows in range inside the backend, so a matching range ships no rows (#701). columns are the compared columns; key the integer key the range is over. Default Ok(None): not supported, the verifier streams the range instead. Two digests compare only when both sides report the same algorithm.