Add pipeline-level driver timing stats. (#17237)

Summary:

Operator stats have gaps and to suitable to determine where time went in a task.
Adding driver timing stats per pipeline to assist in understanding a query bottlenecks.

The current version does not handle the grouped execution well.
Working on the extra to support grouped execution as well.

Differential Revision: D101395498

Author: spershin

velox/exec/Driver.cpp

@@ -228,6 +228,12 @@ void BlockingState::setResume(std::shared_ptr<BlockingState> state) {
         std::lock_guard<std::timed_mutex> l(task->mutex());
         if (!driver->state().isTerminated) {
           state->operator_->recordBlockingTime(state->sinceUs_, state->reason_);
+          // Accumulate driver-level blocked time.
+          const uint64_t blockedNow =
+              std::chrono::duration_cast<std::chrono::microseconds>(
+                  std::chrono::high_resolution_clock::now().time_since_epoch())
+                  .count();
+          driver->addDriverBlockedTime((blockedNow - state->sinceUs_) * 1'000);
         }
         VELOX_CHECK(!driver->state().suspended());
         VELOX_CHECK(driver->state().hasBlockingFuture);
@@ -507,6 +513,12 @@ StopReason Driver::runInternal(
   const auto now = getCurrentTimeMicro();
   const auto queuedTimeUs = now - queueTimeStartUs_;
 
+  // Record driver on-thread time from the very start of runInternal.
+  // Captures everything including Task::enter() mutex wait and close().
+  auto onThreadTimeGuard = folly::makeGuard([this, now]() {
+    totalDriverOnThreadNanos_ += (getCurrentTimeMicro() - now) * 1'000;
+  });
+
   // Update the next operator's queueTime.
   StopReason stop =
       closed_ ? StopReason::kTerminate : task()->enter(state_, now);
@@ -531,6 +543,9 @@ StopReason Driver::runInternal(
         kMetricDriverQueueTimeMs, queuedTimeUs / 1'000);
   }
 
+  // Accumulate driver-level queued time.
+  totalDriverQueuedNanos_ += queuedTimeUs * 1'000;
+
   CancelGuard guard(self, task().get(), &state_, [&](StopReason reason) {
     // This is run on error or cancel exit.
     if (reason == StopReason::kTerminate) {
@@ -870,6 +885,16 @@ void Driver::closeOperators() {
     op->close();
   }
 
+  // Report driver-level lifecycle timing to the Task accumulator.
+  // Use partitionId (0..numDrivers-1) so same-index drivers across split
+  // groups in grouped execution are summed together.
+  task()->addDriverLifecycleStats(
+      static_cast<uint32_t>(ctx_->pipelineId),
+      ctx_->partitionId,
+      totalDriverQueuedNanos_,
+      totalDriverOnThreadNanos_,
+      totalDriverBlockedNanos_);
+
   // Add operator stats to the task.
   for (auto& op : operators_) {
     auto stats = op->stats(true);
@@ -904,15 +929,25 @@ void Driver::updateStats() {
             1'000'000 * state_.totalOffThreadTimeMs,
             RuntimeCounter::Unit::kNanos);
   }
+
   task()->addDriverStats(ctx_->pipelineId, std::move(stats));
 }
 
 void Driver::updateOperatorBlockingStats() {
   // Record blocked time if the driver was blocked when terminated.
   // This ensures we don't lose blocked time metrics when a query is aborted.
-  if (state_.hasBlockingFuture && blockedOperatorId_ < operators_.size()) {
-    operators_[blockedOperatorId_]->recordBlockingTime(
-        state_.blockingStartUs, blockingReason_);
+  if (state_.hasBlockingFuture) {
+    const uint64_t blockedNow =
+        std::chrono::duration_cast<std::chrono::microseconds>(
+            std::chrono::high_resolution_clock::now().time_since_epoch())
+            .count();
+    // Accumulate driver-level blocked time unconditionally.
+    totalDriverBlockedNanos_ += (blockedNow - state_.blockingStartUs) * 1'000;
+    // Record per-operator blocked time if operator is available.
+    if (blockedOperatorId_ < operators_.size()) {
+      operators_[blockedOperatorId_]->recordBlockingTime(
+          state_.blockingStartUs, blockingReason_);
+    }
   }
 }
 

velox/exec/Driver.h

@@ -416,6 +416,12 @@ class Driver : public std::enable_shared_from_this<Driver> {
   /// memory arbitration finishes.
   bool checkUnderArbitration(ContinueFuture* future);
 
+  /// Accumulates blocked time for driver-level lifecycle tracking.
+  /// Called from BlockingState::setResume() when the blocking future resolves.
+  void addDriverBlockedTime(uint64_t nanos) {
+    totalDriverBlockedNanos_ += nanos;
+  }
+
   void initializeOperatorStats(std::vector<OperatorStats>& stats);
 
   /// Close operators and add operator stats to the task.
@@ -700,6 +706,15 @@ class Driver : public std::enable_shared_from_this<Driver> {
   // Timer used to track down the time we are sitting in the driver queue.
   size_t queueTimeStartUs_{0};
 
+  // Driver-level lifecycle timing: independently tracks the three states
+  // a driver can be in (queued, on-thread, blocked) to enable gap analysis.
+  // Reported as RuntimeStats on the source operator at close time.
+  // Atomic because closeByTask() may read these from a different thread
+  // than the one running the onThreadTimeGuard scope guard.
+  std::atomic<uint64_t> totalDriverQueuedNanos_{0};
+  std::atomic<uint64_t> totalDriverOnThreadNanos_{0};
+  std::atomic<uint64_t> totalDriverBlockedNanos_{0};
+
   // Id (index in the vector) of the current operator to run (or the 1st one if
   // we haven't started yet). Used to determine which operator's queueTime we
   // should update.

velox/exec/Task.cpp

@@ -540,6 +540,8 @@ void Task::init(std::optional<common::SpillDiskOptions>&& spillDiskOpts) {
     numDriversPerLeafNode_[factory->leafNodeId()] = factory->numDrivers;
   }
 
+  initDriverLifecycleStatsLocked();
+
   // Create drivers.
   createSplitGroupStateLocked(kUngroupedGroupId);
   std::vector<std::shared_ptr<Driver>> drivers =
@@ -1060,6 +1062,8 @@ void Task::createDriverFactoriesLocked(uint32_t maxDrivers) {
         factory->inputDriver, factory->outputDriver);
   }
 
+  initDriverLifecycleStatsLocked();
+
   validateGroupedExecutionLeafNodes();
 }
 
@@ -2730,6 +2734,37 @@ void Task::addDriverStats(int pipelineId, DriverStats stats) {
   taskStats_.pipelineStats[pipelineId].driverStats.push_back(std::move(stats));
 }
 
+void Task::initDriverLifecycleStatsLocked() {
+  pipelineLifecycleStats_.resize(driverFactories_.size());
+  for (size_t i = 0; i < driverFactories_.size(); ++i) {
+    auto& pls = pipelineLifecycleStats_[i];
+    const auto& leafNode = driverFactories_[i]->planNodes.front();
+    pls.sourceOperatorType = leafNode->name();
+    pls.sourcePlanNodeId = leafNode->id();
+    pls.driverTimes.resize(driverFactories_[i]->numDrivers);
+  }
+}
+
+void Task::addDriverLifecycleStats(
+    uint32_t pipelineId,
+    uint32_t driverIndex,
+    uint64_t queuedNanos,
+    uint64_t onThreadNanos,
+    uint64_t blockedNanos) {
+  std::lock_guard<std::timed_mutex> l(mutex_);
+  if (pipelineId >= pipelineLifecycleStats_.size()) {
+    return;
+  }
+  auto& pls = pipelineLifecycleStats_[pipelineId];
+  if (pls.driverTimes.empty()) {
+    return;
+  }
+  const auto idx = driverIndex % pls.driverTimes.size();
+  pls.driverTimes[idx].queuedNanos += queuedNanos;
+  pls.driverTimes[idx].onThreadNanos += onThreadNanos;
+  pls.driverTimes[idx].blockedNanos += blockedNanos;
+}
+
 TaskStats Task::taskStats() const {
   std::lock_guard<std::timed_mutex> l(mutex_);
 
@@ -2785,6 +2820,41 @@ TaskStats Task::taskStats() const {
     taskStats.longestRunningOpCallMs = 0;
   }
 
+  // Emit per-pipeline driver lifecycle timing. Merged into the first
+  // existing DriverStats entry for each pipeline to avoid inflating the
+  // driverStats vector. Each logical driver index contributes one sample,
+  // giving proper sum/count/min/max aggregation.
+  for (size_t i = 0; i < pipelineLifecycleStats_.size(); ++i) {
+    const auto& pls = pipelineLifecycleStats_[i];
+    if (pls.driverTimes.empty()) {
+      continue;
+    }
+    auto& pipeDriverStats = taskStats.pipelineStats[i].driverStats;
+    if (pipeDriverStats.empty()) {
+      pipeDriverStats.emplace_back();
+    }
+    auto& targetStats = pipeDriverStats[0].runtimeStats;
+    const auto prefix = fmt::format(
+        "P{}-{}.{}", i, pls.sourceOperatorType, pls.sourcePlanNodeId);
+    const auto queuedKey = fmt::format("{}.driverQueuedWallNanos", prefix);
+    const auto onThreadKey = fmt::format("{}.driverOnThreadWallNanos", prefix);
+    const auto blockedKey = fmt::format("{}.driverBlockedWallNanos", prefix);
+    for (const auto& timing : pls.driverTimes) {
+      auto addOrMerge = [&targetStats](const std::string& key, uint64_t nanos) {
+        const auto value = saturateCast(nanos);
+        auto it = targetStats.find(key);
+        if (it != targetStats.end()) {
+          it->second.merge(RuntimeMetric(value, RuntimeCounter::Unit::kNanos));
+        } else {
+          targetStats[key] = RuntimeMetric(value, RuntimeCounter::Unit::kNanos);
+        }
+      };
+      addOrMerge(queuedKey, timing.queuedNanos);
+      addOrMerge(onThreadKey, timing.onThreadNanos);
+      addOrMerge(blockedKey, timing.blockedNanos);
+    }
+  }
+
   auto bufferManager = bufferManager_.lock();
   taskStats.outputBufferUtilization = bufferManager->getUtilization(taskId_);
   taskStats.outputBufferOverutilized = bufferManager->isOverutilized(taskId_);

velox/exec/Task.h

@@ -615,6 +615,16 @@ class Task : public std::enable_shared_from_this<Task> {
   /// Adds per driver statistics.  Called from Drivers upon their closure.
   void addDriverStats(int pipelineId, DriverStats stats);
 
+  /// Accumulates driver lifecycle timing (queued, on-thread, blocked) for gap
+  /// analysis. Same-index drivers across split groups are summed together.
+  /// Called from Driver::closeOperators() upon driver closure.
+  void addDriverLifecycleStats(
+      uint32_t pipelineId,
+      uint32_t driverIndex,
+      uint64_t queuedNanos,
+      uint64_t onThreadNanos,
+      uint64_t blockedNanos);
+
   /// Returns kNone if no pause or terminate is requested. The thread count is
   /// incremented if kNone is returned. If something else is returned the
   /// calling thread should unwind and return itself to its pool. If 'this' goes
@@ -1362,6 +1372,27 @@ class Task : public std::enable_shared_from_this<Task> {
 
   TaskStats taskStats_;
 
+  // Per-pipeline driver lifecycle timing accumulator. For each pipeline, holds
+  // a vector of per-driver timing indexed by driver index within the pipeline.
+  // In grouped execution, same-index drivers across groups accumulate into the
+  // same entry, giving the total time for a logical driver across all groups.
+  // Reported as RuntimeMetrics at task close via taskStats().
+  struct DriverLifecycleTiming {
+    uint64_t queuedNanos{0};
+    uint64_t onThreadNanos{0};
+    uint64_t blockedNanos{0};
+  };
+
+  struct PipelineLifecycleStats {
+    std::string sourceOperatorType;
+    core::PlanNodeId sourcePlanNodeId;
+    std::vector<DriverLifecycleTiming> driverTimes;
+  };
+  std::vector<PipelineLifecycleStats> pipelineLifecycleStats_;
+
+  /// Initializes pipelineLifecycleStats_ from driverFactories_.
+  void initDriverLifecycleStatsLocked();
+
   // Stores inter-operator state (exchange, bridges) per split group. During
   // ungrouped execution we use the [0] entry in this vector.
   std::unordered_map<uint32_t, SplitGroupState> splitGroupStates_;

velox/exec/tests/TaskTest.cpp

@@ -2555,6 +2555,110 @@ TEST_F(TaskTest, updateStatsWhileCloseOffThreadDriver) {
   ASSERT_GE(totalOffThreadTime.sum, 0);
 }
 
+// Verifies that driver-level lifecycle timing metrics
+// (driverQueuedWallNanos, driverOnThreadWallNanos, driverBlockedWallNanos)
+// are reported correctly for each pipeline.
+TEST_F(TaskTest, driverLifecycleTimingStats) {
+  auto data = makeRowVector({
+      makeFlatVector<int64_t>(1'000, [](auto row) { return row % 10; }),
+      makeFlatVector<int64_t>(1'000, [](auto row) { return row; }),
+  });
+
+  auto buildData = makeRowVector(
+      {"u0", "u1"},
+      {
+          makeFlatVector<int64_t>(100, [](auto row) { return row % 10; }),
+          makeFlatVector<int64_t>(100, [](auto row) { return row; }),
+      });
+
+  auto planNodeIdGenerator = std::make_shared<core::PlanNodeIdGenerator>();
+
+  // Build a hash join plan to get multiple pipelines:
+  // Pipeline 0 (probe): Values -> HashProbe -> sink
+  // Pipeline 1 (build): Values -> HashBuild
+  core::PlanNodeId probeScanId;
+  auto plan =
+      PlanBuilder(planNodeIdGenerator)
+          .values({data})
+          .capturePlanNodeId(probeScanId)
+          .hashJoin(
+              {"c0"},
+              {"u0"},
+              PlanBuilder(planNodeIdGenerator).values({buildData}).planNode(),
+              "",
+              {"c0", "c1"})
+          .planNode();
+
+  auto result = AssertQueryBuilder(plan).copyResults(pool_.get());
+  ASSERT_GT(result->size(), 0);
+
+  // Get task stats from the most recently completed task.
+  auto tasks = Task::getRunningTasks();
+  // Tasks may already be cleaned up. Use a cursor-based approach instead.
+
+  // Re-run using CursorParameters to access the task directly.
+  CursorParameters params;
+  params.planNode = plan;
+  params.queryCtx = core::QueryCtx::create(driverExecutor_.get());
+  params.maxDrivers = 2;
+
+  auto cursor = TaskCursor::create(params);
+  while (cursor->moveNext()) {
+  }
+  auto task = cursor->task();
+  ASSERT_TRUE(waitForTaskCompletion(task.get()));
+
+  auto taskStats = task->taskStats();
+  // We should have at least 2 pipelines (probe and build).
+  ASSERT_GE(taskStats.pipelineStats.size(), 2);
+
+  // Helper to find a driver stat by substring match across all pipelines.
+  auto findDriverStat =
+      [&taskStats](
+          const std::string& statSubstr) -> std::optional<RuntimeMetric> {
+    for (const auto& pipeline : taskStats.pipelineStats) {
+      for (const auto& ds : pipeline.driverStats) {
+        for (const auto& [name, metric] : ds.runtimeStats) {
+          if (name.find(statSubstr) != std::string::npos) {
+            return metric;
+          }
+        }
+      }
+    }
+    return std::nullopt;
+  };
+
+  // Verify driver lifecycle metrics exist and have sensible values.
+  auto queued = findDriverStat("driverQueuedWallNanos");
+  auto onThread = findDriverStat("driverOnThreadWallNanos");
+  auto blocked = findDriverStat("driverBlockedWallNanos");
+
+  ASSERT_TRUE(queued.has_value()) << "driverQueuedWallNanos not found";
+  ASSERT_TRUE(onThread.has_value()) << "driverOnThreadWallNanos not found";
+  ASSERT_TRUE(blocked.has_value()) << "driverBlockedWallNanos not found";
+
+  // Each metric should have count >= 1 (at least one driver reported).
+  ASSERT_GE(queued->count, 1);
+  ASSERT_GE(onThread->count, 1);
+  ASSERT_GE(blocked->count, 1);
+
+  // On-thread time must be positive (drivers did real work).
+  ASSERT_GT(onThread->sum, 0);
+  ASSERT_GT(onThread->max, 0);
+
+  // Queued and blocked times must be non-negative.
+  ASSERT_GE(queued->sum, 0);
+  ASSERT_GE(blocked->sum, 0);
+
+  // Verify the stat names contain the pipeline prefix and source operator.
+  // Look for probe pipeline stats (source is Values with probeScanId).
+  auto probePrefix = fmt::format("P0-Values.{}", probeScanId);
+  auto probeOnThread = findDriverStat(probePrefix + ".driverOnThreadWallNanos");
+  ASSERT_TRUE(probeOnThread.has_value())
+      << "Probe pipeline stat not found with prefix: " << probePrefix;
+  ASSERT_GT(probeOnThread->max, 0);
+}
+
 DEBUG_ONLY_TEST_F(TaskTest, driverEnqueAfterFailedAndPausedTask) {
   const auto data = makeRowVector({
       makeFlatVector<int64_t>(50, [](auto row) { return row; }),