diff --git a/mkdocs/docs/api.md b/mkdocs/docs/api.md index 1e17e64f42..ec6eee0b65 100644 --- a/mkdocs/docs/api.md +++ b/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"`: diff --git a/pyiceberg/io/__init__.py b/pyiceberg/io/__init__.py index c44e105e62..48a26fe92b 100644 --- a/pyiceberg/io/__init__.py +++ b/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" diff --git a/pyiceberg/io/fsspec.py b/pyiceberg/io/fsspec.py index 09bbe6f1d6..1f6f939f0b 100644 --- a/pyiceberg/io/fsspec.py +++ b/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): diff --git a/pyiceberg/io/pyarrow.py b/pyiceberg/io/pyarrow.py index c36f1639d9..9bbe81fd4c 100644 --- a/pyiceberg/io/pyarrow.py +++ b/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__) diff --git a/pyiceberg/table/__init__.py b/pyiceberg/table/__init__.py index fca718f5ec..95a279bb70 100644 --- a/pyiceberg/table/__init__.py +++ b/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 diff --git a/pyiceberg/table/maintenance.py b/pyiceberg/table/maintenance/__init__.py similarity index 77% rename from pyiceberg/table/maintenance.py rename to pyiceberg/table/maintenance/__init__.py index 0fcda35ae9..0c8dc983bb 100644 --- a/pyiceberg/table/maintenance.py +++ b/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) diff --git a/pyiceberg/table/maintenance/orphan_files.py b/pyiceberg/table/maintenance/orphan_files.py new file mode 100644 index 0000000000..21c98d8caa --- /dev/null +++ b/pyiceberg/table/maintenance/orphan_files.py @@ -0,0 +1,401 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +"""Action that removes files from storage that are not reachable from table metadata. + +Lists the table's storage location, computes the set of files referenced by any valid +snapshot or metadata file, and deletes the difference. + +Only acts on files older than 3 days by default. +""" + +import logging +import re +from collections.abc import Callable, Iterable, Iterator +from concurrent.futures import as_completed +from dataclasses import dataclass, field +from datetime import datetime, timedelta, timezone +from enum import Enum +from typing import TYPE_CHECKING +from urllib.parse import urlparse + +from pyiceberg.exceptions import ValidationException +from pyiceberg.io import _is_local_path +from pyiceberg.table import TableProperties +from pyiceberg.utils.concurrent import ExecutorFactory +from pyiceberg.utils.properties import property_as_bool + +if TYPE_CHECKING: + from pyiceberg.table import Table + +logger = logging.getLogger(__name__) + + +class PrefixMismatchMode(str, Enum): + """How to treat listed files whose URI scheme or authority differs from the referenced file. + + Files may match a referenced path component-for-component but be served through a different + scheme (s3 vs s3a) or endpoint authority. Use ``equal_schemes`` / ``equal_authorities`` to + declare equivalences; this mode chooses what to do with anything that remains ambiguous. + """ + + ERROR = "ERROR" + IGNORE = "IGNORE" + DELETE = "DELETE" + + +@dataclass(frozen=True) +class RemoveOrphanFilesResult: + """Outcome of a RemoveOrphanFiles execution.""" + + orphan_file_locations: list[str] = field(default_factory=list) + deleted_files: list[str] = field(default_factory=list) + failed_to_delete: list[str] = field(default_factory=list) + total_bytes: int = 0 + + +_DEFAULT_OLDER_THAN = timedelta(days=3) +_DEFAULT_EQUAL_SCHEMES = {"s3a": "s3", "s3n": "s3"} +_HIDDEN_PATH_PREFIXES = ("_", ".") + + +class RemoveOrphanFiles: + r"""Builder for the remove-orphan-files action. + + Usage:: + + result = table.maintenance.remove_orphan_files() \ + .older_than(datetime.now(tz=timezone.utc) - timedelta(days=7)) \ + .execute() + """ + + _table: "Table" + _location: str | None + _older_than_ms: int + _dry_run: bool + _delete_with: Callable[[str], None] | None + _prefix_mismatch_mode: PrefixMismatchMode + _equal_schemes: dict[str, str] + _equal_authorities: dict[str, str] + _compare_to_file_list: Iterable[tuple[str, datetime]] | None + + def __init__(self, table: "Table") -> None: + self._table = table + self._location = None + self._older_than_ms = _now_ms() - int(_DEFAULT_OLDER_THAN.total_seconds() * 1000) + self._dry_run = False + self._delete_with = None + self._prefix_mismatch_mode = PrefixMismatchMode.ERROR + self._equal_schemes = dict(_DEFAULT_EQUAL_SCHEMES) + self._equal_authorities = {} + self._compare_to_file_list = None + + def location(self, location: str) -> "RemoveOrphanFiles": + """Restrict the scan to a specific location. Defaults to the table's root location.""" + self._location = location + return self + + def older_than(self, value: datetime | timedelta) -> "RemoveOrphanFiles": + """Only consider files modified strictly before this point. + + Accepts either an absolute datetime or a timedelta interpreted as "files older + than this much" relative to now. Defaults to 3 days ago. + """ + if isinstance(value, timedelta): + self._older_than_ms = _now_ms() - int(value.total_seconds() * 1000) + else: + if value.tzinfo is None: + value = value.replace(tzinfo=timezone.utc) + self._older_than_ms = int(value.timestamp() * 1000) + return self + + def dry_run(self, enabled: bool = True) -> "RemoveOrphanFiles": + """When enabled, identify orphans but do not delete them.""" + self._dry_run = enabled + return self + + def delete_with(self, delete_func: Callable[[str], None]) -> "RemoveOrphanFiles": + """Use a custom deleter instead of FileIO.delete. + + Useful for dry runs that collect orphans, or for routing deletes through a + different sink. + """ + self._delete_with = delete_func + return self + + def prefix_mismatch_mode(self, mode: PrefixMismatchMode) -> "RemoveOrphanFiles": + """Set how to handle scheme/authority mismatches between listed and referenced files.""" + self._prefix_mismatch_mode = mode + return self + + def equal_schemes(self, schemes: dict[str, str]) -> "RemoveOrphanFiles": + """Declare schemes that should be considered equivalent. + + Keys may be comma-separated lists of schemes that map to the canonical value, e.g. + ``{"s3a,s3n": "s3"}``. Extends (not replaces) the default mapping. + """ + self._equal_schemes = dict(_DEFAULT_EQUAL_SCHEMES) + self._equal_schemes.update(_flatten_mapping(schemes)) + return self + + def equal_authorities(self, authorities: dict[str, str]) -> "RemoveOrphanFiles": + """Declare authorities (host[:port]) that should be considered equivalent. + + Keys may be comma-separated lists. + """ + self._equal_authorities = _flatten_mapping(authorities) + return self + + def compare_to_file_list(self, files: Iterable[tuple[str, datetime]]) -> "RemoveOrphanFiles": + """Skip the storage listing step and use the provided ``(path, last_modified)`` pairs. + + Useful when a caller has already enumerated storage (e.g. from an external inventory). + The same ``location`` and ``older_than`` filters still apply. + """ + self._compare_to_file_list = files + return self + + def execute(self) -> RemoveOrphanFilesResult: + """Run the action and return the result.""" + if not property_as_bool(self._table.metadata.properties, TableProperties.GC_ENABLED, TableProperties.GC_ENABLED_DEFAULT): + raise ValidationException( + "Cannot remove orphan files: gc.enabled is false on this table " + "(deleting files may corrupt other tables that reference them)" + ) + + scan_location = self._location or self._table.metadata.location + + referenced = self._collect_referenced_files() + + candidates = self._collect_candidate_files(scan_location) + orphans, conflicts = _find_orphans( + candidates, + referenced, + self._equal_schemes, + self._equal_authorities, + self._prefix_mismatch_mode, + ) + + if conflicts and self._prefix_mismatch_mode == PrefixMismatchMode.ERROR: + raise ValidationException( + "Unable to determine whether certain files are orphan. Metadata references " + "files that match listed files except for authority/scheme. Resolve by passing " + "equal_schemes() / equal_authorities(), or set prefix_mismatch_mode to IGNORE or " + f"DELETE. Conflicting authorities/schemes: {sorted(conflicts)}" + ) + + total_bytes = sum(size for _, size in orphans) + orphan_locations = [path for path, _ in orphans] + + if self._dry_run: + return RemoveOrphanFilesResult( + orphan_file_locations=orphan_locations, + deleted_files=[], + failed_to_delete=[], + total_bytes=total_bytes, + ) + + deleted, failed = self._delete_files(orphan_locations) + return RemoveOrphanFilesResult( + orphan_file_locations=orphan_locations, + deleted_files=deleted, + failed_to_delete=failed, + total_bytes=total_bytes, + ) + + def _collect_referenced_files(self) -> set[str]: + """Build the full set of file paths reachable from the table's metadata.""" + metadata = self._table.metadata + inspect = self._table.inspect + + referenced: set[str] = set(inspect.metadata_log_entries().column("file").to_pylist()) + referenced.update(stat.statistics_path for stat in metadata.statistics) + referenced.update(pstat.statistics_path for pstat in metadata.partition_statistics) + referenced.update(snapshot.manifest_list for snapshot in metadata.snapshots if snapshot.manifest_list) + referenced.update(inspect.all_manifests().column("path").to_pylist()) + referenced.update(inspect.all_files().column("file_path").to_pylist()) + return referenced + + def _collect_candidate_files(self, scan_location: str) -> list[tuple[str, int]]: + """List files to consider for deletion, applying the ``older_than`` filter.""" + cutoff_ms = self._older_than_ms + results: list[tuple[str, int]] = [] + + if self._compare_to_file_list is not None: + for path, ts in self._compare_to_file_list: + if ts.tzinfo is None: + ts = ts.replace(tzinfo=timezone.utc) + if path.startswith(scan_location) and int(ts.timestamp() * 1000) < cutoff_ms: + results.append((path, 0)) + return results + + hidden_partition_prefixes = _hidden_partition_prefixes(self._table) + for entry in self._table.io.list_prefix(scan_location): + # Without a modification time there is no way to tell a leftover from a file that a + # concurrent writer is about to commit, so leave it alone. + if entry.last_modified is None or int(entry.last_modified.timestamp() * 1000) >= cutoff_ms: + continue + if _is_hidden(entry.location, scan_location, hidden_partition_prefixes): + continue + results.append((entry.location, entry.size)) + return results + + def _delete_files(self, orphan_paths: list[str]) -> tuple[list[str], list[str]]: + """Delete the given files, returning the paths that were deleted and the ones that failed.""" + deleter = self._delete_with or self._table.io.delete + executor = ExecutorFactory.get_or_create() + futures = {executor.submit(deleter, path): path for path in orphan_paths} + + deleted: list[str] = [] + failed: list[str] = [] + for future in as_completed(futures): + path = futures[future] + try: + future.result() + deleted.append(path) + except Exception as e: + logger.warning("Failed to delete orphan file %s: %s", path, e) + failed.append(path) + return deleted, failed + + +def _now_ms() -> int: + return int(datetime.now(tz=timezone.utc).timestamp() * 1000) + + +def _flatten_mapping(mapping: dict[str, str]) -> dict[str, str]: + """Expand comma-separated keys, e.g. ``{"s3a,s3n": "s3"}`` → ``{"s3a": "s3", "s3n": "s3"}``.""" + out: dict[str, str] = {} + for keys, value in mapping.items(): + for key in keys.split(","): + key = key.strip() + if key: + out[key] = value.strip() + return out + + +def _hidden_partition_prefixes(table: "Table") -> tuple[str, ...]: + """Return the directory prefixes of partition fields that would otherwise look hidden. + + A table partitioned by a field named ``_c2`` writes to directories like ``_c2_trunc=AA``, + which must survive the hidden-path filter. + """ + return tuple( + f"{partition_field.name}=" + for spec in table.metadata.partition_specs + for partition_field in spec.fields + if partition_field.name.startswith(_HIDDEN_PATH_PREFIXES) + ) + + +def _is_hidden(location: str, scan_location: str, hidden_partition_prefixes: tuple[str, ...]) -> bool: + """Whether any path component below the scanned location is a hidden file or directory.""" + path = _uri_path(location) + root = _uri_path(scan_location) + if not path.startswith(root): + return False + relative = path[len(root) :] + return any( + component.startswith(_HIDDEN_PATH_PREFIXES) and not component.startswith(hidden_partition_prefixes) + for component in relative.split("/") + ) + + +@dataclass(frozen=True) +class _FileURI: + """A file location split into the components that decide whether two locations are the same.""" + + scheme: str + authority: str + path: str + + def component_match(self, other: "_FileURI") -> bool: + # An absent component on the referenced side matches anything, so metadata that stores + # bare paths does not conflict with a listing that reports full URIs. + return _component_match(self.scheme, other.scheme) and _component_match(self.authority, other.authority) + + +def _component_match(referenced: str, candidate: str) -> bool: + return not referenced or referenced.lower() == candidate.lower() + + +def _find_orphans( + candidates: list[tuple[str, int]], + referenced: Iterable[str], + equal_schemes: dict[str, str], + equal_authorities: dict[str, str], + mode: PrefixMismatchMode, +) -> tuple[list[tuple[str, int]], set[tuple[str, str]]]: + """Return (orphans, prefix-mismatch conflicts) for the given candidate/referenced sets.""" + referenced_by_path: dict[str, list[_FileURI]] = {} + for path in referenced: + uri = _file_uri(path, equal_schemes, equal_authorities) + referenced_by_path.setdefault(uri.path, []).append(uri) + + orphans: list[tuple[str, int]] = [] + conflicts: set[tuple[str, str]] = set() + for path, size in candidates: + candidate = _file_uri(path, equal_schemes, equal_authorities) + matches = referenced_by_path.get(candidate.path) + if not matches: + orphans.append((path, size)) + elif any(match.component_match(candidate) for match in matches): + continue + elif mode == PrefixMismatchMode.DELETE: + orphans.append((path, size)) + else: + conflicts.update(_conflicts(matches, candidate)) + return orphans, conflicts + + +def _conflicts(referenced: list["_FileURI"], candidate: "_FileURI") -> Iterator[tuple[str, str]]: + for match in referenced: + if not _component_match(match.scheme, candidate.scheme): + yield (match.scheme, candidate.scheme) + if not _component_match(match.authority, candidate.authority): + yield (match.authority, candidate.authority) + + +_REPEATED_SLASH = re.compile(r"/+") + + +def _file_uri(location: str, equal_schemes: dict[str, str], equal_authorities: dict[str, str]) -> "_FileURI": + """Split a location into scheme, authority and path, canonicalizing each component. + + Equivalent schemes and authorities are collapsed onto their canonical value, and runs of + slashes in the path are collapsed so ``file:///a///b`` matches ``file:///a/b``. + """ + scheme, authority, path = _split_location(location) + return _FileURI( + scheme=equal_schemes.get(scheme, scheme), + authority=equal_authorities.get(authority, authority), + path=_REPEATED_SLASH.sub("/", path), + ) + + +def _uri_path(location: str) -> str: + return _REPEATED_SLASH.sub("/", _split_location(location)[2]) + + +def _split_location(location: str) -> tuple[str, str, str]: + """Split a location into its scheme, authority and path, treating local paths as scheme-less.""" + # On Windows a drive letter parses as a URI scheme, so local paths are left whole. + if _is_local_path(location): + return "", "", location + parsed = urlparse(location) + if not parsed.scheme: + return "", "", location + return parsed.scheme, parsed.netloc, parsed.path diff --git a/tests/io/test_fsspec.py b/tests/io/test_fsspec.py index 45835a08eb..2756d3df63 100644 --- a/tests/io/test_fsspec.py +++ b/tests/io/test_fsspec.py @@ -20,6 +20,7 @@ import tempfile import threading import uuid +from pathlib import Path from unittest import mock import pytest @@ -57,6 +58,19 @@ def test_fsspec_local_fs_can_create_path_without_parent_dir(fsspec_fileio: Fsspe pytest.fail("Failed to write to file without parent directory") +def test_fsspec_list_prefix(fsspec_fileio: FsspecFileIO, tmp_path: Path) -> None: + """Test recursively listing a directory using FsspecFileIO.list_prefix(...)""" + (tmp_path / "nested").mkdir() + (tmp_path / "a.txt").write_bytes(b"foo") + (tmp_path / "nested" / "b.txt").write_bytes(b"barr") + + entries = sorted(fsspec_fileio.list_prefix(str(tmp_path)), key=lambda entry: entry.location) + + assert [Path(entry.location) for entry in entries] == [tmp_path / "a.txt", tmp_path / "nested" / "b.txt"] + assert [entry.size for entry in entries] == [3, 4] + assert all(entry.last_modified is not None for entry in entries) + + def test_fsspec_get_fs_instance_per_thread_caching(fsspec_fileio: FsspecFileIO) -> None: """Test that filesystem instances are cached per-thread by `FsspecFileIO.get_fs`""" fs_instances: list[AbstractFileSystem] = [] diff --git a/tests/io/test_pyarrow.py b/tests/io/test_pyarrow.py index b31c18949b..b33723ab5a 100644 --- a/tests/io/test_pyarrow.py +++ b/tests/io/test_pyarrow.py @@ -147,6 +147,29 @@ def test_pyarrow_local_fs_can_create_path_without_parent_dir() -> None: pytest.fail("Failed to write to file without parent directory") +def test_pyarrow_list_prefix(tmp_path: Path) -> None: + """Test recursively listing a directory using PyArrowFileIO.list_prefix(...)""" + (tmp_path / "nested").mkdir() + (tmp_path / "a.txt").write_bytes(b"foo") + (tmp_path / "nested" / "b.txt").write_bytes(b"barr") + + entries = sorted(PyArrowFileIO().list_prefix(str(tmp_path)), key=lambda entry: entry.location) + + assert [Path(entry.location) for entry in entries] == [tmp_path / "a.txt", tmp_path / "nested" / "b.txt"] + assert [entry.size for entry in entries] == [3, 4] + assert all(entry.last_modified is not None for entry in entries) + + +@pytest.mark.skipif(sys.platform == "win32", reason="A file:// URI cannot carry a Windows drive letter") +def test_pyarrow_list_prefix_retains_scheme(tmp_path: Path) -> None: + """Test that a location with a scheme is listed as URIs with that same scheme""" + (tmp_path / "a.txt").write_bytes(b"foo") + + entries = list(PyArrowFileIO().list_prefix(f"file://{tmp_path}")) + + assert [entry.location for entry in entries] == [f"file://{tmp_path}/a.txt"] + + def test_pyarrow_input_file() -> None: """Test reading a file using PyArrowFile""" diff --git a/tests/table/test_remove_orphan_files.py b/tests/table/test_remove_orphan_files.py new file mode 100644 index 0000000000..15988a96f3 --- /dev/null +++ b/tests/table/test_remove_orphan_files.py @@ -0,0 +1,467 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +"""Tests for the RemoveOrphanFiles maintenance action.""" + +from __future__ import annotations + +import os +import threading +import time +from datetime import datetime, timedelta, timezone +from functools import partial +from pathlib import Path + +import pyarrow as pa +import pytest + +from pyiceberg.catalog import Catalog +from pyiceberg.catalog.memory import InMemoryCatalog +from pyiceberg.exceptions import ValidationException +from pyiceberg.io import FileIO +from pyiceberg.partitioning import PartitionField, PartitionSpec +from pyiceberg.schema import Schema +from pyiceberg.table import Table +from pyiceberg.table.maintenance.orphan_files import ( + _DEFAULT_EQUAL_SCHEMES, + PrefixMismatchMode, + _find_orphans, + _flatten_mapping, +) +from pyiceberg.table.statistics import StatisticsFile +from pyiceberg.transforms import IdentityTransform +from pyiceberg.types import LongType, NestedField, StringType + + +def _make_table(tmp_path: Path, name: str = "default.t", properties: dict[str, str] | None = None) -> Table: + catalog: Catalog = InMemoryCatalog("test", warehouse=tmp_path.absolute().as_posix()) + catalog.create_namespace("default") + schema = Schema( + NestedField(1, "c1", LongType(), required=False), + NestedField(2, "c2", StringType(), required=False), + NestedField(3, "c3", StringType(), required=False), + ) + table = catalog.create_table(name, schema=schema, properties=properties or {}) + return table + + +def _make_partitioned_table(tmp_path: Path) -> Table: + """A table partitioned on a field whose name starts with an underscore, as in the Java tests.""" + catalog: Catalog = InMemoryCatalog("test", warehouse=tmp_path.absolute().as_posix()) + catalog.create_namespace("default") + schema = Schema( + NestedField(1, "c1", LongType(), required=False), + NestedField(2, "_c2", StringType(), required=False), + NestedField(3, "c3", StringType(), required=False), + ) + spec = PartitionSpec(PartitionField(source_id=2, field_id=1000, transform=IdentityTransform(), name="_c2")) + return catalog.create_table("default.partitioned", schema=schema, partition_spec=spec) + + +def _append(table: Table, rows: list[dict[str, object]]) -> None: + table.append(pa.Table.from_pylist(rows, schema=table.schema().as_arrow())) + + +def _backdate(path: str | Path, hours: int) -> None: + past = time.time() - hours * 3600 + os.utime(path, (past, past)) + + +def _list_data_files(table: Table) -> list[str]: + """List the files under /data in the form that list_prefix reports them.""" + out = [] + for root, _dirs, files in os.walk(f"{table.metadata.location}/data"): + for f in files: + if not f.startswith(".") and not f.startswith("_"): + out.append(_path(Path(root) / f)) + return out + + +def _path(path: Path) -> str: + """A local path in the form that list_prefix reports it, with forward slashes on any platform.""" + return path.as_posix() + + +def _wait_until_after_now() -> None: + """Sleep until the wall clock advances past the moment of the call.""" + target = time.time() + while time.time() <= target: + time.sleep(0.001) + + +def _execute_paths_test( + valid_files: list[str], + actual_files: list[str], + expected_orphans: list[str], + equal_schemes: dict[str, str] | None = None, + equal_authorities: dict[str, str] | None = None, + mode: PrefixMismatchMode = PrefixMismatchMode.IGNORE, +) -> None: + """Drive ``_find_orphans`` directly so path-normalization tests stay one-liners.""" + schemes = dict(_DEFAULT_EQUAL_SCHEMES) + schemes.update(_flatten_mapping(equal_schemes or {})) + authorities = _flatten_mapping(equal_authorities or {}) + + candidates = [(p, 0) for p in actual_files] + orphans, conflicts = _find_orphans(candidates, set(valid_files), schemes, authorities, mode) + + if mode == PrefixMismatchMode.ERROR and conflicts: + raise ValidationException(f"Unable to determine whether certain files are orphan: {sorted(conflicts)}") + + assert [p for p, _ in orphans] == expected_orphans + + +def test_dry_run(tmp_path: Path) -> None: + """Default cutoff finds nothing; explicit cutoff finds the orphan; execute deletes it.""" + table = _make_table(tmp_path) + _append(table, [{"c1": 1, "c2": "AAAAAAAAAA", "c3": "AAAA"}]) + _append(table, [{"c1": 1, "c2": "AAAAAAAAAA", "c3": "AAAA"}]) + + valid_files = set(_list_data_files(table)) + assert len(valid_files) == 2 + + orphan = Path(table.metadata.location) / "data" / "orphan.parquet" + orphan.write_bytes(b"junk") + + all_files = set(_list_data_files(table)) + invalid = sorted(all_files - valid_files) + assert invalid == [_path(orphan)] + + _wait_until_after_now() + + result1 = table.maintenance.remove_orphan_files().delete_with(lambda _: None).execute() + assert result1.orphan_file_locations == [] + + cutoff = datetime.now(tz=timezone.utc) + result2 = table.maintenance.remove_orphan_files().older_than(cutoff).delete_with(lambda _: None).execute() + assert sorted(result2.orphan_file_locations) == invalid + assert orphan.exists() + + result3 = table.maintenance.remove_orphan_files().older_than(cutoff).execute() + assert sorted(result3.deleted_files) == invalid + assert not orphan.exists() + + +def test_all_valid_files_are_kept(tmp_path: Path) -> None: + """Files referenced by any snapshot survive the action even when older than the cutoff.""" + table = _make_table(tmp_path) + _append(table, [{"c1": 1, "c2": "AAAAAAAAAA", "c3": "AAAA"}]) + _append(table, [{"c1": 2, "c2": "AAAAAAAAAA", "c3": "AAAA"}]) + _append(table, [{"c1": 3, "c2": "AAAAAAAAAA", "c3": "AAAA"}]) + + for root, _dirs, files in os.walk(table.metadata.location): + for f in files: + _backdate(Path(root) / f, hours=24 * 7) + + result = table.maintenance.remove_orphan_files().execute() + assert result.orphan_file_locations == [] + assert result.deleted_files == [] + + +def test_orphaned_files_removed_in_parallel(tmp_path: Path) -> None: + """Deletion fans out across the shared executor.""" + table = _make_table(tmp_path) + _append(table, [{"c1": 1, "c2": "AAAAAAAAAA", "c3": "AAAA"}]) + + for i in range(4): + orphan = Path(table.metadata.location) / "data" / f"orphan-{i}.parquet" + orphan.write_bytes(b"junk") + _backdate(orphan, hours=24 * 4) + + seen_threads: set[str] = set() + deleted: list[str] = [] + lock = threading.Lock() + + def record_and_delete(path: str) -> None: + # Hold each worker briefly so all four pick up a task before any returns. + time.sleep(0.05) + with lock: + seen_threads.add(threading.current_thread().name) + deleted.append(path) + + result = table.maintenance.remove_orphan_files().delete_with(record_and_delete).execute() + + assert len(deleted) == 4 + assert len(result.deleted_files) == 4 + assert len(seen_threads) > 1, f"expected parallel deletes, only saw threads: {seen_threads}" + + +def test_metadata_folder_is_intact(tmp_path: Path) -> None: + """When the scan is restricted to a data path, files under metadata/ are never touched.""" + table = _make_table(tmp_path, properties={"write.data.path": f"{tmp_path.as_posix()}/data-redirect"}) + _append(table, [{"c1": 1, "c2": "AAAAAAAAAA", "c3": "AAAA"}]) + + orphan = Path(f"{tmp_path}/data-redirect") / "stray.parquet" + orphan.parent.mkdir(parents=True, exist_ok=True) + orphan.write_bytes(b"junk") + _backdate(orphan, hours=24 * 4) + + metadata_files_before = sorted(p.name for p in Path(table.metadata.location, "metadata").iterdir()) + + result = table.maintenance.remove_orphan_files().location(f"{tmp_path.as_posix()}/data-redirect").execute() + + assert len(result.deleted_files) == 1 + assert result.deleted_files[0].endswith("stray.parquet") + metadata_files_after = sorted(p.name for p in Path(table.metadata.location, "metadata").iterdir()) + assert metadata_files_before == metadata_files_after + + +def test_older_than_timestamp(tmp_path: Path) -> None: + """Only files modified strictly before the cutoff are deleted; fresher files are kept.""" + table = _make_table(tmp_path) + _append(table, [{"c1": 1, "c2": "AAAAAAAAAA", "c3": "AAAA"}]) + + old1 = Path(table.metadata.location) / "data" / "old1.parquet" + old2 = Path(table.metadata.location) / "data" / "old2.parquet" + old1.write_bytes(b"junk") + old2.write_bytes(b"junk") + _backdate(old1, hours=2) + _backdate(old2, hours=2) + + cutoff = datetime.now(tz=timezone.utc) - timedelta(hours=1) + + fresh = Path(table.metadata.location) / "data" / "fresh.parquet" + fresh.write_bytes(b"junk") + + result = table.maintenance.remove_orphan_files().older_than(cutoff).execute() + + deleted = sorted(p.split("/")[-1] for p in result.deleted_files) + assert deleted == ["old1.parquet", "old2.parquet"] + assert fresh.exists() + + +def test_remove_unreachable_metadata_version_files(tmp_path: Path) -> None: + """A metadata.json file not tracked by metadata-log is treated as orphan.""" + table = _make_table(tmp_path) + _append(table, [{"c1": 1, "c2": "AAAAAAAAAA", "c3": "AAAA"}]) + + stray = Path(table.metadata.location) / "metadata" / "v0.unreferenced.metadata.json" + stray.write_bytes(b"{}") + _backdate(stray, hours=24 * 4) + + result = table.maintenance.remove_orphan_files().execute() + + assert any(p.endswith("v0.unreferenced.metadata.json") for p in result.deleted_files) + assert not stray.exists() + + +def test_garbage_collection_disabled(tmp_path: Path) -> None: + """The action refuses to run when the table's gc.enabled property is false.""" + table = _make_table(tmp_path) + _append(table, [{"c1": 1, "c2": "AAAAAAAAAA", "c3": "AAAA"}]) + + table.metadata = table.metadata.model_copy(update={"properties": {**table.metadata.properties, "gc.enabled": "false"}}) + + with pytest.raises(ValidationException, match="gc.enabled is false"): + table.maintenance.remove_orphan_files().execute() + + +def test_compare_to_file_list(tmp_path: Path) -> None: + """The action consumes an explicit (path, last_modified) list and still respects location().""" + table = _make_table(tmp_path) + _append(table, [{"c1": 1, "c2": "AAAAAAAAAA", "c3": "AAAA"}]) + _append(table, [{"c1": 1, "c2": "AAAAAAAAAA", "c3": "AAAA"}]) + + valid_files = set(_list_data_files(table)) + orphan = Path(table.metadata.location) / "data" / "orphan.parquet" + orphan.write_bytes(b"junk") + all_files = set(_list_data_files(table)) + invalid = sorted(all_files - valid_files) + assert invalid == [_path(orphan)] + + now = datetime.now(tz=timezone.utc) + file_list = [(p, now) for p in all_files] + + result1 = table.maintenance.remove_orphan_files().compare_to_file_list(file_list).delete_with(lambda _: None).execute() + assert result1.orphan_file_locations == [] + + cutoff = datetime.now(tz=timezone.utc) + timedelta(seconds=5) + result2 = ( + table.maintenance.remove_orphan_files() + .compare_to_file_list(file_list) + .older_than(cutoff) + .delete_with(lambda _: None) + .execute() + ) + assert sorted(result2.orphan_file_locations) == invalid + assert orphan.exists() + + result3 = table.maintenance.remove_orphan_files().compare_to_file_list(file_list).older_than(cutoff).execute() + assert sorted(result3.deleted_files) == invalid + assert not orphan.exists() + + outside = [("/tmp/mock1", datetime.fromtimestamp(0, tz=timezone.utc))] + result4 = ( + table.maintenance.remove_orphan_files() + .location(table.metadata.location) + .compare_to_file_list(outside) + .delete_with(lambda _: None) + .execute() + ) + assert result4.orphan_file_locations == [] + + +def test_remove_orphan_files_with_statistic_files(tmp_path: Path) -> None: + """Statistics files registered on the table are protected; once unregistered they become orphan.""" + table = _make_table(tmp_path) + _append(table, [{"c1": 1, "c2": "AAAAAAAAAA", "c3": "AAAA"}]) + + current_snapshot = table.metadata.current_snapshot() + assert current_snapshot is not None + snapshot_id = current_snapshot.snapshot_id + stats_path = Path(table.metadata.location) / "data" / "some-stats-file.puffin" + stats_path.parent.mkdir(parents=True, exist_ok=True) + stats_path.write_bytes(b"PFA1stub") + _backdate(stats_path, hours=24 * 4) + + stats_file = StatisticsFile( + snapshot_id=snapshot_id, + statistics_path=_path(stats_path), + file_size_in_bytes=stats_path.stat().st_size, + file_footer_size_in_bytes=4, + blob_metadata=[], + ) + table.metadata = table.metadata.model_copy(update={"statistics": [stats_file]}) + + result1 = table.maintenance.remove_orphan_files().execute() + assert all(not p.endswith("some-stats-file.puffin") for p in result1.deleted_files) + assert stats_path.exists() + + table.metadata = table.metadata.model_copy(update={"statistics": []}) + + result2 = table.maintenance.remove_orphan_files().execute() + assert any(p.endswith("some-stats-file.puffin") for p in result2.deleted_files) + assert not stats_path.exists() + + +def test_paths_with_extra_slashes() -> None: + """Runs of slashes inside a URI are collapsed during normalization.""" + _execute_paths_test( + valid_files=["file:///dir1/dir2/file1"], + actual_files=["file:///dir1/////dir2///file1"], + expected_orphans=[], + ) + + +def test_paths_with_valid_file_having_no_authority() -> None: + """A referenced file with no authority matches an authority-bearing candidate without conflict.""" + _execute_paths_test( + valid_files=["hdfs:///dir1/dir2/file1"], + actual_files=["hdfs://servicename/dir1/dir2/file1"], + expected_orphans=[], + mode=PrefixMismatchMode.ERROR, + ) + + +def test_paths_with_actual_file_having_no_authority() -> None: + """A candidate file with no authority is not flagged when the referenced version has one.""" + _execute_paths_test( + valid_files=["hdfs://servicename/dir1/dir2/file1"], + actual_files=["hdfs:///dir1/dir2/file1"], + expected_orphans=[], + ) + + +def test_paths_with_equal_schemes() -> None: + """ERROR mode raises on a scheme conflict; declaring the schemes equal silences it.""" + valid = ["scheme1://bucket1/dir1/dir2/file1"] + actual = ["scheme2://bucket1/dir1/dir2/file1"] + + with pytest.raises(ValidationException, match="scheme1.*scheme2|scheme2.*scheme1"): + _execute_paths_test(valid, actual, [], mode=PrefixMismatchMode.ERROR) + + _execute_paths_test( + valid, + actual, + [], + equal_schemes={"scheme1,scheme2": "scheme"}, + mode=PrefixMismatchMode.ERROR, + ) + + +def test_paths_with_equal_authorities() -> None: + """ERROR mode raises on an authority conflict; declaring the authorities equal silences it.""" + valid = ["hdfs://servicename1/dir1/dir2/file1"] + actual = ["hdfs://servicename2/dir1/dir2/file1"] + + with pytest.raises(ValidationException, match="servicename1.*servicename2|servicename2.*servicename1"): + _execute_paths_test(valid, actual, [], mode=PrefixMismatchMode.ERROR) + + _execute_paths_test( + valid, + actual, + [], + equal_authorities={"servicename1,servicename2": "servicename"}, + mode=PrefixMismatchMode.ERROR, + ) + + +def test_remove_orphan_file_action_with_delete_mode() -> None: + """DELETE mode treats prefix-mismatched candidates as orphan rather than recording a conflict.""" + _execute_paths_test( + valid_files=["hdfs://servicename1/dir1/dir2/file1"], + actual_files=["hdfs://servicename2/dir1/dir2/file1"], + expected_orphans=["hdfs://servicename2/dir1/dir2/file1"], + mode=PrefixMismatchMode.DELETE, + ) + + +def test_hidden_paths_are_ignored(tmp_path: Path) -> None: + """Files under a path component starting with '_' or '.' are never considered orphan.""" + table = _make_table(tmp_path) + _append(table, [{"c1": 1, "c2": "AAAAAAAAAA", "c3": "AAAA"}]) + + data_dir = Path(table.metadata.location) / "data" + hidden = [data_dir / "_SUCCESS", data_dir / ".staging" / "part.parquet", data_dir / "_temporary" / "part.parquet"] + visible = data_dir / "stray.parquet" + for path in [*hidden, visible]: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(b"junk") + _backdate(path, hours=24 * 4) + + result = table.maintenance.remove_orphan_files().execute() + + assert result.deleted_files == [_path(visible)] + assert all(path.exists() for path in hidden) + + +def test_hidden_partition_paths_are_not_ignored(tmp_path: Path) -> None: + """A partition directory named after a '_'-prefixed field is scanned like any other.""" + table = _make_partitioned_table(tmp_path) + _append(table, [{"c1": 1, "_c2": "AAAAAAAAAA", "c3": "AAAA"}]) + + data_dir = Path(table.metadata.location) / "data" + in_partition = data_dir / "_c2=AAAAAAAAAA" / "stray.parquet" + in_hidden_lookalike = data_dir / "_c2" / "file.txt" + for path in (in_partition, in_hidden_lookalike): + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(b"junk") + _backdate(path, hours=24 * 4) + + result = table.maintenance.remove_orphan_files().execute() + + assert result.deleted_files == [_path(in_partition)] + assert in_hidden_lookalike.exists() + + +def test_file_io_without_list_prefix(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """A FileIO that cannot list storage fails with a clear error.""" + table = _make_table(tmp_path) + _append(table, [{"c1": 1, "c2": "AAAAAAAAAA", "c3": "AAAA"}]) + monkeypatch.setattr(table.io, "list_prefix", partial(FileIO.list_prefix, table.io)) + + with pytest.raises(NotImplementedError, match="does not support list_prefix"): + table.maintenance.remove_orphan_files().execute()