From ab82c45aae4a5f5588d3109e534fa28f76575fcf Mon Sep 17 00:00:00 2001 From: Steven Isaac Date: Wed, 8 Jul 2026 06:41:20 -0400 Subject: [PATCH 1/3] feat(transaction): add RowDeltaAction to commit data and delete files in one snapshot Adds the commit-side primitive for merge-on-read row-level changes, equivalent to Table.newRowDelta() in iceberg-java: a transaction action that commits data files and delete files (position or equality deletes) together in a single overwrite snapshot. Both receive the new snapshot's sequence number, so the delete files apply only to strictly older data, which gives upsert semantics when equality deletes are paired with the rows' new values. SnapshotProducer gains added_delete_files and writes a deletes-content manifest alongside the data manifest; the snapshot summary now counts delete files via the existing SnapshotSummaryCollector support. Includes an end-to-end merge-on-read round trip test: parquet data plus an equality-delete upsert committed via row_delta and read back through the scan path. Co-Authored-By: Claude Fable 5 --- crates/iceberg/src/transaction/append.rs | 1 + crates/iceberg/src/transaction/mod.rs | 8 + crates/iceberg/src/transaction/row_delta.rs | 566 ++++++++++++++++++++ crates/iceberg/src/transaction/snapshot.rs | 80 ++- 4 files changed, 654 insertions(+), 1 deletion(-) create mode 100644 crates/iceberg/src/transaction/row_delta.rs diff --git a/crates/iceberg/src/transaction/append.rs b/crates/iceberg/src/transaction/append.rs index 9d94e18ede..e234642acd 100644 --- a/crates/iceberg/src/transaction/append.rs +++ b/crates/iceberg/src/transaction/append.rs @@ -81,6 +81,7 @@ impl TransactionAction for FastAppendAction { self.commit_uuid.unwrap_or_else(Uuid::now_v7), self.snapshot_properties.clone(), self.added_data_files.clone(), + vec![], ); // validate added files diff --git a/crates/iceberg/src/transaction/mod.rs b/crates/iceberg/src/transaction/mod.rs index d78f41cd42..fa11d58077 100644 --- a/crates/iceberg/src/transaction/mod.rs +++ b/crates/iceberg/src/transaction/mod.rs @@ -55,6 +55,7 @@ mod action; pub use action::*; mod append; mod expire_snapshots; +mod row_delta; mod snapshot; mod sort_order; mod update_location; @@ -75,6 +76,7 @@ use crate::table::Table; use crate::transaction::action::BoxedTransactionAction; use crate::transaction::append::FastAppendAction; use crate::transaction::expire_snapshots::ExpireSnapshotsAction; +use crate::transaction::row_delta::RowDeltaAction; use crate::transaction::sort_order::ReplaceSortOrderAction; use crate::transaction::update_location::UpdateLocationAction; use crate::transaction::update_properties::UpdatePropertiesAction; @@ -151,6 +153,12 @@ impl Transaction { FastAppendAction::new() } + /// Creates a row delta action, which commits data files and delete files + /// (position or equality deletes) together in one `overwrite` snapshot. + pub fn row_delta(&self) -> RowDeltaAction { + RowDeltaAction::new() + } + /// Creates replace sort order action. pub fn replace_sort_order(&self) -> ReplaceSortOrderAction { ReplaceSortOrderAction::new() diff --git a/crates/iceberg/src/transaction/row_delta.rs b/crates/iceberg/src/transaction/row_delta.rs new file mode 100644 index 0000000000..950a7e118c --- /dev/null +++ b/crates/iceberg/src/transaction/row_delta.rs @@ -0,0 +1,566 @@ +// 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. + +use std::collections::HashMap; +use std::sync::Arc; + +use async_trait::async_trait; +use uuid::Uuid; + +use crate::error::Result; +use crate::spec::{DataFile, ManifestEntry, ManifestFile, Operation}; +use crate::table::Table; +use crate::transaction::snapshot::{ + DefaultManifestProcess, SnapshotProduceOperation, SnapshotProducer, +}; +use crate::transaction::{ActionCommit, TransactionAction}; + +/// RowDeltaAction is a transaction action that commits data files and delete +/// files (position or equality deletes) together in a single `overwrite` +/// snapshot. +/// +/// This is the write-side primitive for merge-on-read row-level changes, +/// equivalent to `Table.newRowDelta()` in iceberg-java: both the added data +/// files and the added delete files receive the new snapshot's sequence +/// number, so the delete files apply to data with strictly smaller sequence +/// numbers only. Data files committed in the same row delta are therefore +/// not affected by its own delete files, which gives upsert semantics when a +/// row delta pairs equality deletes with the rows' new values. +pub struct RowDeltaAction { + check_duplicate: bool, + // below are properties used to create SnapshotProducer when commit + commit_uuid: Option, + snapshot_properties: HashMap, + added_data_files: Vec, + added_delete_files: Vec, +} + +impl RowDeltaAction { + pub(crate) fn new() -> Self { + Self { + check_duplicate: true, + commit_uuid: None, + snapshot_properties: HashMap::default(), + added_data_files: vec![], + added_delete_files: vec![], + } + } + + /// Set whether to check duplicate files + pub fn with_check_duplicate(mut self, v: bool) -> Self { + self.check_duplicate = v; + self + } + + /// Add data files to the snapshot. + pub fn add_data_files(mut self, data_files: impl IntoIterator) -> Self { + self.added_data_files.extend(data_files); + self + } + + /// Add delete files (position or equality deletes) to the snapshot. + pub fn add_delete_files(mut self, delete_files: impl IntoIterator) -> Self { + self.added_delete_files.extend(delete_files); + self + } + + /// Set commit UUID for the snapshot. + pub fn set_commit_uuid(mut self, commit_uuid: Uuid) -> Self { + self.commit_uuid = Some(commit_uuid); + self + } + + /// Set snapshot summary properties. + pub fn set_snapshot_properties(mut self, snapshot_properties: HashMap) -> Self { + self.snapshot_properties = snapshot_properties; + self + } +} + +#[async_trait] +impl TransactionAction for RowDeltaAction { + async fn commit(self: Arc, table: &Table) -> Result { + let snapshot_producer = SnapshotProducer::new( + table, + self.commit_uuid.unwrap_or_else(Uuid::now_v7), + self.snapshot_properties.clone(), + self.added_data_files.clone(), + self.added_delete_files.clone(), + ); + + // validate added files + snapshot_producer.validate_added_data_files()?; + snapshot_producer.validate_added_delete_files()?; + + // Checks duplicate files + if self.check_duplicate { + snapshot_producer.validate_duplicate_files().await?; + } + + snapshot_producer + .commit(RowDeltaOperation, DefaultManifestProcess) + .await + } +} + +struct RowDeltaOperation; + +impl SnapshotProduceOperation for RowDeltaOperation { + fn operation(&self) -> Operation { + // Matches iceberg-java's BaseRowDelta: a row delta commits both data + // and delete files, so the snapshot operation is `overwrite`. + Operation::Overwrite + } + + async fn delete_entries( + &self, + _snapshot_produce: &SnapshotProducer<'_>, + ) -> Result> { + Ok(vec![]) + } + + async fn existing_manifest( + &self, + snapshot_produce: &SnapshotProducer<'_>, + ) -> Result> { + let Some(snapshot) = snapshot_produce.table.metadata().current_snapshot() else { + return Ok(vec![]); + }; + + let manifest_list = snapshot_produce + .table + .manifest_list_reader(snapshot) + .load() + .await?; + + Ok(manifest_list + .entries() + .iter() + .filter(|entry| { + // Keep delete-only manifests too: they record which files were removed and + // must persist across snapshots until `expire_snapshots` cleans them up. + // Dropping them lets the removed files reappear as live data (see #2148). + entry.has_added_files() || entry.has_existing_files() || entry.has_deleted_files() + }) + .cloned() + .collect()) + } +} + +#[cfg(test)] +mod tests { + use std::sync::Arc; + + use crate::spec::{ + DataContentType, DataFileBuilder, DataFileFormat, Literal, MAIN_BRANCH, + ManifestContentType, Operation, SnapshotRef, Struct, + }; + use crate::transaction::tests::make_v2_minimal_table; + use crate::transaction::{Transaction, TransactionAction}; + use crate::{TableRequirement, TableUpdate}; + + fn test_data_file(table: &crate::table::Table) -> crate::spec::DataFile { + DataFileBuilder::default() + .content(DataContentType::Data) + .file_path("test/1.parquet".to_string()) + .file_format(DataFileFormat::Parquet) + .file_size_in_bytes(100) + .record_count(1) + .partition_spec_id(table.metadata().default_partition_spec_id()) + .partition(Struct::from_iter([Some(Literal::long(300))])) + .build() + .unwrap() + } + + fn test_equality_delete_file(table: &crate::table::Table) -> crate::spec::DataFile { + DataFileBuilder::default() + .content(DataContentType::EqualityDeletes) + .file_path("test/1-deletes.parquet".to_string()) + .file_format(DataFileFormat::Parquet) + .file_size_in_bytes(50) + .record_count(1) + .equality_ids(Some(vec![1])) + .partition_spec_id(table.metadata().default_partition_spec_id()) + .partition(Struct::from_iter([Some(Literal::long(300))])) + .build() + .unwrap() + } + + #[tokio::test] + async fn test_empty_row_delta_action() { + let table = make_v2_minimal_table(); + let tx = Transaction::new(&table); + let action = tx.row_delta(); + assert!(Arc::new(action).commit(&table).await.is_err()); + } + + #[tokio::test] + async fn test_row_delta_rejects_data_file_in_delete_list() { + let table = make_v2_minimal_table(); + let tx = Transaction::new(&table); + let action = tx + .row_delta() + .add_delete_files(vec![test_data_file(&table)]); + assert!(Arc::new(action).commit(&table).await.is_err()); + } + + #[tokio::test] + async fn test_row_delta_rejects_delete_file_in_data_list() { + let table = make_v2_minimal_table(); + let tx = Transaction::new(&table); + let action = tx + .row_delta() + .add_data_files(vec![test_equality_delete_file(&table)]); + assert!(Arc::new(action).commit(&table).await.is_err()); + } + + #[tokio::test] + async fn test_row_delta() { + let table = make_v2_minimal_table(); + let tx = Transaction::new(&table); + + let data_file = test_data_file(&table); + let delete_file = test_equality_delete_file(&table); + + let action = tx + .row_delta() + .add_data_files(vec![data_file.clone()]) + .add_delete_files(vec![delete_file.clone()]); + let mut action_commit = Arc::new(action).commit(&table).await.unwrap(); + let updates = action_commit.take_updates(); + let requirements = action_commit.take_requirements(); + + // check updates and requirements + assert!( + matches!((&updates[0],&updates[1]), (TableUpdate::AddSnapshot { snapshot },TableUpdate::SetSnapshotRef { reference,ref_name }) if snapshot.snapshot_id() == reference.snapshot_id && ref_name == MAIN_BRANCH) + ); + assert_eq!( + vec![ + TableRequirement::UuidMatch { + uuid: table.metadata().uuid() + }, + TableRequirement::RefSnapshotIdMatch { + r#ref: MAIN_BRANCH.to_string(), + snapshot_id: table.metadata().current_snapshot_id + } + ], + requirements + ); + + // a row delta commits an `overwrite` snapshot + let new_snapshot: SnapshotRef = if let TableUpdate::AddSnapshot { snapshot } = &updates[0] { + SnapshotRef::new(snapshot.clone()) + } else { + unreachable!() + }; + assert_eq!(new_snapshot.summary().operation, Operation::Overwrite); + assert_eq!( + new_snapshot + .summary() + .additional_properties + .get("added-delete-files") + .map(String::as_str), + Some("1") + ); + + // check manifest list: one data manifest and one delete manifest + let manifest_list = table + .manifest_list_reader(&new_snapshot) + .load() + .await + .unwrap(); + assert_eq!(2, manifest_list.entries().len()); + + let data_manifest_file = manifest_list + .entries() + .iter() + .find(|m| m.content == ManifestContentType::Data) + .expect("expected a data manifest"); + let delete_manifest_file = manifest_list + .entries() + .iter() + .find(|m| m.content == ManifestContentType::Deletes) + .expect("expected a delete manifest"); + + // both manifests carry the new snapshot's sequence number, so the + // delete files apply to strictly older data only + assert_eq!( + data_manifest_file.sequence_number, + new_snapshot.sequence_number() + ); + assert_eq!( + delete_manifest_file.sequence_number, + new_snapshot.sequence_number() + ); + + // check the delete manifest contents + let delete_manifest = delete_manifest_file + .load_manifest(table.file_io()) + .await + .unwrap(); + assert_eq!(1, delete_manifest.entries().len()); + assert_eq!( + new_snapshot.sequence_number(), + delete_manifest.entries()[0] + .sequence_number() + .expect("Inherit sequence number by load manifest") + ); + assert_eq!(delete_file, *delete_manifest.entries()[0].data_file()); + + // check the data manifest contents + let data_manifest = data_manifest_file + .load_manifest(table.file_io()) + .await + .unwrap(); + assert_eq!(1, data_manifest.entries().len()); + assert_eq!(data_file, *data_manifest.entries()[0].data_file()); + } + + #[tokio::test] + async fn test_row_delta_deletes_only() { + // A row delta with only delete files is valid (pure row-level delete). + let table = make_v2_minimal_table(); + let tx = Transaction::new(&table); + + let delete_file = test_equality_delete_file(&table); + let action = tx.row_delta().add_delete_files(vec![delete_file]); + let mut action_commit = Arc::new(action).commit(&table).await.unwrap(); + let updates = action_commit.take_updates(); + + let new_snapshot: SnapshotRef = if let TableUpdate::AddSnapshot { snapshot } = &updates[0] { + SnapshotRef::new(snapshot.clone()) + } else { + unreachable!() + }; + let manifest_list = table + .manifest_list_reader(&new_snapshot) + .load() + .await + .unwrap(); + assert_eq!(1, manifest_list.entries().len()); + assert_eq!( + manifest_list.entries()[0].content, + ManifestContentType::Deletes + ); + } + + #[tokio::test] + async fn test_row_delta_write_and_read_back() { + // End-to-end merge-on-read round trip: append rows, then commit a row + // delta pairing an equality delete with a replacement row (an upsert), + // and read the table back through the scan path. + use arrow_array::{Int64Array, RecordBatch, StringArray}; + use futures::TryStreamExt; + use parquet::file::properties::WriterProperties; + + use crate::arrow::schema_to_arrow_schema; + use crate::memory::tests::new_memory_catalog; + use crate::spec::{NestedField, PrimitiveType, Schema, Type}; + use crate::transaction::ApplyTransactionAction; + use crate::writer::base_writer::data_file_writer::DataFileWriterBuilder; + use crate::writer::base_writer::equality_delete_writer::{ + EqualityDeleteFileWriterBuilder, EqualityDeleteWriterConfig, + }; + use crate::writer::file_writer::ParquetWriterBuilder; + use crate::writer::file_writer::location_generator::{ + DefaultFileNameGenerator, DefaultLocationGenerator, + }; + use crate::writer::file_writer::rolling_writer::RollingFileWriterBuilder; + use crate::writer::{IcebergWriter, IcebergWriterBuilder}; + use crate::{Catalog, TableCreation, TableIdent}; + + let catalog = new_memory_catalog().await; + + let schema = Schema::builder() + .with_schema_id(0) + .with_fields(vec![ + NestedField::required(1, "id", Type::Primitive(PrimitiveType::Long)).into(), + NestedField::optional(2, "data", Type::Primitive(PrimitiveType::String)).into(), + ]) + .build() + .unwrap(); + + let table_ident = TableIdent::from_strs(["nsrd", "rowdelta"]).unwrap(); + catalog + .create_namespace(table_ident.namespace(), std::collections::HashMap::new()) + .await + .unwrap(); + let table = catalog + .create_table( + table_ident.namespace(), + TableCreation::builder() + .name(table_ident.name().to_string()) + .schema(schema.clone()) + .build(), + ) + .await + .unwrap(); + + let arrow_schema = + std::sync::Arc::new(schema_to_arrow_schema(table.metadata().current_schema()).unwrap()); + let location_gen = DefaultLocationGenerator::new(table.metadata()).unwrap(); + + let new_data_writer = |prefix: &str| { + let file_name_gen = DefaultFileNameGenerator::new( + prefix.to_string(), + None, + crate::spec::DataFileFormat::Parquet, + ); + let parquet_builder = ParquetWriterBuilder::new( + WriterProperties::builder().build(), + table.metadata().current_schema().clone(), + ); + DataFileWriterBuilder::new(RollingFileWriterBuilder::new_with_default_file_size( + parquet_builder, + table.file_io().clone(), + location_gen.clone(), + file_name_gen, + )) + }; + + // Commit 1: fast append rows (1, "a"), (2, "b"), (3, "c"). + let batch = RecordBatch::try_new(arrow_schema.clone(), vec![ + std::sync::Arc::new(Int64Array::from(vec![1, 2, 3])), + std::sync::Arc::new(StringArray::from(vec![Some("a"), Some("b"), Some("c")])), + ]) + .unwrap(); + let mut writer = new_data_writer("base").build(None).await.unwrap(); + writer.write(batch).await.unwrap(); + let base_files = writer.close().await.unwrap(); + + let tx = Transaction::new(&table); + let tx = tx + .fast_append() + .add_data_files(base_files) + .apply(tx) + .unwrap(); + let table = tx.commit(&catalog).await.unwrap(); + + // Commit 2: a row delta that upserts id=2 -> (2, "b2"): an equality + // delete for id=2 plus a data file with the replacement row. + let equality_config = + EqualityDeleteWriterConfig::new(vec![1], table.metadata().current_schema().clone()) + .unwrap(); + let delete_arrow_schema = equality_config.projected_arrow_schema_ref().clone(); + let delete_iceberg_schema = std::sync::Arc::new( + crate::arrow::arrow_schema_to_schema(&delete_arrow_schema).unwrap(), + ); + let delete_parquet_builder = + ParquetWriterBuilder::new(WriterProperties::builder().build(), delete_iceberg_schema); + let delete_file_name_gen = DefaultFileNameGenerator::new( + "delete".to_string(), + None, + crate::spec::DataFileFormat::Parquet, + ); + let mut delete_writer = EqualityDeleteFileWriterBuilder::new( + RollingFileWriterBuilder::new_with_default_file_size( + delete_parquet_builder, + table.file_io().clone(), + location_gen.clone(), + delete_file_name_gen, + ), + equality_config, + ) + .build(None) + .await + .unwrap(); + // The equality delete writer projects the equality columns itself, so + // it receives full-schema batches carrying the keys to delete. + let delete_batch = RecordBatch::try_new(arrow_schema.clone(), vec![ + std::sync::Arc::new(Int64Array::from(vec![2])), + std::sync::Arc::new(StringArray::from(vec![Some("b")])), + ]) + .unwrap(); + delete_writer.write(delete_batch).await.unwrap(); + let delete_files = delete_writer.close().await.unwrap(); + assert!( + delete_files + .iter() + .all(|f| f.content_type() == DataContentType::EqualityDeletes) + ); + + let upsert_batch = RecordBatch::try_new(arrow_schema.clone(), vec![ + std::sync::Arc::new(Int64Array::from(vec![2])), + std::sync::Arc::new(StringArray::from(vec![Some("b2")])), + ]) + .unwrap(); + let mut writer = new_data_writer("upsert").build(None).await.unwrap(); + writer.write(upsert_batch).await.unwrap(); + let upsert_files = writer.close().await.unwrap(); + + let tx = Transaction::new(&table); + let tx = tx + .row_delta() + .add_data_files(upsert_files) + .add_delete_files(delete_files) + .apply(tx) + .unwrap(); + let table = tx.commit(&catalog).await.unwrap(); + + assert_eq!( + table + .metadata() + .current_snapshot() + .unwrap() + .summary() + .operation, + Operation::Overwrite + ); + + // Read back through the scan path: the equality delete must hide the + // old (2, "b") row while the same commit's (2, "b2") row survives. + let batches: Vec = table + .scan() + .select_all() + .build() + .unwrap() + .to_arrow() + .await + .unwrap() + .try_collect() + .await + .unwrap(); + + let mut rows: Vec<(i64, String)> = batches + .iter() + .flat_map(|b| { + let ids = b + .column(0) + .as_any() + .downcast_ref::() + .unwrap() + .iter() + .map(|v| v.unwrap()); + let data = b + .column(1) + .as_any() + .downcast_ref::() + .unwrap() + .iter() + .map(|v| v.unwrap().to_string()); + ids.zip(data).collect::>() + }) + .collect(); + rows.sort(); + + assert_eq!(rows, vec![ + (1, "a".to_string()), + (2, "b2".to_string()), + (3, "c".to_string()) + ]); + } +} diff --git a/crates/iceberg/src/transaction/snapshot.rs b/crates/iceberg/src/transaction/snapshot.rs index d200c5ba9c..79b10c5f9e 100644 --- a/crates/iceberg/src/transaction/snapshot.rs +++ b/crates/iceberg/src/transaction/snapshot.rs @@ -115,6 +115,10 @@ pub(crate) struct SnapshotProducer<'a> { commit_uuid: Uuid, snapshot_properties: HashMap, added_data_files: Vec, + // Delete files (equality or position deletes) committed in the same + // snapshot. They receive the new snapshot's sequence number, so they + // apply to strictly older data files only (row-delta semantics). + added_delete_files: Vec, // A counter used to generate unique manifest file names. // It starts from 0 and increments for each new manifest file. // Note: This counter is limited to the range of (0..u64::MAX). @@ -127,6 +131,7 @@ impl<'a> SnapshotProducer<'a> { commit_uuid: Uuid, snapshot_properties: HashMap, added_data_files: Vec, + added_delete_files: Vec, ) -> Self { Self { table, @@ -134,6 +139,7 @@ impl<'a> SnapshotProducer<'a> { commit_uuid, snapshot_properties, added_data_files, + added_delete_files, manifest_counter: (0..), } } @@ -162,6 +168,29 @@ impl<'a> SnapshotProducer<'a> { Ok(()) } + pub(crate) fn validate_added_delete_files(&self) -> Result<()> { + for delete_file in &self.added_delete_files { + if delete_file.content_type() == crate::spec::DataContentType::Data { + return Err(Error::new( + ErrorKind::DataInvalid, + "Only delete content types (position or equality) are allowed in the delete-file list", + )); + } + if self.table.metadata().default_partition_spec_id() != delete_file.partition_spec_id { + return Err(Error::new( + ErrorKind::DataInvalid, + "Delete file partition spec id does not match table default partition spec id", + )); + } + Self::validate_partition_value( + delete_file.partition(), + self.table.metadata().default_partition_type(), + )?; + } + + Ok(()) + } + pub(crate) async fn validate_duplicate_files(&self) -> Result<()> { let Some(current_snapshot) = self.table.metadata().current_snapshot() else { return Ok(()); @@ -334,6 +363,37 @@ impl<'a> SnapshotProducer<'a> { writer.write_manifest_file().await } + // Write manifest file for added delete files and return the ManifestFile for ManifestList. + async fn write_added_delete_manifest(&mut self) -> Result { + let added_delete_files = std::mem::take(&mut self.added_delete_files); + if added_delete_files.is_empty() { + return Err(Error::new( + ErrorKind::PreconditionFailed, + "No added delete files found when write an added delete manifest file", + )); + } + + let snapshot_id = self.snapshot_id; + let format_version = self.table.metadata().format_version(); + let manifest_entries = added_delete_files.into_iter().map(|delete_file| { + let builder = ManifestEntry::builder() + .status(crate::spec::ManifestStatus::Added) + .data_file(delete_file); + if format_version == FormatVersion::V1 { + builder.snapshot_id(snapshot_id).build() + } else { + // For format version > 1, we set the snapshot id at the inherited time to avoid rewrite the manifest file when + // commit failed. + builder.build() + } + }); + let mut writer = self.new_manifest_writer(ManifestContentType::Deletes)?; + for entry in manifest_entries { + writer.add_entry(entry)?; + } + writer.write_manifest_file().await + } + /// Creates new manifests for data files added or removed, /// and collects all of the manifests to be included in the new snapshot as [ManifestFile] entries. async fn produce_manifests( @@ -346,7 +406,10 @@ impl<'a> SnapshotProducer<'a> { // TODO: Allowing snapshot property setup with no added data files is a workaround. // We should clean it up after all necessary actions are supported. // For details, please refer to https://github.com/apache/iceberg-rust/issues/1548 - if self.added_data_files.is_empty() && self.snapshot_properties.is_empty() { + if self.added_data_files.is_empty() + && self.added_delete_files.is_empty() + && self.snapshot_properties.is_empty() + { return Err(Error::new( ErrorKind::PreconditionFailed, "No added data files or added snapshot properties found when write a manifest file", @@ -362,6 +425,13 @@ impl<'a> SnapshotProducer<'a> { manifest_files.push(added_manifest); } + // Process added delete files (position or equality deletes committed + // together with the new data files). + if !self.added_delete_files.is_empty() { + let added_delete_manifest = self.write_added_delete_manifest().await?; + manifest_files.push(added_delete_manifest); + } + // # TODO // Support process delete entries. @@ -400,6 +470,14 @@ impl<'a> SnapshotProducer<'a> { ); } + for delete_file in &self.added_delete_files { + summary_collector.add_file( + delete_file, + table_metadata.current_schema().clone(), + table_metadata.default_partition_spec().clone(), + ); + } + let previous_snapshot = table_metadata.current_snapshot(); // User-supplied snapshot properties are applied first, then the computed From 52565b28027c65e040c876384f51cbe9d6004a3a Mon Sep 17 00:00:00 2001 From: Steven Isaac Date: Wed, 8 Jul 2026 07:13:13 -0400 Subject: [PATCH 2/3] chore: regenerate public API snapshots --- crates/iceberg/public-api.txt | 1 + 1 file changed, 1 insertion(+) diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index 41d4d7691b..12dff9d1e4 100644 --- a/crates/iceberg/public-api.txt +++ b/crates/iceberg/public-api.txt @@ -3082,6 +3082,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::row_delta(&self) -> iceberg::transaction::row_delta::RowDeltaAction 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 From 6ed087906b226cd4a3cc05459e10279b6fd1b09d Mon Sep 17 00:00:00 2001 From: Steven Isaac Date: Thu, 9 Jul 2026 13:33:38 -0400 Subject: [PATCH 3/3] feat(transaction): content-based operation + equality-key write guards for RowDelta Review feedback (thanks to Andrei Tserakhau's iceberg-go experience with the same feature): - select the snapshot operation from the delta's content, matching iceberg-java's BaseRowDelta: data-only = append, deletes-only = delete, both = overwrite (was: always overwrite) - write-time guards: reject equality delete files with an empty equality_ids set (would match every row), with unknown field ids, or keyed on float/double columns (equality undefined: NaN, signed zero) - document that BaseRowDelta's conflict-validation surface is deliberately deferred; sequence-number semantics carry row-level correctness (tested: the round-trip test upserts a PK via same-commit equality delete + re-insert and the scan sees the new row) --- crates/iceberg/src/transaction/row_delta.rs | 83 +++++++++++++++++++-- crates/iceberg/src/transaction/snapshot.rs | 42 +++++++++++ 2 files changed, 118 insertions(+), 7 deletions(-) diff --git a/crates/iceberg/src/transaction/row_delta.rs b/crates/iceberg/src/transaction/row_delta.rs index 950a7e118c..9ac85a1a85 100644 --- a/crates/iceberg/src/transaction/row_delta.rs +++ b/crates/iceberg/src/transaction/row_delta.rs @@ -30,8 +30,9 @@ use crate::transaction::snapshot::{ use crate::transaction::{ActionCommit, TransactionAction}; /// RowDeltaAction is a transaction action that commits data files and delete -/// files (position or equality deletes) together in a single `overwrite` -/// snapshot. +/// files (position or equality deletes) together in a single snapshot. The +/// snapshot operation reflects the delta's content: `append` for data files +/// only, `delete` for delete files only, `overwrite` for both. /// /// This is the write-side primitive for merge-on-read row-level changes, /// equivalent to `Table.newRowDelta()` in iceberg-java: both the added data @@ -40,6 +41,15 @@ use crate::transaction::{ActionCommit, TransactionAction}; /// numbers only. Data files committed in the same row delta are therefore /// not affected by its own delete files, which gives upsert semantics when a /// row delta pairs equality deletes with the rows' new values. +/// +/// Not yet implemented (deliberately deferred, matching how other engines +/// phased this in): the conflict-validation surface of iceberg-java's +/// `BaseRowDelta` (`validateFromSnapshot`, `validateNoConflictingDataFiles`, +/// `conflictDetectionFilter`, …). Sequence-number semantics carry row-level +/// correctness on their own — Flink runs with most validation off for the +/// same reason — but concurrent-writer workflows that need +/// serializable-style checks should wait for partition-scoped conflict +/// detection before relying on this action. pub struct RowDeltaAction { check_duplicate: bool, // below are properties used to create SnapshotProducer when commit @@ -111,19 +121,32 @@ impl TransactionAction for RowDeltaAction { snapshot_producer.validate_duplicate_files().await?; } + let operation = RowDeltaOperation { + has_data_files: !self.added_data_files.is_empty(), + has_delete_files: !self.added_delete_files.is_empty(), + }; snapshot_producer - .commit(RowDeltaOperation, DefaultManifestProcess) + .commit(operation, DefaultManifestProcess) .await } } -struct RowDeltaOperation; +struct RowDeltaOperation { + has_data_files: bool, + has_delete_files: bool, +} impl SnapshotProduceOperation for RowDeltaOperation { fn operation(&self) -> Operation { - // Matches iceberg-java's BaseRowDelta: a row delta commits both data - // and delete files, so the snapshot operation is `overwrite`. - Operation::Overwrite + // Matches iceberg-java's BaseRowDelta: the snapshot operation + // reflects what the delta actually contains — data files only is an + // `append`, delete files only is a `delete`, and both together is an + // `overwrite`. + match (self.has_data_files, self.has_delete_files) { + (true, false) => Operation::Append, + (false, true) => Operation::Delete, + _ => Operation::Overwrite, + } } async fn delete_entries( @@ -356,6 +379,52 @@ mod tests { manifest_list.entries()[0].content, ManifestContentType::Deletes ); + // Content-based operation selection: deletes-only is a `delete`. + assert_eq!(new_snapshot.summary().operation, Operation::Delete); + } + + #[tokio::test] + async fn test_row_delta_data_only_is_append() { + // Content-based operation selection: data-only is an `append`. + let table = make_v2_minimal_table(); + let tx = Transaction::new(&table); + let action = tx.row_delta().add_data_files(vec![test_data_file(&table)]); + let mut action_commit = Arc::new(action).commit(&table).await.unwrap(); + let updates = action_commit.take_updates(); + let TableUpdate::AddSnapshot { snapshot } = &updates[0] else { + unreachable!() + }; + assert_eq!(snapshot.summary().operation, Operation::Append); + } + + #[tokio::test] + async fn test_row_delta_rejects_empty_equality_ids() { + // Write-time guard: an equality delete with an empty key set would + // match every row; reject it before it reaches a manifest. + let table = make_v2_minimal_table(); + let tx = Transaction::new(&table); + let mut delete_file = test_equality_delete_file(&table); + delete_file.equality_ids = Some(vec![]); + let action = tx.row_delta().add_delete_files(vec![delete_file]); + let err = match Arc::new(action).commit(&table).await { + Err(e) => e, + Ok(_) => panic!("expected empty equality_ids to be rejected"), + }; + assert!(err.to_string().contains("non-empty equality_ids")); + } + + #[tokio::test] + async fn test_row_delta_rejects_unknown_equality_field() { + let table = make_v2_minimal_table(); + let tx = Transaction::new(&table); + let mut delete_file = test_equality_delete_file(&table); + delete_file.equality_ids = Some(vec![9999]); + let action = tx.row_delta().add_delete_files(vec![delete_file]); + let err = match Arc::new(action).commit(&table).await { + Err(e) => e, + Ok(_) => panic!("expected unknown equality field to be rejected"), + }; + assert!(err.to_string().contains("unknown field id 9999")); } #[tokio::test] diff --git a/crates/iceberg/src/transaction/snapshot.rs b/crates/iceberg/src/transaction/snapshot.rs index 79b10c5f9e..7a386ce624 100644 --- a/crates/iceberg/src/transaction/snapshot.rs +++ b/crates/iceberg/src/transaction/snapshot.rs @@ -186,11 +186,53 @@ impl<'a> SnapshotProducer<'a> { delete_file.partition(), self.table.metadata().default_partition_type(), )?; + if delete_file.content_type() == crate::spec::DataContentType::EqualityDeletes { + self.validate_equality_ids(delete_file)?; + } } Ok(()) } + /// Equality-delete key sanity, enforced at write time so bad files never + /// reach a manifest: the key set must be non-empty (an empty set would + /// match every row), every id must resolve in the current schema, and + /// float/double columns are rejected because equality is undefined for + /// them (NaN, +0.0/-0.0) — mirrors iceberg-java's identifier-field rules. + fn validate_equality_ids(&self, delete_file: &DataFile) -> Result<()> { + let ids = delete_file.equality_ids().unwrap_or_default(); + if ids.is_empty() { + return Err(Error::new( + ErrorKind::DataInvalid, + "Equality delete file must declare a non-empty equality_ids set", + )); + } + let schema = self.table.metadata().current_schema(); + for id in ids { + let field = schema.field_by_id(id).ok_or_else(|| { + Error::new( + ErrorKind::DataInvalid, + format!("Equality delete references unknown field id {id}"), + ) + })?; + if matches!( + field.field_type.as_ref(), + crate::spec::Type::Primitive(crate::spec::PrimitiveType::Float) + | crate::spec::Type::Primitive(crate::spec::PrimitiveType::Double) + ) { + return Err(Error::new( + ErrorKind::DataInvalid, + format!( + "Equality delete on float/double column {:?} (field id {id}) is not \ + allowed: floating-point equality is undefined (NaN, signed zero)", + field.name + ), + )); + } + } + Ok(()) + } + pub(crate) async fn validate_duplicate_files(&self) -> Result<()> { let Some(current_snapshot) = self.table.metadata().current_snapshot() else { return Ok(());