fix: Fix operator finishTiming metrics (#12316)

Summary:
Issue: We record the execution time of `nextOp->noMoreInput()` to `op`'s
finishTiming metrics. This issue is noticeable in the OrderBy, Window, and
HashBuild operators, where computationally intensive tasks are executed within
the `noMoreInput()` function—for example, sorting occurs in `noMoreInput()`. As
a result, the recorded wall time is incorrectly added to upstream operators.

Pull Request resolved: #12316

Reviewed By: kKPulla

Differential Revision: D69635556

Pulled By: pedroerp

fbshipit-source-id: a31c41976f196c007de2140b790c590f359c8bda

velox/exec/Driver.cpp

@@ -671,7 +671,7 @@ StopReason Driver::runInternal(
               });
               if (finished) {
                 withDeltaCpuWallTimer(
-                    op, &OperatorStats::finishTiming, [this, &nextOp]() {
+                    nextOp, &OperatorStats::finishTiming, [this, &nextOp]() {
                       TestValue::adjust(
                           "facebook::velox::exec::Driver::runInternal::noMoreInput",
                           nextOp);

velox/exec/tests/TaskTest.cpp

@@ -1241,7 +1241,7 @@ DEBUG_ONLY_TEST_F(TaskTest, liveStats) {
   EXPECT_EQ(numBatches, operatorStats.outputVectors);
   // isBlocked() should be called at least twice for each batch
   EXPECT_LE(2 * numBatches, operatorStats.isBlockedTiming.count);
-  EXPECT_EQ(2, operatorStats.finishTiming.count);
+  EXPECT_EQ(1, operatorStats.finishTiming.count);
   // No operators with background CPU time yet.
   EXPECT_EQ(0, operatorStats.backgroundTiming.count);
 
@@ -2415,4 +2415,29 @@ DEBUG_ONLY_TEST_F(TaskTest, taskCancellation) {
   task.reset();
   waitForAllTasksToBeDeleted();
 }
+
+TEST_F(TaskTest, finishTiming) {
+  auto data = makeRowVector({
+      makeFlatVector<int64_t>(1'000, [](auto row) { return row; }),
+  });
+  core::PlanNodeId projectId;
+  core::PlanNodeId orderById;
+  auto plan = PlanBuilder()
+                  .values({data, data})
+                  .project({"c0"})
+                  .capturePlanNodeId(projectId)
+                  .orderBy({"c0 DESC NULLS LAST"}, false)
+                  .capturePlanNodeId(orderById)
+                  .planFragment();
+
+  auto [task, _] = executeSerial(plan);
+  auto taskStats = exec::toPlanStats(task->taskStats());
+  auto& projectStats = taskStats.at(projectId);
+  auto& orderByStats = taskStats.at(orderById);
+  // Since the sort is executed in the 'noMoreInput' function of the OrderBy
+  // operator, the finish time of the OrderBy operator should be greater than
+  // that of the Project operator.
+  ASSERT_GT(
+      orderByStats.finishTiming.wallNanos, projectStats.finishTiming.wallNanos);
+}
 } // namespace facebook::velox::exec::test