{"record":{"id":"b3fd4e21ad74e09b","repo":"risingwavelabs/risingwave","slug":"job-id-column-not-found-in-source-backfill-table","errorCode":null,"errorMessage":"`job_id` column not found in source backfill table catalog","messagePattern":"`job_id` column not found in source backfill table catalog","errorType":"validation","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/frontend/src/optimizer/rule/table_function_to_internal_source_backfill_progress.rs","lineNumber":153,"sourceCode":"\nimpl TableFunctionToInternalSourceBackfillProgressRule {\n    pub fn create() -> BoxedRule {\n        Box::new(TableFunctionToInternalSourceBackfillProgressRule {})\n    }\n}\n\nstruct SourceBackfillInfo {\n    job_id: JobId,\n    fragment_id: FragmentId,\n    table_id: TableId,\n    partition_id_column_index: usize,\n    backfill_progress_column_index: usize,\n}\n\nimpl SourceBackfillInfo {\n    fn new(table: &TableCatalog) -> anyhow::Result<Self> {\n        let Some(job_id) = table.job_id else {\n            bail!(\"`job_id` column not found in source backfill table catalog\");\n        };\n        let Some(backfill_progress_column_index) = table\n            .columns\n            .iter()\n            .position(|c| c.name() == StreamSourceScan::BACKFILL_PROGRESS_COLUMN_NAME)\n        else {\n            bail!(\n                \"`{}` column not found in source backfill state table schema\",\n                StreamSourceScan::BACKFILL_PROGRESS_COLUMN_NAME\n            );\n        };\n        let Some(partition_id_column_index) = table\n            .columns\n            .iter()\n            .position(|c| c.name() == StreamSourceScan::PARTITION_ID_COLUMN_NAME)\n        else {\n            bail!(\n                \"`{}` column not found in source backfill state table schema\",","sourceCodeStart":135,"sourceCodeEnd":171,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/frontend/src/optimizer/rule/table_function_to_internal_source_backfill_progress.rs#L135-L171","documentation":"SourceBackfillInfo::new is constructing metadata for an internal source backfill state table while rewriting `internal_source_backfill_progress()` into a scan plan. It expects `table.job_id` (Option<JobId>) to be Some; if it is None the table catalog entry lacks the job id association. This is an internal invariant violation: any table returned by `iter_backfilling_internal_tables()` filtered by `is_source_backfill_table` should carry a job_id.","triggerScenarios":"Calling `SELECT * FROM internal_source_backfill_progress()` (or the optimizer rule applying to it) while the catalog contains a source-backfill internal table whose TableCatalog.job_id is None, e.g. a partially-published or stale catalog entry.","commonSituations":"Racing a catalog read against a streaming job creation/removal so a backfill table is observed before its job id is linked; metadata corruption or a version skew between meta node and frontend where the job_id field was not persisted.","solutions":["Retry the `internal_source_backfill_progress()` query after the ongoing DDL (CREATE MATERIALIZED VIEW / CREATE TABLE with backfill) finishes, so the table catalog is fully published.","Check catalog consistency: verify the offending backfill table has an associated streaming job via internal system catalogs; if a job was dropped mid-creation, drop the leftover objects and recreate.","If reproducible on a healthy cluster, report as a frontend/meta bug — the filter `is_source_backfill_table` should exclude tables without job_id; fix `get_source_backfilling_tables` to skip None job_id entries."],"exampleFix":"// before\nlet Some(job_id) = table.job_id else {\n    bail!(\"`job_id` column not found in source backfill table catalog\");\n};\n// after (skip such tables defensively in the caller)\nlet backfill_info = match SourceBackfillInfo::new(&table) {\n    Ok(info) => info,\n    Err(_) => continue, // skip stale/partially published backfill tables\n};","handlingStrategy":"validation","validationCode":"-- ensure no DDL backfill is mid-flight before querying progress\nSELECT * FROM rw_catalog.rw_table_name WHERE name LIKE '%source_backfill%';\n-- only query internal_source_backfill_progress() after CREATE ... finishes","typeGuard":"// Rust: narrow before use\nfn has_job_id(t: &TableCatalog) -> bool { t.job_id.is_some() }","tryCatchPattern":"// match on Result and skip rather than fail the whole query\nmatch SourceBackfillInfo::new(&table) { Ok(i) => push(i), Err(_) => continue }","preventionTips":["Query internal_source_backfill_progress() only after CREATE statements complete","Filter catalog entries lacking job_id before building SourceBackfillInfo","Keep frontend and meta node versions in sync"],"tags":["frontend","optimizer","catalog","backfill"],"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"}