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-3432-4dd8c946432f5e56a3ccc88e5b438a356ba28287 in repository https://gitbox.apache.org/repos/asf/iceberg-rust.git
commit 1ee5c72a7a56f5b519e7bf5c8c04660f5f070d78 Author: Jegor van Opdorp <[email protected]> AuthorDate: Fri Oct 30 22:40:54 2026 +0000 fix(spec): format dropped-source integer partition values (#3432) * Format dropped-source integer partition values * fix(spec): format dropped-source decimal and uuid partition values * Summarize deleted files with a schema that still has their partition sources * Print int and long partition values across promotion * Resolve the summary schema once per partition spec * Build the dropped-source test history in promotion order --- crates/iceberg/src/spec/partition.rs | 15 ++++ crates/iceberg/src/spec/values/datum.rs | 14 +++- crates/iceberg/src/transaction/snapshot/mod.rs | 95 ++++++++++++++++++++++++-- 3 files changed, 115 insertions(+), 9 deletions(-) diff --git a/crates/iceberg/src/spec/partition.rs b/crates/iceberg/src/spec/partition.rs index e161802c4..bd6ed3542 100644 --- a/crates/iceberg/src/spec/partition.rs +++ b/crates/iceberg/src/spec/partition.rs @@ -2090,6 +2090,21 @@ mod tests { Literal::int(10), "p=10", ), + ( + PrimitiveType::Decimal { + precision: 9, + scale: 2, + }, + Transform::Identity, + Literal::decimal(123), + "p=123", + ), + ( + PrimitiveType::Uuid, + Transform::Identity, + Literal::uuid_from_str("f79c3e09-677c-4bbd-a479-3f349cb785e7").unwrap(), + "p=f79c3e09-677c-4bbd-a479-3f349cb785e7", + ), ] { let schema = Schema::builder() .with_fields(vec![ diff --git a/crates/iceberg/src/spec/values/datum.rs b/crates/iceberg/src/spec/values/datum.rs index 71de11beb..5153151c8 100644 --- a/crates/iceberg/src/spec/values/datum.rs +++ b/crates/iceberg/src/spec/values/datum.rs @@ -300,10 +300,16 @@ 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 | PrimitiveType::Unknown, PrimitiveLiteral::Int(val)) => { + ( + PrimitiveType::Int | PrimitiveType::Long | PrimitiveType::Unknown, + PrimitiveLiteral::Int(val), + ) => { write!(f, "{val}") } - (PrimitiveType::Long | PrimitiveType::Unknown, PrimitiveLiteral::Long(val)) => { + ( + PrimitiveType::Int | PrimitiveType::Long | PrimitiveType::Unknown, + PrimitiveLiteral::Long(val), + ) => { write!(f, "{val}") } (_, PrimitiveLiteral::Float(val)) => write!(f, "{val}"), @@ -327,7 +333,7 @@ impl Display for Datum { write!(f, "{}", timestamptz::nanoseconds_to_datetimetz(*val)) } (_, PrimitiveLiteral::String(val)) => write!(f, r#""{val}""#), - (PrimitiveType::Uuid, PrimitiveLiteral::UInt128(val)) => { + (PrimitiveType::Uuid | PrimitiveType::Unknown, PrimitiveLiteral::UInt128(val)) => { write!(f, "{}", uuid::Uuid::from_u128(*val)) } (_, PrimitiveLiteral::Binary(val)) => display_bytes(val, f), @@ -340,6 +346,8 @@ impl Display for Datum { ) => { write!(f, "{}", decimal_from_i128_with_scale(*val, *scale)) } + // The scale lives on the dropped source type, so only the unscaled value is left. + (PrimitiveType::Unknown, PrimitiveLiteral::Int128(val)) => write!(f, "{val}"), (_, _) => { unreachable!() } diff --git a/crates/iceberg/src/transaction/snapshot/mod.rs b/crates/iceberg/src/transaction/snapshot/mod.rs index 36616d012..eb982e1e4 100644 --- a/crates/iceberg/src/transaction/snapshot/mod.rs +++ b/crates/iceberg/src/transaction/snapshot/mod.rs @@ -259,17 +259,29 @@ fn build_summary( ); } + let mut schemas_by_spec = HashMap::new(); for data_file in removed_files { let partition_spec = metadata .partition_spec_by_id(data_file.partition_spec_id) .ok_or_else(|| { invalid_data!("Unknown partition spec id {}", data_file.partition_spec_id) })?; - collector.remove_file( - data_file, - metadata.current_schema().clone(), - partition_spec.clone(), - ); + // A dropped source column keeps its type, such as a decimal's scale, only in an + // older schema. + let schema = schemas_by_spec + .entry(data_file.partition_spec_id) + .or_insert_with(|| { + metadata + .schemas_iter() + .filter(|schema| { + let mut source_ids = partition_spec.fields().iter().map(|f| f.source_id); + source_ids.all(|id| schema.field_by_id(id).is_some()) + }) + .max_by_key(|schema| schema.schema_id()) + .unwrap_or(metadata.current_schema()) + .clone() + }); + collector.remove_file(data_file, schema.clone(), partition_spec.clone()); } properties.extend(collector.build()); @@ -456,7 +468,7 @@ mod tests { use super::*; use crate::spec::{ DataContentType, DataFileBuilder, Literal, Manifest, NestedField, PartitionSpec, - PrimitiveType, Schema, SnapshotRef, Transform, Type, + PrimitiveType, Schema, SnapshotRef, SortOrder, TableMetadataBuilder, Transform, Type, }; use crate::transaction::tests::make_v2_minimal_table; use crate::{ErrorKind, TableUpdate}; @@ -577,6 +589,77 @@ mod tests { assert_eq!(properties.get("removed-files-size").unwrap(), "100"); } + #[test] + fn test_summary_keeps_decimal_scale_of_dropped_partition_source() { + let decimal = PrimitiveType::Decimal { + precision: 9, + scale: 2, + }; + let d = Arc::new(NestedField::required(1, "d", Type::Primitive(decimal))); + let x = |ty| Arc::new(NestedField::required(2, "x", Type::Primitive(ty))); + let schema = |fields: Vec<_>| Schema::builder().with_fields(fields).build().unwrap(); + let base = schema(vec![d.clone(), x(PrimitiveType::Int)]); + let spec = PartitionSpec::builder(base.clone()) + .add_partition_field("d", "d", Transform::Identity) + .unwrap() + .add_partition_field("x", "x", Transform::Identity) + .unwrap() + .build() + .unwrap(); + let table = make_v2_minimal_table(); + let metadata = TableMetadataBuilder::new( + base, + spec, + SortOrder::unsorted_order(), + table.metadata().location().to_string(), + FormatVersion::V2, + HashMap::from([( + TableProperties::PROPERTY_WRITE_PARTITION_SUMMARY_LIMIT.to_string(), + "10".to_string(), + )]), + ) + .unwrap() + .add_current_schema(schema(vec![d, x(PrimitiveType::Long)])) + .unwrap() + .add_default_partition_spec(Default::default()) + .unwrap() + .add_current_schema(schema(vec![x(PrimitiveType::Long)])) + .unwrap() + .build() + .unwrap() + .metadata; + let table = table.with_metadata(Arc::new(metadata)); + 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(0) + .partition(Struct::from_iter([ + Some(Literal::decimal(123)), + Some(Literal::int(5)), + ])) + .build() + .unwrap(); + + let summary = build_summary( + &table, + Operation::Delete, + HashMap::new(), + &[], + &[deleted], + false, + ) + .unwrap(); + + assert!( + summary + .additional_properties + .contains_key("partitions.d=1.23/x=5") + ); + } + #[cfg(debug_assertions)] #[tokio::test] #[should_panic(expected = "deleted files must match this commit's manifest entries")]
