{"record":{"id":"2122acb811f5b725","repo":"risingwavelabs/risingwave","slug":"expect-pbnodebody-sink-but-got","errorCode":null,"errorMessage":"expect PbNodeBody::Sink but got: {:?}","messagePattern":"expect PbNodeBody::Sink but got: (.+?)","errorType":"validation","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/meta/src/stream/stream_graph/fragment.rs","lineNumber":410,"sourceCode":"        maybe_vnode_count: fragment.maybe_vnode_count,\n        nodes: fragment.nodes.clone(),\n    }\n}\n\npub fn check_sink_fragments_support_refresh_schema(\n    fragments: &BTreeMap<FragmentId, Fragment>,\n) -> MetaResult<()> {\n    if fragments.len() != 1 {\n        return Err(anyhow!(\n            \"sink with auto schema change should have only 1 fragment, but got {:?}\",\n            fragments.len()\n        )\n        .into());\n    }\n    let (_, fragment) = fragments.first_key_value().expect(\"non-empty\");\n    let sink_node = &fragment.nodes;\n    let PbNodeBody::Sink(_) = sink_node.node_body.as_ref().unwrap() else {\n        return Err(anyhow!(\"expect PbNodeBody::Sink but got: {:?}\", sink_node.node_body).into());\n    };\n    let [stream_input_node] = sink_node.input.as_slice() else {\n        panic!(\"Sink has more than 1 input: {:?}\", sink_node.input);\n    };\n    let stream_scan_node = match stream_input_node.node_body.as_ref().unwrap() {\n        PbNodeBody::StreamScan(_) => stream_input_node,\n        PbNodeBody::Project(_) => {\n            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!(","sourceCodeStart":392,"sourceCodeEnd":428,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/stream/stream_graph/fragment.rs#L392-L428","documentation":"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.","triggerScenarios":"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`.","commonSituations":"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.","solutions":["Verify frontend and meta node binaries are built from the same commit/version so the streaming graph layout matches.","Check the SQL statement being run; refresh-schema support is only validated for plain sink jobs whose top fragment is a sink.","Reproduce with `EXPLAIN` on the streaming job and inspect the generated fragment root node type.","If it reproduces on a single version, file a RisingWave bug with the streaming job definition and fragment graph dump."],"exampleFix":"// before: assuming the root is always a Sink\nlet PbNodeBody::Sink(_) = sink_node.node_body.as_ref().unwrap() else { ... };\n// after: guard before validating schema refresh support\nif !matches!(sink_node.node_body.as_ref(), Some(PbNodeBody::Sink(_))) {\n    return Ok(()); // skip refresh-schema check for non-sink root fragments\n}","handlingStrategy":"validation","validationCode":"let root_body = fragment.nodes.node_body.as_ref();\nif !matches!(root_body, Some(PbNodeBody::Sink(_))) {\n    return Err(format!(\"sink fragment root is {:?}, not Sink\", root_body));\n}","typeGuard":"fn is_sink_node(node: &StreamNode) -> bool {\n    matches!(node.node_body.as_ref(), Some(PbNodeBody::Sink(_)))\n}","tryCatchPattern":null,"preventionTips":["Keep frontend and meta binaries on identical versions.","Validate the streaming graph shape immediately after generation, before persistence.","Log the fragment graph on every structural mismatch for faster triage."],"tags":["risingwave","streaming-graph","meta-node","plan-validation"],"backgroundTag":"unexpected-response-shape","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}