From 852f2c1e4e0e8574688764eff1b2dc60e8abc7f2 Mon Sep 17 00:00:00 2001 From: Gerald Berger Date: Wed, 16 Sep 2026 10:51:40 +0200 Subject: [PATCH] fix(transaction): fix correctness bugs in OverwriteAction Prompted by review comments on the upstream PR our OverwriteAction was originally built on top of (apache/iceberg-rust#2185). Checked each comment against our own implementation; three applied here too: - Resurrection bug (the serious one): rewrite_manifest's fallback branch called add_existing_entry for every entry not newly deleted this round, including entries that were already Deleted by a prior overwrite -- add_existing_entry unconditionally resets status to Existing, silently resurrecting previously-deleted files as live data the next time their manifest got rewritten for an unrelated reason. Fixed by routing already-non-alive entries through add_deleted_entry instead, preserving their original snapshot_id as a tombstone. Added test_second_overwrite_does_not_resurrect_deleted_file to lock this in. - Wrong schema-id on rewritten manifests: schema was taken from table.metadata().current_schema() (the table's *current* schema) rather than the manifest being rewritten's own schema. After any schema evolution between the original write and a later overwrite, this stamped the wrong schema-id on old entries, which schema-id-aware readers (Java, PyIceberg) would misinterpret. Fixed by deriving both schema and partition spec directly from the manifest's own ManifestMetadata (which already carries fully resolved objects, not just IDs) -- this is also simpler than the existing partition_spec_by_id table lookup it replaces, and doesn't depend on the table still listing that partition spec. - Manifest naming/location bypassed convention: rewritten manifests were written to a hardcoded `{location}/metadata/` path with a fresh random UUID per manifest, rather than `metadata_location()` (which respects a configured write.metadata.path table property, already used correctly elsewhere in this same file for the manifest list) and the commit's own UUID (shared across every manifest touched by one commit, matching SnapshotProducer::new_manifest_writer's own convention). Added a commit_uuid() getter on SnapshotProducer and an index parameter to rewrite_manifest to keep names unique when a commit rewrites more than one manifest. Also fixed, one level up in shared code: SnapshotProducer::summary() unconditionally applied "truncate full table" semantics (replace computed added/removed counts with the previous snapshot's totals) for any Operation::Overwrite, which is wrong for OverwriteAction's explicit-file-list partial overwrites -- it already knows exactly what was added/removed. Added a `truncate_full_table()` method to SnapshotProduceOperation (default false, only meaningful for an operation that genuinely replaces the whole table) and had OverwriteOperation opt out explicitly. Updated test_delete_only_overwrite_summary, which had documented the old (wrong) truncated counts as expected behavior, to assert the correct ones. Not applicable: the review's manifest-filtering concern (dropping delete-only manifests) is already handled correctly on our side -- both FastAppendAction and OverwriteAction's existing_manifest() were confirmed to already keep manifests with has_deleted_files(). Co-Authored-By: Claude Sonnet 5 --- crates/iceberg/src/transaction/overwrite.rs | 145 +++++++++++++++----- crates/iceberg/src/transaction/snapshot.rs | 17 ++- 2 files changed, 130 insertions(+), 32 deletions(-) diff --git a/crates/iceberg/src/transaction/overwrite.rs b/crates/iceberg/src/transaction/overwrite.rs index 68724bc0b5..051c753753 100644 --- a/crates/iceberg/src/transaction/overwrite.rs +++ b/crates/iceberg/src/transaction/overwrite.rs @@ -149,6 +149,13 @@ impl SnapshotProduceOperation for OverwriteOperation { Operation::Overwrite } + // `OverwriteAction` always works from an explicit, exact list of added/deleted data + // files, so the snapshot summary's computed added/removed counts are already correct + // -- never truncate them to "everything in the previous snapshot was removed." + fn truncate_full_table(&self) -> bool { + false + } + async fn delete_entries( &self, _snapshot_produce: &SnapshotProducer<'_>, @@ -184,6 +191,10 @@ impl SnapshotProduceOperation for OverwriteOperation { } let mut result = Vec::new(); + // Shared across every manifest rewritten by this one commit, so rewritten + // manifest names stay unique within the commit without needing a fresh random + // UUID per manifest (see `rewrite_manifest`). + let mut rewrite_counter: u64 = 0; for manifest_file in manifest_list.entries() { if !manifest_file.has_added_files() @@ -205,8 +216,9 @@ impl SnapshotProduceOperation for OverwriteOperation { if has_deletes { let rewritten = self - .rewrite_manifest(snapshot_produce, manifest_file, &manifest) + .rewrite_manifest(snapshot_produce, manifest_file, &manifest, rewrite_counter) .await?; + rewrite_counter += 1; result.push(rewritten); } else { result.push(manifest_file.clone()); @@ -219,36 +231,37 @@ impl SnapshotProduceOperation for OverwriteOperation { impl OverwriteOperation { /// Rewrite a manifest, marking entries whose file paths are in `deleted_file_paths` - /// as `ManifestStatus::Deleted`. + /// as `ManifestStatus::Deleted`. `index` disambiguates this manifest's name from + /// others rewritten by the same commit (see `rewrite_counter` in `existing_manifest`). async fn rewrite_manifest( &self, snapshot_produce: &SnapshotProducer<'_>, manifest_file: &ManifestFile, manifest: &crate::spec::Manifest, + index: u64, ) -> Result { let table = snapshot_produce.table; + // Match the naming/location convention `SnapshotProducer::new_manifest_writer` + // uses for newly-written manifests: `metadata_location()` (which, unlike a bare + // `location()`, respects a configured `write.metadata.path`) and the commit's own + // UUID, so every manifest touched by one commit is identifiable by that UUID + // rather than each rewritten manifest getting its own unrelated random one. let new_manifest_path = format!( - "{}/metadata/{}-m-overwrite.avro", - table.metadata().location(), - Uuid::now_v7(), + "{}/{}-m-overwrite-{}.avro", + table.metadata().metadata_location()?, + snapshot_produce.commit_uuid(), + index, ); let output_file = table.file_io().new_output(&new_manifest_path)?; - let partition_spec = table - .metadata() - .partition_spec_by_id(manifest_file.partition_spec_id) - .ok_or_else(|| { - Error::new( - ErrorKind::DataInvalid, - format!( - "Partition spec {} not found in table metadata", - manifest_file.partition_spec_id - ), - ) - })? - .as_ref() - .clone(); - let schema = table.metadata().current_schema().clone(); + // Derive schema and partition spec from the manifest being rewritten itself, + // not the table's current/default ones: if the table has since undergone schema + // or partition evolution, stamping the current schema-id/partition-spec-id onto + // entries written under an older one would misrepresent them and cause + // implementations that respect the manifest's own schema-id (e.g. Java, + // PyIceberg) to misread them. + let partition_spec = manifest.metadata().partition_spec().clone(); + let schema = manifest.metadata().schema().clone(); // Preserve the original manifest's own encryption key when rewriting it, rather // than generating a new one, so the rewritten manifest stays decryptable the same @@ -291,6 +304,14 @@ impl OverwriteOperation { let mut deleted: ManifestEntry = (**entry).clone(); deleted.snapshot_id = Some(self.snapshot_id); writer.add_deleted_entry(deleted)?; + } else if !entry.is_alive() { + // Already deleted by a prior snapshot. Preserve it as a tombstone via + // add_deleted_entry (which keeps its original snapshot_id, since we + // don't touch it here) rather than add_existing_entry, which + // unconditionally resets status to Existing and would resurrect this + // file as live data. + let cloned: ManifestEntry = (**entry).clone(); + writer.add_deleted_entry(cloned)?; } else { let cloned: ManifestEntry = (**entry).clone(); writer.add_existing_entry(cloned)?; @@ -705,23 +726,85 @@ mod tests { let props = &snapshot.summary().additional_properties; - // Overwrite semantics use truncate_table_summary which treats overwrite as - // "replace all": deleted-data-files reflects the previous total (3), not the - // number of explicitly deleted files (1). Without Fix 1, previous_snapshot - // would be None so previous_total=0 and u64 subtraction would underflow. + // OverwriteAction always works from an explicit, exact file list, so a partial + // (delete-only) overwrite must report exactly what was removed (1), not the + // previous snapshot's total (3) -- `truncate_full_table()` returning `false` + // for OverwriteOperation is what keeps `update_snapshot_summaries` from + // overwriting these already-correct computed counts with "replace all" + // semantics. assert_eq!( props.get("deleted-data-files").map(|s| s.as_str()), - Some("3"), - "Expected deleted-data-files=3 (overwrite semantics), got: {props:?}" + Some("1"), + "Expected deleted-data-files=1 (only the explicitly deleted file), got: {props:?}" ); - // total-data-files should be 0 after a delete-only overwrite (overwrite - // semantics: 3 removed - 3 previous = 0). Without Fix 1 this would panic - // with u64 underflow in debug mode. + // total-data-files should be 2 after deleting 1 of the 3 previously-appended files. assert_eq!( props.get("total-data-files").map(|s| s.as_str()), - Some("0"), - "Expected total-data-files=0 (overwrite semantics), got: {props:?}" + Some("2"), + "Expected total-data-files=2 (3 appended - 1 deleted), got: {props:?}" + ); + } + + #[tokio::test] + async fn test_second_overwrite_does_not_resurrect_deleted_file() { + use crate::memory::tests::new_memory_catalog; + use crate::transaction::ApplyTransactionAction; + use crate::transaction::tests::make_v3_minimal_table_in_catalog; + + let catalog = new_memory_catalog().await; + let table = make_v3_minimal_table_in_catalog(&catalog).await; + let spec_id = table.metadata().default_partition_spec_id(); + + // Append files A and B together so they land in the same manifest. + let file_a = test_data_file("test/a.parquet", spec_id); + let file_b = test_data_file("test/b.parquet", spec_id); + let tx = Transaction::new(&table); + let action = tx + .fast_append() + .add_data_files(vec![file_a.clone(), file_b.clone()]); + let tx = action.apply(tx).unwrap(); + let table = tx.commit(&catalog).await.unwrap(); + + // First overwrite: delete A. Rewrites the shared manifest: A -> Deleted, B -> + // Existing (carried forward via add_existing_entry). + let tx = Transaction::new(&table); + let action = tx.overwrite().delete_data_files(vec![file_a.clone()]); + let tx = action.apply(tx).unwrap(); + let table = tx.commit(&catalog).await.unwrap(); + + // Second overwrite: delete B. This forces the *same* manifest (which already + // holds A as a Deleted tombstone) to be rewritten again. Before the fix, A's + // entry would fall into rewrite_manifest's `else` branch and go through + // add_existing_entry, which unconditionally resets status to Existing -- + // resurrecting A as live data. + let tx = Transaction::new(&table); + let action = tx.overwrite().delete_data_files(vec![file_b.clone()]); + let tx = action.apply(tx).unwrap(); + let table = tx.commit(&catalog).await.unwrap(); + + let snapshot = table.metadata().current_snapshot().unwrap(); + let manifest_list = table.manifest_list_reader(snapshot).load().await.unwrap(); + + let mut all_entries = vec![]; + for manifest_file in manifest_list.entries() { + let manifest = table.manifest_reader().read(manifest_file).await.unwrap(); + for entry in manifest.entries() { + all_entries.push((entry.status(), entry.file_path().to_string())); + } + } + + assert!( + all_entries.iter().any( + |(status, path)| *status == ManifestStatus::Deleted && path == "test/a.parquet" + ), + "File A must still be Deleted after the second overwrite, not resurrected: {all_entries:?}", + ); + assert!( + all_entries.iter().any( + |(status, path)| *status == ManifestStatus::Deleted && path == "test/b.parquet" + ), + "File B must be Deleted after being targeted by the second overwrite: {all_entries:?}", ); } } diff --git a/crates/iceberg/src/transaction/snapshot.rs b/crates/iceberg/src/transaction/snapshot.rs index aa9cf3818a..9292258e71 100644 --- a/crates/iceberg/src/transaction/snapshot.rs +++ b/crates/iceberg/src/transaction/snapshot.rs @@ -66,6 +66,17 @@ pub(crate) trait SnapshotProduceOperation: Send + Sync { /// which is stored in the snapshot metadata for tracking and auditing purposes. fn operation(&self) -> Operation; + /// Whether the snapshot summary should report a full-table truncate: computed + /// added/removed counts are discarded and every `total-*` property is replaced with + /// the previous snapshot's totals under `deleted-*`/`removed-*` (see + /// `truncate_table_summary`). This only makes sense for an operation that genuinely + /// replaces the entire table's contents; an operation with an explicit, partial set + /// of added/deleted files already has exact counts and must not have them overwritten + /// with the previous snapshot's totals. Defaults to `false`. + fn truncate_full_table(&self) -> bool { + false + } + /// Returns manifest entries that should be marked as deleted in the new snapshot. #[allow(unused)] fn delete_entries( @@ -125,6 +136,10 @@ impl<'a> SnapshotProducer<'a> { self.snapshot_id } + pub(crate) fn commit_uuid(&self) -> Uuid { + self.commit_uuid + } + pub(crate) fn new( table: &'a Table, commit_uuid: Uuid, @@ -442,7 +457,7 @@ impl<'a> SnapshotProducer<'a> { update_snapshot_summaries( summary, previous_snapshot.map(|s| s.summary()), - snapshot_produce_operation.operation() == Operation::Overwrite, + snapshot_produce_operation.truncate_full_table(), ) }