{"record":{"id":"91f9d904192a7d7f","repo":"risingwavelabs/risingwave","slug":"if-you-want-to-use-upsert-please-set-the-keystype","errorCode":null,"errorMessage":"If you want to use upsert, please set the keysType of doris to UNIQUE_KEYS","messagePattern":"If you want to use upsert, please set the keysType of doris to UNIQUE_KEYS","errorType":"validation","errorClass":"SinkError","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/doris.rs","lineNumber":283,"sourceCode":"            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 {\n    pub config: DorisConfig,\n    #[expect(dead_code)]\n    schema: Schema,\n    #[expect(dead_code)]\n    pk_indices: Vec<usize>,\n    inserter_inner_builder: InserterInnerBuilder,\n    is_append_only: bool,\n    client: Option<DorisClient>,\n    row_encoder: JsonEncoder,","sourceCodeStart":265,"sourceCodeEnd":301,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/doris.rs#L265-L301","documentation":"When the Doris sink runs in upsert mode, the target Doris table must use the UNIQUE_KEYS keysType so Doris can merge rows by key. `validate()` queries the Doris FE schema and fails if `keysType` is anything else (e.g. DUPLICATE_KEYS or AGGREGATE_KEYS).","triggerScenarios":"Creating a Doris upsert sink against a Doris table whose `keysType` property is not `UNIQUE_KEYS`, detected during sink validation via get_schema_from_doris.","commonSituations":"Pointing the sink at a pre-existing Doris table created with default DUPLICATE_KEYS; table created for an earlier append-only sink then reused for upsert.","solutions":["Recreate the Doris table with `DUPLICATE KEY` replaced by `UNIQUE KEY(...)` in the DDL.","Verify with `SHOW CREATE TABLE` that keysType is UNIQUE_KEYS before creating the sink.","Alternatively keep the table as-is and define the sink as `type='append-only'`."],"exampleFix":"// before (Doris DDL)\nCREATE TABLE t (id INT, v INT) DUPLICATE KEY(id) DISTRIBUTED BY HASH(id);\n// after\nCREATE TABLE t (id INT, v INT) UNIQUE KEY(id) DISTRIBUTED BY HASH(id);","handlingStrategy":"validation","validationCode":"// run before creating the sink\n// SHOW CREATE TABLE doris_table;  -> must contain UNIQUE KEY(...)","typeGuard":null,"tryCatchPattern":"match sink.validate().await {\n    Err(SinkError::Config(e)) if e.to_string().contains(\"UNIQUE_KEYS\") => {\n        eprintln!(\"recreate the Doris table with UNIQUE KEY\");\n    }\n    other => other?,\n}","preventionTips":["Create Doris tables with UNIQUE KEY when they will receive upserts.","Run SHOW CREATE TABLE to verify keysType before wiring a sink.","Keep one table per write mode; don't reuse append-only tables for upsert."],"tags":["rust","sink","doris","schema","upsert"],"backgroundTag":"incompatible-source-type","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"}