HIVE-29368: More accurate pessimistic stats combining

ql/src/java/org/apache/hadoop/hive/ql/stats/StatsUtils.java

@@ -1578,9 +1578,11 @@ public static ColStatistics getColStatisticsFromExpression(HiveConf conf, Statis
             csList.add(cs);
           }
           if (csList.size() == engfd.getChildren().size()) {
-            Optional<ColStatistics> res = se.estimate(csList);
+            Optional<ColStatistics> res = se.estimate(csList, parentStats);
             if (res.isPresent()) {
               ColStatistics newStats = res.get();
+              // NDV cannot exceed numRows
+              newStats.setCountDistint(Math.min(newStats.getCountDistint(), numRows));
               colType = colType.toLowerCase();
               newStats.setColumnType(colType);
               newStats.setColumnName(colName);
@@ -2109,7 +2111,10 @@ private static List<Long> extractNDVGroupingColumns(List<ColStatistics> colStats
     for (ColStatistics cs : colStats) {
       if (cs != null) {
         long ndv = cs.getCountDistint();
-        if (cs.getNumNulls() > 0) {
+        // NDV needs to be adjusted if a column has a known NDV along with NULL values
+        // or if a column happens to be "const NULL"
+        if ((ndv > 0 && cs.getNumNulls() > 0) ||
+            (ndv == 0 && !cs.isEstimated() && cs.getNumNulls() == parentStats.getNumRows())) {
           ndv = StatsUtils.safeAdd(ndv, 1);
         }
         ndvValues.add(ndv);

ql/src/java/org/apache/hadoop/hive/ql/stats/estimator/PessimisticStatCombiner.java

@@ -21,16 +21,26 @@
 import java.util.Optional;
 
 import org.apache.hadoop.hive.ql.plan.ColStatistics;
+import org.apache.hadoop.hive.ql.stats.StatsUtils;
 
 /**
  * Combines {@link ColStatistics} objects to provide the most pessimistic estimate.
  */
 public class PessimisticStatCombiner {
 
+  private final long numRows;
   private boolean inited;
+  private boolean hasUnknownNDV;
   private ColStatistics result;
 
+  public PessimisticStatCombiner(long numRows) {
+    this.numRows = numRows;
+  }
+
   public void add(ColStatistics stat) {
+    // NDV==0 means unknown, unless it's a NULL constant (numNulls == numRows)
+    hasUnknownNDV = hasUnknownNDV || (stat.getCountDistint() == 0 && stat.getNumNulls() != numRows);
+
     if (!inited) {
       inited = true;
       result = stat.clone();
@@ -41,8 +51,10 @@ public void add(ColStatistics stat) {
     if (stat.getAvgColLen() > result.getAvgColLen()) {
       result.setAvgColLen(stat.getAvgColLen());
     }
-    if (stat.getCountDistint() > result.getCountDistint()) {
-      result.setCountDistint(stat.getCountDistint());
+    if (hasUnknownNDV) {
+      result.setCountDistint(0);
+    } else {
+      result.setCountDistint(StatsUtils.safeAdd(result.getCountDistint(), stat.getCountDistint()));
     }
     if (stat.getNumNulls() < 0 || result.getNumNulls() < 0) {
       result.setNumNulls(-1);
@@ -63,8 +75,8 @@ public void add(ColStatistics stat) {
       result.setFilterColumn();
     }
   }
+
   public Optional<ColStatistics> getResult() {
     return Optional.of(result);
-
   }
 }

ql/src/java/org/apache/hadoop/hive/ql/stats/estimator/StatEstimator.java

@@ -22,6 +22,7 @@
 import java.util.Optional;
 
 import org.apache.hadoop.hive.ql.plan.ColStatistics;
+import org.apache.hadoop.hive.ql.plan.Statistics;
 
 /**
  * Enables statistics related computation on UDFs
@@ -39,5 +40,19 @@ public interface StatEstimator {
    * @param argStats the statistics for every argument of the UDF
    * @return {@link ColStatistics} estimate for the actual UDF.
    */
-  public Optional<ColStatistics> estimate(List<ColStatistics> argStats);
+  default Optional<ColStatistics> estimate(List<ColStatistics> argStats) {
+    throw new UnsupportedOperationException("This estimator requires parentStats");
+  }
+
+  /**
+   * Computes the output statistics with access to parent statistics.
+   * Override this method when the estimator uses more info for accurate estimation.
+   *
+   * @param argStats the statistics for every argument of the UDF
+   * @param parentStats statistics from the parent operator
+   * @return {@link ColStatistics} estimate for the actual UDF.
+   */
+  default Optional<ColStatistics> estimate(List<ColStatistics> argStats, Statistics parentStats) {
+    return estimate(argStats);
+  }
 }

ql/src/java/org/apache/hadoop/hive/ql/udf/generic/GenericUDFCoalesce.java

@@ -26,6 +26,7 @@
 import org.apache.hadoop.hive.ql.exec.vector.VectorizedExpressionsSupportDecimal64;
 import org.apache.hadoop.hive.ql.metadata.HiveException;
 import org.apache.hadoop.hive.ql.plan.ColStatistics;
+import org.apache.hadoop.hive.ql.plan.Statistics;
 import org.apache.hadoop.hive.ql.stats.estimator.StatEstimator;
 import org.apache.hadoop.hive.ql.stats.estimator.StatEstimatorProvider;
 import org.apache.hadoop.hive.ql.stats.estimator.PessimisticStatCombiner;
@@ -89,8 +90,8 @@ public StatEstimator getStatEstimator() {
   static class CoalesceStatEstimator implements StatEstimator {
 
     @Override
-    public Optional<ColStatistics> estimate(List<ColStatistics> argStats) {
-      PessimisticStatCombiner combiner = new PessimisticStatCombiner();
+    public Optional<ColStatistics> estimate(List<ColStatistics> argStats, Statistics parentStats) {
+      PessimisticStatCombiner combiner = new PessimisticStatCombiner(parentStats.getNumRows());
       for (int i = 0; i < argStats.size(); i++) {
         combiner.add(argStats.get(i));
       }

ql/src/java/org/apache/hadoop/hive/ql/udf/generic/GenericUDFIf.java

@@ -29,6 +29,7 @@
 import org.apache.hadoop.hive.ql.exec.vector.VectorizedExpressionsSupportDecimal64;
 import org.apache.hadoop.hive.ql.metadata.HiveException;
 import org.apache.hadoop.hive.ql.plan.ColStatistics;
+import org.apache.hadoop.hive.ql.plan.Statistics;
 import org.apache.hadoop.hive.ql.stats.estimator.StatEstimator;
 import org.apache.hadoop.hive.ql.stats.estimator.StatEstimatorProvider;
 import org.apache.hadoop.hive.ql.stats.estimator.PessimisticStatCombiner;
@@ -187,8 +188,8 @@ public StatEstimator getStatEstimator() {
   static class IfStatEstimator implements StatEstimator {
 
     @Override
-    public Optional<ColStatistics> estimate(List<ColStatistics> argStats) {
-      PessimisticStatCombiner combiner = new PessimisticStatCombiner();
+    public Optional<ColStatistics> estimate(List<ColStatistics> argStats, Statistics parentStats) {
+      PessimisticStatCombiner combiner = new PessimisticStatCombiner(parentStats.getNumRows());
       combiner.add(argStats.get(1));
       combiner.add(argStats.get(2));
       return combiner.getResult();

ql/src/java/org/apache/hadoop/hive/ql/udf/generic/GenericUDFWhen.java

@@ -25,6 +25,7 @@
 import org.apache.hadoop.hive.ql.exec.UDFArgumentTypeException;
 import org.apache.hadoop.hive.ql.metadata.HiveException;
 import org.apache.hadoop.hive.ql.plan.ColStatistics;
+import org.apache.hadoop.hive.ql.plan.Statistics;
 import org.apache.hadoop.hive.ql.stats.estimator.PessimisticStatCombiner;
 import org.apache.hadoop.hive.ql.stats.estimator.StatEstimator;
 import org.apache.hadoop.hive.ql.stats.estimator.StatEstimatorProvider;
@@ -143,8 +144,8 @@ public StatEstimator getStatEstimator() {
   static class WhenStatEstimator implements StatEstimator {
 
     @Override
-    public Optional<ColStatistics> estimate(List<ColStatistics> argStats) {
-      PessimisticStatCombiner combiner = new PessimisticStatCombiner();
+    public Optional<ColStatistics> estimate(List<ColStatistics> argStats, Statistics parentStats) {
+      PessimisticStatCombiner combiner = new PessimisticStatCombiner(parentStats.getNumRows());
       for (int i = 1; i < argStats.size(); i += 2) {
         combiner.add(argStats.get(i));
       }