velox(feat): add Task/Operator/Driver Stats toString() utils

Summary:
Needed a way to summarize plan execution for the native execution project. I was unable to find a TaskStats logging function

Adding Task/Operator/DriverStat logging utilities

The caller is responsible for synchronization, no special effort is made to ensure that the stat structures are not being modified during logging. Ideally a copy would be created and the logger invoked on the copy

Differential Revision: D80673402

Author: hdikeman

velox/exec/CMakeLists.txt

@@ -28,6 +28,7 @@ velox_add_library(
   ContainerRowSerde.cpp
   DistinctAggregations.cpp
   Driver.cpp
+  DriverStats.cpp
   EnforceSingleRow.cpp
   Exchange.cpp
   ExchangeClient.cpp
@@ -56,6 +57,7 @@ velox_add_library(
   NestedLoopJoinBuild.cpp
   NestedLoopJoinProbe.cpp
   Operator.cpp
+  OperatorStats.cpp
   OperatorUtils.cpp
   OrderBy.cpp
   OutputBuffer.cpp
@@ -91,6 +93,7 @@ velox_add_library(
   TableWriteMerge.cpp
   TableWriter.cpp
   Task.cpp
+  TaskStats.cpp
   TopN.cpp
   TopNRowNumber.cpp
   Unnest.cpp

velox/exec/DriverStats.cpp

@@ -0,0 +1,39 @@
+/*
+ * Copyright (c) Facebook, Inc. and its affiliates.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "velox/exec/DriverStats.h"
+#include <sstream>
+
+namespace facebook::velox::exec {
+
+std::string DriverStats::toString() const {
+  std::stringstream out;
+
+  out << "[DriverStats] runtimeStats: {";
+  bool first = true;
+  for (const auto& [name, metric] : runtimeStats) {
+    if (!first) {
+      out << ", ";
+    }
+    out << name << ":" << metric.sum;
+    first = false;
+  }
+  out << "}";
+
+  return out.str();
+}
+
+} // namespace facebook::velox::exec

velox/exec/DriverStats.h

@@ -27,6 +27,8 @@ struct DriverStats {
       "totalDriverOffThreadWallNanos";
 
   std::unordered_map<std::string, RuntimeMetric> runtimeStats;
+
+  std::string toString() const;
 };
 
 } // namespace facebook::velox::exec

velox/exec/OperatorStats.cpp

@@ -0,0 +1,57 @@
+/*
+ * Copyright (c) Facebook, Inc. and its affiliates.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "velox/exec/OperatorStats.h"
+#include <fmt/format.h>
+#include <sstream>
+
+namespace facebook::velox::exec {
+
+std::string OperatorStats::toString() const {
+  std::stringstream out;
+
+  out << "[OperatorStats][" << operatorId << "] " << operatorType << " ["
+      << planNodeId << "]";
+
+  out << "\n        input: " << inputPositions << " rows (" << inputBytes
+      << " bytes, " << inputVectors << " vectors)";
+  out << "\n        output: " << outputPositions << " rows (" << outputBytes
+      << " bytes, " << outputVectors << " vectors)";
+
+  if (numSplits > 0) {
+    out << "\n        splits: " << numSplits;
+  }
+
+  if (spilledBytes > 0) {
+    out << "\n      spilled: " << spilledBytes << " bytes (" << spilledRows
+        << " rows, " << spilledPartitions << " partitions)";
+  }
+
+  if (blockedWallNanos > 0) {
+    out << "\n        blockedWallMs: "
+        << fmt::format("{:.2f}ms", blockedWallNanos / 1e6);
+  }
+
+  out << "\n        runtimeStats: {";
+  for (const auto& [name, metric] : runtimeStats) {
+    out << "\n          " << name << ":" << metric.sum << ",";
+  }
+  out << "\n        }";
+
+  return out.str();
+}
+
+} // namespace facebook::velox::exec

velox/exec/OperatorStats.h

@@ -222,6 +222,9 @@ struct OperatorStats {
   void addRuntimeStat(const std::string& name, const RuntimeCounter& value);
   void add(const OperatorStats& other);
   void clear();
+  std::string toString() const;
 };
 
+std::string toString(const OperatorStats& stats);
+
 } // namespace facebook::velox::exec

velox/exec/TaskStats.cpp

@@ -0,0 +1,58 @@
+/*
+ * Copyright (c) Facebook, Inc. and its affiliates.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "velox/exec/TaskStats.h"
+#include <fmt/format.h>
+#include <sstream>
+
+namespace facebook::velox::exec {
+
+std::string PipelineStats::toString() const {
+  std::stringstream out;
+
+  out << "[PipelineStats] " << (inputPipeline ? "INPUT " : "")
+      << (outputPipeline ? "OUTPUT " : "") << operatorStats.size()
+      << " operators, " << driverStats.size() << " drivers.";
+
+  for (const auto& operatorStat : operatorStats) {
+    out << "\n    " << operatorStat.toString();
+  }
+
+  for (const auto& driverStat : driverStats) {
+    out << "\n    " << driverStat.toString();
+  }
+
+  return out.str();
+}
+
+std::string TaskStats::toString() const {
+  std::stringstream out;
+  out << "[TaskStats] " << numFinishedSplits << "/" << numTotalSplits
+      << " splits completed, " << numCompletedDrivers << "/" << numTotalDrivers
+      << " drivers completed, " << pipelineStats.size() << " pipelines";
+
+  if (executionStartTimeMs > 0 && executionEndTimeMs > 0) {
+    out << " (" << (executionEndTimeMs - executionStartTimeMs) << "ms)";
+  }
+
+  for (const auto& pipelineStat : pipelineStats) {
+    out << "\n  " << pipelineStat.toString();
+  }
+
+  return out.str();
+}
+
+} // namespace facebook::velox::exec

velox/exec/TaskStats.h

@@ -47,6 +47,8 @@ struct PipelineStats {
 
   PipelineStats(bool _inputPipeline, bool _outputPipeline)
       : inputPipeline{_inputPipeline}, outputPipeline{_outputPipeline} {}
+
+  std::string toString() const;
 };
 
 /// Stores execution stats per task.
@@ -126,6 +128,8 @@ struct TaskStats {
   uint32_t memoryReclaimCount{0};
   /// The total memory reclamation time.
   uint64_t memoryReclaimMs{0};
+
+  std::string toString() const;
 };
 
 } // namespace facebook::velox::exec