Struct PostgresSinkWriter 

pub struct PostgresSinkWriter {
    client: Client,
    schema: Schema,
    schema_name: String,
    table_name: String,
    key_indices: Vec<usize>,
    key_types: Vec<Type>,
    schema_types: Vec<Type>,
    max_batch_rows: usize,
    write_statements: HashMap<usize, Statement>,
    delete_statements: HashMap<usize, Statement>,
    pending: PendingRows,
}

Fields

client: Client

schema: Schema

schema_name: String

table_name: String

key_indices: Vec<usize>

Columns identifying a downstream row: the sink pk, or all columns when there is no pk. Never empty.

key_types: Vec<Type>

schema_types: Vec<Type>

max_batch_rows: usize

write_statements: HashMap<usize, Statement>

Prepared statements keyed by tuple count; only power-of-two sizes, so the caches stay small.

delete_statements: HashMap<usize, Statement>

pending: PendingRows

Implementations

impl PostgresSinkWriter

async fn new( config: PostgresConfig, schema: Schema, pk_indices: Vec<usize>, is_append_only: bool, writer_param: &SinkWriterParam, ) -> Result<Self>

fn write_kind(&self) -> StatementKind

fn create_sql(&self, kind: StatementKind, n_tuples: usize) -> String

fn statement_cache( &mut self, kind: StatementKind, ) -> &mut HashMap<usize, Statement>

async fn cached_statement( &mut self, kind: StatementKind, n_tuples: usize, ) -> Result<Statement>

async fn prepare_batches<'a>( &mut self, kind: StatementKind, rows: &'a [Vec<Option<ScalarAdapter>>], ) -> Result<Vec<(Statement, &'a [Vec<Option<ScalarAdapter>>])>>

Pairs each sub-batch with a prepared statement of matching tuple count. Batches are split into power-of-two sizes so that once warm, every size hits the statement cache.

async fn flush(&mut self) -> Result<()>

Writes out all pending rows in a single transaction: all deletes first, then upserts in chronological order. Deletes are not interleaved by time because every upsert surviving dedup is live in RisingWave and must reach the target even if a PG-equal key was deleted after it.

async fn flush_row_by_row( &mut self, deletes: &[Vec<Option<ScalarAdapter>>], upserts: &[Vec<Option<ScalarAdapter>>], ) -> Result<()>

Fallback for a failed batched flush: batched deletes first, then one upsert per statement.

Trait Implementations

impl BatchingSinkWriter for PostgresSinkWriter

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

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

Barriers bound the sink’s visibility latency, so they always flush.

fn write_batch<'life0, 'async_trait>( &'life0 mut self, chunk: StreamChunk, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

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

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

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

Commits buffered data if a batch is ready. Returning true means everything received so far is visible downstream, allowing the log store to truncate up to this point.