Anton-Tarazi commented on code in PR #2369:
URL: https://github.com/apache/iceberg-python/pull/2369#discussion_r2314806466
##########
pyiceberg/table/update/snapshot.py:
##########
@@ -953,7 +953,7 @@ def _get_protected_snapshot_ids(self) -> Set[int]:
if ref.snapshot_ref_type in [SnapshotRefType.TAG,
SnapshotRefType.BRANCH]
}
- def by_id(self, snapshot_id: int) -> ExpireSnapshots:
+ def by_id(self, snapshot_id: int) -> "ExpireSnapshots":
Review Comment:
fwiw since we have `from __future__ import annotations` at the top of the
file I think its cleaner to make things consistent to not have quotes.
Probably outside of the scope of this PR
##########
pyiceberg/table/update/snapshot.py:
##########
@@ -1005,3 +1005,197 @@ 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.
Review Comment:
I believe this should mean `last n per branch`, I don't think this code
handles branches at all
##########
pyiceberg/table/update/snapshot.py:
##########
@@ -1005,3 +1005,197 @@ 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, expiring all others.
+
+ Args:
+ n: Number of most recent snapshots to keep.
+
+ 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")
+
+ protected_ids = self._get_protected_snapshot_ids()
+
+ # Sort snapshots by timestamp (most recent first)
+ sorted_snapshots = sorted(self._transaction.table_metadata.snapshots,
key=lambda s: s.timestamp_ms, reverse=True)
+
+ # Keep the last N snapshots and all protected ones
+ snapshots_to_keep = set()
+ snapshots_to_keep.update(protected_ids)
+
+ # Add the N most recent snapshots
+ for i, snapshot in enumerate(sorted_snapshots):
+ if i < n:
+ snapshots_to_keep.add(snapshot.snapshot_id)
+
+ # Find snapshots to expire
+ snapshots_to_expire = []
+ for snapshot in self._transaction.table_metadata.snapshots:
+ if snapshot.snapshot_id not in snapshots_to_keep:
+ snapshots_to_expire.append(snapshot.snapshot_id)
+
+ self._snapshot_ids_to_expire.update(snapshots_to_expire)
+ return self
+
+ 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.
+ """
+ protected_ids = self._get_protected_snapshot_ids()
+
+ # Sort snapshots by timestamp (most recent first)
+ sorted_snapshots = sorted(self._transaction.table_metadata.snapshots,
key=lambda s: s.timestamp_ms, reverse=True)
+
+ # Start with all snapshots that could be expired
+ candidates_for_expiration = []
+ snapshots_to_keep = set(protected_ids)
+
+ # Apply retain_last_n constraint
+ if retain_last_n is not None:
+ for i, snapshot in enumerate(sorted_snapshots):
+ if i < retain_last_n:
+ snapshots_to_keep.add(snapshot.snapshot_id)
Review Comment:
this code is the same as in `retain_last_n`, can we refactor to its own
function? I think we also need to handle branches and take the last n of each
branch
##########
pyiceberg/table/update/snapshot.py:
##########
@@ -1005,3 +1005,197 @@ 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, expiring all others.
+
+ Args:
+ n: Number of most recent snapshots to keep.
+
+ 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")
+
+ protected_ids = self._get_protected_snapshot_ids()
+
+ # Sort snapshots by timestamp (most recent first)
+ sorted_snapshots = sorted(self._transaction.table_metadata.snapshots,
key=lambda s: s.timestamp_ms, reverse=True)
+
+ # Keep the last N snapshots and all protected ones
+ snapshots_to_keep = set()
+ snapshots_to_keep.update(protected_ids)
+
+ # Add the N most recent snapshots
+ for i, snapshot in enumerate(sorted_snapshots):
+ if i < n:
+ snapshots_to_keep.add(snapshot.snapshot_id)
+
+ # Find snapshots to expire
+ snapshots_to_expire = []
+ for snapshot in self._transaction.table_metadata.snapshots:
+ if snapshot.snapshot_id not in snapshots_to_keep:
+ snapshots_to_expire.append(snapshot.snapshot_id)
Review Comment:
small syntax change to make more pythonic :)
##########
pyiceberg/table/update/snapshot.py:
##########
@@ -1005,3 +1005,197 @@ 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, expiring all others.
+
+ Args:
+ n: Number of most recent snapshots to keep.
+
+ 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")
+
+ protected_ids = self._get_protected_snapshot_ids()
+
+ # Sort snapshots by timestamp (most recent first)
+ sorted_snapshots = sorted(self._transaction.table_metadata.snapshots,
key=lambda s: s.timestamp_ms, reverse=True)
+
+ # Keep the last N snapshots and all protected ones
+ snapshots_to_keep = set()
+ snapshots_to_keep.update(protected_ids)
+
+ # Add the N most recent snapshots
+ for i, snapshot in enumerate(sorted_snapshots):
+ if i < n:
+ snapshots_to_keep.add(snapshot.snapshot_id)
+
+ # Find snapshots to expire
+ snapshots_to_expire = []
+ for snapshot in self._transaction.table_metadata.snapshots:
+ if snapshot.snapshot_id not in snapshots_to_keep:
+ snapshots_to_expire.append(snapshot.snapshot_id)
Review Comment:
```suggestion
snapshots_to_keep = self._get_protected_snapshot_ids()
# 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[:n])
snapshots_to_expire = [id for snapshot in
self._transaction.table_metadata.snapshots if (id := snapshot.snapshot_id) not
in snapshots_to_keep]
```
##########
pyiceberg/table/update/snapshot.py:
##########
@@ -1005,3 +1005,197 @@ 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, expiring all others.
+
+ Args:
+ n: Number of most recent snapshots to keep.
+
+ 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")
+
+ protected_ids = self._get_protected_snapshot_ids()
+
+ # Sort snapshots by timestamp (most recent first)
+ sorted_snapshots = sorted(self._transaction.table_metadata.snapshots,
key=lambda s: s.timestamp_ms, reverse=True)
+
+ # Keep the last N snapshots and all protected ones
+ snapshots_to_keep = set()
+ snapshots_to_keep.update(protected_ids)
+
+ # Add the N most recent snapshots
+ for i, snapshot in enumerate(sorted_snapshots):
+ if i < n:
+ snapshots_to_keep.add(snapshot.snapshot_id)
+
+ # Find snapshots to expire
+ snapshots_to_expire = []
+ for snapshot in self._transaction.table_metadata.snapshots:
+ if snapshot.snapshot_id not in snapshots_to_keep:
+ snapshots_to_expire.append(snapshot.snapshot_id)
+
+ self._snapshot_ids_to_expire.update(snapshots_to_expire)
+ return self
+
+ 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.
+ """
+ protected_ids = self._get_protected_snapshot_ids()
+
+ # Sort snapshots by timestamp (most recent first)
+ sorted_snapshots = sorted(self._transaction.table_metadata.snapshots,
key=lambda s: s.timestamp_ms, reverse=True)
+
+ # Start with all snapshots that could be expired
+ candidates_for_expiration = []
+ snapshots_to_keep = set(protected_ids)
+
+ # Apply retain_last_n constraint
+ if retain_last_n is not None:
+ for i, snapshot in enumerate(sorted_snapshots):
+ if i < retain_last_n:
+ snapshots_to_keep.add(snapshot.snapshot_id)
+
+ # Apply timestamp constraint
+ 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)
Review Comment:
make more pythonic with comprehension?
##########
pyiceberg/table/update/snapshot.py:
##########
@@ -1005,3 +1005,197 @@ 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, expiring all others.
+
+ Args:
+ n: Number of most recent snapshots to keep.
+
+ 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")
+
+ protected_ids = self._get_protected_snapshot_ids()
+
+ # Sort snapshots by timestamp (most recent first)
+ sorted_snapshots = sorted(self._transaction.table_metadata.snapshots,
key=lambda s: s.timestamp_ms, reverse=True)
+
+ # Keep the last N snapshots and all protected ones
+ snapshots_to_keep = set()
+ snapshots_to_keep.update(protected_ids)
+
+ # Add the N most recent snapshots
+ for i, snapshot in enumerate(sorted_snapshots):
+ if i < n:
+ snapshots_to_keep.add(snapshot.snapshot_id)
+
+ # Find snapshots to expire
+ snapshots_to_expire = []
+ for snapshot in self._transaction.table_metadata.snapshots:
+ if snapshot.snapshot_id not in snapshots_to_keep:
+ snapshots_to_expire.append(snapshot.snapshot_id)
+
+ self._snapshot_ids_to_expire.update(snapshots_to_expire)
+ return self
+
+ 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.
+ """
+ protected_ids = self._get_protected_snapshot_ids()
+
+ # Sort snapshots by timestamp (most recent first)
+ sorted_snapshots = sorted(self._transaction.table_metadata.snapshots,
key=lambda s: s.timestamp_ms, reverse=True)
+
+ # Start with all snapshots that could be expired
+ candidates_for_expiration = []
+ snapshots_to_keep = set(protected_ids)
+
+ # Apply retain_last_n constraint
+ if retain_last_n is not None:
+ for i, snapshot in enumerate(sorted_snapshots):
+ if i < retain_last_n:
+ snapshots_to_keep.add(snapshot.snapshot_id)
+
+ # Apply timestamp constraint
+ 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
+ total_snapshots = len(self._transaction.table_metadata.snapshots)
+ snapshots_to_expire: List[int] = []
+
+ for candidate in candidates_for_expiration:
+ # Check if expiring this snapshot would violate
min_snapshots_to_keep
+ remaining_after_expiration = total_snapshots -
len(snapshots_to_expire) - 1
+
+ if min_snapshots_to_keep is None or remaining_after_expiration >=
min_snapshots_to_keep:
+ snapshots_to_expire.append(candidate.snapshot_id)
+ else:
+ # Stop expiring to maintain minimum count
+ break
Review Comment:
double check that I didn't make an off-by-one error here but I believe this
is a more concise way to express things :)
##########
pyiceberg/table/update/snapshot.py:
##########
@@ -1005,3 +1005,197 @@ 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, expiring all others.
+
+ Args:
+ n: Number of most recent snapshots to keep.
+
+ 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")
+
+ protected_ids = self._get_protected_snapshot_ids()
+
+ # Sort snapshots by timestamp (most recent first)
+ sorted_snapshots = sorted(self._transaction.table_metadata.snapshots,
key=lambda s: s.timestamp_ms, reverse=True)
+
+ # Keep the last N snapshots and all protected ones
+ snapshots_to_keep = set()
+ snapshots_to_keep.update(protected_ids)
+
+ # Add the N most recent snapshots
+ for i, snapshot in enumerate(sorted_snapshots):
+ if i < n:
+ snapshots_to_keep.add(snapshot.snapshot_id)
+
+ # Find snapshots to expire
+ snapshots_to_expire = []
+ for snapshot in self._transaction.table_metadata.snapshots:
+ if snapshot.snapshot_id not in snapshots_to_keep:
+ snapshots_to_expire.append(snapshot.snapshot_id)
+
+ self._snapshot_ids_to_expire.update(snapshots_to_expire)
+ return self
+
+ 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.
+ """
+ protected_ids = self._get_protected_snapshot_ids()
+
+ # Sort snapshots by timestamp (most recent first)
+ sorted_snapshots = sorted(self._transaction.table_metadata.snapshots,
key=lambda s: s.timestamp_ms, reverse=True)
+
+ # Start with all snapshots that could be expired
+ candidates_for_expiration = []
+ snapshots_to_keep = set(protected_ids)
+
+ # Apply retain_last_n constraint
+ if retain_last_n is not None:
+ for i, snapshot in enumerate(sorted_snapshots):
+ if i < retain_last_n:
+ snapshots_to_keep.add(snapshot.snapshot_id)
+
+ # Apply timestamp constraint
+ 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
+ total_snapshots = len(self._transaction.table_metadata.snapshots)
+ snapshots_to_expire: List[int] = []
+
+ for candidate in candidates_for_expiration:
+ # Check if expiring this snapshot would violate
min_snapshots_to_keep
+ remaining_after_expiration = total_snapshots -
len(snapshots_to_expire) - 1
+
+ if min_snapshots_to_keep is None or remaining_after_expiration >=
min_snapshots_to_keep:
+ snapshots_to_expire.append(candidate.snapshot_id)
+ else:
+ # Stop expiring to maintain minimum count
+ break
+
+ 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)
+ """
+ properties = self._transaction.table_metadata.properties
+
+ max_snapshot_age_ms =
properties.get("history.expire.max-snapshot-age-ms")
Review Comment:
should this string and the default value be a named constant somewhere?
What do you think about using `property_as_int` from `properties.py` to be
consistent with how properties are handled elsewhere?
##########
pyiceberg/table/update/snapshot.py:
##########
@@ -1005,3 +1005,197 @@ 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, expiring all others.
+
+ Args:
+ n: Number of most recent snapshots to keep.
+
+ 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")
+
+ protected_ids = self._get_protected_snapshot_ids()
+
+ # Sort snapshots by timestamp (most recent first)
+ sorted_snapshots = sorted(self._transaction.table_metadata.snapshots,
key=lambda s: s.timestamp_ms, reverse=True)
+
+ # Keep the last N snapshots and all protected ones
+ snapshots_to_keep = set()
+ snapshots_to_keep.update(protected_ids)
+
+ # Add the N most recent snapshots
+ for i, snapshot in enumerate(sorted_snapshots):
+ if i < n:
+ snapshots_to_keep.add(snapshot.snapshot_id)
+
+ # Find snapshots to expire
+ snapshots_to_expire = []
+ for snapshot in self._transaction.table_metadata.snapshots:
+ if snapshot.snapshot_id not in snapshots_to_keep:
+ snapshots_to_expire.append(snapshot.snapshot_id)
+
+ self._snapshot_ids_to_expire.update(snapshots_to_expire)
+ return self
+
+ 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.
+ """
+ protected_ids = self._get_protected_snapshot_ids()
+
+ # Sort snapshots by timestamp (most recent first)
+ sorted_snapshots = sorted(self._transaction.table_metadata.snapshots,
key=lambda s: s.timestamp_ms, reverse=True)
+
+ # Start with all snapshots that could be expired
+ candidates_for_expiration = []
+ snapshots_to_keep = set(protected_ids)
+
+ # Apply retain_last_n constraint
+ if retain_last_n is not None:
+ for i, snapshot in enumerate(sorted_snapshots):
+ if i < retain_last_n:
+ snapshots_to_keep.add(snapshot.snapshot_id)
+
+ # Apply timestamp constraint
+ 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
+ total_snapshots = len(self._transaction.table_metadata.snapshots)
+ snapshots_to_expire: List[int] = []
+
+ for candidate in candidates_for_expiration:
+ # Check if expiring this snapshot would violate
min_snapshots_to_keep
+ remaining_after_expiration = total_snapshots -
len(snapshots_to_expire) - 1
+
+ if min_snapshots_to_keep is None or remaining_after_expiration >=
min_snapshots_to_keep:
+ snapshots_to_expire.append(candidate.snapshot_id)
+ else:
+ # Stop expiring to maintain minimum count
+ break
Review Comment:
```suggestion
# Sort candidates by timestamp (newest first) for potential
expiration
candidates_for_expiration.sort(key=lambda s: s.timestamp_ms,
reverse=True)
snapshots_to_expire =
candidates_for_expiration[min_snapshots_to_keep:]
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
