Make query JSON more compact for UI

- Serialize data sizes using succinct form (only query json download)
- Do not output raw digest in query.json
- Do not serialize type information for metrics
- Do not serialize type information for operator infos
- Cleanup old prunning logic to reduce copying

Author: wendigo

core/trino-main/src/main/java/io/trino/execution/DistributionSnapshot.java

@@ -18,34 +18,17 @@
 import com.google.common.base.MoreObjects.ToStringHelper;
 import io.trino.spi.metrics.Distribution;
 import io.trino.spi.metrics.Metric;
-import io.trino.spi.metrics.Metrics;
 
 import java.util.Locale;
-import java.util.Map;
 
 import static com.google.common.base.MoreObjects.toStringHelper;
-import static com.google.common.collect.ImmutableMap.toImmutableMap;
 import static java.lang.String.format;
 
 @JsonTypeInfo(use = JsonTypeInfo.Id.NONE) // Do not add @class property
 @JsonSerialize
 public record DistributionSnapshot(long total, double min, double max, double p01, double p05, double p10, double p25, double p50, double p75, double p90, double p95, double p99)
         implements Metric<DistributionSnapshot>
 {
-    public static Metrics pruneMetrics(Metrics metrics)
-    {
-        return new Metrics(metrics.getMetrics().entrySet().stream()
-                .collect(toImmutableMap(
-                        Map.Entry::getKey,
-                        entry -> {
-                            Metric<?> metric = entry.getValue();
-                            if (metric instanceof Distribution) {
-                                return new DistributionSnapshot((Distribution<?>) metric);
-                            }
-                            return metric;
-                        })));
-    }
-
     public DistributionSnapshot(Distribution<?> distribution)
     {
         this(

core/trino-main/src/main/java/io/trino/execution/QueryInfo.java

@@ -448,45 +448,4 @@ public String toString()
                 .add("fieldNames", fieldNames)
                 .toString();
     }
-
-    public QueryInfo pruneDigests()
-    {
-        return new QueryInfo(
-                queryId,
-                session,
-                state,
-                self,
-                fieldNames,
-                query,
-                preparedQuery,
-                queryStats,
-                setCatalog,
-                setSchema,
-                setPath,
-                setAuthorizationUser,
-                resetAuthorizationUser,
-                setOriginalRoles,
-                setSessionProperties,
-                resetSessionProperties,
-                setRoles,
-                addedPreparedStatements,
-                deallocatedPreparedStatements,
-                startedTransactionId,
-                clearTransactionId,
-                updateType,
-                stages.map(StagesInfo::pruneDigests),
-                failureInfo,
-                errorCode,
-                warnings,
-                inputs,
-                output,
-                referencedTables,
-                routines,
-                finalQueryInfo,
-                resourceGroupId,
-                queryType,
-                retryPolicy,
-                pruned,
-                version);
-    }
 }

core/trino-main/src/main/java/io/trino/execution/QueryStats.java

@@ -32,7 +32,6 @@
 import java.util.Set;
 
 import static com.google.common.base.Preconditions.checkArgument;
-import static com.google.common.collect.ImmutableList.toImmutableList;
 import static io.airlift.units.DataSize.succinctBytes;
 import static io.trino.server.DynamicFilterService.DynamicFiltersStats;
 import static java.util.Objects.requireNonNull;
@@ -323,10 +322,7 @@ public QueryStats(
         this.stageGcStatistics = ImmutableList.copyOf(requireNonNull(stageGcStatistics, "stageGcStatistics is null"));
 
         this.dynamicFiltersStats = requireNonNull(dynamicFiltersStats, "dynamicFiltersStats is null");
-
-        requireNonNull(operatorSummaries, "operatorSummaries is null");
-        this.operatorSummaries = operatorSummaries.stream().map(OperatorStats::pruneDigests).collect(toImmutableList());
-
+        this.operatorSummaries = ImmutableList.copyOf(operatorSummaries);
         this.optimizerRulesSummaries = ImmutableList.copyOf(requireNonNull(optimizerRulesSummaries, "optimizerRulesSummaries is null"));
     }
 

core/trino-main/src/main/java/io/trino/execution/StageInfo.java

@@ -28,7 +28,6 @@
 import java.util.Map;
 
 import static com.google.common.base.MoreObjects.toStringHelper;
-import static com.google.common.collect.ImmutableList.toImmutableList;
 import static java.util.Objects.requireNonNull;
 
 @Immutable
@@ -167,21 +166,6 @@ public StageInfo withSubStages(List<StageId> subStages)
                 failureCause);
     }
 
