You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "rambleraptor (via GitHub)" <gi...@apache.org> on 2026/05/14 22:12:59 UTC
[PR] remove orphan files [iceberg-python]
rambleraptor opened a new pull request, #3361:
URL: https://github.com/apache/iceberg-python/pull/3361
<!--
Thanks for opening a pull request!
-->
<!-- In the case this PR will resolve an issue, please replace ${GITHUB_ISSUE_ID} below with the actual Github issue id. -->
<!-- Closes #${GITHUB_ISSUE_ID} -->
# Rationale for this change
This adds support for the RemoveOrphanFiles metadata maintenance task. The goal is to match the Java implementation.
I had to add a list method to FileIO in order to fully implement this. I can separate that work into a separate PR if that's more useful.
A good follow-up would be to wire this into the CLI. Doing these ad-hoc actions without having to write a script / spin up a Spark cluster is a huge win!
## Are these changes tested?
I did some local testing where I took a table with orphaned files and tried both the Java/PyIceberg implementations against it. Results were the same.
There's also plenty of tests.
## Are there any user-facing changes?
- Adds support for the RemoveOrphanFiles maintenance action
<!-- In the case of user-facing changes, please add the changelog label. -->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] remove orphan files [iceberg-python]
Posted by "jayceslesar (via GitHub)" <gi...@apache.org>.
jayceslesar commented on code in PR #3361:
URL: https://github.com/apache/iceberg-python/pull/3361#discussion_r3928819359
##########
pyiceberg/table/maintenance/orphan_files.py:
##########
@@ -0,0 +1,414 @@
+# 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.
+"""
+
+from __future__ import annotations
+
+import logging
+import re
+from collections.abc import Callable, Iterable
+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.utils.concurrent import ExecutorFactory
+from pyiceberg.utils.properties import property_as_bool
+
+if TYPE_CHECKING:
+ from pyiceberg.io import FileEntry
+ 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"}
+
+
+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
+ _max_concurrency: int | 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._max_concurrency = 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 max_concurrency(self, max_workers: int) -> RemoveOrphanFiles:
+ """Override the worker count for manifest reads and deletes."""
+ if max_workers <= 0:
+ raise ValueError(f"max_concurrency must be positive, got {max_workers}")
+ self._max_concurrency = max_workers
+ 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."""
+ properties = self._table.metadata.properties
+ if not property_as_bool(properties, "gc.enabled", True):
+ raise ValueError(
+ "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 ValueError(
+ "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 prefixes: {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 current metadata."""
Review Comment:
Why reimplement this instead of leaning on the inspect interface?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] remove orphan files [iceberg-python]
Posted by "jayceslesar (via GitHub)" <gi...@apache.org>.
jayceslesar commented on PR #3361:
URL: https://github.com/apache/iceberg-python/pull/3361#issuecomment-5586093387
probably worth also having an integration test that runs the same setup twice, and calls remove orphans via spark and another time via this, and confirms results are the same
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] remove orphan files [iceberg-python]
Posted by "jayceslesar (via GitHub)" <gi...@apache.org>.
jayceslesar commented on code in PR #3361:
URL: https://github.com/apache/iceberg-python/pull/3361#discussion_r3951349904
##########
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":
Review Comment:
i think what you have here matches the java super well, ill need to do another review but this should match too imo (only taking a timestamp, or a datetime for us in python)
https://iceberg.apache.org/javadoc/1.11.0/org/apache/iceberg/actions/DeleteOrphanFiles.html#olderThan(long)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] remove orphan files [iceberg-python]
Posted by "djouallah (via GitHub)" <gi...@apache.org>.
djouallah commented on PR #3361:
URL: https://github.com/apache/iceberg-python/pull/3361#issuecomment-4583059021
please i need this, i am writing using duckdb, but they don't support any maintenance and i am drowning in old data :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] DeleteOrphanFiles maintenance task [iceberg-python]
Posted by "jayceslesar (via GitHub)" <gi...@apache.org>.
jayceslesar commented on code in PR #3361:
URL: https://github.com/apache/iceberg-python/pull/3361#discussion_r3996674705
##########
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":
Review Comment:
also better matches https://github.com/apache/iceberg-python/blob/main/pyiceberg/table/update/snapshot.py#L1412
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] DeleteOrphanFiles maintenance task [iceberg-python]
Posted by "rambleraptor (via GitHub)" <gi...@apache.org>.
rambleraptor commented on PR #3361:
URL: https://github.com/apache/iceberg-python/pull/3361#issuecomment-6074132384
That's great to hear!
@Fokko could we look into getting this reviewed or maybe even merged in 😄?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] remove orphan files [iceberg-python]
Posted by "rambleraptor (via GitHub)" <gi...@apache.org>.
rambleraptor commented on code in PR #3361:
URL: https://github.com/apache/iceberg-python/pull/3361#discussion_r3929834093
##########
pyiceberg/table/maintenance/orphan_files.py:
##########
@@ -0,0 +1,414 @@
+# 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.
+"""
+
+from __future__ import annotations
+
+import logging
+import re
+from collections.abc import Callable, Iterable
+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.utils.concurrent import ExecutorFactory
+from pyiceberg.utils.properties import property_as_bool
+
+if TYPE_CHECKING:
+ from pyiceberg.io import FileEntry
+ 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"}
+
+
+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
+ _max_concurrency: int | 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._max_concurrency = 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 max_concurrency(self, max_workers: int) -> RemoveOrphanFiles:
Review Comment:
Yep, great call. Fixed!
##########
pyiceberg/table/maintenance/orphan_files.py:
##########
@@ -0,0 +1,414 @@
+# 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.
+"""
+
+from __future__ import annotations
+
+import logging
+import re
+from collections.abc import Callable, Iterable
+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.utils.concurrent import ExecutorFactory
+from pyiceberg.utils.properties import property_as_bool
+
+if TYPE_CHECKING:
+ from pyiceberg.io import FileEntry
+ 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"}
+
+
+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
+ _max_concurrency: int | 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._max_concurrency = 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 max_concurrency(self, max_workers: int) -> RemoveOrphanFiles:
+ """Override the worker count for manifest reads and deletes."""
+ if max_workers <= 0:
+ raise ValueError(f"max_concurrency must be positive, got {max_workers}")
+ self._max_concurrency = max_workers
+ 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."""
+ properties = self._table.metadata.properties
+ if not property_as_bool(properties, "gc.enabled", True):
+ raise ValueError(
+ "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 ValueError(
+ "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 prefixes: {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 current metadata."""
Review Comment:
Yeah, we can do that. It's much cleaner. Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] remove orphan files [iceberg-python]
Posted by "jayceslesar (via GitHub)" <gi...@apache.org>.
jayceslesar commented on code in PR #3361:
URL: https://github.com/apache/iceberg-python/pull/3361#discussion_r3928799058
##########
pyiceberg/table/maintenance/orphan_files.py:
##########
@@ -0,0 +1,414 @@
+# 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.
+"""
+
+from __future__ import annotations
Review Comment:
```suggestion
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] remove orphan files [iceberg-python]
Posted by "qzyu999 (via GitHub)" <gi...@apache.org>.
qzyu999 commented on PR #3361:
URL: https://github.com/apache/iceberg-python/pull/3361#issuecomment-4494900244
Hi @rambleraptor, thanks for the PR, quick comment, it may make sense to link that this Closes #1200. I do see a related PR #1958, but it looks potentially abandoned (CC: @jayceslesar).
_NOTE:_ I notice that you call it remove orphan files, while the linked issue is called delete orphan files. The Java code itself calls it `DeleteOrphanFiles`, however the API for Spark and Trino use `remove_orphan_files`. Did a bit of research to try and understand why. Seems to potentially avoid ambiguity regarding doing row-level deleting, making it more user-friendly where "remove" is known to be a house-keeping type operation. The Java naming of choosing "delete" actually matches their CRU**D** interface since they are performing a delete operation on files.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] remove orphan files [iceberg-python]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #3361:
URL: https://github.com/apache/iceberg-python/pull/3361#issuecomment-5223468975
This pull request has been marked as stale due to 30 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that's incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@iceberg.apache.org list. Thank you for your contributions.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] DeleteOrphanFiles maintenance task [iceberg-python]
Posted by "djouallah (via GitHub)" <gi...@apache.org>.
djouallah commented on PR #3361:
URL: https://github.com/apache/iceberg-python/pull/3361#issuecomment-6072278482
We're running this in production now: a build of `4adf237` in the daily maintenance of 25 Iceberg tables written by DuckDB to the OneLake (Microsoft Fabric) Iceberg REST catalog, right after `expire_snapshots`. The first pass deleted 14,563 orphan files (~11.5 GB), with no failed deletes. Our tests passed afterwards, and the next pass found only a handful of files that had just crossed the 3-day line. Thank you!
A few notes from running it on ADLS:
1. **The fsspec listing drops the ADLS account.** adlfs lists `abfss://<container>/...` while the metadata references `abfss://<container>@<account-host>/...`. As a result, every table hit the prefix-mismatch `ValidationException`. The default `ERROR` mode did exactly the right thing there. We worked around it with `equal_authorities({container: f"{container}@{host}"})`. #3974's `list_prefix` already puts the netloc back for `_ADLS_SCHEMES`, so rebasing this PR on #3974 instead of carrying its own copy would fix it. An ADLS case in the tests would catch it.
2. **`compare_to_file_list` skips the hidden-path filter.** It returns before `_is_hidden`, so an inventory that includes `_delta_log/` or `.crc` files would delete them. It also reports a size of 0 for every file, so `total_bytes` is wrong on that path.
3. **Deletes run one call per file.** 6,995 files took about 7 minutes on ADLS. A bulk path would help a lot when there is a backlog, for example `fs.rm(list)` in fsspec, the counterpart of Java's `SupportsBulkOperations`.
Happy to test a new revision against the same tables.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] remove orphan files [iceberg-python]
Posted by "rambleraptor (via GitHub)" <gi...@apache.org>.
rambleraptor commented on PR #3361:
URL: https://github.com/apache/iceberg-python/pull/3361#issuecomment-5589598334
@jayceslesar that's a fantastic idea! Added in the integration test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] remove orphan files [iceberg-python]
Posted by "jayceslesar (via GitHub)" <gi...@apache.org>.
jayceslesar commented on code in PR #3361:
URL: https://github.com/apache/iceberg-python/pull/3361#discussion_r3928832014
##########
pyiceberg/table/maintenance/orphan_files.py:
##########
@@ -0,0 +1,414 @@
+# 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.
+"""
+
+from __future__ import annotations
+
+import logging
+import re
+from collections.abc import Callable, Iterable
+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.utils.concurrent import ExecutorFactory
+from pyiceberg.utils.properties import property_as_bool
+
+if TYPE_CHECKING:
+ from pyiceberg.io import FileEntry
+ 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"}
+
+
+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
+ _max_concurrency: int | 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._max_concurrency = 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 max_concurrency(self, max_workers: int) -> RemoveOrphanFiles:
Review Comment:
why not just use https://github.com/apache/iceberg-python/blob/main/pyiceberg/utils/concurrent.py#L30
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] DeleteOrphanFiles maintenance task [iceberg-python]
Posted by "jayceslesar (via GitHub)" <gi...@apache.org>.
jayceslesar commented on code in PR #3361:
URL: https://github.com/apache/iceberg-python/pull/3361#discussion_r3996684933
##########
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":
Review Comment:
yuck lol different kwarg names for essentially the same meaning
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org