risingwavelabs/risingwave · error

expect PbNodeBody::Sink but got: {:?}

Error message

expect PbNodeBody::Sink but got: {:?}

What it means

During auto schema-change validation of a sink streaming job, RisingWave inspects the top node of the sink fragment and asserts its protobuf node body is a Sink (`PbNodeBody::Sink`). This `anyhow!` error is returned when the fragment's root node has some other node body variant, meaning the generated graph does not match the expected sink shape. It is a defensive structural check on the streaming plan rather than a user-facing SQL error.

Source

Thrown at src/meta/src/stream/stream_graph/fragment.rs:410

        maybe_vnode_count: fragment.maybe_vnode_count,
        nodes: fragment.nodes.clone(),
    }
}

pub fn check_sink_fragments_support_refresh_schema(
    fragments: &BTreeMap<FragmentId, Fragment>,
) -> MetaResult<()> {
    if fragments.len() != 1 {
        return Err(anyhow!(
            "sink with auto schema change should have only 1 fragment, but got {:?}",
            fragments.len()
        )
        .into());
    }
    let (_, fragment) = fragments.first_key_value().expect("non-empty");
    let sink_node = &fragment.nodes;
    let PbNodeBody::Sink(_) = sink_node.node_body.as_ref().unwrap() else {
        return Err(anyhow!("expect PbNodeBody::Sink but got: {:?}", sink_node.node_body).into());
    };
    let [stream_input_node] = sink_node.input.as_slice() else {
        panic!("Sink has more than 1 input: {:?}", sink_node.input);
    };
    let stream_scan_node = match stream_input_node.node_body.as_ref().unwrap() {
        PbNodeBody::StreamScan(_) => stream_input_node,
        PbNodeBody::Project(_) => {
            let [stream_scan_node] = stream_input_node.input.as_slice() else {
                return Err(anyhow!(
                    "Project node must have exactly 1 input for auto schema change, but got {:?}",
                    stream_input_node.input.len()
                )
                .into());
            };
            stream_scan_node
        }
        _ => {
            return Err(anyhow!(

View on GitHub (pinned to 6469eb736d)

Solutions

1. Verify frontend and meta node binaries are built from the same commit/version so the streaming graph layout matches.
2. Check the SQL statement being run; refresh-schema support is only validated for plain sink jobs whose top fragment is a sink.
3. Reproduce with `EXPLAIN` on the streaming job and inspect the generated fragment root node type.
4. If it reproduces on a single version, file a RisingWave bug with the streaming job definition and fragment graph dump.

Example fix

// before: assuming the root is always a Sink
let PbNodeBody::Sink(_) = sink_node.node_body.as_ref().unwrap() else { ... };
// after: guard before validating schema refresh support
if !matches!(sink_node.node_body.as_ref(), Some(PbNodeBody::Sink(_))) {
    return Ok(()); // skip refresh-schema check for non-sink root fragments
}

let root_body = fragment.nodes.node_body.as_ref();
if !matches!(root_body, Some(PbNodeBody::Sink(_))) {
    return Err(format!("sink fragment root is {:?}, not Sink", root_body));
}

fn is_sink_node(node: &StreamNode) -> bool {
    matches!(node.node_body.as_ref(), Some(PbNodeBody::Sink(_)))
}

When it happens

Trigger: Calling `generate_streaming_job` which invokes `check_sink_fragments_support_refresh_schema` when the first (and only) fragment of the streaming job has a root node whose `node_body` is not `PbNodeBody::Sink`.

Common situations: Internal inconsistency after frontend plan generation or a version drift where the frontend emits a different fragment layout (e.g. a Project or StreamScan pushed to the sink fragment root) than the meta node's validation expects; typically seen when upgrading frontend/meta components out of lockstep.

Related errors

• expect PbNodeBody::Merge but got: {:?}
• Project node must have exactly 1 input for auto schema chang
• expect PbNodeBody::StreamScan or PbNodeBody::Project but got
• expect PbNodeBody::StreamScan but got: {:?}
• empty meta addresses

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