{"record":{"id":"81ede7fb37a9db20","repo":"risingwavelabs/risingwave","slug":"pk-index-sink-pending-row-at-epoch-missing-meta","errorCode":null,"errorMessage":"pk-index sink pending row at epoch {} missing metadata blob","messagePattern":"pk-index sink pending row at epoch (.+?) missing metadata blob","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/meta/src/manager/iceberg_pk_index_sink/coordinator.rs","lineNumber":404,"sourceCode":") -> Result<(Option<u64>, Vec<EpochCommit>)> {\n    fail::fail_point!(\"iceberg_v3_recovery_fail\", |_| Err(anyhow::anyhow!(\n        \"injected: iceberg_v3_recovery_fail\"\n    )));\n    let rows = list_sink_states_ordered_by_epoch(db, sink_id)\n        .await\n        .context(\"list pending sink states for pk-index sink recovery\")?;\n\n    let mut prev_committed_epoch = None;\n    let mut pending = Vec::new();\n    let mut aborted_epochs = Vec::new();\n    for (epoch, state, metadata, _schema_change) in rows {\n        match state {\n            SinkState::Committed => {\n                prev_committed_epoch = Some(epoch);\n            }\n            SinkState::Pending => {\n                let blob = metadata.ok_or_else(|| {\n                    anyhow!(\n                        \"pk-index sink pending row at epoch {} missing metadata blob\",\n                        epoch\n                    )\n                })?;\n                let (merged, snapshot_id) = decode_pre_commit_state(&blob).with_context(|| {\n                    format!(\"decode pk-index sink pre-commit state at epoch {}\", epoch)\n                })?;\n                pending.push(EpochCommit {\n                    epoch,\n                    merged,\n                    snapshot_id,\n                    // Recovered from the persisted blob; the commit materializes files from\n                    // `merged` once.\n                    materialized_add_files: None,\n                });\n            }\n            SinkState::Aborted => {\n                // V3 doesn't produce Aborted rows; tolerate them defensively and drop them so they don't","sourceCodeStart":386,"sourceCodeEnd":422,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/manager/iceberg_pk_index_sink/coordinator.rs#L386-L422","documentation":"On meta restart, `recovery` reads persisted sink-state rows ordered by epoch. A row in `Pending` state must contain the serialized pre-commit metadata blob; if the `metadata` column is NULL, the coordinator cannot reconstruct the pending commit and throws this error.","triggerScenarios":"A `Pending` row was written without its metadata blob — e.g. partial/non-transactional write of the state row, manual DB pruning, or a bug that cleared metadata on state update.","commonSituations":"Meta store corrupted by manual intervention or a restore from a backup taken mid-write; schema migration that dropped the column's data; historical bug writing state before metadata.","solutions":["Inspect the meta store row (sink_states table) for the given sink_id/epoch to confirm the metadata column is NULL.","Remove the offending Pending row (or reset the sink) so recovery skips it and the epoch re-commits from fresh writer reports.","If reproducible, file/fix a bug ensuring the state row and metadata blob are written atomically."],"exampleFix":"null","handlingStrategy":"validation","validationCode":"// before recovery, check rows are self-consistent\nfor row in rows {\n    if row.state == SinkState::Pending {\n        anyhow::ensure!(row.metadata.is_some(), \"pending epoch {} has no metadata blob\", row.epoch);\n    }\n}","typeGuard":"fn pending_has_metadata(row: &SinkStateRow) -> bool {\n    row.state != SinkState::Pending || row.metadata.is_some()\n}","tryCatchPattern":null,"preventionTips":["Never manually edit or prune the meta store sink-state rows.","Write state row and metadata blob atomically in one transaction.","Test restore/backup procedures against pending sinks."],"tags":["iceberg","recovery","internal-invariant-violation","meta-store"],"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"}