refactor: update snapshot expiration implementation and remove legacy code

Author: ForeverAngry

iceberg-rust/examples/expire_snapshots.rs

@@ -1,13 +1,12 @@
-//! Snapshot expiration functionality for Iceberg tables
+//! Snapshot expiration functionality for Iceberg tables (Legacy implementation)
 //!
-//! This module provides the ability to expire old snapshots and clean up associated
-//! manifest and data files. The implementation follows Iceberg's atomic commit model
-//! and supports various expiration criteria including:
+//! **Note: This module contains the standalone implementation of snapshot expiration.** 
+//! **The recommended approach is to use `table.expire_snapshots()` which integrates**
+//! **with the Operation framework for better transaction support.**
 //!
-//! * Time-based expiration (older than timestamp)
-//! * Count-based retention (keep only last N snapshots)
-//! * Reference-aware cleanup (preserve snapshots referenced by branches/tags)
-//! * Optional orphaned file cleanup
+//! This module provides functionality to expire (remove) old snapshots from Iceberg tables
+//! based on various retention policies. Snapshot expiration helps manage storage costs
+//! by removing metadata for old table versions while preserving data integrity.
 
 use std::collections::HashSet;
 

iceberg-rust/src/table/maintenance/expire_snapshots.rs

@@ -41,11 +41,90 @@ use crate::{
     table::transaction::TableTransaction,
 };
 
-pub mod maintenance;
 pub mod manifest;
 pub mod manifest_list;
 pub mod transaction;
 
+/// Builder for configuring and executing snapshot expiration operations
+///
+/// This builder provides a fluent API for configuring how snapshots should be expired:
+/// * [`expire_older_than`](ExpireSnapshotsBuilder::expire_older_than) - Remove snapshots older than a timestamp
+/// * [`retain_last`](ExpireSnapshotsBuilder::retain_last) - Keep only the most recent N snapshots
+/// * [`clean_orphan_files`](ExpireSnapshotsBuilder::clean_orphan_files) - Also remove unreferenced data files
+/// * [`dry_run`](ExpireSnapshotsBuilder::dry_run) - Preview what would be deleted without actually deleting
+pub struct ExpireSnapshotsBuilder<'a> {
+    table: &'a mut Table,
+    older_than: Option<i64>,
+    retain_last: Option<usize>,
+    clean_orphan_files: bool,
+    retain_ref_snapshots: bool,
+    dry_run: bool,
+}
+
+impl<'a> ExpireSnapshotsBuilder<'a> {
+    /// Create a new snapshot expiration builder for the given table
+    fn new(table: &'a mut Table) -> Self {
+        Self {
+            table,
+            older_than: None,
+            retain_last: None,
+            clean_orphan_files: false,
+            retain_ref_snapshots: true,
+            dry_run: false,
+        }
+    }
+
+    /// Expire snapshots older than the given timestamp (in milliseconds since Unix epoch)
+    pub fn expire_older_than(mut self, timestamp_ms: i64) -> Self {
+        self.older_than = Some(timestamp_ms);
+        self
+    }
+
+    /// Retain only the most recent N snapshots, expiring all others
+    pub fn retain_last(mut self, count: usize) -> Self {
+        self.retain_last = Some(count);
+        self
+    }
+
+    /// Enable or disable cleanup of orphaned data files
+    pub fn clean_orphan_files(mut self, enabled: bool) -> Self {
+        self.clean_orphan_files = enabled;
+        self
+    }
+
+    /// Control whether snapshots referenced by branches/tags should be preserved
+    pub fn retain_ref_snapshots(mut self, enabled: bool) -> Self {
+        self.retain_ref_snapshots = enabled;
+        self
+    }
+
+    /// Enable dry run mode to preview what would be deleted without actually deleting
+    pub fn dry_run(mut self, enabled: bool) -> Self {
+        self.dry_run = enabled;
+        self
+    }
+
+    /// Execute the snapshot expiration operation
+    pub async fn execute(self) -> Result<Vec<i64>, Error> {
+        let _result = self.table.new_transaction(None)
+            .expire_snapshots(
+                self.older_than,
+                self.retain_last,
+                self.clean_orphan_files,
+                self.retain_ref_snapshots,
+                self.dry_run,
+            )
+            .commit()
+            .await?;
+
+        // Extract the expired snapshot IDs from the commit result
+        // For now, we'll need to return empty vec since the transaction commit
+        // doesn't directly return the expired snapshot IDs
+        // TODO: Enhance transaction result to include operation-specific details
+        Ok(vec![])
+    }
+}
+
 #[derive(Debug, Clone)]
 /// Iceberg table
 pub struct Table {
@@ -299,14 +378,20 @@ impl Table {
         TableTransaction::new(self, branch)
     }
 
-    /// Creates a new snapshot expiration builder for cleaning up old snapshots
+    /// Configures snapshot expiration for this table
+    ///
+    /// Returns a builder that allows configuring snapshot expiration policies:
+    /// * Time-based expiration: Remove snapshots older than a timestamp
+    /// * Count-based retention: Keep only the most recent N snapshots  
+    /// * Orphan file cleanup: Remove data files no longer referenced by any snapshot
+    /// * Reference preservation: Protect snapshots referenced by branches/tags
+    /// * Dry run mode: Preview what would be deleted without actually deleting
     ///
-    /// This method returns a builder that can be configured to expire snapshots based on
-    /// various criteria such as age, count, or reference status. The operation is atomic
-    /// and will either succeed completely or not modify the table at all.
+    /// The operation is executed through the table's transaction system, ensuring
+    /// atomicity and consistency with other table operations.
     ///
     /// # Returns
-    /// * `ExpireSnapshots` - A builder for configuring and executing snapshot expiration
+    /// * `ExpireSnapshotsBuilder` - A builder for configuring expiration parameters
     ///
     /// # Examples
     /// ```rust,no_run
@@ -319,12 +404,12 @@ impl Table {
     ///     .execute()
     ///     .await?;
     /// 
-    /// println!("Expired {} snapshots", result.expired_snapshot_ids.len());
+    /// println!("Expired {} snapshots", result.len());
     /// # Ok(())
     /// # }
     /// ```
-    pub fn expire_snapshots(&mut self) -> maintenance::ExpireSnapshots<'_> {
-        maintenance::ExpireSnapshots::new(self)
+    pub fn expire_snapshots(&mut self) -> ExpireSnapshotsBuilder<'_> {
+        ExpireSnapshotsBuilder::new(self)
     }
 }
 

