include dry run and older than

Author: jayceslesar

pyiceberg/table/__init__.py

@@ -24,6 +24,7 @@
 import warnings
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
+from datetime import timedelta
 from functools import cached_property
 from itertools import chain
 from types import TracebackType
@@ -1375,10 +1376,10 @@ def to_polars(self) -> pl.LazyFrame:
 
         return pl.scan_iceberg(self)
 
-    def delete_orphaned_files(self) -> None:
+    def delete_orphaned_files(self, older_than: Optional[timedelta] = timedelta(days=3), dry_run: bool = False) -> None:
         """Delete orphaned files in the table."""
         location = self.location()
-        orphaned_files = self.inspect.orphaned_files(location)
+        orphaned_files = self.inspect.orphaned_files(location, older_than)
         logger.info(f"Found {len(orphaned_files)} orphaned files at {location}!")
 
         def _delete(file: str) -> None:
@@ -1388,11 +1389,14 @@ def _delete(file: str) -> None:
                 self.io.delete(file)
 
         if orphaned_files:
-            executor = ExecutorFactory.get_or_create()
-            deletes = executor.map(_delete, orphaned_files)
-            # exhaust
-            list(deletes)
-            logger.info(f"Deleted {len(orphaned_files)} orphaned files at {location}!")
+            if dry_run:
+                logger.info(f"(Dry Run) Deleted {len(orphaned_files)} orphaned files at {location}!")
+            else:
+                executor = ExecutorFactory.get_or_create()
+                deletes = executor.map(_delete, orphaned_files)
+                # exhaust
+                list(deletes)
+                logger.info(f"Deleted {len(orphaned_files)} orphaned files at {location}!")
 
 
 class StaticTable(Table):

pyiceberg/table/inspect.py

@@ -16,7 +16,7 @@
 # under the License.
 from __future__ import annotations
 
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 from functools import reduce
 from typing import TYPE_CHECKING, Any, Dict, Iterator, List, Optional, Set, Tuple, Union, cast
 
@@ -666,7 +666,7 @@ def all_manifests(self, snapshots: Optional[Union[list[Snapshot], list[int]]] =
         )
         return pa.concat_tables(manifests_by_snapshots)
 
-    def orphaned_files(self, location: str) -> Set[str]:
+    def orphaned_files(self, location: str, older_than: Optional[timedelta] = timedelta(days=3)) -> Set[str]:
         try:
             import pyarrow as pa  # noqa: F401
         except ModuleNotFoundError as e:
@@ -692,9 +692,10 @@ def orphaned_files(self, location: str) -> Set[str]:
 
         _, _, path = _parse_location(location)
         selector = FileSelector(path, recursive=True)
-        # filter to just files as it may return directories
-        all_files = [f.path for f in fs.get_file_info(selector) if f.type == FileType.File]
+        # filter to just files as it may return directories, and filter on time
+        as_of = datetime.now(timezone.utc) - older_than if older_than else None
+        all_files = [f for f in fs.get_file_info(selector) if f.type == FileType.File and (as_of is None or (f.mtime < as_of))]
 
-        orphaned_files = set(all_files).difference(set(all_known_files))
+        orphaned_files = set(all_files).difference(all_known_files)
 
         return orphaned_files

tests/table/test_delete_orphans.py

@@ -14,6 +14,8 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
+import os
+from datetime import datetime, timedelta
 from pathlib import Path, PosixPath
 from unittest.mock import PropertyMock, patch
 
@@ -66,8 +68,13 @@ def test_delete_orphaned_files(catalog: Catalog) -> None:
     orphaned_file.touch()
     assert orphaned_file.exists()
 
+    # should not delete because it was just created...
     tbl.delete_orphaned_files()
-    assert not orphaned_file.exists()
+    assert orphaned_file.exists()
+
+    # modify creation date to be older than 3 days
+    five_days_ago = (datetime.now() - timedelta(days=5)).timestamp()
+    os.utime(orphaned_file, (five_days_ago, five_days_ago))
 
 
 def test_delete_orphaned_files_with_invalid_file_doesnt_error(catalog: Catalog) -> None:
@@ -100,7 +107,7 @@ def test_delete_orphaned_files_with_invalid_file_doesnt_error(catalog: Catalog)
 
     file_that_does_not_exist = "foo/bar.baz"
     with patch.object(type(tbl), "inspect", new_callable=PropertyMock) as mock_inspect:
-        mock_inspect.return_value.orphaned_files = lambda x: {file_that_does_not_exist}
+        mock_inspect.return_value.orphaned_files = lambda location, older_than: {file_that_does_not_exist}
         with patch.object(tbl.io, "delete", wraps=tbl.io.delete) as mock_delete:
             tbl.delete_orphaned_files()
             mock_delete.assert_called_with(file_that_does_not_exist)