Add metrics iceberg.numSplits and iceberg.numDeletes

Author: unidevel

velox/connectors/hive/iceberg/IcebergSplitReader.cpp

@@ -75,6 +75,9 @@ void IcebergSplitReader::prepareSplit(
   splitOffset_ = baseRowReader_->nextRowNumber();
   positionalDeleteFileReaders_.clear();
 
+  runtimeStats_ = &runtimeStats;
+  runtimeStats.unitLoaderStats.addCounter("iceberg.numSplits", RuntimeCounter(1));
+
   const auto& deleteFiles = icebergSplit->deleteFiles;
   for (const auto& deleteFile : deleteFiles) {
     if (deleteFile.content == FileContent::kPositionalDeletes) {
@@ -138,6 +141,11 @@ uint64_t IcebergSplitReader::next(uint64_t size, VectorPtr& output) {
 
   auto rowsScanned = baseRowReader_->next(actualSize, output, &mutation);
 
+  if (mutation.deletedRows != nullptr && runtimeStats_ != nullptr) {
+    auto numDeletes = bits::countBits(mutation.deletedRows, 0, actualSize);
+    runtimeStats_->unitLoaderStats.addCounter("iceberg.numDeletes", RuntimeCounter(numDeletes));
+  }
+
   return rowsScanned;
 }
 

velox/connectors/hive/iceberg/IcebergSplitReader.h

@@ -101,5 +101,7 @@ class IcebergSplitReader : public SplitReader {
   std::list<std::unique_ptr<PositionalDeleteFileReader>>
       positionalDeleteFileReaders_;
   BufferPtr deleteBitmap_;
+  // Pointer to runtime stats for tracking iceberg metrics
+  dwio::common::RuntimeStatistics* runtimeStats_{nullptr};
 };
 } // namespace facebook::velox::connector::hive::iceberg

velox/connectors/hive/iceberg/tests/IcebergReadTest.cpp

@@ -26,6 +26,7 @@
 #include "velox/exec/tests/utils/AssertQueryBuilder.h"
 #include "velox/exec/tests/utils/HiveConnectorTestBase.h"
 #include "velox/exec/tests/utils/PlanBuilder.h"
+#include "velox/exec/TaskStats.h"
 
 #ifdef VELOX_ENABLE_PARQUET
 #include "velox/dwio/parquet/RegisterParquetReader.h"
@@ -181,7 +182,7 @@ class HiveIcebergTest : public HiveConnectorTestBase {
   /// positions for data_file_1 and data_file_2. THere are 3 RowGroups in this
   /// delete file, the first two contain positions for data_file_1, and the last
   /// contain positions for data_file_2
-  void assertPositionalDeletes(
+  std::shared_ptr<exec::Task> assertPositionalDeletes(
       const std::map<std::string, std::vector<int64_t>>& rowGroupSizesForFiles,
       const std::unordered_map<
           std::string,
@@ -242,8 +243,11 @@ class HiveIcebergTest : public HiveConnectorTestBase {
     auto planStats = toPlanStats(task->taskStats());
 
     auto it = planStats.find(plan->id());
-    ASSERT_TRUE(it != planStats.end());
-    ASSERT_TRUE(it->second.peakMemoryBytes > 0);
+    EXPECT_TRUE(it != planStats.end());
+    if (it != planStats.end()) {
+      EXPECT_TRUE(it->second.peakMemoryBytes > 0);
+    }
+    return task;
   }
 
   const static int rowCount = 20000;
@@ -947,4 +951,48 @@ TEST_F(HiveIcebergTest, positionalDeleteFileWithRowGroupFilter) {
       0);
 }
 #endif
+
+TEST_F(HiveIcebergTest, icebergMetrics) {
+  folly::SingletonVault::singleton()->registrationComplete();
+
+  // Helper function to aggregate a runtime metric across all pipelines and operators
+  auto getAggregatedRuntimeMetric = [](const exec::TaskStats& taskStats, const std::string& metricName) -> int64_t {
+    int64_t total = 0;
+    for (const auto& pipelineStats : taskStats.pipelineStats) {
+      for (const auto& operatorStats : pipelineStats.operatorStats) {
+        auto it = operatorStats.runtimeStats.find(metricName);
+        if (it != operatorStats.runtimeStats.end()) {
+          total += it->second.sum;
+        }
+      }
+    }
+    return total;
+  };
+
+  std::map<std::string, std::vector<int64_t>> rowGroupSizesForFiles = {
+      {"data_file_1", {100, 85}}};
+  std::unordered_map<
+      std::string,
+      std::multimap<std::string, std::vector<int64_t>>>
+      deleteFilesForBaseDatafiles;
+  deleteFilesForBaseDatafiles["delete_file_1"] = {{"data_file_1", {0, 1, 99}}};
+  auto task =
+      assertPositionalDeletes(rowGroupSizesForFiles, deleteFilesForBaseDatafiles);
+  const auto& taskStats = task->taskStats();
+  
+  ASSERT_EQ(getAggregatedRuntimeMetric(taskStats, "iceberg.numSplits"), 1);
+  ASSERT_EQ(getAggregatedRuntimeMetric(taskStats, "iceberg.numDeletes"), 3);
+
+  rowGroupSizesForFiles = {
+      {"data_file_1", {100, 85}}, {"data_file_2", {99, 1}}};
+  deleteFilesForBaseDatafiles.clear();
+  deleteFilesForBaseDatafiles["delete_file_1"] = {
+      {"data_file_1", {0, 100, 102, 184}}, {"data_file_2", {1, 98, 99}}};
+  task =
+      assertPositionalDeletes(rowGroupSizesForFiles, deleteFilesForBaseDatafiles);
+  const auto& taskStats2 = task->taskStats();
+  
+  ASSERT_EQ(getAggregatedRuntimeMetric(taskStats2, "iceberg.numSplits"), 2);
+  ASSERT_EQ(getAggregatedRuntimeMetric(taskStats2, "iceberg.numDeletes"), 7);
+}
 } // namespace facebook::velox::connector::hive::iceberg

velox/docs/develop/debugging/print-plan-with-stats.rst

@@ -283,3 +283,43 @@ TableScan operator shows how many rows were processed by pushing down aggregatio
 .. code-block::
 
     loadedToValueHook          sum: 50000, count: 5, min: 10000, max: 10000
+
+For Iceberg tables, TableScan operator reports Iceberg-specific statistics:
+
+.. code-block::
+
+   -> TableScan[Table: iceberg_table]
+          iceberg.numSplits     sum: 2, count: 1, min: 2, max: 2
+          iceberg.numDeletes    sum: 7, count: 1, min: 7, max: 7
+
+The `iceberg.numSplits` metric shows the total number of Iceberg splits processed,
+while `iceberg.numDeletes` shows the total number of rows deleted.
+These metrics help understand the overhead of Iceberg's delete
+operations and split processing.
+
+**Note:** These Iceberg metrics are collected by the IcebergSplitReader and are
+only available in the TableScan operator statistics after all splits have been
+processed (when the query completes or when no more splits are available).
+
+To programmatically aggregate these metrics across all operators in a task:
+
+.. code-block:: cpp
+
+   auto getAggregatedRuntimeMetric = [](const exec::TaskStats& taskStats,
+                                        const std::string& metricName) -> int64_t {
+     int64_t total = 0;
+     for (const auto& pipelineStats : taskStats.pipelineStats) {
+       for (const auto& operatorStats : pipelineStats.operatorStats) {
+         auto it = operatorStats.runtimeStats.find(metricName);
+         if (it != operatorStats.runtimeStats.end()) {
+           total += it->second.sum;
+         }
+       }
+     }
+     return total;
+   };
+
+   // Example usage:
+   const auto& taskStats = task->taskStats();
+   int64_t totalSplits = getAggregatedRuntimeMetric(taskStats, "iceberg.numSplits");
+   int64_t totalDeletes = getAggregatedRuntimeMetric(taskStats, "iceberg.numDeletes");