{"record":{"id":"d9a7f3125506c893","repo":"risingwavelabs/risingwave","slug":"there-should-be-no-exchangenode-on-the-top-of-the","errorCode":null,"errorMessage":"there should be no ExchangeNode on the top of the plan node: {:#?}","messagePattern":"there should be no ExchangeNode on the top of the plan node: (.+?)","errorType":"exception","errorClass":"MetaError","httpStatus":null,"severity":"error","filePath":"src/meta/src/stream/stream_graph/actor.rs","lineNumber":54,"sourceCode":"\nimpl FragmentActorBuilder {\n    /// Rewrite the actor body.\n    ///\n    /// During this process, the following things will be done:\n    /// 1. Replace the logical `Exchange` in node's input with `Merge`, which can be executed on the\n    ///    compute nodes.\n    fn rewrite(&self) -> MetaResult<StreamNode> {\n        self.rewrite_inner(&self.node, 0)\n    }\n\n    fn rewrite_inner(&self, stream_node: &StreamNode, depth: usize) -> MetaResult<StreamNode> {\n        match stream_node.get_node_body()? {\n            // Leaf node `Exchange`.\n            NodeBody::Exchange(exchange) => {\n                // The exchange node should always be the bottom of the plan node. If we find one\n                // when the depth is 0, it means that the plan node is not well-formed.\n                if depth == 0 {\n                    bail!(\n                        \"there should be no ExchangeNode on the top of the plan node: {:#?}\",\n                        stream_node\n                    )\n                }\n                assert!(!stream_node.get_fields().is_empty());\n                assert!(stream_node.input.is_empty());\n\n                // Index the upstreams by the an internal edge ID.\n                let (upstream_fragment_id, _) = &self.upstreams[&EdgeId::Internal {\n                    link_id: stream_node.get_operator_id().as_raw_id(),\n                }];\n\n                let upstream_fragment_id = upstream_fragment_id.as_global_id();\n\n                Ok(StreamNode {\n                    node_body: Some(NodeBody::Merge(Box::new({\n                        MergeNode {\n                            upstream_fragment_id,","sourceCodeStart":36,"sourceCodeEnd":72,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/stream/stream_graph/actor.rs#L36-L72","documentation":"In actor rewriting (rewrite_inner), an Exchange node may only appear at the bottom of a plan tree (depth 0, where it becomes the merge of an upstream fragment). Finding one at depth > 0 means the streamed plan is malformed — exchanges must have been rewritten/hoisted before this stage.","triggerScenarios":"Building the actor graph from a stream plan whose tree contains an Exchange node above other operators, i.e. planner/frontend emitted an exchange at a non-leaf position.","commonSituations":"Frontend/planner bug after a new plan node type or rewrite rule; custom internal development; version skew between frontend and meta binary.","solutions":["Inspect the dumped plan in the error to find where the Exchange sits","Check for frontend/meta version mismatch and align binaries","Reproduce with the offending query and file/upgrade a planner bug","Restart and rebuild the MV/job to regenerate the plan"],"exampleFix":null,"handlingStrategy":"validation","validationCode":"// before graph build, assert plan well-formedness: exchanges only at depth 0\nfn exchange_only_at_root(node: &StreamNode, depth: usize) -> bool {\n    let is_exchange = matches!(node.get_node_body(), Ok(NodeBody::Exchange(_)));\n    !(is_exchange && depth > 0)\n        && node.get_input().iter().all(|c| exchange_only_at_root(c, depth + 1))\n}","typeGuard":null,"tryCatchPattern":"match rewrite(stream_node) {\n    Err(e) if e.to_string().contains(\"ExchangeNode on the top\") => {\n        // plan malformed: capture the dumped plan for a planner bug report\n    }\n    r => r?,\n}","preventionTips":["Keep frontend and meta binaries version-aligned","Run planner tests when adding exchange-affecting rewrite rules","Include the dumped plan node when reporting this error"],"tags":["rust","meta","stream-graph","planner"],"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"}