From 3c12d2d975c778dc2dcbe3643ca5d19f0b581a6a Mon Sep 17 00:00:00 2001 From: osipovartem Date: Thu, 15 Oct 2026 03:13:39 +0300 Subject: [PATCH] Preserve promoted partition bounds across V2 manifest rewrites --- iceberg-rust-spec/src/spec/manifest.rs | 47 ++ iceberg-rust-spec/src/spec/manifest_list.rs | 47 +- iceberg-rust/src/table/manifest.rs | 31 ++ iceberg-rust/src/table/manifest_list.rs | 248 ++++++++- iceberg-rust/src/table/transaction/append.rs | 29 +- .../src/table/transaction/operation.rs | 495 +++++++++++++++++- .../src/table/transaction/overwrite.rs | 40 +- 7 files changed, 895 insertions(+), 42 deletions(-) diff --git a/iceberg-rust-spec/src/spec/manifest.rs b/iceberg-rust-spec/src/spec/manifest.rs index bcb88d96..be288548 100644 --- a/iceberg-rust-spec/src/spec/manifest.rs +++ b/iceberg-rust-spec/src/spec/manifest.rs @@ -707,6 +707,22 @@ pub(crate) fn decode_bound(bytes: &[u8], data_type: &Type) -> Result Result { + match (data_type, bytes.len()) { + (Type::Primitive(PrimitiveType::Long), 4) => { + Value::try_from_bytes(bytes, &Type::Primitive(PrimitiveType::Int)) + } + (Type::Primitive(PrimitiveType::Double), 4) => { + Value::try_from_bytes(bytes, &Type::Primitive(PrimitiveType::Float)) + } + (Type::Primitive(PrimitiveType::Timestamp), 4) => { + Value::try_from_bytes(bytes, &Type::Primitive(PrimitiveType::Date)) + } + _ => Value::try_from_bytes(bytes, data_type), + } +} + impl From> for AvroMap { fn from(value: HashMap) -> Self { AvroMap(HashMap::from_iter( @@ -776,6 +792,37 @@ impl DataFile { DataFileBuilder::default() } + /// Promotes partition values when moving a file entry into a manifest written with a newer schema. + pub fn promote_partition_to_types( + &mut self, + partition_spec: &PartitionSpec, + target_types: &[Type], + ) -> Result<(), Error> { + if partition_spec.fields().len() != target_types.len() { + return Err(Error::InvalidFormat( + "Partition type count mismatch".to_owned(), + )); + } + for (field, target_type) in partition_spec.fields().iter().zip(target_types.iter()) { + let Some(value) = self.partition.get_mut(field.name()) else { + return Err(Error::NotFound(format!("Partition field {}", field.name()))); + }; + let Some(value) = value else { + continue; + }; + let source_type = value.datatype(); + let decimal_unchanged = matches!( + (&*value, target_type), + (Value::Decimal(decimal), Type::Primitive(PrimitiveType::Decimal { scale, .. })) + if decimal_scale(decimal) == *scale + ); + if source_type != *target_type && !decimal_unchanged { + *value = value.clone().promote_iceberg(&source_type, target_type)?; + } + } + Ok(()) + } + /// Promote numeric bounds to the current schema before rewriting this file /// into a manifest whose embedded schema has evolved. pub fn promote_bounds_to_schema(&mut self, schema: &Schema) { diff --git a/iceberg-rust-spec/src/spec/manifest_list.rs b/iceberg-rust-spec/src/spec/manifest_list.rs index c7988a38..8a22778c 100644 --- a/iceberg-rust-spec/src/spec/manifest_list.rs +++ b/iceberg-rust-spec/src/spec/manifest_list.rs @@ -394,12 +394,13 @@ impl ManifestListEntry { schema: &Schema, ) -> Result { let types = partition_field_types(table_metadata, entry.partition_spec_id, schema)?; - Self::try_from_v3_with_types(entry, &types) + Self::try_from_v3_with_types(entry, &types, true) } fn try_from_v3_with_types( entry: _serde::ManifestListEntryV3, types: &[Option], + promote: bool, ) -> Result { Ok(ManifestListEntry { format_version: FormatVersion::V3, @@ -416,7 +417,7 @@ impl ManifestListEntry { added_rows_count: Some(entry.added_rows_count), existing_rows_count: Some(entry.existing_rows_count), deleted_rows_count: Some(entry.deleted_rows_count), - partitions: decode_partition_summaries(entry.partitions, types)?, + partitions: decode_partition_summaries(entry.partitions, types, promote)?, key_metadata: entry.key_metadata, first_row_id: entry.first_row_id, }) @@ -440,12 +441,13 @@ impl ManifestListEntry { schema: &Schema, ) -> Result { let types = partition_field_types(table_metadata, entry.partition_spec_id, schema)?; - Self::try_from_v2_with_types(entry, &types) + Self::try_from_v2_with_types(entry, &types, true) } fn try_from_v2_with_types( entry: _serde::ManifestListEntryV2, types: &[Option], + promote: bool, ) -> Result { Ok(ManifestListEntry { format_version: FormatVersion::V2, @@ -462,7 +464,7 @@ impl ManifestListEntry { added_rows_count: Some(entry.added_rows_count), existing_rows_count: Some(entry.existing_rows_count), deleted_rows_count: Some(entry.deleted_rows_count), - partitions: decode_partition_summaries(entry.partitions, types)?, + partitions: decode_partition_summaries(entry.partitions, types, promote)?, key_metadata: entry.key_metadata, first_row_id: None, }) @@ -486,12 +488,13 @@ impl ManifestListEntry { schema: &Schema, ) -> Result { let types = partition_field_types(table_metadata, entry.partition_spec_id, schema)?; - Self::try_from_v1_with_types(entry, &types) + Self::try_from_v1_with_types(entry, &types, true) } fn try_from_v1_with_types( entry: _serde::ManifestListEntryV1, types: &[Option], + promote: bool, ) -> Result { Ok(ManifestListEntry { format_version: FormatVersion::V1, @@ -508,7 +511,7 @@ impl ManifestListEntry { added_rows_count: entry.added_rows_count, existing_rows_count: entry.existing_rows_count, deleted_rows_count: entry.deleted_rows_count, - partitions: decode_partition_summaries(entry.partitions, types)?, + partitions: decode_partition_summaries(entry.partitions, types, promote)?, key_metadata: entry.key_metadata, first_row_id: None, }) @@ -602,9 +605,10 @@ impl<'a> ManifestListEntryDecoder<'a> { match format_version { FormatVersion::V1 => { let entry = apache_avro::from_value::(&entry)?; + let promote = self.schema.is_some(); let (types, has_partition_types) = self.partition_types(entry.partition_spec_id, entry.added_snapshot_id)?; - let mut decoded = ManifestListEntry::try_from_v1_with_types(entry, types)?; + let mut decoded = ManifestListEntry::try_from_v1_with_types(entry, types, promote)?; if !has_partition_types { decoded.partitions = None; } @@ -612,9 +616,10 @@ impl<'a> ManifestListEntryDecoder<'a> { } FormatVersion::V2 => { let entry = apache_avro::from_value::(&entry)?; + let promote = self.schema.is_some(); let (types, has_partition_types) = self.partition_types(entry.partition_spec_id, entry.added_snapshot_id)?; - let mut decoded = ManifestListEntry::try_from_v2_with_types(entry, types)?; + let mut decoded = ManifestListEntry::try_from_v2_with_types(entry, types, promote)?; if !has_partition_types { decoded.partitions = None; } @@ -622,9 +627,10 @@ impl<'a> ManifestListEntryDecoder<'a> { } FormatVersion::V3 => { let entry = apache_avro::from_value::(&entry)?; + let promote = self.schema.is_some(); let (types, has_partition_types) = self.partition_types(entry.partition_spec_id, entry.added_snapshot_id)?; - let mut decoded = ManifestListEntry::try_from_v3_with_types(entry, types)?; + let mut decoded = ManifestListEntry::try_from_v3_with_types(entry, types, promote)?; if !has_partition_types { decoded.partitions = None; } @@ -634,12 +640,11 @@ impl<'a> ManifestListEntryDecoder<'a> { } } -/// Decodes an entry's partition field summaries, resolving types from the entry's own -/// `partition_spec_id` against the scan schema and decoding each bound with the spec's -/// width-inference rule ([`super::manifest::decode_bound`]). +/// Decodes partition summaries against the selected schema, promoting bounds only for scans. fn decode_partition_summaries( partitions: Option>, types: &[Option], + promote: bool, ) -> Result>, Error> { let Some(summaries) = partitions else { return Ok(None); @@ -650,18 +655,32 @@ fn decode_partition_summaries( summaries .into_iter() .zip(types.iter()) - .map(|(summary, data_type)| FieldSummary::from_serde(summary, data_type.as_ref())) + .map(|(summary, data_type)| { + FieldSummary::from_serde_with_mode(summary, data_type.as_ref(), promote) + }) .collect::, Error>>() .map(Some) } impl FieldSummary { + #[cfg(test)] fn from_serde(value: FieldSummarySerde, data_type: Option<&Type>) -> Result { + Self::from_serde_with_mode(value, data_type, true) + } + + fn from_serde_with_mode( + value: FieldSummarySerde, + data_type: Option<&Type>, + promote: bool, + ) -> Result { let decode = |bytes: Option| -> Result, Error> { match (bytes, data_type) { - (Some(bytes), Some(data_type)) => { + (Some(bytes), Some(data_type)) if promote => { super::manifest::decode_bound(&bytes, data_type).map(Some) } + (Some(bytes), Some(data_type)) => { + super::manifest::decode_bound_for_copy(&bytes, data_type).map(Some) + } _ => Ok(None), } }; diff --git a/iceberg-rust/src/table/manifest.rs b/iceberg-rust/src/table/manifest.rs index a138acb4..b4c084d8 100644 --- a/iceberg-rust/src/table/manifest.rs +++ b/iceberg-rust/src/table/manifest.rs @@ -488,6 +488,11 @@ impl<'schema, 'metadata> ManifestWriter<'schema, 'metadata> { schema: &'schema AvroSchema, table_metadata: &'metadata TableMetadata, ) -> Result<(Self, FilteredManifestStats), Error> { + if manifest.partition_spec_id != table_metadata.default_spec_id { + return Err(Error::NotSupported( + "COW rewrite across partition-spec evolution".to_owned(), + )); + } let inherited_snapshot_id = manifest.added_snapshot_id; let manifest_reader = ManifestReader::new(bytes)?; let mut next_inherited_row_id = manifest.first_row_id; @@ -496,6 +501,22 @@ impl<'schema, 'metadata> ManifestWriter<'schema, 'metadata> { let mut filtered_stats = FilteredManifestStats::default(); let mut existing_files = 0; let mut existing_rows = 0; + let current_spec = table_metadata.default_partition_spec()?; + let current_partition_types = + current_spec.data_types(table_metadata.current_schema()?.fields())?; + manifest.partitions = Some( + current_spec + .fields() + .iter() + .map(|_| FieldSummary { + contains_null: false, + contains_nan: None, + lower_bound: None, + upper_bound: None, + }) + .collect(), + ); + manifest.partition_spec_id = table_metadata.default_spec_id; writer.add_user_metadata( "format-version".to_string(), @@ -578,6 +599,16 @@ impl<'schema, 'metadata> ManifestWriter<'schema, 'metadata> { existing_files += 1; existing_rows += entry.data_file().record_count(); *entry.status_mut() = Status::Existing; + entry + .data_file_mut() + .promote_partition_to_types(current_spec, ¤t_partition_types)?; + update_partitions( + manifest.partitions.as_mut().ok_or_else(|| { + Error::InvalidFormat("Missing partition summaries".to_owned()) + })?, + entry.data_file().partition(), + current_spec.fields(), + )?; writer.append_ser(entry)?; } } diff --git a/iceberg-rust/src/table/manifest_list.rs b/iceberg-rust/src/table/manifest_list.rs index 0ac657a3..aad2ae66 100644 --- a/iceberg-rust/src/table/manifest_list.rs +++ b/iceberg-rust/src/table/manifest_list.rs @@ -19,6 +19,7 @@ use iceberg_rust_spec::{ schema::Schema, snapshot::Snapshot, table_metadata::{FormatVersion, TableMetadata}, + types::Type, util::strip_prefix, }; @@ -65,6 +66,55 @@ pub(crate) struct ManifestListReader<'a, 'metadata, R: Read> { writer_format_version: FormatVersion, } +pub(crate) struct ManifestRewriteCompatibility { + current_spec_id: i32, + compatible_schema_ids: HashMap, + partition_types: Vec, +} + +impl ManifestRewriteCompatibility { + pub(crate) fn new(metadata: &TableMetadata) -> Result { + let spec = metadata.default_partition_spec()?; + let partition_types = spec.data_types(metadata.current_schema()?.fields())?; + Ok(Self { + current_spec_id: metadata.default_spec_id, + compatible_schema_ids: HashMap::new(), + partition_types, + }) + } + + pub(crate) fn allows( + &mut self, + manifest: &ManifestListEntry, + metadata: &TableMetadata, + ) -> bool { + if manifest.partition_spec_id != self.current_spec_id { + return false; + } + if let Some(schema_id) = metadata + .snapshots + .get(&manifest.added_snapshot_id) + .and_then(|snapshot| *snapshot.schema_id()) + { + if let Some(schema) = metadata.schemas.get(&schema_id) { + if let Some(compatible) = self.compatible_schema_ids.get(&schema_id) { + return *compatible; + } + let compatible = metadata + .partition_specs + .get(&self.current_spec_id) + .and_then(|spec| spec.data_types(schema.fields()).ok()) + .is_some_and(|types| types == self.partition_types); + self.compatible_schema_ids.insert(schema_id, compatible); + return compatible; + } + } + + // Width inference can make old partition values look current after snapshot expiry. + false + } +} + impl Iterator for ManifestListReader<'_, '_, R> { type Item = Result; fn next(&mut self) -> Option { @@ -827,13 +877,14 @@ impl<'schema, 'metadata> ManifestListWriter<'schema, 'metadata> { &mut writer, None, &bounding_partition_values, + table_metadata, )? }; Ok(Self { table_metadata, writer, - selected_data_manifest: Some(data_manifest), + selected_data_manifest: data_manifest, selected_delete_manifest: delete_manifest, bounding_partition_values, n_existing_files: file_count_all_entries, @@ -971,6 +1022,7 @@ impl<'schema, 'metadata> ManifestListWriter<'schema, 'metadata> { row_id_assigner.as_mut(), &bounding_partition_values, manifests_to_overwrite, + table_metadata, )? }; @@ -978,7 +1030,7 @@ impl<'schema, 'metadata> ManifestListWriter<'schema, 'metadata> { Self { table_metadata, writer, - selected_data_manifest: Some(manifest), + selected_data_manifest: manifest, selected_delete_manifest: None, bounding_partition_values, n_existing_files: file_count_all_entries, @@ -1804,14 +1856,17 @@ impl<'schema, 'metadata> ManifestListWriter<'schema, 'metadata> { mod tests { use super::*; use iceberg_rust_spec::{ - manifest_list::{manifest_list_schema_v1, manifest_list_schema_v3, FieldSummary}, + manifest::FileFormat, + manifest_list::{ + manifest_list_schema_v1, manifest_list_schema_v2, manifest_list_schema_v3, FieldSummary, + }, spec::{ partition::{PartitionField, PartitionSpec, Transform}, schema::SchemaBuilder, snapshot::{SnapshotBuilder, Summary}, table_metadata::TableMetadataBuilder, types::{PrimitiveType, StructField, Type}, - values::Value, + values::{Struct, Value}, }, }; @@ -2078,6 +2133,191 @@ mod tests { } } + #[test] + fn v2_writes_after_partition_promotion_keep_old_manifests() { + let schema = |schema_id, primitive_type| { + SchemaBuilder::default() + .with_schema_id(schema_id) + .with_struct_field(StructField { + id: 1, + name: "partition_source".to_string(), + required: true, + field_type: Type::Primitive(primitive_type), + doc: None, + initial_default: None, + write_default: None, + }) + .build() + .unwrap() + }; + let snapshot = |snapshot_id, schema_id| { + SnapshotBuilder::default() + .with_snapshot_id(snapshot_id) + .with_sequence_number(snapshot_id) + .with_timestamp_ms(snapshot_id) + .with_manifest_list(format!("snapshot-{snapshot_id}.avro")) + .with_summary(Summary::default()) + .with_schema_id(schema_id) + .build() + .unwrap() + }; + let metadata = TableMetadataBuilder::default() + .format_version(FormatVersion::V2) + .location("s3://tests/table".to_string()) + .current_schema_id(2) + .schemas(HashMap::from([ + (1, schema(1, PrimitiveType::Int)), + (2, schema(2, PrimitiveType::Long)), + ])) + .default_spec_id(0) + .partition_specs(HashMap::from([( + 0, + PartitionSpec::builder() + .with_spec_id(0) + .with_partition_field(PartitionField::new( + 1, + 1000, + "partition_source", + Transform::Identity, + )) + .build() + .unwrap(), + )])) + .snapshots(HashMap::from([(7, snapshot(7, 1)), (8, snapshot(8, 2))])) + .build() + .unwrap(); + let old_entry = ManifestListEntry { + format_version: FormatVersion::V2, + manifest_path: "old-manifest.avro".to_string(), + manifest_length: 1, + partition_spec_id: 0, + content: Content::Data, + sequence_number: 1, + min_sequence_number: 1, + added_snapshot_id: 7, + added_files_count: Some(1), + existing_files_count: Some(0), + deleted_files_count: Some(0), + added_rows_count: Some(1), + existing_rows_count: Some(0), + deleted_rows_count: Some(0), + partitions: Some(vec![FieldSummary { + contains_null: false, + contains_nan: None, + lower_bound: Some(Value::Int(42)), + upper_bound: Some(Value::Int(42)), + }]), + key_metadata: None, + first_row_id: None, + }; + let new_file = DataFile::builder() + .with_content(iceberg_rust_spec::manifest::Content::Data) + .with_file_path("new-file.parquet".to_string()) + .with_file_format(FileFormat::Parquet) + .with_partition(Struct::from_iter([( + "partition_source".to_string(), + Some(Value::LongInt(-100)), + )])) + .with_record_count(1) + .with_file_size_in_bytes(100) + .with_column_sizes(None) + .with_value_counts(None) + .with_null_value_counts(None) + .with_nan_value_counts(None) + .with_distinct_counts(None) + .with_lower_bounds(None) + .with_upper_bounds(None) + .build() + .unwrap(); + let mut current_entry = old_entry.clone(); + current_entry.manifest_path = "current-manifest.avro".to_string(); + current_entry.added_snapshot_id = 8; + current_entry.partitions = Some(vec![FieldSummary { + contains_null: false, + contains_nan: None, + lower_bound: Some(Value::LongInt(10)), + upper_bound: Some(Value::LongInt(10)), + }]); + let mut compatibility = ManifestRewriteCompatibility::new(&metadata).unwrap(); + let mut missing_snapshot = current_entry.clone(); + missing_snapshot.added_snapshot_id = 999; + assert!(!compatibility.allows(&missing_snapshot, &metadata)); + missing_snapshot.partitions = None; + assert!(!compatibility.allows(&missing_snapshot, &metadata)); + let mut old_missing_snapshot = old_entry.clone(); + old_missing_snapshot.added_snapshot_id = 999; + assert!(!compatibility.allows(&old_missing_snapshot, &metadata)); + + for entries in [ + vec![old_entry.clone()], + vec![old_entry.clone(), current_entry.clone()], + ] { + let mut avro = AvroWriter::new(manifest_list_schema_v2(), Vec::new()); + for entry in &entries { + avro.append_ser(entry).unwrap(); + } + let bytes = avro.into_inner().unwrap(); + let selected_path = + (entries.len() == 2).then_some(current_entry.manifest_path.as_str()); + + let append = ManifestListWriter::from_existing( + &bytes, + std::iter::once(&new_file), + manifest_list_schema_v2(), + &metadata, + ) + .unwrap(); + assert_eq!( + append + .selected_data_manifest() + .map(|entry| entry.manifest_path.as_str()), + selected_path + ); + let copied_bytes = append.writer.into_inner().unwrap(); + let copied = ManifestListReader::new(copied_bytes.as_slice(), &metadata) + .unwrap() + .collect::, _>>() + .unwrap(); + assert_eq!(copied, vec![old_entry.clone()]); + + let (overwrite, affected) = ManifestListWriter::from_existing_without_overwrites( + &bytes, + std::iter::once(&new_file), + &HashSet::from([old_entry.manifest_path.clone()]), + manifest_list_schema_v2(), + &metadata, + ) + .unwrap(); + assert_eq!( + overwrite + .selected_data_manifest() + .map(|entry| entry.manifest_path.as_str()), + selected_path + ); + assert_eq!(affected, vec![old_entry.clone()]); + } + + let mut expired_metadata = metadata.clone(); + expired_metadata.snapshots.remove(&7); + let mut avro = AvroWriter::new(manifest_list_schema_v2(), Vec::new()); + avro.append_ser(&old_entry).unwrap(); + let bytes = avro.into_inner().unwrap(); + let copied = ManifestListWriter::from_existing( + &bytes, + std::iter::once(&new_file), + manifest_list_schema_v2(), + &expired_metadata, + ) + .unwrap(); + assert!(copied.selected_data_manifest().is_none()); + let copied_bytes = copied.writer.into_inner().unwrap(); + let copied_entries = ManifestListReader::new(copied_bytes.as_slice(), &expired_metadata) + .unwrap() + .collect::, _>>() + .unwrap(); + assert_eq!(copied_entries, vec![old_entry]); + } + fn manifest( path: &str, content: Content, diff --git a/iceberg-rust/src/table/transaction/append.rs b/iceberg-rust/src/table/transaction/append.rs index 8f213fea..7c063ceb 100644 --- a/iceberg-rust/src/table/transaction/append.rs +++ b/iceberg-rust/src/table/transaction/append.rs @@ -1,6 +1,6 @@ use iceberg_rust_spec::{ manifest::Content, manifest::DataFile, manifest::ManifestEntry, - manifest_list::ManifestListEntry, + manifest_list::ManifestListEntry, table_metadata::TableMetadata, }; use smallvec::SmallVec; use std::cmp::Ordering; @@ -8,7 +8,9 @@ use std::collections::HashMap; use crate::{ error::Error, - table::manifest_list::{append_manifest, ManifestListReader, RowIdAssigner}, + table::manifest_list::{ + append_manifest, ManifestListReader, ManifestRewriteCompatibility, RowIdAssigner, + }, util::{cmp_with_priority, partition_struct_to_vec, summary_to_rectangle, try_sub, Rectangle}, }; @@ -116,7 +118,7 @@ pub(crate) fn split_datafiles( } pub(crate) struct SelectedManifest { - pub data_manifest: ManifestListEntry, + pub data_manifest: Option, pub delete_manifest: Option, pub file_count_all_entries: usize, } @@ -128,13 +130,25 @@ pub(crate) fn select_manifest_partitioned( manifest_list_writer: &mut apache_avro::Writer>, mut row_id_assigner: Option<&mut RowIdAssigner>, bounding_partition_values: &Rectangle, + table_metadata: &TableMetadata, ) -> Result { + let mut compatibility = ManifestRewriteCompatibility::new(table_metadata)?; let mut selected_data_state = None; let mut selected_delete_state = None; let mut file_count_all_entries = 0; for manifest_res in manifest_list_reader { let manifest = manifest_res?; + file_count_all_entries += manifest.added_files_count.unwrap_or(0) as usize; + if !compatibility.allows(&manifest, table_metadata) { + append_manifest( + manifest_list_writer, + row_id_assigner.as_deref_mut(), + manifest, + )?; + continue; + } + let mut bounds = summary_to_rectangle(manifest.partitions.as_ref().ok_or(Error::NotFound(format!( "Partition struct in manifest {}", @@ -143,8 +157,6 @@ pub(crate) fn select_manifest_partitioned( bounds.expand(bounding_partition_values); - file_count_all_entries += manifest.added_files_count.unwrap_or(0) as usize; - match manifest.content { iceberg_rust_spec::manifest_list::Content::Data => { let Some((selected_bounds, _)) = &selected_data_state else { @@ -212,11 +224,8 @@ pub(crate) fn select_manifest_partitioned( } } } - let (_, data_manifest) = - selected_data_state.ok_or(Error::NotFound("Manifest for insert".to_owned()))?; - Ok(SelectedManifest { - data_manifest, + data_manifest: selected_data_state.map(|(_, manifest)| manifest), delete_manifest: selected_delete_state.map(|(_, x)| x), file_count_all_entries, }) @@ -310,7 +319,7 @@ pub(crate) fn select_manifest_unpartitioned( selected_data_state.ok_or(Error::NotFound("Manifest for insert".to_owned()))?; Ok(SelectedManifest { - data_manifest, + data_manifest: Some(data_manifest), delete_manifest: selected_delete_state.map(|(_, x)| x), file_count_all_entries, }) diff --git a/iceberg-rust/src/table/transaction/operation.rs b/iceberg-rust/src/table/transaction/operation.rs index b8d702bb..4e34a633 100644 --- a/iceberg-rust/src/table/transaction/operation.rs +++ b/iceberg-rust/src/table/transaction/operation.rs @@ -797,7 +797,18 @@ impl Operation { if !filtered_stats.filtered_entries.is_empty() { let n_filtered_files = filtered_stats.filtered_entries.len(); - let filtered_iter = filtered_stats.filtered_entries.clone().into_iter().map(Ok); + let current_spec = table_metadata.default_partition_spec()?; + let current_schema = table_metadata.current_schema()?; + let current_partition_types = + current_spec.data_types(current_schema.fields())?; + for entry in &mut filtered_stats.filtered_entries { + entry + .data_file_mut() + .promote_partition_to_types(current_spec, ¤t_partition_types)?; + } + let filtered_iter = std::mem::take(&mut filtered_stats.filtered_entries) + .into_iter() + .map(Ok); let n_filtered_splits = manifest_list_writer.n_splits(n_filtered_files, ManifestListContent::Data); @@ -1300,6 +1311,7 @@ mod tests { use crate::table::ManifestReader; use futures::{executor::block_on, TryStreamExt}; use iceberg_rust_spec::manifest::FileFormat; + use iceberg_rust_spec::spec::partition::{PartitionField, PartitionSpec, Transform}; use iceberg_rust_spec::spec::schema::SchemaBuilder; use iceberg_rust_spec::spec::table_metadata::TableMetadataBuilder; use iceberg_rust_spec::spec::types::{PrimitiveType, StructField, Type}; @@ -1417,6 +1429,487 @@ mod tests { .unwrap() } + fn partitioned_data_file(path: &str, partition_value: Value) -> DataFile { + DataFile::builder() + .with_content(Content::Data) + .with_file_path(path.to_string()) + .with_file_format(FileFormat::Parquet) + .with_partition(Struct::from_iter([( + "partition_source".to_string(), + Some(partition_value), + )])) + .with_record_count(1) + .with_file_size_in_bytes(100) + .with_column_sizes(None) + .with_value_counts(None) + .with_null_value_counts(None) + .with_nan_value_counts(None) + .with_distinct_counts(None) + .with_lower_bounds(None) + .with_upper_bounds(None) + .build() + .unwrap() + } + + #[tokio::test] + async fn v2_append_reuses_only_manifests_with_current_partition_type() { + let mut metadata = sample_metadata(&[], None, &[]); + metadata.schemas.insert( + 0, + SchemaBuilder::default() + .with_schema_id(0) + .with_struct_field(StructField { + id: 1, + name: "partition_source".to_string(), + required: true, + field_type: Type::Primitive(PrimitiveType::Int), + doc: None, + initial_default: None, + write_default: None, + }) + .build() + .unwrap(), + ); + metadata.partition_specs.insert( + 0, + PartitionSpec::builder() + .with_spec_id(0) + .with_partition_field(PartitionField::new( + 1, + 1000, + "partition_source", + Transform::Identity, + )) + .build() + .unwrap(), + ); + let store = Arc::new(InMemory::new()); + + for (path, value) in [ + ("old.parquet", Value::Int(42)), + ("new.parquet", Value::LongInt(-100)), + ("next.parquet", Value::LongInt(10)), + ] { + if path == "new.parquet" { + metadata.schemas.insert( + 1, + SchemaBuilder::default() + .with_schema_id(1) + .with_struct_field(StructField { + id: 1, + name: "partition_source".to_string(), + required: true, + field_type: Type::Primitive(PrimitiveType::Long), + doc: None, + initial_default: None, + write_default: None, + }) + .build() + .unwrap(), + ); + metadata.current_schema_id = 1; + } + let mut data_files = vec![partitioned_data_file(path, value)]; + if path == "old.parquet" { + data_files.push(partitioned_data_file( + "old-survivor.parquet", + Value::Int(43), + )); + } + let append = Operation::Append { + branch: None, + data_files, + delete_files: Vec::new(), + additional_summary: None, + }; + let (_, updates) = append.execute(&metadata, store.clone()).await.unwrap(); + crate::catalog::commit::apply_table_updates(&mut metadata, updates).unwrap(); + } + + let snapshot = metadata.current_snapshot(None).unwrap().unwrap(); + let bytes = store + .get(&strip_prefix(snapshot.manifest_list()).into()) + .await + .unwrap() + .bytes() + .await + .unwrap(); + let manifests = ManifestListReader::new(&bytes[..], &metadata) + .unwrap() + .collect::, _>>() + .unwrap(); + assert_eq!(manifests.len(), 2); + let mut files = crate::table::datafiles(store.clone(), &manifests, None, (None, None)) + .await + .unwrap() + .map_ok(|(_, entry)| entry.data_file().file_path().clone()) + .try_collect::>() + .await + .unwrap(); + files.sort(); + assert_eq!( + files, + vec![ + "new.parquet", + "next.parquet", + "old-survivor.parquet", + "old.parquet" + ] + ); + + let old_manifest = manifests + .iter() + .find(|manifest| { + metadata + .snapshots + .get(&manifest.added_snapshot_id) + .and_then(|snapshot| *snapshot.schema_id()) + == Some(0) + }) + .unwrap(); + let overwrite = Operation::Overwrite { + branch: None, + data_files: vec![partitioned_data_file( + "replacement.parquet", + Value::LongInt(42), + )], + files_to_overwrite: HashMap::from([( + old_manifest.manifest_path.clone(), + vec!["old.parquet".to_string()], + )]), + additional_summary: None, + }; + let (_, updates) = overwrite.execute(&metadata, store.clone()).await.unwrap(); + crate::catalog::commit::apply_table_updates(&mut metadata, updates).unwrap(); + let snapshot = metadata.current_snapshot(None).unwrap().unwrap(); + let bytes = store + .get(&strip_prefix(snapshot.manifest_list()).into()) + .await + .unwrap() + .bytes() + .await + .unwrap(); + let manifests = ManifestListReader::new(&bytes[..], &metadata) + .unwrap() + .collect::, _>>() + .unwrap(); + assert!(manifests.iter().any(|manifest| { + manifest.partitions.as_ref().is_some_and(|partitions| { + partitions.first().is_some_and(|summary| { + summary.lower_bound == Some(Value::LongInt(43)) + && summary.upper_bound == Some(Value::LongInt(43)) + }) + }) + })); + let entries = crate::table::datafiles(store.clone(), &manifests, None, (None, None)) + .await + .unwrap() + .map_ok(|(_, entry)| { + ( + entry.data_file().file_path().clone(), + *entry.status(), + entry + .data_file() + .partition() + .get("partition_source") + .cloned(), + ) + }) + .try_collect::>() + .await + .unwrap(); + assert!(entries + .iter() + .any(|(path, status, _)| { path == "old.parquet" && *status == Status::Deleted })); + assert!(entries.iter().any(|(path, status, partition)| { + path == "old-survivor.parquet" + && *status == Status::Existing + && partition == &Some(Some(Value::LongInt(43))) + })); + let mut files = entries + .into_iter() + .filter_map(|(path, status, _)| (status != Status::Deleted).then_some(path)) + .collect::>(); + files.sort(); + assert_eq!( + files, + vec![ + "new.parquet", + "next.parquet", + "old-survivor.parquet", + "replacement.parquet", + ] + ); + + let (_, updates) = Operation::Append { + branch: None, + data_files: vec![partitioned_data_file( + "post-cow.parquet", + Value::LongInt(50), + )], + delete_files: Vec::new(), + additional_summary: None, + } + .execute(&metadata, store.clone()) + .await + .unwrap(); + crate::catalog::commit::apply_table_updates(&mut metadata, updates).unwrap(); + let snapshot = metadata.current_snapshot(None).unwrap().unwrap(); + let bytes = store + .get(&strip_prefix(snapshot.manifest_list()).into()) + .await + .unwrap() + .bytes() + .await + .unwrap(); + let manifests = ManifestListReader::new(&bytes[..], &metadata) + .unwrap() + .collect::, _>>() + .unwrap(); + let files = crate::table::datafiles(store, &manifests, None, (None, None)) + .await + .unwrap() + .map_ok(|(_, entry)| (entry.data_file().file_path().clone(), *entry.status())) + .try_collect::>() + .await + .unwrap(); + assert!(files + .iter() + .any(|(path, status)| path == "post-cow.parquet" && *status != Status::Deleted)); + } + + #[tokio::test] + async fn v2_cow_split_promotes_old_partition_tombstones() { + let mut metadata = sample_metadata(&[], None, &[]); + let schema = |schema_id, primitive_type| { + SchemaBuilder::default() + .with_schema_id(schema_id) + .with_struct_field(StructField { + id: 1, + name: "partition_source".to_string(), + required: true, + field_type: Type::Primitive(primitive_type), + doc: None, + initial_default: None, + write_default: None, + }) + .build() + .unwrap() + }; + metadata.schemas.insert(0, schema(0, PrimitiveType::Int)); + metadata.partition_specs.insert( + 0, + PartitionSpec::builder() + .with_spec_id(0) + .with_partition_field(PartitionField::new( + 1, + 1000, + "partition_source", + Transform::Identity, + )) + .build() + .unwrap(), + ); + let store = Arc::new(InMemory::new()); + let old_files = (0..8) + .map(|value| partitioned_data_file(&format!("old-{value}.parquet"), Value::Int(value))) + .collect(); + let (_, updates) = Operation::Append { + branch: None, + data_files: old_files, + delete_files: Vec::new(), + additional_summary: None, + } + .execute(&metadata, store.clone()) + .await + .unwrap(); + crate::catalog::commit::apply_table_updates(&mut metadata, updates).unwrap(); + + metadata.schemas.insert(1, schema(1, PrimitiveType::Long)); + metadata.current_schema_id = 1; + let (_, updates) = Operation::Append { + branch: None, + data_files: vec![partitioned_data_file("new.parquet", Value::LongInt(20))], + delete_files: Vec::new(), + additional_summary: None, + } + .execute(&metadata, store.clone()) + .await + .unwrap(); + crate::catalog::commit::apply_table_updates(&mut metadata, updates).unwrap(); + + let snapshot = metadata.current_snapshot(None).unwrap().unwrap(); + let bytes = store + .get(&strip_prefix(snapshot.manifest_list()).into()) + .await + .unwrap() + .bytes() + .await + .unwrap(); + let manifests = ManifestListReader::new(&bytes[..], &metadata) + .unwrap() + .collect::, _>>() + .unwrap(); + let mut files_to_overwrite = HashMap::>::new(); + let entries = crate::table::datafiles(store.clone(), &manifests, None, (None, None)) + .await + .unwrap() + .try_collect::>() + .await + .unwrap(); + for (manifest_path, entry) in entries { + files_to_overwrite + .entry(manifest_path) + .or_default() + .push(entry.data_file().file_path().clone()); + } + assert_eq!(files_to_overwrite.values().map(Vec::len).sum::(), 9); + + let (_, updates) = Operation::Overwrite { + branch: None, + data_files: vec![ + partitioned_data_file("replacement-low.parquet", Value::LongInt(-100)), + partitioned_data_file("replacement-high.parquet", Value::LongInt(100)), + ], + files_to_overwrite, + additional_summary: None, + } + .execute(&metadata, store.clone()) + .await + .unwrap(); + crate::catalog::commit::apply_table_updates(&mut metadata, updates).unwrap(); + + let snapshot = metadata.current_snapshot(None).unwrap().unwrap(); + let bytes = store + .get(&strip_prefix(snapshot.manifest_list()).into()) + .await + .unwrap() + .bytes() + .await + .unwrap(); + let manifests = ManifestListReader::new(&bytes[..], &metadata) + .unwrap() + .collect::, _>>() + .unwrap(); + let entries = crate::table::datafiles(store, &manifests, None, (None, None)) + .await + .unwrap() + .map_ok(|(_, entry)| { + ( + entry.data_file().file_path().clone(), + *entry.status(), + entry + .data_file() + .partition() + .get("partition_source") + .cloned(), + ) + }) + .try_collect::>() + .await + .unwrap(); + let deleted = entries + .iter() + .filter(|(_, status, _)| *status == Status::Deleted) + .collect::>(); + assert_eq!(deleted.len(), 9); + assert!(deleted + .iter() + .all(|(_, _, partition)| { matches!(partition, Some(Some(Value::LongInt(_)))) })); + let mut live = entries + .iter() + .filter_map(|(path, status, _)| (*status != Status::Deleted).then_some(path.as_str())) + .collect::>(); + live.sort_unstable(); + assert_eq!( + live, + vec!["replacement-high.parquet", "replacement-low.parquet"] + ); + } + + #[tokio::test] + async fn v2_cow_rejects_changed_partition_transform() { + let mut metadata = sample_metadata(&[], None, &[]); + metadata.schemas.insert( + 0, + SchemaBuilder::default() + .with_schema_id(0) + .with_struct_field(StructField { + id: 1, + name: "partition_source".to_string(), + required: true, + field_type: Type::Primitive(PrimitiveType::Int), + doc: None, + initial_default: None, + write_default: None, + }) + .build() + .unwrap(), + ); + let spec = |id, transform| { + PartitionSpec::builder() + .with_spec_id(id) + .with_partition_field(PartitionField::new( + 1, + 1000 + id, + "partition_source", + transform, + )) + .build() + .unwrap() + }; + metadata + .partition_specs + .insert(0, spec(0, Transform::Identity)); + let store = Arc::new(InMemory::new()); + let (_, updates) = Operation::Append { + branch: None, + data_files: vec![ + partitioned_data_file("old.parquet", Value::Int(42)), + partitioned_data_file("survivor.parquet", Value::Int(43)), + ], + delete_files: Vec::new(), + additional_summary: None, + } + .execute(&metadata, store.clone()) + .await + .unwrap(); + crate::catalog::commit::apply_table_updates(&mut metadata, updates).unwrap(); + let snapshot = metadata.current_snapshot(None).unwrap().unwrap(); + let bytes = store + .get(&strip_prefix(snapshot.manifest_list()).into()) + .await + .unwrap() + .bytes() + .await + .unwrap(); + let old_manifest = ManifestListReader::new(&bytes[..], &metadata) + .unwrap() + .next() + .unwrap() + .unwrap(); + + metadata + .partition_specs + .insert(1, spec(1, Transform::Bucket(16))); + metadata.default_spec_id = 1; + let error = Operation::Overwrite { + branch: None, + data_files: vec![partitioned_data_file("replacement.parquet", Value::Int(5))], + files_to_overwrite: HashMap::from([( + old_manifest.manifest_path, + vec!["old.parquet".to_string()], + )]), + additional_summary: None, + } + .execute(&metadata, store) + .await + .unwrap_err(); + assert!( + matches!(error, Error::NotSupported(message) if message == "COW rewrite across partition-spec evolution") + ); + } + async fn clear_current_v1_manifest_list_counts( metadata: &TableMetadata, store: &Arc, diff --git a/iceberg-rust/src/table/transaction/overwrite.rs b/iceberg-rust/src/table/transaction/overwrite.rs index c7157045..f5a34498 100644 --- a/iceberg-rust/src/table/transaction/overwrite.rs +++ b/iceberg-rust/src/table/transaction/overwrite.rs @@ -1,15 +1,17 @@ use std::{cmp::Ordering, collections::HashSet}; -use iceberg_rust_spec::manifest_list::ManifestListEntry; +use iceberg_rust_spec::{manifest_list::ManifestListEntry, table_metadata::TableMetadata}; use crate::{ error::Error, - table::manifest_list::{append_manifest, ManifestListReader, RowIdAssigner}, + table::manifest_list::{ + append_manifest, ManifestListReader, ManifestRewriteCompatibility, RowIdAssigner, + }, util::{summary_to_rectangle, Rectangle}, }; pub(crate) struct OverwriteManifest { - pub manifest: ManifestListEntry, + pub manifest: Option, pub file_count_all_entries: usize, pub manifests_to_overwrite: Vec, } @@ -22,13 +24,29 @@ pub(crate) fn select_manifest_without_overwrites_partitioned( mut row_id_assigner: Option<&mut RowIdAssigner>, bounding_partition_values: &Rectangle, overwrites: &HashSet, + table_metadata: &TableMetadata, ) -> Result { + let mut compatibility = ManifestRewriteCompatibility::new(table_metadata)?; let mut selected_state = None; let mut file_count_all_entries = 0; let mut manifests_to_overwrite = Vec::new(); for manifest_res in manifest_list_reader { let manifest = manifest_res?; + file_count_all_entries += manifest.added_files_count.unwrap_or(0) as usize; + if !compatibility.allows(&manifest, table_metadata) { + if overwrites.contains(&manifest.manifest_path) { + manifests_to_overwrite.push(manifest); + } else { + append_manifest( + manifest_list_writer, + row_id_assigner.as_deref_mut(), + manifest, + )?; + } + continue; + } + let mut bounds = summary_to_rectangle(manifest.partitions.as_ref().ok_or(Error::NotFound(format!( "Partition struct in manifest {}", @@ -37,8 +55,6 @@ pub(crate) fn select_manifest_without_overwrites_partitioned( bounds.expand(bounding_partition_values); - file_count_all_entries += manifest.added_files_count.unwrap_or(0) as usize; - let Some((selected_bounds, _)) = &selected_state else { selected_state = Some((bounds, manifest)); continue; @@ -71,13 +87,11 @@ pub(crate) fn select_manifest_without_overwrites_partitioned( } } } - selected_state - .map(|(_, entry)| OverwriteManifest { - manifest: entry, - file_count_all_entries, - manifests_to_overwrite, - }) - .ok_or(Error::NotFound("Manifest for insert".to_owned())) + Ok(OverwriteManifest { + manifest: selected_state.map(|(_, entry)| entry), + file_count_all_entries, + manifests_to_overwrite, + }) } /// Select the manifest with the smallest number of rows. @@ -132,7 +146,7 @@ pub(crate) fn select_manifest_without_overwrites_unpartitioned( } selected_state .map(|(_, entry)| OverwriteManifest { - manifest: entry, + manifest: Some(entry), file_count_all_entries, manifests_to_overwrite, })