flyingImer commented on code in PR #4115: URL: https://github.com/apache/polaris/pull/4115#discussion_r3313540500
########## extensions/metrics-reports/impl/src/main/java/org/apache/polaris/extension/metrics/reports/MetricsReportsService.java: ########## @@ -0,0 +1,277 @@ +/* + * 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.extension.metrics.reports; + +import jakarta.annotation.Nonnull; +import jakarta.enterprise.context.RequestScoped; +import jakarta.inject.Inject; +import jakarta.ws.rs.core.Response; +import jakarta.ws.rs.core.SecurityContext; +import java.util.Arrays; +import java.util.List; +import java.util.stream.Collectors; +import org.apache.iceberg.catalog.Namespace; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.exceptions.NotFoundException; +import org.apache.polaris.core.auth.PolarisAuthorizableOperation; +import org.apache.polaris.core.auth.PolarisAuthorizer; +import org.apache.polaris.core.auth.PolarisPrincipal; +import org.apache.polaris.core.catalog.PolarisCatalogHelpers; +import org.apache.polaris.core.context.CallContext; +import org.apache.polaris.core.context.RealmContext; +import org.apache.polaris.core.entity.PolarisEntity; +import org.apache.polaris.core.entity.PolarisEntitySubType; +import org.apache.polaris.core.entity.PolarisEntityType; +import org.apache.polaris.core.persistence.PolarisMetaStoreManager; +import org.apache.polaris.core.persistence.PolarisResolvedPathWrapper; +import org.apache.polaris.core.persistence.metrics.CommitMetricsRecord; +import org.apache.polaris.core.persistence.metrics.ScanMetricsRecord; +import org.apache.polaris.core.persistence.pagination.Page; +import org.apache.polaris.core.persistence.pagination.PageToken; +import org.apache.polaris.core.persistence.resolver.PolarisResolutionManifest; +import org.apache.polaris.core.persistence.resolver.ResolutionManifestFactory; +import org.apache.polaris.core.persistence.resolver.ResolvedPathKey; +import org.apache.polaris.core.persistence.resolver.ResolverPath; +import org.apache.polaris.core.persistence.resolver.ResolverStatus; +import org.apache.polaris.service.metrics.api.PolarisCatalogsApiService; + +/** + * Service implementation for the Metrics Reports API. + * + * <p>Resolves catalog/namespace/table names to internal IDs, performs authorization, and delegates + * to the {@link org.apache.polaris.core.persistence.metrics.MetricsPersistence} layer to retrieve + * persisted metrics reports. + */ +@RequestScoped +public class MetricsReportsService implements PolarisCatalogsApiService { + + private final CallContext callContext; + private final PolarisMetaStoreManager metaStoreManager; + private final PolarisAuthorizer authorizer; + private final PolarisPrincipal polarisPrincipal; + private final ResolutionManifestFactory resolutionManifestFactory; + + @Inject + public MetricsReportsService( + @Nonnull CallContext callContext, + @Nonnull PolarisMetaStoreManager metaStoreManager, + @Nonnull PolarisAuthorizer authorizer, + @Nonnull PolarisPrincipal polarisPrincipal, + @Nonnull ResolutionManifestFactory resolutionManifestFactory) { + this.callContext = callContext; + this.metaStoreManager = metaStoreManager; + this.authorizer = authorizer; + this.polarisPrincipal = polarisPrincipal; + this.resolutionManifestFactory = resolutionManifestFactory; + } + + @Override + public Response listTableMetrics( + String catalogName, + String namespace, + String table, + String metricType, + String pageToken, + Integer pageSize, + Long snapshotId, + String principalName, + Long timestampFrom, + Long timestampTo, + RealmContext realmContext, + SecurityContext securityContext) { + + Namespace ns = decodeNamespace(namespace); + TableIdentifier identifier = TableIdentifier.of(ns, table); + + // Resolve and authorize + PolarisResolvedPathWrapper tableWrapper = + resolveAndAuthorizeTableMetrics(catalogName, identifier); + + PolarisEntity tableEntity = tableWrapper.getResolvedLeafEntity().getEntity(); + long catalogId = tableEntity.getCatalogId(); + long tableId = tableEntity.getId(); + + PageToken token = PageToken.build(pageToken, pageSize, () -> true); + + return switch (metricType) { + case "scan" -> { + Page<ScanMetricsRecord> page = + metaStoreManager.listScanMetrics( + callContext.getPolarisCallContext(), + catalogId, + tableId, + snapshotId, + principalName, + timestampFrom, + timestampTo, + token); + yield Response.ok(buildScanResponse(page)).build(); + } + case "commit" -> { + Page<CommitMetricsRecord> page = + metaStoreManager.listCommitMetrics( + callContext.getPolarisCallContext(), + catalogId, + tableId, + snapshotId, + principalName, + timestampFrom, + timestampTo, + token); + yield Response.ok(buildCommitResponse(page)).build(); + } + default -> + throw new IllegalArgumentException( + "Invalid metricType: " + metricType + "; must be 'scan' or 'commit'"); + }; + } + + private PolarisResolvedPathWrapper resolveAndAuthorizeTableMetrics( + String catalogName, TableIdentifier identifier) { + PolarisResolutionManifest manifest = + resolutionManifestFactory.createResolutionManifest(polarisPrincipal, catalogName); + manifest.addPassthroughPath( + new ResolverPath( + Arrays.asList(identifier.namespace().levels()), PolarisEntityType.NAMESPACE)); + manifest.addPassthroughPath( + new ResolverPath( + PolarisCatalogHelpers.tableIdentifierToList(identifier), PolarisEntityType.TABLE_LIKE)); + ResolverStatus status = manifest.resolveAll(); + + if (status.getStatus() == ResolverStatus.StatusEnum.ENTITY_COULD_NOT_BE_RESOLVED) { + throw new NotFoundException( + "TopLevelEntity of type %s does not exist: %s", + status.getFailedToResolvedEntityType(), status.getFailedToResolvedEntityName()); + } + if (status.getStatus() == ResolverStatus.StatusEnum.PATH_COULD_NOT_BE_FULLY_RESOLVED) { + throw new NotFoundException("Table not found: %s", identifier); + } + + PolarisResolvedPathWrapper tableWrapper = + manifest.getResolvedPath( + ResolvedPathKey.ofTableLike(identifier), PolarisEntitySubType.ANY_SUBTYPE, true); + + if (tableWrapper == null) { + throw new NotFoundException("Table not found: %s", identifier); + } + + authorizer.authorizeOrThrow( + polarisPrincipal, + manifest.getAllActivatedCatalogRoleAndPrincipalRoles(), + PolarisAuthorizableOperation.LIST_TABLE_METRICS, + tableWrapper, + null); + + return tableWrapper; + } + + private static Namespace decodeNamespace(String encodedNamespace) { + if (encodedNamespace == null || encodedNamespace.isEmpty()) { + throw new IllegalArgumentException("namespace must not be empty"); + } + // Multi-level namespaces use unit separator (0x1F / %1F) between levels + String[] levels = encodedNamespace.split("\u001F", -1); + return Namespace.of(levels); + } + + private static MetricsListResponse<ScanMetricsReport> buildScanResponse( + Page<ScanMetricsRecord> page) { + List<ScanMetricsReport> reports = + page.items().stream().map(MetricsReportsService::toScanReport).toList(); + return new MetricsListResponse<>("scan", page.encodedResponseToken(), reports); + } + + private static MetricsListResponse<CommitMetricsReport> buildCommitResponse( + Page<CommitMetricsRecord> page) { + List<CommitMetricsReport> reports = + page.items().stream().map(MetricsReportsService::toCommitReport).toList(); + return new MetricsListResponse<>("commit", page.encodedResponseToken(), reports); + } + + private static ScanMetricsReport toScanReport(ScanMetricsRecord r) { + MetricsReportActor actor = new MetricsReportActor(r.principalName()); + MetricsReportRequest request = + new MetricsReportRequest(r.requestId(), r.otelTraceId(), r.otelSpanId()); + ScanMetricsReport.TableObject object = + new ScanMetricsReport.TableObject(r.snapshotId().orElse(null)); + ScanMetricsReport.Payload.Data data = + new ScanMetricsReport.Payload.Data( + r.schemaId().orElse(null), + r.filterExpression().orElse(null), + r.projectedFieldIds().isEmpty() Review Comment: Should `projectedFieldIds` and `projectedFieldNames` be arrays in the JSON response instead of comma-separated strings? The SPI record already carries these as `List<Integer>` / `List<String>`, so the comma-joining here seems to leak the current JDBC storage encoding into the REST shape. JSON arrays would be more natural for clients and avoid delimiter ambiguity, especially for field names. ########## polaris-core/src/main/java/org/apache/polaris/core/persistence/metrics/PolarisMetricsManager.java: ########## @@ -75,4 +80,34 @@ default void writeCommitMetrics( @NonNull PolarisCallContext callCtx, @NonNull CommitMetricsRecord record) { callCtx.getMetaStore().writeCommitReport(record); } + + default Page<ScanMetricsRecord> listScanMetrics( Review Comment: This mirrors the existing write-path pattern, so I don't think it should block this PR by itself. But since #4397 is specifically moving `MetricsPersistence` out from the old aggregate persistence surface, can we make sure this PR is either rebased on that before merge or has an explicit follow-up? Otherwise these new read methods extend the exact transitional layering that the metrics SPI split is trying to unwind. ########## spec/metrics-reports-service.yml: ########## @@ -0,0 +1,479 @@ +# +# 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. +# + +openapi: 3.0.3 +info: + title: Apache Polaris Metrics Reports API + description: > + **Beta**: Read-only API for querying persisted Iceberg table metrics (scan and commit reports) + from Apache Polaris. This API is in beta and may change in future releases. + Requires TABLE_READ_METRICS privilege on the target table. + version: 0.1.0 + license: + name: Apache 2.0 + url: https://www.apache.org/licenses/LICENSE-2.0.html + +servers: + - url: "{scheme}://{host}/api/metrics-reports/v1" + variables: + scheme: + default: https + host: + default: localhost + +paths: + /catalogs/{catalogName}/namespaces/{namespace}/tables/{table}: + parameters: + - $ref: '#/components/parameters/catalogName' + - $ref: '#/components/parameters/namespace' + - $ref: '#/components/parameters/table' + get: + operationId: listTableMetrics + summary: List metrics reports for a table + description: > + Returns persisted metrics reports for the specified table. The required `metricType` + parameter selects between scan reports (produced during table reads) and commit reports + (produced during table writes). Results are ordered by timestamp descending. + Requires TABLE_READ_METRICS privilege on the target table. + tags: + - Metrics + parameters: + - name: metricType + in: query + required: true + description: Type of metrics to retrieve + schema: + type: string + enum: [scan, commit] + - name: pageToken + in: query + required: false + schema: + type: string + description: Opaque cursor from a previous response's nextPageToken field + - name: pageSize + in: query + required: false + schema: + type: integer + minimum: 1 + default: 100 + description: Maximum number of results to return per page + - name: snapshotId + in: query + required: false + schema: + type: integer + format: int64 + description: Filter results to a specific snapshot ID + - name: principalName + in: query + required: false + schema: + type: string + description: Filter results to a specific principal (e.g. service account name) + - name: timestampFrom + in: query + required: false + schema: + type: integer + format: int64 + description: Inclusive lower bound on report timestamp (Unix epoch milliseconds) + - name: timestampTo + in: query + required: false + schema: + type: integer + format: int64 + description: Exclusive upper bound on report timestamp (Unix epoch milliseconds) + responses: + '200': + description: Paginated list of metrics reports + content: + application/json: + schema: + $ref: '#/components/schemas/ListMetricsResponse' + '400': + description: Bad request (missing or invalid parameters) + content: + application/json: + schema: + $ref: '#/components/schemas/ErrorResponse' + '403': + description: Insufficient privileges (TABLE_READ_METRICS required) + content: + application/json: + schema: + $ref: '#/components/schemas/ErrorResponse' + '404': + description: Catalog, namespace, or table not found + content: + application/json: + schema: + $ref: '#/components/schemas/ErrorResponse' + +components: + parameters: + catalogName: + name: catalogName + in: path + required: true + schema: + type: string + namespace: + name: namespace + in: path + required: true + description: > + Namespace path encoded using the same convention as the Polaris Iceberg REST API: + multi-level namespaces use the unit separator (0x1F) between levels, URL-encoded as %1F. + schema: + type: string + table: + name: table + in: path + required: true + schema: + type: string + + schemas: + ListMetricsResponse: + description: > + Polymorphic response for metrics queries. The concrete type is determined by the + metricType discriminator field, which echoes the requested metricType query parameter. + oneOf: + - $ref: '#/components/schemas/ListScanMetricsResponse' + - $ref: '#/components/schemas/ListCommitMetricsResponse' + discriminator: + propertyName: metricType + mapping: + scan: '#/components/schemas/ListScanMetricsResponse' + commit: '#/components/schemas/ListCommitMetricsResponse' + + ListScanMetricsResponse: + type: object + required: + - metricType + - reports + properties: + nextPageToken: + type: string + nullable: true + description: > + Opaque cursor for fetching the next page. Null or absent when no further pages exist. + metricType: + type: string + enum: [scan] + description: Discriminator — always "scan" for this response type + reports: + type: array + items: + $ref: '#/components/schemas/ScanMetricsReport' + + ListCommitMetricsResponse: + type: object + required: + - metricType + - reports + properties: + nextPageToken: + type: string + nullable: true + description: > + Opaque cursor for fetching the next page. Null or absent when no further pages exist. + metricType: + type: string + enum: [commit] + description: Discriminator — always "commit" for this response type + reports: + type: array + items: + $ref: '#/components/schemas/CommitMetricsReport' + + MetricsActor: + type: object + description: Identity of the principal who triggered the operation + properties: + principalName: + type: string + nullable: true + + MetricsRequest: + type: object + description: Request context for correlation with logs and traces + properties: + requestId: + type: string + nullable: true + otelTraceId: + type: string + nullable: true + description: OpenTelemetry trace ID + otelSpanId: + type: string + nullable: true + description: OpenTelemetry span ID + + ScanMetricsObject: + type: object + description: Resource context for the scanned table operation + properties: + snapshotId: + type: integer + format: int64 + nullable: true + + CommitMetricsObject: + type: object + description: Resource context for the committed table operation + required: + - snapshotId + properties: + snapshotId: + type: integer + format: int64 + + ScanPayloadData: + type: object + description: Iceberg scan metrics data + properties: + schemaId: + type: integer + nullable: true + filterExpression: + type: string + nullable: true + projectedFieldIds: Review Comment: Same API-shape question as the service conversion: should these be arrays instead of comma-separated strings? If we keep strings, clients need to parse a secondary encoding inside JSON, and names containing commas become awkward. Since this is a new beta API and the internal record is already list-shaped, exposing arrays now seems cleaner. ########## persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/models/ModelCommitMetricsReport.java: ########## @@ -309,12 +308,54 @@ private static String toJsonString(Map<String, String> map) { return "{}"; } try { - return OBJECT_MAPPER.writeValueAsString(map); + return MetricsModelUtils.OBJECT_MAPPER.writeValueAsString(map); } catch (JsonProcessingException e) { return "{}"; } } + /** + * Converts this JDBC model back to the backend-agnostic SPI record. + * + * @return a CommitMetricsRecord built from this model's fields + */ + default CommitMetricsRecord toRecord() { + return CommitMetricsRecord.builder() + .reportId(getReportId()) + .catalogId(getCatalogId()) + .tableId(getTableId()) + .timestamp(Instant.ofEpochMilli(getTimestampMs())) + .metadata(MetricsModelUtils.parseMetadata(getMetadata())) + .principalName(getPrincipalName()) + .requestId(getRequestId()) + .otelTraceId(getOtelTraceId()) + .otelSpanId(getOtelSpanId()) + .snapshotId(getSnapshotId()) + .sequenceNumber(Optional.ofNullable(getSequenceNumber())) + .operation(getOperation()) + .addedDataFiles(getAddedDataFiles()) + .removedDataFiles(getRemovedDataFiles()) + .totalDataFiles(getTotalDataFiles()) + .addedDeleteFiles(getAddedDeleteFiles()) + .removedDeleteFiles(getRemovedDeleteFiles()) + .totalDeleteFiles(getTotalDeleteFiles()) + .addedEqualityDeleteFiles(getAddedEqualityDeleteFiles()) + .removedEqualityDeleteFiles(getRemovedEqualityDeleteFiles()) + .addedPositionalDeleteFiles(getAddedPositionalDeleteFiles()) + .removedPositionalDeleteFiles(getRemovedPositionalDeleteFiles()) + .addedRecords(getAddedRecords()) + .removedRecords(getRemovedRecords()) + .totalRecords(getTotalRecords()) + .addedFileSizeBytes(getAddedFileSizeBytes()) + .removedFileSizeBytes(getRemovedFileSizeBytes()) + .totalFileSizeBytes(getTotalFileSizeBytes()) + // The write path stores Optional.empty() as 0L; treat 0 as "unknown" on read. + .totalDurationMs( + getTotalDurationMs() == 0L ? Optional.empty() : Optional.of(getTotalDurationMs())) Review Comment: This loses one bit of fidelity: `Optional.empty()` is stored as `0L`, then read back as empty, so a real zero-duration commit would also come back as empty. Not necessarily asking this PR to redesign the schema, but can we avoid baking this in silently? Either make the DB column nullable now, or add a TODO/comment tying this to the planned schema consolidation where `null` vs `0` can be represented without ambiguity. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
