Prune files on large IN predicates with a min and max check

An IN predicate with more than 200 values returned ROWS_MIGHT_MATCH
before the file bounds were checked. Above that limit, plan_files kept
every data file, so a delete or an upsert scanned the whole table.

Keep the exact check at 200 values or fewer. Above the limit, compare
only the smallest and the largest value against the file bounds. The
result is always a superset of the exact check.

Closes #4003

Claude-Session: https://claude.ai/code/session_01XwKs9csmvNooLVfK61qYwA

Author: azwanzuharimi

pyiceberg/expressions/visitors.py

@@ -575,7 +575,7 @@ def visit_in(self, term: BoundTerm, literals: set[L]) -> bool:
             return ROWS_CANNOT_MATCH
 
         if len(literals) > IN_PREDICATE_LIMIT:
-            return ROWS_MIGHT_MATCH
+            literals = {min(literals), max(literals)}
 
         lower = _from_byte_buffer(term.ref().field.field_type, field.lower_bound)
 
@@ -1388,23 +1388,26 @@ 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}")
 
+        # only compare the smallest and largest values if the number of values is too big
+        above_limit = len(literals) > IN_PREDICATE_LIMIT
+
         lower_bound_bytes = self.lower_bounds.get(field_id)
         if lower_bound_bytes is not None:
             lower_bound = from_bytes(field.field_type, lower_bound_bytes)
             if self._is_nan(lower_bound):
                 # NaN indicates unreliable bounds. See the InclusiveMetricsEvaluator docs for more.
                 return ROWS_MIGHT_MATCH
 
-            literals = {lit for lit in literals if lower_bound <= lit}  # type: ignore[operator]
-            if len(literals) == 0:
-                return ROWS_CANNOT_MATCH
+            if above_limit:
+                if max(literals) < lower_bound:  # type: ignore[operator]
+                    return ROWS_CANNOT_MATCH
+            else:
+                literals = {lit for lit in literals if lower_bound <= lit}  # type: ignore[operator]
+                if len(literals) == 0:
+                    return ROWS_CANNOT_MATCH
 
         upper_bound_bytes = self.upper_bounds.get(field_id)
         if upper_bound_bytes is not None:
@@ -1413,9 +1416,13 @@ def visit_in(self, term: BoundTerm, literals: set[L]) -> bool:
             if self._is_nan(upper_bound):
                 return ROWS_MIGHT_MATCH
 
-            literals = {lit for lit in literals if upper_bound >= lit}  # type: ignore[operator]
-            if len(literals) == 0:
-                return ROWS_CANNOT_MATCH
+            if above_limit:
+                if min(literals) > upper_bound:  # type: ignore[operator]
+                    return ROWS_CANNOT_MATCH
+            else:
+                literals = {lit for lit in literals if upper_bound >= lit}  # type: ignore[operator]
+                if len(literals) == 0:
+                    return ROWS_CANNOT_MATCH
 
         return ROWS_MIGHT_MATCH
 

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 should_read, "Should read: id range covers bounds, values between min and max are not checked"
+
+    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