-    public StageInfo pruneDigests()
-    {
-        return new StageInfo(
-                stageId,
-                state,
-                plan,
-                coordinatorOnly,
-                types,
-                stageStats.pruneDigests(),
-                tasks.stream().map(TaskInfo::pruneDigests).collect(toImmutableList()),
-                subStages,
-                tables,
-                failureCause);
-    }
-
     public static StageInfo createInitial(QueryId queryId, StageState state, PlanFragment fragment)
     {
         return new StageInfo(

core/trino-main/src/main/java/io/trino/execution/StageStats.java

@@ -36,9 +36,7 @@
 import java.util.Set;
 
 import static com.google.common.base.Preconditions.checkArgument;
-import static com.google.common.collect.ImmutableList.toImmutableList;
 import static io.airlift.units.DataSize.Unit.BYTE;
-import static io.trino.execution.DistributionSnapshot.pruneMetrics;
 import static io.trino.execution.StageState.RUNNING;
 import static java.lang.Math.min;
 import static java.util.Objects.requireNonNull;
@@ -277,9 +275,7 @@ public StageStats(
         this.failedPhysicalWrittenDataSize = requireNonNull(failedPhysicalWrittenDataSize, "failedPhysicalWrittenDataSize is null");
 
         this.gcInfo = requireNonNull(gcInfo, "gcInfo is null");
-
-        requireNonNull(operatorSummaries, "operatorSummaries is null");
-        this.operatorSummaries = operatorSummaries.stream().map(OperatorStats::pruneDigests).collect(toImmutableList());
+        this.operatorSummaries = ImmutableList.copyOf(operatorSummaries);
     }
 
     @JsonProperty
@@ -661,69 +657,6 @@ public BasicStageStats toBasicStageStats(StageState stageState)
                 runningPercentage);
     }
 
-    public StageStats pruneDigests()
-    {
-        return new StageStats(
-                schedulingComplete,
-                getSplitDistribution,
-                splitSourceMetrics,
-                totalTasks,
-                runningTasks,
-                completedTasks,
-                failedTasks,
-                totalDrivers,
-                queuedDrivers,
-                runningDrivers,
-                blockedDrivers,
-                completedDrivers,
-                cumulativeUserMemory,
-                failedCumulativeUserMemory,
-                userMemoryReservation,
-                revocableMemoryReservation,
-                totalMemoryReservation,
-                peakUserMemoryReservation,
-                peakRevocableMemoryReservation,
-                spilledDataSize,
-                totalScheduledTime,
-                failedScheduledTime,
-                totalCpuTime,
-                failedCpuTime,
-                totalBlockedTime,
-                fullyBlocked,
-                blockedReasons,
-                physicalInputDataSize,
-                failedPhysicalInputDataSize,
-                physicalInputPositions,
-                failedPhysicalInputPositions,
-                physicalInputReadTime,
-                failedPhysicalInputReadTime,
-                internalNetworkInputDataSize,
-                failedInternalNetworkInputDataSize,
-                internalNetworkInputPositions,
-                failedInternalNetworkInputPositions,
-                processedInputDataSize,
-                failedProcessedInputDataSize,
-                processedInputPositions,
-                failedProcessedInputPositions,
-                inputBlockedTime,
-                failedInputBlockedTime,
-                bufferedDataSize,
-                outputBufferUtilization,
-                outputDataSize,
-                failedOutputDataSize,
-                outputPositions,
-                failedOutputPositions,
-                pruneMetrics(outputBufferMetrics),
-                outputBlockedTime,
-                failedOutputBlockedTime,
-                physicalWrittenDataSize,
-                failedPhysicalWrittenDataSize,
-                gcInfo,
-                operatorSummaries.stream()
-                        .map(OperatorStats::pruneDigests)
-                        .collect(toImmutableList()));
-    }
-
     public static StageStats createInitial()
     {
         DataSize zeroBytes = DataSize.of(0, BYTE);

core/trino-main/src/main/java/io/trino/execution/StagesInfo.java

@@ -66,13 +66,6 @@ public List<StageInfo> getStages()
         return stages;
     }
 
