{"record":{"id":"5143af8f7cfdfd32","repo":"risingwavelabs/risingwave","slug":"iceberg-pk-index-writer-input-must-be-a-merge-afte","errorCode":null,"errorMessage":"iceberg pk-index writer input must be a merge after actor rewrite","messagePattern":"iceberg pk-index writer input must be a merge after actor rewrite","errorType":"exception","errorClass":"MetaError","httpStatus":null,"severity":"error","filePath":"src/meta/src/stream/stream_graph/actor.rs","lineNumber":206,"sourceCode":"                ];\n                Ok(StreamNode {\n                    input,\n                    ..stream_node.clone()\n                })\n            }\n\n            NodeBody::IcebergWithPkIndexWriter(_) => {\n                let mut new_stream_node = stream_node.clone();\n                // Compaction alternates between the normal and resolver inputs, so either merge\n                // must be able to remain alive while temporarily disconnected.\n                for (input, new_input) in stream_node\n                    .input\n                    .iter()\n                    .zip_eq_fast(&mut new_stream_node.input)\n                {\n                    *new_input = self.rewrite_inner(input, depth + 1)?;\n                    let Some(NodeBody::Merge(merge)) = new_input.node_body.as_mut() else {\n                        bail!(\"iceberg pk-index writer input must be a merge after actor rewrite\");\n                    };\n                    merge.allow_empty_upstream = true;\n                }\n                Ok(new_stream_node)\n            }\n\n            // For other nodes, visit the children recursively.\n            _ => {\n                let mut new_stream_node = stream_node.clone();\n                for (input, new_input) in stream_node\n                    .input\n                    .iter()\n                    .zip_eq_fast(&mut new_stream_node.input)\n                {\n                    *new_input = self.rewrite_inner(input, depth + 1)?;\n                }\n                Ok(new_stream_node)\n            }","sourceCodeStart":188,"sourceCodeEnd":224,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/stream/stream_graph/actor.rs#L188-L224","documentation":"During actor rewriting of the stream graph, an Iceberg primary-key index writer node's inputs must be rewritten into Merge nodes so `allow_empty_upstream` can be set. If a rewritten input is not a `NodeBody::Merge`, the invariant that pk-index writers always consume merge inputs after rewriting is broken, and the meta service bails out. This is an internal consistency check in the frontend/meta actor-rewrite pass.","triggerScenarios":"Calling the stream graph actor rewrite (`rewrite`/`rewrite_inner` in src/meta/src/stream/stream_graph/actor.rs) on a plan containing an Iceberg pk-index writer whose upstream node does not rewrite to a Merge node body.","commonSituations":"Developing a new stream node type or changing node rewriting logic so a pk-index writer is wired to a non-merge input; using an internal/dev build where fragment planning produced an unexpected topology around Iceberg sinks with primary keys; version mismatch between frontend plan generation and meta rewrite expectations.","solutions":["Inspect the fragment/plan topology feeding the Iceberg pk-index writer and ensure its input is a Merge node before actor rewrite.","Check recent changes to `rewrite_inner` or NodeBody handling in src/meta/src/stream/stream_graph/actor.rs and fix the rewrite so the input becomes a Merge.","Verify the frontend does not build pk-index writers for Iceberg sinks with unusual upstreams; adjust plan generation instead of relaxing this check.","If reproducible, capture the plan (explain/graph dump) and file an issue — this is an internal invariant violation, not a user-facing config error."],"exampleFix":"// before\nlet Some(NodeBody::Merge(merge)) = new_input.node_body.as_mut() else {\n    bail!(\"iceberg pk-index writer input must be a merge after actor rewrite\");\n};\n// after\nmatch new_input.node_body.as_mut() {\n    Some(NodeBody::Merge(merge)) => merge.allow_empty_upstream = true,\n    other => return Err(anyhow::anyhow!(\n        \"iceberg pk-index writer input must be a merge after actor rewrite, got {:?}\",\n        other.map(|b| b.kind())\n    )),\n}","handlingStrategy":"try-catch","validationCode":"// Rust, before calling rewrite on the graph\nfn input_is_merge_after_rewrite(node: &StreamNode) -> bool {\n    node.input.iter().all(|i| matches!(i.node_body, Some(NodeBody::Merge(_))))\n}","typeGuard":"fn as_merge(node: &mut StreamNode) -> Option<&mut Merge> {\n    match node.node_body.as_mut() {\n        Some(NodeBody::Merge(m)) => Some(m),\n        _ => None,\n    }\n}","tryCatchPattern":"match actor::rewrite(graph) {\n    Ok(rewritten) => deploy(rewritten),\n    Err(e) if e.to_string().contains(\"must be a merge\") => {\n        tracing::error!(\"plan topology bug around iceberg pk-index writer: {e:#}\");\n        // surface as internal error, do not retry blindly\n    }\n    Err(e) => return Err(e),\n}","preventionTips":["Add unit tests covering iceberg pk-index writer rewrites after any change to `rewrite_inner`.","Keep pk-index writers wired only to Merge-typed upstream nodes in plan generation.","Log the offending node body kind in the bail message for faster triage."],"tags":["internal-invariant","stream-graph","iceberg","meta"],"backgroundTag":"internal-invariant-violation","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"}