Struct PostgresSink 

pub struct PostgresSink { /* private fields */ }

A sink that writes JSON records to a PostgreSQL table.

Implementations

impl PostgresSink

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

Create a new PostgreSQL sink. Establishes a connection pool.

Trait Implementations

impl Sink for PostgresSink

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

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

Create the staging table as an empty clone of the target’s columns (CREATE TABLE staging (LIKE target INCLUDING DEFAULTS)), dropping any leftover staging from a crashed run first. The target must already exist (the sink never auto-creates it) — overwrite replaces its rows, not its definition.

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

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

Atomically replace the destination in one transaction. Full overwrite: TRUNCATE target; INSERT INTO target SELECT * FROM staging; DROP staging. Scoped/windowed overwrite (#518): DELETE FROM target WHERE <scope>; INSERT …; DROP staging — only the in-scope rows are replaced, the rest preserved. Postgres runs TRUNCATE and DDL transactionally, so a failure rolls the whole swap back and the prior rows survive.

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

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

Drop the staging table so a failed/cancelled overwrite leaves nothing behind. Best-effort — the destination was never touched.

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

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

Read the live destination schema from pg_catalog as an infer_schema-shaped object ({"type":"object","properties":{…}}), or None when the target table does not exist yet (issue #194).

Reuses the AutoMap column-discovery query shape (scoped to the exact relation via to_regclass), additionally reading a.attnotnull so nullability round-trips through pg_udt_to_json_schema.

fn evolve_schema<'life0, 'life1, 'async_trait>( &'life0 self, evolution: &'life1 SchemaEvolution, ) -> Pin<Box<dyn Future<Output = Result<(), FaucetError>> + Send + 'async_trait>>

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

Apply an additive schema evolution (new columns, lossless widenings, nullability relaxations) to the destination table. Idempotent — ADD COLUMN IF NOT EXISTS, and re-running the same TYPE / DROP NOT NULL is a no-op (issue #194).

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 connectivity probe (faucet doctor).

Acquires a connection from the existing pool and runs SELECT 1. This is non-mutating and idempotent — it validates that the database is reachable and the credentials are accepted without writing anything.

fn write_batch<'life0, 'life1, 'async_trait>( &'life0 self, records: &'life1 [Value], ) -> Pin<Box<dyn Future<Output = Result<usize, FaucetError>> + Send + 'async_trait>>

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

Write records to PostgreSQL.

When config.batch_size > 0 and the input slice is larger than batch_size, the slice is split into chunks of batch_size rows and each chunk is sent as a separate multi-row INSERT. When config.batch_size == 0, the entire slice is sent in a single INSERT — useful when upstream StreamPages are already sized for Postgres’ per-statement bind-parameter limit (~65 535 / num_columns in AutoMap mode).

Acquires one connection from the pool and routes all chunks through it. Each INSERT executes as its own autocommit statement — identical observable behaviour to executing directly on the pool, while keeping the same connection for the entire call (avoids repeated pool-checkout overhead on large batches).

fn write_batch_partial<'life0, 'life1, 'async_trait>( &'life0 self, records: &'life1 [Value], ) -> Pin<Box<dyn Future<Output = Result<Vec<RowOutcome>, FaucetError>> + Send + 'async_trait>>

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

Write a batch and report per-row outcomes.

In append mode this delegates to write_batch and maps a single success onto an all-Ok(()) vector (the trait default). In upsert/delete mode the good rows are applied (upserts + deletes), and only the rows whose key could not be extracted (missing / null key) are reported as Err so the pipeline routes them to the DLQ per-row instead of sending the whole page.

fn connector_name(&self) -> &'static str

Stable identifier used as the connector label on metrics and the connector attribute on spans. See Source::connector_name.

fn config_schema(&self) -> Value

Return a JSON Schema describing the configuration this sink accepts.

fn supports_cleanup(&self) -> bool

Whether this sink can delete a scoped set of rows for scoped cleanup (#478). Default false; the upsert-capable sinks override it.

fn cleanup_scope<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, scope: &'life1 BTreeMap<String, Value>, seen: &'life2 SeenKeys, ) -> Pin<Box<dyn Future<Output = Result<u64, FaucetError>> + Send + 'async_trait>>

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

Delete rows matching scope whose key is not in seen.

fn supported_write_modes(&self) -> &'static [WriteMode]

Write modes this sink can apply. Default: append-only. Sinks that implement key-based merge override this to include WriteMode::Upsert / WriteMode::Delete. The CLI rejects a configured mode that is not in this set at config-load time.

fn is_overwrite(&self) -> bool

Whether this sink instance is configured for full-destination replacement (WriteMode::Overwrite).

fn dedups_by_key(&self) -> bool

Whether this sink instance is configured to dedup by key — i.e. write_mode: upsert (or delete) with a non-empty key, so re-applying a record with the same key converges instead of duplicating. Default: false.

fn supports_schema_evolution(&self) -> bool

Whether this sink can apply additive/widening DDL via evolve_schema. Default: false. The CLI rejects on_drift: evolve against a sink that returns false at config-load.

fn dataset_uri(&self) -> String

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

fn supports_idempotent_writes(&self) -> bool

Whether this sink can durably commit a page’s rows and a commit token in a single atomic transaction. Default: false (at-least-once only).

fn last_committed_token<'life0, 'life1, 'async_trait>( &'life0 self, scope: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Option<String>, FaucetError>> + Send + 'async_trait>>

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

The last token durably committed for scope, or None if this sink has never committed under that scope. Default: None.

fn write_batch_idempotent<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, records: &'life1 [Value], scope: &'life2 str, token: &'life3 str, ) -> Pin<Box<dyn Future<Output = Result<usize, FaucetError>> + Send + 'async_trait>>

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

Write records AND durably record token for scope, atomically.

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

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

Flush any buffered data to the destination.

fn sink_guarantee(&self) -> SinkGuarantee

The strongest delivery guarantee this sink can uphold — see SinkGuarantee.

fn supports_staged_load(&self) -> bool

Whether this sink can bulk-load via an object-store stage — write the page to S3/GCS/Azure, then pull it with the warehouse’s native load command (COPY / COPY INTO / s3() table function) (#528). Default false; warehouse sinks that honour a StagingSpec override it. Object-safe (no args, no generics) so Box<dyn Sink> is unaffected.