Make CPU Operator Stats Optional (facebookincubator#2638)

Kevin Wilfong · facebook-github-bot · commit b536895b1a8f · 2022-10-10T13:21:55.000-07:00
Summary: Pull Request resolved: facebookincubator#2638 Similar to what facebookincubator#1534 did for Expressions, this adds a configuration property "driver.track_operator_cpu_usage" to control whether cpu usage tracking for operators is on of off. The default is on to maintain the behavior for existing use cases. We have seen in use cases with very small batches (128 or less) measuring the CPU time can itself account for ~10% of the CPU spent in processing batches. For example, running the SimpleArithmetic benchmark with batches of 100 rows With CPU usage tracking: ``` multiplySmall 4.58ns 218.29M multiplySameColumnSmall 4.37ns 228.67M multiplyHalfNullSmall 7.86ns 127.17M multiplyConstantSmall 5.08ns 196.67M multiplyNestedSmall 6.80ns 147.16M multiplyNestedDeepSmall 12.06ns 82.89M ``` Without CPU usage tracking: ``` multiplySmall 4.22ns 236.99M multiplySameColumnSmall 3.99ns 250.70M multiplyHalfNullSmall 7.12ns 140.39M multiplyConstantSmall 4.67ns 214.27M multiplyNestedSmall 5.94ns 168.39M multiplyNestedDeepSmall 11.03ns 90.65M ``` Reviewed By: mbasmanova Differential Revision: D39753640 fbshipit-source-id: f0d10dd5bfc938804713d069bec6d6cd5ef69c0e
diff --git a/velox/core/QueryConfig.h b/velox/core/QueryConfig.h
@@ -58,6 +58,11 @@ class QueryConfig {
   static constexpr const char* kExprTrackCpuUsage =
       "expression.track_cpu_usage";
 
+  // Whether to track CPU usage for stages of individual operators. True by
+  // default. Can be expensive when processing small batches, e.g. < 10K rows.
+  static constexpr const char* kOperatorTrackCpuUsage =
+      "driver.track_operator_cpu_usage";
+
   // Flags used to configure the CAST operator:
 
   // This flag makes the Row conversion to by applied
@@ -271,6 +276,10 @@ class QueryConfig {
     return get<bool>(kExprTrackCpuUsage, false);
   }
 
+  bool operatorTrackCpuUsage() const {
+    return get<bool>(kOperatorTrackCpuUsage, true);
+  }
+
   template <typename T>
   T get(const std::string& key, const T& defaultValue) const {
     return config_->get<T>(key, defaultValue);
diff --git a/velox/exec/Driver.cpp b/velox/exec/Driver.cpp
@@ -160,6 +160,7 @@ Driver::Driver(
   curOpIndex_ = operators_.size() - 1;
   // Operators need access to their Driver for adaptation.
   ctx_->driver = this;
+  trackOperatorCpuUsage_ = ctx_->queryConfig().operatorTrackCpuUsage();
 }
 
 namespace {
@@ -347,7 +348,7 @@ StopReason Driver::runInternal(
             uint64_t resultBytes = 0;
             RowVectorPtr result;
             {
-              CpuWallTimer timer(op->stats().getOutputTiming);
+              auto timer = cpuWallTimer(op->stats().getOutputTiming);
               result = op->getOutput();
               if (result) {
                 VELOX_CHECK(
@@ -362,7 +363,7 @@ StopReason Driver::runInternal(
             }
             pushdownFilters(i);
             if (result) {
-              CpuWallTimer timer(nextOp->stats().addInputTiming);
+              auto timer = cpuWallTimer(op->stats().addInputTiming);
               nextOp->stats().inputVectors += 1;
               nextOp->stats().inputPositions += result->size();
               nextOp->stats().inputBytes += resultBytes;
@@ -391,7 +392,7 @@ StopReason Driver::runInternal(
                 return StopReason::kBlock;
               }
               if (op->isFinished()) {
-                CpuWallTimer timer(nextOp->stats().finishTiming);
+                auto timer = cpuWallTimer(op->stats().finishTiming);
                 nextOp->noMoreInput();
                 break;
               }
@@ -403,7 +404,7 @@ StopReason Driver::runInternal(
           // this will be detected when trying to add input, and we
           // will come back here after this is again on thread.
           {
-            CpuWallTimer timer(op->stats().getOutputTiming);
+            auto timer = cpuWallTimer(op->stats().getOutputTiming);
             result = op->getOutput();
             if (result) {
               VELOX_CHECK(
diff --git a/velox/exec/Driver.h b/velox/exec/Driver.h
@@ -17,8 +17,10 @@
 #include <folly/executors/CPUThreadPoolExecutor.h>
 #include <folly/futures/Future.h>
 #include <folly/portability/SysSyscall.h>
+#include <memory>
 
 #include "velox/common/future/VeloxPromise.h"
+#include "velox/common/time/CpuWallTimer.h"
 #include "velox/connectors/Connector.h"
 #include "velox/core/PlanNode.h"
 #include "velox/core/QueryCtx.h"
@@ -295,6 +297,11 @@ class Driver : public std::enable_shared_from_this<Driver> {
   // position in the pipeline.
   void pushdownFilters(int operatorIndex);
 
+  std::unique_ptr<CpuWallTimer> cpuWallTimer(CpuWallTiming& timing) {
+    return trackOperatorCpuUsage_ ? std::make_unique<CpuWallTimer>(timing)
+                                  : nullptr;
+  }
+
   std::unique_ptr<DriverCtx> ctx_;
   std::atomic_bool closed_{false};
 
@@ -310,6 +317,8 @@ class Driver : public std::enable_shared_from_this<Driver> {
   std::vector<std::unique_ptr<Operator>> operators_;
 
   BlockingReason blockingReason_{BlockingReason::kNotBlocked};
+
+  bool trackOperatorCpuUsage_;
 };
 
 using OperatorSupplier = std::function<std::unique_ptr<Operator>(
diff --git a/velox/exec/tests/DriverTest.cpp b/velox/exec/tests/DriverTest.cpp
@@ -448,6 +448,11 @@ TEST_F(DriverTest, pause) {
       [](int64_t num) { return num % 10 > 0; },
       &hits);
   params.maxDrivers = 10;
+  params.queryCtx =
+      core::QueryCtx::createForTest(std::make_shared<core::MemConfig>(
+          std::unordered_map<std::string, std::string>{
+              // Make sure CPU usage tracking is enabled.
+              {core::QueryConfig::kOperatorTrackCpuUsage, "true"}}));
   int32_t numRead = 0;
   readResults(params, ResultOperation::kPause, 370'000'000, &numRead);
   // Each thread will fully read the 1M rows in values.
