feat: support rewrite manifest action

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

@@ -136,6 +136,12 @@ impl ManifestEntry {
         self.sequence_number
     }
 
+    /// File sequence number.
+    #[inline]
+    pub fn file_sequence_number(&self) -> Option<i64> {
+        self.file_sequence_number
+    }
+
     /// File size in bytes.
     #[inline]
     pub fn file_size_in_bytes(&self) -> u64 {

crates/iceberg/src/spec/snapshot_summary.rs

@@ -335,17 +335,6 @@ pub(crate) fn update_snapshot_summaries(
     previous_summary: Option<&Summary>,
     truncate_full_table: bool,
 ) -> Result<Summary> {
-    // Validate that the operation is supported
-    if summary.operation != Operation::Append
-        && summary.operation != Operation::Overwrite
-        && summary.operation != Operation::Delete
-    {
-        return Err(Error::new(
-            ErrorKind::DataInvalid,
-            "Operation is not supported.",
-        ));
-    }
-
     let mut summary = match previous_summary {
         Some(prev_summary) if truncate_full_table && summary.operation == Operation::Overwrite => {
             truncate_table_summary(summary, prev_summary)

crates/iceberg/src/transaction/append.rs

@@ -284,7 +284,7 @@ mod tests {
             new_snapshot.sequence_number()
         );
 
-        // check manifset
+        // check manifest
         let manifest = manifest_list.entries()[0]
             .load_manifest(table.file_io())
             .await

crates/iceberg/src/transaction/mod.rs

@@ -18,14 +18,20 @@
 //! This module contains transaction api.
 
 mod append;
+mod rewrite_manifest;
+pub use rewrite_manifest::{
+    CREATED_MANIFESTS_COUNT, KEPT_MANIFESTS_COUNT, PROCESSED_ENTRY_COUNT, REPLACED_MANIFESTS_COUNT,
+};
 mod snapshot;
 mod sort_order;
 
 use std::cmp::Ordering;
 use std::collections::HashMap;
+use std::hash::Hash;
 use std::mem::discriminant;
 use std::sync::Arc;
 
+use rewrite_manifest::{ClusterFunc, PredicateFunc, RewriteManifestAction};
 use uuid::Uuid;
 
 use crate::error::Result;
@@ -166,6 +172,42 @@ impl<'a> Transaction<'a> {
         )
     }
 
+    /// Rewrite manifest file.
+    pub fn rewrite_manifest<T: Hash + Eq>(
+        self,
+        cluster_by_func: Option<ClusterFunc<T>>,
+        manifest_predicate: Option<PredicateFunc>,
+        snapshot_id: Option<i64>,
+        commit_uuid: Option<Uuid>,
+        key_metadata: Vec<u8>,
+    ) -> Result<RewriteManifestAction<'a, T>> {
+        let snapshot_id = if let Some(snapshot_id) = snapshot_id {
+            if self
+                .current_table
+                .metadata()
+                .snapshots()
+                .any(|s| s.snapshot_id() == snapshot_id)
+            {
+                return Err(Error::new(
+                    ErrorKind::DataInvalid,
+                    format!("Snapshot id {} already exists", snapshot_id),
+                ));
+            }
+            snapshot_id
+        } else {
+            self.generate_unique_snapshot_id()
+        };
+        RewriteManifestAction::new(
+            self,
+            cluster_by_func,
+            manifest_predicate,
+            snapshot_id,
+            commit_uuid.unwrap_or_else(Uuid::now_v7),
+            key_metadata,
+            HashMap::new(),
+        )
+    }
+
     /// Creates replace sort order action.
     pub fn replace_sort_order(self) -> ReplaceSortOrderAction<'a> {
         ReplaceSortOrderAction {