{"record":{"id":"296e610794a900a9","repo":"risingwavelabs/risingwave","slug":"primary-key-mismatch-postgres-table-has-primary-k","errorCode":null,"errorMessage":"Primary key mismatch: Postgres table has primary key on columns {:?}, but sink schema defines primary key on columns {:?}","messagePattern":"Primary key mismatch: Postgres table has primary key on columns (.+?), but sink schema defines primary key on columns (.+?)","errorType":"validation","errorClass":"SinkError::Config","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/postgres.rs","lineNumber":315,"sourceCode":"                                    pg_column,\n                                    sink_column.data_type()\n                                )));\n                            }\n                        }\n                    }\n                }\n            }\n\n            // check that pk matches\n            {\n                let pg_pk_names = pg_table.pk_names();\n                let sink_pk_names = self\n                    .pk_indices\n                    .iter()\n                    .map(|i| &self.schema.fields()[*i].name)\n                    .collect::<HashSet<_>>();\n                if pg_pk_names.len() != sink_pk_names.len() {\n                    return Err(SinkError::Config(anyhow!(\n                        \"Primary key mismatch: Postgres table has primary key on columns {:?}, but sink schema defines primary key on columns {:?}\",\n                        pg_pk_names,\n                        sink_pk_names\n                    )));\n                }\n                for name in pg_pk_names {\n                    if !sink_pk_names.contains(name) {\n                        return Err(SinkError::Config(anyhow!(\n                            \"Primary key mismatch: Postgres table has primary key on column `{}`, but sink schema does not define it as a primary key\",\n                            name\n                        )));\n                    }\n                }\n            }\n        }\n\n        Ok(())\n    }","sourceCodeStart":297,"sourceCodeEnd":333,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/postgres.rs#L297-L333","documentation":"For non-append-only sinks, the sink's declared primary key column-name set must exactly match the Postgres table's primary key column-name set, starting with a count check. If the number of PK columns differs (or the sets otherwise diverge), CREATE SINK fails with the full sets printed via {:?}.","triggerScenarios":"`primary_key` in WITH options names a different number of columns than the PG table's actual PRIMARY KEY constraint; duplicate names collapsing in the HashSet comparison.","commonSituations":"Defining a single-column sink PK while the PG table uses a composite PK (or vice versa); PG table PK changed after sink creation config was written.","solutions":["Make the sink `primary_key` list exactly the PG table's primary key columns (ALTER the WITH options and recreate the sink).","ALTER the Postgres table's primary key to match the sink's declared PK.","If the stream is append-only, drop the primary_key requirement by restructuring the sink as append-only."],"exampleFix":"-- before (pg t has PK (a,b))\nCREATE SINK s FROM mv WITH (connector='postgres', table='t', primary_key='a');\n-- after\nCREATE SINK s FROM mv WITH (connector='postgres', table='t', primary_key='a,b');","handlingStrategy":"validation","validationCode":"SELECT a.attname FROM pg_index i\nJOIN pg_attribute a ON a.attrelid = i.indrelid AND a.attnum = ANY(i.indkey)\nWHERE i.indrelid = 't'::regclass AND i.indisprimary;\n-- sink primary_key must list exactly these columns","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Mirror the composite PK exactly in the primary_key option.","Re-check the PG PK after migrations."],"tags":["rust","postgres","sink","primary-key"],"backgroundTag":"schema-validation-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"}