Merge PR apache#1606: Add RewriteFilesAction for truncate support

Author: wasifaleem

crates/iceberg/src/lib.rs

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

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

@@ -295,10 +295,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;
@@ -341,7 +337,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

@@ -38,7 +38,7 @@ pub const UNASSIGNED_SNAPSHOT_ID: i64 = -1;
 
 /// Reference to [`Snapshot`].
 pub type SnapshotRef = Arc<Snapshot>;
-#[derive(Debug, Default, Serialize, Deserialize, PartialEq, Eq, Clone)]
+#[derive(Debug, Default, 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/spec/snapshot_summary.rs

@@ -339,6 +339,7 @@ pub(crate) fn update_snapshot_summaries(
     if summary.operation != Operation::Append
         && summary.operation != Operation::Overwrite
         && summary.operation != Operation::Delete
+        && summary.operation != Operation::Replace
     {
         return Err(Error::new(
             ErrorKind::DataInvalid,

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.
@@ -103,6 +104,8 @@ impl TransactionAction for FastAppendAction {
             self.snapshot_properties.clone(),
             self.added_data_files.clone(),
             self.added_delete_files.clone(),
+            vec![],
+            vec![],
         );
 
         snapshot_producer.validate_added_data_files(&self.added_data_files)?;
@@ -121,6 +124,8 @@ impl TransactionAction for FastAppendAction {
 
 struct FastAppendOperation;
 
+impl SnapshotValidator for FastAppendOperation {}
+
 impl SnapshotProduceOperation for FastAppendOperation {
     fn operation(&self) -> Operation {
         Operation::Append
@@ -135,7 +140,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

@@ -54,13 +54,15 @@ mod action;
 
 pub use action::*;
 mod append;
+mod rewrite_files;
 mod snapshot;
 mod sort_order;
 mod update_location;
 mod update_properties;
 mod update_schema;
 mod update_statistics;
 mod upgrade_format_version;
+mod validate;
 
 use std::sync::Arc;
 use std::time::Duration;
@@ -73,6 +75,7 @@ use crate::spec::TableProperties;
 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;
@@ -154,6 +157,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()