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