Add pipeline-level driver timing stats. (facebookincubator#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

@@ -32,6 +32,14 @@ Driver::~Driver() = default;
 
 namespace {
 
+/// Returns current time in microseconds using high_resolution_clock.
+/// Used for driver-level lifecycle timing to match BlockingState::sinceUs_.
+inline uint64_t currentTimeMicrosHires() {
+  return std::chrono::duration_cast<std::chrono::microseconds>(
+             std::chrono::high_resolution_clock::now().time_since_epoch())
+      .count();
+}
+
 // Checks if output channel is produced using identity projection and returns
 // input channel if so.
 std::optional<column_index_t> getIdentityProjection(
@@ -228,6 +236,10 @@ 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 using high_resolution_clock,
+          // matching sinceUs_ and all other driver lifecycle timing.
+          driver->addDriverBlockedTime(
+              (currentTimeMicrosHires() - state->sinceUs_) * 1'000);
         }
         VELOX_CHECK(!driver->state().suspended());
         VELOX_CHECK(driver->state().hasBlockingFuture);
@@ -358,7 +370,7 @@ void Driver::enqueueInternal() {
   VELOX_CHECK(!state_.isEnqueued);
   state_.isEnqueued = true;
   // When enqueuing, starting timing the queue time.
-  queueTimeStartUs_ = getCurrentTimeMicro();
+  queueTimeStartUs_ = currentTimeMicrosHires();
 }
 
 // Call an Operator method. record silenced throws, but not a query
@@ -504,9 +516,24 @@ StopReason Driver::runInternal(
     std::shared_ptr<Driver>& self,
     std::shared_ptr<BlockingState>& blockingState,
     RowVectorPtr& result) {
-  const auto now = getCurrentTimeMicro();
+  // All driver timing uses high_resolution_clock consistently
+  // (matching BlockingState::sinceUs_ used for blocked time).
+  const auto now = currentTimeMicrosHires();
   const auto queuedTimeUs = now - queueTimeStartUs_;
 
+  totalDriverQueuedNanos_ += queuedTimeUs * 1'000;
+  onThreadStartUs_ = now;
+  // For the normal close path, closeOperators() finalizes and clears
+  // onThreadStartUs_ before reporting. This guard handles early returns
+  // (e.g. Task::enter() failure) and non-close exit paths.
+  auto onThreadTimeGuard = folly::makeGuard([this]() {
+    if (onThreadStartUs_ > 0) {
+      totalDriverOnThreadNanos_ +=
+          (currentTimeMicrosHires() - onThreadStartUs_) * 1'000;
+      onThreadStartUs_ = 0;
+    }
+  });
+
   // Update the next operator's queueTime.
   StopReason stop =
       closed_ ? StopReason::kTerminate : task()->enter(state_, now);
@@ -870,6 +897,23 @@ 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.
+  // Finalize on-thread time here (the onThreadTimeGuard in runInternal
+  // hasn't fired yet since CancelGuard destructs before it).
+  if (onThreadStartUs_ > 0) {
+    totalDriverOnThreadNanos_ +=
+        (currentTimeMicrosHires() - onThreadStartUs_) * 1'000;
+    onThreadStartUs_ = 0; // Prevent double-counting in the guard.
+  }
+  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 +948,22 @@ 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) {
+    // Accumulate driver-level blocked time unconditionally.
+    totalDriverBlockedNanos_ +=
+        (currentTimeMicrosHires() - 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,22 @@ 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.
+  // All three use high_resolution_clock for consistency, matching
+  // BlockingState::sinceUs_ and enabling accurate gap analysis
+  // (queued + on-thread + blocked ≈ elapsed 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};
+  // Timestamp (micros, high_resolution_clock) when the current on-thread
+  // period started. Set at the beginning of runInternal, used by
+  // closeOperators to snapshot on-thread time before the scope guard fires.
+  uint64_t onThreadStartUs_{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_;