feat: metrics reporting for scan and commit

src/iceberg/CMakeLists.txt

@@ -58,6 +58,7 @@ set(ICEBERG_SOURCES
     manifest/v3_metadata.cc
     metadata_columns.cc
     metrics_config.cc
+    metrics/metrics_reporters.cc
     name_mapping.cc
     partition_field.cc
     partition_spec.cc
@@ -172,6 +173,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/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,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/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/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/metrics_reporter.h"
 #include "iceberg/result.h"
 
 /// \file iceberg/catalog/rest/rest_catalog.h
@@ -110,7 +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,
-              SnapshotMode snapshot_mode);
+              std::shared_ptr<MetricsReporter> reporter, SnapshotMode snapshot_mode);
 
   Result<std::string> LoadTableInternal(const TableIdentifier& identifier) const;
 
@@ -128,6 +129,7 @@ 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/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/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/meson.build

@@ -75,6 +75,7 @@ iceberg_sources = files(
     'manifest/v2_metadata.cc',
     'manifest/v3_metadata.cc',
     'metadata_columns.cc',
+    'metrics/metrics_reporters.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)