From e7db2087397ddb05cbbc1b748eeb09932515afd1 Mon Sep 17 00:00:00 2001 From: Noritaka Sekiyama Date: Wed, 27 May 2026 10:24:22 +0900 Subject: [PATCH 1/6] Core: Add table-name filter for MetricsReporter Add an optional filtering layer above any MetricsReporter implementation that drops ScanReports and CommitReports whose tableName() does not pass the configured include / exclude regex. Two new catalog properties control the filter: metrics-reporter.table-name.include and metrics-reporter.table-name.exclude. Both are Java regex patterns matched against the table name; when both are set, exclude wins over include. When neither property is set, CatalogUtil.loadMetricsReporter returns the underlying reporter unchanged, so the default code path incurs no runtime overhead. Empty values are treated as not set to avoid accidentally silencing all metrics on misconfiguration. Invalid regex values fail fast at catalog initialization with a clear error pointing at the offending property. The filter applies uniformly across all reporter implementations (LoggingMetricsReporter, RESTMetricsReporter, and custom user-supplied ones). Reports whose subtype does not expose a table name are forwarded without filtering. Closes #16573 --- .../org/apache/iceberg/CatalogProperties.java | 19 ++ .../java/org/apache/iceberg/CatalogUtil.java | 54 +++--- .../metrics/FilteringMetricsReporter.java | 118 +++++++++++ .../org/apache/iceberg/TestCatalogUtil.java | 23 +++ .../metrics/TestFilteringMetricsReporter.java | 183 ++++++++++++++++++ docs/docs/metrics-reporting.md | 21 ++ 6 files changed, 392 insertions(+), 26 deletions(-) create mode 100644 core/src/main/java/org/apache/iceberg/metrics/FilteringMetricsReporter.java create mode 100644 core/src/test/java/org/apache/iceberg/metrics/TestFilteringMetricsReporter.java diff --git a/core/src/main/java/org/apache/iceberg/CatalogProperties.java b/core/src/main/java/org/apache/iceberg/CatalogProperties.java index 6b85ccbc87bc..79ea4525d6cc 100644 --- a/core/src/main/java/org/apache/iceberg/CatalogProperties.java +++ b/core/src/main/java/org/apache/iceberg/CatalogProperties.java @@ -33,6 +33,25 @@ private CatalogProperties() {} public static final String VIEW_OVERRIDE_PREFIX = "view-override."; public static final String METRICS_REPORTER_IMPL = "metrics-reporter-impl"; + /** + * Java regex 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 the pattern are forwarded to the configured {@link + * org.apache.iceberg.metrics.MetricsReporter}. Empty values are treated as not set. + */ + public static final String METRICS_REPORTER_TABLE_NAME_INCLUDE = + "metrics-reporter.table-name.include"; + + /** + * Java regex 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 the + * pattern are dropped before reaching the configured {@link + * org.apache.iceberg.metrics.MetricsReporter}. When both include and exclude are set, exclude + * wins. Empty values are treated as not set. + */ + public static final String METRICS_REPORTER_TABLE_NAME_EXCLUDE = + "metrics-reporter.table-name.exclude"; + /** * Controls whether the catalog will cache table entries upon load. * diff --git a/core/src/main/java/org/apache/iceberg/CatalogUtil.java b/core/src/main/java/org/apache/iceberg/CatalogUtil.java index 4fa9fc30f1d0..8fbb4a126c9a 100644 --- a/core/src/main/java/org/apache/iceberg/CatalogUtil.java +++ b/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; @@ -514,37 +515,38 @@ public static void configureHadoopConf(Object maybeConfigurable, Object conf) { */ public static MetricsReporter loadMetricsReporter(Map 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 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 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, properties); } public static String fullTableName(String catalogName, TableIdentifier identifier) { diff --git a/core/src/main/java/org/apache/iceberg/metrics/FilteringMetricsReporter.java b/core/src/main/java/org/apache/iceberg/metrics/FilteringMetricsReporter.java new file mode 100644 index 000000000000..c5c14feca5f7 --- /dev/null +++ b/core/src/main/java/org/apache/iceberg/metrics/FilteringMetricsReporter.java @@ -0,0 +1,118 @@ +/* + * 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.Map; +import java.util.regex.Pattern; +import java.util.regex.PatternSyntaxException; +import org.apache.iceberg.CatalogProperties; + +/** + * A {@link MetricsReporter} wrapper that drops {@link ScanReport} and {@link CommitReport} + * instances whose {@code tableName()} does not pass the configured include / exclude patterns + * before forwarding to a delegate reporter. + * + *

