Introduce CommitWindow to fix validation in multi-producer retry

_validate_concurrency was resolving the catalog HEAD via
_parent_snapshot_id, which breaks when a Transaction has multiple
producers (_DeleteFiles + _OverwriteFiles). On retry, the second
producer's parent points to the first producer's uncommitted snapshot,
which does not exist in the catalog metadata, causing a spurious
ValidationException.

CommitWindow explicitly separates the two snapshot references that
validation needs: starting_snapshot_id (fixed at operation start) and
catalog_head_snapshot_id (the actual catalog HEAD after refresh).
_rebuild_snapshot_updates builds one CommitWindow and shares it across
all producers, so each validates against the real catalog state.

Add test_mixed_delete_overwrite_retries_successfully to cover the
multi-producer retry path end-to-end.

Signed-off-by: Sotaro Hikita <bering1814@gmail.com>

Author: lawofcycles

pyiceberg/table/__init__.py

@@ -1133,11 +1133,20 @@ def _cleanup_uncommitted_manifests(self) -> None:
     def _rebuild_snapshot_updates(self) -> None:
         """Rebuild snapshot updates for retry by re-executing registered producers."""
         from pyiceberg.table.update import AddSnapshotUpdate, AssertRefSnapshotId, SetSnapshotRefUpdate
+        from pyiceberg.table.update.snapshot import CommitWindow
 
         self._updates = tuple(u for u in self._updates if not isinstance(u, (AddSnapshotUpdate, SetSnapshotRefUpdate)))
         self._requirements = tuple(r for r in self._requirements if not isinstance(r, AssertRefSnapshotId))
 
+        # Build CommitWindow: starting_snapshot_id is from the first producer (fixed at operation start),
+        # catalog_head_snapshot_id is the current catalog HEAD after refresh.
+        starting_id = self._snapshot_producers[0]._starting_snapshot_id if self._snapshot_producers else None
+        current_snapshot = self._table.metadata.current_snapshot()
+        catalog_head_id = current_snapshot.snapshot_id if current_snapshot else None
+        commit_window = CommitWindow(starting_snapshot_id=starting_id, catalog_head_snapshot_id=catalog_head_id)
+
         for producer in self._snapshot_producers:
+            producer._commit_window = commit_window
             producer._refresh_for_retry()
             producer._validate_concurrency()
             updates, requirements = producer._commit()

pyiceberg/table/update/snapshot.py

@@ -22,6 +22,7 @@
 from abc import abstractmethod
 from collections import defaultdict
 from collections.abc import Callable
+from dataclasses import dataclass
 from datetime import datetime
 from functools import cached_property
 from typing import TYPE_CHECKING, Generic
@@ -95,6 +96,18 @@ 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
@@ -109,6 +122,7 @@ class _SnapshotProducer(UpdateTableMetadata[U], Generic[U]):
     _target_branch: str | None
     _predicate: BooleanExpression
     _case_sensitive: bool
+    _commit_window: CommitWindow | None
     _written_manifests: list[str]
     _uncommitted_manifests: list[str]
 
@@ -144,6 +158,7 @@ def __init__(
         self._starting_snapshot_id = self._parent_snapshot_id
         self._predicate = AlwaysFalse()
         self._case_sensitive = True
+        self._commit_window = None
         self._isolation_level_property: str = TableProperties.WRITE_DELETE_ISOLATION_LEVEL
 
     def _validate_target_branch(self, branch: str | None) -> str | None:
@@ -407,8 +422,10 @@ def _refresh_for_retry(self) -> None:
     def _validate_concurrency(self) -> None:
         """Validate that concurrent changes do not conflict with this operation.
 
-        Checks isolation level and uses the conflict detection filter to determine
-        whether concurrent commits introduced conflicting data or delete files.
+        Uses the CommitWindow to determine which catalog commits to validate against.
+        The window spans from starting_snapshot (when the operation began) to the
+        catalog HEAD (latest committed snapshot), covering all external concurrent commits.
+
         Subclasses that do not require validation (e.g. fast append) should override
         with a no-op.
         """
@@ -421,8 +438,11 @@ def _validate_concurrency(self) -> None:
             _validate_no_new_deletes_for_data_files,
         )
 
