{"record":{"id":"36ba08c1b8eb5e08","repo":"risingwavelabs/risingwave","slug":"icebergwithpkindexwriterexecutor-requires-exactly","errorCode":null,"errorMessage":"IcebergWithPkIndexWriterExecutor requires exactly two inputs","messagePattern":"IcebergWithPkIndexWriterExecutor requires exactly two inputs","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/stream/src/from_proto/iceberg_with_pk_index/writer.rs","lineNumber":49,"sourceCode":"use crate::executor::{Executor, IcebergWriterImpl, StreamExecutorError, WriterExecutor};\nuse crate::from_proto::ExecutorBuilder;\nuse crate::task::ExecutorParams;\npub struct IcebergWithPkIndexWriterExecutorBuilder;\n\nimpl_stream_node_body!(IcebergWithPkIndexWriter(IcebergWithPkIndexWriterNode) => IcebergWithPkIndexWriterExecutorBuilder);\n\nimpl ExecutorBuilder for IcebergWithPkIndexWriterExecutorBuilder {\n    type Node = IcebergWithPkIndexWriterNode;\n\n    async fn new_boxed_executor(\n        params: ExecutorParams,\n        node: &Self::Node,\n        store: impl StateStore,\n    ) -> StreamResult<Executor> {\n        let [input, resolver_input] = params\n            .input\n            .try_into()\n            .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\");","sourceCodeStart":31,"sourceCodeEnd":67,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/stream/src/from_proto/iceberg_with_pk_index/writer.rs#L31-L67","documentation":"The `IcebergWithPkIndexWriterExecutor` is a two-input operator: a main input plus a dedicated resolver (compaction) input. During executor construction, `params.input.try_into()` into an exact two-element array fails when the frontend attached any other number of inputs, and the code maps that failure to this anyhow error. It protects against malformed or version-skewed plan fragments.","triggerScenarios":"`new_boxed_executor` receives a `StreamNode` for the Iceberg pk writer whose `input` list has fewer or more than two children (e.g. one input because the resolver edge was dropped, or three after a buggy rewrite).","commonSituations":"Frontend/backend version skew where the plan serializer omitted the resolver input; hand-crafted or replayed plan fragments; internal planner bugs producing the wrong arity for this sink node.","solutions":["Regenerate/re-plan the streaming job so the frontend emits both the writer input and the resolver input.","Ensure frontend and compute-node binaries are from the same version (upgrade together).","Inspect the serialized plan fragment (actor proto) to confirm the writer node has exactly two inputs.","If reproducible with matching versions, file a planner bug — the frontend should never produce wrong arity."],"exampleFix":"// before\nlet [input] = params.input.try_into().unwrap(); // wrong arity for this executor\n// after\nlet [input, resolver_input] = params.input.try_into().map_err(|_| anyhow!(\"IcebergWithPkIndexWriterExecutor requires exactly two inputs\"))?;","handlingStrategy":"validation","validationCode":"// verify plan arity before dispatch to the builder\nif node.input.len() != 2 { return Err(anyhow!(\"expected writer + resolver inputs\")); }","typeGuard":"fn two_inputs(inputs: &[StreamNode]) -> Option<(&StreamNode, &StreamNode)> {\n    match inputs { [a, b] => Some((a, b)), _ => None }\n}","tryCatchPattern":"let [input, resolver_input] = params.input.try_into().map_err(|_| anyhow!(\"IcebergWithPkIndexWriterExecutor requires exactly two inputs\"))?;","preventionTips":["Keep frontend and compute binaries on the same version.","Add planner tests asserting the pk writer node always has two inputs.","When replaying plan fragments, validate arity before executing."],"tags":["rust","streaming","plan-validation","iceberg"],"backgroundTag":"unexpected-response-shape","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"}