Bind conflict-detection filters with the operation's case sensitivity

pyiceberg/table/update/snapshot.py

@@ -498,19 +498,26 @@ def _validate_concurrency(self) -> None:
         conflict_detection_filter = self._predicate if self._predicate != AlwaysFalse() else None
 
         if isolation_level == IsolationLevel.SERIALIZABLE:
-            _validate_added_data_files(table, catalog_head, conflict_detection_filter, starting_snapshot)
+            _validate_added_data_files(table, catalog_head, conflict_detection_filter, starting_snapshot, self._case_sensitive)
 
         if self._predicate != AlwaysFalse():
             # partition_set is None, so any concurrently committed delete file counts as
             # a conflict even in unrelated partitions. Scope this via partition projection
             # (as Java does) when delete-file write support lands.
-            _validate_no_new_delete_files(table, catalog_head, conflict_detection_filter, None, starting_snapshot)
-            _validate_deleted_data_files(table, catalog_head, conflict_detection_filter, starting_snapshot)
+            _validate_no_new_delete_files(
+                table, catalog_head, conflict_detection_filter, None, starting_snapshot, self._case_sensitive
+            )
+            _validate_deleted_data_files(table, catalog_head, conflict_detection_filter, starting_snapshot, self._case_sensitive)
 
         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
+                table,
+                catalog_head,
+                conflict_detection_filter,
+                self._deleted_data_files,
+                starting_snapshot,
+                self._case_sensitive,
             )
 
     def _build_partition_projection(self, spec_id: int) -> BooleanExpression:

pyiceberg/table/update/validate.py

@@ -105,6 +105,7 @@ def _filter_manifest_entries(
     partition_set: dict[int, set[Record]] | None,
     entry_status: ManifestEntryStatus | None,
     schema: Schema,
+    case_sensitive: bool = True,
 ) -> bool:
     """Filter manifest entries based on data filter and partition set.
 
@@ -115,6 +116,7 @@ def _filter_manifest_entries(
         partition_set: Optional set of partitions to match data files
         entry_status: Optional status to match data files
         schema: schema for filtering
+        case_sensitive: Whether the data filter binds column names case-sensitively
 
     Returns:
         True if the entry should be included, False otherwise
@@ -126,7 +128,7 @@ def _filter_manifest_entries(
         return False
 
     if data_filter is not None:
-        evaluator = _InclusiveMetricsEvaluator(schema, data_filter)
+        evaluator = _InclusiveMetricsEvaluator(schema, data_filter, case_sensitive=case_sensitive)
         if evaluator.eval(entry.data_file) is ROWS_CANNOT_MATCH:
             return False
 
@@ -145,6 +147,7 @@ def _deleted_data_files(
     data_filter: BooleanExpression | None,
     partition_set: dict[int, set[Record]] | None,
     parent_snapshot: Snapshot | None,
+    case_sensitive: bool = True,
 ) -> Iterator[ManifestEntry]:
     """Find deleted data files matching a filter since a starting snapshot.
 
