Add thread-per-driver executor

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

@@ -40,9 +40,9 @@
 import io.trino.execution.buffer.BufferResult;
 import io.trino.execution.buffer.OutputBuffers;
 import io.trino.execution.buffer.PipelinedOutputBuffers;
-import io.trino.execution.executor.PrioritizedSplitRunner;
+import io.trino.execution.executor.RunningSplitInfo;
 import io.trino.execution.executor.TaskExecutor;
-import io.trino.execution.executor.TaskExecutor.RunningSplitInfo;
+import io.trino.execution.executor.timesharing.PrioritizedSplitRunner;
 import io.trino.memory.LocalMemoryManager;
 import io.trino.memory.NodeMemoryConfig;
 import io.trino.memory.QueryContext;
@@ -90,7 +90,7 @@
 import static io.trino.SystemSessionProperties.resourceOvercommit;
 import static io.trino.cache.SafeCaches.buildNonEvictableCache;
 import static io.trino.execution.SqlTask.createSqlTask;
-import static io.trino.execution.executor.PrioritizedSplitRunner.SPLIT_RUN_QUANTA;
+import static io.trino.execution.executor.timesharing.PrioritizedSplitRunner.SPLIT_RUN_QUANTA;
 import static io.trino.operator.RetryPolicy.TASK;
 import static io.trino.spi.StandardErrorCode.ABANDONED_TASK;
 import static io.trino.spi.StandardErrorCode.GENERIC_USER_ERROR;

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

@@ -45,6 +45,7 @@
         "task.level-absolute-priority"})
 public class TaskManagerConfig
 {
+    private boolean threadPerDriverSchedulerEnabled;
     private boolean perOperatorCpuTimerEnabled = true;
     private boolean taskCpuTimerEnabled = true;
     private boolean statisticsCpuTimerEnabled = true;
@@ -107,6 +108,18 @@ public class TaskManagerConfig
 
     private BigDecimal levelTimeMultiplier = new BigDecimal(2.0);
 
+    @Config("experimental.thread-per-driver-scheduler-enabled")
+    public TaskManagerConfig setThreadPerDriverSchedulerEnabled(boolean enabled)
+    {
+        this.threadPerDriverSchedulerEnabled = enabled;
+        return this;
+    }
+
+    public boolean isThreadPerDriverSchedulerEnabled()
+    {
+        return threadPerDriverSchedulerEnabled;
+    }
+
     @MinDuration("1ms")
     @MaxDuration("10s")
     @NotNull

core/trino-main/src/main/java/io/trino/execution/executor/RunningSplitInfo.java

@@ -0,0 +1,98 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.trino.execution.executor;
+
+import com.google.common.collect.ComparisonChain;
+import io.trino.execution.TaskId;
+
+import java.util.function.Supplier;
+
+import static java.util.Objects.requireNonNull;
+
+/**
+ * A class representing a split that is running on the TaskRunner.
+ * It has a Thread object that gets assigned while assigning the split
+ * to the taskRunner. However, when the TaskRunner moves to a different split,
+ * the thread stored here will not remain assigned to this split anymore.
+ */
+public class RunningSplitInfo
+        implements Comparable<RunningSplitInfo>
+{
+    private final long startTime;
+    private final String threadId;
+    private final Thread thread;
+    private boolean printed;
+    private final TaskId taskId;
+    private final Supplier<String> splitInfo;
+
+    public RunningSplitInfo(long startTime, String threadId, Thread thread, TaskId taskId, Supplier<String> splitInfo)
+    {
+        this.startTime = startTime;
+        this.threadId = requireNonNull(threadId, "threadId is null");
+        this.thread = requireNonNull(thread, "thread is null");
+        this.taskId = requireNonNull(taskId, "taskId is null");
+        this.splitInfo = requireNonNull(splitInfo, "split is null");
+        this.printed = false;
+    }
+
+    public long getStartTime()
+    {
+        return startTime;
+    }
+
+    public String getThreadId()
+    {
+        return threadId;
+    }
+
+    public Thread getThread()
+    {
+        return thread;
+    }
+
+    public TaskId getTaskId()
+    {
+        return taskId;
+    }
+
+    /**
+     * {@link PrioritizedSplitRunner#getInfo()} provides runtime statistics for the split (such as total cpu utilization so far).
+     * A value returned from this method changes over time and cannot be cached as a field of {@link RunningSplitInfo}.
+     *
+     * @return Formatted string containing runtime statistics for the split.
+     */
+    public String getSplitInfo()
+    {
+        return splitInfo.get();
+    }
+
+    public boolean isPrinted()
+    {
+        return printed;
+    }
+
+    public void setPrinted()
+    {
+        printed = true;
+    }
+
+    @Override
+    public int compareTo(RunningSplitInfo o)
+    {
+        return ComparisonChain.start()
+                .compare(startTime, o.getStartTime())
+                .compare(threadId, o.getThreadId())
+                .result();
+    }
+}