analytics-engine: dedicated reduce thread pool (opensearch-project#21800)

* analytics-engine: dedicated reduce thread pool

Add a fixed analytics_reduce pool (size=max(2, processors/2), queue=100)
for coordinator reduce drains. The reduce blocks for the entire query
duration via a synchronous FFM call (streamNext), so sharing the SEARCH
pool causes deadlock: concurrent reduces occupy all search threads,
preventing local shard fragments from executing and feeding the reduce.

The reduce pool is separate from both SEARCH (which runs shard fragments)
and analytics_scheduler (which orchestrates stage lifecycle). Pool size
is adjustable via thread_pool.analytics_reduce.size cluster setting.

Signed-off-by: Marc Handalian <marc.handalian@gmail.com>

Author: mch2

sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/AnalyticsPlugin.java

@@ -69,6 +69,17 @@ public class AnalyticsPlugin extends Plugin implements ExtensiblePlugin, ActionP
     public static final String SCHEDULER_THREAD_POOL_NAME = "analytics_scheduler";
     private static final int SCHEDULER_QUEUE_SIZE = 200;
 
+    public static final String REDUCE_THREAD_POOL_NAME = "analytics_reduce";
+
+    // The reduce pool exists to isolate coordinator-reduce drains from the SEARCH
+    // pool, preventing deadlock when reduces and local shard fragments compete for
+    // the same threads. Each reduce thread blocks on a synchronous FFM call
+    // (streamNext) with negligible CPU usage — memory is the real constraint,
+    // bounded by the DataFusion pool and phantom reservations. Size is generous
+    // so the thread pool isn't the throughput bottleneck.
+    private static final int REDUCE_POOL_SIZE = Math.max(8, Runtime.getRuntime().availableProcessors() * 4);
+    private static final int REDUCE_QUEUE_SIZE = 200;
+
     public static final Setting<Long> COORDINATOR_BUFFER_LIMIT = Setting.longSetting(
         "analytics.coordinator.buffer_limit",
         256L * 1024 * 1024,
@@ -152,7 +163,10 @@ public List<Setting<?>> getSettings() {
     @Override
     public List<ExecutorBuilder<?>> getExecutorBuilders(Settings settings) {
         int poolSize = schedulerPoolSize();
-        return List.of(new FixedExecutorBuilder(settings, SCHEDULER_THREAD_POOL_NAME, poolSize, SCHEDULER_QUEUE_SIZE, "analytics"));
+        return List.of(
+            new FixedExecutorBuilder(settings, SCHEDULER_THREAD_POOL_NAME, poolSize, SCHEDULER_QUEUE_SIZE, "analytics"),
+            new FixedExecutorBuilder(settings, REDUCE_THREAD_POOL_NAME, REDUCE_POOL_SIZE, REDUCE_QUEUE_SIZE, "analytics_reduce")
+        );
     }
 
     static int schedulerPoolSize() {

sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/QueryContext.java

@@ -83,6 +83,10 @@ public Executor schedulerExecutor() {
         return threadPool != null ? threadPool.executor(AnalyticsPlugin.SCHEDULER_THREAD_POOL_NAME) : Runnable::run;
     }
 
+    public Executor reduceExecutor() {
+        return threadPool != null ? threadPool.executor(AnalyticsPlugin.REDUCE_THREAD_POOL_NAME) : Runnable::run;
+    }
+
     public AnalyticsQueryTask parentTask() {
         return parentTask;
     }

sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/stage/coordinator/ReduceStageExecution.java

@@ -38,13 +38,13 @@ public final class ReduceStageExecution extends AbstractStageExecution implement
 
     private final ReducingExchangeSink backendSink;
     private final ExchangeSink downstream;
-    private final Executor searchExecutor;
+    private final Executor reduceExecutor;
 
     public ReduceStageExecution(Stage stage, QueryContext config, ReducingExchangeSink backendSink, ExchangeSink downstream) {
         super(stage, config.queryId(), config.operationListeners(), config.parentTask());
         this.backendSink = backendSink;
         this.downstream = downstream;
-        this.searchExecutor = config.searchExecutor();
+        this.reduceExecutor = config.reduceExecutor();
         this.runner = new LocalTaskRunner(config.schedulerExecutor());
     }
 
@@ -81,7 +81,7 @@ public ExchangeSource outputSource() {
     @Override
     protected List<StageTask> materializeTasks() {
         return List.of(new LocalStageTask(new StageTaskId(getStageId(), 0), listener -> {
-            searchExecutor.execute(() -> {
+            reduceExecutor.execute(() -> {
                 try {
                     backendSink.reduce(listener);
                 } catch (Exception e) {

sandbox/plugins/analytics-engine/src/test/java/org/opensearch/analytics/exec/stage/ReduceStageExecutionTests.java

@@ -257,7 +257,7 @@ private static ExecutorService inlineExecutor() {
 
     private static QueryContext mockContext() {
         QueryContext ctx = mock(QueryContext.class);
-        when(ctx.searchExecutor()).thenReturn(inlineExecutor());
+        when(ctx.reduceExecutor()).thenReturn(inlineExecutor());
         when(ctx.schedulerExecutor()).thenReturn(inlineExecutor());
         when(ctx.parentTask()).thenReturn(mock(AnalyticsQueryTask.class));
         return ctx;