From 7fdc8fdef462efc3e04bda46e184113145bae9ed Mon Sep 17 00:00:00 2001 From: hotcache Date: Tue, 25 Aug 2026 11:35:09 +0900 Subject: [PATCH 01/17] 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 83ac7a5df89c9d640984eb432542595fdaacec08 Mon Sep 17 00:00:00 2001 From: Jegor van Opdorp Date: Fri, 18 Sep 2026 17:25:46 +0200 Subject: [PATCH 02/17] 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 6fc7bf1b912c74dd5a09542d80ca8147d228967f Mon Sep 17 00:00:00 2001 From: Jegor van Opdorp Date: Fri, 18 Sep 2026 17:28:56 +0200 Subject: [PATCH 03/17] 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 9d40b4cacaf0668440e3fb37d7c309bc378b05ef Mon Sep 17 00:00:00 2001 From: Jegor van Opdorp Date: Fri, 18 Sep 2026 17:33:23 +0200 Subject: [PATCH 04/17] 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 d466a48f370b6bd9bc8e0d44d13a38f496b17a94 Mon Sep 17 00:00:00 2001 From: Jegor van Opdorp Date: Fri, 18 Sep 2026 17:42:32 +0200 Subject: [PATCH 05/17] 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 ee3fbddcec463fd0e9d84024ff70ec15dfd182eb Mon Sep 17 00:00:00 2001 From: Jegor van Opdorp Date: Fri, 18 Sep 2026 17:54:11 +0200 Subject: [PATCH 06/17] 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 4f26afcd7fc59a0831e36a0a51f5e81ba7f11897 Mon Sep 17 00:00:00 2001 From: Jegor van Opdorp Date: Fri, 18 Sep 2026 17:59:07 +0200 Subject: [PATCH 07/17] 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 6fcdb66179b8322b618fbac9ddc3539f5c4bfc64 Mon Sep 17 00:00:00 2001 From: Jegor van Opdorp Date: Fri, 18 Sep 2026 18:01:53 +0200 Subject: [PATCH 08/17] 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 8d461af10464a415177d118620ccc49e137d542d Mon Sep 17 00:00:00 2001 From: Jegor van Opdorp Date: Fri, 18 Sep 2026 18:11:13 +0200 Subject: [PATCH 09/17] 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 69aa1b3fd84b9506a877324b0ef85d6067154f22 Mon Sep 17 00:00:00 2001 From: abel Date: Mon, 28 Sep 2026 12:28:51 +0000 Subject: [PATCH 10/17] 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 b8c4c70d6327e882cf894479d3b840c7019f1e57 Mon Sep 17 00:00:00 2001 From: abel Date: Mon, 28 Sep 2026 12:33:52 +0000 Subject: [PATCH 11/17] 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 d22436b2ffe554552308cda4ecf18da37e8f7bb4 Mon Sep 17 00:00:00 2001 From: abel Date: Mon, 28 Sep 2026 12:39:59 +0000 Subject: [PATCH 12/17] 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 f5818cce3ed24083bff92c5872ef120d98acd07b Mon Sep 17 00:00:00 2001 From: abel Date: Mon, 28 Sep 2026 12:42:59 +0000 Subject: [PATCH 13/17] 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 d4dbf46e024d43ecc9877b47f1ac02525798b807 Mon Sep 17 00:00:00 2001 From: abel Date: Mon, 28 Sep 2026 12:46:28 +0000 Subject: [PATCH 14/17] 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 e7db3c84b20b774bf5b081ffd82ddd0436054309 Mon Sep 17 00:00:00 2001 From: abel Date: Mon, 28 Sep 2026 13:01:45 +0000 Subject: [PATCH 15/17] 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 e147c40a99129c2cdc89c753e03a2fa5e55bfbe0 Mon Sep 17 00:00:00 2001 From: Rexwell Minnis Date: Mon, 28 Sep 2026 11:24:04 -0500 Subject: [PATCH 16/17] 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 4bc7e4fe0578ca20999952480c4e5dce6936a806 Mon Sep 17 00:00:00 2001 From: Rexwell Minnis Date: Mon, 28 Sep 2026 11:24:34 -0500 Subject: [PATCH 17/17] 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:?}" + ); + } }