Struct PostgresSink 

pub struct PostgresSink { /* private fields */ }

PostgreSQL sink connector.

Writes Arrow RecordBatch to PostgreSQL tables using COPY BINARY (append) or UNNEST-based upsert with durable at-least-once semantics.

Implementations

impl PostgresSink

pub fn new( schema: SchemaRef, config: PostgresSinkConfig, registry: Option<&Registry>, ) -> Self

Creates a new PostgreSQL sink connector.

pub fn state(&self) -> ConnectorState

Returns the current connector state.

pub fn buffered_rows(&self) -> usize

Returns the number of buffered rows pending flush.

pub fn sink_metrics(&self) -> &PostgresSinkMetrics

Returns a reference to the sink metrics.

pub fn build_copy_sql( schema: &SchemaRef, config: &PostgresSinkConfig, ) -> Result<String, ConnectorError>

Builds the COPY BINARY SQL statement.

COPY "public"."events" ("id", "value", "ts") FROM STDIN BINARY

Errors

Returns an error when the schema or sink configuration cannot produce a valid COPY.

pub fn build_upsert_sql( schema: &SchemaRef, config: &PostgresSinkConfig, ) -> Result<String, ConnectorError>

Builds the UNNEST-based upsert SQL statement.

INSERT INTO "public"."target" ("id", "value", "updated_at")
SELECT * FROM UNNEST($1::int8[], $2::text[], $3::timestamptz[])
ON CONFLICT (id) DO UPDATE SET
    value = EXCLUDED.value,
    updated_at = EXCLUDED.updated_at

Errors

Returns an error when the schema or sink configuration cannot produce a valid upsert.

pub fn build_delete_sql( schema: &SchemaRef, config: &PostgresSinkConfig, ) -> Result<String, ConnectorError>

Builds the DELETE SQL for changelog deletes. One array parameter is bound per primary-key column ($1, $2, …), each holding that column’s values for the batch’s deleted keys.

-- single PK
DELETE FROM "public"."events" WHERE "id" = ANY($1::int8[])
-- composite PK: match tuple-wise via UNNEST, not the cross-product
DELETE FROM "public"."events" AS "target"
  USING UNNEST($1::int8[], $2::text[]) AS "keys"("id", "name")

Errors

Returns an error when the configured primary key cannot produce a valid delete.

pub fn build_create_table_sql( schema: &SchemaRef, config: &PostgresSinkConfig, ) -> Result<String, ConnectorError>

Builds CREATE TABLE DDL from the Arrow schema.

CREATE TABLE IF NOT EXISTS "public"."events" (
    "id" BIGINT NOT NULL,
    "value" TEXT,
    "ts" TIMESTAMPTZ,
    PRIMARY KEY ("id")
)

Errors

Returns an error when an Arrow field or identifier cannot be represented in PostgreSQL.

pub fn split_changelog_batch( batch: &RecordBatch, ) -> Result<(RecordBatch, RecordBatch), ConnectorError>

Splits a changelog RecordBatch into insert and delete batches.

Uses the _op metadata column:

• "I" (insert), "U" (update-after), "r" (snapshot read) → insert batch
• "D" (delete) → delete batch

The returned batches exclude the exact changelog metadata field set.

Errors

Returns ConnectorError::ConfigurationError if the _op column is missing or not a string type.

Trait Implementations

impl Debug for PostgresSink

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl SinkConnector for PostgresSink

fn contract( &self, config: &ConnectorConfig, ) -> Result<SinkContract, ConnectorError>

Declare durability, placement, and input semantics for this exact configuration without opening files, sockets, clients, or transactions.

fn open<'life0, 'life1, 'async_trait>( &'life0 mut self, config: &'life1 ConnectorConfig, ) -> Pin<Box<dyn Future<Output = Result<(), ConnectorError>> + Send + 'async_trait>>

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

Open the connection and prepare to accept writes.

fn write_batch<'life0, 'life1, 'async_trait>( &'life0 mut self, batch: &'life1 RecordBatch, ) -> Pin<Box<dyn Future<Output = Result<WriteResult, ConnectorError>> + Send + 'async_trait>>

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

Implementations using ConnectorCancellationPolicy::CancelSafe must remain valid when this future is dropped at a deadline. For ConnectorCancellationPolicy::RetireConnector, cancellation makes the complete connector instance terminal before later work can be processed.

fn schema(&self) -> SchemaRef

Expected Arrow schema of input batches.

fn suggested_write_timeout(&self) -> Duration

Default per-call write_batch I/O timeout. Users can override this via the sink.write.timeout.ms connector property.

fn flush_interval(&self) -> Duration

Maximum residence time for a non-empty sink buffer before the runtime invokes flush. Checkpoint-committable sinks ignore the periodic timer and flush only through their checkpoint protocol.

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

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

Must be internally bounded — the sink task’s periodic timer calls this on every tick. Thorough drains belong in pre_commit / coordinated commit / close, not here.

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

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

Close the sink and release resources.

fn cancellation_policy(&self) -> ConnectorCancellationPolicy

Deadline behavior required by the underlying client implementation.

fn terminal_task_tracker(&self) -> Option<ConnectorTaskTracker>

Observe detached tasks whose lifetime may outlast this connector value.

fn begin_epoch<'life0, 'async_trait>( &'life0 mut self, _epoch: u64, ) -> Pin<Box<dyn Future<Output = Result<(), ConnectorError>> + Send + 'async_trait>>

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

Begin checkpoint-owned staging. Called only for an admitted checkpoint-committable contract; weaker sinks use the no-op default.

fn pre_commit<'life0, 'async_trait>( &'life0 mut self, _epoch: u64, ) -> Pin<Box<dyn Future<Output = Result<Option<Vec<u8>>, ConnectorError>> + Send + 'async_trait>>

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

Flush + prepare, but do not finalize externally. The runtime persists the checkpoint decision before a designated committer finalizes the collected descriptors; on failure it calls rollback_epoch.

fn rollback_epoch<'life0, 'async_trait>( &'life0 mut self, _epoch: u64, ) -> Pin<Box<dyn Future<Output = Result<(), ConnectorError>> + Send + 'async_trait>>

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

Must be idempotent. The runtime calls this on every checkpoint-committable sink after proving a pre-decision failure, including sinks that never completed pre_commit.

fn as_coordinated_committer(&self) -> Option<&dyn CoordinatedCommitter>

Leader-side committer for a checkpoint-committable contract; None for every weaker contract.