IMPALA-8042: Assign BETWEEN selectivity for discrete-unique column

Impala frontend can not evaluate BETWEEN/NOT BETWEEN predicate directly.
It needs to transform a BetweenPredicate into a CompoundPredicate
consisting of upper bound and lower bound BinaryPredicate through
BetweenToCompoundRule.java. The BinaryPredicate can then be pushed down
or rewritten into other form by another expression rewrite rule.
However, the selectivity of BetweenPredicate or its derivatives remains
unassigned and often collapses with other unknown selectivity predicates
to have collective selectivity equals Expr.DEFAULT_SELECTIVITY (0.1).

This patch adds a narrow optimization of BetweenPredicate selectivity
when the following criteria are met:

1. The BetweenPredicate is bound to a slot reference of a single column
   of a table.
2. The column type is discrete, such as INTEGER or DATE.
3. The column stats are available.
4. The column is sufficiently unique based on available stats.
5. The BETWEEN/NOT BETWEEN predicate is in good form (lower bound value
   <= upper bound value).
6. The final calculated selectivity is less than or equal to
   Expr.DEFAULT_SELECTIVITY.

If these criteria are unmet, the Planner will revert to the old
behavior, which is letting the selectivity unassigned.

Since this patch only target BetweenPredicate over unique column, the
following query will still have the default scan selectivity (0.1):

select count(*) from tpch.customer c
where c.c_custkey >= 1234 and c.c_custkey <= 2345;

While this equivalent query written with BETWEEN predicate will have
lower scan selectivity:

select count(*) from tpch.customer c
where c.c_custkey between 1234 and 2345;

This patch calculates the BetweenPredicate selectivity during
transformation at BetweenToCompoundRule.java. The selectivity is
piggy-backed into the resulting CompoundPredicate and BinaryPredicate as
betweenSelectivity_ field, separate from the selectivity_ field.
Analyzer.getBoundPredicates() is modified to prioritize the derived
BinaryPredicate over ordinary BinaryPredicate in its return value to
prevent the derived BinaryPredicate from being eliminated by a matching
ordinary BinaryPredicate.

Testing:
- Add table functional_parquet.unique_with_nulls.
- Add FE tests in ExprCardinalityTest#testBetweenSelectivity,
  ExprCardinalityTest#testNotBetweenSelectivity, and
  PlannerTest#testScanCardinality.
- Pass core tests.

Change-Id: Ib349d97349d1ee99788645a66be1b81749684d10
Reviewed-on: http://gerrit.cloudera.org:8080/21377
Reviewed-by: Impala Public Jenkins <impala-public-jenkins@cloudera.com>
Tested-by: Impala Public Jenkins <impala-public-jenkins@cloudera.com>

Author: rizaon

fe/src/main/java/org/apache/impala/analysis/Analyzer.java

