risingwavelabs/risingwave · critical · SinkError

{batch write failure, with context()}

Error message

{batch write failure, with context()}

What it means

When a batched transaction write to Postgres fails and the subsequent rollback also fails, the original error `e` is re-wrapped with a context() description (which sink/table/batch) and returned — surfacing as a batch write failure. If rollback succeeds, the sink falls back to flushing row by row. This error means a whole-batch transaction commit failed AND a clean rollback could not be confirmed.

Source

Thrown at src/connector/src/sink/postgres.rs:678

        .await;
        if let Err(e) = result {
            // Retry any failed batch row by row: keys distinct to RisingWave but equal to
            // PostgreSQL fail a multi-row upsert with SQLSTATE 21000 yet apply cleanly one row at
            // a time; other errors recover on retry or resurface localized to a single row.
            let context = || {
                format!(
                    "failed to execute batched {} statements ({} delete rows, {} write rows)",
                    write_kind.as_str(),
                    deletes.len(),
                    upserts.len()
                )
            };
            if let Err(rollback_err) = transaction.rollback().await {
                tracing::warn!(
                    error = %rollback_err.as_report(),
                    "failed to roll back failed batch"
                );
                return Err(anyhow::Error::new(e).context(context()).into());
            }
            tracing::warn!(error = %e.as_report(), "{}, retrying row by row", context());
            return self.flush_row_by_row(&deletes, &upserts).await;
        }
        transaction.commit().await?;

        Ok(())
    }

    /// Fallback for a failed batched flush: batched deletes first, then one upsert per statement.
    async fn flush_row_by_row(&mut self, deletes: &[PgRow], upserts: &[PgRow]) -> Result<()> {
        let delete_batches = self.prepare_batches(StatementKind::Delete, deletes).await?;
        let statement = self.cached_statement(self.write_kind(), 1).await?;

        let transaction = self.client.transaction().await?;
        execute_batches(&transaction, &delete_batches)
            .await
            .with_context(|| {

View on GitHub (pinned to 6469eb736d)

Solutions

1. Check connectivity to Postgres (network, PG logs, max_connections) — both the write and rollback failing usually means the connection is dead.
2. Rely on RisingWave's retry: the sink retries failed batches downstream; verify sink retry/backoff config and let it recover.
3. Reduce `max_batch_rows`/batch size if constraint conflicts within large batches are implicated.
4. Inspect the wrapped original error (via context) in the logs to find the root cause of the failed batch.

// preflight: ensure PG is reachable and credentials valid
psql 'postgres://user@host/db' -c 'SELECT 1';

match sink.write_err {
    e if e.to_string().contains("batch write failure") => {
        // check PG connectivity, then rely on sink retry / reduce batch size
    }
    _ => {}
}

When it happens

Trigger: A batch INSERT/DELETE fails inside the transaction (constraint violation, connection drop, serialization failure) and `transaction.rollback().await` returns Err.

Common situations: Broken network connection to Postgres mid-batch (both the statement and rollback fail); PG server restarting; dead connections after idle timeouts.

Related errors

• SinkError::SqlServer(anyhow!(err))
• SinkError::Postgres(anyhow!(err))
• sending bulk write command failed, database: {}
• `max_batch_rows` must be between 1 and {}, got {}
• Primary key not defined for upsert Postgres sink (please def

AI-assisted analysis of risingwavelabs/risingwave@6469eb736d (2026-09-11). Data as JSON: /api/errors/ef0149f93799c7a0. Report an issue: GitHub.