{"record":{"id":"c538da4e17d86ee4","repo":"risingwavelabs/risingwave","slug":"expect-pbnodebody-streamscan-or-pbnodebody-proje","errorCode":null,"errorMessage":"expect PbNodeBody::StreamScan or PbNodeBody::Project but got: {:?}","messagePattern":"expect PbNodeBody::StreamScan or PbNodeBody::Project but got: (.+?)","errorType":"validation","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/meta/src/stream/stream_graph/fragment.rs","lineNumber":428,"sourceCode":"        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!(\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","sourceCodeStart":410,"sourceCodeEnd":446,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/stream/stream_graph/fragment.rs#L410-L446","documentation":"In `check_sink_fragments_support_refresh_schema`, after the sink node the code expects the next node to be either a StreamScan or a Project (Project being the schema-change indirection). Any other node body variant reaches this catch-all branch and produces this error, meaning the sink fragment topology is not one of the shapes the auto-refresh-schema feature understands.","triggerScenarios":"`generate_streaming_job` -> `check_sink_fragments_support_refresh_schema` where the sink's single input node has a `node_body` other than `PbNodeBody::StreamScan` or `PbNodeBody::Project` (e.g. Exchange, Filter, TopN).","commonSituations":"Creating a sink over a materialized view or query whose streaming graph inserts extra operators (filters, transforms) between the sink and the backfill scan, then relying on auto schema refresh; plan layout changes across RisingWave versions.","solutions":["Create the sink directly over a table/backfill arrangement without intermediate operators if schema refresh is required.","Verify the sink's upstream is a table with backfill; auto refresh-schema only supports the StreamScan(ArrangementBackfill) shape.","Align frontend/meta versions and retry the DDL.","If supported SQL still triggers it, file a RisingWave bug with the job definition."],"exampleFix":"// before\n_ => { return Err(anyhow!(\"expect PbNodeBody::StreamScan or PbNodeBody::Project but got: {:?}\", ...).into()); }\n// after: skip validation for unsupported shapes instead of failing job creation\n_ => { tracing::info!(\"sink fragment shape {:?} not eligible for refresh schema; skipping\", stream_input_node.node_body); return Ok(()); }","handlingStrategy":"validation","validationCode":"match stream_input_node.node_body.as_ref() {\n    Some(PbNodeBody::StreamScan(_) | PbNodeBody::Project(_)) => {},\n    other => return Err(format!(\"unsupported sink input for refresh schema: {:?}\", other)),\n}","typeGuard":"fn supports_refresh_schema(node: &StreamNode) -> bool {\n    matches!(node.node_body.as_ref(), Some(PbNodeBody::StreamScan(_) | PbNodeBody::Project(_)))\n}","tryCatchPattern":null,"preventionTips":["Create sinks directly over tables/backfill scans when schema refresh is needed.","Check the feature's supported topology docs before enabling auto schema change.","Test schema evolution flows in staging after version upgrades."],"tags":["risingwave","streaming-graph","schema-change","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-14T16:17:12.679Z"}