Enum PendingRows 

enum PendingRows {
    Insert(Vec<Vec<Option<ScalarAdapter>>>),
    Upsert {
        absorbed: usize,
        rows: HashMap<OwnedRow, (usize, PendingOp)>,
    },
}

Rows accumulated across chunks, written out on flush.

Variants

Insert(Vec<Vec<Option<ScalarAdapter>>>)

Upsert

Keeps only the last operation per key: PostgreSQL rejects an INSERT .. ON CONFLICT DO UPDATE affecting the same row twice. Unlike ChangeBuffer, insert-then-delete still emits the delete: at-least-once replay may regroup a committed batch with new chunks, so same-key duplicates are legal and the downstream state is unknown.

Fields

absorbed: usize

Rows absorbed since the last flush; drives the flush threshold so log-store truncation keeps pace even when dedup keeps the map small.

rows: HashMap<OwnedRow, (usize, PendingOp)>

Last op per key, tagged with its arrival sequence: keys distinct to RisingWave may be equal to PostgreSQL (e.g. char(n) padding), so execution order is observable.

Implementations

impl PendingRows

fn absorbed(&self) -> usize

fn absorb( &mut self, chunk: &StreamChunk, key_indices: &[usize], schema_types: &[PgType], )

fn take( &mut self, key_types: &[PgType], ) -> (Vec<Vec<Option<ScalarAdapter>>>, Vec<Vec<Option<ScalarAdapter>>>)

Takes all pending rows out, split into deletes (rebuilt from the keys) and upserts, each in chronological order.