Make AdaptivePlanner optional in the FTE scheduler

... instead of having a separate boolean field indicating whether the planner is to be used.

Reduces potential for inadvertent erroneous use; simplifies method signatures.

Author: dekimir

core/trino-main/src/main/java/io/trino/execution/scheduler/faulttolerant/EventDrivenFaultTolerantQueryScheduler.java

@@ -219,8 +219,7 @@ public class EventDrivenFaultTolerantQueryScheduler
     private final FailureDetector failureDetector;
     private final DynamicFilterService dynamicFilterService;
     private final TaskExecutionStats taskExecutionStats;
-    private final AdaptivePlanner adaptivePlanner;
-    private final boolean adaptiveQueryPlanningEnabled;
+    private final Optional<AdaptivePlanner> adaptivePlanner;
     private final SubPlan originalPlan;
     private final boolean stageEstimationForEagerParentEnabled;
 
@@ -275,8 +274,9 @@ public EventDrivenFaultTolerantQueryScheduler(
         this.failureDetector = requireNonNull(failureDetector, "failureDetector is null");
         this.dynamicFilterService = requireNonNull(dynamicFilterService, "dynamicFilterService is null");
         this.taskExecutionStats = requireNonNull(taskExecutionStats, "taskExecutionStats is null");
-        this.adaptivePlanner = requireNonNull(adaptivePlanner, "adaptivePlanner is null");
-        this.adaptiveQueryPlanningEnabled = isFaultTolerantExecutionAdaptiveQueryPlanningEnabled(queryStateMachine.getSession());
+        this.adaptivePlanner = isFaultTolerantExecutionAdaptiveQueryPlanningEnabled(queryStateMachine.getSession()) ?
+                Optional.of(requireNonNull(adaptivePlanner, "adaptivePlanner is null")) :
+                Optional.empty();
         this.originalPlan = requireNonNull(originalPlan, "originalPlan is null");
 
         this.stageEstimationForEagerParentEnabled = isFaultTolerantExecutionStageEstimationForEagerParentEnabled(queryStateMachine.getSession());
@@ -361,8 +361,7 @@ public synchronized void start()
                     getFaultTolerantExecutionRuntimeAdaptivePartitioningPartitionCount(session),
                     getFaultTolerantExecutionRuntimeAdaptivePartitioningMaxTaskSize(session),
                     stageEstimationForEagerParentEnabled,
-                    adaptivePlanner,
-                    adaptiveQueryPlanningEnabled);
+                    adaptivePlanner);
             queryExecutor.submit(scheduler::run);
         }
         catch (Throwable t) {
@@ -717,8 +716,7 @@ private static class Scheduler
         private SubPlan plan;
         private List<SubPlan> planInTopologicalOrder;
 
-        private final AdaptivePlanner adaptivePlanner;
-        private final boolean adaptiveQueryPlanningEnabled;
+        private final Optional<AdaptivePlanner> adaptivePlanner;
 
         private final Map<StageId, StageExecution> stageExecutions = new HashMap<>();
         private final Map<SubPlan, IsReadyForExecutionResult> isReadyForExecutionCache = new HashMap<>();
@@ -764,8 +762,7 @@ public Scheduler(
                 int runtimeAdaptivePartitioningPartitionCount,
                 DataSize runtimeAdaptivePartitioningMaxTaskSize,
                 boolean stageEstimationForEagerParentEnabled,
-                AdaptivePlanner adaptivePlanner,
-                boolean adaptiveQueryPlanningEnabled)
+                Optional<AdaptivePlanner> adaptivePlanner)
         {
             this.queryStateMachine = requireNonNull(queryStateMachine, "queryStateMachine is null");
             this.metadata = requireNonNull(metadata, "metadata is null");
@@ -798,7 +795,6 @@ public Scheduler(
             this.runtimeAdaptivePartitioningPartitionCount = runtimeAdaptivePartitioningPartitionCount;
             this.runtimeAdaptivePartitioningMaxTaskSizeInBytes = requireNonNull(runtimeAdaptivePartitioningMaxTaskSize, "runtimeAdaptivePartitioningMaxTaskSize is null").toBytes();
             this.adaptivePlanner = requireNonNull(adaptivePlanner, "adaptivePlanner is null");
-            this.adaptiveQueryPlanningEnabled = adaptiveQueryPlanningEnabled;
             this.stageEstimationForEagerParentEnabled = stageEstimationForEagerParentEnabled;
             this.schedulerSpan = tracer.spanBuilder("scheduler")
                 .setParent(Context.current().with(queryStateMachine.getSession().getQuerySpan()))
@@ -1094,7 +1090,7 @@ private SubPlan optimizePlan(SubPlan plan)
         {
             // Re-optimize plan here based on available runtime statistics.
             // Fragments changed due to re-optimization as well as their downstream stages are expected to be assigned new fragment ids.
-            if (!adaptiveQueryPlanningEnabled) {
+            if (adaptivePlanner.isEmpty()) {
                 return plan;
             }
 
@@ -1124,7 +1120,7 @@ private SubPlan optimizePlan(SubPlan plan)
                 // already ready for execution.
                 if (isReadyForExecutionResult.isReadyForExecution()
                         && (oldValue == null || !oldValue.isReadyForExecution())) {
-                    return adaptivePlanner.optimize(plan, createRuntimeInfoProvider());
+                    return adaptivePlanner.get().optimize(plan, createRuntimeInfoProvider());
                 }
             }