[Enhancement] Forward cluster planning settings to the Analytics Engine unified query path

plugin/src/main/java/org/opensearch/sql/plugin/rest/RestUnifiedQueryAction.java

@@ -10,9 +10,11 @@
 import static org.opensearch.sql.opensearch.executor.OpenSearchQueryManager.SQL_WORKER_THREAD_POOL_NAME;
 import static org.opensearch.sql.protocol.response.format.JsonResponseFormatter.Style.PRETTY;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.google.gson.JsonElement;
 import com.google.gson.JsonObject;
 import com.google.gson.JsonParser;
+import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import org.apache.calcite.rel.RelNode;
@@ -40,6 +42,7 @@
 import org.opensearch.sql.calcite.CalcitePlanContext;
 import org.opensearch.sql.calcite.plan.rel.LogicalSystemLimit;
 import org.opensearch.sql.common.response.ResponseListener;
+import org.opensearch.sql.common.setting.Settings.Key;
 import org.opensearch.sql.executor.ExecutionEngine.QueryResponse;
 import org.opensearch.sql.executor.QueryType;
 import org.opensearch.sql.executor.analytics.AnalyticsExecutionEngine;
@@ -346,26 +349,42 @@ private static QueryRequestContext withParentTask(QueryRequestContext ctx, Task
     return new QueryRequestContext(ctx.clusterState(), ctx.schema(), ctx.querySource(), parentTask);
   }
 
+  /**
+   * Cluster settings forwarded into every {@link UnifiedQueryContext}, so the Analytics Engine
+   * plans against the same configuration as the default pipeline. A planning setting the AE path
+   * must honor belongs here; otherwise the value {@link UnifiedQueryContext.Builder} seeds silently
+   * wins and the configured cluster value is ignored.
+   *
+   * <p>The other settings the AE path reads are deliberately not forwarded; {@code
+   * RestUnifiedQueryActionTest#DELIBERATELY_NOT_FORWARDED} enumerates them with a reason each, and
+   * pins both lists.
+   */
+  @VisibleForTesting
+  static final List<Key> FORWARDED_CLUSTER_SETTINGS =
+      List.of(
+          Key.QUERY_SIZE_LIMIT,
+          Key.PPL_REX_MAX_MATCH_LIMIT,
+          Key.PPL_SYNTAX_LEGACY_PREFERRED,
+          Key.MAX_EXPRESSION_DEPTH,
+          Key.PATTERN_METHOD,
+          Key.PATTERN_MODE,
+          Key.PATTERN_MAX_SAMPLE_COUNT,
+          Key.PATTERN_BUFFER_LIMIT,
+          Key.PATTERN_SHOW_NUMBERED_TOKEN);
+
   /**
    * Routes operator-configured cluster overrides into the builder via the existing {@code
    * setting(String, Object)} API, keeping {@link UnifiedQueryContext} decoupled from any specific
-   * {@link org.opensearch.sql.common.setting.Settings} implementation.
-   *
-   * <p>Add keys here if a future PR / IT depends on cluster-side fidelity for one of the other
-   * planning settings.
+   * {@link org.opensearch.sql.common.setting.Settings} implementation. The forwarded keys are
+   * {@link #FORWARDED_CLUSTER_SETTINGS}.
    */
