Add memory requirements debug logs to stage scheduler

trinodb · Apr 9, 2022 · 6fba6a4 · 6fba6a4
1 parent 588209a
commit 6fba6a4
Show file tree

Hide file tree

Showing 2 changed files with 27 additions and 0 deletions.
diff --git a/...src/main/java/io/trino/execution/scheduler/ExponentialGrowthPartitionMemoryEstimator.java b/...src/main/java/io/trino/execution/scheduler/ExponentialGrowthPartitionMemoryEstimator.java
@@ -13,13 +13,17 @@
  */
 package io.trino.execution.scheduler;
 
+import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Ordering;
+import com.google.common.collect.Streams;
 import io.airlift.stats.TDigest;
 import io.airlift.units.DataSize;
 import io.trino.Session;
 import io.trino.spi.ErrorCode;
 
+import java.util.List;
 import java.util.Optional;
+import java.util.stream.Collectors;
 
 import static io.trino.SystemSessionProperties.getFaultTolerantExecutionTaskMemoryEstimationQuantile;
 import static io.trino.SystemSessionProperties.getFaultTolerantExecutionTaskMemoryGrowthFactor;
@@ -88,4 +92,25 @@ private synchronized DataSize getEstimatedMemoryUsage(Session session)
         }
         return DataSize.ofBytes((long) estimation);
     }
+
+    private String memoryUsageDistributionInfo()
+    {
+        List<Double> quantiles = ImmutableList.of(0.01, 0.05, 0.1, 0.2, 0.5, 0.8, 0.9, 0.95, 0.99);
+        List<Double> values;
+        synchronized (this) {
+            values = memoryUsageDistribution.valuesAt(quantiles);
+        }
+
+        return Streams.zip(
+                        quantiles.stream(),
+                        values.stream(),
+                        (quantile, value) -> "" + quantile + "=" + value)
+                .collect(Collectors.joining(", ", "[", "]"));
+    }
+
+    @Override
+    public String toString()
+    {
+        return "memoryUsageDistribution=" + memoryUsageDistributionInfo();
+    }
 }
diff --git a/core/trino-main/src/main/java/io/trino/execution/scheduler/FaultTolerantStageScheduler.java b/core/trino-main/src/main/java/io/trino/execution/scheduler/FaultTolerantStageScheduler.java
@@ -266,6 +266,7 @@ public synchronized void schedule()
             TaskDescriptor taskDescriptor = taskDescriptorOptional.get();
 
             MemoryRequirements memoryRequirements = partitionMemoryRequirements.computeIfAbsent(partition, ignored -> partitionMemoryEstimator.getInitialMemoryRequirements(session, taskDescriptor.getNodeRequirements().getMemory()));
+            log.debug("Computed initial memory requirements for task from stage %s; requirements=%s; estimator=%s", stage.getStageId(), memoryRequirements, partitionMemoryEstimator);
             if (nodeLease == null) {
                 NodeRequirements nodeRequirements = taskDescriptor.getNodeRequirements();
                 nodeRequirements = nodeRequirements.withMemory(memoryRequirements.getRequiredMemory());
@@ -553,6 +554,7 @@ private void updateTaskStatus(TaskStatus taskStatus, Optional<ExchangeSinkInstan
 
                                 // update memory limits for next attempt
                                 MemoryRequirements newMemoryLimits = partitionMemoryEstimator.getNextRetryMemoryRequirements(session, memoryLimits, taskStatus.getPeakMemoryReservation(), errorCode);
+                                log.debug("Computed next memory requirements for task from stage %s; previous=%s; new=%s; peak=%s; estimator=%s", stage.getStageId(), memoryLimits, newMemoryLimits, taskStatus.getPeakMemoryReservation(), partitionMemoryEstimator);
                                 partitionMemoryRequirements.put(partitionId, newMemoryLimits);
 
                                 // reschedule