Struct FileSource 

pub struct FileSource { /* private fields */ }

A local-filesystem (or single http(s):// file) source.

Implementations

impl FileSource

pub fn new(config: FileSourceConfig) -> Result<Self, FaucetError>

Validate the config and build the HTTP client when path is a URL. Performs no filesystem I/O.

Trait Implementations

impl Source for FileSource

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

Stream every selected file in order. Pages carry up to batch_size records; in incremental mode a page never spans two files and the last page of each file carries the bookmark past it, so an interrupted run resumes at the next unread file.

fn supports_columnar(&self) -> bool

Available on crate feature arrow only.

The columnar path is taken for an explicit avro, orc or parquet format; auto cannot promise every file decodes to Arrow.

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

The bookmark lives under a key derived from path, so two sources reading different paths never share a position.

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

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

One dataset per readable file under path — no file is opened.

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

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

Reachability without reading data: HEAD for a URL, a listing for a local path.

fn fetch_with_context<'life0, 'life1, 'async_trait>( &'life0 self, context: &'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 stream_batches<'a>( &'a self, context: &'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 source natively as Arrow ColumnarPages.

fn config_schema(&self) -> Value

Return a JSON Schema describing the configuration this source accepts.

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 dataset_uri(&self) -> String

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

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

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

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

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 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, ) -> Pin<Box<dyn Future<Output = Result<(), FaucetError>> + Send + 'async_trait>>

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 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_with_context_incremental<'life0, 'life1, 'async_trait>( &'life0 self, context: &'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, Self: 'async_trait,

Incremental fetch with parent context support.

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 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 legacy_state_keys(&self) -> Vec<String>

Earlier renderings of this source’s natural state_key, newest first. When the current key holds nothing, the pipeline moves a bookmark found under one of these to it before reading (migrate_state_key), so changing how a source derives its key does not restart a resumable run. Return nothing when the key was configured explicitly. Default: none.

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, Self: '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, ) -> Pin<Box<dyn Future<Output = Result<Option<SourceLag>, FaucetError>> + Send + 'async_trait>>

where 'life0: 'async_trait, Self: '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) -> Result<Value, FaucetError>

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 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 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait, Self: '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.

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

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

The columns this stream emits, with their exact types (#955), so a sink that creates its destination types it from the source rather than from the first page’s JSON values. See crate::declared.

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 read_completeness(&self) -> Option<ReadCompleteness>

The signal through which this source reports, during a run, that it is reading its whole dataset (see crate::cleanup). After such a run the pipeline may delete destination rows the run did not write. Default None: the source never makes that claim.

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 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.