fix: raise on unresolvable snapshots and fix _starting_snapshot_id propagation

Address review feedback on commit retry validation:

- _validate_concurrency() now raises ValidationException when
parent_snapshot_id or starting_snapshot_id is non-null but cannot
be resolved, instead of silently skipping validation.
- Fix _starting_snapshot_id not being propagated from _DeleteFiles to
_OverwriteFiles in Transaction.delete(), which caused the conflict
detection window to collapse when both producers are involved.
- Move ValidationException import to module level per project convention.
- Add commit retry and isolation level properties to configuration docs.

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

Author: lawofcycles

mkdocs/docs/configuration.md

@@ -85,6 +85,26 @@ Iceberg tables support table properties to configure table behavior.
 
 <!-- prettier-ignore-end -->
 
+### Commit retry options
+
+When a concurrent commit is detected, PyIceberg automatically retries the operation with exponential backoff. If the retry detects a real data conflict (e.g. concurrent deletes on the same partition), it raises `ValidationException` instead of retrying.
+
+| Key                              | Options          | Default   | Description                                                        |
+| -------------------------------- | ---------------- | --------- | ------------------------------------------------------------------ |
+| `commit.retry.num-retries`       | Integer          | 4         | Maximum number of retry attempts after a commit conflict            |
+| `commit.retry.min-wait-ms`      | Integer (ms)     | 100       | Minimum wait time before the first retry                            |
+| `commit.retry.max-wait-ms`      | Integer (ms)     | 60000     | Maximum wait time between retries (caps exponential backoff)        |
+| `commit.retry.total-timeout-ms` | Integer (ms)     | 1800000   | Total time allowed for all retry attempts before giving up          |
+
+### Isolation level options
+
+These properties control conflict detection behavior during concurrent writes.
+
+| Key                              | Options                          | Default      | Description                                                                                     |
+| -------------------------------- | -------------------------------- | ------------ | ----------------------------------------------------------------------------------------------- |
+| `write.delete.isolation-level`   | `{serializable,snapshot}`        | serializable | Isolation level for delete operations. Under `serializable`, concurrent appends to affected partitions cause `ValidationException`. Under `snapshot`, only conflicting deletes are rejected. |
+| `write.update.isolation-level`   | `{serializable,snapshot}`        | serializable | Isolation level for overwrite operations. Same semantics as `write.delete.isolation-level`.      |
+
 ## FileIO
 
 Iceberg works with the concept of a FileIO which is a pluggable module for reading, writing, and deleting files. By default, PyIceberg will try to initialize the FileIO that's suitable for the scheme (`s3://`, `gs://`, etc.) and will use the first one that's installed.

pyiceberg/table/__init__.py

@@ -816,6 +816,7 @@ def delete(
                 ).overwrite() as overwrite_snapshot:
                     if _isolation_level_property is not None:
                         overwrite_snapshot._isolation_level_property = _isolation_level_property
+                    overwrite_snapshot._starting_snapshot_id = delete_snapshot._starting_snapshot_id
                     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:

pyiceberg/table/update/snapshot.py

@@ -27,6 +27,7 @@
 from typing import TYPE_CHECKING, Generic
 
 from pyiceberg.avro.codecs import AvroCompressionCodec
