Keep rows where a delete filter evaluates to null

pyiceberg/io/pyarrow.py

@@ -69,7 +69,7 @@
 
 from pyiceberg.conversions import to_bytes
 from pyiceberg.exceptions import ResolveError
-from pyiceberg.expressions import AlwaysTrue, BooleanExpression, BoundIsNaN, BoundIsNull, BoundTerm, Not, Or
+from pyiceberg.expressions import AlwaysFalse, AlwaysTrue, BooleanExpression, BoundIsNaN, BoundTerm, Or
 from pyiceberg.expressions.literals import Literal
 from pyiceberg.expressions.visitors import (
     BoundBooleanExpressionVisitor,
@@ -1125,25 +1125,24 @@ def _expression_to_complementary_pyarrow(expr: BooleanExpression, schema: Schema
     """Complementary filter conversion function of expression_to_pyarrow.
 
     Could not use expression_to_pyarrow(Not(expr)) to achieve this complementary effect because
-    ~ in pyarrow.compute.Expression does not handle null.
+    ~ in pyarrow.compute.Expression does not handle null: where expr evaluates to null, Not(expr) is null too
+    and the row is dropped. Only the rows where expr is true are deleted, so null is read as false.
     """
     collector = _NullNaNUnmentionedTermsCollector()
     collector.collect(expr)
 
     # Convert the set of terms to a sorted list so that layout of the expression to build is deterministic.
-    null_unmentioned_bound_terms: list[BoundTerm] = sorted(
-        collector.null_unmentioned_bound_terms, key=lambda term: term.ref().field.name
-    )
     nan_unmentioned_bound_terms: list[BoundTerm] = sorted(
         collector.nan_unmentioned_bound_terms, key=lambda term: term.ref().field.name
     )
 
-    preserve_expr: BooleanExpression = Not(expr)
-    for term in null_unmentioned_bound_terms:
-        preserve_expr = Or(preserve_expr, BoundIsNull(term=term))
+    preserve_expr = ~pc.coalesce(expression_to_pyarrow(expr, schema), pc.scalar(False))
+    nan_expr: BooleanExpression = AlwaysFalse()
     for term in nan_unmentioned_bound_terms:
-        preserve_expr = Or(preserve_expr, BoundIsNaN(term=term))
-    return expression_to_pyarrow(preserve_expr, schema)
+        nan_expr = Or(nan_expr, BoundIsNaN(term=term))
+    if nan_expr == AlwaysFalse():
+        return preserve_expr
+    return preserve_expr | expression_to_pyarrow(nan_expr, schema)
 
 
 @lru_cache

tests/catalog/test_catalog_behaviors.py

@@ -1295,6 +1295,31 @@ def test_overwrite_record_batch_reader(catalog: Catalog) -> None:
     assert len(tbl.scan().to_arrow()) == total_rows
 
 
+def test_delete_and_overwrite_filter_with_null(catalog: Catalog) -> None:
+    catalog.create_namespace("default")
+    pa_table = pa.table({"x": [1, 2, 3], "p": [1, 2, 2], "v": [None, None, "c"]})
+    filters = [
+        # A null v makes the second branch null for x=2, which matches neither branch
+        "(p == 1 AND v IS NULL) OR (p == 2 AND v == 'c')",
+        # A null v leaves the filter true for x=1
+        "p == 1 OR v == 'c'",
+        "v != 'c'",
+        "NOT (v == 'c')",
+    ]
+    for i, row_filter in enumerate(filters):
+        tbl = catalog.create_table(identifier=f"default.delete_filter_with_null_{i}_{catalog.name}", schema=pa_table.schema)
+        tbl.append(pa_table)
+        matched = set(tbl.scan(row_filter=row_filter).to_arrow()["x"].to_pylist())
+        expected = sorted({1, 2, 3} - matched)
+
+        tbl.delete(row_filter)
+        assert sorted(tbl.scan().to_arrow()["x"].to_pylist()) == expected, row_filter
+
+        tbl.overwrite(pa_table, overwrite_filter="x > 0")
+        tbl.overwrite(pa_table.slice(0, 0), overwrite_filter=row_filter)
+        assert sorted(tbl.scan().to_arrow()["x"].to_pylist()) == expected, row_filter
+
+
 def test_append_record_batch_reader_to_partitioned_table_raises(catalog: Catalog) -> None:
     catalog.create_namespace("default")
     identifier = f"default.append_record_batch_reader_partitioned_{catalog.name}"

tests/io/test_pyarrow_visitor.py

@@ -856,10 +856,9 @@ def test_expression_to_complementary_pyarrow(
     # Notice an isNan predicate on a str column is automatically converted to always false and removed from Or
     # and thus will not appear in the pc.expr.
     expected_repr = (
-        '<pyarrow.compute.Expression (((invert(((((string_field == "hello") and (float_field > 100)) '
+        '<pyarrow.compute.Expression (invert(coalesce(((((string_field == "hello") and (float_field > 100)) '
         "or ((is_nan(float_field) and (double_field == 0)) or (float_field > 100))) "
-        "and invert(is_null(double_field, {nan_is_null=false})))) "
-        "or is_null(float_field, {nan_is_null=false})) "
-        "or is_null(string_field, {nan_is_null=false})) or is_nan(double_field))>"
+        "and invert(is_null(double_field, {nan_is_null=false}))), false)) "
+        "or is_nan(double_field))>"
     )
     assert repr(result) == expected_repr