Struct DynamoDbSource 

pub struct DynamoDbSource { /* private fields */ }

Amazon DynamoDB source. See the crate README for semantics.

Implementations

impl DynamoDbSource

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

Create a new DynamoDB source. Validates the config and builds the AWS clients; no network I/O until streaming.

Trait Implementations

impl Source for DynamoDbSource

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

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

Streams mode: every current shard opened at its trim horizon, so a CDC run started after a snapshot replays the retained window and converges. Other modes: None.

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

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

Streams mode: now minus the creation time of the newest record read from each open shard, the largest across shards; before this run has read, the age of each open shard’s oldest unconsumed record at the bookmark (one Limit: 1 read per shard). A shard with nothing to read counts as caught up. A failed probe is logged once and reported as no lag. Other modes: None.

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,

One shard per scan segment: segments when configured above 1, otherwise target.

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,

ListTables + DescribeTable per table (metadata only).

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,

Side-effect-free probe: DescribeTable, plus DescribeStream in streams mode (no records read, no iterator opened).

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_pages<'a>( &'a self, _context: &'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, ) -> 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 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 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 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 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 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_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_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 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 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_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 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 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 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 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.