{"record":{"id":"e259df3c0c40460a","repo":"risingwavelabs/risingwave","slug":"primary-key-not-defined-for-upsert-postgres-sink","errorCode":null,"errorMessage":"Primary key not defined for upsert Postgres sink (please define in `primary_key` field)","messagePattern":"Primary key not defined for upsert Postgres sink \\(please define in `primary_key` field\\)","errorType":"validation","errorClass":"SinkError::Config","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/postgres.rs","lineNumber":242,"sourceCode":"\nimpl Sink for PostgresSink {\n    type LogSinker = BatchingLogSinker<PostgresSinkWriter>;\n\n    const SINK_NAME: &'static str = POSTGRES_SINK;\n\n    crate::impl_validate_sink_unknown_fields!();\n\n    async fn validate(&self) -> Result<()> {\n        if !(1..=MAX_BATCH_ROWS_LIMIT).contains(&self.config.max_batch_rows) {\n            return Err(SinkError::Config(anyhow!(\n                \"`max_batch_rows` must be between 1 and {}, got {}\",\n                MAX_BATCH_ROWS_LIMIT,\n                self.config.max_batch_rows\n            )));\n        }\n\n        if !self.is_append_only && self.pk_indices.is_empty() {\n            return Err(SinkError::Config(anyhow!(\n                \"Primary key not defined for upsert Postgres sink (please define in `primary_key` field)\"\n            )));\n        }\n\n        ensure_no_foreign_key(&self.config).await?;\n\n        // Verify our sink schema is compatible with Postgres\n        {\n            let pg_conn = self.config.pg_connection_config();\n            let pg_table = PostgresExternalTable::connect(\n                &pg_conn,\n                &self.config.schema,\n                &self.config.table,\n                self.is_append_only,\n                None,\n            )\n            .await\n            .context(format!(","sourceCodeStart":224,"sourceCodeEnd":260,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/postgres.rs#L224-L260","documentation":"An upsert (non-append-only) PostgreSQL sink requires a primary key so rows can be keyed for upsert/delete operations. If the streaming fragment is not append-only and `pk_indices` is empty, validate() rejects the sink and asks the user to define a primary key via the `primary_key` field.","triggerScenarios":"CREATE SINK from a non-append-only source/MV without specifying `primary_key` in the WITH options, so the sink has empty pk_indices.","commonSituations":"Sinking an append-only=false materialized view (or one containing updates/deletes) to Postgres; forgetting that the upstream stream's PK is not automatically propagated to the sink.","solutions":["Add `primary_key = '<column>'` (or a comma-separated list) to the WITH options of CREATE SINK.","Ensure the queried relation has a primary key so pk_indices is populated automatically.","If the data is truly append-only, adjust the query/sink so it is treated as append-only."],"exampleFix":"// before\nCREATE SINK s FROM mv WITH (connector = 'postgres');\n// after\nCREATE SINK s FROM mv WITH (connector = 'postgres', primary_key = 'id');","handlingStrategy":"validation","validationCode":"-- ensure the relation is append-only or a PK is supplied\n-- if relation is not append-only: WITH (... , primary_key = 'id')","typeGuard":null,"tryCatchPattern":"match err {\n    SinkError::Config(e) if e.to_string().contains(\"Primary key not defined\") => {\n        // recreate sink with primary_key option\n    }\n    _ => {}\n}","preventionTips":["Always set primary_key when sinking non-append-only streams.","Check EXPLAIN/query type: updates/deletes imply a PK is required."],"tags":["rust","postgres","sink","primary-key"],"backgroundTag":"missing-required-config-field","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"}