### Features & Enhancements

- **Duplicate File Remediation #2130**
  - Added `deduplicate_data_files` to the `MaintenanceTable` class.
  - Enables detection and removal of duplicate data files, improving table hygiene and storage efficiency.

- **Support `retainLast` and `setMinSnapshotsToKeep` Snapshot Retention Policies #2150**
  - Added new snapshot retention methods to `MaintenanceTable` for feature parity with the Java API:
    - `retain_last_n_snapshots(n)`: Retain only the last N snapshots.
    - `expire_snapshots_older_than_with_retention(timestamp_ms, retain_last_n=None, min_snapshots_to_keep=None)`: Expire snapshots older than a timestamp, with additional retention constraints.
    - `expire_snapshots_with_retention_policy(timestamp_ms=None, retain_last_n=None, min_snapshots_to_keep=None)`: Unified retention policy supporting time-based and count-based constraints.
  - All retention logic respects protected snapshots (branches/tags) and includes guardrails to prevent over-aggressive expiration.

### Bug Fixes & Cleanups

- **Remove unrelated instance variable from the `ManageSnapshots` class #2151**
  - Removed an errant member variable from the `ManageSnapshots` class, aligning the implementation with the intended design and the Java reference.

### Testing & Documentation

- Consolidated all snapshot expiration and retention tests into a single file (`test_retention_strategies.py`), covering:
  - Basic expiration by ID and timestamp.
  - Protection of branch/tag snapshots.
  - Retention guardrails and combined policies.
  - Deduplication of data files.
- Added and updated documentation to describe all new retention strategies, deduplication, and API parity improvements.

Author: ForeverAngry

pyiceberg/table/__init__.py

@@ -116,7 +116,7 @@
     update_table_metadata,
 )
 from pyiceberg.table.update.schema import UpdateSchema
-from pyiceberg.table.update.snapshot import ExpireSnapshots, ManageSnapshots, UpdateSnapshot, _FastAppendFiles
+from pyiceberg.table.update.snapshot import ManageSnapshots, UpdateSnapshot, _FastAppendFiles
 from pyiceberg.table.update.spec import UpdateSpec
 from pyiceberg.table.update.statistics import UpdateStatistics
 from pyiceberg.transforms import IdentityTransform

pyiceberg/table/maintenance.py

@@ -17,7 +17,6 @@
 from __future__ import annotations
 
 import logging
-from concurrent.futures import ThreadPoolExecutor
 from typing import TYPE_CHECKING, List, Optional, Set, Union
 
 from pyiceberg.manifest import DataFile

test_retention_strategies.py

@@ -1,26 +1,138 @@
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements.  See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership.  The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License.  You may obtain a copy of the License at
-#
-#   http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing,
-# software distributed under the License is distributed on an
-# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-# KIND, either express or implied.  See the License for the
-# specific language governing permissions and limitations
-# under the License.
-from unittest.mock import MagicMock
-from uuid import uuid4
+#!/usr/bin/env python3
+"""
+Test script to validate the retention strategies implementation in MaintenanceTable.
+
+This script demonstrates the new retention features:
+1. retain_last_n_snapshots() - Keep only the last N snapshots
+2. expire_snapshots_older_than_with_retention() - Time-based expiration with retention constraints
+3. expire_snapshots_with_retention_policy() - Comprehensive retention policy
+"""
+
+# Example usage (commented out since we don't have an actual table)
+"""
+from pyiceberg.table.maintenance import MaintenanceTable
+from pyiceberg.table import Table
+
+# Assume we have a table instance
+table = Table(...)  # Initialize your table
+maintenance = MaintenanceTable(table)
+
+# Example 1: Keep only the last 5 snapshots regardless of age
+# This is helpful when regular snapshot creation occurs and users always want 
+# to keep the last few for rollback
+maintenance.retain_last_n_snapshots(5)
+
+# Example 2: Expire snapshots older than a timestamp but keep at least 3 total
+# This acts as a guardrail to prevent aggressive expiration logic from removing too many snapshots
+import time
+one_week_ago = int((time.time() - 7 * 24 * 60 * 60) * 1000)  # 7 days ago in milliseconds
+maintenance.expire_snapshots_older_than_with_retention(
+    timestamp_ms=one_week_ago,
+    min_snapshots_to_keep=3
+)
+
+# Example 3: Combined policy - expire old snapshots but keep last 10 and at least 5 total
+# This provides comprehensive control combining both strategies
+maintenance.expire_snapshots_with_retention_policy(
+    timestamp_ms=one_week_ago,
+    retain_last_n=10,
+    min_snapshots_to_keep=5
+)
+
+# Example 4: Just keep the last 20 snapshots (no time constraint)
+expired_ids = maintenance.expire_snapshots_with_retention_policy(retain_last_n=20)
+print(f"Expired {len(expired_ids)} snapshots")
+"""
 
 import pytest
-
+from unittest.mock import MagicMock
+from uuid import uuid4
+from types import SimpleNamespace
 from pyiceberg.table import CommitTableResponse
 
