What causes "Node {} cannot be convert to stream node" in risingwavelabs/risingwave?

Confirm the plan is meant to be a streaming plan; if not, route it through batch plan-to-prost conversion instead.

Node {} cannot be convert to stream node

Error message

Node {} cannot be convert to stream node

What it means

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.

Source

Thrown at src/frontend/src/optimizer/plan_node/to_prost.rs:50

pub trait ToBatchPb {
    fn to_batch_prost_body(&self) -> pb_batch_node::NodeBody;
}

impl<T: ToBatchPb> TryToBatchPb for T {
    fn try_to_batch_prost_body(&self) -> SchedulerResult<pb_batch_node::NodeBody> {
        Ok(self.to_batch_prost_body())
    }
}

pub trait TryToStreamPb {
    fn try_to_stream_prost_body(
        &self,
        _state: &mut BuildFragmentGraphState,
    ) -> SchedulerResult<pb_stream_node::NodeBody> {
        // Originally we panic in the following way
        // panic!("convert into distributed is only allowed on stream plan")
        Err(anyhow!(
            "Node {} cannot be convert to stream node",
            std::any::type_name::<Self>()
        )
        .into())
    }
}

impl<T: StreamNode> TryToStreamPb for T {
    fn try_to_stream_prost_body(
        &self,
        state: &mut BuildFragmentGraphState,
    ) -> SchedulerResult<pb_stream_node::NodeBody> {
        Ok(self.to_stream_prost_body(state))
    }
}

pub trait StreamNode {
    fn to_stream_prost_body(&self, state: &mut BuildFragmentGraphState)

View on GitHub (pinned to 6469eb736d)

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.

Example fix

// before
impl TryToStreamPb for MyFilterNode {}
// after
impl TryToStreamPb for MyFilterNode {
    fn try_to_stream_prost_body(&self, _state: &mut BuildFragmentGraphState) -> SchedulerResult<pb_stream_node::NodeBody> {
        Ok(pb_stream_node::NodeBody::Filter(self.to_prost()))
    }
}

Defensive patterns

Strategy: try-catch

Validate before calling

// ensure node kind is stream before conversion
if !matches!(plan_node, PlanNode::Stream(_)) { return Err(anyhow!("expected stream plan node")); }

Type guard

fn as_stream(node: &PlanRef) -> Option<&StreamPlanNode> { node.as_stream() }

Try / catch

match node.try_to_stream_prost_body(&mut state) {
    Ok(body) => body,
    Err(e) if e.to_string().contains("cannot be convert to stream node") => return Err(anyhow!("batch-only node in stream fragment: {}", e)),
    Err(e) => return Err(e),
}

Prevention

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.

When it happens

Trigger: 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.

Common situations: A new plan node was added without a TryToStreamPb impl; a materialization decision incorrectly routes a batch-only plan into stream fragment building.

Understand the failure class

Background: UnsupportedOperationException and "is not supported" errors: when a library deliberately refuses a call — this error's family across 30 libraries.

Related errors

AI-assisted analysis of risingwavelabs/risingwave@6469eb736d (2026-09-11). Data as JSON: /api/errors/e4cbda397d58379d. Report an issue: GitHub.