This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-3253-b3470c8463c281c6e00ea9eee402881bb6c6d1fd in repository https://gitbox.apache.org/repos/asf/iceberg-rust.git
commit e4bec84bcca5ba4f3e19583e95b1de7c1fab2559 Author: Jegor van Opdorp <[email protected]> AuthorDate: Thu Oct 15 18:28:41 2026 +0000 feat(transaction): let a snapshot operation report removed files and rewrite manifests (#3253) * 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. * 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. * 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. * test(transaction): cover partial overwrite and per-file spec in snapshot summary Co-authored-by: Marco Kroll <[email protected]> * 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. * Omit a summary total that would go negative, as Java does * Name the removed-file fields and take the manifest spec by ref * Take the added files again instead of cloning them for the summary * Assert an operation reports exactly the files it marked deleted * Say in the removed_data_files doc what it must match * 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. * Check that deletion summaries match manifest entries * Format dropped-source integer partition values --------- Co-authored-by: Marco Kroll <[email protected]> --- crates/iceberg/src/spec/partition.rs | 47 ++++++ crates/iceberg/src/spec/snapshot_summary.rs | 24 ++- crates/iceberg/src/spec/values/datum.rs | 8 +- crates/iceberg/src/transaction/append.rs | 1 + crates/iceberg/src/transaction/snapshot.rs | 223 +++++++++++++++++++++++++--- 5 files changed, 278 insertions(+), 25 deletions(-) diff --git a/crates/iceberg/src/spec/partition.rs b/crates/iceberg/src/spec/partition.rs index f3c9f9e54..fdb5a4a7b 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/snapshot_summary.rs b/crates/iceberg/src/spec/snapshot_summary.rs index f8e69c4c8..d68a49d14 100644 --- a/crates/iceberg/src/spec/snapshot_summary.rs +++ b/crates/iceberg/src/spec/snapshot_summary.rs @@ -530,7 +530,10 @@ fn update_totals( return; }; - let new_total = previous_total + added - 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()); @@ -1274,4 +1277,23 @@ 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_omits_total_when_removed_exceeds_previous_total() { + let previous_summary = Summary { + operation: Operation::Append, + additional_properties: HashMap::from([(TOTAL_DATA_FILES.to_string(), "2".to_string())]), + }; + let summary = Summary { + operation: Operation::Delete, + additional_properties: HashMap::from([( + DELETED_DATA_FILES.to_string(), + "3".to_string(), + )]), + }; + + let updated = update_snapshot_summaries(summary, Some(&previous_summary), false).unwrap(); + + assert!(!updated.additional_properties.contains_key(TOTAL_DATA_FILES)); + } } diff --git a/crates/iceberg/src/spec/values/datum.rs b/crates/iceberg/src/spec/values/datum.rs index a501466b3..2a7bd3f44 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)) => { diff --git a/crates/iceberg/src/transaction/append.rs b/crates/iceberg/src/transaction/append.rs index 2b6f01044..e3e74338c 100644 --- a/crates/iceberg/src/transaction/append.rs +++ b/crates/iceberg/src/transaction/append.rs @@ -100,6 +100,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 diff --git a/crates/iceberg/src/transaction/snapshot.rs b/crates/iceberg/src/transaction/snapshot.rs index a0ce1fba1..c30175001 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; @@ -26,9 +26,10 @@ use uuid::Uuid; use crate::error::{Result, invalid_data}; 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; @@ -85,6 +86,12 @@ pub(crate) trait SnapshotProduceOperation: Send + Sync { &self, snapshot_produce: &SnapshotProducer<'_>, ) -> impl Future<Output = Result<Vec<ManifestFile>>> + Send; + + /// 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 + } } pub(crate) struct DefaultManifestProcess; @@ -113,10 +120,10 @@ pub(crate) struct SnapshotProducer<'a> { commit_uuid: Uuid, snapshot_properties: HashMap<String, String>, added_data_files: Vec<DataFile>, - // 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<u64>, + /// Must match the data files marked Deleted by this commit, including their metrics. + deleted_data_files: Vec<DataFile>, + // Numbers this commit's manifest files. Atomic so `new_manifest_writer` can take `&self`. + manifest_counter: AtomicU64, } impl<'a> SnapshotProducer<'a> { @@ -125,6 +132,7 @@ impl<'a> SnapshotProducer<'a> { commit_uuid: Uuid, snapshot_properties: HashMap<String, String>, added_data_files: Vec<DataFile>, + deleted_data_files: Vec<DataFile>, ) -> Self { Self { table, @@ -132,7 +140,8 @@ impl<'a> SnapshotProducer<'a> { commit_uuid, snapshot_properties, added_data_files, - manifest_counter: (0..), + deleted_data_files, + manifest_counter: AtomicU64::new(0), } } @@ -229,22 +238,23 @@ impl<'a> SnapshotProducer<'a> { snapshot_id } - fn new_manifest_writer(&mut self, content: ManifestContentType) -> Result<ManifestWriter> { + /// 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<ManifestWriter> { let new_manifest_path = 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)?; - let partition_spec = self - .table - .metadata() - .default_partition_spec() - .as_ref() - .clone(); - let schema = self.table.metadata().current_schema().clone(); + let partition_spec = partition_spec.as_ref().clone(); let builder = if let Some(em) = self.table.encryption_manager() { ManifestWriterBuilder::new_from_encrypted( @@ -323,7 +333,11 @@ 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().clone(), + )?; for entry in manifest_entries { writer.add_entry(entry)?; } @@ -342,10 +356,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", )); } @@ -361,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::<u64>(), + 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) } @@ -396,6 +425,19 @@ impl<'a> SnapshotProducer<'a> { ); } + 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( + data_file, + table_metadata.current_schema().clone(), + partition_spec.clone(), + ); + } + let previous_snapshot = table_metadata.current_snapshot(); // User-supplied snapshot properties are applied first, then the computed @@ -414,7 +456,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(), ) } @@ -558,3 +600,140 @@ impl<'a> SnapshotProducer<'a> { Ok(ActionCommit::new(updates, requirements)) } } + +#[cfg(test)] +mod tests { + use std::collections::HashMap; + use std::sync::Arc; + + use uuid::Uuid; + + use super::*; + use crate::spec::{ + DataContentType, DataFileBuilder, Literal, Manifest, NestedField, PartitionSpec, + PrimitiveType, Schema, Transform, Type, + }; + use crate::transaction::tests::make_v2_minimal_table; + + struct DeleteOperation; + + impl SnapshotProduceOperation for DeleteOperation { + fn operation(&self) -> Operation { + Operation::Delete + } + + async fn delete_entries(&self, _: &SnapshotProducer<'_>) -> Result<Vec<ManifestEntry>> { + Ok(vec![]) + } + + async fn existing_manifest(&self, _: &SnapshotProducer<'_>) -> Result<Vec<ManifestFile>> { + Ok(vec![]) + } + } + + #[test] + fn test_summary_counts_deleted_data_files() { + 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(); + + 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("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(); + 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 = 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![], 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); + } +}
