Validate each branch against its own commits on retry

pyiceberg/table/__init__.py

@@ -1194,17 +1194,10 @@ def commit_transaction(self) -> Table:
         elif self._snapshot_producers:
             # An empty staged output (e.g. a delete whose plan matched nothing) skips the
             # commit loop above, so run concurrency validation explicitly before reporting success.
-            from pyiceberg.table.update.snapshot import CommitWindow
-
             try:
                 self._table.refresh()
-                commit_window = CommitWindow.resolve(
-                    self._table.metadata,
-                    self._snapshot_producers[0]._starting_snapshot_id,
-                    self._snapshot_producers[0]._target_branch,
-                )
+                self._resolve_commit_windows()
                 for producer in self._snapshot_producers:
-                    producer._commit_window = commit_window
                     producer._validate_concurrency()
             except Exception:
                 for producer in self._snapshot_producers:
@@ -1224,20 +1217,30 @@ def _cleanup_uncommitted_manifests(self) -> None:
         for producer in self._snapshot_producers:
             producer._cleanup_uncommitted()
 
+    def _resolve_commit_windows(self) -> None:
+        """Resolve each producer's commit window against the refreshed metadata of its own branch.
+
+        The window starts where the transaction first wrote to the branch. Later producers on the same
+        branch start from a snapshot staged by this transaction, which the catalog has never seen.
+        """
+        from pyiceberg.table.update.snapshot import CommitWindow
+
+        windows: dict[str | None, CommitWindow] = {}
+        for producer in self._snapshot_producers:
+            branch = producer._target_branch
+            if branch not in windows:
+                windows[branch] = CommitWindow.resolve(self._table.metadata, producer._starting_snapshot_id, branch)
+            producer._commit_window = windows[branch]
+
     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))
 
-        starting_id = self._snapshot_producers[0]._starting_snapshot_id if self._snapshot_producers else None
-        target_branch = self._snapshot_producers[0]._target_branch if self._snapshot_producers else None
-        commit_window = CommitWindow.resolve(self._table.metadata, starting_id, target_branch)
-
+        self._resolve_commit_windows()
         for producer in self._snapshot_producers:
-            producer._commit_window = commit_window
             producer._refresh_for_retry()
             producer._validate_concurrency()
             updates, requirements = producer._commit()

tests/table/test_commit_retry.py

@@ -1363,3 +1363,72 @@ def test_transaction_stages_one_ref_assertion_per_ref(catalog: Catalog) -> None:
     ref_assertions = {r.ref: r.snapshot_id for r in tx._requirements if isinstance(r, AssertRefSnapshotId)}
     assert ref_assertions == {"b": base_snapshot_id, "main": base_snapshot_id}
     assert len([r for r in tx._requirements if isinstance(r, AssertRefSnapshotId)]) == 2
+
+
+def _table_with_branch(catalog: Catalog, identifier: str) -> Any:
+    import pyarrow as pa
+
+    tbl = catalog.create_table(
+        identifier,
+        schema=_test_schema(),
+        properties={TableProperties.COMMIT_MIN_RETRY_WAIT_MS: "1", TableProperties.COMMIT_MAX_RETRY_WAIT_MS: "2"},
+    )
+    tbl.append(pa.table({"x": [0, 1]}))
+    snapshot_id = tbl.metadata.current_snapshot_id
+    assert snapshot_id is not None
+    tbl.manage_snapshots().create_branch(snapshot_id=snapshot_id, branch_name="b").commit()
+    return tbl
+
+
+def test_retry_validates_each_branch_against_its_own_commits(catalog: Catalog) -> None:
+    """A retried delete on a second branch is validated against that branch, not the first producer's branch."""
+    import pyarrow as pa
+
+    catalog.create_namespace("default")
+    tx = _table_with_branch(catalog, "default.retry_window_per_branch").transaction()
+    tx.append(pa.table({"x": [9]}))
+    tx.delete("x == 1", branch="b")
+
+    # A row matching the delete lands on b after the delete was planned; main is untouched.
+    catalog.load_table("default.retry_window_per_branch").append(pa.table({"x": [1]}), branch="b")
+
+    with pytest.raises(ValidationException):
+        tx.commit_transaction()
+
+    refreshed = catalog.load_table("default.retry_window_per_branch")
+    assert sorted(refreshed.scan().use_ref("b").to_arrow()["x"].to_pylist()) == [0, 1, 1]
+    assert sorted(refreshed.scan().to_arrow()["x"].to_pylist()) == [0, 1]
+
+
+def test_empty_plan_validation_checks_each_branch_against_its_own_commits(catalog: Catalog) -> None:
+    """Empty-plan deletes on two branches are each validated against their own branch."""
+    import pyarrow as pa
+
+    catalog.create_namespace("default")
+    tx = _table_with_branch(catalog, "default.empty_plan_window_per_branch").transaction()
+    with pytest.warns(UserWarning):  # neither delete matches anything at staging time
+        tx.delete("x > 45")
+        tx.delete("x > 45", branch="b")
+
+    catalog.load_table("default.empty_plan_window_per_branch").append(pa.table({"x": [50]}), branch="b")
+
+    with pytest.raises(ValidationException):
+        tx.commit_transaction()
+
+
+def test_retry_with_two_writes_to_one_branch_succeeds(catalog: Catalog) -> None:
+    """The second write to a branch starts from a staged snapshot; its retry window must use the committed base."""
+    import pyarrow as pa
+
+    catalog.create_namespace("default")
+    tx = _table_with_branch(catalog, "default.retry_two_writes_one_branch").transaction()
+    tx.append(pa.table({"x": [9]}))
+    tx.delete("x == 0")
+
+    # Non-conflicting concurrent append to the same branch forces a retry.
+    catalog.load_table("default.retry_two_writes_one_branch").append(pa.table({"x": [5]}))
+
+    tx.commit_transaction()
+
+    rows = sorted(catalog.load_table("default.retry_two_writes_one_branch").scan().to_arrow()["x"].to_pylist())
+    assert rows == [1, 5, 9]