{"record":{"id":"0950ef83f5e7cdc8","repo":"risingwavelabs/risingwave","slug":"primary-key-must-not-be-empty-in","errorCode":null,"errorMessage":"`primary-key` must not be empty in {}","messagePattern":"`primary-key` must not be empty in (.+?)","errorType":"validation","errorClass":"SinkError::Config","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/iceberg/config.rs","lineNumber":592,"sourceCode":"                .map_err(|e| SinkError::Config(anyhow!(e)))?;\n\n        if config.enable_compaction && !values.contains_key(COMPACTION_MAX_SNAPSHOTS_NUM) {\n            config.max_snapshots_num_before_compaction = Some(DEFAULT_COMPACTION_MAX_SNAPSHOTS_NUM);\n        }\n\n        if config.r#type != SINK_TYPE_APPEND_ONLY && config.r#type != SINK_TYPE_UPSERT {\n            return Err(SinkError::Config(anyhow!(\n                \"`{}` must be {}, or {}\",\n                SINK_TYPE_OPTION,\n                SINK_TYPE_APPEND_ONLY,\n                SINK_TYPE_UPSERT\n            )));\n        }\n\n        if config.r#type == SINK_TYPE_UPSERT {\n            if let Some(primary_key) = &config.primary_key {\n                if primary_key.is_empty() {\n                    return Err(SinkError::Config(anyhow!(\n                        \"`primary-key` must not be empty in {}\",\n                        SINK_TYPE_UPSERT\n                    )));\n                }\n            } else if !config.enable_pk_index {\n                // When `enable_pk_index = true`, the planner auto-derives the iceberg pk\n                // from the upstream stream key, so the user does not need to spell it out\n                // in WITH options. The derived pk is written back into properties before\n                // this validation is consulted again at sink-construction time.\n                return Err(SinkError::Config(anyhow!(\n                    \"Must set `primary-key` in {}\",\n                    SINK_TYPE_UPSERT\n                )));\n            }\n        }\n\n        // Enforce merge-on-read for append-only sinks\n        Self::validate_append_only_write_mode(&config.r#type, config.write_mode)?;","sourceCodeStart":574,"sourceCodeEnd":610,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/iceberg/config.rs#L574-L610","documentation":"This error is raised by IcebergSinkConfig::from_btreemap when a sink is created with type = 'upsert' and the `primary_key` option is present but set to an empty string. RisingWave requires an explicit non-empty primary key for upsert sinks so it can deduplicate and merge rows in the Iceberg table. The check rejects empty strings before the sink is constructed.","triggerScenarios":"Creating an Iceberg sink with `connector='iceberg'`, `type='upsert'`, and `primary_key=''` (empty string) in the WITH options, e.g. via CREATE SINK. The value parses as Some(\"\") which fails `primary_key.is_empty()`.","commonSituations":"Templates or generated SQL that interpolate a pk column list variable that is empty; users copying upsert sink DDL and deleting the key list; tooling that writes `primary_key` unconditionally even when no pk columns were selected.","solutions":["Provide a non-empty `primary_key` in the WITH options listing the upsert key columns, e.g. `primary_key='id'` or `primary_key='id,name'`.","If no meaningful pk exists, switch the sink `type` from `upsert` to `append-only`.","If the stream key should be used automatically, set `enable_pk_index=true` (or omit primary_key with pk index enabled) so the planner derives the pk."],"exampleFix":"// before\nWITH (\n  connector = 'iceberg',\n  type = 'upsert',\n  primary_key = ''\n)\n// after\nWITH (\n  connector = 'iceberg',\n  type = 'upsert',\n  primary_key = 'user_id'\n)","handlingStrategy":"validation","validationCode":"const pk = opts['primary_key'];\nif (opts.type === 'upsert' && pk !== undefined && pk.trim() === '') {\n  throw new Error(\"`primary-key` must not be empty in upsert\");\n}","typeGuard":"function hasNonEmptyPk(opts) {\n  return typeof opts.primary_key === 'string' && opts.primary_key.length > 0;\n}","tryCatchPattern":"try {\n  await createIcebergSink(opts);\n} catch (e) {\n  if (String(e).includes('`primary-key` must not be empty')) {\n    opts.primary_key = inferPrimaryKeyColumns(opts);\n    return createIcebergSink(opts);\n  }\n  throw e;\n}","preventionTips":["Only emit the primary_key option when a non-empty column list is available.","Default to append-only sink type when no key exists.","Validate WITH options in your SQL-generation tooling before submitting DDL."],"tags":["iceberg","sink","config-validation","upsert"],"backgroundTag":"empty-required-field","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}