fix: use starting_snapshot_id for validation window in commit retry

The concurrency validation was using parent_snapshot (current head) for
both the starting point and ending point of the validation window. When
multiple concurrent commits occur during retry sleep, the validation
would only inspect the latest head and miss conflicting commits below it.

Introduce _starting_snapshot_id that is fixed at operation init time and
does not change on retry. Also fix _validation_history to use exclusive
semantics for from_snapshot, matching Java Iceberg's ancestorsBetween.

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

Author: lawofcycles

pyiceberg/table/update/snapshot.py

@@ -100,6 +100,7 @@ class _SnapshotProducer(UpdateTableMetadata[U], Generic[U]):
     _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]
@@ -139,6 +140,7 @@ def __init__(
         self._parent_snapshot_id = (
             snapshot.snapshot_id if (snapshot := self._transaction.table_metadata.snapshot_by_name(self._target_branch)) else None
         )
+        self._starting_snapshot_id = self._parent_snapshot_id
         self._predicate = AlwaysFalse()
         self._case_sensitive = True
         self._isolation_level_property: str = TableProperties.WRITE_DELETE_ISOLATION_LEVEL
@@ -579,22 +581,27 @@ def _validate_concurrency(self) -> None:
         if parent_snapshot is None:
             return
 
+        starting_snapshot_id = self._starting_snapshot_id if self._starting_snapshot_id is not None else self._parent_snapshot_id
+        starting_snapshot = table.metadata.snapshot_by_id(starting_snapshot_id)
+        if starting_snapshot is None:
+            return
+
         isolation_level_str = table.metadata.properties.get(
             self._isolation_level_property, 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_added_data_files(table, parent_snapshot, conflict_detection_filter, starting_snapshot)
 
         if conflict_detection_filter is not None:
-            _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)
+            _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)
 
         if self._deleted_data_files:
             _validate_no_new_deletes_for_data_files(
-                table, parent_snapshot, conflict_detection_filter, self._deleted_data_files, parent_snapshot
+                table, parent_snapshot, conflict_detection_filter, self._deleted_data_files, starting_snapshot
             )
 
 
@@ -792,22 +799,27 @@ def _validate_concurrency(self) -> None:
         if parent_snapshot is None:
             return
 
+        starting_snapshot_id = self._starting_snapshot_id if self._starting_snapshot_id is not None else self._parent_snapshot_id
+        starting_snapshot = table.metadata.snapshot_by_id(starting_snapshot_id)
+        if starting_snapshot is None:
+            return
+
         isolation_level_str = table.metadata.properties.get(
             self._isolation_level_property, 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_added_data_files(table, parent_snapshot, conflict_detection_filter, starting_snapshot)
 
         if conflict_detection_filter is not None:
-            _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)
+            _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)
 
         if self._deleted_data_files:
             _validate_no_new_deletes_for_data_files(
-                table, parent_snapshot, conflict_detection_filter, self._deleted_data_files, parent_snapshot
+                table, parent_snapshot, conflict_detection_filter, self._deleted_data_files, starting_snapshot
             )
 
 

pyiceberg/table/update/validate.py

@@ -47,10 +47,13 @@ def _validation_history(
 ) -> tuple[list[ManifestFile], set[int]]:
     """Return newly added manifests and snapshot IDs between the starting snapshot and parent snapshot.
 
+    Walks from to_snapshot backwards towards from_snapshot, collecting manifests from
+    snapshots whose operations match. from_snapshot is excluded from results.
+
     Args:
         table: Table to get the history from
-        from_snapshot: Parent snapshot to get the history from
-        to_snapshot: Starting snapshot
+        from_snapshot: Snapshot where the walk stops (exclusive)
+        to_snapshot: Snapshot where the walk starts
         matching_operations: Operations to match on
         manifest_content_filter: Manifest content type to filter
 
@@ -60,11 +63,17 @@ def _validation_history(
     Returns:
         List of manifest files and set of snapshots ID's matching conditions
     """
+    if from_snapshot.snapshot_id == to_snapshot.snapshot_id:
+        return [], set()
+
     manifests_files: list[ManifestFile] = []
     snapshots: set[int] = set()
 
     last_snapshot = None
     for snapshot in ancestors_between(from_snapshot, to_snapshot, table.metadata):
+        if snapshot.snapshot_id == from_snapshot.snapshot_id:
+            last_snapshot = snapshot
+            break
         last_snapshot = snapshot
         summary = snapshot.summary
         if summary is None:
@@ -73,7 +82,6 @@ def _validation_history(
             continue
 
         snapshots.add(snapshot.snapshot_id)
-        # TODO: Maybe do the IO in a separate thread at some point, and collect at the bottom (we can easily merge the sets
         manifests_files.extend(
             [
                 manifest
@@ -82,7 +90,7 @@ def _validation_history(
             ]
         )
 
-    if last_snapshot is not None and last_snapshot.snapshot_id != from_snapshot.snapshot_id:
+    if last_snapshot is None or last_snapshot.snapshot_id != from_snapshot.snapshot_id:
         raise ValidationException("No matching snapshot found.")
 
     return manifests_files, snapshots

tests/table/test_validate.py

@@ -65,11 +65,18 @@ def test_validation_history(table_v2_with_extensive_snapshots_and_manifests: tup
     """Test the validation history function."""
     table, mock_manifests = table_v2_with_extensive_snapshots_and_manifests
 
-    expected_manifest_data_counts = len([m for m in mock_manifests.values() if m[0].content == ManifestContent.DATA])
-
     oldest_snapshot = table.snapshots()[0]
     newest_snapshot = cast(Snapshot, table.current_snapshot())
 
+    # from_snapshot (oldest) is excluded from the results since it's the base
+    expected_manifest_data_counts = len(
+        [
+            m
+            for snap_id, m in mock_manifests.items()
+            if m[0].content == ManifestContent.DATA and snap_id != oldest_snapshot.snapshot_id
+        ]
+    )
+
     def mock_read_manifest_side_effect(self: Snapshot, io: FileIO) -> list[ManifestFile]:
         """Mock the manifests method to use the snapshot_id for lookup."""
         snapshot_id = self.snapshot_id
@@ -246,7 +253,8 @@ def test_validate_added_data_files_conflicting_count(
         update={"snapshots": snapshots},
     )
 
-    oldest_snapshot = table.snapshots()[-snapshot_history]
+    # Use one snapshot before the altered range as the boundary (exclusive stop point)
+    boundary_snapshot = table.snapshots()[-(snapshot_history + 1)]
     newest_snapshot = cast(Snapshot, table.current_snapshot())
 
     def mock_read_manifest_side_effect(self: Snapshot, io: FileIO) -> list[ManifestFile]:
@@ -273,7 +281,7 @@ def mock_fetch_manifest_entry(self: ManifestFile, io: FileIO, discard_deleted: b
                 table=table,
                 starting_snapshot=newest_snapshot,
                 data_filter=None,
-                parent_snapshot=oldest_snapshot,
+                parent_snapshot=boundary_snapshot,
                 partition_set=None,
             )
         )