Add default.execution.threads server config to decouple default query parallelism from max

pinot-core/src/main/java/org/apache/pinot/core/plan/maker/InstancePlanMakerImplV2.java

@@ -105,6 +105,7 @@ public class InstancePlanMakerImplV2 implements PlanMaker {
 
   private final FetchPlanner _fetchPlanner = FetchPlannerRegistry.getPlanner();
   private int _maxExecutionThreads = Server.DEFAULT_QUERY_EXECUTOR_MAX_EXECUTION_THREADS;
+  private int _defaultExecutionThreads = Server.DEFAULT_QUERY_EXECUTOR_DEFAULT_EXECUTION_THREADS;
   private int _maxInitialResultHolderCapacity = Server.DEFAULT_QUERY_EXECUTOR_MAX_INITIAL_RESULT_HOLDER_CAPACITY;
   private int _minInitialIndexedTableCapacity = Server.DEFAULT_QUERY_EXECUTOR_MIN_INITIAL_INDEXED_TABLE_CAPACITY;
   // Limit on number of groups stored for each segment, beyond which no new group will be created
@@ -120,6 +121,9 @@ public class InstancePlanMakerImplV2 implements PlanMaker {
   public void init(PinotConfiguration queryExecutorConfig) {
     _maxExecutionThreads = queryExecutorConfig.getProperty(Server.MAX_EXECUTION_THREADS,
         Server.DEFAULT_QUERY_EXECUTOR_MAX_EXECUTION_THREADS);
+    _defaultExecutionThreads = queryExecutorConfig.getProperty(Server.DEFAULT_EXECUTION_THREADS,
+        Server.DEFAULT_QUERY_EXECUTOR_DEFAULT_EXECUTION_THREADS);
+    validateExecutionThreadConfig();
     _maxInitialResultHolderCapacity = queryExecutorConfig.getProperty(Server.MAX_INITIAL_RESULT_HOLDER_CAPACITY,
         Server.DEFAULT_QUERY_EXECUTOR_MAX_INITIAL_RESULT_HOLDER_CAPACITY);
     _minInitialIndexedTableCapacity = queryExecutorConfig.getProperty(Server.MIN_INITIAL_INDEXED_TABLE_CAPACITY,
@@ -142,10 +146,31 @@ public void init(PinotConfiguration queryExecutorConfig) {
         Server.DEFAULT_QUERY_EXECUTOR_GROUPBY_TRIM_THRESHOLD);
     Preconditions.checkState(_groupByTrimThreshold > 0,
         "Invalid configurable: groupByTrimThreshold: %d must be positive", _groupByTrimThreshold);
-    LOGGER.info("Initialized plan maker with maxExecutionThreads: {}, maxInitialResultHolderCapacity: {}, "
-            + "numGroupsLimit: {}, minSegmentGroupTrimSize: {}, minServerGroupTrimSize: {}, groupByTrimThreshold: {}",
-        _maxExecutionThreads, _maxInitialResultHolderCapacity, _numGroupsLimit, _minSegmentGroupTrimSize,
-        _minServerGroupTrimSize, _groupByTrimThreshold);
+    LOGGER.info("Initialized plan maker with maxExecutionThreads: {}, defaultExecutionThreads: {}, "
+            + "maxInitialResultHolderCapacity: {}, numGroupsLimit: {}, minSegmentGroupTrimSize: {}, "
+            + "minServerGroupTrimSize: {}, groupByTrimThreshold: {}",
+        _maxExecutionThreads, _defaultExecutionThreads, _maxInitialResultHolderCapacity, _numGroupsLimit,
+        _minSegmentGroupTrimSize, _minServerGroupTrimSize, _groupByTrimThreshold);
+  }
+
+  @VisibleForTesting
+  public void setMaxExecutionThreads(int maxExecutionThreads) {
+    _maxExecutionThreads = maxExecutionThreads;
+    validateExecutionThreadConfig();
+  }
+
+  @VisibleForTesting
+  public void setDefaultExecutionThreads(int defaultExecutionThreads) {
+    _defaultExecutionThreads = defaultExecutionThreads;
+    validateExecutionThreadConfig();
+  }
+
+  private void validateExecutionThreadConfig() {
+    if (_defaultExecutionThreads > 0 && _maxExecutionThreads > 0) {
+      Preconditions.checkState(_defaultExecutionThreads <= _maxExecutionThreads,
+          "Invalid configuration: defaultExecutionThreads: %d must be <= maxExecutionThreads: %d",
+          _defaultExecutionThreads, _maxExecutionThreads);
+    }
   }
 
   @VisibleForTesting
@@ -208,7 +233,8 @@ public Plan makeInstancePlan(List<SegmentContext> segmentContexts, QueryContext
         new InstanceResponsePlanNode(combinePlanNode, segmentContexts, fetchContexts, queryContext));
   }
 
-  private void applyQueryOptions(QueryContext queryContext) {
+  @VisibleForTesting
+  void applyQueryOptions(QueryContext queryContext) {
     Map<String, String> queryOptions = queryContext.getQueryOptions();
 
     // Set skipUpsert
@@ -227,15 +253,20 @@ private void applyQueryOptions(QueryContext queryContext) {
     queryContext.setSkipIndexes(QueryOptionsUtils.getSkipIndexes(queryOptions));
 
     // Set maxExecutionThreads
+    // Resolution order:
+    //   1. Per-query override (SET maxExecutionThreads=N) — capped by server max
+    //   2. Server-level default (default.execution.threads) — decoupled from max, but still capped by it
+    //   3. Server-level max (max.execution.threads) — legacy fallback
     int maxExecutionThreads;
     Integer maxExecutionThreadsFromQuery = QueryOptionsUtils.getMaxExecutionThreads(queryOptions);
     if (maxExecutionThreadsFromQuery != null) {
-      // Do not allow query to override the execution threads over the instance-level limit
       if (_maxExecutionThreads > 0) {
         maxExecutionThreads = Math.min(_maxExecutionThreads, maxExecutionThreadsFromQuery);
       } else {
         maxExecutionThreads = maxExecutionThreadsFromQuery;
       }
+    } else if (_defaultExecutionThreads > 0) {
+      maxExecutionThreads = _defaultExecutionThreads;
     } else {
       maxExecutionThreads = _maxExecutionThreads;
     }

pinot-core/src/test/java/org/apache/pinot/core/plan/maker/InstancePlanMakerImplV2Test.java

@@ -0,0 +1,104 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.pinot.core.plan.maker;
+
+import java.util.Map;
+import org.apache.pinot.core.query.request.context.QueryContext;
+import org.apache.pinot.core.query.request.context.utils.QueryContextConverterUtils;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.utils.CommonConstants.Broker.Request.QueryOptionKey;
+import org.apache.pinot.spi.utils.CommonConstants.Server;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+
+
+/**
+ * Tests for execution-thread resolution in {@link InstancePlanMakerImplV2}, covering the interplay
+ * between {@code max.execution.threads}, {@code default.execution.threads}, and per-query overrides.
+ */
+public class InstancePlanMakerImplV2Test {
+
+  private static final String BASE_QUERY = "SELECT * FROM testTable";
+
+  private static QueryContext buildQueryContext(String maxExecutionThreadsOption) {
+    QueryContext queryContext = QueryContextConverterUtils.getQueryContext(BASE_QUERY);
+    if (maxExecutionThreadsOption != null) {
+      Map<String, String> queryOptions = queryContext.getQueryOptions();
+      queryOptions.put(QueryOptionKey.MAX_EXECUTION_THREADS, maxExecutionThreadsOption);
+    }
+    return queryContext;
+  }
+
+  @Test
+  public void testDefaultNotSetFallsBackToMax() {
+    InstancePlanMakerImplV2 planMaker = new InstancePlanMakerImplV2();
+    planMaker.setMaxExecutionThreads(12);
+
+    QueryContext queryContext = buildQueryContext(null);
+    planMaker.applyQueryOptions(queryContext);
+
+    assertEquals(queryContext.getMaxExecutionThreads(), 12);
+  }
+
+  @Test
+  public void testDefaultExecutionThreadsUsedWhenSet() {
+    InstancePlanMakerImplV2 planMaker = new InstancePlanMakerImplV2();
+    planMaker.setMaxExecutionThreads(16);
+    planMaker.setDefaultExecutionThreads(4);
+
+    QueryContext queryContext = buildQueryContext(null);
+    planMaker.applyQueryOptions(queryContext);
+
+    assertEquals(queryContext.getMaxExecutionThreads(), 4);
+  }
+
+  @Test
+  public void testQueryOverrideOverridesDefault() {
+    InstancePlanMakerImplV2 planMaker = new InstancePlanMakerImplV2();
+    planMaker.setMaxExecutionThreads(16);
+    planMaker.setDefaultExecutionThreads(4);
+
+    QueryContext queryContext = buildQueryContext("10");
+    planMaker.applyQueryOptions(queryContext);
+
+    assertEquals(queryContext.getMaxExecutionThreads(), 10);
+  }
+
+  @Test
+  public void testQueryOverrideCappedByMax() {
+    InstancePlanMakerImplV2 planMaker = new InstancePlanMakerImplV2();
+    planMaker.setMaxExecutionThreads(8);
+    planMaker.setDefaultExecutionThreads(4);
+
+    QueryContext queryContext = buildQueryContext("20");
+    planMaker.applyQueryOptions(queryContext);
+
+    assertEquals(queryContext.getMaxExecutionThreads(), 8);
+  }
+
+  @Test(expectedExceptions = IllegalStateException.class)
+  public void testInitRejectsDefaultExceedingMax() {
+    InstancePlanMakerImplV2 planMaker = new InstancePlanMakerImplV2();
+    PinotConfiguration config = new PinotConfiguration();
+    config.setProperty(Server.MAX_EXECUTION_THREADS, 4);
+    config.setProperty(Server.DEFAULT_EXECUTION_THREADS, 8);
+    planMaker.init(config);
+  }
+}

pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java

@@ -1321,6 +1321,10 @@ public static class Server {
     public static final String CONFIG_OF_QUERY_EXECUTOR_MAX_EXECUTION_THREADS =
         QUERY_EXECUTOR_CONFIG_PREFIX + "." + MAX_EXECUTION_THREADS;
     public static final int DEFAULT_QUERY_EXECUTOR_MAX_EXECUTION_THREADS = -1;  // Use number of CPU cores
+    public static final String DEFAULT_EXECUTION_THREADS = "default.execution.threads";
+    public static final String CONFIG_OF_QUERY_EXECUTOR_DEFAULT_EXECUTION_THREADS =
+        QUERY_EXECUTOR_CONFIG_PREFIX + "." + DEFAULT_EXECUTION_THREADS;
+    public static final int DEFAULT_QUERY_EXECUTOR_DEFAULT_EXECUTION_THREADS = -1;  // Not set; fall back to max
 
     // OOM protection: heap usage throttle configuration
     public static final String CONFIG_OF_HEAP_USAGE_THROTTLE_QUEUE_MAX_SIZE =