feat(iceberg): introduce remove snapshot action (#21)

* feat(iceberg): basic remove snapshot

* feat(iceberg): introduce new properties for remove snapshots

* feat(iceberg): support remove schemas

* refactor(iceberg): refactor file org

* address comments

* refactor(iceberg): refactor and ut

* fix(iceberg): fix integration-test

Author: Li0k

Cargo.lock

@@ -175,7 +175,6 @@ impl Snapshot {
     }
 
     /// Get parent snapshot.
-    #[cfg(test)]
     pub(crate) fn parent_snapshot(&self, table_metadata: &TableMetadata) -> Option<SnapshotRef> {
         match self.parent_snapshot_id {
             Some(id) => table_metadata.snapshot_by_id(id).cloned(),
@@ -401,6 +400,33 @@ impl SnapshotRetention {
     }
 }
 
+/// An iterator over the ancestors of a snapshot.
+pub struct AncestorIterator<'a> {
+    current: Option<SnapshotRef>,
+    table_metadata: &'a TableMetadata,
+}
+
+impl<'a> Iterator for AncestorIterator<'a> {
+    type Item = SnapshotRef;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        let current = self.current.take()?;
+
+        let next = current.parent_snapshot(self.table_metadata);
+        self.current = next;
+
+        Some(current)
+    }
+}
+
+/// Returns an iterator over the ancestors of a snapshot.
+pub fn ancestors_of(snapshot: SnapshotRef, table_metadata: &TableMetadata) -> AncestorIterator<'_> {
+    AncestorIterator {
+        current: Some(snapshot),
+        table_metadata,
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use std::collections::HashMap;

crates/iceberg/src/spec/snapshot.rs

@@ -78,6 +78,19 @@ pub const PROPERTY_METADATA_PREVIOUS_VERSIONS_MAX: &str = "write.metadata.previo
 /// Default value for max number of previous versions to keep.
 pub const PROPERTY_METADATA_PREVIOUS_VERSIONS_MAX_DEFAULT: usize = 100;
 
+/// Property key for max snapshot age in milliseconds.
+pub const MAX_SNAPSHOT_AGE_MS: &str = "history.expire.max-snapshot-age-ms";
+/// Default value for max snapshot age in milliseconds.
+pub const MAX_SNAPSHOT_AGE_MS_DEFAULT: i64 = 5 * 24 * 60 * 60 * 1000; // 5 days
+/// Property key for min snapshots to keep.
+pub const MIN_SNAPSHOTS_TO_KEEP: &str = "history.expire.min-snapshots-to-keep";
+/// Default value for min snapshots to keep.
+pub const MIN_SNAPSHOTS_TO_KEEP_DEFAULT: i32 = 1;
+/// Property key for max reference age in milliseconds.
+pub const MAX_REF_AGE_MS: &str = "history.expire.max-ref-age-ms";
+/// Default value for max reference age in milliseconds.
+pub const MAX_REF_AGE_MS_DEFAULT: i64 = i64::MAX;
+
 /// Reserved Iceberg table properties list.
 ///
 /// Reserved table properties are only used to control behaviors when creating or updating a

crates/iceberg/src/spec/table_metadata.rs

@@ -18,6 +18,7 @@
 //! This module contains transaction api.
 
 mod append;
+pub mod remove_snapshots;
 mod snapshot;
 mod sort_order;
 
@@ -26,6 +27,7 @@ use std::collections::HashMap;
 use std::mem::discriminant;
 use std::sync::Arc;
 
+use remove_snapshots::RemoveSnapshotAction;
 use uuid::Uuid;
 
 use crate::error::Result;
@@ -190,6 +192,11 @@ impl<'a> Transaction<'a> {
         }
     }
 
+    /// Creates remove snapshot action.
+    pub fn expire_snapshot(self) -> RemoveSnapshotAction<'a> {
+        RemoveSnapshotAction::new(self)
+    }
+
     /// Remove properties in table.
     pub fn remove_properties(mut self, keys: Vec<String>) -> Result<Self> {
         self.apply(