{"record":{"id":"e4cbda397d58379d","repo":"risingwavelabs/risingwave","slug":"node-cannot-be-convert-to-stream-node","errorCode":null,"errorMessage":"Node {} cannot be convert to stream node","messagePattern":"Node (.+?) cannot be convert to stream node","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/frontend/src/optimizer/plan_node/to_prost.rs","lineNumber":50,"sourceCode":"\npub trait ToBatchPb {\n    fn to_batch_prost_body(&self) -> pb_batch_node::NodeBody;\n}\n\nimpl<T: ToBatchPb> TryToBatchPb for T {\n    fn try_to_batch_prost_body(&self) -> SchedulerResult<pb_batch_node::NodeBody> {\n        Ok(self.to_batch_prost_body())\n    }\n}\n\npub trait TryToStreamPb {\n    fn try_to_stream_prost_body(\n        &self,\n        _state: &mut BuildFragmentGraphState,\n    ) -> SchedulerResult<pb_stream_node::NodeBody> {\n        // Originally we panic in the following way\n        // panic!(\"convert into distributed is only allowed on stream plan\")\n        Err(anyhow!(\n            \"Node {} cannot be convert to stream node\",\n            std::any::type_name::<Self>()\n        )\n        .into())\n    }\n}\n\nimpl<T: StreamNode> TryToStreamPb for T {\n    fn try_to_stream_prost_body(\n        &self,\n        state: &mut BuildFragmentGraphState,\n    ) -> SchedulerResult<pb_stream_node::NodeBody> {\n        Ok(self.to_stream_prost_body(state))\n    }\n}\n\npub trait StreamNode {\n    fn to_stream_prost_body(&self, state: &mut BuildFragmentGraphState)","sourceCodeStart":32,"sourceCodeEnd":68,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/frontend/src/optimizer/plan_node/to_prost.rs#L32-L68","documentation":"TryToStreamPb::try_to_stream_prost_body's default implementation always returns this error: only stream plan nodes may be serialized into a protobuf StreamNode body. Calling it on a node that did not override the method (a batch-only node) yields this SchedulerResult error instead of the original panic.","triggerScenarios":"Calling try_to_stream_prost_body() during stream plan graph construction (BuildFragmentGraphState) on a plan node lacking an override, i.e. a batch plan erroneously sent down the streaming execution path.","commonSituations":"A new plan node was added without a TryToStreamPb impl; a materialization decision incorrectly routes a batch-only plan into stream fragment building.","solutions":["Confirm the plan is meant to be a streaming plan; if not, route it through batch plan-to-prost conversion instead.","Implement TryToStreamPb for the concrete node type and return its StreamNode body.","Fix the optimizer/streaming decision so batch-only operators are never placed in stream fragments."],"exampleFix":"// before\nimpl TryToStreamPb for MyFilterNode {}\n// after\nimpl TryToStreamPb for MyFilterNode {\n    fn try_to_stream_prost_body(&self, _state: &mut BuildFragmentGraphState) -> SchedulerResult<pb_stream_node::NodeBody> {\n        Ok(pb_stream_node::NodeBody::Filter(self.to_prost()))\n    }\n}","handlingStrategy":"try-catch","validationCode":"// ensure node kind is stream before conversion\nif !matches!(plan_node, PlanNode::Stream(_)) { return Err(anyhow!(\"expected stream plan node\")); }","typeGuard":"fn as_stream(node: &PlanRef) -> Option<&StreamPlanNode> { node.as_stream() }","tryCatchPattern":"match node.try_to_stream_prost_body(&mut state) {\n    Ok(body) => body,\n    Err(e) if e.to_string().contains(\"cannot be convert to stream node\") => return Err(anyhow!(\"batch-only node in stream fragment: {}\", e)),\n    Err(e) => return Err(e),\n}","preventionTips":["Implement TryToStreamPb for every new stream plan node type.","Cover fragment building for all node kinds in unit tests.","Route batch-only plans exclusively through the batch scheduler."],"tags":["frontend","plan-serialization","streaming","protobuf"],"backgroundTag":"unsupported-operation","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"}