risingwavelabs/risingwave · error

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

Error message

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

What it means

The StreamScan of an arrangement backfill must have exactly two inputs: a Merge (fetching from the upstream fragment) and a batch plan node. This error fires when the first input's node body is not `PbNodeBody::Merge`, so the backfill scan's upstream linkage cannot be validated.

Source

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

        )
        .into());
    };
    let stream_scan_type = PbStreamScanType::try_from(scan.stream_scan_type).unwrap();
    if stream_scan_type != PbStreamScanType::ArrangementBackfill {
        return Err(anyhow!(
            "unsupported stream_scan_type for auto refresh schema: {:?}",
            stream_scan_type
        )
        .into());
    }
    let [merge_node, _batch_plan_node] = stream_scan_node.input.as_slice() else {
        panic!(
            "the number of StreamScan inputs is not 2: {:?}",
            stream_scan_node.input
        );
    };
    let NodeBody::Merge(_) = merge_node.node_body.as_ref().unwrap() else {
        return Err(anyhow!(
            "expect PbNodeBody::Merge but got: {:?}",
            merge_node.node_body
        )
        .into());
    };
    Ok(())
}

/// Output mapping info after rewriting a `StreamScan` node.
struct ScanRewriteResult {
    old_output_index_to_new_output_index: HashMap<u32, u32>,
    new_output_index_by_column_id: HashMap<ColumnId, u32>,
    output_fields: Vec<risingwave_pb::plan_common::Field>,
}

/// Append new columns to a sink/log-store column list with updated names/ids.
fn extend_sink_columns(
    sink_columns: &mut Vec<PbColumnCatalog>,

View on GitHub (pinned to 6469eb736d)

Solutions

1. Ensure frontend and meta binaries are the same version and retry the DDL.
2. Dump the fragment graph from meta logs and inspect the StreamScan inputs.
3. Restart the meta node to rebuild in-memory graph state.
4. If reproducible on aligned versions, file a RisingWave bug with the job definition.

Example fix

// before
let NodeBody::Merge(_) = merge_node.node_body.as_ref().unwrap() else { ... };
// after: accept the non-merge shape with a diagnostic
let Some(NodeBody::Merge(m)) = merge_node.node_body.as_ref() else {
    return Err(anyhow!("backfill scan first input is {:?}, expected Merge", merge_node.node_body).into());
};

if !matches!(merge_node.node_body.as_ref(), Some(NodeBody::Merge(_))) {
    return Err("backfill scan first input must be Merge".into());
}

fn first_input_is_merge(scan_node: &StreamNode) -> bool {
    matches!(
        scan_node.input.first().and_then(|n| n.node_body.as_ref()),
        Some(NodeBody::Merge(_))
    )
}

When it happens

Trigger: `generate_streaming_job` -> `check_sink_fragments_support_refresh_schema` where `stream_scan_node.input[0].node_body` is not `PbNodeBody::Merge`.

Common situations: Internal graph-generation inconsistency; frontend/meta version mismatch producing a different StreamScan input layout; corrupted fragment metadata.

Related errors

• expect PbNodeBody::Sink 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/3b17625af0eaba3b. Report an issue: GitHub.