@@ -154,6 +157,7 @@ def _deleted_data_files(
         data_filter: Expression used to find deleted data files
         partition_set: dict of {spec_id: set[partition]} to filter on
         parent_snapshot: Ending snapshot on the branch being validated
+        case_sensitive: Whether the data filter binds column names case-sensitively
 
     Returns:
         List of conflicting manifest-entries
@@ -170,7 +174,7 @@ def _deleted_data_files(
     for manifest in manifests:
         for entry in manifest.fetch_manifest_entry(table.io, discard_deleted=False):
             if _filter_manifest_entries(
-                entry, snapshot_ids, data_filter, partition_set, ManifestEntryStatus.DELETED, table.schema()
+                entry, snapshot_ids, data_filter, partition_set, ManifestEntryStatus.DELETED, table.schema(), case_sensitive
             ):
                 yield entry
 
@@ -180,6 +184,7 @@ def _validate_deleted_data_files(
     starting_snapshot: Snapshot,
     data_filter: BooleanExpression | None,
     parent_snapshot: Snapshot | None,
+    case_sensitive: bool = True,
 ) -> None:
     """Validate that no files matching a filter have been deleted from the table since a starting snapshot.
 
@@ -188,9 +193,10 @@ def _validate_deleted_data_files(
         starting_snapshot: Snapshot current at the start of the operation
         data_filter: Expression used to find deleted data files
         parent_snapshot: Ending snapshot on the branch being validated
+        case_sensitive: Whether the data filter binds column names case-sensitively
 
     """
-    conflicting_entries = _deleted_data_files(table, starting_snapshot, data_filter, None, parent_snapshot)
+    conflicting_entries = _deleted_data_files(table, starting_snapshot, data_filter, None, parent_snapshot, case_sensitive)
     if any(conflicting_entries):
         conflicting_snapshots = {entry.snapshot_id for entry in conflicting_entries}
         raise ValidationException(f"Deleted data files were found matching the filter for snapshots {conflicting_snapshots}!")
@@ -229,6 +235,7 @@ def _added_data_files(
     data_filter: BooleanExpression | None,
     partition_set: dict[int, set[Record]] | None,
     parent_snapshot: Snapshot | None,
+    case_sensitive: bool = True,
 ) -> Iterator[ManifestEntry]:
     """Return manifest entries for data files added between the starting snapshot and parent snapshot.
 
@@ -238,6 +245,7 @@ def _added_data_files(
         data_filter: Optional filter to match data files
         partition_set: Optional set of partitions to match data files
         parent_snapshot: Parent snapshot to get the history from
+        case_sensitive: Whether the data filter binds column names case-sensitively
 
     Returns:
         Iterator of manifest entries for added data files matching the conditions
@@ -252,7 +260,7 @@ def _added_data_files(
 
     for manifest in manifests:
         for entry in manifest.fetch_manifest_entry(table.io):
-            if _filter_manifest_entries(entry, snapshot_ids, data_filter, partition_set, None, table.schema()):
+            if _filter_manifest_entries(entry, snapshot_ids, data_filter, partition_set, None, table.schema(), case_sensitive):
                 yield entry
 
 
@@ -262,6 +270,7 @@ def _added_delete_files(
     data_filter: BooleanExpression | None,
     partition_set: dict[int, set[Record]] | None,
     parent_snapshot: Snapshot | None,
+    case_sensitive: bool = True,
 ) -> DeleteFileIndex:
     """Return matching delete files that have been added to the table since a starting snapshot.
 
@@ -271,6 +280,7 @@ def _added_delete_files(
         data_filter: Optional filter to match data files
         partition_set: Optional set of partitions to match data files
         parent_snapshot: Parent snapshot to get the history from
+        case_sensitive: Whether the data filter binds column names case-sensitively
 
     Returns:
         DeleteFileIndex
@@ -287,7 +297,7 @@ def _added_delete_files(
     for manifest in manifests:
         for entry in manifest.fetch_manifest_entry(table.io, discard_deleted=True):
             if _filter_manifest_entries(
-                entry, snapshot_ids, data_filter, partition_set, ManifestEntryStatus.ADDED, table.schema()
+                entry, snapshot_ids, data_filter, partition_set, ManifestEntryStatus.ADDED, table.schema(), case_sensitive
             ):
                 dfi.add_delete_file(entry, entry.data_file.partition)
 
@@ -315,6 +325,7 @@ def _validate_added_data_files(
     starting_snapshot: Snapshot,
     data_filter: BooleanExpression | None,
     parent_snapshot: Snapshot | None,
+    case_sensitive: bool = True,
 ) -> None:
     """Validate that no files matching a filter have been added to the table since a starting snapshot.
 
@@ -323,9 +334,10 @@ def _validate_added_data_files(
         starting_snapshot: Snapshot current at the start of the operation
         data_filter: Expression used to find added data files
         parent_snapshot: Ending snapshot on the branch being validated
+        case_sensitive: Whether the data filter binds column names case-sensitively
 
     """
-    conflicting_entries = _added_data_files(table, starting_snapshot, data_filter, None, parent_snapshot)
+    conflicting_entries = _added_data_files(table, starting_snapshot, data_filter, None, parent_snapshot, case_sensitive)
     if any(conflicting_entries):
         conflicting_snapshots = {entry.snapshot_id for entry in conflicting_entries if entry.snapshot_id is not None}
         raise ValidationException(f"Added data files were found matching the filter for snapshots {conflicting_snapshots}!")
@@ -337,6 +349,7 @@ def _validate_no_new_delete_files(
     data_filter: BooleanExpression | None,
     partition_set: dict[int, set[Record]] | None,
     parent_snapshot: Snapshot | None,
+    case_sensitive: bool = True,
 ) -> None:
     """Validate no new delete files matching a filter have been added to the table since starting a snapshot.
 
@@ -346,8 +359,9 @@ def _validate_no_new_delete_files(
         data_filter: Expression used to find added data files
         partition_set: Dictionary of partition spec to set of partition records
         parent_snapshot: Ending snapshot on the branch being validated
+        case_sensitive: Whether the data filter binds column names case-sensitively
     """
-    deletes = _added_delete_files(table, starting_snapshot, data_filter, partition_set, parent_snapshot)
+    deletes = _added_delete_files(table, starting_snapshot, data_filter, partition_set, parent_snapshot, case_sensitive)
 
     if deletes.is_empty():
         return
@@ -364,6 +378,7 @@ def _validate_no_new_deletes_for_data_files(
     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.
 
@@ -373,12 +388,13 @@ def _validate_no_new_deletes_for_data_files(
         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)
+    deletes = _added_delete_files(table, starting_snapshot, data_filter, None, parent_snapshot, case_sensitive)
     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

@@ -207,6 +207,96 @@ def test_concurrent_append_delete_raises_validation_exception(catalog: Catalog)
         tbl2.delete("x == 1")
 
 
+def test_concurrent_append_case_insensitive_delete_retries_successfully(catalog: Catalog) -> None:
+    """A case-insensitive delete that does not conflict with a concurrent append retries and commits."""
+    catalog.create_namespace("default")
+    schema = Schema(NestedField(1, "Id", LongType(), required=False))
+    catalog.create_table("default.app_del_case_test", schema=schema)
+
+    import pyarrow as pa
+
+    tbl = catalog.load_table("default.app_del_case_test")
+    tbl.append(pa.table({"Id": [6, 7, 8]}))
+
+    tbl1 = catalog.load_table("default.app_del_case_test")
+    tbl2 = catalog.load_table("default.app_del_case_test")
+
+    tbl1.append(pa.table({"Id": [0]}))
+    tbl2.delete("id > 5", case_sensitive=False)
+
+    refreshed = catalog.load_table("default.app_del_case_test")
+    assert refreshed.scan().to_arrow()["Id"].to_pylist() == [0]
+
+
+def test_concurrent_append_case_insensitive_delete_raises_validation_exception(catalog: Catalog) -> None:
+    """A case-insensitive delete still detects a conflicting concurrent append."""
+    catalog.create_namespace("default")
+    schema = Schema(NestedField(1, "Id", LongType(), required=False))
+    catalog.create_table("default.app_del_case_conflict_test", schema=schema)
+
+    import pyarrow as pa
+
+    tbl = catalog.load_table("default.app_del_case_conflict_test")
+    tbl.append(pa.table({"Id": [6, 7, 8]}))
+
+    tbl1 = catalog.load_table("default.app_del_case_conflict_test")
+    tbl2 = catalog.load_table("default.app_del_case_conflict_test")
+
+    tbl1.append(pa.table({"Id": [9]}))
+
+    with pytest.raises(ValidationException):
+        tbl2.delete("id > 5", case_sensitive=False)
+
+
+def test_snapshot_isolation_case_insensitive_delete_after_concurrent_delete_retries_successfully(catalog: Catalog) -> None:
+    """Under snapshot isolation, a case-insensitive delete retries after a non-conflicting concurrent delete."""
+    catalog.create_namespace("default")
+    schema = Schema(NestedField(1, "Id", LongType(), required=False))
+    catalog.create_table("default.del_del_case_test", schema=schema, properties={"write.delete.isolation-level": "snapshot"})
+
+    import pyarrow as pa
+
+    tbl = catalog.load_table("default.del_del_case_test")
+    tbl.append(pa.table({"Id": [0, 1, 2]}))
+    tbl.append(pa.table({"Id": [6, 7, 8]}))
+
+    tbl1 = catalog.load_table("default.del_del_case_test")
+    tbl2 = catalog.load_table("default.del_del_case_test")
+
+    # Rewrites the file holding Ids 0 to 2, which has no rows matching id > 5
+    tbl1.delete("Id == 0")
+    tbl2.delete("id > 5", case_sensitive=False)
+
+    refreshed = catalog.load_table("default.del_del_case_test")
+    assert refreshed.scan().to_arrow()["Id"].to_pylist() == [1, 2]
+
+
+def test_snapshot_isolation_case_insensitive_delete_after_concurrent_delete_raises_validation_exception(
+    catalog: Catalog,
+) -> None:
+    """Under snapshot isolation, a case-insensitive delete still detects a conflicting concurrent delete."""
+    catalog.create_namespace("default")
+    schema = Schema(NestedField(1, "Id", LongType(), required=False))
+    catalog.create_table(
+        "default.del_del_case_conflict_test", schema=schema, properties={"write.delete.isolation-level": "snapshot"}
+    )
+
+    import pyarrow as pa
+
+    tbl = catalog.load_table("default.del_del_case_conflict_test")
+    tbl.append(pa.table({"Id": [0, 1, 2]}))
+    tbl.append(pa.table({"Id": [6, 7, 8]}))
+
+    tbl1 = catalog.load_table("default.del_del_case_conflict_test")
+    tbl2 = catalog.load_table("default.del_del_case_conflict_test")
+
+    # Rewrites the file holding Ids 6 to 8, which the delete below also matches
+    tbl1.delete("Id == 8")
+
+    with pytest.raises(ValidationException):
+        tbl2.delete("id > 5", case_sensitive=False)
+
+
 def test_concurrent_delete_append_retries_successfully(catalog: Catalog) -> None:
     """Append after a concurrent delete should succeed via retry."""
     catalog.create_namespace("default")

tests/table/test_validate.py

@@ -21,6 +21,7 @@
 import pytest
 
 from pyiceberg.exceptions import ValidationException
+from pyiceberg.expressions import EqualTo
 from pyiceberg.io import FileIO
 from pyiceberg.manifest import DataFile, DataFileContent, ManifestContent, ManifestEntry, ManifestEntryStatus, ManifestFile
 from pyiceberg.table import Table
@@ -508,6 +509,78 @@ def mock_fetch_manifest_entry(self: ManifestFile, io: FileIO, discard_deleted: b
         assert dfi.referenced_delete_files()[0] == mock_delete_file
 
 
+def test_added_delete_files_case_insensitive_filter(
+    table_v2_with_extensive_snapshots_and_manifests: tuple[Table, dict[int, list[ManifestFile]]],
+) -> None:
+    table, mock_manifests = table_v2_with_extensive_snapshots_and_manifests
+
+    snapshot_history = 100
+    snapshots = table.snapshots()
+    for i in range(1, snapshot_history + 1):
+        snapshots[-i] = snapshots[-i].model_copy(update={"summary": Summary(operation=Operation.DELETE)})
+    table.metadata = table.metadata.model_copy(update={"snapshots": snapshots})
+
+    oldest_snapshot = table.snapshots()[-snapshot_history]
+    newest_snapshot = cast(Snapshot, table.current_snapshot())
+
+    mock_delete_file = DataFile.from_args(
+        content=DataFileContent.POSITION_DELETES,
+        file_path="s3://dummy/path",
+        record_count=1,
+    )
+    mock_delete_file.spec_id = 0
+
+    def mock_read_manifest_side_effect(self: Snapshot, io: FileIO) -> list[ManifestFile]:
+        """Mock the manifests method to use the snapshot_id for lookup."""
+        return mock_manifests.get(self.snapshot_id, [])
+
+    def mock_fetch_manifest_entry(self: ManifestFile, io: FileIO, discard_deleted: bool = True) -> list[ManifestEntry]:
+        return [
+            ManifestEntry.from_args(
+                status=ManifestEntryStatus.ADDED,
+                snapshot_id=self.added_snapshot_id,
+                sequence_number=self.min_sequence_number,
+                data_file=mock_delete_file,
+            )
+        ]
+
+    # The table's column is "x", so this filter only binds case-insensitively
+    data_filter = EqualTo("X", 1)
+    with (
+        patch("pyiceberg.table.snapshots.Snapshot.manifests", new=mock_read_manifest_side_effect),
+        patch("pyiceberg.manifest.ManifestFile.fetch_manifest_entry", new=mock_fetch_manifest_entry),
+    ):
+        dfi = _added_delete_files(
+            table=table,
+            starting_snapshot=newest_snapshot,
+            data_filter=data_filter,
+            partition_set=None,
+            parent_snapshot=oldest_snapshot,
+            case_sensitive=False,
+        )
+        assert not dfi.is_empty()
+        assert dfi.referenced_delete_files()[0] == mock_delete_file
+
+        with pytest.raises(ValidationException):
+            _validate_no_new_delete_files(
+                table=table,
+                starting_snapshot=newest_snapshot,
+                data_filter=data_filter,
+                partition_set=None,
+                parent_snapshot=oldest_snapshot,
+                case_sensitive=False,
+            )
+
+        with pytest.raises(ValueError, match="Could not find field with name X"):
+            _added_delete_files(
+                table=table,
+                starting_snapshot=newest_snapshot,
+                data_filter=data_filter,
+                partition_set=None,
+                parent_snapshot=oldest_snapshot,
+            )
+
+
 def test_validate_no_new_delete_files_raises_on_conflict(
     table_v2_with_extensive_snapshots_and_manifests: tuple[Table, dict[int, list[ManifestFile]]],
 ) -> None: