feat: metrics reporting for scan and commit (apache#589)

Initial commit for addressing
apache#533

---------

Co-authored-by: Innocent <idjiofack@jabode.com>

Author: evindj

src/iceberg/CMakeLists.txt

@@ -59,6 +59,13 @@ set(ICEBERG_SOURCES
     manifest/v3_metadata.cc
     metadata_columns.cc
     metrics_config.cc
+    metrics/commit_report.cc
+    metrics/counter.cc
+    metrics/json_serde.cc
+    metrics/metrics_context.cc
+    metrics/metrics_reporters.cc
+    metrics/scan_report.cc
+    metrics/timer.cc
     name_mapping.cc
     partition_field.cc
     partition_spec.cc
@@ -227,6 +234,7 @@ add_subdirectory(puffin)
 add_subdirectory(row)
 add_subdirectory(update)
 add_subdirectory(util)
+add_subdirectory(metrics)
 
 if(ICEBERG_BUILD_BUNDLE)
   set(ICEBERG_BUNDLE_SOURCES

src/iceberg/constants.h

@@ -33,6 +33,7 @@ namespace iceberg {
 constexpr std::string_view kParquetFieldIdKey = "PARQUET:field_id";
 constexpr int64_t kInvalidSnapshotId = -1;
 constexpr int64_t kInvalidSequenceNumber = -1;
+constexpr int64_t kInvalidSchemaId = -1;
 /// \brief Stand-in for the current sequence number that will be assigned when the commit
 /// is successful. This is replaced when writing a manifest list by the ManifestFile
 /// adapter.

src/iceberg/meson.build

@@ -80,6 +80,13 @@ iceberg_sources = files(
     'manifest/v2_metadata.cc',
     'manifest/v3_metadata.cc',
     'metadata_columns.cc',
+    'metrics/commit_report.cc',
+    'metrics/counter.cc',
+    'metrics/json_serde.cc',
+    'metrics/metrics_context.cc',
+    'metrics/metrics_reporters.cc',
+    'metrics/scan_report.cc',
+    'metrics/timer.cc',
     'metrics_config.cc',
     'name_mapping.cc',
     'partition_field.cc',
@@ -282,6 +289,7 @@ subdir('data')
 subdir('deletes')
 subdir('expression')
 subdir('manifest')
+subdir('metrics')
 subdir('puffin')
 subdir('row')
 subdir('update')

src/iceberg/metrics/CMakeLists.txt

@@ -0,0 +1,18 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you 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.
+
+iceberg_install_all_headers(iceberg/metrics)

src/iceberg/metrics/commit_report.cc

@@ -0,0 +1,111 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 "iceberg/metrics/commit_report.h"
+
+#include "iceberg/snapshot.h"
+#include "iceberg/util/string_util.h"
+
+namespace iceberg {
+
+std::unique_ptr<CommitMetrics> CommitMetrics::Make(MetricsContext& context) {
+  auto m = std::unique_ptr<CommitMetrics>(new CommitMetrics());
+  m->total_duration = context.GetTimer("total-duration", TimerUnit::kNanoseconds);
+  m->attempts = context.GetCounter("attempts");
+  return m;
+}
+
+std::unique_ptr<CommitMetrics> CommitMetrics::Noop() {
+  return CommitMetrics::Make(*MetricsContext::Noop());
+}
+
+CommitMetricsResult CommitMetrics::ToResult() const {
+  CommitMetricsResult result;
+  if (total_duration && !total_duration->IsNoop()) {
+    result.total_duration =
+        TimerResult{.unit = std::string(total_duration->Unit()),
+                    .count = total_duration->Count(),
+                    .total_duration = total_duration->TotalDuration()};
+  }
+  if (attempts && !attempts->IsNoop()) {
+    result.attempts = CounterResult{.unit = attempts->unit(), .value = attempts->value()};
+  }
+  return result;
+}
+
+CommitMetricsResult CommitMetricsResult::From(
+    const CommitMetrics& live_metrics,
+    const std::unordered_map<std::string, std::string>& snapshot_summary) {
+  auto result = live_metrics.ToResult();
+
+  auto count_field =
+      [&snapshot_summary](const std::string& key) -> std::optional<CounterResult> {
+    auto it = snapshot_summary.find(key);
+    if (it == snapshot_summary.end()) return std::nullopt;
+    auto parsed = StringUtils::ParseNumber<int64_t>(it->second);
+    if (!parsed.has_value()) return std::nullopt;
+    return CounterResult{.unit = CounterUnit::kCount, .value = parsed.value()};
+  };
+  auto bytes_field =
+      [&snapshot_summary](const std::string& key) -> std::optional<CounterResult> {
+    auto it = snapshot_summary.find(key);
+    if (it == snapshot_summary.end()) return std::nullopt;
+    auto parsed = StringUtils::ParseNumber<int64_t>(it->second);
+    if (!parsed.has_value()) return std::nullopt;
+    return CounterResult{.unit = CounterUnit::kBytes, .value = parsed.value()};
+  };
+
+  result.added_data_files = count_field(SnapshotSummaryFields::kAddedDataFiles);
+  result.removed_data_files = count_field(SnapshotSummaryFields::kDeletedDataFiles);
+  result.total_data_files = count_field(SnapshotSummaryFields::kTotalDataFiles);
+  result.added_delete_files = count_field(SnapshotSummaryFields::kAddedDeleteFiles);
+  result.added_equality_delete_files =
+      count_field(SnapshotSummaryFields::kAddedEqDeleteFiles);
+  result.added_positional_delete_files =
+      count_field(SnapshotSummaryFields::kAddedPosDeleteFiles);
+  result.added_dvs = count_field(SnapshotSummaryFields::kAddedDVs);
+  result.removed_delete_files = count_field(SnapshotSummaryFields::kRemovedDeleteFiles);
+  result.removed_positional_delete_files =
+      count_field(SnapshotSummaryFields::kRemovedPosDeleteFiles);
+  result.removed_dvs = count_field(SnapshotSummaryFields::kRemovedDVs);
+  result.removed_equality_delete_files =
+      count_field(SnapshotSummaryFields::kRemovedEqDeleteFiles);
+  result.total_delete_files = count_field(SnapshotSummaryFields::kTotalDeleteFiles);
+  result.added_records = count_field(SnapshotSummaryFields::kAddedRecords);
+  result.removed_records = count_field(SnapshotSummaryFields::kDeletedRecords);
+  result.total_records = count_field(SnapshotSummaryFields::kTotalRecords);
+  result.added_files_size_bytes = bytes_field(SnapshotSummaryFields::kAddedFileSize);
+  result.removed_files_size_bytes = bytes_field(SnapshotSummaryFields::kRemovedFileSize);
+  result.total_files_size_bytes = bytes_field(SnapshotSummaryFields::kTotalFileSize);
+  result.added_positional_deletes = count_field(SnapshotSummaryFields::kAddedPosDeletes);
+  result.removed_positional_deletes =
+      count_field(SnapshotSummaryFields::kRemovedPosDeletes);
+  result.total_positional_deletes = count_field(SnapshotSummaryFields::kTotalPosDeletes);
+  result.added_equality_deletes = count_field(SnapshotSummaryFields::kAddedEqDeletes);
+  result.removed_equality_deletes = count_field(SnapshotSummaryFields::kRemovedEqDeletes);
+  result.total_equality_deletes = count_field(SnapshotSummaryFields::kTotalEqDeletes);
+  result.kept_manifest_count = count_field(SnapshotSummaryFields::kManifestsKept);
+  result.created_manifest_count = count_field(SnapshotSummaryFields::kManifestsCreated);
+  result.replaced_manifest_count = count_field(SnapshotSummaryFields::kManifestsReplaced);
+  result.processed_manifest_entries_count =
+      count_field(SnapshotSummaryFields::kEntriesProcessed);
+  return result;
+}
+
+}  // namespace iceberg

src/iceberg/metrics/commit_report.h

@@ -0,0 +1,159 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.
+ */
+
+#pragma once
+
+#include <memory>
+#include <optional>
+#include <string>
+#include <unordered_map>
+
+#include "iceberg/constants.h"
+#include "iceberg/iceberg_export.h"
+#include "iceberg/metrics/metrics_context.h"
+#include "iceberg/metrics/metrics_types.h"
+#include "iceberg/metrics/timer.h"
+
+namespace iceberg {
+
+// Forward declaration: CommitMetrics is defined later in this header.
+class CommitMetrics;
+
+/// \brief Immutable snapshot of commit metrics for use in CommitReport.
+///
+/// Populated by CommitMetrics::ToResult() after a commit completes. File and
+/// record counts can also be combined from the snapshot summary with From().
+struct ICEBERG_EXPORT CommitMetricsResult {
+  /// \brief Total wall-clock duration of the commit attempt.
+  std::optional<TimerResult> total_duration;
+  /// \brief Number of commit attempts (1 on success without retries).
+  std::optional<CounterResult> attempts;
+  /// \brief Number of data files added in this commit.
+  std::optional<CounterResult> added_data_files;
+  /// \brief Number of data files removed in this commit.
+  std::optional<CounterResult> removed_data_files;
+  /// \brief Total live data files after this commit.
+  std::optional<CounterResult> total_data_files;
+  /// \brief Number of delete files added in this commit.
+  std::optional<CounterResult> added_delete_files;
+  /// \brief Equality delete files added.
+  std::optional<CounterResult> added_equality_delete_files;
+  /// \brief Positional delete files added.
+  std::optional<CounterResult> added_positional_delete_files;
+  /// \brief Deletion vectors added.
+  std::optional<CounterResult> added_dvs;
+  /// \brief Number of delete files removed in this commit.
+  std::optional<CounterResult> removed_delete_files;
+  /// \brief Positional delete files removed.
+  std::optional<CounterResult> removed_positional_delete_files;
+  /// \brief Deletion vectors removed.
+  std::optional<CounterResult> removed_dvs;
+  /// \brief Equality delete files removed.
+  std::optional<CounterResult> removed_equality_delete_files;
+  /// \brief Total live delete files after this commit.
+  std::optional<CounterResult> total_delete_files;
+  /// \brief Number of records added in this commit.
+  std::optional<CounterResult> added_records;
+  /// \brief Number of records removed in this commit.
+  std::optional<CounterResult> removed_records;
+  /// \brief Total live records after this commit.
+  std::optional<CounterResult> total_records;
+  /// \brief Total byte size of files added.
+  std::optional<CounterResult> added_files_size_bytes;
+  /// \brief Total byte size of files removed.
+  std::optional<CounterResult> removed_files_size_bytes;
+  /// \brief Total byte size of all live files after this commit.
+  std::optional<CounterResult> total_files_size_bytes;
+  /// \brief Positional delete records added.
+  std::optional<CounterResult> added_positional_deletes;
+  /// \brief Positional delete records removed.
+  std::optional<CounterResult> removed_positional_deletes;
+  /// \brief Total positional delete records after this commit.
+  std::optional<CounterResult> total_positional_deletes;
+  /// \brief Equality delete records added.
+  std::optional<CounterResult> added_equality_deletes;
+  /// \brief Equality delete records removed.
+  std::optional<CounterResult> removed_equality_deletes;
+  /// \brief Total equality delete records after this commit.
+  std::optional<CounterResult> total_equality_deletes;
+  /// \brief Manifest files kept unchanged in this commit.
+  std::optional<CounterResult> kept_manifest_count;
+  /// \brief Manifest files created in this commit.
+  std::optional<CounterResult> created_manifest_count;
+  /// \brief Manifest files replaced in this commit.
+  std::optional<CounterResult> replaced_manifest_count;
+  /// \brief Manifest entries processed in this commit.
+  std::optional<CounterResult> processed_manifest_entries_count;
+
+  bool operator==(const CommitMetricsResult&) const = default;
+
+  /// \brief Build a CommitMetricsResult from live metrics and a snapshot summary map.
+  ///
+  /// Combines timer/retry measurements from \p live_metrics with records parsed
+  /// from \p snapshot_summary. Missing or unparseable summary keys are omitted.
+  static CommitMetricsResult From(
+      const CommitMetrics& live_metrics,
+      const std::unordered_map<std::string, std::string>& snapshot_summary);
+};
+
+/// \brief Live commit metrics collected during a table commit operation.
+///
+/// Tracks the overall commit duration and retry count. File/record counts come
+/// from the snapshot summary after the commit succeeds and are stored separately
+/// in CommitMetricsResult.
+class ICEBERG_EXPORT CommitMetrics {
+ public:
+  /// \brief Create a CommitMetrics instance backed by the given MetricsContext.
+  static std::unique_ptr<CommitMetrics> Make(MetricsContext& context);
+
+  /// \brief Create a CommitMetrics instance with all-noop timer and counter.
+  static std::unique_ptr<CommitMetrics> Noop();
+
+  /// \brief Snapshot current timer and counter values into a CommitMetricsResult.
+  CommitMetricsResult ToResult() const;
+
+  /// \brief Timer measuring total wall-clock time of the commit call.
+  std::shared_ptr<Timer> total_duration;
+
+  /// \brief Counter for the number of commit attempts (including retries).
+  std::shared_ptr<Counter> attempts;
+
+ private:
+  CommitMetrics() = default;
+};
+
+/// \brief Report generated after a commit operation.
+///
+/// Contains metrics about the changes made in a commit.
+struct ICEBERG_EXPORT CommitReport {
+  /// \brief The fully qualified name of the table that was modified.
+  std::string table_name;
+  /// \brief The snapshot ID created by this commit.
+  int64_t snapshot_id = kInvalidSnapshotId;
+  /// \brief The sequence number assigned to this commit.
+  int64_t sequence_number = kInvalidSequenceNumber;
+  /// \brief The operation that was performed (write, delete, etc.).
+  std::string operation;
+  /// \brief Metrics collected during the commit operation.
+  CommitMetricsResult commit_metrics;
+  /// \brief Additional key-value metadata.
+  std::unordered_map<std::string, std::string> metadata;
+};
+
+}  // namespace iceberg

src/iceberg/metrics/counter.cc

@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 "iceberg/metrics/counter.h"
+
+#include <memory>
+
+namespace iceberg {
+
+namespace {
+
+class NoopCounter final : public Counter {
+ public:
+  using Counter::Increment;
+  void Increment(int64_t) override {}
+  int64_t value() const override { return -1; }
+  CounterUnit unit() const override { return CounterUnit::kUndefined; }
+  bool IsNoop() const override { return true; }
+};
+
+}  // namespace
+
+std::shared_ptr<Counter> Counter::Noop() {
+  static std::shared_ptr<Counter> instance = std::make_shared<NoopCounter>();
+  return instance;
+}
+
+DefaultCounter::DefaultCounter(CounterUnit unit) : unit_(unit) {}
+
+void DefaultCounter::Increment(int64_t amount) {
+  count_.fetch_add(amount, std::memory_order_relaxed);
+}
+
+int64_t DefaultCounter::value() const { return count_.load(std::memory_order_relaxed); }
+
+}  // namespace iceberg