{"record":{"id":"e8bba59e194fe0d8","repo":"risingwavelabs/risingwave","slug":"expect-pbnodebody-streamscan-but-got","errorCode":null,"errorMessage":"expect PbNodeBody::StreamScan but got: {:?}","messagePattern":"expect PbNodeBody::StreamScan but got: (.+?)","errorType":"validation","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/meta/src/stream/stream_graph/fragment.rs","lineNumber":436,"sourceCode":"            let [stream_scan_node] = stream_input_node.input.as_slice() else {\n                return Err(anyhow!(\n                    \"Project node must have exactly 1 input for auto schema change, but got {:?}\",\n                    stream_input_node.input.len()\n                )\n                .into());\n            };\n            stream_scan_node\n        }\n        _ => {\n            return Err(anyhow!(\n                \"expect PbNodeBody::StreamScan or PbNodeBody::Project but got: {:?}\",\n                stream_input_node.node_body\n            )\n            .into());\n        }\n    };\n    let PbNodeBody::StreamScan(scan) = stream_scan_node.node_body.as_ref().unwrap() else {\n        return Err(anyhow!(\n            \"expect PbNodeBody::StreamScan but got: {:?}\",\n            stream_scan_node.node_body\n        )\n        .into());\n    };\n    let stream_scan_type = PbStreamScanType::try_from(scan.stream_scan_type).unwrap();\n    if stream_scan_type != PbStreamScanType::ArrangementBackfill {\n        return Err(anyhow!(\n            \"unsupported stream_scan_type for auto refresh schema: {:?}\",\n            stream_scan_type\n        )\n        .into());\n    }\n    let [merge_node, _batch_plan_node] = stream_scan_node.input.as_slice() else {\n        panic!(\n            \"the number of StreamScan inputs is not 2: {:?}\",\n            stream_scan_node.input\n        );","sourceCodeStart":418,"sourceCodeEnd":454,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/stream/stream_graph/fragment.rs#L418-L454","documentation":"After resolving the stream scan node (directly or via a Project indirection), the code destructures its body as `PbNodeBody::StreamScan`. If the node is not a StreamScan, this error is returned. This is an internal invariant: the previous match should guarantee the node is a StreamScan, so hitting this indicates a logic bug or unexpected graph mutation.","triggerScenarios":"`generate_streaming_job` -> `check_sink_fragments_support_refresh_schema` where `stream_scan_node.node_body` is not `PbNodeBody::StreamScan` despite earlier matching — effectively unreachable except via inconsistent fragment data.","commonSituations":"Corrupted or concurrently modified fragment metadata in meta's memory store; non-deterministic node rewriting between validation passes; version drift between components.","solutions":["Restart the meta node to clear in-memory fragment state and retry the streaming job.","Verify all components run the same RisingWave version.","Capture the fragment graph dump from meta logs and inspect the node body.","If reproducible, file a RisingWave bug — this is an internal invariant breach."],"exampleFix":"// before\nlet PbNodeBody::StreamScan(scan) = stream_scan_node.node_body.as_ref().unwrap() else { ... };\n// after: use the match result directly to avoid re-unwrapping\nlet PbNodeBody::StreamScan(scan) = node_body else {\n    unreachable!(\"match above guarantees StreamScan\")\n};","handlingStrategy":"type-guard","validationCode":"if !matches!(stream_scan_node.node_body.as_ref(), Some(PbNodeBody::StreamScan(_))) {\n    panic!(\"invariant breached: scan node is {:?}\", stream_scan_node.node_body);\n}","typeGuard":"fn as_stream_scan(node: &StreamNode) -> Option<&PbStreamScan> {\n    match node.node_body.as_ref() {\n        Some(PbNodeBody::StreamScan(scan)) => Some(scan),\n        _ => None,\n    }\n}","tryCatchPattern":"match result {\n    Ok(()) => {},\n    Err(e) if e.to_string().contains(\"expect PbNodeBody::StreamScan\") => {\n        // internal invariant issue: restart meta / report bug\n    }\n    Err(e) => return Err(e),\n}","preventionTips":["Do not mutate fragment nodes between validation and use.","Cover graph-shape invariants with unit tests on generated graphs.","Keep check and rewrite passes in the same module to avoid drift."],"tags":["risingwave","streaming-graph","internal-invariant","meta-node"],"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"}