Set value of writerCount to availableProcessors * 2

Since writers are both CPU and IO bound, we
should increase them to the same value as the
number of worker threads.

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

@@ -79,18 +79,17 @@ public class TaskManagerConfig
     private Duration interruptStuckSplitTasksDetectionInterval = new Duration(2, TimeUnit.MINUTES);
 
     private boolean scaleWritersEnabled = true;
-    // Set the value of default max writer count to the number of processors and cap it to 32. We can do this
-    // because preferred write partitioning is always enabled for local exchange thus partitioned inserts will never
-    // use this property. Hence, there is no risk in terms of more numbers of physical writers which can cause high
-    // resource utilization.
-    private int scaleWritersMaxWriterCount = min(getAvailablePhysicalProcessorCount(), 32);
+    // Set the value of default max writer count to the number of processors * 2 and cap it to 64. We can set this value
+    // higher because preferred write partitioning is always enabled for local exchange thus partitioned inserts will never
+    // use this property. Additionally, we have a mechanism to stop scaling if local memory utilization is high.
+    private int scaleWritersMaxWriterCount = min(getAvailablePhysicalProcessorCount(), 32) * 2;
     private int writerCount = 1;
     // Default value of partitioned task writer count should be above 1, otherwise it can create a plan
     // with a single gather exchange node on the coordinator due to a single available processor. Whereas,
     // on the worker nodes due to more available processors, the default value could be above 1. Therefore,
-    // it can cause error due to config mismatch during execution. Additionally, cap it to 32 in order to
+    // it can cause error due to config mismatch during execution. Additionally, cap it to 64 in order to
     // avoid small pages produced by local partitioning exchanges.
-    private int partitionedWriterCount = min(max(nextPowerOfTwo(getAvailablePhysicalProcessorCount()), 2), 32);
+    private int partitionedWriterCount = min(max(nextPowerOfTwo(getAvailablePhysicalProcessorCount()), 2), 32) * 2;
     // Default value of task concurrency should be above 1, otherwise it can create a plan with a single gather
     // exchange node on the coordinator due to a single available processor. Whereas, on the worker nodes due to
     // more available processors, the default value could be above 1. Therefore, it can cause error due to config

core/trino-main/src/test/java/io/trino/execution/TestTaskManagerConfig.java

@@ -34,7 +34,7 @@
 public class TestTaskManagerConfig
 {
     private static final int DEFAULT_PROCESSOR_COUNT = min(max(nextPowerOfTwo(getAvailablePhysicalProcessorCount()), 2), 32);
-    private static final int DEFAULT_SCALE_WRITERS_MAX_WRITER_COUNT = min(getAvailablePhysicalProcessorCount(), 32);
+    private static final int DEFAULT_SCALE_WRITERS_MAX_WRITER_COUNT = min(getAvailablePhysicalProcessorCount(), 32) * 2;
 
     @Test
     public void testDefaults()
@@ -66,7 +66,7 @@ public void testDefaults()
                 .setScaleWritersEnabled(true)
                 .setScaleWritersMaxWriterCount(DEFAULT_SCALE_WRITERS_MAX_WRITER_COUNT)
                 .setWriterCount(1)
-                .setPartitionedWriterCount(DEFAULT_PROCESSOR_COUNT)
+                .setPartitionedWriterCount(DEFAULT_PROCESSOR_COUNT * 2)
                 .setTaskConcurrency(DEFAULT_PROCESSOR_COUNT)
                 .setHttpResponseThreads(100)
                 .setHttpTimeoutThreads(3)