refactor: simplify snapshot expiration logic and clean up unused imports

Author: ForeverAngry

pyiceberg/table/maintenance.py

@@ -104,8 +104,7 @@ def expire_snapshots_older_than(self, timestamp_ms: int) -> None:
                 snapshots_to_expire.append(snapshot.snapshot_id)
 
         if snapshots_to_expire:
-            with self.tbl.transaction():
-                self.expire_snapshots_by_ids(snapshots_to_expire)
+            self.expire_snapshots_by_ids(snapshots_to_expire)
 
     def expire_snapshots_older_than_with_retention(
         self, timestamp_ms: int, retain_last_n: Optional[int] = None, min_snapshots_to_keep: Optional[int] = None
@@ -122,8 +121,7 @@ def expire_snapshots_older_than_with_retention(
         )
 
         if snapshots_to_expire:
-            with self.tbl.transaction():
-                self.expire_snapshots_by_ids(snapshots_to_expire)
+            self.expire_snapshots_by_ids(snapshots_to_expire)
 
     def retain_last_n_snapshots(self, n: int) -> None:
         """Keep only the last N snapshots, expiring all others.
@@ -158,8 +156,7 @@ def retain_last_n_snapshots(self, n: int) -> None:
                 snapshots_to_expire.append(snapshot.snapshot_id)
 
         if snapshots_to_expire:
-            with self.tbl.transaction():
-                self.expire_snapshots_by_ids(snapshots_to_expire)
+            self.expire_snapshots_by_ids(snapshots_to_expire)
 
     def _get_snapshots_to_expire_with_retention(
         self, timestamp_ms: Optional[int] = None, retain_last_n: Optional[int] = None, min_snapshots_to_keep: Optional[int] = None
@@ -215,7 +212,7 @@ def _get_snapshots_to_expire_with_retention(
 
     def expire_snapshots_with_retention_policy(
         self, timestamp_ms: Optional[int] = None, retain_last_n: Optional[int] = None, min_snapshots_to_keep: Optional[int] = None
-    ) -> List[int]:
+    ) -> None:
         """Comprehensive snapshot expiration with multiple retention strategies.
 
         This method provides a unified interface for snapshot expiration with various
@@ -265,12 +262,7 @@ def expire_snapshots_with_retention_policy(
         )
 
         if snapshots_to_expire:
-            with self.tbl.transaction() as txn:
-                from pyiceberg.table.update import RemoveSnapshotsUpdate
-
-                txn._apply((RemoveSnapshotsUpdate(snapshot_ids=snapshots_to_expire),))
-
-        return snapshots_to_expire
+            self.expire_snapshots_by_ids(snapshots_to_expire)
 
     def _get_protected_snapshot_ids(self, table_metadata: TableMetadata) -> Set[int]:
         """Get the IDs of protected snapshots.

pyiceberg/table/update/snapshot.py

@@ -68,7 +68,6 @@
     AddSnapshotUpdate,
     AssertRefSnapshotId,
     RemoveSnapshotRefUpdate,
-    RemoveSnapshotsUpdate,
     SetSnapshotRefUpdate,
     TableRequirement,
     TableUpdate,
@@ -903,4 +902,4 @@ def remove_branch(self, branch_name: str) -> ManageSnapshots:
         Returns:
             This for method chaining
         """
-        return self._remove_ref_snapshot(ref_name=branch_name)
+        return self._remove_ref_snapshot(ref_name=branch_name)

tests/table/test_dedup_data_file_filepaths.py

@@ -136,4 +136,4 @@ def test_deduplicate_data_files_removes_duplicates_in_current_snapshot(
     file_names_after: List[str] = [os.path.basename(df.file_path) for df in all_datafiles_after]
     # Only one reference should remain after deduplication
     assert file_names_after.count(dupe_data_file_path.name) == 1
-    assert all(isinstance(df, DataFile) for df in removed)
+    assert all(isinstance(df, DataFile) for df in removed)