+def _make_snapshots(ids_and_timestamps):
+    return [SimpleNamespace(snapshot_id=sid, timestamp_ms=ts, parent_snapshot_id=None) for sid, ts in ids_and_timestamps]
+
+def test_retain_last_n_snapshots(table_v2):
+    # Setup: 5 snapshots, keep last 3
+    ids_and_ts = [
+        (1, 1000),
+        (2, 2000),
+        (3, 3000),
+        (4, 4000),
+        (5, 5000),
+    ]
+    snapshots = _make_snapshots(ids_and_ts)
+    table_v2.metadata = table_v2.metadata.model_copy(update={"snapshots": snapshots, "refs": {}})
+    table_v2.catalog = MagicMock()
+    # Simulate commit response with only last 3 snapshots
+    keep_ids = [3, 4, 5]
+    mock_response = CommitTableResponse(
+        metadata=table_v2.metadata.model_copy(update={"snapshots": [s for s in snapshots if s.snapshot_id in keep_ids]}),
+        metadata_location="mock://metadata/location",
+        uuid=uuid4(),
+    )
+    table_v2.catalog.commit_table.return_value = mock_response
+    table_v2.maintenance.retain_last_n_snapshots(3)
+    table_v2.catalog.commit_table.assert_called_once()
+    # Update metadata to reflect commit
+    table_v2.metadata = mock_response.metadata
+    remaining_ids = {s.snapshot_id for s in table_v2.metadata.snapshots}
+    assert remaining_ids == set(keep_ids)
+
+def test_min_snapshots_to_keep(table_v2):
+    # Setup: 5 snapshots, expire all older than 4500, but keep at least 3
+    ids_and_ts = [
+        (1, 1000),
+        (2, 2000),
+        (3, 3000),
+        (4, 4000),
+        (5, 5000),
+    ]
+    snapshots = _make_snapshots(ids_and_ts)
+    table_v2.metadata = table_v2.metadata.model_copy(update={"snapshots": snapshots, "refs": {}})
+    table_v2.catalog = MagicMock()
+    # Only 1,2 should be expired (to keep 3 total)
+    keep_ids = [3, 4, 5]
+    mock_response = CommitTableResponse(
+        metadata=table_v2.metadata.model_copy(update={"snapshots": [s for s in snapshots if s.snapshot_id in keep_ids]}),
+        metadata_location="mock://metadata/location",
+        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.catalog.commit_table.assert_called_once()
+    table_v2.metadata = mock_response.metadata
+    remaining_ids = {s.snapshot_id for s in table_v2.metadata.snapshots}
+    assert remaining_ids == set(keep_ids)
+
+def test_combined_constraints(table_v2):
+    # Setup: 5 snapshots, expire all older than 3500, keep last 2, min 4 total
+    ids_and_ts = [
+        (1, 1000),
+        (2, 2000),
+        (3, 3000),
+        (4, 4000),
+        (5, 5000),
+    ]
+    snapshots = _make_snapshots(ids_and_ts)
+    table_v2.metadata = table_v2.metadata.model_copy(update={"snapshots": snapshots, "refs": {}})
+    table_v2.catalog = MagicMock()
+    # Only 1 should be expired (to keep last 2 and min 4 total)
+    keep_ids = [2, 3, 4, 5]
+    mock_response = CommitTableResponse(
+        metadata=table_v2.metadata.model_copy(update={"snapshots": [s for s in snapshots if s.snapshot_id in keep_ids]}),
+        metadata_location="mock://metadata/location",
+        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.catalog.commit_table.assert_called_once()
+    table_v2.metadata = mock_response.metadata
+    remaining_ids = {s.snapshot_id for s in table_v2.metadata.snapshots}
+    assert remaining_ids == set(keep_ids)
 
 def test_cannot_expire_protected_head_snapshot(table_v2) -> None:
     """Test that a HEAD (branch) snapshot cannot be expired."""
@@ -47,7 +159,6 @@ def test_cannot_expire_protected_head_snapshot(table_v2) -> None:
 
     table_v2.catalog.commit_table.assert_not_called()
 
-
 def test_cannot_expire_tagged_snapshot(table_v2) -> None:
     """Test that a tagged snapshot cannot be expired."""
     TAGGED_SNAPSHOT = 3051729675574597004
@@ -70,7 +181,6 @@ def test_cannot_expire_tagged_snapshot(table_v2) -> None:
 
     table_v2.catalog.commit_table.assert_not_called()
 
-
 def test_expire_unprotected_snapshot(table_v2) -> None:
     """Test that an unprotected snapshot can be expired."""
     EXPIRE_SNAPSHOT = 3051729675574597004
@@ -105,7 +215,6 @@ def test_expire_unprotected_snapshot(table_v2) -> None:
     assert EXPIRE_SNAPSHOT not in remaining_snapshots
     assert len(table_v2.metadata.snapshots) == 1
 
-
 def test_expire_nonexistent_snapshot_raises(table_v2) -> None:
     """Test that trying to expire a non-existent snapshot raises an error."""
     NONEXISTENT_SNAPSHOT = 9999999999999999999
@@ -118,7 +227,6 @@ def test_expire_nonexistent_snapshot_raises(table_v2) -> None:
 
     table_v2.catalog.commit_table.assert_not_called()
 
-
 def test_expire_snapshots_by_timestamp_skips_protected(table_v2) -> None:
     # Setup: two snapshots; both are old, but one is head/tag protected
     HEAD_SNAPSHOT = 3051729675574597004
@@ -169,7 +277,6 @@ def test_expire_snapshots_by_timestamp_skips_protected(table_v2) -> None:
     assert remove_update is not None
     assert remove_update.snapshot_ids == []
 
-
 def test_expire_snapshots_by_ids(table_v2) -> None:
     """Test that multiple unprotected snapshots can be expired by IDs."""
     EXPIRE_SNAPSHOT_1 = 3051729675574597004
@@ -221,4 +328,4 @@ def test_expire_snapshots_by_ids(table_v2) -> None:
     remaining_snapshots = table_v2.metadata.snapshots
     assert EXPIRE_SNAPSHOT_1 not in remaining_snapshots
     assert EXPIRE_SNAPSHOT_2 not in remaining_snapshots
-    assert len(table_v2.metadata.snapshots) == 1
+    assert len(table_v2.metadata.snapshots) == 1