From fed895fde70456a11c4b17c1ce244ab79d215326 Mon Sep 17 00:00:00 2001 From: Jegor van Opdorp Date: Fri, 18 Sep 2026 19:38:03 +0200 Subject: [PATCH 01/13] Let a snapshot operation report the files it removed The snapshot summary could only count what the caller asked to remove, so an operation that skips a path it never had, or is handed the same path twice, inflated deleted-data-files and deleted-records. SnapshotProduceOperation now reports the entries it actually marked Deleted, each with the schema and partition spec of the manifest that recorded it, and the summary counts those. Two changes follow from it. The manifests are produced before the summary, since the counts are only known once the operation has run, which means the added data files are cloned rather than taken. And the full-table truncate that drops the previous totals is gated on a new operation hook instead of on the operation being Overwrite: a partial overwrite keeps the totals, and with truncate off update_totals needs a saturating subtraction so removing more than the previous total does not underflow. --- crates/iceberg/src/spec/snapshot_summary.rs | 40 +++++- crates/iceberg/src/transaction/append.rs | 2 +- crates/iceberg/src/transaction/snapshot.rs | 135 +++++++++++++++++--- 3 files changed, 160 insertions(+), 17 deletions(-) diff --git a/crates/iceberg/src/spec/snapshot_summary.rs b/crates/iceberg/src/spec/snapshot_summary.rs index c9fd022724..105952957f 100644 --- a/crates/iceberg/src/spec/snapshot_summary.rs +++ b/crates/iceberg/src/spec/snapshot_summary.rs @@ -532,7 +532,7 @@ fn update_totals( return; }; - let new_total = previous_total + added - removed; + let new_total = (previous_total + added).saturating_sub(removed); summary .additional_properties .insert(total_property.to_string(), new_total.to_string()); @@ -1274,4 +1274,42 @@ mod tests { assert_eq!(props.get(TOTAL_POSITION_DELETES).unwrap(), "2"); assert_eq!(props.get(TOTAL_EQUALITY_DELETES).unwrap(), "1"); } + + #[test] + fn test_update_totals_saturates_when_removed_exceeds_previous_total() { + let prev_props: HashMap = [ + (TOTAL_DATA_FILES, "2"), + (TOTAL_RECORDS, "10"), + (TOTAL_FILE_SIZE, "100"), + ] + .into_iter() + .map(|(k, v)| (k.to_string(), v.to_string())) + .collect(); + + let previous_summary = Summary { + operation: Operation::Append, + additional_properties: prev_props, + }; + + let new_props: HashMap = [ + (DELETED_DATA_FILES, "3"), + (DELETED_RECORDS, "25"), + (REMOVED_FILE_SIZE, "250"), + ] + .into_iter() + .map(|(k, v)| (k.to_string(), v.to_string())) + .collect(); + + let summary = Summary { + operation: Operation::Delete, + additional_properties: new_props, + }; + + let updated = update_snapshot_summaries(summary, Some(&previous_summary), false).unwrap(); + let props = &updated.additional_properties; + + assert_eq!(props.get(TOTAL_DATA_FILES).unwrap(), "0"); + assert_eq!(props.get(TOTAL_RECORDS).unwrap(), "0"); + assert_eq!(props.get(TOTAL_FILE_SIZE).unwrap(), "0"); + } } diff --git a/crates/iceberg/src/transaction/append.rs b/crates/iceberg/src/transaction/append.rs index 872a1d89c1..3533c62d27 100644 --- a/crates/iceberg/src/transaction/append.rs +++ b/crates/iceberg/src/transaction/append.rs @@ -124,7 +124,7 @@ impl SnapshotProduceOperation for FastAppendOperation { } async fn existing_manifest( - &self, + &mut self, snapshot_produce: &SnapshotProducer<'_>, ) -> Result> { let Some(snapshot) = snapshot_produce.table.metadata().current_snapshot() else { diff --git a/crates/iceberg/src/transaction/snapshot.rs b/crates/iceberg/src/transaction/snapshot.rs index 1f09a79a08..62ef9c6d60 100644 --- a/crates/iceberg/src/transaction/snapshot.rs +++ b/crates/iceberg/src/transaction/snapshot.rs @@ -26,9 +26,10 @@ use uuid::Uuid; use crate::error::Result; use crate::spec::{ DataFile, DataFileFormat, FormatVersion, MAIN_BRANCH, ManifestContentType, ManifestEntry, - ManifestFile, ManifestListWriter, ManifestWriter, ManifestWriterBuilder, Operation, Snapshot, - SnapshotReference, SnapshotRetention, SnapshotSummaryCollector, Struct, StructType, Summary, - TableProperties, update_snapshot_summaries, + ManifestFile, ManifestListWriter, ManifestWriter, ManifestWriterBuilder, Operation, + PartitionSpecRef, SchemaRef, Snapshot, SnapshotReference, SnapshotRetention, + SnapshotSummaryCollector, Struct, StructType, Summary, TableProperties, + update_snapshot_summaries, }; use crate::table::Table; use crate::transaction::ActionCommit; @@ -82,9 +83,24 @@ pub(crate) trait SnapshotProduceOperation: Send + Sync { /// - **Overwrite operations**: May exclude manifests for partitions being overwritten /// - **Delete operations**: May exclude manifests for partitions being deleted fn existing_manifest( - &self, + &mut self, snapshot_produce: &SnapshotProducer<'_>, ) -> impl Future>> + Send; + + /// Returns the data files this operation actually removed, each with the schema and + /// partition spec of the manifest that recorded it. + /// + /// Only populated once [`Self::existing_manifest`] has run, so the snapshot summary counts + /// what was really removed rather than what the caller asked to remove. + fn removed_data_files(&self) -> &[(DataFile, SchemaRef, PartitionSpecRef)] { + &[] + } + + /// Returns whether this operation replaces the whole table, dropping the previous totals + /// from the snapshot summary. A partial overwrite must return `false`. + fn truncate_full_table(&self) -> bool { + false + } } pub(crate) struct DefaultManifestProcess; @@ -308,7 +324,9 @@ impl<'a> SnapshotProducer<'a> { // Write manifest file for added data files and return the ManifestFile for ManifestList. async fn write_added_manifest(&mut self) -> Result { - let added_data_files = std::mem::take(&mut self.added_data_files); + // Cloned rather than taken: the summary is built after the manifests, so the added + // files must still be here. + let added_data_files = self.added_data_files.clone(); if added_data_files.is_empty() { return Err(Error::new( ErrorKind::PreconditionFailed, @@ -341,7 +359,7 @@ impl<'a> SnapshotProducer<'a> { /// and collects all of the manifests to be included in the new snapshot as [ManifestFile] entries. async fn produce_manifests( &mut self, - snapshot_produce_operation: &OP, + snapshot_produce_operation: &mut OP, manifest_process: &MP, ) -> Result> { // Assert current snapshot producer contains new content to add to new snapshot. @@ -403,6 +421,10 @@ impl<'a> SnapshotProducer<'a> { ); } + for (data_file, schema, partition_spec) in snapshot_produce_operation.removed_data_files() { + summary_collector.remove_file(data_file, schema.clone(), partition_spec.clone()); + } + let previous_snapshot = table_metadata.current_snapshot(); // User-supplied snapshot properties are applied first, then the computed @@ -421,7 +443,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(), ) } @@ -439,7 +461,7 @@ impl<'a> SnapshotProducer<'a> { /// Finished building the action and return the [`ActionCommit`] to the transaction. pub(crate) async fn commit( mut self, - snapshot_produce_operation: OP, + mut snapshot_produce_operation: OP, process: MP, ) -> Result { let manifest_list_path = self.generate_manifest_list_file_path(0)?; @@ -479,17 +501,16 @@ impl<'a> SnapshotProducer<'a> { ), }; - // Calling self.summary() before self.produce_manifests() is important because self.added_data_files - // will be set to an empty vec after self.produce_manifests() returns, resulting in an empty summary - // being generated. + // The manifests are produced first so the summary can count the entries the operation + // actually marked Deleted. + let new_manifests = self + .produce_manifests(&mut snapshot_produce_operation, &process) + .await?; + let summary = self.summary(&snapshot_produce_operation).map_err(|err| { Error::new(ErrorKind::Unexpected, "Failed to create snapshot summary.").with_source(err) })?; - let new_manifests = self - .produce_manifests(&snapshot_produce_operation, &process) - .await?; - manifest_list_writer.add_manifests(new_manifests.into_iter())?; let writer_next_row_id = manifest_list_writer.next_row_id(); manifest_list_writer.close().await?; @@ -556,3 +577,87 @@ impl<'a> SnapshotProducer<'a> { Ok(ActionCommit::new(updates, requirements)) } } + +#[cfg(test)] +mod tests { + use std::collections::HashMap; + + use uuid::Uuid; + + use super::*; + use crate::spec::{DataContentType, DataFileBuilder, Literal}; + use crate::transaction::tests::make_v2_minimal_table; + + /// An operation that adds no existing manifests and reports one removed data file. + struct RemoveOneFileOperation { + removed: Vec<(DataFile, SchemaRef, PartitionSpecRef)>, + } + + impl SnapshotProduceOperation for RemoveOneFileOperation { + fn operation(&self) -> Operation { + Operation::Delete + } + + async fn delete_entries(&self, _: &SnapshotProducer<'_>) -> Result> { + Ok(vec![]) + } + + async fn existing_manifest( + &mut self, + _: &SnapshotProducer<'_>, + ) -> Result> { + Ok(vec![]) + } + + fn removed_data_files(&self) -> &[(DataFile, SchemaRef, PartitionSpecRef)] { + &self.removed + } + } + + fn data_file(table: &Table, path: &str, record_count: u64) -> DataFile { + DataFileBuilder::default() + .content(DataContentType::Data) + .file_path(path.to_string()) + .file_format(DataFileFormat::Parquet) + .file_size_in_bytes(100) + .record_count(record_count) + .partition_spec_id(table.metadata().default_partition_spec_id()) + .partition(Struct::from_iter([Some(Literal::long(300))])) + .build() + .unwrap() + } + + #[tokio::test] + async fn test_summary_counts_files_the_operation_reports_as_removed() { + let table = make_v2_minimal_table(); + let added = data_file(&table, "test/added.parquet", 3); + let removed = data_file(&table, "test/removed.parquet", 2); + + let operation = RemoveOneFileOperation { + removed: vec![( + removed, + table.metadata().current_schema().clone(), + table.metadata().default_partition_spec().clone(), + )], + }; + + 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; + + 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("total-data-files").unwrap(), "0"); + assert_eq!(properties.get("total-records").unwrap(), "1"); + } +} From 3fc4e80621568c9f5601cb9e7b7003fe6d2eb053 Mon Sep 17 00:00:00 2001 From: Jegor van Opdorp Date: Fri, 18 Sep 2026 19:38:47 +0200 Subject: [PATCH 02/13] Keep manifest names unique across a commit The manifest name is built inline in new_manifest_writer from the metadata location, the commit uuid and a counter, so anything else that needs to write a manifest for the same commit has to reinvent it, and a name invented elsewhere is neither unique within the commit nor findable by orphan cleanup after a failed one. Pull the name out into new_manifest_path, and make the counter an AtomicU64 so it can be bumped behind a shared reference. --- crates/iceberg/src/transaction/snapshot.rs | 20 ++++++++++++-------- 1 file changed, 12 insertions(+), 8 deletions(-) diff --git a/crates/iceberg/src/transaction/snapshot.rs b/crates/iceberg/src/transaction/snapshot.rs index 62ef9c6d60..0347257290 100644 --- a/crates/iceberg/src/transaction/snapshot.rs +++ b/crates/iceberg/src/transaction/snapshot.rs @@ -17,7 +17,7 @@ use std::collections::{HashMap, HashSet}; use std::future::Future; -use std::ops::RangeFrom; +use std::sync::atomic::{AtomicU64, Ordering}; use futures::TryStreamExt; use futures::stream::FuturesUnordered; @@ -132,7 +132,7 @@ pub(crate) struct SnapshotProducer<'a> { // A counter used to generate unique manifest file names. // It starts from 0 and increments for each new manifest file. // Note: This counter is limited to the range of (0..u64::MAX). - manifest_counter: RangeFrom, + manifest_counter: AtomicU64, } impl<'a> SnapshotProducer<'a> { @@ -148,7 +148,7 @@ impl<'a> SnapshotProducer<'a> { commit_uuid, snapshot_properties, added_data_files, - manifest_counter: (0..), + manifest_counter: AtomicU64::new(0), } } @@ -250,15 +250,19 @@ impl<'a> SnapshotProducer<'a> { snapshot_id } - fn new_manifest_writer(&mut self, content: ManifestContentType) -> Result { - let new_manifest_path = format!( + /// Returns the path for the next manifest file of this commit. + fn new_manifest_path(&self) -> Result { + Ok(format!( "{}/{}-m{}.{}", self.table.metadata().metadata_location()?, self.commit_uuid, - self.manifest_counter.next().unwrap(), + self.manifest_counter.fetch_add(1, Ordering::Relaxed), DataFileFormat::Avro - ); - let output_file = self.table.file_io().new_output(new_manifest_path)?; + )) + } + + fn new_manifest_writer(&mut self, content: ManifestContentType) -> Result { + let output_file = self.table.file_io().new_output(self.new_manifest_path()?)?; let partition_spec = self .table .metadata() From 6325d2270424c92fb2f9734721de4ff91708c86e Mon Sep 17 00:00:00 2001 From: Jegor van Opdorp Date: Fri, 18 Sep 2026 19:41:28 +0200 Subject: [PATCH 03/13] Write a manifest under the schema and spec it was written with new_manifest_writer hardcoded the table's current schema and default partition spec, so an operation that rewrites an existing manifest could not keep the schema and spec that manifest's entries were recorded under, and building a writer by hand instead means losing the encryption manager the method already branches on: on an encrypted table the manifest would go out in plaintext while its key metadata said otherwise. The schema and partition spec become arguments, and the added-files path passes the table's current ones. --- crates/iceberg/src/transaction/snapshot.rs | 89 +++++++++++++++++++--- 1 file changed, 78 insertions(+), 11 deletions(-) diff --git a/crates/iceberg/src/transaction/snapshot.rs b/crates/iceberg/src/transaction/snapshot.rs index 0347257290..1770e12f6c 100644 --- a/crates/iceberg/src/transaction/snapshot.rs +++ b/crates/iceberg/src/transaction/snapshot.rs @@ -27,7 +27,7 @@ use crate::error::Result; use crate::spec::{ DataFile, DataFileFormat, FormatVersion, MAIN_BRANCH, ManifestContentType, ManifestEntry, ManifestFile, ManifestListWriter, ManifestWriter, ManifestWriterBuilder, Operation, - PartitionSpecRef, SchemaRef, Snapshot, SnapshotReference, SnapshotRetention, + PartitionSpec, PartitionSpecRef, SchemaRef, Snapshot, SnapshotReference, SnapshotRetention, SnapshotSummaryCollector, Struct, StructType, Summary, TableProperties, update_snapshot_summaries, }; @@ -261,15 +261,15 @@ impl<'a> SnapshotProducer<'a> { )) } - fn new_manifest_writer(&mut self, content: ManifestContentType) -> Result { + /// Returns a writer for the next manifest file of this commit, routed through the table's + /// encryption manager when one is configured. + pub(crate) fn new_manifest_writer( + &self, + content: ManifestContentType, + schema: SchemaRef, + partition_spec: PartitionSpec, + ) -> Result { let output_file = self.table.file_io().new_output(self.new_manifest_path()?)?; - let partition_spec = self - .table - .metadata() - .default_partition_spec() - .as_ref() - .clone(); - let schema = self.table.metadata().current_schema().clone(); let builder = if let Some(em) = self.table.encryption_manager() { ManifestWriterBuilder::new_from_encrypted( @@ -352,7 +352,15 @@ impl<'a> SnapshotProducer<'a> { builder.build() } }); - let mut writer = self.new_manifest_writer(ManifestContentType::Data)?; + let mut writer = self.new_manifest_writer( + ManifestContentType::Data, + self.table.metadata().current_schema().clone(), + self.table + .metadata() + .default_partition_spec() + .as_ref() + .clone(), + )?; for entry in manifest_entries { writer.add_entry(entry)?; } @@ -585,11 +593,15 @@ impl<'a> SnapshotProducer<'a> { #[cfg(test)] mod tests { use std::collections::HashMap; + use std::sync::Arc; use uuid::Uuid; use super::*; - use crate::spec::{DataContentType, DataFileBuilder, Literal}; + use crate::spec::{ + DataContentType, DataFileBuilder, Literal, Manifest, NestedField, PartitionSpec, + PrimitiveType, Schema, Transform, Type, + }; use crate::transaction::tests::make_v2_minimal_table; /// An operation that adds no existing manifests and reports one removed data file. @@ -664,4 +676,59 @@ mod tests { assert_eq!(properties.get("total-data-files").unwrap(), "0"); assert_eq!(properties.get("total-records").unwrap(), "1"); } + + #[tokio::test] + async fn test_new_manifest_writer_uses_the_given_schema_and_spec() { + let table = make_v2_minimal_table(); + let schema = Arc::new( + Schema::builder() + .with_schema_id(7) + .with_fields(vec![ + NestedField::required(1, "a", Type::Primitive(PrimitiveType::Long)).into(), + ]) + .build() + .unwrap(), + ); + let partition_spec = PartitionSpec::builder(schema.clone()) + .with_spec_id(3) + .add_partition_field("a", "a", Transform::Identity) + .unwrap() + .build() + .unwrap(); + + let producer = SnapshotProducer::new(&table, Uuid::now_v7(), HashMap::new(), vec![]); + let first = producer + .new_manifest_writer( + ManifestContentType::Data, + schema.clone(), + partition_spec.clone(), + ) + .unwrap() + .write_manifest_file() + .await + .unwrap(); + let second = producer + .new_manifest_writer(ManifestContentType::Data, schema.clone(), partition_spec) + .unwrap() + .write_manifest_file() + .await + .unwrap(); + + assert_ne!(first.manifest_path, second.manifest_path); + + let bytes = table + .file_io() + .new_input(&first.manifest_path) + .unwrap() + .read() + .await + .unwrap(); + let metadata = Manifest::parse_avro(&bytes).unwrap().into_parts().1; + + // Not the table's defaults, which are schema 0 and spec 0. + assert_eq!(metadata.schema_id(), 7); + assert_eq!(metadata.schema().as_ref(), schema.as_ref()); + assert_eq!(metadata.partition_spec().spec_id(), 3); + assert_eq!(first.partition_spec_id, 3); + } } From 1cad7313308ad4bc7168d4111b12aa921a6379fc Mon Sep 17 00:00:00 2001 From: Jegor van Opdorp Date: Tue, 29 Sep 2026 21:55:37 +0200 Subject: [PATCH 04/13] test(transaction): cover partial overwrite and per-file spec in snapshot summary Co-authored-by: Marco Kroll <216147982+mkroll-db@users.noreply.github.com> --- crates/iceberg/src/transaction/snapshot.rs | 200 ++++++++++++++++++++- 1 file changed, 198 insertions(+), 2 deletions(-) diff --git a/crates/iceberg/src/transaction/snapshot.rs b/crates/iceberg/src/transaction/snapshot.rs index 1770e12f6c..c1b4feb40b 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> { + Ok(vec![]) + } + + async fn existing_manifest( + &mut self, + producer: &SnapshotProducer<'_>, + ) -> Result> { + 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")); + } } From d5623e3c29b10db073fe60b48a39da32dd44423b Mon Sep 17 00:00:00 2001 From: Jegor van Opdorp Date: Fri, 2 Oct 2026 20:20:33 +0200 Subject: [PATCH 05/13] Drop the full-table truncation from the snapshot summary Java and PyIceberg derive totals from the files an operation marks deleted, with no truncation path. --- crates/iceberg/src/spec/snapshot_summary.rs | 283 +------------------- crates/iceberg/src/transaction/snapshot.rs | 12 +- 2 files changed, 8 insertions(+), 287 deletions(-) diff --git a/crates/iceberg/src/spec/snapshot_summary.rs b/crates/iceberg/src/spec/snapshot_summary.rs index 105952957f..b13b35bf2a 100644 --- a/crates/iceberg/src/spec/snapshot_summary.rs +++ b/crates/iceberg/src/spec/snapshot_summary.rs @@ -330,9 +330,8 @@ where T: PartialOrd + Default + ToString { } pub(crate) fn update_snapshot_summaries( - summary: Summary, + mut summary: Summary, previous_summary: Option<&Summary>, - truncate_full_table: bool, ) -> Result { // Validate that the operation is supported if summary.operation != Operation::Append @@ -345,15 +344,6 @@ pub(crate) fn update_snapshot_summaries( )); } - let mut summary = match previous_summary { - Some(prev_summary) if truncate_full_table && summary.operation == Operation::Overwrite => { - truncate_table_summary(summary, prev_summary).map_err(|err| { - Error::new(ErrorKind::Unexpected, "Failed to truncate table summary.") - .with_source(err) - })? - } - _ => summary, - }; update_totals( &mut summary, @@ -405,77 +395,6 @@ pub(crate) fn update_snapshot_summaries( Ok(summary) } -fn get_prop(previous_summary: &Summary, prop: &str) -> Result { - let value_str = previous_summary - .additional_properties - .get(prop) - .map(String::as_str) - .unwrap_or("0"); - value_str.parse::().map_err(|err| { - Error::new( - ErrorKind::Unexpected, - format!("Failed to parse summary property '{prop}' value '{value_str}' as u64."), - ) - .with_source(err) - }) -} - -fn truncate_table_summary(mut summary: Summary, previous_summary: &Summary) -> Result { - for prop in [ - TOTAL_DATA_FILES, - TOTAL_DELETE_FILES, - TOTAL_RECORDS, - TOTAL_FILE_SIZE, - TOTAL_POSITION_DELETES, - TOTAL_EQUALITY_DELETES, - ] { - summary - .additional_properties - .insert(prop.to_string(), "0".to_string()); - } - - let value = get_prop(previous_summary, TOTAL_DATA_FILES)?; - if value != 0 { - summary - .additional_properties - .insert(DELETED_DATA_FILES.to_string(), value.to_string()); - } - let value = get_prop(previous_summary, TOTAL_DELETE_FILES)?; - if value != 0 { - summary - .additional_properties - .insert(REMOVED_DELETE_FILES.to_string(), value.to_string()); - } - let value = get_prop(previous_summary, TOTAL_RECORDS)?; - if value != 0 { - summary - .additional_properties - .insert(DELETED_RECORDS.to_string(), value.to_string()); - } - let value = get_prop(previous_summary, TOTAL_FILE_SIZE)?; - if value != 0 { - summary - .additional_properties - .insert(REMOVED_FILE_SIZE.to_string(), value.to_string()); - } - - let value = get_prop(previous_summary, TOTAL_POSITION_DELETES)?; - if value != 0 { - summary - .additional_properties - .insert(REMOVED_POSITION_DELETES.to_string(), value.to_string()); - } - - let value = get_prop(previous_summary, TOTAL_EQUALITY_DELETES)?; - if value != 0 { - summary - .additional_properties - .insert(REMOVED_EQUALITY_DELETES.to_string(), value.to_string()); - } - - Ok(summary) -} - fn update_totals( summary: &mut Summary, previous_summary: Option<&Summary>, @@ -589,7 +508,7 @@ mod tests { additional_properties: new_props, }; - let updated = update_snapshot_summaries(summary, Some(&previous_summary), false).unwrap(); + let updated = update_snapshot_summaries(summary, Some(&previous_summary)).unwrap(); assert_eq!( updated.additional_properties.get(TOTAL_DATA_FILES).unwrap(), @@ -626,194 +545,6 @@ mod tests { ); } - #[test] - fn test_truncate_table_summary() { - let prev_props: HashMap = [ - (TOTAL_DATA_FILES.to_string(), "10".to_string()), - (TOTAL_DELETE_FILES.to_string(), "5".to_string()), - (TOTAL_RECORDS.to_string(), "100".to_string()), - (TOTAL_FILE_SIZE.to_string(), "1000".to_string()), - (TOTAL_POSITION_DELETES.to_string(), "3".to_string()), - (TOTAL_EQUALITY_DELETES.to_string(), "2".to_string()), - ] - .into_iter() - .collect(); - - let previous_summary = Summary { - operation: Operation::Overwrite, - additional_properties: prev_props, - }; - - let mut new_props = HashMap::new(); - new_props.insert("dummy".to_string(), "value".to_string()); - let summary = Summary { - operation: Operation::Overwrite, - additional_properties: new_props, - }; - - let truncated = truncate_table_summary(summary, &previous_summary).unwrap(); - - assert_eq!( - truncated - .additional_properties - .get(TOTAL_DATA_FILES) - .unwrap(), - "0" - ); - assert_eq!( - truncated - .additional_properties - .get(TOTAL_DELETE_FILES) - .unwrap(), - "0" - ); - assert_eq!( - truncated.additional_properties.get(TOTAL_RECORDS).unwrap(), - "0" - ); - assert_eq!( - truncated - .additional_properties - .get(TOTAL_FILE_SIZE) - .unwrap(), - "0" - ); - assert_eq!( - truncated - .additional_properties - .get(TOTAL_POSITION_DELETES) - .unwrap(), - "0" - ); - assert_eq!( - truncated - .additional_properties - .get(TOTAL_EQUALITY_DELETES) - .unwrap(), - "0" - ); - - assert_eq!( - truncated - .additional_properties - .get(DELETED_DATA_FILES) - .unwrap(), - "10" - ); - assert_eq!( - truncated - .additional_properties - .get(REMOVED_DELETE_FILES) - .unwrap(), - "5" - ); - assert_eq!( - truncated - .additional_properties - .get(DELETED_RECORDS) - .unwrap(), - "100" - ); - assert_eq!( - truncated - .additional_properties - .get(REMOVED_FILE_SIZE) - .unwrap(), - "1000" - ); - assert_eq!( - truncated - .additional_properties - .get(REMOVED_POSITION_DELETES) - .unwrap(), - "3" - ); - assert_eq!( - truncated - .additional_properties - .get(REMOVED_EQUALITY_DELETES) - .unwrap(), - "2" - ); - } - - #[test] - fn test_update_snapshot_summaries_overwrite_truncate_handles_totals_above_i32_max() { - // A table can legitimately accumulate more than i32::MAX rows or files - // over its lifetime. Truncating such a table on overwrite must succeed - // and surface the previous totals into the deleted-* counters. - let big = (i32::MAX as u64 + 1).to_string(); // 2_147_483_648 - let prev_props: HashMap = [ - (TOTAL_DATA_FILES.to_string(), big.clone()), - (TOTAL_DELETE_FILES.to_string(), "0".to_string()), - (TOTAL_RECORDS.to_string(), big.clone()), - (TOTAL_FILE_SIZE.to_string(), "0".to_string()), - (TOTAL_POSITION_DELETES.to_string(), "0".to_string()), - (TOTAL_EQUALITY_DELETES.to_string(), "0".to_string()), - ] - .into_iter() - .collect(); - - let previous_summary = Summary { - operation: Operation::Overwrite, - additional_properties: prev_props, - }; - - let summary = Summary { - operation: Operation::Overwrite, - additional_properties: HashMap::new(), - }; - - let updated = update_snapshot_summaries(summary, Some(&previous_summary), true) - .expect("overwrite truncation should accept totals above i32::MAX"); - assert_eq!( - updated - .additional_properties - .get(DELETED_DATA_FILES) - .unwrap(), - &big - ); - assert_eq!( - updated.additional_properties.get(DELETED_RECORDS).unwrap(), - &big - ); - } - - #[test] - fn test_update_snapshot_summaries_overwrite_truncate_returns_err_on_malformed_total() { - // Non-numeric values in the previous summary (corruption, manual edits, - // a foreign implementation) must surface as a recoverable Err - not - // crash the process. - let prev_props: HashMap = [ - (TOTAL_DATA_FILES.to_string(), "not_a_number".to_string()), - (TOTAL_DELETE_FILES.to_string(), "0".to_string()), - (TOTAL_RECORDS.to_string(), "0".to_string()), - (TOTAL_FILE_SIZE.to_string(), "0".to_string()), - (TOTAL_POSITION_DELETES.to_string(), "0".to_string()), - (TOTAL_EQUALITY_DELETES.to_string(), "0".to_string()), - ] - .into_iter() - .collect(); - - let previous_summary = Summary { - operation: Operation::Overwrite, - additional_properties: prev_props, - }; - - let summary = Summary { - operation: Operation::Overwrite, - additional_properties: HashMap::new(), - }; - - let err = update_snapshot_summaries(summary, Some(&previous_summary), true) - .expect_err("malformed previous summary must produce an Err, not a panic"); - assert!( - err.message().contains("truncate table summary"), - "expected wrapped 'Failed to truncate table summary' context, got: {}", - err.message() - ); - } - #[test] fn test_snapshot_summary_collector_build() { let schema = Arc::new( @@ -1146,7 +877,7 @@ mod tests { additional_properties: new_props, }; - let updated = update_snapshot_summaries(summary, Some(&previous_summary), false).unwrap(); + let updated = update_snapshot_summaries(summary, Some(&previous_summary)).unwrap(); let props = &updated.additional_properties; assert_eq!(props.get(TOTAL_DATA_FILES).unwrap(), "12"); @@ -1192,7 +923,7 @@ mod tests { }; // Must not panic. - let updated = update_snapshot_summaries(summary, Some(&previous_summary), false).unwrap(); + let updated = update_snapshot_summaries(summary, Some(&previous_summary)).unwrap(); let props = &updated.additional_properties; // The total whose added delta was unparsable is skipped... @@ -1220,7 +951,7 @@ mod tests { additional_properties: new_props, }; - let updated = update_snapshot_summaries(summary, None, false).unwrap(); + let updated = update_snapshot_summaries(summary, None).unwrap(); let props = &updated.additional_properties; assert_eq!(props.get(TOTAL_DATA_FILES).unwrap(), "4"); @@ -1264,7 +995,7 @@ mod tests { additional_properties: new_props, }; - let updated = update_snapshot_summaries(summary, Some(&previous_summary), false).unwrap(); + let updated = update_snapshot_summaries(summary, Some(&previous_summary)).unwrap(); let props = &updated.additional_properties; assert_eq!(props.get(TOTAL_DATA_FILES).unwrap(), "8"); @@ -1305,7 +1036,7 @@ mod tests { additional_properties: new_props, }; - let updated = update_snapshot_summaries(summary, Some(&previous_summary), false).unwrap(); + let updated = update_snapshot_summaries(summary, Some(&previous_summary)).unwrap(); let props = &updated.additional_properties; assert_eq!(props.get(TOTAL_DATA_FILES).unwrap(), "0"); diff --git a/crates/iceberg/src/transaction/snapshot.rs b/crates/iceberg/src/transaction/snapshot.rs index c1b4feb40b..3aeb2af39a 100644 --- a/crates/iceberg/src/transaction/snapshot.rs +++ b/crates/iceberg/src/transaction/snapshot.rs @@ -95,12 +95,6 @@ pub(crate) trait SnapshotProduceOperation: Send + Sync { fn removed_data_files(&self) -> &[(DataFile, SchemaRef, PartitionSpecRef)] { &[] } - - /// Returns whether this operation replaces the whole table, dropping the previous totals - /// from the snapshot summary. A partial overwrite must return `false`. - fn truncate_full_table(&self) -> bool { - false - } } pub(crate) struct DefaultManifestProcess; @@ -452,11 +446,7 @@ impl<'a> SnapshotProducer<'a> { additional_properties, }; - update_snapshot_summaries( - summary, - previous_snapshot.map(|s| s.summary()), - snapshot_produce_operation.truncate_full_table(), - ) + update_snapshot_summaries(summary, previous_snapshot.map(|s| s.summary())) } fn generate_manifest_list_file_path(&self, attempt: i64) -> Result { From 296532e6ba26caf89334ce1bcbacd0872674abc3 Mon Sep 17 00:00:00 2001 From: Jegor van Opdorp Date: Fri, 2 Oct 2026 20:20:41 +0200 Subject: [PATCH 06/13] Omit a summary total that would go negative, as Java does --- crates/iceberg/src/spec/snapshot_summary.rs | 13 ++++++++----- 1 file changed, 8 insertions(+), 5 deletions(-) diff --git a/crates/iceberg/src/spec/snapshot_summary.rs b/crates/iceberg/src/spec/snapshot_summary.rs index b13b35bf2a..89af2081ba 100644 --- a/crates/iceberg/src/spec/snapshot_summary.rs +++ b/crates/iceberg/src/spec/snapshot_summary.rs @@ -451,7 +451,10 @@ fn update_totals( return; }; - let new_total = (previous_total + added).saturating_sub(removed); + // Like Java, a total that would go negative is omitted rather than written as a false basis. + let Some(new_total) = previous_total.saturating_add(added).checked_sub(removed) else { + return; + }; summary .additional_properties .insert(total_property.to_string(), new_total.to_string()); @@ -1007,7 +1010,7 @@ mod tests { } #[test] - fn test_update_totals_saturates_when_removed_exceeds_previous_total() { + fn test_update_totals_omits_total_when_removed_exceeds_previous_total() { let prev_props: HashMap = [ (TOTAL_DATA_FILES, "2"), (TOTAL_RECORDS, "10"), @@ -1039,8 +1042,8 @@ mod tests { let updated = update_snapshot_summaries(summary, Some(&previous_summary)).unwrap(); let props = &updated.additional_properties; - assert_eq!(props.get(TOTAL_DATA_FILES).unwrap(), "0"); - assert_eq!(props.get(TOTAL_RECORDS).unwrap(), "0"); - assert_eq!(props.get(TOTAL_FILE_SIZE).unwrap(), "0"); + assert!(!props.contains_key(TOTAL_DATA_FILES)); + assert!(!props.contains_key(TOTAL_RECORDS)); + assert!(!props.contains_key(TOTAL_FILE_SIZE)); } } From eeff802555161e76e8a8bfb7b2027307f3d005a1 Mon Sep 17 00:00:00 2001 From: Jegor van Opdorp Date: Fri, 2 Oct 2026 20:21:11 +0200 Subject: [PATCH 07/13] Name the removed-file fields and take the manifest spec by ref --- crates/iceberg/src/transaction/snapshot.rs | 82 +++++++++++++--------- 1 file changed, 48 insertions(+), 34 deletions(-) diff --git a/crates/iceberg/src/transaction/snapshot.rs b/crates/iceberg/src/transaction/snapshot.rs index 3aeb2af39a..43345fdeb2 100644 --- a/crates/iceberg/src/transaction/snapshot.rs +++ b/crates/iceberg/src/transaction/snapshot.rs @@ -27,7 +27,7 @@ use crate::error::Result; use crate::spec::{ DataFile, DataFileFormat, FormatVersion, MAIN_BRANCH, ManifestContentType, ManifestEntry, ManifestFile, ManifestListWriter, ManifestWriter, ManifestWriterBuilder, Operation, - PartitionSpec, PartitionSpecRef, SchemaRef, Snapshot, SnapshotReference, SnapshotRetention, + PartitionSpecRef, SchemaRef, Snapshot, SnapshotReference, SnapshotRetention, SnapshotSummaryCollector, Struct, StructType, Summary, TableProperties, update_snapshot_summaries, }; @@ -87,16 +87,23 @@ pub(crate) trait SnapshotProduceOperation: Send + Sync { snapshot_produce: &SnapshotProducer<'_>, ) -> impl Future>> + Send; - /// Returns the data files this operation actually removed, each with the schema and - /// partition spec of the manifest that recorded it. + /// Returns the data files this operation actually removed. /// /// Only populated once [`Self::existing_manifest`] has run, so the snapshot summary counts /// what was really removed rather than what the caller asked to remove. - fn removed_data_files(&self) -> &[(DataFile, SchemaRef, PartitionSpecRef)] { + fn removed_data_files(&self) -> &[RemovedDataFile] { &[] } } +/// A data file an operation removed, with the schema and partition spec of the manifest that +/// recorded it. +pub(crate) struct RemovedDataFile { + pub(crate) data_file: DataFile, + pub(crate) schema: SchemaRef, + pub(crate) partition_spec: PartitionSpecRef, +} + pub(crate) struct DefaultManifestProcess; impl ManifestProcess for DefaultManifestProcess { @@ -261,9 +268,10 @@ impl<'a> SnapshotProducer<'a> { &self, content: ManifestContentType, schema: SchemaRef, - partition_spec: PartitionSpec, + partition_spec: PartitionSpecRef, ) -> Result { let output_file = self.table.file_io().new_output(self.new_manifest_path()?)?; + let partition_spec = partition_spec.as_ref().clone(); let builder = if let Some(em) = self.table.encryption_manager() { ManifestWriterBuilder::new_from_encrypted( @@ -349,11 +357,7 @@ impl<'a> SnapshotProducer<'a> { let mut writer = self.new_manifest_writer( ManifestContentType::Data, self.table.metadata().current_schema().clone(), - self.table - .metadata() - .default_partition_spec() - .as_ref() - .clone(), + self.table.metadata().default_partition_spec().clone(), )?; for entry in manifest_entries { writer.add_entry(entry)?; @@ -427,8 +431,12 @@ impl<'a> SnapshotProducer<'a> { ); } - for (data_file, schema, partition_spec) in snapshot_produce_operation.removed_data_files() { - summary_collector.remove_file(data_file, schema.clone(), partition_spec.clone()); + for removed in snapshot_produce_operation.removed_data_files() { + summary_collector.remove_file( + &removed.data_file, + removed.schema.clone(), + removed.partition_spec.clone(), + ); } let previous_snapshot = table_metadata.current_snapshot(); @@ -597,7 +605,7 @@ mod tests { /// An operation that adds no existing manifests and reports one removed data file. struct RemoveOneFileOperation { - removed: Vec<(DataFile, SchemaRef, PartitionSpecRef)>, + removed: Vec, } impl SnapshotProduceOperation for RemoveOneFileOperation { @@ -616,14 +624,14 @@ mod tests { Ok(vec![]) } - fn removed_data_files(&self) -> &[(DataFile, SchemaRef, PartitionSpecRef)] { + fn removed_data_files(&self) -> &[RemovedDataFile] { &self.removed } } struct RemoveExistingFileOperation { file_path: String, - removed: Vec<(DataFile, SchemaRef, PartitionSpecRef)>, + removed: Vec, } impl SnapshotProduceOperation for RemoveExistingFileOperation { @@ -650,7 +658,7 @@ mod tests { .read(&manifest_list.entries()[0]) .await?; let schema = manifest.metadata().schema().clone(); - let partition_spec = manifest.metadata().partition_spec().clone(); + let partition_spec = Arc::new(manifest.metadata().partition_spec().clone()); let mut writer = producer.new_manifest_writer( ManifestContentType::Data, schema.clone(), @@ -661,11 +669,11 @@ mod tests { 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()), - )); + removed.push(RemovedDataFile { + data_file: entry.data_file().clone(), + schema: schema.clone(), + partition_spec: partition_spec.clone(), + }); } else if entry.is_alive() { writer.add_existing_entry(entry.as_ref().clone())?; } @@ -677,7 +685,7 @@ mod tests { Ok(vec![rewritten_manifest]) } - fn removed_data_files(&self) -> &[(DataFile, SchemaRef, PartitionSpecRef)] { + fn removed_data_files(&self) -> &[RemovedDataFile] { &self.removed } } @@ -710,11 +718,11 @@ mod tests { let removed = data_file(&table, "test/removed.parquet", 2); let operation = RemoveOneFileOperation { - removed: vec![( - removed, - table.metadata().current_schema().clone(), - table.metadata().default_partition_spec().clone(), - )], + removed: vec![RemovedDataFile { + data_file: removed, + schema: table.metadata().current_schema().clone(), + partition_spec: table.metadata().default_partition_spec().clone(), + }], }; let mut action_commit = @@ -823,12 +831,14 @@ mod tests { .build() .unwrap(), ); - let partition_spec = PartitionSpec::builder(schema.clone()) - .with_spec_id(3) - .add_partition_field("a", "a", Transform::Identity) - .unwrap() - .build() - .unwrap(); + let partition_spec = Arc::new( + PartitionSpec::builder(schema.clone()) + .with_spec_id(3) + .add_partition_field("a", "a", Transform::Identity) + .unwrap() + .build() + .unwrap(), + ); let producer = SnapshotProducer::new(&table, Uuid::now_v7(), HashMap::new(), vec![]); let first = producer @@ -896,7 +906,11 @@ mod tests { 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)], + removed: vec![RemovedDataFile { + data_file: removed, + schema: reported_schema, + partition_spec: reported_spec, + }], }; let mut action_commit = From 495f3adc4e8bd2a349e59a6866a14b55b8ad8fdd Mon Sep 17 00:00:00 2001 From: Jegor van Opdorp Date: Fri, 2 Oct 2026 20:21:32 +0200 Subject: [PATCH 08/13] Take the added files again instead of cloning them for the summary --- crates/iceberg/src/transaction/snapshot.rs | 35 ++++++++++++++-------- 1 file changed, 22 insertions(+), 13 deletions(-) diff --git a/crates/iceberg/src/transaction/snapshot.rs b/crates/iceberg/src/transaction/snapshot.rs index 43345fdeb2..5c335a5b9d 100644 --- a/crates/iceberg/src/transaction/snapshot.rs +++ b/crates/iceberg/src/transaction/snapshot.rs @@ -330,9 +330,7 @@ impl<'a> SnapshotProducer<'a> { // Write manifest file for added data files and return the ManifestFile for ManifestList. async fn write_added_manifest(&mut self) -> Result { - // Cloned rather than taken: the summary is built after the manifests, so the added - // files must still be here. - let added_data_files = self.added_data_files.clone(); + let added_data_files = std::mem::take(&mut self.added_data_files); if added_data_files.is_empty() { return Err(Error::new( ErrorKind::PreconditionFailed, @@ -400,11 +398,8 @@ impl<'a> SnapshotProducer<'a> { Ok(manifest_files) } - // Returns a `Summary` of the current snapshot - fn summary( - &self, - snapshot_produce_operation: &OP, - ) -> Result { + // Collects the added files, before `produce_manifests` takes them. + fn added_files_summary(&self) -> SnapshotSummaryCollector { let mut summary_collector = SnapshotSummaryCollector::default(); let table_metadata = self.table.metadata_ref(); @@ -430,6 +425,16 @@ impl<'a> SnapshotProducer<'a> { table_metadata.default_partition_spec().clone(), ); } + summary_collector + } + + // Returns a `Summary` of the current snapshot + fn summary( + &self, + mut summary_collector: SnapshotSummaryCollector, + snapshot_produce_operation: &OP, + ) -> Result { + let table_metadata = self.table.metadata_ref(); for removed in snapshot_produce_operation.removed_data_files() { summary_collector.remove_file( @@ -511,15 +516,19 @@ impl<'a> SnapshotProducer<'a> { ), }; - // The manifests are produced first so the summary can count the entries the operation - // actually marked Deleted. + // The manifests are produced before the summary is finished so it can count the entries + // the operation actually marked Deleted. + let summary_collector = self.added_files_summary(); let new_manifests = self .produce_manifests(&mut snapshot_produce_operation, &process) .await?; - let summary = self.summary(&snapshot_produce_operation).map_err(|err| { - Error::new(ErrorKind::Unexpected, "Failed to create snapshot summary.").with_source(err) - })?; + let summary = self + .summary(summary_collector, &snapshot_produce_operation) + .map_err(|err| { + Error::new(ErrorKind::Unexpected, "Failed to create snapshot summary.") + .with_source(err) + })?; manifest_list_writer.add_manifests(new_manifests.into_iter())?; let writer_next_row_id = manifest_list_writer.next_row_id(); From 96da179920b3bf82c1fed8ff149b53b43c7ee3ed Mon Sep 17 00:00:00 2001 From: Jegor van Opdorp Date: Fri, 2 Oct 2026 20:23:03 +0200 Subject: [PATCH 09/13] Assert an operation reports exactly the files it marked deleted --- crates/iceberg/src/spec/snapshot_summary.rs | 1 - crates/iceberg/src/transaction/snapshot.rs | 31 +++++++++++++++++---- 2 files changed, 25 insertions(+), 7 deletions(-) diff --git a/crates/iceberg/src/spec/snapshot_summary.rs b/crates/iceberg/src/spec/snapshot_summary.rs index 89af2081ba..ccdad4589a 100644 --- a/crates/iceberg/src/spec/snapshot_summary.rs +++ b/crates/iceberg/src/spec/snapshot_summary.rs @@ -344,7 +344,6 @@ pub(crate) fn update_snapshot_summaries( )); } - update_totals( &mut summary, previous_summary, diff --git a/crates/iceberg/src/transaction/snapshot.rs b/crates/iceberg/src/transaction/snapshot.rs index 5c335a5b9d..5c6aca8c79 100644 --- a/crates/iceberg/src/transaction/snapshot.rs +++ b/crates/iceberg/src/transaction/snapshot.rs @@ -130,9 +130,7 @@ pub(crate) struct SnapshotProducer<'a> { commit_uuid: Uuid, snapshot_properties: HashMap, added_data_files: Vec, - // A counter used to generate unique manifest file names. - // It starts from 0 and increments for each new manifest file. - // Note: This counter is limited to the range of (0..u64::MAX). + // Numbers this commit's manifest files. Atomic so `new_manifest_writer` can take `&self`. manifest_counter: AtomicU64, } @@ -394,6 +392,17 @@ impl<'a> SnapshotProducer<'a> { // # TODO // Support process delete entries. + debug_assert_eq!( + manifest_files + .iter() + .filter(|m| m.added_snapshot_id == self.snapshot_id + && m.content == ManifestContentType::Data) + .filter_map(|m| m.deleted_files_count) + .sum::() as usize, + snapshot_produce_operation.removed_data_files().len(), + "an operation must report exactly the data files it marked deleted" + ); + let manifest_files = manifest_process.process_manifests(self, manifest_files); Ok(manifest_files) } @@ -612,7 +621,7 @@ mod tests { 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. + /// An operation that writes each of its removed files as a deleted entry in its own manifest. struct RemoveOneFileOperation { removed: Vec, } @@ -628,9 +637,19 @@ mod tests { async fn existing_manifest( &mut self, - _: &SnapshotProducer<'_>, + producer: &SnapshotProducer<'_>, ) -> Result> { - Ok(vec![]) + let mut manifests = vec![]; + for removed in &self.removed { + let mut writer = producer.new_manifest_writer( + ManifestContentType::Data, + removed.schema.clone(), + removed.partition_spec.clone(), + )?; + writer.add_delete_file(removed.data_file.clone(), 0, Some(0))?; + manifests.push(writer.write_manifest_file().await?); + } + Ok(manifests) } fn removed_data_files(&self) -> &[RemovedDataFile] { From a5653bceda2812696483842a1cc032a164ab8546 Mon Sep 17 00:00:00 2001 From: Jegor van Opdorp Date: Fri, 2 Oct 2026 20:23:53 +0200 Subject: [PATCH 10/13] Say in the removed_data_files doc what it must match --- crates/iceberg/src/transaction/snapshot.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/iceberg/src/transaction/snapshot.rs b/crates/iceberg/src/transaction/snapshot.rs index 5c6aca8c79..3c0c3c2f86 100644 --- a/crates/iceberg/src/transaction/snapshot.rs +++ b/crates/iceberg/src/transaction/snapshot.rs @@ -87,7 +87,7 @@ pub(crate) trait SnapshotProduceOperation: Send + Sync { snapshot_produce: &SnapshotProducer<'_>, ) -> impl Future>> + Send; - /// Returns the data files this operation actually removed. + /// Returns the data files this operation marked deleted in the manifests it wrote. /// /// Only populated once [`Self::existing_manifest`] has run, so the snapshot summary counts /// what was really removed rather than what the caller asked to remove. From 29772614f6e37c8338e46ed7c682297ad032da2d Mon Sep 17 00:00:00 2001 From: Jegor van Opdorp Date: Fri, 2 Oct 2026 20:29:21 +0200 Subject: [PATCH 11/13] Summarize deleted files from the producer's list, as pyiceberg does Drops the removed_data_files hook and the manifest-first reorder, and truncation defaults back to every overwrite. --- crates/iceberg/src/spec/snapshot_summary.rs | 315 +++++++++++++-- crates/iceberg/src/transaction/append.rs | 3 +- crates/iceberg/src/transaction/snapshot.rs | 403 ++++---------------- 3 files changed, 356 insertions(+), 365 deletions(-) diff --git a/crates/iceberg/src/spec/snapshot_summary.rs b/crates/iceberg/src/spec/snapshot_summary.rs index 7a639aa08e..d68a49d149 100644 --- a/crates/iceberg/src/spec/snapshot_summary.rs +++ b/crates/iceberg/src/spec/snapshot_summary.rs @@ -331,8 +331,9 @@ where T: PartialOrd + Default + ToString { } pub(crate) fn update_snapshot_summaries( - mut summary: Summary, + summary: Summary, previous_summary: Option<&Summary>, + truncate_full_table: bool, ) -> Result { // Validate that the operation is supported if summary.operation != Operation::Append @@ -342,6 +343,16 @@ pub(crate) fn update_snapshot_summaries( return Err(invalid_data!("Operation is not supported.")); } + let mut summary = match previous_summary { + Some(prev_summary) if truncate_full_table && summary.operation == Operation::Overwrite => { + truncate_table_summary(summary, prev_summary).map_err(|err| { + Error::new(ErrorKind::Unexpected, "Failed to truncate table summary.") + .with_source(err) + })? + } + _ => summary, + }; + update_totals( &mut summary, previous_summary, @@ -392,6 +403,77 @@ pub(crate) fn update_snapshot_summaries( Ok(summary) } +fn get_prop(previous_summary: &Summary, prop: &str) -> Result { + let value_str = previous_summary + .additional_properties + .get(prop) + .map(String::as_str) + .unwrap_or("0"); + value_str.parse::().map_err(|err| { + Error::new( + ErrorKind::Unexpected, + format!("Failed to parse summary property '{prop}' value '{value_str}' as u64."), + ) + .with_source(err) + }) +} + +fn truncate_table_summary(mut summary: Summary, previous_summary: &Summary) -> Result { + for prop in [ + TOTAL_DATA_FILES, + TOTAL_DELETE_FILES, + TOTAL_RECORDS, + TOTAL_FILE_SIZE, + TOTAL_POSITION_DELETES, + TOTAL_EQUALITY_DELETES, + ] { + summary + .additional_properties + .insert(prop.to_string(), "0".to_string()); + } + + let value = get_prop(previous_summary, TOTAL_DATA_FILES)?; + if value != 0 { + summary + .additional_properties + .insert(DELETED_DATA_FILES.to_string(), value.to_string()); + } + let value = get_prop(previous_summary, TOTAL_DELETE_FILES)?; + if value != 0 { + summary + .additional_properties + .insert(REMOVED_DELETE_FILES.to_string(), value.to_string()); + } + let value = get_prop(previous_summary, TOTAL_RECORDS)?; + if value != 0 { + summary + .additional_properties + .insert(DELETED_RECORDS.to_string(), value.to_string()); + } + let value = get_prop(previous_summary, TOTAL_FILE_SIZE)?; + if value != 0 { + summary + .additional_properties + .insert(REMOVED_FILE_SIZE.to_string(), value.to_string()); + } + + let value = get_prop(previous_summary, TOTAL_POSITION_DELETES)?; + if value != 0 { + summary + .additional_properties + .insert(REMOVED_POSITION_DELETES.to_string(), value.to_string()); + } + + let value = get_prop(previous_summary, TOTAL_EQUALITY_DELETES)?; + if value != 0 { + summary + .additional_properties + .insert(REMOVED_EQUALITY_DELETES.to_string(), value.to_string()); + } + + Ok(summary) +} + fn update_totals( summary: &mut Summary, previous_summary: Option<&Summary>, @@ -508,7 +590,7 @@ mod tests { additional_properties: new_props, }; - let updated = update_snapshot_summaries(summary, Some(&previous_summary)).unwrap(); + let updated = update_snapshot_summaries(summary, Some(&previous_summary), false).unwrap(); assert_eq!( updated.additional_properties.get(TOTAL_DATA_FILES).unwrap(), @@ -545,6 +627,194 @@ mod tests { ); } + #[test] + fn test_truncate_table_summary() { + let prev_props: HashMap = [ + (TOTAL_DATA_FILES.to_string(), "10".to_string()), + (TOTAL_DELETE_FILES.to_string(), "5".to_string()), + (TOTAL_RECORDS.to_string(), "100".to_string()), + (TOTAL_FILE_SIZE.to_string(), "1000".to_string()), + (TOTAL_POSITION_DELETES.to_string(), "3".to_string()), + (TOTAL_EQUALITY_DELETES.to_string(), "2".to_string()), + ] + .into_iter() + .collect(); + + let previous_summary = Summary { + operation: Operation::Overwrite, + additional_properties: prev_props, + }; + + let mut new_props = HashMap::new(); + new_props.insert("dummy".to_string(), "value".to_string()); + let summary = Summary { + operation: Operation::Overwrite, + additional_properties: new_props, + }; + + let truncated = truncate_table_summary(summary, &previous_summary).unwrap(); + + assert_eq!( + truncated + .additional_properties + .get(TOTAL_DATA_FILES) + .unwrap(), + "0" + ); + assert_eq!( + truncated + .additional_properties + .get(TOTAL_DELETE_FILES) + .unwrap(), + "0" + ); + assert_eq!( + truncated.additional_properties.get(TOTAL_RECORDS).unwrap(), + "0" + ); + assert_eq!( + truncated + .additional_properties + .get(TOTAL_FILE_SIZE) + .unwrap(), + "0" + ); + assert_eq!( + truncated + .additional_properties + .get(TOTAL_POSITION_DELETES) + .unwrap(), + "0" + ); + assert_eq!( + truncated + .additional_properties + .get(TOTAL_EQUALITY_DELETES) + .unwrap(), + "0" + ); + + assert_eq!( + truncated + .additional_properties + .get(DELETED_DATA_FILES) + .unwrap(), + "10" + ); + assert_eq!( + truncated + .additional_properties + .get(REMOVED_DELETE_FILES) + .unwrap(), + "5" + ); + assert_eq!( + truncated + .additional_properties + .get(DELETED_RECORDS) + .unwrap(), + "100" + ); + assert_eq!( + truncated + .additional_properties + .get(REMOVED_FILE_SIZE) + .unwrap(), + "1000" + ); + assert_eq!( + truncated + .additional_properties + .get(REMOVED_POSITION_DELETES) + .unwrap(), + "3" + ); + assert_eq!( + truncated + .additional_properties + .get(REMOVED_EQUALITY_DELETES) + .unwrap(), + "2" + ); + } + + #[test] + fn test_update_snapshot_summaries_overwrite_truncate_handles_totals_above_i32_max() { + // A table can legitimately accumulate more than i32::MAX rows or files + // over its lifetime. Truncating such a table on overwrite must succeed + // and surface the previous totals into the deleted-* counters. + let big = (i32::MAX as u64 + 1).to_string(); // 2_147_483_648 + let prev_props: HashMap = [ + (TOTAL_DATA_FILES.to_string(), big.clone()), + (TOTAL_DELETE_FILES.to_string(), "0".to_string()), + (TOTAL_RECORDS.to_string(), big.clone()), + (TOTAL_FILE_SIZE.to_string(), "0".to_string()), + (TOTAL_POSITION_DELETES.to_string(), "0".to_string()), + (TOTAL_EQUALITY_DELETES.to_string(), "0".to_string()), + ] + .into_iter() + .collect(); + + let previous_summary = Summary { + operation: Operation::Overwrite, + additional_properties: prev_props, + }; + + let summary = Summary { + operation: Operation::Overwrite, + additional_properties: HashMap::new(), + }; + + let updated = update_snapshot_summaries(summary, Some(&previous_summary), true) + .expect("overwrite truncation should accept totals above i32::MAX"); + assert_eq!( + updated + .additional_properties + .get(DELETED_DATA_FILES) + .unwrap(), + &big + ); + assert_eq!( + updated.additional_properties.get(DELETED_RECORDS).unwrap(), + &big + ); + } + + #[test] + fn test_update_snapshot_summaries_overwrite_truncate_returns_err_on_malformed_total() { + // Non-numeric values in the previous summary (corruption, manual edits, + // a foreign implementation) must surface as a recoverable Err - not + // crash the process. + let prev_props: HashMap = [ + (TOTAL_DATA_FILES.to_string(), "not_a_number".to_string()), + (TOTAL_DELETE_FILES.to_string(), "0".to_string()), + (TOTAL_RECORDS.to_string(), "0".to_string()), + (TOTAL_FILE_SIZE.to_string(), "0".to_string()), + (TOTAL_POSITION_DELETES.to_string(), "0".to_string()), + (TOTAL_EQUALITY_DELETES.to_string(), "0".to_string()), + ] + .into_iter() + .collect(); + + let previous_summary = Summary { + operation: Operation::Overwrite, + additional_properties: prev_props, + }; + + let summary = Summary { + operation: Operation::Overwrite, + additional_properties: HashMap::new(), + }; + + let err = update_snapshot_summaries(summary, Some(&previous_summary), true) + .expect_err("malformed previous summary must produce an Err, not a panic"); + assert!( + err.message().contains("truncate table summary"), + "expected wrapped 'Failed to truncate table summary' context, got: {}", + err.message() + ); + } + #[test] fn test_snapshot_summary_collector_build() { let schema = Arc::new( @@ -879,7 +1149,7 @@ mod tests { additional_properties: new_props, }; - let updated = update_snapshot_summaries(summary, Some(&previous_summary)).unwrap(); + let updated = update_snapshot_summaries(summary, Some(&previous_summary), false).unwrap(); let props = &updated.additional_properties; assert_eq!(props.get(TOTAL_DATA_FILES).unwrap(), "12"); @@ -925,7 +1195,7 @@ mod tests { }; // Must not panic. - let updated = update_snapshot_summaries(summary, Some(&previous_summary)).unwrap(); + let updated = update_snapshot_summaries(summary, Some(&previous_summary), false).unwrap(); let props = &updated.additional_properties; // The total whose added delta was unparsable is skipped... @@ -953,7 +1223,7 @@ mod tests { additional_properties: new_props, }; - let updated = update_snapshot_summaries(summary, None).unwrap(); + let updated = update_snapshot_summaries(summary, None, false).unwrap(); let props = &updated.additional_properties; assert_eq!(props.get(TOTAL_DATA_FILES).unwrap(), "4"); @@ -997,7 +1267,7 @@ mod tests { additional_properties: new_props, }; - let updated = update_snapshot_summaries(summary, Some(&previous_summary)).unwrap(); + let updated = update_snapshot_summaries(summary, Some(&previous_summary), false).unwrap(); let props = &updated.additional_properties; assert_eq!(props.get(TOTAL_DATA_FILES).unwrap(), "8"); @@ -1010,39 +1280,20 @@ mod tests { #[test] fn test_update_totals_omits_total_when_removed_exceeds_previous_total() { - let prev_props: HashMap = [ - (TOTAL_DATA_FILES, "2"), - (TOTAL_RECORDS, "10"), - (TOTAL_FILE_SIZE, "100"), - ] - .into_iter() - .map(|(k, v)| (k.to_string(), v.to_string())) - .collect(); - let previous_summary = Summary { operation: Operation::Append, - additional_properties: prev_props, + additional_properties: HashMap::from([(TOTAL_DATA_FILES.to_string(), "2".to_string())]), }; - - let new_props: HashMap = [ - (DELETED_DATA_FILES, "3"), - (DELETED_RECORDS, "25"), - (REMOVED_FILE_SIZE, "250"), - ] - .into_iter() - .map(|(k, v)| (k.to_string(), v.to_string())) - .collect(); - let summary = Summary { operation: Operation::Delete, - additional_properties: new_props, + additional_properties: HashMap::from([( + DELETED_DATA_FILES.to_string(), + "3".to_string(), + )]), }; - let updated = update_snapshot_summaries(summary, Some(&previous_summary)).unwrap(); - let props = &updated.additional_properties; + let updated = update_snapshot_summaries(summary, Some(&previous_summary), false).unwrap(); - assert!(!props.contains_key(TOTAL_DATA_FILES)); - assert!(!props.contains_key(TOTAL_RECORDS)); - assert!(!props.contains_key(TOTAL_FILE_SIZE)); + assert!(!updated.additional_properties.contains_key(TOTAL_DATA_FILES)); } } diff --git a/crates/iceberg/src/transaction/append.rs b/crates/iceberg/src/transaction/append.rs index 5701be3d13..881bd5d752 100644 --- a/crates/iceberg/src/transaction/append.rs +++ b/crates/iceberg/src/transaction/append.rs @@ -93,6 +93,7 @@ impl TransactionAction for FastAppendAction { self.commit_uuid.unwrap_or_else(Uuid::now_v7), self.snapshot_properties.clone(), self.dedupe_added_files(), + vec![], ); // validate added files @@ -124,7 +125,7 @@ impl SnapshotProduceOperation for FastAppendOperation { } async fn existing_manifest( - &mut self, + &self, snapshot_produce: &SnapshotProducer<'_>, ) -> Result> { let Some(snapshot) = snapshot_produce.table.metadata().current_snapshot() else { diff --git a/crates/iceberg/src/transaction/snapshot.rs b/crates/iceberg/src/transaction/snapshot.rs index a3b6e70dbd..0a64594422 100644 --- a/crates/iceberg/src/transaction/snapshot.rs +++ b/crates/iceberg/src/transaction/snapshot.rs @@ -83,27 +83,17 @@ pub(crate) trait SnapshotProduceOperation: Send + Sync { /// - **Overwrite operations**: May exclude manifests for partitions being overwritten /// - **Delete operations**: May exclude manifests for partitions being deleted fn existing_manifest( - &mut self, + &self, snapshot_produce: &SnapshotProducer<'_>, ) -> impl Future>> + Send; - /// Returns the data files this operation marked deleted in the manifests it wrote. - /// - /// Only populated once [`Self::existing_manifest`] has run, so the snapshot summary counts - /// what was really removed rather than what the caller asked to remove. - fn removed_data_files(&self) -> &[RemovedDataFile] { - &[] + /// Returns whether the snapshot summary resets the table totals. Defaults to every + /// overwrite; an operation that overwrites only some files returns `false`. + fn truncate_full_table(&self) -> bool { + self.operation() == Operation::Overwrite } } -/// A data file an operation removed, with the schema and partition spec of the manifest that -/// recorded it. -pub(crate) struct RemovedDataFile { - pub(crate) data_file: DataFile, - pub(crate) schema: SchemaRef, - pub(crate) partition_spec: PartitionSpecRef, -} - pub(crate) struct DefaultManifestProcess; impl ManifestProcess for DefaultManifestProcess { @@ -130,6 +120,7 @@ pub(crate) struct SnapshotProducer<'a> { commit_uuid: Uuid, snapshot_properties: HashMap, added_data_files: Vec, + deleted_data_files: Vec, // Numbers this commit's manifest files. Atomic so `new_manifest_writer` can take `&self`. manifest_counter: AtomicU64, } @@ -140,6 +131,7 @@ impl<'a> SnapshotProducer<'a> { commit_uuid: Uuid, snapshot_properties: HashMap, added_data_files: Vec, + deleted_data_files: Vec, ) -> Self { Self { table, @@ -147,6 +139,7 @@ impl<'a> SnapshotProducer<'a> { commit_uuid, snapshot_properties, added_data_files, + deleted_data_files, manifest_counter: AtomicU64::new(0), } } @@ -244,26 +237,22 @@ impl<'a> SnapshotProducer<'a> { snapshot_id } - /// Returns the path for the next manifest file of this commit. - fn new_manifest_path(&self) -> Result { - Ok(format!( - "{}/{}-m{}.{}", - self.table.metadata().metadata_location()?, - self.commit_uuid, - self.manifest_counter.fetch_add(1, Ordering::Relaxed), - DataFileFormat::Avro - )) - } - - /// Returns a writer for the next manifest file of this commit, routed through the table's - /// encryption manager when one is configured. + /// Returns a writer for the next manifest file of this commit, written under the given + /// schema and partition spec. pub(crate) fn new_manifest_writer( &self, content: ManifestContentType, schema: SchemaRef, partition_spec: PartitionSpecRef, ) -> Result { - let output_file = self.table.file_io().new_output(self.new_manifest_path()?)?; + let new_manifest_path = format!( + "{}/{}-m{}.{}", + self.table.metadata().metadata_location()?, + self.commit_uuid, + self.manifest_counter.fetch_add(1, Ordering::Relaxed), + DataFileFormat::Avro + ); + let output_file = self.table.file_io().new_output(new_manifest_path)?; let partition_spec = partition_spec.as_ref().clone(); let builder = if let Some(em) = self.table.encryption_manager() { @@ -358,7 +347,7 @@ impl<'a> SnapshotProducer<'a> { /// and collects all of the manifests to be included in the new snapshot as [ManifestFile] entries. async fn produce_manifests( &mut self, - snapshot_produce_operation: &mut OP, + snapshot_produce_operation: &OP, manifest_process: &MP, ) -> Result> { // Assert current snapshot producer contains new content to add to new snapshot. @@ -366,10 +355,13 @@ impl<'a> SnapshotProducer<'a> { // TODO: Allowing snapshot property setup with no added data files is a workaround. // We should clean it up after all necessary actions are supported. // For details, please refer to https://github.com/apache/iceberg-rust/issues/1548 - if self.added_data_files.is_empty() && self.snapshot_properties.is_empty() { + if self.added_data_files.is_empty() + && self.deleted_data_files.is_empty() + && self.snapshot_properties.is_empty() + { return Err(Error::new( ErrorKind::PreconditionFailed, - "No added data files or added snapshot properties found when write a manifest file", + "No added or deleted data files or added snapshot properties found when write a manifest file", )); } @@ -385,23 +377,15 @@ impl<'a> SnapshotProducer<'a> { // # TODO // Support process delete entries. - debug_assert_eq!( - manifest_files - .iter() - .filter(|m| m.added_snapshot_id == self.snapshot_id - && m.content == ManifestContentType::Data) - .filter_map(|m| m.deleted_files_count) - .sum::() as usize, - snapshot_produce_operation.removed_data_files().len(), - "an operation must report exactly the data files it marked deleted" - ); - let manifest_files = manifest_process.process_manifests(self, manifest_files); Ok(manifest_files) } - // Collects the added files, before `produce_manifests` takes them. - fn added_files_summary(&self) -> SnapshotSummaryCollector { + // Returns a `Summary` of the current snapshot + fn summary( + &self, + snapshot_produce_operation: &OP, + ) -> Result { let mut summary_collector = SnapshotSummaryCollector::default(); let table_metadata = self.table.metadata_ref(); @@ -427,22 +411,17 @@ impl<'a> SnapshotProducer<'a> { table_metadata.default_partition_spec().clone(), ); } - summary_collector - } - - // Returns a `Summary` of the current snapshot - fn summary( - &self, - mut summary_collector: SnapshotSummaryCollector, - snapshot_produce_operation: &OP, - ) -> Result { - let table_metadata = self.table.metadata_ref(); - for removed in snapshot_produce_operation.removed_data_files() { + for data_file in &self.deleted_data_files { + let partition_spec = table_metadata + .partition_spec_by_id(data_file.partition_spec_id) + .ok_or_else(|| { + invalid_data!("Unknown partition spec id {}", data_file.partition_spec_id) + })?; summary_collector.remove_file( - &removed.data_file, - removed.schema.clone(), - removed.partition_spec.clone(), + data_file, + table_metadata.current_schema().clone(), + partition_spec.clone(), ); } @@ -461,7 +440,11 @@ impl<'a> SnapshotProducer<'a> { additional_properties, }; - update_snapshot_summaries(summary, previous_snapshot.map(|s| s.summary())) + update_snapshot_summaries( + summary, + previous_snapshot.map(|s| s.summary()), + snapshot_produce_operation.truncate_full_table(), + ) } fn generate_manifest_list_file_path(&self, attempt: i64) -> Result { @@ -478,7 +461,7 @@ impl<'a> SnapshotProducer<'a> { /// Finished building the action and return the [`ActionCommit`] to the transaction. pub(crate) async fn commit( mut self, - mut snapshot_produce_operation: OP, + snapshot_produce_operation: OP, process: MP, ) -> Result { let manifest_list_path = self.generate_manifest_list_file_path(0)?; @@ -518,20 +501,17 @@ impl<'a> SnapshotProducer<'a> { ), }; - // The manifests are produced before the summary is finished so it can count the entries - // the operation actually marked Deleted. - let summary_collector = self.added_files_summary(); + // Calling self.summary() before self.produce_manifests() is important because self.added_data_files + // will be set to an empty vec after self.produce_manifests() returns, resulting in an empty summary + // being generated. + let summary = self.summary(&snapshot_produce_operation).map_err(|err| { + Error::new(ErrorKind::Unexpected, "Failed to create snapshot summary.").with_source(err) + })?; + let new_manifests = self - .produce_manifests(&mut snapshot_produce_operation, &process) + .produce_manifests(&snapshot_produce_operation, &process) .await?; - let summary = self - .summary(summary_collector, &snapshot_produce_operation) - .map_err(|err| { - Error::new(ErrorKind::Unexpected, "Failed to create snapshot summary.") - .with_source(err) - })?; - manifest_list_writer.add_manifests(new_manifests.into_iter())?; let writer_next_row_id = manifest_list_writer.next_row_id(); let file_metadata = manifest_list_writer.close().await?; @@ -617,18 +597,14 @@ mod tests { use super::*; use crate::spec::{ - DataContentType, DataFileBuilder, Literal, Manifest, ManifestStatus, NestedField, - PartitionSpec, PrimitiveType, Schema, Transform, Type, + DataContentType, DataFileBuilder, Literal, Manifest, NestedField, PartitionSpec, + PrimitiveType, Schema, Transform, Type, }; use crate::transaction::tests::make_v2_minimal_table; - use crate::transaction::{Transaction, TransactionAction}; - /// An operation that writes each of its removed files as a deleted entry in its own manifest. - struct RemoveOneFileOperation { - removed: Vec, - } + struct DeleteOperation; - impl SnapshotProduceOperation for RemoveOneFileOperation { + impl SnapshotProduceOperation for DeleteOperation { fn operation(&self) -> Operation { Operation::Delete } @@ -637,216 +613,34 @@ mod tests { Ok(vec![]) } - async fn existing_manifest( - &mut self, - producer: &SnapshotProducer<'_>, - ) -> Result> { - let mut manifests = vec![]; - for removed in &self.removed { - let mut writer = producer.new_manifest_writer( - ManifestContentType::Data, - removed.schema.clone(), - removed.partition_spec.clone(), - )?; - writer.add_delete_file(removed.data_file.clone(), 0, Some(0))?; - manifests.push(writer.write_manifest_file().await?); - } - Ok(manifests) - } - - fn removed_data_files(&self) -> &[RemovedDataFile] { - &self.removed - } - } - - struct RemoveExistingFileOperation { - file_path: String, - removed: Vec, - } - - impl SnapshotProduceOperation for RemoveExistingFileOperation { - fn operation(&self) -> Operation { - Operation::Overwrite - } - - async fn delete_entries(&self, _: &SnapshotProducer<'_>) -> Result> { + async fn existing_manifest(&self, _: &SnapshotProducer<'_>) -> Result> { Ok(vec![]) } - - async fn existing_manifest( - &mut self, - producer: &SnapshotProducer<'_>, - ) -> Result> { - 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 = Arc::new(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(RemovedDataFile { - data_file: entry.data_file().clone(), - schema: schema.clone(), - partition_spec: 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) -> &[RemovedDataFile] { - &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() + #[test] + fn test_summary_counts_deleted_data_files() { + let table = make_v2_minimal_table(); + let deleted = DataFileBuilder::default() .content(DataContentType::Data) - .file_path(path.to_string()) + .file_path("test/deleted.parquet".to_string()) .file_format(DataFileFormat::Parquet) .file_size_in_bytes(100) - .record_count(record_count) + .record_count(2) .partition_spec_id(table.metadata().default_partition_spec_id()) .partition(Struct::from_iter([Some(Literal::long(300))])) .build() - .unwrap() - } - - #[tokio::test] - async fn test_summary_counts_files_the_operation_reports_as_removed() { - let table = make_v2_minimal_table(); - let added = data_file(&table, "test/added.parquet", 3); - let removed = data_file(&table, "test/removed.parquet", 2); - - let operation = RemoveOneFileOperation { - removed: vec![RemovedDataFile { - data_file: removed, - schema: table.metadata().current_schema().clone(), - partition_spec: table.metadata().default_partition_spec().clone(), - }], - }; - - let mut action_commit = - SnapshotProducer::new(&table, Uuid::now_v7(), HashMap::new(), vec![added]) - .commit(operation, DefaultManifestProcess) - .await - .unwrap(); + .unwrap(); - let updates = action_commit.take_updates(); - let TableUpdate::AddSnapshot { snapshot } = &updates[0] else { - unreachable!() - }; - let properties = &snapshot.summary().additional_properties; + let producer = SnapshotProducer::new(&table, Uuid::now_v7(), HashMap::new(), vec![], vec![ + deleted, + ]); + let summary = producer.summary(&DeleteOperation).unwrap(); + let properties = &summary.additional_properties; 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("total-data-files").unwrap(), "0"); - 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] @@ -870,7 +664,8 @@ mod tests { .unwrap(), ); - let producer = SnapshotProducer::new(&table, Uuid::now_v7(), HashMap::new(), vec![]); + let producer = + SnapshotProducer::new(&table, Uuid::now_v7(), HashMap::new(), vec![], vec![]); let first = producer .new_manifest_writer( ManifestContentType::Data, @@ -905,60 +700,4 @@ 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![RemovedDataFile { - data_file: removed, - schema: reported_schema, - partition_spec: 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")); - } } From 683fae4fd1b4c43d44e699ca13cb872bf10a63d1 Mon Sep 17 00:00:00 2001 From: Jegor van Opdorp Date: Tue, 13 Oct 2026 14:07:32 +0200 Subject: [PATCH 12/13] Check that deletion summaries match manifest entries --- crates/iceberg/src/transaction/snapshot.rs | 36 ++++++++++++++++++++++ 1 file changed, 36 insertions(+) diff --git a/crates/iceberg/src/transaction/snapshot.rs b/crates/iceberg/src/transaction/snapshot.rs index 0a64594422..c301750012 100644 --- a/crates/iceberg/src/transaction/snapshot.rs +++ b/crates/iceberg/src/transaction/snapshot.rs @@ -120,6 +120,7 @@ pub(crate) struct SnapshotProducer<'a> { commit_uuid: Uuid, snapshot_properties: HashMap, added_data_files: Vec, + /// Must match the data files marked Deleted by this commit, including their metrics. deleted_data_files: Vec, // Numbers this commit's manifest files. Atomic so `new_manifest_writer` can take `&self`. manifest_counter: AtomicU64, @@ -377,6 +378,18 @@ impl<'a> SnapshotProducer<'a> { // # TODO // Support process delete entries. + debug_assert_eq!( + manifest_files + .iter() + .filter(|m| m.added_snapshot_id == self.snapshot_id + && m.content == ManifestContentType::Data) + .filter_map(|m| m.deleted_files_count) + .map(u64::from) + .sum::(), + self.deleted_data_files.len() as u64, + "deleted files must match this commit's manifest entries" + ); + let manifest_files = manifest_process.process_manifests(self, manifest_files); Ok(manifest_files) } @@ -643,6 +656,29 @@ mod tests { assert_eq!(properties.get("removed-files-size").unwrap(), "100"); } + #[cfg(debug_assertions)] + #[tokio::test] + #[should_panic(expected = "deleted files must match this commit's manifest entries")] + async fn test_commit_rejects_unrecorded_deletions() { + let table = make_v2_minimal_table(); + let deleted = DataFileBuilder::default() + .content(DataContentType::Data) + .file_path("test/deleted.parquet".to_string()) + .file_format(DataFileFormat::Parquet) + .file_size_in_bytes(100) + .record_count(2) + .partition_spec_id(table.metadata().default_partition_spec_id()) + .partition(Struct::from_iter([Some(Literal::long(300))])) + .build() + .unwrap(); + SnapshotProducer::new(&table, Uuid::now_v7(), HashMap::new(), vec![], vec![ + deleted, + ]) + .commit(DeleteOperation, DefaultManifestProcess) + .await + .unwrap(); + } + #[tokio::test] async fn test_new_manifest_writer_uses_the_given_schema_and_spec() { let table = make_v2_minimal_table(); From 5f3693808baeab1f04e4cdc295963a5326f76e28 Mon Sep 17 00:00:00 2001 From: Jegor van Opdorp Date: Tue, 13 Oct 2026 14:07:56 +0200 Subject: [PATCH 13/13] Format dropped-source integer partition values --- crates/iceberg/src/spec/partition.rs | 47 +++++++++++++++++++++++++ crates/iceberg/src/spec/values/datum.rs | 8 +++-- 2 files changed, 53 insertions(+), 2 deletions(-) diff --git a/crates/iceberg/src/spec/partition.rs b/crates/iceberg/src/spec/partition.rs index 17aa3de968..fa8de99a59 100644 --- a/crates/iceberg/src/spec/partition.rs +++ b/crates/iceberg/src/spec/partition.rs @@ -1913,6 +1913,53 @@ mod tests { assert!(!spec.has_sequential_ids()); } + #[test] + fn test_partition_to_path_with_dropped_source_column() { + for (source_type, transform, value, expected) in [ + ( + PrimitiveType::Int, + Transform::Identity, + Literal::int(1), + "p=1", + ), + ( + PrimitiveType::Timestamp, + Transform::Identity, + Literal::long(1), + "p=1", + ), + ( + PrimitiveType::Int, + Transform::Truncate(10), + Literal::int(10), + "p=10", + ), + ] { + let schema = Schema::builder() + .with_fields(vec![ + NestedField::required(1, "kept", PrimitiveType::Int.into()).into(), + NestedField::required(2, "dropped", source_type.into()).into(), + ]) + .build() + .unwrap(); + let spec = PartitionSpec::builder(schema) + .add_partition_field("dropped", "p", transform) + .unwrap() + .build() + .unwrap(); + let current_schema = Schema::builder() + .with_fields(vec![ + NestedField::required(1, "kept", PrimitiveType::Int.into()).into(), + ]) + .build() + .unwrap(); + assert_eq!( + spec.partition_to_path(&Struct::from_iter([Some(value)]), current_schema.into()), + expected + ); + } + } + #[test] fn test_partition_to_path() { let schema = Schema::builder() diff --git a/crates/iceberg/src/spec/values/datum.rs b/crates/iceberg/src/spec/values/datum.rs index 57d9675736..7fa54954c8 100644 --- a/crates/iceberg/src/spec/values/datum.rs +++ b/crates/iceberg/src/spec/values/datum.rs @@ -294,8 +294,12 @@ impl Display for Datum { fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { match (&self.r#type, &self.literal) { (_, PrimitiveLiteral::Boolean(val)) => write!(f, "{val}"), - (PrimitiveType::Int, PrimitiveLiteral::Int(val)) => write!(f, "{val}"), - (PrimitiveType::Long, PrimitiveLiteral::Long(val)) => write!(f, "{val}"), + (PrimitiveType::Int | PrimitiveType::Unknown, PrimitiveLiteral::Int(val)) => { + write!(f, "{val}") + } + (PrimitiveType::Long | PrimitiveType::Unknown, PrimitiveLiteral::Long(val)) => { + write!(f, "{val}") + } (_, PrimitiveLiteral::Float(val)) => write!(f, "{val}"), (_, PrimitiveLiteral::Double(val)) => write!(f, "{val}"), (PrimitiveType::Date, PrimitiveLiteral::Int(val)) => {