refactor: introduce ExpireSnapshots builder for snapshot expiration

Add ExpireSnapshots builder class to consolidate snapshot expiration
operations under a fluent API. This change moves from direct method
calls on MaintenanceTable to a builder pattern accessed via
maintenance.expire_snapshots().

Key changes:
- Add ExpireSnapshots class with fluent API methods
- Migrate existing snapshot expiration functionality to builder pattern
- Update API documentation examples to use new builder syntax
- Update tests to use new expire_snapshots() builder methods
- Maintain backward compatibility in MaintenanceTable implementation

This provides a more intuitive and JAVA aligned API as suggested by @kevinjqliu and @aammar5

Author: ForeverAngry

mkdocs/docs/api.md

@@ -1089,7 +1089,7 @@ Iceberg tables accumulate snapshots over time. Retaining too many can waste stor
 from pyiceberg.table.maintenance import MaintenanceTable
 
 maintenance = MaintenanceTable(table)
-maintenance.retain_last_n_snapshots(5)
+maintenance.expire_snapshots().retain_last_n(5)
 ```
 
 #### Example: Expire snapshots older than 30 days, but keep at least 3
@@ -1100,7 +1100,7 @@ from pyiceberg.table.maintenance import MaintenanceTable
 
 maintenance = MaintenanceTable(table)
 thirty_days_ago = int((time.time() - 30 * 24 * 60 * 60) * 1000)
-maintenance.expire_snapshots_with_retention_policy(
+maintenance.expire_snapshots().with_retention_policy(
     timestamp_ms=thirty_days_ago,
     min_snapshots_to_keep=3
 )
@@ -1209,7 +1209,7 @@ PyIceberg makes it easy to filter out data from a huge table and pull it into a
 
 ```python
 # Expire old snapshots, but always keep last 10 and at least 5 total
