feat:metrics reporting integration

Author: Innocent

.gitignore

@@ -16,6 +16,9 @@
 # under the License.
 
 build/
+build_bundle/
+build_core/
+builddir/
 cmake-build/
 cmake-build-debug/
 cmake-build-release/

src/iceberg/catalog/memory/in_memory_catalog.cc

@@ -22,6 +22,7 @@
 #include <algorithm>
 #include <iterator>
 
+#include "iceberg/metrics/metrics_reporters.h"
 #include "iceberg/table.h"
 #include "iceberg/table_identifier.h"
 #include "iceberg/table_metadata.h"
@@ -351,7 +352,15 @@ InMemoryCatalog::InMemoryCatalog(
       properties_(std::move(properties)),
       file_io_(std::move(file_io)),
       warehouse_location_(std::move(warehouse_location)),
-      root_namespace_(std::make_unique<InMemoryNamespace>()) {}
+      root_namespace_(std::make_unique<InMemoryNamespace>()) {
+  auto it = properties_.find(std::string(kMetricsReporterImpl));
+  if (it != properties_.end() && !it->second.empty() &&
+      it->second != kMetricsReporterTypeNoop) {
+    if (auto r = MetricsReporters::Load(properties_); r.has_value()) {
+      reporter_ = std::shared_ptr<MetricsReporter>(std::move(r.value()));
+    }
+  }
+}
 
 InMemoryCatalog::~InMemoryCatalog() = default;
 
@@ -428,7 +437,8 @@ 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());
+                     std::move(metadata_file_location), file_io_, shared_from_this(),
+                     reporter_);
 }
 
 Result<std::shared_ptr<Table>> InMemoryCatalog::UpdateTable(
@@ -479,7 +489,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());
+                     file_io_, shared_from_this(), reporter_);
 }
 
 Result<std::shared_ptr<Transaction>> InMemoryCatalog::StageCreateTable(
@@ -500,7 +510,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()));
+                                    shared_from_this(), reporter_));
   return Transaction::Make(std::move(table), TransactionKind::kCreate);
 }
 
@@ -537,7 +547,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());
+                     file_io_, shared_from_this(), reporter_);
 }
 
 Result<std::shared_ptr<Table>> InMemoryCatalog::RegisterTable(
@@ -557,7 +567,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());
+                     shared_from_this(), reporter_);
 }
 
 }  // namespace iceberg

src/iceberg/catalog/memory/in_memory_catalog.h

@@ -19,12 +19,15 @@
 
 #pragma once
 
