Bridge live cluster settings into UnifiedQueryContext for PPL_REX_MAX_MATCH_LIMIT

The previous commit defaulted `PPL_REX_MAX_MATCH_LIMIT=10` in the
`UnifiedQueryContext.Builder`'s static defaults map to fix the NPE in
`AstBuilder.visitRexCommand` on the unified path. The default is correct, but
it doesn't respect mid-run cluster overrides — every key in the static map
returns its hardcoded value regardless of `_cluster/settings` updates. This
breaks `CalciteRexCommandIT.testRexMaxMatchConfigurableLimit`, which
explicitly sets the cluster-side limit to 5 and asserts that `max_match=0`
caps at 5; on the unified path it stayed at 10.

This change introduces a `Builder.settings(Settings)` hook that the REST
handler can use to inject the cluster's live `OpenSearchSettings` instance.
Inside `buildSettings()`, the Builder snapshots the live value of
`PPL_REX_MAX_MATCH_LIMIT` into the defaults map, overriding the hardcoded
default when the operator has set a cluster value. Snapshot-at-build matches
the per-HTTP-request lifecycle of `UnifiedQueryContext` and avoids per-call
lookup overhead.

The Builder's existing static defaults map is renamed `settings -> defaults`
to free up the `settings` identifier for the live-cluster-settings concept.

`RestUnifiedQueryAction` gains a `pluginSettings` field (the same
`OpenSearchSettings` instance bound in the Guice module) and forwards it to
the Builder in both `buildContext` and `buildParsingContext`. Both
construction sites — `SQLPlugin.createSqlAnalyticsRouter` and
`TransportPPLQueryAction.<init>` — are updated.

## Why scoped to `PPL_REX_MAX_MATCH_LIMIT` only

The same architectural gap exists for every key in the defaults map
(`QUERY_SIZE_LIMIT`, `PPL_SUBSEARCH_MAXOUT`, `PPL_JOIN_SUBSEARCH_MAXOUT`,
`CALCITE_ENGINE_ENABLED`). For three of those, the static defaults are fine
in practice (no test overrides them mid-run; `head N` covers `QUERY_SIZE_LIMIT`
per-query). `CALCITE_ENGINE_ENABLED` is intentionally pinned to `true` for
the unified path — a cluster override toggling it off would defeat the point
of routing here. So this PR widens only the one key that demonstrably needs
it; widening the snapshot to the rest is a future scope decision tied to
whichever new IT first depends on it.

Signed-off-by: Jialiang Liang <jiallian@amazon.com>

Author: RyanL1997

api/src/main/java/org/opensearch/sql/api/UnifiedQueryContext.java

