Add debug logging to PhasedExecutionSchedule

core/trino-main/src/main/java/io/trino/execution/scheduler/PipelinedStageExecution.java

@@ -58,6 +58,7 @@
 import java.util.function.Consumer;
 import java.util.stream.Stream;
 
+import static com.google.common.base.MoreObjects.toStringHelper;
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkState;
 import static com.google.common.collect.ImmutableList.toImmutableList;
@@ -559,6 +560,12 @@ public Optional<ExecutionFailureInfo> getFailureCause()
         return stateMachine.getFailureCause();
     }
 
+    @Override
+    public String toString()
+    {
+        return stateMachine.toString();
+    }
+
     private static Split createExchangeSplit(RemoteTask sourceTask, RemoteTask destinationTask)
     {
         // Fetch the results from the buffer assigned to the task based on id
@@ -657,5 +664,14 @@ public void addStateChangeListener(StateChangeListener<State> stateChangeListene
         {
             state.addStateChangeListener(stateChangeListener);
         }
+
+        @Override
+        public String toString()
+        {
+            return toStringHelper(this)
+                    .add("stageId", stageId)
+                    .add("state", state)
+                    .toString();
+        }
     }
 }

core/trino-main/src/main/java/io/trino/execution/scheduler/policy/PhasedExecutionSchedule.java

@@ -18,6 +18,7 @@
 import com.google.common.collect.Ordering;
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.SettableFuture;
+import io.airlift.log.Logger;
 import io.trino.execution.scheduler.StageExecution;
 import io.trino.execution.scheduler.StageExecution.State;
 import io.trino.server.DynamicFilterService;
@@ -75,6 +76,8 @@
 public class PhasedExecutionSchedule
         implements ExecutionSchedule
 {
+    private static final Logger log = Logger.get(PhasedExecutionSchedule.class);
+
     /**
      * Graph representing a before -> after relationship between fragments.
      * Destination fragment should be started only when source stage is completed.
@@ -127,6 +130,12 @@ private void init(Collection<StageExecution> stages)
                 .forEach(fragmentsToExecute::add);
         fragmentOrdering = Ordering.explicit(sortedFragments);
         selectForExecution(fragmentsToExecute.build());
+        log.debug(
+                "fragmentDependency: %s, fragmentTopology: %s, sortedFragments: %s, stagesByFragmentId: %s",
+                fragmentDependency,
+                fragmentTopology,
+                sortedFragments,
+                stagesByFragmentId);
     }
 
     @Override
@@ -186,6 +195,7 @@ private Set<PlanFragmentId> removeCompletedStages()
                 .filter(this::isStageCompleted)
                 .collect(toImmutableSet());
         // remove completed stages outside of Java stream to prevent concurrent modification
+        log.debug("completedStages: %s", completedStages);
         return completedStages.stream()
                 .flatMap(stage -> removeCompletedStage(stage).stream())
                 .collect(toImmutableSet());
@@ -217,6 +227,7 @@ private Set<PlanFragmentId> unblockStagesWithFullOutputBuffer()
                 .filter(StageExecution::isAnyTaskBlocked)
                 .map(stage -> stage.getFragment().getId())
                 .collect(toImmutableSet());
+        log.debug("blockedFragments: %s", blockedFragments);
         // start immediate downstream stages so that data can be consumed
         return blockedFragments.stream()
                 .flatMap(fragmentId -> fragmentTopology.outgoingEdgesOf(fragmentId).stream())
@@ -227,10 +238,12 @@ private Set<PlanFragmentId> unblockStagesWithFullOutputBuffer()
     private void selectForExecution(Set<PlanFragmentId> fragmentIds)
     {
         requireNonNull(fragmentOrdering, "fragmentOrdering is null");
-        fragmentIds.stream()
+        List<StageExecution> selectedForExecution = fragmentIds.stream()
                 .sorted(fragmentOrdering)
                 .map(stagesByFragmentId::get)
-                .forEach(this::selectForExecution);
+                .collect(toImmutableList());
+        log.debug("selectedForExecution: %s", selectedForExecution);
+        selectedForExecution.forEach(this::selectForExecution);
     }
 
     private void selectForExecution(StageExecution stage)
@@ -245,20 +258,21 @@ private void selectForExecution(StageExecution stage)
             // if there are any dependent stages then reschedule when stage is completed
             stage.addStateChangeListener(state -> {
                 if (isStageCompleted(stage)) {
-                    notifyReschedule();
+                    notifyReschedule(stage);
                 }
             });
         }
     }
 
-    private void notifyReschedule()
+    private void notifyReschedule(StageExecution stage)
     {
         SettableFuture<Void> rescheduleFuture;
         synchronized (this) {
             rescheduleFuture = this.rescheduleFuture;
             this.rescheduleFuture = SettableFuture.create();
         }
         // notify listeners outside of the critical section
+        log.debug("notifyReschedule by %s", stage);
         rescheduleFuture.set(null);
     }