{"record":{"id":"f6a566ec1109e1d5","repo":"risingwavelabs/risingwave","slug":"primary-key-must-be-specified-for-upsert-iceberg-s","errorCode":null,"errorMessage":"primary key must be specified for upsert iceberg sink","messagePattern":"primary key must be specified for upsert iceberg sink","errorType":"validation","errorClass":"SinkError::Config","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/iceberg/mod.rs","lineNumber":177,"sourceCode":"\n    pub fn new(config: IcebergConfig, param: SinkParam) -> Result<Self> {\n        if let Some(order_key) = &config.order_key {\n            validate_order_key_columns(\n                order_key,\n                param.columns.iter().map(|column| column.name.as_str()),\n            )\n            .context(\"invalid order_key\")\n            .map_err(SinkError::Config)?;\n        }\n\n        let upsert_primary_key_column_names =\n            if config.r#type == SINK_TYPE_UPSERT && !config.force_append_only {\n                let pk_indices = param\n                    .downstream_pk\n                    .as_ref()\n                    .filter(|pk| !pk.is_empty())\n                    .ok_or_else(|| {\n                        SinkError::Config(anyhow!(\n                            \"primary key must be specified for upsert iceberg sink\"\n                        ))\n                    })?;\n                Some(\n                    pk_indices\n                        .iter()\n                        .map(|&idx| {\n                            param\n                                .columns\n                                .get(idx)\n                                .map(|column| column.name.clone())\n                                .ok_or_else(|| {\n                                    SinkError::Config(anyhow!(\n                                        \"primary key column index {} out of range in sink schema\",\n                                        idx\n                                    ))\n                                })\n                        })","sourceCodeStart":159,"sourceCodeEnd":195,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/iceberg/mod.rs#L159-L195","documentation":"An upsert-type iceberg sink must know the downstream primary key to perform key-based upserts. If no primary key is available (param.downstream_pk is None or empty) and force_append_only is not set, sink construction fails.","triggerScenarios":"CREATE SINK with type='upsert' (or an upsert-capable default) from a source without a primary key, without specifying a `primary_key` option and without force_append_only=true.","commonSituations":"Sink on an append-only MV without PK; forgetting the primary_key option; upgrading from append-only behavior where upsert is now the inferred type.","solutions":["Add a `primary_key` option to the sink listing key column(s)","Or define a primary key on the source materialized view","Or use type='append-only' / set force_append_only=true if upsert semantics are not needed"],"exampleFix":"// before\nCREATE SINK s FROM mv WITH (connector='iceberg', type='upsert');\n// after\nCREATE SINK s FROM mv WITH (connector='iceberg', type='upsert', primary_key='user_id');","handlingStrategy":"validation","validationCode":"if sink_type == \"upsert\" && !force_append_only && (primary_key.is_none() || primary_key.as_deref() == Some(\"\")) {\n    return Err(\"upsert iceberg sink requires primary_key\");\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Always specify primary_key for upsert sinks","Define PKs on upstream MVs feeding upsert sinks","Use type='append-only' when upsert semantics are unnecessary"],"tags":["iceberg","sink","upsert","primary-key"],"backgroundTag":"missing-required-argument","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"}