feat: metrics implementation

Author: Innocent

src/iceberg/catalog/memory/in_memory_catalog.cc

@@ -22,6 +22,7 @@
 #include <algorithm>
 #include <iterator>
 
+#include "iceberg/metrics_reporters.h"
 #include "iceberg/table.h"
 #include "iceberg/table_identifier.h"
 #include "iceberg/table_metadata.h"
@@ -351,7 +352,12 @@ 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 reporter_result = MetricsReporters::Load(properties_);
+  if (reporter_result.has_value()) {
+    reporter_ = std::move(reporter_result.value());
+  }
+}
 
 InMemoryCatalog::~InMemoryCatalog() = default;
 
@@ -428,7 +434,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 +486,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 +507,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 +544,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 +564,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

@@ -22,6 +22,7 @@
 #include <shared_mutex>
 
 #include "iceberg/catalog.h"
+#include "iceberg/metrics_reporter.h"
 
 namespace iceberg {
 
@@ -105,6 +106,7 @@ 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,6 +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/partition_spec.h"
 #include "iceberg/result.h"
 #include "iceberg/schema.h"
@@ -168,10 +169,17 @@ 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), snapshot_mode));
+                      std::move(catalog_session), std::move(reporter), snapshot_mode));
 }
 
 RestCatalog::RestCatalog(RestCatalogProperties config, std::shared_ptr<FileIO> file_io,
@@ -180,6 +188,7 @@ 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)),
@@ -189,6 +198,7 @@ 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");
 }
@@ -355,7 +365,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(),
+                     reporter_);
 }
 
 Result<std::shared_ptr<Table>> RestCatalog::UpdateTable(
@@ -386,7 +397,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(), reporter_);
 }
 
 Result<std::shared_ptr<Transaction>> RestCatalog::StageCreateTable(
@@ -400,7 +411,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()));
+                                            shared_from_this(), reporter_));
   return Transaction::Make(std::move(staged_table), TransactionKind::kCreate);
 }
 
@@ -469,7 +480,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());
+                     shared_from_this(), reporter_);
 }
 
 Result<std::shared_ptr<Table>> RestCatalog::RegisterTable(
@@ -492,7 +503,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(), reporter_);
 }
 
 }  // namespace iceberg::rest

src/iceberg/catalog/rest/rest_catalog.h

@@ -28,6 +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/result.h"
 
 /// \file iceberg/catalog/rest/rest_catalog.h
@@ -110,6 +111,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);
 
   Result<std::string> LoadTableInternal(const TableIdentifier& identifier) const;
@@ -128,7 +130,9 @@ 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_;
+  
 };
 
 }  // namespace iceberg::rest

src/iceberg/manifest/manifest_group.cc

