{"record":{"id":"b4fb99cd3979044f","repo":"risingwavelabs/risingwave","slug":"primary-key-not-defined-for-upsert-doris-sink-ple","errorCode":null,"errorMessage":"Primary key not defined for upsert doris sink (please define in `primary_key` field)","messagePattern":"Primary key not defined for upsert doris sink \\(please define in `primary_key` field\\)","errorType":"validation","errorClass":"SinkError","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/doris.rs","lineNumber":274,"sourceCode":"\n    const SINK_NAME: &'static str = DORIS_SINK;\n\n    crate::impl_validate_sink_unknown_fields!();\n\n    async fn new_log_sinker(&self, writer_param: SinkWriterParam) -> Result<Self::LogSinker> {\n        Ok(DorisSinkWriter::new(\n            self.config.clone(),\n            self.schema.clone(),\n            self.pk_indices.clone(),\n            self.is_append_only,\n        )\n        .await?\n        .into_log_sinker(SinkWriterMetrics::new(&writer_param)))\n    }\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 doris sink (please define in `primary_key` field)\"\n            )));\n        }\n        // check reachability\n        let client = self.config.common.build_get_client();\n        let doris_schema = client.get_schema_from_doris().await?;\n\n        if !self.is_append_only && doris_schema.keys_type.ne(\"UNIQUE_KEYS\") {\n            return Err(SinkError::Config(anyhow!(\n                \"If you want to use upsert, please set the keysType of doris to UNIQUE_KEYS\"\n            )));\n        }\n        self.check_column_name_and_type(doris_schema.properties)?;\n        Ok(())\n    }\n}\n\npub struct DorisSinkWriter {","sourceCodeStart":256,"sourceCodeEnd":292,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/doris.rs#L256-L292","documentation":"The Doris sink in upsert mode requires a primary key so updates/deletes can be applied row-by-row on the Doris side. `validate()` raises this error when the sink is not append-only and `pk_indices` is empty, meaning no `primary_key` was defined.","triggerScenarios":"Creating a Doris sink with `type='upsert'` (or on a non-append-only stream) while omitting the `primary_key` option in the WITH clause or the underlying source having no PK.","commonSituations":"Upserting from a source table without a primary key; forgetting `primary_key` in the sink definition; relying on implicit keys that RisingWave does not propagate to the sink.","solutions":["Add a `primary_key` option to the sink WITH clause listing the key columns.","If the data is truly append-only, switch the sink `type` to `append-only`.","Ensure the upstream materialized view defines a primary key so pk_indices is non-empty."],"exampleFix":"// before\nCREATE SINK s FROM mv WITH ('connector'='doris', 'type'='upsert');\n// after\nCREATE SINK s FROM mv WITH ('connector'='doris', 'type'='upsert', 'primary_key'='id');","handlingStrategy":"validation","validationCode":"if sink_type == \"upsert\" && with_options.get(\"primary_key\").is_none() {\n    return Err(\"upsert doris sink requires a `primary_key` option\");\n}","typeGuard":null,"tryCatchPattern":"if let Err(SinkError::Config(e)) = sink.validate().await {\n    if e.to_string().contains(\"Primary key not defined\") {\n        eprintln!(\"add `primary_key` to the sink WITH clause\");\n    }\n}","preventionTips":["Always define `primary_key` when sink type is upsert.","Ensure upstream MVs/sources declare primary keys.","Use append-only type when no key semantics are needed."],"tags":["rust","sink","doris","config","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"}