PARQUET-116: Move the config object from keep method to a configure method in udp predicate

Author: Yash Datta

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

@@ -344,16 +344,16 @@ public static final class UserDefined<T extends Comparable<T>, U extends UserDef
     private final Column<T> column;
     private final Class<U> udpClass;
     private final String toString;
-    private final S o;
+    private final S udpConfig;
     private static final String INSTANTIATION_ERROR_MESSAGE =
         "Could not instantiate custom filter: %s. User defined predicates must be static classes with a default constructor.";
 
-    UserDefined(Column<T> column, Class<U> udpClass, S o) {
+    UserDefined(Column<T> column, Class<U> udpClass, S udpConfigParam) {
       this.column = checkNotNull(column, "column");
       this.udpClass = checkNotNull(udpClass, "udpClass");
       String name = getClass().getSimpleName().toLowerCase();
       this.toString = name + "(" + column.getColumnPath().toDotString() + ", " + udpClass.getName() + ")";
-      this.o = o;
+      this.udpConfig = udpConfigParam;
 
       // defensively try to instantiate the class early to make sure that it's possible
       getUserDefinedPredicate();
@@ -367,13 +367,11 @@ public Class<U> getUserDefinedPredicateClass() {
       return udpClass;
     }
 
-    public S getFilterObject() {
-      return o;
-    }
-
     public U getUserDefinedPredicate() {
       try {
-        return udpClass.newInstance();
+        U udpInstance = udpClass.newInstance();
+        udpInstance.configure(udpConfig);
+        return udpInstance;
       } catch (InstantiationException e) {
         throw new RuntimeException(String.format(INSTANTIATION_ERROR_MESSAGE, udpClass), e);
       } catch (IllegalAccessException e) {

parquet-column/src/main/java/parquet/filter2/predicate/UserDefinedPredicate.java

@@ -14,6 +14,10 @@
 // TODO: downside is that's fairly unwieldy for users
 public abstract class UserDefinedPredicate<T extends Comparable<T>, S extends Serializable> {
 
+  /*
+   * An object that can be used for filtering in the keep method 
+   */
+  protected S udpConfig;
   /**
    * A udp must have a default constructor.
    * The udp passed to {@link FilterApi} will not be serialized along with its state.
@@ -22,11 +26,19 @@ public abstract class UserDefinedPredicate<T extends Comparable<T>, S extends Se
    */
   public UserDefinedPredicate() { }
 
+  /*
+   * This method is used to set the object that is used in the keep method for filtering.
+   * Called before returning the new instance of this class.
+   */
+  public void configure(S udpConfigParam) { 
+    this.udpConfig = udpConfigParam; 
+  }
+
   /**
    * Return true to keep the record with this value, false to drop it.
    * o is a filter object that can be used for filtering the value.
    */
-  public abstract boolean keep(T value, S o);
+  public abstract boolean keep(T value);
 
   /**
    * Given information about a group of records (eg, the min and max value)

parquet-column/src/test/java/parquet/filter2/predicate/DummyUdp.java

@@ -5,7 +5,7 @@
 public class DummyUdp extends UserDefinedPredicate<Integer, Serializable> {
 
   @Override
-  public boolean keep(Integer value, Serializable o) {
+  public boolean keep(Integer value) {
     return false;
   }
 

parquet-column/src/test/java/parquet/filter2/predicate/TestSchemaCompatibilityValidator.java

@@ -53,7 +53,7 @@ public class TestSchemaCompatibilityValidator {
 
   static class LongDummyUdp extends UserDefinedPredicate<Long, Serializable> {
     @Override
-    public boolean keep(Long value, Serializable o) {
+    public boolean keep(Long value) {
       return false;
     }
 

parquet-generator/src/main/java/parquet/filter2/IncrementallyUpdatedFilterPredicateGenerator.java

@@ -222,8 +222,6 @@ private void addUdpBegin() throws IOException {
         "    ValueInspector valueInspector = null;\n" +
         "\n" +
         "    final U udp = pred.getUserDefinedPredicate();\n" +
-        "\n" +
-        "    final S o = pred.getFilterObject();\n" +
         "\n");
   }
 
@@ -232,13 +230,13 @@ private void addUdpCase(TypeInfo info, boolean invert)throws IOException {
         "      valueInspector = new ValueInspector() {\n" +
         "        @Override\n" +
         "        public void updateNull() {\n" +
-        "          setResult(" + (invert ? "!" : "") + "udp.keep(null, o));\n" +
+        "          setResult(" + (invert ? "!" : "") + "udp.keep(null));\n" +
         "        }\n" +
         "\n" +
         "        @SuppressWarnings(\"unchecked\")\n" +
         "        @Override\n" +
         "        public void update(" + info.primitiveName + " value) {\n" +
-        "          setResult(" + (invert ? "!" : "") + "udp.keep((T) (Object) value, o));\n" +
+        "          setResult(" + (invert ? "!" : "") + "udp.keep((T) (Object) value));\n" +
         "        }\n" +
         "      };\n" +
         "    }\n\n");

parquet-hadoop/src/main/java/parquet/filter2/statisticslevel/StatisticsFilter.java

@@ -1,5 +1,6 @@
 package parquet.filter2.statisticslevel;
 
+import java.io.Serializable;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -216,7 +217,7 @@ public Boolean visit(Not not) {
         "This predicate contains a not! Did you forget to run this predicate through LogicalInverseRewriter? " + not);
   }
 
-  private <T extends Comparable<T>, U extends UserDefinedPredicate<T>> Boolean visit(UserDefined<T, U> ud, boolean inverted) {
+  private <T extends Comparable<T>, U extends UserDefinedPredicate<T, S>, S extends Serializable> Boolean visit(UserDefined<T, U, S> ud, boolean inverted) {
     Column<T> filterColumn = ud.getColumn();
     ColumnChunkMetaData columnChunk = getColumnChunk(filterColumn.getColumnPath());
     U udp = ud.getUserDefinedPredicate();
@@ -232,12 +233,12 @@ private <T extends Comparable<T>, U extends UserDefinedPredicate<T>> Boolean vis
   }
 
   @Override
-  public <T extends Comparable<T>, U extends UserDefinedPredicate<T>> Boolean visit(UserDefined<T, U> ud) {
+  public <T extends Comparable<T>, U extends UserDefinedPredicate<T, S>, S extends Serializable> Boolean visit(UserDefined<T, U, S> ud) {
     return visit(ud, false);
   }
 
   @Override
-  public <T extends Comparable<T>, U extends UserDefinedPredicate<T>> Boolean visit(LogicalNotUserDefined<T, U> lnud) {
+  public <T extends Comparable<T>, U extends UserDefinedPredicate<T, S>, S extends Serializable> Boolean visit(LogicalNotUserDefined<T, U, S> lnud) {
     return visit(lnud.getUserDefined(), true);
   }
 

parquet-hadoop/src/test/java/parquet/filter2/recordlevel/TestRecordLevelFilters.java

@@ -148,7 +148,7 @@ public boolean keep(User u) {
   public static class StartWithP extends UserDefinedPredicate<Binary, Serializable> {
 
     @Override
-    public boolean keep(Binary value, Serializable o) {
+    public boolean keep(Binary value) {
       if (value == null) {
         return false;
       }
@@ -169,12 +169,12 @@ public boolean inverseCanDrop(Statistics<Binary> statistics) {
   public static class SetInFilter extends UserDefinedPredicate<Long, HashSet<Long>> {
 
     @Override
-    public boolean keep(Long value, HashSet o) {
+    public boolean keep(Long value) {
       if (value == null) {
         return false;
       }
 
-      return o.contains(value);
+      return udpConfig.contains(value);
     }
 
     @Override

parquet-hadoop/src/test/java/parquet/filter2/statisticslevel/TestStatisticsFilter.java

@@ -223,7 +223,7 @@ public void testOr() {
   public static class SevensAndEightsUdp extends UserDefinedPredicate<Integer, Serializable> {
 
     @Override
-    public boolean keep(Integer value, Serializable o) {
+    public boolean keep(Integer value) {
       throw new RuntimeException("this method should not be called");
     }
 

parquet-scala/src/test/scala/parquet/filter2/dsl/DslTest.scala

@@ -10,7 +10,7 @@ import parquet.filter2.predicate.Operators.{Or, UserDefined, DoubleColumn => JDo
 import parquet.filter2.predicate.{FilterApi, Statistics, UserDefinedPredicate}
 
 class DummyFilter extends UserDefinedPredicate[JInt, java.io.Serializable] {
-  override def keep(value: JInt, o: java.io.Serializable): Boolean = false
+  override def keep(value: JInt): Boolean = false
 
   override def canDrop(statistics: Statistics[JInt]): Boolean = false