feat: complex query thread pool

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

@@ -81,7 +81,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);
@@ -229,6 +234,7 @@ public static void clearTimewrapSignals() {
     stripNullColumns.set(false);
     timewrapUnitName.set(null);
     timewrapSeries.set(null);
+    executionPool.set(null);
   }
 
   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

@@ -8,7 +8,6 @@
 import java.util.List;
 import java.util.Optional;
 import javax.annotation.Nullable;
-import lombok.AllArgsConstructor;
 import lombok.Getter;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.log4j.Log4j2;
@@ -42,6 +41,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.common.error.ErrorReport;
 import org.opensearch.sql.common.error.QueryProcessingStage;
 import org.opensearch.sql.common.error.StageErrorHandler;
@@ -64,14 +64,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);
@@ -207,9 +237,15 @@ private void executeCalcitePlan(
       CalcitePlanContext context,
       ResponseListener<ExecutionEngine.QueryResponse> listener) {
     try {
+      // Optimize before dispatch so the dispatcher's ScriptDetector
+      // sees the post-optimization plan for accurate routing.
+      RelNode optimizedPlan = CalciteToolsHelper.optimize(calcitePlan, context);
+
+      // 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.
    *

integ-test/src/test/java/org/opensearch/sql/calcite/standalone/CalcitePPLRelNodeIntegTestCase.java

@@ -25,6 +25,7 @@
 import org.apache.calcite.tools.RelBuilder;
 import org.opensearch.sql.calcite.CalcitePlanContext;
 import org.opensearch.sql.calcite.SysLimit;
+import org.opensearch.sql.calcite.utils.CalciteToolsHelper;
 import org.opensearch.sql.calcite.utils.CalciteToolsHelper.OpenSearchRelRunners;
 import org.opensearch.sql.executor.QueryType;
 
@@ -87,7 +88,8 @@ protected RexNode createStringArray(RexBuilder rexBuilder, String... values) {
   protected void executeRelNodeAndVerify(
       CalcitePlanContext planContext, RelNode relNode, ResultVerifier verifier)
       throws SQLException {
-    try (PreparedStatement statement = OpenSearchRelRunners.run(planContext, relNode)) {
+    try (PreparedStatement statement =
+        OpenSearchRelRunners.run(planContext, CalciteToolsHelper.optimize(relNode, planContext))) {
       ResultSet resultSet = statement.executeQuery();
       verifier.verify(resultSet);
     }

integ-test/src/test/java/org/opensearch/sql/security/FGACIndexScanningIT.java

@@ -656,4 +656,88 @@ public void testRowLevelSecurity(boolean useCalcite) throws IOException {
         expectedPublicDocs,
         totalDocs);
   }
+
+  /**
+   * Verifies that document-level security is enforced when queries are dispatched to the complex
+   * worker pool. Queries containing window functions (eventstats) are routed to sql-complex-worker;
+   * this test ensures the OpenSearch ThreadContext (which carries DLS filters) is correctly
+   * propagated across that thread pool boundary.
+   */
+  @ParameterizedTest
+  @ValueSource(booleans = {false, true})
+  public void testRowLevelSecurityEnforcedOnComplexPool(boolean useCalcite) throws IOException {
+    configureEngine(useCalcite);
+    String engineLabel = useCalcite ? "V3" : "V2";
+
+    // eventstats creates a Window node, which ScriptDetector flags as expensive,
+    // routing the query to the sql-complex-worker pool.
+    String query =
+        String.format(
+            "search source=%s | eventstats count() as total_count by security_level"
+                + " | stats count() by security_level",
+            SECURE_LOGS);
+    JSONObject result = executeQueryAsUser(query, LIMITED_USER);
+
+    var datarows = result.getJSONArray("datarows");
+    var schema = result.getJSONArray("schema");
+    int levelIdx = -1;
+    for (int i = 0; i < schema.length(); i++) {
+      String name = schema.getJSONObject(i).getString("name");
+      if ("security_level".equals(name)) {
+        levelIdx = i;
+      }
+    }
+    assertTrue("Expected security_level in schema", levelIdx >= 0);
+
+    for (int i = 0; i < datarows.length(); i++) {
+      var row = datarows.getJSONArray(i);
+      String securityLevel = row.getString(levelIdx);
+      assertFalse(
+          String.format(
+              "[%s] SECURITY VIOLATION on complex pool: limited_user saw '%s' documents. "
+                  + "DLS ThreadContext may not be propagated to sql-complex-worker pool.",
+              engineLabel, securityLevel),
+          "confidential".equals(securityLevel) || "internal".equals(securityLevel));
+    }
+  }
+
+  /**
+   * Verifies that field-level security is enforced when queries are dispatched to the slow worker
+   * pool. The eventstats command creates a Window node that triggers complex pool dispatch; this
+   * test ensures the restricted field (ssn) remains invisible.
+   */
+  @ParameterizedTest
+  @ValueSource(booleans = {false, true})
+  public void testFieldLevelSecurityEnforcedOnSlowPool(boolean useCalcite) throws IOException {
+    configureEngine(useCalcite);
+
+    // eventstats creates a Window node, which ScriptDetector flags as expensive,
+    // routing the query to the sql-complex-worker pool.
+    // manager_user should still NOT see ssn.
+    String query =
+        String.format(
+            "search source=%s | eventstats avg(salary) as avg_salary by department"
+                + " | fields name, department, salary, avg_salary | head 10",
+            EMPLOYEE_RECORDS);
+    JSONObject result = executeQueryAsUser(query, MANAGER_USER);
+
+    var resultSchema = result.getJSONArray("schema");
+    boolean hasSSN = false;
+    boolean hasName = false;
+    boolean hasAvgSalary = false;
+
+    for (int i = 0; i < resultSchema.length(); i++) {
+      String fieldName = resultSchema.getJSONObject(i).getString("name");
+      if ("ssn".equals(fieldName)) hasSSN = true;
+      if ("name".equals(fieldName)) hasName = true;
+      if ("avg_salary".equals(fieldName)) hasAvgSalary = true;
+    }
+
+    assertTrue("manager_user should see 'name' field on complex pool", hasName);
+    assertTrue("manager_user should see computed 'avg_salary' field on complex pool", hasAvgSalary);
+    assertFalse(
+        "SECURITY VIOLATION on complex pool: manager_user saw 'ssn' field. "
+            + "FLS ThreadContext may not be propagated to sql-complex-worker pool.",
+        hasSSN);
+  }
 }