continue on validation, added validate_no_new_delete_files_for_data_files

Author: CTTY

crates/iceberg/src/spec/snapshot.rs

@@ -38,7 +38,7 @@ pub const UNASSIGNED_SNAPSHOT_ID: i64 = -1;
 
 /// Reference to [`Snapshot`].
 pub type SnapshotRef = Arc<Snapshot>;
-#[derive(Debug, Serialize, Deserialize, PartialEq, Eq, Clone)]
+#[derive(Debug, Serialize, Deserialize, PartialEq, Eq, Clone, Hash)]
 #[serde(rename_all = "lowercase")]
 /// The operation field is used by some operations, like snapshot expiration, to skip processing certain snapshots.
 pub enum Operation {

crates/iceberg/src/transaction/append.rs

@@ -94,6 +94,7 @@ impl TransactionAction for FastAppendAction {
             vec![],
             vec![],
             vec![],
+            None,
         );
 
         // validate added files

crates/iceberg/src/transaction/rewrite_files.rs

@@ -40,13 +40,16 @@ pub struct RewriteFilesAction {
     added_delete_files: Vec<DataFile>,
     deleted_data_files: Vec<DataFile>,
     deleted_delete_files: Vec<DataFile>,
+    starting_sequence_number: Option<i64>,
+    starting_snapshot_id: Option<i64>,
 }
 
 pub struct RewriteFilesOperation {
     added_data_files: Vec<DataFile>,
     added_delete_files: Vec<DataFile>,
     deleted_data_files: Vec<DataFile>,
     deleted_delete_files: Vec<DataFile>,
+    starting_snapshot_id: Option<i64>,
 }
 
 impl RewriteFilesAction {
@@ -59,6 +62,8 @@ impl RewriteFilesAction {
             added_delete_files: vec![],
             deleted_data_files: vec![],
             deleted_delete_files: vec![],
+            starting_sequence_number: None,
+            starting_snapshot_id: None,
         }
     }
 
@@ -112,6 +117,19 @@ impl RewriteFilesAction {
         self.snapshot_properties = snapshot_properties;
         self
     }
+
+    /// 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);
+        self
+    }
+
+    /// Set the snapshot ID used in any reads for this operation.
+    pub fn set_starting_snapshot_id(mut self, snapshot_id: i64) -> Self {
+        self.starting_snapshot_id = Some(snapshot_id);
+        self
+    }
 }
 
 impl Default for RewriteFilesAction {
@@ -132,13 +150,15 @@ impl TransactionAction for RewriteFilesAction {
             self.added_delete_files.clone(),
             self.deleted_data_files.clone(),
             self.deleted_delete_files.clone(),
+            self.starting_sequence_number.clone(),
         );
 
         let rewrite_operation = RewriteFilesOperation {
             added_data_files: self.added_data_files.clone(),
             added_delete_files: self.added_delete_files.clone(),
             deleted_data_files: self.deleted_data_files.clone(),
             deleted_delete_files: self.deleted_delete_files.clone(),
+            starting_snapshot_id: self.starting_snapshot_id.clone(),
         };
 
         // todo should be able to configure merge manifest process

crates/iceberg/src/transaction/snapshot.rs

@@ -82,9 +82,11 @@ pub(crate) struct SnapshotProducer<'a> {
     // It starts from 0 and increments for each new manifest file.
     // Note: This counter is limited to the range of (0..u64::MAX).
     manifest_counter: RangeFrom<u64>,
+    starting_sequence_number: Option<i64>,
 }
 
 impl<'a> SnapshotProducer<'a> {
+    // todo add a builder for this to fix the clippy
     #[allow(clippy::too_many_arguments)]
     pub(crate) fn new(
         table: &'a Table,
@@ -95,6 +97,7 @@ impl<'a> SnapshotProducer<'a> {
         added_delete_files: Vec<DataFile>,
         deleted_data_files: Vec<DataFile>,
         deleted_delete_files: Vec<DataFile>,
+        starting_sequence_number: Option<i64>,
     ) -> Self {
         Self {
             table,
@@ -107,6 +110,7 @@ impl<'a> SnapshotProducer<'a> {
             deleted_data_files,
             deleted_delete_files,
             manifest_counter: (0..),
+            starting_sequence_number,
         }
     }
 

crates/iceberg/src/transaction/validate.rs

@@ -16,27 +16,136 @@
 // under the License.
 
 use std::collections::HashSet;
+use std::sync::Arc;
 
+use futures::{Sink, SinkExt};
+use futures::future::try_join_all;
+use once_cell::sync::Lazy;
+
+use crate::delete_file_index::DeleteFileIndex;
 use crate::error::Result;
-use crate::spec::{ManifestContentType, ManifestFile, Operation, SnapshotRef};
+use crate::scan::DeleteFileContext;
+use crate::spec::{
+    DataFile, ManifestContentType, ManifestFile, Operation, SnapshotRef,
+};
 use crate::table::Table;
+use crate::util::snapshot::ancestors_between;
+use crate::{Error, ErrorKind};
+
+static VALIDATE_ADDED_DELETE_FILES_OPERATIONS: Lazy<HashSet<Operation>> =
+    Lazy::new(|| HashSet::from([Operation::Overwrite, Operation::Delete]));
 
 pub(crate) trait SnapshotValidator {
+    // todo doc
+    // table: base table
+    // 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<()> {
         // todo: add default implementation
         Ok(())
     }
 
-    #[allow(dead_code)]
+    // todo doc
     async fn validation_history(
         &self,
-        _base: &Table,
-        _from_snapshot: Option<&SnapshotRef>,
-        _to_snapshot: &SnapshotRef,
-        _matching_operations: HashSet<Operation>,
-        _manifest_content_type: ManifestContentType,
+        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>,
+        matching_operations: &HashSet<Operation>,
+        manifest_content_type: ManifestContentType,
     ) -> Result<(Vec<ManifestFile>, HashSet<i64>)> {
-        // todo: add default implementation
-        Ok((vec![], HashSet::new()))
+        let mut manifests: Vec<ManifestFile> = vec![];
+        let mut new_snapshots = HashSet::new();
+        let mut last_snapshot: Option<SnapshotRef> = None;
+
+        let snapshots = ancestors_between(
+            &Arc::new(base.metadata().clone()),
+            to_snapshot.snapshot_id(),
+            from_snapshot_id.clone(),
+        );
+
+        for current_snapshot in snapshots {
+            last_snapshot = Some(current_snapshot.clone());
+
+            // Find all snapshots with the matching operations
+            // and their manifest files with the matching content type
+            if matching_operations.contains(&current_snapshot.summary().operation) {
+                new_snapshots.insert(current_snapshot.snapshot_id());
+                current_snapshot
+                    .load_manifest_list(base.file_io(), base.metadata())
+                    .await?
+                    .entries()
+                    .iter()
+                    .for_each(|manifest| {
+                        if manifest.content == manifest_content_type
+                            && manifest.added_snapshot_id == current_snapshot.snapshot_id()
+                        {
+                            manifests.push(manifest.clone());
+                        }
+                    });
+            }
+        }
+
+        if last_snapshot.is_some()
+            && last_snapshot.clone().unwrap().parent_snapshot_id() != from_snapshot_id
+        {
+            return Err(Error::new(
+                ErrorKind::DataInvalid,
+                format!(
+                    "Cannot determine history between starting snapshot {} and the last known ancestor {}",
+                    from_snapshot_id.unwrap_or(-1),
+                    last_snapshot.unwrap().snapshot_id()
+                ),
+            ));
+        }
+
+        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,
+    ) -> Result<()> {
+        // Get matching delete files have been added since the from_snapshot_id
+        let (delete_manifests, snapshot_ids) = self
+            .validation_history(
+                base,
+                to_snapshot,
+                from_snapshot_id,
+                &VALIDATE_ADDED_DELETE_FILES_OPERATIONS,
+                ManifestContentType::Deletes,
+            )
+            .await?;
+
+        // Building 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 {
+            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?
+        }
+        
+        // todo validate if there are deletes
+
+        Ok(())
     }
 }