feat: metrics integration for commit and scan (#701)

This change implements part 2 of the core metrics. it covers RestCatalog
spec, integration with commit and scan workflows. It leaves the logging
metrics reporter out of scope, it will be added in a subsequent diff.
some other differences is that some of the http calls for the
RestCatalogMetrics reporter is still synchronous a subsequent change
will cover that.

---------

Co-authored-by: Innocent <idjiofack@jabode.com>

Author: evindj

example/demo_example.cc

@@ -18,6 +18,7 @@
  */
 
 #include <iostream>
+#include <utility>
 
 #include "iceberg/arrow/arrow_io_util.h"
 #include "iceberg/avro/avro_register.h"
@@ -42,8 +43,14 @@ int main(int argc, char** argv) {
   iceberg::avro::RegisterAll();
   iceberg::parquet::RegisterAll();
 
-  auto catalog = iceberg::InMemoryCatalog::Make("test", iceberg::arrow::MakeLocalFileIO(),
-                                                warehouse_location, properties);
+  auto catalog_result = iceberg::InMemoryCatalog::Make(
+      "test", iceberg::arrow::MakeLocalFileIO(), warehouse_location, properties);
+  if (!catalog_result.has_value()) {
+    std::cerr << "Failed to create catalog: " << catalog_result.error().message
+              << std::endl;
+    return 1;
+  }
+  auto catalog = std::move(catalog_result.value());
 
   auto register_result = catalog->RegisterTable({.name = table_name}, table_location);
   if (!register_result.has_value()) {

src/iceberg/CMakeLists.txt

@@ -21,6 +21,7 @@ set(ICEBERG_SOURCES
     arrow_c_data_guard_internal.cc
     arrow_c_data_util.cc
     arrow_row_builder.cc
+    catalog/catalog_util.cc
     catalog/memory/in_memory_catalog.cc
     catalog/session_catalog.cc
     catalog/session_context.cc
@@ -39,6 +40,7 @@ set(ICEBERG_SOURCES
     expression/projections.cc
     expression/residual_evaluator.cc
     expression/rewrite_not.cc
+    expression/sanitize_expression.cc
     expression/strict_metrics_evaluator.cc
     expression/term.cc
     file_io.cc

src/iceberg/catalog/catalog_util.cc

@@ -0,0 +1,49 @@
+/*
+ * 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/catalog_util.h"
+
+#include "iceberg/table_identifier.h"
+
+namespace iceberg {
+
+std::string CatalogUtil::FullTableName(std::string_view catalog_name,
+                                       const TableIdentifier& identifier) {
+  if (catalog_name.empty()) {
+    return identifier.ToString();
+  }
+
+  std::string result;
+  if (catalog_name.contains('/') || catalog_name.contains(':')) {
+    result = catalog_name;
+    if (!catalog_name.ends_with('/')) {
+      result += '/';
+    }
+  } else {
+    result = std::string(catalog_name) + '.';
+  }
+
+  for (const auto& level : identifier.ns.levels) {
+    result += level + '.';
+  }
+  result += identifier.name;
+  return result;
+}
+
+}  // namespace iceberg

src/iceberg/catalog/catalog_util.h

@@ -0,0 +1,38 @@
+/*
+ * 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>
+#include <string_view>
+
+#include "iceberg/iceberg_export.h"
+#include "iceberg/type_fwd.h"
+
+namespace iceberg {
+
+/// \brief Utilities for working with catalogs.
+class ICEBERG_EXPORT CatalogUtil {
+ public:
+  /// \brief Return the fully-qualified name for a table in a catalog.
+  static std::string FullTableName(std::string_view catalog_name,
+                                   const TableIdentifier& identifier);
+};
+
+}  // namespace iceberg

src/iceberg/catalog/memory/in_memory_catalog.cc

@@ -22,7 +22,9 @@
 #include <algorithm>
 #include <iterator>
 
+#include "iceberg/catalog/catalog_util.h"
 #include "iceberg/file_io.h"
+#include "iceberg/metrics/metrics_reporters.h"
 #include "iceberg/table.h"
 #include "iceberg/table_identifier.h"
 #include "iceberg/table_metadata.h"
@@ -337,22 +339,30 @@ Status InMemoryNamespace::UpdateTableMetadataLocation(
   return {};
 }
 
-std::shared_ptr<InMemoryCatalog> InMemoryCatalog::Make(
+Result<std::shared_ptr<InMemoryCatalog>> InMemoryCatalog::Make(
     const std::string& name, const std::shared_ptr<FileIO>& file_io,
     const std::string& warehouse_location,
     const std::unordered_map<std::string, std::string>& properties) {
-  return std::make_shared<InMemoryCatalog>(name, file_io, warehouse_location, properties);
+  std::shared_ptr<MetricsReporter> reporter;
+  auto it = properties.find(std::string(kMetricsReporterImpl));
+  if (it != properties.end() && !it->second.empty() &&
+      it->second != kMetricsReporterTypeNoop) {
+    ICEBERG_ASSIGN_OR_RAISE(reporter, MetricsReporters::Load(properties));
+  }
+  return std::make_shared<InMemoryCatalog>(name, file_io, warehouse_location, properties,
+                                           std::move(reporter));
 }
 
-InMemoryCatalog::InMemoryCatalog(
-    const std::string& name, const std::shared_ptr<FileIO>& file_io,
-    const std::string& warehouse_location,
-    const std::unordered_map<std::string, std::string>& properties)
+InMemoryCatalog::InMemoryCatalog(std::string name, std::shared_ptr<FileIO> file_io,
+                                 std::string warehouse_location,
+                                 std::unordered_map<std::string, std::string> properties,
+                                 std::shared_ptr<MetricsReporter> reporter)
     : catalog_name_(std::move(name)),
       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>()),
+      reporter_(std::move(reporter)) {}
 
 InMemoryCatalog::~InMemoryCatalog() = default;
 
@@ -429,7 +439,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(),
+                     CatalogUtil::FullTableName(name(), identifier), reporter_);
 }
 
 Result<std::shared_ptr<Table>> InMemoryCatalog::UpdateTable(
@@ -480,7 +491,8 @@ 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(),
+                     CatalogUtil::FullTableName(name(), identifier), reporter_);
 }
 
 Result<std::shared_ptr<Transaction>> InMemoryCatalog::StageCreateTable(
@@ -500,8 +512,10 @@ Result<std::shared_ptr<Transaction>> InMemoryCatalog::StageCreateTable(
       auto table_metadata,
       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()));
+      auto table,
+      StagedTable::Make(identifier, std::move(table_metadata), "", file_io_,
+                        shared_from_this(),
+                        CatalogUtil::FullTableName(name(), identifier), reporter_));
   return Transaction::Make(std::move(table), TransactionKind::kCreate);
 }
 
@@ -581,7 +595,8 @@ 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(),
+                     CatalogUtil::FullTableName(name(), identifier), reporter_);
 }
 
 Result<std::shared_ptr<Table>> InMemoryCatalog::RegisterTable(
@@ -601,7 +616,8 @@ 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(), CatalogUtil::FullTableName(name(), identifier),
+                     reporter_);
 }
 
 }  // namespace iceberg

src/iceberg/catalog/memory/in_memory_catalog.h

@@ -22,6 +22,7 @@
 /// \file iceberg/catalog/memory/in_memory_catalog.h
 /// \brief Provide an in-memory catalog implementation.
 
+#include <memory>
 #include <shared_mutex>
 
 #include "iceberg/catalog.h"
@@ -42,12 +43,13 @@ class ICEBERG_EXPORT InMemoryCatalog
     : public Catalog,
       public std::enable_shared_from_this<InMemoryCatalog> {
  public:
-  InMemoryCatalog(std::string const& name, std::shared_ptr<FileIO> const& file_io,
-                  std::string const& warehouse_location,
-                  std::unordered_map<std::string, std::string> const& properties);
+  InMemoryCatalog(std::string name, std::shared_ptr<FileIO> file_io,
+                  std::string warehouse_location,
+                  std::unordered_map<std::string, std::string> properties,
+                  std::shared_ptr<MetricsReporter> reporter = nullptr);
   ~InMemoryCatalog() override;
 
-  static std::shared_ptr<InMemoryCatalog> Make(
+  static Result<std::shared_ptr<InMemoryCatalog>> Make(
       std::string const& name, std::shared_ptr<FileIO> const& file_io,
       std::string const& warehouse_location,
       std::unordered_map<std::string, std::string> const& properties);
@@ -109,6 +111,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/meson.build

@@ -18,7 +18,7 @@
 subdir('memory')
 
 install_headers(
-    ['session_catalog.h', 'session_context.h'],
+    ['catalog_util.h', 'session_catalog.h', 'session_context.h'],
     subdir: 'iceberg/catalog',
 )
 

src/iceberg/catalog/rest/CMakeLists.txt

@@ -33,6 +33,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,9 @@ 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).
+  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',
 )