Add physicalWrittenBytes to PlanNodeStats (#8213)

Summary:
The physicalWrittenBytes metric is available in the OperatorStats class, but not
in the PlanNodeStats. When accessing metrics from Task::taskStats(), we are
unable to retrieve the physicalWrittenBytes' value.

Ad physicalWrittenBytes metric to PlanNodeStats.

Pull Request resolved: #8213

Reviewed By: xiaoxmeng

Differential Revision: D52554834

Pulled By: mbasmanova

fbshipit-source-id: 7edb3b468cc8c32c5861a0bd3309a4473b837079

velox/exec/PlanNodeStats.cpp

@@ -54,6 +54,8 @@ void PlanNodeStats::addTotals(const OperatorStats& stats) {
   peakMemoryBytes += stats.memoryStats.peakTotalMemoryReservation;
   numMemoryAllocations += stats.memoryStats.numMemoryAllocations;
 
+  physicalWrittenBytes += stats.physicalWrittenBytes;
+
   for (const auto& [name, runtimeStats] : stats.runtimeStats) {
     if (UNLIKELY(customStats.count(name) == 0)) {
       customStats.insert(std::make_pair(name, runtimeStats));
@@ -154,6 +156,7 @@ folly::dynamic toPlanStatsJson(const facebook::velox::exec::TaskStats& stats) {
       stat["blockedWallNanos"] = operatorStat.second->blockedWallNanos;
       stat["peakMemoryBytes"] = operatorStat.second->peakMemoryBytes;
       stat["numMemoryAllocations"] = operatorStat.second->numMemoryAllocations;
+      stat["physicalWrittenBytes"] = operatorStat.second->physicalWrittenBytes;
       stat["numDrivers"] = operatorStat.second->numDrivers;
       stat["numSplits"] = operatorStat.second->numSplits;
       stat["spilledInputBytes"] = operatorStat.second->spilledInputBytes;

velox/exec/PlanNodeStats.h

@@ -92,6 +92,8 @@ struct PlanNodeStats {
 
   uint64_t numMemoryAllocations{0};
 
+  uint64_t physicalWrittenBytes{0};
+
   /// Operator-specific counters.
   std::unordered_map<std::string, RuntimeMetric> customStats;
 

velox/exec/tests/TableWriteTest.cpp

@@ -2972,17 +2972,18 @@ TEST_P(AllTableWriterTest, tableWriterStats) {
   const int32_t ORC_HEADER_LEN{3};
   const auto fixedWrittenBytes =
       numWrittenFiles * (fileFormat_ == FileFormat::DWRF ? ORC_HEADER_LEN : 0);
-  for (int i = 0; i < task->taskStats().pipelineStats.size(); ++i) {
-    auto operatorStats = task->taskStats().pipelineStats.at(i).operatorStats;
-    for (int j = 0; j < operatorStats.size(); ++j) {
-      if (operatorStats.at(j).operatorType == "TableWrite") {
-        ASSERT_GT(operatorStats.at(j).physicalWrittenBytes, fixedWrittenBytes);
-        ASSERT_EQ(
-            operatorStats.at(j).runtimeStats.at("numWrittenFiles").sum,
-            numWrittenFiles);
-      }
-    }
-  }
+
+  auto planStats = exec::toPlanStats(task->taskStats());
+  auto& stats = planStats.at(tableWriteNodeId_);
+  ASSERT_GT(stats.physicalWrittenBytes, fixedWrittenBytes);
+  ASSERT_GT(
+      stats.operatorStats.at("TableWrite")->physicalWrittenBytes,
+      fixedWrittenBytes);
+  ASSERT_EQ(
+      stats.operatorStats.at("TableWrite")
+          ->customStats.at("numWrittenFiles")
+          .sum,
+      numWrittenFiles);
 }
 
 DEBUG_ONLY_TEST_P(