qzyu999 commented on code in PR #3320:
URL: https://github.com/apache/iceberg-python/pull/3320#discussion_r3425488390
##########
pyiceberg/table/update/snapshot.py:
##########
@@ -91,19 +96,35 @@ def _new_manifest_list_file_name(snapshot_id: int, attempt:
int, commit_uuid: uu
return f"snap-{snapshot_id}-{attempt}-{commit_uuid}.avro"
+@dataclass
+class CommitWindow:
+ """Tracks the commit range to validate against during retry.
+
+ starting_snapshot_id: The snapshot when the operation began (fixed across
retries).
+ catalog_head_snapshot_id: The catalog's latest HEAD snapshot (updated on
each retry).
+ """
+
+ starting_snapshot_id: int | None
+ catalog_head_snapshot_id: int | None
+
+
class _SnapshotProducer(UpdateTableMetadata[U], Generic[U]):
commit_uuid: uuid.UUID
_io: FileIO
_operation: Operation
_snapshot_id: int
_parent_snapshot_id: int | None
+ _starting_snapshot_id: int | None
_added_data_files: list[DataFile]
_manifest_num_counter: itertools.count[int]
_deleted_data_files: set[DataFile]
_compression: AvroCompressionCodec
_target_branch: str | None
_predicate: BooleanExpression
_case_sensitive: bool
+ _commit_window: CommitWindow | None
+ _written_manifests: list[str]
+ _uncommitted_manifests: list[str]
Review Comment:
We can make sure that `_SnapshotProducer` has a variable to track all
manifest lists like with `_written_manifests`:
```python
_written_manifest_lists: list[str]
```
##########
pyiceberg/table/update/snapshot.py:
##########
Review Comment:
It's at this point where we create the manifest list, but never save it
anywhere so it becomes orphaned in the case of a failed commit. So we add it to
our `_written_manifest_lists` variable here:
```python
self._written_manifest_lists.append(manifest_list_file_path)
```
##########
pyiceberg/table/update/snapshot.py:
##########
@@ -353,11 +379,123 @@ def new_manifest_output(self) -> OutputFile:
location_provider = self._transaction._table.location_provider()
file_name =
_new_manifest_file_name(num=next(self._manifest_num_counter),
commit_uuid=self.commit_uuid)
file_path = location_provider.new_metadata_location(file_name)
+ self._written_manifests.append(file_path)
return self._io.new_output(file_path)
def fetch_manifest_entry(self, manifest: ManifestFile, discard_deleted:
bool = True) -> list[ManifestEntry]:
return manifest.fetch_manifest_entry(io=self._io,
discard_deleted=discard_deleted)
+ def commit(self) -> None:
+ self._transaction._register_snapshot_producer(self)
+ self._transaction._apply(*self._commit())
+
+ def _cleanup_uncommitted(self) -> None:
+ """Delete manifest files from failed retry attempts."""
+ for path in self._uncommitted_manifests:
+ try:
+ self._io.delete(path)
+ except Exception:
+ logger.warning("Failed to delete uncommitted manifest: %s",
path, exc_info=True)
+ self._uncommitted_manifests.clear()
+
+ def _clean_all_uncommitted(self) -> None:
+ """Clean up all manifests written during this producer's lifecycle on
abort."""
+ for path in itertools.chain(self._uncommitted_manifests,
self._written_manifests):
+ try:
+ self._io.delete(path)
+ except Exception:
+ logger.warning("Failed to delete uncommitted manifest: %s",
path, exc_info=True)
+ self._uncommitted_manifests.clear()
+ self._written_manifests.clear()
Review Comment:
In `_cleanup_uncommitted` and `_clean_all_uncommitted`, we make sure to also
delete the potentially orphaned manifest lists, copying the pattern in the Java
implementation where after a successful commit, all but the manifest list that
was actually used:
```python
def _cleanup_uncommitted(self) -> None:
"""Delete manifest files and manifest lists from failed retry
attempts."""
for path in self._uncommitted_manifests:
try:
self._io.delete(path)
except Exception:
logger.warning("Failed to delete uncommitted manifest: %s",
path, exc_info=True)
self._uncommitted_manifests.clear()
# Delete all manifest lists except the last one (which the committed
snapshot references)
if len(self._written_manifest_lists) > 1:
for path in self._written_manifest_lists[:-1]:
try:
self._io.delete(path)
except Exception:
logger.warning("Failed to delete uncommitted manifest
list: %s", path, exc_info=True)
self._written_manifest_lists = self._written_manifest_lists[-1:]
def _clean_all_uncommitted(self) -> None:
"""Clean up all manifests and manifest lists on abort."""
for path in itertools.chain(self._uncommitted_manifests,
self._written_manifests):
try:
self._io.delete(path)
except Exception:
logger.warning("Failed to delete uncommitted manifest: %s",
path, exc_info=True)
for path in self._written_manifest_lists:
try:
self._io.delete(path)
except Exception:
logger.warning("Failed to delete uncommitted manifest list:
%s", path, exc_info=True)
self._uncommitted_manifests.clear()
self._written_manifests.clear()
self._written_manifest_lists.clear()
```
##########
pyiceberg/table/update/snapshot.py:
##########
@@ -123,6 +144,8 @@ def __init__(
self._deleted_data_files = set()
self.snapshot_properties = snapshot_properties
self._manifest_num_counter = itertools.count(0)
+ self._written_manifests = []
+ self._uncommitted_manifests = []
from pyiceberg.table import TableProperties
Review Comment:
The variable is initialized as an empty list like `_written_manifests`:
```python
self._written_manifest_lists = []
```
##########
tests/table/test_validate.py:
##########
Review Comment:
Here we can add two tests, checking that the manifest list clean up works on
retry and abort:
```python
def test_manifest_list_cleanup_on_retry(catalog: Catalog) -> None:
"""Verify that manifest list files from failed retry attempts are
cleaned up.
Background:
- Each commit attempt writes a manifest list file
(snap-{id}-0-{uuid}.avro)
- When a commit fails and retries, the OLD manifest list becomes orphaned
- This test verifies that after a successful retry, the orphaned manifest
list from the failed attempt is deleted from storage
Flow:
1. tbl1 and tbl2 both load the table at snapshot S1
2. tbl1.append() succeeds, advancing catalog to S2
3. tbl2.append() attempt 1:
- Writes manifest list "snap-A-0-{uuid1}.avro" to storage
- _do_commit() fails because tbl2 expected S1 but catalog is at S2
4. tbl2.append() attempt 2 (retry):
- Writes NEW manifest list "snap-B-0-{uuid2}.avro"
- _do_commit() succeeds
5. Cleanup:
- "snap-A-0-{uuid1}.avro" should be DELETED (it's orphaned)
- "snap-B-0-{uuid2}.avro" stays (referenced by committed snapshot)
"""
catalog.create_namespace("default")
schema = _test_schema()
catalog.create_table("default.manifest_list_cleanup_test", schema=schema)
import pyarrow as pa
df = pa.table({"x": [1, 2, 3]})
# Seed the table with one snapshot so there's a branch HEAD to conflict
with
tbl = catalog.load_table("default.manifest_list_cleanup_test")
tbl.append(df)
# Load the same table twice to simulate two concurrent writers.
# Both see the SAME snapshot (S1). Whoever commits second will get
# CommitFailedException on their first attempt.
tbl1 = catalog.load_table("default.manifest_list_cleanup_test")
tbl2 = catalog.load_table("default.manifest_list_cleanup_test")
# tbl1 commits first — advances catalog HEAD to S2.
# tbl2 still thinks HEAD is S1, so its next commit will fail on first
attempt.
tbl1.append(df)
# Intercept tbl2.io.delete() to record every file path that gets deleted.
# This lets us verify cleanup happened without inspecting internal state.
deleted_paths: list[str] = []
original_delete = tbl2.io.delete
def tracking_delete(path: str) -> None:
deleted_paths.append(path)
original_delete(path)
# tbl2.append(df) will:
# - Attempt 1: write manifest list → fail with CommitFailedException
# - Retry: refresh metadata, rebuild snapshot, write NEW manifest list
→ succeed
# - Cleanup: delete the manifest list from attempt 1
with patch.object(tbl2.io, "delete", side_effect=tracking_delete):
tbl2.append(df)
# ASSERTION: At least one manifest list file (snap-*.avro) was deleted.
# This proves the orphaned manifest list from the failed first attempt
# was tracked and cleaned up. Without the _written_manifest_lists
tracking,
# this list would be empty — only individual manifests ({uuid}-m0.avro)
would
# appear in deleted_paths.
manifest_list_deletes = [p for p in deleted_paths if "snap-" in p and
p.endswith(".avro")]
assert len(manifest_list_deletes) >= 1, (
f"Expected at least one orphaned manifest list (snap-*-*.avro) to be
cleaned up.\n"
f"Deleted paths were: {deleted_paths}\n"
f"This means the manifest list from the failed retry attempt was NOT
tracked for cleanup."
)
# Sanity check: the actual data commit succeeded (both appends visible)
refreshed = catalog.load_table("default.manifest_list_cleanup_test")
assert len(refreshed.scan().to_arrow()) == 9 # 3 (seed) + 3 (tbl1) + 3
(tbl2)
def test_manifest_list_cleanup_on_abort(catalog: Catalog) -> None:
"""Verify that ALL manifest lists are cleaned up when a commit
permanently fails.
Background:
- When validation detects an unresolvable conflict (e.g. two writers
deleted
the same rows), it raises ValidationException — no more retries.
- On abort, _clean_all_uncommitted() should delete ALL written files,
including manifest list files, not just manifests.
Flow:
1. Table has rows [1, 2, 3]. tbl1 and tbl2 both load at this snapshot.
2. tbl1.delete("x == 1") succeeds — removes row 1, advances catalog HEAD.
3. tbl2.delete("x == 1") attempt 1:
- Writes manifest list to storage
- _do_commit() fails (stale snapshot)
4. Retry validation:
- Refreshes metadata, sees tbl1 already deleted files matching "x ==
1"
- _validate_concurrency() raises ValidationException (conflicting
deletes)
5. Abort path (except Exception):
- _clean_all_uncommitted() should delete ALL manifests AND manifest
lists
- The manifest list from attempt 1 should be deleted
"""
catalog.create_namespace("default")
schema = _test_schema()
catalog.create_table("default.manifest_list_abort_test", schema=schema)
import pyarrow as pa
from pyiceberg.exceptions import ValidationException
df = pa.table({"x": [1, 2, 3]})
# Seed the table
tbl = catalog.load_table("default.manifest_list_abort_test")
tbl.append(df)
# Two writers looking at the same snapshot
tbl1 = catalog.load_table("default.manifest_list_abort_test")
tbl2 = catalog.load_table("default.manifest_list_abort_test")
# tbl1 deletes x==1 first. This means the data file containing x==1
# has been removed. When tbl2 tries the same delete, validation will
# detect this as a conflict (the file it wants to delete is already
gone).
tbl1.delete("x == 1")
# Intercept deletes to verify cleanup
deleted_paths: list[str] = []
original_delete = tbl2.io.delete
def tracking_delete(path: str) -> None:
deleted_paths.append(path)
original_delete(path)
# tbl2.delete("x == 1") will:
# - Attempt 1: write manifest + manifest list → CommitFailedException
# - Retry: refresh, _validate_concurrency() sees tbl1 already deleted
# the same files → raises ValidationException (permanent failure)
# - Abort: _clean_all_uncommitted() should delete ALL written artifacts
with patch.object(tbl2.io, "delete", side_effect=tracking_delete):
with pytest.raises(ValidationException):
tbl2.delete("x == 1")
# ASSERTION: At least one manifest list file was deleted on abort.
# _clean_all_uncommitted() iterates _written_manifest_lists and deletes
them all.
# Without tracking, this would be empty — only manifests in
_written_manifests
# and _uncommitted_manifests would be cleaned.
manifest_list_deletes = [p for p in deleted_paths if "snap-" in p and
p.endswith(".avro")]
assert len(manifest_list_deletes) >= 1, (
f"Expected manifest list cleanup on abort (ValidationException).\n"
f"Deleted paths were: {deleted_paths}\n"
f"This means _clean_all_uncommitted() did not clean manifest list
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
