only focus on metrics definition and core components

Author: Innocent

src/iceberg/CMakeLists.txt

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

src/iceberg/catalog/memory/in_memory_catalog.cc

@@ -22,7 +22,6 @@
 #include <algorithm>
 #include <iterator>
 
-#include "iceberg/metrics/metrics_reporters.h"
 #include "iceberg/table.h"
 #include "iceberg/table_identifier.h"
 #include "iceberg/table_metadata.h"
@@ -352,12 +351,7 @@ InMemoryCatalog::InMemoryCatalog(
       properties_(std::move(properties)),
       file_io_(std::move(file_io)),
       warehouse_location_(std::move(warehouse_location)),
-      root_namespace_(std::make_unique<InMemoryNamespace>()) {
-  auto reporter_result = MetricsReporters::Load(properties_);
-  if (reporter_result.has_value()) {
-    reporter_ = std::move(reporter_result.value());
-  }
-}
+      root_namespace_(std::make_unique<InMemoryNamespace>()) {}
 
 InMemoryCatalog::~InMemoryCatalog() = default;
 
@@ -434,8 +428,7 @@ Result<std::shared_ptr<Table>> InMemoryCatalog::CreateTable(
   ICEBERG_RETURN_UNEXPECTED(
       root_namespace_->UpdateTableMetadataLocation(identifier, metadata_file_location));
   return Table::Make(identifier, std::move(table_metadata),
-                     std::move(metadata_file_location), file_io_, shared_from_this(),
-                     reporter_);
+                     std::move(metadata_file_location), file_io_, shared_from_this());
 }
 
 Result<std::shared_ptr<Table>> InMemoryCatalog::UpdateTable(
@@ -486,7 +479,7 @@ Result<std::shared_ptr<Table>> InMemoryCatalog::UpdateTable(
   TableMetadataUtil::DeleteRemovedMetadataFiles(*file_io_, base.get(), *updated);
 
   return Table::Make(identifier, std::move(updated), std::move(new_metadata_location),
-                     file_io_, shared_from_this(), reporter_);
+                     file_io_, shared_from_this());
 }
 
 Result<std::shared_ptr<Transaction>> InMemoryCatalog::StageCreateTable(
@@ -507,7 +500,7 @@ Result<std::shared_ptr<Transaction>> InMemoryCatalog::StageCreateTable(
       TableMetadata::Make(*schema, *spec, *order, base_location, properties));
   ICEBERG_ASSIGN_OR_RAISE(
       auto table, StagedTable::Make(identifier, std::move(table_metadata), "", file_io_,
-                                    shared_from_this(), reporter_));
+                                    shared_from_this()));
   return Transaction::Make(std::move(table), TransactionKind::kCreate);
 }
 
@@ -544,7 +537,7 @@ Result<std::shared_ptr<Table>> InMemoryCatalog::LoadTable(
   ICEBERG_ASSIGN_OR_RAISE(auto metadata,
                           TableMetadataUtil::Read(*file_io_, metadata_location));
   return Table::Make(identifier, std::move(metadata), std::move(metadata_location),
-                     file_io_, shared_from_this(), reporter_);
+                     file_io_, shared_from_this());
 }
 
 Result<std::shared_ptr<Table>> InMemoryCatalog::RegisterTable(
@@ -564,7 +557,7 @@ Result<std::shared_ptr<Table>> InMemoryCatalog::RegisterTable(
     return UnknownError("The registry failed.");
   }
   return Table::Make(identifier, std::move(metadata), metadata_file_location, file_io_,
-                     shared_from_this(), reporter_);
+                     shared_from_this());
 }
 
 }  // namespace iceberg

src/iceberg/catalog/memory/in_memory_catalog.h

@@ -22,7 +22,6 @@
 #include <shared_mutex>
 
 #include "iceberg/catalog.h"
-#include "iceberg/metrics/metrics_reporter.h"
 
 namespace iceberg {
 
@@ -106,7 +105,6 @@ class ICEBERG_EXPORT InMemoryCatalog
   std::shared_ptr<FileIO> file_io_;
   std::string warehouse_location_;
   std::unique_ptr<class InMemoryNamespace> root_namespace_;
-  std::shared_ptr<MetricsReporter> reporter_;
   mutable std::shared_mutex mutex_;
 };
 

src/iceberg/catalog/rest/rest_catalog.cc

@@ -37,7 +37,6 @@
 #include "iceberg/catalog/rest/rest_util.h"
 #include "iceberg/catalog/rest/types.h"
 #include "iceberg/json_serde_internal.h"
-#include "iceberg/metrics/metrics_reporters.h"
 #include "iceberg/partition_spec.h"
 #include "iceberg/result.h"
 #include "iceberg/schema.h"
