add ability to combine reporters

Author: Innocent

src/iceberg/CMakeLists.txt

@@ -58,7 +58,7 @@ set(ICEBERG_SOURCES
     manifest/v3_metadata.cc
     metadata_columns.cc
     metrics_config.cc
-    metrics_reporters.cc
+    metrics/metrics_reporters.cc
     name_mapping.cc
     partition_field.cc
     partition_spec.cc

src/iceberg/catalog/memory/in_memory_catalog.cc

@@ -22,7 +22,7 @@
 #include <algorithm>
 #include <iterator>
 
-#include "iceberg/metrics_reporters.h"
+#include "iceberg/metrics/metrics_reporters.h"
 #include "iceberg/table.h"
 #include "iceberg/table_identifier.h"
 #include "iceberg/table_metadata.h"

src/iceberg/catalog/memory/in_memory_catalog.h

@@ -22,7 +22,7 @@
 #include <shared_mutex>
 
 #include "iceberg/catalog.h"
-#include "iceberg/metrics_reporter.h"
+#include "iceberg/metrics/metrics_reporter.h"
 
 namespace iceberg {
 

src/iceberg/catalog/rest/rest_catalog.cc

@@ -37,7 +37,7 @@
 #include "iceberg/catalog/rest/rest_util.h"
 #include "iceberg/catalog/rest/types.h"
 #include "iceberg/json_serde_internal.h"
-#include "iceberg/metrics_reporters.h"
+#include "iceberg/metrics/metrics_reporters.h"
 #include "iceberg/partition_spec.h"
 #include "iceberg/result.h"
 #include "iceberg/schema.h"

src/iceberg/catalog/rest/rest_catalog.h

@@ -28,7 +28,7 @@
 #include "iceberg/catalog/rest/endpoint.h"
 #include "iceberg/catalog/rest/iceberg_rest_export.h"
 #include "iceberg/catalog/rest/type_fwd.h"
-#include "iceberg/metrics_reporter.h"
+#include "iceberg/metrics/metrics_reporter.h"
 #include "iceberg/result.h"
 
 /// \file iceberg/catalog/rest/rest_catalog.h

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,8 +75,8 @@ iceberg_sources = files(
     'manifest/v2_metadata.cc',
     'manifest/v3_metadata.cc',
     'metadata_columns.cc',
+    'metrics/metrics_reporters.cc',
     'metrics_config.cc',
-    'metrics_reporters.cc',
     'name_mapping.cc',
     'partition_field.cc',
     'partition_spec.cc',

src/iceberg/metrics/commit_report.h

@@ -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.
+ */
+
+#pragma once
+
+#include <cstdint>
+#include <string>
+#include <unordered_map>
+
+#include "iceberg/constants.h"
+#include "iceberg/iceberg_export.h"
+
+namespace iceberg {
+
+/// \brief Metrics collected during a table commit (snapshot creation).
+///
+struct ICEBERG_EXPORT CommitMetrics {
+  /// \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 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.
+  CommitMetrics commit_metrics;
+  /// \brief Additional key-value metadata.
+  std::unordered_map<std::string, std::string> metadata;
+};
+
+}  // namespace iceberg

src/iceberg/metrics/metrics_reporter.h

@@ -0,0 +1,89 @@
+/*
+ * 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 <string_view>
+#include <variant>
+
+#include "iceberg/iceberg_export.h"
+#include "iceberg/metrics/commit_report.h"
+#include "iceberg/metrics/scan_report.h"
+
+namespace iceberg {
+
+/// \brief The type of a metrics report.
+enum class MetricsReportType {
+  kScanReport,
+  kCommitReport,
+};
+
+/// \brief Get the string representation of a metrics report type.
+ICEBERG_EXPORT constexpr std::string_view ToString(MetricsReportType type) noexcept {
+  switch (type) {
+    case MetricsReportType::kScanReport:
+      return "scan";
+    case MetricsReportType::kCommitReport:
+      return "commit";
+  }
+  std::unreachable();
+}
+
+/// \brief A metrics report, which can be either a ScanReport or CommitReport.
+///
+/// This variant type allows handling both report types uniformly through
+/// the MetricsReporter interface.
+using MetricsReport = std::variant<ScanReport, CommitReport>;
+
+/// \brief Get the type of a metrics report.
+///
+/// \param report The metrics report to get the type of.
+/// \return The type of the metrics report.
+ICEBERG_EXPORT inline MetricsReportType GetReportType(const MetricsReport& report) {
+  return std::visit(
+      [](const auto& r) -> MetricsReportType {
+        using T = std::decay_t<decltype(r)>;
+        if constexpr (std::is_same_v<T, ScanReport>) {
+          return MetricsReportType::kScanReport;
+        } else {
+          return MetricsReportType::kCommitReport;
+        }
+      },
+      report);
+}
+
+/// \brief Interface for reporting metrics from Iceberg operations.
+///
+/// Implementations of this interface can be used to collect and report
+/// metrics about scan and commit operations. Common implementations include
+/// logging reporters, metrics collectors, and the noop reporter for testing.
+class ICEBERG_EXPORT MetricsReporter {
+ public:
+  virtual ~MetricsReporter() = default;
+
+  /// \brief Report a metrics report.
+  ///
+  /// Implementations should handle the report according to their purpose
+  /// (e.g., logging, sending to a metrics service, etc.).
+  ///
+  /// \param report The metrics report to process.
+  virtual void Report(const MetricsReport& report) = 0;
+};
+
+}  // namespace iceberg

src/iceberg/metrics_reporters.cc src/iceberg/metrics/metrics_reporters.ccsrc/iceberg/metrics_reporters.cc renamed to src/iceberg/metrics/metrics_reporters.cc

@@ -17,8 +17,9 @@
  * under the License.
  */
 
-#include "iceberg/metrics_reporters.h"
+#include "iceberg/metrics/metrics_reporters.h"
 
+#include <iostream>
 #include <unordered_set>
 
 #include "iceberg/util/string_util.h"
@@ -27,10 +28,10 @@ namespace iceberg {
 
 namespace {
 
-/// \brief Registry type for MetricsReporter factories with heterogeneous lookup support.
+/// \brief Registry type for MetricsReporter factories.
 using MetricsReporterRegistry = std::unordered_map<std::string, MetricsReporterFactory>;
 
-/// \brief Get the set of known metrics reporter types.
+/// \brief Get the set of known built-in metrics reporter types.
 const std::unordered_set<std::string>& DefaultReporterTypes() {
   static const std::unordered_set<std::string> kReporterTypes = {
       std::string(kMetricsReporterTypeNoop),
@@ -45,48 +46,63 @@ std::string InferReporterType(
   if (it != properties.end() && !it->second.empty()) {
     return StringUtils::ToLower(it->second);
   }
-  // Default to noop reporter
   return std::string(kMetricsReporterTypeNoop);
 }
 
 /// \brief Metrics reporter that does nothing.
-///
-/// This is the default reporter used when no reporter is configured.
-/// It silently discards all reports.
 class NoopMetricsReporter : public MetricsReporter {
  public:
   static Result<std::unique_ptr<MetricsReporter>> Make(
       [[maybe_unused]] const std::unordered_map<std::string, std::string>& properties) {
     return std::make_unique<NoopMetricsReporter>();
   }
 
-  void Report([[maybe_unused]] const MetricsReport& report) override {
-    // Intentionally empty - noop implementation discards all reports
-  }
+  void Report([[maybe_unused]] const MetricsReport& report) override {}
 };
 
-/// \brief Template helper to create factory functions for reporter types.
 template <typename T>
 MetricsReporterFactory MakeReporterFactory() {
   return [](const std::unordered_map<std::string, std::string>& props)
              -> Result<std::unique_ptr<MetricsReporter>> { return T::Make(props); };
 }
 
-/// \brief Create the default registry with built-in reporters.
 MetricsReporterRegistry CreateDefaultRegistry() {
   return {
       {std::string(kMetricsReporterTypeNoop), MakeReporterFactory<NoopMetricsReporter>()},
   };
 }
 
-/// \brief Get the global registry of metrics reporter factories.
 MetricsReporterRegistry& GetRegistry() {
   static MetricsReporterRegistry registry = CreateDefaultRegistry();
   return registry;
 }
 
 }  // namespace
 
+// --- CompositeMetricsReporter ---
+
+CompositeMetricsReporter::CompositeMetricsReporter(
+    std::unordered_set<std::shared_ptr<MetricsReporter>> reporters)
+    : reporters_(std::move(reporters)) {}
+
+void CompositeMetricsReporter::Report(const MetricsReport& report) {
+  for (const auto& reporter : reporters_) {
+    try {
+      reporter->Report(report);
+    } catch (...) {
+      // Catch all exceptions to ensure one failing reporter doesn't prevent others from
+      // receiving the report.
+    }
+  }
+}
+
+const std::unordered_set<std::shared_ptr<MetricsReporter>>&
+CompositeMetricsReporter::Reporters() const {
+  return reporters_;
+}
+
+// --- MetricsReporters ---
+
 void MetricsReporters::Register(std::string_view reporter_type,
                                 MetricsReporterFactory factory) {
   GetRegistry()[StringUtils::ToLower(reporter_type)] = std::move(factory);
@@ -109,4 +125,29 @@ Result<std::unique_ptr<MetricsReporter>> MetricsReporters::Load(
   return it->second(properties);
 }
 
+std::shared_ptr<MetricsReporter> MetricsReporters::Combine(
+    std::shared_ptr<MetricsReporter> first, std::shared_ptr<MetricsReporter> second) {
+  if (!first) return second;
+  if (!second || first.get() == second.get()) return first;
+
+  // Single-pass collection: insert into the set to flatten nested composites
+  // and deduplicate by shared_ptr identity simultaneously.
+  std::unordered_set<std::shared_ptr<MetricsReporter>> reporters;
+
+  auto collect = [&reporters](const std::shared_ptr<MetricsReporter>& r) {
+    if (auto* composite = dynamic_cast<CompositeMetricsReporter*>(r.get())) {
+      for (const auto& inner : composite->Reporters()) {
+        reporters.insert(inner);
+      }
+    } else {
+      reporters.insert(r);
+    }
+  };
+
+  collect(first);
+  collect(second);
+
+  return std::make_shared<CompositeMetricsReporter>(std::move(reporters));
+}
+
 }  // namespace iceberg