Check every new delete file for rewritten files on a retry

pyiceberg/table/update/snapshot.py

@@ -514,14 +514,7 @@ def _validate_concurrency(self) -> None:
 
         if self._deleted_data_files:
             _validate_data_files_exist(table, catalog_head, self._deleted_data_files, starting_snapshot)
-            _validate_no_new_deletes_for_data_files(
-                table,
-                catalog_head,
-                conflict_detection_filter,
-                self._deleted_data_files,
-                starting_snapshot,
-                self._case_sensitive,
-            )
+            _validate_no_new_deletes_for_data_files(table, catalog_head, self._deleted_data_files, starting_snapshot)
 
     def _build_partition_projection(self, spec_id: int) -> BooleanExpression:
         project = inclusive_projection(self.schema(), self.spec(spec_id), self._case_sensitive)

pyiceberg/table/update/validate.py

@@ -375,26 +375,23 @@ def _validate_no_new_delete_files(
 def _validate_no_new_deletes_for_data_files(
     table: Table,
     starting_snapshot: Snapshot,
-    data_filter: BooleanExpression | None,
     data_files: set[DataFile],
     parent_snapshot: Snapshot | None,
-    case_sensitive: bool = True,
 ) -> None:
     """Validate no new delete files must be applied for data files that have been added to the table since a starting snapshot.
 
     Args:
         table: Table to validate
         starting_snapshot: Snapshot current at the start of the operation
-        data_filter: Expression used to find added data files
         data_files: data files to validate have no new deletes
         parent_snapshot: Ending snapshot on the branch being validated
-        case_sensitive: Whether the data filter binds column names case-sensitively
     """
     # If there is no current state, or no files has been added
     if table.format_version < 2:
         return
 
-    deletes = _added_delete_files(table, starting_snapshot, data_filter, None, parent_snapshot, case_sensitive)
+    # The data files are rewritten in full, so a delete file outside the operation's filter still applies to them
+    deletes = _added_delete_files(table, starting_snapshot, None, None, parent_snapshot)
     seq_num = _starting_sequence_number(table, starting_snapshot)
 
     # Fail to any delete file found that applies to files written in or before the starting snapshot

tests/table/test_commit_retry.py

@@ -636,6 +636,28 @@ def test_file_overwrite_validates_concurrent_position_delete_in_nan_partition(ca
     assert catalog.load_table(identifier).scan().to_arrow()["id"].to_pylist() == [2]
 
 
+def test_partial_delete_validates_concurrent_delete_outside_the_filter(catalog: Catalog) -> None:
+    """A retried partial delete must fail when a concurrent delete file applies to a rewritten file, whatever its bounds."""
+    import pyarrow as pa
+
+    catalog.create_namespace("default")
+    schema = Schema(NestedField(1, "k", LongType(), required=False), NestedField(2, "v", StringType(), required=False))
+    identifier = "default.concurrent_delete_outside_the_filter"
+    table = catalog.create_table(identifier, schema=schema, properties={"format-version": "2"})
+    table.append(pa.table({"k": [1, 2, 3], "v": ["a", "b", "c"]}))
+    data_file = next(iter(table.scan().plan_files())).file
+
+    deleting_table = catalog.load_table(identifier)
+
+    # The delete file stores the deleted row (2, 'b'), so its bounds can't match k = 1.
+    commit_partition_position_delete(table, Record(), data_file.file_path, pos=1, row={"k": 2, "v": "b"})
+
+    with pytest.raises(ValidationException, match="found new delete for replace data file"):
+        deleting_table.delete("k = 1")
+
+    assert catalog.load_table(identifier).scan().to_arrow()["k"].to_pylist() == [1, 3]
+
+
 def test_concurrent_overwrite_append_retries_successfully(catalog: Catalog) -> None:
     """Append after a concurrent overwrite should succeed via retry."""
     catalog.create_namespace("default")

tests/table/test_validate.py

@@ -613,7 +613,6 @@ def test_validate_no_new_delete_files_for_data_files_raises_on_conflict(
             _validate_no_new_deletes_for_data_files(
                 table=table,
                 starting_snapshot=newest_snapshot,
-                data_filter=None,
                 data_files={mocked_data_file},
                 parent_snapshot=oldest_snapshot,
             )