From 6e4fc17a486abb5769efbcc77f0b1bc0da6ef7ba Mon Sep 17 00:00:00 2001 From: hotcache Date: Fri, 21 Aug 2026 13:49:30 +0900 Subject: [PATCH 01/23] feat(transaction): add MergingSnapshotProducer and RewriteFilesAction Add support for atomically replacing data files (compaction). - MergingSnapshotProducer: shared engine for operations that both add and delete files, with manifest filtering and summary computation - ManifestFilterManager: rewrites existing manifests to remove deleted file entries, with encryption support - RewriteFilesAction: thin action layer using Operation::Replace - Refactor SnapshotProducer to extract write_snapshot() shared method - Allow Operation::Replace in snapshot summary computation Closes #1607 Related: #2242, #624 --- crates/iceberg/src/spec/snapshot_summary.rs | 1 + crates/iceberg/src/transaction/merging.rs | 386 ++++++++++++++++++++ crates/iceberg/src/transaction/mod.rs | 12 + crates/iceberg/src/transaction/rewrite.rs | 323 ++++++++++++++++ crates/iceberg/src/transaction/snapshot.rs | 52 ++- 5 files changed, 761 insertions(+), 13 deletions(-) create mode 100644 crates/iceberg/src/transaction/merging.rs create mode 100644 crates/iceberg/src/transaction/rewrite.rs diff --git a/crates/iceberg/src/spec/snapshot_summary.rs b/crates/iceberg/src/spec/snapshot_summary.rs index c9fd022724..c7669eda21 100644 --- a/crates/iceberg/src/spec/snapshot_summary.rs +++ b/crates/iceberg/src/spec/snapshot_summary.rs @@ -338,6 +338,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(Error::new( ErrorKind::DataInvalid, diff --git a/crates/iceberg/src/transaction/merging.rs b/crates/iceberg/src/transaction/merging.rs new file mode 100644 index 0000000000..2c9ee86c03 --- /dev/null +++ b/crates/iceberg/src/transaction/merging.rs @@ -0,0 +1,386 @@ +// 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 uuid::Uuid; + +use crate::error::Result; +use crate::io::OutputFile; +use crate::spec::{ + DataContentType, DataFile, DataFileFormat, FormatVersion, ManifestContentType, ManifestEntry, + ManifestFile, ManifestStatus, ManifestWriter, ManifestWriterBuilder, Operation, PartitionSpec, + SchemaRef, SnapshotSummaryCollector, Summary, update_snapshot_summaries, +}; +use crate::table::Table; +use crate::transaction::ActionCommit; +use crate::transaction::snapshot::SnapshotProducer; +use crate::{Error, ErrorKind}; + +/// Create a manifest writer that handles encryption when available. +fn new_manifest_writer( + table: &Table, + output_file: OutputFile, + snapshot_id: Option, + 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 => Ok(builder.build_v2_data()), + FormatVersion::V3 => Ok(builder.build_v3_data()), + } +} + +/// 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, + ) -> 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(); + + let schema = table.metadata().current_schema().clone(); + let partition_spec = table.metadata().default_partition_spec().clone(); + + 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 = manifest_file.load_manifest(table.file_io()).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; + } + + // Manifest requires the current partition spec to match so we can + // rewrite it correctly. Non-default specs need a writer + // parameterised by the source spec, which is not yet implemented. + if manifest_file.partition_spec_id != table.metadata().default_partition_spec_id() { + return Err(Error::new( + ErrorKind::FeatureUnsupported, + format!( + "Cannot rewrite manifest with partition spec {} (table default is {}). \ + Rewriting manifests with non-default partition specs is not yet supported.", + manifest_file.partition_spec_id, + table.metadata().default_partition_spec_id(), + ), + )); + } + + // Rewrite: keep surviving entries as EXISTING, drop deleted ones. + let mut surviving_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(), + partition_spec.clone(), + ); + } else if entry.is_alive() { + // Surviving entry — re-emit as EXISTING with original ids preserved. + let existing = ManifestEntry::builder() + .status(ManifestStatus::Existing) + .snapshot_id( + entry + .snapshot_id() + .unwrap_or(manifest_file.added_snapshot_id), + ) + .sequence_number(entry.sequence_number().unwrap_or(0)) + .file_sequence_number(entry.file_sequence_number.unwrap_or(0)) + .data_file(entry.data_file().clone()) + .build(); + surviving_entries.push(existing); + } + // Already-deleted entries (status == Deleted) are dropped. + } + + if surviving_entries.is_empty() { + // Manifest is now empty — omit entirely. + continue; + } + + // Write the filtered manifest. + let new_manifest_path = format!( + "{}/{}-m-filter-{}.{}", + table.metadata().metadata_location()?, + Uuid::now_v7(), + manifest_file.partition_spec_id, + 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), + schema.clone(), + partition_spec.as_ref().clone(), + )?; + for entry in surviving_entries { + writer.add_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(Error::new( + ErrorKind::DataInvalid, + format!( + "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, + pub(crate) added_data_files: Vec, + pub(crate) deleted_data_files: Vec, + filter_manager: ManifestFilterManager, + snapshot_properties: HashMap, + commit_uuid: Uuid, +} + +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), + snapshot_properties: HashMap::new(), + commit_uuid: Uuid::now_v7(), + } + } + + pub(crate) fn add_data_file(&mut self, file: DataFile) { + self.added_data_files.push(file); + } + + 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); + } + + /// Produce manifests, compute summary, and commit a new snapshot. + pub(crate) async fn commit_snapshot(&self, table: &Table) -> Result { + // 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(), Vec::new()); + 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() + .filter(|e| { + e.has_added_files() || e.has_existing_files() || e.has_deleted_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) + .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).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) -> Result { + let new_manifest_path = format!( + "{}/{}-m-added.{}", + table.metadata().metadata_location()?, + self.commit_uuid, + 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), + schema, + partition_spec, + )?; + + let format_version = table.metadata().format_version(); + for data_file in &self.added_data_files { + if data_file.content_type() != DataContentType::Data { + return Err(Error::new( + ErrorKind::DataInvalid, + "Only data content type is allowed for rewrite files", + )); + } + 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(); + + let mut collector = SnapshotSummaryCollector::default(); + for file in &self.added_data_files { + collector.add_file(file, schema.clone(), partition_spec.clone()); + } + // Merge removal metrics from the filter manager. + collector.merge(removed_collector); + + let mut additional_properties = self.snapshot_properties.clone(); + additional_properties.extend(collector.build()); + + let summary = Summary { + operation: self.operation.clone(), + additional_properties, + }; + + 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 02acbc232c..b8f51a298a 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; use crate::transaction::append::FastAppendAction; use crate::transaction::expire_snapshots::ExpireSnapshotsAction; +use crate::transaction::rewrite::RewriteFilesAction; use crate::transaction::sort_order::ReplaceSortOrderAction; use crate::transaction::update_location::UpdateLocationAction; 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() + } + /// Creates replace sort order action. pub fn replace_sort_order(&self) -> ReplaceSortOrderAction { ReplaceSortOrderAction::new() diff --git a/crates/iceberg/src/transaction/rewrite.rs b/crates/iceberg/src/transaction/rewrite.rs new file mode 100644 index 0000000000..80f1d4ef8b --- /dev/null +++ b/crates/iceberg/src/transaction/rewrite.rs @@ -0,0 +1,323 @@ +// 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; +use crate::spec::{DataFile, Operation}; +use crate::table::Table; +use crate::transaction::merging::MergingSnapshotProducer; +use crate::transaction::{ActionCommit, TransactionAction}; +use crate::{Error, ErrorKind}; + +/// 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?; +/// ``` +pub struct RewriteFilesAction { + producer: MergingSnapshotProducer, +} + +impl RewriteFilesAction { + pub(crate) fn new() -> Self { + Self { + producer: MergingSnapshotProducer::new(Operation::Replace), + } + } + + /// 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 + } + + fn validate(&self) -> Result<()> { + if self.producer.deleted_data_files.is_empty() { + return Err(Error::new( + ErrorKind::DataInvalid, + "Rewrite files requires at least one file to delete", + )); + } + if self.producer.added_data_files.is_empty() { + return Err(Error::new( + ErrorKind::DataInvalid, + "Rewrite files requires at least one file to add", + )); + } + Ok(()) + } +} + +#[async_trait] +impl TransactionAction for RewriteFilesAction { + async fn commit(self: Arc, table: &Table) -> Result { + self.validate()?; + self.producer.commit_snapshot(table).await + } +} + +#[cfg(test)] +mod tests { + use crate::Catalog; + use crate::memory::tests::new_memory_catalog; + use crate::spec::{ + DataContentType, DataFileBuilder, DataFileFormat, Literal, Operation, Struct, + }; + use crate::transaction::tests::make_v3_minimal_table_in_catalog; + use crate::transaction::{ApplyTransactionAction, Transaction}; + + fn make_data_file( + table: &crate::table::Table, + path: &str, + records: u64, + size: u64, + ) -> crate::spec::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() + } + + /// Helper: append data files to a table and return the updated table. + async fn append_files( + catalog: &impl Catalog, + table: &crate::table::Table, + files: Vec, + ) -> crate::table::Table { + let tx = Transaction::new(table); + let tx = tx.fast_append().add_data_files(files).apply(tx).unwrap(); + tx.commit(catalog).await.unwrap() + } + + /// 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; + + // 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 = manifest_entry.load_manifest(table.file_io()).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; + + 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; + + // 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 = manifest_entry.load_manifest(table.file_io()).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"]); + } + + /// 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") + ); + } +} diff --git a/crates/iceberg/src/transaction/snapshot.rs b/crates/iceberg/src/transaction/snapshot.rs index 1f09a79a08..532671a6d1 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, @@ -441,6 +441,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(); @@ -479,18 +516,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(); manifest_list_writer.close().await?; From 64a5b1f0c83ecc1a1fb55057f89c54a5272b12a8 Mon Sep 17 00:00:00 2001 From: hotcache Date: Fri, 21 Aug 2026 22:56:38 +0900 Subject: [PATCH 02/23] fix(transaction): support non-default partition specs in manifest filtering Previously, ManifestFilterManager rejected manifests with non-default partition spec IDs with a FeatureUnsupported error. This broke tables that had undergone partition evolution, as old manifests carry their original spec ID. Now the filter manager resolves each manifest's own partition spec from table metadata and uses it for rewriting, allowing correct handling of partition-evolved tables. Co-Authored-By: Claude Opus 4.6 (1M context) --- crates/iceberg/src/transaction/merging.rs | 40 +++++++++++------------ 1 file changed, 20 insertions(+), 20 deletions(-) diff --git a/crates/iceberg/src/transaction/merging.rs b/crates/iceberg/src/transaction/merging.rs index 2c9ee86c03..b21df75c7c 100644 --- a/crates/iceberg/src/transaction/merging.rs +++ b/crates/iceberg/src/transaction/merging.rs @@ -104,9 +104,6 @@ impl ManifestFilterManager { let mut removed_collector = SnapshotSummaryCollector::default(); let mut found_paths: HashSet = HashSet::new(); - let schema = table.metadata().current_schema().clone(); - let partition_spec = table.metadata().default_partition_spec().clone(); - for manifest_file in &manifests { // Only filter data manifests; pass delete manifests through unchanged. if manifest_file.content != ManifestContentType::Data { @@ -128,20 +125,23 @@ impl ManifestFilterManager { continue; } - // Manifest requires the current partition spec to match so we can - // rewrite it correctly. Non-default specs need a writer - // parameterised by the source spec, which is not yet implemented. - if manifest_file.partition_spec_id != table.metadata().default_partition_spec_id() { - return Err(Error::new( - ErrorKind::FeatureUnsupported, - format!( - "Cannot rewrite manifest with partition spec {} (table default is {}). \ - Rewriting manifests with non-default partition specs is not yet supported.", - manifest_file.partition_spec_id, - table.metadata().default_partition_spec_id(), - ), - )); - } + // 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(|| { + Error::new( + ErrorKind::DataInvalid, + format!( + "Manifest references unknown partition spec {}", + manifest_file.partition_spec_id, + ), + ) + })? + .clone(); + let schema = table.metadata().current_schema().clone(); // Rewrite: keep surviving entries as EXISTING, drop deleted ones. let mut surviving_entries: Vec = Vec::new(); @@ -152,7 +152,7 @@ impl ManifestFilterManager { removed_collector.remove_file( entry.data_file(), schema.clone(), - partition_spec.clone(), + manifest_spec.clone(), ); } else if entry.is_alive() { // Surviving entry — re-emit as EXISTING with original ids preserved. @@ -177,7 +177,7 @@ impl ManifestFilterManager { continue; } - // Write the filtered manifest. + // Write the filtered manifest using the manifest's own partition spec. let new_manifest_path = format!( "{}/{}-m-filter-{}.{}", table.metadata().metadata_location()?, @@ -193,7 +193,7 @@ impl ManifestFilterManager { output_file, Some(snapshot_id), schema.clone(), - partition_spec.as_ref().clone(), + manifest_spec.as_ref().clone(), )?; for entry in surviving_entries { writer.add_entry(entry)?; From b02fb8eb24ad953644b41102e7fe44d617b83198 Mon Sep 17 00:00:00 2001 From: hotcache Date: Fri, 21 Aug 2026 23:02:04 +0900 Subject: [PATCH 03/23] refactor(transaction): encapsulate MergingSnapshotProducer fields Replace pub(crate) field access with accessor methods (has_added_data_files, has_deleted_data_files) to improve encapsulation. Downstream actions now use these methods instead of reaching into internal Vec fields directly. Co-Authored-By: Claude Opus 4.6 (1M context) --- crates/iceberg/src/transaction/merging.rs | 12 ++++++++++-- crates/iceberg/src/transaction/rewrite.rs | 4 ++-- 2 files changed, 12 insertions(+), 4 deletions(-) diff --git a/crates/iceberg/src/transaction/merging.rs b/crates/iceberg/src/transaction/merging.rs index b21df75c7c..af4f844a38 100644 --- a/crates/iceberg/src/transaction/merging.rs +++ b/crates/iceberg/src/transaction/merging.rs @@ -240,8 +240,8 @@ impl ManifestFilterManager { /// and validation rules. pub(crate) struct MergingSnapshotProducer { operation: Operation, - pub(crate) added_data_files: Vec, - pub(crate) deleted_data_files: Vec, + added_data_files: Vec, + deleted_data_files: Vec, filter_manager: ManifestFilterManager, snapshot_properties: HashMap, commit_uuid: Uuid, @@ -268,6 +268,14 @@ impl MergingSnapshotProducer { 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() + } + /// Produce manifests, compute summary, and commit a new snapshot. pub(crate) async fn commit_snapshot(&self, table: &Table) -> Result { // Create the SnapshotProducer first so we can use its snapshot_id diff --git a/crates/iceberg/src/transaction/rewrite.rs b/crates/iceberg/src/transaction/rewrite.rs index 80f1d4ef8b..a184c8d093 100644 --- a/crates/iceberg/src/transaction/rewrite.rs +++ b/crates/iceberg/src/transaction/rewrite.rs @@ -81,13 +81,13 @@ impl RewriteFilesAction { } fn validate(&self) -> Result<()> { - if self.producer.deleted_data_files.is_empty() { + if !self.producer.has_deleted_data_files() { return Err(Error::new( ErrorKind::DataInvalid, "Rewrite files requires at least one file to delete", )); } - if self.producer.added_data_files.is_empty() { + if !self.producer.has_added_data_files() { return Err(Error::new( ErrorKind::DataInvalid, "Rewrite files requires at least one file to add", From 1f412f8cd3eea8e7072f7778204933659b0be4f2 Mon Sep 17 00:00:00 2001 From: hotcache Date: Fri, 21 Aug 2026 23:29:19 +0900 Subject: [PATCH 04/23] refactor(transaction): address review feedback for PR1 - Add ManifestContentType param to new_manifest_writer for future delete manifest support - Remove unused snapshot_properties field from MergingSnapshotProducer - Move test helpers (make_data_file, append_files) to mod.rs as pub(crate) for reuse across test modules - Add starting_snapshot_id to RewriteFilesAction for future conflict detection support - Fix action-specific error message in generic engine Co-Authored-By: Claude Opus 4.6 (1M context) --- crates/iceberg/src/transaction/merging.rs | 24 ++++++----- crates/iceberg/src/transaction/mod.rs | 29 +++++++++++++- crates/iceberg/src/transaction/rewrite.rs | 49 ++++++----------------- 3 files changed, 54 insertions(+), 48 deletions(-) diff --git a/crates/iceberg/src/transaction/merging.rs b/crates/iceberg/src/transaction/merging.rs index af4f844a38..d8fc7d9b1b 100644 --- a/crates/iceberg/src/transaction/merging.rs +++ b/crates/iceberg/src/transaction/merging.rs @@ -43,6 +43,7 @@ fn new_manifest_writer( table: &Table, output_file: OutputFile, snapshot_id: Option, + content: ManifestContentType, schema: SchemaRef, partition_spec: PartitionSpec, ) -> Result { @@ -59,8 +60,14 @@ fn new_manifest_writer( match table.metadata().format_version() { FormatVersion::V1 => Ok(builder.build_v1()), - FormatVersion::V2 => Ok(builder.build_v2_data()), - FormatVersion::V3 => Ok(builder.build_v3_data()), + 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()), + }, } } @@ -111,7 +118,7 @@ impl ManifestFilterManager { continue; } - let manifest = manifest_file.load_manifest(table.file_io()).await?; + let manifest = table.manifest_reader().read(manifest_file).await?; // Check whether this manifest contains any files we want to delete. let has_deletes = manifest @@ -192,6 +199,7 @@ impl ManifestFilterManager { table, output_file, Some(snapshot_id), + ManifestContentType::Data, schema.clone(), manifest_spec.as_ref().clone(), )?; @@ -243,7 +251,6 @@ pub(crate) struct MergingSnapshotProducer { added_data_files: Vec, deleted_data_files: Vec, filter_manager: ManifestFilterManager, - snapshot_properties: HashMap, commit_uuid: Uuid, } @@ -254,7 +261,6 @@ impl MergingSnapshotProducer { added_data_files: Vec::new(), deleted_data_files: Vec::new(), filter_manager: ManifestFilterManager::new(true), - snapshot_properties: HashMap::new(), commit_uuid: Uuid::now_v7(), } } @@ -337,6 +343,7 @@ impl MergingSnapshotProducer { table, output_file, Some(snapshot_id), + ManifestContentType::Data, schema, partition_spec, )?; @@ -346,7 +353,7 @@ impl MergingSnapshotProducer { if data_file.content_type() != DataContentType::Data { return Err(Error::new( ErrorKind::DataInvalid, - "Only data content type is allowed for rewrite files", + "Only data content type is allowed in added_data_files", )); } let entry_builder = ManifestEntry::builder() @@ -380,12 +387,9 @@ impl MergingSnapshotProducer { // Merge removal metrics from the filter manager. collector.merge(removed_collector); - let mut additional_properties = self.snapshot_properties.clone(); - additional_properties.extend(collector.build()); - let summary = Summary { operation: self.operation.clone(), - additional_properties, + additional_properties: collector.build(), }; let previous_snapshot = table_metadata.current_snapshot(); diff --git a/crates/iceberg/src/transaction/mod.rs b/crates/iceberg/src/transaction/mod.rs index b8f51a298a..3fe1b161e7 100644 --- a/crates/iceberg/src/transaction/mod.rs +++ b/crates/iceberg/src/transaction/mod.rs @@ -160,7 +160,7 @@ impl Transaction { /// the logical table contents unchanged. The resulting snapshot uses /// [`Operation::Replace`](crate::spec::Operation::Replace). pub fn rewrite_files(&self) -> RewriteFilesAction { - RewriteFilesAction::new() + RewriteFilesAction::new(self.table.metadata().current_snapshot_id()) } /// Creates replace sort order action. @@ -266,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; @@ -274,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/{}", diff --git a/crates/iceberg/src/transaction/rewrite.rs b/crates/iceberg/src/transaction/rewrite.rs index a184c8d093..3bf7202ea9 100644 --- a/crates/iceberg/src/transaction/rewrite.rs +++ b/crates/iceberg/src/transaction/rewrite.rs @@ -54,12 +54,17 @@ use crate::{Error, ErrorKind}; /// ``` pub struct RewriteFilesAction { producer: MergingSnapshotProducer, + /// The snapshot ID at which this rewrite started reading. Used to + /// detect conflicting deletes added after this point. + #[allow(dead_code)] // Will be used for conflict detection in a follow-up PR. + starting_snapshot_id: Option, } impl RewriteFilesAction { - pub(crate) fn new() -> Self { + pub(crate) fn new(starting_snapshot_id: Option) -> Self { Self { producer: MergingSnapshotProducer::new(Operation::Replace), + starting_snapshot_id, } } @@ -107,43 +112,13 @@ impl TransactionAction for RewriteFilesAction { #[cfg(test)] mod tests { - use crate::Catalog; use crate::memory::tests::new_memory_catalog; - use crate::spec::{ - DataContentType, DataFileBuilder, DataFileFormat, Literal, Operation, Struct, + use crate::spec::Operation; + use crate::transaction::tests::{ + append_files, make_data_file, make_v3_minimal_table_in_catalog, }; - use crate::transaction::tests::make_v3_minimal_table_in_catalog; use crate::transaction::{ApplyTransactionAction, Transaction}; - fn make_data_file( - table: &crate::table::Table, - path: &str, - records: u64, - size: u64, - ) -> crate::spec::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() - } - - /// Helper: append data files to a table and return the updated table. - async fn append_files( - catalog: &impl Catalog, - table: &crate::table::Table, - files: Vec, - ) -> crate::table::Table { - let tx = Transaction::new(table); - let tx = tx.fast_append().add_data_files(files).apply(tx).unwrap(); - tx.commit(catalog).await.unwrap() - } - /// E2E: Compact 3 small files into 1 merged file. /// Verify: operation=Replace, file counts, record counts. #[tokio::test] @@ -193,7 +168,8 @@ mod tests { 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 = manifest_entry.load_manifest(table.file_io()).await.unwrap(); + 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()); @@ -289,7 +265,8 @@ mod tests { 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 = manifest_entry.load_manifest(table.file_io()).await.unwrap(); + 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()); From 0fb348cda5be5e79b17942e500b937495a3dc406 Mon Sep 17 00:00:00 2001 From: hotcache Date: Sat, 22 Aug 2026 08:29:10 +0900 Subject: [PATCH 05/23] test(transaction): add coverage for empty table rewrite and manifest omission - test_rewrite_files_empty_table: rewrite on table with no snapshot - test_rewrite_files_removes_empty_manifest: verify empty manifests are omitted, not kept as empty files Co-Authored-By: Claude Opus 4.6 (1M context) --- crates/iceberg/src/transaction/rewrite.rs | 51 +++++++++++++++++++++++ 1 file changed, 51 insertions(+) diff --git a/crates/iceberg/src/transaction/rewrite.rs b/crates/iceberg/src/transaction/rewrite.rs index 3bf7202ea9..8b9e3f3136 100644 --- a/crates/iceberg/src/transaction/rewrite.rs +++ b/crates/iceberg/src/transaction/rewrite.rs @@ -277,6 +277,57 @@ mod tests { assert_eq!(live_files, vec!["test/3.parquet", "test/merged.parquet"]); } + /// 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 should be omitted. + 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(); + + // Verify: only 1 manifest (the new one), not 2. + let snapshot = table.metadata().current_snapshot().unwrap(); + let manifest_list = table.manifest_list_reader(snapshot).load().await.unwrap(); + assert_eq!( + manifest_list.entries().len(), + 1, + "empty manifest should be omitted, not kept" + ); + } + /// Rewrite with no adds should fail validation. #[tokio::test] async fn test_rewrite_files_no_adds() { From 4208c44525330bfe8104d6bb78d89b473ac3679b Mon Sep 17 00:00:00 2001 From: hotcache Date: Tue, 25 Aug 2026 11:35:09 +0900 Subject: [PATCH 06/23] fix(transaction): address review feedback from #3046 - Return explicit errors instead of silently defaulting to 0 for sequence_number and file_sequence_number in manifest entries - Use standard manifest naming scheme (-m.avro) instead of custom names (-m-filter-, -m-added) Co-Authored-By: Claude Opus 4.6 (1M context) --- crates/iceberg/src/transaction/merging.rs | 52 +++++++++++++++++++---- 1 file changed, 43 insertions(+), 9 deletions(-) diff --git a/crates/iceberg/src/transaction/merging.rs b/crates/iceberg/src/transaction/merging.rs index d8fc7d9b1b..70299a1001 100644 --- a/crates/iceberg/src/transaction/merging.rs +++ b/crates/iceberg/src/transaction/merging.rs @@ -23,6 +23,7 @@ //! [`SnapshotProducer`]. use std::collections::{HashMap, HashSet}; +use std::ops::RangeFrom; use uuid::Uuid; @@ -102,6 +103,8 @@ impl ManifestFilterManager { table: &Table, manifests: Vec, snapshot_id: i64, + commit_uuid: Uuid, + manifest_counter: &mut RangeFrom, ) -> Result<(Vec, SnapshotSummaryCollector)> { if self.deleted_file_paths.is_empty() { return Ok((manifests, SnapshotSummaryCollector::default())); @@ -163,6 +166,24 @@ impl ManifestFilterManager { ); } else if entry.is_alive() { // Surviving entry — re-emit as EXISTING with original ids preserved. + let seq = entry.sequence_number().ok_or_else(|| { + Error::new( + ErrorKind::DataInvalid, + format!( + "Manifest entry for {} is missing sequence_number", + entry.file_path(), + ), + ) + })?; + let file_seq = entry.file_sequence_number.ok_or_else(|| { + Error::new( + ErrorKind::DataInvalid, + format!( + "Manifest entry for {} is missing file_sequence_number", + entry.file_path(), + ), + ) + })?; let existing = ManifestEntry::builder() .status(ManifestStatus::Existing) .snapshot_id( @@ -170,8 +191,8 @@ impl ManifestFilterManager { .snapshot_id() .unwrap_or(manifest_file.added_snapshot_id), ) - .sequence_number(entry.sequence_number().unwrap_or(0)) - .file_sequence_number(entry.file_sequence_number.unwrap_or(0)) + .sequence_number(seq) + .file_sequence_number(file_seq) .data_file(entry.data_file().clone()) .build(); surviving_entries.push(existing); @@ -186,10 +207,10 @@ impl ManifestFilterManager { // Write the filtered manifest using the manifest's own partition spec. let new_manifest_path = format!( - "{}/{}-m-filter-{}.{}", + "{}/{}-m{}.{}", table.metadata().metadata_location()?, - Uuid::now_v7(), - manifest_file.partition_spec_id, + commit_uuid, + manifest_counter.next().unwrap(), DataFileFormat::Avro, ); let output_file = table.file_io().new_output(new_manifest_path)?; @@ -290,6 +311,7 @@ impl MergingSnapshotProducer { let snapshot_producer = SnapshotProducer::new(table, self.commit_uuid, HashMap::new(), Vec::new()); let snapshot_id = snapshot_producer.snapshot_id; + let mut manifest_counter: RangeFrom = (0..); // 1. Load existing manifests from the current snapshot. let existing_manifests = match table.metadata().current_snapshot() { @@ -310,12 +332,18 @@ impl MergingSnapshotProducer { // 2. Filter existing manifests — remove deleted file entries. let (mut filtered_manifests, removed_collector) = self .filter_manager - .filter_manifests(table, existing_manifests, snapshot_id) + .filter_manifests( + table, + existing_manifests, + snapshot_id, + self.commit_uuid, + &mut 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).await?; + let added_manifest = self.write_added_manifest(table, snapshot_id, &mut manifest_counter).await?; filtered_manifests.push(added_manifest); } @@ -328,11 +356,17 @@ impl MergingSnapshotProducer { .await } - async fn write_added_manifest(&self, table: &Table, snapshot_id: i64) -> Result { + async fn write_added_manifest( + &self, + table: &Table, + snapshot_id: i64, + manifest_counter: &mut RangeFrom, + ) -> Result { let new_manifest_path = format!( - "{}/{}-m-added.{}", + "{}/{}-m{}.{}", table.metadata().metadata_location()?, self.commit_uuid, + manifest_counter.next().unwrap(), DataFileFormat::Avro, ); let output_file = table.file_io().new_output(new_manifest_path)?; From 1dd97f04a9781f374df8dff85f85e7b1b07367ed Mon Sep 17 00:00:00 2001 From: Jegor van Opdorp Date: Fri, 18 Sep 2026 17:25:46 +0200 Subject: [PATCH 07/23] Preserve survivor entries as Existing when filtering manifests `ManifestFilterManager::filter_manifests` built each surviving entry with status Existing and the original snapshot id and sequence numbers, then handed it to `ManifestWriter::add_entry`, which rewrites the status to Added, replaces the snapshot id with the new snapshot's, and clears the file sequence number. Survivors of a rewrite were therefore recorded as newly added files of the rewrite snapshot. Use `add_existing_entry`, which only sets the status and leaves the ids the filter already resolved, and drop its `#[allow(dead_code)]`. --- crates/iceberg/src/spec/manifest/writer.rs | 4 ---- crates/iceberg/src/transaction/merging.rs | 2 +- crates/iceberg/src/transaction/rewrite.rs | 22 +++++++++++++++++++--- 3 files changed, 20 insertions(+), 8 deletions(-) diff --git a/crates/iceberg/src/spec/manifest/writer.rs b/crates/iceberg/src/spec/manifest/writer.rs index 7256692950..c58425ac31 100644 --- a/crates/iceberg/src/spec/manifest/writer.rs +++ b/crates/iceberg/src/spec/manifest/writer.rs @@ -371,10 +371,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/transaction/merging.rs b/crates/iceberg/src/transaction/merging.rs index 70299a1001..2c594e60df 100644 --- a/crates/iceberg/src/transaction/merging.rs +++ b/crates/iceberg/src/transaction/merging.rs @@ -225,7 +225,7 @@ impl ManifestFilterManager { manifest_spec.as_ref().clone(), )?; for entry in surviving_entries { - writer.add_entry(entry)?; + writer.add_existing_entry(entry)?; } let new_manifest = writer.write_manifest_file().await?; result.push(new_manifest); diff --git a/crates/iceberg/src/transaction/rewrite.rs b/crates/iceberg/src/transaction/rewrite.rs index 8b9e3f3136..c8b3524681 100644 --- a/crates/iceberg/src/transaction/rewrite.rs +++ b/crates/iceberg/src/transaction/rewrite.rs @@ -113,7 +113,7 @@ impl TransactionAction for RewriteFilesAction { #[cfg(test)] mod tests { use crate::memory::tests::new_memory_catalog; - use crate::spec::Operation; + use crate::spec::{ManifestStatus, Operation}; use crate::transaction::tests::{ append_files, make_data_file, make_v3_minimal_table_in_catalog, }; @@ -240,6 +240,7 @@ mod tests { 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); @@ -264,17 +265,32 @@ mod tests { // 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(); + let mut survivor = None; for manifest_entry in manifest_list.entries() { - let manifest = table.manifest_reader().read(manifest_entry).await.unwrap() -; + 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()); } + if entry.file_path() == "test/3.parquet" { + survivor = Some(entry.clone()); + } } } live_files.sort(); assert_eq!(live_files, vec!["test/3.parquet", "test/merged.parquet"]); + + let survivor = survivor.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()) + ); } /// Rewrite on an empty table (no snapshot) should fail because From 243f8f32b676a16d5d2970b57b72911b53f2f0c2 Mon Sep 17 00:00:00 2001 From: Jegor van Opdorp Date: Fri, 18 Sep 2026 17:28:56 +0200 Subject: [PATCH 08/23] Allow a rewrite to set the data sequence number of added files Java's `RewriteFiles.dataSequenceNumber` lets a compaction record the sequence number of the files it replaces, so the rewritten file does not shadow deletes written concurrently at a higher sequence number. Without it the added files inherit the rewrite snapshot's own sequence number. `RewriteFilesAction::data_sequence_number` sets it, and `write_added_manifest` then writes the added files through `ManifestWriter::add_file`, which records the given number on the entry. A number above the one the new snapshot will carry is rejected at commit; on a V1 table that is 0, matching the V1 entry layout, which records no sequence number at all. The parameter is `i64`, not `u64`, to match `ManifestWriter::add_file` and every other sequence number in the crate. --- crates/iceberg/src/transaction/merging.rs | 32 ++++++- crates/iceberg/src/transaction/rewrite.rs | 108 ++++++++++++++++++++-- 2 files changed, 131 insertions(+), 9 deletions(-) diff --git a/crates/iceberg/src/transaction/merging.rs b/crates/iceberg/src/transaction/merging.rs index 2c594e60df..8759c7e0c7 100644 --- a/crates/iceberg/src/transaction/merging.rs +++ b/crates/iceberg/src/transaction/merging.rs @@ -273,6 +273,7 @@ pub(crate) struct MergingSnapshotProducer { deleted_data_files: Vec, filter_manager: ManifestFilterManager, commit_uuid: Uuid, + data_sequence_number: Option, } impl MergingSnapshotProducer { @@ -283,6 +284,7 @@ impl MergingSnapshotProducer { deleted_data_files: Vec::new(), filter_manager: ManifestFilterManager::new(true), commit_uuid: Uuid::now_v7(), + data_sequence_number: None, } } @@ -290,6 +292,10 @@ impl MergingSnapshotProducer { 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); @@ -303,8 +309,26 @@ impl MergingSnapshotProducer { !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(Error::new( + ErrorKind::DataInvalid, + format!( + "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. @@ -343,7 +367,9 @@ impl MergingSnapshotProducer { // 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, &mut manifest_counter).await?; + let added_manifest = self + .write_added_manifest(table, snapshot_id, &mut manifest_counter) + .await?; filtered_manifests.push(added_manifest); } @@ -390,6 +416,10 @@ impl MergingSnapshotProducer { "Only data content type is allowed in 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()); diff --git a/crates/iceberg/src/transaction/rewrite.rs b/crates/iceberg/src/transaction/rewrite.rs index c8b3524681..9f4339c42f 100644 --- a/crates/iceberg/src/transaction/rewrite.rs +++ b/crates/iceberg/src/transaction/rewrite.rs @@ -85,6 +85,18 @@ impl RewriteFilesAction { 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(Error::new( @@ -112,13 +124,31 @@ impl TransactionAction for RewriteFilesAction { #[cfg(test)] mod tests { + use crate::ErrorKind; use crate::memory::tests::new_memory_catalog; - use crate::spec::{ManifestStatus, Operation}; + use crate::spec::{ManifestEntryRef, ManifestStatus, Operation, SnapshotRef}; + use crate::table::Table; use crate::transaction::tests::{ append_files, make_data_file, make_v3_minimal_table_in_catalog, }; use crate::transaction::{ApplyTransactionAction, Transaction}; + /// 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] @@ -168,8 +198,7 @@ mod tests { 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() -; + 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()); @@ -265,22 +294,20 @@ mod tests { // 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(); - let mut survivor = None; 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()); } - if entry.file_path() == "test/3.parquet" { - survivor = Some(entry.clone()); - } } } live_files.sort(); assert_eq!(live_files, vec!["test/3.parquet", "test/merged.parquet"]); - let survivor = survivor.expect("surviving file should still be in a manifest"); + 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!( @@ -293,6 +320,71 @@ mod tests { ); } + /// 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] From 291b9b6813f79c96cac14ccec7715ea0a8c4fae6 Mon Sep 17 00:00:00 2001 From: Jegor van Opdorp Date: Fri, 18 Sep 2026 17:33:23 +0200 Subject: [PATCH 09/23] Run the producer validations before committing a rewrite `commit_snapshot` handed `SnapshotProducer` an empty added-file list and went straight to `commit_with_manifests`, so neither `validate_added_data_files` nor `validate_duplicate_files` ever saw the files a rewrite adds. A file carrying a partition value the default spec cannot hold, or a path that is already live, was written into a manifest and committed. Pass the added files to the producer and run both validations first, and drop `write_added_manifest`'s own content-type check, which `validate_added_data_files` now makes unreachable. --- crates/iceberg/src/transaction/merging.rs | 22 +++++----- crates/iceberg/src/transaction/rewrite.rs | 51 ++++++++++++++++++++++- 2 files changed, 61 insertions(+), 12 deletions(-) diff --git a/crates/iceberg/src/transaction/merging.rs b/crates/iceberg/src/transaction/merging.rs index 8759c7e0c7..f3f38a6f7b 100644 --- a/crates/iceberg/src/transaction/merging.rs +++ b/crates/iceberg/src/transaction/merging.rs @@ -30,9 +30,9 @@ use uuid::Uuid; use crate::error::Result; use crate::io::OutputFile; use crate::spec::{ - DataContentType, DataFile, DataFileFormat, FormatVersion, ManifestContentType, ManifestEntry, - ManifestFile, ManifestStatus, ManifestWriter, ManifestWriterBuilder, Operation, PartitionSpec, - SchemaRef, SnapshotSummaryCollector, Summary, update_snapshot_summaries, + DataFile, DataFileFormat, FormatVersion, ManifestContentType, ManifestEntry, ManifestFile, + ManifestStatus, ManifestWriter, ManifestWriterBuilder, Operation, PartitionSpec, SchemaRef, + SnapshotSummaryCollector, Summary, update_snapshot_summaries, }; use crate::table::Table; use crate::transaction::ActionCommit; @@ -332,8 +332,14 @@ impl MergingSnapshotProducer { // 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(), Vec::new()); + 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; let mut manifest_counter: RangeFrom = (0..); @@ -410,12 +416,6 @@ impl MergingSnapshotProducer { let format_version = table.metadata().format_version(); for data_file in &self.added_data_files { - if data_file.content_type() != DataContentType::Data { - return Err(Error::new( - ErrorKind::DataInvalid, - "Only data content type is allowed in added_data_files", - )); - } if let Some(sequence_number) = self.data_sequence_number { writer.add_file(data_file.clone(), sequence_number)?; continue; diff --git a/crates/iceberg/src/transaction/rewrite.rs b/crates/iceberg/src/transaction/rewrite.rs index 9f4339c42f..a0f89e6094 100644 --- a/crates/iceberg/src/transaction/rewrite.rs +++ b/crates/iceberg/src/transaction/rewrite.rs @@ -126,7 +126,7 @@ impl TransactionAction for RewriteFilesAction { mod tests { use crate::ErrorKind; use crate::memory::tests::new_memory_catalog; - use crate::spec::{ManifestEntryRef, ManifestStatus, Operation, SnapshotRef}; + use crate::spec::{Literal, ManifestEntryRef, ManifestStatus, Operation, SnapshotRef, Struct}; use crate::table::Table; use crate::transaction::tests::{ append_files, make_data_file, make_v3_minimal_table_in_catalog, @@ -436,6 +436,55 @@ mod tests { ); } + /// 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}" + ); + } + + /// 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() { From d969ab3746d7983ed4eceec51e1c357acc6c690e Mon Sep 17 00:00:00 2001 From: Jegor van Opdorp Date: Fri, 18 Sep 2026 17:42:32 +0200 Subject: [PATCH 10/23] Keep manifest names unique across commit attempts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `commit_snapshot` created a fresh `(0..)` counter on every call while the commit uuid stayed fixed for the life of the action, so a retried commit rewrote `{uuid}-m0.avro` over the manifest the previous attempt had written under the same name. Java keeps `manifestCount` on the snapshot producer and never resets it. The counter cannot live on Rust's `SnapshotProducer`: that is built fresh inside each `commit_snapshot` call and so resets with it. The object that survives a retry is the action, which the transaction re-invokes through an `Arc`, so the counter goes on `MergingSnapshotProducer` as an `AtomicU64` — `commit` only has `&self`. `fast_append` is unaffected: it draws a new commit uuid per attempt unless the caller pins one. --- crates/iceberg/src/transaction/merging.rs | 20 +++++---- crates/iceberg/src/transaction/rewrite.rs | 55 ++++++++++++++++++++++- 2 files changed, 66 insertions(+), 9 deletions(-) diff --git a/crates/iceberg/src/transaction/merging.rs b/crates/iceberg/src/transaction/merging.rs index f3f38a6f7b..5cd25e0b5d 100644 --- a/crates/iceberg/src/transaction/merging.rs +++ b/crates/iceberg/src/transaction/merging.rs @@ -23,7 +23,7 @@ //! [`SnapshotProducer`]. use std::collections::{HashMap, HashSet}; -use std::ops::RangeFrom; +use std::sync::atomic::{AtomicU64, Ordering}; use uuid::Uuid; @@ -104,7 +104,7 @@ impl ManifestFilterManager { manifests: Vec, snapshot_id: i64, commit_uuid: Uuid, - manifest_counter: &mut RangeFrom, + manifest_counter: &AtomicU64, ) -> Result<(Vec, SnapshotSummaryCollector)> { if self.deleted_file_paths.is_empty() { return Ok((manifests, SnapshotSummaryCollector::default())); @@ -210,7 +210,7 @@ impl ManifestFilterManager { "{}/{}-m{}.{}", table.metadata().metadata_location()?, commit_uuid, - manifest_counter.next().unwrap(), + manifest_counter.fetch_add(1, Ordering::Relaxed), DataFileFormat::Avro, ); let output_file = table.file_io().new_output(new_manifest_path)?; @@ -274,6 +274,10 @@ pub(crate) struct MergingSnapshotProducer { 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 { @@ -285,6 +289,7 @@ impl MergingSnapshotProducer { filter_manager: ManifestFilterManager::new(true), commit_uuid: Uuid::now_v7(), data_sequence_number: None, + manifest_counter: AtomicU64::new(0), } } @@ -341,7 +346,6 @@ impl MergingSnapshotProducer { snapshot_producer.validate_added_data_files()?; snapshot_producer.validate_duplicate_files().await?; let snapshot_id = snapshot_producer.snapshot_id; - let mut manifest_counter: RangeFrom = (0..); // 1. Load existing manifests from the current snapshot. let existing_manifests = match table.metadata().current_snapshot() { @@ -367,14 +371,14 @@ impl MergingSnapshotProducer { existing_manifests, snapshot_id, self.commit_uuid, - &mut manifest_counter, + &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, &mut manifest_counter) + .write_added_manifest(table, snapshot_id, &self.manifest_counter) .await?; filtered_manifests.push(added_manifest); } @@ -392,13 +396,13 @@ impl MergingSnapshotProducer { &self, table: &Table, snapshot_id: i64, - manifest_counter: &mut RangeFrom, + manifest_counter: &AtomicU64, ) -> Result { let new_manifest_path = format!( "{}/{}-m{}.{}", table.metadata().metadata_location()?, self.commit_uuid, - manifest_counter.next().unwrap(), + manifest_counter.fetch_add(1, Ordering::Relaxed), DataFileFormat::Avro, ); let output_file = table.file_io().new_output(new_manifest_path)?; diff --git a/crates/iceberg/src/transaction/rewrite.rs b/crates/iceberg/src/transaction/rewrite.rs index a0f89e6094..86186a98fe 100644 --- a/crates/iceberg/src/transaction/rewrite.rs +++ b/crates/iceberg/src/transaction/rewrite.rs @@ -124,7 +124,9 @@ impl TransactionAction for RewriteFilesAction { #[cfg(test)] mod tests { - use crate::ErrorKind; + use std::sync::Arc; + + use super::*; use crate::memory::tests::new_memory_catalog; use crate::spec::{Literal, ManifestEntryRef, ManifestStatus, Operation, SnapshotRef, Struct}; use crate::table::Table; @@ -132,6 +134,7 @@ mod tests { append_files, make_data_file, make_v3_minimal_table_in_catalog, }; use crate::transaction::{ApplyTransactionAction, Transaction}; + use crate::{ErrorKind, TableUpdate}; /// Read back the manifest entry for `path` from the manifests of `snapshot`. async fn find_entry( @@ -436,6 +439,56 @@ mod tests { ); } + /// 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] From 82ad3ae5ffcade761fab4b20e4f64c7ab39c04db Mon Sep 17 00:00:00 2001 From: Jegor van Opdorp Date: Fri, 18 Sep 2026 17:54:11 +0200 Subject: [PATCH 11/23] Drop manifests an earlier snapshot left all-deleted The filter carried a manifest forward on `has_deleted_files()`, so a manifest whose entries are all Deleted stayed in every following snapshot for the rest of the table's life. Java keeps an existing manifest on `hasAddedFiles() || hasExistingFiles()`; its third arm matches the snapshot being produced, which no manifest read from the current snapshot can carry. The manifests this commit rewrites are produced by the filter pass and kept on their own. --- crates/iceberg/src/transaction/merging.rs | 5 +- crates/iceberg/src/transaction/rewrite.rs | 123 +++++++++++++++++++++- 2 files changed, 124 insertions(+), 4 deletions(-) diff --git a/crates/iceberg/src/transaction/merging.rs b/crates/iceberg/src/transaction/merging.rs index 5cd25e0b5d..3b4363a409 100644 --- a/crates/iceberg/src/transaction/merging.rs +++ b/crates/iceberg/src/transaction/merging.rs @@ -354,9 +354,8 @@ impl MergingSnapshotProducer { manifest_list .entries() .iter() - .filter(|e| { - e.has_added_files() || e.has_existing_files() || e.has_deleted_files() - }) + // 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() } diff --git a/crates/iceberg/src/transaction/rewrite.rs b/crates/iceberg/src/transaction/rewrite.rs index 86186a98fe..7348fdf4d6 100644 --- a/crates/iceberg/src/transaction/rewrite.rs +++ b/crates/iceberg/src/transaction/rewrite.rs @@ -126,9 +126,14 @@ impl TransactionAction for RewriteFilesAction { mod tests { use std::sync::Arc; + use uuid::Uuid; + use super::*; use crate::memory::tests::new_memory_catalog; - use crate::spec::{Literal, ManifestEntryRef, ManifestStatus, Operation, SnapshotRef, Struct}; + use crate::spec::{ + Literal, ManifestEntryRef, ManifestListWriter, ManifestStatus, ManifestWriterBuilder, + Operation, SnapshotRef, Struct, + }; use crate::table::Table; use crate::transaction::tests::{ append_files, make_data_file, make_v3_minimal_table_in_catalog, @@ -439,6 +444,122 @@ mod tests { ); } + /// 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 snapshot = table.metadata().current_snapshot().unwrap(); + 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()); + + let mut entries = table + .manifest_list_reader(snapshot) + .load() + .await + .unwrap() + .consume_entries() + .into_iter() + .collect::>(); + entries.push(delete_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(); + + delete_manifest_path + } + /// 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] From ddafede1a48c65b7cd80e56a1efb997232879d94 Mon Sep 17 00:00:00 2001 From: Jegor van Opdorp Date: Fri, 18 Sep 2026 17:59:07 +0200 Subject: [PATCH 12/23] Refuse a rewrite when deletes landed after its starting snapshot `starting_snapshot_id` was recorded and never read. A rewrite planned against one snapshot and committed against a later one silently drops any delete written in between: the replacement file keeps the deleted rows. Java resolves this per file in `validateNoNewDeletesForDataFiles`, which needs a delete-file index this crate does not have yet. Until then the commit fails closed when the current snapshot carries a delete manifest with a sequence number above the starting snapshot's. --- crates/iceberg/src/transaction/rewrite.rs | 120 ++++++++++++++++++++-- 1 file changed, 111 insertions(+), 9 deletions(-) diff --git a/crates/iceberg/src/transaction/rewrite.rs b/crates/iceberg/src/transaction/rewrite.rs index 7348fdf4d6..9d264a4aa0 100644 --- a/crates/iceberg/src/transaction/rewrite.rs +++ b/crates/iceberg/src/transaction/rewrite.rs @@ -29,7 +29,7 @@ use std::sync::Arc; use async_trait::async_trait; use crate::error::Result; -use crate::spec::{DataFile, Operation}; +use crate::spec::{DataFile, ManifestContentType, Operation}; use crate::table::Table; use crate::transaction::merging::MergingSnapshotProducer; use crate::transaction::{ActionCommit, TransactionAction}; @@ -56,7 +56,6 @@ pub struct RewriteFilesAction { producer: MergingSnapshotProducer, /// The snapshot ID at which this rewrite started reading. Used to /// detect conflicting deletes added after this point. - #[allow(dead_code)] // Will be used for conflict detection in a follow-up PR. starting_snapshot_id: Option, } @@ -112,12 +111,49 @@ impl RewriteFilesAction { } 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<()> { + let (Some(starting_snapshot_id), Some(current_snapshot)) = + (self.starting_snapshot_id, table.metadata().current_snapshot()) + else { + return Ok(()); + }; + let starting_sequence_number = table + .metadata() + .snapshot_by_id(starting_snapshot_id) + .map_or(0, |snapshot| 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(Error::new( + ErrorKind::DataInvalid, + format!( + "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 } } @@ -131,8 +167,8 @@ mod tests { use super::*; use crate::memory::tests::new_memory_catalog; use crate::spec::{ - Literal, ManifestEntryRef, ManifestListWriter, ManifestStatus, ManifestWriterBuilder, - Operation, SnapshotRef, Struct, + DataContentType, Literal, ManifestEntryRef, ManifestFile, ManifestListWriter, + ManifestStatus, ManifestWriterBuilder, Operation, SnapshotRef, Struct, }; use crate::table::Table; use crate::transaction::tests::{ @@ -487,7 +523,6 @@ mod tests { /// 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 snapshot = table.metadata().current_snapshot().unwrap(); let added_sequence_number = table .metadata() .snapshot_by_id(added_snapshot_id) @@ -529,15 +564,54 @@ mod tests { assert!(!delete_manifest.has_added_files()); assert!(!delete_manifest.has_existing_files()); - let mut entries = table + 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(delete_manifest); + .collect(); + entries.push(manifest); let mut manifest_list_writer = ManifestListWriter::v3( table @@ -556,8 +630,36 @@ mod tests { .add_manifests(entries.into_iter()) .unwrap(); manifest_list_writer.close().await.unwrap(); + } - delete_manifest_path + /// 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 retried commit must not reuse the manifest paths of the attempt From 5f81639c91e06e48e79cc47666fe6ccc906f1000 Mon Sep 17 00:00:00 2001 From: Jegor van Opdorp Date: Fri, 18 Sep 2026 18:01:53 +0200 Subject: [PATCH 13/23] Run the compaction and partial rewrites against a V2 table Every rewrite test used the V3 fixture, so nothing covered the V2 manifest and entry layout the action also writes. Both cases now run on V2 as well, through a V2 catalog fixture alongside the V3 one. --- crates/iceberg/src/transaction/mod.rs | 26 ++++++++++++++++-- crates/iceberg/src/transaction/rewrite.rs | 32 +++++++++++++++++++---- 2 files changed, 51 insertions(+), 7 deletions(-) diff --git a/crates/iceberg/src/transaction/mod.rs b/crates/iceberg/src/transaction/mod.rs index 3fe1b161e7..12242cb364 100644 --- a/crates/iceberg/src/transaction/mod.rs +++ b/crates/iceberg/src/transaction/mod.rs @@ -359,7 +359,29 @@ mod tests { .unwrap() } + 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(); @@ -372,7 +394,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); @@ -383,7 +405,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 index 9d264a4aa0..6a509f075f 100644 --- a/crates/iceberg/src/transaction/rewrite.rs +++ b/crates/iceberg/src/transaction/rewrite.rs @@ -171,8 +171,10 @@ mod tests { ManifestStatus, ManifestWriterBuilder, Operation, SnapshotRef, Struct, }; use crate::table::Table; + use crate::Catalog; use crate::transaction::tests::{ - append_files, make_data_file, make_v3_minimal_table_in_catalog, + append_files, make_data_file, make_v2_minimal_table_in_catalog, + make_v3_minimal_table_in_catalog, }; use crate::transaction::{ApplyTransactionAction, Transaction}; use crate::{ErrorKind, TableUpdate}; @@ -199,12 +201,22 @@ mod tests { 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_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; + let table = append_files(catalog, &table, vec![f1.clone(), f2.clone(), f3.clone()]).await; // Verify pre-compaction state. let summary = &table @@ -226,7 +238,7 @@ mod tests { .delete_file(f3) .add_file(merged); let tx = action.apply(tx).unwrap(); - let table = tx.commit(&catalog).await.unwrap(); + let table = tx.commit(catalog).await.unwrap(); // Verify post-compaction snapshot. let snapshot = table.metadata().current_snapshot().unwrap(); @@ -308,11 +320,21 @@ mod tests { 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_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 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. @@ -324,7 +346,7 @@ mod tests { .delete_file(f2) .add_file(merged); let tx = action.apply(tx).unwrap(); - let table = tx.commit(&catalog).await.unwrap(); + let table = tx.commit(catalog).await.unwrap(); let snapshot = table.metadata().current_snapshot().unwrap(); assert_eq!(snapshot.summary().operation, Operation::Replace); From 021c3b3407ca5f18f82ce49ab75a57f1ed94fe22 Mon Sep 17 00:00:00 2001 From: Jegor van Opdorp Date: Fri, 18 Sep 2026 18:11:13 +0200 Subject: [PATCH 14/23] Format the new code and record the rewrite action in the public API `cargo fmt` over the two new modules, and `public-api.txt` regenerated for `Transaction::rewrite_files`. The clippy warning on the unnecessary parentheses around the manifest counter went with the counter itself. --- crates/iceberg/public-api.txt | 1 + crates/iceberg/src/transaction/rewrite.rs | 15 ++++++++------- 2 files changed, 9 insertions(+), 7 deletions(-) diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index 3d9abf2275..960672689e 100644 --- a/crates/iceberg/public-api.txt +++ b/crates/iceberg/public-api.txt @@ -3235,6 +3235,7 @@ pub fn iceberg::transaction::Transaction::expire_snapshots(&self) -> iceberg::tr pub fn iceberg::transaction::Transaction::fast_append(&self) -> iceberg::transaction::append::FastAppendAction pub fn iceberg::transaction::Transaction::new(table: &iceberg::table::Table) -> Self pub fn iceberg::transaction::Transaction::replace_sort_order(&self) -> iceberg::transaction::sort_order::ReplaceSortOrderAction +pub fn iceberg::transaction::Transaction::rewrite_files(&self) -> iceberg::transaction::rewrite::RewriteFilesAction pub fn iceberg::transaction::Transaction::update_location(&self) -> iceberg::transaction::update_location::UpdateLocationAction pub fn iceberg::transaction::Transaction::update_schema(&self) -> iceberg::transaction::update_schema::UpdateSchemaAction pub fn iceberg::transaction::Transaction::update_statistics(&self) -> iceberg::transaction::update_statistics::UpdateStatisticsAction diff --git a/crates/iceberg/src/transaction/rewrite.rs b/crates/iceberg/src/transaction/rewrite.rs index 6a509f075f..5c9c7104e6 100644 --- a/crates/iceberg/src/transaction/rewrite.rs +++ b/crates/iceberg/src/transaction/rewrite.rs @@ -120,9 +120,10 @@ impl RewriteFilesAction { /// 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<()> { - let (Some(starting_snapshot_id), Some(current_snapshot)) = - (self.starting_snapshot_id, table.metadata().current_snapshot()) - else { + let (Some(starting_snapshot_id), Some(current_snapshot)) = ( + self.starting_snapshot_id, + table.metadata().current_snapshot(), + ) else { return Ok(()); }; let starting_sequence_number = table @@ -171,13 +172,12 @@ mod tests { ManifestStatus, ManifestWriterBuilder, Operation, SnapshotRef, Struct, }; use crate::table::Table; - use crate::Catalog; use crate::transaction::tests::{ append_files, make_data_file, make_v2_minimal_table_in_catalog, make_v3_minimal_table_in_catalog, }; use crate::transaction::{ApplyTransactionAction, Transaction}; - use crate::{ErrorKind, TableUpdate}; + use crate::{Catalog, ErrorKind, TableUpdate}; /// Read back the manifest entry for `path` from the manifests of `snapshot`. async fn find_entry( @@ -678,8 +678,9 @@ mod tests { }; assert_eq!(err.kind(), ErrorKind::DataInvalid); assert!( - err.to_string() - .contains(&format!("delete manifest {delete_manifest_path} was added after")), + err.to_string().contains(&format!( + "delete manifest {delete_manifest_path} was added after" + )), "{err}" ); } From 7d630619eca4f9f47547cc3c01a49792f581ecbb Mon Sep 17 00:00:00 2001 From: hotcache Date: Mon, 28 Sep 2026 12:28:51 +0000 Subject: [PATCH 15/23] test(transaction): run the rewrite tests against a V1 table V1 manifest entries carry no sequence numbers, and `ManifestEntryV1` deserialization fills both in as 0. Nothing in the rewrite path treats that as missing, so a V1 table compacts like V2 and V3 and the surviving entries keep Existing status, their snapshot id and their zero sequence numbers. Cover it so it stays that way. --- crates/iceberg/src/transaction/mod.rs | 9 +++++++++ crates/iceberg/src/transaction/rewrite.rs | 18 ++++++++++++++++-- 2 files changed, 25 insertions(+), 2 deletions(-) diff --git a/crates/iceberg/src/transaction/mod.rs b/crates/iceberg/src/transaction/mod.rs index d13ccd080d..da80cf4167 100644 --- a/crates/iceberg/src/transaction/mod.rs +++ b/crates/iceberg/src/transaction/mod.rs @@ -359,6 +359,15 @@ 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, diff --git a/crates/iceberg/src/transaction/rewrite.rs b/crates/iceberg/src/transaction/rewrite.rs index 5c9c7104e6..c76e0d0ceb 100644 --- a/crates/iceberg/src/transaction/rewrite.rs +++ b/crates/iceberg/src/transaction/rewrite.rs @@ -173,8 +173,8 @@ mod tests { }; use crate::table::Table; use crate::transaction::tests::{ - append_files, make_data_file, make_v2_minimal_table_in_catalog, - make_v3_minimal_table_in_catalog, + 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, TableUpdate}; @@ -204,6 +204,13 @@ mod tests { 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; @@ -323,6 +330,13 @@ mod tests { 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; From 318758c87ebd3eee5f218d39680c289b77f4d1fb Mon Sep 17 00:00:00 2001 From: hotcache Date: Mon, 28 Sep 2026 12:33:52 +0000 Subject: [PATCH 16/23] fix(transaction): name the file in the added-file content type error `validate_added_data_files` is shared by fast append and rewrite, but its content-type error said "for fast append", so a rewrite that was handed a delete file reported the wrong operation and not which file was wrong. Drop the operation from the message and name the file and its content type instead. --- crates/iceberg/src/transaction/rewrite.rs | 26 ++++++++++++++++++++++ crates/iceberg/src/transaction/snapshot.rs | 4 +++- 2 files changed, 29 insertions(+), 1 deletion(-) diff --git a/crates/iceberg/src/transaction/rewrite.rs b/crates/iceberg/src/transaction/rewrite.rs index c76e0d0ceb..436be2fa2b 100644 --- a/crates/iceberg/src/transaction/rewrite.rs +++ b/crates/iceberg/src/transaction/rewrite.rs @@ -774,6 +774,32 @@ mod tests { ); } + /// 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() { diff --git a/crates/iceberg/src/transaction/snapshot.rs b/crates/iceberg/src/transaction/snapshot.rs index aa5c779f43..2918e31b77 100644 --- a/crates/iceberg/src/transaction/snapshot.rs +++ b/crates/iceberg/src/transaction/snapshot.rs @@ -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. From 1af9f1d27fe1067cebf9ce4d606c6e9a7efa63c9 Mon Sep 17 00:00:00 2001 From: hotcache Date: Mon, 28 Sep 2026 12:39:59 +0000 Subject: [PATCH 17/23] fix(transaction): fail a rewrite whose starting snapshot is gone `validate_no_new_deletes` fell back to sequence number 0 when the starting snapshot was no longer in the table, which let a rewrite commit when the table held no delete manifest at all and otherwise blamed an unrelated delete manifest for the rejection. An expired starting snapshot means no delete can be ruled out, so say that instead. --- crates/iceberg/src/transaction/rewrite.rs | 60 +++++++++++++++++++++-- 1 file changed, 56 insertions(+), 4 deletions(-) diff --git a/crates/iceberg/src/transaction/rewrite.rs b/crates/iceberg/src/transaction/rewrite.rs index 436be2fa2b..c15ace74f3 100644 --- a/crates/iceberg/src/transaction/rewrite.rs +++ b/crates/iceberg/src/transaction/rewrite.rs @@ -126,10 +126,17 @@ impl RewriteFilesAction { ) else { return Ok(()); }; - let starting_sequence_number = table - .metadata() - .snapshot_by_id(starting_snapshot_id) - .map_or(0, |snapshot| snapshot.sequence_number()); + 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(Error::new( + ErrorKind::DataInvalid, + format!( + "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| { @@ -699,6 +706,51 @@ mod tests { ); } + /// 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] From 9c9f1edbb819c3f7e3a77350c46ccd7f39bcceae Mon Sep 17 00:00:00 2001 From: hotcache Date: Mon, 28 Sep 2026 12:42:59 +0000 Subject: [PATCH 18/23] fix(transaction): fail a rewrite planned before the table had a snapshot `validate_no_new_deletes` skipped the check whenever either the starting snapshot id or the current snapshot was absent. A missing starting snapshot id means the table held nothing when the rewrite was planned, so everything it holds at commit time, delete files included, arrived after that -- the case the check is least able to skip. Reject it, and keep returning early only when the table still has no snapshot, where the filter rejects the rewrite for its missing sources anyway. --- crates/iceberg/src/transaction/rewrite.rs | 42 ++++++++++++++++++++--- 1 file changed, 38 insertions(+), 4 deletions(-) diff --git a/crates/iceberg/src/transaction/rewrite.rs b/crates/iceberg/src/transaction/rewrite.rs index c15ace74f3..6ad74e40df 100644 --- a/crates/iceberg/src/transaction/rewrite.rs +++ b/crates/iceberg/src/transaction/rewrite.rs @@ -120,12 +120,19 @@ impl RewriteFilesAction { /// 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<()> { - let (Some(starting_snapshot_id), Some(current_snapshot)) = ( - self.starting_snapshot_id, - table.metadata().current_snapshot(), - ) else { + // 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(Error::new( + ErrorKind::DataInvalid, + "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. @@ -706,6 +713,33 @@ mod tests { ); } + /// 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] From 544a91f9e7242b4a2ca4d339a67403852ccdc2f3 Mon Sep 17 00:00:00 2001 From: hotcache Date: Mon, 28 Sep 2026 12:46:28 +0000 Subject: [PATCH 19/23] docs(transaction): describe when a rewrite refuses to commit The conflict check is table-wide and always on, so a compaction fails on a delete committed anywhere in the table while it ran, not only on the files it replaces. Say so on the action, along with the two other cases that cannot rule out a delete and what the check does not cover, so a caller can tell an expected rejection from a bug. Also stop the public docs linking to the private MergingSnapshotProducer, which rustdoc drops on the generated page. --- crates/iceberg/src/transaction/rewrite.rs | 23 ++++++++++++++++++++--- 1 file changed, 20 insertions(+), 3 deletions(-) diff --git a/crates/iceberg/src/transaction/rewrite.rs b/crates/iceberg/src/transaction/rewrite.rs index 6ad74e40df..1e9a5169d0 100644 --- a/crates/iceberg/src/transaction/rewrite.rs +++ b/crates/iceberg/src/transaction/rewrite.rs @@ -38,7 +38,7 @@ use crate::{Error, ErrorKind}; /// 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, +/// `MergingSnapshotProducer` to handle manifest filtering and creation, /// and commits a snapshot with [`Operation::Replace`]. /// /// # Example @@ -52,10 +52,27 @@ use crate::{Error, ErrorKind}; /// 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`]. 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 ID at which this rewrite started reading. Used to - /// detect conflicting deletes added after this point. + /// The snapshot the rewrite was planned against, if the table had one. + /// Deletes newer than it fail the commit. starting_snapshot_id: Option, } From ebf19464118702c9ddb71908ddce037afe0352d0 Mon Sep 17 00:00:00 2001 From: hotcache Date: Mon, 28 Sep 2026 13:01:45 +0000 Subject: [PATCH 20/23] refactor(transaction): build the rewrite errors with invalid_data! The macro from #2928 landed while this branch was open and snapshot.rs already uses it. Use it for the merging producer and the rewrite action too, so the whole transaction module reads the same way. --- crates/iceberg/src/transaction/merging.rs | 46 ++++++++--------------- crates/iceberg/src/transaction/rewrite.rs | 37 +++++++----------- 2 files changed, 29 insertions(+), 54 deletions(-) diff --git a/crates/iceberg/src/transaction/merging.rs b/crates/iceberg/src/transaction/merging.rs index 3b4363a409..d3b3bf0edc 100644 --- a/crates/iceberg/src/transaction/merging.rs +++ b/crates/iceberg/src/transaction/merging.rs @@ -27,7 +27,7 @@ use std::sync::atomic::{AtomicU64, Ordering}; use uuid::Uuid; -use crate::error::Result; +use crate::error::{Result, invalid_data}; use crate::io::OutputFile; use crate::spec::{ DataFile, DataFileFormat, FormatVersion, ManifestContentType, ManifestEntry, ManifestFile, @@ -37,7 +37,6 @@ use crate::spec::{ use crate::table::Table; use crate::transaction::ActionCommit; use crate::transaction::snapshot::SnapshotProducer; -use crate::{Error, ErrorKind}; /// Create a manifest writer that handles encryption when available. fn new_manifest_writer( @@ -142,12 +141,9 @@ impl ManifestFilterManager { .metadata() .partition_spec_by_id(manifest_file.partition_spec_id) .ok_or_else(|| { - Error::new( - ErrorKind::DataInvalid, - format!( - "Manifest references unknown partition spec {}", - manifest_file.partition_spec_id, - ), + invalid_data!( + "Manifest references unknown partition spec {}", + manifest_file.partition_spec_id, ) })? .clone(); @@ -167,21 +163,15 @@ impl ManifestFilterManager { } else if entry.is_alive() { // Surviving entry — re-emit as EXISTING with original ids preserved. let seq = entry.sequence_number().ok_or_else(|| { - Error::new( - ErrorKind::DataInvalid, - format!( - "Manifest entry for {} is missing sequence_number", - entry.file_path(), - ), + invalid_data!( + "Manifest entry for {} is missing sequence_number", + entry.file_path(), ) })?; let file_seq = entry.file_sequence_number.ok_or_else(|| { - Error::new( - ErrorKind::DataInvalid, - format!( - "Manifest entry for {} is missing file_sequence_number", - entry.file_path(), - ), + invalid_data!( + "Manifest entry for {} is missing file_sequence_number", + entry.file_path(), ) })?; let existing = ManifestEntry::builder() @@ -240,12 +230,9 @@ impl ManifestFilterManager { .map(String::as_str) .collect(); if !missing.is_empty() { - return Err(Error::new( - ErrorKind::DataInvalid, - format!( - "Failed to find the following files to delete in the current snapshot: {}", - missing.join(", "), - ), + return Err(invalid_data!( + "Failed to find the following files to delete in the current snapshot: {}", + missing.join(", "), )); } } @@ -321,11 +308,8 @@ impl MergingSnapshotProducer { }; let snapshot_sequence_number = table.metadata().next_sequence_number(); if sequence_number > snapshot_sequence_number { - return Err(Error::new( - ErrorKind::DataInvalid, - format!( - "Data sequence number {sequence_number} is greater than the snapshot's {snapshot_sequence_number}." - ), + return Err(invalid_data!( + "Data sequence number {sequence_number} is greater than the snapshot's {snapshot_sequence_number}." )); } Ok(()) diff --git a/crates/iceberg/src/transaction/rewrite.rs b/crates/iceberg/src/transaction/rewrite.rs index 1e9a5169d0..de31d723bc 100644 --- a/crates/iceberg/src/transaction/rewrite.rs +++ b/crates/iceberg/src/transaction/rewrite.rs @@ -28,12 +28,11 @@ use std::sync::Arc; use async_trait::async_trait; -use crate::error::Result; +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}; -use crate::{Error, ErrorKind}; /// A transaction action that rewrites (replaces) data files. /// @@ -60,7 +59,8 @@ use crate::{Error, ErrorKind}; /// 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`]. So do a +/// 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. @@ -115,15 +115,13 @@ impl RewriteFilesAction { fn validate(&self) -> Result<()> { if !self.producer.has_deleted_data_files() { - return Err(Error::new( - ErrorKind::DataInvalid, - "Rewrite files requires at least one file to delete", + return Err(invalid_data!( + "Rewrite files requires at least one file to delete" )); } if !self.producer.has_added_data_files() { - return Err(Error::new( - ErrorKind::DataInvalid, - "Rewrite files requires at least one file to add", + return Err(invalid_data!( + "Rewrite files requires at least one file to add" )); } Ok(()) @@ -145,19 +143,15 @@ impl RewriteFilesAction { // 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(Error::new( - ErrorKind::DataInvalid, - "Cannot rewrite files: the rewrite was planned against a table with no snapshot, and the table has been committed to since.", + 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(Error::new( - ErrorKind::DataInvalid, - format!( - "Cannot rewrite files: the starting snapshot {starting_snapshot_id} is no longer in the table, so deletes added since it cannot 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(); @@ -169,12 +163,9 @@ impl RewriteFilesAction { }); match conflict { - Some(entry) => Err(Error::new( - ErrorKind::DataInvalid, - format!( - "Cannot rewrite files: delete manifest {} was added after the starting snapshot {starting_snapshot_id}.", - entry.manifest_path, - ), + Some(entry) => Err(invalid_data!( + "Cannot rewrite files: delete manifest {} was added after the starting snapshot {starting_snapshot_id}.", + entry.manifest_path, )), None => Ok(()), } From 43a09a5027c833cb3dbf9f9f993a8709d14d0e6c Mon Sep 17 00:00:00 2001 From: Rexwell Minnis Date: Mon, 28 Sep 2026 11:24:04 -0500 Subject: [PATCH 21/23] test(transaction): cover rewrites on a spec-evolved table The filter resolves each manifest's own partition spec, but every test ran on the default spec. Add an identity(ts) -> day(ts) fixture on V2 and V3: a survivor stays Existing in a spec-0 manifest with its timestamptz value and partition summary intact, one rewrite filters a manifest of each spec, and retiring the last old-spec file drops the spec-0 manifest with the chained totals still exact. --- crates/iceberg/src/transaction/rewrite.rs | 281 +++++++++++++++++++++- 1 file changed, 278 insertions(+), 3 deletions(-) diff --git a/crates/iceberg/src/transaction/rewrite.rs b/crates/iceberg/src/transaction/rewrite.rs index de31d723bc..508b432106 100644 --- a/crates/iceberg/src/transaction/rewrite.rs +++ b/crates/iceberg/src/transaction/rewrite.rs @@ -183,6 +183,7 @@ impl TransactionAction for RewriteFilesAction { #[cfg(test)] mod tests { + use std::collections::HashMap; use std::sync::Arc; use uuid::Uuid; @@ -190,8 +191,10 @@ mod tests { use super::*; use crate::memory::tests::new_memory_catalog; use crate::spec::{ - DataContentType, Literal, ManifestEntryRef, ManifestFile, ManifestListWriter, - ManifestStatus, ManifestWriterBuilder, Operation, SnapshotRef, Struct, + DataContentType, DataFileBuilder, DataFileFormat, Datum, FormatVersion, Literal, + ManifestEntryRef, ManifestFile, ManifestListWriter, ManifestStatus, ManifestWriterBuilder, + NestedField, Operation, PrimitiveType, Schema, SnapshotRef, Struct, Transform, Type, + UnboundPartitionSpec, }; use crate::table::Table; use crate::transaction::tests::{ @@ -199,7 +202,7 @@ mod tests { make_v2_minimal_table_in_catalog, make_v3_minimal_table_in_catalog, }; use crate::transaction::{ApplyTransactionAction, Transaction}; - use crate::{Catalog, ErrorKind, TableUpdate}; + use crate::{Catalog, ErrorKind, NamespaceIdent, TableCommit, TableCreation, TableUpdate}; /// Read back the manifest entry for `path` from the manifests of `snapshot`. async fn find_entry( @@ -938,4 +941,276 @@ mod tests { .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(2, "ts", Transform::Identity) + .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(2, "ts_day", Transform::Day) + .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(); + let summary = &old_spec_manifest.partitions.as_ref().unwrap()[0]; + let evening = Datum::timestamptz_micros(TS_EVENING).to_bytes().unwrap(); + assert_eq!(summary.lower_bound.as_ref(), Some(&evening)); + 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() + .all(|m| m.partition_spec_id == 1), + "{format_version}: a spec-0 manifest outlived its last file" + ); + let totals = &snapshot.summary().additional_properties; + assert_eq!(totals.get("total-data-files").unwrap(), "2"); + assert_eq!(totals.get("total-records").unwrap(), "30"); + } + } } From 89786a86e9a25ea60ba2c3b78acc247d55e6e440 Mon Sep 17 00:00:00 2001 From: Rexwell Minnis Date: Mon, 28 Sep 2026 11:24:34 -0500 Subject: [PATCH 22/23] fix(transaction): keep partition metrics in the rewrite summary build_summary merged the filter's removal collector into a fresh one; merge drops partition metrics unless both sides trust them and neither did, so a rewrite never reported changed-partition-count or the partitions.* summaries. Add the new files to the filter's collector instead, and honour write.summary.partition-limit as fast append does. --- crates/iceberg/src/transaction/merging.rs | 15 ++++-- crates/iceberg/src/transaction/rewrite.rs | 61 +++++++++++++++++++++++ 2 files changed, 72 insertions(+), 4 deletions(-) diff --git a/crates/iceberg/src/transaction/merging.rs b/crates/iceberg/src/transaction/merging.rs index d3b3bf0edc..3b619717d0 100644 --- a/crates/iceberg/src/transaction/merging.rs +++ b/crates/iceberg/src/transaction/merging.rs @@ -32,7 +32,7 @@ use crate::io::OutputFile; use crate::spec::{ DataFile, DataFileFormat, FormatVersion, ManifestContentType, ManifestEntry, ManifestFile, ManifestStatus, ManifestWriter, ManifestWriterBuilder, Operation, PartitionSpec, SchemaRef, - SnapshotSummaryCollector, Summary, update_snapshot_summaries, + SnapshotSummaryCollector, Summary, TableProperties, update_snapshot_summaries, }; use crate::table::Table; use crate::transaction::ActionCommit; @@ -431,12 +431,19 @@ impl MergingSnapshotProducer { let schema = table_metadata.current_schema().clone(); let partition_spec = table_metadata.default_partition_spec().clone(); - let mut collector = SnapshotSummaryCollector::default(); + // 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()); } - // Merge removal metrics from the filter manager. - collector.merge(removed_collector); let summary = Summary { operation: self.operation.clone(), diff --git a/crates/iceberg/src/transaction/rewrite.rs b/crates/iceberg/src/transaction/rewrite.rs index 508b432106..5f5f821d64 100644 --- a/crates/iceberg/src/transaction/rewrite.rs +++ b/crates/iceberg/src/transaction/rewrite.rs @@ -1213,4 +1213,65 @@ mod tests { assert_eq!(totals.get("total-records").unwrap(), "30"); } } + + /// 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:?}" + ); + } } From 8a4f780596f296068a3bfe90afd564a4d3004df3 Mon Sep 17 00:00:00 2001 From: hotcache Date: Tue, 29 Sep 2026 14:29:43 +0000 Subject: [PATCH 23/23] fix(transaction): record removed files as DELETED manifest entries The filter dropped the entries of the files it removed, so nothing in the metadata said which snapshot removed them. Java's expiry walks exactly those entries -- IncrementalFileCleanup finds a data file to delete only through a DELETED entry whose snapshot has expired -- so a table compacted by this action kept every replaced file in storage for good, and a changelog reader saw no removals. Read back with Trino, the manifests reported deleted_data_files_count = 0 while the snapshot summary claimed deleted-data-files = 2. Write each removed entry with add_delete_entry, which stamps the removing snapshot and keeps the file's own sequence numbers. A manifest the commit empties is now written rather than omitted, holding only those entries; the next merging commit drops it, as it already did for manifests left all-deleted by an earlier snapshot. Reported by @rexminnis against Apache Polaris with Trino and PyIceberg. Two tests change with the contract. The emptied manifest is no longer absent, so they assert it has no live entry and that a later rewrite drops it -- an append carries it along, here as in Java, because only a merging commit rebuilds the manifest list. A filtered manifest's partition summary now spans its deleted entries too, matching Java, where every entry updates it. --- crates/iceberg/src/spec/manifest/writer.rs | 4 -- crates/iceberg/src/transaction/merging.rs | 16 ++++- crates/iceberg/src/transaction/rewrite.rs | 82 +++++++++++++++++++--- 3 files changed, 85 insertions(+), 17 deletions(-) diff --git a/crates/iceberg/src/spec/manifest/writer.rs b/crates/iceberg/src/spec/manifest/writer.rs index ba0d02b43f..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; diff --git a/crates/iceberg/src/transaction/merging.rs b/crates/iceberg/src/transaction/merging.rs index 3b619717d0..d66ae4df21 100644 --- a/crates/iceberg/src/transaction/merging.rs +++ b/crates/iceberg/src/transaction/merging.rs @@ -149,8 +149,11 @@ impl ManifestFilterManager { .clone(); let schema = table.metadata().current_schema().clone(); - // Rewrite: keep surviving entries as EXISTING, drop deleted ones. + // 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. @@ -160,6 +163,7 @@ impl ManifestFilterManager { 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(|| { @@ -190,8 +194,8 @@ impl ManifestFilterManager { // Already-deleted entries (status == Deleted) are dropped. } - if surviving_entries.is_empty() { - // Manifest is now empty — omit entirely. + if surviving_entries.is_empty() && removed_entries.is_empty() { + // Nothing left to say about this manifest — omit entirely. continue; } @@ -217,6 +221,12 @@ impl ManifestFilterManager { 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); } diff --git a/crates/iceberg/src/transaction/rewrite.rs b/crates/iceberg/src/transaction/rewrite.rs index 5f5f821d64..5ed8e36bc9 100644 --- a/crates/iceberg/src/transaction/rewrite.rs +++ b/crates/iceberg/src/transaction/rewrite.rs @@ -524,20 +524,52 @@ mod tests { 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 should be omitted. + // 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).add_file(merged); + 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(); - // Verify: only 1 manifest (the new one), not 2. 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(), - 1, - "empty manifest should be omitted, not kept" + 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" ); } @@ -1119,9 +1151,12 @@ mod tests { .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(&evening)); + assert_eq!(summary.lower_bound.as_ref(), Some(&morning)); assert_eq!(summary.upper_bound.as_ref(), Some(&evening)); let totals = &snapshot.summary().additional_properties; @@ -1205,12 +1240,39 @@ mod tests { manifest_list .entries() .iter() - .all(|m| m.partition_spec_id == 1), - "{format_version}: a spec-0 manifest outlived its last file" + .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" + ); } }