Added ExpireSnapshots Feature

.gitignore

@@ -50,3 +50,5 @@ htmlcov
 pyiceberg/avro/decoder_fast.c
 pyiceberg/avro/*.html
 pyiceberg/avro/*.so
+.vscode/settings.json
+pyiceberg/table/update/expire_snapshot.md

pyiceberg/table/update/snapshot.py

@@ -66,7 +66,9 @@
     AddSnapshotUpdate,
     AssertRefSnapshotId,
     RemoveSnapshotRefUpdate,
+    RemoveSnapshotsUpdate,
     SetSnapshotRefUpdate,
+    TableMetadata,
     TableRequirement,
     TableUpdate,
     U,
@@ -82,7 +84,11 @@
 from pyiceberg.utils.properties import property_as_bool, property_as_int
 
 if TYPE_CHECKING:
-    from pyiceberg.table import Transaction
+    pass
+
+
+from pyiceberg.table.metadata import Snapshot, TableMetadata
+from pyiceberg.table.snapshots import Snapshot
 
 
 def _new_manifest_file_name(num: int, commit_uuid: uuid.UUID) -> str:
@@ -239,7 +245,7 @@ def _summary(self, snapshot_properties: Dict[str, str] = EMPTY_DICT) -> Summary:
             truncate_full_table=self._operation == Operation.OVERWRITE,
         )
 
-    def _commit(self) -> UpdatesAndRequirements:
+    def _commit(self, base_metadata: TableMetadata) -> UpdatesAndRequirements:
         new_manifests = self._manifests()
         next_sequence_number = self._transaction.table_metadata.next_sequence_number()
 
@@ -740,11 +746,11 @@ class ManageSnapshots(UpdateTableMetadata["ManageSnapshots"]):
        ms.create_tag(snapshot_id1, "Tag_A").create_tag(snapshot_id2, "Tag_B")
     """
 
+    _snapshot_ids_to_expire = set()
     _updates: Tuple[TableUpdate, ...] = ()
     _requirements: Tuple[TableRequirement, ...] = ()
 
     def _commit(self) -> UpdatesAndRequirements:
-        """Apply the pending changes and commit."""
         return self._updates, self._requirements
 
     def _remove_ref_snapshot(self, ref_name: str) -> ManageSnapshots:
@@ -844,3 +850,52 @@ def remove_branch(self, branch_name: str) -> ManageSnapshots:
             This for method chaining
         """
         return self._remove_ref_snapshot(ref_name=branch_name)
+
+    def _commit(self) -> UpdatesAndRequirements:
+        """
+        Commit the staged updates and requirements.
+        This will remove the snapshots with the given IDs.
+
+        Returns:
+            Tuple of updates and requirements to be committed,
+            as requried by the calling parent apply functions.
+        """
+        update = RemoveSnapshotsUpdate(snapshot_ids=self._snapshot_ids_to_expire)
+        self._updates += (update,)
+        return self._updates, self._requirements
+
+    def expire_snapshot_by_id(self, snapshot_id: int) -> ManageSnapshots:
+        """
+        Expire a snapshot by its ID.
+
+        Args:
+            snapshot_id (int): The ID of the snapshot to expire.
+
+        Returns:
+            This for method chaining.
+        """
+        if self._transaction.table_metadata.snapshot_by_id(snapshot_id) is None:
+            raise ValueError(f"Snapshot with ID {snapshot_id} does not exist.")
+        self._snapshot_ids_to_expire.add(snapshot_id)
+        return self
+
+    def expire_snapshots_older_than(self, timestamp_ms: int) -> ManageSnapshots:
+        """
+        Expire snapshots older than the given timestamp.
+
+        Args:
+            timestamp_ms (int): The timestamp in milliseconds. Snapshots older than this will be expired.
+
+        Returns:
+            This for method chaining.
+        """
+        # Collect IDs of snapshots to be expired
+        snapshots_to_remove = [
+            snapshot.snapshot_id
+            for snapshot in self._transaction.table_metadata.snapshots
+            if snapshot.timestamp_ms < timestamp_ms
+        ]
+        if snapshots_to_remove:
+            for snapshot_id in snapshots_to_remove:
+                self._snapshot_ids_to_expire.add(snapshot_id)
+        return self

pyproject.toml

@@ -84,8 +84,9 @@ pyiceberg-core = { version = "^0.4.0", optional = true }
 polars = { version = "^1.21.0", optional = true }
 thrift-sasl = { version = ">=0.4.3", optional = true }
 
+
 [tool.poetry.group.dev.dependencies]
-pytest = "7.4.4"
+pytest = ">=8.3.5,<9.0.0"
 pytest-checkdocs = "2.13.0"
 pytest-lazy-fixture = "0.6.3"
 pre-commit = "4.2.0"

tests/conftest.py

@@ -2349,7 +2349,7 @@ def table_v2_with_extensive_snapshots(example_table_metadata_v2_with_extensive_s
         identifier=("database", "table"),
         metadata=table_metadata,
         metadata_location=f"{table_metadata.location}/uuid.metadata.json",
-        io=load_file_io(),
+        io=load_file_io(location=metadata_location),
         catalog=NoopCatalog("NoopCatalog"),
     )
 

tests/table/test_expire_snapshots.py

@@ -0,0 +1,43 @@
+from unittest.mock import MagicMock
+from uuid import uuid4
+
+from pyiceberg.table import CommitTableResponse, Table
+
+
+def test_expire_snapshot(table_v2: Table) -> None:
+    EXPIRE_SNAPSHOT = 3051729675574597004
+    KEEP_SNAPSHOT = 3055729675574597004
+    # Mock the catalog's commit_table method
+    mock_response = CommitTableResponse(
+        # Use the table's current metadata but keep only the snapshot not to be expired
+        metadata=table_v2.metadata.model_copy(update={"snapshots": [KEEP_SNAPSHOT]}),
+        metadata_location="mock://metadata/location",
+        uuid=uuid4(),
+    )
+
+    # Mock the commit_table method to return the mock response
+    table_v2.catalog.commit_table = MagicMock(return_value=mock_response)
+
+    # Print snapshot IDs for debugging
+    print(f"Snapshot IDs before expiration: {[snapshot.snapshot_id for snapshot in table_v2.metadata.snapshots]}")
+
+    # Assert fixture data to validate test assumptions
+    assert len(table_v2.metadata.snapshots) == 2
+    assert len(table_v2.metadata.snapshot_log) == 2
+    assert len(table_v2.metadata.refs) == 2
+
+    # Expire the snapshot directly without using a transaction
+    try:
+        table_v2.manage_snapshots().expire_snapshot_by_id(EXPIRE_SNAPSHOT).commit()
+    except Exception as e:
+        assert False, f"Commit failed with error: {e}"
+
+    # Assert that commit_table was called once
+    table_v2.catalog.commit_table.assert_called_once()
+
+    # Assert the expired snapshot ID is no longer present
+    remaining_snapshots = table_v2.metadata.snapshots
+    assert EXPIRE_SNAPSHOT not in remaining_snapshots
+
+    # Assert the length of snapshots after expiration
+    assert len(table_v2.metadata.snapshots) == 1