{"record":{"id":"d0d19560ea479617","repo":"risingwavelabs/risingwave","slug":"since-timestamp-requires-at-least-one-upstream-tab","errorCode":null,"errorMessage":"since_timestamp requires at least one upstream table","messagePattern":"since_timestamp requires at least one upstream table","errorType":"validation","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/meta/src/barrier/context/context_impl.rs","lineNumber":196,"sourceCode":"    }\n\n    fn mark_ready(&self, options: MarkReadyOptions) {\n        let is_global = matches!(&options, MarkReadyOptions::Global { .. });\n        self.scheduled_barriers.mark_ready(options);\n        if is_global {\n            self.set_status(BarrierManagerStatus::Running);\n        }\n    }\n\n    async fn resolve_log_store_epoch<'a>(\n        &'a self,\n        upstream_table_ids: impl Iterator<Item = TableId> + Send + 'a,\n        since_epoch: u64,\n    ) -> MetaResult<SinceTimestampResolvedEpoch> {\n        let upstream_table_ids = upstream_table_ids.collect::<Vec<_>>();\n        if upstream_table_ids.is_empty() {\n            return Err(\n                anyhow::anyhow!(\"since_timestamp requires at least one upstream table\").into(),\n            );\n        }\n\n        self.hummock_manager\n            .on_current_version_and_table_change_log(|version, table_change_log| {\n                let mut unified_log_epochs = None;\n                for &upstream_table_id in &upstream_table_ids {\n                    let upstream_committed_epoch = version\n                        .state_table_info\n                        .info()\n                        .get(&upstream_table_id)\n                        .map(|info| info.committed_epoch)\n                        .ok_or_else(|| {\n                            anyhow::anyhow!(\n                                \"cannot get committed epoch for upstream table {}\",\n                                upstream_table_id\n                            )\n                        })?;","sourceCodeStart":178,"sourceCodeEnd":214,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/barrier/context/context_impl.rs#L178-L214","documentation":"resolve_log_store_epoch resolves the unified change-log epoch for a streaming job's upstream tables. It rejects the call up-front when the caller supplies zero upstream table IDs, because computing a 'since' timestamp is meaningless without any table to read change logs from. This is a defensive validation against calling the API with an empty table set.","triggerScenarios":"Calling GlobalBarrierManager::resolve_log_store_epoch (via the barrier manager context) with an upstream_table_ids iterator that yields no elements — e.g. a streaming job with no upstream materialized-source/table dependencies, or an upstream ID collection bug upstream of the call.","commonSituations":"Creating or recovering a sink/job whose upstream resolution returned an empty set; a planner or catalog bug mapping a job to its upstream tables; races where upstream tables were dropped before this call.","solutions":["Inspect the job whose upstream tables resolved to empty; verify it actually has upstream materialized views/sources in the catalog","Fix the caller so it only invokes resolve_log_store_epoch when the upstream table list is non-empty (guard before collecting)","If the job legitimately has no upstream table, change its design (e.g. use a source or datagen) — log-store resolution cannot work tableless","If it's a recovery-time failure, restart recovery after fixing the catalog so upstream mappings are correct"],"exampleFix":"// before\nlet epochs = manager.resolve_log_store_epoch(upstream_table_ids, since_epoch).await?;\n// after\nlet tables: Vec<_> = upstream_table_ids.collect();\nanyhow::ensure!(!tables.is_empty(), \"job has no upstream tables; skip since_timestamp resolution\");\nlet epochs = manager.resolve_log_store_epoch(tables.into_iter(), since_epoch).await?;","handlingStrategy":"validation","validationCode":"let tables: Vec<TableId> = upstream_table_ids.collect();\nif tables.is_empty() {\n    return Err(anyhow::anyhow!(\"no upstream tables to resolve since_timestamp\"));\n}\nlet resolved = manager.resolve_log_store_epoch(tables.into_iter(), since_epoch).await?;","typeGuard":null,"tryCatchPattern":"match manager.resolve_log_store_epoch(ids, epoch).await {\n    Ok(resolved) => use(resolved),\n    Err(e) if e.to_string().contains(\"requires at least one upstream table\") => log::warn!(\"job has no upstream tables, skipping\"),\n    Err(e) => return Err(e),\n}","preventionTips":["Guard callers against empty upstream table sets before invoking","Add unit tests for jobs with zero upstream dependencies","Log upstream table collection results to catch resolution bugs early"],"tags":["rust","meta","barrier","recovery"],"backgroundTag":"empty-required-field","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"}