Struct RedshiftSource 

pub struct RedshiftSource { /* private fields */ }

A source that executes a SQL query against Amazon Redshift and returns rows as JSON objects.

Implementations

impl RedshiftSource

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

Create a new source. Validates config and builds a lazily-connected pool (no I/O — connectivity is verified on first query or via Source::check).

Trait Implementations

impl Source for RedshiftSource

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

The query’s columns from a describe of it (#955). The server reports no decimal precision, so a NUMERIC is declared unbounded.

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 rows from the sqlx cursor without buffering the full result set. Each emitted StreamPage holds up to RedshiftSourceConfig::batch_size rows.

For incremental replication rows arrive in cursor order, the rows already emitted at the bookmark value are dropped, and every page carries the bookmark after its last row (the final page carries one even when empty).

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,

Preflight probe for faucet doctor: acquire a connection and run SELECT 1 (non-scanning — never executes the configured query).

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

Incremental fetch with parent context support.

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