finished validate_no_new_delete_files_for_data_files

Author: CTTY

Cargo.lock

@@ -26,7 +26,7 @@ use super::{ActionCommit, TransactionAction};
 use crate::error::{Error, ErrorKind, Result};
 use crate::spec::{
     DataContentType, DataFile, ManifestContentType, ManifestEntry, ManifestEntryRef, ManifestFile,
-    ManifestStatus, Operation, SnapshotRef,
+    ManifestStatus, Operation,
 };
 use crate::table::Table;
 use crate::transaction::validate::SnapshotValidator;
@@ -40,7 +40,7 @@ pub struct RewriteFilesAction {
     added_delete_files: Vec<DataFile>,
     deleted_data_files: Vec<DataFile>,
     deleted_delete_files: Vec<DataFile>,
-    starting_sequence_number: Option<i64>,
+    data_sequence_number: Option<i64>,
     starting_snapshot_id: Option<i64>,
 }
 
@@ -50,6 +50,7 @@ pub struct RewriteFilesOperation {
     deleted_data_files: Vec<DataFile>,
     deleted_delete_files: Vec<DataFile>,
     starting_snapshot_id: Option<i64>,
+    data_sequence_number: Option<i64>,
 }
 
 impl RewriteFilesAction {
@@ -62,7 +63,7 @@ impl RewriteFilesAction {
             added_delete_files: vec![],
             deleted_data_files: vec![],
             deleted_delete_files: vec![],
-            starting_sequence_number: None,
+            data_sequence_number: None,
             starting_snapshot_id: None,
         }
     }
