{"record":{"id":"e97c9b84357dcd13","repo":"risingwavelabs/risingwave","slug":"variant-column-cannot-be-used-as-the-primary","errorCode":null,"errorMessage":"VARIANT column `{}` cannot be used as the primary key of an upsert iceberg sink","messagePattern":"VARIANT column `(.+?)` cannot be used as the primary key of an upsert iceberg sink","errorType":"validation","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/iceberg/mod.rs","lineNumber":260,"sourceCode":"            self.config.write_mode,\n        )?;\n        validate_explicit_compaction_type(&self.config)?;\n        validate_compaction_option_compatibility(&self.config)?;\n\n        // VARIANT is not comparable, so it can never be an equality-delete key.\n        if self.config.r#type == SINK_TYPE_UPSERT\n            && !self.config.force_append_only\n            && let Some(pk_indices) = self\n                .param\n                .downstream_pk\n                .as_ref()\n                .filter(|pk| !pk.is_empty())\n        {\n            for &idx in pk_indices {\n                if let Some(column) = self.param.columns.get(idx)\n                    && column.data_type.contains_variant()\n                {\n                    bail!(\n                        \"VARIANT column `{}` cannot be used as the primary key of an upsert iceberg sink\",\n                        column.name\n                    );\n                }\n            }\n        }\n\n        let table = self.create_and_validate_table().await?;\n        self.config\n            .validate_manifest_rewrite_format(table.metadata().format_version())?;\n        Ok(())\n    }\n\n    fn support_schema_change() -> bool {\n        true\n    }\n\n    fn validate_alter_config_change(","sourceCodeStart":242,"sourceCodeEnd":278,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/iceberg/mod.rs#L242-L278","documentation":"For upsert (primary-key) Iceberg sinks, the primary key columns must be stored as Iceberg identity fields, but Iceberg does not support VARIANT (semi-structured) columns as identity/primary-key fields. Validation therefore rejects any sink whose PK includes a column containing a VARIANT type.","triggerScenarios":"Creating an upsert Iceberg sink whose primary key includes a `jsonb`/VARIANT column, including via nested `contains_variant()` types (e.g. struct fields containing variant).","commonSituations":"Streaming CDC tables where the natural key was modeled as jsonb; users materializing a deduplicated table keyed on a variant payload instead of concrete key columns.","solutions":["Choose a primary key made of non-variant columns (integers, strings, timestamps, etc.).","Extract the key into a concrete typed column in the MV and use it as the sink PK.","Change the sink to an append-only sink (no primary key) if upsert semantics are not required."],"exampleFix":"-- before\nCREATE SINK s FROM mv WITH (connector='iceberg', primary_key='payload'); -- payload is jsonb\n-- after\nCREATE SINK s FROM mv WITH (connector='iceberg', primary_key='id'); -- id is BIGINT","handlingStrategy":"validation","validationCode":"-- ensure no PK column is jsonb/variant\nSELECT c.name\nFROM rw_columns c\nWHERE c.relation_id = <sink_upstream_mv_id>\n  AND c.name = ANY(ARRAY[<pk_columns>])\n  AND c.data_type IN ('jsonb', 'variant');","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Never use jsonb/variant columns as primary keys in upsert sinks.","Model dedup keys as concrete typed columns in the upstream MV.","Prefer append-only sinks when no true key exists."],"tags":["iceberg","sink","variant","primary-key","upsert"],"backgroundTag":"invalid-argument-value","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"}