{"record":{"id":"9e763a8c8a229222","repo":"risingwavelabs/risingwave","slug":"primary-key-not-defined-for-upsert-clickhouse-sink","errorCode":null,"errorMessage":"Primary key not defined for upsert clickhouse sink (please define in `primary_key` field)","messagePattern":"Primary key not defined for upsert clickhouse sink \\(please define in `primary_key` field\\)","errorType":"validation","errorClass":"SinkError::Config","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/clickhouse.rs","lineNumber":568,"sourceCode":"                ck_column.name, fields_type, ck_column.r#type\n            )));\n        }\n\n        Ok(())\n    }\n}\n\nimpl Sink for ClickHouseSink {\n    type LogSinker = DecoupleCheckpointLogSinkerOf<ClickHouseSinkWriter>;\n\n    const SINK_NAME: &'static str = CLICKHOUSE_SINK;\n\n    crate::impl_validate_sink_unknown_fields!();\n\n    async fn validate(&self) -> Result<()> {\n        // For upsert clickhouse sink, the primary key must be defined.\n        if !self.is_append_only && self.pk_indices.is_empty() {\n            return Err(SinkError::Config(anyhow!(\n                \"Primary key not defined for upsert clickhouse sink (please define in `primary_key` field)\"\n            )));\n        }\n\n        // check reachability\n        let client = self.config.common.build_client()?;\n\n        let (clickhouse_column, clickhouse_engine) =\n            query_column_engine_from_ck(client, &self.config).await?;\n        if clickhouse_engine.is_shared_tree() {\n            risingwave_common::license::Feature::ClickHouseSharedEngine\n                .check_available()\n                .map_err(|e| anyhow::anyhow!(e))?;\n        }\n\n        if !self.is_append_only\n            && !clickhouse_engine.is_collapsing_engine()\n            && !clickhouse_engine.is_delete_replacing_engine()","sourceCodeStart":550,"sourceCodeEnd":586,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/clickhouse.rs#L550-L586","documentation":"The ClickHouse sink's validate() requires an upsert sink to have a primary key. If the sink is not append-only and no primary key columns were derived (pk_indices empty), validation fails, telling users to define it via the `primary_key` field.","triggerScenarios":"CREATE SINK ... WITH (connector='clickhouse', type='upsert', ...) where neither the WITH options nor the underlying table supply primary key columns, leaving pk_indices empty.","commonSituations":"Sinking a table without a primary key in upsert mode; forgetting the `primary_key` property; expecting ClickHouse's own ORDER BY to substitute for a RisingWave primary key.","solutions":["Add primary_key='col1,col2' to the WITH options listing the key columns.","Switch the sink to type='append-only' if upsert semantics are not needed.","Create the sink from a materialized view that includes a primary key."],"exampleFix":"// before\nWITH (connector='clickhouse', type='upsert', ...)\n// after\nWITH (connector='clickhouse', type='upsert', primary_key='user_id', ...)","handlingStrategy":"validation","validationCode":"if (props.type === 'upsert' && !props.primary_key) throw new Error(\"ClickHouse upsert sink requires primary_key in WITH options\");","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Always pair type='upsert' with an explicit primary_key option.","Ensure the source relation has a primary key when sinking in upsert mode.","Use append-only sinks when no key semantics are needed."],"tags":["clickhouse","sink","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-14T16:17:12.679Z"}