Simplify AssertTableUUID and clean up manifests on abort

AssertTableUUID is constant across retries, so add it once before the
loop instead of adding/removing on each iteration.

Add _clean_all_uncommitted() that deletes both _uncommitted_manifests
and _written_manifests on permanent failure, fixing orphaned manifests
from the last attempt.

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

Author: lawofcycles

pyiceberg/table/__init__.py

@@ -997,27 +997,32 @@ def commit_transaction(self) -> Table:
                 total_timeout_val if total_timeout_val is not None else TableProperties.COMMIT_TOTAL_RETRY_TIME_MS_DEFAULT
             )
             start_time = time.monotonic()
+            self._requirements += (AssertTableUUID(uuid=self.table_metadata.table_uuid),)
 
-            for attempt in range(num_retries + 1):
-                try:
-                    self._requirements += (AssertTableUUID(uuid=self.table_metadata.table_uuid),)
-                    self._table._do_commit(  # pylint: disable=W0212
-                        updates=self._updates,
-                        requirements=self._requirements,
-                    )
-                    self._cleanup_uncommitted_manifests()
-                    break
-                except CommitFailedException:
-                    elapsed_ms = (time.monotonic() - start_time) * 1000
-                    if attempt == num_retries or not self._snapshot_producers or elapsed_ms >= total_timeout_ms:
-                        raise
-
-                    wait = min(min_wait_ms * (2**attempt), max_wait_ms)
-                    jitter = random.uniform(0, 0.25 * wait)
-                    time.sleep((wait + jitter) / 1000.0)
-
-                    self._table.refresh()
-                    self._rebuild_snapshot_updates()
+            try:
+                for attempt in range(num_retries + 1):
+                    try:
+                        self._table._do_commit(  # pylint: disable=W0212
+                            updates=self._updates,
+                            requirements=self._requirements,
+                        )
+                        self._cleanup_uncommitted_manifests()
+                        break
+                    except CommitFailedException:
+                        elapsed_ms = (time.monotonic() - start_time) * 1000
+                        if attempt == num_retries or not self._snapshot_producers or elapsed_ms >= total_timeout_ms:
+                            raise
+
+                        wait = min(min_wait_ms * (2**attempt), max_wait_ms)
+                        jitter = random.uniform(0, 0.25 * wait)
+                        time.sleep((wait + jitter) / 1000.0)
+
+                        self._table.refresh()
+                        self._rebuild_snapshot_updates()
+            except Exception:
+                for producer in self._snapshot_producers:
+                    producer._clean_all_uncommitted()
+                raise
 
         self._updates = ()
         self._requirements = ()
@@ -1034,7 +1039,7 @@ def _rebuild_snapshot_updates(self) -> None:
         from pyiceberg.table.update import AddSnapshotUpdate, AssertRefSnapshotId, SetSnapshotRefUpdate
 
         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, AssertTableUUID)))
+        self._requirements = tuple(r for r in self._requirements if not isinstance(r, AssertRefSnapshotId))
 
         for producer in self._snapshot_producers:
             producer._refresh_for_retry()

pyiceberg/table/update/snapshot.py

@@ -378,6 +378,16 @@ def _cleanup_uncommitted(self) -> None:
                 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()
+
     def _refresh_for_retry(self) -> None:
         """Reset state for a retry attempt with refreshed metadata."""
         self._uncommitted_manifests.extend(self._written_manifests)

tests/table/test_commit_retry.py

@@ -556,3 +556,45 @@ def test_overwrite_with_serializable_update_isolation_raises(catalog: Catalog) -
 
     with pytest.raises(ValidationException):
         tbl2.overwrite(pa.table({"x": [10, 20, 30]}), overwrite_filter="x > 0")
+
+
+def test_clean_all_uncommitted_on_validation_exception(catalog: Catalog) -> None:
+    """Verify that all manifests are cleaned up when commit aborts with ValidationException."""
+    catalog.create_namespace("default")
+    schema = _test_schema()
+    catalog.create_table("default.clean_abort_test", schema=schema)
+
+    import pyarrow as pa
+
+    df = pa.table({"x": [1, 2, 3]})
+
+    tbl = catalog.load_table("default.clean_abort_test")
+    tbl.append(df)
+
+    tbl1 = catalog.load_table("default.clean_abort_test")
+    tbl2 = catalog.load_table("default.clean_abort_test")
+
+    tbl1.delete("x == 1")
+
+    captured_producers: list = []
+
+    original_clean_all = None
+
+    def capturing_clean_all(self_producer: Any) -> None:
+        captured_producers.append(self_producer)
+        original_clean_all(self_producer)
+
+    from pyiceberg.table.update.snapshot import _SnapshotProducer
+
+    original_clean_all = _SnapshotProducer._clean_all_uncommitted
+
+    with patch.object(_SnapshotProducer, "_clean_all_uncommitted", capturing_clean_all):
+        with pytest.raises(ValidationException):
+            tbl2.delete("x == 1")
+
+    # _clean_all_uncommitted was called on abort
+    assert len(captured_producers) > 0
+    # All manifest lists should be cleared
+    for producer in captured_producers:
+        assert producer._written_manifests == []
+        assert producer._uncommitted_manifests == []