@@ -120,8 +121,8 @@ impl RewriteFilesAction {
 
     /// Set the data sequence number for this rewrite operation.
     /// The number will be used for all new data files that are added in this rewrite.
-    pub fn set_starting_sequence_number(mut self, sequence_number: i64) -> Self {
-        self.starting_sequence_number = Some(sequence_number);
+    pub fn set_data_sequence_number(mut self, sequence_number: i64) -> Self {
+        self.data_sequence_number = Some(sequence_number);
         self
     }
 
@@ -150,7 +151,7 @@ impl TransactionAction for RewriteFilesAction {
             self.added_delete_files.clone(),
             self.deleted_data_files.clone(),
             self.deleted_delete_files.clone(),
-            self.starting_sequence_number.clone(),
+            self.data_sequence_number.clone(),
         );
 
         let rewrite_operation = RewriteFilesOperation {
@@ -159,17 +160,18 @@ impl TransactionAction for RewriteFilesAction {
             deleted_data_files: self.deleted_data_files.clone(),
             deleted_delete_files: self.deleted_delete_files.clone(),
             starting_snapshot_id: self.starting_snapshot_id.clone(),
+            data_sequence_number: self.data_sequence_number.clone(),
         };
 
-        // todo should be able to configure merge manifest process
+        // todo should be able to configure to use the merge manifest process
         snapshot_producer
             .commit(rewrite_operation, DefaultManifestProcess)
             .await
     }
 }
 
 fn copy_with_deleted_status(entry: &ManifestEntryRef) -> Result<ManifestEntry> {
-    // todo should we fail on missing properties?
+    // todo should we fail on missing properties or should we ignore them when they don't exist?
     let builder = ManifestEntry::builder()
         .status(ManifestStatus::Deleted)
         .snapshot_id(entry.snapshot_id().ok_or_else(|| {
@@ -197,7 +199,7 @@ fn copy_with_deleted_status(entry: &ManifestEntryRef) -> Result<ManifestEntry> {
 }
 
 impl SnapshotValidator for RewriteFilesOperation {
-    fn validate(&self, _table: &Table, _snapshot: Option<&SnapshotRef>) -> Result<()> {
+    async fn validate(&self, base: &Table, parent_snapshot_id: Option<i64>) -> Result<()> {
         // Validate replaced and added files
         if self.deleted_data_files.is_empty() && self.deleted_delete_files.is_empty() {
             return Err(Error::new(
@@ -218,9 +220,18 @@ impl SnapshotValidator for RewriteFilesOperation {
             ));
         }
 
-        // todo add use_starting_seq_number to help the validation
-        // todo validate no new deletes since the current base
-        // if there are replaced data files, there cannot be any new row-level deletes for those data files
+        // todo add use_starting_seq_number to determine if we want to use data_sequence_number
+        // If there are replaced data files, there cannot be any new row-level deletes for those data files
+        if !self.deleted_data_files.is_empty() {
+            self.validate_no_new_delete_files_for_data_files(
+                base,
+                self.starting_snapshot_id,
+                parent_snapshot_id,
+                &self.deleted_data_files,
+                self.data_sequence_number.is_some(),
+            )
+            .await?;
+        }
 
         Ok(())
     }

crates/iceberg/src/transaction/rewrite_files.rs

@@ -476,7 +476,8 @@ impl<'a> SnapshotProducer<'a> {
     ) -> Result<ActionCommit> {
         // Validate to avoid conflicts
         snapshot_produce_operation
-            .validate(self.table, self.table.metadata().current_snapshot())?;
+            .validate(self.table, self.table.metadata().current_snapshot_id)
+            .await?;
 
         let new_manifests = self
             .manifest_file(&snapshot_produce_operation, &process)

crates/iceberg/src/transaction/snapshot.rs

@@ -18,14 +18,17 @@
 use std::collections::HashSet;
 use std::sync::Arc;
 
+use futures::SinkExt;
 use futures::future::try_join_all;
-use futures::{Sink, SinkExt};
 use once_cell::sync::Lazy;
 
 use crate::delete_file_index::DeleteFileIndex;
 use crate::error::Result;
 use crate::scan::DeleteFileContext;
-use crate::spec::{DataFile, ManifestContentType, ManifestFile, Operation, SnapshotRef};
+use crate::spec::{
+    DataContentType, DataFile, FormatVersion, INITIAL_SEQUENCE_NUMBER, ManifestContentType,
+    ManifestFile, Operation, SnapshotRef,
+};
 use crate::table::Table;
 use crate::util::snapshot::ancestors_between;
 use crate::{Error, ErrorKind};
@@ -39,7 +42,7 @@ pub(crate) trait SnapshotValidator {
     // snapshot: parent snapshot
     // usually snapshot is the latest snapshot of base table, unless it's non-main branch
     // but we don't support writing to branches as of now
-    fn validate(&self, _table: &Table, _snapshot: Option<&SnapshotRef>) -> Result<()> {
+    async fn validate(&self, _base: &Table, _parent_snapshot_id: Option<i64>) -> Result<()> {
         // todo: add default implementation
         Ok(())
     }
@@ -48,8 +51,8 @@ pub(crate) trait SnapshotValidator {
     async fn validation_history(
         &self,
         base: &Table,
-        to_snapshot: SnapshotRef, // todo maybe the naming/variable order can be better, or just snapshot id is better? this is parent
         from_snapshot_id: Option<i64>,
+        to_snapshot_id: i64,
         matching_operations: &HashSet<Operation>,
         manifest_content_type: ManifestContentType,
     ) -> Result<(Vec<ManifestFile>, HashSet<i64>)> {
@@ -59,7 +62,7 @@ pub(crate) trait SnapshotValidator {
 
         let snapshots = ancestors_between(
             &Arc::new(base.metadata().clone()),
-            to_snapshot.snapshot_id(),
+            to_snapshot_id,
             from_snapshot_id.clone(),
         );
 
@@ -101,49 +104,95 @@ pub(crate) trait SnapshotValidator {
         Ok((manifests, new_snapshots))
     }
 
-    #[allow(dead_code)]
     async fn validate_no_new_delete_files_for_data_files(
         &self,
         base: &Table,
         from_snapshot_id: Option<i64>,
-        _data_files: &[DataFile],
-        to_snapshot: SnapshotRef,
+        to_snapshot_id: Option<i64>,
+        data_files: &[DataFile],
+        ignore_equality_deletes: bool,
     ) -> Result<()> {
+        // If there is no current table state, no files have been added
+        if to_snapshot_id.is_none() || base.metadata().format_version() != FormatVersion::V1 {
+            return Ok(());
+        }
+        let to_snapshot_id = to_snapshot_id.unwrap();
+
         // Get matching delete files have been added since the from_snapshot_id
-        let (delete_manifests, snapshot_ids) = self
+        let (delete_manifests, _) = self
             .validation_history(
                 base,
-                to_snapshot,
                 from_snapshot_id,
+                to_snapshot_id,
                 &VALIDATE_ADDED_DELETE_FILES_OPERATIONS,
                 ManifestContentType::Deletes,
             )
             .await?;
 
-        // Building delete file index
-        let (_delete_file_index, mut delete_file_tx) = DeleteFileIndex::new();
+        // Build delete file index
+        let (delete_file_index, mut delete_file_tx) = DeleteFileIndex::new();
         let manifests = try_join_all(
             delete_manifests
                 .iter()
                 .map(|f| f.load_manifest(base.file_io()))
                 .collect::<Vec<_>>(),
         )
         .await?;
-
-        let delete_files_ctx = manifests
-            .iter()
-            .flat_map(|manifest| manifest.entries())
-            .map(|entry| DeleteFileContext {
+        let manifest_entries = manifests.iter().flat_map(|manifest| manifest.entries());
+        for entry in manifest_entries {
+            let delete_file_ctx = DeleteFileContext {
                 manifest_entry: entry.clone(),
                 partition_spec_id: entry.data_file().partition_spec_id,
-            })
-            .collect::<Vec<_>>();
-
-        for ctx in delete_files_ctx {
-            delete_file_tx.send(ctx).await?
+            };
+            delete_file_tx.send(delete_file_ctx).await?;
         }
 
-        // todo validate if there are deletes
+        // Get starting seq num from starting snapshot if available
+        let starting_sequence_number = if from_snapshot_id.is_some()
+            && base
+                .metadata()
+                .snapshots
+                .get(&from_snapshot_id.unwrap())
+                .is_some()
+        {
+            base.metadata()
+                .snapshots
+                .get(&from_snapshot_id.unwrap())
+                .unwrap()
+                .sequence_number()
+        } else {
+            INITIAL_SEQUENCE_NUMBER
+        };
+
+        // Validate if there are deletes using delete file index
+        for data_file in data_files {
+            let delete_files = delete_file_index
+                .get_deletes_for_data_file(data_file, Some(starting_sequence_number))
+                .await;
+
+            if ignore_equality_deletes {
+                if delete_files
+                    .iter()
+                    .any(|delete_file| delete_file.file_type == DataContentType::PositionDeletes)
+                {
+                    return Err(Error::new(
+                        ErrorKind::DataInvalid,
+                        format!(
+                            "Cannot commit, found new positional delete for added data file: {}",
+                            data_file.file_path
+                        ),
+                    ));
+                }
+            } else if !delete_files.is_empty() {
+                return Err(Error::new(
+                    ErrorKind::DataInvalid,
+                    format!(
+                        "Cannot commit, found new delete for added data file: {}",
+                        data_file.file_path
+                    ),
+                ));
+            }
+        }
 
         Ok(())
     }