Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down
41 changes: 0 additions & 41 deletions core/trino-main/src/main/java/io/trino/execution/QueryInfo.java
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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"));
}

Expand Down
16 changes: 0 additions & 16 deletions core/trino-main/src/main/java/io/trino/execution/StageInfo.java
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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(
Expand Down
69 changes: 1 addition & 68 deletions core/trino-main/src/main/java/io/trino/execution/StageStats.java
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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()
{
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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()
{
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand Down Expand Up @@ -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()
{
Expand Down
46 changes: 0 additions & 46 deletions core/trino-main/src/main/java/io/trino/operator/OperatorStats.java
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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()) {
Expand Down
Loading