@@ -117,6 +117,7 @@ public static class Builder {
     private String defaultNamespace;
     private boolean cacheMetadata = false;
     private boolean profiling = false;
+    private Settings settings;
 
     /**
      * Setting values with defaults from SysLimit.DEFAULT. Only includes planning-required settings
@@ -136,7 +137,7 @@ public static class Builder {
      * {@code OpenSearchSettings.PPL_REX_MAX_MATCH_LIMIT_SETTING}, so unified-path behavior matches
      * v2-path behavior when neither has an explicit cluster override.
      */
-    private final Map<Settings.Key, Object> settings =
+    private final Map<Settings.Key, Object> defaults =
         new HashMap<Settings.Key, Object>(
             Map.of(
                 QUERY_SIZE_LIMIT, SysLimit.DEFAULT.querySizeLimit(),
@@ -215,7 +216,25 @@ public Builder setting(String name, Object value) {
       Settings.Key key =
           Settings.Key.of(name)
               .orElseThrow(() -> new IllegalArgumentException("Unknown setting name: " + name));
-      settings.put(key, value);
+      defaults.put(key, value);
+      return this;
+    }
+
+    /**
+     * Injects the live cluster {@link Settings} so dynamic cluster overrides reach the unified-path
+     * settings map. Without this, every key in {@link #defaults} resolves to its hardcoded default
+     * and {@code _cluster/settings} updates are invisible to the unified path.
+     *
+     * <p>Scoped to {@link Settings.Key#PPL_REX_MAX_MATCH_LIMIT} for now — required so {@code rex
+     * max_match=0} caps at the operator-configured limit (CalciteRexCommandIT's
+     * testRexMaxMatchConfigurableLimit). The other keys in {@link #defaults} keep their static
+     * defaults; widening the snapshot to all keys is a separate scope.
+     *
+     * @param settings the cluster's live {@code OpenSearchSettings} instance
+     * @return this builder instance
+     */
+    public Builder settings(Settings settings) {
+      this.settings = settings;
       return this;
     }
 
@@ -232,6 +251,7 @@ public UnifiedQueryContext build() {
             case SQL -> UnifiedSqlSpec.extended();
             case PPL -> UnifiedPplSpec.create();
           };
+
       Settings settings = buildSettings();
       CalcitePlanContext planContext =
           CalcitePlanContext.create(
@@ -249,16 +269,27 @@ private UnifiedQueryParser<?> createParser(CalcitePlanContext planContext, Setti
     }
 
     private Settings buildSettings() {
+      // Snapshot PPL_REX_MAX_MATCH_LIMIT from live cluster settings if present so a
+      // mid-test `_cluster/settings` PUT takes effect on the unified path. Snapshot
+      // semantics (not per-call lookup) — UnifiedQueryContext is built per HTTP
+      // request, so a one-shot read at build() is sufficient. Other keys are
+      // intentionally not bridged here; that's a follow-up scope decision.
+      if (settings != null) {
+        Object liveRexLimit = settings.getSettingValue(PPL_REX_MAX_MATCH_LIMIT);
+        if (liveRexLimit != null) {
+          defaults.put(PPL_REX_MAX_MATCH_LIMIT, liveRexLimit);
+        }
+      }
       return new Settings() {
         @Override
         @SuppressWarnings("unchecked")
         public <T> T getSettingValue(Key key) {
-          return (T) settings.get(key);
+          return (T) defaults.get(key);
         }
 
         @Override
         public List<?> getSettings() {
-          return List.copyOf(settings.entrySet());
+          return List.copyOf(defaults.entrySet());
         }
       };
     }

plugin/src/main/java/org/opensearch/sql/plugin/SQLPlugin.java

@@ -208,7 +208,8 @@ private BiFunction<SQLQueryRequest, RestChannel, Boolean> createSqlAnalyticsRout
             if (executor == null) {
               return null;
             }
-            cached[0] = new RestUnifiedQueryAction(client, clusterService, executor);
+            cached[0] =
+                new RestUnifiedQueryAction(client, clusterService, executor, pluginSettings);
           }
           return cached[0];
         };

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

@@ -59,14 +59,17 @@ public class RestUnifiedQueryAction {
   private final AnalyticsExecutionEngine analyticsEngine;
   private final NodeClient client;
   private final ClusterService clusterService;
+  private final org.opensearch.sql.common.setting.Settings pluginSettings;
 
   public RestUnifiedQueryAction(
       NodeClient client,
       ClusterService clusterService,
-      QueryPlanExecutor<RelNode, Iterable<Object[]>> planExecutor) {
+      QueryPlanExecutor<RelNode, Iterable<Object[]>> planExecutor,
+      org.opensearch.sql.common.setting.Settings pluginSettings) {
     this.client = client;
     this.clusterService = clusterService;
     this.analyticsEngine = new AnalyticsExecutionEngine(planExecutor);
+    this.pluginSettings = pluginSettings;
   }
 
   /**
@@ -154,13 +157,14 @@ public void explain(
    * Build a lightweight context for parsing only (index name extraction). Does not require cluster
    * state or catalog schema.
    */
-  private static UnifiedQueryContext buildParsingContext(QueryType queryType) {
-    return UnifiedQueryContext.builder().language(queryType).build();
+  private UnifiedQueryContext buildParsingContext(QueryType queryType) {
+    return UnifiedQueryContext.builder().language(queryType).settings(pluginSettings).build();
   }
 
   private UnifiedQueryContext buildContext(QueryType queryType, boolean profiling) {
     return UnifiedQueryContext.builder()
         .language(queryType)
+        .settings(pluginSettings)
         .catalog(SCHEMA_NAME, OpenSearchSchemaBuilder.buildSchema(clusterService.state()))
         .defaultNamespace(SCHEMA_NAME)
         .profiling(profiling)

plugin/src/main/java/org/opensearch/sql/plugin/transport/TransportPPLQueryAction.java

@@ -67,6 +67,7 @@ public class TransportPPLQueryAction
 
   private final NodeClient clientRef;
   private final ClusterService clusterServiceRef;
+  private final org.opensearch.sql.common.setting.Settings pluginSettingsRef;
 
   @Inject
   public TransportPPLQueryAction(
@@ -82,11 +83,13 @@ public TransportPPLQueryAction(
 
     ModulesBuilder modules = new ModulesBuilder();
     modules.add(new OpenSearchPluginModule());
+    org.opensearch.sql.common.setting.Settings pluginSettings =
+        new OpenSearchSettings(clusterService.getClusterSettings());
+    this.pluginSettingsRef = pluginSettings;
     modules.add(
         b -> {
           b.bind(NodeClient.class).toInstance(client);
-          b.bind(org.opensearch.sql.common.setting.Settings.class)
-              .toInstance(new OpenSearchSettings(clusterService.getClusterSettings()));
+          b.bind(org.opensearch.sql.common.setting.Settings.class).toInstance(pluginSettings);
           b.bind(DataSourceService.class).toInstance(dataSourceService);
         });
     this.injector = Guice.createInjector(modules);
@@ -105,7 +108,8 @@ public void setQueryPlanExecutor(
       QueryPlanExecutor<RelNode, Iterable<Object[]>> queryPlanExecutor) {
     AnalyticsExecutorHolder.set(queryPlanExecutor);
     this.unifiedQueryHandler =
-        new RestUnifiedQueryAction(clientRef, clusterServiceRef, queryPlanExecutor);
+        new RestUnifiedQueryAction(
+            clientRef, clusterServiceRef, queryPlanExecutor, pluginSettingsRef);
   }
 
   /**