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 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..9ac85a1a85 --- /dev/null +++ b/crates/iceberg/src/transaction/row_delta.rs @@ -0,0 +1,635 @@ +// 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 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 +/// 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. +/// +/// 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 + 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?; + } + + let operation = RowDeltaOperation { + has_data_files: !self.added_data_files.is_empty(), + has_delete_files: !self.added_delete_files.is_empty(), + }; + snapshot_producer + .commit(operation, DefaultManifestProcess) + .await + } +} + +struct RowDeltaOperation { + has_data_files: bool, + has_delete_files: bool, +} + +impl SnapshotProduceOperation for RowDeltaOperation { + fn operation(&self) -> Operation { + // 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( + &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 + ); + // 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] + 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..7a386ce624 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,71 @@ 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(), + )?; + 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(()); @@ -334,6 +405,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 +448,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 +467,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 +512,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