Wires EngineContextProvider (analytics-engine API) into the SQL plugin for schema resolution (#5486)

* Wires `EngineContextProvider` (analytics-engine API) into the SQL plugin so PPL/SQL queries routed to the analytics engine bind to a single per-query `QueryRequestContext` (ClusterState + Calcite schema)
  captured once at REST entry. The same snapshot drives both Calcite planning and analytics-engine execution, eliminating drift between schema resolution at parse time and state lookups at execution time.

  - `RestUnifiedQueryAction` calls `contextProvider.getContext()` per query; the result feeds `UnifiedQueryContext.catalog(...)` and `AnalyticsExecutionEngine.execute(..., QueryRequestContext, listener)`.
  - `EngineContextProviderHolder` — small static bridge from Guice (where the binding lives) to the REST handler (constructed before cross-plugin Guice bindings are satisfied). Mirrors the existing
  `AnalyticsExecutorHolder` pattern.
  - `TransportPPLQueryAction` picks up the provider via `@Inject(optional = true) setEngineContextProvider` and populates the holder. `SQLPlugin`'s router waits for both executor + provider to be available.
  - `isAnalyticsIndex` short-circuits on `cluster.pluggable.dataformat="composite"` so aliases, wildcards, comma-lists, and data streams (which `Metadata#index()` can't resolve) route to the analytics engine
  without a per-index lookup.

  Pairs with the OpenSearch-side PR that introduces `EngineContextProvider` / `QueryRequestContext` and threads the snapshot through `DefaultPlanExecutor`.

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

* spotless

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

---------

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

Author: mch2

core/src/main/java/org/opensearch/sql/executor/analytics/AnalyticsExecutionEngine.java

@@ -77,6 +77,20 @@ public void explain(PhysicalPlan plan, ResponseListener<ExplainResponse> listene
   @Override
   public void execute(
       RelNode plan, CalcitePlanContext context, ResponseListener<QueryResponse> listener) {
+    execute(plan, context, null, listener);
+  }
+
+  /**
+   * Overload that threads a {@link org.opensearch.analytics.QueryRequestContext} snapshot down to
+   * the analytics-engine's plan executor via {@link
+   * org.opensearch.analytics.exec.QueryPlanExecutor}'s opaque context slot. The snapshot is
+   * captured once at query entry so planner and executor see the same cluster state.
+   */
+  public void execute(
+      RelNode plan,
+      CalcitePlanContext context,
+      org.opensearch.analytics.QueryRequestContext queryCtx,
+      ResponseListener<QueryResponse> listener) {
     // QueryPlanExecutor became asynchronous in analytics-framework 3.7 — execution is dispatched
     // to a worker pool and results arrive on the listener. Record the execute metric in the
     // listener callback, before delegating to the user-supplied listener, so the metric snapshot
@@ -86,7 +100,7 @@ public void execute(
 
     planExecutor.execute(
         plan,
-        null,
+        queryCtx,
         new ActionListener<>() {
           @Override
           public void onResponse(Iterable<Object[]> rows) {

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

@@ -225,11 +225,13 @@ private BiFunction<SQLQueryRequest, RestChannel, Boolean> createSqlAnalyticsRout
         () -> {
           if (cached[0] == null) {
             var executor = AnalyticsExecutorHolder.get();
-            if (executor == null) {
+            var contextProvider = org.opensearch.sql.plugin.rest.EngineContextProviderHolder.get();
+            if (executor == null || contextProvider == null) {
               return null;
             }
             cached[0] =
-                new RestUnifiedQueryAction(client, clusterService, executor, pluginSettings);
+                new RestUnifiedQueryAction(
+                    client, clusterService, executor, contextProvider, pluginSettings);
           }
           return cached[0];
         };

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

@@ -0,0 +1,34 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.sql.plugin.rest;
+
+import org.opensearch.analytics.EngineContextProvider;
+
+/**
+ * Bridge for sharing the analytics-engine {@link EngineContextProvider} singleton between the PPL
+ * transport action (where Guice resolves the binding via {@code @Inject}) and the REST-only SQL
+ * router (where Guice cannot, because {@code SQLPlugin#getRestHandlers} runs before the Node-level
+ * injector satisfies {@code @Inject} parameters).
+ *
+ * <p>Mirrors {@link AnalyticsExecutorHolder} — same cross-plugin Guice gap, same solution. The
+ * holder is populated once by {@link org.opensearch.sql.plugin.transport.TransportPPLQueryAction}'s
+ * ctor and then read by the SQL router. The context itself is intended to be a stable singleton
+ * produced by AnalyticsPlugin#createComponents.
+ */
+public final class EngineContextProviderHolder {
+
+  private static volatile EngineContextProvider context;
+
+  private EngineContextProviderHolder() {}
+
+  public static void set(EngineContextProvider instance) {
+    context = instance;
+  }
+
+  public static EngineContextProvider get() {
+    return context;
+  }
+}

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

@@ -17,11 +17,11 @@
 import org.apache.logging.log4j.Logger;
 import org.apache.logging.log4j.ThreadContext;
 import org.opensearch.analytics.exec.QueryPlanExecutor;
-import org.opensearch.analytics.schema.OpenSearchSchemaBuilder;
 import org.opensearch.cluster.service.ClusterService;
 import org.opensearch.common.unit.TimeValue;
 import org.opensearch.core.action.ActionListener;
 import org.opensearch.index.IndexSettings;
+import org.opensearch.indices.IndicesService;
 import org.opensearch.sql.api.UnifiedQueryContext;
 import org.opensearch.sql.api.UnifiedQueryPlanner;
 import org.opensearch.sql.ast.AbstractNodeVisitor;
@@ -54,16 +54,19 @@ public class RestUnifiedQueryAction {
   private final AnalyticsExecutionEngine analyticsEngine;
   private final NodeClient client;
   private final ClusterService clusterService;
+  private final org.opensearch.analytics.EngineContextProvider contextProvider;
   private final org.opensearch.sql.common.setting.Settings pluginSettings;
 
   public RestUnifiedQueryAction(
       NodeClient client,
       ClusterService clusterService,
       QueryPlanExecutor<RelNode, Iterable<Object[]>> planExecutor,
+      org.opensearch.analytics.EngineContextProvider contextProvider,
       org.opensearch.sql.common.setting.Settings pluginSettings) {
     this.client = client;
     this.clusterService = clusterService;
     this.analyticsEngine = new AnalyticsExecutionEngine(planExecutor);
+    this.contextProvider = contextProvider;
     this.pluginSettings = pluginSettings;
   }
 
@@ -82,6 +85,17 @@ public boolean isAnalyticsIndex(String query, QueryType queryType) {
     if (query == null || query.isEmpty()) {
       return false;
     }
+    // Cluster-level opt-in: when `cluster.pluggable.dataformat="composite"`, new indices
+    // inherit `index.pluggable.dataformat="composite"` at creation (see
+    // MetadataCreateIndexService), so every queryable target is analytics-eligible. Skip
+    // the per-index lookup — it doesn't work for aliases, wildcards, comma-lists, or data
+    // streams (Metadata#index() only resolves concrete names).
+    if ("composite"
+        .equals(
+            IndicesService.CLUSTER_PLUGGABLE_DATAFORMAT_VALUE_SETTING.get(
+                clusterService.getSettings()))) {
+      return true;
+    }
     try (UnifiedQueryContext context = buildParsingContext(queryType)) {
       return extractIndexName(query, queryType, context)
           .map(this::stripSchemaPrefix)
@@ -118,7 +132,12 @@ public void execute(
         .schedule(
             withCurrentContext(
                 () -> {
-                  UnifiedQueryContext context = buildContext(queryType, profiling);
+                  // Ask the engine for a per-query context — it binds the snapshot
+                  // (cluster state + schema built from it) and returns the pair, so the
+                  // schema we plan against and the state the executor uses are the same view.
+                  org.opensearch.analytics.QueryRequestContext queryCtx =
+                      contextProvider.getContext();
+                  UnifiedQueryContext context = buildContext(queryType, profiling, queryCtx);
                   ActionListener<TransportPPLQueryResponse> closingListener =
                       wrapWithContextClose(context, listener);
                   try {
@@ -127,7 +146,10 @@ public void execute(
                     CalcitePlanContext planContext = context.getPlanContext();
                     plan = addQuerySizeLimit(plan, planContext);
                     analyticsEngine.execute(
-                        plan, planContext, createQueryListener(queryType, closingListener));
+                        plan,
+                        planContext,
+                        queryCtx,
+                        createQueryListener(queryType, closingListener));
                   } catch (Exception e) {
                     closingListener.onFailure(e);
                   }
@@ -150,7 +172,9 @@ public void explain(
         .schedule(
             withCurrentContext(
                 () -> {
-                  try (UnifiedQueryContext context = buildContext(queryType, false)) {
+                  org.opensearch.analytics.QueryRequestContext queryCtx =
+                      contextProvider.getContext();
+                  try (UnifiedQueryContext context = buildContext(queryType, false, queryCtx)) {
                     UnifiedQueryPlanner planner = new UnifiedQueryPlanner(context);
                     RelNode plan = planner.plan(query);
                     CalcitePlanContext planContext = context.getPlanContext();
@@ -172,11 +196,15 @@ private UnifiedQueryContext buildParsingContext(QueryType queryType) {
     return applyClusterOverrides(UnifiedQueryContext.builder().language(queryType)).build();
   }
 
-  private UnifiedQueryContext buildContext(QueryType queryType, boolean profiling) {
+  private UnifiedQueryContext buildContext(
+      QueryType queryType,
+      boolean profiling,
+      org.opensearch.analytics.QueryRequestContext queryCtx) {
     return applyClusterOverrides(
             UnifiedQueryContext.builder()
                 .language(queryType)
-                .catalog(SCHEMA_NAME, OpenSearchSchemaBuilder.buildSchema(clusterService.state()))
+                // Schema captured by queryCtx — same cluster state the executor will use.
+                .catalog(SCHEMA_NAME, queryCtx.schema())
                 .defaultNamespace(SCHEMA_NAME)
                 .profiling(profiling))
         .build();

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

@@ -109,9 +109,27 @@ public TransportPPLQueryAction(
   public void setQueryPlanExecutor(
       QueryPlanExecutor<RelNode, Iterable<Object[]>> queryPlanExecutor) {
     AnalyticsExecutorHolder.set(queryPlanExecutor);
-    this.unifiedQueryHandler =
-        new RestUnifiedQueryAction(
-            clientRef, clusterServiceRef, queryPlanExecutor, pluginSettingsRef);
+    // Build the SQL router once both bridges are populated (engine context might arrive
+    // first or last depending on Guice ordering). buildUnifiedQueryHandler is idempotent.
+    buildUnifiedQueryHandlerIfReady();
+  }
+
+  /** Invoked by Guice iff analytics-engine bound {@code EngineContextProvider}. */
+  @Inject(optional = true)
+  public void setEngineContext(org.opensearch.analytics.EngineContextProvider contextProvider) {
+    org.opensearch.sql.plugin.rest.EngineContextProviderHolder.set(contextProvider);
+    buildUnifiedQueryHandlerIfReady();
+  }
+
+  private void buildUnifiedQueryHandlerIfReady() {
+    QueryPlanExecutor<RelNode, Iterable<Object[]>> executor = AnalyticsExecutorHolder.get();
+    org.opensearch.analytics.EngineContextProvider contextProvider =
+        org.opensearch.sql.plugin.rest.EngineContextProviderHolder.get();
+    if (executor != null && contextProvider != null) {
+      this.unifiedQueryHandler =
+          new RestUnifiedQueryAction(
+              clientRef, clusterServiceRef, executor, contextProvider, pluginSettingsRef);
+    }
   }
 
   /**

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

@@ -13,6 +13,7 @@
 import org.apache.calcite.rel.RelNode;
 import org.junit.Before;
 import org.junit.Test;
+import org.opensearch.analytics.EngineContextProvider;
 import org.opensearch.analytics.exec.QueryPlanExecutor;
 import org.opensearch.cluster.ClusterState;
 import org.opensearch.cluster.metadata.IndexMetadata;
@@ -40,6 +41,9 @@ public void setUp() {
     metadata = mock(Metadata.class);
     when(clusterService.state()).thenReturn(clusterState);
     when(clusterState.metadata()).thenReturn(metadata);
+    // isAnalyticsIndex short-circuits on the cluster.pluggable.dataformat setting; the per-index
+    // path is only exercised when this returns something other than "composite".
+    when(clusterService.getSettings()).thenReturn(Settings.EMPTY);
 
     @SuppressWarnings("unchecked")
     QueryPlanExecutor<RelNode, Iterable<Object[]>> executor = mock(QueryPlanExecutor.class);
@@ -48,6 +52,7 @@ public void setUp() {
             mock(NodeClient.class),
             clusterService,
             executor,
+            mock(EngineContextProvider.class),
             mock(org.opensearch.sql.common.setting.Settings.class));
   }