+#include <memory>
 #include <shared_mutex>
 
 #include "iceberg/catalog.h"
 
 namespace iceberg {
 
+class MetricsReporter;
+
 /**
  * @brief An in-memory implementation of the Iceberg Catalog interface.
  *
@@ -106,6 +109,7 @@ class ICEBERG_EXPORT InMemoryCatalog
   std::string warehouse_location_;
   std::unique_ptr<class InMemoryNamespace> root_namespace_;
   mutable std::shared_mutex mutex_;
+  std::shared_ptr<MetricsReporter> reporter_;
 };
 
 }  // namespace iceberg

src/iceberg/catalog/rest/CMakeLists.txt

@@ -32,6 +32,7 @@ set(ICEBERG_REST_SOURCES
     resource_paths.cc
     rest_catalog.cc
     rest_file_io.cc
+    rest_metrics_reporter.cc
     rest_util.cc
     types.cc)
 

src/iceberg/catalog/rest/catalog_properties.h

@@ -55,6 +55,12 @@ class ICEBERG_REST_EXPORT RestCatalogProperties
   inline static Entry<std::string> kNamespaceSeparator{"namespace-separator", "%1F"};
   /// \brief The snapshot loading mode (ALL or REFS).
   inline static Entry<std::string> kSnapshotLoadingMode{"snapshot-loading-mode", "ALL"};
+  /// \brief Whether to report metrics to the REST catalog server (default: true).
+  ///
+  /// When true and the server advertises the ReportMetrics endpoint, RestCatalog
+  /// automatically POSTs scan and commit reports to the per-table metrics endpoint.
+  inline static Entry<std::string> kMetricsReportingEnabled{
+      "rest-metrics-reporting-enabled", "true"};
   /// \brief The prefix for HTTP headers.
   inline static constexpr std::string_view kHeaderPrefix = "header.";
 

src/iceberg/catalog/rest/meson.build

@@ -30,6 +30,7 @@ iceberg_rest_sources = files(
     'resource_paths.cc',
     'rest_catalog.cc',
     'rest_file_io.cc',
+    'rest_metrics_reporter.cc',
     'rest_util.cc',
     'types.cc',
 )

src/iceberg/catalog/rest/rest_catalog.cc

@@ -36,9 +36,11 @@
 #include "iceberg/catalog/rest/json_serde_internal.h"
 #include "iceberg/catalog/rest/resource_paths.h"
 #include "iceberg/catalog/rest/rest_file_io.h"
+#include "iceberg/catalog/rest/rest_metrics_reporter.h"
 #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"
@@ -48,6 +50,7 @@
 #include "iceberg/table_update.h"
 #include "iceberg/transaction.h"
 #include "iceberg/util/macros.h"
+#include "iceberg/util/string_util.h"
 
 namespace iceberg::rest {
 
@@ -171,7 +174,7 @@ Result<std::shared_ptr<RestCatalog>> RestCatalog::Make(
   // Get snapshot loading mode
   ICEBERG_ASSIGN_OR_RAISE(auto snapshot_mode, final_config.SnapshotLoadingMode());
 
-  auto client = std::make_unique<HttpClient>(final_config.ExtractHeaders());
+  auto client = std::make_shared<HttpClient>(final_config.ExtractHeaders());
   ICEBERG_ASSIGN_OR_RAISE(auto catalog_session,
                           auth_manager->CatalogSession(*client, final_config.configs()));
 
@@ -185,7 +188,7 @@ Result<std::shared_ptr<RestCatalog>> RestCatalog::Make(
 }
 
 RestCatalog::RestCatalog(RestCatalogProperties config, std::shared_ptr<FileIO> file_io,
-                         std::unique_ptr<HttpClient> client,
+                         std::shared_ptr<HttpClient> client,
                          std::unique_ptr<ResourcePaths> paths,
                          std::unordered_set<Endpoint> endpoints,
                          std::unique_ptr<auth::AuthManager> auth_manager,
@@ -201,10 +204,33 @@ RestCatalog::RestCatalog(RestCatalogProperties config, std::shared_ptr<FileIO> f
       catalog_session_(std::move(catalog_session)),
       snapshot_mode_(snapshot_mode) {
   ICEBERG_DCHECK(catalog_session_ != nullptr, "catalog_session must not be null");
+  const auto& props = config_.configs();
+  auto it = props.find(std::string(kMetricsReporterImpl));
+  if (it != props.end() && !it->second.empty() &&
+      it->second != kMetricsReporterTypeNoop) {
+    if (auto r = MetricsReporters::Load(props); r.has_value()) {
+      reporter_ = std::shared_ptr<MetricsReporter>(std::move(r.value()));
+    }
+  }
 }
 
 std::string_view RestCatalog::name() const { return name_; }
 
+std::shared_ptr<MetricsReporter> RestCatalog::MakeTableReporter(
+    const TableIdentifier& identifier) const {
+  auto enabled = config_.Get(RestCatalogProperties::kMetricsReportingEnabled);
+  if (StringUtils::ToLower(enabled) == "true" &&
+      supported_endpoints_.contains(Endpoint::ReportMetrics())) {
+    auto path = paths_->Metrics(identifier);
+    if (path.has_value()) {
+      auto rest_reporter =
+          std::make_shared<RestMetricsReporter>(client_, *path, catalog_session_);
+      return MetricsReporters::Combine(reporter_, rest_reporter);
+    }
+  }
+  return reporter_;
+}
+
 Result<std::vector<Namespace>> RestCatalog::ListNamespaces(const Namespace& ns) const {
   ICEBERG_ENDPOINT_CHECK(supported_endpoints_, Endpoint::ListNamespaces());
   ICEBERG_ASSIGN_OR_RAISE(auto path, paths_->Namespaces());
@@ -367,7 +393,8 @@ 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());
+                     std::move(result.metadata_location), file_io_, shared_from_this(),
+                     MakeTableReporter(identifier));
 }
 
 Result<std::shared_ptr<Table>> RestCatalog::UpdateTable(
@@ -398,7 +425,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());
+                     shared_from_this(), MakeTableReporter(identifier));
 }
 
 Result<std::shared_ptr<Transaction>> RestCatalog::StageCreateTable(
@@ -409,10 +436,11 @@ Result<std::shared_ptr<Transaction>> RestCatalog::StageCreateTable(
   ICEBERG_ASSIGN_OR_RAISE(auto result,
                           CreateTableInternal(identifier, schema, spec, order, location,
                                               properties, /*stage_create=*/true));
-  ICEBERG_ASSIGN_OR_RAISE(auto staged_table,
-                          StagedTable::Make(identifier, std::move(result.metadata),
-                                            std::move(result.metadata_location), file_io_,
-                                            shared_from_this()));
+  ICEBERG_ASSIGN_OR_RAISE(
+      auto staged_table,
+      StagedTable::Make(identifier, std::move(result.metadata),
+                        std::move(result.metadata_location), file_io_, shared_from_this(),
+                        MakeTableReporter(identifier)));
   return Transaction::Make(std::move(staged_table), TransactionKind::kCreate);
 }
 
@@ -480,9 +508,11 @@ Result<std::shared_ptr<Table>> RestCatalog::LoadTable(const TableIdentifier& ide
   ICEBERG_ASSIGN_OR_RAISE(auto json, FromJsonString(body));
   ICEBERG_ASSIGN_OR_RAISE(auto load_result, LoadTableResultFromJson(json));
   /// FIXME: support per-table FileIO creation
+  /// FIXME: support per-table auth session (currently uses catalog-level
+  /// catalog_session_)
   return Table::Make(identifier, std::move(load_result.metadata),
                      std::move(load_result.metadata_location), file_io_,
-                     shared_from_this());
+                     shared_from_this(), MakeTableReporter(identifier));
 }
 
 Result<std::shared_ptr<Table>> RestCatalog::RegisterTable(
@@ -505,7 +535,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());
+                     shared_from_this(), MakeTableReporter(identifier));
 }
 
 }  // namespace iceberg::rest

