{"record":{"id":"65df22fd2e2f78e4","repo":"risingwavelabs/risingwave","slug":"snapshot-backfill-epoch-set-again","errorCode":null,"errorMessage":"snapshot backfill epoch set again: {} {} {}","messagePattern":"snapshot backfill epoch set again: (.+?) (.+?) (.+?)","errorType":"error_code","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/meta/src/stream/stream_graph/fragment.rs","lineNumber":1627,"sourceCode":"        {\n            result = try {\n                let table_id = stream_scan.table_id;\n                let snapshot_epoch = cross_db_snapshot_backfill_info\n                    .upstream_mv_table_id_to_backfill_epoch\n                    .get(&table_id)\n                    .or_else(|| {\n                        snapshot_backfill_info.and_then(|snapshot_backfill_info| {\n                            snapshot_backfill_info\n                                .upstream_mv_table_id_to_backfill_epoch\n                                .get(&table_id)\n                        })\n                    })\n                    .ok_or_else(|| anyhow!(\"upstream table id not covered: {}\", table_id))?\n                    .ok_or_else(|| anyhow!(\"upstream table id not set: {}\", table_id))?;\n                if let Some(prev_snapshot_epoch) =\n                    stream_scan.snapshot_backfill_epoch.replace(snapshot_epoch)\n                {\n                    Err(anyhow!(\n                        \"snapshot backfill epoch set again: {} {} {}\",\n                        table_id,\n                        prev_snapshot_epoch,\n                        snapshot_epoch\n                    ))?;\n                }\n                applied = true;\n            };\n            result.is_ok()\n        } else {\n            true\n        }\n    });\n    result.map_err(MetaError::from).map(|_| applied)\n}\n\nstatic EMPTY_HASHMAP: LazyLock<HashMap<GlobalFragmentId, StreamFragmentEdge>> =\n    LazyLock::new(HashMap::new);","sourceCodeStart":1609,"sourceCodeEnd":1645,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/stream/stream_graph/fragment.rs#L1609-L1645","documentation":"During stream graph building, the meta service records the snapshot epoch used for a table's snapshot backfill stream scan. `fill_snapshot_backfill_epoch` asserts that each upstream table's stream scan gets exactly one snapshot epoch; if `replace()` finds a previously set epoch for the same table, the graph builder throws. This is an internal invariant: a fragment should never scan the same upstream table twice with different (or duplicate) snapshot backfill epochs.","triggerScenarios":"Building a graph where a streaming job has two stream-scan operators reading the same upstream table id (e.g. a job created via CREATE TABLE AS or a sink referencing the same MV table from multiple upstream fragments), so `fill_snapshot_backfill_epoch` is invoked twice for one `table_id` with an epoch already recorded.","commonSituations":"Rare meta-side invariant violations surfaced during CREATE MATERIALIZED VIEW / CREATE SINK graph construction, typically after refactors to stream-scan/upstream-table wiring or during upgrading where a job references the same upstream table from multiple fragments.","solutions":["Inspect the built graph (fragment plan) and remove the duplicate stream scan over the same upstream table so each table id is scanned once for snapshot backfill.","If duplicate reference is intentional, restructure the job so downstream fragments reuse a single upstream fragment instead of directly scanning the table twice.","Check for recent changes to `fill_snapshot_backfill_epoch` call sites and ensure only one call happens per table id; file a bug with the SQL/job definition if triggered by an ordinary query."],"exampleFix":"// before: two stream scans both call fill_snapshot_backfill_epoch for table_id 1001\n// after: reuse one stream scan fragment's output for both consumers\n// (restructure the graph so only one stream scan references table_id 1001)","handlingStrategy":"validation","validationCode":"// Before building the graph, ensure each upstream table id appears in exactly one stream scan\nlet mut seen = std::collections::HashSet::new();\nfor scan in collect_stream_scans(&graph) {\n    if !seen.insert(scan.upstream_table_id) {\n        return Err(format!(\"table {} scanned by multiple stream scans\", scan.upstream_table_id));\n    }\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Ensure each upstream table is scanned by a single stream-scan fragment in the built graph","After graph-building refactors, add a unit test asserting one snapshot backfill epoch per table id","Include the job/fragment ids in bug reports when this invariant trips"],"tags":["meta","streaming","internal-invariant"],"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-14T11:17:12.474Z"}