@@ -321,30 +321,36 @@ ManifestGroup::ReadEntries() {
     ICEBERG_ASSIGN_OR_RAISE(bool should_match, manifest_evaluator->Evaluate(manifest));
     if (!should_match) {
       // Skip this manifest because it doesn't match partition filter
+      scan_counters_.skipped_data_manifests++;
       continue;
     }
 
     if (ignore_deleted_) {
       // only scan manifests that have entries other than deletes
       if (!manifest.has_added_files() && !manifest.has_existing_files()) {
+        scan_counters_.skipped_data_manifests++;
         continue;
       }
     }
 
     if (ignore_existing_) {
       // only scan manifests that have entries other than existing
       if (!manifest.has_added_files() && !manifest.has_deleted_files()) {
+        scan_counters_.skipped_data_manifests++;
         continue;
       }
     }
 
+    scan_counters_.scanned_data_manifests++;
+
     // Read manifest entries
     ICEBERG_ASSIGN_OR_RAISE(auto reader, MakeReader(manifest));
     ICEBERG_ASSIGN_OR_RAISE(auto entries,
                             ignore_deleted_ ? reader->LiveEntries() : reader->Entries());
 
     for (auto& entry : entries) {
       if (ignore_existing_ && entry.status == ManifestStatus::kExisting) {
+        scan_counters_.skipped_data_files++;
         continue;
       }
 
@@ -354,6 +360,7 @@ ManifestGroup::ReadEntries() {
       }
 
       if (!manifest_entry_predicate_(entry)) {
+        scan_counters_.skipped_data_files++;
         continue;
       }
 

src/iceberg/manifest/manifest_group.h

@@ -39,6 +39,16 @@
 
 namespace iceberg {
 
+/// \brief Counters for tracking scan metrics during manifest processing.
+struct ICEBERG_EXPORT ScanMetricsCounters {
+  int64_t scanned_data_manifests = 0;
+  int64_t skipped_data_manifests = 0;
+  int64_t scanned_delete_manifests = 0;
+  int64_t skipped_delete_manifests = 0;
+  int64_t skipped_data_files = 0;
+  int64_t skipped_delete_files = 0;
+};
+
 /// \brief Context passed to task creation functions.
 struct ICEBERG_EXPORT TaskContext {
  public:
@@ -120,6 +130,9 @@ class ICEBERG_EXPORT ManifestGroup : public ErrorCollector {
   /// \param column_ids Field IDs of columns whose statistics should be preserved.
   ManifestGroup& ColumnsToKeepStats(std::unordered_set<int32_t> column_ids);
 
+  /// \brief Returns the scan metrics counters accumulated during plan operations.
+  const ScanMetricsCounters& scan_counters() const { return scan_counters_; }
+
   /// \brief Plan scan tasks for all matching data files.
   Result<std::vector<std::shared_ptr<FileScanTask>>> PlanFiles();
 
@@ -162,6 +175,7 @@ class ICEBERG_EXPORT ManifestGroup : public ErrorCollector {
   bool ignore_deleted_ = false;
   bool ignore_existing_ = false;
   bool ignore_residuals_ = false;
+  ScanMetricsCounters scan_counters_;
 };
 
 }  // namespace iceberg

src/iceberg/metrics_reporter.h

@@ -23,8 +23,10 @@
 #include <memory>
 #include <string>
 #include <string_view>
+#include <unordered_map>
 #include <utility>
 #include <variant>
+#include <vector>
 
 #include "iceberg/iceberg_export.h"
 
@@ -85,6 +87,25 @@ struct ICEBERG_EXPORT ScanReport {
 
   /// \brief Number of delete manifests that were skipped.
   int64_t skipped_delete_manifests = 0;
+
+  /// \brief Projected field IDs from the scan schema.
+  std::vector<int32_t> projected_field_ids;
+  /// \brief Projected field names from the scan schema.
+  std::vector<std::string> projected_field_names;
+  /// \brief Total size in bytes of all result data files.
+  int64_t total_file_size_in_bytes = 0;
+  /// \brief Total size in bytes of all result delete files.
+  int64_t total_delete_file_size_in_bytes = 0;
+  /// \brief Number of indexed delete files.
+  int64_t indexed_delete_files = 0;
+  /// \brief Number of equality delete files in the scan result.
+  int64_t equality_delete_files = 0;
+  /// \brief Number of positional delete files in the scan result.
+  int64_t positional_delete_files = 0;
+  /// \brief Number of deletion vectors in the scan result.
+  int64_t dvs = 0;
+  /// \brief Additional key-value metadata.
+  std::unordered_map<std::string, std::string> metadata;
 };
 
 /// \brief Report generated after a commit operation.
@@ -136,6 +157,47 @@ struct ICEBERG_EXPORT CommitReport {
 
   /// \brief Size in bytes of files removed.
   int64_t removed_files_size = 0;
+
+  /// \brief Total duration of the commit operation.
+  DurationMs total_duration{0};
+  /// \brief Total records after this commit.
+  int64_t total_records = 0;
+  /// \brief Total file size in bytes after this commit.
+  int64_t total_files_size = 0;
+  /// \brief Equality delete files added.
+  int64_t added_equality_delete_files = 0;
+  /// \brief Equality delete files removed.
+  int64_t removed_equality_delete_files = 0;
+  /// \brief Positional delete files added.
+  int64_t added_positional_delete_files = 0;
+  /// \brief Positional delete files removed.
+  int64_t removed_positional_delete_files = 0;
+  /// \brief Position delete records added.
+  int64_t added_positional_deletes = 0;
+  /// \brief Position delete records removed.
+  int64_t removed_positional_deletes = 0;
+  /// \brief Total position delete records.
+  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.
+  int64_t total_equality_deletes = 0;
+  /// \brief Deletion vectors added.
+  int64_t added_dvs = 0;
+  /// \brief Deletion vectors removed.
+  int64_t removed_dvs = 0;
+  /// \brief Manifests created in this commit.
+  int64_t manifests_created = 0;
+  /// \brief Manifests replaced in this commit.
+  int64_t manifests_replaced = 0;
+  /// \brief Manifests kept in this commit.
+  int64_t manifests_kept = 0;
+  /// \brief Manifest entries processed.
+  int64_t manifest_entries_processed = 0;
+  /// \brief Additional key-value metadata.
+  std::unordered_map<std::string, std::string> metadata;
 };
 
 /// \brief The type of a metrics report.