Add AnalyticsFrontEndExtension SPI + producer-side wiring in AnalyticsPlugin

sandbox/libs/analytics-framework/src/main/java/org/opensearch/analytics/spi/AnalyticsFrontEndExtension.java

@@ -0,0 +1,42 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.analytics.spi;
+
+/**
+ * SPI for frontend plugins (e.g., opensearch-sql) that integrate with analytics-engine.
+ *
+ * <p>Implementers are discovered by {@code AnalyticsPlugin} via
+ * {@link org.opensearch.plugins.ExtensiblePlugin#loadExtensions}; analytics-engine pushes its
+ * services to each consumer once Guice has constructed them. Mirrors the
+ * {@code JobSchedulerExtension} pattern from opensearch-job-scheduler — the consumer plugin
+ * declares its capability via this interface; the publishing plugin (analytics-engine) handles
+ * discovery and lifecycle.
+ *
+ * <p>This SPI lets a frontend declare analytics-engine as an OPTIONAL extended plugin
+ * ({@code extendedPlugins = ['analytics-engine;optional=true']}). When analytics-engine is not
+ * installed, no consumer ever receives a callback; analytics-routing code paths stay inert and
+ * the frontend plugin boots normally.
+ *
+ * <p><b>Lifecycle.</b> {@link #setAnalyticsServices} is invoked exactly once per consumer per node
+ * lifecycle, AFTER the node Guice injector is built (i.e., after every plugin's
+ * {@code createComponents} returns) and BEFORE the first analytics query is dispatched.
+ * Implementations should stash the bundle for later use; do not assume the services are available
+ * during {@code createComponents}.
+ *
+ * @opensearch.internal
+ */
+public interface AnalyticsFrontEndExtension {
+
+    /**
+     * Receives the bundle of analytics-engine services. Called exactly once after the services are
+     * constructed and before any analytics query is dispatched. Each service inside the bundle is
+     * safe to invoke from any thread once received.
+     */
+    void setAnalyticsServices(AnalyticsServices services);
+}

sandbox/libs/analytics-framework/src/main/java/org/opensearch/analytics/spi/AnalyticsServices.java

@@ -0,0 +1,29 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.analytics.spi;
+
+import org.apache.calcite.rel.RelNode;
+import org.opensearch.analytics.exec.QueryPlanExecutor;
+import org.opensearch.analytics.schema.SchemaProvider;
+
+/**
+ * Bundle of services that {@code AnalyticsPlugin} pushes to each {@link AnalyticsFrontEndExtension}
+ * consumer once Guice has constructed them.
+ *
+ * <p>Bundled rather than passed through individual setters so future analytics-engine services can
+ * be added without changing the {@link AnalyticsFrontEndExtension} signature — frontends that do
+ * not consume the new service simply ignore the new accessor.
+ *
+ * @param queryPlanExecutor coordinator-level query plan executor
+ * @param schemaProvider    builds a Calcite {@code SchemaPlus} from the current cluster state
+ *
+ * @opensearch.internal
+ */
+public record AnalyticsServices(QueryPlanExecutor<RelNode, Iterable<Object[]>> queryPlanExecutor, SchemaProvider schemaProvider) {
+}

sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/AnalyticsPlugin.java

@@ -24,11 +24,19 @@
 import org.opensearch.analytics.planner.CapabilityRegistry;
 import org.opensearch.analytics.planner.FieldStorageResolver;
 import org.opensearch.analytics.schema.OpenSearchSchemaBuilder;
+import org.opensearch.analytics.schema.SchemaProvider;
+import org.opensearch.analytics.spi.AnalyticsFrontEndExtension;
 import org.opensearch.analytics.spi.AnalyticsSearchBackendPlugin;
+import org.opensearch.analytics.spi.AnalyticsServices;
+import org.opensearch.cluster.ClusterState;
 import org.opensearch.cluster.metadata.IndexNameExpressionResolver;
 import org.opensearch.cluster.service.ClusterService;
 import org.opensearch.common.inject.Module;
 import org.opensearch.common.inject.TypeLiteral;
+import org.opensearch.common.inject.matcher.Matchers;
+import org.opensearch.common.inject.spi.InjectionListener;
+import org.opensearch.common.inject.spi.TypeEncounter;
+import org.opensearch.common.inject.spi.TypeListener;
 import org.opensearch.core.action.ActionResponse;
 import org.opensearch.core.common.io.stream.NamedWriteableRegistry;
 import org.opensearch.core.xcontent.NamedXContentRegistry;
@@ -66,12 +74,14 @@ public class AnalyticsPlugin extends Plugin implements ExtensiblePlugin, ActionP
     public AnalyticsPlugin() {}
 
     private final List<AnalyticsSearchBackendPlugin> backEnds = new ArrayList<>();
+    private final List<AnalyticsFrontEndExtension> frontEnds = new ArrayList<>();
     private SqlOperatorTable operatorTable;
 
     @SuppressWarnings("rawtypes")
     @Override
     public void loadExtensions(ExtensionLoader loader) {
         backEnds.addAll(loader.loadExtensions(AnalyticsSearchBackendPlugin.class));
+        frontEnds.addAll(loader.loadExtensions(AnalyticsFrontEndExtension.class));
     }
 
     @Override
@@ -112,9 +122,40 @@ public Collection<Module> createGuiceModules() {
             }).to(DefaultPlanExecutor.class);
             b.bind(EngineContext.class).to(DefaultEngineContext.class);
             b.bind(Scheduler.class).to(QueryScheduler.class);
+            // Push the executor + schemaProvider bundle to every registered AnalyticsFrontEndExtension
+            // once Guice constructs DefaultPlanExecutor. The InjectionListener fires on the singleton
+            // construction; pushAnalyticsServices guards against re-firing if Guice ever instantiates
+            // more than once.
+            b.bindListener(Matchers.any(), new TypeListener() {
+                @Override
+                public <I> void hear(TypeLiteral<I> type, TypeEncounter<I> encounter) {
+                    if (!DefaultPlanExecutor.class.isAssignableFrom(type.getRawType())) {
+                        return;
+                    }
+                    encounter.register((InjectionListener<I>) instance -> pushAnalyticsServices((DefaultPlanExecutor) instance));
+                }
+            });
         });
     }
 
+    private boolean servicesPushed = false;
+
+    private synchronized void pushAnalyticsServices(DefaultPlanExecutor executor) {
+        if (servicesPushed) {
+            return;
+        }
+        servicesPushed = true;
+        SchemaProvider schemaProvider = clusterState -> OpenSearchSchemaBuilder.buildSchema((ClusterState) clusterState);
+        AnalyticsServices services = new AnalyticsServices(executor, schemaProvider);
+        for (AnalyticsFrontEndExtension consumer : frontEnds) {
+            try {
+                consumer.setAnalyticsServices(services);
+            } catch (Exception e) {
+                logger.warn("AnalyticsFrontEndExtension {} threw on setAnalyticsServices", consumer.getClass().getName(), e);
+            }
+        }
+    }
+
     @Override
     public List<ActionHandler<? extends ActionRequest, ? extends ActionResponse>> getActions() {
         return List.of(new ActionHandler<>(AnalyticsQueryAction.INSTANCE, DefaultPlanExecutor.class));