dimas-b commented on code in PR #4115:
URL: https://github.com/apache/polaris/pull/4115#discussion_r3251281700
##########
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:
Access via `BasePersistence` will conflict with recent PR #4397.
@obelix74 : Please review that PR and share your opinion.
##########
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:
nit: an interface with all default methods looks a bit odd. Not a blocker
for this PR, but it feels like our general approach to wiring REST services to
backend code may needs a review (if not a revamp 😅 )
##########
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:
nit: Maps work here, but WDYT about using light-weight Immutable classes
with JSON serializers instead? It would offer better control over attribute
names and JSON structure if we have to reuse the in multiple places.
--
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]