-    public StagesInfo pruneDigests()
-    {
-        return new StagesInfo(
-                outputStageId,
-                stages.stream().map(StageInfo::pruneDigests).collect(toImmutableList()));
-    }
-
     @JsonIgnore
     public StageInfo getOutputStage()
     {

core/trino-main/src/main/java/io/trino/execution/TaskInfo.java

@@ -64,11 +64,6 @@ public TaskInfo pruneSpoolingOutputStats()
         return new TaskInfo(taskStatus, lastHeartbeat, outputBuffers.pruneSpoolingOutputStats(), noMoreSplits, stats, estimatedMemory, needsPlan);
     }
 
-    public TaskInfo pruneDigests()
-    {
-        return new TaskInfo(taskStatus, lastHeartbeat, outputBuffers.pruneDigests(), noMoreSplits, stats.pruneDigests(), estimatedMemory, needsPlan);
-    }
-
     @Override
     public String toString()
     {

core/trino-main/src/main/java/io/trino/execution/buffer/OutputBufferInfo.java

@@ -16,7 +16,6 @@
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.google.common.collect.ImmutableList;
-import io.trino.execution.DistributionSnapshot;
 import io.trino.plugin.base.metrics.TDigestHistogram;
 import io.trino.spi.metrics.Metrics;
 
@@ -193,23 +192,6 @@ public OutputBufferInfo pruneSpoolingOutputStats()
                 metrics);
     }
 
-    public OutputBufferInfo pruneDigests()
-    {
-        return new OutputBufferInfo(
-                type,
-                state,
-                canAddBuffers,
-                canAddPages,
-                totalBufferedBytes,
-                totalBufferedPages,
-                totalRowsSent,
-                totalPagesSent,
-                pipelinedBufferStates,
-                Optional.empty(),
-                spoolingOutputStats,
-                metrics.map(DistributionSnapshot::pruneMetrics));
-    }
-
     @Override
     public String toString()
     {

core/trino-main/src/main/java/io/trino/operator/OperatorStats.java

@@ -29,7 +29,6 @@
 
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Verify.verify;
-import static io.trino.execution.DistributionSnapshot.pruneMetrics;
 import static java.lang.Math.max;
 import static java.util.Objects.requireNonNull;
 import static java.util.concurrent.TimeUnit.NANOSECONDS;
@@ -636,51 +635,6 @@ private static <T> Mergeable<T> mergeInfos(Mergeable<T> base, List<T> others)
         return (Mergeable<T>) base.mergeWith(others);
     }
 
-    public OperatorStats pruneDigests()
-    {
-        return new OperatorStats(
-                stageId,
-                pipelineId,
-                operatorId,
-                planNodeId,
-                sourceId,
-                operatorType,
-                totalDrivers,
-                addInputCalls,
-                addInputWall,
-                addInputCpu,
-                physicalInputDataSize,
-                physicalInputPositions,
-                physicalInputReadTime,
-                internalNetworkInputDataSize,
-                internalNetworkInputPositions,
-                inputDataSize,
-                inputPositions,
-                sumSquaredInputPositions,
-                getOutputCalls,
-                getOutputWall,
-                getOutputCpu,
-                outputDataSize,
-                outputPositions,
-                dynamicFilterSplitsProcessed,
-                pruneMetrics(metrics),
-                pruneMetrics(connectorMetrics),
-                pruneMetrics(pipelineMetrics),
-                physicalWrittenDataSize,
-                blockedWall,
-                finishCalls,
-                finishWall,
-                finishCpu,
-                userMemoryReservation,
-                revocableMemoryReservation,
-                peakUserMemoryReservation,
-                peakRevocableMemoryReservation,
-                peakTotalMemoryReservation,
-                spilledDataSize,
-                blockedReason,
-                info);
-    }
-
     public OperatorStats summarize()
     {
         if (info == null || info.isFinal()) {