-        parent_snapshot = self._resolve_parent_snapshot()
-        if parent_snapshot is None:
+        if self._commit_window is None:
+            return
+
+        catalog_head = self._resolve_catalog_head_snapshot()
+        if catalog_head is None:
             return
 
         starting_snapshot = self._resolve_starting_snapshot()
@@ -435,16 +455,27 @@ def _validate_concurrency(self) -> None:
         conflict_detection_filter = self._predicate if self._predicate != AlwaysFalse() else None
 
         if isolation_level == IsolationLevel.SERIALIZABLE:
-            _validate_added_data_files(table, parent_snapshot, conflict_detection_filter, starting_snapshot)
+            _validate_added_data_files(table, catalog_head, conflict_detection_filter, starting_snapshot)
 
         if conflict_detection_filter is not None:
-            _validate_no_new_delete_files(table, parent_snapshot, conflict_detection_filter, None, starting_snapshot)
-            _validate_deleted_data_files(table, parent_snapshot, conflict_detection_filter, starting_snapshot)
+            _validate_no_new_delete_files(table, catalog_head, conflict_detection_filter, None, starting_snapshot)
+            _validate_deleted_data_files(table, catalog_head, conflict_detection_filter, starting_snapshot)
 
         if self._deleted_data_files:
             _validate_no_new_deletes_for_data_files(
-                table, parent_snapshot, conflict_detection_filter, self._deleted_data_files, starting_snapshot
+                table, catalog_head, conflict_detection_filter, self._deleted_data_files, starting_snapshot
+            )
+
+    def _resolve_catalog_head_snapshot(self) -> Snapshot | None:
+        """Resolve the catalog HEAD snapshot from the CommitWindow."""
+        if self._commit_window is None or self._commit_window.catalog_head_snapshot_id is None:
+            return None
+        snapshot = self._transaction._table.metadata.snapshot_by_id(self._commit_window.catalog_head_snapshot_id)
+        if snapshot is None:
+            raise ValidationException(
+                f"Cannot find catalog head snapshot {self._commit_window.catalog_head_snapshot_id} in table metadata"
             )
+        return snapshot
 
     def _resolve_parent_snapshot(self) -> Snapshot | None:
         """Resolve parent snapshot, raising ValidationException if ID is set but snapshot is missing."""
@@ -456,8 +487,8 @@ def _resolve_parent_snapshot(self) -> Snapshot | None:
         return snapshot
 
     def _resolve_starting_snapshot(self) -> Snapshot:
-        """Resolve starting snapshot for the conflict detection window."""
-        starting_id = self._starting_snapshot_id if self._starting_snapshot_id is not None else self._parent_snapshot_id
+        """Resolve starting snapshot for the conflict detection window from the CommitWindow."""
+        starting_id = self._commit_window.starting_snapshot_id if self._commit_window else self._starting_snapshot_id
         if starting_id is None:
             raise ValidationException("Cannot resolve starting snapshot: both starting and parent snapshot IDs are None")
         snapshot = self._transaction._table.metadata.snapshot_by_id(starting_id)

tests/table/test_commit_retry.py

@@ -632,8 +632,8 @@ def test_mixed_delete_overwrite_starts_from_catalog_snapshot(catalog: Catalog) -
     assert overwrite_producer._starting_snapshot_id == base_snapshot_id
 
 
-def test_validate_concurrency_raises_on_missing_parent_snapshot(catalog: Catalog) -> None:
-    """Validation should raise when parent_snapshot_id is non-null but cannot be resolved."""
+def test_validate_concurrency_raises_on_missing_catalog_head_snapshot(catalog: Catalog) -> None:
+    """Validation should raise when catalog_head_snapshot_id is non-null but cannot be resolved."""
     catalog.create_namespace("default")
     schema = _test_schema()
     table = catalog.create_table("default.missing_parent_test", schema=schema)