+from pyiceberg.exceptions import ValidationException
 from pyiceberg.expressions import AlwaysFalse, BooleanExpression, Or
 from pyiceberg.expressions.visitors import (
     ROWS_MIGHT_NOT_MATCH,
@@ -579,12 +580,12 @@ def _validate_concurrency(self) -> None:
         table = self._transaction._table
         parent_snapshot = table.metadata.snapshot_by_id(self._parent_snapshot_id)
         if parent_snapshot is None:
-            return
+            raise ValidationException(f"Cannot find parent snapshot {self._parent_snapshot_id} in table metadata")
 
         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
+            raise ValidationException(f"Cannot find starting snapshot {starting_snapshot_id} in table metadata")
 
         isolation_level_str = table.metadata.properties.get(
             self._isolation_level_property, TableProperties.WRITE_ISOLATION_LEVEL_DEFAULT
@@ -797,12 +798,12 @@ def _validate_concurrency(self) -> None:
         table = self._transaction._table
         parent_snapshot = table.metadata.snapshot_by_id(self._parent_snapshot_id)
         if parent_snapshot is None:
-            return
+            raise ValidationException(f"Cannot find parent snapshot {self._parent_snapshot_id} in table metadata")
 
         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
+            raise ValidationException(f"Cannot find starting snapshot {starting_snapshot_id} in table metadata")
 
         isolation_level_str = table.metadata.properties.get(
             self._isolation_level_property, TableProperties.WRITE_ISOLATION_LEVEL_DEFAULT

tests/table/test_commit_retry.py

@@ -595,3 +595,90 @@ def capturing_clean_all(self_producer: Any) -> None:
     for producer in captured_producers:
         assert producer._written_manifests == []
         assert producer._uncommitted_manifests == []
+
+
+def test_mixed_delete_overwrite_starts_from_catalog_snapshot(catalog: Catalog) -> None:
+    """Mixed full-file and partial deletes should validate from the original table snapshot."""
+    from pyiceberg.partitioning import PartitionField, PartitionSpec
+    from pyiceberg.table.update.snapshot import _DeleteFiles, _OverwriteFiles
+    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"))
+    table = catalog.create_table("default.mixed_delete_start_snapshot", schema=schema, partition_spec=spec)
+
+    import pyarrow as pa
+
+    # Partition "a" has one row (will be fully deleted) and partition "b" has two rows (partial delete)
+    table.append(pa.table({"category": ["a", "b", "b"], "value": [1, 2, 3]}))
+
+    base_snapshot_id = table.metadata.current_snapshot_id
+
+    tx = Transaction(table, autocommit=False)
+    # "value == 1" deletes the entire file in partition "a" (full-file delete)
+    # "value == 2" partially deletes from partition "b" (CoW rewrite)
+    tx.delete("value <= 2")
+
+    assert len(tx._snapshot_producers) == 2
+
+    delete_producer = next(p for p in tx._snapshot_producers if isinstance(p, _DeleteFiles))
+    overwrite_producer = next(p for p in tx._snapshot_producers if isinstance(p, _OverwriteFiles))
+
+    assert delete_producer._starting_snapshot_id == base_snapshot_id
+    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."""
+    catalog.create_namespace("default")
+    schema = _test_schema()
+    table = catalog.create_table("default.missing_parent_test", schema=schema)
+
+    import pyarrow as pa
+
+    table.append(pa.table({"x": [1, 2, 3]}))
+
+    from pyiceberg.table.update.snapshot import _DeleteFiles
+
+    tx = Transaction(table, autocommit=False)
+    producer = _DeleteFiles(
+        operation=Operation.DELETE,
+        transaction=tx,
+        io=table.io,
+    )
+
+    # Artificially set a non-existent parent snapshot ID
+    producer._parent_snapshot_id = 99999999
+
+    with pytest.raises(ValidationException, match="Cannot find parent snapshot"):
+        producer._validate_concurrency()
+
+
+def test_validate_concurrency_raises_on_missing_starting_snapshot(catalog: Catalog) -> None:
+    """Validation should raise when starting_snapshot_id is non-null but cannot be resolved."""
+    catalog.create_namespace("default")
+    schema = _test_schema()
+    table = catalog.create_table("default.missing_starting_test", schema=schema)
+
+    import pyarrow as pa
+
+    table.append(pa.table({"x": [1, 2, 3]}))
+
+    from pyiceberg.table.update.snapshot import _DeleteFiles
+
+    tx = Transaction(table, autocommit=False)
+    producer = _DeleteFiles(
+        operation=Operation.DELETE,
+        transaction=tx,
+        io=table.io,
+    )
+
+    # parent is valid but starting is corrupted
+    producer._starting_snapshot_id = 99999999
+
+    with pytest.raises(ValidationException, match="Cannot find starting snapshot"):
+        producer._validate_concurrency()