@@ -2460,6 +2460,8 @@ public boolean canEvalAntiJoinedConjunct(Expr e, List<TupleId> nodeTupleIds) {
    */
   public List<Expr> getBoundPredicates(TupleId destTid, Set<SlotId> ignoreSlots,
       boolean markAssigned) {
+    // Map that tracks BinaryPredicates that derived from the same BetweenPredicate.
+    Map<ExprId, List<BinaryPredicate>> betweenPredicates = new HashMap<>();
     List<Expr> result = new ArrayList<>();
     for (ExprId srcConjunctId: globalState_.singleTidConjuncts) {
       Expr srcConjunct = globalState_.conjuncts.get(srcConjunctId);
@@ -2610,9 +2612,31 @@ public List<Expr> getBoundPredicates(TupleId destTid, Set<SlotId> ignoreSlots,
           }
         }
 
-        // check if we already created this predicate
-        if (!result.contains(p)) result.add(p);
+        if ((p instanceof BinaryPredicate)
+            && ((BinaryPredicate) p).derivedFromBetween()) {
+          BinaryPredicate b = (BinaryPredicate) p;
+          betweenPredicates.computeIfAbsent(b.getBetweenExprId(), k -> new ArrayList<>());
+          betweenPredicates.get(b.getBetweenExprId()).add(b);
+        } else {
+          // check if we already created this predicate
+          if (!result.contains(p)) result.add(p);
+        }
+      }
+    }
+
+    if (!betweenPredicates.isEmpty()) {
+      // Prioritize members of 'betweenPredicates' ahead of 'result'.
+      // BinaryPredicates that derived from BetweenPredicates may have lower selectivity
+      // estimate from BetweenToCompoundRule. Placing them in-front will ensure that
+      // they are retained over other matching BinaryPredicates that do not come from
+      // BetweenPredicates when passed through Expr.removeDuplicates().
+      List<Expr> prioritizedExprs = new ArrayList<>();
+      for (List<BinaryPredicate> predicates : betweenPredicates.values()) {
+        prioritizedExprs.addAll(predicates);
       }
+      prioritizedExprs.addAll(result);
+      Expr.removeDuplicates(prioritizedExprs);
+      result = prioritizedExprs;
     }
     return result;
   }

fe/src/main/java/org/apache/impala/analysis/BinaryPredicate.java

