diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index 58c7db7dd6..a476258218 100644 --- a/crates/iceberg/public-api.txt +++ b/crates/iceberg/public-api.txt @@ -3262,6 +3262,11 @@ pub fn iceberg::transaction::ReplaceSortOrderAction::asc(self, name: &str, null_ pub fn iceberg::transaction::ReplaceSortOrderAction::asc_with_transform(self, name: &str, transform: iceberg::spec::Transform, null_order: iceberg::spec::NullOrder) -> Self pub fn iceberg::transaction::ReplaceSortOrderAction::desc(self, name: &str, null_order: iceberg::spec::NullOrder) -> Self pub fn iceberg::transaction::ReplaceSortOrderAction::desc_with_transform(self, name: &str, transform: iceberg::spec::Transform, null_order: iceberg::spec::NullOrder) -> Self +pub struct iceberg::transaction::RewriteFilesAction +impl iceberg::transaction::RewriteFilesAction +pub fn iceberg::transaction::RewriteFilesAction::add_file(self, file: iceberg::spec::DataFile) -> Self +pub fn iceberg::transaction::RewriteFilesAction::data_sequence_number(self, sequence_number: i64) -> Self +pub fn iceberg::transaction::RewriteFilesAction::delete_file(self, file: iceberg::spec::DataFile) -> Self pub struct iceberg::transaction::Transaction impl iceberg::transaction::Transaction pub async fn iceberg::transaction::Transaction::commit(self, catalog: &dyn iceberg::Catalog) -> iceberg::Result @@ -3269,6 +3274,7 @@ pub fn iceberg::transaction::Transaction::expire_snapshots(&self) -> iceberg::tr pub fn iceberg::transaction::Transaction::fast_append(&self) -> iceberg::transaction::FastAppendAction pub fn iceberg::transaction::Transaction::new(table: &iceberg::table::Table) -> Self pub fn iceberg::transaction::Transaction::replace_sort_order(&self) -> iceberg::transaction::ReplaceSortOrderAction +pub fn iceberg::transaction::Transaction::rewrite_files(&self) -> iceberg::transaction::RewriteFilesAction pub fn iceberg::transaction::Transaction::update_location(&self) -> iceberg::transaction::UpdateLocationAction pub fn iceberg::transaction::Transaction::update_schema(&self) -> iceberg::transaction::UpdateSchemaAction pub fn iceberg::transaction::Transaction::update_statistics(&self) -> iceberg::transaction::UpdateStatisticsAction diff --git a/crates/iceberg/src/spec/manifest/writer.rs b/crates/iceberg/src/spec/manifest/writer.rs index 15608a63f8..106b462627 100644 --- a/crates/iceberg/src/spec/manifest/writer.rs +++ b/crates/iceberg/src/spec/manifest/writer.rs @@ -315,10 +315,6 @@ impl ManifestWriter { /// Add a delete manifest entry. This method will update following status of the entry: /// - Update the entry status to `Deleted` /// - Set the snapshot id to the current snapshot id - /// - /// # TODO - /// Remove this allow later - #[allow(dead_code)] pub(crate) fn add_delete_entry(&mut self, mut entry: ManifestEntry) -> Result<()> { self.check_data_file(&entry.data_file)?; entry.status = ManifestStatus::Deleted; @@ -350,10 +346,6 @@ impl ManifestWriter { /// Add an existing manifest entry. This method will update following status of the entry: /// - Update the entry status to `Existing` - /// - /// # TODO - /// Remove this allow later - #[allow(dead_code)] pub(crate) fn add_existing_entry(&mut self, mut entry: ManifestEntry) -> Result<()> { self.check_data_file(&entry.data_file)?; entry.status = ManifestStatus::Existing; diff --git a/crates/iceberg/src/spec/snapshot_summary.rs b/crates/iceberg/src/spec/snapshot_summary.rs index f8e69c4c86..ebecc825b2 100644 --- a/crates/iceberg/src/spec/snapshot_summary.rs +++ b/crates/iceberg/src/spec/snapshot_summary.rs @@ -339,6 +339,7 @@ pub(crate) fn update_snapshot_summaries( if summary.operation != Operation::Append && summary.operation != Operation::Overwrite && summary.operation != Operation::Delete + && summary.operation != Operation::Replace { return Err(invalid_data!("Operation is not supported.")); } diff --git a/crates/iceberg/src/transaction/merging.rs b/crates/iceberg/src/transaction/merging.rs new file mode 100644 index 0000000000..d66ae4df21 --- /dev/null +++ b/crates/iceberg/src/transaction/merging.rs @@ -0,0 +1,466 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Shared engine for snapshot-producing operations that both add and delete files. +//! +//! [`MergingSnapshotProducer`] is the Rust equivalent of Java's +//! `MergingSnapshotProducer`. It handles manifest filtering, new manifest +//! creation, summary computation, and delegates the final snapshot commit to +//! [`SnapshotProducer`]. + +use std::collections::{HashMap, HashSet}; +use std::sync::atomic::{AtomicU64, Ordering}; + +use uuid::Uuid; + +use crate::error::{Result, invalid_data}; +use crate::io::OutputFile; +use crate::spec::{ + DataFile, DataFileFormat, FormatVersion, ManifestContentType, ManifestEntry, ManifestFile, + ManifestStatus, ManifestWriter, ManifestWriterBuilder, Operation, PartitionSpec, SchemaRef, + SnapshotSummaryCollector, Summary, TableProperties, update_snapshot_summaries, +}; +use crate::table::Table; +use crate::transaction::ActionCommit; +use crate::transaction::snapshot::SnapshotProducer; + +/// Create a manifest writer that handles encryption when available. +fn new_manifest_writer( + table: &Table, + output_file: OutputFile, + snapshot_id: Option, + content: ManifestContentType, + schema: SchemaRef, + partition_spec: PartitionSpec, +) -> Result { + let builder = if let Some(em) = table.encryption_manager() { + ManifestWriterBuilder::new_from_encrypted( + em.encrypt(output_file), + snapshot_id, + schema, + partition_spec, + )? + } else { + ManifestWriterBuilder::new(output_file, snapshot_id, schema, partition_spec) + }; + + match table.metadata().format_version() { + FormatVersion::V1 => Ok(builder.build_v1()), + FormatVersion::V2 => match content { + ManifestContentType::Data => Ok(builder.build_v2_data()), + ManifestContentType::Deletes => Ok(builder.build_v2_deletes()), + }, + FormatVersion::V3 => match content { + ManifestContentType::Data => Ok(builder.build_v3_data()), + ManifestContentType::Deletes => Ok(builder.build_v3_deletes()), + }, + } +} + +/// Filters existing manifests by removing entries for deleted data files. +/// +/// This is equivalent to Java's `ManifestFilterManager`. When a rewrite or +/// overwrite operation deletes files, the filter manager rewrites affected +/// manifests so that deleted entries are dropped and surviving entries are +/// re-emitted with [`ManifestStatus::Existing`]. +pub(crate) struct ManifestFilterManager { + deleted_file_paths: HashSet, + fail_missing_delete_paths: bool, +} + +impl ManifestFilterManager { + pub(crate) fn new(fail_missing_delete_paths: bool) -> Self { + Self { + deleted_file_paths: HashSet::new(), + fail_missing_delete_paths, + } + } + + pub(crate) fn add_delete(&mut self, path: String) { + self.deleted_file_paths.insert(path); + } + + /// Filter `manifests` by removing entries whose file path is in the delete + /// set. Returns the surviving manifests plus a [`SnapshotSummaryCollector`] + /// that recorded metrics for every removed file. + pub(crate) async fn filter_manifests( + &self, + table: &Table, + manifests: Vec, + snapshot_id: i64, + commit_uuid: Uuid, + manifest_counter: &AtomicU64, + ) -> Result<(Vec, SnapshotSummaryCollector)> { + if self.deleted_file_paths.is_empty() { + return Ok((manifests, SnapshotSummaryCollector::default())); + } + + let mut result: Vec = Vec::with_capacity(manifests.len()); + let mut removed_collector = SnapshotSummaryCollector::default(); + let mut found_paths: HashSet = HashSet::new(); + + for manifest_file in &manifests { + // Only filter data manifests; pass delete manifests through unchanged. + if manifest_file.content != ManifestContentType::Data { + result.push(manifest_file.clone()); + continue; + } + + let manifest = table.manifest_reader().read(manifest_file).await?; + + // Check whether this manifest contains any files we want to delete. + let has_deletes = manifest + .entries() + .iter() + .any(|e| e.is_alive() && self.deleted_file_paths.contains(e.file_path())); + + if !has_deletes { + // Manifest is unaffected — pass through verbatim. + result.push(manifest_file.clone()); + continue; + } + + // Resolve the partition spec for this manifest. After partition + // evolution, old manifests may carry a non-default spec — we must + // use the manifest's own spec so the rewritten manifest stays valid. + let manifest_spec = table + .metadata() + .partition_spec_by_id(manifest_file.partition_spec_id) + .ok_or_else(|| { + invalid_data!( + "Manifest references unknown partition spec {}", + manifest_file.partition_spec_id, + ) + })? + .clone(); + let schema = table.metadata().current_schema().clone(); + + // Rewrite: keep surviving entries as EXISTING, and record the + // removed ones as DELETED so that the file they name can be + // cleaned up when this snapshot expires. + let mut surviving_entries: Vec = Vec::new(); + let mut removed_entries: Vec = Vec::new(); + for entry in manifest.entries() { + if entry.is_alive() && self.deleted_file_paths.contains(entry.file_path()) { + // Record removal metrics. + found_paths.insert(entry.file_path().to_string()); + removed_collector.remove_file( + entry.data_file(), + schema.clone(), + manifest_spec.clone(), + ); + removed_entries.push(entry.as_ref().clone()); + } else if entry.is_alive() { + // Surviving entry — re-emit as EXISTING with original ids preserved. + let seq = entry.sequence_number().ok_or_else(|| { + invalid_data!( + "Manifest entry for {} is missing sequence_number", + entry.file_path(), + ) + })?; + let file_seq = entry.file_sequence_number.ok_or_else(|| { + invalid_data!( + "Manifest entry for {} is missing file_sequence_number", + entry.file_path(), + ) + })?; + let existing = ManifestEntry::builder() + .status(ManifestStatus::Existing) + .snapshot_id( + entry + .snapshot_id() + .unwrap_or(manifest_file.added_snapshot_id), + ) + .sequence_number(seq) + .file_sequence_number(file_seq) + .data_file(entry.data_file().clone()) + .build(); + surviving_entries.push(existing); + } + // Already-deleted entries (status == Deleted) are dropped. + } + + if surviving_entries.is_empty() && removed_entries.is_empty() { + // Nothing left to say about this manifest — omit entirely. + continue; + } + + // Write the filtered manifest using the manifest's own partition spec. + let new_manifest_path = format!( + "{}/{}-m{}.{}", + table.metadata().metadata_location()?, + commit_uuid, + manifest_counter.fetch_add(1, Ordering::Relaxed), + DataFileFormat::Avro, + ); + let output_file = table.file_io().new_output(new_manifest_path)?; + // Use the current snapshot_id so that the manifest list writer + // can assign sequence numbers to this rewritten manifest. + let mut writer = new_manifest_writer( + table, + output_file, + Some(snapshot_id), + ManifestContentType::Data, + schema.clone(), + manifest_spec.as_ref().clone(), + )?; + for entry in surviving_entries { + writer.add_existing_entry(entry)?; + } + for entry in removed_entries { + // `add_delete_entry` stamps the removing snapshot on the entry + // and keeps the file's own sequence numbers, which is what an + // expiring reader needs to attribute the removal. + writer.add_delete_entry(entry)?; + } + let new_manifest = writer.write_manifest_file().await?; + result.push(new_manifest); + } + + // Validate that every delete target was found. + if self.fail_missing_delete_paths { + let missing: Vec<&str> = self + .deleted_file_paths + .iter() + .filter(|p| !found_paths.contains(p.as_str())) + .map(String::as_str) + .collect(); + if !missing.is_empty() { + return Err(invalid_data!( + "Failed to find the following files to delete in the current snapshot: {}", + missing.join(", "), + )); + } + } + + Ok((result, removed_collector)) + } +} + +/// Shared engine for operations that both add and remove data files. +/// +/// This struct is the Rust equivalent of Java's `MergingSnapshotProducer`. +/// It manages: +/// - Tracking files to add and delete +/// - Filtering existing manifests to remove deleted files +/// - Writing new manifests for added files +/// - Computing snapshot summaries that account for both additions and removals +/// - Delegating final snapshot creation to [`SnapshotProducer`] +/// +/// Concrete transaction actions like [`RewriteFilesAction`] own an instance +/// of this struct and configure it with the appropriate [`Operation`] type +/// and validation rules. +pub(crate) struct MergingSnapshotProducer { + operation: Operation, + added_data_files: Vec, + deleted_data_files: Vec, + filter_manager: ManifestFilterManager, + commit_uuid: Uuid, + data_sequence_number: Option, + /// Names the manifests this producer writes. It is never reset, so a + /// commit retry cannot overwrite a manifest an earlier attempt wrote + /// under the same commit uuid. + manifest_counter: AtomicU64, +} + +impl MergingSnapshotProducer { + pub(crate) fn new(operation: Operation) -> Self { + Self { + operation, + added_data_files: Vec::new(), + deleted_data_files: Vec::new(), + filter_manager: ManifestFilterManager::new(true), + commit_uuid: Uuid::now_v7(), + data_sequence_number: None, + manifest_counter: AtomicU64::new(0), + } + } + + pub(crate) fn add_data_file(&mut self, file: DataFile) { + self.added_data_files.push(file); + } + + pub(crate) fn set_data_sequence_number(&mut self, sequence_number: i64) { + self.data_sequence_number = Some(sequence_number); + } + + pub(crate) fn delete_data_file(&mut self, file: DataFile) { + self.filter_manager.add_delete(file.file_path.clone()); + self.deleted_data_files.push(file); + } + + pub(crate) fn has_added_data_files(&self) -> bool { + !self.added_data_files.is_empty() + } + + pub(crate) fn has_deleted_data_files(&self) -> bool { + !self.deleted_data_files.is_empty() + } + + /// Reject a data sequence number above the one the new snapshot will carry. + fn validate_data_sequence_number(&self, table: &Table) -> Result<()> { + let Some(sequence_number) = self.data_sequence_number else { + return Ok(()); + }; + let snapshot_sequence_number = table.metadata().next_sequence_number(); + if sequence_number > snapshot_sequence_number { + return Err(invalid_data!( + "Data sequence number {sequence_number} is greater than the snapshot's {snapshot_sequence_number}." + )); + } + Ok(()) + } + + /// Produce manifests, compute summary, and commit a new snapshot. + pub(crate) async fn commit_snapshot(&self, table: &Table) -> Result { + self.validate_data_sequence_number(table)?; + // Create the SnapshotProducer first so we can use its snapshot_id + // for new manifests. This ensures the manifest list writer can + // assign sequence numbers to manifests from this snapshot. + let snapshot_producer = SnapshotProducer::new( + table, + self.commit_uuid, + HashMap::new(), + self.added_data_files.clone(), + ); + snapshot_producer.validate_added_data_files()?; + snapshot_producer.validate_duplicate_files().await?; + let snapshot_id = snapshot_producer.snapshot_id; + + // 1. Load existing manifests from the current snapshot. + let existing_manifests = match table.metadata().current_snapshot() { + Some(snapshot) => { + let manifest_list = table.manifest_list_reader(snapshot).load().await?; + manifest_list + .entries() + .iter() + // Java's third arm is the new snapshot's id, which no existing manifest has. + .filter(|e| e.has_added_files() || e.has_existing_files()) + .cloned() + .collect() + } + None => Vec::new(), + }; + + // 2. Filter existing manifests — remove deleted file entries. + let (mut filtered_manifests, removed_collector) = self + .filter_manager + .filter_manifests( + table, + existing_manifests, + snapshot_id, + self.commit_uuid, + &self.manifest_counter, + ) + .await?; + + // 3. Write a new manifest for added files. + if !self.added_data_files.is_empty() { + let added_manifest = self + .write_added_manifest(table, snapshot_id, &self.manifest_counter) + .await?; + filtered_manifests.push(added_manifest); + } + + // 4. Compute summary (added + removed). + let summary = self.build_summary(table, removed_collector)?; + + // 5. Delegate to SnapshotProducer for manifest list + snapshot creation. + snapshot_producer + .commit_with_manifests(filtered_manifests, summary) + .await + } + + async fn write_added_manifest( + &self, + table: &Table, + snapshot_id: i64, + manifest_counter: &AtomicU64, + ) -> Result { + let new_manifest_path = format!( + "{}/{}-m{}.{}", + table.metadata().metadata_location()?, + self.commit_uuid, + manifest_counter.fetch_add(1, Ordering::Relaxed), + DataFileFormat::Avro, + ); + let output_file = table.file_io().new_output(new_manifest_path)?; + let schema = table.metadata().current_schema().clone(); + let partition_spec = table.metadata().default_partition_spec().as_ref().clone(); + + let mut writer = new_manifest_writer( + table, + output_file, + Some(snapshot_id), + ManifestContentType::Data, + schema, + partition_spec, + )?; + + let format_version = table.metadata().format_version(); + for data_file in &self.added_data_files { + if let Some(sequence_number) = self.data_sequence_number { + writer.add_file(data_file.clone(), sequence_number)?; + continue; + } + let entry_builder = ManifestEntry::builder() + .status(ManifestStatus::Added) + .data_file(data_file.clone()); + let entry = if format_version == FormatVersion::V1 { + // V1 requires snapshot_id on each entry. + entry_builder.snapshot_id(snapshot_id).build() + } else { + entry_builder.build() + }; + writer.add_entry(entry)?; + } + + writer.write_manifest_file().await + } + + fn build_summary( + &self, + table: &Table, + removed_collector: SnapshotSummaryCollector, + ) -> Result { + let table_metadata = table.metadata_ref(); + let schema = table_metadata.current_schema().clone(); + let partition_spec = table_metadata.default_partition_spec().clone(); + + // Add to the filter's collector rather than merging two: `merge` drops + // partition metrics unless both sides trust them, and neither does. + let mut collector = removed_collector; + collector.set_partition_summary_limit( + table_metadata + .properties() + .get(TableProperties::PROPERTY_WRITE_PARTITION_SUMMARY_LIMIT) + .and_then(|limit| limit.parse::().ok()) + .unwrap_or(TableProperties::PROPERTY_WRITE_PARTITION_SUMMARY_LIMIT_DEFAULT), + ); + for file in &self.added_data_files { + collector.add_file(file, schema.clone(), partition_spec.clone()); + } + + let summary = Summary { + operation: self.operation.clone(), + additional_properties: collector.build(), + }; + + let previous_snapshot = table_metadata.current_snapshot(); + update_snapshot_summaries(summary, previous_snapshot.map(|s| s.summary()), false) + } +} diff --git a/crates/iceberg/src/transaction/mod.rs b/crates/iceberg/src/transaction/mod.rs index 3622471277..da80cf4167 100644 --- a/crates/iceberg/src/transaction/mod.rs +++ b/crates/iceberg/src/transaction/mod.rs @@ -55,6 +55,8 @@ mod action; pub use action::*; mod append; mod expire_snapshots; +mod merging; +mod rewrite; mod snapshot; mod sort_order; mod update_location; @@ -75,6 +77,7 @@ use crate::table::Table; use crate::transaction::action::BoxedTransactionAction; pub use crate::transaction::append::FastAppendAction; pub use crate::transaction::expire_snapshots::ExpireSnapshotsAction; +pub use crate::transaction::rewrite::RewriteFilesAction; pub use crate::transaction::sort_order::ReplaceSortOrderAction; pub use crate::transaction::update_location::UpdateLocationAction; pub use crate::transaction::update_properties::UpdatePropertiesAction; @@ -151,6 +154,15 @@ impl Transaction { FastAppendAction::new() } + /// Creates a rewrite files action for compaction. + /// + /// This action replaces a set of data files with a new set while keeping + /// the logical table contents unchanged. The resulting snapshot uses + /// [`Operation::Replace`](crate::spec::Operation::Replace). + pub fn rewrite_files(&self) -> RewriteFilesAction { + RewriteFilesAction::new(self.table.metadata().current_snapshot_id()) + } + /// Creates replace sort order action. pub fn replace_sort_order(&self) -> ReplaceSortOrderAction { ReplaceSortOrderAction::new() @@ -254,7 +266,7 @@ mod tests { use crate::io::FileIO; use crate::memory::tests::new_memory_catalog; use crate::spec::{ - DataContentType, DataFileBuilder, DataFileFormat, Literal, Struct, TableMetadata, + DataContentType, DataFile, DataFileBuilder, DataFileFormat, Literal, Struct, TableMetadata, TableProperties, }; use crate::table::Table; @@ -262,6 +274,31 @@ mod tests { use crate::transaction::{ApplyTransactionAction, Transaction}; use crate::{Catalog, Error, ErrorKind, TableCreation, TableIdent}; + /// Create a test data file with the given path, record count, and file size. + pub(crate) fn make_data_file(table: &Table, path: &str, records: u64, size: u64) -> DataFile { + DataFileBuilder::default() + .content(DataContentType::Data) + .file_path(path.to_string()) + .file_format(DataFileFormat::Parquet) + .file_size_in_bytes(size) + .record_count(records) + .partition(Struct::from_iter([Some(Literal::long(1))])) + .partition_spec_id(table.metadata().default_partition_spec_id()) + .build() + .unwrap() + } + + /// Append data files to a table via fast_append and return the updated table. + pub(crate) async fn append_files( + catalog: &impl Catalog, + table: &Table, + files: Vec, + ) -> Table { + let tx = Transaction::new(table); + let tx = tx.fast_append().add_data_files(files).apply(tx).unwrap(); + tx.commit(catalog).await.unwrap() + } + pub fn make_v1_table() -> Table { let file = File::open(format!( "{}/testdata/table_metadata/{}", @@ -322,7 +359,38 @@ mod tests { .unwrap() } + pub(crate) async fn make_v1_minimal_table_in_catalog(catalog: &impl Catalog) -> Table { + make_minimal_table_in_catalog( + catalog, + "TableMetadataV1Valid.json", + crate::spec::FormatVersion::V1, + ) + .await + } + + pub(crate) async fn make_v2_minimal_table_in_catalog(catalog: &impl Catalog) -> Table { + make_minimal_table_in_catalog( + catalog, + "TableMetadataV2ValidMinimal.json", + crate::spec::FormatVersion::V2, + ) + .await + } + pub(crate) async fn make_v3_minimal_table_in_catalog(catalog: &impl Catalog) -> Table { + make_minimal_table_in_catalog( + catalog, + "TableMetadataV3ValidMinimal.json", + crate::spec::FormatVersion::V3, + ) + .await + } + + async fn make_minimal_table_in_catalog( + catalog: &impl Catalog, + metadata_file: &str, + format_version: crate::spec::FormatVersion, + ) -> Table { let table_ident = TableIdent::from_strs([format!("ns1-{}", uuid::Uuid::new_v4()), "test1".to_string()]) .unwrap(); @@ -335,7 +403,7 @@ mod tests { let file = File::open(format!( "{}/testdata/table_metadata/{}", env!("CARGO_MANIFEST_DIR"), - "TableMetadataV3ValidMinimal.json" + metadata_file )) .unwrap(); let reader = BufReader::new(file); @@ -346,7 +414,7 @@ mod tests { .partition_spec((**base_metadata.default_partition_spec()).clone()) .sort_order((**base_metadata.default_sort_order()).clone()) .name(table_ident.name().to_string()) - .format_version(crate::spec::FormatVersion::V3) + .format_version(format_version) .build(); catalog diff --git a/crates/iceberg/src/transaction/rewrite.rs b/crates/iceberg/src/transaction/rewrite.rs new file mode 100644 index 0000000000..b8b297939c --- /dev/null +++ b/crates/iceberg/src/transaction/rewrite.rs @@ -0,0 +1,1353 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Transaction action for rewriting data files (compaction). +//! +//! [`RewriteFilesAction`] replaces a set of data files with a new set while +//! keeping the logical table contents unchanged. This is used for compaction — +//! merging many small files into fewer large ones. +//! +//! The resulting snapshot uses [`Operation::Replace`] to indicate that files +//! were reorganised without changing the data. + +use std::sync::Arc; + +use async_trait::async_trait; + +use crate::error::{Result, invalid_data}; +use crate::spec::{DataFile, ManifestContentType, Operation}; +use crate::table::Table; +use crate::transaction::merging::MergingSnapshotProducer; +use crate::transaction::{ActionCommit, TransactionAction}; + +/// A transaction action that rewrites (replaces) data files. +/// +/// This is the Rust equivalent of Java's `BaseRewriteFiles`. It uses +/// `MergingSnapshotProducer` to handle manifest filtering and creation, +/// and commits a snapshot with [`Operation::Replace`]. +/// +/// # Example +/// +/// ```ignore +/// let tx = Transaction::new(&table); +/// let action = tx.rewrite_files() +/// .delete_file(old_file_1) +/// .delete_file(old_file_2) +/// .add_file(merged_file); +/// let tx = action.apply(tx)?; +/// let table = tx.commit(&catalog).await?; +/// ``` +/// +/// # Concurrent deletes +/// +/// The action records the table's current snapshot when it is created, and +/// refuses to commit if the table gained a delete manifest after it. Java +/// narrows that to the files being replaced in +/// `validateNoNewDeletesForDataFiles`; until that exists here the check is +/// table-wide, so a delete committed to an unrelated partition while the +/// rewrite ran also fails the commit with +/// [`ErrorKind::DataInvalid`](crate::ErrorKind::DataInvalid). So do a +/// rewrite planned against a table that had no snapshot yet, and one whose +/// starting snapshot has since been expired: neither can rule out a delete. +/// Replan the rewrite against the current table and run it again. +/// +/// Deletes that were already committed when the rewrite was planned are not +/// covered by this check. Either apply them while rewriting, or keep them +/// applicable to the new file with +/// [`data_sequence_number`](RewriteFilesAction::data_sequence_number). +pub struct RewriteFilesAction { + producer: MergingSnapshotProducer, + /// The snapshot the rewrite was planned against, if the table had one. + /// Deletes newer than it fail the commit. + starting_snapshot_id: Option, +} + +impl RewriteFilesAction { + pub(crate) fn new(starting_snapshot_id: Option) -> Self { + Self { + producer: MergingSnapshotProducer::new(Operation::Replace), + starting_snapshot_id, + } + } + + /// Register a data file to be removed from the table. + /// + /// The file must exist in the current snapshot; otherwise the commit + /// will fail with a validation error. + pub fn delete_file(mut self, file: DataFile) -> Self { + self.producer.delete_data_file(file); + self + } + + /// Register a data file to be added to the table. + /// + /// Typically this is the merged output of the files being deleted. + pub fn add_file(mut self, file: DataFile) -> Self { + self.producer.add_data_file(file); + self + } + + /// Set the data sequence number recorded for every added file. + /// + /// Without this the added files inherit the sequence number of the new + /// snapshot. A compaction that must not shadow concurrently written + /// deletes sets the sequence number of the files it replaces instead. + /// V1 manifest entries carry no sequence number, so this has no effect on + /// a V1 table. + pub fn data_sequence_number(mut self, sequence_number: i64) -> Self { + self.producer.set_data_sequence_number(sequence_number); + self + } + + fn validate(&self) -> Result<()> { + if !self.producer.has_deleted_data_files() { + return Err(invalid_data!( + "Rewrite files requires at least one file to delete" + )); + } + if !self.producer.has_added_data_files() { + return Err(invalid_data!( + "Rewrite files requires at least one file to add" + )); + } + Ok(()) + } + + /// Reject the rewrite when a delete manifest was added after the snapshot + /// it was planned against. + /// + /// Java re-checks those deletes file by file in + /// `validateNoNewDeletesForDataFiles`. Until that exists here the commit + /// fails closed, because a rewritten file carries the deletes of the files + /// it replaces only if nothing was deleted from them in the meantime. + async fn validate_no_new_deletes(&self, table: &Table) -> Result<()> { + // Nothing has been committed, so there is nothing to conflict with. + // The filter rejects the rewrite for its missing sources instead. + let Some(current_snapshot) = table.metadata().current_snapshot() else { + return Ok(()); + }; + // The rewrite was planned against a table without a snapshot, so every + // snapshot it has now, deletes included, landed after that. + let Some(starting_snapshot_id) = self.starting_snapshot_id else { + return Err(invalid_data!( + "Cannot rewrite files: the rewrite was planned against a table with no snapshot, and the table has been committed to since." + )); + }; + let Some(starting_snapshot) = table.metadata().snapshot_by_id(starting_snapshot_id) else { + // Without the starting snapshot there is no sequence number to + // compare against, so no delete can be ruled out. + return Err(invalid_data!( + "Cannot rewrite files: the starting snapshot {starting_snapshot_id} is no longer in the table, so deletes added since it cannot be ruled out." + )); + }; + let starting_sequence_number = starting_snapshot.sequence_number(); + + let manifest_list = table.manifest_list_reader(current_snapshot).load().await?; + let conflict = manifest_list.entries().iter().find(|entry| { + entry.content == ManifestContentType::Deletes + && entry.sequence_number > starting_sequence_number + }); + + match conflict { + Some(entry) => Err(invalid_data!( + "Cannot rewrite files: delete manifest {} was added after the starting snapshot {starting_snapshot_id}.", + entry.manifest_path, + )), + None => Ok(()), + } + } +} + +#[async_trait] +impl TransactionAction for RewriteFilesAction { + async fn commit(self: Arc, table: &Table) -> Result { + self.validate()?; + self.validate_no_new_deletes(table).await?; + self.producer.commit_snapshot(table).await + } +} + +#[cfg(test)] +mod tests { + use std::collections::HashMap; + use std::sync::Arc; + + use uuid::Uuid; + + use super::*; + use crate::memory::tests::new_memory_catalog; + use crate::spec::{ + DataContentType, DataFileBuilder, DataFileFormat, Datum, FormatVersion, Literal, + ManifestEntryRef, ManifestFile, ManifestListWriter, ManifestStatus, ManifestWriterBuilder, + NestedField, Operation, PrimitiveType, Schema, SnapshotRef, Struct, Transform, Type, + UnboundPartitionField, UnboundPartitionSpec, + }; + use crate::table::Table; + use crate::transaction::tests::{ + append_files, make_data_file, make_v1_minimal_table_in_catalog, + make_v2_minimal_table_in_catalog, make_v3_minimal_table_in_catalog, + }; + use crate::transaction::{ApplyTransactionAction, Transaction}; + use crate::{Catalog, ErrorKind, NamespaceIdent, TableCommit, TableCreation, TableUpdate}; + + /// Read back the manifest entry for `path` from the manifests of `snapshot`. + async fn find_entry( + table: &Table, + snapshot: &SnapshotRef, + path: &str, + ) -> Option { + let manifest_list = table.manifest_list_reader(snapshot).load().await.unwrap(); + for manifest_file in manifest_list.entries() { + let manifest = table.manifest_reader().read(manifest_file).await.unwrap(); + if let Some(entry) = manifest.entries().iter().find(|e| e.file_path() == path) { + return Some(entry.clone()); + } + } + None + } + + /// E2E: Compact 3 small files into 1 merged file. + /// Verify: operation=Replace, file counts, record counts. + #[tokio::test] + async fn test_rewrite_files_compaction() { + let catalog = new_memory_catalog().await; + let table = make_v3_minimal_table_in_catalog(&catalog).await; + rewrite_files_compaction(&catalog, table).await; + } + + #[tokio::test] + async fn test_rewrite_files_compaction_on_v1_table() { + let catalog = new_memory_catalog().await; + let table = make_v1_minimal_table_in_catalog(&catalog).await; + rewrite_files_compaction(&catalog, table).await; + } + + #[tokio::test] + async fn test_rewrite_files_compaction_on_v2_table() { + let catalog = new_memory_catalog().await; + let table = make_v2_minimal_table_in_catalog(&catalog).await; + rewrite_files_compaction(&catalog, table).await; + } + + async fn rewrite_files_compaction(catalog: &impl Catalog, table: Table) { + // Append 3 small files (10 records each). + let f1 = make_data_file(&table, "test/1.parquet", 10, 100); + let f2 = make_data_file(&table, "test/2.parquet", 10, 100); + let f3 = make_data_file(&table, "test/3.parquet", 10, 100); + let table = append_files(catalog, &table, vec![f1.clone(), f2.clone(), f3.clone()]).await; + + // Verify pre-compaction state. + let summary = &table + .metadata() + .current_snapshot() + .unwrap() + .summary() + .additional_properties; + assert_eq!(summary.get("total-data-files").unwrap(), "3"); + assert_eq!(summary.get("total-records").unwrap(), "30"); + + // Rewrite: delete 3 files, add 1 merged file (30 records). + let merged = make_data_file(&table, "test/merged.parquet", 30, 300); + let tx = Transaction::new(&table); + let action = tx + .rewrite_files() + .delete_file(f1) + .delete_file(f2) + .delete_file(f3) + .add_file(merged); + let tx = action.apply(tx).unwrap(); + let table = tx.commit(catalog).await.unwrap(); + + // Verify post-compaction snapshot. + let snapshot = table.metadata().current_snapshot().unwrap(); + assert_eq!(snapshot.summary().operation, Operation::Replace); + + let summary = &snapshot.summary().additional_properties; + assert_eq!(summary.get("total-data-files").unwrap(), "1"); + assert_eq!(summary.get("total-records").unwrap(), "30"); + assert_eq!(summary.get("added-data-files").unwrap(), "1"); + assert_eq!(summary.get("deleted-data-files").unwrap(), "3"); + + // Verify manifest list: merged file is the only live file. + let manifest_list = table.manifest_list_reader(snapshot).load().await.unwrap(); + let mut live_files: Vec = Vec::new(); + for manifest_entry in manifest_list.entries() { + let manifest = table.manifest_reader().read(manifest_entry).await.unwrap(); + for entry in manifest.entries() { + if entry.is_alive() { + live_files.push(entry.file_path().to_string()); + } + } + } + assert_eq!(live_files, vec!["test/merged.parquet"]); + } + + /// Rewrite with a non-existent delete target should fail. + #[tokio::test] + async fn test_rewrite_files_missing_delete_target() { + let catalog = new_memory_catalog().await; + let table = make_v3_minimal_table_in_catalog(&catalog).await; + + // Append 1 file. + let f1 = make_data_file(&table, "test/1.parquet", 10, 100); + let table = append_files(&catalog, &table, vec![f1]).await; + + // Try to delete a file that doesn't exist. + let ghost = make_data_file(&table, "test/ghost.parquet", 10, 100); + let merged = make_data_file(&table, "test/merged.parquet", 10, 100); + let tx = Transaction::new(&table); + let action = tx.rewrite_files().delete_file(ghost).add_file(merged); + let tx = action.apply(tx).unwrap(); + let result = tx.commit(&catalog).await; + + assert!(result.is_err()); + let err = result.unwrap_err(); + assert!( + err.message() + .contains("Failed to find the following files to delete"), + "unexpected error: {}", + err.message() + ); + } + + /// Rewrite with no deletes should fail validation. + #[tokio::test] + async fn test_rewrite_files_no_deletes() { + let catalog = new_memory_catalog().await; + let table = make_v3_minimal_table_in_catalog(&catalog).await; + + let merged = make_data_file(&table, "test/merged.parquet", 10, 100); + let tx = Transaction::new(&table); + let action = tx.rewrite_files().add_file(merged); + let tx = action.apply(tx).unwrap(); + let result = tx.commit(&catalog).await; + + assert!(result.is_err()); + assert!( + result + .unwrap_err() + .message() + .contains("at least one file to delete") + ); + } + + /// Partial rewrite: delete 2 of 3 files, keeping one. + /// Verifies that the rewritten manifest for the surviving file + /// has the correct snapshot_id for sequence number assignment. + #[tokio::test] + async fn test_rewrite_files_partial() { + let catalog = new_memory_catalog().await; + let table = make_v3_minimal_table_in_catalog(&catalog).await; + rewrite_files_partial(&catalog, table).await; + } + + #[tokio::test] + async fn test_rewrite_files_partial_on_v1_table() { + let catalog = new_memory_catalog().await; + let table = make_v1_minimal_table_in_catalog(&catalog).await; + rewrite_files_partial(&catalog, table).await; + } + + #[tokio::test] + async fn test_rewrite_files_partial_on_v2_table() { + let catalog = new_memory_catalog().await; + let table = make_v2_minimal_table_in_catalog(&catalog).await; + rewrite_files_partial(&catalog, table).await; + } + + async fn rewrite_files_partial(catalog: &impl Catalog, table: Table) { + let f1 = make_data_file(&table, "test/1.parquet", 10, 100); + let f2 = make_data_file(&table, "test/2.parquet", 10, 100); + let f3 = make_data_file(&table, "test/3.parquet", 10, 100); + let table = append_files(catalog, &table, vec![f1.clone(), f2.clone(), f3.clone()]).await; + let append_snapshot = table.metadata().current_snapshot().unwrap().clone(); + + // Rewrite only f1 and f2, keep f3. + let merged = make_data_file(&table, "test/merged.parquet", 20, 200); + let tx = Transaction::new(&table); + let action = tx + .rewrite_files() + .delete_file(f1) + .delete_file(f2) + .add_file(merged); + let tx = action.apply(tx).unwrap(); + let table = tx.commit(catalog).await.unwrap(); + + let snapshot = table.metadata().current_snapshot().unwrap(); + assert_eq!(snapshot.summary().operation, Operation::Replace); + + let summary = &snapshot.summary().additional_properties; + assert_eq!(summary.get("total-data-files").unwrap(), "2"); + assert_eq!(summary.get("total-records").unwrap(), "30"); + assert_eq!(summary.get("deleted-data-files").unwrap(), "2"); + assert_eq!(summary.get("added-data-files").unwrap(), "1"); + + // Verify live files: f3 (surviving) + merged (new). + let manifest_list = table.manifest_list_reader(snapshot).load().await.unwrap(); + let mut live_files: Vec = Vec::new(); + for manifest_entry in manifest_list.entries() { + let manifest = table.manifest_reader().read(manifest_entry).await.unwrap(); + for entry in manifest.entries() { + if entry.is_alive() { + live_files.push(entry.file_path().to_string()); + } + } + } + live_files.sort(); + assert_eq!(live_files, vec!["test/3.parquet", "test/merged.parquet"]); + + let survivor = find_entry(&table, snapshot, "test/3.parquet") + .await + .expect("surviving file should still be in a manifest"); + assert_eq!(survivor.status(), ManifestStatus::Existing); + assert_eq!(survivor.snapshot_id(), Some(append_snapshot.snapshot_id())); + assert_eq!( + survivor.sequence_number(), + Some(append_snapshot.sequence_number()) + ); + assert_eq!( + survivor.file_sequence_number, + Some(append_snapshot.sequence_number()) + ); + } + + /// An explicit data sequence number is recorded on the added file instead + /// of the one it would inherit from the rewrite snapshot. + #[tokio::test] + async fn test_rewrite_files_applies_data_sequence_number() { + let catalog = new_memory_catalog().await; + let table = make_v3_minimal_table_in_catalog(&catalog).await; + + let f1 = make_data_file(&table, "test/1.parquet", 10, 100); + let table = append_files(&catalog, &table, vec![f1.clone()]).await; + let append_sequence_number = table + .metadata() + .current_snapshot() + .unwrap() + .sequence_number(); + + let merged = make_data_file(&table, "test/merged.parquet", 10, 100); + let tx = Transaction::new(&table); + let action = tx + .rewrite_files() + .delete_file(f1) + .add_file(merged) + .data_sequence_number(append_sequence_number); + let tx = action.apply(tx).unwrap(); + let table = tx.commit(&catalog).await.unwrap(); + + let snapshot = table.metadata().current_snapshot().unwrap(); + assert_ne!(snapshot.sequence_number(), append_sequence_number); + + let merged_entry = find_entry(&table, snapshot, "test/merged.parquet") + .await + .expect("added file should be in a manifest"); + assert_eq!(merged_entry.sequence_number(), Some(append_sequence_number)); + } + + /// A data sequence number above the one the new snapshot will carry is + /// rejected. + #[tokio::test] + async fn test_rewrite_files_rejects_data_sequence_number_above_the_snapshot() { + let catalog = new_memory_catalog().await; + let table = make_v3_minimal_table_in_catalog(&catalog).await; + + let f1 = make_data_file(&table, "test/1.parquet", 10, 100); + let table = append_files(&catalog, &table, vec![f1.clone()]).await; + let snapshot_sequence_number = table.metadata().next_sequence_number(); + + let merged = make_data_file(&table, "test/merged.parquet", 10, 100); + let tx = Transaction::new(&table); + let action = tx + .rewrite_files() + .delete_file(f1) + .add_file(merged) + .data_sequence_number(snapshot_sequence_number + 1); + let tx = action.apply(tx).unwrap(); + let err = tx.commit(&catalog).await.unwrap_err(); + + assert_eq!(err.kind(), ErrorKind::DataInvalid); + assert!( + err.to_string().contains(&format!( + "Data sequence number {} is greater than the snapshot's {snapshot_sequence_number}.", + snapshot_sequence_number + 1 + )), + "{err}" + ); + } + + /// Rewrite on an empty table (no snapshot) should fail because + /// the delete target doesn't exist. + #[tokio::test] + async fn test_rewrite_files_empty_table() { + let catalog = new_memory_catalog().await; + let table = make_v3_minimal_table_in_catalog(&catalog).await; + + let f1 = make_data_file(&table, "test/1.parquet", 10, 100); + let merged = make_data_file(&table, "test/merged.parquet", 10, 100); + let tx = Transaction::new(&table); + let action = tx.rewrite_files().delete_file(f1).add_file(merged); + let tx = action.apply(tx).unwrap(); + let result = tx.commit(&catalog).await; + + assert!(result.is_err()); + assert!( + result + .unwrap_err() + .message() + .contains("Failed to find the following files to delete"), + ); + } + + /// Rewrite where all files in a manifest are deleted should omit + /// the manifest entirely (not leave an empty one). + #[tokio::test] + async fn test_rewrite_files_removes_empty_manifest() { + let catalog = new_memory_catalog().await; + let table = make_v3_minimal_table_in_catalog(&catalog).await; + + // Append 1 file → 1 manifest. + let f1 = make_data_file(&table, "test/1.parquet", 10, 100); + let table = append_files(&catalog, &table, vec![f1.clone()]).await; + + // Rewrite: delete f1, add merged. f1's manifest keeps no live entry, + // but it still has to record the removal for this snapshot. + let merged = make_data_file(&table, "test/merged.parquet", 10, 100); + let tx = Transaction::new(&table); + let action = tx.rewrite_files().delete_file(f1.clone()).add_file(merged); + let tx = action.apply(tx).unwrap(); + let table = tx.commit(&catalog).await.unwrap(); + + let snapshot = table.metadata().current_snapshot().unwrap(); + let emptied = find_entry(&table, snapshot, "test/1.parquet") + .await + .expect("the removal has to be recorded somewhere"); + assert_eq!(emptied.status(), ManifestStatus::Deleted); + assert_eq!(emptied.snapshot_id(), Some(snapshot.snapshot_id())); + let manifest_list = table.manifest_list_reader(snapshot).load().await.unwrap(); + assert_eq!(manifest_list.entries().len(), 2); + assert!( + manifest_list + .entries() + .iter() + .any(|m| !m.has_added_files() && !m.has_existing_files()), + "the emptied manifest holds only the deleted entry" + ); + + // A later rewrite drops it: it has nothing live, and no removal of its + // own left to record. An append would carry it along, as in Java, + // where only a merging commit rebuilds the manifest list. + let merged = find_entry(&table, snapshot, "test/merged.parquet") + .await + .unwrap() + .data_file() + .clone(); + let merged_again = make_data_file(&table, "test/merged-2.parquet", 10, 100); + let tx = Transaction::new(&table); + let action = tx + .rewrite_files() + .delete_file(merged) + .add_file(merged_again); + let table = action.apply(tx).unwrap().commit(&catalog).await.unwrap(); + + let snapshot = table.metadata().current_snapshot().unwrap(); + assert!( + find_entry(&table, snapshot, "test/1.parquet") + .await + .is_none(), + "an all-deleted manifest should not survive a later rewrite" + ); + } + + /// A manifest left with only deleted entries records nothing live, whether + /// the current snapshot or an older one left it that way. + #[tokio::test] + async fn test_rewrite_files_drops_all_deleted_manifests() { + let catalog = new_memory_catalog().await; + let table = make_v3_minimal_table_in_catalog(&catalog).await; + + let f1 = make_data_file(&table, "test/1.parquet", 10, 100); + let f2 = make_data_file(&table, "test/2.parquet", 10, 100); + let table = append_files(&catalog, &table, vec![f1.clone()]).await; + let table = append_files(&catalog, &table, vec![f2]).await; + let current_snapshot_id = table.metadata().current_snapshot().unwrap().snapshot_id(); + let earlier_snapshot_id = table + .metadata() + .current_snapshot() + .unwrap() + .parent_snapshot_id() + .unwrap(); + let older = add_delete_only_manifest(&table, earlier_snapshot_id).await; + let current = add_delete_only_manifest(&table, current_snapshot_id).await; + + let merged = make_data_file(&table, "test/merged.parquet", 10, 100); + let tx = Transaction::new(&table); + let action = tx.rewrite_files().delete_file(f1).add_file(merged); + let tx = action.apply(tx).unwrap(); + let table = tx.commit(&catalog).await.unwrap(); + + let snapshot = table.metadata().current_snapshot().unwrap(); + let manifest_list = table.manifest_list_reader(snapshot).load().await.unwrap(); + for path in [older, current] { + assert!( + !manifest_list + .entries() + .iter() + .any(|m| m.manifest_path == path), + "{path} holds no live entry and should have been dropped" + ); + } + } + + /// Add a delete-only manifest, attributed to `added_snapshot_id`, to the + /// manifest list of the table's current snapshot, and return its path. + async fn add_delete_only_manifest(table: &Table, added_snapshot_id: i64) -> String { + let added_sequence_number = table + .metadata() + .snapshot_by_id(added_snapshot_id) + .unwrap() + .sequence_number(); + let output = table + .file_io() + .new_output(format!( + "{}/delete-only-{}.avro", + table.metadata().metadata_location().unwrap(), + Uuid::new_v4() + )) + .unwrap(); + let mut writer = ManifestWriterBuilder::new( + output, + Some(added_snapshot_id), + table.metadata().current_schema().clone(), + table.metadata().default_partition_spec().as_ref().clone(), + ) + .build_v3_data(); + writer + .add_delete_file( + make_data_file( + table, + &format!("test/removed-{added_snapshot_id}.parquet"), + 10, + 100, + ), + added_sequence_number, + Some(added_sequence_number), + ) + .unwrap(); + let mut delete_manifest = writer.write_manifest_file().await.unwrap(); + // The manifest list writer only assigns sequence numbers to manifests of + // the snapshot being written; this one belongs to an earlier snapshot. + delete_manifest.sequence_number = added_sequence_number; + delete_manifest.min_sequence_number = added_sequence_number; + let delete_manifest_path = delete_manifest.manifest_path.clone(); + assert!(!delete_manifest.has_added_files()); + assert!(!delete_manifest.has_existing_files()); + + append_to_manifest_list(table, delete_manifest).await; + delete_manifest_path + } + + /// Add a delete manifest, attributed to the table's current snapshot, to + /// that snapshot's manifest list. + async fn add_delete_manifest(table: &Table) -> String { + let snapshot = table.metadata().current_snapshot().unwrap(); + let output = table + .file_io() + .new_output(format!( + "{}/deletes-{}.avro", + table.metadata().metadata_location().unwrap(), + Uuid::new_v4() + )) + .unwrap(); + let mut writer = ManifestWriterBuilder::new( + output, + Some(snapshot.snapshot_id()), + table.metadata().current_schema().clone(), + table.metadata().default_partition_spec().as_ref().clone(), + ) + .build_v3_deletes(); + let mut delete_file = make_data_file(table, "test/1-deletes.parquet", 1, 100); + delete_file.content = DataContentType::PositionDeletes; + writer + .add_file(delete_file, snapshot.sequence_number()) + .unwrap(); + let delete_manifest = writer.write_manifest_file().await.unwrap(); + let delete_manifest_path = delete_manifest.manifest_path.clone(); + + append_to_manifest_list(table, delete_manifest).await; + delete_manifest_path + } + + /// Rewrite the manifest list of the table's current snapshot so it also + /// contains `manifest`. + async fn append_to_manifest_list(table: &Table, manifest: ManifestFile) { + let snapshot = table.metadata().current_snapshot().unwrap(); + let mut entries: Vec = table + .manifest_list_reader(snapshot) + .load() + .await + .unwrap() + .consume_entries() + .into_iter() + .collect(); + entries.push(manifest); + + let mut manifest_list_writer = ManifestListWriter::v3( + table + .file_io() + .new_output(snapshot.manifest_list()) + .unwrap() + .writer() + .await + .unwrap(), + snapshot.snapshot_id(), + snapshot.parent_snapshot_id(), + snapshot.sequence_number(), + Some(table.metadata().next_row_id()), + ); + manifest_list_writer + .add_manifests(entries.into_iter()) + .unwrap(); + manifest_list_writer.close().await.unwrap(); + } + + /// A delete manifest added after the snapshot the rewrite was planned + /// against must stop the commit. + #[tokio::test] + async fn test_rewrite_files_rejects_deletes_added_after_starting_snapshot() { + let catalog = new_memory_catalog().await; + let table = make_v3_minimal_table_in_catalog(&catalog).await; + + let f1 = make_data_file(&table, "test/1.parquet", 10, 100); + let f2 = make_data_file(&table, "test/2.parquet", 10, 100); + let table = append_files(&catalog, &table, vec![f1.clone()]).await; + + // Plan the rewrite against this snapshot, then let another commit land. + let tx = Transaction::new(&table); + let merged = make_data_file(&table, "test/merged.parquet", 10, 100); + let action = Arc::new(tx.rewrite_files().delete_file(f1).add_file(merged)); + + let table = append_files(&catalog, &table, vec![f2]).await; + let delete_manifest_path = add_delete_manifest(&table).await; + + let Err(err) = action.commit(&table).await else { + panic!("a delete manifest newer than the starting snapshot must be rejected"); + }; + assert_eq!(err.kind(), ErrorKind::DataInvalid); + assert!( + err.to_string().contains(&format!( + "delete manifest {delete_manifest_path} was added after" + )), + "{err}" + ); + } + + /// A rewrite planned against a table with no snapshot knows nothing about + /// what the table holds once something is committed to it. + #[tokio::test] + async fn test_rewrite_files_rejects_commit_after_empty_starting_table() { + let catalog = new_memory_catalog().await; + let table = make_v3_minimal_table_in_catalog(&catalog).await; + + // Planned before the table had any snapshot. + let f1 = make_data_file(&table, "test/1.parquet", 10, 100); + let merged = make_data_file(&table, "test/merged.parquet", 10, 100); + let tx = Transaction::new(&table); + let action = Arc::new(tx.rewrite_files().delete_file(f1.clone()).add_file(merged)); + + // The file it wants to replace only exists because of a later commit. + let table = append_files(&catalog, &table, vec![f1]).await; + + let Err(err) = action.commit(&table).await else { + panic!("a rewrite planned against an empty table must be rejected"); + }; + assert_eq!(err.kind(), ErrorKind::DataInvalid); + assert!( + err.to_string() + .contains("planned against a table with no snapshot"), + "{err}" + ); + } + + /// A rewrite whose starting snapshot has been expired cannot rule out + /// deletes added since, and says so rather than comparing against zero. + #[tokio::test] + async fn test_rewrite_files_rejects_expired_starting_snapshot() { + let catalog = new_memory_catalog().await; + let table = make_v3_minimal_table_in_catalog(&catalog).await; + + let f1 = make_data_file(&table, "test/1.parquet", 10, 100); + let f2 = make_data_file(&table, "test/2.parquet", 10, 100); + let table = append_files(&catalog, &table, vec![f1.clone()]).await; + let starting_snapshot_id = table.metadata().current_snapshot().unwrap().snapshot_id(); + + // Plan the rewrite against this snapshot, then expire it behind us. + let tx = Transaction::new(&table); + let merged = make_data_file(&table, "test/merged.parquet", 10, 100); + let action = Arc::new(tx.rewrite_files().delete_file(f1).add_file(merged)); + + let table = append_files(&catalog, &table, vec![f2]).await; + let tx = Transaction::new(&table); + let tx = tx + .expire_snapshots() + .expire_snapshot_ids([starting_snapshot_id]) + .apply(tx) + .unwrap(); + let table = tx.commit(&catalog).await.unwrap(); + assert!( + table + .metadata() + .snapshot_by_id(starting_snapshot_id) + .is_none(), + "the starting snapshot should have been expired" + ); + + let Err(err) = action.commit(&table).await else { + panic!("an expired starting snapshot must be rejected"); + }; + assert_eq!(err.kind(), ErrorKind::DataInvalid); + assert!( + err.to_string().contains(&format!( + "the starting snapshot {starting_snapshot_id} is no longer in the table" + )), + "{err}" + ); + } + + /// A retried commit must not reuse the manifest paths of the attempt + /// before it: the commit uuid is fixed, so only the counter separates them. + #[tokio::test] + async fn test_rewrite_files_writes_distinct_manifests_per_attempt() { + let catalog = new_memory_catalog().await; + let table = make_v3_minimal_table_in_catalog(&catalog).await; + + let f1 = make_data_file(&table, "test/1.parquet", 10, 100); + let f2 = make_data_file(&table, "test/2.parquet", 10, 100); + let table = append_files(&catalog, &table, vec![f1.clone(), f2.clone()]).await; + + let merged = make_data_file(&table, "test/merged.parquet", 10, 100); + let tx = Transaction::new(&table); + let action = Arc::new(tx.rewrite_files().delete_file(f1).add_file(merged)); + + let first = manifest_paths(&table, Arc::clone(&action)).await; + let second = manifest_paths(&table, action).await; + + assert_eq!(first.len(), 2, "{first:?}"); + assert_eq!(first.len(), second.len()); + for path in &first { + assert!(!second.contains(path), "{path} was written twice"); + } + } + + /// Commit `action` against `table` and return the paths of the manifests + /// the resulting snapshot points at. + async fn manifest_paths(table: &Table, action: Arc) -> Vec { + let mut commit = action.commit(table).await.unwrap(); + let snapshot = commit + .take_updates() + .into_iter() + .find_map(|update| match update { + TableUpdate::AddSnapshot { snapshot } => Some(snapshot), + _ => None, + }) + .expect("commit should add a snapshot"); + + let manifest_list = table + .manifest_list_reader(&Arc::new(snapshot)) + .load() + .await + .unwrap(); + manifest_list + .entries() + .iter() + .map(|entry| entry.manifest_path.clone()) + .collect() + } + + /// An added file whose partition value does not fit the default spec is + /// rejected before anything is written. + #[tokio::test] + async fn test_rewrite_files_rejects_incompatible_partition_value() { + let catalog = new_memory_catalog().await; + let table = make_v3_minimal_table_in_catalog(&catalog).await; + + let f1 = make_data_file(&table, "test/1.parquet", 10, 100); + let table = append_files(&catalog, &table, vec![f1.clone()]).await; + + let mut merged = make_data_file(&table, "test/merged.parquet", 10, 100); + merged.partition = Struct::from_iter([Some(Literal::string("not-a-long"))]); + let tx = Transaction::new(&table); + let action = tx.rewrite_files().delete_file(f1).add_file(merged); + let tx = action.apply(tx).unwrap(); + let err = tx.commit(&catalog).await.unwrap_err(); + + assert_eq!(err.kind(), ErrorKind::DataInvalid); + assert!( + err.to_string() + .contains("Partition value is not compatible partition type"), + "{err}" + ); + } + + /// A delete file handed to `add_file` is rejected: a rewrite replaces data + /// files, and the producer validation names the offending file. + #[tokio::test] + async fn test_rewrite_files_rejects_non_data_content_type() { + let catalog = new_memory_catalog().await; + let table = make_v3_minimal_table_in_catalog(&catalog).await; + + let f1 = make_data_file(&table, "test/1.parquet", 10, 100); + let table = append_files(&catalog, &table, vec![f1.clone()]).await; + + let mut deletes = make_data_file(&table, "test/1-deletes.parquet", 1, 100); + deletes.content = DataContentType::PositionDeletes; + let tx = Transaction::new(&table); + let action = tx.rewrite_files().delete_file(f1).add_file(deletes); + let tx = action.apply(tx).unwrap(); + let err = tx.commit(&catalog).await.unwrap_err(); + + assert_eq!(err.kind(), ErrorKind::DataInvalid); + assert!( + err.to_string().contains( + "Only data content type is allowed in added data files, but test/1-deletes.parquet is PositionDeletes" + ), + "{err}" + ); + } + + /// An added file that is already live in the current snapshot is rejected. + #[tokio::test] + async fn test_rewrite_files_rejects_already_referenced_file() { + let catalog = new_memory_catalog().await; + let table = make_v3_minimal_table_in_catalog(&catalog).await; + + let f1 = make_data_file(&table, "test/1.parquet", 10, 100); + let f2 = make_data_file(&table, "test/2.parquet", 10, 100); + let table = append_files(&catalog, &table, vec![f1.clone(), f2.clone()]).await; + + let tx = Transaction::new(&table); + let action = tx.rewrite_files().delete_file(f1).add_file(f2); + let tx = action.apply(tx).unwrap(); + let err = tx.commit(&catalog).await.unwrap_err(); + + assert_eq!(err.kind(), ErrorKind::DataInvalid); + assert!( + err.to_string().contains( + "Cannot add files that are already referenced by table, files: test/2.parquet" + ), + "{err}" + ); + } + + /// Rewrite with no adds should fail validation. + #[tokio::test] + async fn test_rewrite_files_no_adds() { + let catalog = new_memory_catalog().await; + let table = make_v3_minimal_table_in_catalog(&catalog).await; + + let f1 = make_data_file(&table, "test/1.parquet", 10, 100); + let tx = Transaction::new(&table); + let action = tx.rewrite_files().delete_file(f1); + let tx = action.apply(tx).unwrap(); + let result = tx.commit(&catalog).await; + + assert!(result.is_err()); + assert!( + result + .unwrap_err() + .message() + .contains("at least one file to add") + ); + } + + // Partition-spec evolution: `ts` is identity-partitioned under spec 0, then + // `day(ts)` becomes the default spec. Old files keep spec 0 and a + // timestamptz partition value; new files carry spec 1 and a date. + + /// 2026-08-24T09:30:00Z and 2026-08-24T17:45:00Z, in microseconds. + const TS_MORNING: i64 = 1_787_563_800_000_000; + const TS_EVENING: i64 = 1_787_593_500_000_000; + /// 2026-08-24, in days since the epoch. + const DAY: i32 = 20_689; + + /// A table partitioned by `identity(ts)`, still on its first spec. + async fn make_identity_ts_table( + catalog: &impl Catalog, + format_version: FormatVersion, + ) -> Table { + let namespace = NamespaceIdent::new(format!("ns1-{}", Uuid::new_v4())); + catalog + .create_namespace(&namespace, HashMap::new()) + .await + .unwrap(); + + let schema = Schema::builder() + .with_fields(vec![ + NestedField::required(1, "id", Type::Primitive(PrimitiveType::Long)).into(), + NestedField::required(2, "ts", Type::Primitive(PrimitiveType::Timestamptz)).into(), + ]) + .build() + .unwrap(); + let spec = UnboundPartitionSpec::builder() + .add_partition_field( + UnboundPartitionField::builder() + .source_ids(vec![2]) + .name("ts") + .transform(Transform::Identity) + .build() + .unwrap(), + ) + .unwrap() + .build(); + let creation = TableCreation::builder() + .name("events".to_string()) + .schema(schema) + .partition_spec(spec) + .format_version(format_version) + .build(); + catalog.create_table(&namespace, creation).await.unwrap() + } + + /// Replace `identity(ts)` with `day(ts)` as the default spec. + async fn evolve_to_day_ts(catalog: &impl Catalog, table: &Table) -> Table { + let spec = UnboundPartitionSpec::builder() + .add_partition_field( + UnboundPartitionField::builder() + .source_ids(vec![2]) + .name("ts_day") + .transform(Transform::Day) + .build() + .unwrap(), + ) + .unwrap() + .build(); + let commit = TableCommit::builder() + .ident(table.identifier().clone()) + .requirements(vec![]) + .updates(vec![ + TableUpdate::AddSpec { spec }, + TableUpdate::SetDefaultSpec { spec_id: -1 }, + ]) + .build(); + let table = catalog.update_table(commit).await.unwrap(); + assert_eq!(table.metadata().default_partition_spec_id(), 1); + table + } + + fn make_file_in_spec( + path: &str, + records: u64, + spec_id: i32, + partition_value: Literal, + ) -> DataFile { + DataFileBuilder::default() + .content(DataContentType::Data) + .file_path(path.to_string()) + .file_format(DataFileFormat::Parquet) + .file_size_in_bytes(records * 10) + .record_count(records) + .partition(Struct::from_iter([Some(partition_value)])) + .partition_spec_id(spec_id) + .build() + .unwrap() + } + + /// Live files of the current snapshot, keyed by path, with the spec id of + /// the manifest that lists them. + async fn live_files_by_manifest_spec(table: &Table) -> Vec<(String, i32)> { + let snapshot = table.metadata().current_snapshot().unwrap(); + let manifest_list = table.manifest_list_reader(snapshot).load().await.unwrap(); + let mut live = Vec::new(); + for manifest_file in manifest_list.entries() { + let manifest = table.manifest_reader().read(manifest_file).await.unwrap(); + for entry in manifest.entries().iter().filter(|e| e.is_alive()) { + assert_eq!( + entry.data_file().partition_spec_id, + manifest_file.partition_spec_id, + "{} is listed by a manifest of another spec", + entry.file_path() + ); + live.push(( + entry.file_path().to_string(), + manifest_file.partition_spec_id, + )); + } + } + live.sort(); + live + } + + /// Two old-spec files and one new-spec file; returns the table and the + /// snapshot that appended the old-spec files. + async fn spec_evolved_table_with_files( + catalog: &impl Catalog, + format_version: FormatVersion, + ) -> (Table, SnapshotRef, [DataFile; 3]) { + let table = make_identity_ts_table(catalog, format_version).await; + let old_1 = make_file_in_spec( + "test/old-1.parquet", + 10, + 0, + Literal::timestamptz(TS_MORNING), + ); + let old_2 = make_file_in_spec( + "test/old-2.parquet", + 10, + 0, + Literal::timestamptz(TS_EVENING), + ); + let table = append_files(catalog, &table, vec![old_1.clone(), old_2.clone()]).await; + let old_snapshot = table.metadata().current_snapshot().unwrap().clone(); + + let table = evolve_to_day_ts(catalog, &table).await; + let new_1 = make_file_in_spec("test/new-1.parquet", 10, 1, Literal::date(DAY)); + let table = append_files(catalog, &table, vec![new_1.clone()]).await; + (table, old_snapshot, [old_1, old_2, new_1]) + } + + /// Compacting one old-spec file keeps the other in a manifest written under + /// spec 0: its entry, partition value and the manifest's partition summary + /// are still those of `identity(ts)`, not re-encoded under `day(ts)`. + #[tokio::test] + async fn test_rewrite_files_keeps_survivors_on_their_partition_spec() { + for format_version in [FormatVersion::V2, FormatVersion::V3] { + let catalog = new_memory_catalog().await; + let (table, old_snapshot, [old_1, _old_2, _new_1]) = + spec_evolved_table_with_files(&catalog, format_version).await; + + let merged = make_file_in_spec("test/merged.parquet", 10, 1, Literal::date(DAY)); + let tx = Transaction::new(&table); + let action = tx.rewrite_files().delete_file(old_1).add_file(merged); + let table = action.apply(tx).unwrap().commit(&catalog).await.unwrap(); + + assert_eq!( + live_files_by_manifest_spec(&table).await, + vec![ + ("test/merged.parquet".to_string(), 1), + ("test/new-1.parquet".to_string(), 1), + ("test/old-2.parquet".to_string(), 0), + ], + "{format_version}" + ); + + let snapshot = table.metadata().current_snapshot().unwrap(); + let survivor = find_entry(&table, snapshot, "test/old-2.parquet") + .await + .unwrap(); + assert_eq!(survivor.status(), ManifestStatus::Existing); + assert_eq!(survivor.snapshot_id(), Some(old_snapshot.snapshot_id())); + assert_eq!( + survivor.sequence_number(), + Some(old_snapshot.sequence_number()) + ); + assert_eq!( + survivor.data_file().partition, + Struct::from_iter([Some(Literal::timestamptz(TS_EVENING))]) + ); + + let manifest_list = table.manifest_list_reader(snapshot).load().await.unwrap(); + let old_spec_manifest = manifest_list + .entries() + .iter() + .find(|m| m.partition_spec_id == 0) + .unwrap(); + // The summary spans the deleted entry as well as the survivor, + // as it does in Java, where every entry updates it. + let summary = &old_spec_manifest.partitions.as_ref().unwrap()[0]; + let morning = Datum::timestamptz_micros(TS_MORNING).to_bytes().unwrap(); + let evening = Datum::timestamptz_micros(TS_EVENING).to_bytes().unwrap(); + assert_eq!(summary.lower_bound.as_ref(), Some(&morning)); + assert_eq!(summary.upper_bound.as_ref(), Some(&evening)); + + let totals = &snapshot.summary().additional_properties; + assert_eq!(totals.get("total-data-files").unwrap(), "3"); + assert_eq!(totals.get("total-records").unwrap(), "30"); + } + } + + /// One rewrite removing a file from each spec: both manifests are filtered + /// under their own spec and the merged file lands under the default spec. + #[tokio::test] + async fn test_rewrite_files_across_partition_specs() { + for format_version in [FormatVersion::V2, FormatVersion::V3] { + let catalog = new_memory_catalog().await; + let (table, _, [old_1, _old_2, new_1]) = + spec_evolved_table_with_files(&catalog, format_version).await; + + let merged = make_file_in_spec("test/merged.parquet", 20, 1, Literal::date(DAY)); + let tx = Transaction::new(&table); + let action = tx + .rewrite_files() + .delete_file(old_1) + .delete_file(new_1) + .add_file(merged); + let table = action.apply(tx).unwrap().commit(&catalog).await.unwrap(); + + assert_eq!( + live_files_by_manifest_spec(&table).await, + vec![ + ("test/merged.parquet".to_string(), 1), + ("test/old-2.parquet".to_string(), 0), + ], + "{format_version}" + ); + let totals = &table + .metadata() + .current_snapshot() + .unwrap() + .summary() + .additional_properties; + assert_eq!(totals.get("deleted-data-files").unwrap(), "2"); + assert_eq!(totals.get("total-data-files").unwrap(), "2"); + assert_eq!(totals.get("total-records").unwrap(), "30"); + } + } + + /// A second compaction that removes the last old-spec file drops the spec-0 + /// manifest, and the totals chained through both rewrites stay exact. + #[tokio::test] + async fn test_rewrite_files_retires_the_old_spec() { + for format_version in [FormatVersion::V2, FormatVersion::V3] { + let catalog = new_memory_catalog().await; + let (table, _, [old_1, old_2, new_1]) = + spec_evolved_table_with_files(&catalog, format_version).await; + + let merged = make_file_in_spec("test/merged-1.parquet", 10, 1, Literal::date(DAY)); + let tx = Transaction::new(&table); + let action = tx.rewrite_files().delete_file(old_1).add_file(merged); + let table = action.apply(tx).unwrap().commit(&catalog).await.unwrap(); + + let merged = make_file_in_spec("test/merged-2.parquet", 20, 1, Literal::date(DAY)); + let tx = Transaction::new(&table); + let action = tx + .rewrite_files() + .delete_file(old_2) + .delete_file(new_1) + .add_file(merged); + let table = action.apply(tx).unwrap().commit(&catalog).await.unwrap(); + + assert_eq!( + live_files_by_manifest_spec(&table).await, + vec![ + ("test/merged-1.parquet".to_string(), 1), + ("test/merged-2.parquet".to_string(), 1), + ], + "{format_version}" + ); + let snapshot = table.metadata().current_snapshot().unwrap(); + let manifest_list = table.manifest_list_reader(snapshot).load().await.unwrap(); + assert!( + manifest_list + .entries() + .iter() + .filter(|m| m.partition_spec_id == 0) + .all(|m| !m.has_added_files() && !m.has_existing_files()), + "{format_version}: a spec-0 manifest kept a live file" + ); + + let totals = &snapshot.summary().additional_properties; + assert_eq!(totals.get("total-data-files").unwrap(), "2"); + assert_eq!(totals.get("total-records").unwrap(), "30"); + + // The spec-0 manifest only carries the removal made by the snapshot + // that emptied it. The next rewrite has no reason to keep it. + let merged = make_file_in_spec("test/merged-3.parquet", 10, 1, Literal::date(DAY)); + let tx = Transaction::new(&table); + let action = tx + .rewrite_files() + .delete_file(make_file_in_spec( + "test/merged-1.parquet", + 10, + 1, + Literal::date(DAY), + )) + .add_file(merged); + let table = action.apply(tx).unwrap().commit(&catalog).await.unwrap(); + + let snapshot = table.metadata().current_snapshot().unwrap(); + let manifest_list = table.manifest_list_reader(snapshot).load().await.unwrap(); + assert!( + manifest_list + .entries() + .iter() + .all(|m| m.partition_spec_id == 1), + "{format_version}: a spec-0 manifest outlived its last file" + ); + } + } + + /// The summary counts and names the partitions a rewrite changed, each + /// under its file's own spec: one `identity(ts)` partition lost a file, one + /// `day(ts)` partition gained one. + #[tokio::test] + async fn test_rewrite_files_summarizes_partitions_of_both_specs() { + let catalog = new_memory_catalog().await; + let (table, _, [old_1, _old_2, _new_1]) = + spec_evolved_table_with_files(&catalog, FormatVersion::V2).await; + // Fast append reports the partition it changed. + let appended = &table + .metadata() + .current_snapshot() + .unwrap() + .summary() + .additional_properties; + assert_eq!( + appended.get("changed-partition-count").map(String::as_str), + Some("1") + ); + + let tx = Transaction::new(&table); + let tx = tx + .update_table_properties() + .set( + "write.summary.partition-limit".to_string(), + "10".to_string(), + ) + .apply(tx) + .unwrap(); + let table = tx.commit(&catalog).await.unwrap(); + + let merged = make_file_in_spec("test/merged.parquet", 10, 1, Literal::date(DAY)); + let tx = Transaction::new(&table); + let action = tx.rewrite_files().delete_file(old_1).add_file(merged); + let table = action.apply(tx).unwrap().commit(&catalog).await.unwrap(); + + let summary = &table + .metadata() + .current_snapshot() + .unwrap() + .summary() + .additional_properties; + assert_eq!( + summary.get("changed-partition-count").map(String::as_str), + Some("2"), + "{summary:?}" + ); + // The removed file is named under `identity(ts)`, the added one under + // `day(ts)`. + let mut partitions: Vec<&str> = summary + .keys() + .filter_map(|k| k.strip_prefix("partitions.")) + .collect(); + partitions.sort(); + assert_eq!( + partitions, + vec!["ts=2026-08-24+09%3A30%3A00+UTC", "ts_day=2026-08-24"], + "{summary:?}" + ); + } +} diff --git a/crates/iceberg/src/transaction/snapshot.rs b/crates/iceberg/src/transaction/snapshot.rs index a0ce1fba14..2918e31b77 100644 --- a/crates/iceberg/src/transaction/snapshot.rs +++ b/crates/iceberg/src/transaction/snapshot.rs @@ -109,7 +109,7 @@ pub(crate) trait ManifestProcess: Send + Sync { pub(crate) struct SnapshotProducer<'a> { pub(crate) table: &'a Table, - snapshot_id: i64, + pub(crate) snapshot_id: i64, commit_uuid: Uuid, snapshot_properties: HashMap, added_data_files: Vec, @@ -140,7 +140,9 @@ impl<'a> SnapshotProducer<'a> { for data_file in &self.added_data_files { if data_file.content_type() != crate::spec::DataContentType::Data { return Err(invalid_data!( - "Only data content type is allowed for fast append" + "Only data content type is allowed in added data files, but {} is {:?}", + data_file.file_path, + data_file.content_type() )); } // Check if the data file partition spec id matches the table default partition spec id. @@ -434,6 +436,43 @@ impl<'a> SnapshotProducer<'a> { mut self, snapshot_produce_operation: OP, process: MP, + ) -> Result { + // Calling self.summary() before self.produce_manifests() is important because self.added_data_files + // will be set to an empty vec after self.produce_manifests() returns, resulting in an empty summary + // being generated. + let summary = self.summary(&snapshot_produce_operation).map_err(|err| { + Error::new(ErrorKind::Unexpected, "Failed to create snapshot summary.").with_source(err) + })?; + + let new_manifests = self + .produce_manifests(&snapshot_produce_operation, &process) + .await?; + + self.write_snapshot(new_manifests, summary).await + } + + /// Commit a snapshot from externally produced manifests and summary. + /// + /// This is used by [`MergingSnapshotProducer`] which manages manifest + /// creation (including filtering deleted files) on its own and only needs + /// `SnapshotProducer` to write the manifest list, build the snapshot object, + /// and return the [`ActionCommit`]. + pub(crate) async fn commit_with_manifests( + self, + manifests: Vec, + summary: Summary, + ) -> Result { + self.write_snapshot(manifests, summary).await + } + + /// Writes the manifest list, builds the snapshot, and returns the [`ActionCommit`]. + /// + /// This is the shared implementation used by both [`commit`] and + /// [`commit_with_manifests`]. + async fn write_snapshot( + self, + manifests: Vec, + summary: Summary, ) -> Result { let manifest_list_path = self.generate_manifest_list_file_path(0)?; let next_seq_num = self.table.metadata().next_sequence_number(); @@ -472,18 +511,7 @@ impl<'a> SnapshotProducer<'a> { ), }; - // Calling self.summary() before self.produce_manifests() is important because self.added_data_files - // will be set to an empty vec after self.produce_manifests() returns, resulting in an empty summary - // being generated. - let summary = self.summary(&snapshot_produce_operation).map_err(|err| { - Error::new(ErrorKind::Unexpected, "Failed to create snapshot summary.").with_source(err) - })?; - - let new_manifests = self - .produce_manifests(&snapshot_produce_operation, &process) - .await?; - - manifest_list_writer.add_manifests(new_manifests.into_iter())?; + manifest_list_writer.add_manifests(manifests.into_iter())?; let writer_next_row_id = manifest_list_writer.next_row_id(); let file_metadata = manifest_list_writer.close().await?; let encryption_key_id = match encrypted_output {