src/iceberg/catalog/rest/rest_catalog.h

@@ -30,6 +30,10 @@
 #include "iceberg/catalog/rest/type_fwd.h"
 #include "iceberg/result.h"
 
+namespace iceberg {
+class MetricsReporter;
+}  // namespace iceberg
+
 /// \file iceberg/catalog/rest/rest_catalog.h
 /// RestCatalog implementation for Iceberg REST API.
 
@@ -101,14 +105,23 @@ class ICEBERG_REST_EXPORT RestCatalog : public Catalog,
 
  private:
   RestCatalog(RestCatalogProperties config, std::shared_ptr<FileIO> file_io,
-              std::unique_ptr<HttpClient> client, std::unique_ptr<ResourcePaths> paths,
+              std::shared_ptr<HttpClient> client, std::unique_ptr<ResourcePaths> paths,
               std::unordered_set<Endpoint> endpoints,
               std::unique_ptr<auth::AuthManager> auth_manager,
               std::shared_ptr<auth::AuthSession> catalog_session,
               SnapshotMode snapshot_mode);
 
   Result<std::string> LoadTableInternal(const TableIdentifier& identifier) const;
 
+  /// \brief Build the per-table metrics reporter.
+  ///
+  /// When rest-metrics-reporting-enabled is true and the server advertises the
+  /// ReportMetrics endpoint, returns a CompositeMetricsReporter combining configured
+  /// reporter with a RestMetricsReporter targeting this table. Otherwise returns the
+  /// configured reporter.
+  std::shared_ptr<MetricsReporter> MakeTableReporter(
+      const TableIdentifier& identifier) const;
+
   Result<LoadTableResult> CreateTableInternal(
       const TableIdentifier& identifier, const std::shared_ptr<Schema>& schema,
       const std::shared_ptr<PartitionSpec>& spec, const std::shared_ptr<SortOrder>& order,
@@ -117,13 +130,14 @@ class ICEBERG_REST_EXPORT RestCatalog : public Catalog,
 
   RestCatalogProperties config_;
   std::shared_ptr<FileIO> file_io_;
-  std::unique_ptr<HttpClient> client_;
+  std::shared_ptr<HttpClient> client_;
   std::unique_ptr<ResourcePaths> paths_;
   std::string name_;
   std::unordered_set<Endpoint> supported_endpoints_;
   std::unique_ptr<auth::AuthManager> auth_manager_;
   std::shared_ptr<auth::AuthSession> catalog_session_;
   SnapshotMode snapshot_mode_;
+  std::shared_ptr<MetricsReporter> reporter_;
 };
 
 }  // namespace iceberg::rest

src/iceberg/catalog/rest/rest_metrics_reporter.cc

@@ -0,0 +1,69 @@
+/*
+ * 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/catalog/rest/rest_metrics_reporter.h"
+
+#include <cstdio>
+#include <utility>
+
+#include <nlohmann/json.hpp>
+
+#include "iceberg/catalog/rest/auth/auth_session.h"
+#include "iceberg/catalog/rest/error_handlers.h"
+#include "iceberg/catalog/rest/http_client.h"
+#include "iceberg/metrics/json_serde_internal.h"
+#include "iceberg/metrics/metrics_reporter.h"
+
+namespace iceberg::rest {
+
+namespace {
+
+constexpr std::string_view kReportType = "report-type";
+constexpr std::string_view kScanReportType = "scan-report";
+constexpr std::string_view kCommitReportType = "commit-report";
+
+}  // namespace
+
+RestMetricsReporter::RestMetricsReporter(std::shared_ptr<HttpClient> client,
+                                         std::string metrics_endpoint,
+                                         std::shared_ptr<auth::AuthSession> session)
+    : client_(std::move(client)),
+      metrics_endpoint_(std::move(metrics_endpoint)),
+      session_(std::move(session)) {}
+
+Status RestMetricsReporter::Report(const MetricsReport& report) {
+  // Serialize the report variant to JSON.
+  Result<nlohmann::json> json_result = std::visit(
+      [](const auto& r) -> Result<nlohmann::json> { return ToJson(r); }, report);
+  if (!json_result) {
+    return {};
+  }
+
+  // Inject "report-type" required by the REST spec (not included in core ToJson).
+  auto& json = json_result.value();
+  json[kReportType] =
+      std::holds_alternative<ScanReport>(report) ? kScanReportType : kCommitReportType;
+
+  // POST to the metrics endpoint; suppress errors to match Java fire-and-forget behavior.
+  std::ignore = client_->Post(metrics_endpoint_, json.dump(), /*headers=*/{},
+                              *DefaultErrorHandler::Instance(), *session_);
+  return {};
+}
+
+}  // namespace iceberg::rest