remove orphan files

Author: rambleraptor

mkdocs/docs/api.md

@@ -1516,6 +1516,28 @@ def cleanup_old_snapshots(table_name: str, snapshot_ids: list[int]):
 cleanup_old_snapshots("analytics.user_events", [12345, 67890, 11111])
 ```
 
+### Remove Orphan Files
+
+Remove files in the table's storage location that are not reachable from any valid snapshot or metadata file. This typically happens after failed writes or aborted compactions leave residual data files behind. Table property `gc.enabled` must be set.
+
+!!! warning
+    Removing orphan files is destructive and irreversible. Always start with `dry_run()` to inspect the candidates, and make sure no other table or in-flight writer is reading from the same storage location.
+
+```python
+from datetime import datetime, timedelta, timezone
+
+# Dry run — list orphans without deleting
+result = table.maintenance.remove_orphan_files() \
+    .older_than(timedelta(days=7)) \
+    .dry_run() \
+    .execute()
+
+# Actually delete
+table.maintenance.remove_orphan_files() \
+    .older_than(datetime.now(tz=timezone.utc) - timedelta(days=7)) \
+    .execute()
+```
+
 ## Views
 
 PyIceberg supports view operations.

pyiceberg/io/__init__.py

@@ -29,6 +29,9 @@
 import logging
 import warnings
 from abc import ABC, abstractmethod
+from collections.abc import Iterator
+from dataclasses import dataclass
+from datetime import datetime
 from io import SEEK_SET
 from types import TracebackType
 from typing import (
@@ -254,6 +257,15 @@ def create(self, overwrite: bool = False) -> OutputStream:
         """
 
 
+@dataclass(frozen=True)
+class FileEntry:
+    """Metadata only for a single file."""
+
+    location: str
+    size: int
+    last_modified: datetime | None = None
+
+
 class FileIO(ABC):
     """A base class for FileIO implementations."""
 
@@ -291,6 +303,20 @@ def delete(self, location: str | InputFile | OutputFile) -> None:
             FileNotFoundError: When the file at the provided location does not exist.
         """
 
+    def list_prefix(self, location: str) -> Iterator[FileEntry]:
+        """Recursively list every file under the given location.
+
+        Returns metadata-only FileEntry objects.
+
+
+        Args:
+            location (str): A URI or path to recursively list.
+
+        Raises:
+            NotImplementedError: If the FileIO implementation does not support listing.
+        """
+        raise NotImplementedError(f"{type(self).__name__} does not support list_prefix")
+
 
 LOCATION = "location"
 WAREHOUSE = "warehouse"

pyiceberg/io/fsspec.py

@@ -22,8 +22,9 @@
 import logging
 import os
 import threading
-from collections.abc import Callable
+from collections.abc import Callable, Iterator
 from copy import copy
+from datetime import datetime, timezone
 from functools import lru_cache
 from typing import (
     TYPE_CHECKING,
@@ -83,6 +84,7 @@
     S3_SIGNER_ENDPOINT,
     S3_SIGNER_ENDPOINT_DEFAULT,
     S3_SIGNER_URI,
+    FileEntry,
     FileIO,
     InputFile,
     InputStream,
@@ -467,6 +469,30 @@ def delete(self, location: str | InputFile | OutputFile) -> None:
         fs = self._get_fs_from_uri(uri)
         fs.rm(str_location)
 
+    def list_prefix(self, location: str) -> Iterator[FileEntry]:
+        """Recursively list every file under the given location."""
+        uri = urlparse(location)
+        fs = self._get_fs_from_uri(uri)
+
+        for path, info in fs.find(location, detail=True).items():
+            if info.get("type") not in (None, "file"):
+                continue
+
+            mtime = info.get("mtime") or info.get("LastModified") or info.get("last_modified")
+            last_modified: datetime | None
+            if isinstance(mtime, datetime):
+                last_modified = mtime
+            elif isinstance(mtime, (int, float)):
+                last_modified = datetime.fromtimestamp(mtime, tz=timezone.utc)
+            else:
+                last_modified = None
+
+            yield FileEntry(
+                location=path if uri.scheme in ("", "file") else f"{uri.scheme}://{path}",
+                size=int(info.get("size") or 0),
+                last_modified=last_modified,
+            )
+
     def _get_fs_from_uri(self, uri: "ParseResult") -> AbstractFileSystem:
         """Get a filesystem from a parsed URI, using hostname for ADLS account resolution."""
         if uri.scheme in _ADLS_SCHEMES:

pyiceberg/io/pyarrow.py

@@ -60,6 +60,7 @@
 from pyarrow._s3fs import S3RetryStrategy
 from pyarrow.fs import (
     FileInfo,
+    FileSelector,
     FileSystem,
     FileType,
 )
@@ -114,6 +115,7 @@
     S3_ROLE_SESSION_NAME,
     S3_SECRET_ACCESS_KEY,
     S3_SESSION_TOKEN,
+    FileEntry,
     FileIO,
     InputFile,
     InputStream,
@@ -674,6 +676,36 @@ def delete(self, location: str | InputFile | OutputFile) -> None:
                 raise PermissionError(f"Cannot delete file, access denied: {location}") from e
             raise  # pragma: no cover - If some other kind of OSError, raise the raw error
 
+    def list_prefix(self, location: str) -> Iterator[FileEntry]:
+        """Recursively list every file under the given location."""
+        original = urlparse(location)
+        scheme, netloc, path = self.parse_location(location, self.properties)
+        fs = self.fs_by_scheme(scheme, netloc)
+        selector = FileSelector(path, recursive=True, allow_not_found=True)
+
+        if original.scheme in ("hdfs", "viewfs"):
+            uri_prefix = f"{original.scheme}://{netloc}"
+            ensure_leading_slash = True
+        elif original.scheme:
+            # Cloud filesystem paths from pyarrow already start with the bucket/container.
+            uri_prefix = f"{original.scheme}://"
+            ensure_leading_slash = False
+        else:
+            uri_prefix = ""
+            ensure_leading_slash = False
+
+        for info in fs.get_file_info(selector):
+            if info.type != FileType.File:
+                continue
+            info_path = info.path
+            if ensure_leading_slash and not info_path.startswith("/"):
+                info_path = "/" + info_path
+            yield FileEntry(
+                location=f"{uri_prefix}{info_path}",
+                size=info.size or 0,
+                last_modified=info.mtime,
+            )
+
     def __getstate__(self) -> dict[str, Any]:
         """Create a dictionary of the PyArrowFileIO fields used when pickling."""
         fileio_copy = copy(self.__dict__)

pyiceberg/table/__init__.py

@@ -205,6 +205,9 @@ class TableProperties:
     MIN_SNAPSHOTS_TO_KEEP = "history.expire.min-snapshots-to-keep"
     MIN_SNAPSHOTS_TO_KEEP_DEFAULT = 1
 
+    GC_ENABLED = "gc.enabled"
+    GC_ENABLED_DEFAULT = True
+
 
 class Transaction:
     _table: Table

pyiceberg/table/maintenance.py pyiceberg/table/maintenance/__init__.pypyiceberg/table/maintenance.py renamed to pyiceberg/table/maintenance/__init__.py

@@ -24,6 +24,7 @@
 
 if TYPE_CHECKING:
     from pyiceberg.table import Table
+    from pyiceberg.table.maintenance.orphan_files import RemoveOrphanFiles
     from pyiceberg.table.update.snapshot import ExpireSnapshots
 
 
@@ -43,3 +44,13 @@ def expire_snapshots(self) -> ExpireSnapshots:
         from pyiceberg.table.update.snapshot import ExpireSnapshots
 
         return ExpireSnapshots(transaction=Transaction(self.tbl, autocommit=True))
+
+    def remove_orphan_files(self) -> RemoveOrphanFiles:
+        """Return a RemoveOrphanFiles builder for removing files unreachable from the table.
+
+        Returns:
+            RemoveOrphanFiles builder for configuring and executing orphan file removal.
+        """
+        from pyiceberg.table.maintenance.orphan_files import RemoveOrphanFiles
+
+        return RemoveOrphanFiles(self.tbl)