{"record":{"id":"869ab106ddd53f7d","repo":"risingwavelabs/risingwave","slug":"unexpected-exchange-detected-we-are-either-conver","errorCode":null,"errorMessage":"Unexpected exchange detected. We are either converting a single stage plan or converting the second stage of the plan.","messagePattern":"Unexpected exchange detected\\. We are either converting a single stage plan or converting the second stage of the plan\\.","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/frontend/src/scheduler/local.rs","lineNumber":276,"sourceCode":"    }\n\n    fn convert_plan_node<'a>(\n        &'a self,\n        execution_plan_node: &ExecutionPlanNode,\n        second_stages: &mut Option<HashMap<StageId, &'a QueryStage>>,\n        partition: Option<PartitionInfo>,\n        next_executor_id: Arc<AtomicU32>,\n    ) -> SchedulerResult<PbPlanNode> {\n        let identity = format!(\n            \"{:?}-{}\",\n            execution_plan_node.plan_node_type,\n            next_executor_id.fetch_add(1, Ordering::Relaxed)\n        );\n        match execution_plan_node.plan_node_type {\n            BatchPlanNodeType::BatchExchange => {\n                let exchange_source_stage_id = execution_plan_node\n                    .source_stage_id\n                    .expect(\"We expect stage id for Exchange Operator\");\n                let Some(second_stages) = second_stages.as_mut() else {\n                    bail!(\n                        \"Unexpected exchange detected. We are either converting a single stage plan or converting the second stage of the plan.\"\n                    )\n                };\n                let second_stage = second_stages.remove(&exchange_source_stage_id).expect(\n                    \"We expect child stage fragment for Exchange Operator running in the frontend\",\n                );\n                let mut node_body = execution_plan_node.node.clone();\n                let sources = match &mut node_body {\n                    NodeBody::Exchange(exchange_node) => &mut exchange_node.sources,\n                    NodeBody::MergeSortExchange(merge_sort_exchange_node) => {\n                        &mut merge_sort_exchange_node\n                            .exchange\n                            .as_mut()\n                            .expect(\"MergeSortExchangeNode must have a exchange node\")\n                            .sources\n                    }","sourceCodeStart":258,"sourceCodeEnd":294,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/frontend/src/scheduler/local.rs#L258-L294","documentation":"Raised in `DistributedQueryStageScheduler`/local conversion (`convert_plan_node`, src/frontend/src/scheduler/local.rs:276) when a `BatchExchange` node is encountered while `second_stages` is `None`. Exchanges are only valid during the second-stage conversion pass, so this indicates the plan-to-executor converter saw an exchange in a context where it expected a single-stage plan (or already consumed the second stages).","triggerScenarios":"Calling `convert_plan_node` via `create_plan_fragment` (including recursive conversion) with a batch plan that contains a `BatchExchange` node but no second-stage map supplied, or after all second stages have been removed and another exchange appears.","commonSituations":"Internal scheduler bugs after plan-shape changes (new operators introducing exchanges), or running the local/single-node executor path with a plan built for distributed execution.","solutions":["Check that the query is scheduled through the correct path (distributed vs local) for its plan shape.","Inspect the generated batch plan (`EXPLAIN (TRACE)`) to see why an exchange appears where single-stage execution was assumed.","Retry with a simpler query shape or disable the optimization producing the unexpected exchange.","If reproducible on a supported plan, report with the EXPLAIN output — it is an invariant violation in stage splitting."],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":null,"typeGuard":"// Ensure second_stages exists before processing exchanges\nfn require_second_stages(second_stages: &Option<HashMap<u32, _>>) -> Result<&HashMap<u32, _>> {\n    second_stages.as_ref().ok_or_else(|| anyhow!(\"no second stages supplied for exchange-containing plan\"))\n}","tryCatchPattern":"match result {\n    Err(e) if e.to_string().contains(\"Unexpected exchange detected\") => {\n        log::error!(\"stage splitting invariant broken; retry via distributed scheduler\");\n    }\n    r => r?,\n}","preventionTips":["Run exchange-containing plans through the distributed scheduler path, not the local single-stage path.","Re-run EXPLAIN (TRACE) after planner changes to check plan shape.","Cover stage-splitting invariants with unit tests."],"tags":["scheduler","batch","exchange","invariant"],"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"}