The patterns are Java regular expressions sourced from the catalog properties {@link + * CatalogProperties#METRICS_REPORTER_TABLE_NAME_INCLUDE} and {@link + * CatalogProperties#METRICS_REPORTER_TABLE_NAME_EXCLUDE}. When both are set, {@code exclude} wins + * over {@code include} (an explicit deny overrides an include). When neither is set, {@link + * #wrap(MetricsReporter, Map)} returns the delegate unchanged. + * + *

{@link MetricsReport} subtypes other than {@link ScanReport} and {@link CommitReport} are + * forwarded to the delegate without filtering, since they do not expose a {@code tableName()}. + */ +public class FilteringMetricsReporter implements MetricsReporter { + + private final MetricsReporter delegate; + private final Pattern include; + private final Pattern exclude; + + private FilteringMetricsReporter(MetricsReporter delegate, Pattern include, Pattern exclude) { + this.delegate = delegate; + this.include = include; + this.exclude = exclude; + } + + /** + * Wraps the given delegate in a {@code FilteringMetricsReporter} when either include or exclude + * 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 properties catalog properties; consulted for the table-name include / exclude regex + * @return either the delegate unchanged, or a new filtering wrapper around it + */ + public static MetricsReporter wrap(MetricsReporter delegate, Map properties) { + Pattern include = + compileIfPresent( + properties.get(CatalogProperties.METRICS_REPORTER_TABLE_NAME_INCLUDE), + CatalogProperties.METRICS_REPORTER_TABLE_NAME_INCLUDE); + Pattern exclude = + compileIfPresent( + properties.get(CatalogProperties.METRICS_REPORTER_TABLE_NAME_EXCLUDE), + CatalogProperties.METRICS_REPORTER_TABLE_NAME_EXCLUDE); + if (include == null && exclude == null) { + return delegate; + } + return new FilteringMetricsReporter(delegate, include, exclude); + } + + private static Pattern compileIfPresent(String value, String propertyName) { + if (value == null || value.isEmpty()) { + return null; + } + try { + return Pattern.compile(value); + } catch (PatternSyntaxException e) { + throw new IllegalArgumentException( + String.format("Invalid regex for %s: %s", propertyName, value), e); + } + } + + @Override + public void report(MetricsReport report) { + String tableName = extractTableName(report); + if (tableName == null) { + delegate.report(report); + return; + } + if (exclude != null && exclude.matcher(tableName).matches()) { + return; + } + if (include != null && !include.matcher(tableName).matches()) { + return; + } + delegate.report(report); + } + + private static String extractTableName(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(); + } +} diff --git a/core/src/test/java/org/apache/iceberg/TestCatalogUtil.java b/core/src/test/java/org/apache/iceberg/TestCatalogUtil.java index 84e79e35c9b5..6a5a8dc52d99 100644 --- a/core/src/test/java/org/apache/iceberg/TestCatalogUtil.java +++ b/core/src/test/java/org/apache/iceberg/TestCatalogUtil.java @@ -43,6 +43,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.MetricsReport; import org.apache.iceberg.metrics.MetricsReporter; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; @@ -266,6 +267,28 @@ public void loadCustomMetricsReporter_badClass() { .hasMessageContaining("does not implement MetricsReporter"); } + @Test + public void loadMetricsReporter_wrappedWhenTableNameFilterPresent() { + MetricsReporter metricsReporter = + CatalogUtil.loadMetricsReporter( + ImmutableMap.of( + CatalogProperties.METRICS_REPORTER_IMPL, + TestMetricsReporterDefault.class.getName(), + CatalogProperties.METRICS_REPORTER_TABLE_NAME_INCLUDE, + "prod\\..*")); + assertThat(metricsReporter).isInstanceOf(FilteringMetricsReporter.class); + } + + @Test + public void loadMetricsReporter_notWrappedWhenFilterAbsent() { + MetricsReporter metricsReporter = + CatalogUtil.loadMetricsReporter( + ImmutableMap.of( + CatalogProperties.METRICS_REPORTER_IMPL, + TestMetricsReporterDefault.class.getName())); + assertThat(metricsReporter).isInstanceOf(TestMetricsReporterDefault.class); + } + @Test public void fullTableNameWithDifferentValues() { String uriTypeCatalogName = "thrift://host:port/db.table"; diff --git a/core/src/test/java/org/apache/iceberg/metrics/TestFilteringMetricsReporter.java b/core/src/test/java/org/apache/iceberg/metrics/TestFilteringMetricsReporter.java new file mode 100644 index 000000000000..8259315d3207 --- /dev/null +++ b/core/src/test/java/org/apache/iceberg/metrics/TestFilteringMetricsReporter.java @@ -0,0 +1,183 @@ +/* + * 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 static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import java.util.List; +import java.util.regex.PatternSyntaxException; +import org.apache.iceberg.CatalogProperties; +import org.apache.iceberg.expressions.Expressions; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; +import org.apache.iceberg.relocated.com.google.common.collect.Lists; +import org.junit.jupiter.api.Test; + +public class TestFilteringMetricsReporter { + + private static final ScanReport SCAN_PROD = newScanReport("prod_db.orders"); + private static final ScanReport SCAN_TMP = newScanReport("prod_db.tmp_staging"); + private static final ScanReport SCAN_DEV = newScanReport("dev_db.orders"); + private static final CommitReport COMMIT_PROD = newCommitReport("prod_db.orders"); + + @Test + public void wrapReturnsDelegateWhenNoPropertiesSet() { + CapturingMetricsReporter delegate = new CapturingMetricsReporter(); + MetricsReporter wrapped = FilteringMetricsReporter.wrap(delegate, ImmutableMap.of()); + assertThat(wrapped).isSameAs(delegate); + } + + @Test + public void wrapReturnsDelegateWhenPropertiesAreEmpty() { + CapturingMetricsReporter delegate = new CapturingMetricsReporter(); + MetricsReporter wrapped = + FilteringMetricsReporter.wrap( + delegate, + ImmutableMap.of( + CatalogProperties.METRICS_REPORTER_TABLE_NAME_INCLUDE, "", + CatalogProperties.METRICS_REPORTER_TABLE_NAME_EXCLUDE, "")); + assertThat(wrapped).isSameAs(delegate); + } + + @Test + public void includeOnlyForwardsMatchingTableNames() { + CapturingMetricsReporter delegate = new CapturingMetricsReporter(); + MetricsReporter wrapped = + FilteringMetricsReporter.wrap( + delegate, + ImmutableMap.of(CatalogProperties.METRICS_REPORTER_TABLE_NAME_INCLUDE, "prod_db\\..*")); + + wrapped.report(SCAN_PROD); + wrapped.report(SCAN_DEV); + wrapped.report(COMMIT_PROD); + + assertThat(delegate.reports).containsExactly(SCAN_PROD, COMMIT_PROD); + } + + @Test + public void excludeOnlyDropsMatchingTableNames() { + CapturingMetricsReporter delegate = new CapturingMetricsReporter(); + MetricsReporter wrapped = + FilteringMetricsReporter.wrap( + delegate, + ImmutableMap.of(CatalogProperties.METRICS_REPORTER_TABLE_NAME_EXCLUDE, ".*\\.tmp_.*")); + + wrapped.report(SCAN_PROD); + wrapped.report(SCAN_TMP); + + assertThat(delegate.reports).containsExactly(SCAN_PROD); + } + + @Test + public void excludeWinsOverInclude() { + CapturingMetricsReporter delegate = new CapturingMetricsReporter(); + MetricsReporter wrapped = + FilteringMetricsReporter.wrap( + delegate, + ImmutableMap.of( + CatalogProperties.METRICS_REPORTER_TABLE_NAME_INCLUDE, "prod_db\\..*", + CatalogProperties.METRICS_REPORTER_TABLE_NAME_EXCLUDE, ".*\\.tmp_.*")); + + wrapped.report(SCAN_PROD); + wrapped.report(SCAN_TMP); + wrapped.report(SCAN_DEV); + + assertThat(delegate.reports).containsExactly(SCAN_PROD); + } + + @Test + public void unknownReportSubtypeIsForwardedWithoutFiltering() { + CapturingMetricsReporter delegate = new CapturingMetricsReporter(); + MetricsReporter wrapped = + FilteringMetricsReporter.wrap( + delegate, + ImmutableMap.of(CatalogProperties.METRICS_REPORTER_TABLE_NAME_INCLUDE, "no_such\\..*")); + + MetricsReport unknown = new MetricsReport() {}; + wrapped.report(unknown); + + assertThat(delegate.reports).containsExactly(unknown); + } + + @Test + public void wrapThrowsClearErrorForInvalidRegex() { + assertThatThrownBy( + () -> + FilteringMetricsReporter.wrap( + new CapturingMetricsReporter(), + ImmutableMap.of( + CatalogProperties.METRICS_REPORTER_TABLE_NAME_INCLUDE, "[invalid"))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining(CatalogProperties.METRICS_REPORTER_TABLE_NAME_INCLUDE) + .hasMessageContaining("[invalid") + .hasCauseInstanceOf(PatternSyntaxException.class); + } + + @Test + public void closeIsDelegated() { + CapturingMetricsReporter delegate = new CapturingMetricsReporter(); + MetricsReporter wrapped = + FilteringMetricsReporter.wrap( + delegate, ImmutableMap.of(CatalogProperties.METRICS_REPORTER_TABLE_NAME_INCLUDE, ".*")); + + wrapped.close(); + + assertThat(delegate.closed).isTrue(); + } + + private static ScanReport newScanReport(String tableName) { + return ImmutableScanReport.builder() + .tableName(tableName) + .snapshotId(1L) + .filter(Expressions.alwaysTrue()) + .schemaId(1) + .projectedFieldIds(ImmutableList.of()) + .projectedFieldNames(ImmutableList.of()) + .scanMetrics(ImmutableScanMetricsResult.builder().build()) + .metadata(ImmutableMap.of()) + .build(); + } + + private static CommitReport newCommitReport(String tableName) { + return ImmutableCommitReport.builder() + .tableName(tableName) + .snapshotId(1L) + .sequenceNumber(1L) + .operation("append") + .commitMetrics(ImmutableCommitMetricsResult.builder().build()) + .metadata(ImmutableMap.of()) + .build(); + } + + private static class CapturingMetricsReporter implements MetricsReporter { + private final List reports = Lists.newArrayList(); + private boolean closed = false; + + @Override + public void report(MetricsReport report) { + reports.add(report); + } + + @Override + public void close() { + this.closed = true; + } + } +} diff --git a/docs/docs/metrics-reporting.md b/docs/docs/metrics-reporting.md index 4ca452b0d503..7e1ac1dfc402 100644 --- a/docs/docs/metrics-reporting.md +++ b/docs/docs/metrics-reporting.md @@ -147,6 +147,27 @@ public class InMemoryMetricsReporter implements MetricsReporter { The [catalog property](catalog-properties.md) `metrics-reporter-impl` allows registering a given [`MetricsReporter`](https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/metrics/MetricsReporter.java) by specifying its fully-qualified class name, e.g. `metrics-reporter-impl=org.apache.iceberg.metrics.InMemoryMetricsReporter`. +### Table-name filtering + +Reports forwarded to the configured `MetricsReporter` can be filtered by table name using two additional catalog properties. Both accept Java regular expressions matched against `ScanReport.tableName()` and `CommitReport.tableName()`: + +| Property | Effect | +|---|---| +| `metrics-reporter.table-name.include` | Forward only reports whose table name matches; drop the rest. | +| `metrics-reporter.table-name.exclude` | Drop reports whose table name matches; forward the rest. | + +When both are set, `exclude` wins over `include` (an explicit deny overrides an include). When neither is set, behavior is identical to today (every report is forwarded, with no runtime overhead). Empty values are treated as not set to avoid accidentally silencing all metrics on misconfiguration. + +For example, to forward metrics only for tables in the `prod_db` namespace while still dropping any temporary tables under it: + +``` +metrics-reporter-impl=org.apache.iceberg.metrics.LoggingMetricsReporter +metrics-reporter.table-name.include=prod_db\..* +metrics-reporter.table-name.exclude=.*\.tmp_.* +``` + +The filter applies uniformly to all `MetricsReporter` implementations (`LoggingMetricsReporter`, `RESTMetricsReporter`, and custom user-supplied ones). Reports whose subtype does not expose a table name (i.e. anything other than `ScanReport` and `CommitReport`) are forwarded without filtering. + ### Via the Java API during Scan planning Independently of the [`MetricsReporter`](https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/metrics/MetricsReporter.java) being registered at the catalog level via the `metrics-reporter-impl` property, it is also possible to supply additional reporters during scan planning as shown below: From 8e3988ae83a22e078afd176ed1d8fa8186258c13 Mon Sep 17 00:00:00 2001 From: Noritaka Sekiyama Date: Wed, 27 May 2026 21:58:43 +0900 Subject: [PATCH 2/6] REST: Apply table-name filter to RESTMetricsReporter as well The table-name filter introduced earlier in this PR is applied via CatalogUtil.loadMetricsReporter to the user-configured reporter, but RESTSessionCatalog injects an additional RESTMetricsReporter per table inside metricsReporter(...), which previously bypassed the filter. Wrap that RESTMetricsReporter with the same FilteringMetricsReporter (using the catalog properties stored at init) before combining with the user reporter, so both sides of the combined reporter honor the configured table-name filter. Add two tests: - TestRESTCatalog.metricsFilterAppliesToRestMetricsReporter exercises RESTSessionCatalog.metricsReporter(...) with filter properties and a mock RESTClient, verifying that one filtered + one unfiltered scan report produce exactly one post() and that the filtered report short-circuits before reaching the client. - TestFilteringMetricsReporter.loadMetricsReporterFiltersThroughUserConfiguredReporter goes through CatalogUtil.loadMetricsReporter with a static-singleton capturing reporter, demonstrating that the wrap applies at the catalog wiring level when metrics-reporter-impl plus the filter properties are configured together. --- .../iceberg/rest/RESTSessionCatalog.java | 9 +++- .../metrics/TestFilteringMetricsReporter.java | 37 ++++++++++++++ .../apache/iceberg/rest/TestRESTCatalog.java | 48 +++++++++++++++++++ 3 files changed, 92 insertions(+), 2 deletions(-) diff --git a/core/src/main/java/org/apache/iceberg/rest/RESTSessionCatalog.java b/core/src/main/java/org/apache/iceberg/rest/RESTSessionCatalog.java index 1d8e9da6d43c..943ef10f1ea6 100644 --- a/core/src/main/java/org/apache/iceberg/rest/RESTSessionCatalog.java +++ b/core/src/main/java/org/apache/iceberg/rest/RESTSessionCatalog.java @@ -62,6 +62,7 @@ import org.apache.iceberg.io.FileIOTracker; import org.apache.iceberg.io.StorageCredential; import org.apache.iceberg.io.SupportsStorageCredentials; +import org.apache.iceberg.metrics.FilteringMetricsReporter; import org.apache.iceberg.metrics.MetricsReporter; import org.apache.iceberg.metrics.MetricsReporters; import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting; @@ -169,6 +170,7 @@ public class RESTSessionCatalog extends BaseViewSessionCatalog private FileIO io = null; private MetricsReporter reporter = null; private ExecutorService metricsExecutor = null; + private Map reporterFilterProperties = ImmutableMap.of(); private boolean reportingViaRestEnabled; private Integer pageSize = null; private CloseableGroup closeables = null; @@ -274,6 +276,7 @@ public void initialize(String name, Map unresolved) { .toUpperCase(Locale.US)); this.reporter = CatalogUtil.loadMetricsReporter(mergedProps); + this.reporterFilterProperties = mergedProps; this.closeables.addCloseable(reporter); this.reportingViaRestEnabled = @@ -666,11 +669,13 @@ private void trackFileIO(RESTTableOperations ops) { } } - private MetricsReporter metricsReporter(String metricsEndpoint, RESTClient restClient) { + @VisibleForTesting + MetricsReporter metricsReporter(String metricsEndpoint, RESTClient restClient) { if (reportingViaRestEnabled && endpoints.contains(Endpoint.V1_REPORT_METRICS)) { RESTMetricsReporter restMetricsReporter = new RESTMetricsReporter(restClient, metricsEndpoint, Map::of, metricsExecutor); - return MetricsReporters.combine(reporter, restMetricsReporter); + return MetricsReporters.combine( + reporter, FilteringMetricsReporter.wrap(restMetricsReporter, reporterFilterProperties)); } else { return this.reporter; } diff --git a/core/src/test/java/org/apache/iceberg/metrics/TestFilteringMetricsReporter.java b/core/src/test/java/org/apache/iceberg/metrics/TestFilteringMetricsReporter.java index 8259315d3207..fc0a5b3e0b10 100644 --- a/core/src/test/java/org/apache/iceberg/metrics/TestFilteringMetricsReporter.java +++ b/core/src/test/java/org/apache/iceberg/metrics/TestFilteringMetricsReporter.java @@ -24,6 +24,7 @@ import java.util.List; import java.util.regex.PatternSyntaxException; import org.apache.iceberg.CatalogProperties; +import org.apache.iceberg.CatalogUtil; import org.apache.iceberg.expressions.Expressions; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; @@ -130,6 +131,28 @@ public void wrapThrowsClearErrorForInvalidRegex() { .hasCauseInstanceOf(PatternSyntaxException.class); } + @Test + public void loadMetricsReporterFiltersThroughUserConfiguredReporter() { + StaticCapturingReporter.REPORTS.clear(); + MetricsReporter reporter = + CatalogUtil.loadMetricsReporter( + ImmutableMap.of( + CatalogProperties.METRICS_REPORTER_IMPL, + StaticCapturingReporter.class.getName(), + CatalogProperties.METRICS_REPORTER_TABLE_NAME_INCLUDE, + "prod_db\\..*")); + + reporter.report(SCAN_PROD); + reporter.report(SCAN_DEV); + reporter.report(COMMIT_PROD); + + assertThat(StaticCapturingReporter.REPORTS) + .as( + "Reports configured via metrics-reporter-impl receive only table names that pass the" + + " include filter") + .containsExactly(SCAN_PROD, COMMIT_PROD); + } + @Test public void closeIsDelegated() { CapturingMetricsReporter delegate = new CapturingMetricsReporter(); @@ -180,4 +203,18 @@ public void close() { this.closed = true; } } + + /** + * Public no-arg reporter usable via {@code metrics-reporter-impl}. Captured reports live on a + * static list so the test can inspect what reached the underlying reporter after CatalogUtil + * instantiated it via reflection. + */ + public static class StaticCapturingReporter implements MetricsReporter { + static final List REPORTS = Lists.newCopyOnWriteArrayList(); + + @Override + public void report(MetricsReport report) { + REPORTS.add(report); + } + } } diff --git a/core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java b/core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java index d2d45c6173f5..efe1e654f580 100644 --- a/core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java +++ b/core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java @@ -30,6 +30,7 @@ import static org.mockito.ArgumentMatchers.argThat; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.atLeastOnce; +import static org.mockito.Mockito.mock; import static org.mockito.Mockito.timeout; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; @@ -93,6 +94,10 @@ import org.apache.iceberg.io.StorageCredential; import org.apache.iceberg.io.SupportsStorageCredentials; import org.apache.iceberg.metrics.CommitReport; +import org.apache.iceberg.metrics.ImmutableScanMetricsResult; +import org.apache.iceberg.metrics.ImmutableScanReport; +import org.apache.iceberg.metrics.MetricsReporter; +import org.apache.iceberg.metrics.ScanReport; import org.apache.iceberg.relocated.com.google.common.base.Preconditions; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; @@ -3990,6 +3995,49 @@ public Table registerTable( } } + @Test + public void metricsFilterAppliesToRestMetricsReporter() throws Exception { + RESTCatalog catalog = + initCatalog( + "withFilter", + ImmutableMap.of(CatalogProperties.METRICS_REPORTER_TABLE_NAME_INCLUDE, "prod_db\\..*")); + + String metricsEndpoint = "/v1/prefix/namespaces/x/tables/y/metrics"; + RESTClient mockClient = mock(RESTClient.class); + MetricsReporter combined = + catalog.sessionCatalog().metricsReporter(metricsEndpoint, mockClient); + + ScanReport excluded = scanReport("dev_db.scratch"); + ScanReport included = scanReport("prod_db.orders"); + + combined.report(excluded); + combined.report(included); + + // RESTMetricsReporter.report() submits to its executor and waits for completion via + // Tasks.range(1).run(...), so by the time both report() calls above have returned, all + // resulting client.post() invocations have already happened. The included report should + // produce exactly one post; the excluded one should produce none because the + // FilteringMetricsReporter wrapping the RESTMetricsReporter drops it before it reaches the + // REST client. + verify(mockClient, times(1)) + .post(eq(metricsEndpoint), any(RESTRequest.class), any(), any(Supplier.class), any()); + + catalog.close(); + } + + private static ScanReport scanReport(String tableName) { + return ImmutableScanReport.builder() + .tableName(tableName) + .snapshotId(1L) + .filter(Expressions.alwaysTrue()) + .schemaId(1) + .projectedFieldIds(ImmutableList.of()) + .projectedFieldNames(ImmutableList.of()) + .scanMetrics(ImmutableScanMetricsResult.builder().build()) + .metadata(ImmutableMap.of()) + .build(); + } + private RESTCatalog catalog(RESTCatalogAdapter adapter) { RESTCatalog catalog = new RESTCatalog(SessionCatalog.SessionContext.createEmpty(), (config) -> adapter); From e1ffab81f3ac2e25b8098b575fe6adc8589f4b4d Mon Sep 17 00:00:00 2001 From: Noritaka Sekiyama Date: Wed, 27 May 2026 22:12:59 +0900 Subject: [PATCH 3/6] Retrigger CI after transient gradle-wrapper download failure From 228a9ab27f2321eb7bedc6c3d2d78339b9029ec2 Mon Sep 17 00:00:00 2001 From: Noritaka Sekiyama Date: Mon, 29 Jun 2026 10:00:23 +0900 Subject: [PATCH 4/6] REST: Wait for async metrics export in filter test The rebase onto main picked up the change that makes RESTMetricsReporter export asynchronously via a single-threaded executor. The filter test verified post() synchronously, so it raced the background export and failed intermittently in CI (zero interactions). Use timeout()-based verification, matching the existing async CommitReport tests. --- .../org/apache/iceberg/rest/TestRESTCatalog.java | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java b/core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java index efe1e654f580..276d8bc2f0df 100644 --- a/core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java +++ b/core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java @@ -4013,13 +4013,13 @@ public void metricsFilterAppliesToRestMetricsReporter() throws Exception { combined.report(excluded); combined.report(included); - // RESTMetricsReporter.report() submits to its executor and waits for completion via - // Tasks.range(1).run(...), so by the time both report() calls above have returned, all - // resulting client.post() invocations have already happened. The included report should - // produce exactly one post; the excluded one should produce none because the - // FilteringMetricsReporter wrapping the RESTMetricsReporter drops it before it reaches the - // REST client. - verify(mockClient, times(1)) + // RESTMetricsReporter dispatches each report to a single-threaded executor asynchronously, so + // the resulting client.post() happens off-thread. The excluded report is submitted first and + // dropped by the FilteringMetricsReporter before it reaches the REST client; the included one + // is submitted next and produces exactly one post. Because the executor is single-threaded and + // FIFO, observing the included post (waited for via timeout) guarantees the excluded report has + // already been processed, so times(1) confirms the filter suppressed it. + verify(mockClient, timeout(5000).times(1)) .post(eq(metricsEndpoint), any(RESTRequest.class), any(), any(Supplier.class), any()); catalog.close(); From 4a5c96a1ab16ca10246a908cf83e9b654fa32155 Mon Sep 17 00:00:00 2001 From: Noritaka Sekiyama Date: Fri, 7 Aug 2026 14:02:30 +0900 Subject: [PATCH 5/6] Core: Accept a list of patterns for the MetricsReporter table-name filter The table-name include / exclude properties took a single regex, so covering several databases meant hand-writing an alternation. Accept a comma-separated list instead, matching how the same problem is configured elsewhere: Debezium's table.include.list, DataHub's table_pattern allow/deny, and OpenMetadata's tableFilterPattern all take a list of expressions. Patterns were already matched against the whole name via Matcher.matches(), which is the same choice Debezium documents as an "anchored regular expression". That property is worth stating explicitly, since an unanchored prod.* would otherwise capture production.orders and prod_sandbox.orders -- a silent over-match that is hard to notice. Documented it and added a test that pins the behavior. --- .../org/apache/iceberg/CatalogProperties.java | 25 ++-- .../metrics/FilteringMetricsReporter.java | 111 ++++++++++++------ .../metrics/TestFilteringMetricsReporter.java | 52 ++++++++ docs/docs/metrics-reporting.md | 8 +- 4 files changed, 147 insertions(+), 49 deletions(-) diff --git a/core/src/main/java/org/apache/iceberg/CatalogProperties.java b/core/src/main/java/org/apache/iceberg/CatalogProperties.java index 79ea4525d6cc..ce0be76946dc 100644 --- a/core/src/main/java/org/apache/iceberg/CatalogProperties.java +++ b/core/src/main/java/org/apache/iceberg/CatalogProperties.java @@ -34,20 +34,27 @@ private CatalogProperties() {} public static final String METRICS_REPORTER_IMPL = "metrics-reporter-impl"; /** - * Java regex 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 the pattern are forwarded to the configured {@link - * org.apache.iceberg.metrics.MetricsReporter}. Empty values are treated as not set. + * 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}. + * + *

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"; /** - * Java regex 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 the - * pattern are dropped before reaching the configured {@link - * org.apache.iceberg.metrics.MetricsReporter}. When both include and exclude are set, exclude - * wins. Empty values are treated as not set. + * 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. + * + *

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"; diff --git a/core/src/main/java/org/apache/iceberg/metrics/FilteringMetricsReporter.java b/core/src/main/java/org/apache/iceberg/metrics/FilteringMetricsReporter.java index c5c14feca5f7..d15f21e1e2a1 100644 --- a/core/src/main/java/org/apache/iceberg/metrics/FilteringMetricsReporter.java +++ b/core/src/main/java/org/apache/iceberg/metrics/FilteringMetricsReporter.java @@ -18,35 +18,42 @@ */ 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 whose {@code tableName()} does not pass the configured include / exclude patterns - * before forwarding to a delegate reporter. + * instances whose {@code tableName()} does not pass the configured include / exclude filters before + * forwarding to a delegate reporter. * - *

The patterns are Java regular expressions sourced from the catalog properties {@link + *

The filters come from the catalog properties {@link * CatalogProperties#METRICS_REPORTER_TABLE_NAME_INCLUDE} and {@link - * CatalogProperties#METRICS_REPORTER_TABLE_NAME_EXCLUDE}. When both are set, {@code exclude} wins - * over {@code include} (an explicit deny overrides an include). When neither is set, {@link - * #wrap(MetricsReporter, Map)} returns the delegate unchanged. + * CatalogProperties#METRICS_REPORTER_TABLE_NAME_EXCLUDE}, each holding a comma-separated list of + * Java regular expressions. An exclude match wins over an include match. When neither is set, + * {@link #wrap(MetricsReporter, Map)} returns the delegate unchanged so the default path incurs no + * overhead. + * + *

Patterns are 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}. * *

{@link MetricsReport} subtypes other than {@link ScanReport} and {@link CommitReport} are - * forwarded to the delegate without filtering, since they do not expose a {@code tableName()}. + * forwarded without filtering, since they do not identify a table. */ public class FilteringMetricsReporter implements MetricsReporter { private final MetricsReporter delegate; - private final Pattern include; - private final Pattern exclude; + private final List tableNameInclude; + private final List tableNameExclude; - private FilteringMetricsReporter(MetricsReporter delegate, Pattern include, Pattern exclude) { + private FilteringMetricsReporter( + MetricsReporter delegate, List tableNameInclude, List tableNameExclude) { this.delegate = delegate; - this.include = include; - this.exclude = exclude; + this.tableNameInclude = tableNameInclude; + this.tableNameExclude = tableNameExclude; } /** @@ -55,59 +62,89 @@ private FilteringMetricsReporter(MetricsReporter delegate, Pattern include, Patt * case incurs no runtime overhead. * * @param delegate the underlying reporter that receives forwarded reports - * @param properties catalog properties; consulted for the table-name include / exclude regex + * @param properties catalog properties; consulted for the table-name include / exclude filters * @return either the delegate unchanged, or a new filtering wrapper around it */ public static MetricsReporter wrap(MetricsReporter delegate, Map properties) { - Pattern include = - compileIfPresent( - properties.get(CatalogProperties.METRICS_REPORTER_TABLE_NAME_INCLUDE), - CatalogProperties.METRICS_REPORTER_TABLE_NAME_INCLUDE); - Pattern exclude = - compileIfPresent( - properties.get(CatalogProperties.METRICS_REPORTER_TABLE_NAME_EXCLUDE), - CatalogProperties.METRICS_REPORTER_TABLE_NAME_EXCLUDE); - if (include == null && exclude == null) { + List tableNameInclude = + compilePatterns(properties, CatalogProperties.METRICS_REPORTER_TABLE_NAME_INCLUDE); + List tableNameExclude = + compilePatterns(properties, CatalogProperties.METRICS_REPORTER_TABLE_NAME_EXCLUDE); + + if (tableNameInclude.isEmpty() && tableNameExclude.isEmpty()) { return delegate; } - return new FilteringMetricsReporter(delegate, include, exclude); + + return new FilteringMetricsReporter(delegate, tableNameInclude, tableNameExclude); } - private static Pattern compileIfPresent(String value, String propertyName) { - if (value == null || value.isEmpty()) { - return null; + private static List compilePatterns( + Map properties, String propertyName) { + String value = properties.get(propertyName); + if (value == null || value.trim().isEmpty()) { + return ImmutableList.of(); } - try { - return Pattern.compile(value); - } catch (PatternSyntaxException e) { - throw new IllegalArgumentException( - String.format("Invalid regex for %s: %s", propertyName, value), e); + + ImmutableList.Builder 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 = extractTableName(report); + String tableName = tableName(report); if (tableName == null) { delegate.report(report); return; } - if (exclude != null && exclude.matcher(tableName).matches()) { - return; - } - if (include != null && !include.matcher(tableName).matches()) { + + if (!passes(tableName, tableNameInclude, tableNameExclude)) { return; } + delegate.report(report); } - private static String extractTableName(MetricsReport report) { + private static boolean passes(String value, List include, List exclude) { + if (matchesAny(value, exclude)) { + return false; + } + + return include.isEmpty() || matchesAny(value, include); + } + + private static boolean matchesAny(String value, List patterns) { + for (Pattern pattern : patterns) { + if (pattern.matcher(value).matches()) { + return true; + } + } + + return false; + } + + private static String tableName(MetricsReport report) { if (report instanceof ScanReport) { return ((ScanReport) report).tableName(); } + if (report instanceof CommitReport) { return ((CommitReport) report).tableName(); } + return null; } diff --git a/core/src/test/java/org/apache/iceberg/metrics/TestFilteringMetricsReporter.java b/core/src/test/java/org/apache/iceberg/metrics/TestFilteringMetricsReporter.java index fc0a5b3e0b10..06014d531d2d 100644 --- a/core/src/test/java/org/apache/iceberg/metrics/TestFilteringMetricsReporter.java +++ b/core/src/test/java/org/apache/iceberg/metrics/TestFilteringMetricsReporter.java @@ -165,6 +165,58 @@ public void closeIsDelegated() { assertThat(delegate.closed).isTrue(); } + @Test + public void includeAcceptsCommaSeparatedPatterns() { + CapturingMetricsReporter delegate = new CapturingMetricsReporter(); + MetricsReporter wrapped = + FilteringMetricsReporter.wrap( + delegate, + ImmutableMap.of( + CatalogProperties.METRICS_REPORTER_TABLE_NAME_INCLUDE, + "prod_db\\..*, analytics_db\\..*")); + + ScanReport analytics = newScanReport("analytics_db.events"); + wrapped.report(SCAN_PROD); + wrapped.report(analytics); + wrapped.report(SCAN_DEV); + + assertThat(delegate.reports).containsExactly(SCAN_PROD, analytics); + } + + @Test + public void excludeAcceptsCommaSeparatedPatterns() { + CapturingMetricsReporter delegate = new CapturingMetricsReporter(); + MetricsReporter wrapped = + FilteringMetricsReporter.wrap( + delegate, + ImmutableMap.of( + CatalogProperties.METRICS_REPORTER_TABLE_NAME_EXCLUDE, ".*\\.tmp_.*,dev_db\\..*")); + + wrapped.report(SCAN_PROD); + wrapped.report(SCAN_TMP); + wrapped.report(SCAN_DEV); + + assertThat(delegate.reports).containsExactly(SCAN_PROD); + } + + @Test + public void patternsMatchWholeNameNotSubstring() { + CapturingMetricsReporter delegate = new CapturingMetricsReporter(); + MetricsReporter wrapped = + FilteringMetricsReporter.wrap( + delegate, + ImmutableMap.of(CatalogProperties.METRICS_REPORTER_TABLE_NAME_INCLUDE, "prod\\..*")); + + ScanReport prod = newScanReport("prod.orders"); + // names that a substring match would wrongly accept + wrapped.report(prod); + wrapped.report(newScanReport("production.orders")); + wrapped.report(newScanReport("prod_sandbox.orders")); + wrapped.report(newScanReport("staging.prod.orders")); + + assertThat(delegate.reports).containsExactly(prod); + } + private static ScanReport newScanReport(String tableName) { return ImmutableScanReport.builder() .tableName(tableName) diff --git a/docs/docs/metrics-reporting.md b/docs/docs/metrics-reporting.md index 7e1ac1dfc402..1431ade068d6 100644 --- a/docs/docs/metrics-reporting.md +++ b/docs/docs/metrics-reporting.md @@ -149,20 +149,22 @@ The [catalog property](catalog-properties.md) `metrics-reporter-impl` allows reg ### Table-name filtering -Reports forwarded to the configured `MetricsReporter` can be filtered by table name using two additional catalog properties. Both accept Java regular expressions matched against `ScanReport.tableName()` and `CommitReport.tableName()`: +Reports forwarded to the configured `MetricsReporter` can be filtered by table name using two additional catalog properties. Both accept a comma-separated list of Java regular expressions matched against `ScanReport.tableName()` and `CommitReport.tableName()`: | Property | Effect | |---|---| | `metrics-reporter.table-name.include` | Forward only reports whose table name matches; drop the rest. | | `metrics-reporter.table-name.exclude` | Drop reports whose table name matches; forward the rest. | +Patterns are matched against the **entire** table name rather than any substring of it. This matters in practice: `prod\..*` matches `prod.db.table` but not `production.db.table` or `prod_sandbox.db.table`, which a substring match would wrongly accept. + When both are set, `exclude` wins over `include` (an explicit deny overrides an include). When neither is set, behavior is identical to today (every report is forwarded, with no runtime overhead). Empty values are treated as not set to avoid accidentally silencing all metrics on misconfiguration. -For example, to forward metrics only for tables in the `prod_db` namespace while still dropping any temporary tables under it: +For example, to forward metrics for the `prod_db` and `analytics_db` databases while still dropping any temporary tables under them: ``` metrics-reporter-impl=org.apache.iceberg.metrics.LoggingMetricsReporter -metrics-reporter.table-name.include=prod_db\..* +metrics-reporter.table-name.include=prod_db\..*,analytics_db\..* metrics-reporter.table-name.exclude=.*\.tmp_.* ``` From 5da80bf6a4d77346cf3f7f8d9c5115de0fc8618d Mon Sep 17 00:00:00 2001 From: Noritaka Sekiyama Date: Fri, 7 Aug 2026 14:03:45 +0900 Subject: [PATCH 6/6] Core: Add a namespace filter for MetricsReporter Filtering only by table name relies on a naming convention: a pattern written today silently captures tables created tomorrow, and the failure mode is invisible because metrics simply stop appearing for a table nobody thought about. A namespace is part of a table's identity instead, so a table added to an included namespace later is picked up automatically and a table outside it cannot match by accident. Add metrics-reporter.namespace.include / .exclude alongside the existing table-name filters. Both levels are applied independently, so a namespace can be selected while individual noisy tables inside it are still excluded by name -- the coarse-plus-fine combination that database.include.list plus table.include.list provides in Debezium, and schema_pattern plus table_pattern in DataHub. The namespace is derived from the reported table name by removing the catalog name, which the catalog supplies when it loads the reporter. Knowing the catalog name is what makes this unambiguous: the reported name is built by CatalogUtil#fullTableName, which switches its separator for URI-like catalog names and offers no way to tell a dotted catalog name from the namespace that follows it. Tests cover both shapes, along with multi-level and empty namespaces. Configuring a namespace filter where no catalog name is available fails at initialization rather than silently dropping every report. --- .../apache/iceberg/BaseMetastoreCatalog.java | 2 +- .../org/apache/iceberg/CatalogProperties.java | 31 ++++ .../java/org/apache/iceberg/CatalogUtil.java | 20 ++- .../metrics/FilteringMetricsReporter.java | 110 ++++++++++-- .../iceberg/rest/RESTSessionCatalog.java | 9 +- .../metrics/TestFilteringMetricsReporter.java | 166 +++++++++++++++++- docs/docs/metrics-reporting.md | 24 ++- 7 files changed, 331 insertions(+), 31 deletions(-) diff --git a/core/src/main/java/org/apache/iceberg/BaseMetastoreCatalog.java b/core/src/main/java/org/apache/iceberg/BaseMetastoreCatalog.java index 940d7fa05ec6..7d2329a1464f 100644 --- a/core/src/main/java/org/apache/iceberg/BaseMetastoreCatalog.java +++ b/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; diff --git a/core/src/main/java/org/apache/iceberg/CatalogProperties.java b/core/src/main/java/org/apache/iceberg/CatalogProperties.java index ce0be76946dc..ab36a93aefa3 100644 --- a/core/src/main/java/org/apache/iceberg/CatalogProperties.java +++ b/core/src/main/java/org/apache/iceberg/CatalogProperties.java @@ -59,6 +59,37 @@ private CatalogProperties() {} 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}. + * + *

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. + * + *

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. + * + *

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. * diff --git a/core/src/main/java/org/apache/iceberg/CatalogUtil.java b/core/src/main/java/org/apache/iceberg/CatalogUtil.java index 8fbb4a126c9a..96c6b0848b86 100644 --- a/core/src/main/java/org/apache/iceberg/CatalogUtil.java +++ b/core/src/main/java/org/apache/iceberg/CatalogUtil.java @@ -514,6 +514,24 @@ public static void configureHadoopConf(Object maybeConfigurable, Object conf) { * loaded class cannot be cast to the given interface type */ public static MetricsReporter loadMetricsReporter(Map properties) { + return loadMetricsReporter(null, properties); + } + + /** + * Load a custom {@link MetricsReporter} implementation. + * + *

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 properties) { String impl = properties.get(CatalogProperties.METRICS_REPORTER_IMPL); MetricsReporter reporter; if (impl == null) { @@ -546,7 +564,7 @@ public static MetricsReporter loadMetricsReporter(Map properties reporter.initialize(properties); } - return FilteringMetricsReporter.wrap(reporter, properties); + return FilteringMetricsReporter.wrap(reporter, catalogName, properties); } public static String fullTableName(String catalogName, TableIdentifier identifier) { diff --git a/core/src/main/java/org/apache/iceberg/metrics/FilteringMetricsReporter.java b/core/src/main/java/org/apache/iceberg/metrics/FilteringMetricsReporter.java index d15f21e1e2a1..de023d0422ff 100644 --- a/core/src/main/java/org/apache/iceberg/metrics/FilteringMetricsReporter.java +++ b/core/src/main/java/org/apache/iceberg/metrics/FilteringMetricsReporter.java @@ -27,18 +27,27 @@ /** * A {@link MetricsReporter} wrapper that drops {@link ScanReport} and {@link CommitReport} - * instances whose {@code tableName()} does not pass the configured include / exclude filters before - * forwarding to a delegate reporter. + * instances that do not pass the configured namespace and table-name filters before forwarding to a + * delegate reporter. * - *

The filters come from the catalog properties {@link - * CatalogProperties#METRICS_REPORTER_TABLE_NAME_INCLUDE} and {@link - * CatalogProperties#METRICS_REPORTER_TABLE_NAME_EXCLUDE}, each holding a comma-separated list of - * Java regular expressions. An exclude match wins over an include match. When neither is set, - * {@link #wrap(MetricsReporter, Map)} returns the delegate unchanged so the default path incurs no + *

Filtering happens on two levels, either of which may be configured on its own: + * + *

    + *
  • namespace, via {@link CatalogProperties#METRICS_REPORTER_NAMESPACE_INCLUDE} and {@link + * CatalogProperties#METRICS_REPORTER_NAMESPACE_EXCLUDE} + *
  • table name, via {@link CatalogProperties#METRICS_REPORTER_TABLE_NAME_INCLUDE} and {@link + * CatalogProperties#METRICS_REPORTER_TABLE_NAME_EXCLUDE} + *
+ * + *

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. * - *

Patterns are 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}. + *

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}. * *

{@link MetricsReport} subtypes other than {@link ScanReport} and {@link CommitReport} are * forwarded without filtering, since they do not identify a table. @@ -46,36 +55,72 @@ public class FilteringMetricsReporter implements MetricsReporter { private final MetricsReporter delegate; + private final String catalogName; + private final List namespaceInclude; + private final List namespaceExclude; private final List tableNameInclude; private final List tableNameExclude; private FilteringMetricsReporter( - MetricsReporter delegate, List tableNameInclude, List tableNameExclude) { + MetricsReporter delegate, + String catalogName, + List namespaceInclude, + List namespaceExclude, + List tableNameInclude, + List 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 either include or exclude - * is configured in {@code properties}; otherwise returns the delegate unchanged so the default - * case incurs no runtime overhead. + * 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 properties catalog properties; consulted for the table-name include / exclude filters + * @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, Map properties) { + public static MetricsReporter wrap( + MetricsReporter delegate, String catalogName, Map properties) { + List namespaceInclude = + compilePatterns(properties, CatalogProperties.METRICS_REPORTER_NAMESPACE_INCLUDE); + List namespaceExclude = + compilePatterns(properties, CatalogProperties.METRICS_REPORTER_NAMESPACE_EXCLUDE); List tableNameInclude = compilePatterns(properties, CatalogProperties.METRICS_REPORTER_TABLE_NAME_INCLUDE); List tableNameExclude = compilePatterns(properties, CatalogProperties.METRICS_REPORTER_TABLE_NAME_EXCLUDE); - if (tableNameInclude.isEmpty() && tableNameExclude.isEmpty()) { + if (namespaceInclude.isEmpty() + && namespaceExclude.isEmpty() + && tableNameInclude.isEmpty() + && tableNameExclude.isEmpty()) { return delegate; } - return new FilteringMetricsReporter(delegate, tableNameInclude, tableNameExclude); + 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 compilePatterns( @@ -115,6 +160,13 @@ public void report(MetricsReport report) { return; } + if (!namespaceInclude.isEmpty() || !namespaceExclude.isEmpty()) { + String namespace = namespace(tableName); + if (namespace == null || !passes(namespace, namespaceInclude, namespaceExclude)) { + return; + } + } + delegate.report(report); } @@ -136,6 +188,30 @@ private static boolean matchesAny(String value, List patterns) { 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(); diff --git a/core/src/main/java/org/apache/iceberg/rest/RESTSessionCatalog.java b/core/src/main/java/org/apache/iceberg/rest/RESTSessionCatalog.java index 943ef10f1ea6..002ff99bc1fb 100644 --- a/core/src/main/java/org/apache/iceberg/rest/RESTSessionCatalog.java +++ b/core/src/main/java/org/apache/iceberg/rest/RESTSessionCatalog.java @@ -171,6 +171,7 @@ public class RESTSessionCatalog extends BaseViewSessionCatalog private MetricsReporter reporter = null; private ExecutorService metricsExecutor = null; private Map reporterFilterProperties = ImmutableMap.of(); + private String reporterCatalogName = null; private boolean reportingViaRestEnabled; private Integer pageSize = null; private CloseableGroup closeables = null; @@ -275,7 +276,9 @@ public void initialize(String name, Map unresolved) { RESTCatalogProperties.SNAPSHOT_LOADING_MODE_DEFAULT.name()) .toUpperCase(Locale.US)); - this.reporter = CatalogUtil.loadMetricsReporter(mergedProps); + // name() is only available after super.initialize below, so use the name argument here + this.reporter = CatalogUtil.loadMetricsReporter(name, mergedProps); + this.reporterCatalogName = name; this.reporterFilterProperties = mergedProps; this.closeables.addCloseable(reporter); @@ -675,7 +678,9 @@ MetricsReporter metricsReporter(String metricsEndpoint, RESTClient restClient) { RESTMetricsReporter restMetricsReporter = new RESTMetricsReporter(restClient, metricsEndpoint, Map::of, metricsExecutor); return MetricsReporters.combine( - reporter, FilteringMetricsReporter.wrap(restMetricsReporter, reporterFilterProperties)); + reporter, + FilteringMetricsReporter.wrap( + restMetricsReporter, reporterCatalogName, reporterFilterProperties)); } else { return this.reporter; } diff --git a/core/src/test/java/org/apache/iceberg/metrics/TestFilteringMetricsReporter.java b/core/src/test/java/org/apache/iceberg/metrics/TestFilteringMetricsReporter.java index 06014d531d2d..f5ad404d7d16 100644 --- a/core/src/test/java/org/apache/iceberg/metrics/TestFilteringMetricsReporter.java +++ b/core/src/test/java/org/apache/iceberg/metrics/TestFilteringMetricsReporter.java @@ -41,7 +41,7 @@ public class TestFilteringMetricsReporter { @Test public void wrapReturnsDelegateWhenNoPropertiesSet() { CapturingMetricsReporter delegate = new CapturingMetricsReporter(); - MetricsReporter wrapped = FilteringMetricsReporter.wrap(delegate, ImmutableMap.of()); + MetricsReporter wrapped = FilteringMetricsReporter.wrap(delegate, null, ImmutableMap.of()); assertThat(wrapped).isSameAs(delegate); } @@ -51,6 +51,7 @@ public void wrapReturnsDelegateWhenPropertiesAreEmpty() { MetricsReporter wrapped = FilteringMetricsReporter.wrap( delegate, + null, ImmutableMap.of( CatalogProperties.METRICS_REPORTER_TABLE_NAME_INCLUDE, "", CatalogProperties.METRICS_REPORTER_TABLE_NAME_EXCLUDE, "")); @@ -63,6 +64,7 @@ public void includeOnlyForwardsMatchingTableNames() { MetricsReporter wrapped = FilteringMetricsReporter.wrap( delegate, + null, ImmutableMap.of(CatalogProperties.METRICS_REPORTER_TABLE_NAME_INCLUDE, "prod_db\\..*")); wrapped.report(SCAN_PROD); @@ -78,6 +80,7 @@ public void excludeOnlyDropsMatchingTableNames() { MetricsReporter wrapped = FilteringMetricsReporter.wrap( delegate, + null, ImmutableMap.of(CatalogProperties.METRICS_REPORTER_TABLE_NAME_EXCLUDE, ".*\\.tmp_.*")); wrapped.report(SCAN_PROD); @@ -92,6 +95,7 @@ public void excludeWinsOverInclude() { MetricsReporter wrapped = FilteringMetricsReporter.wrap( delegate, + null, ImmutableMap.of( CatalogProperties.METRICS_REPORTER_TABLE_NAME_INCLUDE, "prod_db\\..*", CatalogProperties.METRICS_REPORTER_TABLE_NAME_EXCLUDE, ".*\\.tmp_.*")); @@ -109,6 +113,7 @@ public void unknownReportSubtypeIsForwardedWithoutFiltering() { MetricsReporter wrapped = FilteringMetricsReporter.wrap( delegate, + null, ImmutableMap.of(CatalogProperties.METRICS_REPORTER_TABLE_NAME_INCLUDE, "no_such\\..*")); MetricsReport unknown = new MetricsReport() {}; @@ -123,6 +128,7 @@ public void wrapThrowsClearErrorForInvalidRegex() { () -> FilteringMetricsReporter.wrap( new CapturingMetricsReporter(), + null, ImmutableMap.of( CatalogProperties.METRICS_REPORTER_TABLE_NAME_INCLUDE, "[invalid"))) .isInstanceOf(IllegalArgumentException.class) @@ -158,7 +164,9 @@ public void closeIsDelegated() { CapturingMetricsReporter delegate = new CapturingMetricsReporter(); MetricsReporter wrapped = FilteringMetricsReporter.wrap( - delegate, ImmutableMap.of(CatalogProperties.METRICS_REPORTER_TABLE_NAME_INCLUDE, ".*")); + delegate, + null, + ImmutableMap.of(CatalogProperties.METRICS_REPORTER_TABLE_NAME_INCLUDE, ".*")); wrapped.close(); @@ -171,6 +179,7 @@ public void includeAcceptsCommaSeparatedPatterns() { MetricsReporter wrapped = FilteringMetricsReporter.wrap( delegate, + null, ImmutableMap.of( CatalogProperties.METRICS_REPORTER_TABLE_NAME_INCLUDE, "prod_db\\..*, analytics_db\\..*")); @@ -189,6 +198,7 @@ public void excludeAcceptsCommaSeparatedPatterns() { MetricsReporter wrapped = FilteringMetricsReporter.wrap( delegate, + null, ImmutableMap.of( CatalogProperties.METRICS_REPORTER_TABLE_NAME_EXCLUDE, ".*\\.tmp_.*,dev_db\\..*")); @@ -205,6 +215,7 @@ public void patternsMatchWholeNameNotSubstring() { MetricsReporter wrapped = FilteringMetricsReporter.wrap( delegate, + null, ImmutableMap.of(CatalogProperties.METRICS_REPORTER_TABLE_NAME_INCLUDE, "prod\\..*")); ScanReport prod = newScanReport("prod.orders"); @@ -217,6 +228,157 @@ public void patternsMatchWholeNameNotSubstring() { assertThat(delegate.reports).containsExactly(prod); } + @Test + public void namespaceIncludeFiltersOnNamespaceOnly() { + CapturingMetricsReporter delegate = new CapturingMetricsReporter(); + MetricsReporter wrapped = + FilteringMetricsReporter.wrap( + delegate, + "cat", + ImmutableMap.of( + CatalogProperties.METRICS_REPORTER_NAMESPACE_INCLUDE, "prod,analytics")); + + ScanReport prod = newScanReport("cat.prod.orders"); + ScanReport analytics = newScanReport("cat.analytics.events"); + wrapped.report(prod); + wrapped.report(analytics); + wrapped.report(newScanReport("cat.staging.orders")); + // a namespace that only shares a prefix must not match + wrapped.report(newScanReport("cat.production.orders")); + + assertThat(delegate.reports).containsExactly(prod, analytics); + } + + @Test + public void namespaceExcludeDropsMatchingNamespaces() { + CapturingMetricsReporter delegate = new CapturingMetricsReporter(); + MetricsReporter wrapped = + FilteringMetricsReporter.wrap( + delegate, + "cat", + ImmutableMap.of( + CatalogProperties.METRICS_REPORTER_NAMESPACE_EXCLUDE, "staging,sandbox")); + + ScanReport prod = newScanReport("cat.prod.orders"); + wrapped.report(prod); + wrapped.report(newScanReport("cat.staging.orders")); + wrapped.report(newScanReport("cat.sandbox.orders")); + + assertThat(delegate.reports).containsExactly(prod); + } + + @Test + public void namespaceAndTableNameFiltersCombine() { + CapturingMetricsReporter delegate = new CapturingMetricsReporter(); + MetricsReporter wrapped = + FilteringMetricsReporter.wrap( + delegate, + "cat", + ImmutableMap.of( + CatalogProperties.METRICS_REPORTER_NAMESPACE_INCLUDE, "prod", + CatalogProperties.METRICS_REPORTER_TABLE_NAME_EXCLUDE, ".*\\.bench_.*")); + + ScanReport prod = newScanReport("cat.prod.orders"); + wrapped.report(prod); + // in the included namespace, but excluded by table name + wrapped.report(newScanReport("cat.prod.bench_scratch")); + // outside the included namespace + wrapped.report(newScanReport("cat.staging.orders")); + + assertThat(delegate.reports).containsExactly(prod); + } + + @Test + public void namespaceFilterHandlesMultiLevelAndEmptyNamespaces() { + CapturingMetricsReporter delegate = new CapturingMetricsReporter(); + MetricsReporter wrapped = + FilteringMetricsReporter.wrap( + delegate, + "cat", + ImmutableMap.of(CatalogProperties.METRICS_REPORTER_NAMESPACE_INCLUDE, "a\\.b")); + + ScanReport nested = newScanReport("cat.a.b.orders"); + wrapped.report(nested); + wrapped.report(newScanReport("cat.a.orders")); + // table directly under the catalog has an empty namespace + wrapped.report(newScanReport("cat.orders")); + + assertThat(delegate.reports).containsExactly(nested); + } + + @Test + public void namespaceFilterHandlesCatalogNameContainingDots() { + CapturingMetricsReporter delegate = new CapturingMetricsReporter(); + MetricsReporter wrapped = + FilteringMetricsReporter.wrap( + delegate, + "my.cat", + ImmutableMap.of(CatalogProperties.METRICS_REPORTER_NAMESPACE_INCLUDE, "db")); + + ScanReport report = newScanReport("my.cat.db.orders"); + wrapped.report(report); + wrapped.report(newScanReport("my.cat.other.orders")); + + assertThat(delegate.reports).containsExactly(report); + } + + @Test + public void namespaceFilterHandlesUriStyleCatalogNames() { + CapturingMetricsReporter delegate = new CapturingMetricsReporter(); + MetricsReporter wrapped = + FilteringMetricsReporter.wrap( + delegate, + "thrift://localhost:9083", + ImmutableMap.of(CatalogProperties.METRICS_REPORTER_NAMESPACE_INCLUDE, "db")); + + ScanReport report = newScanReport("thrift://localhost:9083/db.orders"); + wrapped.report(report); + wrapped.report(newScanReport("thrift://localhost:9083/other.orders")); + + assertThat(delegate.reports).containsExactly(report); + } + + @Test + public void namespaceFilterDropsReportsWithoutExpectedCatalogPrefix() { + CapturingMetricsReporter delegate = new CapturingMetricsReporter(); + MetricsReporter wrapped = + FilteringMetricsReporter.wrap( + delegate, + "cat", + ImmutableMap.of(CatalogProperties.METRICS_REPORTER_NAMESPACE_INCLUDE, ".*")); + + // the namespace cannot be derived, so the report cannot be shown to pass the filter + wrapped.report(newScanReport("other.db.orders")); + + assertThat(delegate.reports).isEmpty(); + } + + @Test + public void namespaceFilterWithoutCatalogNameIsRejected() { + assertThatThrownBy( + () -> + FilteringMetricsReporter.wrap( + new CapturingMetricsReporter(), + null, + ImmutableMap.of(CatalogProperties.METRICS_REPORTER_NAMESPACE_INCLUDE, "prod"))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining(CatalogProperties.METRICS_REPORTER_NAMESPACE_INCLUDE); + } + + @Test + public void wrapThrowsClearErrorForInvalidNamespaceRegex() { + assertThatThrownBy( + () -> + FilteringMetricsReporter.wrap( + new CapturingMetricsReporter(), + "cat", + ImmutableMap.of( + CatalogProperties.METRICS_REPORTER_NAMESPACE_EXCLUDE, "[invalid"))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining(CatalogProperties.METRICS_REPORTER_NAMESPACE_EXCLUDE) + .hasMessageContaining("[invalid"); + } + private static ScanReport newScanReport(String tableName) { return ImmutableScanReport.builder() .tableName(tableName) diff --git a/docs/docs/metrics-reporting.md b/docs/docs/metrics-reporting.md index 1431ade068d6..eb0b042befdd 100644 --- a/docs/docs/metrics-reporting.md +++ b/docs/docs/metrics-reporting.md @@ -147,28 +147,36 @@ public class InMemoryMetricsReporter implements MetricsReporter { The [catalog property](catalog-properties.md) `metrics-reporter-impl` allows registering a given [`MetricsReporter`](https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/metrics/MetricsReporter.java) by specifying its fully-qualified class name, e.g. `metrics-reporter-impl=org.apache.iceberg.metrics.InMemoryMetricsReporter`. -### Table-name filtering +### Filtering which tables are reported -Reports forwarded to the configured `MetricsReporter` can be filtered by table name using two additional catalog properties. Both accept a comma-separated list of Java regular expressions matched against `ScanReport.tableName()` and `CommitReport.tableName()`: +Reports forwarded to the configured `MetricsReporter` can be filtered on two levels, either of which may be used on its own: | Property | Effect | |---|---| +| `metrics-reporter.namespace.include` | Forward only reports for tables whose namespace matches; drop the rest. | +| `metrics-reporter.namespace.exclude` | Drop reports for tables whose namespace matches; forward the rest. | | `metrics-reporter.table-name.include` | Forward only reports whose table name matches; drop the rest. | | `metrics-reporter.table-name.exclude` | Drop reports whose table name matches; forward the rest. | -Patterns are matched against the **entire** table name rather than any substring of it. This matters in practice: `prod\..*` matches `prod.db.table` but not `production.db.table` or `prod_sandbox.db.table`, which a substring match would wrongly accept. +Each property accepts a comma-separated list of Java regular expressions. Namespace patterns are matched against the table's namespace levels joined by dots, with the catalog name removed; table-name patterns are matched against `ScanReport.tableName()` and `CommitReport.tableName()`, which include the catalog name. -When both are set, `exclude` wins over `include` (an explicit deny overrides an include). When neither is set, behavior is identical to today (every report is forwarded, with no runtime overhead). Empty values are treated as not set to avoid accidentally silencing all metrics on misconfiguration. +Patterns are matched against the **entire** namespace or table name rather than any substring of it. This matters in practice: `prod\..*` matches `prod.db.table` but not `production.db.table` or `prod_sandbox.db.table`, which a substring match would wrongly accept. -For example, to forward metrics for the `prod_db` and `analytics_db` databases while still dropping any temporary tables under them: +An `exclude` match always wins over an `include` match, and the two levels are applied independently — a report must survive both to be forwarded. When no property is set, behavior is identical to today: every report is forwarded, and no wrapper is instantiated. Empty values are treated as not set, to avoid accidentally silencing all metrics on misconfiguration. + +Filtering by namespace is less error-prone than filtering by table name, because a namespace is part of a table's identity rather than a naming convention: a table added to an included namespace later is picked up automatically, and no table outside that namespace can match by accident. Table-name patterns remain available for cases a namespace cannot express, such as excluding a few noisy tables inside an otherwise interesting namespace. + +For example, to report on the `prod` and `analytics` namespaces while dropping benchmark tables anywhere: ``` metrics-reporter-impl=org.apache.iceberg.metrics.LoggingMetricsReporter -metrics-reporter.table-name.include=prod_db\..*,analytics_db\..* -metrics-reporter.table-name.exclude=.*\.tmp_.* +metrics-reporter.namespace.include=prod,analytics +metrics-reporter.table-name.exclude=.*\.bench_.* ``` -The filter applies uniformly to all `MetricsReporter` implementations (`LoggingMetricsReporter`, `RESTMetricsReporter`, and custom user-supplied ones). Reports whose subtype does not expose a table name (i.e. anything other than `ScanReport` and `CommitReport`) are forwarded without filtering. +The filter applies uniformly to all `MetricsReporter` implementations (`LoggingMetricsReporter`, `RESTMetricsReporter`, and custom user-supplied ones). Reports whose subtype does not identify a table (i.e. anything other than `ScanReport` and `CommitReport`) are forwarded without filtering. + +Namespace filtering requires the catalog name, which the catalog supplies when it loads the reporter. Configuring a namespace filter where no catalog name is available fails at initialization with a clear error rather than silently dropping reports. ### Via the Java API during Scan planning