This is an automated email from the ASF dual-hosted git repository.
dimas pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/polaris.git
The following commit(s) were added to refs/heads/main by this push:
new b2eb9b683 feat(metrics): Add MetricsPersistence SPI for
backend-agnostic metrics storage (#3337) (#3616)
b2eb9b683 is described below
commit b2eb9b6833494f93dad2948c9708acb19efde732
Author: Anand K Sankaran <[email protected]>
AuthorDate: Fri Feb 13 15:07:09 2026 -0800
feat(metrics): Add MetricsPersistence SPI for backend-agnostic metrics
storage (#3337) (#3616)
* feat(metrics): Add MetricsPersistence SPI for backend-agnostic metrics
storage
This commit introduces a Service Provider Interface (SPI) for persisting
Iceberg metrics reports, addressing the extensibility concerns raised in
the design review.
Key components:
- MetricsPersistence: Main SPI interface with write and query operations
- NoOpMetricsPersistence: Default do-nothing implementation for backends
that don't support metrics persistence
- ScanMetricsRecord: Immutable interface for scan metrics data
- CommitMetricsRecord: Immutable interface for commit metrics data
- MetricsQueryCriteria: Query parameters with filtering and pagination
- MetricsContext: Context for conversion (realm, catalog, principal info)
- MetricsPersistenceFactory: Factory for realm-scoped instances
- MetricsRecordConverter: Converts Iceberg reports to SPI records
Design principles:
- Backend-agnostic: Can be implemented by JDBC, NoSQL, or custom backends
- No instanceof checks: Service code calls interface methods directly
- Idempotent writes: Same reportId written twice has no effect
- Graceful degradation: Unsupported backends return empty results
Relates to: #3337
* refactor(metrics): Remove ambient context fields from SPI records
Remove fields that can be obtained from ambient request context at write
time:
- principalName: Available from SecurityContext/PolarisPrincipal
- requestId: Not well-defined in Polaris; unclear what request it refers to
- otelTraceId/otelSpanId: Available from OTel context via Span.current()
Keep reportTraceId as it's a client-provided value from the report metadata
that cannot be obtained from the ambient context.
Rename otelTraceId filter in MetricsQueryCriteria to reportTraceId to match
the field that is actually stored in the records.
This keeps the SPI focused on business data (the metrics themselves) rather
than infrastructure concerns (tracing, authentication) which the persistence
implementation can obtain from the ambient context at write time if needed.
* refactor(metrics): Replace offset-based pagination with PageToken pattern
- Create ReportIdToken for cursor-based pagination using report ID (UUID)
- Remove limit() and offset() from MetricsQueryCriteria
- Update MetricsPersistence to use PageToken parameter and return Page<T>
- Update NoOpMetricsPersistence to return empty Page objects
- Register ReportIdToken via service loader
This change makes the SPI truly backend-agnostic by using the existing
Polaris PageToken pattern instead of RDBMS-specific offset pagination.
Each backend can implement cursors in their optimal way (keyset for RDBMS,
continuation tokens for NoSQL).
Addresses reviewer feedback on MetricsQueryCriteria.offset() field.
* Review comments
* Review comments
* refactor: Remove TableIdentifier and catalogName from SPI records
Per reviewer feedback:
- Replace Iceberg's TableIdentifier with separate namespace/tableName
strings
in MetricsRecordIdentity to avoid Iceberg dependencies in Polaris SPI
- Remove catalogName from records, keep only catalogId since catalog names
can change over time (via rename operations)
- Update MetricsQueryCriteria to use catalogId (OptionalLong) instead of
catalogName
- Update MetricsRecordConverter to extract namespace/tableName from
TableIdentifier
The service layer (MetricsRecordConverter) still accepts TableIdentifier and
performs the conversion to primitives for the SPI records.
* refactor: Use List<String> for namespace instead of dot-separated string
Per reviewer feedback, namespace is now represented as a List<String>
of individual levels rather than a dot-separated string. This avoids
ambiguity when namespace segments contain dots.
Changes:
- MetricsRecordIdentity: namespace() now returns List<String>
- MetricsQueryCriteria: namespace() now returns List<String>
- MetricsRecordConverter: namespaceToList() converts Iceberg Namespace
to List<String> using Arrays.asList()
The persistence implementation handles the serialization format.
* refactor: Use tableId instead of tableName in metrics records
Per reviewer feedback:
- r2766326028: Use table ID (same as catalog ID) since table names can
change
- r2766343275: Avoid denormalizing table names to prevent correctness issues
- r2766321215: Return builder with table info, add time ranges at call site
Changes:
- MetricsRecordIdentity: tableName() -> tableId() (long)
- MetricsQueryCriteria: tableName() -> tableId() (OptionalLong)
- MetricsQueryCriteria.forTable(): Returns builder with catalogId/tableId
- MetricsRecordConverter: tableIdentifier(TableIdentifier) -> tableId(long)
+ namespace(List<String>)
The caller (PersistingMetricsReporter) now needs to resolve table entity ID
before creating records, similar to how catalogId is resolved.
* refactor: Remove namespace from MetricsQueryCriteria
Per reviewer feedback - since we query by tableId, namespace is implicit.
If users want to query by namespace, the service layer should resolve
namespace to table IDs using the current catalog state, then query by
those IDs. This avoids confusion with table moves over time.
Namespace is still stored in MetricsRecordIdentity for display purposes.
* feat: mark Metrics Persistence SPI as @Beta (experimental)
Added @Beta annotation from Guava to all public types in the
Metrics Persistence SPI package to signal that this API is
experimental and may change in future releases.
Annotated types:
- MetricsPersistence
- ScanMetricsRecord
- CommitMetricsRecord
- MetricsQueryCriteria
- MetricsRecordIdentity
- ReportIdToken
* feat: Add timestamp() method to MetricsRecordConverter builders
Allow callers to specify the timestamp for metrics records, defaulting
to Instant.now() if not provided. This enables the reporter to use the
received timestamp rather than the conversion time.
* refactor: Remove namespace from MetricsRecordIdentity and
MetricsRecordConverter
Address PR review comment singhpk234#r2772722282 - remove namespace entirely
since tableId uniquely identifies the table. Namespace can be derived from
the table entity if needed.
---------
Co-authored-by: Anand Kumar Sankaran <[email protected]>
---
.../metrics/iceberg/MetricsRecordConverter.java | 272 +++++++++++++++++++++
.../persistence/metrics/CommitMetricsRecord.java | 126 ++++++++++
.../persistence/metrics/MetricsPersistence.java | 152 ++++++++++++
.../persistence/metrics/MetricsQueryCriteria.java | 165 +++++++++++++
.../persistence/metrics/MetricsRecordIdentity.java | 95 +++++++
.../metrics/NoOpMetricsPersistence.java | 60 +++++
.../core/persistence/metrics/ReportIdToken.java | 136 +++++++++++
.../persistence/metrics/ScanMetricsRecord.java | 125 ++++++++++
...ris.core.persistence.pagination.Token$TokenType | 1 +
9 files changed, 1132 insertions(+)
diff --git
a/polaris-core/src/main/java/org/apache/polaris/core/metrics/iceberg/MetricsRecordConverter.java
b/polaris-core/src/main/java/org/apache/polaris/core/metrics/iceberg/MetricsRecordConverter.java
new file mode 100644
index 000000000..0289ce276
--- /dev/null
+++
b/polaris-core/src/main/java/org/apache/polaris/core/metrics/iceberg/MetricsRecordConverter.java
@@ -0,0 +1,272 @@
+/*
+ * 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.polaris.core.metrics.iceberg;
+
+import java.time.Instant;
+import java.util.Collections;
+import java.util.Map;
+import java.util.Optional;
+import java.util.UUID;
+import org.apache.iceberg.metrics.CommitMetricsResult;
+import org.apache.iceberg.metrics.CommitReport;
+import org.apache.iceberg.metrics.CounterResult;
+import org.apache.iceberg.metrics.ScanMetricsResult;
+import org.apache.iceberg.metrics.ScanReport;
+import org.apache.iceberg.metrics.TimerResult;
+import org.apache.polaris.core.persistence.metrics.CommitMetricsRecord;
+import org.apache.polaris.core.persistence.metrics.ScanMetricsRecord;
+
+/**
+ * Converts Iceberg metrics reports to SPI record types using a fluent builder
API.
+ *
+ * <p>This converter extracts all relevant metrics from Iceberg's {@link
ScanReport} and {@link
+ * CommitReport} and combines them with context information to create
persistence-ready records.
+ *
+ * <p>Example usage:
+ *
+ * <pre>{@code
+ * ScanMetricsRecord record = MetricsRecordConverter.forScanReport(scanReport)
+ * .catalogId(catalog.getId())
+ * .tableId(tableEntity.getId())
+ * .build();
+ * }</pre>
+ */
+public final class MetricsRecordConverter {
+
+ private MetricsRecordConverter() {
+ // Utility class
+ }
+
+ /**
+ * Creates a builder for converting a ScanReport to a ScanMetricsRecord.
+ *
+ * @param scanReport the Iceberg scan report
+ * @return builder for configuring the conversion
+ */
+ public static ScanReportBuilder forScanReport(ScanReport scanReport) {
+ return new ScanReportBuilder(scanReport);
+ }
+
+ /**
+ * Creates a builder for converting a CommitReport to a CommitMetricsRecord.
+ *
+ * @param commitReport the Iceberg commit report
+ * @return builder for configuring the conversion
+ */
+ public static CommitReportBuilder forCommitReport(CommitReport commitReport)
{
+ return new CommitReportBuilder(commitReport);
+ }
+
+ /** Builder for converting ScanReport to ScanMetricsRecord. */
+ public static final class ScanReportBuilder {
+ private final ScanReport scanReport;
+ private long catalogId;
+ private long tableId;
+ private Instant timestamp;
+
+ private ScanReportBuilder(ScanReport scanReport) {
+ this.scanReport = scanReport;
+ }
+
+ public ScanReportBuilder catalogId(long catalogId) {
+ this.catalogId = catalogId;
+ return this;
+ }
+
+ /**
+ * Sets the table entity ID.
+ *
+ * <p>This is the internal Polaris entity ID for the table.
+ *
+ * @param tableId the table entity ID
+ * @return this builder
+ */
+ public ScanReportBuilder tableId(long tableId) {
+ this.tableId = tableId;
+ return this;
+ }
+
+ /**
+ * Sets the timestamp for the metrics record.
+ *
+ * <p>This should be the time the metrics report was received by the
server, which may differ
+ * from the time it was recorded by the client.
+ *
+ * @param timestamp the timestamp
+ * @return this builder
+ */
+ public ScanReportBuilder timestamp(Instant timestamp) {
+ this.timestamp = timestamp;
+ return this;
+ }
+
+ public ScanMetricsRecord build() {
+ ScanMetricsResult metrics = scanReport.scanMetrics();
+ Map<String, String> reportMetadata =
+ scanReport.metadata() != null ? scanReport.metadata() :
Collections.emptyMap();
+
+ return ScanMetricsRecord.builder()
+ .reportId(UUID.randomUUID().toString())
+ .catalogId(catalogId)
+ .tableId(tableId)
+ .timestamp(timestamp != null ? timestamp : Instant.now())
+ .snapshotId(Optional.of(scanReport.snapshotId()))
+ .schemaId(Optional.of(scanReport.schemaId()))
+ .filterExpression(
+ scanReport.filter() != null
+ ? Optional.of(scanReport.filter().toString())
+ : Optional.empty())
+ .projectedFieldIds(
+ scanReport.projectedFieldIds() != null
+ ? scanReport.projectedFieldIds()
+ : Collections.emptyList())
+ .projectedFieldNames(
+ scanReport.projectedFieldNames() != null
+ ? scanReport.projectedFieldNames()
+ : Collections.emptyList())
+ .resultDataFiles(getCounterValue(metrics.resultDataFiles()))
+ .resultDeleteFiles(getCounterValue(metrics.resultDeleteFiles()))
+ .totalFileSizeBytes(getCounterValue(metrics.totalFileSizeInBytes()))
+ .totalDataManifests(getCounterValue(metrics.totalDataManifests()))
+
.totalDeleteManifests(getCounterValue(metrics.totalDeleteManifests()))
+
.scannedDataManifests(getCounterValue(metrics.scannedDataManifests()))
+
.scannedDeleteManifests(getCounterValue(metrics.scannedDeleteManifests()))
+
.skippedDataManifests(getCounterValue(metrics.skippedDataManifests()))
+
.skippedDeleteManifests(getCounterValue(metrics.skippedDeleteManifests()))
+ .skippedDataFiles(getCounterValue(metrics.skippedDataFiles()))
+ .skippedDeleteFiles(getCounterValue(metrics.skippedDeleteFiles()))
+
.totalPlanningDurationMs(getTimerValueMs(metrics.totalPlanningDuration()))
+ .equalityDeleteFiles(getCounterValue(metrics.equalityDeleteFiles()))
+
.positionalDeleteFiles(getCounterValue(metrics.positionalDeleteFiles()))
+ .indexedDeleteFiles(getCounterValue(metrics.indexedDeleteFiles()))
+
.totalDeleteFileSizeBytes(getCounterValue(metrics.totalDeleteFileSizeInBytes()))
+ .metadata(reportMetadata)
+ .build();
+ }
+ }
+
+ /** Builder for converting CommitReport to CommitMetricsRecord. */
+ public static final class CommitReportBuilder {
+ private final CommitReport commitReport;
+ private long catalogId;
+ private long tableId;
+ private Instant timestamp;
+
+ private CommitReportBuilder(CommitReport commitReport) {
+ this.commitReport = commitReport;
+ }
+
+ public CommitReportBuilder catalogId(long catalogId) {
+ this.catalogId = catalogId;
+ return this;
+ }
+
+ /**
+ * Sets the table entity ID.
+ *
+ * <p>This is the internal Polaris entity ID for the table.
+ *
+ * @param tableId the table entity ID
+ * @return this builder
+ */
+ public CommitReportBuilder tableId(long tableId) {
+ this.tableId = tableId;
+ return this;
+ }
+
+ /**
+ * Sets the timestamp for the metrics record.
+ *
+ * <p>This should be the time the metrics report was received by the
server, which may differ
+ * from the time it was recorded by the client.
+ *
+ * @param timestamp the timestamp
+ * @return this builder
+ */
+ public CommitReportBuilder timestamp(Instant timestamp) {
+ this.timestamp = timestamp;
+ return this;
+ }
+
+ public CommitMetricsRecord build() {
+ CommitMetricsResult metrics = commitReport.commitMetrics();
+ Map<String, String> reportMetadata =
+ commitReport.metadata() != null ? commitReport.metadata() :
Collections.emptyMap();
+
+ return CommitMetricsRecord.builder()
+ .reportId(UUID.randomUUID().toString())
+ .catalogId(catalogId)
+ .tableId(tableId)
+ .timestamp(timestamp != null ? timestamp : Instant.now())
+ .snapshotId(commitReport.snapshotId())
+ .sequenceNumber(Optional.of(commitReport.sequenceNumber()))
+ .operation(commitReport.operation())
+ .addedDataFiles(getCounterValue(metrics.addedDataFiles()))
+ .removedDataFiles(getCounterValue(metrics.removedDataFiles()))
+ .totalDataFiles(getCounterValue(metrics.totalDataFiles()))
+ .addedDeleteFiles(getCounterValue(metrics.addedDeleteFiles()))
+ .removedDeleteFiles(getCounterValue(metrics.removedDeleteFiles()))
+ .totalDeleteFiles(getCounterValue(metrics.totalDeleteFiles()))
+
.addedEqualityDeleteFiles(getCounterValue(metrics.addedEqualityDeleteFiles()))
+
.removedEqualityDeleteFiles(getCounterValue(metrics.removedEqualityDeleteFiles()))
+
.addedPositionalDeleteFiles(getCounterValue(metrics.addedPositionalDeleteFiles()))
+
.removedPositionalDeleteFiles(getCounterValue(metrics.removedPositionalDeleteFiles()))
+ .addedRecords(getCounterValue(metrics.addedRecords()))
+ .removedRecords(getCounterValue(metrics.removedRecords()))
+ .totalRecords(getCounterValue(metrics.totalRecords()))
+ .addedFileSizeBytes(getCounterValue(metrics.addedFilesSizeInBytes()))
+
.removedFileSizeBytes(getCounterValue(metrics.removedFilesSizeInBytes()))
+ .totalFileSizeBytes(getCounterValue(metrics.totalFilesSizeInBytes()))
+ .totalDurationMs(getTimerValueMsOpt(metrics.totalDuration()))
+ .attempts(getCounterValueInt(metrics.attempts()))
+ .metadata(reportMetadata)
+ .build();
+ }
+ }
+
+ // === Helper Methods ===
+
+ private static long getCounterValue(CounterResult counter) {
+ if (counter == null) {
+ return 0L;
+ }
+ return counter.value();
+ }
+
+ private static int getCounterValueInt(CounterResult counter) {
+ if (counter == null) {
+ return 0;
+ }
+ return (int) counter.value();
+ }
+
+ private static long getTimerValueMs(TimerResult timer) {
+ if (timer == null || timer.totalDuration() == null) {
+ return 0L;
+ }
+ return timer.totalDuration().toMillis();
+ }
+
+ private static Optional<Long> getTimerValueMsOpt(TimerResult timer) {
+ if (timer == null || timer.totalDuration() == null) {
+ return Optional.empty();
+ }
+ return Optional.of(timer.totalDuration().toMillis());
+ }
+}
diff --git
a/polaris-core/src/main/java/org/apache/polaris/core/persistence/metrics/CommitMetricsRecord.java
b/polaris-core/src/main/java/org/apache/polaris/core/persistence/metrics/CommitMetricsRecord.java
new file mode 100644
index 000000000..6d67408cb
--- /dev/null
+++
b/polaris-core/src/main/java/org/apache/polaris/core/persistence/metrics/CommitMetricsRecord.java
@@ -0,0 +1,126 @@
+/*
+ * 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.polaris.core.persistence.metrics;
+
+import com.google.common.annotations.Beta;
+import java.util.Optional;
+import org.apache.polaris.immutables.PolarisImmutable;
+
+/**
+ * Backend-agnostic representation of an Iceberg commit metrics report.
+ *
+ * <p>This record captures all relevant metrics from an Iceberg {@code
CommitReport} along with
+ * contextual information such as catalog identification and table location.
+ *
+ * <p>Common identification fields are inherited from {@link
MetricsRecordIdentity}.
+ *
+ * <p>Note: Realm ID is not included in this record. Multi-tenancy realm
context should be obtained
+ * from the CDI-injected {@code RealmContext} at persistence time.
+ *
+ * <p><b>Note:</b> This type is part of the experimental Metrics Persistence
SPI and may change in
+ * future releases.
+ */
+@Beta
+@PolarisImmutable
+public interface CommitMetricsRecord extends MetricsRecordIdentity {
+
+ // === Commit Context ===
+
+ /** Snapshot ID created by this commit. */
+ long snapshotId();
+
+ /** Sequence number of the snapshot. */
+ Optional<Long> sequenceNumber();
+
+ /** Operation type (e.g., "append", "overwrite", "delete"). */
+ String operation();
+
+ // === File Metrics - Data Files ===
+
+ /** Number of data files added. */
+ long addedDataFiles();
+
+ /** Number of data files removed. */
+ long removedDataFiles();
+
+ /** Total number of data files after commit. */
+ long totalDataFiles();
+
+ // === File Metrics - Delete Files ===
+
+ /** Number of delete files added. */
+ long addedDeleteFiles();
+
+ /** Number of delete files removed. */
+ long removedDeleteFiles();
+
+ /** Total number of delete files after commit. */
+ long totalDeleteFiles();
+
+ /** Number of equality delete files added. */
+ long addedEqualityDeleteFiles();
+
+ /** Number of equality delete files removed. */
+ long removedEqualityDeleteFiles();
+
+ /** Number of positional delete files added. */
+ long addedPositionalDeleteFiles();
+
+ /** Number of positional delete files removed. */
+ long removedPositionalDeleteFiles();
+
+ // === Record Metrics ===
+
+ /** Number of records added. */
+ long addedRecords();
+
+ /** Number of records removed. */
+ long removedRecords();
+
+ /** Total number of records after commit. */
+ long totalRecords();
+
+ // === Size Metrics ===
+
+ /** Size of added files in bytes. */
+ long addedFileSizeBytes();
+
+ /** Size of removed files in bytes. */
+ long removedFileSizeBytes();
+
+ /** Total file size in bytes after commit. */
+ long totalFileSizeBytes();
+
+ // === Timing ===
+
+ /** Total duration of the commit in milliseconds. */
+ Optional<Long> totalDurationMs();
+
+ /** Number of commit attempts. */
+ int attempts();
+
+ /**
+ * Creates a new builder for CommitMetricsRecord.
+ *
+ * @return a new builder instance
+ */
+ static ImmutableCommitMetricsRecord.Builder builder() {
+ return ImmutableCommitMetricsRecord.builder();
+ }
+}
diff --git
a/polaris-core/src/main/java/org/apache/polaris/core/persistence/metrics/MetricsPersistence.java
b/polaris-core/src/main/java/org/apache/polaris/core/persistence/metrics/MetricsPersistence.java
new file mode 100644
index 000000000..1e9865701
--- /dev/null
+++
b/polaris-core/src/main/java/org/apache/polaris/core/persistence/metrics/MetricsPersistence.java
@@ -0,0 +1,152 @@
+/*
+ * 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.polaris.core.persistence.metrics;
+
+import com.google.common.annotations.Beta;
+import jakarta.annotation.Nonnull;
+import org.apache.polaris.core.persistence.pagination.Page;
+import org.apache.polaris.core.persistence.pagination.PageToken;
+
+/**
+ * Service Provider Interface (SPI) for persisting Iceberg metrics reports.
+ *
+ * <p>This interface enables different persistence backends (JDBC, NoSQL,
custom) to implement
+ * metrics storage in a way appropriate for their storage model, while
allowing service code to
+ * remain backend-agnostic.
+ *
+ * <p>Implementations should be idempotent - writing the same reportId twice
should have no effect.
+ * Implementations that don't support metrics persistence can use {@link
#NOOP} which silently
+ * ignores write operations and returns empty pages for queries.
+ *
+ * <h3>Dependency Injection</h3>
+ *
+ * <p>This interface is designed to be injected via CDI (Contexts and
Dependency Injection). The
+ * deployment module (e.g., {@code polaris-quarkus-service}) should provide a
{@code @Produces}
+ * method that creates the appropriate implementation based on the configured
persistence backend.
+ *
+ * <p>Example producer:
+ *
+ * <pre>{@code
+ * @Produces
+ * @RequestScoped
+ * MetricsPersistence metricsPersistence(RealmContext realmContext,
PersistenceBackend backend) {
+ * if (backend.supportsMetrics()) {
+ * return backend.createMetricsPersistence(realmContext);
+ * }
+ * return MetricsPersistence.NOOP;
+ * }
+ * }</pre>
+ *
+ * <h3>Multi-Tenancy</h3>
+ *
+ * <p>Realm context is not passed in the record objects. Implementations
should obtain the realm
+ * from the CDI-injected {@code RealmContext} at write/query time. This keeps
catalog-specific code
+ * from needing to manage realm concerns directly.
+ *
+ * <h3>Pagination</h3>
+ *
+ * <p>Query methods use the standard Polaris pagination pattern with {@link
PageToken} for requests
+ * and {@link Page} for responses. This enables:
+ *
+ * <ul>
+ * <li>Backend-specific cursor implementations (RDBMS offset, NoSQL
continuation tokens, etc.)
+ * <li>Consistent pagination interface across all Polaris persistence APIs
+ * <li>Efficient cursor-based pagination that works with large result sets
+ * </ul>
+ *
+ * <p>The {@link ReportIdToken} provides a reference cursor implementation
based on report ID
+ * (UUID), but backends may use other cursor strategies internally.
+ *
+ * <p><b>Note:</b> This SPI is currently experimental and not yet implemented
in all persistence
+ * backends. The API may change in future releases.
+ *
+ * @see PageToken
+ * @see Page
+ * @see ReportIdToken
+ */
+@Beta
+public interface MetricsPersistence {
+
+ /** A no-op implementation for backends that don't support metrics
persistence. */
+ MetricsPersistence NOOP = new NoOpMetricsPersistence();
+
+ //
============================================================================
+ // Write Operations
+ //
============================================================================
+
+ /**
+ * Persists a scan metrics record.
+ *
+ * <p>This operation is idempotent - writing the same reportId twice has no
effect.
+ *
+ * @param record the scan metrics record to persist
+ */
+ void writeScanReport(@Nonnull ScanMetricsRecord record);
+
+ /**
+ * Persists a commit metrics record.
+ *
+ * <p>This operation is idempotent - writing the same reportId twice has no
effect.
+ *
+ * @param record the commit metrics record to persist
+ */
+ void writeCommitReport(@Nonnull CommitMetricsRecord record);
+
+ //
============================================================================
+ // Query Operations
+ //
============================================================================
+
+ /**
+ * Queries scan metrics reports based on the specified criteria.
+ *
+ * <p>Example usage:
+ *
+ * <pre>{@code
+ * // First page
+ * PageToken pageToken = PageToken.fromLimit(100);
+ * Page<ScanMetricsRecord> page = persistence.queryScanReports(criteria,
pageToken);
+ *
+ * // Next page (if available)
+ * String nextPageToken = page.encodedResponseToken();
+ * if (nextPageToken != null) {
+ * pageToken = PageToken.build(nextPageToken, null, () -> true);
+ * Page<ScanMetricsRecord> nextPage =
persistence.queryScanReports(criteria, pageToken);
+ * }
+ * }</pre>
+ *
+ * @param criteria the query criteria (filters)
+ * @param pageToken pagination parameters (page size and optional cursor)
+ * @return page of matching scan metrics records with continuation token if
more results exist
+ */
+ @Nonnull
+ Page<ScanMetricsRecord> queryScanReports(
+ @Nonnull MetricsQueryCriteria criteria, @Nonnull PageToken pageToken);
+
+ /**
+ * Queries commit metrics reports based on the specified criteria.
+ *
+ * @param criteria the query criteria (filters)
+ * @param pageToken pagination parameters (page size and optional cursor)
+ * @return page of matching commit metrics records with continuation token
if more results exist
+ * @see #queryScanReports(MetricsQueryCriteria, PageToken) for pagination
example
+ */
+ @Nonnull
+ Page<CommitMetricsRecord> queryCommitReports(
+ @Nonnull MetricsQueryCriteria criteria, @Nonnull PageToken pageToken);
+}
diff --git
a/polaris-core/src/main/java/org/apache/polaris/core/persistence/metrics/MetricsQueryCriteria.java
b/polaris-core/src/main/java/org/apache/polaris/core/persistence/metrics/MetricsQueryCriteria.java
new file mode 100644
index 000000000..210fa3909
--- /dev/null
+++
b/polaris-core/src/main/java/org/apache/polaris/core/persistence/metrics/MetricsQueryCriteria.java
@@ -0,0 +1,165 @@
+/*
+ * 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.polaris.core.persistence.metrics;
+
+import com.google.common.annotations.Beta;
+import java.time.Instant;
+import java.util.Map;
+import java.util.Optional;
+import java.util.OptionalLong;
+import org.apache.polaris.immutables.PolarisImmutable;
+
+/**
+ * Query criteria for retrieving metrics reports.
+ *
+ * <p>This class defines the filter parameters for metrics queries. Pagination
is handled separately
+ * via {@link org.apache.polaris.core.persistence.pagination.PageToken}, which
is passed as a
+ * separate parameter to query methods. This separation of concerns allows:
+ *
+ * <ul>
+ * <li>Different backends to implement pagination in their optimal way
+ * <li>Cursor-based pagination that works with both RDBMS and NoSQL backends
+ * <li>Reuse of the existing Polaris pagination infrastructure
+ * </ul>
+ *
+ * <h3>Supported Query Patterns</h3>
+ *
+ * <table>
+ * <tr><th>Pattern</th><th>Fields Used</th><th>Index Required</th></tr>
+ * <tr><td>By Table + Time</td><td>catalogId, tableId, startTime,
endTime</td><td>Yes (OSS)</td></tr>
+ * <tr><td>By Time Only</td><td>startTime, endTime</td><td>Partial (timestamp
index)</td></tr>
+ * </table>
+ *
+ * <p>Additional query patterns (e.g., by trace ID) can be implemented by
persistence backends using
+ * the {@link #metadata()} filter map. Client-provided correlation data should
be stored in the
+ * metrics record's metadata map and can be filtered using the metadata
criteria.
+ *
+ * <p><b>Note:</b> This type is part of the experimental Metrics Persistence
SPI and may change in
+ * future releases.
+ *
+ * <h3>Pagination</h3>
+ *
+ * <p>Pagination is handled via the {@link
org.apache.polaris.core.persistence.pagination.PageToken}
+ * passed to query methods. The token contains:
+ *
+ * <ul>
+ * <li>{@code pageSize()} - Maximum number of results to return
+ * <li>{@code value()} - Optional cursor token (e.g., {@link ReportIdToken})
for continuation
+ * </ul>
+ *
+ * <p>Query results are returned as {@link
org.apache.polaris.core.persistence.pagination.Page}
+ * which includes an encoded token for fetching the next page.
+ *
+ * @see org.apache.polaris.core.persistence.pagination.PageToken
+ * @see org.apache.polaris.core.persistence.pagination.Page
+ * @see ReportIdToken
+ */
+@Beta
+@PolarisImmutable
+public interface MetricsQueryCriteria {
+
+ // === Table Identification (optional) ===
+
+ /**
+ * Catalog ID to filter by.
+ *
+ * <p>This is the internal catalog entity ID. Callers should resolve catalog
names to IDs before
+ * querying, as catalog names can change over time.
+ */
+ OptionalLong catalogId();
+
+ /**
+ * Table entity ID to filter by.
+ *
+ * <p>This is the internal table entity ID. Callers should resolve table
names to IDs before
+ * querying, as table names can change over time.
+ *
+ * <p>Note: Namespace is intentionally not included as a query filter. Since
we query by table ID,
+ * the namespace is implicit. If users want to query by namespace, the
service layer should
+ * resolve namespace to table IDs using the current catalog state, then
query by those IDs. This
+ * avoids confusion with table moves over time.
+ */
+ OptionalLong tableId();
+
+ // === Time Range ===
+
+ /** Start time for the query (inclusive). */
+ Optional<Instant> startTime();
+
+ /** End time for the query (exclusive). */
+ Optional<Instant> endTime();
+
+ // === Metadata Filtering ===
+
+ /**
+ * Metadata key-value pairs to filter by.
+ *
+ * <p>This enables filtering metrics by client-provided correlation data
stored in the record's
+ * metadata map. For example, clients may include a trace ID in the metadata
that can be queried
+ * later.
+ *
+ * <p>Note: Metadata filtering may require custom indexes depending on the
persistence backend.
+ * The OSS codebase provides basic support, but performance optimizations
may be needed for
+ * high-volume deployments.
+ */
+ Map<String, String> metadata();
+
+ // === Factory Methods ===
+
+ /**
+ * Creates a new builder for MetricsQueryCriteria.
+ *
+ * @return a new builder instance
+ */
+ static ImmutableMetricsQueryCriteria.Builder builder() {
+ return ImmutableMetricsQueryCriteria.builder();
+ }
+
+ /**
+ * Creates a builder pre-populated with table identification info.
+ *
+ * <p>This allows the caller to add time ranges and other filters at the
call site. This pattern
+ * is useful when table info is resolved in one place and time ranges are
added elsewhere.
+ *
+ * <p>Example usage:
+ *
+ * <pre>{@code
+ * MetricsQueryCriteria criteria = MetricsQueryCriteria.forTable(catalogId,
tableId)
+ * .startTime(startTime)
+ * .endTime(endTime)
+ * .build();
+ * }</pre>
+ *
+ * @param catalogId the catalog entity ID
+ * @param tableId the table entity ID
+ * @return a builder pre-populated with table info, ready for adding time
ranges
+ */
+ static ImmutableMetricsQueryCriteria.Builder forTable(long catalogId, long
tableId) {
+ return builder().catalogId(catalogId).tableId(tableId);
+ }
+
+ /**
+ * Creates empty criteria (no filters). Useful for pagination-only queries.
+ *
+ * @return empty query criteria
+ */
+ static MetricsQueryCriteria empty() {
+ return builder().build();
+ }
+}
diff --git
a/polaris-core/src/main/java/org/apache/polaris/core/persistence/metrics/MetricsRecordIdentity.java
b/polaris-core/src/main/java/org/apache/polaris/core/persistence/metrics/MetricsRecordIdentity.java
new file mode 100644
index 000000000..51f819ea0
--- /dev/null
+++
b/polaris-core/src/main/java/org/apache/polaris/core/persistence/metrics/MetricsRecordIdentity.java
@@ -0,0 +1,95 @@
+/*
+ * 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.polaris.core.persistence.metrics;
+
+import com.google.common.annotations.Beta;
+import java.time.Instant;
+import java.util.Map;
+
+/**
+ * Base interface containing common identification fields shared by all
metrics records.
+ *
+ * <p>This interface defines the common fields that identify the source of a
metrics report,
+ * including the report ID, catalog ID, table ID, timestamp, and metadata.
+ *
+ * <p>Both {@link ScanMetricsRecord} and {@link CommitMetricsRecord} extend
this interface to
+ * inherit these common fields while adding their own specific metrics.
+ *
+ * <h3>Design Decisions</h3>
+ *
+ * <p><b>Entity IDs only (no names):</b> We store only catalog ID and table
ID, not their names or
+ * namespace paths. Names can change over time (via rename operations), which
would make querying
+ * historical metrics by name challenging and lead to correctness issues.
Queries should resolve
+ * names to IDs using the current catalog state. The table ID uniquely
identifies the table, and the
+ * namespace can be derived from the table entity if needed.
+ *
+ * <p><b>Realm ID:</b> Realm ID is intentionally not included in this
interface. Multi-tenancy realm
+ * context should be obtained from the CDI-injected {@code RealmContext} at
persistence time. This
+ * keeps catalog-specific code from needing to manage realm concerns.
+ *
+ * <p><b>Note:</b> This type is part of the experimental Metrics Persistence
SPI and may change in
+ * future releases.
+ */
+@Beta
+public interface MetricsRecordIdentity {
+
+ /**
+ * Unique identifier for this report (UUID).
+ *
+ * <p>This ID is generated when the record is created and serves as the
primary key for the
+ * metrics record in persistence storage.
+ */
+ String reportId();
+
+ /**
+ * Internal catalog ID.
+ *
+ * <p>This matches the catalog entity ID in Polaris persistence, as defined
by {@code
+ * PolarisEntityCore#getId()}. The catalog name is not stored since it can
change over time;
+ * queries should resolve names to IDs using the current catalog state.
+ */
+ long catalogId();
+
+ /**
+ * Internal table entity ID.
+ *
+ * <p>This matches the table entity ID in Polaris persistence, as defined by
{@code
+ * PolarisEntityCore#getId()}. The table name is not stored since it can
change over time; queries
+ * should resolve names to IDs using the current catalog state. The
namespace can be derived from
+ * the table entity if needed.
+ */
+ long tableId();
+
+ /**
+ * Timestamp when the report was received.
+ *
+ * <p>This is the server-side timestamp when the metrics report was
processed, not the client-side
+ * timestamp when the operation occurred.
+ */
+ Instant timestamp();
+
+ /**
+ * Additional metadata as key-value pairs.
+ *
+ * <p>This map can contain additional contextual information from the
original Iceberg report,
+ * including client-provided trace IDs or other correlation data.
Persistence implementations can
+ * store and index specific metadata fields as needed.
+ */
+ Map<String, String> metadata();
+}
diff --git
a/polaris-core/src/main/java/org/apache/polaris/core/persistence/metrics/NoOpMetricsPersistence.java
b/polaris-core/src/main/java/org/apache/polaris/core/persistence/metrics/NoOpMetricsPersistence.java
new file mode 100644
index 000000000..b33c095dc
--- /dev/null
+++
b/polaris-core/src/main/java/org/apache/polaris/core/persistence/metrics/NoOpMetricsPersistence.java
@@ -0,0 +1,60 @@
+/*
+ * 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.polaris.core.persistence.metrics;
+
+import jakarta.annotation.Nonnull;
+import java.util.Collections;
+import org.apache.polaris.core.persistence.pagination.Page;
+import org.apache.polaris.core.persistence.pagination.PageToken;
+
+/**
+ * A no-op implementation of {@link MetricsPersistence} for backends that
don't support metrics
+ * persistence.
+ *
+ * <p>This implementation is used as the default when a persistence backend
does not support metrics
+ * storage. All write operations are silently ignored, and all query
operations return empty pages.
+ */
+final class NoOpMetricsPersistence implements MetricsPersistence {
+
+ NoOpMetricsPersistence() {}
+
+ @Override
+ public void writeScanReport(@Nonnull ScanMetricsRecord record) {
+ // No-op
+ }
+
+ @Override
+ public void writeCommitReport(@Nonnull CommitMetricsRecord record) {
+ // No-op
+ }
+
+ @Nonnull
+ @Override
+ public Page<ScanMetricsRecord> queryScanReports(
+ @Nonnull MetricsQueryCriteria criteria, @Nonnull PageToken pageToken) {
+ return Page.fromItems(Collections.emptyList());
+ }
+
+ @Nonnull
+ @Override
+ public Page<CommitMetricsRecord> queryCommitReports(
+ @Nonnull MetricsQueryCriteria criteria, @Nonnull PageToken pageToken) {
+ return Page.fromItems(Collections.emptyList());
+ }
+}
diff --git
a/polaris-core/src/main/java/org/apache/polaris/core/persistence/metrics/ReportIdToken.java
b/polaris-core/src/main/java/org/apache/polaris/core/persistence/metrics/ReportIdToken.java
new file mode 100644
index 000000000..f3e384695
--- /dev/null
+++
b/polaris-core/src/main/java/org/apache/polaris/core/persistence/metrics/ReportIdToken.java
@@ -0,0 +1,136 @@
+/*
+ * 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.polaris.core.persistence.metrics;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+import com.google.common.annotations.Beta;
+import jakarta.annotation.Nullable;
+import org.apache.polaris.core.persistence.pagination.Token;
+import org.apache.polaris.immutables.PolarisImmutable;
+
+/**
+ * Pagination {@linkplain Token token} for metrics queries, backed by the
report ID (UUID).
+ *
+ * <p><strong>Note:</strong> This is a reference implementation provided for
convenience. It is
+ * <em>not required</em> by the {@link MetricsPersistence} SPI contract.
Persistence backends are
+ * free to implement their own {@link Token} subclass optimized for their
storage model (e.g.,
+ * timestamp-based cursors, composite keys, continuation tokens).
+ *
+ * <p>Only {@link org.apache.polaris.core.persistence.pagination.PageToken}
(for requests) and
+ * {@link org.apache.polaris.core.persistence.pagination.Page} (for responses)
are required by the
+ * SPI contract.
+ *
+ * <p>This token enables cursor-based pagination for metrics queries across
different storage
+ * backends. The report ID is used as the cursor because it is:
+ *
+ * <ul>
+ * <li>Guaranteed unique across all reports
+ * <li>Present in both scan and commit metrics records
+ * <li>Stable (doesn't change over time)
+ * </ul>
+ *
+ * <p>Each backend implementation can use this cursor value to implement
efficient pagination in
+ * whatever way is optimal for that storage system:
+ *
+ * <ul>
+ * <li>RDBMS: {@code WHERE report_id > :lastReportId ORDER BY report_id}
+ * <li>NoSQL: Use report ID as partition/sort key cursor
+ * <li>Time-series: Combine with timestamp for efficient range scans
+ * </ul>
+ *
+ * <p><b>Note:</b> This type is part of the experimental Metrics Persistence
SPI and may change in
+ * future releases.
+ */
+@Beta
+@PolarisImmutable
+@JsonSerialize(as = ImmutableReportIdToken.class)
+@JsonDeserialize(as = ImmutableReportIdToken.class)
+public interface ReportIdToken extends Token {
+
+ /** Token type identifier. Short to minimize serialized token size. */
+ String ID = "r";
+
+ /**
+ * The report ID to use as the cursor.
+ *
+ * <p>Results should start after this report ID. This is typically the
{@code reportId} of the
+ * last item from the previous page.
+ */
+ @JsonProperty("r")
+ String reportId();
+
+ @Override
+ default String getT() {
+ return ID;
+ }
+
+ /**
+ * Creates a token from a report ID.
+ *
+ * @param reportId the report ID to use as cursor
+ * @return the token, or null if reportId is null
+ */
+ static @Nullable ReportIdToken fromReportId(@Nullable String reportId) {
+ if (reportId == null) {
+ return null;
+ }
+ return ImmutableReportIdToken.builder().reportId(reportId).build();
+ }
+
+ /**
+ * Creates a token from a metrics record.
+ *
+ * @param record the record whose report ID should be used as cursor
+ * @return the token, or null if record is null
+ */
+ static @Nullable ReportIdToken fromRecord(@Nullable ScanMetricsRecord
record) {
+ if (record == null) {
+ return null;
+ }
+ return fromReportId(record.reportId());
+ }
+
+ /**
+ * Creates a token from a commit metrics record.
+ *
+ * @param record the record whose report ID should be used as cursor
+ * @return the token, or null if record is null
+ */
+ static @Nullable ReportIdToken fromRecord(@Nullable CommitMetricsRecord
record) {
+ if (record == null) {
+ return null;
+ }
+ return fromReportId(record.reportId());
+ }
+
+ /** Token type registration for service loader. */
+ final class ReportIdTokenType implements TokenType {
+ @Override
+ public String id() {
+ return ID;
+ }
+
+ @Override
+ public Class<? extends Token> javaType() {
+ return ReportIdToken.class;
+ }
+ }
+}
diff --git
a/polaris-core/src/main/java/org/apache/polaris/core/persistence/metrics/ScanMetricsRecord.java
b/polaris-core/src/main/java/org/apache/polaris/core/persistence/metrics/ScanMetricsRecord.java
new file mode 100644
index 000000000..44947d8f7
--- /dev/null
+++
b/polaris-core/src/main/java/org/apache/polaris/core/persistence/metrics/ScanMetricsRecord.java
@@ -0,0 +1,125 @@
+/*
+ * 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.polaris.core.persistence.metrics;
+
+import com.google.common.annotations.Beta;
+import java.util.List;
+import java.util.Optional;
+import org.apache.polaris.immutables.PolarisImmutable;
+
+/**
+ * Backend-agnostic representation of an Iceberg scan metrics report.
+ *
+ * <p>This record captures all relevant metrics from an Iceberg {@code
ScanReport} along with
+ * contextual information such as catalog identification and table location.
+ *
+ * <p>Common identification fields are inherited from {@link
MetricsRecordIdentity}.
+ *
+ * <p>Note: Realm ID is not included in this record. Multi-tenancy realm
context should be obtained
+ * from the CDI-injected {@code RealmContext} at persistence time.
+ *
+ * <p><b>Note:</b> This type is part of the experimental Metrics Persistence
SPI and may change in
+ * future releases.
+ */
+@Beta
+@PolarisImmutable
+public interface ScanMetricsRecord extends MetricsRecordIdentity {
+
+ // === Scan Context ===
+
+ /** Snapshot ID that was scanned. */
+ Optional<Long> snapshotId();
+
+ /** Schema ID used for the scan. */
+ Optional<Integer> schemaId();
+
+ /** Filter expression applied to the scan (as string). */
+ Optional<String> filterExpression();
+
+ /** List of projected field IDs. */
+ List<Integer> projectedFieldIds();
+
+ /** List of projected field names. */
+ List<String> projectedFieldNames();
+
+ // === Scan Metrics - File Counts ===
+
+ /** Number of data files in the result. */
+ long resultDataFiles();
+
+ /** Number of delete files in the result. */
+ long resultDeleteFiles();
+
+ /** Total size of files in bytes. */
+ long totalFileSizeBytes();
+
+ // === Scan Metrics - Manifest Counts ===
+
+ /** Total number of data manifests. */
+ long totalDataManifests();
+
+ /** Total number of delete manifests. */
+ long totalDeleteManifests();
+
+ /** Number of data manifests that were scanned. */
+ long scannedDataManifests();
+
+ /** Number of delete manifests that were scanned. */
+ long scannedDeleteManifests();
+
+ /** Number of data manifests that were skipped. */
+ long skippedDataManifests();
+
+ /** Number of delete manifests that were skipped. */
+ long skippedDeleteManifests();
+
+ /** Number of data files that were skipped. */
+ long skippedDataFiles();
+
+ /** Number of delete files that were skipped. */
+ long skippedDeleteFiles();
+
+ // === Scan Metrics - Timing ===
+
+ /** Total planning duration in milliseconds. */
+ long totalPlanningDurationMs();
+
+ // === Scan Metrics - Delete Files ===
+
+ /** Number of equality delete files. */
+ long equalityDeleteFiles();
+
+ /** Number of positional delete files. */
+ long positionalDeleteFiles();
+
+ /** Number of indexed delete files. */
+ long indexedDeleteFiles();
+
+ /** Total size of delete files in bytes. */
+ long totalDeleteFileSizeBytes();
+
+ /**
+ * Creates a new builder for ScanMetricsRecord.
+ *
+ * @return a new builder instance
+ */
+ static ImmutableScanMetricsRecord.Builder builder() {
+ return ImmutableScanMetricsRecord.builder();
+ }
+}
diff --git
a/polaris-core/src/main/resources/META-INF/services/org.apache.polaris.core.persistence.pagination.Token$TokenType
b/polaris-core/src/main/resources/META-INF/services/org.apache.polaris.core.persistence.pagination.Token$TokenType
index 3579dd29b..d496ebedd 100644
---
a/polaris-core/src/main/resources/META-INF/services/org.apache.polaris.core.persistence.pagination.Token$TokenType
+++
b/polaris-core/src/main/resources/META-INF/services/org.apache.polaris.core.persistence.pagination.Token$TokenType
@@ -18,3 +18,4 @@
#
org.apache.polaris.core.persistence.pagination.EntityIdToken$EntityIdTokenType
+org.apache.polaris.core.persistence.metrics.ReportIdToken$ReportIdTokenType
