{"record":{"id":"f73359a9da801c24","repo":"risingwavelabs/risingwave","slug":"if-you-want-to-use-upsert-please-set-the-keystype-f73359","errorCode":null,"errorMessage":"If you want to use upsert, please set the keysType of starrocks to PRIMARY_KEY","messagePattern":"If you want to use upsert, please set the keysType of starrocks to PRIMARY_KEY","errorType":"validation","errorClass":"SinkError::Config","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/starrocks.rs","lineNumber":393,"sourceCode":"        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\"\n            )));\n        }\n\n        for (index, filed) in self.schema.fields().iter().enumerate() {\n            if self.pk_indices.contains(&index) && !pks.contains(&filed.name) {\n                return Err(SinkError::Starrocks(format!(\n                    \"Can't find pk {:?} in starrocks\",\n                    filed.name\n                )));\n            }\n        }\n\n        let starrocks_columns_desc = client.get_columns_from_starrocks().await?;\n\n        self.check_column_name_and_type(starrocks_columns_desc)?;\n        Ok(())\n    }","sourceCodeStart":375,"sourceCodeEnd":411,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/starrocks.rs#L375-L411","documentation":"Thrown during sink validation when the sink is upsert (non-append-only) but the target StarRocks table's `table_model` (from `information_schema.tables_config`) is not `PRIMARY_KEYS`. Upsert semantics require the StarRocks table to use the PRIMARY_KEY keys type so incoming rows replace existing rows.","triggerScenarios":"Sink validated against an existing StarRocks table whose keysType is DUPLICATE_KEYS, AGG_KEYS, or UNIQUE_KEYS while the RW sink is created with a primary key / non-append-only data.","commonSituations":"Pointing a RisingWave upsert sink at a pre-existing StarRocks detail or aggregate table; the StarRocks table was created without `PRIMARY KEY(...)` / `PRIMARY KEYS(...)` in its DDL.","solutions":["Alter/recreate the StarRocks table with `PRIMARY KEY (...) DISTRIBUTED BY ...` so its model is PRIMARY_KEYS","Change the RisingWave sink to append-only if you do not need upsert semantics","Point the sink at a different StarRocks table that uses PRIMARY_KEYS"],"exampleFix":"// before (StarRocks DDL)\nCREATE TABLE t (id INT, v INT) DUPLICATE KEY(id) DISTRIBUTED BY HASH(id);\n// after\nCREATE TABLE t (id INT, v INT) PRIMARY KEY(id) DISTRIBUTED BY HASH(id);","handlingStrategy":"validation","validationCode":"// Before creating the sink, check the target model:\nSELECT table_model FROM information_schema.tables_config\nWHERE table_name = 't' AND table_schema = 'db';\n-- must return PRIMARY_KEYS for upsert sinks","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Create StarRocks tables with PRIMARY KEY(...) when planning upsert sinks","Re-check the table model after any StarRocks-side schema changes"],"tags":["starrocks","schema","upsert","validation"],"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-23T08:17:48.524Z"}