[PARQUET-1968] FilterApi support In predicate

parquet-column/src/main/java/org/apache/parquet/column/MinMax.java

@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.parquet.column;
+
+import org.apache.parquet.schema.PrimitiveComparator;
+
+/**
+ * This class calculates the max and min values of an iterable collection.
+ */
+public final class MinMax<T> {
+  private T min = null;
+  private T max = null;
+
+  public MinMax(PrimitiveComparator<T> comparator, Iterable<T> iterable) {
+    getMinAndMax(comparator, iterable);
+  }
+
+  public T getMin() {
+    return min;
+  }
+
+  public T getMax() {
+    return max;
+  }
+
+  private void getMinAndMax(PrimitiveComparator<T> comparator, Iterable<T> iterable) {
+    iterable.forEach(element -> {
+      if (max == null) {
+        max = element;
+      } else if (element != null && comparator.compare(max, element) < 0) {
+          max = element;
+      }
+      if (min == null) {
+        min = element;
+      } else if (element != null && comparator.compare(min, element) > 0) {
+          min = element;
+      }
+    });
+  }
+}

parquet-column/src/main/java/org/apache/parquet/filter2/predicate/FilterApi.java

@@ -19,6 +19,7 @@
 package org.apache.parquet.filter2.predicate;
 
 import java.io.Serializable;
+import java.util.Set;
 
 import org.apache.parquet.hadoop.metadata.ColumnPath;
 import org.apache.parquet.filter2.predicate.Operators.And;
@@ -30,12 +31,14 @@
 import org.apache.parquet.filter2.predicate.Operators.FloatColumn;
 import org.apache.parquet.filter2.predicate.Operators.Gt;
 import org.apache.parquet.filter2.predicate.Operators.GtEq;
+import org.apache.parquet.filter2.predicate.Operators.In;
 import org.apache.parquet.filter2.predicate.Operators.IntColumn;
 import org.apache.parquet.filter2.predicate.Operators.LongColumn;
 import org.apache.parquet.filter2.predicate.Operators.Lt;
 import org.apache.parquet.filter2.predicate.Operators.LtEq;
 import org.apache.parquet.filter2.predicate.Operators.Not;
 import org.apache.parquet.filter2.predicate.Operators.NotEq;
+import org.apache.parquet.filter2.predicate.Operators.NotIn;
 import org.apache.parquet.filter2.predicate.Operators.Or;
 import org.apache.parquet.filter2.predicate.Operators.SupportsEqNotEq;
 import org.apache.parquet.filter2.predicate.Operators.SupportsLtGt;
@@ -204,6 +207,56 @@ public static <T extends Comparable<T>, C extends Column<T> & SupportsLtGt> GtEq
     return new GtEq<>(column, value);
   }
 
