{"record":{"id":"4fea5b74ab48362d","repo":"risingwavelabs/risingwave","slug":"apply-iceberg-pk-index-sink-overwrite-files-action","errorCode":null,"errorMessage":"apply iceberg pk-index sink overwrite_files action","messagePattern":"apply iceberg pk-index sink overwrite_files action","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/meta/src/manager/iceberg_pk_index_sink/coordinator.rs","lineNumber":532,"sourceCode":"                        .map(&materialize)\n                        .collect::<Result<Vec<_>, _>>()?,\n                };\n                let overwrite_files: Vec<DataFile> = merged\n                    .overwrite_files\n                    .iter()\n                    .map(&materialize)\n                    .collect::<Result<Vec<_>, _>>()?;\n\n                let txn = Transaction::new(&table);\n                let action = txn\n                    .overwrite_files()\n                    .set_snapshot_id(snapshot_id)\n                    .set_target_branch(target_branch)\n                    .add_data_files(add_files)\n                    .delete_files(overwrite_files);\n                let txn = action.apply(txn).map_err(|err| {\n                    CommitError::Commit(\n                        anyhow!(err).context(\"apply iceberg pk-index sink overwrite_files action\"),\n                    )\n                })?;\n                let table = txn.commit(catalog.as_ref()).await.map_err(|err| {\n                    CommitError::Commit(\n                        anyhow!(err).context(\"commit iceberg pk-index sink transaction\"),\n                    )\n                })?;\n                Ok(table)\n            }\n        },\n    )\n    .await\n    .map_err(CommitError::Commit)\n}\n\n#[derive(Clone, Serialize, Deserialize)]\nstruct IcebergPkIndexSinkAggResult {\n    schema_id: i32,","sourceCodeStart":514,"sourceCodeEnd":550,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/manager/iceberg_pk_index_sink/coordinator.rs#L514-L550","documentation":"The coordinator builds an Iceberg `overwrite_files` transaction action (adding data/delete files and deleting overwritten files on a target branch) and applies it to the transaction. If the action's in-memory application fails (invalid requirement evaluation, empty/invalid operations), this error is thrown as `CommitError::Commit`.","triggerScenarios":"`action.apply(txn)` fails in `commit_one_epoch`: e.g. invalid overwrite operation given the table's requirements, snapshot/branch mismatch, or an iceberg-rust validation error when applying the update against the current table metadata.","commonSituations":"Target branch doesn't exist or was repointed; concurrent commits changed table state between table load and apply; iceberg-rust library version with stricter overwrite validation.","solutions":["Read the inner error from the log to see which requirement/operation failed the apply step.","Confirm the target branch exists and is correct in the sink config ('target branch' option).","Rely on/verify the commit retry loop: if the failure is from concurrent commits, ensure `run_with_retry` reloads the latest table before reapplying."],"exampleFix":"null","handlingStrategy":"retry","validationCode":"// ensure branch exists before applying\nlet branches: Vec<_> = table.metadata().branches().unwrap_or_default().collect();\nanyhow::ensure!(branches.iter().any(|b| b == &target_branch) || target_branch == \"main\", \"target branch {target_branch} missing\");","typeGuard":null,"tryCatchPattern":"let txn = match action.apply(txn) {\n    Ok(t) => t,\n    Err(e) => return Err(CommitError::Commit(anyhow!(e).context(\"apply iceberg pk-index sink overwrite_files action\"))),\n};","preventionTips":["Configure a valid existing target branch in sink options.","Reload the latest table snapshot before each retry of the commit.","Keep iceberg-rust library versions consistent across upgrades."],"tags":["iceberg","transaction","commit","concurrency"],"backgroundTag":"invalid-state-transition","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"}