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

Reviewer feedback: bind should not drop what the caller wrote. Converting a
literal outside the field's range yields an AboveMax/BelowMin sentinel carrying
the clamped boundary value, so keep the original literal instead of the
sentinel. The bound set then round-trips, and `value_set` leaves the value out
so no evaluator sees it -- every BoundBooleanExpressionVisitor receives
`value_set`, so that is the one place the range has to be taken into account.

Storing the sentinel is not an option: it is `==` and hash-equal to the boundary
literal it clamps to, so the set collapses and a boundary value the caller did
write can be lost.

The tests now assert evaluation results rather than the bound shape, which no
longer folds, plus the round trip itself.

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

Author: jackylee-ch

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."""
 
@@ -698,14 +704,11 @@ def __init__(
     def bind(self, schema: Schema, case_sensitive: bool = True) -> BoundSetPredicate:
         bound_term = self.term.bind(schema, case_sensitive)
         field_type = bound_term.ref().field.field_type
-        # Literals outside the field's range can never match, so drop them rather than
-        # keep the clamped AboveMax/BelowMin sentinel in the bound set. Filter while
-        # building the set: a sentinel is equal to the boundary literal it clamps to,
-        # so collecting first would let it absorb a boundary value the user did write.
+        # 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 = {
-            bound_literal
-            for bound_literal in (lit.to(field_type) for lit in self.literals)
-            if not isinstance(bound_literal, (AboveMax, BelowMin))
+            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
 
@@ -744,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

@@ -1912,31 +1912,40 @@ def test_strict_metrics_eval_bounds_after_promotion(
     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
 
-    assert In("id", [1, above_max]).bind(schema) == EqualTo("id", 1).bind(schema)
-    assert NotIn("id", [1, above_max]).bind(schema) == NotEqualTo("id", 1).bind(schema)
-    assert In("id", [above_max]).bind(schema) == AlwaysFalse()
-    assert NotIn("id", [above_max]).bind(schema) == AlwaysTrue()
-
-    # The clamped literal used to match the field's maximum
+    # 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 In("id", [1, below_min]).bind(schema) == EqualTo("id", 1).bind(schema)
-    assert NotIn("id", [1, below_min]).bind(schema) == NotEqualTo("id", 1).bind(schema)
-    assert In("id", [below_min]).bind(schema) == AlwaysFalse()
-    assert NotIn("id", [below_min]).bind(schema) == AlwaysTrue()
-
     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(
@@ -1952,8 +1961,6 @@ def test_int_bounds_in_all_literals_out_of_range(literals: list[int]) -> None:
     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
@@ -1965,8 +1972,6 @@ def test_int_bounds_in_keeps_multiple_literals() -> None:
     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)
@@ -1988,13 +1993,13 @@ def test_int_bounds_in_metrics(schema_data_file: Schema, boundary: int, out_of_r
 
 
 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."""
+    """A converted out-of-range literal clamps to the boundary, so it must not shadow it."""
     schema = Schema(NestedField(1, "id", IntegerType(), required=False))
 
-    assert In("id", [IntegerType.max, IntegerType.max + 1]).bind(schema) == EqualTo("id", IntegerType.max).bind(schema)
-    assert NotIn("id", [IntegerType.max, IntegerType.max + 1]).bind(schema) == NotEqualTo("id", IntegerType.max).bind(schema)
-    assert In("id", [IntegerType.min, IntegerType.min - 1]).bind(schema) == EqualTo("id", IntegerType.min).bind(schema)
-    assert NotIn("id", [IntegerType.min, IntegerType.min - 1]).bind(schema) == NotEqualTo("id", IntegerType.min).bind(schema)
-
-    assert expression_evaluator(schema, In("id", [IntegerType.max, IntegerType.max + 1]), True)(Record(IntegerType.max)) is True
-    assert expression_evaluator(schema, In("id", [IntegerType.min, IntegerType.min - 1]), True)(Record(IntegerType.min)) is True
+    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