refactor: consolidate snapshot expiration into MaintenanceTable

- Move ExpireSnapshots functionality from standalone class to MaintenanceTable
- Replace fluent API (table.expire_snapshots().method().commit()) with direct execution (table.maintenance.method())
- Remove ExpireSnapshots class and integrate logic into maintenance operations
- Update all tests to use new unified maintenance API
- Maintain all existing validation and protection logic for snapshots

This change consolidates table maintenance operations under a single interface
and simplifies the API by removing the need for explicit commit calls.

Breaking change: table.expire_snapshots() API is replaced with table.maintenance.expire_*() methods

Author: ForeverAngry

pyiceberg/table/maintenance.py

@@ -28,6 +28,7 @@
 
 if TYPE_CHECKING:
     from pyiceberg.table import Table
+    from pyiceberg.table.metadata import TableMetadata
 
 
 class MaintenanceTable:
@@ -139,6 +140,7 @@ def expire_snapshot_by_id(self, snapshot_id: int) -> None:
 
             # Remove the snapshot
             from pyiceberg.table.update import RemoveSnapshotsUpdate
+
             txn._apply((RemoveSnapshotsUpdate(snapshot_ids=[snapshot_id]),))
 
     def expire_snapshots_by_ids(self, snapshot_ids: List[int]) -> None:
@@ -152,7 +154,7 @@ def expire_snapshots_by_ids(self, snapshot_ids: List[int]) -> None:
         """
         with self.tbl.transaction() as txn:
             protected_ids = self._get_protected_snapshot_ids(txn.table_metadata)
-            
+
             # Validate all snapshots before expiring any
             for snapshot_id in snapshot_ids:
                 if txn.table_metadata.snapshot_by_id(snapshot_id) is None:
@@ -162,6 +164,7 @@ def expire_snapshots_by_ids(self, snapshot_ids: List[int]) -> None:
 
             # Remove all snapshots
             from pyiceberg.table.update import RemoveSnapshotsUpdate
+
             txn._apply((RemoveSnapshotsUpdate(snapshot_ids=snapshot_ids),))
 
     def expire_snapshots_older_than(self, timestamp_ms: int) -> None:
@@ -173,17 +176,18 @@ def expire_snapshots_older_than(self, timestamp_ms: int) -> None:
         # First check if there are any snapshots to expire to avoid unnecessary transactions
         protected_ids = self._get_protected_snapshot_ids(self.tbl.metadata)
         snapshots_to_expire = []
-        
+
         for snapshot in self.tbl.metadata.snapshots:
             if snapshot.timestamp_ms < timestamp_ms and snapshot.snapshot_id not in protected_ids:
                 snapshots_to_expire.append(snapshot.snapshot_id)
-        
+
         if snapshots_to_expire:
             with self.tbl.transaction() as txn:
                 from pyiceberg.table.update import RemoveSnapshotsUpdate
+
                 txn._apply((RemoveSnapshotsUpdate(snapshot_ids=snapshots_to_expire),))
 
-    def _get_protected_snapshot_ids(self, table_metadata) -> Set[int]:
+    def _get_protected_snapshot_ids(self, table_metadata: TableMetadata) -> Set[int]:
         """Get the IDs of protected snapshots.
 
         These are the HEAD snapshots of all branches and all tagged snapshots.
@@ -196,7 +200,7 @@ def _get_protected_snapshot_ids(self, table_metadata) -> Set[int]:
             Set of protected snapshot IDs to exclude from expiration.
         """
         from pyiceberg.table.refs import SnapshotRefType
-        
+
         protected_ids: Set[int] = set()
         for ref in table_metadata.refs.values():
             if ref.snapshot_ref_type in [SnapshotRefType.TAG, SnapshotRefType.BRANCH]:

pyiceberg/table/update/snapshot.py

@@ -56,7 +56,6 @@
 from pyiceberg.partitioning import (
     PartitionSpec,
 )
-from pyiceberg.table.refs import SnapshotRefType
 from pyiceberg.table.snapshots import (
     Operation,
     Snapshot,
@@ -68,7 +67,6 @@
     AddSnapshotUpdate,
     AssertRefSnapshotId,
     RemoveSnapshotRefUpdate,
-    RemoveSnapshotsUpdate,
     SetSnapshotRefUpdate,
     TableRequirement,
     TableUpdate,