From 68c7ca2917048970356cd98bf25d95e7739d4e97 Mon Sep 17 00:00:00 2001 From: Rahul-cs50 Date: Sat, 31 Oct 2026 12:25:59 +0530 Subject: [PATCH] Keep rows where a delete filter evaluates to null delete() and overwrite() rewrite a data file with the rows the delete filter does not select. The complementary filter was Not(filter) OR'd with IS NULL for each column the filter does not null-check, so a null kept rows the filter matches (p == 1 OR v == 'c' with a null v) and dropped rows it does not match, where Not(filter) is null. Coalesce the filter to false before inverting it, so exactly the rows where the filter is true are deleted, the rows scan(row_filter=filter) returns. Closes #4184 --- pyiceberg/io/pyarrow.py | 19 +++++++++---------- tests/catalog/test_catalog_behaviors.py | 25 +++++++++++++++++++++++++ tests/io/test_pyarrow_visitor.py | 7 +++---- 3 files changed, 37 insertions(+), 14 deletions(-) diff --git a/pyiceberg/io/pyarrow.py b/pyiceberg/io/pyarrow.py index 1dd816d5d3..69f0ca9d06 100644 --- a/pyiceberg/io/pyarrow.py +++ b/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 diff --git a/tests/catalog/test_catalog_behaviors.py b/tests/catalog/test_catalog_behaviors.py index 1148ca8527..394c9efc15 100644 --- a/tests/catalog/test_catalog_behaviors.py +++ b/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}" diff --git a/tests/io/test_pyarrow_visitor.py b/tests/io/test_pyarrow_visitor.py index e98d76e262..2183d5b6f9 100644 --- a/tests/io/test_pyarrow_visitor.py +++ b/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 = ( - ' 100)) ' + ' 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