dsl-query-executor: route _search through DslCalciteGrammar with codec fallback

sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/DslQueryExecutorPlugin.java

@@ -12,12 +12,16 @@
 import org.opensearch.action.support.ActionFilter;
 import org.opensearch.cluster.metadata.IndexNameExpressionResolver;
 import org.opensearch.cluster.service.ClusterService;
+import org.opensearch.common.settings.Setting;
 import org.opensearch.core.action.ActionResponse;
 import org.opensearch.core.common.io.stream.NamedWriteableRegistry;
 import org.opensearch.core.xcontent.NamedXContentRegistry;
 import org.opensearch.dsl.action.DslExecuteAction;
 import org.opensearch.dsl.action.SearchActionFilter;
 import org.opensearch.dsl.action.TransportDslExecuteAction;
+import org.opensearch.dsl.aggregation.AggregationRegistryFactory;
+import org.opensearch.dsl.query.QueryRegistryFactory;
+import org.opensearch.dsl.router.DslCalciteGrammar;
 import org.opensearch.env.Environment;
 import org.opensearch.env.NodeEnvironment;
 import org.opensearch.plugins.ActionPlugin;
@@ -59,7 +63,8 @@ public Collection<Object> createComponents(
         IndexNameExpressionResolver indexNameExpressionResolver,
         Supplier<RepositoriesService> repositoriesServiceSupplier
     ) {
-        this.searchActionFilter = new SearchActionFilter((NodeClient) client);
+        DslCalciteGrammar grammar = new DslCalciteGrammar(QueryRegistryFactory.create(), AggregationRegistryFactory.create());
+        this.searchActionFilter = new SearchActionFilter((NodeClient) client, clusterService, grammar);
         return Collections.emptyList();
     }
 
@@ -72,4 +77,9 @@ public Collection<Object> createComponents(
     public List<ActionFilter> getActionFilters() {
         return searchActionFilter != null ? List.of(searchActionFilter) : List.of();
     }
+
+    @Override
+    public List<Setting<?>> getSettings() {
+        return List.of(DslQueryExecutorSettings.CALCITE_ENABLED);
+    }
 }

sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/DslQueryExecutorSettings.java

@@ -0,0 +1,37 @@
+/*
+ * 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.dsl;
+
+import org.opensearch.common.settings.Setting;
+
+/**
+ * Cluster-scoped settings owned by the DSL query executor plugin. Registered via
+ * {@link DslQueryExecutorPlugin#getSettings()}.
+ */
+public final class DslQueryExecutorSettings {
+
+    /**
+     * Master switch for Calcite-path routing. Defaults to {@code true} — the plugin routes
+     * {@code _search} through the grammar, with codec fallback on grammar rejection or
+     * conversion failure. Set to {@code false} to force every request through the codec
+     * path unchanged (operational escape hatch: mitigations, benchmarking, incident
+     * response).
+     *
+     * <p>Node-scoped + dynamic → cluster-wide, updatable via {@code PUT _cluster/settings}
+     * (use {@code persistent} to survive a full cluster restart).
+     */
+    public static final Setting<Boolean> CALCITE_ENABLED = Setting.boolSetting(
+        "dsl.query_executor.calcite.enabled",
+        true,
+        Setting.Property.NodeScope,
+        Setting.Property.Dynamic
+    );
+
+    private DslQueryExecutorSettings() {}
+}

sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/action/SearchActionFilter.java

@@ -8,36 +8,75 @@
 
 package org.opensearch.dsl.action;
 
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 import org.opensearch.action.ActionRequest;
 import org.opensearch.action.search.SearchAction;
 import org.opensearch.action.search.SearchRequest;
 import org.opensearch.action.search.SearchResponse;
 import org.opensearch.action.support.ActionFilter;
 import org.opensearch.action.support.ActionFilterChain;
 import org.opensearch.action.support.ActionRequestMetadata;
+import org.opensearch.cluster.service.ClusterService;
 import org.opensearch.core.action.ActionListener;
 import org.opensearch.core.action.ActionResponse;
+import org.opensearch.dsl.DslQueryExecutorSettings;
+import org.opensearch.dsl.converter.ConversionException;
+import org.opensearch.dsl.router.DslCalciteGrammar;
+import org.opensearch.dsl.router.RouteDecision;
 import org.opensearch.tasks.Task;
 import org.opensearch.transport.client.node.NodeClient;
 