@@ -46,6 +46,9 @@ public class BinaryPredicate extends Predicate {
   // true if this BinaryPredicate is inferred from slot equivalences, false otherwise.
   private boolean isInferred_ = false;
 
+  private ExprId betweenExprId_ = null;
+  private double betweenSelectivity_ = -1;
+
   public enum Operator {
     EQ("=", "eq", TComparisonOp.EQ),
     NE("!=", "ne", TComparisonOp.NE),
@@ -155,6 +158,8 @@ protected BinaryPredicate(BinaryPredicate other) {
     super(other);
     op_ = other.op_;
     isInferred_ = other.isInferred_;
+    betweenExprId_ = other.betweenExprId_;
+    betweenSelectivity_ = other.betweenSelectivity_;
   }
 
   public boolean isNullMatchingEq() { return op_ == Operator.NULL_MATCHING_EQ; }
@@ -193,6 +198,10 @@ public String debugString() {
     toStrHelper.add("op", op_).addValue(super.debugString());
     if (isAuxExpr()) toStrHelper.add("isAux", true);
     if (isInferred_) toStrHelper.add("isInferred", true);
+    if (derivedFromBetween()) {
+      toStrHelper.add("betweenExprId", betweenExprId_);
+      toStrHelper.add("betweenSelectivity", betweenSelectivity_);
+    }
     return toStrHelper.toString();
   }
 
@@ -415,4 +424,13 @@ public boolean localEquals(Expr that) {
 
   @Override
   public Expr clone() { return new BinaryPredicate(this); }
+
+  public void setBetweenSelectivity(ExprId betweenExprId, double betweenSelectivity) {
+    betweenExprId_ = betweenExprId;
+    betweenSelectivity_ = betweenSelectivity;
+  }
+
+  public boolean derivedFromBetween() { return betweenExprId_ != null; }
+  public ExprId getBetweenExprId() { return betweenExprId_; }
+  public double getBetweenSelectivity() { return betweenSelectivity_; }
 }

fe/src/main/java/org/apache/impala/analysis/CompoundPredicate.java

@@ -28,6 +28,7 @@
 import org.apache.impala.thrift.TExprNodeType;
 
 import com.google.common.base.MoreObjects;
+import com.google.common.base.MoreObjects.ToStringHelper;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.Lists;
 
@@ -54,6 +55,11 @@ public String toString() {
   }
   private final Operator op_;
 
+  // Selectivity estimate from BetweenToCompoundRule.computeBetweenSelectivity().
+  // Only set if this CompoundPredicate is a rewrite from BetweenPredicate.
+  // Otherwise, set to -1.
+  private final double betweenSelectivity_;
+
   public static void initBuiltins(Db db) {
     // AND and OR are implemented as custom exprs, so they do not have a function symbol.
     db.addBuiltin(ScalarFunction.createBuiltinOperator(
@@ -67,7 +73,13 @@ public static void initBuiltins(Db db) {
         Lists.<Type>newArrayList(Type.BOOLEAN), Type.BOOLEAN));
   }
 
-  public CompoundPredicate(Operator op, Expr e1, Expr e2) {
+  public static CompoundPredicate createFromBetweenPredicate(Operator op,
+      BinaryPredicate lower, BinaryPredicate upper, double betweenSelectivity) {
+    Preconditions.checkArgument(betweenSelectivity >= 0.0);
+    return new CompoundPredicate(op, lower, upper, betweenSelectivity);
+  }
+
+  private CompoundPredicate(Operator op, Expr e1, Expr e2, double betweenSelectivity) {
     super();
     this.op_ = op;
     Preconditions.checkNotNull(e1);
@@ -84,14 +96,18 @@ public CompoundPredicate(Operator op, Expr e1, Expr e2) {
         setHasAlwaysTrueHint(true);
       }
     }
+    betweenSelectivity_ = betweenSelectivity;
   }
 
+  public CompoundPredicate(Operator op, Expr e1, Expr e2) { this(op, e1, e2, -1); }
+
   /**
    * Copy c'tor used in clone().
    */
   protected CompoundPredicate(CompoundPredicate other) {
     super(other);
     op_ = other.op_;
+    betweenSelectivity_ = other.betweenSelectivity_;
   }
 
   public Operator getOp() { return op_; }
@@ -103,10 +119,11 @@ public boolean localEquals(Expr that) {
 
   @Override
   public String debugString() {
-    return MoreObjects.toStringHelper(this)
+    ToStringHelper helper = MoreObjects.toStringHelper(this)
         .add("op", op_)
-        .addValue(super.debugString())
-        .toString();
+        .addValue(super.debugString());
+    if (betweenSelectivity_ != -1) helper.add("betweenSelectivity", betweenSelectivity_);
+    return helper.toString();
   }
 
   @Override
@@ -163,8 +180,11 @@ protected void computeSelectivity(Analyzer analyzer) {
 
   @Deprecated
   protected void computeSelectivity() {
-    if (!getChild(0).hasSelectivity() ||
-        (children_.size() == 2 && !getChild(1).hasSelectivity())) {
+    if (betweenSelectivity_ != -1) {
+      selectivity_ = betweenSelectivity_;
+      return;
+    } else if (!getChild(0).hasSelectivity()
+        || (children_.size() == 2 && !getChild(1).hasSelectivity())) {
       // Give up if one of our children has an unknown selectivity.
       selectivity_ = -1;
       return;
@@ -238,4 +258,6 @@ public static Expr createConjunction(Expr lhs, Expr rhs) {
     if (rhs == null) return lhs;
     return new CompoundPredicate(Operator.AND, rhs, lhs);
   }
+
+  public boolean derivedFromBetween() { return betweenSelectivity_ != -1; }
 }

fe/src/main/java/org/apache/impala/analysis/Expr.java

@@ -1851,9 +1851,11 @@ public static String getExplainString(
    * Analyzes and evaluates expression to an integral value, returned as a long.
    * Throws if the expression cannot be evaluated or if the value evaluates to null.
    * The 'name' parameter is used in exception messages, e.g. "LIMIT expression
-   * evaluates to NULL".
+   * evaluates to NULL". If 'acceptDate' is true, treat Date expression as integral
+   * expression as well.
    */
-  public long evalToInteger(Analyzer analyzer, String name) throws AnalysisException {
+  public long evalToInteger(Analyzer analyzer, String name, boolean acceptDate)
+      throws AnalysisException {
     // Check for slotrefs and subqueries before analysis so we can provide a more
     // helpful error message.
     if (contains(SlotRef.class) || contains(Subquery.class)) {
@@ -1865,7 +1867,7 @@ public long evalToInteger(Analyzer analyzer, String name) throws AnalysisExcepti
       throw new AnalysisException(name + " expression must be a constant expression: " +
           toSql());
     }
-    if (!getType().isIntegerType()) {
+    if (!getType().isIntegerType() && !(acceptDate && getType().isDate())) {
       throw new AnalysisException(name + " expression must be an integer type but is '" +
           getType() + "': " + toSql());
     }
@@ -1875,6 +1877,16 @@ public long evalToInteger(Analyzer analyzer, String name) throws AnalysisExcepti
     } catch (InternalException e) {
       throw new AnalysisException("Failed to evaluate expr: " + toSql(), e);
     }
+
+    try {
+      return evalToInteger(val, acceptDate);
+    } catch (AnalysisException e) {
+      throw new AnalysisException(name + " expression evaluates to NULL: " + toSql());
+    }
+  }
+
+  public static long evalToInteger(TColumnValue val, boolean acceptDate)
+      throws AnalysisException {
     long value;
     if (val.isSetLong_val()) {
       value = val.getLong_val();
@@ -1884,12 +1896,18 @@ public long evalToInteger(Analyzer analyzer, String name) throws AnalysisExcepti
       value = val.getShort_val();
     } else if (val.isSetByte_val()) {
       value = val.getByte_val();
+    } else if (acceptDate && val.isSetDate_val()) {
+      value = val.getDate_val();
     } else {
-      throw new AnalysisException(name + " expression evaluates to NULL: " + toSql());
+      throw new AnalysisException("TColumnValue evaluates to NULL: " + val);
     }
     return value;
   }
 
+  public long evalToInteger(Analyzer analyzer, String name) throws AnalysisException {
+    return evalToInteger(analyzer, name, false);
+  }
+
   /**
    * Analyzes and evaluates expression to a non-negative integral value, returned as a
    * long. Throws if the expression cannot be evaluated, if the value evaluates to null,

fe/src/main/java/org/apache/impala/catalog/Type.java

@@ -234,6 +234,7 @@ public boolean isDateOrTimeType() {
         || isScalarType(PrimitiveType.TIMESTAMP);
   }
 
+  public boolean isIntegerOrDateType() { return isIntegerType() || isDate(); }
   public boolean isComplexType() { return isStructType() || isCollectionType(); }
   public boolean isCollectionType() { return isMapType() || isArrayType(); }
   public boolean isMapType() { return this instanceof MapType; }

fe/src/main/java/org/apache/impala/planner/PlanNode.java

@@ -21,18 +21,22 @@
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
-import java.util.Iterator;
+import java.util.HashMap;
 import java.util.HashSet;
+import java.util.Iterator;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 import java.util.Stack;
 
 import org.apache.impala.analysis.Analyzer;
 import org.apache.impala.analysis.BinaryPredicate;
+import org.apache.impala.analysis.CompoundPredicate;
 import org.apache.impala.analysis.Expr;
 import org.apache.impala.analysis.ExprId;
 import org.apache.impala.analysis.ExprSubstitutionMap;
 import org.apache.impala.analysis.SlotDescriptor;
+import org.apache.impala.analysis.SlotId;
 import org.apache.impala.analysis.SlotRef;
 import org.apache.impala.analysis.ToSqlOptions;
 import org.apache.impala.analysis.TupleDescriptor;
@@ -751,19 +755,59 @@ protected void computeMemLayout(Analyzer analyzer) {
    * 1. The individual selectivities of conjuncts may be unknown.
    * 2. Two selectivities, whether known or unknown, could be correlated. Assuming
    *    independence can lead to significant underestimation.
+   * 3. Two BinaryPredicate may be derived from single BetweenPredicate that have
+   *    lower selectivity if analyzed as a pair.
    *
    * The first issue is addressed by using a single default selectivity that is
    * representative of all conjuncts with unknown selectivities.
    * The second issue is addressed by an exponential backoff when multiplying each
    * additional selectivity into the final result.
+   * The third issue is addressed by checking BinaryPredicate.derivedFromBetween()
+   * property. If it is True, calculate the expression selectivity from
+   * BinaryPredicate.getBetweenSelectivity(). Both of the BinaryPredicates should have
+   * the same getBetweenSelectivity() value, thus it is OK to inspect just from one of
+   * them.
+   * TODO: Fix the third issue with more general solution when there are multiple range
+   * predicates involved, even if some of them are not derived from BetweenPredicate.
    */
   static protected double computeCombinedSelectivity(List<Expr> conjuncts) {
     // Collect all estimated selectivities.
     List<Double> selectivities = new ArrayList<>();
+    // Map between a slot id and the lowest between selectivity targeting that slot.
+    Map<SlotId, Double> perSlotBetweenSelectivities = new HashMap<>();
+
+    int conjunctSize = 0;
     for (Expr e: conjuncts) {
-      if (e.hasSelectivity()) selectivities.add(e.getSelectivity());
+      if (e instanceof BinaryPredicate && ((BinaryPredicate) e).derivedFromBetween()) {
+        // This is one of two BinaryPredicate that derived from a BetweenPredicate.
+        // A pair of BetweenPredicate must have been assigned and pushed down to the same
+        // 'targetSlotId', or one of them maybe removed by Expr.removeDuplicates()
+        // somewhere by Analyzer. But it is enough to get the selectivity from one of
+        // them. Analyzer.getBoundPredicates() ensure that BinaryPredicate derived from
+        // BetweenPredicate gets priority over regular BinaryPredicate.
+        // If multiple BetweenPredicate target the same 'targetSlotId', pick the least
+        // selectivity.
+        BinaryPredicate pred = (BinaryPredicate) e;
+        insertBetweenSelectivity(pred, perSlotBetweenSelectivities);
+      } else if (e instanceof CompoundPredicate
+          && ((CompoundPredicate) e).derivedFromBetween()
+          && (((CompoundPredicate) e).getChild(0) instanceof BinaryPredicate)) {
+        BinaryPredicate pred = (BinaryPredicate) ((CompoundPredicate) e).getChild(0);
+        insertBetweenSelectivity(pred, perSlotBetweenSelectivities);
+      } else {
+        // For everything else.
+        if (e.hasSelectivity()) selectivities.add(e.getSelectivity());
+        conjunctSize++;
+      }
     }
-    if (selectivities.size() != conjuncts.size()) {
+
+    // Add all values from perSlotBetweenSelectivities to selectivities.
+    if (!perSlotBetweenSelectivities.isEmpty()) {
+      selectivities.addAll(perSlotBetweenSelectivities.values());
+      conjunctSize += perSlotBetweenSelectivities.size();
+    }
+
+    if (selectivities.size() != conjunctSize) {
       // Some conjuncts have no estimated selectivity. Use a single default
       // representative selectivity for all those conjuncts.
       selectivities.add(Expr.DEFAULT_SELECTIVITY);
@@ -781,6 +825,15 @@ static protected double computeCombinedSelectivity(List<Expr> conjuncts) {
     return Math.max(0.0, Math.min(1.0, result));
   }
 
+  static private void insertBetweenSelectivity(
+      BinaryPredicate pred, Map<SlotId, Double> perSlotBetweenSelectivities) {
+    Preconditions.checkNotNull(pred.getBoundSlot());
+    Preconditions.checkState(pred.getBetweenSelectivity() >= 0.0);
+    SlotId targetSlotId = pred.getBoundSlot().getSlotId();
+    double sel = pred.getBetweenSelectivity();
+    perSlotBetweenSelectivities.merge(targetSlotId, sel, (v1, v2) -> Math.min(v1, v2));
+  }
+
   protected double computeSelectivity() {
     return computeCombinedSelectivity(conjuncts_);
   }