Add commit retry with data conflict validation

Add automatic retry with exponential backoff when catalog commits fail
due to concurrent transactions (CommitFailedException), and integrate
the existing validation functions from validate.py into the write path
to detect incompatible concurrent modifications (ValidationException).

The retry loop is placed in Transaction.commit_transaction(). On each
retry attempt, table metadata is refreshed, registered snapshot
producers are re-executed to regenerate manifests, and data conflict
validation is run. Uncommitted manifests from failed attempts are
cleaned up after a successful commit.

Validation is performed for _OverwriteFiles and _DeleteFiles based on
the table's isolation level (serializable/snapshot). _FastAppendFiles
and _MergeAppendFiles do not require validation since appends never
conflict.

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

Author: lawofcycles

pyiceberg/table/__init__.py

@@ -31,6 +31,7 @@
 from pydantic import Field
 
 import pyiceberg.expressions.parser as parser
+from pyiceberg.exceptions import CommitFailedException
 from pyiceberg.expressions import AlwaysFalse, AlwaysTrue, And, BooleanExpression, EqualTo, IsNull, Or, Reference
 from pyiceberg.expressions.visitors import (
     ResidualEvaluator,
@@ -205,6 +206,23 @@ class TableProperties:
     MIN_SNAPSHOTS_TO_KEEP = "history.expire.min-snapshots-to-keep"
     MIN_SNAPSHOTS_TO_KEEP_DEFAULT = 1
 
+    COMMIT_NUM_RETRIES = "commit.retry.num-retries"
+    COMMIT_NUM_RETRIES_DEFAULT = 4
+
+    COMMIT_MIN_RETRY_WAIT_MS = "commit.retry.min-wait-ms"
+    COMMIT_MIN_RETRY_WAIT_MS_DEFAULT = 100
+
+    COMMIT_MAX_RETRY_WAIT_MS = "commit.retry.max-wait-ms"
+    COMMIT_MAX_RETRY_WAIT_MS_DEFAULT = 60000
+
+    COMMIT_TOTAL_RETRY_TIME_MS = "commit.retry.total-timeout-ms"
+    COMMIT_TOTAL_RETRY_TIME_MS_DEFAULT = 1800000  # 30 minutes
+
+    WRITE_DELETE_ISOLATION_LEVEL = "write.delete.isolation-level"
+    WRITE_UPDATE_ISOLATION_LEVEL = "write.update.isolation-level"
+    WRITE_MERGE_ISOLATION_LEVEL = "write.merge.isolation-level"
+    WRITE_ISOLATION_LEVEL_DEFAULT = "serializable"
+
 
 class Transaction:
     _table: Table
@@ -223,6 +241,7 @@ def __init__(self, table: Table, autocommit: bool = False):
         self._autocommit = autocommit
         self._updates = ()
         self._requirements = ()
+        self._snapshot_producers: list[Any] = []
 
     @property
     def table_metadata(self) -> TableMetadata:
@@ -265,6 +284,10 @@ def _stage(
 
         return self
 
+    def _register_snapshot_producer(self, producer: Any) -> None:
+        """Register a snapshot producer for retry support."""
+        self._snapshot_producers.append(producer)
+
     def _apply(
         self,
         updates: tuple[TableUpdate, ...],
@@ -703,6 +726,7 @@ def delete(
                     snapshot_properties=snapshot_properties, branch=branch
                 ).overwrite() as overwrite_snapshot:
                     overwrite_snapshot.commit_uuid = commit_uuid
+                    overwrite_snapshot.delete_by_predicate(delete_filter, case_sensitive)
                     for original_data_file, replaced_data_files in replaced_files:
                         overwrite_snapshot.delete_data_file(original_data_file)
                         for replaced_data_file in replaced_data_files:
@@ -939,17 +963,67 @@ def commit_transaction(self) -> Table:
             The table with the updates applied.
         """
         if len(self._updates) > 0:
-            self._requirements += (AssertTableUUID(uuid=self.table_metadata.table_uuid),)
-            self._table._do_commit(  # pylint: disable=W0212
-                updates=self._updates,
-                requirements=self._requirements,
+            from pyiceberg.utils.properties import property_as_int
+
+            properties = self._table.metadata.properties
+            num_retries_val = property_as_int(
+                properties, TableProperties.COMMIT_NUM_RETRIES, TableProperties.COMMIT_NUM_RETRIES_DEFAULT
+            )
+            num_retries = num_retries_val if num_retries_val is not None else TableProperties.COMMIT_NUM_RETRIES_DEFAULT
+            min_wait_val = property_as_int(
+                properties, TableProperties.COMMIT_MIN_RETRY_WAIT_MS, TableProperties.COMMIT_MIN_RETRY_WAIT_MS_DEFAULT
             )
+            min_wait_ms = min_wait_val if min_wait_val is not None else TableProperties.COMMIT_MIN_RETRY_WAIT_MS_DEFAULT
+            max_wait_val = property_as_int(
+                properties, TableProperties.COMMIT_MAX_RETRY_WAIT_MS, TableProperties.COMMIT_MAX_RETRY_WAIT_MS_DEFAULT
+            )
+            max_wait_ms = max_wait_val if max_wait_val is not None else TableProperties.COMMIT_MAX_RETRY_WAIT_MS_DEFAULT
+
+            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:
+                    if attempt == num_retries or not self._snapshot_producers:
+                        raise
+                    import random
+                    import time
+
+                    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()
 
         self._updates = ()
         self._requirements = ()
 
         return self._table
 
+    def _cleanup_uncommitted_manifests(self) -> None:
+        """Clean up manifests from failed retry attempts after a successful commit."""
+        for producer in self._snapshot_producers:
+            producer._cleanup_uncommitted()
+
+    def _rebuild_snapshot_updates(self) -> None:
+        """Rebuild snapshot updates for retry by re-executing registered producers."""
+        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)))
+
+        for producer in self._snapshot_producers:
+            producer._refresh_for_retry()
+            producer._validate_concurrency()
+            updates, requirements = producer._commit()
+            self._stage(updates, requirements)
+
 
 class CreateTableTransaction(Transaction):
     """A transaction that involves the creation of a new table."""
@@ -1961,13 +2035,11 @@ def _build_residual_evaluator(self, spec_id: int) -> Callable[[DataFile], Residu
         # The lambda created here is run in multiple threads.
         # So we avoid creating _EvaluatorExpression methods bound to a single
         # shared instance across multiple threads.
-        return lambda datafile: (
-            residual_evaluator_of(
-                spec=spec,
-                expr=self.row_filter,
-                case_sensitive=self.case_sensitive,
-                schema=self.table_metadata.schema(),
-            )
+        return lambda datafile: residual_evaluator_of(
+            spec=spec,
+            expr=self.row_filter,
+            case_sensitive=self.case_sensitive,
+            schema=self.table_metadata.schema(),
         )
 
     @staticmethod

pyiceberg/table/snapshots.py

@@ -86,6 +86,13 @@ def __repr__(self) -> str:
         return f"Operation.{self.name}"
 
 
+class IsolationLevel(str, Enum):
+    """Transaction isolation level for concurrent write validation."""
+
+    SERIALIZABLE = "serializable"
+    SNAPSHOT = "snapshot"
+
+
 class UpdateMetrics:
     added_file_size: int
     removed_file_size: int

pyiceberg/table/update/snapshot.py

@@ -17,6 +17,7 @@
 from __future__ import annotations
 
 import itertools
+import logging
 import uuid
 from abc import abstractmethod
 from collections import defaultdict
@@ -80,6 +81,8 @@
 if TYPE_CHECKING:
     from pyiceberg.table import Transaction
 
+logger = logging.getLogger(__name__)
+
 
 def _new_manifest_file_name(num: int, commit_uuid: uuid.UUID) -> str:
     return f"{commit_uuid}-m{num}.avro"
@@ -104,6 +107,8 @@ class _SnapshotProducer(UpdateTableMetadata[U], Generic[U]):
     _target_branch: str | None
     _predicate: BooleanExpression
     _case_sensitive: bool
+    _written_manifests: list[str]
+    _uncommitted_manifests: list[str]
 
     def __init__(
         self,
@@ -123,6 +128,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
 
         self._compression = self._transaction.table_metadata.properties.get(  # type: ignore
@@ -351,11 +358,39 @@ 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 _refresh_for_retry(self) -> None:
+        """Reset state for a retry attempt with refreshed metadata."""
+        self._uncommitted_manifests.extend(self._written_manifests)
+        self._written_manifests.clear()
+        self._parent_snapshot_id = (
+            snapshot.snapshot_id if (snapshot := self._transaction.table_metadata.snapshot_by_name(self._target_branch)) else None
+        )
+        self._snapshot_id = self._transaction.table_metadata.new_snapshot_id()
+        self._manifest_num_counter = itertools.count(0)
+        self.commit_uuid = uuid.uuid4()
+
+    def _validate_concurrency(self) -> None:
+        """Validate that concurrent changes do not conflict with this operation. No-op by default."""
+
     def _build_partition_projection(self, spec_id: int) -> BooleanExpression:
         project = inclusive_projection(self.schema(), self.spec(spec_id), self._case_sensitive)
         return project(self._predicate)
@@ -495,6 +530,48 @@ def files_affected(self) -> bool:
         """Indicate if any manifest-entries can be dropped."""
         return len(self._deleted_entries()) > 0
 
+    def _refresh_for_retry(self) -> None:
+        """Reset state for a retry attempt, clearing the cached delete computation."""
+        super()._refresh_for_retry()
+        if "_compute_deletes" in self.__dict__:
+            del self.__dict__["_compute_deletes"]
+
+    def _validate_concurrency(self) -> None:
+        """Validate that concurrent changes do not conflict with this delete."""
+        from pyiceberg.table import TableProperties
+        from pyiceberg.table.snapshots import IsolationLevel
+        from pyiceberg.table.update.validate import (
+            _validate_added_data_files,
+            _validate_deleted_data_files,
+            _validate_no_new_delete_files,
+            _validate_no_new_deletes_for_data_files,
+        )
+
+        if self._parent_snapshot_id is None:
+            return
+
+        table = self._transaction._table
+        parent_snapshot = table.metadata.snapshot_by_id(self._parent_snapshot_id)
+        if parent_snapshot is None:
+            return
+
+        isolation_level_str = table.metadata.properties.get(
+            TableProperties.WRITE_DELETE_ISOLATION_LEVEL, TableProperties.WRITE_ISOLATION_LEVEL_DEFAULT
+        )
+        isolation_level = IsolationLevel(isolation_level_str)
+        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, parent_snapshot)
+
+        _validate_no_new_delete_files(table, parent_snapshot, conflict_detection_filter, None, parent_snapshot)
+        _validate_deleted_data_files(table, parent_snapshot, conflict_detection_filter, parent_snapshot)
+
+        if self._deleted_data_files:
+            _validate_no_new_deletes_for_data_files(
+                table, parent_snapshot, conflict_detection_filter, self._deleted_data_files, parent_snapshot
+            )
+
 
 class _FastAppendFiles(_SnapshotProducer["_FastAppendFiles"]):
     def _existing_manifests(self) -> list[ManifestFile]:
@@ -666,6 +743,42 @@ def _get_entries(manifest: ManifestFile) -> list[ManifestEntry]:
         else:
             return []
 
+    def _validate_concurrency(self) -> None:
+        """Validate that concurrent changes do not conflict with this overwrite."""
+        from pyiceberg.table import TableProperties
+        from pyiceberg.table.snapshots import IsolationLevel
+        from pyiceberg.table.update.validate import (
+            _validate_added_data_files,
+            _validate_deleted_data_files,
+            _validate_no_new_delete_files,
+            _validate_no_new_deletes_for_data_files,
+        )
+
+        if self._parent_snapshot_id is None:
+            return
+
+        table = self._transaction._table
+        parent_snapshot = table.metadata.snapshot_by_id(self._parent_snapshot_id)
+        if parent_snapshot is None:
+            return
+
+        isolation_level_str = table.metadata.properties.get(
+            TableProperties.WRITE_DELETE_ISOLATION_LEVEL, TableProperties.WRITE_ISOLATION_LEVEL_DEFAULT
+        )
+        isolation_level = IsolationLevel(isolation_level_str)
+        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, parent_snapshot)
+
+        _validate_no_new_delete_files(table, parent_snapshot, conflict_detection_filter, None, parent_snapshot)
+        _validate_deleted_data_files(table, parent_snapshot, conflict_detection_filter, parent_snapshot)
+
+        if self._deleted_data_files:
+            _validate_no_new_deletes_for_data_files(
+                table, parent_snapshot, conflict_detection_filter, self._deleted_data_files, parent_snapshot
+            )
+
 
 class UpdateSnapshot:
     _transaction: Transaction