obelix74 commented on code in PR #4115: URL: https://github.com/apache/polaris/pull/4115#discussion_r3251981177
########## extensions/metrics-reports/impl/src/main/java/org/apache/polaris/extension/metrics/reports/MetricsReportsService.java: ########## @@ -0,0 +1,311 @@ +/* + * 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.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.function.Function; +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(buildResponse("scan", page, MetricsReportsService::scanRecordToMap)) + .build(); + } + case "commit" -> { + Page<CommitMetricsRecord> page = + metaStoreManager.listCommitMetrics( + callContext.getPolarisCallContext(), + catalogId, + tableId, + snapshotId, + principalName, + timestampFrom, + timestampTo, + token); + yield Response.ok(buildResponse("commit", page, MetricsReportsService::commitRecordToMap)) + .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 <T> Map<String, Object> buildResponse( + String metricType, Page<T> page, Function<T, Map<String, Object>> toMap) { + List<Map<String, Object>> reports = page.items().stream().map(toMap).toList(); + Map<String, Object> response = new LinkedHashMap<>(); + response.put("metricType", metricType); + response.put("nextPageToken", page.encodedResponseToken()); + response.put("reports", reports); + return response; + } + + private static Map<String, Object> scanRecordToMap(ScanMetricsRecord r) { + Map<String, Object> actor = new LinkedHashMap<>(); + actor.put("principalName", r.principalName()); + + Map<String, Object> request = new LinkedHashMap<>(); + request.put("requestId", r.requestId()); + request.put("otelTraceId", r.otelTraceId()); + request.put("otelSpanId", r.otelSpanId()); + + Map<String, Object> object = new LinkedHashMap<>(); + r.snapshotId().ifPresent(v -> object.put("snapshotId", v)); + + Map<String, Object> data = new LinkedHashMap<>(); + r.schemaId().ifPresent(v -> data.put("schemaId", v)); + r.filterExpression().ifPresent(v -> data.put("filterExpression", v)); + if (!r.projectedFieldIds().isEmpty()) { + data.put( + "projectedFieldIds", + r.projectedFieldIds().stream().map(Object::toString).collect(Collectors.joining(","))); + } + if (!r.projectedFieldNames().isEmpty()) { + data.put("projectedFieldNames", String.join(",", r.projectedFieldNames())); + } + data.put("resultDataFiles", r.resultDataFiles()); + data.put("resultDeleteFiles", r.resultDeleteFiles()); + data.put("totalFileSizeBytes", r.totalFileSizeBytes()); + data.put("totalDataManifests", r.totalDataManifests()); + data.put("totalDeleteManifests", r.totalDeleteManifests()); + data.put("scannedDataManifests", r.scannedDataManifests()); + data.put("scannedDeleteManifests", r.scannedDeleteManifests()); + data.put("skippedDataManifests", r.skippedDataManifests()); + data.put("skippedDeleteManifests", r.skippedDeleteManifests()); + data.put("skippedDataFiles", r.skippedDataFiles()); + data.put("skippedDeleteFiles", r.skippedDeleteFiles()); + data.put("totalPlanningDurationMs", r.totalPlanningDurationMs()); + data.put("equalityDeleteFiles", r.equalityDeleteFiles()); + data.put("positionalDeleteFiles", r.positionalDeleteFiles()); + data.put("indexedDeleteFiles", r.indexedDeleteFiles()); + data.put("totalDeleteFileSizeBytes", r.totalDeleteFileSizeBytes()); + + Map<String, Object> payload = new LinkedHashMap<>(); + payload.put("type", "iceberg.metrics.scan"); + payload.put("version", 1); + payload.put("data", data); + + Map<String, Object> m = new LinkedHashMap<>(); + m.put("id", r.reportId()); + m.put("timestampMs", r.timestamp().toEpochMilli()); + m.put("actor", actor); + m.put("request", request); + m.put("object", object); + m.put("payload", payload); + return m; + } + + private static Map<String, Object> commitRecordToMap(CommitMetricsRecord r) { + Map<String, Object> actor = new LinkedHashMap<>(); + actor.put("principalName", r.principalName()); + + Map<String, Object> request = new LinkedHashMap<>(); + request.put("requestId", r.requestId()); + request.put("otelTraceId", r.otelTraceId()); + request.put("otelSpanId", r.otelSpanId()); + + Map<String, Object> object = new LinkedHashMap<>(); + object.put("snapshotId", r.snapshotId()); + + Map<String, Object> data = new LinkedHashMap<>(); + r.sequenceNumber().ifPresent(v -> data.put("sequenceNumber", v)); + data.put("operation", r.operation()); Review Comment: Done — replaced all `Map<String, Object>` response construction with typed Java records (`ScanMetricsReport`, `CommitMetricsReport`, `MetricsListResponse`, `MetricsReportActor`, `MetricsReportRequest`). Each record uses `@JsonProperty` to pin the JSON field names, so renaming an accessor won't silently change the wire format. The nested `Payload.Data` records also carry `@JsonInclude(NON_NULL)` so optional scan fields (`schemaId`, `filterExpression`, etc.) are omitted when absent. ########## 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( + @Nonnull PolarisCallContext callCtx, + long catalogId, + long tableId, + @Nullable Long snapshotId, + @Nullable String principalName, + @Nullable Long timestampFrom, + @Nullable Long timestampTo, + @Nonnull PageToken pageToken) { + return callCtx + .getMetaStore() + .listScanReports( Review Comment: Good catch. I've reviewed #4397 — it decomposes `BasePersistence` into disjoint SPIs where `MetricsPersistence` becomes a top-level interface, which is exactly what we want here too. Once #4397 lands, `PolarisMetricsManager` will delegate to `MetricsPersistence` directly (injected as its own SPI) instead of going through `callCtx.getMetaStore()`. I'll rebase on top of that change rather than trying to reconcile both in this PR. -- 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]
