Preserve nulls when simplifying out-of-range NOT IN predicates

Keep null rows when NOT IN simplifies to NotEqualTo so Arrow scans agree
with the expression evaluator. Cover bounds, metrics, and int-to-long
schema evolution with evaluator and file scan regression tests.

Generated-by: Codex

Author: jackylee-ch

pyiceberg/io/pyarrow.py

@@ -919,7 +919,8 @@ 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)
+        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/expressions/test_evaluator.py

@@ -1939,6 +1939,54 @@ def test_below_int_bounds_in() -> None:
     assert expression_evaluator(schema, NotIn("id", [1, below_min]), True)(Record(IntegerType.min)) is True
 
 
+@pytest.mark.parametrize(
+    "literals",
+    [
+        [IntegerType.max + 1, IntegerType.max + 2],
+        [IntegerType.min - 1, IntegerType.min - 2],
+        [IntegerType.min - 1, IntegerType.max + 1],
+    ],
+)
+def test_int_bounds_in_all_literals_out_of_range(literals: list[int]) -> None:
+    schema = Schema(NestedField(1, "id", IntegerType(), required=False))
+    in_expr = In("id", literals)
+    not_in_expr = NotIn("id", literals)
+
+    assert in_expr.bind(schema) == AlwaysFalse()
+    assert not_in_expr.bind(schema) == AlwaysTrue()
+    for value in [None, IntegerType.min, 0, IntegerType.max]:
+        assert expression_evaluator(schema, in_expr, True)(Record(value)) is False
+        assert expression_evaluator(schema, not_in_expr, True)(Record(value)) is True
+
+
+def test_int_bounds_in_keeps_multiple_literals() -> None:
+    schema = Schema(NestedField(1, "id", IntegerType(), required=False))
+    literals = [1, 2, IntegerType.min - 1, IntegerType.max + 1]
+    in_expr = In("id", literals)
+    not_in_expr = NotIn("id", literals)
+
+    assert in_expr.bind(schema) == In("id", [1, 2]).bind(schema)
+    assert not_in_expr.bind(schema) == NotIn("id", [1, 2]).bind(schema)
+    values = [None, 1, 2, 3, IntegerType.min, IntegerType.max]
+    eval_in = expression_evaluator(schema, in_expr, True)
+    eval_not_in = expression_evaluator(schema, not_in_expr, True)
+    assert [value for value in values if eval_in(Record(value))] == [1, 2]
+    assert [value for value in values if eval_not_in(Record(value))] == [None, 3, IntegerType.min, IntegerType.max]
+
+
+@pytest.mark.parametrize(
+    "boundary,out_of_range",
+    [(IntegerType.min, IntegerType.min - 1), (IntegerType.max, IntegerType.max + 1)],
+)
+def test_int_bounds_in_metrics(schema_data_file: Schema, boundary: int, out_of_range: int) -> None:
+    bounds = {1: to_bytes(IntegerType(), boundary)}
+    data_file = _single_value_metrics_file(boundary, lower_bounds=bounds, upper_bounds=bounds)
+
+    assert _InclusiveMetricsEvaluator(schema_data_file, In("id", [1, out_of_range])).eval(data_file) == ROWS_CANNOT_MATCH
+    assert _StrictMetricsEvaluator(schema_data_file, NotIn("id", [1, out_of_range])).eval(data_file) == ROWS_MUST_MATCH
+    assert _StrictMetricsEvaluator(schema_data_file, In("id", [1, boundary, out_of_range])).eval(data_file) == ROWS_MUST_MATCH
+
+
 def test_int_bounds_in_keeps_the_boundary_value() -> None:
     """A sentinel is equal to the boundary literal it clamps to, so it must not absorb it."""
     schema = Schema(NestedField(1, "id", IntegerType(), required=False))

tests/io/test_pyarrow.py

@@ -59,7 +59,9 @@
     BoundReference,
     BoundStartsWith,
     GreaterThan,
+    In,
     Not,
+    NotIn,
     Or,
 )
 from pyiceberg.expressions.literals import literal
@@ -783,10 +785,41 @@ 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")>'
-    )
+    table = pa.table({"foo": [None, "hello", "world"]})
+    expression = expression_to_pyarrow(BoundNotEqualTo(bound_reference, literal("hello")))
+    assert table.filter(expression).column("foo").to_pylist() == [None, "world"]
+
+
+@pytest.mark.parametrize("boundary", [IntegerType.min, IntegerType.max])
+@pytest.mark.parametrize("valid_values", [[], [1], [1, 2], [IntegerType.min], [IntegerType.max]])
+def test_scan_in_out_of_range_literals(catalog: InMemoryCatalog, tmp_path: Path, boundary: int, valid_values: list[int]) -> None:
+    schema = Schema(NestedField(1, "id", IntegerType(), required=False))
+    catalog.create_namespace("default")
+    table = catalog.create_table("default.out_of_range", schema=schema, location=str(tmp_path))
+    values = [None, IntegerType.min, 1, 2, IntegerType.max]
+    table.append(pa.table({"id": pa.array(values, type=pa.int32())}))
+    out_of_range = boundary - 1 if boundary == IntegerType.min else boundary + 1
+    literals = [*valid_values, out_of_range, out_of_range * 2]
+
+    assert table.scan(row_filter=In("id", literals)).to_arrow().column("id").to_pylist() == [
+        value for value in values if value in valid_values
+    ]
+    assert table.scan(row_filter=NotIn("id", literals)).to_arrow().column("id").to_pylist() == [
+        value for value in values if value not in valid_values
+    ]
+
+
+def test_scan_in_out_of_range_literals_after_type_promotion(catalog: InMemoryCatalog, tmp_path: Path) -> None:
+    schema = Schema(NestedField(1, "id", IntegerType(), required=False))
+    catalog.create_namespace("default")
+    table = catalog.create_table("default.promoted_int", schema=schema, location=str(tmp_path))
+    table.append(pa.table({"id": pa.array([None, 1, IntegerType.max], type=pa.int32())}))
+    with table.update_schema() as update:
+        update.update_column("id", field_type=LongType())
+    table.append(pa.table({"id": pa.array([2**40], type=pa.int64())}))
+
+    assert sorted(table.scan(row_filter=In("id", [1, 2**40])).to_arrow().column("id").to_pylist()) == [1, 2**40]
+    assert table.scan(row_filter=NotIn("id", [1, 2**40])).to_arrow().column("id").to_pylist() == [None, IntegerType.max]
 
 
 def test_expr_greater_than_or_equal_equal_to_pyarrow(bound_reference: BoundReference) -> None: