feat: metrics reporting for scan and commit

Author: Innocent

src/iceberg/CMakeLists.txt

@@ -58,6 +58,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
@@ -172,6 +179,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

@@ -75,6 +75,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',

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,40 @@
+/*
+ * 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"
+
+namespace iceberg {
+
+CommitMetrics CommitMetrics::Of(MetricsContext& context) {
+  CommitMetrics m;
+  m.total_duration = context.GetTimer("totalDuration");
+  m.attempts = context.GetCounter("attempts");
+  return m;
+}
+
+CommitMetrics CommitMetrics::Noop() { return CommitMetrics::Of(MetricsContext::Null()); }
+
+void CommitMetrics::PopulateResult(CommitMetricsResult& result) const {
+  result.total_duration = total_duration ? TimerResult{total_duration->Count(),
+                                                       total_duration->TotalDuration()}
+                                         : TimerResult{};
+  result.attempts = attempts ? attempts->Value() : 0;
+}
+
+}  // namespace iceberg

src/iceberg/metrics/commit_report.h

@@ -0,0 +1,151 @@
+/*
+ * 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 <cstdint>
+#include <memory>
+#include <string>
+#include <unordered_map>
+
+#include "iceberg/constants.h"
+#include "iceberg/iceberg_export.h"
+#include "iceberg/metrics/counter.h"
+#include "iceberg/metrics/metrics_context.h"
+#include "iceberg/metrics/scan_report.h"  // TimerResult
+#include "iceberg/metrics/timer.h"
+
+namespace iceberg {
+
+/// \brief Immutable snapshot of commit metrics for use in CommitReport.
+///
+/// Populated by CommitMetrics::ToResult() (for total_duration and attempts) and
+/// by parsing snapshot summary fields (for file/record counts).
+/// Mirrors org.apache.iceberg.metrics.CommitMetricsResult.
+struct ICEBERG_EXPORT CommitMetricsResult {
+  /// \brief Total wall-clock duration of the commit attempt (count + nanoseconds).
+  TimerResult total_duration;
+  /// \brief Number of commit attempts (1 on success without retries).
+  int64_t attempts = 0;
+  /// \brief Number of data files added in this commit.
+  int64_t added_data_files = 0;
+  /// \brief Number of data files removed in this commit.
+  int64_t removed_data_files = 0;
+  /// \brief Total live data files after this commit.
+  int64_t total_data_files = 0;
+  /// \brief Number of delete files added in this commit.
+  int64_t added_delete_files = 0;
+  /// \brief Equality delete files added.
+  int64_t added_equality_delete_files = 0;
+  /// \brief Positional delete files added.
+  int64_t added_positional_delete_files = 0;
+  /// \brief Deletion vectors added.
+  int64_t added_dvs = 0;
+  /// \brief Positional delete files removed.
+  int64_t removed_positional_delete_files = 0;
+  /// \brief Deletion vectors removed.
+  int64_t removed_dvs = 0;
+  /// \brief Equality delete files removed.
+  int64_t removed_equality_delete_files = 0;
+  /// \brief Number of delete files removed in this commit.
+  int64_t removed_delete_files = 0;
+  /// \brief Total live delete files after this commit.
+  int64_t total_delete_files = 0;
+  /// \brief Number of records added in this commit.
+  int64_t added_records = 0;
+  /// \brief Number of records removed in this commit.
+  int64_t removed_records = 0;
+  /// \brief Total live records after this commit.
+  int64_t total_records = 0;
+  /// \brief Total byte size of files added.
+  int64_t added_files_size_bytes = 0;
+  /// \brief Total byte size of files removed.
+  int64_t removed_files_size_bytes = 0;
+  /// \brief Total byte size of all live files after this commit.
+  int64_t total_files_size_bytes = 0;
+  /// \brief Positional delete records added.
+  int64_t added_positional_deletes = 0;
+  /// \brief Positional delete records removed.
+  int64_t removed_positional_deletes = 0;
+  /// \brief Total positional delete records after this commit.
+  int64_t total_positional_deletes = 0;
+  /// \brief Equality delete records added.
+  int64_t added_equality_deletes = 0;
+  /// \brief Equality delete records removed.
+  int64_t removed_equality_deletes = 0;
+  /// \brief Total equality delete records after this commit.
+  int64_t total_equality_deletes = 0;
+  /// \brief Manifest files kept unchanged in this commit.
+  int64_t kept_manifest_count = 0;
+  /// \brief Manifest files created in this commit.
+  int64_t created_manifest_count = 0;
+  /// \brief Manifest files replaced in this commit.
+  int64_t replaced_manifest_count = 0;
+  /// \brief Manifest entries processed in this commit.
+  int64_t processed_manifest_entries_count = 0;
+};
+
+/// \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.
+///
+/// Mirrors org.apache.iceberg.metrics.CommitMetrics.
+class ICEBERG_EXPORT CommitMetrics {
+ public:
+  /// \brief Create a CommitMetrics instance backed by the given MetricsContext.
+  static CommitMetrics Of(MetricsContext& context);
+
+  /// \brief Create a CommitMetrics instance with all-noop timer and counter.
+  static CommitMetrics Noop();
+
+  /// \brief Snapshot timer and counter values into the corresponding fields of result.
+  ///
+  /// Only total_duration and attempts are written; the caller is responsible for
+  /// populating the remaining snapshot-summary fields.
+  void PopulateResult(CommitMetricsResult& result) 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;
+};
+
+/// \brief Report generated after a commit operation.
+///
+/// Contains metrics about the changes made in a commit, including
+/// files added/removed and retry information.
+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 (append, overwrite, 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,51 @@
+/*
+ * 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"
+
+namespace iceberg {
+
+namespace {
+
+class NoopCounter final : public Counter {
+ public:
+  void Increment() override {}
+  void Increment(int64_t) override {}
+  int64_t Value() const override { return 0; }
+  bool IsNoop() const override { return true; }
+};
+
+}  // namespace
+
+Counter& Counter::Noop() {
+  static NoopCounter instance;
+  return instance;
+}
+
+DefaultCounter::DefaultCounter(CounterUnit unit) : unit_(unit) {}
+
+void DefaultCounter::Increment() { count_.fetch_add(1, std::memory_order_relaxed); }
+
+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