Reject out-of-range literals when binding IN / NOT IN

pyiceberg/expressions/__init__.py

@@ -30,7 +30,7 @@
 from pyiceberg.expressions.literals import AboveMax, BelowMin, Literal, literal
 from pyiceberg.schema import Accessor, Schema
 from pyiceberg.typedef import IcebergBaseModel, IcebergRootModel, L, LiteralValue, StructProtocol
-from pyiceberg.types import DoubleType, FloatType, NestedField
+from pyiceberg.types import DoubleType, FloatType, IcebergType, NestedField
 from pyiceberg.utils.singleton import Singleton
 
 
@@ -49,6 +49,12 @@ def _to_literal(value: L | Literal[L]) -> Literal[L]:
         return literal(value)
 
 
+def _to_bound_literal(lit: LiteralValue, field_type: IcebergType) -> LiteralValue | None:
+    """Convert a literal to the field's type, or None when the field can never hold its value."""
+    converted = lit.to(field_type)
+    return None if isinstance(converted, (AboveMax, BelowMin)) else converted
+
+
 class BooleanExpression(IcebergBaseModel, ABC):
     """An expression that evaluates to a boolean."""
 
@@ -697,8 +703,14 @@ def __init__(
 
     def bind(self, schema: Schema, case_sensitive: bool = True) -> BoundSetPredicate:
         bound_term = self.term.bind(schema, case_sensitive)
-        literal_set = self.literals
-        return self.as_bound(bound_term, {lit.to(bound_term.ref().field.field_type) for lit in literal_set})  # type: ignore
+        field_type = bound_term.ref().field.field_type
+        # An out-of-range literal converts to an AboveMax/BelowMin sentinel that carries the
+        # clamped boundary value, which would then match rows at the boundary. Keep the original
+        # literal instead, so the bound set still round-trips to what the caller wrote.
+        bound_literals = {
+            converted if (converted := _to_bound_literal(lit, field_type)) is not None else lit for lit in self.literals
+        }
+        return self.as_bound(bound_term, bound_literals)  # type: ignore
 
     def __str__(self) -> str:
         """Return the string representation of the SetPredicate class."""
@@ -735,7 +747,10 @@ def __init__(self, term: BoundTerm, literals: set[LiteralValue]) -> None:
 
     @cached_property
     def value_set(self) -> set[Any]:
-        return {lit.value for lit in self.literals}
+        field_type = self.term.ref().field.field_type
+        # `literals` keeps every literal the caller wrote so the predicate round-trips, but a
+        # value the field can never hold must not reach an evaluator.
+        return {converted.value for lit in self.literals if (converted := _to_bound_literal(lit, field_type)) is not None}
 
     def __str__(self) -> str:
         """Return the string representation of the BoundSetPredicate class."""

tests/expressions/test_evaluator.py

@@ -24,6 +24,8 @@
 
 from pyiceberg.conversions import to_bytes
 from pyiceberg.expressions import (
+    AlwaysFalse,
+    AlwaysTrue,
     And,
     BooleanExpression,
     EqualTo,
@@ -51,6 +53,7 @@
     ROWS_MUST_MATCH,
     _InclusiveMetricsEvaluator,
     _StrictMetricsEvaluator,
+    expression_evaluator,
 )
 from pyiceberg.manifest import DataFile, FileFormat
 from pyiceberg.schema import Schema
@@ -1907,3 +1910,96 @@ def test_strict_metrics_eval_bounds_after_promotion(
 
     evaluator = _StrictMetricsEvaluator(schema, op("col", lit))
     assert evaluator.eval(data_file) == expected
+
+
+def test_bind_preserves_out_of_range_literals() -> None:
+    """Binding keeps the literals the caller wrote, so the bound predicate still round-trips."""
+    schema = Schema(NestedField(1, "id", IntegerType(), required=False))
+    literals = [1, IntegerType.max + 1]
+
+    for predicate in [In("id", literals), NotIn("id", literals)]:
+        bound = predicate.bind(schema)
+        assert {lit.value for lit in bound.literals} == set(literals)
+        assert bound.as_unbound(bound.term.ref().field.name, bound.literals) == predicate
+        # Only the values the field can hold reach an evaluator
+        assert bound.value_set == {1}
+
+
+def test_above_int_bounds_in() -> None:
+    schema = Schema(NestedField(1, "id", IntegerType(), required=False))
+    above_max = IntegerType.max + 1
+
+    # The out-of-range literal used to be clamped to the maximum and match rows there
+    assert expression_evaluator(schema, In("id", [1, above_max]), True)(Record(IntegerType.max)) is False
+    assert expression_evaluator(schema, NotIn("id", [1, above_max]), True)(Record(IntegerType.max)) is True
+    assert expression_evaluator(schema, In("id", [1, above_max]), True)(Record(1)) is True
+    assert In("id", [above_max]).bind(schema) == AlwaysFalse()
+    assert NotIn("id", [above_max]).bind(schema) == AlwaysTrue()
+
+
+def test_below_int_bounds_in() -> None:
+    schema = Schema(NestedField(1, "id", IntegerType(), required=False))
+    below_min = IntegerType.min - 1
+
+    assert expression_evaluator(schema, In("id", [1, below_min]), True)(Record(IntegerType.min)) is False
+    assert expression_evaluator(schema, NotIn("id", [1, below_min]), True)(Record(IntegerType.min)) is True
+    assert expression_evaluator(schema, In("id", [1, below_min]), True)(Record(1)) is True
+    assert In("id", [below_min]).bind(schema) == AlwaysFalse()
+    assert NotIn("id", [below_min]).bind(schema) == AlwaysTrue()
+
+
+@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)
+
+    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)
+
+    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 converted out-of-range literal clamps to the boundary, so it must not shadow it."""
+    schema = Schema(NestedField(1, "id", IntegerType(), required=False))
+
+    for boundary, out_of_range in [(IntegerType.max, IntegerType.max + 1), (IntegerType.min, IntegerType.min - 1)]:
+        eval_in = expression_evaluator(schema, In("id", [boundary, out_of_range]), True)
+        eval_not_in = expression_evaluator(schema, NotIn("id", [boundary, out_of_range]), True)
+        assert eval_in(Record(boundary)) is True
+        assert eval_not_in(Record(boundary)) is False
+        assert eval_in(Record(0)) is False
+        assert eval_not_in(Record(0)) is True

tests/io/test_pyarrow.py

@@ -59,7 +59,9 @@
     BoundReference,
     BoundStartsWith,
     GreaterThan,
+    In,
     Not,
+    NotIn,
     Or,
 )
 from pyiceberg.expressions.literals import literal
@@ -789,6 +791,38 @@ def test_expr_not_equal_to_pyarrow(bound_reference: BoundReference) -> None:
     )
 
 
+@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 = [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([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() == [IntegerType.max]
+
+
 def test_expr_greater_than_or_equal_equal_to_pyarrow(bound_reference: BoundReference) -> None:
     assert (
         repr(expression_to_pyarrow(BoundGreaterThanOrEqual(bound_reference, literal("hello"))))