Remove the IN predicate limit so large IN predicates prune files

pyiceberg/expressions/visitors.py

@@ -574,9 +574,6 @@ def visit_in(self, term: BoundTerm, literals: set[L]) -> bool:
         if field.lower_bound is None:
             return ROWS_CANNOT_MATCH
 
-        if len(literals) > IN_PREDICATE_LIMIT:
-            return ROWS_MIGHT_MATCH
-
         lower = _from_byte_buffer(term.ref().field.field_type, field.lower_bound)
 
         if all(lower > val for val in literals):
@@ -1388,10 +1385,6 @@ def visit_in(self, term: BoundTerm, literals: set[L]) -> bool:
         if self._contains_nulls_only(field_id) or self._contains_nans_only(field_id):
             return ROWS_CANNOT_MATCH
 
-        if len(literals) > IN_PREDICATE_LIMIT:
-            # skip evaluating the predicate if the number of values is too big
-            return ROWS_MIGHT_MATCH
-
         if not isinstance(field.field_type, PrimitiveType):
             raise ValueError(f"Expected PrimitiveType: {field.field_type}")
 

tests/expressions/test_evaluator.py

@@ -45,6 +45,7 @@
     StartsWith,
 )
 from pyiceberg.expressions.visitors import (
+    IN_PREDICATE_LIMIT,
     ROWS_CANNOT_MATCH,
     ROWS_MIGHT_MATCH,
     ROWS_MIGHT_NOT_MATCH,
@@ -674,6 +675,47 @@ def test_integer_in(schema_data_file: Schema, data_file: DataFile) -> None:
     assert should_read, "Should read: large in expression"
 
 
+def test_integer_in_above_limit(schema_data_file: Schema, data_file: DataFile) -> None:
+    below_lower = set(range(INT_MIN_VALUE - IN_PREDICATE_LIMIT - 1, INT_MIN_VALUE))
+    should_read = _InclusiveMetricsEvaluator(schema_data_file, In("id", below_lower)).eval(data_file)
+    assert not should_read, "Should not read: id below lower bound (max 29 < 30)"
+
+    above_upper = set(range(INT_MAX_VALUE + 1, INT_MAX_VALUE + IN_PREDICATE_LIMIT + 2))
+    should_read = _InclusiveMetricsEvaluator(schema_data_file, In("id", above_upper)).eval(data_file)
+    assert not should_read, "Should not read: id above upper bound (min 80 > 79)"
+
+    equal_lower = set(range(INT_MIN_VALUE - IN_PREDICATE_LIMIT, INT_MIN_VALUE + 1))
+    should_read = _InclusiveMetricsEvaluator(schema_data_file, In("id", equal_lower)).eval(data_file)
+    assert should_read, "Should read: id equal to lower bound (max 30 == 30)"
+
+    equal_upper = set(range(INT_MAX_VALUE, INT_MAX_VALUE + IN_PREDICATE_LIMIT + 1))
+    should_read = _InclusiveMetricsEvaluator(schema_data_file, In("id", equal_upper)).eval(data_file)
+    assert should_read, "Should read: id equal to upper bound (min 79 == 79)"
+
+    straddle = set(range(INT_MIN_VALUE - IN_PREDICATE_LIMIT, INT_MAX_VALUE + 2))
+    should_read = _InclusiveMetricsEvaluator(schema_data_file, In("id", straddle)).eval(data_file)
+    assert should_read, "Should read: id range overlaps bounds"
+
+    outside_both = {*range(INT_MIN_VALUE - IN_PREDICATE_LIMIT, INT_MIN_VALUE), INT_MAX_VALUE + 1}
+    should_read = _InclusiveMetricsEvaluator(schema_data_file, In("id", outside_both)).eval(data_file)
+    assert not should_read, "Should not read: no id between lower and upper bounds"
+
+    should_read = _InclusiveMetricsEvaluator(
+        schema_data_file, In("all_nulls", {str(i) for i in range(IN_PREDICATE_LIMIT + 1)})
+    ).eval(data_file)
+    assert not should_read, "Should skip: in on all nulls column"
+
+
+def test_integer_in_at_limit(schema_data_file: Schema, data_file: DataFile) -> None:
+    below_lower = set(range(INT_MIN_VALUE - IN_PREDICATE_LIMIT, INT_MIN_VALUE))
+    should_read = _InclusiveMetricsEvaluator(schema_data_file, In("id", below_lower)).eval(data_file)
+    assert not should_read, "Should not read: id below lower bound (max 29 < 30)"
+
+    outside_both = {*range(INT_MIN_VALUE - IN_PREDICATE_LIMIT + 1, INT_MIN_VALUE), INT_MAX_VALUE + 1}
+    should_read = _InclusiveMetricsEvaluator(schema_data_file, In("id", outside_both)).eval(data_file)
+    assert not should_read, "Should not read: no id between lower and upper bounds"
+
+
 def test_integer_not_in(schema_data_file: Schema, data_file: DataFile) -> None:
     should_read = _InclusiveMetricsEvaluator(schema_data_file, NotIn("id", {INT_MIN_VALUE - 25, INT_MIN_VALUE - 24})).eval(
         data_file
@@ -909,6 +951,34 @@ def test_inclusive_metrics_evaluator_in(schema_data_file_nan: Schema, data_file_
     assert should_read, "Should match: overlap with upper bounds"
 
 
+def test_inclusive_metrics_evaluator_in_above_limit(schema_data_file_nan: Schema, data_file_nan: DataFile) -> None:
+    below_seven = {float(i) for i in range(-IN_PREDICATE_LIMIT, 1)}
+    above_twenty_two = {float(i) for i in range(30, 30 + IN_PREDICATE_LIMIT + 1)}
+
+    should_read = _InclusiveMetricsEvaluator(schema_data_file_nan, In("all_nan", below_seven)).eval(data_file_nan)
+    assert not should_read, "Should not match: all nan column doesn't contain number"
+
+    should_read = _InclusiveMetricsEvaluator(schema_data_file_nan, In("max_nan", below_seven)).eval(data_file_nan)
+    assert not should_read, "Should not match: all values are smaller than lower bound"
+
+    should_read = _InclusiveMetricsEvaluator(schema_data_file_nan, In("max_nan", above_twenty_two)).eval(data_file_nan)
+    assert should_read, "Should match: upper bound is nan"
+
+    should_read = _InclusiveMetricsEvaluator(schema_data_file_nan, In("min_max_nan", below_seven)).eval(data_file_nan)
+    assert should_read, "Should match: no visibility"
+
+    should_read = _InclusiveMetricsEvaluator(schema_data_file_nan, In("all_nan_null_bounds", below_seven)).eval(data_file_nan)
+    assert not should_read, "Should not match: all nan column doesn't contain number"
+
+    should_read = _InclusiveMetricsEvaluator(schema_data_file_nan, In("some_nan_correct_bounds", below_seven)).eval(data_file_nan)
+    assert not should_read, "Should not match: all values are smaller than lower bound"
+
+    should_read = _InclusiveMetricsEvaluator(schema_data_file_nan, In("some_nan_correct_bounds", above_twenty_two)).eval(
+        data_file_nan
+    )
+    assert not should_read, "Should not match: all values are larger than upper bound"
+
+
 def test_inclusive_metrics_evaluator_not_in(schema_data_file_nan: Schema, data_file_nan: DataFile) -> None:
     should_read = _InclusiveMetricsEvaluator(schema_data_file_nan, NotIn("all_nan", (1, 10, 30))).eval(data_file_nan)
     assert should_read, "Should match: no visibility"

tests/expressions/test_visitors.py

@@ -66,6 +66,7 @@
 )
 from pyiceberg.expressions.literals import Literal, literal
 from pyiceberg.expressions.visitors import (
+    IN_PREDICATE_LIMIT,
     MAX_DNF_TERMS,
     BindVisitor,
     BooleanExpressionVisitor,
@@ -1360,6 +1361,38 @@ def test_integer_in(schema: Schema, manifest: ManifestFile) -> None:
     )
 
 
+def test_integer_in_above_limit(schema: Schema, manifest: ManifestFile) -> None:
+    below_lower = range(INT_MIN_VALUE - IN_PREDICATE_LIMIT - 1, INT_MIN_VALUE)
+    assert not _ManifestEvalVisitor(schema, In(Reference("id"), below_lower), case_sensitive=True).eval(manifest), (
+        "Should not read: id below lower bound (max 29 < 30)"
+    )
+
+    above_upper = range(INT_MAX_VALUE + 1, INT_MAX_VALUE + IN_PREDICATE_LIMIT + 2)
+    assert not _ManifestEvalVisitor(schema, In(Reference("id"), above_upper), case_sensitive=True).eval(manifest), (
+        "Should not read: id above upper bound (min 80 > 79)"
+    )
+
+    equal_lower = range(INT_MIN_VALUE - IN_PREDICATE_LIMIT, INT_MIN_VALUE + 1)
+    assert _ManifestEvalVisitor(schema, In(Reference("id"), equal_lower), case_sensitive=True).eval(manifest), (
+        "Should read: id equal to lower bound (max 30 == 30)"
+    )
+
+    equal_upper = range(INT_MAX_VALUE, INT_MAX_VALUE + IN_PREDICATE_LIMIT + 1)
+    assert _ManifestEvalVisitor(schema, In(Reference("id"), equal_upper), case_sensitive=True).eval(manifest), (
+        "Should read: id equal to upper bound (min 79 == 79)"
+    )
+
+    straddle = range(INT_MIN_VALUE - IN_PREDICATE_LIMIT, INT_MAX_VALUE + 2)
+    assert _ManifestEvalVisitor(schema, In(Reference("id"), straddle), case_sensitive=True).eval(manifest), (
+        "Should read: id range overlaps bounds"
+    )
+
+    outside_both = [*range(INT_MIN_VALUE - IN_PREDICATE_LIMIT, INT_MIN_VALUE), INT_MAX_VALUE + 1]
+    assert _ManifestEvalVisitor(schema, In(Reference("id"), outside_both), case_sensitive=True).eval(manifest), (
+        "Should read: id range covers bounds"
+    )
+
+
 def test_integer_not_in(schema: Schema, manifest: ManifestFile) -> None:
     assert _ManifestEvalVisitor(
         schema, NotIn(Reference("id"), (INT_MIN_VALUE - 25, INT_MIN_VALUE - 24)), case_sensitive=True