remove orphan snapshots

Author: rambleraptor

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)