DRILL-6259: Support parquet filter push down for complex types

close #1173

exec/java-exec/src/main/java/org/apache/drill/exec/compile/sig/ConstantExpressionIdentifier.java

@@ -31,6 +31,7 @@
 import org.apache.drill.common.expression.LogicalExpression;
 import org.apache.drill.common.expression.NullExpression;
 import org.apache.drill.common.expression.SchemaPath;
+import org.apache.drill.common.expression.TypedFieldExpr;
 import org.apache.drill.common.expression.TypedNullConstant;
 import org.apache.drill.common.expression.ValueExpressions;
 import org.apache.drill.common.expression.ValueExpressions.BooleanExpression;
@@ -231,4 +232,9 @@ public Boolean visitConvertExpression(ConvertExpression e,
   public Boolean visitParameter(ValueExpressions.ParameterExpression e, IdentityHashMap<LogicalExpression, Object> value) throws RuntimeException {
     return false;
   }
+
+  @Override
+  public Boolean visitTypedFieldExpr(TypedFieldExpr e, IdentityHashMap<LogicalExpression, Object> value) throws RuntimeException {
+    return false;
+  }
 }

exec/java-exec/src/main/java/org/apache/drill/exec/expr/ExpressionTreeMaterializer.java

@@ -44,6 +44,7 @@
 import org.apache.drill.common.expression.LogicalExpression;
 import org.apache.drill.common.expression.NullExpression;
 import org.apache.drill.common.expression.SchemaPath;
+import org.apache.drill.common.expression.TypedFieldExpr;
 import org.apache.drill.common.expression.TypedNullConstant;
 import org.apache.drill.common.expression.ValueExpressions;
 import org.apache.drill.common.expression.ValueExpressions.BooleanExpression;
@@ -78,7 +79,6 @@
 import org.apache.drill.exec.expr.fn.DrillFuncHolder;
 import org.apache.drill.exec.expr.fn.ExceptionFunction;
 import org.apache.drill.exec.expr.fn.FunctionLookupContext;
-import org.apache.drill.exec.expr.stat.TypedFieldExpr;
 import org.apache.drill.exec.record.TypedFieldId;
 import org.apache.drill.exec.record.VectorAccessible;
 import org.apache.drill.exec.resolver.FunctionResolver;
@@ -323,7 +323,6 @@ public LogicalExpression visitSchemaPath(SchemaPath path, FunctionLookupContext
       } else {
         logger.warn("Unable to find value vector of path {}, returning null-int instance.", path);
         return new TypedFieldExpr(path, Types.OPTIONAL_INT);
-        // return NullExpression.INSTANCE;
       }
     }
   }

exec/java-exec/src/main/java/org/apache/drill/exec/expr/stat/ParquetIsPredicates.java

@@ -18,6 +18,8 @@
 
 import org.apache.drill.common.expression.LogicalExpression;
 import org.apache.drill.common.expression.LogicalExpressionBase;
+import org.apache.drill.common.expression.SchemaPath;
+import org.apache.drill.common.expression.TypedFieldExpr;
 import org.apache.drill.common.expression.visitors.ExprVisitor;
 import org.apache.parquet.column.statistics.Statistics;
 
