{"record":{"id":"713d1a6aee031b7a","repo":"risingwavelabs/risingwave","slug":"database-should-exist-for-streaming-job","errorCode":null,"errorMessage":"database should exist for streaming job","messagePattern":"database should exist for streaming job","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"critical","filePath":"src/meta/src/controller/scale.rs","lineNumber":355,"sourceCode":"            ensembles,\n            mut job_fragments,\n            mut job_map,\n            streaming_job_databases,\n            mut database_map,\n            mut fragment_source_ids,\n            mut fragment_splits,\n        } = self;\n\n        let mut contexts = HashMap::<DatabaseId, Self>::new();\n        let mut fragment_databases = HashMap::<FragmentId, DatabaseId>::new();\n        let mut unresolved_ensembles = 0usize;\n        let mut unresolved_ensemble_sample: Option<Vec<FragmentId>> = None;\n\n        for (job_id, database_id) in streaming_job_databases {\n            let context = contexts.entry(database_id).or_insert_with(|| {\n                let database_model = database_map\n                    .remove(&database_id)\n                    .expect(\"database should exist for streaming job\");\n                Self {\n                    ensembles: Vec::new(),\n                    job_fragments: HashMap::new(),\n                    job_map: HashMap::new(),\n                    streaming_job_databases: HashMap::new(),\n                    database_map: HashMap::from([(database_id, database_model)]),\n                    fragment_source_ids: HashMap::new(),\n                    fragment_splits: HashMap::new(),\n                }\n            });\n\n            let fragments = job_fragments\n                .remove(&job_id)\n                .expect(\"job fragments should exist for streaming job\");\n            for fragment_id in fragments.keys().copied() {\n                fragment_databases.insert(fragment_id, database_id);\n                if let Some(source_id) = fragment_source_ids.remove(&fragment_id) {\n                    context.fragment_source_ids.insert(fragment_id, source_id);","sourceCodeStart":337,"sourceCodeEnd":373,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/controller/scale.rs#L337-L373","documentation":"In scale.rs, when building per-database resharding contexts, each streaming job's database_id must resolve to a DatabaseModel in database_map; .expect panics if not. It means a streaming job is recorded against a database that no longer exists in the catalog — catalog inconsistency.","triggerScenarios":"Running resharding/ALTER ... SET PARALLELISM scale operations while the streaming_job_databases map references a database_id missing from database_map (database dropped while jobs remain, or stale catalog rows).","commonSituations":"Dropping a database with lingering jobs due to prior partial failure, meta store corruption, manual row deletion, or racing a DROP DATABASE with a scale operation.","solutions":["Check the database referenced by the job still exists: SELECT from the databases/objects tables for the database_id.","Drop or clean the orphaned streaming job(s) referencing the missing database, or restore the database.","Avoid concurrent DROP DATABASE and resharding operations; retry the scale operation after the catalog is consistent."],"exampleFix":null,"handlingStrategy":"validation","validationCode":"let missing: Vec<_> = streaming_job_databases.iter()\n    .map(|(_, db)| db)\n    .filter(|db| !database_map.contains_key(db))\n    .collect();\nif !missing.is_empty() { bail!(\"databases missing for jobs: {:?}\", missing); }","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Ensure DROP DATABASE completes job cleanup before removal","Avoid concurrent DROP DATABASE and resharding/scale operations","Run catalog consistency audits (jobs referencing missing databases)","Restore from meta backup if orphaned jobs are detected"],"tags":["panic","scale","catalog","database"],"backgroundTag":"entity-not-found","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-23T08:17:48.524Z"}