{"record":{"id":"bdb37142eb44d716","repo":"risingwavelabs/risingwave","slug":"invalid-backfill-state-backfill-finished-bdb371","errorCode":null,"errorMessage":"invalid backfill state: backfill_finished","messagePattern":"invalid backfill state: backfill_finished","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/stream/src/executor/backfill/cdc/state_v2.rs","lineNumber":70,"sourceCode":"\n    pub async fn init_epoch(&mut self, epoch: EpochPair) -> StreamExecutorResult<()> {\n        self.state_table.init_epoch(epoch).await\n    }\n\n    /// Restore the backfill state from storage\n    pub async fn restore_state(&mut self, split_id: i64) -> StreamExecutorResult<CdcStateRecord> {\n        let key = Some(split_id);\n        match self\n            .state_table\n            .get_row(row::once(key.map(ScalarImpl::from)))\n            .await?\n        {\n            Some(row) => {\n                tracing::info!(\"restored cdc backfill state: {:?}\", row);\n                let state = row.into_inner().into_vec();\n                let is_finished = match state[1] {\n                    Some(ScalarImpl::Bool(val)) => val,\n                    _ => return Err(anyhow!(\"invalid backfill state: backfill_finished\").into()),\n                };\n                let row_count = match state[2] {\n                    Some(ScalarImpl::Int64(val)) => val,\n                    _ => return Err(anyhow!(\"invalid backfill state: row_count\").into()),\n                };\n                let (cdc_offset_low, cdc_offset_high) = if !self.is_legacy_state {\n                    let cdc_offset_low = match state[3] {\n                        Some(ScalarImpl::Jsonb(ref jsonb)) => {\n                            serde_json::from_value(jsonb.clone().take()).unwrap()\n                        }\n                        None => None,\n                        _ => return Err(anyhow!(\"invalid backfill state: cdc_offset_low\").into()),\n                    };\n                    let cdc_offset_high = match state[4] {\n                        Some(ScalarImpl::Jsonb(ref jsonb)) => {\n                            serde_json::from_value(jsonb.clone().take()).unwrap()\n                        }\n                        None => None,","sourceCodeStart":52,"sourceCodeEnd":88,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/stream/src/executor/backfill/cdc/state_v2.rs#L52-L88","documentation":"Thrown by CdcBackfillStateV2::restore_state when element [1] of the persisted state row is not a Bool, so the backfill_finished flag cannot be decoded. It guards against reading a state row written with a different layout or corrupted value.","triggerScenarios":"Restoring a v2 CDC backfill state where state[1] is None or holds a non-Bool ScalarImpl (e.g. Int64/Bytea from a legacy or foreign row layout).","commonSituations":"Version upgrade/migration where the state schema changed; a row written by the legacy no-shuffle backfill being read by the v2 state reader; corrupted persisted state after a crash mid-write.","solutions":["Cancel and recreate the streaming job so the CDC backfill is rebuilt with a fresh state row","Verify the writer/reader versions match; downgrade to the version that wrote the state and let backfill finish before upgrading","Inspect the internal state table row to confirm the field type mismatch","Restore the state table from backup if the row was corrupted"],"exampleFix":null,"handlingStrategy":"validation","validationCode":"let is_finished = matches!(state[STATE_FINISHED_IDX], Some(ScalarImpl::Bool(_)));\nif !is_finished { /* treat state as incompatible; recreate job */ }","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Upgrade via the documented migration path instead of skipping versions","Back up state tables before major version upgrades","Watch migration test suites for state-layout changes"],"tags":["rust","streaming","cdc","state-restore"],"backgroundTag":"type-mismatch","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"}