@@ -642,7 +642,7 @@ def test_validate_concurrency_raises_on_missing_parent_snapshot(catalog: Catalog
 
     table.append(pa.table({"x": [1, 2, 3]}))
 
-    from pyiceberg.table.update.snapshot import _DeleteFiles
+    from pyiceberg.table.update.snapshot import CommitWindow, _DeleteFiles
 
     tx = Transaction(table, autocommit=False)
     producer = _DeleteFiles(
@@ -651,10 +651,12 @@ def test_validate_concurrency_raises_on_missing_parent_snapshot(catalog: Catalog
         io=table.io,
     )
 
-    # Artificially set a non-existent parent snapshot ID
-    producer._parent_snapshot_id = 99999999
+    # Set a CommitWindow with a non-existent catalog head snapshot ID
+    producer._commit_window = CommitWindow(
+        starting_snapshot_id=table.metadata.current_snapshot_id, catalog_head_snapshot_id=99999999
+    )
 
-    with pytest.raises(ValidationException, match="Cannot find parent snapshot"):
+    with pytest.raises(ValidationException, match="Cannot find catalog head snapshot"):
         producer._validate_concurrency()
 
 
@@ -668,7 +670,7 @@ def test_validate_concurrency_raises_on_missing_starting_snapshot(catalog: Catal
 
     table.append(pa.table({"x": [1, 2, 3]}))
 
-    from pyiceberg.table.update.snapshot import _DeleteFiles
+    from pyiceberg.table.update.snapshot import CommitWindow, _DeleteFiles
 
     tx = Transaction(table, autocommit=False)
     producer = _DeleteFiles(
@@ -677,8 +679,46 @@ def test_validate_concurrency_raises_on_missing_starting_snapshot(catalog: Catal
         io=table.io,
     )
 
-    # parent is valid but starting is corrupted
-    producer._starting_snapshot_id = 99999999
+    # Set a CommitWindow with a non-existent starting snapshot ID
+    producer._commit_window = CommitWindow(
+        starting_snapshot_id=99999999, catalog_head_snapshot_id=table.metadata.current_snapshot_id
+    )
 
     with pytest.raises(ValidationException, match="Cannot find starting snapshot"):
         producer._validate_concurrency()
+
+
+def test_mixed_delete_overwrite_retries_successfully(catalog: Catalog) -> None:
+    """A mixed full-file + partial delete should succeed via retry, not raise ValidationException."""
+    from pyiceberg.partitioning import PartitionField, PartitionSpec
+    from pyiceberg.transforms import IdentityTransform
+
+    catalog.create_namespace("default")
+    schema = Schema(
+        NestedField(1, "category", StringType(), required=False),
+        NestedField(2, "value", LongType(), required=False),
+    )
+    spec = PartitionSpec(PartitionField(source_id=1, field_id=1000, transform=IdentityTransform(), name="category"))
+    catalog.create_table("default.mixed_retry_test", schema=schema, partition_spec=spec)
+
+    import pyarrow as pa
+
+    tbl = catalog.load_table("default.mixed_retry_test")
+
+    # 3 partitions, one data file each: a->[1,2], b->[3,4], c->[5,6]
+    tbl.append(pa.table({"category": ["a", "a", "b", "b", "c", "c"], "value": [1, 2, 3, 4, 5, 6]}))
+
+    tbl1 = catalog.load_table("default.mixed_retry_test")
+    tbl2 = catalog.load_table("default.mixed_retry_test")
+
+    # Concurrent append to partition 'c' (commits first, advances the HEAD)
+    tbl1.append(pa.table({"category": ["c"], "value": [7]}))
+
+    # Mixed delete on tbl2 (stale snapshot):
+    # partition 'a' is a partial rewrite (value==1 deleted, value==2 remains) -> _OverwriteFiles
+    # partition 'b' is a full delete (category == 'b') -> _DeleteFiles
+    # This should NOT conflict with the append to 'c', so retry should succeed.
+    tbl2.delete("value == 1 or category == 'b'")
+
+    result = catalog.load_table("default.mixed_retry_test").scan().to_arrow()
+    assert sorted(result.column("value").to_pylist()) == [2, 5, 6, 7]