{"record":{"id":"59db3c8556adc580","repo":"risingwavelabs/risingwave","slug":"primary-key-not-defined-for-upsert-starrocks-sink","errorCode":null,"errorMessage":"Primary key not defined for upsert starrocks sink (please define in `primary_key` field)","messagePattern":"Primary key not defined for upsert starrocks sink \\(please define in `primary_key` field\\)","errorType":"validation","errorClass":"SinkError::Config","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/starrocks.rs","lineNumber":376,"sourceCode":"                \"MAP is not supported for Starrocks sink.\".to_owned(),\n            )),\n            DataType::Vector(_) => Err(SinkError::Starrocks(\n                \"VECTOR is not supported for Starrocks sink.\".to_owned(),\n            )),\n        }\n    }\n}\n\nimpl Sink for StarrocksSink {\n    type LogSinker = DecoupleCheckpointLogSinkerOf<StarrocksSinkWriter>;\n\n    const SINK_NAME: &'static str = STARROCKS_SINK;\n\n    crate::impl_validate_sink_unknown_fields!();\n\n    async fn validate(&self) -> Result<()> {\n        if !self.is_append_only && self.pk_indices.is_empty() {\n            return Err(SinkError::Config(anyhow!(\n                \"Primary key not defined for upsert starrocks sink (please define in `primary_key` field)\"\n            )));\n        }\n        // check reachability\n        let mut client = StarrocksSchemaClient::new(\n            self.config.common.host.clone(),\n            self.config.common.mysql_port.clone(),\n            self.config.common.table.clone(),\n            self.config.common.database.clone(),\n            self.config.common.user.clone(),\n            self.config.common.password.clone(),\n        )\n        .await?;\n        let (read_model, pks) = client.get_pk_from_starrocks().await?;\n\n        if !self.is_append_only && read_model.ne(\"PRIMARY_KEYS\") {\n            return Err(SinkError::Config(anyhow!(\n                \"If you want to use upsert, please set the keysType of starrocks to PRIMARY_KEY\"","sourceCodeStart":358,"sourceCodeEnd":394,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/starrocks.rs#L358-L394","documentation":"Thrown from the sink `validate` hook when a StarRocks sink is declared as non-append-only (upsert mode) but `pk_indices` is empty, meaning no primary key columns were defined. Upsert requires a primary key to know which rows to overwrite in StarRocks.","triggerScenarios":"`CREATE SINK ... INTO starrocks` where the upstream materialized view is not append-only (contains updates/deletes) and no `primary_key` field is given in the WITH options or the sink definition.","commonSituations":"Sinking an MV with aggregations or updates into StarRocks upsert-style but forgetting the `primary_key` option; users assuming the primary key is inferred from the source table.","solutions":["Add `'primary_key' = 'col1,col2'` to the sink WITH options","Declare the sink with an explicit `primary_key` clause matching the target StarRocks table's key columns","If data is truly append-only, create the sink in append-only mode so the PK requirement is skipped"],"exampleFix":"// before\nCREATE SINK s FROM mv INTO starrocks WITH (\n  'connector' = 'starrocks'\n);\n// after\nCREATE SINK s FROM mv INTO starrocks WITH (\n  'connector' = 'starrocks',\n  'primary_key' = 'order_id'\n);","handlingStrategy":"validation","validationCode":"// Ensure PK is declared for non-append-only sources\nif !source_is_append_only && !opts.contains_key(\"primary_key\") {\n    throw \"upsert StarRocks sink requires 'primary_key' option\";\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Always specify primary_key when sinking MVs with updates/aggregations","Check `rw_columns`/pk of the source MV and mirror it in the sink config"],"tags":["config","starrocks","primary-key","upsert"],"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-23T08:17:48.524Z"}