[jopdorp]

Which issue does this PR close?

Closes #3252.

What changes are included in this PR?

Producer-level changes that #2185 / #3300 need, split out so they can land on their own:

• SnapshotProducer takes the data files a commit deletes and the summary counts them like the added ones, each under its own partition spec, as pyiceberg's _SnapshotProducer._summary does. A total that would go negative is left out of the summary instead of written as 0, matching Java's updateTotal.
• truncate_full_table() on the operation keeps main's default (every overwrite resets the totals), so an operation that only replaces some files can opt out.
• The manifest counter is atomic and new_manifest_writer takes &self plus the schema and spec to write under, so an operation can rewrite a manifest under the spec it was written with.

The caller must provide matching deletion metrics; a debug assertion checks the deletion count against the data manifests written by the commit.

It also includes #3432 for the reported int/long panic in old-spec deletion summaries; that diff drops out once the fix merges.

No public API change.

Are these changes tested?

Unit tests cover the deleted-file summary, the omitted total, a manifest reading back with its given schema and spec, an unrecorded deletion failing the debug assertion, and the reported dropped-column partition cases. cargo test -p iceberg, the public API check, fmt and clippy with -D warnings are clean.

AI Disclosure

Drafted with an AI coding assistant, then reviewed and tested by us. Worth a close look: the summary trusts the caller's delete list, so an operation has to reject deletes it can't find (the overwrite in #3300 does).

[mkroll-db]

Thanks a lot. The changes look good. My only suggestion would be to add tests to cover:

• truncate_full_table with a partial overwrite because the cxurrent test only uses Operation::Delete
• checking that the manifest is produced before the summary (since the ordering changed)
• checking per-file (schema, spec) tuple

Proposed test diff

diff --git a/crates/iceberg/src/transaction/snapshot.rs b/crates/iceberg/src/transaction/snapshot.rs
index 1770e12f6..c1b4feb40 100644
--- a/crates/iceberg/src/transaction/snapshot.rs
+++ b/crates/iceberg/src/transaction/snapshot.rs
@@ -599,10 +599,11 @@ mod tests {
 
     use super::*;
     use crate::spec::{
-        DataContentType, DataFileBuilder, Literal, Manifest, NestedField, PartitionSpec,
-        PrimitiveType, Schema, Transform, Type,
+        DataContentType, DataFileBuilder, Literal, Manifest, ManifestStatus, NestedField,
+        PartitionSpec, PrimitiveType, Schema, Transform, Type,
     };
     use crate::transaction::tests::make_v2_minimal_table;
+    use crate::transaction::{Transaction, TransactionAction};
 
     /// An operation that adds no existing manifests and reports one removed data file.
     struct RemoveOneFileOperation {
@@ -630,6 +631,75 @@ mod tests {
         }
     }
 
+    struct RemoveExistingFileOperation {
+        file_path: String,
+        removed: Vec<(DataFile, SchemaRef, PartitionSpecRef)>,
+    }
+
+    impl SnapshotProduceOperation for RemoveExistingFileOperation {
+        fn operation(&self) -> Operation {
+            Operation::Overwrite
+        }
+
+        async fn delete_entries(&self, _: &SnapshotProducer<'_>) -> Result<Vec<ManifestEntry>> {
+            Ok(vec![])
+        }
+
+        async fn existing_manifest(
+            &mut self,
+            producer: &SnapshotProducer<'_>,
+        ) -> Result<Vec<ManifestFile>> {
+            let snapshot = producer.table.metadata().current_snapshot().unwrap();
+            let manifest_list = producer.table.manifest_list_reader(snapshot).load().await?;
+            // The initial append writes both files into one manifest.
+            assert_eq!(manifest_list.entries().len(), 1);
+
+            let manifest = producer
+                .table
+                .manifest_reader()
+                .read(&manifest_list.entries()[0])
+                .await?;
+            let schema = manifest.metadata().schema().clone();
+            let partition_spec = manifest.metadata().partition_spec().clone();
+            let mut writer = producer.new_manifest_writer(
+                ManifestContentType::Data,
+                schema.clone(),
+                partition_spec.clone(),
+            )?;
+
+            let mut removed = Vec::new();
+            for entry in manifest.entries() {
+                if entry.file_path() == self.file_path && entry.is_alive() {
+                    writer.add_delete_entry(entry.as_ref().clone())?;
+                    removed.push((
+                        entry.data_file().clone(),
+                        schema.clone(),
+                        Arc::new(partition_spec.clone()),
+                    ));
+                } else if entry.is_alive() {
+                    writer.add_existing_entry(entry.as_ref().clone())?;
+                }
+            }
+            assert_eq!(removed.len(), 1);
+
+            let rewritten_manifest = writer.write_manifest_file().await?;
+            self.removed.extend(removed);
+            Ok(vec![rewritten_manifest])
+        }
+
+        fn removed_data_files(&self) -> &[(DataFile, SchemaRef, PartitionSpecRef)] {
+            &self.removed
+        }
+    }
+
+    fn apply_commit(table: Table, mut commit: ActionCommit) -> Table {
+        let mut builder = table.metadata().clone().into_builder(None);
+        for update in commit.take_updates() {
+            builder = update.apply(builder).unwrap();
+        }
+        table.with_metadata(Arc::new(builder.build().unwrap().metadata))
+    }
+
     fn data_file(table: &Table, path: &str, record_count: u64) -> DataFile {
         DataFileBuilder::default()
             .content(DataContentType::Data)
@@ -677,6 +747,80 @@ mod tests {
         assert_eq!(properties.get("total-records").unwrap(), "1");
     }
 
+    #[tokio::test]
+    async fn test_partial_overwrite_reports_actual_removed_file() {
+        let table = make_v2_minimal_table();
+        let removed = data_file(&table, "test/removed.parquet", 2);
+        let kept = data_file(&table, "test/kept.parquet", 5);
+        let append = Transaction::new(&table)
+            .fast_append()
+            .add_data_files(vec![removed.clone(), kept]);
+        let initial_commit = Arc::new(append).commit(&table).await.unwrap();
+        let table = apply_commit(table, initial_commit);
+
+        let initial_summary = &table
+            .metadata()
+            .current_snapshot()
+            .unwrap()
+            .summary()
+            .additional_properties;
+        assert_eq!(initial_summary.get("total-data-files").unwrap(), "2");
+        assert_eq!(initial_summary.get("total-records").unwrap(), "7");
+
+        let mut replacement = data_file(&table, "test/replacement.parquet", 3);
+        replacement.file_size_in_bytes = 300;
+        let operation = RemoveExistingFileOperation {
+            file_path: removed.file_path.clone(),
+            removed: vec![],
+        };
+        let commit =
+            SnapshotProducer::new(&table, Uuid::now_v7(), HashMap::new(), vec![replacement])
+                .commit(operation, DefaultManifestProcess)
+                .await
+                .unwrap();
+        let table = apply_commit(table, commit);
+        let snapshot = table.metadata().current_snapshot().unwrap();
+        let properties = &snapshot.summary().additional_properties;
+
+        assert_eq!(snapshot.summary().operation, Operation::Overwrite);
+        assert_eq!(properties.get("deleted-data-files").unwrap(), "1");
+        assert_eq!(properties.get("deleted-records").unwrap(), "2");
+        assert_eq!(properties.get("added-data-files").unwrap(), "1");
+        assert_eq!(properties.get("added-records").unwrap(), "3");
+        assert_eq!(properties.get("removed-files-size").unwrap(), "100");
+        assert_eq!(properties.get("added-files-size").unwrap(), "300");
+        assert_eq!(properties.get("total-data-files").unwrap(), "2");
+        assert_eq!(properties.get("total-records").unwrap(), "8");
+        assert_eq!(properties.get("total-files-size").unwrap(), "400");
+
+        let manifest_list = table.manifest_list_reader(snapshot).load().await.unwrap();
+        assert_eq!(manifest_list.entries().len(), 2);
+        let mut statuses = HashMap::new();
+        for manifest_file in manifest_list.entries() {
+            let manifest = table.manifest_reader().read(manifest_file).await.unwrap();
+            for entry in manifest.entries() {
+                assert!(
+                    statuses
+                        .insert(entry.file_path().to_string(), entry.status())
+                        .is_none()
+                );
+            }
+        }
+        assert_eq!(statuses.len(), 3);
+        assert_eq!(
+            statuses.get("test/removed.parquet"),
+            Some(&ManifestStatus::Deleted)
+        );
+        assert_eq!(
+            statuses.get("test/kept.parquet"),
+            Some(&ManifestStatus::Existing)
+        );
+        assert_eq!(
+            statuses.get("test/replacement.parquet"),
+            Some(&ManifestStatus::Added)
+        );
+    }
+
     #[tokio::test]
     async fn test_new_manifest_writer_uses_the_given_schema_and_spec() {
         let table = make_v2_minimal_table();
@@ -731,4 +875,56 @@ mod tests {
         assert_eq!(metadata.partition_spec().spec_id(), 3);
         assert_eq!(first.partition_spec_id, 3);
     }
+
+    #[tokio::test]
+    async fn test_removed_file_summary_uses_the_reported_partition_spec() {
+        let table = make_v2_minimal_table();
+        let metadata = table
+            .metadata()
+            .clone()
+            .into_builder(None)
+            .set_properties(HashMap::from([(
+                TableProperties::PROPERTY_WRITE_PARTITION_SUMMARY_LIMIT.to_string(),
+                "10".to_string(),
+            )]))
+            .unwrap()
+            .build()
+            .unwrap()
+            .metadata;
+        let table = table.with_metadata(Arc::new(metadata));
+
+        let reported_schema = table.metadata().current_schema().clone();
+        let reported_spec = Arc::new(
+            PartitionSpec::builder(reported_schema.clone())
+                .with_spec_id(1)
+                .add_partition_field("y", "y", Transform::Identity)
+                .unwrap()
+                .build()
+                .unwrap(),
+        );
+
+        let added = data_file(&table, "test/added.parquet", 3);
+        let removed = data_file(&table, "test/removed.parquet", 2);
+        let operation = RemoveOneFileOperation {
+            removed: vec![(removed, reported_schema, reported_spec)],
+        };
+
+        let mut action_commit =
+            SnapshotProducer::new(&table, Uuid::now_v7(), HashMap::new(), vec![added])
+                .commit(operation, DefaultManifestProcess)
+                .await
+                .unwrap();
+
+        let updates = action_commit.take_updates();
+        let TableUpdate::AddSnapshot { snapshot } = &updates[0] else {
+            unreachable!()
+        };
+        let properties = &snapshot.summary().additional_properties;
+
+        let removed_partition = properties
+            .get("partitions.y=300")
+            .expect("removal must be summarized under the reported spec's partition path");
+        assert!(removed_partition.contains("deleted-data-files=1"));
+        assert!(properties.contains_key("partitions.x=300"));
+    }
 }

[jopdorp]

Thanks @mkroll-db, added your three tests as proposed, all green.

[laskoviymishka]

The ordering fix and the move off operation() == Overwrite to an explicit truncate_full_table() are the right calls — the partial-overwrite path reads cleanly now, and the test coverage is genuinely thorough.

What gives me pause is that this is step 1 of the COW series, so every operation built on top inherits these semantics — and two of them are contracts the type system doesn't hold. truncate_full_table() defaulting to false is correct for partial overwrites but silently wrong for a future full-table overwrite that returns Overwrite and forgets to override it: totals accumulate instead of resetting, with no repair path. And the summary's deleted-* counts are built entirely from removed_data_files() while the manifest's Deleted entries come from existing_manifest() — nothing binds them, and test_summary_counts_files_the_operation_reports_as_removed already commits a summary that disagrees with its manifests and passes. I'd rather we make both hard to get wrong here (a debug_assert on each, or return-type/enum changes) before the follow-on PRs build on the loose version.

Smaller, but worth folding into this pass: previous_total + added can still overflow before the saturating_sub, and saturating to 0 diverges from Java's omit-on-underflow — a Java reader treats the 0 as a real basis. Details inline, along with a couple of idiom nits (spec-by-value clone, the added-files clone on the commit path).

None of this is far off — the shape is right, it's mostly about locking the contracts down before they spread. Happy to take another look once the two invariants are enforced.

[jopdorp]

@laskoviymishka we reshaped this to follow pyiceberg and Java (the summary counts the producer's deleted files, main's truncation default stays), which removes both invariants instead of asserting them. Could you take another look?

[laskoviymishka]

Thanks for reshaping this — the round-1 line items all landed cleanly: saturating_add + omit-on-underflow in update_totals (with the test asserting the key is dropped), new_manifest_writer taking a PartitionSpecRef and cloning once inside, the restored mem::take on added files, and the dropped positional tuple. No need to revisit any of those.

Where I'm still holding is the two concerns from last round that moved but didn't close:

• Concern #2 (summary vs. manifests): dropping removed_data_files() removed the two-hook disagreement, but the summary still counts deleted_data_files while the Deleted manifest entries come from the operation independently, and nothing here reconciles them. The body conceding "the summary trusts the caller's delete list" is the trust gap I flagged last round, now explicit rather than enforced — and since total-* is cumulative, drift here poisons every later snapshot. For a primitive the whole COW series stacks on, I'd want the binding visible in this diff: either produce_manifests owns the Deleted entries so one list drives both, or keep summary-only with a loud doc contract plus a debug_assert! in commit().
• Concern #1 (truncate_full_table): inverting the default to opt-out fixed the original accumulate-forever risk but flipped it — a partial overwrite that returns Overwrite and forgets to override now silently truncates the totals. Main-parity is fine; I'd just pin the contract at the definition with a doc note or a debug_assert! so it can't be mis-declared silently.

One new thing this pass: the deleted-file summary loop pairs each file's historical spec with current_schema(), so partition_to_path's unwrap can panic on commit when an old spec references a dropped column — details inline, and a partitioned-fixture test would pin it alongside the unknown-spec-id branch.

This is close now — nail those down and I'm good.

[jopdorp]

@laskoviymishka the remaining fixes are pushed and the addressed threads are resolved. the library suite, fmt, clippy and public API check pass locally. could you take another look?

[laskoviymishka]

LGTM, ready to go. The truncate_full_table doc note, the deleted_data_files contract, and the debug_assert_eq! in produce_manifests are what I was after on the two contracts, so both threads are closed from my side. And thx @JeonDaehong for tracing the panic past the partition_type unwrap to Datum's Display — that was the right layer for the fix.

The inlines are optional follow-ups, nothing that should hold this: two more Unknown arms in Display for dropped decimal/uuid sources (fits naturally in #3432 or a small follow-up), a test for the unknown-spec-id branch, and a test rename.

[laskoviymishka]

Optional follow-up: this covers Int/Long literals only. partition_type gives Unknown for identity/truncate over a dropped column, so a dropped decimal or uuid source should end up as (Unknown, Int128) / (Unknown, UInt128), miss the Decimal/Uuid arms below and land on the unreachable!() — same commit-path panic, two more types. I haven't run it, this is from reading the arms.

Two arms before the catch-all plus decimal/uuid cases in test_partition_to_path_with_dropped_source_column would close it:

(PrimitiveType::Unknown, PrimitiveLiteral::Int128(val)) => write!(f, "{val}"),
(PrimitiveType::Unknown, PrimitiveLiteral::UInt128(val)) => {
    write!(f, "{}", uuid::Uuid::from_u128(*val))
}

Since this is #3432's change, that PR or a small follow-up is a fine home for it.

[JeonDaehong]

I ran this one: identity(decimal), truncate(decimal) and identity(uuid) on a dropped source all hit that unreachable!(). #3432 adds those two arms in b765bbf (uuid prints like Java, decimal prints the unscaled value since the scale is gone with the source).

[laskoviymishka]

Optional: this branch still has no test. A deleted DataFile with a partition_spec_id that isn't in the metadata, asserting summary() returns Err, would pin it. Happy for that to come in a follow-up.

[laskoviymishka]

nit, optional: _rejects_ reads like an Err path, but this pins a debug-only panic and isn't compiled under --release. Something like test_commit_panics_on_unrecorded_deletions_in_debug would make that obvious.

[JeonDaehong]

Thanks @laskoviymishka. For the record, the dropped decimal/uuid follow-up is now handled in #3432 (b765bbf).