Add validation guards and isolation level property routing

Skip _validate_no_new_delete_files and _validate_deleted_data_files
when conflict_detection_filter is None, matching Java's
BaseOverwriteFiles.validate() behavior for rowFilter == AlwaysFalse().

Route isolation level property based on the calling operation.
Transaction.delete() uses write.delete.isolation-level (default).
Transaction.overwrite(), dynamic_partition_overwrite(), and upsert()
use write.update.isolation-level via _isolation_level_property on
the snapshot producer.

Remove unused WRITE_MERGE_ISOLATION_LEVEL constant.`

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

Author: lawofcycles

pyiceberg/table/__init__.py

@@ -220,7 +220,6 @@ class TableProperties:
 
     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"
 
 
@@ -569,7 +568,12 @@ def dynamic_partition_overwrite(
         delete_filter = self._build_partition_predicate(
             partition_records=partitions_to_overwrite, spec=self.table_metadata.spec(), schema=self.table_metadata.schema()
         )
-        self.delete(delete_filter=delete_filter, snapshot_properties=snapshot_properties, branch=branch)
+        self.delete(
+            delete_filter=delete_filter,
+            snapshot_properties=snapshot_properties,
+            branch=branch,
+            _isolation_level_property=TableProperties.WRITE_UPDATE_ISOLATION_LEVEL,
+        )
 
         with self._append_snapshot_producer(snapshot_properties, branch=branch) as append_files:
             append_files.commit_uuid = append_snapshot_commit_uuid
@@ -626,6 +630,7 @@ def overwrite(
                 case_sensitive=case_sensitive,
                 snapshot_properties=snapshot_properties,
                 branch=branch,
+                _isolation_level_property=TableProperties.WRITE_UPDATE_ISOLATION_LEVEL,
             )
 
         with self._append_snapshot_producer(snapshot_properties, branch=branch) as append_files:
@@ -643,6 +648,7 @@ def delete(
         snapshot_properties: dict[str, str] = EMPTY_DICT,
         case_sensitive: bool = True,
         branch: str | None = MAIN_BRANCH,
+        _isolation_level_property: str | None = None,
     ) -> None:
         """
         Shorthand for deleting record from a table.
@@ -670,6 +676,8 @@ def delete(
             delete_filter = _parse_row_filter(delete_filter)
 
         with self.update_snapshot(snapshot_properties=snapshot_properties, branch=branch).delete() as delete_snapshot:
+            if _isolation_level_property is not None:
+                delete_snapshot._isolation_level_property = _isolation_level_property
             delete_snapshot.delete_by_predicate(delete_filter, case_sensitive)
 
         # Check if there are any files that require an actual rewrite of a data file
@@ -725,6 +733,8 @@ def delete(
                 with self.update_snapshot(
                     snapshot_properties=snapshot_properties, branch=branch
                 ).overwrite() as overwrite_snapshot:
+                    if _isolation_level_property is not None:
+                        overwrite_snapshot._isolation_level_property = _isolation_level_property
                     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

@@ -141,6 +141,7 @@ def __init__(
         )
         self._predicate = AlwaysFalse()
         self._case_sensitive = True
+        self._isolation_level_property: str = TableProperties.WRITE_DELETE_ISOLATION_LEVEL
 
     def _validate_target_branch(self, branch: str | None) -> str | None:
         # if branch is none, write will be written into a staging snapshot
@@ -556,16 +557,17 @@ def _validate_concurrency(self) -> None:
             return
 
         isolation_level_str = table.metadata.properties.get(
-            TableProperties.WRITE_DELETE_ISOLATION_LEVEL, TableProperties.WRITE_ISOLATION_LEVEL_DEFAULT
+            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_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 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)
 
         if self._deleted_data_files:
             _validate_no_new_deletes_for_data_files(
@@ -763,16 +765,17 @@ def _validate_concurrency(self) -> None:
             return
 
         isolation_level_str = table.metadata.properties.get(
-            TableProperties.WRITE_DELETE_ISOLATION_LEVEL, TableProperties.WRITE_ISOLATION_LEVEL_DEFAULT
+            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_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 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)
 
         if self._deleted_data_files:
             _validate_no_new_deletes_for_data_files(

tests/table/test_commit_retry.py

@@ -47,7 +47,6 @@ def test_commit_retry_table_properties() -> None:
 def test_isolation_level_table_properties() -> None:
     assert TableProperties.WRITE_DELETE_ISOLATION_LEVEL == "write.delete.isolation-level"
     assert TableProperties.WRITE_UPDATE_ISOLATION_LEVEL == "write.update.isolation-level"
-    assert TableProperties.WRITE_MERGE_ISOLATION_LEVEL == "write.merge.isolation-level"
     assert TableProperties.WRITE_ISOLATION_LEVEL_DEFAULT == "serializable"
 
 
@@ -484,3 +483,66 @@ def test_concurrent_partial_deletes_on_different_partitions_succeed(catalog: Cat
     result = refreshed.scan().to_arrow()
     # Original 4 rows, minus value==1 and value==3 = 2 rows remaining
     assert len(result) == 2
+
+
+def test_overwrite_uses_update_isolation_level(catalog: Catalog) -> None:
+    """Verify that overwrite() reads write.update.isolation-level, not write.delete.isolation-level."""
+    catalog.create_namespace("default")
+    schema = _test_schema()
+    catalog.create_table(
+        "default.update_iso_test",
+        schema=schema,
+        properties={
+            "write.delete.isolation-level": "serializable",
+            "write.update.isolation-level": "snapshot",
+        },
+    )
+
+    import pyarrow as pa
+
+    df = pa.table({"x": [1, 2, 3]})
+
+    tbl = catalog.load_table("default.update_iso_test")
+    tbl.append(df)
+
+    tbl1 = catalog.load_table("default.update_iso_test")
+    tbl2 = catalog.load_table("default.update_iso_test")
+
+    tbl1.append(df)
+
+    # Under write.delete.isolation-level=serializable this would raise ValidationException.
+    # But overwrite() uses write.update.isolation-level=snapshot, so it succeeds.
+    tbl2.overwrite(pa.table({"x": [10, 20, 30]}), overwrite_filter="x > 0")
+
+    refreshed = catalog.load_table("default.update_iso_test")
+    result = refreshed.scan().to_arrow()
+    # overwrite with x > 0 deletes all rows (including tbl1's append), then adds 3 new rows
+    assert len(result) == 3
+
+
+def test_overwrite_with_serializable_update_isolation_raises(catalog: Catalog) -> None:
+    """Verify that overwrite() raises ValidationException when write.update.isolation-level=serializable."""
+    catalog.create_namespace("default")
+    schema = _test_schema()
+    catalog.create_table(
+        "default.update_serial_test",
+        schema=schema,
+        properties={
+            "write.update.isolation-level": "serializable",
+        },
+    )
+
+    import pyarrow as pa
+
+    df = pa.table({"x": [1, 2, 3]})
+
+    tbl = catalog.load_table("default.update_serial_test")
+    tbl.append(df)
+
+    tbl1 = catalog.load_table("default.update_serial_test")
+    tbl2 = catalog.load_table("default.update_serial_test")
+
+    tbl1.append(df)
+
+    with pytest.raises(ValidationException):
+        tbl2.overwrite(pa.table({"x": [10, 20, 30]}), overwrite_filter="x > 0")