{"record":{"id":"5594cdb8db09ce33","repo":"risingwavelabs/risingwave","slug":"missing-downstream-pk-in-iceberg-sink-desc","errorCode":null,"errorMessage":"missing downstream pk in iceberg sink desc","messagePattern":"missing downstream pk in iceberg sink desc","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/stream/src/from_proto/iceberg_with_pk_index/writer.rs","lineNumber":67,"sourceCode":"            .map_err(|_| anyhow!(\"IcebergWithPkIndexWriterExecutor requires exactly two inputs\"))?;\n        let sink_desc = node.sink_desc.as_ref().unwrap();\n        let sink_id: SinkId = sink_desc.get_id();\n        let sink_name = sink_desc.get_name().to_owned();\n\n        let properties_with_secret = LocalSecretManager::global().fill_secrets(\n            sink_desc.get_properties().clone(),\n            sink_desc.get_secret_refs().clone(),\n        )?;\n        let config = IcebergConfig::from_btreemap(properties_with_secret.clone())\n            .map_err(|err| StreamExecutorError::from((err, sink_id)))?;\n\n        let pk_indices = sink_desc\n            .downstream_pk\n            .iter()\n            .map(|&idx| idx as usize)\n            .collect::<Vec<_>>();\n        if pk_indices.is_empty() {\n            bail!(\"missing downstream pk in iceberg sink desc\");\n        }\n\n        let (sink_param, _) = build_sink_param(sink_desc, properties_with_secret, ICEBERG_SINK)?;\n\n        let table = create_and_validate_table_impl(&config, &sink_param)\n            .await\n            .map_err(|e| StreamExecutorError::sink_error(e, sink_id))?;\n\n        let pk_index_state_table = StateTableBuilder::new(\n            node.get_pk_index_table()?,\n            store,\n            params.vnode_bitmap.clone().map(Arc::new),\n        )\n        .enable_preload_all_rows_by_config(&params.config)\n        .with_op_consistency_level(StateTableOpConsistencyLevel::Inconsistent)\n        .build()\n        .await;\n","sourceCodeStart":49,"sourceCodeEnd":85,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/stream/src/from_proto/iceberg_with_pk_index/writer.rs#L49-L85","documentation":"An Iceberg sink with primary key requires a non-empty `downstream_pk` in its sink description, because pk-index maintenance (compaction resolver, position-delete handling) keys off the primary key column indices. `new_boxed_executor` bails when `pk_indices` collected from `sink_desc.downstream_pk` is empty, treating the plan fragment as corrupt.","triggerScenarios":"Creating an Iceberg pk-index sink whose proto `sink_desc.downstream_pk` is empty or unset — e.g. the sink was defined without a primary key but routed to the pk-index writer node, or the frontend failed to fill downstream_pk.","commonSituations":"Creating `CREATE SINK ... ` on a table/MV without a PRIMARY KEY but hitting the pk writer path; version skew where older frontends did not populate `downstream_pk`; manual proto editing or corrupted plan fragments.","solutions":["Define the sink's upstream with an explicit PRIMARY KEY so the frontend populates `downstream_pk`.","Re-create the sink with a current frontend version that fills `downstream_pk` for pk sinks.","Verify the sink desc proto in the actor plan actually contains downstream_pk entries.","If the sink genuinely has no key, use the plain (non-pk) Iceberg sink path instead of the pk-index writer."],"exampleFix":"-- before\nCREATE SINK s FROM mv WITH (connector='iceberg', ...); -- upstream has no PK\n-- after\nCREATE MATERIALIZED VIEW mv AS SELECT id, ... FROM t; -- id PRIMARY KEY\nCREATE SINK s FROM mv WITH (connector='iceberg', primary_key='id', ...);","handlingStrategy":"validation","validationCode":"-- SQL-level guard\n-- ensure the upstream has a PRIMARY KEY before creating a pk Iceberg sink\nSELECT count(*) FROM rw_catalog.rw_columns WHERE relation = 'mv' AND is_primary_key; -- must be > 0","typeGuard":"fn has_downstream_pk(desc: &SinkDesc) -> bool { !desc.downstream_pk.is_empty() }","tryCatchPattern":"if pk_indices.is_empty() { bail!(\"missing downstream pk in iceberg sink desc\"); }","preventionTips":["Always declare PRIMARY KEY on the source table/MV feeding a pk Iceberg sink.","Include `primary_key` in the sink's WITH options consistently.","Add frontend assertion that the pk writer path is only used with non-empty downstream_pk."],"tags":["rust","streaming","iceberg","plan-validation"],"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-14T16:17:12.679Z"}