feat: add snapshot expiration methods with retention strategies

mkdocs/docs/api.md

@@ -1376,6 +1376,82 @@ def cleanup_old_snapshots(table_name: str, snapshot_ids: list[int]):
 cleanup_old_snapshots("analytics.user_events", [12345, 67890, 11111])
 ```
 
+#### Advanced Retention Strategies
+
+PyIceberg provides additional retention helpers on `ExpireSnapshots` to balance safety and cleanup:
+
+Key table properties used as defaults (all optional):
+
+- `history.expire.max-snapshot-age-ms`: Default age threshold for `with_retention_policy`
+- `history.expire.min-snapshots-to-keep`: Minimum total snapshots to retain
+- `history.expire.max-ref-age-ms`: (Reserved for future protected ref/branch cleanup logic)
+
+Protected snapshots (referenced by branches or tags) are never expired by these APIs.
+
+Keep only the last N snapshots (plus protected):
+
+```python
+table.maintenance.expire_snapshots().retain_last_n(5).commit()
+```
+
+Expire older snapshots but always keep the most recent N and a safety floor:
+
+```python
+from datetime import datetime, timedelta
+
+cutoff = int((datetime.now() - timedelta(days=7)).timestamp() * 1000)
+table.maintenance.expire_snapshots().older_than_with_retention(
+    timestamp_ms=cutoff,
+    retain_last_n=3,
+    min_snapshots_to_keep=4,
+).commit()
+```
+
+Unified policy that also reads table property defaults:
+
+```python
+# Uses table properties if arguments omitted
+table.maintenance.expire_snapshots().with_retention_policy().commit()
+
+# Override selectively
+table.maintenance.expire_snapshots().with_retention_policy(
+    retain_last_n=2,          # keep 2 newest regardless of age
+    min_snapshots_to_keep=5,  # never go below 5 total
+    # timestamp_ms omitted -> falls back to history.expire.max-snapshot-age-ms if set
+).commit()
+```
+
+##### Using a context manager
+
+You can use a context manager to automatically commit on successful exit (and skip commit if an exception occurs):
+
+```python
+# Keep the 3 newest snapshots (plus protected refs) and enforce a floor of 8 total
+with table.maintenance.expire_snapshots() as expire:
+    expire.with_retention_policy(retain_last_n=3, min_snapshots_to_keep=8)
+
+# Only keep the last 5 snapshots
+with table.maintenance.expire_snapshots() as expire:
+    expire.retain_last_n(5)
+
+# Combine explicit cutoff with other guards
+from datetime import datetime, timedelta
+cutoff = int((datetime.utcnow() - timedelta(days=14)).timestamp() * 1000)
+
+with table.maintenance.expire_snapshots() as expire:
+    expire.older_than_with_retention(timestamp_ms=cutoff, retain_last_n=2, min_snapshots_to_keep=6)
+```
+
+Parameter interaction rules:
+
+- `retain_last_n` snapshots are always kept (plus protected refs)
+- `timestamp_ms` filters candidates (older than only)
+- `min_snapshots_to_keep` stops expiration once the floor would be violated
+- If all of (`timestamp_ms`, `retain_last_n`, `min_snapshots_to_keep`) are None in `with_retention_policy`, nothing is expired
+- Passing invalid values (`< 1`) for counts raises `ValueError`
+
+Safety tip: Start with higher `min_snapshots_to_keep` when first enabling automated cleanup.
+
 ## Views
 
 PyIceberg supports view operations.

pyiceberg/table/update/snapshot.py

@@ -1005,3 +1005,183 @@ def older_than(self, dt: datetime) -> "ExpireSnapshots":
             if snapshot.timestamp_ms < expire_from and snapshot.snapshot_id not in protected_ids:
                 self._snapshot_ids_to_expire.add(snapshot.snapshot_id)
         return self
+
+    def older_than_with_retention(
+        self, timestamp_ms: int, retain_last_n: Optional[int] = None, min_snapshots_to_keep: Optional[int] = None
+    ) -> ExpireSnapshots:
+        """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.
+
+        Returns:
+            This for method chaining.
+        """
+        snapshots_to_expire = self._get_snapshots_to_expire_with_retention(
+            timestamp_ms=timestamp_ms, retain_last_n=retain_last_n, min_snapshots_to_keep=min_snapshots_to_keep
+        )
+        self._snapshot_ids_to_expire.update(snapshots_to_expire)
+        return self
+
+    def with_retention_policy(
+        self, timestamp_ms: Optional[int] = None, retain_last_n: Optional[int] = None, min_snapshots_to_keep: Optional[int] = None
+    ) -> ExpireSnapshots:
+        """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.
+
+        Returns:
+            This for method chaining.
+
+        Raises:
+            ValueError: If retain_last_n or min_snapshots_to_keep is less than 1.
+
+        Examples:
+            # Use table property defaults
+            table.expire_snapshots().with_retention_policy().commit()
+
+            # Override defaults with explicit values
+            table.expire_snapshots().with_retention_policy(
+                timestamp_ms=1234567890000,
+                retain_last_n=10,
+                min_snapshots_to_keep=5
+            ).commit()
+        """
+        # Get default values from table properties
+        default_max_age, default_min_snapshots, _ = self._get_expiration_properties()
+
+        # Use defaults from table properties if not explicitly provided
+        if timestamp_ms is None:
+            timestamp_ms = default_max_age
+
+        if min_snapshots_to_keep is None:
+            min_snapshots_to_keep = default_min_snapshots
+
+        # If no expiration criteria are provided, don't expire anything
+        if timestamp_ms is None and retain_last_n is None and min_snapshots_to_keep is None:
+            return self
+
+        if retain_last_n is not None and retain_last_n < 1:
+            raise ValueError("retain_last_n must be at least 1")
+
+        if min_snapshots_to_keep is not None and min_snapshots_to_keep < 1:
+            raise ValueError("min_snapshots_to_keep must be at least 1")
+
+        snapshots_to_expire = self._get_snapshots_to_expire_with_retention(
+            timestamp_ms=timestamp_ms, retain_last_n=retain_last_n, min_snapshots_to_keep=min_snapshots_to_keep
+        )
+        self._snapshot_ids_to_expire.update(snapshots_to_expire)
+        return self
+
+    def retain_last_n(self, n: int) -> ExpireSnapshots:
+        """Keep only the last N snapshots globally across all branches, expiring all others.
+
+        Note: This method considers snapshots globally across the entire table history,
+        not per-branch. Protected snapshots (branch/tag heads) are always preserved
+        regardless of the retention count.
+
+        Args:
+            n: Number of most recent snapshots to keep globally.
+
+        Returns:
+            This for method chaining.
+
+        Raises:
+            ValueError: If n is less than 1.
+        """
+        if n < 1:
+            raise ValueError("Number of snapshots to retain must be at least 1")
+
+        snapshots_to_keep = self._get_snapshots_to_keep(retain_last_n=n)
+        snapshots_to_expire = [
+            id for snapshot in self._transaction.table_metadata.snapshots if (id := snapshot.snapshot_id) not in snapshots_to_keep
+        ]
+
+        self._snapshot_ids_to_expire.update(snapshots_to_expire)
+        return self
+
+    def _get_snapshots_to_keep(self, retain_last_n: Optional[int] = None) -> Set[int]:
+        """Get set of snapshot IDs that should be kept based on protection and retention rules.
+
+        Args:
+            retain_last_n: Number of most recent snapshots to keep.
+
+        Returns:
+            Set of snapshot IDs to keep.
+        """
+        snapshots_to_keep = self._get_protected_snapshot_ids()
+
+        if retain_last_n is not None:
+            # Sort snapshots by timestamp (most recent first), and get most recent N
+            sorted_snapshots = sorted(self._transaction.table_metadata.snapshots, key=lambda s: s.timestamp_ms, reverse=True)
+            snapshots_to_keep.update(snapshot.snapshot_id for snapshot in sorted_snapshots[:retain_last_n])
+
+        return snapshots_to_keep
+
+    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
+    ) -> List[int]:
+        """Get snapshots to expire considering retention strategies.
+
+        Args:
+            timestamp_ms: Only snapshots with timestamp_ms < this value will be considered for expiration.
+            retain_last_n: Always keep the last N snapshots regardless of age.
+            min_snapshots_to_keep: Minimum number of snapshots to keep in total.
+
+        Returns:
+            List of snapshot IDs to expire.
+        """
+        snapshots_to_keep = self._get_snapshots_to_keep(retain_last_n=retain_last_n)
+
+        # Apply timestamp constraint
+        candidates_for_expiration = []
+        for snapshot in self._transaction.table_metadata.snapshots:
+            if snapshot.snapshot_id not in snapshots_to_keep and (timestamp_ms is None or snapshot.timestamp_ms < timestamp_ms):
+                candidates_for_expiration.append(snapshot)
+
+        # Sort candidates by timestamp (oldest first) for potential expiration
+        candidates_for_expiration.sort(key=lambda s: s.timestamp_ms)
+
+        # Apply min_snapshots_to_keep constraint
+        if min_snapshots_to_keep is not None:
+            total_snapshots = len(self._transaction.table_metadata.snapshots)
+            max_to_expire = total_snapshots - min_snapshots_to_keep
+            snapshots_to_expire = [candidate.snapshot_id for candidate in candidates_for_expiration[:max_to_expire]]
+        else:
+            snapshots_to_expire = [candidate.snapshot_id for candidate in candidates_for_expiration]
+
+        return snapshots_to_expire
+
+    def _get_expiration_properties(self) -> tuple[Optional[int], Optional[int], Optional[int]]:
+        """Get the default expiration properties from table properties.
+
+        Returns:
+            Tuple of (max_snapshot_age_ms, min_snapshots_to_keep, max_ref_age_ms)
+        """
+        from pyiceberg.table import TableProperties
+
+        properties = self._transaction.table_metadata.properties
+
+        max_snapshot_age = property_as_int(properties, TableProperties.MAX_SNAPSHOT_AGE_MS)
+        min_snapshots_to_keep = property_as_int(properties, TableProperties.MIN_SNAPSHOTS_TO_KEEP)
+        max_ref_age_ms = property_as_int(properties, "history.expire.max-ref-age-ms")
+
+        return max_snapshot_age, min_snapshots_to_keep, max_ref_age_ms