{"record":{"id":"90ed47eb9a22a11f","repo":"risingwavelabs/risingwave","slug":"the-streaming-job-shouldn-t-have-an-upstream-fragm","errorCode":null,"errorMessage":"the streaming job shouldn't have an upstream fragment, job_type: {:?}","messagePattern":"the streaming job shouldn't have an upstream fragment, job_type: (.+?)","errorType":"error_code","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/meta/src/stream/stream_graph/fragment.rs","lineNumber":1885,"sourceCode":"                                        downstream_fragment_id: id,\n                                    },\n                                    // We always use `NoShuffle` for the exchange between the upstream\n                                    // `Source` and the downstream `StreamScan` of the new MV.\n                                    dispatch_strategy: DispatchStrategy {\n                                        r#type: DispatcherType::NoShuffle as _,\n                                        dist_key_indices: vec![], // not used for `NoShuffle`\n                                        output_mapping: Some(output_mapping),\n                                    },\n                                }\n                            } else {\n                                bail!(\n                                    \"the upstream fragment should be a MView or Source, got fragment type: {:b}\",\n                                    upstream_fragment.fragment_type_mask\n                                )\n                            }\n                        }\n                        StreamingJobType::Source | StreamingJobType::Table(_) => {\n                            bail!(\n                                \"the streaming job shouldn't have an upstream fragment, job_type: {:?}\",\n                                job_type\n                            )\n                        }\n                    };\n\n                    // put the edge into the extra edges\n                    extra_downstreams\n                        .entry(upstream_root_fragment_id)\n                        .or_insert_with(HashMap::new)\n                        .try_insert(id, edge.clone())\n                        .unwrap();\n                    extra_upstreams\n                        .entry(id)\n                        .or_insert_with(HashMap::new)\n                        .try_insert(upstream_root_fragment_id, edge)\n                        .unwrap();\n                }","sourceCodeStart":1867,"sourceCodeEnd":1903,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/stream/stream_graph/fragment.rs#L1867-L1903","documentation":"For streaming jobs of type Source or Table, the graph builder expects the job to have no upstream fragment in this replacement/edge-building path. If an upstream fragment is present for such a job type, the builder bails. This is a structural check that source- and table-type jobs never receive an upstream edge here.","triggerScenarios":"Building a replacement graph for a `StreamingJobType::Source` or `StreamingJobType::Table(_)` job while the code path finds a non-empty upstream fragment for it, hitting the bail in the match arm.","commonSituations":"Schema-change/replacement plan construction where the job definition unexpectedly retains an upstream edge; bugs in replacement plan generation or hand-edited/incorrect job metadata.","solutions":["Verify the streaming job's type metadata: a Source/Table job should not carry an upstream fragment here; fix the replacement plan generation.","If this is a CREATE TABLE with a connector, ensure no upstream fragment id was mistakenly attached to the job.","If triggered by normal DDL, capture the job type and fragment metadata and file a bug; try recreating the job as a workaround."],"exampleFix":"null","handlingStrategy":"type-guard","validationCode":"if matches!(job_type, StreamingJobType::Source | StreamingJobType::Table(_)) && upstream_fragment.is_some() {\n    return Err(\"Source/Table jobs must not have an upstream fragment here\");\n}","typeGuard":"fn requires_no_upstream(job_type: &StreamingJobType) -> bool {\n    matches!(job_type, StreamingJobType::Source | StreamingJobType::Table(_))\n}","tryCatchPattern":null,"preventionTips":["Never attach upstream fragment edges to Source/Table type jobs in replacement plans","Validate replacement plans against job type before graph building","Add tests covering Source and Table jobs in the replacement pipeline"],"tags":["meta","streaming","replacement"],"backgroundTag":"invalid-state-transition","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"}