@@ -169,17 +168,10 @@ Result<std::shared_ptr<RestCatalog>> RestCatalog::Make(
   ICEBERG_ASSIGN_OR_RAISE(auto catalog_session,
                           auth_manager->CatalogSession(*client, final_config.configs()));
 
-  // Load metrics reporter from catalog properties
-  std::shared_ptr<MetricsReporter> reporter;
-  auto reporter_result = MetricsReporters::Load(final_config.configs());
-  if (reporter_result.has_value()) {
-    reporter = std::move(reporter_result.value());
-  }
-
   return std::shared_ptr<RestCatalog>(
       new RestCatalog(std::move(final_config), std::move(file_io), std::move(client),
                       std::move(paths), std::move(endpoints), std::move(auth_manager),
-                      std::move(catalog_session), std::move(reporter), snapshot_mode));
+                      std::move(catalog_session), snapshot_mode));
 }
 
 RestCatalog::RestCatalog(RestCatalogProperties config, std::shared_ptr<FileIO> file_io,
@@ -188,7 +180,6 @@ RestCatalog::RestCatalog(RestCatalogProperties config, std::shared_ptr<FileIO> f
                          std::unordered_set<Endpoint> endpoints,
                          std::unique_ptr<auth::AuthManager> auth_manager,
                          std::shared_ptr<auth::AuthSession> catalog_session,
-                         std::shared_ptr<MetricsReporter> reporter,
                          SnapshotMode snapshot_mode)
     : config_(std::move(config)),
       file_io_(std::move(file_io)),
@@ -198,7 +189,6 @@ RestCatalog::RestCatalog(RestCatalogProperties config, std::shared_ptr<FileIO> f
       supported_endpoints_(std::move(endpoints)),
       auth_manager_(std::move(auth_manager)),
       catalog_session_(std::move(catalog_session)),
-      reporter_(std::move(reporter)),
       snapshot_mode_(snapshot_mode) {
   ICEBERG_DCHECK(catalog_session_ != nullptr, "catalog_session must not be null");
 }
@@ -365,8 +355,7 @@ Result<std::shared_ptr<Table>> RestCatalog::CreateTable(
                           CreateTableInternal(identifier, schema, spec, order, location,
                                               properties, /*stage_create=*/false));
   return Table::Make(identifier, std::move(result.metadata),
-                     std::move(result.metadata_location), file_io_, shared_from_this(),
-                     reporter_);
+                     std::move(result.metadata_location), file_io_, shared_from_this());
 }
 
 Result<std::shared_ptr<Table>> RestCatalog::UpdateTable(
@@ -397,7 +386,7 @@ Result<std::shared_ptr<Table>> RestCatalog::UpdateTable(
 
   return Table::Make(identifier, std::move(commit_response.metadata),
                      std::move(commit_response.metadata_location), file_io_,
-                     shared_from_this(), reporter_);
+                     shared_from_this());
 }
 
 Result<std::shared_ptr<Transaction>> RestCatalog::StageCreateTable(
@@ -411,7 +400,7 @@ Result<std::shared_ptr<Transaction>> RestCatalog::StageCreateTable(
   ICEBERG_ASSIGN_OR_RAISE(auto staged_table,
                           StagedTable::Make(identifier, std::move(result.metadata),
                                             std::move(result.metadata_location), file_io_,
-                                            shared_from_this(), reporter_));
+                                            shared_from_this()));
   return Transaction::Make(std::move(staged_table), TransactionKind::kCreate);
 }
 
@@ -480,7 +469,7 @@ Result<std::shared_ptr<Table>> RestCatalog::LoadTable(const TableIdentifier& ide
   ICEBERG_ASSIGN_OR_RAISE(auto load_result, LoadTableResultFromJson(json));
   return Table::Make(identifier, std::move(load_result.metadata),
                      std::move(load_result.metadata_location), file_io_,
-                     shared_from_this(), reporter_);
+                     shared_from_this());
 }
 
 Result<std::shared_ptr<Table>> RestCatalog::RegisterTable(
@@ -503,7 +492,7 @@ Result<std::shared_ptr<Table>> RestCatalog::RegisterTable(
   ICEBERG_ASSIGN_OR_RAISE(auto load_result, LoadTableResultFromJson(json));
   return Table::Make(identifier, std::move(load_result.metadata),
                      std::move(load_result.metadata_location), file_io_,
-                     shared_from_this(), reporter_);
+                     shared_from_this());
 }
 
 }  // namespace iceberg::rest

src/iceberg/catalog/rest/rest_catalog.h

@@ -28,7 +28,6 @@
 #include "iceberg/catalog/rest/endpoint.h"
 #include "iceberg/catalog/rest/iceberg_rest_export.h"
 #include "iceberg/catalog/rest/type_fwd.h"
-#include "iceberg/metrics/metrics_reporter.h"
 #include "iceberg/result.h"
 
 /// \file iceberg/catalog/rest/rest_catalog.h
@@ -111,7 +110,7 @@ class ICEBERG_REST_EXPORT RestCatalog : public Catalog,
               std::unordered_set<Endpoint> endpoints,
               std::unique_ptr<auth::AuthManager> auth_manager,
               std::shared_ptr<auth::AuthSession> catalog_session,
-              std::shared_ptr<MetricsReporter> reporter, SnapshotMode snapshot_mode);
+              SnapshotMode snapshot_mode);
 
   Result<std::string> LoadTableInternal(const TableIdentifier& identifier) const;
 
@@ -129,7 +128,6 @@ class ICEBERG_REST_EXPORT RestCatalog : public Catalog,
   std::unordered_set<Endpoint> supported_endpoints_;
   std::unique_ptr<auth::AuthManager> auth_manager_;
   std::shared_ptr<auth::AuthSession> catalog_session_;
-  std::shared_ptr<MetricsReporter> reporter_;
   SnapshotMode snapshot_mode_;
 };
 

src/iceberg/meson.build

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

@@ -20,17 +20,29 @@
 #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 Metrics collected during a table commit (snapshot creation).
+/// \brief Immutable snapshot of commit metrics for use in CommitReport.
 ///
-struct ICEBERG_EXPORT CommitMetrics {
+/// 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.
@@ -89,6 +101,34 @@ struct ICEBERG_EXPORT CommitMetrics {
   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
@@ -103,7 +143,7 @@ struct ICEBERG_EXPORT CommitReport {
   /// \brief The operation that was performed (append, overwrite, delete, etc.).
   std::string operation;
   /// \brief Metrics collected during the commit operation.
-  CommitMetrics commit_metrics;
+  CommitMetricsResult commit_metrics;
   /// \brief Additional key-value metadata.
   std::unordered_map<std::string, std::string> metadata;
 };

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