+  /**
+   * Keeps records if their value is in the provided values.
+   * The provided values set could not be null, but could contains a null value.
+   * <p>
+   * For example:
+   * <pre>
+   *   {@code
+   *   Set<Integer> set = new HashSet<>();
+   *   set.add(9);
+   *   set.add(null);
+   *   set.add(50);
+   *   in(column, set);}
+   * </pre>
+   * will keep all records whose values are 9, null, or 50.
+   *
+   * @param column a column reference created by FilterApi
+   * @param values a set of values that match the column's type
+   * @param <T> the Java type of values in the column
+   * @param <C> the column type that corresponds to values of type T
+   * @return an in predicate for the given column and value
+   */
+  public static <T extends Comparable<T>, C extends Column<T> & SupportsEqNotEq> In<T> in(C column, Set<T> values) {
+    return new In<>(column, values);
+  }
+
+  /**
+   * Keeps records if their value is not in the provided values.
+   * The provided values set could not be null, but could contains a null value.
+   * <p>
+   * For example:
+   * <pre>
+   *   {@code
+   *   Set<Integer> set = new HashSet<>();
+   *   set.add(9);
+   *   set.add(null);
+   *   set.add(50);
+   *   notIn(column, set);}
+   * </pre>
+   * will keep all records whose values are not 9, null, and 50.
+   *
+   * @param column a column reference created by FilterApi
+   * @param values a set of values that match the column's type
+   * @param <T> the Java type of values in the column
+   * @param <C> the column type that corresponds to values of type T
+   * @return an notIn predicate for the given column and value
+   */
+  public static <T extends Comparable<T>, C extends Column<T> & SupportsEqNotEq> NotIn<T> notIn(C column, Set<T> values) {
+    return new NotIn<>(column, values);
+  }
+
   /**
    * Keeps records that pass the provided {@link UserDefinedPredicate}
    * <p>

parquet-column/src/main/java/org/apache/parquet/filter2/predicate/FilterPredicate.java

@@ -22,11 +22,13 @@
 import org.apache.parquet.filter2.predicate.Operators.Eq;
 import org.apache.parquet.filter2.predicate.Operators.Gt;
 import org.apache.parquet.filter2.predicate.Operators.GtEq;
+import org.apache.parquet.filter2.predicate.Operators.In;
 import org.apache.parquet.filter2.predicate.Operators.LogicalNotUserDefined;
 import org.apache.parquet.filter2.predicate.Operators.Lt;
 import org.apache.parquet.filter2.predicate.Operators.LtEq;
 import org.apache.parquet.filter2.predicate.Operators.Not;
 import org.apache.parquet.filter2.predicate.Operators.NotEq;
+import org.apache.parquet.filter2.predicate.Operators.NotIn;
 import org.apache.parquet.filter2.predicate.Operators.Or;
 import org.apache.parquet.filter2.predicate.Operators.UserDefined;
 
@@ -66,6 +68,12 @@ public static interface Visitor<R> {
     <T extends Comparable<T>> R visit(LtEq<T> ltEq);
     <T extends Comparable<T>> R visit(Gt<T> gt);
     <T extends Comparable<T>> R visit(GtEq<T> gtEq);
+    default <T extends Comparable<T>> R visit(In<T> in) {
+      throw new UnsupportedOperationException("visit in is not supported.");
+    }
+    default <T extends Comparable<T>> R visit(NotIn<T> notIn) {
+      throw new UnsupportedOperationException("visit NotIn is not supported.");
+    }
     R visit(And and);
     R visit(Or or);
     R visit(Not not);

parquet-column/src/main/java/org/apache/parquet/filter2/predicate/LogicalInverseRewriter.java

@@ -23,11 +23,13 @@
 import org.apache.parquet.filter2.predicate.Operators.Eq;
 import org.apache.parquet.filter2.predicate.Operators.Gt;
 import org.apache.parquet.filter2.predicate.Operators.GtEq;
+import org.apache.parquet.filter2.predicate.Operators.In;
 import org.apache.parquet.filter2.predicate.Operators.LogicalNotUserDefined;
 import org.apache.parquet.filter2.predicate.Operators.Lt;
 import org.apache.parquet.filter2.predicate.Operators.LtEq;
 import org.apache.parquet.filter2.predicate.Operators.Not;
 import org.apache.parquet.filter2.predicate.Operators.NotEq;
+import org.apache.parquet.filter2.predicate.Operators.NotIn;
 import org.apache.parquet.filter2.predicate.Operators.Or;
 import org.apache.parquet.filter2.predicate.Operators.UserDefined;
 
@@ -87,6 +89,16 @@ public <T extends Comparable<T>> FilterPredicate visit(GtEq<T> gtEq) {
     return gtEq;
   }
 
+  @Override
+  public <T extends Comparable<T>> FilterPredicate visit(In<T> in) {
+    return in;
+  }
+
+  @Override
+  public <T extends Comparable<T>> FilterPredicate visit(NotIn<T> notIn) {
+    return notIn;
+  }
+
   @Override
   public FilterPredicate visit(And and) {
     return and(and.getLeft().accept(this), and.getRight().accept(this));

parquet-column/src/main/java/org/apache/parquet/filter2/predicate/LogicalInverter.java

@@ -23,11 +23,13 @@
 import org.apache.parquet.filter2.predicate.Operators.Eq;
 import org.apache.parquet.filter2.predicate.Operators.Gt;
 import org.apache.parquet.filter2.predicate.Operators.GtEq;
+import org.apache.parquet.filter2.predicate.Operators.In;
 import org.apache.parquet.filter2.predicate.Operators.LogicalNotUserDefined;
 import org.apache.parquet.filter2.predicate.Operators.Lt;
 import org.apache.parquet.filter2.predicate.Operators.LtEq;
 import org.apache.parquet.filter2.predicate.Operators.Not;
 import org.apache.parquet.filter2.predicate.Operators.NotEq;
+import org.apache.parquet.filter2.predicate.Operators.NotIn;
 import org.apache.parquet.filter2.predicate.Operators.Or;
 import org.apache.parquet.filter2.predicate.Operators.UserDefined;
 
@@ -81,6 +83,16 @@ public <T extends Comparable<T>> FilterPredicate visit(GtEq<T> gtEq) {
     return new Lt<>(gtEq.getColumn(), gtEq.getValue());
   }
 
+  @Override
+  public <T extends Comparable<T>> FilterPredicate visit(In<T> in) {
+    return new NotIn<>(in.getColumn(), in.getValues());
+  }
+
+  @Override
+  public <T extends Comparable<T>> FilterPredicate visit(NotIn<T> notIn) {
+    return new In<>(notIn.getColumn(), notIn.getValues());
+  }
+
   @Override
   public FilterPredicate visit(And and) {
     return new Or(and.getLeft().accept(this), and.getRight().accept(this));

parquet-column/src/main/java/org/apache/parquet/filter2/predicate/Operators.java

@@ -21,10 +21,13 @@
 import java.io.Serializable;
 import java.util.Locale;
 import java.util.Objects;
+import java.util.Set;
 
 import org.apache.parquet.hadoop.metadata.ColumnPath;
 import org.apache.parquet.io.api.Binary;
 
+import static org.apache.parquet.Preconditions.checkArgument;
+
 /**
  * These are the operators in a filter predicate expression tree.
  * They are constructed by using the methods in {@link FilterApi}
@@ -169,7 +172,7 @@ public int hashCode() {
   public static final class Eq<T extends Comparable<T>> extends ColumnFilterPredicate<T> {
 
     // value can be null
-    Eq(Column<T> column, T value) {
+    public Eq(Column<T> column, T value) {
       super(column, value);
     }
 
@@ -247,6 +250,82 @@ public <R> R accept(Visitor<R> visitor) {
     }
   }
 
+  /**
+   * Base class for {@link In} and {@link NotIn}. {@link In} is used to filter data based on a list of values.
+   * {@link NotIn} is used to filter data that are not in the list of values.
+   */
+  public static abstract class SetColumnFilterPredicate<T extends Comparable<T>> implements FilterPredicate, Serializable {
+    private final Column<T> column;
+    private final Set<T> values;
+
+    protected SetColumnFilterPredicate(Column<T> column, Set<T> values) {
+      this.column = Objects.requireNonNull(column, "column cannot be null");
+      this.values = Objects.requireNonNull(values, "values cannot be null");
+      checkArgument(!values.isEmpty(), "values in SetColumnFilterPredicate shouldn't be empty!");
+    }
+
+    public Column<T> getColumn() {
+      return column;
+    }
+
+    public Set<T> getValues() {
+      return values;
+    }
+
+    @Override
+    public String toString() {
+      String name = getClass().getSimpleName().toLowerCase(Locale.ENGLISH);
+      StringBuilder str = new StringBuilder();
+      str.append(name).append("(").append(column.getColumnPath().toDotString()).append(", ");
+      int iter = 0;
+      for (T value : values) {
+        if (iter >= 100) break;
+        str.append(value).append(", ");
+        iter++;
+      }
+      int length = str.length();
+      str = values.size() <= 100 ? str.delete(length - 2, length) : str.append("...");
+      return str.append(")").toString();
+    }
+
+    @Override
+    public boolean equals(Object o) {
+      if (this == o) return true;
+      if (o == null || getClass() != o.getClass()) return false;
+      SetColumnFilterPredicate<?> that = (SetColumnFilterPredicate<?>) o;
+      return column.equals(that.column) && values.equals(that.values);
+    }
+
+    @Override
+    public int hashCode() {
+      return Objects.hash(column, values);
+    }
+  }
+
+  public static final class In<T extends Comparable<T>> extends SetColumnFilterPredicate<T> {
+
+    public In(Column<T> column, Set<T> values) {
+      super(column, values);
+    }
+
+    @Override
+    public <R> R accept(Visitor<R> visitor) {
+      return visitor.visit(this);
+    }
+  }
+
+  public static final class NotIn<T extends Comparable<T>> extends SetColumnFilterPredicate<T> {
+
+    NotIn(Column<T> column, Set<T> values) {
+      super(column, values);
+    }
+
+    @Override
+    public <R> R accept(Visitor<R> visitor) {
+      return visitor.visit(this);
+    }
+  }
+
   // base class for And, Or
   private static abstract class BinaryLogicalFilterPredicate implements FilterPredicate, Serializable {
     private final FilterPredicate left;

parquet-column/src/main/java/org/apache/parquet/filter2/predicate/SchemaCompatibilityValidator.java

@@ -29,12 +29,15 @@
 import org.apache.parquet.filter2.predicate.Operators.Eq;
 import org.apache.parquet.filter2.predicate.Operators.Gt;
 import org.apache.parquet.filter2.predicate.Operators.GtEq;
+import org.apache.parquet.filter2.predicate.Operators.In;
 import org.apache.parquet.filter2.predicate.Operators.LogicalNotUserDefined;
 import org.apache.parquet.filter2.predicate.Operators.Lt;
 import org.apache.parquet.filter2.predicate.Operators.LtEq;
 import org.apache.parquet.filter2.predicate.Operators.Not;
 import org.apache.parquet.filter2.predicate.Operators.NotEq;
+import org.apache.parquet.filter2.predicate.Operators.NotIn;
 import org.apache.parquet.filter2.predicate.Operators.Or;
+import org.apache.parquet.filter2.predicate.Operators.SetColumnFilterPredicate;
 import org.apache.parquet.filter2.predicate.Operators.UserDefined;
 import org.apache.parquet.hadoop.metadata.ColumnPath;
 import org.apache.parquet.schema.MessageType;
@@ -114,6 +117,18 @@ public <T extends Comparable<T>> Void visit(GtEq<T> pred) {
     return null;
   }
 
+  @Override
+  public <T extends Comparable<T>> Void visit(In<T> pred) {
+    validateColumnFilterPredicate(pred);
+    return null;
+  }
+
+  @Override
+  public <T extends Comparable<T>> Void visit(NotIn<T> pred) {
+    validateColumnFilterPredicate(pred);
+    return null;
+  }
+
   @Override
   public Void visit(And and) {
     and.getLeft().accept(this);
@@ -149,6 +164,10 @@ private <T extends Comparable<T>> void validateColumnFilterPredicate(ColumnFilte
     validateColumn(pred.getColumn());
   }
 
+  private <T extends Comparable<T>> void validateColumnFilterPredicate(SetColumnFilterPredicate<T> pred) {
+    validateColumn(pred.getColumn());
+  }
+
   private <T extends Comparable<T>> void validateColumn(Column<T> column) {
     ColumnPath path = column.getColumnPath();