feat: complex query thread pool

.gitignore

@@ -25,6 +25,7 @@ src/site-server/node_modules
 build/
 gen/
 *.tokens
+.attach_pid*
 
 # various IDE files
 .vscode
@@ -59,4 +60,4 @@ http-client.env.json
 !.claude/harness/
 .claude/settings.local.json
 .clinerules
-memory-bank
+memory-bank

common/src/main/java/org/opensearch/sql/common/setting/Settings.java

@@ -79,7 +79,10 @@ public enum Key {
     ASYNC_QUERY_EXTERNAL_SCHEDULER_INTERVAL(
         "plugins.query.executionengine.async_query.external_scheduler.interval"),
     STREAMING_JOB_HOUSEKEEPER_INTERVAL(
-        "plugins.query.executionengine.spark.streamingjobs.housekeeper.interval");
+        "plugins.query.executionengine.spark.streamingjobs.housekeeper.interval"),
+
+    /** Thread Pool Settings. */
+    SQL_COMPLEX_WORKER_POOL_ENABLED("plugins.sql.complex_worker_pool.enabled");
 
     @Getter private final String keyValue;
 

core/src/main/java/org/opensearch/sql/calcite/CalcitePlanContext.java

@@ -58,6 +58,11 @@ public class CalcitePlanContext {
   /** Timewrap series mode: "relative", "short", or "exact". */
   public static final ThreadLocal<String> timewrapSeries = new ThreadLocal<>();
 
+  /**
+   * Thread-local tracking which pool executed this query ("sql-worker" or "sql-complex-worker").
+   */
+  public static final ThreadLocal<String> executionPool = new ThreadLocal<>();
+
   /** Thread-local switch that tells whether the current query prefers legacy behavior. */
   private static final ThreadLocal<Boolean> legacyPreferredFlag =
       ThreadLocal.withInitial(() -> true);
@@ -245,6 +250,51 @@ public static void clearTimewrapSignals() {
     stripNullColumns.set(false);
     timewrapUnitName.set(null);
     timewrapSeries.set(null);
+    executionPool.set(null);
+  }
+
+  /**
+   * Snapshot of all thread-local state in CalcitePlanContext. Used when dispatching queries to the
+   * complex worker pool — capture state on the caller thread, restore on the worker thread.
+   */
+  public static class ThreadLocalSnapshot {
+    final boolean skipEncoding;
+    final boolean stripNullColumns;
+    final String timewrapUnitName;
+    final String timewrapSeries;
+    final String executionPool;
+
+    private ThreadLocalSnapshot(
+        boolean skipEncoding,
+        boolean stripNullColumns,
+        String timewrapUnitName,
+        String timewrapSeries,
+        String executionPool) {
+      this.skipEncoding = skipEncoding;
+      this.stripNullColumns = stripNullColumns;
+      this.timewrapUnitName = timewrapUnitName;
+      this.timewrapSeries = timewrapSeries;
+      this.executionPool = executionPool;
+    }
+  }
+
+  /** Capture current thread-local state for cross-thread propagation. */
+  public static ThreadLocalSnapshot snapshotThreadLocals() {
+    return new ThreadLocalSnapshot(
+        skipEncoding.get(),
+        stripNullColumns.get(),
+        timewrapUnitName.get(),
+        timewrapSeries.get(),
+        executionPool.get());
+  }
+
+  /** Restore thread-local state from a snapshot. */
+  public static void restoreThreadLocals(ThreadLocalSnapshot snapshot) {
+    skipEncoding.set(snapshot.skipEncoding);
+    stripNullColumns.set(snapshot.stripNullColumns);
+    timewrapUnitName.set(snapshot.timewrapUnitName);
+    timewrapSeries.set(snapshot.timewrapSeries);
+    executionPool.set(snapshot.executionPool);
   }
 
   public void pushForeachBindings(

core/src/main/java/org/opensearch/sql/calcite/utils/CalciteToolsHelper.java

@@ -516,8 +516,6 @@ private static void enrichErrorsForSpecialCases(ErrorReport.Builder report, SQLE
     public static PreparedStatement run(CalcitePlanContext context, RelNode rel) {
       ProfileMetric optimizeTime = QueryProfiling.current().getOrCreateMetric(OPTIMIZE);
       long startTime = System.nanoTime();
-      // Optimize the plan by Calcite's HepPlanner before using VolcanoPlanner in prepareStatement.
-      rel = CalciteToolsHelper.optimize(rel, context);
       final RelShuttle shuttle =
           new RelHomogeneousShuttle() {
             @Override

core/src/main/java/org/opensearch/sql/executor/DirectExecutionDispatcher.java

@@ -0,0 +1,26 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.sql.executor;
+
+import org.apache.calcite.rel.RelNode;
+import org.opensearch.sql.calcite.CalcitePlanContext;
+import org.opensearch.sql.common.response.ResponseListener;
+
+/**
+ * Default no-op dispatcher that executes inline on the current thread. Used when complex-pool
+ * routing is disabled or as a fallback.
+ */
+public class DirectExecutionDispatcher implements ExecutionDispatcher {
+
+  @Override
+  public void dispatch(
+      RelNode plan,
+      CalcitePlanContext context,
+      ResponseListener<ExecutionEngine.QueryResponse> listener,
+      ExecutionEngine engine) {
+    engine.execute(plan, context, listener);
+  }
+}

core/src/main/java/org/opensearch/sql/executor/ExecutionDispatcher.java

@@ -0,0 +1,45 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.sql.executor;
+
+import org.apache.calcite.rel.RelNode;
+import org.opensearch.sql.calcite.CalcitePlanContext;
+import org.opensearch.sql.common.response.ResponseListener;
+
+/**
+ * Dispatches query execution to an appropriate thread pool based on plan characteristics. After
+ * query analysis and optimization, the dispatcher inspects the plan and routes execution to either
+ * the fast worker pool (for queries fully pushed to OpenSearch) or the complex worker pool (for
+ * queries requiring scripts/table scans).
+ */
+public interface ExecutionDispatcher {
+
+  /**
+   * Dispatch execution of the given plan via the standard ExecutionEngine.
+   *
+   * @param plan the optimized Calcite plan
+   * @param context the plan context
+   * @param listener response listener for query results
+   * @param engine the execution engine to invoke
+   */
+  void dispatch(
+      RelNode plan,
+      CalcitePlanContext context,
+      ResponseListener<ExecutionEngine.QueryResponse> listener,
+      ExecutionEngine engine);
+
+  /**
+   * Dispatch a task to the appropriate thread pool based on plan characteristics. Use this when the
+   * execution path differs from the standard ExecutionEngine interface (e.g., analytics engine).
+   *
+   * @param plan the optimized Calcite plan used for routing decisions
+   * @param context the plan context
+   * @param task the execution task to run
+   */
+  default void dispatchTask(RelNode plan, CalcitePlanContext context, Runnable task) {
+    task.run();
+  }
+}

core/src/main/java/org/opensearch/sql/executor/QueryService.java

@@ -16,7 +16,6 @@
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicReference;
 import javax.annotation.Nullable;
-import lombok.AllArgsConstructor;
 import lombok.Getter;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.log4j.Log4j2;
@@ -54,6 +53,7 @@
 import org.opensearch.sql.calcite.plan.rel.LogicalSystemLimit;
 import org.opensearch.sql.calcite.plan.rel.LogicalSystemLimit.SystemLimitType;
 import org.opensearch.sql.calcite.utils.CalciteClassLoaderHelper;
+import org.opensearch.sql.calcite.utils.CalciteToolsHelper;
 import org.opensearch.sql.calcite.utils.CalciteToolsHelper.OpenSearchRelRunners;
 import org.opensearch.sql.common.error.ErrorReport;
 import org.opensearch.sql.common.error.QueryProcessingStage;
@@ -78,14 +78,44 @@
 
 /** The low level interface of core engine. */
 @RequiredArgsConstructor
-@AllArgsConstructor
 @Log4j2
 public class QueryService {
   private final Analyzer analyzer;
   private final ExecutionEngine executionEngine;
   private final Planner planner;
   private DataSourceService dataSourceService;
   private Settings settings;
+  private ExecutionDispatcher executionDispatcher = new DirectExecutionDispatcher();
+
+  public QueryService(
+      Analyzer analyzer,
+      ExecutionEngine executionEngine,
+      Planner planner,
+      DataSourceService dataSourceService,
+      Settings settings) {
+    this(
+        analyzer,
+        executionEngine,
+        planner,
+        dataSourceService,
+        settings,
+        new DirectExecutionDispatcher());
+  }
+
+  public QueryService(
+      Analyzer analyzer,
+      ExecutionEngine executionEngine,
+      Planner planner,
+      DataSourceService dataSourceService,
+      Settings settings,
+      ExecutionDispatcher executionDispatcher) {
+    this.analyzer = analyzer;
+    this.executionEngine = executionEngine;
+    this.planner = planner;
+    this.dataSourceService = dataSourceService;
+    this.settings = settings;
+    this.executionDispatcher = executionDispatcher;
+  }
 
   @Getter(lazy = true)
   private final CalciteRelNodeVisitor relNodeVisitor = new CalciteRelNodeVisitor(dataSourceService);
@@ -199,9 +229,7 @@ public void executeWithCalcite(
                                   convertToCalcitePlan(relNode, context), context),
                           "while converting the query to an executable plan");
 
-                  analyzeMetric.set(System.nanoTime() - analyzeStart);
-
-                  executeCalcitePlan(calcitePlan, context, listener);
+                  executeCalcitePlan(calcitePlan, context, listener, analyzeMetric, analyzeStart);
                 },
                 QueryService.class);
           } catch (Throwable t) {
@@ -219,11 +247,20 @@ public void executeWithCalcite(
   private void executeCalcitePlan(
       RelNode calcitePlan,
       CalcitePlanContext context,
-      ResponseListener<ExecutionEngine.QueryResponse> listener) {
+      ResponseListener<ExecutionEngine.QueryResponse> listener,
+      ProfileMetric analyzeMetric,
+      long analyzeStart) {
     try {
+      // Optimize before dispatch so the dispatcher's ScriptDetector
+      // sees the post-optimization plan for accurate routing.
+      RelNode optimizedPlan = CalciteToolsHelper.optimize(calcitePlan, context);
+      analyzeMetric.set(System.nanoTime() - analyzeStart);
+
+      // Wrap execution with EXECUTING stage tracking — dispatch via
+      // ExecutionDispatcher which may route to a complex worker pool
       StageErrorHandler.executeStageVoid(
           QueryProcessingStage.EXECUTING,
-          () -> executionEngine.execute(calcitePlan, context, listener),
+          () -> executionDispatcher.dispatch(optimizedPlan, context, listener, executionEngine),
           "while running the query");
     } catch (RuntimeException e) {
       ArithmeticException overflow = findArithmeticOverflow(e);

core/src/main/java/org/opensearch/sql/monitor/profile/DefaultProfileContext.java

@@ -9,6 +9,7 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.concurrent.ConcurrentHashMap;
+import org.opensearch.sql.calcite.CalcitePlanContext;
 
 /** Default implementation that records profiling metrics. */
 public class DefaultProfileContext implements ProfileContext {
@@ -63,7 +64,8 @@ public synchronized QueryProfile finish() {
     double totalMillis = ProfileUtils.roundToMillis(endNanos - startNanos);
     Object planSnapshot =
         enginePlan != null ? enginePlan : (planRoot == null ? null : planRoot.snapshot());
-    profile = new QueryProfile(totalMillis, snapshot, planSnapshot);
+    String threadPool = CalcitePlanContext.executionPool.get();
+    profile = new QueryProfile(totalMillis, snapshot, planSnapshot, threadPool);
     return profile;
   }
 }

core/src/main/java/org/opensearch/sql/monitor/profile/QueryProfile.java

@@ -24,14 +24,17 @@ public final class QueryProfile {
   /** Execution-engine-specific plan profile: a {@link PlanNode} tree, or a pre-rendered object. */
   private final Object plan;
 
+  @SerializedName("thread_pool")
+  private final String threadPool;
+
   /**
    * Create a new query profile snapshot.
    *
    * @param totalTimeMillis total elapsed milliseconds for the query (rounded to two decimals)
    * @param phases metric values keyed by {@link MetricName}
    */
   public QueryProfile(double totalTimeMillis, Map<MetricName, Double> phases) {
-    this(totalTimeMillis, phases, null);
+    this(totalTimeMillis, phases, null, null);
   }
 
   /**
@@ -42,9 +45,23 @@ public QueryProfile(double totalTimeMillis, Map<MetricName, Double> phases) {
    * @param plan plan tree profiling output
    */
   public QueryProfile(double totalTimeMillis, Map<MetricName, Double> phases, Object plan) {
+    this(totalTimeMillis, phases, plan, null);
+  }
+
+  /**
+   * Create a new query profile snapshot.
+   *
+   * @param totalTimeMillis total elapsed milliseconds for the query (rounded to two decimals)
+   * @param phases metric values keyed by {@link MetricName}
+   * @param plan plan tree profiling output
+   * @param threadPool thread pool name that executed the query
+   */
+  public QueryProfile(
+      double totalTimeMillis, Map<MetricName, Double> phases, Object plan, String threadPool) {
     this.summary = new Summary(totalTimeMillis);
     this.phases = buildPhases(phases);
     this.plan = plan;
+    this.threadPool = threadPool;
   }
 
   private Map<String, Phase> buildPhases(Map<MetricName, Double> phases) {

core/src/main/java/org/opensearch/sql/monitor/profile/QueryProfiling.java

@@ -57,6 +57,16 @@ public static void clear() {
     CURRENT.remove();
   }
 
+  /**
+   * Set the profiling context for the current thread. Used when propagating context across thread
+   * boundaries.
+   *
+   * @param ctx profiling context to bind
+   */
+  public static void set(ProfileContext ctx) {
+    CURRENT.set(Objects.requireNonNull(ctx, "ctx"));
+  }
+
   /**
    * Run a supplier with the provided profiling context bound to the current thread.
    *

docs/user/ppl/interfaces/endpoint.md

@@ -303,7 +303,8 @@ Expected output (trimmed):
          "children": [
             { "node": "CalciteEnumerableIndexScan", "time_ms": 4.12, "rows": 2 }
          ]
-      }
+      },
+      "thread_pool": "sql-worker"
    }
 }
 ```
@@ -315,6 +316,7 @@ Expected output (trimmed):
 - Plan node names use Calcite physical operator names (for example, `EnumerableCalc` or `CalciteEnumerableIndexScan`).
 - Plan `time_ms` is inclusive of child operators and represents wall-clock time; overlapping work can make summed plan times exceed `summary.total_time_ms`.
 - Scan nodes reflect operator wall-clock time; background prefetch can make scan time smaller than total request latency.
+- `thread_pool` indicates which thread pool executed the query. Possible values are `sql-worker` (default, pushdown-only queries) and `sql-complex-worker` (queries requiring in-memory evaluation such as scripted fields).
 
 ## Highlight