feat(transaction): RewriteFilesAction + Validation Support + Process Delete Manifests

crates/iceberg/src/lib.rs

@@ -87,3 +87,5 @@ pub mod writer;
 
 mod delete_vector;
 pub mod puffin;
+/// Utility functions and modules.
+pub mod util;

crates/iceberg/src/spec/manifest/writer.rs

@@ -236,10 +236,6 @@ impl ManifestWriter {
     /// Add a delete manifest entry. This method will update following status of the entry:
     /// - Update the entry status to `Deleted`
     /// - Set the snapshot id to the current snapshot id
-    ///
-    /// # TODO
-    /// Remove this allow later
-    #[allow(dead_code)]
     pub(crate) fn add_delete_entry(&mut self, mut entry: ManifestEntry) -> Result<()> {
         self.check_data_file(&entry.data_file)?;
         entry.status = ManifestStatus::Deleted;
@@ -282,7 +278,7 @@ impl ManifestWriter {
         Ok(())
     }
 
-    /// Add an file as existing manifest entry. The original data and file sequence numbers, snapshot ID,
+    /// Add a file as existing manifest entry. The original data and file sequence numbers, snapshot ID,
     /// which were assigned at commit, must be preserved when adding an existing entry.
     pub fn add_existing_file(
         &mut self,

crates/iceberg/src/spec/snapshot.rs

@@ -39,7 +39,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

@@ -27,6 +27,7 @@ use crate::table::Table;
 use crate::transaction::snapshot::{
     DefaultManifestProcess, SnapshotProduceOperation, SnapshotProducer,
 };
+use crate::transaction::validate::SnapshotValidator;
 use crate::transaction::{ActionCommit, TransactionAction};
 
 /// FastAppendAction is a transaction action for fast append data files to the table.
@@ -90,6 +91,10 @@ impl TransactionAction for FastAppendAction {
             self.key_metadata.clone(),
             self.snapshot_properties.clone(),
             self.added_data_files.clone(),
+            vec![],
+            vec![],
+            vec![],
+            None,
         );
 
         // validate added files
@@ -110,6 +115,8 @@ impl TransactionAction for FastAppendAction {
 
 struct FastAppendOperation;
 
+impl SnapshotValidator for FastAppendOperation {}
+
 impl SnapshotProduceOperation for FastAppendOperation {
     fn operation(&self) -> Operation {
         Operation::Append
@@ -124,7 +131,7 @@ impl SnapshotProduceOperation for FastAppendOperation {
 
     async fn existing_manifest(
         &self,
-        snapshot_produce: &SnapshotProducer<'_>,
+        snapshot_produce: &mut SnapshotProducer<'_>,
     ) -> Result<Vec<ManifestFile>> {
         let Some(snapshot) = snapshot_produce.table.metadata().current_snapshot() else {
             return Ok(vec![]);

crates/iceberg/src/transaction/mod.rs

@@ -56,12 +56,14 @@ use std::collections::HashMap;
 
 pub use action::*;
 mod append;
+mod rewrite_files;
 mod snapshot;
 mod sort_order;
 mod update_location;
 mod update_properties;
 mod update_statistics;
 mod upgrade_format_version;
+mod validate;
 
 use std::sync::Arc;
 use std::time::Duration;
@@ -78,6 +80,7 @@ use crate::spec::{
 use crate::table::Table;
 use crate::transaction::action::BoxedTransactionAction;
 use crate::transaction::append::FastAppendAction;
+use crate::transaction::rewrite_files::RewriteFilesAction;
 use crate::transaction::sort_order::ReplaceSortOrderAction;
 use crate::transaction::update_location::UpdateLocationAction;
 use crate::transaction::update_properties::UpdatePropertiesAction;
@@ -153,6 +156,11 @@ impl Transaction {
         ReplaceSortOrderAction::new()
     }
 
+    /// Rewrite a set of data files of table
+    pub fn rewrite_files(&self) -> RewriteFilesAction {
+        RewriteFilesAction::new()
+    }
+
     /// Set the location of table
     pub fn update_location(&self) -> UpdateLocationAction {
         UpdateLocationAction::new()