@@ -29,6 +31,7 @@
  * IS predicates for parquet filter pushdown.
  */
 public class ParquetIsPredicates {
+
   public static abstract class ParquetIsPredicate extends LogicalExpressionBase implements ParquetFilterPredicate {
     protected final LogicalExpression expr;
 
@@ -54,12 +57,22 @@ public <T, V, E extends Exception> T accept(ExprVisitor<T, V, E> visitor, V valu
    * IS NULL predicate.
    */
   public static class IsNullPredicate extends ParquetIsPredicate {
+    private final boolean isArray;
+
     public IsNullPredicate(LogicalExpression expr) {
       super(expr);
+      this.isArray = isArray(expr);
     }
 
     @Override
     public boolean canDrop(RangeExprEvaluator evaluator) {
+
+      // for arrays we are not able to define exact number of nulls
+      // [1,2,3] vs [1,2] -> in second case 3 is absent and thus it's null but statistics shows no nulls
+      if (isArray) {
+        return false;
+      }
+
       Statistics exprStat = expr.accept(evaluator, null);
 
       if (!ParquetPredicatesHelper.hasStats(exprStat)) {
@@ -73,6 +86,16 @@ public boolean canDrop(RangeExprEvaluator evaluator) {
         return false;
       }
     }
+
+    private boolean isArray(LogicalExpression expression) {
+      if (expression instanceof TypedFieldExpr) {
+        TypedFieldExpr typedFieldExpr = (TypedFieldExpr) expression;
+        SchemaPath schemaPath = typedFieldExpr.getPath();
+        return schemaPath.isArray();
+      }
+      return false;
+    }
+
   }
 
   /**

exec/java-exec/src/main/java/org/apache/drill/exec/expr/stat/RangeExprEvaluator.java

@@ -17,11 +17,11 @@
  */
 package org.apache.drill.exec.expr.stat;
 
-import com.google.common.base.Preconditions;
 import org.apache.drill.common.exceptions.DrillRuntimeException;
 import org.apache.drill.common.expression.FunctionHolderExpression;
 import org.apache.drill.common.expression.LogicalExpression;
 import org.apache.drill.common.expression.SchemaPath;
+import org.apache.drill.common.expression.TypedFieldExpr;
 import org.apache.drill.common.expression.ValueExpressions;
 import org.apache.drill.common.expression.fn.CastFunctions;
 import org.apache.drill.common.expression.fn.FuncHolder;
@@ -70,17 +70,20 @@ public long getRowCount() {
 
   @Override
   public Statistics visitUnknown(LogicalExpression e, Void value) throws RuntimeException {
-    if (e instanceof TypedFieldExpr) {
-      TypedFieldExpr fieldExpr = (TypedFieldExpr) e;
-      final ColumnStatistics columnStatistics = columnStatMap.get(fieldExpr.getPath());
-      if (columnStatistics != null) {
-        return columnStatistics.getStatistics();
-      } else if (fieldExpr.getMajorType().equals(Types.OPTIONAL_INT)) {
-        // field does not exist.
-        IntStatistics intStatistics = new IntStatistics();
-        intStatistics.setNumNulls(rowCount); // all values are nulls
-        return intStatistics;
-      }
+    // do nothing for the unknown expression
+    return null;
+  }
+
+  @Override
+  public Statistics visitTypedFieldExpr(TypedFieldExpr typedFieldExpr, Void value) throws RuntimeException {
+    final ColumnStatistics columnStatistics = columnStatMap.get(typedFieldExpr.getPath());
+    if (columnStatistics != null) {
+      return columnStatistics.getStatistics();
+    } else if (typedFieldExpr.getMajorType().equals(Types.OPTIONAL_INT)) {
+      // field does not exist.
+      IntStatistics intStatistics = new IntStatistics();
+      intStatistics.setNumNulls(rowCount); // all values are nulls
+      return intStatistics;
     }
     return null;
   }

exec/java-exec/src/main/java/org/apache/drill/exec/planner/common/DrillRelOptUtil.java

@@ -18,6 +18,7 @@
 package org.apache.drill.exec.planner.common;
 
 import java.util.AbstractList;
+import java.util.Collection;
 import java.util.List;
 
 import com.google.common.collect.Lists;
@@ -178,22 +179,20 @@ private static boolean containIdentity(List<? extends RexNode> exps,
   }
 
   /**
-   * Travesal RexNode to find the item/flattern operator. Continue search if RexNode has a
+   * Travesal RexNode to find at least one operator in the given collection. Continue search if RexNode has a
    * RexInputRef which refers to a RexNode in project expressions.
    *
    * @param node : RexNode to search
    * @param projExprs : the list of project expressions. Empty list means there is No project operator underneath.
+   * @param operators collection of operators to find
    * @return : Return null if there is NONE; return the first appearance of item/flatten RexCall.
    */
-  public static RexCall findItemOrFlatten(
-      final RexNode node,
-      final List<RexNode> projExprs) {
+  public static RexCall findOperators(final RexNode node, final List<RexNode> projExprs, final Collection<String> operators) {
     try {
       RexVisitor<Void> visitor =
           new RexVisitorImpl<Void>(true) {
             public Void visitCall(RexCall call) {
-              if ("item".equals(call.getOperator().getName().toLowerCase()) ||
-                  "flatten".equals(call.getOperator().getName().toLowerCase())) {
+              if (operators.contains(call.getOperator().getName().toLowerCase())) {
                 throw new Util.FoundOne(call); /* throw exception to interrupt tree walk (this is similar to
                                               other utility methods in RexUtil.java */
               }
@@ -208,8 +207,7 @@ public Void visitInputRef(RexInputRef inputRef) {
                 RexNode n = projExprs.get(index);
                 if (n instanceof RexCall) {
                   RexCall r = (RexCall) n;
-                  if ("item".equals(r.getOperator().getName().toLowerCase()) ||
-                      "flatten".equals(r.getOperator().getName().toLowerCase())) {
+                  if (operators.contains(r.getOperator().getName().toLowerCase())) {
                     throw new Util.FoundOne(r);
                   }
                 }

exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillPushFilterPastProjectRule.java

@@ -30,13 +30,23 @@
 import org.apache.calcite.util.Pair;
 import org.apache.drill.exec.planner.common.DrillRelOptUtil;
 
+import java.util.ArrayList;
+import java.util.Collection;
 import java.util.List;
 
 public class DrillPushFilterPastProjectRule extends RelOptRule {
 
   public final static RelOptRule INSTANCE = new DrillPushFilterPastProjectRule();
 
-  protected DrillPushFilterPastProjectRule() {
+  private static final Collection<String> BANNED_OPERATORS;
+
+  static {
+    BANNED_OPERATORS = new ArrayList<>(2);
+    BANNED_OPERATORS.add("flatten");
+    BANNED_OPERATORS.add("item");
+  }
+
+  private DrillPushFilterPastProjectRule() {
     super(
         operand(
             LogicalFilter.class,
@@ -60,7 +70,7 @@ public void onMatch(RelOptRuleCall call) {
 
 
     for (final RexNode pred : predList) {
-      if (DrillRelOptUtil.findItemOrFlatten(pred, projRel.getProjects()) == null) {
+      if (DrillRelOptUtil.findOperators(pred, projRel.getProjects(), BANNED_OPERATORS) == null) {
         qualifiedPredList.add(pred);
       } else {
         unqualifiedPredList.add(pred);

exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetFilterBuilder.java

@@ -19,8 +19,7 @@
 import org.apache.drill.common.expression.BooleanOperator;
 import org.apache.drill.common.expression.FunctionHolderExpression;
 import org.apache.drill.common.expression.LogicalExpression;
-import org.apache.drill.common.expression.PathSegment;
-import org.apache.drill.common.expression.SchemaPath;
+import org.apache.drill.common.expression.TypedFieldExpr;
 import org.apache.drill.common.expression.ValueExpressions;
 import org.apache.drill.common.expression.fn.CastFunctions;
 import org.apache.drill.common.expression.fn.FuncHolder;
@@ -41,7 +40,6 @@
 import org.apache.drill.exec.expr.stat.ParquetBooleanPredicates;
 import org.apache.drill.exec.expr.stat.ParquetComparisonPredicates;
 import org.apache.drill.exec.expr.stat.ParquetIsPredicates;
-import org.apache.drill.exec.expr.stat.TypedFieldExpr;
 import org.apache.drill.exec.ops.UdfUtilities;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -62,11 +60,12 @@ public class ParquetFilterBuilder extends AbstractExprVisitor<LogicalExpression,
   /**
    * @param expr materialized filter expression
    * @param constantBoundaries set of constant expressions
-   * @param udfUtilities
+   * @param udfUtilities udf utilities
+   *
+   * @return logical expression
    */
   public static LogicalExpression buildParquetFilterPredicate(LogicalExpression expr, final Set<LogicalExpression> constantBoundaries, UdfUtilities udfUtilities) {
-    final LogicalExpression predicate = expr.accept(new ParquetFilterBuilder(udfUtilities), constantBoundaries);
-    return predicate;
+    return expr.accept(new ParquetFilterBuilder(udfUtilities), constantBoundaries);
   }
 
   private ParquetFilterBuilder(UdfUtilities udfUtilities) {
@@ -75,18 +74,15 @@ private ParquetFilterBuilder(UdfUtilities udfUtilities) {
 
   @Override
   public LogicalExpression visitUnknown(LogicalExpression e, Set<LogicalExpression> value) {
-    if (e instanceof TypedFieldExpr &&
-        ! containsArraySeg(((TypedFieldExpr) e).getPath()) &&
-        e.getMajorType().getMode() != TypeProtos.DataMode.REPEATED) {
-      // A filter is not qualified for push down, if
-      // 1. it contains an array segment : a.b[1], a.b[1].c.d
-      // 2. it's repeated type.
-      return e;
-    }
-
+    // for the unknown expression, do nothing
     return null;
   }
 
+  @Override
+  public LogicalExpression visitTypedFieldExpr(TypedFieldExpr typedFieldExpr, Set<LogicalExpression> value) throws RuntimeException {
+    return typedFieldExpr;
+  }
+
   @Override
   public LogicalExpression visitIntConstant(ValueExpressions.IntExpression intExpr, Set<LogicalExpression> value)
       throws RuntimeException {
@@ -161,18 +157,6 @@ public LogicalExpression visitBooleanOperator(BooleanOperator op, Set<LogicalExp
     }
   }
 
-  private boolean containsArraySeg(final SchemaPath schemaPath) {
-    PathSegment seg = schemaPath.getRootSegment();
-
-    while (seg != null) {
-      if (seg.isArray()) {
-        return true;
-      }
-      seg = seg.getChild();
-    }
-    return false;
-  }
-
   private LogicalExpression getValueExpressionFromConst(ValueHolder holder, TypeProtos.MinorType type) {
     switch (type) {
     case INT:
@@ -229,13 +213,9 @@ public LogicalExpression visitFunctionHolderExpression(FunctionHolderExpression
     }
 
     if (CastFunctions.isCastFunction(funcName)) {
-      List<LogicalExpression> newArgs = new ArrayList();
-      for (LogicalExpression arg : funcHolderExpr.args) {
-        final LogicalExpression newArg = arg.accept(this, value);
-        if (newArg == null) {
-          return null;
-        }
-        newArgs.add(newArg);
+      List<LogicalExpression> newArgs = generateNewExpressions(funcHolderExpr.args, value);
+      if (newArgs == null) {
+        return null;
       }
 
       return funcHolderExpr.copy(newArgs);
@@ -244,15 +224,22 @@ public LogicalExpression visitFunctionHolderExpression(FunctionHolderExpression
     }
   }
 
-  private LogicalExpression handleCompareFunction(FunctionHolderExpression functionHolderExpression, Set<LogicalExpression> value) {
-    List<LogicalExpression> newArgs = new ArrayList();
-
-    for (LogicalExpression arg : functionHolderExpression.args) {
-      LogicalExpression newArg = arg.accept(this, value);
+  private List<LogicalExpression> generateNewExpressions(List<LogicalExpression> expressions, Set<LogicalExpression> value) {
+    List<LogicalExpression> newExpressions = new ArrayList<>();
+    for (LogicalExpression arg : expressions) {
+      final LogicalExpression newArg = arg.accept(this, value);
       if (newArg == null) {
         return null;
       }
-      newArgs.add(newArg);
+      newExpressions.add(newArg);
+    }
+    return newExpressions;
+  }
+
+  private LogicalExpression handleCompareFunction(FunctionHolderExpression functionHolderExpression, Set<LogicalExpression> value) {
+    List<LogicalExpression> newArgs = generateNewExpressions(functionHolderExpression.args, value);
+    if (newArgs == null) {
+      return null;
     }
 
     String funcName = ((DrillSimpleFuncHolder) functionHolderExpression.getHolder()).getRegisteredNames()[0];
@@ -306,19 +293,6 @@ private LogicalExpression handleIsFunction(FunctionHolderExpression functionHold
     }
   }
 
-  private LogicalExpression handleCastFunction(FunctionHolderExpression functionHolderExpression, Set<LogicalExpression> value) {
-    for (LogicalExpression arg : functionHolderExpression.args) {
-      LogicalExpression newArg = arg.accept(this, value);
-      if (newArg == null) {
-        return null;
-      }
-    }
-
-    String funcName = ((DrillSimpleFuncHolder) functionHolderExpression.getHolder()).getRegisteredNames()[0];
-
-    return null;
-  }
-
   private static boolean isCompareFunction(String funcName) {
     return COMPARE_FUNCTIONS_SET.contains(funcName);
   }