{"record":{"id":"c76c9cd207972991","repo":"risingwavelabs/risingwave","slug":"primary-key-does-not-match-between-risingwave-sink","errorCode":null,"errorMessage":"primary key does not match between RisingWave sink ({}: [{}]) and SQL Server table {} ({}: [{}])","messagePattern":"primary key does not match between RisingWave sink \\((.+?): \\[(.+?)\\]\\) and SQL Server table (.+?) \\((.+?): \\[(.+?)\\]\\)","errorType":"validation","errorClass":"SinkError::SqlServer","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/sqlserver.rs","lineNumber":261,"sourceCode":"                    }\n                }\n            }\n        }\n\n        if !self.is_append_only && sql_server_pk_count != self.pk_indices.len() {\n            let sql_server_pk_columns = sql_server_table_metadata\n                .values()\n                .filter(|metadata| metadata.is_pk)\n                .map(|metadata| metadata.name.as_str())\n                .collect::<Vec<_>>()\n                .join(\",\");\n            let rw_pk_columns = self\n                .pk_indices\n                .iter()\n                .map(|idx| self.schema[*idx].name.as_str())\n                .collect::<Vec<_>>()\n                .join(\",\");\n            return Err(SinkError::SqlServer(anyhow!(format!(\n                \"primary key does not match between RisingWave sink ({}: [{}]) and SQL Server table {} ({}: [{}])\",\n                self.pk_indices.len(),\n                rw_pk_columns,\n                self.config.full_object_path(),\n                sql_server_pk_count,\n                sql_server_pk_columns,\n            ))));\n        }\n\n        Ok(())\n    }\n\n    async fn new_log_sinker(&self, writer_param: SinkWriterParam) -> Result<Self::LogSinker> {\n        Ok(SqlServerSinkWriter::new(\n            self.config.clone(),\n            self.schema.clone(),\n            self.pk_indices.clone(),\n            self.is_append_only,","sourceCodeStart":243,"sourceCodeEnd":279,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/sqlserver.rs#L243-L279","documentation":"This is the fallback/mismatch report when the per-column PK checks cannot simply pinpoint the difference (e.g. multi-column PKs): it compares the count and ordered names of the RisingWave sink PK against the SQL Server table PK and fails with a full listing of both sides.","triggerScenarios":"validate() detects a PK mismatch (count or composition) between the sink's pk_indices-derived columns and the downstream table's PK, and formats both column lists into this message.","commonSituations":"Composite primary keys defined differently on each side; reordered or partially overlapping PK columns; sink created against a table whose PK was later changed.","solutions":["Read both column lists in the message and align the sink's PRIMARY KEY to exactly the SQL Server table's PK columns (same set and order)","Recreate the SQL Server table with the PK matching the sink definition","Use an append_only sink if PK-based upsert matching is not required"],"exampleFix":"// before\nCREATE SINK ... PRIMARY KEY (a, b) ...; -- table PK is (a)\n// after\nCREATE SINK ... PRIMARY KEY (a) ...;","handlingStrategy":"validation","validationCode":"fn check_pk_match(rw_pk: &[&str], sqlserver_pk: &[&str]) -> Result<(), String> {\n    if rw_pk != sqlserver_pk {\n        Err(format!(\"PK mismatch: rw={:?} sqlserver={:?}\", rw_pk, sqlserver_pk))\n    } else { Ok(()) }\n}","typeGuard":null,"tryCatchPattern":"match sink.validate().await {\n    Err(e) if e.to_string().contains(\"primary key does not match between RisingWave sink\") => {\n        // message lists both sides; rebuild sink PK from the SQL Server side\n        rebuild_sink_pk_from_error(&e)\n    }\n    other => other,\n}","preventionTips":["Compare full PK column lists (name + order) on both sides before sink creation","Automate sink validation in CI against live table metadata","Treat any downstream PK change as requiring sink re-validation"],"tags":["sqlserver","sink","primary-key","schema-mismatch"],"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"}