Fix IN predicate performance

api/src/main/java/org/apache/iceberg/expressions/InclusiveMetricsEvaluator.java

@@ -46,6 +46,8 @@
  * return value of {@code eval} is false.
  */
 public class InclusiveMetricsEvaluator {
+  private static final int IN_PREDICATE_LIMIT = 200;
+
   private final Expression expr;
 
   public InclusiveMetricsEvaluator(Schema schema, Expression unbound) {
@@ -274,6 +276,11 @@ public <T> Boolean in(BoundReference<T> ref, Set<T> literalSet) {
 
       Collection<T> literals = literalSet;
 
+      if (literals.size() > IN_PREDICATE_LIMIT) {
+        // skip evaluating the predicate if the number of values is too big
+        return ROWS_MIGHT_MATCH;
+      }
+
       if (lowerBounds != null && lowerBounds.containsKey(id)) {
         T lower = Conversions.fromByteBuffer(ref.type(), lowerBounds.get(id));
         literals = literals.stream().filter(v -> ref.comparator().compare(lower, v) <= 0).collect(Collectors.toList());

api/src/main/java/org/apache/iceberg/expressions/ManifestEvaluator.java

@@ -48,6 +48,8 @@
  * return value of {@code eval} is false.
  */
 public class ManifestEvaluator {
+  private static final int IN_PREDICATE_LIMIT = 200;
+
   private final StructType struct;
   private final Expression expr;
 
@@ -252,6 +254,11 @@ public <T> Boolean in(BoundReference<T> ref, Set<T> literalSet) {
 
       Collection<T> literals = literalSet;
 
+      if (literals.size() > IN_PREDICATE_LIMIT) {
+        // skip evaluating the predicate if the number of values is too big
+        return ROWS_MIGHT_MATCH;
+      }
+
       T lower = Conversions.fromByteBuffer(ref.type(), fieldStats.lowerBound());
       literals = literals.stream().filter(v -> ref.comparator().compare(lower, v) <= 0).collect(Collectors.toList());
       if (literals.isEmpty()) { // if all values are less than lower bound, rows cannot match.

api/src/test/java/org/apache/iceberg/expressions/TestInclusiveMetricsEvaluator.java

@@ -19,13 +19,15 @@
 
 package org.apache.iceberg.expressions;
 
+import java.util.List;
 import org.apache.iceberg.AssertHelpers;
 import org.apache.iceberg.DataFile;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.TestHelpers.Row;
 import org.apache.iceberg.TestHelpers.TestDataFile;
 import org.apache.iceberg.exceptions.ValidationException;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.types.Types;
 import org.apache.iceberg.types.Types.IntegerType;
 import org.apache.iceberg.types.Types.StringType;
@@ -484,6 +486,14 @@ public void testIntegerIn() {
     shouldRead = new InclusiveMetricsEvaluator(SCHEMA,
         in("no_nulls", "abc", "def")).eval(FILE);
     Assert.assertTrue("Should read: in on no nulls column", shouldRead);
+
+    // should read as the number of elements in the in expression is too big
+    List<Integer> ids = Lists.newArrayListWithExpectedSize(400);
+    for (int id = -400; id <= 0; id++) {
+      ids.add(id);
+    }
+    shouldRead = new InclusiveMetricsEvaluator(SCHEMA, in("id", ids)).eval(FILE);
+    Assert.assertTrue("Should read: large in expression", shouldRead);
   }
 
   @Test

data/src/test/java/org/apache/iceberg/data/TestMetricsRowGroupFilter.java

@@ -22,6 +22,7 @@
 import java.io.File;
 import java.io.IOException;
 import java.io.UncheckedIOException;
+import java.util.List;
 import java.util.Locale;
 import java.util.UUID;
 import org.apache.avro.generic.GenericData.Record;
@@ -750,6 +751,22 @@ public void testSomeNullsNotEq() {
     Assert.assertTrue("Should read: notEqual on some nulls column", shouldRead);
   }
 
+  @Test
+  public void testInLimitParquet() {
+    Assume.assumeTrue(format == FileFormat.PARQUET);
+
+    boolean shouldRead = shouldRead(in("id", 1, 2));
+    Assert.assertFalse("Should not read if IN is evaluated", shouldRead);
+
+    List<Integer> ids = Lists.newArrayListWithExpectedSize(400);
+    for (int id = -400; id <= 0; id++) {
+      ids.add(id);
+    }
+
+    shouldRead = shouldRead(in("id", ids));
+    Assert.assertTrue("Should read if IN is not evaluated", shouldRead);
+  }
+
   private boolean shouldRead(Expression expression) {
     return shouldRead(expression, true);
   }

parquet/src/main/java/org/apache/iceberg/parquet/ParquetMetricsRowGroupFilter.java

@@ -47,6 +47,8 @@
 import org.apache.parquet.schema.PrimitiveType;
 
 public class ParquetMetricsRowGroupFilter {
+  private static final int IN_PREDICATE_LIMIT = 200;
+
   private final Schema schema;
   private final Expression expr;
 
@@ -373,6 +375,11 @@ public <T> Boolean in(BoundReference<T> ref, Set<T> literalSet) {
 
         Collection<T> literals = literalSet;
 
+        if (literals.size() > IN_PREDICATE_LIMIT) {
+          // skip evaluating the predicate if the number of values is too big
+          return ROWS_MIGHT_MATCH;
+        }
+
         T lower = min(colStats, id);
         literals = literals.stream().filter(v -> ref.comparator().compare(lower, v) <= 0).collect(Collectors.toList());
         if (literals.isEmpty()) {  // if all values are less than lower bound, rows cannot match.