-maintenance.expire_snapshots_with_retention_policy(
+maintenance.expire_snapshots().with_retention_policy(
     timestamp_ms=thirty_days_ago,
     retain_last_n=10,
     min_snapshots_to_keep=5

pyiceberg/table/maintenance.py

@@ -36,6 +36,14 @@ class MaintenanceTable:
     def __init__(self, tbl: Table) -> None:
         self.tbl = tbl
 
+    def expire_snapshots(self) -> "ExpireSnapshots":
+        """Return an ExpireSnapshots builder for snapshot expiration operations.
+
+        Returns:
+            ExpireSnapshots builder for configuring and executing snapshot expiration.
+        """
+        return ExpireSnapshots(self.tbl)
+
     def expire_snapshot_by_id(self, snapshot_id: int) -> None:
         """Expire a single snapshot by its ID.
 
@@ -332,3 +340,107 @@ def _get_expiration_properties(self) -> tuple[Optional[int], Optional[int], Opti
         max_ref_age_ms = int(max_ref_age) if max_ref_age is not None else None
 
         return max_snapshot_age, min_snapshots_to_keep, max_ref_age_ms
+
+
+class ExpireSnapshots:
+    """Builder for snapshot expiration operations."""
+
+    def __init__(self, tbl: Table) -> None:
+        self.tbl = tbl
+        self.maintenance = MaintenanceTable(tbl)
+
+    def by_id(self, snapshot_id: int) -> None:
+        """Expire a single snapshot by its ID.
+
+        Args:
+            snapshot_id: The ID of the snapshot to expire.
+
+        Raises:
+            ValueError: If the snapshot does not exist or is protected.
+        """
+        self.maintenance.expire_snapshot_by_id(snapshot_id)
+
+    def by_ids(self, snapshot_ids: List[int]) -> None:
+        """Expire multiple snapshots by their IDs.
+
+        Args:
+            snapshot_ids: List of snapshot IDs to expire.
+
+        Raises:
+            ValueError: If any snapshot does not exist or is protected.
+        """
+        self.maintenance._expire_snapshots_by_ids(snapshot_ids)
+
+    def older_than(self, timestamp_ms: int) -> None:
+        """Expire all unprotected snapshots with a timestamp older than a given value.
+
+        Args:
+            timestamp_ms: Only snapshots with timestamp_ms < this value will be expired.
+        """
+        self.maintenance.expire_snapshots_older_than(timestamp_ms)
+
+    def older_than_with_retention(
+        self, timestamp_ms: int, retain_last_n: Optional[int] = None, min_snapshots_to_keep: Optional[int] = None
+    ) -> None:
+        """Expire all unprotected snapshots with a timestamp older than a given value, with retention strategies.
+
+        Args:
+            timestamp_ms: Only snapshots with timestamp_ms < this value will be expired.
+            retain_last_n: Always keep the last N snapshots regardless of age.
+            min_snapshots_to_keep: Minimum number of snapshots to keep in total.
+        """
+        self.maintenance.expire_snapshots_older_than_with_retention(
+            timestamp_ms=timestamp_ms, retain_last_n=retain_last_n, min_snapshots_to_keep=min_snapshots_to_keep
+        )
+
+    def with_retention_policy(
+        self, timestamp_ms: Optional[int] = None, retain_last_n: Optional[int] = None, min_snapshots_to_keep: Optional[int] = None
+    ) -> None:
+        """Comprehensive snapshot expiration with multiple retention strategies.
+
+        This method provides a unified interface for snapshot expiration with various
+        retention policies to ensure operational resilience while allowing space reclamation.
+
+        The method will use table properties as defaults if they are set:
+        - history.expire.max-snapshot-age-ms: Default for timestamp_ms if not provided
+        - history.expire.min-snapshots-to-keep: Default for min_snapshots_to_keep if not provided
+        - history.expire.max-ref-age-ms: Used for ref expiration (branches/tags)
+
+        Args:
+            timestamp_ms: Only snapshots with timestamp_ms < this value will be considered for expiration.
+                         If None, will use history.expire.max-snapshot-age-ms table property if set.
+            retain_last_n: Always keep the last N snapshots regardless of age.
+                          Useful when regular snapshot creation occurs and users want to keep
+                          the last few for rollback purposes.
+            min_snapshots_to_keep: Minimum number of snapshots to keep in total.
+                                 Acts as a guardrail to prevent aggressive expiration logic.
+                                 If None, will use history.expire.min-snapshots-to-keep table property if set.
+
+        Raises:
+            ValueError: If retain_last_n or min_snapshots_to_keep is less than 1.
+
+        Examples:
+            # Use table property defaults
+            table.maintenance().expire_snapshots().with_retention_policy()
+
+            # Override defaults with explicit values
+            table.maintenance().expire_snapshots().with_retention_policy(
+                timestamp_ms=1234567890000,
+                retain_last_n=10,
+                min_snapshots_to_keep=5
+            )
+        """
+        self.maintenance.expire_snapshots_with_retention_policy(
+            timestamp_ms=timestamp_ms, retain_last_n=retain_last_n, min_snapshots_to_keep=min_snapshots_to_keep
+        )
+
+    def retain_last_n(self, n: int) -> None:
+        """Keep only the last N snapshots, expiring all others.
+
+        Args:
+            n: Number of most recent snapshots to keep.
+
+        Raises:
+            ValueError: If n is less than 1.
+        """
+        self.maintenance.retain_last_n_snapshots(n)

tests/table/test_retention_strategies.py

@@ -52,7 +52,7 @@ def test_retain_last_n_snapshots(table_v2: Table) -> None:
             uuid=uuid4(),
         )
         table_v2.catalog.commit_table.return_value = mock_response
-        table_v2.maintenance.retain_last_n_snapshots(3)
+        table_v2.maintenance.expire_snapshots().retain_last_n(3)
         table_v2.catalog.commit_table.assert_called_once()
         # Update metadata to reflect commit
         table_v2.metadata = mock_response.metadata
@@ -93,7 +93,7 @@ def test_min_snapshots_to_keep(table_v2: Table) -> None:
             uuid=uuid4(),
         )
         table_v2.catalog.commit_table.return_value = mock_response
-        table_v2.maintenance.expire_snapshots_older_than_with_retention(timestamp_ms=4500, min_snapshots_to_keep=3)
+        table_v2.maintenance.expire_snapshots().older_than_with_retention(timestamp_ms=4500, min_snapshots_to_keep=3)
         table_v2.catalog.commit_table.assert_called_once()
         table_v2.metadata = mock_response.metadata
         remaining_ids = {s.snapshot_id for s in table_v2.metadata.snapshots}
@@ -128,7 +128,7 @@ def test_combined_constraints(table_v2: Table) -> None:
         uuid=uuid4(),
     )
     table_v2.catalog.commit_table.return_value = mock_response
-    table_v2.maintenance.expire_snapshots_with_retention_policy(timestamp_ms=3500, retain_last_n=2, min_snapshots_to_keep=4)
+    table_v2.maintenance.expire_snapshots().with_retention_policy(timestamp_ms=3500, retain_last_n=2, min_snapshots_to_keep=4)
     table_v2.catalog.commit_table.assert_called_once()
     table_v2.metadata = mock_response.metadata
     remaining_ids = {s.snapshot_id for s in table_v2.metadata.snapshots}
@@ -156,7 +156,7 @@ def test_cannot_expire_protected_head_snapshot(table_v2: Table) -> None:
 
     # Attempt to expire the HEAD snapshot and expect a ValueError
     with pytest.raises(ValueError, match=f"Snapshot with ID {HEAD_SNAPSHOT} is protected and cannot be expired."):
-        table_v2.maintenance.expire_snapshot_by_id(HEAD_SNAPSHOT)
+        table_v2.maintenance.expire_snapshots().by_id(HEAD_SNAPSHOT)
 
     table_v2.catalog.commit_table.assert_not_called()
 
@@ -179,7 +179,7 @@ def test_cannot_expire_tagged_snapshot(table_v2: Table) -> None:
     assert any(ref.snapshot_id == TAGGED_SNAPSHOT for ref in table_v2.metadata.refs.values())
 
     with pytest.raises(ValueError, match=f"Snapshot with ID {TAGGED_SNAPSHOT} is protected and cannot be expired."):
-        table_v2.maintenance.expire_snapshot_by_id(TAGGED_SNAPSHOT)
+        table_v2.maintenance.expire_snapshots().by_id(TAGGED_SNAPSHOT)
 
     table_v2.catalog.commit_table.assert_not_called()
 
@@ -211,7 +211,7 @@ def test_expire_unprotected_snapshot(table_v2: Table) -> None:
     assert all(ref.snapshot_id != EXPIRE_SNAPSHOT for ref in table_v2.metadata.refs.values())
 
     # Expire the snapshot
-    table_v2.maintenance.expire_snapshot_by_id(EXPIRE_SNAPSHOT)
+    table_v2.maintenance.expire_snapshots().by_id(EXPIRE_SNAPSHOT)
 
     table_v2.catalog.commit_table.assert_called_once()
     remaining_snapshots = table_v2.metadata.snapshots
@@ -227,7 +227,7 @@ def test_expire_nonexistent_snapshot_raises(table_v2: Table) -> None:
     table_v2.metadata = table_v2.metadata.model_copy(update={"refs": {}})
 
     with pytest.raises(ValueError, match=f"Snapshot with ID {NONEXISTENT_SNAPSHOT} does not exist."):
-        table_v2.maintenance.expire_snapshot_by_id(NONEXISTENT_SNAPSHOT)
+        table_v2.maintenance.expire_snapshots().by_id(NONEXISTENT_SNAPSHOT)
 
     table_v2.catalog.commit_table.assert_not_called()
 
@@ -265,7 +265,7 @@ def test_expire_snapshots_by_timestamp_skips_protected(table_v2: Table) -> None:
     )
     table_v2.catalog.commit_table.return_value = mock_response
 
-    table_v2.maintenance.expire_snapshots_older_than(future_timestamp)
+    table_v2.maintenance.expire_snapshots().older_than(future_timestamp)
 
     # Both protected snapshots should remain
     remaining_ids = {s.snapshot_id for s in table_v2.metadata.snapshots}
@@ -326,7 +326,7 @@ def test_expire_snapshots_by_ids(table_v2: Table) -> None:
         assert all(ref.snapshot_id not in (EXPIRE_SNAPSHOT_1, EXPIRE_SNAPSHOT_2) for ref in table_v2.metadata.refs.values())
 
         # Expire the snapshots
-        table_v2.maintenance._expire_snapshots_by_ids([EXPIRE_SNAPSHOT_1, EXPIRE_SNAPSHOT_2])
+        table_v2.maintenance.expire_snapshots().by_ids([EXPIRE_SNAPSHOT_1, EXPIRE_SNAPSHOT_2])
 
         table_v2.catalog.commit_table.assert_called_once()
         remaining_snapshots = table_v2.metadata.snapshots
@@ -380,7 +380,7 @@ def test_expire_snapshots_with_table_property_defaults(table_v2: Table) -> None:
     table_v2.catalog.commit_table.return_value = mock_response
 
     # Call expire without explicit parameters - should use table properties
-    table_v2.maintenance.expire_snapshots_with_retention_policy()
+    table_v2.maintenance.expire_snapshots().with_retention_policy()
 
     table_v2.catalog.commit_table.assert_called_once()
     table_v2.metadata = mock_response.metadata
@@ -425,7 +425,7 @@ def test_explicit_parameters_override_table_properties(table_v2: Table) -> None:
     table_v2.catalog.commit_table.return_value = mock_response
 
     # Call expire with explicit parameters that should override the properties
-    table_v2.maintenance.expire_snapshots_with_retention_policy(
+    table_v2.maintenance.expire_snapshots().with_retention_policy(
         timestamp_ms=1500,  # Only expire snapshots older than this
         min_snapshots_to_keep=4,  # Keep at least 4 snapshots (overrides property of 2)
     )
@@ -458,7 +458,7 @@ def test_expire_snapshots_no_properties_no_parameters(table_v2: Table) -> None:
     table_v2.catalog = MagicMock()
 
     # Call expire with no parameters
-    table_v2.maintenance.expire_snapshots_with_retention_policy()
+    table_v2.maintenance.expire_snapshots().with_retention_policy()
 
     # Should not attempt to expire anything since no criteria were provided
     table_v2.catalog.commit_table.assert_not_called()