updates from review!

Author: jayceslesar

pyiceberg/table/__init__.py

@@ -16,6 +16,7 @@
 # under the License.
 from __future__ import annotations
 
+import contextlib
 import itertools
 import logging
 import os
@@ -32,7 +33,6 @@
     Callable,
     Dict,
     Iterable,
-    Iterator,
     List,
     Optional,
     Set,
@@ -64,7 +64,7 @@
     inclusive_projection,
     manifest_evaluator,
 )
-from pyiceberg.io import FileIO, _parse_location, load_file_io
+from pyiceberg.io import FileIO, load_file_io
 from pyiceberg.manifest import (
     POSITIONAL_DELETE_SCHEMA,
     DataFile,
@@ -1377,39 +1377,19 @@ def to_polars(self) -> pl.LazyFrame:
 
     def delete_orphaned_files(self) -> None:
         """Delete orphaned files in the table."""
-        try:
-            import pyarrow as pa  # noqa: F401
-        except ModuleNotFoundError as e:
-            raise ModuleNotFoundError("For deleting orphaned files PyArrow needs to be installed") from e
-
-        from pyarrow.fs import FileSelector, FileType
-
-        from pyiceberg.io.pyarrow import _fs_from_file_path
-
         location = self.location()
-
-        all_known_files = []
-        snapshots = self.snapshots()
-        snapshot_ids = [snapshot.snapshot_id for snapshot in snapshots]
-        all_manifests_table = self.inspect.all_manifests(snapshots)
-        all_known_files.extend(all_manifests_table["path"].to_pylist())
-
-        executor = ExecutorFactory.get_or_create()
-        files_by_snapshots: Iterator["pa.Table"] = executor.map(lambda snapshot_id: self.inspect.files(snapshot_id), snapshot_ids)
-        all_known_files.extend(pa.concat_tables(files_by_snapshots)["file_path"].to_pylist())
-
-        fs = _fs_from_file_path(self.io, location)
-
-        _, _, 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]
-
-        orphaned_files = set(all_files).difference(set(all_known_files))
+        orphaned_files = self.inspect.orphaned_files(location)
         logger.info(f"Found {len(orphaned_files)} orphaned files at {location}!")
 
+        def _delete(file: str) -> None:
+            # don't error if the file doesn't exist
+            # still catch ctrl-c, etc.
+            with contextlib.suppress(Exception):
+                self.io.delete(file)
+
         if orphaned_files:
-            deletes = executor.map(self.io.delete, 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}!")

pyiceberg/table/inspect.py

@@ -17,9 +17,11 @@
 from __future__ import annotations
 
 from datetime import datetime, timezone
-from typing import TYPE_CHECKING, Any, Dict, Iterator, List, Optional, Set, Tuple
+from functools import reduce
+from typing import TYPE_CHECKING, Any, Dict, Iterator, List, Optional, Set, Tuple, Union, cast
 
 from pyiceberg.conversions import from_bytes
+from pyiceberg.io import _parse_location
 from pyiceberg.manifest import DataFile, DataFileContent, ManifestContent, PartitionFieldSummary
 from pyiceberg.partitioning import PartitionSpec
 from pyiceberg.table.snapshots import Snapshot, ancestors_of
@@ -645,10 +647,16 @@ def data_files(self, snapshot_id: Optional[int] = None) -> "pa.Table":
     def delete_files(self, snapshot_id: Optional[int] = None) -> "pa.Table":
         return self._files(snapshot_id, {DataFileContent.POSITION_DELETES, DataFileContent.EQUALITY_DELETES})
 
-    def all_manifests(self, snapshots: Optional[list[Snapshot]] = None) -> "pa.Table":
+    def all_manifests(self, snapshots: Optional[Union[list[Snapshot], list[int]]] = None) -> "pa.Table":
         import pyarrow as pa
 
-        snapshots = snapshots or self.tbl.snapshots()
+        # coerce into snapshot objects if users passes in snapshot ids
+        if snapshots is not None:
+            if isinstance(snapshots[0], int):
+                snapshots = cast(list[Snapshot], [self.tbl.metadata.snapshot_by_id(snapshot_id) for snapshot_id in snapshots])
+        else:
+            snapshots = self.tbl.snapshots()
+
         if not snapshots:
             return pa.Table.from_pylist([], schema=self._get_all_manifests_schema())
 
@@ -657,3 +665,36 @@ def all_manifests(self, snapshots: Optional[list[Snapshot]] = None) -> "pa.Table
             lambda args: self._generate_manifests_table(*args), [(snapshot, True) for snapshot in snapshots]
         )
         return pa.concat_tables(manifests_by_snapshots)
+
+    def orphaned_files(self, location: str) -> Set[str]:
+        try:
+            import pyarrow as pa  # noqa: F401
+        except ModuleNotFoundError as e:
+            raise ModuleNotFoundError("For deleting orphaned files PyArrow needs to be installed") from e
+
+        from pyarrow.fs import FileSelector, FileType
+
+        from pyiceberg.io.pyarrow import _fs_from_file_path
+
+        all_known_files = set()
+        snapshots = self.tbl.snapshots()
+        manifests_paths = self.all_manifests(snapshots)["path"].to_pylist()
+        all_known_files.update(manifests_paths)
+
+        executor = ExecutorFactory.get_or_create()
+        files_by_snapshots: Iterator[Set[str]] = executor.map(
+            lambda snapshot_id: set(self.files(snapshot_id)["file_path"].to_pylist())
+        )
+        datafile_paths: set[str] = reduce(set.union, files_by_snapshots, set())
+        all_known_files.update(datafile_paths)
+
+        fs = _fs_from_file_path(self.tbl.io, location)
+
+        _, _, 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]
+
+        orphaned_files = set(all_files).difference(set(all_known_files))
+
+        return orphaned_files

tests/table/test_delete_orphans.py

@@ -15,6 +15,7 @@
 # specific language governing permissions and limitations
 # under the License.
 from pathlib import Path, PosixPath
+from unittest.mock import PropertyMock, patch
 
 import pyarrow as pa
 import pytest
@@ -67,3 +68,39 @@ def test_delete_orphaned_files(catalog: Catalog) -> None:
 
     tbl.delete_orphaned_files()
     assert not orphaned_file.exists()
+
+
+def test_delete_orphaned_files_with_invalid_file_doesnt_error(catalog: Catalog) -> None:
+    identifier = "default.test_delete_orphaned_files"
+
+    schema = Schema(
+        NestedField(1, "city", StringType(), required=True),
+        NestedField(2, "inhabitants", IntegerType(), required=True),
+        # Mark City as the identifier field, also known as the primary-key
+        identifier_field_ids=[1],
+    )
+
+    tbl = catalog.create_table(identifier, schema=schema)
+
+    arrow_schema = pa.schema(
+        [
+            pa.field("city", pa.string(), nullable=False),
+            pa.field("inhabitants", pa.int32(), nullable=False),
+        ]
+    )
+
+    df = pa.Table.from_pylist(
+        [
+            {"city": "Drachten", "inhabitants": 45019},
+            {"city": "Drachten", "inhabitants": 45019},
+        ],
+        schema=arrow_schema,
+    )
+    tbl.append(df)
+
+    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}
+        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)