Reduce ES|QL driver parallelism

Author: dnhatn

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/QueryPragmas.java

@@ -18,7 +18,6 @@
 import org.elasticsearch.compute.operator.Driver;
 import org.elasticsearch.compute.operator.DriverStatus;
 import org.elasticsearch.core.TimeValue;
-import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.xpack.esql.core.expression.Expression;
 
 import java.io.IOException;
@@ -34,9 +33,19 @@ public final class QueryPragmas implements Writeable {
 
     private static final Setting<Integer> TASK_CONCURRENCY = Setting.intSetting(
         "task_concurrency",
-        ThreadPool.searchOrGetThreadPoolSize(EsExecutors.allocatedProcessors(Settings.EMPTY))
+        allocatedProcessorsMinusOneMaxTen(EsExecutors.allocatedProcessors(Settings.EMPTY))
     );
 
+    /**
+     * Returns the number of processors to be used in driver parallelism, starting with Lucene operators (and reading values).
+     * One processor is reserved for node-level reduction.
+     * The concurrency is capped at 10 as drivers can contain both I/O-bound and CPU-bound tasks. This avoids blocking the thread pool
+     * due to waiting for I/O and reduces the resource usage by many pipelines, minimizing the cost for reduction.
+     */
+    static int allocatedProcessorsMinusOneMaxTen(final int allocatedProcessors) {
+        return Math.min(Math.max(allocatedProcessors - 1, 2), 10);
+    }
+
     public static final Setting<DataPartitioning> DATA_PARTITIONING = Setting.enumSetting(
         DataPartitioning.class,
         "data_partitioning",

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/plugin/QueryPragmasTests.java

@@ -0,0 +1,33 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.esql.plugin;
+
+import org.elasticsearch.test.ESTestCase;
+
+import static org.hamcrest.Matchers.equalTo;
+
+public class QueryPragmasTests extends ESTestCase {
+
+    public void testTaskConcurrency() {
+        assertThat(QueryPragmas.allocatedProcessorsMinusOneMaxTen(1), equalTo(2));
+        assertThat(QueryPragmas.allocatedProcessorsMinusOneMaxTen(2), equalTo(2));
+        assertThat(QueryPragmas.allocatedProcessorsMinusOneMaxTen(3), equalTo(2));
+        assertThat(QueryPragmas.allocatedProcessorsMinusOneMaxTen(4), equalTo(3));
+        assertThat(QueryPragmas.allocatedProcessorsMinusOneMaxTen(5), equalTo(4));
+        assertThat(QueryPragmas.allocatedProcessorsMinusOneMaxTen(6), equalTo(5));
+        assertThat(QueryPragmas.allocatedProcessorsMinusOneMaxTen(7), equalTo(6));
+        assertThat(QueryPragmas.allocatedProcessorsMinusOneMaxTen(8), equalTo(7));
+        assertThat(QueryPragmas.allocatedProcessorsMinusOneMaxTen(9), equalTo(8));
+        assertThat(QueryPragmas.allocatedProcessorsMinusOneMaxTen(10), equalTo(9));
+        assertThat(QueryPragmas.allocatedProcessorsMinusOneMaxTen(11), equalTo(10));
+        for (int i = 12; i < 64; i++) {
+            assertThat(QueryPragmas.allocatedProcessorsMinusOneMaxTen(i), equalTo(10));
+        }
+        assertThat(QueryPragmas.allocatedProcessorsMinusOneMaxTen(between(12, 4096)), equalTo(10));
+    }
+}