Struct S3Source 

pub struct S3Source { /* private fields */ }

Available on crate features source-rest and source-s3 only.

An S3 source that lists and reads objects from a bucket.

Implementations

impl S3Source

pub async fn new(config: S3SourceConfig) -> Result<S3Source, FaucetError>

Create a new S3 source from the given configuration.

Builds the S3 client eagerly so it is reused across calls.

Trait Implementations

impl Source for S3Source

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 records from listed S3 objects without buffering the full scan. Each emitted StreamPage holds up to S3SourceConfig::batch_size records.

The trait-level batch_size argument is ignored in favour of the config field — the config is the user-facing knob the README documents, and routing the pipeline-supplied hint through it would silently override an explicit config value.

Behaviour by format:

• JsonLines / RawText: the object body is decoded line-by-line via tokio::io::AsyncBufReadExt::lines so client-side memory is bounded at O(batch_size) per object. Multi-object scans are flattened — a single page may carry lines drawn from any object.
• JsonArray: each object is buffered fully (the JSON value can only be parsed once the array is complete) and then its records are chunked into pages of batch_size. See the README “Streaming and batching” section for the caveat.

batch_size = 0 is the “no batching” sentinel: one StreamPage is emitted per S3 object (no within-object chunking and no cross-object accumulation). The S3 source has no incremental-replication mode today, so every emitted page carries bookmark: None.

fn is_shardable(&self) -> bool

The S3 source is always shardable: any object set can be split by hash-of-key. Sharding only takes effect when the cluster coordinator calls apply_shard; a plain faucet run reads every object.

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, S3Source: 'async_trait,

Enumerate target hash-modulo shards. Each shard i will read the objects whose key hashes to i (mod target). No I/O: the partition is defined by the hash function, so enumeration is cheap and stable as new objects appear. target <= 1 yields a single whole-dataset shard.

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, S3Source: 'async_trait,

Narrow this source to one hash-modulo shard. The whole-dataset shard clears any filter (reads every object).

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

where 'life0: 'async_trait, S3Source: 'async_trait,

Enumerate the “directories” directly under the configured prefix via one ListObjectsV2 delimiter (/) listing — each common prefix becomes a prefix dataset. When the listing returns no common prefixes but does return objects directly under the prefix, each object (first page only, capped at DISCOVER_MAX_OBJECTS = 1000) becomes an object dataset instead. No recursion and no data scan — object counts would require paging the whole listing, so estimated_rows is never set.

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 'life0: 'async_trait, 'life1: 'async_trait, S3Source: 'async_trait,

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

fn config_schema(&self) -> Value

Return a JSON Schema describing the configuration this source accepts.

fn dataset_uri(&self) -> String

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

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

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

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