[impala] 01/02: IMPALA-8912: Avoid sampling hbase table twice for HBaseScanNode

[jo...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

joemcdonnell pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git

commit d21f00ef2f2219551e3501e3cf1cf45ff4afdcef
Author: stiga-huang <hu...@gmail.com>
AuthorDate: Sat Aug 31 01:57:39 2019 -0700

    IMPALA-8912: Avoid sampling hbase table twice for HBaseScanNode
    
    HBaseScanNode#computeStats do sampling on the target hbase table to
    estimate cardinality and row size. The sampling can be time consuming
    since it requires HBase RPCs. As mentioned in the JIRA description,
    HBaseScanNode#computeStats can be called twice if it's the root of the
    single node plan. The sampling can be skipped in the second pass since
    startKey_, endKey_ and keyConjucts_ are not changed.
    
    inputCardinality_ will have a valid value if the sampling has been done
    without errors. We check this to determine whether we can skip sampling
    in the second pass.
    
    Tests:
     - Testing in my local env with TRACE log level. Observing the sampling
    be done only once.
     - Run CORE tests.
    
    Change-Id: I4650483c0504128048630714e993b481737fd1e2
    Reviewed-on: http://gerrit.cloudera.org:8080/14167
    Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
    Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
 .../org/apache/impala/catalog/FeHBaseTable.java    |  9 ++++++
 .../org/apache/impala/planner/HBaseScanNode.java   | 32 +++++++++++++++++-----
 2 files changed, 34 insertions(+), 7 deletions(-)

diff --git a/fe/src/main/java/org/apache/impala/catalog/FeHBaseTable.java b/fe/src/main/java/org/apache/impala/catalog/FeHBaseTable.java
index df4f12e..f2449ff 100644
--- a/fe/src/main/java/org/apache/impala/catalog/FeHBaseTable.java
+++ b/fe/src/main/java/org/apache/impala/catalog/FeHBaseTable.java
@@ -393,6 +393,9 @@ public interface FeHBaseTable extends FeTable {
 
         // Collects stats samples from at least MIN_NUM_REGIONS_TO_CHECK
         // and at most all regions until the delta is small enough.
+        if (LOG.isTraceEnabled()) {
+          LOG.trace("Start rows sampling on " + locations.size() + " regions");
+        }
         while ((statsSize.count() < MIN_NUM_REGIONS_TO_CHECK ||
             statsSize.stddev() > statsSize.mean() * DELTA_FROM_AVERAGE) &&
             statsSize.count() < locations.size()) {
@@ -402,6 +405,12 @@ public interface FeHBaseTable extends FeTable {
                   clusterStatus);
           totalEstimatedRows += tmp.first;
           statsSize.addSample(tmp.second);
+          if (LOG.isTraceEnabled()) {
+            LOG.trace(String.format("Estimation state: totalEstimatedRows=%d, " +
+                    "statsSize.count=%d, statsSize.stddev=%f, statsSize.mean=%f",
+                totalEstimatedRows, statsSize.count(), statsSize.stddev(),
+                statsSize.mean()));
+          }
         }
 
         // Sum up the total size for all regions in range.
diff --git a/fe/src/main/java/org/apache/impala/planner/HBaseScanNode.java b/fe/src/main/java/org/apache/impala/planner/HBaseScanNode.java
index 25ccae1..36f7b63 100644
--- a/fe/src/main/java/org/apache/impala/planner/HBaseScanNode.java
+++ b/fe/src/main/java/org/apache/impala/planner/HBaseScanNode.java
@@ -21,6 +21,7 @@ import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.ListIterator;
@@ -99,11 +100,11 @@ public class HBaseScanNode extends ScanNode {
   // One range per clustering column. The range bounds are expected to be constants.
   // A null entry means there's no range restriction for that particular key.
   // If keyRanges is non-null it always contains as many entries as there are clustering
-  // cols.
+  // cols. Don't replace this variable after init().
   private List<ValueRange> keyRanges_ = new ArrayList<>();
 
-  // The list of conjuncts used to create the key ranges. Used if we
-  // must estimate cardinality based on row count stats.
+  // The list of conjuncts used to create the key ranges. Used if we must estimate
+  // cardinality based on row count stats. Don't replace this variable after init().
   private List<Expr> keyConjuncts_ = new ArrayList<>();
 
   // derived from keyRanges_; empty means unbounded;
@@ -156,6 +157,12 @@ public class HBaseScanNode extends ScanNode {
     computeScanRangeLocations(analyzer);
     Preconditions.checkState(!scanRangeSpecs_.isSetSplit_specs());
 
+    // Make sure key ranges are not changed any more. So startKey_ and stopKey_ are
+    // stable too. These invariants make it safe to reuse row estimation results in
+    // computeStats.
+    keyRanges_ = Collections.unmodifiableList(keyRanges_);
+    keyConjuncts_ = Collections.unmodifiableList(keyConjuncts_);
+
     // Call computeStats() after materializing slots and computing the mem layout.
     computeStats(analyzer);
   }
@@ -285,12 +292,23 @@ public class HBaseScanNode extends ScanNode {
   public void computeStats(Analyzer analyzer) {
     super.computeStats(analyzer);
     FeHBaseTable tbl = (FeHBaseTable) desc_.getTable();
-
+    if (LOG.isTraceEnabled()) {
+      LOG.trace("computing stats for HbaseScan on " + tbl.getHBaseTableName());
+    }
     ValueRange rowRange = keyRanges_.get(0);
     if (isEmpty_) {
       cardinality_ = 0;
     } else if (rowRange != null && rowRange.isEqRange()) {
       cardinality_ = 1;
+    } else if (inputCardinality_ >= 0) {
+      // We have run computeStats successfully. Don't need to estimate cardinality again
+      // (IMPALA-8912). Check some invariants if computeStats has been called.
+      Preconditions.checkState(numNodes_ > 0);
+      Preconditions.checkState(cardinality_ >= 0);
+      cardinality_ = inputCardinality_;
+      if (LOG.isTraceEnabled()) {
+        LOG.trace("Reuse last stats: inputCardinality_=" + inputCardinality_);
+      }
     } else {
       Pair<Long, Long> estimate;
       if (analyzer.getQueryCtx().isDisable_hbase_row_est()) {
@@ -319,7 +337,7 @@ public class HBaseScanNode extends ScanNode {
         cardinality_ = estimate.first;
         if (estimate.second > 0) {
           suggestedCaching_ = (int)
-              Math.max(MAX_HBASE_FETCH_BATCH_SIZE / estimate.second.longValue(), 1);
+              Math.max(MAX_HBASE_FETCH_BATCH_SIZE / estimate.second, 1);
         }
       }
     }
@@ -329,7 +347,7 @@ public class HBaseScanNode extends ScanNode {
     cardinality_ = Math.max(1, cardinality_);
     cardinality_ = capCardinalityAtLimit(cardinality_);
     if (LOG.isTraceEnabled()) {
-      LOG.trace("computeStats HbaseScan: cardinality=" + Long.toString(cardinality_));
+      LOG.trace("computeStats HbaseScan: cardinality=" + cardinality_);
     }
 
     // Assume that each executor in the cluster gets a scan range, unless there are fewer
@@ -337,7 +355,7 @@ public class HBaseScanNode extends ScanNode {
     numNodes_ = Math.max(1, Math.min(scanRangeSpecs_.getConcrete_rangesSize(),
                                 ExecutorMembershipSnapshot.getCluster().numExecutors()));
     if (LOG.isTraceEnabled()) {
-      LOG.trace("computeStats HbaseScan: #nodes=" + Integer.toString(numNodes_));
+      LOG.trace("computeStats HbaseScan: #nodes=" + numNodes_);
     }
   }