{"record":{"id":"d4b80cd2eaed3e36","repo":"risingwavelabs/risingwave","slug":"primary-key-not-defined-for-upsert-sql-server-sink","errorCode":null,"errorMessage":"Primary key not defined for upsert SQL Server sink (please define in `primary_key` field)","messagePattern":"Primary key not defined for upsert SQL Server sink \\(please define in `primary_key` field\\)","errorType":"validation","errorClass":"SinkError::Config","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/sqlserver.rs","lineNumber":187,"sourceCode":"        let config = SqlServerConfig::from_btreemap(param.properties)?;\n        SqlServerSink::new(config, schema, pk_indices, param.sink_type.is_append_only())\n    }\n}\n\nimpl Sink for SqlServerSink {\n    type LogSinker = LogSinkerOf<SqlServerSinkWriter>;\n\n    const SINK_NAME: &'static str = SQLSERVER_SINK;\n\n    crate::impl_validate_sink_unknown_fields!();\n\n    async fn validate(&self) -> Result<()> {\n        risingwave_common::license::Feature::SqlServerSink\n            .check_available()\n            .map_err(|e| anyhow::anyhow!(e))?;\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 SQL Server sink (please define in `primary_key` field)\"\n            )));\n        }\n\n        for f in self.schema.fields() {\n            check_data_type_compatibility(&f.data_type)?;\n        }\n\n        let mut sql_client = SqlServerClient::new(&self.config).await?;\n        validate_sql_server_write_permission(&mut sql_client, &self.config, self.is_append_only)\n            .await?;\n        let sql_server_table_metadata =\n            query_sql_server_table_metadata(&mut sql_client, &self.config).await?;\n        let sql_server_pk_count = sql_server_table_metadata\n            .iter()\n            .filter(|metadata| metadata.is_pk)\n            .count();\n        let sql_server_table_metadata = sql_server_table_metadata","sourceCodeStart":169,"sourceCodeEnd":205,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/sqlserver.rs#L169-L205","documentation":"An upsert SQL Server sink needs primary key indices to know which columns to match on for UPDATE/MERGE semantics. If the sink is not append-only and pk_indices is empty, validate() rejects the sink definition.","triggerScenarios":"Creating a SQL Server sink with `type = 'upsert'` (or non-append-only) but without a `primary_key` field defined in the sink definition.","commonSituations":"Upsert sink over a source without a primary key and forgetting to specify `primary_key` in the CREATE SINK statement; append-only data being sunk as upsert.","solutions":["Add `PRIMARY KEY (...)` (or the `primary_key` field) to the CREATE SINK statement naming the key columns","Change the sink type to 'append_only' if the data stream is truly append-only and no upsert semantics are needed","Ensure the upstream materialized view/table has a primary key so it is inherited by the sink"],"exampleFix":"// before\nCREATE SINK s FROM mv INTO sqlserver ... WITH ( 'type' = 'upsert' );\n// after\nCREATE SINK s FROM mv INTO sqlserver ... WITH ( 'type' = 'upsert' ) AS SELECT ... PRIMARY KEY (id);","handlingStrategy":"validation","validationCode":"fn check_upsert_pk(sink_type: &str, pk_columns: &[&str]) -> Result<(), String> {\n    if sink_type == \"upsert\" && pk_columns.is_empty() {\n        Err(\"upsert SQL Server sink requires a primary_key\".into())\n    } else { Ok(()) }\n}","typeGuard":null,"tryCatchPattern":"match create_sink(cfg).await {\n    Err(e) if e.to_string().contains(\"Primary key not defined for upsert SQL Server sink\") => return Err(\"add PRIMARY KEY to sink definition\".into()),\n    other => other,\n}","preventionTips":["Always define PRIMARY KEY when sink type is upsert","Ensure upstream MVs/tables have PKs so sinks inherit them","Default to append_only when no key semantics are needed"],"tags":["sqlserver","sink","primary-key","upsert","config"],"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"}