support rewrite manifest

Author: ZENOTME

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/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, RewriteManifsetAction};
 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<RewriteManifsetAction<'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()
+        };
+        RewriteManifsetAction::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 {