diff --git a/iceberg-rust-spec/src/spec/manifest.rs b/iceberg-rust-spec/src/spec/manifest.rs index be288548..9318ade8 100644 --- a/iceberg-rust-spec/src/spec/manifest.rs +++ b/iceberg-rust-spec/src/spec/manifest.rs @@ -74,6 +74,11 @@ impl ManifestEntry { &mut self.status } + /// Returns a mutable reference to the encoding version used when serializing this entry. + pub fn format_version_mut(&mut self) -> &mut FormatVersion { + &mut self.format_version + } + /// Returns a mutable reference to the sequence number field of this manifest entry. /// /// The sequence number tracks the order of changes to a table. Modifying this allows diff --git a/iceberg-rust/src/table/manifest.rs b/iceberg-rust/src/table/manifest.rs index b4c084d8..7d2ce334 100644 --- a/iceberg-rust/src/table/manifest.rs +++ b/iceberg-rust/src/table/manifest.rs @@ -62,6 +62,7 @@ type ReaderMap<'a, R> = Map< /// * `R` - The type implementing `Read` that provides the manifest data pub(crate) struct ManifestReader<'a, R: Read> { reader: ReaderMap<'a, R>, + metadata: Arc<(Schema, PartitionSpec, FormatVersion)>, } impl Iterator for ManifestReader<'_, R> { @@ -136,12 +137,18 @@ impl ManifestReader<'_, R> { .with_spec_id(spec_id) .with_fields(partition_fields) .build()?; + let metadata = Arc::new((schema, partition_spec, format_version)); Ok(Self { reader: reader - .zip(repeat(Arc::new((schema, partition_spec, format_version)))) + .zip(repeat(metadata.clone())) .map(avro_value_to_manifest_entry), + metadata, }) } + + pub(crate) fn metadata(&self) -> Arc<(Schema, PartitionSpec, FormatVersion)> { + self.metadata.clone() + } } /// A writer for Iceberg manifest files that handles creating and updating manifest entries. @@ -323,11 +330,11 @@ impl<'schema, 'metadata> ManifestWriter<'schema, 'metadata> { /// sequence numbers and snapshot IDs as needed. /// /// # Arguments - /// * `bytes` - The raw bytes of the existing manifest file + /// * `manifest_reader` - The existing manifest and its embedded metadata /// * `manifest` - The manifest list entry describing the existing manifest + /// * `snapshot_id` - The snapshot ID for the rewritten manifest /// * `schema` - The Avro schema used for serializing manifest entries /// * `table_metadata` - The table metadata containing schema and partition information - /// * `branch` - Optional branch name to get the current schema from /// /// # Returns /// * `Result` - A new ManifestWriter instance or an error if initialization fails @@ -349,6 +356,7 @@ impl<'schema, 'metadata> ManifestWriter<'schema, 'metadata> { let mut writer = AvroWriter::new(schema, Vec::new()); let mut existing_files = 0; let mut existing_rows = 0; + let mut min_live_sequence = None; writer.add_user_metadata( "format-version".to_string(), match table_metadata.format_version { @@ -406,10 +414,15 @@ impl<'schema, 'metadata> ManifestWriter<'schema, 'metadata> { if *entry.status() == Status::Deleted { continue; } + *entry.format_version_mut() = table_metadata.format_version; *entry.status_mut() = Status::Existing; if entry.sequence_number().is_none() { *entry.sequence_number_mut() = Some(manifest.sequence_number); } + if let Some(sequence) = entry.sequence_number() { + min_live_sequence = + Some(min_live_sequence.map_or(*sequence, |min: i64| min.min(*sequence))); + } if entry.snapshot_id().is_none() { *entry.snapshot_id_mut() = Some(inherited_snapshot_id); } @@ -420,9 +433,7 @@ impl<'schema, 'metadata> ManifestWriter<'schema, 'metadata> { manifest.sequence_number = table_metadata.last_sequence_number + 1; manifest.added_snapshot_id = snapshot_id; - if existing_files == 0 { - manifest.min_sequence_number = manifest.sequence_number; - } + manifest.min_sequence_number = min_live_sequence.unwrap_or(manifest.sequence_number); manifest.existing_files_count = Some(existing_files); manifest.existing_rows_count = Some(existing_rows); @@ -447,21 +458,21 @@ impl<'schema, 'metadata> ManifestWriter<'schema, 'metadata> { /// Creates a ManifestWriter from an existing manifest file with selective filtering of entries. /// - /// This method reads an existing manifest file and creates a new writer that includes - /// only the entries whose file paths are NOT in the provided filter set. Entries that - /// pass the filter have their status updated to "Existing" and their sequence numbers - /// and snapshot IDs updated as needed. + /// This method reads an existing manifest file and creates a new writer with surviving + /// entries marked "Existing". For an old partition spec, filtered entries are retained + /// as tombstones in the source-spec manifest rather than relabeled with the current spec. /// /// This is particularly useful for overwrite operations where specific files need to be /// excluded from the new manifest while preserving other existing entries. /// /// # Arguments - /// * `bytes` - The raw bytes of the existing manifest file + /// * `manifest_reader` - The existing manifest and its embedded metadata /// * `manifest` - The manifest list entry describing the existing manifest /// * `filter` - A set of file paths to exclude from the new manifest + /// * `snapshot_id` - The snapshot ID for the rewritten manifest /// * `schema` - The Avro schema used for serializing manifest entries /// * `table_metadata` - The table metadata containing schema and partition information - /// * `branch` - Optional branch name to get the current schema from + /// * `source_metadata` - Original schema and spec for a manifest written under an older spec /// /// # Returns /// * `Result` - A new ManifestWriter instance or an error if initialization fails @@ -474,38 +485,54 @@ impl<'schema, 'metadata> ManifestWriter<'schema, 'metadata> { /// * The partition spec ID is not found in table metadata /// /// # Behavior - /// - Entries whose file paths are in the `filter` set are excluded from the new manifest + /// - Filtered entries are emitted separately for the current spec or retained for an old spec /// - Remaining entries have their status set to `Status::Existing` /// - Sequence numbers are updated for entries that don't have them /// - Snapshot IDs are updated for entries that don't have them /// - The manifest's sequence number is incremented /// - File counts are updated to reflect the filtered entries - pub(crate) fn from_existing_with_filter( - bytes: &[u8], + pub(crate) fn from_existing_with_filter( + manifest_reader: ManifestReader<'_, R>, mut manifest: ManifestListEntry, filter: &HashSet, snapshot_id: i64, schema: &'schema AvroSchema, table_metadata: &'metadata TableMetadata, + source_metadata: Option<(&Schema, &PartitionSpec)>, ) -> Result<(Self, FilteredManifestStats), Error> { - if manifest.partition_spec_id != table_metadata.default_spec_id { + let old_spec = manifest.partition_spec_id != table_metadata.default_spec_id; + if old_spec && source_metadata.is_none() { return Err(Error::NotSupported( "COW rewrite across partition-spec evolution".to_owned(), )); } + let (write_schema, write_spec) = match source_metadata { + Some(metadata) => metadata, + None => ( + table_metadata.current_schema()?, + table_metadata.default_partition_spec()?, + ), + }; + if *write_spec.spec_id() != manifest.partition_spec_id + || table_metadata.partition_specs.get(write_spec.spec_id()) != Some(write_spec) + { + return Err(Error::InvalidFormat( + "Manifest partition spec does not match table metadata".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; let mut writer = AvroWriter::new(schema, Vec::new()); 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())?; + let mut deleted_files = 0; + let mut deleted_rows = 0; + let mut min_live_sequence = None; + let partition_types = write_spec.data_types(write_schema.fields())?; manifest.partitions = Some( - current_spec + write_spec .fields() .iter() .map(|_| FieldSummary { @@ -516,7 +543,7 @@ impl<'schema, 'metadata> ManifestWriter<'schema, 'metadata> { }) .collect(), ); - manifest.partition_spec_id = table_metadata.default_spec_id; + manifest.partition_spec_id = *write_spec.spec_id(); writer.add_user_metadata( "format-version".to_string(), @@ -530,31 +557,25 @@ impl<'schema, 'metadata> ManifestWriter<'schema, 'metadata> { writer.add_user_metadata( "schema".to_string(), match table_metadata.format_version { - FormatVersion::V1 => serde_json::to_string(&Into::::into( - table_metadata.current_schema()?.clone(), - ))?, - FormatVersion::V2 | FormatVersion::V3 => serde_json::to_string( - &Into::::into(table_metadata.current_schema()?.clone()), - )?, + FormatVersion::V1 => { + serde_json::to_string(&Into::::into(write_schema.clone()))? + } + FormatVersion::V2 | FormatVersion::V3 => { + serde_json::to_string(&Into::::into(write_schema.clone()))? + } }, )?; writer.add_user_metadata( "schema-id".to_string(), - serde_json::to_string(&table_metadata.current_schema()?.schema_id())?, + serde_json::to_string(write_schema.schema_id())?, )?; - let spec_id = table_metadata.default_spec_id; + let spec_id = *write_spec.spec_id(); writer.add_user_metadata( "partition-spec".to_string(), - serde_json::to_string( - &table_metadata - .partition_specs - .get(&spec_id) - .ok_or(Error::NotFound(format!("Partition spec with id {spec_id}")))? - .fields(), - )?, + serde_json::to_string(write_spec.fields())?, )?; writer.add_user_metadata( @@ -586,6 +607,7 @@ impl<'schema, 'metadata> ManifestWriter<'schema, 'metadata> { if entry.snapshot_id().is_none() { *entry.snapshot_id_mut() = Some(inherited_snapshot_id); } + *entry.format_version_mut() = table_metadata.format_version; if filter.contains(entry.data_file().file_path()) { if *entry.data_file().content() == Content::Data { @@ -594,20 +616,42 @@ impl<'schema, 'metadata> ManifestWriter<'schema, 'metadata> { filtered_stats.removed_file_size_bytes += entry.data_file().file_size_in_bytes(); filtered_stats.removed_data_files += 1; *entry.status_mut() = Status::Deleted; - filtered_stats.filtered_entries.push(entry); + *entry.snapshot_id_mut() = Some(snapshot_id); + if old_spec { + // A tombstone must use the same partition spec as the file it removes. + entry + .data_file_mut() + .promote_partition_to_types(write_spec, &partition_types)?; + update_partitions( + manifest.partitions.as_mut().ok_or_else(|| { + Error::InvalidFormat("Missing partition summaries".to_owned()) + })?, + entry.data_file().partition(), + write_spec.fields(), + )?; + deleted_files += 1; + deleted_rows += entry.data_file().record_count(); + writer.append_ser(entry)?; + } else { + filtered_stats.filtered_entries.push(entry); + } } else { existing_files += 1; existing_rows += entry.data_file().record_count(); + if let Some(sequence) = entry.sequence_number() { + min_live_sequence = + Some(min_live_sequence.map_or(*sequence, |min: i64| min.min(*sequence))); + } *entry.status_mut() = Status::Existing; entry .data_file_mut() - .promote_partition_to_types(current_spec, ¤t_partition_types)?; + .promote_partition_to_types(write_spec, &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(), + write_spec.fields(), )?; writer.append_ser(entry)?; } @@ -615,9 +659,7 @@ impl<'schema, 'metadata> ManifestWriter<'schema, 'metadata> { manifest.sequence_number = table_metadata.last_sequence_number + 1; manifest.added_snapshot_id = snapshot_id; - if existing_files == 0 { - manifest.min_sequence_number = manifest.sequence_number; - } + manifest.min_sequence_number = min_live_sequence.unwrap_or(manifest.sequence_number); manifest.existing_files_count = Some(existing_files); manifest.existing_rows_count = Some(existing_rows); @@ -630,8 +672,8 @@ impl<'schema, 'metadata> ManifestWriter<'schema, 'metadata> { // `add_file` call to repopulate them. manifest.added_files_count = Some(0); manifest.added_rows_count = Some(0); - manifest.deleted_files_count = Some(0); - manifest.deleted_rows_count = Some(0); + manifest.deleted_files_count = Some(deleted_files); + manifest.deleted_rows_count = Some(deleted_rows); Ok(( ManifestWriter { @@ -776,7 +818,7 @@ impl<'schema, 'metadata> ManifestWriter<'schema, 'metadata> { let manifest_length: i64 = manifest_bytes.len() as i64; - self.manifest.manifest_length += manifest_length; + self.manifest.manifest_length = manifest_length; object_store .put( @@ -817,7 +859,7 @@ impl<'schema, 'metadata> ManifestWriter<'schema, 'metadata> { let manifest_length: i64 = manifest_bytes.len() as i64; - self.manifest.manifest_length += manifest_length; + self.manifest.manifest_length = manifest_length; let path = strip_prefix(&self.manifest.manifest_path).as_str().into(); let future = async move { @@ -878,6 +920,20 @@ fn update_partitions( summary.contains_null = true; } if let Some(value) = value { + match value { + Value::Float(number) if number.0.is_nan() => { + summary.contains_nan = Some(true); + continue; + } + Value::Double(number) if number.0.is_nan() => { + summary.contains_nan = Some(true); + continue; + } + Value::Float(_) | Value::Double(_) => { + summary.contains_nan.get_or_insert(false); + } + _ => {} + } if summary.lower_bound.is_none() { summary.lower_bound = Some(value.clone()); } else if let Some(lower_bound) = &mut summary.lower_bound { diff --git a/iceberg-rust/src/table/manifest_list.rs b/iceberg-rust/src/table/manifest_list.rs index 4ea3606d..73b3b920 100644 --- a/iceberg-rust/src/table/manifest_list.rs +++ b/iceberg-rust/src/table/manifest_list.rs @@ -18,7 +18,7 @@ use iceberg_rust_spec::{ }, schema::Schema, snapshot::Snapshot, - table_metadata::{FormatVersion, TableMetadata}, + table_metadata::{partition_fields as bound_partition_fields, FormatVersion, TableMetadata}, types::Type, util::strip_prefix, }; @@ -1332,12 +1332,13 @@ impl<'schema, 'metadata> ManifestListWriter<'schema, 'metadata> { if let Some(filter) = filter { let (manifest_writer, filtered_stats) = ManifestWriter::from_existing_with_filter( - manifest_bytes.as_ref(), + ManifestReader::new(manifest_bytes.as_ref())?, manifest, &filter, snapshot_id, &manifest_schema, self.table_metadata, + None, )?; (manifest_writer, Some(filtered_stats)) } else { @@ -1799,15 +1800,36 @@ impl<'schema, 'metadata> ManifestListWriter<'schema, 'metadata> { .bytes() .await?; + let manifest_reader = ManifestReader::new(bytes.as_ref())?; + let source_metadata = (manifest.partition_spec_id + != table_metadata.default_spec_id) + .then(|| manifest_reader.metadata()); + let old_spec_schema = source_metadata + .as_ref() + .map(|metadata| { + let (source_schema, source_spec, _) = metadata.as_ref(); + let fields = bound_partition_fields(source_spec, source_schema)?; + Ok::<_, Error>(ManifestEntry::schema( + &partition_value_schema(&fields)?, + &table_metadata.format_version, + )?) + }) + .transpose()?; + let schema = old_spec_schema.as_ref().unwrap_or(manifest_schema.as_ref()); + let rewrite_metadata = source_metadata.as_ref().map(|metadata| { + let (schema, spec, _) = metadata.as_ref(); + (schema, spec) + }); manifest.manifest_path = manifest_location; let (manifest_writer, filtered_stats) = ManifestWriter::from_existing_with_filter( - &bytes, + manifest_reader, manifest, &data_files_to_filter, snapshot_id, - &manifest_schema, + schema, table_metadata, + rewrite_metadata, )?; let new_manifest = manifest_writer.finish(object_store.clone()).await?; @@ -1827,6 +1849,7 @@ impl<'schema, 'metadata> ManifestListWriter<'schema, 'metadata> { if manifest.added_files_count.unwrap_or(0) > 0 || manifest.existing_files_count.unwrap_or(0) > 0 + || manifest.deleted_files_count.unwrap_or(0) > 0 { self.append_new_manifest(manifest)?; } diff --git a/iceberg-rust/src/table/transaction/mod.rs b/iceberg-rust/src/table/transaction/mod.rs index 6f7f4fd8..c9be795a 100644 --- a/iceberg-rust/src/table/transaction/mod.rs +++ b/iceberg-rust/src/table/transaction/mod.rs @@ -49,6 +49,46 @@ pub(crate) static REPLACE_SORT_ORDER_INDEX: usize = 9; pub(crate) static NUM_OPERATIONS: usize = 10; +fn reject_spec_change_with_file_write( + operations: &[Option], + current_spec_id: i32, +) -> Result<(), Error> { + let changes_spec = matches!( + operations[SET_DEFAULT_SPEC_INDEX].as_ref(), + Some(Operation::SetDefaultSpec(spec_id)) if *spec_id != current_spec_id + ); + if changes_spec + && operations + .iter() + .flatten() + .any(|operation| match operation { + Operation::Append { + data_files, + delete_files, + .. + } => !data_files.is_empty() || !delete_files.is_empty(), + Operation::AppendSequenceGroups { + sequence_groups, .. + } => sequence_groups + .iter() + .any(|group| !group.data_files.is_empty() || !group.delete_files.is_empty()), + Operation::Replace { files, .. } => !files.is_empty(), + Operation::Overwrite { + data_files, + files_to_overwrite, + .. + } => !data_files.is_empty() || !files_to_overwrite.is_empty(), + _ => false, + }) + { + return Err(Error::NotSupported( + "Changing the default partition spec and writing files in one transaction is not supported" + .to_owned(), + )); + } + Ok(()) +} + /// A transaction that can perform multiple operations on a table atomically /// /// TableTransaction allows grouping multiple table operations (like schema updates, @@ -502,9 +542,13 @@ impl<'table> TableTransaction<'table> { { return Err(Error::NotSupported( "writes to Iceberg tables without a table UUID cannot be fenced against DROP/recreate" - .to_owned(), + .to_owned(), )); } + reject_spec_change_with_file_write( + &self.operations, + self.table.metadata().default_spec_id, + )?; let catalog = self.table.catalog(); let identifier = self.table.identifier.clone(); @@ -554,8 +598,78 @@ impl<'table> TableTransaction<'table> { #[cfg(test)] mod tests { + use std::collections::HashMap; + + use super::{ + reject_spec_change_with_file_write, Operation, APPEND_INDEX, APPEND_SEQUENCE_GROUPS_INDEX, + NUM_OPERATIONS, OVERWRITE_INDEX, REPLACE_INDEX, SET_DEFAULT_SPEC_INDEX, + }; + use crate::error::Error; use rstest::rstest; + #[test] + fn spec_change_and_file_write_fail_closed() { + let mut operations: Vec> = (0..NUM_OPERATIONS).map(|_| None).collect(); + operations[SET_DEFAULT_SPEC_INDEX] = Some(Operation::SetDefaultSpec(1)); + assert!(reject_spec_change_with_file_write(&operations, 0).is_ok()); + + let writes = [ + ( + APPEND_INDEX, + Operation::Append { + branch: None, + data_files: Vec::new(), + delete_files: Vec::new(), + additional_summary: None, + }, + ), + ( + APPEND_SEQUENCE_GROUPS_INDEX, + Operation::AppendSequenceGroups { + branch: None, + sequence_groups: Vec::new(), + }, + ), + ( + REPLACE_INDEX, + Operation::Replace { + branch: None, + files: Vec::new(), + additional_summary: None, + }, + ), + ( + OVERWRITE_INDEX, + Operation::Overwrite { + branch: None, + data_files: Vec::new(), + files_to_overwrite: Default::default(), + additional_summary: None, + }, + ), + ]; + for (index, operation) in writes { + operations[index] = Some(operation); + assert!(reject_spec_change_with_file_write(&operations, 0).is_ok()); + assert!(reject_spec_change_with_file_write(&operations, 1).is_ok()); + operations[index] = None; + } + + operations[OVERWRITE_INDEX] = Some(Operation::Overwrite { + branch: None, + data_files: Vec::new(), + files_to_overwrite: HashMap::from([( + "manifest.avro".to_owned(), + vec!["data.parquet".to_owned()], + )]), + additional_summary: None, + }); + assert!(matches!( + reject_spec_change_with_file_write(&operations, 0), + Err(Error::NotSupported(_)) + )); + } + // ----------------------------------------------------------------------- // Placeholders for table-operation + snapshot-producer + partition-spec // evolution surfaces. diff --git a/iceberg-rust/src/table/transaction/operation.rs b/iceberg-rust/src/table/transaction/operation.rs index 7f57ee99..0a85cde6 100644 --- a/iceberg-rust/src/table/transaction/operation.rs +++ b/iceberg-rust/src/table/transaction/operation.rs @@ -1430,13 +1430,17 @@ mod tests { } fn partitioned_data_file(path: &str, partition_value: Value) -> DataFile { + partitioned_data_file_with_value(path, Some(partition_value)) + } + + fn partitioned_data_file_with_value(path: &str, partition_value: Option) -> 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), + partition_value, )])) .with_record_count(1) .with_file_size_in_bytes(100) @@ -1828,7 +1832,7 @@ mod tests { } #[tokio::test] - async fn v2_cow_rejects_changed_partition_transform() { + async fn v2_cow_preserves_changed_partition_transform() { let mut metadata = sample_metadata(&[], None, &[]); metadata.schemas.insert( 0, @@ -1893,7 +1897,7 @@ mod tests { .partition_specs .insert(1, spec(1, Transform::Bucket(16))); metadata.default_spec_id = 1; - let error = Operation::Overwrite { + let (_, updates) = Operation::Overwrite { branch: None, data_files: vec![partitioned_data_file("replacement.parquet", Value::Int(5))], files_to_overwrite: HashMap::from([( @@ -1902,17 +1906,245 @@ mod tests { )]), additional_summary: None, } - .execute(&metadata, store) + .execute(&metadata, store.clone()) .await - .unwrap_err(); + .unwrap(); + crate::catalog::commit::apply_table_updates(&mut metadata, updates).unwrap(); + let snapshot = metadata.current_snapshot(None).unwrap().unwrap(); assert!( - matches!(error, Error::NotSupported(message) if message == "COW rewrite across partition-spec evolution") + crate::table::manifest_list::snapshot_partition_bounds_with_nulls( + snapshot, + &metadata, + store.clone(), + ) + .await + .unwrap() + .is_none() + ); + 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 old = manifests + .iter() + .find(|manifest| manifest.partition_spec_id == 0) + .unwrap(); + assert_eq!(old.existing_files_count, Some(1)); + assert_eq!(old.deleted_files_count, Some(1)); + assert_eq!( + old.partitions.as_ref().unwrap()[0].lower_bound, + Some(Value::Int(42)) + ); + assert_eq!( + old.partitions.as_ref().unwrap()[0].upper_bound, + Some(Value::Int(43)) + ); + let new = manifests + .iter() + .find(|manifest| manifest.partition_spec_id == 1) + .unwrap(); + assert_eq!(new.added_files_count, Some(1)); + assert_eq!( + new.partitions.as_ref().unwrap()[0].lower_bound, + Some(Value::Int(5)) + ); + + let old_bytes = store + .get(&strip_prefix(&old.manifest_path).into()) + .await + .unwrap() + .bytes() + .await + .unwrap(); + assert_eq!( + usize::try_from(old.manifest_length).unwrap(), + old_bytes.len() + ); + let reader = ManifestReader::new(old_bytes.as_ref()).unwrap(); + let (_, source_spec, _) = reader.metadata().as_ref().clone(); + assert_eq!(*source_spec.spec_id(), 0); + let old_entries = reader.collect::, _>>().unwrap(); + assert_eq!( + *old_entries + .iter() + .find(|entry| entry.data_file().file_path() == "old.parquet") + .unwrap() + .snapshot_id(), + Some(*snapshot.snapshot_id()) + ); + let mut 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(); + entries.sort_by(|a, b| a.0.cmp(&b.0)); + assert_eq!( + entries, + vec![ + ( + "old.parquet".to_string(), + Status::Deleted, + Some(Some(Value::Int(42))) + ), + ( + "replacement.parquet".to_string(), + Status::Added, + Some(Some(Value::Int(5))) + ), + ( + "survivor.parquet".to_string(), + Status::Existing, + Some(Some(Value::Int(43))) + ), + ] + ); + + let (_, updates) = Operation::Append { + branch: None, + data_files: vec![partitioned_data_file("later.parquet", Value::Int(8))], + delete_files: Vec::new(), + additional_summary: None, + } + .execute(&metadata, store.clone()) + .await + .unwrap(); + crate::catalog::commit::apply_table_updates(&mut metadata, updates).unwrap(); + let mut metadata: TableMetadata = + serde_json::from_slice(&serde_json::to_vec(&metadata).unwrap()).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 live_files = crate::table::datafiles(store.clone(), &manifests, None, (None, None)) + .await + .unwrap() + .try_filter_map(|(_, entry)| async move { + Ok((entry.status() != &Status::Deleted) + .then(|| entry.data_file().file_path().clone())) + }) + .try_collect::>() + .await + .unwrap(); + live_files.sort(); + assert_eq!( + live_files, + vec!["later.parquet", "replacement.parquet", "survivor.parquet"] ); + + let mut files_to_overwrite: HashMap> = HashMap::new(); + for (manifest_path, entry) in + crate::table::datafiles(store.clone(), &manifests, None, (None, None)) + .await + .unwrap() + .try_collect::>() + .await + .unwrap() + { + if ["survivor.parquet", "later.parquet"] + .contains(&entry.data_file().file_path().as_str()) + { + files_to_overwrite + .entry(manifest_path) + .or_default() + .push(entry.data_file().file_path().clone()); + } + } + assert_eq!(files_to_overwrite.len(), 2); + let (_, updates) = Operation::Overwrite { + branch: None, + data_files: vec![partitioned_data_file("mixed.parquet", Value::Int(9))], + 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 old = manifests + .iter() + .find(|manifest| manifest.partition_spec_id == 0) + .unwrap(); + assert_eq!(old.existing_files_count, Some(0)); + assert_eq!(old.min_sequence_number, old.sequence_number); + for manifest in &manifests { + let bytes = store + .get(&strip_prefix(&manifest.manifest_path).into()) + .await + .unwrap() + .bytes() + .await + .unwrap(); + for entry in ManifestReader::new(bytes.as_ref()).unwrap() { + let entry = entry.unwrap(); + if ["survivor.parquet", "later.parquet"] + .contains(&entry.data_file().file_path().as_str()) + { + assert_eq!(*entry.snapshot_id(), Some(*snapshot.snapshot_id())); + } + } + } + let mut entries = 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(); + entries.sort_by(|a, b| a.0.cmp(&b.0)); + let live = entries + .iter() + .filter_map(|(path, status)| (status != &Status::Deleted).then_some(path.as_str())) + .collect::>(); + assert_eq!(live, vec!["mixed.parquet", "replacement.parquet"]); + assert!(entries.contains(&("survivor.parquet".to_string(), Status::Deleted))); + assert!(entries.contains(&("later.parquet".to_string(), Status::Deleted))); } #[tokio::test] - async fn v2_unpartitioned_spec_does_not_reuse_old_partitioned_manifest() { + async fn v3_cow_full_old_spec_preserves_null_partition_and_row_lineage() { let mut metadata = sample_metadata(&[], None, &[]); + metadata.format_version = FormatVersion::V3; + metadata.next_row_id = 10; metadata.schemas.insert( 0, SchemaBuilder::default() @@ -1920,7 +2152,7 @@ mod tests { .with_struct_field(StructField { id: 1, name: "partition_source".to_string(), - required: true, + required: false, field_type: Type::Primitive(PrimitiveType::Int), doc: None, initial_default: None, @@ -1929,25 +2161,27 @@ mod tests { .build() .unwrap(), ); - metadata.partition_specs.insert( - 0, + let spec = |id, transform| { PartitionSpec::builder() - .with_spec_id(0) + .with_spec_id(id) .with_partition_field(PartitionField::new( 1, - 1000, + 1000 + id, "partition_source", - Transform::Identity, + transform, )) .build() - .unwrap(), - ); + .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_with_value("null.parquet", None), partitioned_data_file("old.parquet", Value::Int(42)), - partitioned_data_file("survivor.parquet", Value::Int(43)), ], delete_files: Vec::new(), additional_summary: None, @@ -1956,41 +2190,39 @@ mod tests { .await .unwrap(); crate::catalog::commit::apply_table_updates(&mut metadata, updates).unwrap(); - let old_snapshot = metadata.current_snapshot(None).unwrap().unwrap(); - let old_bytes = store - .get(&strip_prefix(old_snapshot.manifest_list()).into()) + assert_eq!(metadata.next_row_id, 12); + 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(&old_bytes[..], &metadata) + let old_manifest = ManifestListReader::new(&bytes[..], &metadata) .unwrap() .next() .unwrap() .unwrap(); - metadata.partition_specs.insert( - 1, - PartitionSpec::builder() - .with_spec_id(1) - .with_fields(Vec::new()) - .build() - .unwrap(), - ); + metadata + .partition_specs + .insert(1, spec(1, Transform::Bucket(16))); metadata.default_spec_id = 1; - let (_, updates) = Operation::Append { + let (_, updates) = Operation::Overwrite { branch: None, - data_files: (0..8) - .map(|index| data_file(&format!("new-{index}.parquet"), 1)) - .collect(), - delete_files: Vec::new(), + data_files: vec![partitioned_data_file("replacement.parquet", Value::Int(5))], + files_to_overwrite: HashMap::from([( + old_manifest.manifest_path, + vec!["null.parquet".to_string(), "old.parquet".to_string()], + )]), additional_summary: None, } .execute(&metadata, store.clone()) .await .unwrap(); crate::catalog::commit::apply_table_updates(&mut metadata, updates).unwrap(); + assert_eq!(metadata.next_row_id, 13); let snapshot = metadata.current_snapshot(None).unwrap().unwrap(); let bytes = store .get(&strip_prefix(snapshot.manifest_list()).into()) @@ -2003,57 +2235,608 @@ mod tests { .unwrap() .collect::, _>>() .unwrap(); - assert!(manifests.len() > 2, "append must use the split path"); - assert!(manifests.iter().any(|entry| { - entry.manifest_path == old_manifest.manifest_path && entry.partition_spec_id == 0 - })); - assert!(manifests.iter().any(|entry| entry.partition_spec_id == 1)); - let mut file_paths = crate::table::datafiles(store.clone(), &manifests, None, (None, None)) + assert_eq!(manifests.len(), 2); + let old = manifests + .iter() + .find(|manifest| manifest.partition_spec_id == 0) + .unwrap(); + assert_eq!(old.existing_files_count, Some(0)); + assert_eq!(old.deleted_files_count, Some(2)); + assert_eq!(old.first_row_id, Some(10)); + assert_eq!(old.min_sequence_number, old.sequence_number); + let partition = &old.partitions.as_ref().unwrap()[0]; + assert!(partition.contains_null); + assert_eq!(partition.lower_bound, Some(Value::Int(42))); + assert_eq!(partition.upper_bound, Some(Value::Int(42))); + + let mut entries = crate::table::datafiles(store, &manifests, None, (None, None)) .await .unwrap() - .map_ok(|(_, entry)| entry.data_file().file_path().clone()) + .map_ok(|(_, entry)| { + ( + entry.data_file().file_path().clone(), + *entry.status(), + *entry.data_file().first_row_id(), + ) + }) .try_collect::>() .await .unwrap(); - file_paths.sort(); - let mut expected = (0..8) - .map(|index| format!("new-{index}.parquet")) - .chain(["old.parquet".to_string(), "survivor.parquet".to_string()]) - .collect::>(); - expected.sort(); - assert_eq!(file_paths, expected); - let replacement = data_file("replacement.parquet", 1); - let (selector, affected) = ManifestListWriter::from_existing_without_overwrites( - &bytes, - std::iter::once(&replacement), - &HashSet::from([old_manifest.manifest_path.clone()]), - manifest_list_schema_v2(), - &metadata, - ) - .unwrap(); + entries.sort_by(|a, b| a.0.cmp(&b.0)); assert_eq!( - selector - .selected_data_manifest() - .map(|entry| entry.partition_spec_id), - Some(1) + entries, + vec![ + ("null.parquet".to_string(), Status::Deleted, Some(10)), + ("old.parquet".to_string(), Status::Deleted, Some(11)), + ("replacement.parquet".to_string(), Status::Added, Some(12)), + ] ); - assert_eq!(affected, vec![old_manifest.clone()]); + } - let error = Operation::Overwrite { - branch: None, - data_files: vec![replacement], - 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") - ); + #[tokio::test] + async fn v3_cow_rewrites_v2_manifest_with_old_partition_spec() { + 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.format_version = FormatVersion::V3; + metadata.next_row_id = 0; + metadata + .partition_specs + .insert(1, spec(1, Transform::Bucket(16))); + metadata.default_spec_id = 1; + let (_, updates) = 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.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); + assert!(manifests + .iter() + .all(|manifest| manifest.first_row_id.is_some())); + let old = manifests + .iter() + .find(|manifest| manifest.partition_spec_id == 0) + .unwrap(); + let bytes = store + .get(&strip_prefix(&old.manifest_path).into()) + .await + .unwrap() + .bytes() + .await + .unwrap(); + assert_eq!( + ManifestReader::new(bytes.as_ref()).unwrap().metadata().2, + FormatVersion::V3 + ); + let mut live_files = crate::table::datafiles(store, &manifests, None, (None, None)) + .await + .unwrap() + .try_filter_map(|(_, entry)| async move { + Ok((entry.status() != &Status::Deleted) + .then(|| entry.data_file().file_path().clone())) + }) + .try_collect::>() + .await + .unwrap(); + live_files.sort(); + assert_eq!(live_files, vec!["replacement.parquet", "survivor.parquet"]); + } + + #[tokio::test] + async fn v2_cow_old_float_spec_excludes_nan_from_bounds() { + 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: false, + field_type: Type::Primitive(PrimitiveType::Float), + 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()); + let (_, updates) = Operation::Append { + branch: None, + data_files: vec![ + partitioned_data_file("nan.parquet", Value::Float(f32::NAN.into())), + partitioned_data_file("finite.parquet", Value::Float(2.5_f32.into())), + ], + 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(); + assert_eq!( + old_manifest.partitions.as_ref().unwrap()[0].contains_nan, + Some(true) + ); + + metadata.partition_specs.insert( + 1, + PartitionSpec::builder() + .with_spec_id(1) + .with_fields(Vec::new()) + .build() + .unwrap(), + ); + metadata.default_spec_id = 1; + let (_, updates) = Operation::Overwrite { + branch: None, + data_files: vec![data_file("replacement.parquet", 1)], + files_to_overwrite: HashMap::from([( + old_manifest.manifest_path, + vec!["finite.parquet".to_string()], + )]), + 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 old = manifests + .iter() + .find(|manifest| manifest.partition_spec_id == 0) + .unwrap(); + let summary = &old.partitions.as_ref().unwrap()[0]; + assert_eq!(summary.contains_nan, Some(true)); + assert_eq!(summary.lower_bound, Some(Value::Float(2.5_f32.into()))); + assert_eq!(summary.upper_bound, Some(Value::Float(2.5_f32.into()))); + let mut live_files = crate::table::datafiles(store, &manifests, None, (None, None)) + .await + .unwrap() + .try_filter_map(|(_, entry)| async move { + Ok((entry.status() != &Status::Deleted) + .then(|| entry.data_file().file_path().clone())) + }) + .try_collect::>() + .await + .unwrap(); + live_files.sort(); + assert_eq!(live_files, vec!["nan.parquet", "replacement.parquet"]); + } + + #[tokio::test] + #[ignore = "local file-level COW benchmark; run with --ignored --nocapture"] + async fn benchmark_cow_partition_spec_evolution() { + use std::hint::black_box; + use std::time::Instant; + + const FILES: i32 = 1024; + const REPEATS: usize = 3; + for changed_spec in [false, true] { + 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()); + let (_, updates) = Operation::Append { + branch: None, + data_files: (0..FILES) + .map(|index| { + partitioned_data_file(&format!("old-{index}.parquet"), Value::Int(index)) + }) + .collect(), + delete_files: Vec::new(), + additional_summary: None, + } + .execute(&metadata, store.clone()) + .await + .unwrap(); + crate::catalog::commit::apply_table_updates(&mut metadata, updates).unwrap(); + + if changed_spec { + metadata.partition_specs.insert( + 1, + PartitionSpec::builder() + .with_spec_id(1) + .with_partition_field(PartitionField::new( + 1, + 1001, + "partition_source", + Transform::Bucket(16), + )) + .build() + .unwrap(), + ); + metadata.default_spec_id = 1; + } + 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 selected = (0..FILES) + .step_by(2) + .map(|index| format!("old-{index}.parquet")) + .collect::>(); + let mut files_to_overwrite: HashMap> = HashMap::new(); + for (manifest_path, entry) in + crate::table::datafiles(store.clone(), &manifests, None, (None, None)) + .await + .unwrap() + .try_collect::>() + .await + .unwrap() + { + if selected.contains(entry.data_file().file_path()) { + 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::(), + 512 + ); + let replacements = (0..FILES) + .step_by(2) + .map(|index| { + let partition = if changed_spec { index % 16 } else { index }; + partitioned_data_file( + &format!("replacement-{index}.parquet"), + Value::Int(partition), + ) + }) + .collect::>(); + let mut elapsed = Vec::with_capacity(REPEATS); + for _ in 0..REPEATS { + let operation = Operation::Overwrite { + branch: None, + data_files: replacements.clone(), + files_to_overwrite: files_to_overwrite.clone(), + additional_summary: None, + }; + let start = Instant::now(); + black_box(operation.execute(&metadata, store.clone()).await.unwrap()); + elapsed.push(start.elapsed()); + } + elapsed.sort_unstable(); + eprintln!( + "COW 1024 source files / 512 replacements, changed_spec={changed_spec}: median {:.1} ms", + elapsed[REPEATS / 2].as_secs_f64() * 1000.0, + ); + } + } + + #[tokio::test] + async fn v2_unpartitioned_spec_does_not_reuse_old_partitioned_manifest() { + 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()); + 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 old_snapshot = metadata.current_snapshot(None).unwrap().unwrap(); + let old_bytes = store + .get(&strip_prefix(old_snapshot.manifest_list()).into()) + .await + .unwrap() + .bytes() + .await + .unwrap(); + let old_manifest = ManifestListReader::new(&old_bytes[..], &metadata) + .unwrap() + .next() + .unwrap() + .unwrap(); + + metadata.partition_specs.insert( + 1, + PartitionSpec::builder() + .with_spec_id(1) + .with_fields(Vec::new()) + .build() + .unwrap(), + ); + metadata.default_spec_id = 1; + let (_, updates) = Operation::Append { + branch: None, + data_files: (0..8) + .map(|index| data_file(&format!("new-{index}.parquet"), 1)) + .collect(), + 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(); + assert!(manifests.len() > 2, "append must use the split path"); + assert!(manifests.iter().any(|entry| { + entry.manifest_path == old_manifest.manifest_path && entry.partition_spec_id == 0 + })); + assert!(manifests.iter().any(|entry| entry.partition_spec_id == 1)); + let mut file_paths = crate::table::datafiles(store.clone(), &manifests, None, (None, None)) + .await + .unwrap() + .map_ok(|(_, entry)| entry.data_file().file_path().clone()) + .try_collect::>() + .await + .unwrap(); + file_paths.sort(); + let mut expected = (0..8) + .map(|index| format!("new-{index}.parquet")) + .chain(["old.parquet".to_string(), "survivor.parquet".to_string()]) + .collect::>(); + expected.sort(); + assert_eq!(file_paths, expected); + let replacement = data_file("replacement.parquet", 1); + let (selector, affected) = ManifestListWriter::from_existing_without_overwrites( + &bytes, + std::iter::once(&replacement), + &HashSet::from([old_manifest.manifest_path.clone()]), + manifest_list_schema_v2(), + &metadata, + ) + .unwrap(); + assert_eq!( + selector + .selected_data_manifest() + .map(|entry| entry.partition_spec_id), + Some(1) + ); + assert_eq!(affected, vec![old_manifest.clone()]); + + let (_, updates) = Operation::Overwrite { + branch: None, + data_files: vec![replacement], + files_to_overwrite: HashMap::from([( + old_manifest.manifest_path, + vec!["old.parquet".to_string()], + )]), + 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 old = manifests + .iter() + .find(|entry| entry.partition_spec_id == 0) + .unwrap(); + assert_eq!(old.existing_files_count, Some(1)); + assert_eq!(old.deleted_files_count, Some(1)); + assert!(manifests.iter().any(|entry| entry.partition_spec_id == 1)); + let mut live_files = crate::table::datafiles(store, &manifests, None, (None, None)) + .await + .unwrap() + .try_filter_map(|(_, entry)| async move { + Ok((entry.status() != &Status::Deleted) + .then(|| entry.data_file().file_path().clone())) + }) + .try_collect::>() + .await + .unwrap(); + live_files.sort(); + let mut expected = (0..8) + .map(|index| format!("new-{index}.parquet")) + .chain([ + "replacement.parquet".to_string(), + "survivor.parquet".to_string(), + ]) + .collect::>(); + expected.sort(); + assert_eq!(live_files, expected); } async fn clear_current_v1_manifest_list_counts(