-  private UnifiedQueryContext.Builder applyClusterOverrides(UnifiedQueryContext.Builder builder) {
-    forwardClusterSetting(
-        builder, org.opensearch.sql.common.setting.Settings.Key.PPL_REX_MAX_MATCH_LIMIT);
-    forwardClusterSetting(
-        builder, org.opensearch.sql.common.setting.Settings.Key.PPL_SYNTAX_LEGACY_PREFERRED);
-    forwardClusterSetting(
-        builder, org.opensearch.sql.common.setting.Settings.Key.MAX_EXPRESSION_DEPTH);
+  @VisibleForTesting
+  UnifiedQueryContext.Builder applyClusterOverrides(UnifiedQueryContext.Builder builder) {
+    FORWARDED_CLUSTER_SETTINGS.forEach(key -> forwardClusterSetting(builder, key));
     return builder;
   }
 
-  private void forwardClusterSetting(
-      UnifiedQueryContext.Builder builder, org.opensearch.sql.common.setting.Settings.Key key) {
+  private void forwardClusterSetting(UnifiedQueryContext.Builder builder, Key key) {
     Object value = pluginSettings.getSettingValue(key);
     if (value != null) {
       builder.setting(key.getKeyValue(), value);

plugin/src/test/java/org/opensearch/sql/plugin/rest/RestUnifiedQueryActionTest.java

@@ -5,11 +5,16 @@
 
 package org.opensearch.sql.plugin.rest;
 
+import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotEquals;
 import static org.junit.Assert.assertTrue;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
+import static org.opensearch.sql.plugin.rest.RestUnifiedQueryAction.FORWARDED_CLUSTER_SETTINGS;
 
+import java.util.List;
+import java.util.Map;
 import org.apache.calcite.rel.RelNode;
 import org.junit.Before;
 import org.junit.Test;
@@ -22,6 +27,8 @@
 import org.opensearch.common.settings.Settings;
 import org.opensearch.index.IndexSettings;
 import org.opensearch.indices.IndicesService;
+import org.opensearch.sql.api.UnifiedQueryContext;
+import org.opensearch.sql.common.setting.Settings.Key;
 import org.opensearch.sql.executor.QueryType;
 import org.opensearch.transport.client.node.NodeClient;
 
@@ -33,6 +40,7 @@ public class RestUnifiedQueryActionTest {
 
   private ClusterService clusterService;
   private Metadata metadata;
+  private org.opensearch.sql.common.setting.Settings pluginSettings;
   private RestUnifiedQueryAction action;
 
   @Before
@@ -46,6 +54,7 @@ public void setUp() {
     // path is only exercised when this returns something other than "composite".
     when(clusterService.getSettings()).thenReturn(Settings.EMPTY);
 
+    pluginSettings = mock(org.opensearch.sql.common.setting.Settings.class);
     @SuppressWarnings("unchecked")
     QueryPlanExecutor<RelNode, Iterable<Object[]>> executor = mock(QueryPlanExecutor.class);
     action =
@@ -54,7 +63,7 @@ public void setUp() {
             clusterService,
             executor,
             mock(EngineContextProvider.class),
-            mock(org.opensearch.sql.common.setting.Settings.class),
+            pluginSettings,
             new org.opensearch.sql.executor.DirectExecutionDispatcher());
   }
 
@@ -237,6 +246,132 @@ public void pplUnparseableQueryRoutesToAnalyticsUnderClusterComposite() {
     assertTrue(action.isAnalyticsIndex("source = parquet_logs | | fields ts", QueryType.PPL));
   }
 
+  @Test
+  public void clusterQuerySizeLimitReachesAnalyticsContext() {
+    // Regression: the AE path pinned QUERY_SIZE_LIMIT to the builder's hardcoded default (10000),
+    // silently ignoring the configured cluster value. Forwarding must carry the live value through
+    // to the plan context, since addQuerySizeLimit reads it from there.
+    when(pluginSettings.getSettingValue(Key.QUERY_SIZE_LIMIT)).thenReturn(500);
+
+    assertEquals(
+        "Cluster plugins.query.size_limit must reach the AE plan context",
+        Integer.valueOf(500),
+        buildAnalyticsContext().getPlanContext().sysLimit.querySizeLimit());
+  }
+
+  @Test
+  public void calciteEngineEnabledNotOverriddenByCluster() {
+    // CALCITE_ENGINE_ENABLED is deliberately excluded from forwarding: the unified path is
+    // Calcite-based by definition and must stay true even if the cluster disables it.
+    when(pluginSettings.getSettingValue(Key.CALCITE_ENGINE_ENABLED)).thenReturn(false);
+
+    assertEquals(
+        "Unified path must force Calcite on regardless of the cluster setting",
+        Boolean.TRUE,
+        buildAnalyticsContext().getSettings().getSettingValue(Key.CALCITE_ENGINE_ENABLED));
+  }
+
+  /**
+   * Every setting the AE path reads that {@link RestUnifiedQueryAction} deliberately does not
+   * forward, with the reason it stays unforwarded. Derived from the call sites reachable from the
+   * unified context's {@code Settings}: {@code SysLimit.fromSettings}, the {@code AstBuilder} /
+   * {@code AstExpressionBuilder} / {@code AstBuildGuard} behind its parser, and {@code
+   * UnresolvedPlanHelper}.
+   *
+   * <ul>
+   *   <li>{@link Key#CALCITE_ENGINE_ENABLED} — the unified path is Calcite-based by definition.
+   *   <li>{@link Key#PPL_SUBSEARCH_MAXOUT} / {@link Key#PPL_JOIN_SUBSEARCH_MAXOUT} — seeded to
+   *       {@code 0} (unlimited) on purpose, for external consumers of the unified query API.
+   *   <li>{@link Key#PPL_VALUES_MAX_LIMIT} — forwarding it lowers {@code values()} to {@code
+   *       array_agg(DISTINCT x, limit)}, which the backend cannot bind, so the query fails outright
+   *       where today it merely ignores the cap.
+   *   <li>{@link Key#CALCITE_SUPPORT_ALL_JOIN_TYPES} — never seeded, so {@code
+   *       AstBuilder.validateJoinType} reads {@code null} and skips the high-cost-join guard.
+   *       Restoring it is a user-visible tightening.
+   * </ul>
+   */
+  private static final List<Key> DELIBERATELY_NOT_FORWARDED =
+      List.of(
+          Key.CALCITE_ENGINE_ENABLED,
+          Key.PPL_SUBSEARCH_MAXOUT,
+          Key.PPL_JOIN_SUBSEARCH_MAXOUT,
+          Key.PPL_VALUES_MAX_LIMIT,
+          Key.CALCITE_SUPPORT_ALL_JOIN_TYPES);
+
+  /**
+   * Drift guard for the defect class this forwarding exists to prevent: the builder's seed map and
+   * the handler's forward list are maintained independently, so a planning setting seeded but not
+   * forwarded silently regresses to its hardcoded default (this is how {@code
+   * plugins.query.size_limit} came to be ignored on the AE path). Every seeded key must therefore
+   * be classified — forwarded, or explicitly excluded with a reason.
+   */
+  @Test
+  public void everySeededPlanningSettingIsClassified() {
+    UnifiedQueryContext defaults = UnifiedQueryContext.builder().language(QueryType.PPL).build();
+
+    for (Object entry : defaults.getSettings().getSettings()) {
+      Key seeded = ((Map.Entry<Key, ?>) entry).getKey();
+      assertTrue(
+          "Setting "
+              + seeded.getKeyValue()
+              + " is seeded with a hardcoded default by UnifiedQueryContext.Builder but is neither"
+              + " forwarded from the cluster nor listed in DELIBERATELY_NOT_FORWARDED. Add it to"
+              + " RestUnifiedQueryAction.FORWARDED_CLUSTER_SETTINGS so the configured cluster value"
+              + " reaches the Analytics Engine, or document why it must stay hardcoded.",
+          FORWARDED_CLUSTER_SETTINGS.contains(seeded)
+              || DELIBERATELY_NOT_FORWARDED.contains(seeded));
+    }
+  }
+
+  @Test
+  public void clusterPatternSettingsReachAnalyticsContext() {
+    // patterns command defaults are read straight off the context's settings in AstBuilder, so a
+    // cluster-configured method/mode/limit must be visible there rather than the seeded default.
+    when(pluginSettings.getSettingValue(Key.PATTERN_METHOD)).thenReturn("BRAIN");
+    when(pluginSettings.getSettingValue(Key.PATTERN_MODE)).thenReturn("AGGREGATION");
+    when(pluginSettings.getSettingValue(Key.PATTERN_MAX_SAMPLE_COUNT)).thenReturn(42);
+    when(pluginSettings.getSettingValue(Key.PATTERN_BUFFER_LIMIT)).thenReturn(60000);
+    when(pluginSettings.getSettingValue(Key.PATTERN_SHOW_NUMBERED_TOKEN)).thenReturn(true);
+
+    org.opensearch.sql.common.setting.Settings forwarded = buildAnalyticsContext().getSettings();
+
+    assertEquals("BRAIN", forwarded.getSettingValue(Key.PATTERN_METHOD));
+    assertEquals("AGGREGATION", forwarded.getSettingValue(Key.PATTERN_MODE));
+    assertEquals(Integer.valueOf(42), forwarded.getSettingValue(Key.PATTERN_MAX_SAMPLE_COUNT));
+    assertEquals(Integer.valueOf(60000), forwarded.getSettingValue(Key.PATTERN_BUFFER_LIMIT));
+    assertEquals(Boolean.TRUE, forwarded.getSettingValue(Key.PATTERN_SHOW_NUMBERED_TOKEN));
+  }
+
+  @Test
+  public void documentedExclusionsAreNotForwarded() {
+    // Pins the exclusion list in RestUnifiedQueryAction's javadoc: each of these is a conscious
+    // decision, not an oversight, so a well-meaning "just forward everything" change fails here.
+    // PPL_VALUES_MAX_LIMIT in particular must stay out: forwarding it lowers values() to
+    // array_agg(DISTINCT x, limit), which the AE backend cannot bind, turning a silently-ignored
+    // cap into a 500 (verified on a live composite/parquet cluster).
+    DELIBERATELY_NOT_FORWARDED.forEach(
+        key -> when(pluginSettings.getSettingValue(key)).thenReturn(SENTINEL));
+
+    org.opensearch.sql.common.setting.Settings forwarded = buildAnalyticsContext().getSettings();
+
+    for (Key key : DELIBERATELY_NOT_FORWARDED) {
+      assertNotEquals(
+          key.getKeyValue() + " must not be forwarded to the Analytics Engine context",
+          SENTINEL,
+          forwarded.getSettingValue(key));
+    }
+  }
+
+  /** Value no seeded default uses, so "forwarded" is distinguishable from "seeded" or "absent". */
+  private static final Integer SENTINEL = -12345;
+
+  /** Builds the context the AE path plans against, with the mocked cluster settings applied. */
+  private UnifiedQueryContext buildAnalyticsContext() {
+    return action
+        .applyClusterOverrides(UnifiedQueryContext.builder().language(QueryType.PPL))
+        .build();
+  }
+
   private void enableClusterComposite() {
     when(clusterService.getSettings())
         .thenReturn(