fix(io): keep nulls when pushing NotEqualTo down to Arrow

An Arrow comparison yields null for a null input, so `field != value` dropped
every row where the column is null. All the other evaluators treat a null as
satisfying NotEqualTo: `_ExpressionEvaluationVisitor.visit_not_equal` returns
`None != value`, `_StrictMetricsEvaluationVisitor` reports ROWS_MUST_MATCH for
a column proven to hold only nulls, and the Arrow NOT IN path already keeps
them because `~isin(...)` is false for a null.

The reference implementation agrees: `Evaluator.notEq` is `!eq`, and `eq`
compares with a nulls-first comparator, so a null never equals the literal.

Keep the null explicitly. NOT IN needs no change.

Co-Authored-By: Claude Code <noreply@anthropic.com>

Author: jackylee-ch

pyiceberg/io/pyarrow.py

@@ -919,7 +919,10 @@ def visit_equal(self, term: BoundTerm, literal: Literal[Any]) -> pc.Expression:
         return pc.field(self._get_field_name(term)) == _convert_scalar(literal.value, term.ref().field.field_type)
 
     def visit_not_equal(self, term: BoundTerm, literal: Literal[Any]) -> pc.Expression:
-        return pc.field(self._get_field_name(term)) != _convert_scalar(literal.value, term.ref().field.field_type)
+        # A null is not equal to the literal, but an Arrow comparison yields null for it and
+        # the row would be dropped. Keep it explicitly to match the other evaluators.
+        ref = pc.field(self._get_field_name(term))
+        return ref.is_null(nan_is_null=False) | (ref != _convert_scalar(literal.value, term.ref().field.field_type))
 
     def visit_greater_than_or_equal(self, term: BoundTerm, literal: Literal[Any]) -> pc.Expression:
         return pc.field(self._get_field_name(term)) >= _convert_scalar(literal.value, term.ref().field.field_type)

tests/io/test_pyarrow.py

@@ -60,6 +60,7 @@
     BoundStartsWith,
     GreaterThan,
     Not,
+    NotEqualTo,
     Or,
 )
 from pyiceberg.expressions.literals import literal
@@ -785,10 +786,23 @@ def test_expr_equal_to_pyarrow(bound_reference: BoundReference) -> None:
 def test_expr_not_equal_to_pyarrow(bound_reference: BoundReference) -> None:
     assert (
         repr(expression_to_pyarrow(BoundNotEqualTo(bound_reference, literal("hello"))))
-        == '<pyarrow.compute.Expression (foo != "hello")>'
+        == '<pyarrow.compute.Expression (is_null(foo, {nan_is_null=false}) or (foo != "hello"))>'
     )
 
 
+def test_expr_not_equal_to_pyarrow_keeps_nulls(bound_reference: BoundReference) -> None:
+    """A null is not equal to the literal, so it satisfies NotEqualTo, as it does in the other evaluators."""
+    from pyiceberg.expressions.visitors import expression_evaluator
+
+    values = ["hello", "world", None]
+    pushed_down = pa.table({"foo": values}).filter(expression_to_pyarrow(BoundNotEqualTo(bound_reference, literal("hello"))))
+    schema = Schema(NestedField(field_id=1, name="foo", field_type=StringType(), required=False))
+    in_memory = expression_evaluator(schema, NotEqualTo("foo", "hello"), True)
+
+    assert pushed_down.column("foo").to_pylist() == ["world", None]
+    assert [value for value in values if in_memory(Record(value))] == ["world", None]
+
+
 def test_expr_greater_than_or_equal_equal_to_pyarrow(bound_reference: BoundReference) -> None:
     assert (
         repr(expression_to_pyarrow(BoundGreaterThanOrEqual(bound_reference, literal("hello"))))