iceberg-rust/src/table/maintenance/mod.rs

@@ -38,8 +38,9 @@ pub(crate) static REPLACE_INDEX: usize = 4;
 pub(crate) static OVERWRITE_INDEX: usize = 5;
 pub(crate) static UPDATE_PROPERTIES_INDEX: usize = 6;
 pub(crate) static SET_SNAPSHOT_REF_INDEX: usize = 7;
+pub(crate) static EXPIRE_SNAPSHOTS_INDEX: usize = 8;
 
-pub(crate) static NUM_OPERATIONS: usize = 8;
+pub(crate) static NUM_OPERATIONS: usize = 9;
 
 /// A transaction that can perform multiple operations on a table atomically
 ///
@@ -395,6 +396,54 @@ impl<'table> TableTransaction<'table> {
         self.operations[SET_SNAPSHOT_REF_INDEX] = Some(Operation::SetSnapshotRef(entry));
         self
     }
+
+    /// Expire snapshots based on the provided configuration
+    ///
+    /// This operation expires snapshots according to the retention policies specified.
+    /// It can expire snapshots older than a certain timestamp, retain only the most recent N snapshots,
+    /// and optionally clean up orphaned data files.
+    ///
+    /// # Arguments
+    /// * `older_than` - Optional timestamp (ms since Unix epoch) to expire snapshots older than this time
+    /// * `retain_last` - Optional number of most recent snapshots to keep, regardless of timestamp
+    /// * `clean_orphan_files` - Whether to clean up data files that are no longer referenced
+    /// * `retain_ref_snapshots` - Whether to preserve snapshots that are referenced by branches/tags
+    /// * `dry_run` - Whether to perform a dry run without actually deleting anything
+    ///
+    /// # Returns
+    /// * `Self` - The transaction builder for method chaining
+    ///
+    /// # Examples
+    /// ```
+    /// let result = table.new_transaction(None)
+    ///     .expire_snapshots(
+    ///         Some(chrono::Utc::now().timestamp_millis() - 7 * 24 * 60 * 60 * 1000),
+    ///         Some(5),
+    ///         true,
+    ///         true,
+    ///         false
+    ///     )
+    ///     .commit()
+    ///     .await?;
+    /// ```
+    pub fn expire_snapshots(
+        mut self,
+        older_than: Option<i64>,
+        retain_last: Option<usize>,
+        clean_orphan_files: bool,
+        retain_ref_snapshots: bool,
+        dry_run: bool,
+    ) -> Self {
+        self.operations[EXPIRE_SNAPSHOTS_INDEX] = Some(Operation::ExpireSnapshots {
+            older_than,
+            retain_last,
+            clean_orphan_files,
+            retain_ref_snapshots,
+            dry_run,
+        });
+        self
+    }
+
     /// Commits all operations in this transaction atomically
     ///
     /// This method executes all operations in the transaction and updates the table