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/snapshot_summary.rs b/crates/iceberg/src/spec/snapshot_summary.rs index f8e69c4c86..d68a49d149 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 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)) => { diff --git a/crates/iceberg/src/transaction/append.rs b/crates/iceberg/src/transaction/append.rs index 3260f9ffa2..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 diff --git a/crates/iceberg/src/transaction/snapshot.rs b/crates/iceberg/src/transaction/snapshot.rs index a0ce1fba14..c301750012 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>> + 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, 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). - manifest_counter: RangeFrom, + /// 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, } impl<'a> SnapshotProducer<'a> { @@ -125,6 +132,7 @@ impl<'a> SnapshotProducer<'a> { commit_uuid: Uuid, snapshot_properties: HashMap, added_data_files: Vec, + deleted_data_files: Vec, ) -> 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 { + /// 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 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::(), + 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> { + Ok(vec![]) + } + + async fn existing_manifest(&self, _: &SnapshotProducer<'_>) -> Result> { + 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); + } +}