remove orphan files

mkdocs/docs/api.md

@@ -1527,6 +1527,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
 
 If PyIceberg is unable to automatically determine view support on your REST Catalog, you can manually specify, `"view-endpoints-supported": "true"`:

pyiceberg/io/__init__.py

@@ -30,6 +30,9 @@
 import os
 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 (
@@ -269,6 +272,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."""
 
@@ -306,6 +318,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.
+
+        Args:
+            location (str): A URI or path to recursively list.
+
+        Returns:
+            Iterator[FileEntry]: The metadata of every file under the location.
+
+        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,
@@ -86,6 +87,7 @@
     S3_SIGNER_ENDPOINT_DEFAULT,
     S3_SIGNER_URI,
     S3_SSE_KMS_KEY_ID,
+    FileEntry,
     FileIO,
     InputFile,
     InputStream,
@@ -491,6 +493,40 @@ def delete(self, location: str | InputFile | OutputFile) -> None:
         fs = self._get_fs_from_uri(uri, str_location)
         fs.rm(str_location)
 
+    @override
+    def list_prefix(self, location: str) -> Iterator[FileEntry]:
+        """Recursively list every file under the given location.
+
+        Args:
+            location (str): A URI or a path to recursively list.
+
+        Returns:
+            Iterator[FileEntry]: The metadata of every file under the location.
+        """
+        uri = urlparse(location)
+        fs = self._get_fs_from_uri(uri, location)
+        # On Windows a drive letter parses as a URI scheme, so local paths are reported as-is.
+        scheme = "" if _is_local_path(location) else uri.scheme
+
+        for path, info in fs.find(location, detail=True).items():
+            if info.get("type", "file") != "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 scheme in ("", "file") else f"{scheme}://{path}",
+                size=int(info.get("size") or 0),
+                last_modified=last_modified,
+            )
+
     def _get_fs_from_uri(self, uri: "ParseResult", location: str = "") -> AbstractFileSystem:
         """Get a filesystem from a parsed URI, using hostname for ADLS account resolution."""
         if _is_local_path(location):

pyiceberg/io/pyarrow.py

@@ -61,6 +61,7 @@
 from pyarrow._s3fs import S3RetryStrategy
 from pyarrow.fs import (
     FileInfo,
+    FileSelector,
     FileSystem,
     FileType,
 )
@@ -116,6 +117,7 @@
     S3_ROLE_SESSION_NAME,
     S3_SECRET_ACCESS_KEY,
     S3_SESSION_TOKEN,
+    FileEntry,
     FileIO,
     InputFile,
     InputStream,
@@ -694,6 +696,38 @@ 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
 
+    @override
+    def list_prefix(self, location: str) -> Iterator[FileEntry]:
+        """Recursively list every file under the given location.
+
+        Args:
+            location (str): A URI or a path to recursively list.
+
+        Returns:
+            Iterator[FileEntry]: The metadata of every file under the 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)
+
+        # PyArrow reports paths without a scheme, and for object stores the bucket is part of
+        # the path, so the prefix that reconstructs the original URI differs per scheme.
+        original_scheme = "" if _is_local_path(location) else urlparse(location).scheme
+        if original_scheme in ("hdfs", "viewfs"):
+            uri_prefix = f"{original_scheme}://{netloc}"
+        elif original_scheme:
+            uri_prefix = f"{original_scheme}://"
+        else:
+            uri_prefix = ""
+
+        for info in fs.get_file_info(selector):
+            if info.type == FileType.File:
+                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

@@ -235,6 +235,9 @@ class TableProperties:
     WRITE_UPDATE_ISOLATION_LEVEL = "write.update.isolation-level"
     WRITE_ISOLATION_LEVEL_DEFAULT = "serializable"
 
+    GC_ENABLED = "gc.enabled"
+    GC_ENABLED_DEFAULT = True
+
 
 class Transaction:
     _table: Table

pyiceberg/table/maintenance.py → 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)