+import java.util.Collections;
+import java.util.IdentityHashMap;
+import java.util.Set;
+
 /**
- * Intercepts all {@code _search} requests and dispatches them to {@link DslExecuteAction}
- * for execution through the Calcite pipeline. Non-search actions pass through unchanged.
+ * Intercepts {@code _search} requests and, when Calcite routing is enabled, decides between
+ * the Calcite path and the codec path.
+ *
+ * <p>Layer 1: {@link DslQueryExecutorSettings#CALCITE_ENABLED} — defaults to {@code true};
+ * setting it to {@code false} forces every request through the codec path unchanged
+ * (operational escape hatch).
+ *
+ * <p>Layer 2 (only when the setting is on):
+ * <ul>
+ *   <li>{@link DslCalciteGrammar#validate} classifies the request. Grammar-rejected requests
+ *       fall back to the codec path.</li>
+ *   <li>Grammar-accepted requests are dispatched to {@link DslExecuteAction}. A
+ *       {@link ConversionException} during execution triggers codec fallback (schema-side
+ *       checks the grammar can't run may still fail at conversion time). Non-conversion
+ *       errors surface to the caller unchanged.</li>
+ * </ul>
  */
 public class SearchActionFilter implements ActionFilter {
 
+    private static final Logger logger = LogManager.getLogger(SearchActionFilter.class);
+
     /** Runs after the Security plugin's authorization filter (order 0). */
     static final int FILTER_ORDER = 1;
 
     private final NodeClient client;
+    private final DslCalciteGrammar grammar;
+    /**
+     * Kept in sync with {@link DslQueryExecutorSettings#CALCITE_ENABLED} via a
+     * cluster-settings update consumer, so a {@code PUT _cluster/settings} change propagates
+     * without any per-request settings lookup. {@code volatile} because updates land on the
+     * cluster-state-applier thread while reads happen on transport/search threads.
+     */
+    private volatile boolean calciteEnabled;
 
     /**
-     * Creates a filter that dispatches intercepted searches via the given client.
-     *
      * @param client node client for dispatching to {@link DslExecuteAction}
+     * @param clusterService source of the dynamic {@code CALCITE_ENABLED} setting value
+     * @param grammar route-decision oracle
      */
-    public SearchActionFilter(NodeClient client) {
+    public SearchActionFilter(NodeClient client, ClusterService clusterService, DslCalciteGrammar grammar) {
         this.client = client;
+        this.grammar = grammar;
+        this.calciteEnabled = DslQueryExecutorSettings.CALCITE_ENABLED.get(clusterService.getSettings());
+        clusterService.getClusterSettings()
+            .addSettingsUpdateConsumer(DslQueryExecutorSettings.CALCITE_ENABLED, v -> this.calciteEnabled = v);
     }
 
     @Override
@@ -56,12 +95,47 @@ public <Request extends ActionRequest, Response extends ActionResponse> void app
         ActionFilterChain<Request, Response> chain
     ) {
         // TODO: add support for other search-related APIs (_msearch, _count, _search_shards, etc.).
-        // Consider two categories: APIs that execute search vs APIs that only explain/validate.
-        if (SearchAction.NAME.equals(action)) {
-            SearchRequest searchRequest = (SearchRequest) request;
-            client.execute(DslExecuteAction.INSTANCE, searchRequest, (ActionListener<SearchResponse>) listener);
-        } else {
+        if (!calciteEnabled || !SearchAction.NAME.equals(action)) {
+            chain.proceed(task, action, request, listener);
+            return;
+        }
+
+        SearchRequest searchRequest = (SearchRequest) request;
+        RouteDecision decision = grammar.validate(searchRequest.source());
+
+        if (!decision.supported()) {
+            logger.debug("Grammar rejected _search, falling back to codec: {}", decision.unsupportedFeatures());
             chain.proceed(task, action, request, listener);
+            return;
+        }
+
+        ActionListener<SearchResponse> calciteListener = ActionListener.wrap(
+            (SearchResponse resp) -> ((ActionListener<SearchResponse>) listener).onResponse(resp),
+            error -> {
+                if (isFallbackable(error)) {
+                    logger.debug("Calcite path threw ConversionException, falling back to codec", error);
+                    chain.proceed(task, action, request, listener);
+                } else {
+                    listener.onFailure(error);
+                }
+            }
+        );
+        client.execute(DslExecuteAction.INSTANCE, searchRequest, calciteListener);
+    }
+
+    /**
+     * Only fall back on failures that mean "Calcite can't handle this request" — grammar
+     * gaps or schema mismatches the grammar can't see. Runtime engine errors, timeouts,
+     * and other operational failures must surface to the caller so they can be diagnosed.
+     *
+     * <p>Uses an identity-based seen set to defend against pathological self-referential
+     * or cyclic exception chains (e.g. legacy code that calls {@code initCause(this)}).
+     */
+    private static boolean isFallbackable(Throwable e) {
+        Set<Throwable> seen = Collections.newSetFromMap(new IdentityHashMap<>());
+        for (Throwable t = e; t != null && seen.add(t); t = t.getCause()) {
+            if (t instanceof ConversionException) return true;
         }
+        return false;
     }
 }

sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/aggregation/AggregationRegistry.java

@@ -36,6 +36,17 @@ public void register(AggregationTranslator<?> translator) {
         translators.put(translator.getAggregationType(), translator);
     }
 
+    /**
+     * Returns {@code true} when a translator is registered for the given aggregation type. Used
+     * by routing (grammar) to decide whether a request can be sent to the Calcite path before
+     * any conversion work is attempted.
+     *
+     * @param type the concrete {@link AggregationBuilder} class
+     */
+    public boolean hasTranslator(Class<? extends AggregationBuilder> type) {
+        return translators.containsKey(type);
+    }
+
     /**
      * Returns the translator for the given class, or null.
      * Caller checks {@code instanceof MetricTranslator} or {@code instanceof BucketTranslator}.

sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/aggregation/AggregationRegistryFactory.java

@@ -15,14 +15,19 @@
 import org.opensearch.dsl.aggregation.metric.SumMetricTranslator;
 
 /**
- * Creates an {@link AggregationRegistry} populated with all supported translators.
+ * Returns the process-wide {@link AggregationRegistry} populated with all supported
+ * metric and bucket translators. Registrations are effectively immutable after class
+ * init, and the registry is safe to share across threads (concurrent reads, no writes
+ * at steady state).
  */
 public class AggregationRegistryFactory {
 
+    /** Built once at class init and cached forever. */
+    private static final AggregationRegistry INSTANCE = build();
+
     private AggregationRegistryFactory() {}
 
-    /** Creates a registry with all supported metric and bucket translators. */
-    public static AggregationRegistry create() {
+    private static AggregationRegistry build() {
         AggregationRegistry registry = new AggregationRegistry();
         registry.register(new AvgMetricTranslator());
         registry.register(new SumMetricTranslator());
@@ -32,4 +37,9 @@ public static AggregationRegistry create() {
         // TODO: add other aggregation translators
         return registry;
     }
+
+    /** Returns the shared registry. All callers see the same instance. */
+    public static AggregationRegistry create() {
+        return INSTANCE;
+    }
 }

sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/query/QueryRegistry.java

@@ -37,6 +37,17 @@ public void register(QueryTranslator translator) {
         translators.put(translator.getQueryType(), translator);
     }
 
+    /**
+     * Returns {@code true} when a translator is registered for the given query type. Used by
+     * routing (grammar) to decide whether a request can be sent to the Calcite path before any
+     * conversion work is attempted.
+     *
+     * @param type the concrete {@link QueryBuilder} class
+     */
+    public boolean hasTranslator(Class<? extends QueryBuilder> type) {
+        return translators.containsKey(type);
+    }
+
     /**
      * Converts a query using the registered translator for its type.
      * If no translator is registered, wraps the query in an {@link UnresolvedQueryCall}

sandbox/plugins/dsl-query-executor/src/main/java/org/opensearch/dsl/query/QueryRegistryFactory.java

@@ -9,14 +9,21 @@
 package org.opensearch.dsl.query;
 
 /**
- * Creates a {@link QueryRegistry} populated with all supported query translators.
+ * Returns the process-wide {@link QueryRegistry} populated with all supported query
+ * translators. Registrations are effectively immutable after class init, and the registry
+ * is safe to share across threads (concurrent reads, no writes at steady state).
  */
 public class QueryRegistryFactory {
 
+    /**
+     * Built once at class init. The registry is populated in a private helper (not inline
+     * so we can keep the {@code register(...)} calls readable) and cached forever.
+     */
+    private static final QueryRegistry INSTANCE = build();
+
     private QueryRegistryFactory() {}
 
-    /** Creates a registry with all supported query translators. */
-    public static QueryRegistry create() {
+    private static QueryRegistry build() {
         QueryRegistry registry = new QueryRegistry();
         registry.register(new TermQueryTranslator());
         registry.register(new TermsQueryTranslator());
@@ -25,4 +32,9 @@ public static QueryRegistry create() {
         // TODO: add other query translators
         return registry;
     }
+
+    /** Returns the shared registry. All callers see the same instance. */
+    public static QueryRegistry create() {
+        return INSTANCE;
+    }
 }