Core: Add table-name filter for MetricsReporter

core/src/main/java/org/apache/iceberg/BaseMetastoreCatalog.java

@@ -293,7 +293,7 @@ protected static String fullTableName(String catalogName, TableIdentifier identi
 
   protected MetricsReporter metricsReporter() {
     if (metricsReporter == null) {
-      metricsReporter = CatalogUtil.loadMetricsReporter(properties());
+      metricsReporter = CatalogUtil.loadMetricsReporter(name(), properties());
     }
 
     return metricsReporter;

core/src/main/java/org/apache/iceberg/CatalogProperties.java

@@ -33,6 +33,63 @@ private CatalogProperties() {}
   public static final String VIEW_OVERRIDE_PREFIX = "view-override.";
   public static final String METRICS_REPORTER_IMPL = "metrics-reporter-impl";
 
+  /**
+   * Comma-separated list of Java regexes applied to {@code tableName()} of {@link
+   * org.apache.iceberg.metrics.ScanReport} and {@link org.apache.iceberg.metrics.CommitReport}.
+   * When set, only reports whose table name matches at least one pattern are forwarded to the
+   * configured {@link org.apache.iceberg.metrics.MetricsReporter}.
+   *
+   * <p>Each pattern is matched against the entire table name rather than any substring of it, so
+   * {@code prod\..*} matches {@code prod.db.table} but not {@code production.db.table}. Empty
+   * values are treated as not set.
+   */
+  public static final String METRICS_REPORTER_TABLE_NAME_INCLUDE =
+      "metrics-reporter.table-name.include";
+
+  /**
+   * Comma-separated list of Java regexes applied to {@code tableName()} of {@link
+   * org.apache.iceberg.metrics.ScanReport} and {@link org.apache.iceberg.metrics.CommitReport}.
+   * When set, reports whose table name matches any pattern are dropped before reaching the
+   * configured {@link org.apache.iceberg.metrics.MetricsReporter}. An exclude match wins over an
+   * include match.
+   *
+   * <p>Each pattern is matched against the entire table name rather than any substring of it. Empty
+   * values are treated as not set.
+   */
+  public static final String METRICS_REPORTER_TABLE_NAME_EXCLUDE =
+      "metrics-reporter.table-name.exclude";
+
+  /**
+   * Comma-separated list of Java regexes applied to the namespace of the table a {@link
+   * org.apache.iceberg.metrics.ScanReport} or {@link org.apache.iceberg.metrics.CommitReport} was
+   * produced for. When set, only reports for tables whose namespace matches at least one pattern
+   * are forwarded to the configured {@link org.apache.iceberg.metrics.MetricsReporter}.
+   *
+   * <p>The namespace is the dot-joined form of the table's namespace levels, with the catalog name
+   * removed, so {@code db} matches a table reported as {@code prod.db.table} in a catalog named
+   * {@code prod}. Filtering on the namespace is less prone to unintended matches than filtering on
+   * the table name, because a namespace is part of the table's identity rather than a naming
+   * convention.
+   *
+   * <p>Each pattern is matched against the entire namespace rather than any substring of it. Empty
+   * values are treated as not set.
+   */
+  public static final String METRICS_REPORTER_NAMESPACE_INCLUDE =
+      "metrics-reporter.namespace.include";
+
+  /**
+   * Comma-separated list of Java regexes applied to the namespace of the table a {@link
+   * org.apache.iceberg.metrics.ScanReport} or {@link org.apache.iceberg.metrics.CommitReport} was
+   * produced for. When set, reports for tables whose namespace matches any pattern are dropped
+   * before reaching the configured {@link org.apache.iceberg.metrics.MetricsReporter}. An exclude
+   * match wins over an include match.
+   *
+   * <p>Each pattern is matched against the entire namespace rather than any substring of it. Empty
+   * values are treated as not set.
+   */
+  public static final String METRICS_REPORTER_NAMESPACE_EXCLUDE =
+      "metrics-reporter.namespace.exclude";
+
   /**
    * Controls whether the catalog will cache table entries upon load.
    *

core/src/main/java/org/apache/iceberg/CatalogUtil.java

@@ -40,6 +40,7 @@
 import org.apache.iceberg.io.StorageCredential;
 import org.apache.iceberg.io.SupportsBulkOperations;
 import org.apache.iceberg.io.SupportsStorageCredentials;
+import org.apache.iceberg.metrics.FilteringMetricsReporter;
 import org.apache.iceberg.metrics.LoggingMetricsReporter;
 import org.apache.iceberg.metrics.MetricsReporter;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
@@ -513,38 +514,57 @@ public static void configureHadoopConf(Object maybeConfigurable, Object conf) {
    *     loaded class cannot be cast to the given interface type
    */
   public static MetricsReporter loadMetricsReporter(Map<String, String> properties) {
+    return loadMetricsReporter(null, properties);
+  }
+
+  /**
+   * Load a custom {@link MetricsReporter} implementation.
+   *
+   * <p>The implementation must have a no-arg constructor.
+   *
+   * @param catalogName name of the catalog the reporter is loaded for, used to derive a table's
+   *     namespace when filtering by namespace is configured
+   * @param properties catalog properties which contains class name of a custom {@link
+   *     MetricsReporter} implementation
+   * @return An initialized {@link MetricsReporter}.
+   * @throws IllegalArgumentException if class path not found or right constructor not found or the
+   *     loaded class cannot be cast to the given interface type
+   */
+  public static MetricsReporter loadMetricsReporter(
+      String catalogName, Map<String, String> properties) {
     String impl = properties.get(CatalogProperties.METRICS_REPORTER_IMPL);
+    MetricsReporter reporter;
     if (impl == null) {
-      return LoggingMetricsReporter.instance();
-    }
+      reporter = LoggingMetricsReporter.instance();
+    } else {
+      LOG.info("Loading custom MetricsReporter implementation: {}", impl);
+      DynConstructors.Ctor<MetricsReporter> ctor;
+      try {
+        ctor =
+            DynConstructors.builder(MetricsReporter.class)
+                .loader(CatalogUtil.class.getClassLoader())
+                .impl(impl)
+                .buildChecked();
+      } catch (NoSuchMethodException e) {
+        throw new IllegalArgumentException(
+            String.format(
+                "Cannot initialize MetricsReporter, missing no-arg constructor: %s", impl),
+            e);
+      }
 
-    LOG.info("Loading custom MetricsReporter implementation: {}", impl);
-    DynConstructors.Ctor<MetricsReporter> ctor;
-    try {
-      ctor =
-          DynConstructors.builder(MetricsReporter.class)
-              .loader(CatalogUtil.class.getClassLoader())
-              .impl(impl)
-              .buildChecked();
-    } catch (NoSuchMethodException e) {
-      throw new IllegalArgumentException(
-          String.format("Cannot initialize MetricsReporter, missing no-arg constructor: %s", impl),
-          e);
-    }
+      try {
+        reporter = ctor.newInstance();
+      } catch (ClassCastException e) {
+        throw new IllegalArgumentException(
+            String.format(
+                "Cannot initialize MetricsReporter, %s does not implement MetricsReporter.", impl),
+            e);
+      }
 
-    MetricsReporter reporter;
-    try {
-      reporter = ctor.newInstance();
-    } catch (ClassCastException e) {
-      throw new IllegalArgumentException(
-          String.format(
-              "Cannot initialize MetricsReporter, %s does not implement MetricsReporter.", impl),
-          e);
+      reporter.initialize(properties);
     }
 
-    reporter.initialize(properties);
-
-    return reporter;
+    return FilteringMetricsReporter.wrap(reporter, catalogName, properties);
   }
 
   public static String fullTableName(String catalogName, TableIdentifier identifier) {

core/src/main/java/org/apache/iceberg/metrics/FilteringMetricsReporter.java

@@ -0,0 +1,231 @@
+/*
+ * 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.
+ */
+package org.apache.iceberg.metrics;
+
+import java.util.List;
+import java.util.Map;
+import java.util.regex.Pattern;
+import java.util.regex.PatternSyntaxException;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+
+/**
+ * A {@link MetricsReporter} wrapper that drops {@link ScanReport} and {@link CommitReport}
+ * instances that do not pass the configured namespace and table-name filters before forwarding to a
+ * delegate reporter.
+ *
+ * <p>Filtering happens on two levels, either of which may be configured on its own:
+ *
+ * <ul>
+ *   <li>namespace, via {@link CatalogProperties#METRICS_REPORTER_NAMESPACE_INCLUDE} and {@link
+ *       CatalogProperties#METRICS_REPORTER_NAMESPACE_EXCLUDE}
+ *   <li>table name, via {@link CatalogProperties#METRICS_REPORTER_TABLE_NAME_INCLUDE} and {@link
+ *       CatalogProperties#METRICS_REPORTER_TABLE_NAME_EXCLUDE}
+ * </ul>
+ *
+ * <p>Each property holds a comma-separated list of Java regular expressions. A report is forwarded
+ * unless a filter drops it, and the levels are applied independently: an exclude match on either
+ * level drops the report, and when a level has an include list configured the report must match it.
+ * An exclude match always wins over an include match. When no property is set, {@link
+ * #wrap(MetricsReporter, String, Map)} returns the delegate unchanged so the default path incurs no
+ * overhead.
+ *
+ * <p>Patterns are matched against the entire namespace or table name rather than any substring of
+ * it, so {@code prod\..*} matches {@code prod.db.table} but not {@code production.db.table}.
+ *
+ * <p>{@link MetricsReport} subtypes other than {@link ScanReport} and {@link CommitReport} are
+ * forwarded without filtering, since they do not identify a table.
+ */
+public class FilteringMetricsReporter implements MetricsReporter {
+
+  private final MetricsReporter delegate;
+  private final String catalogName;
+  private final List<Pattern> namespaceInclude;
+  private final List<Pattern> namespaceExclude;
+  private final List<Pattern> tableNameInclude;
+  private final List<Pattern> tableNameExclude;
+
+  private FilteringMetricsReporter(
+      MetricsReporter delegate,
+      String catalogName,
+      List<Pattern> namespaceInclude,
+      List<Pattern> namespaceExclude,
+      List<Pattern> tableNameInclude,
+      List<Pattern> tableNameExclude) {
+    this.delegate = delegate;
+    this.catalogName = catalogName;
+    this.namespaceInclude = namespaceInclude;
+    this.namespaceExclude = namespaceExclude;
+    this.tableNameInclude = tableNameInclude;
+    this.tableNameExclude = tableNameExclude;
+  }
+
+  /**
+   * Wraps the given delegate in a {@code FilteringMetricsReporter} when any of the namespace or
+   * table-name filters is configured in {@code properties}; otherwise returns the delegate
+   * unchanged so the default case incurs no runtime overhead.
+   *
+   * @param delegate the underlying reporter that receives forwarded reports
+   * @param catalogName name of the catalog the reports originate from, used to derive a table's
+   *     namespace from the reported table name. When null, namespace filters cannot be applied and
+   *     configuring them is rejected.
+   * @param properties catalog properties; consulted for the namespace and table-name filters
+   * @return either the delegate unchanged, or a new filtering wrapper around it
+   */
+  public static MetricsReporter wrap(
+      MetricsReporter delegate, String catalogName, Map<String, String> properties) {
+    List<Pattern> namespaceInclude =
+        compilePatterns(properties, CatalogProperties.METRICS_REPORTER_NAMESPACE_INCLUDE);
+    List<Pattern> namespaceExclude =
+        compilePatterns(properties, CatalogProperties.METRICS_REPORTER_NAMESPACE_EXCLUDE);
+    List<Pattern> tableNameInclude =
+        compilePatterns(properties, CatalogProperties.METRICS_REPORTER_TABLE_NAME_INCLUDE);
+    List<Pattern> tableNameExclude =
+        compilePatterns(properties, CatalogProperties.METRICS_REPORTER_TABLE_NAME_EXCLUDE);
+
+    if (namespaceInclude.isEmpty()
+        && namespaceExclude.isEmpty()
+        && tableNameInclude.isEmpty()
+        && tableNameExclude.isEmpty()) {
+      return delegate;
+    }
+
+    if (catalogName == null && !(namespaceInclude.isEmpty() && namespaceExclude.isEmpty())) {
+      throw new IllegalArgumentException(
+          String.format(
+              "Cannot filter metrics by namespace without a catalog name: %s, %s",
+              CatalogProperties.METRICS_REPORTER_NAMESPACE_INCLUDE,
+              CatalogProperties.METRICS_REPORTER_NAMESPACE_EXCLUDE));
+    }
+
+    return new FilteringMetricsReporter(
+        delegate,
+        catalogName,
+        namespaceInclude,
+        namespaceExclude,
+        tableNameInclude,
+        tableNameExclude);
+  }
+
+  private static List<Pattern> compilePatterns(
+      Map<String, String> properties, String propertyName) {
+    String value = properties.get(propertyName);
+    if (value == null || value.trim().isEmpty()) {
+      return ImmutableList.of();
+    }
+
+    ImmutableList.Builder<Pattern> patterns = ImmutableList.builder();
+    for (String pattern : value.split(",", -1)) {
+      String trimmed = pattern.trim();
+      if (trimmed.isEmpty()) {
+        continue;
+      }
+
+      try {
+        patterns.add(Pattern.compile(trimmed));
+      } catch (PatternSyntaxException e) {
+        throw new IllegalArgumentException(
+            String.format("Invalid regex for %s: %s", propertyName, trimmed), e);
+      }
+    }
+
+    return patterns.build();
+  }
+
+  @Override
+  public void report(MetricsReport report) {
+    String tableName = tableName(report);
+    if (tableName == null) {
+      delegate.report(report);
+      return;
+    }
+
+    if (!passes(tableName, tableNameInclude, tableNameExclude)) {
+      return;
+    }
+
+    if (!namespaceInclude.isEmpty() || !namespaceExclude.isEmpty()) {
+      String namespace = namespace(tableName);
+      if (namespace == null || !passes(namespace, namespaceInclude, namespaceExclude)) {
+        return;
+      }
+    }
+
+    delegate.report(report);
+  }
+
+  private static boolean passes(String value, List<Pattern> include, List<Pattern> exclude) {
+    if (matchesAny(value, exclude)) {
+      return false;
+    }
+
+    return include.isEmpty() || matchesAny(value, include);
+  }
+
+  private static boolean matchesAny(String value, List<Pattern> patterns) {
+    for (Pattern pattern : patterns) {
+      if (pattern.matcher(value).matches()) {
+        return true;
+      }
+    }
+
+    return false;
+  }
+
+  /**
+   * Derives the namespace of a reported table by removing the catalog name prefix and the table
+   * name, mirroring how {@code CatalogUtil#fullTableName} builds the reported name. Returns an
+   * empty string for a table directly under the catalog, or null when the name does not carry the
+   * expected catalog prefix.
+   */
+  private String namespace(String tableName) {
+    String prefix;
+    if (catalogName.contains("/") || catalogName.contains(":")) {
+      // URI-like catalog names are joined with /, as in thrift://host:port/db.table
+      prefix = catalogName.endsWith("/") ? catalogName : catalogName + "/";
+    } else {
+      prefix = catalogName + ".";
+    }
+
+    if (!tableName.startsWith(prefix)) {
+      return null;
+    }
+
+    String withoutCatalog = tableName.substring(prefix.length());
+    int lastDot = withoutCatalog.lastIndexOf('.');
+    return lastDot < 0 ? "" : withoutCatalog.substring(0, lastDot);
+  }
+
+  private static String tableName(MetricsReport report) {
+    if (report instanceof ScanReport) {
+      return ((ScanReport) report).tableName();
+    }
+
+    if (report instanceof CommitReport) {
+      return ((CommitReport) report).tableName();
+    }
+
+    return null;
+  }
+
+  @Override
+  public void close() {
+    delegate.close();
+  }
+}