risingwavelabs/risingwave · error · SinkError::Postgres

SinkError::Postgres(anyhow!(err))

Error message

SinkError::Postgres(anyhow!(err))

What it means

RisingWave converts any tokio_postgres::Error into the Postgres SinkError variant. Any failure from the tokio-postgres driver while sinking rows to PostgreSQL is wrapped into this variant, preserving the display string via anyhow.

Source

Thrown at src/connector/src/sink/mod.rs:1309

        SinkError::SqlServer(anyhow!(err))
    }
}

impl From<::elasticsearch::Error> for SinkError {
    fn from(err: ::elasticsearch::Error) -> Self {
        SinkError::ElasticSearchOpenSearch(anyhow!(err))
    }
}

impl From<::opensearch::Error> for SinkError {
    fn from(err: ::opensearch::Error) -> Self {
        SinkError::ElasticSearchOpenSearch(anyhow!(err))
    }
}

impl From<tokio_postgres::Error> for SinkError {
    fn from(err: tokio_postgres::Error) -> Self {
        SinkError::Postgres(anyhow!(err))
    }
}

#[cfg(test)]
mod tests {
    use std::collections::BTreeMap;

    use super::*;

    fn btreemap<const N: usize>(entries: [(&str, &str); N]) -> BTreeMap<String, String> {
        entries
            .into_iter()
            .map(|(key, value)| (key.to_owned(), value.to_owned()))
            .collect()
    }

    #[test]
    fn test_validate_sink_unknown_fields() {

View on GitHub (pinned to 6469eb736d)

Solutions

1. Test the connection string with psql using the exact DSN from the sink config.
2. Check Postgres server logs for the corresponding error to see the SQLSTATE code.
3. Verify the target table schema matches the RisingWave sink schema (types and column names).
4. For unique violations, ensure the sink uses upsert semantics with a proper primary key.
5. Confirm SSL mode requirements (e.g. sslmode=require) are supported by the configured connection.

Example fix

// before: opaque wrapped error
SinkError::Postgres(anyhow!(err))
// after: surface SQLSTATE for diagnosability
if let Some(db_err) = err.as_db_error() {
    tracing::error!("postgres error code: {}", db_err.code());
}
SinkError::Postgres(anyhow!(err))

// pre-flight the DSN before sink creation
psql "$POSTGRES_SINK_DSN" -c 'SELECT 1' || echo 'sink DSN unusable'

fn is_retryable_pg_error(err: &tokio_postgres::Error) -> bool {
    use tokio_postgres::error::SqlState;
    err.code().map(|c| !matches!(c, SqlState::UNIQUE_VIOLATION | SqlState::NOT_NULL_VIOLATION | SqlState::SYNTAX_ERROR)).unwrap_or(true)
}

// use SQLSTATE to decide retry vs fatal
if let Some(db) = err.as_db_error() {
    match db.code() {
        SqlState::UNIQUE_VIOLATION => fail_fast(),
        SqlState::CONNECTION_FAILURE | SqlState::ADMIN_SHUTDOWN => retry_with_backoff(),
        _ => fail_fast(),
    }
}

When it happens

Trigger: Connection establishment failure, lost connection during writes, SQL syntax errors in generated statements, constraint violations (unique/PK/not-null), or type mismatches between RisingWave and Postgres columns.

Common situations: Wrong host/port/user/password in the sink DSN, target database or table missing, duplicate-key violations on upsert sinks, schema drift after the sink was created, or Postgres SSL requirements not met.

Related errors

• `max_batch_rows` must be between 1 and {}, got {}
• Primary key not defined for upsert Postgres sink (please def
• Column count mismatch: Postgres table has {} columns, but si
• Column `{}` not found in Postgres table `{}`
• Column `{}` in Postgres table `{}` has type `{}`, but sink s

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