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 optional exactly-once semantics via co-transactional epoch storage.

Implementations

impl PostgresSink

pub fn new(schema: SchemaRef, config: PostgresSinkConfig) -> Self

Creates a new PostgreSQL sink connector.

pub fn state(&self) -> ConnectorState

Returns the current connector state.

pub fn current_epoch(&self) -> u64

Returns the current epoch.

pub fn last_committed_epoch(&self) -> u64

Returns the last committed epoch.

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

Builds the COPY BINARY SQL statement.

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

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

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

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

Builds the DELETE SQL for changelog deletes.

DELETE FROM public.events WHERE id = ANY($1::int8[])

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

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

pub fn build_offset_table_sql() -> &'static str

Builds CREATE TABLE DDL for the offset tracking table.

pub fn build_epoch_commit_sql() -> &'static str

Builds the epoch commit SQL.

pub fn build_epoch_recover_sql() -> &'static str

Builds the epoch recovery SQL.

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 metadata columns (those starting with _).

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

Opens the connector and initializes the connection.

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,

Writes a batch of records to the external system.

fn schema(&self) -> SchemaRef

Returns the expected input schema for this sink.

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,

Begins a new epoch for exactly-once processing.

fn pre_commit<'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,

Pre-commits the current epoch (phase 1 of two-phase commit).

fn commit_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,

Commits the current epoch (phase 2 of two-phase commit).

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,

Rolls back the current epoch.

fn health_check(&self) -> HealthStatus

Returns the current health status of the connector.

fn metrics(&self) -> ConnectorMetrics

Returns current metrics from the connector.

fn capabilities(&self) -> SinkConnectorCapabilities

Returns the capabilities of this sink connector.

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,

Flushes any buffered data to the external system.

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,

Closes the connector and releases all resources.

fn as_schema_registry_aware(&self) -> Option<&dyn SchemaRegistryAware>

Returns this connector as a SchemaRegistryAware, if supported.

fn as_schema_evolvable(&self) -> Option<&dyn SchemaEvolvable>

Returns this connector as a SchemaEvolvable, if supported.