SQLite Connection Cleanup: Added proper cleanup of SQLAlchemy engine connections using engine.dispose() in test fixtures

Test Resource Management: Added try/finally blocks to ensure cleanup happens even if tests fail

Catalog Connection Handling: Modified both the iceberg_catalog and prepopulated_table fixtures to properly clean up database connections

Mock Catalog Cleanup: Added cleanup for tests that replace the table catalog with mock objects

Author: ForeverAngry

pyiceberg/table/maintenance.py

@@ -213,15 +213,20 @@ def expire_snapshots_with_retention_policy(
         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, all snapshots are candidates (subject to other constraints).
+                         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
-                                 from removing too many snapshots.
+                                 Acts as a guardrail to prevent aggressive expiration logic.
+                                 If None, will use history.expire.min-snapshots-to-keep table property if set.
 
         Returns:
             List of snapshot IDs that were expired.
@@ -230,48 +235,62 @@ def expire_snapshots_with_retention_policy(
             ValueError: If retain_last_n or min_snapshots_to_keep is less than 1.
 
         Examples:
-            # Keep last 5 snapshots regardless of age
-            maintenance.expire_snapshots_with_retention_policy(retain_last_n=5)
-
-            # Expire snapshots older than timestamp but keep at least 3 total
-            maintenance.expire_snapshots_with_retention_policy(
-                timestamp_ms=1234567890000,
-                min_snapshots_to_keep=3
-            )
+            # Use table property defaults
+            maintenance.expire_snapshots_with_retention_policy()
 
-            # Combined policy: expire old snapshots but keep last 10 and at least 5 total
+            # Override defaults with explicit values
             maintenance.expire_snapshots_with_retention_policy(
                 timestamp_ms=1234567890000,
                 retain_last_n=10,
                 min_snapshots_to_keep=5
             )
         """
+        # 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
+
         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
+            timestamp_ms=timestamp_ms,
+            retain_last_n=retain_last_n,
+            min_snapshots_to_keep=min_snapshots_to_keep
         )
 
         if snapshots_to_expire:
             self._expire_snapshots_by_ids(snapshots_to_expire)
 
-    def _get_protected_snapshot_ids(self) -> Set[int]:
+    def _get_protected_snapshot_ids(self, table_metadata: Optional[TableMetadata] = None) -> Set[int]:
         """Get the IDs of protected snapshots.
 
         These are the HEAD snapshots of all branches and all tagged snapshots.
         These ids are to be excluded from expiration.
 
         Args:
-            table_metadata: The table metadata to check for protected snapshots.
+            table_metadata: Optional table metadata to check for protected snapshots.
+                          If not provided, uses the table's current metadata.
 
         Returns:
             Set of protected snapshot IDs to exclude from expiration.
         """
-        return set(self.tbl.inspect.refs()["snapshot_id"].to_pylist())
+        # Prefer provided metadata, fall back to current table metadata
+        metadata = table_metadata or self.tbl.metadata
+        refs = metadata.refs if metadata else {}
+        return {ref.snapshot_id for ref in refs.values()}
 
     def _get_all_datafiles(self) -> List[DataFile]:
         """Collect all DataFiles in the current snapshot only."""
@@ -359,3 +378,22 @@ def deduplicate_data_files(self) -> List[DataFile]:
             self.tbl = self.tbl.refresh()
 
         return removed
+
+    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.tbl.properties
+
+        max_snapshot_age_ms = properties.get("history.expire.max-snapshot-age-ms")
+        max_snapshot_age = int(max_snapshot_age_ms) if max_snapshot_age_ms is not None else None
+
+        min_snapshots = properties.get("history.expire.min-snapshots-to-keep")
+        min_snapshots_to_keep = int(min_snapshots) if min_snapshots is not None else None
+
+        max_ref_age = properties.get("history.expire.max-ref-age-ms")
+        max_ref_age_ms = int(max_ref_age) if max_ref_age is not None else None
+
+        return max_snapshot_age, min_snapshots_to_keep, max_ref_age_ms

tests/table/test_dedup_data_file_filepaths.py

@@ -17,7 +17,7 @@
 import os
 import uuid
 from pathlib import Path
-from typing import List, Set
+from typing import List, Set, Generator
 
 import pyarrow as pa
 import pyarrow.parquet as pq
@@ -30,10 +30,16 @@
 
 
 @pytest.fixture
-def iceberg_catalog(tmp_path: Path) -> InMemoryCatalog:
+def iceberg_catalog(tmp_path: Path) -> Generator[InMemoryCatalog, None, None]:
     catalog = InMemoryCatalog("test.in_memory.catalog", warehouse=tmp_path.absolute().as_posix())
     catalog.create_namespace("default")
-    return catalog
+    yield catalog
+    # Clean up SQLAlchemy engine connections
+    if hasattr(catalog, 'engine'):
+        try:
+            catalog.engine.dispose()
+        except Exception:
+            pass
 
 
 @pytest.fixture
@@ -43,7 +49,7 @@ def dupe_data_file_path(tmp_path: Path) -> Path:
 
 
 @pytest.fixture
-def prepopulated_table(iceberg_catalog: InMemoryCatalog, dupe_data_file_path: Path) -> Table:
+def prepopulated_table(iceberg_catalog: InMemoryCatalog, dupe_data_file_path: Path) -> Generator[Table, None, None]:
     identifier = "default.test_table"
     try:
         iceberg_catalog.drop_table(identifier)
@@ -85,7 +91,14 @@ def prepopulated_table(iceberg_catalog: InMemoryCatalog, dupe_data_file_path: Pa
     tx2.add_files([str(dupe_data_file_path)], check_duplicate_files=False)
     tx2.commit_transaction()
 
-    return table
+    yield table
+    
+    # Cleanup table's catalog connections
+    if hasattr(table, '_catalog') and hasattr(table._catalog, 'engine'):
+        try:
+            table._catalog.engine.dispose()
+        except Exception:
+            pass
 
 
 def test_overwrite_removes_only_selected_datafile(prepopulated_table: Table, dupe_data_file_path: Path) -> None:
@@ -112,11 +125,21 @@ def test_get_all_datafiles_current_snapshot(prepopulated_table: Table, dupe_data
 
 
 def test_get_all_datafiles_all_snapshots(prepopulated_table: Table, dupe_data_file_path: Path) -> None:
-    mt = MaintenanceTable(tbl=prepopulated_table)
+    try:
+        mt = MaintenanceTable(tbl=prepopulated_table)
 
-    datafiles: List[DataFile] = mt._get_all_datafiles()
-    file_paths: Set[str] = {df.file_path.split("/")[-1] for df in datafiles}
-    assert dupe_data_file_path.name in file_paths
+        datafiles: List[DataFile] = mt._get_all_datafiles()
+        file_paths: Set[str] = {df.file_path.split("/")[-1] for df in datafiles}
+        assert dupe_data_file_path.name in file_paths
+    finally:
+        # Ensure catalog connections are properly closed
+        if hasattr(prepopulated_table, '_catalog'):
+            catalog = prepopulated_table._catalog
+            if hasattr(catalog, '_connection') and catalog._connection is not None:
+                try:
+                    catalog._connection.close()
+                except Exception:
+                    pass
 
 
 def test_deduplicate_data_files_removes_duplicates_in_current_snapshot(