{"record":{"id":"c1f1722d73f51a5b","repo":"risingwavelabs/risingwave","slug":"sinkerror-postgres-anyhow-err","errorCode":null,"errorMessage":"SinkError::Postgres(anyhow!(err))","messagePattern":"SinkError::Postgres\\(anyhow!\\(err\\)\\)","errorType":"exception","errorClass":"SinkError::Postgres","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/mod.rs","lineNumber":1309,"sourceCode":"        SinkError::SqlServer(anyhow!(err))\n    }\n}\n\nimpl From<::elasticsearch::Error> for SinkError {\n    fn from(err: ::elasticsearch::Error) -> Self {\n        SinkError::ElasticSearchOpenSearch(anyhow!(err))\n    }\n}\n\nimpl From<::opensearch::Error> for SinkError {\n    fn from(err: ::opensearch::Error) -> Self {\n        SinkError::ElasticSearchOpenSearch(anyhow!(err))\n    }\n}\n\nimpl From<tokio_postgres::Error> for SinkError {\n    fn from(err: tokio_postgres::Error) -> Self {\n        SinkError::Postgres(anyhow!(err))\n    }\n}\n\n#[cfg(test)]\nmod tests {\n    use std::collections::BTreeMap;\n\n    use super::*;\n\n    fn btreemap<const N: usize>(entries: [(&str, &str); N]) -> BTreeMap<String, String> {\n        entries\n            .into_iter()\n            .map(|(key, value)| (key.to_owned(), value.to_owned()))\n            .collect()\n    }\n\n    #[test]\n    fn test_validate_sink_unknown_fields() {","sourceCodeStart":1291,"sourceCodeEnd":1327,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/mod.rs#L1291-L1327","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Test the connection string with psql using the exact DSN from the sink config.","Check Postgres server logs for the corresponding error to see the SQLSTATE code.","Verify the target table schema matches the RisingWave sink schema (types and column names).","For unique violations, ensure the sink uses upsert semantics with a proper primary key.","Confirm SSL mode requirements (e.g. sslmode=require) are supported by the configured connection."],"exampleFix":"// before: opaque wrapped error\nSinkError::Postgres(anyhow!(err))\n// after: surface SQLSTATE for diagnosability\nif let Some(db_err) = err.as_db_error() {\n    tracing::error!(\"postgres error code: {}\", db_err.code());\n}\nSinkError::Postgres(anyhow!(err))","handlingStrategy":"try-catch","validationCode":"// pre-flight the DSN before sink creation\npsql \"$POSTGRES_SINK_DSN\" -c 'SELECT 1' || echo 'sink DSN unusable'\n","typeGuard":"fn is_retryable_pg_error(err: &tokio_postgres::Error) -> bool {\n    use tokio_postgres::error::SqlState;\n    err.code().map(|c| !matches!(c, SqlState::UNIQUE_VIOLATION | SqlState::NOT_NULL_VIOLATION | SqlState::SYNTAX_ERROR)).unwrap_or(true)\n}","tryCatchPattern":"// use SQLSTATE to decide retry vs fatal\nif let Some(db) = err.as_db_error() {\n    match db.code() {\n        SqlState::UNIQUE_VIOLATION => fail_fast(),\n        SqlState::CONNECTION_FAILURE | SqlState::ADMIN_SHUTDOWN => retry_with_backoff(),\n        _ => fail_fast(),\n    }\n}","preventionTips":["Validate the DSN with psql before creating the sink.","Keep the Postgres table schema in sync with the RisingWave sink schema.","Use upsert (ON CONFLICT) semantics when duplicate keys are possible.","Configure SSL mode explicitly if the server requires it."],"tags":["rust","postgres","sink","database"],"backgroundTag":"database-write-failed","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}