diff --git a/pyiceberg/expressions/__init__.py b/pyiceberg/expressions/__init__.py index ef4cb2506e..1de8b4aa00 100644 --- a/pyiceberg/expressions/__init__.py +++ b/pyiceberg/expressions/__init__.py @@ -697,8 +697,17 @@ 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 + # 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. + bound_literals = { + bound_literal + for bound_literal in (lit.to(field_type) for lit in self.literals) + if not isinstance(bound_literal, (AboveMax, BelowMin)) + } + return self.as_bound(bound_term, bound_literals) # type: ignore def __str__(self) -> str: """Return the string representation of the SetPredicate class.""" diff --git a/tests/expressions/test_evaluator.py b/tests/expressions/test_evaluator.py index bba4156e99..fd5bf0047e 100644 --- a/tests/expressions/test_evaluator.py +++ b/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,91 @@ def test_strict_metrics_eval_bounds_after_promotion( evaluator = _StrictMetricsEvaluator(schema, op("col", lit)) assert evaluator.eval(data_file) == expected + + +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 + 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 + + +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 + + +@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)) + + 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 diff --git a/tests/io/test_pyarrow.py b/tests/io/test_pyarrow.py index b31c18949b..980be2c44d 100644 --- a/tests/io/test_pyarrow.py +++ b/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"))))