Planner initial impl

.github/workflows/codeql-analysis.yml

@@ -51,6 +51,9 @@ jobs:
         # Prefix the list here with "+" to use these queries and those in the config file.
         # queries: ./path/to/local/query, your-org/your-repo/queries@main
 
+    - name: Set up protoc
+      uses: arduino/setup-protoc@v3
+
     # Autobuild attempts to build any compiled languages  (C/C++, C#, or Java).
     # If this step fails, then you should remove it and run the build manually (see below)
     - name: Autobuild

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

@@ -8,6 +8,9 @@
 
 package org.opensearch.analytics.spi;
 
+import org.opensearch.analytics.backend.EngineResultStream;
+import org.opensearch.analytics.backend.ExecutionContext;
+import org.opensearch.analytics.backend.SearchExecEngine;
 import org.opensearch.index.engine.dataformat.DataFormat;
 
 import java.util.Collections;
@@ -17,9 +20,26 @@
 /**
  * SPI extension point for back-end query engines for query planning and execution capabilities
  * as needed by the {@link org.opensearch.analytics.exec.QueryPlanExecutor}
+ *
+ * <p>TODO: separate capability declaration (planner, coordinator) from execution engine factory
+ * (data node) into two interfaces. AnalyticsSearchBackendPlugin should only declare capabilities.
+ * SearchExecEngineProvider should be discovered separately by the executor. Remove the extends
+ * relationship and the default createSearchExecEngine() below once that separation is done.
  */
 public interface AnalyticsSearchBackendPlugin extends SearchExecEngineProvider {
 
+    /** Unique engine name (e.g., "lucene", "datafusion"). */
+    String name();
+
+    /**
+     * {@inheritDoc}
+     * Temporary default — remove once SearchExecEngineProvider is separated from this interface.
+     */
+    @Override
+    default SearchExecEngine<ExecutionContext, EngineResultStream> createSearchExecEngine(ExecutionContext ctx) {
+        throw new UnsupportedOperationException("createSearchExecEngine not implemented for " + name());
+    }
+
     /** Returns the data formats supported by this backend. */
     List<DataFormat> getSupportedFormats();
 
@@ -71,5 +91,4 @@ default Set<ShuffleCapability> supportedShuffleCapabilities() {
     default byte[] convertFragment(Object fragment) {
         throw new UnsupportedOperationException("convertFragment not yet implemented for " + name());
     }
-
 }

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

@@ -69,7 +69,6 @@ public Iterable<Object[]> execute(RelNode logicalFragment, Object context) {
             new PlannerContext(
                 new CapabilityRegistry(new ArrayList<>(backEnds.values())),
                 clusterService.state()));
-
         String tableName = extractTableName(logicalFragment);
         AnalyticsSearchBackendPlugin provider = selectBackEnd();
         if (provider == null) {

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

@@ -39,6 +39,21 @@ public class FieldStorageResolver {
 
     /**
      * Production: resolves per-field storage from IndexMetadata and backend capabilities.
+     *
+     * <p>LIMITATION: This infers storage from what each DataFormat DECLARES it can support
+     * (via FieldTypeCapabilities), not from what the index ACTUALLY stores. A format declaring
+     * COLUMNAR_STORAGE for "integer" doesn't mean this index has integer doc values in that format.
+     *
+     * <p>The indexing side has no per-field format metadata at the coordinator level today:
+     * - DataFormat.supportedFields() = capability, not actual storage
+     * - Segment.dfGroupedSearchableFiles = per-segment format info, data node only
+     * - DataFormatRegistry has TODOs to filter by index settings/mapper service
+     * - primary_data_format index setting is the only coordinator-level hint (index-level, not field-level)
+     *
+     * <p>TODO: Replace inference with actual per-field format metadata once the indexing team adds
+     * it to MappingMetadata or IndexMetadata. Until then, this over-estimates viable backends —
+     * the shard-level cost function must handle the mismatch. Consider using primary_data_format
+     * as a narrowing hint in the interim.
      */
     @SuppressWarnings("unchecked")
     public FieldStorageResolver(IndexMetadata indexMetadata, List<AnalyticsSearchBackendPlugin> backends) {

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

@@ -14,29 +14,35 @@
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.opensearch.analytics.planner.rel.OpenSearchDistributionTraitDef;
+import org.opensearch.analytics.planner.dag.DAGBuilder;
+import org.opensearch.analytics.planner.dag.QueryDAG;
 import org.opensearch.analytics.planner.rules.OpenSearchAggregateRule;
 import org.opensearch.analytics.planner.rules.OpenSearchAggregateSplitRule;
 import org.opensearch.analytics.planner.rules.OpenSearchFilterRule;
 import org.opensearch.analytics.planner.rules.OpenSearchProjectRule;
 import org.opensearch.analytics.planner.rules.OpenSearchSortRule;
 import org.opensearch.analytics.planner.rules.OpenSearchTableScanRule;
 
+import org.opensearch.analytics.planner.dag.DAGBuilder;
+import org.opensearch.analytics.planner.dag.QueryDAG;
+
 import java.util.List;
 
 /**
  * Central planner for the Analytics Plugin.
  *
- * <p>Two phases:
+ * <p>Three phases:
  * <ol>
  *   <li>HepPlanner (RBO): converts LogicalXxx → OpenSearchXxx with backend
  *       assignment, predicate annotation, and distribution traits.</li>
  *   <li>VolcanoPlanner (CBO): requests SINGLETON at root (coordinator must
  *       gather all results). Split rule fires on aggregates, Volcano inserts
  *       exchanges via trait enforcement where distribution mismatches.</li>
+ *   <li>DAG construction: cuts at exchange boundaries, builds stage tree.</li>
  * </ol>
  *
  * <p>TODO: eliminate copyToCluster — have frontends create RelNodes with Volcano cluster.
- * <p>TODO: DAG construction (cut at exchange boundaries)
+ * <p>TODO: Per-stage plan forking (multiple plan generation)
  * <p>TODO: Fragment conversion (backend.convertFragment)
  * <p>TODO: Join strategy selection, sort removal via CBO
  *
@@ -93,6 +99,11 @@ public static RelNode createPlan(RelNode rawRelNode, PlannerContext context) {
         RelNode result = volcanoPlanner.findBestExp();
 
         LOGGER.info("After CBO:\n{}", RelOptUtil.toString(result));
+
+        // Phase 3: DAG construction — cut at exchange boundaries
+        QueryDAG dag = DAGBuilder.build(result);
+        LOGGER.info("QueryDAG:\n{}", dag);
+
         return result;
     }
 }

sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/dag/DAGBuilder.java

@@ -0,0 +1,159 @@
+/*
+ * 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.planner.dag;
+
+import org.apache.calcite.rel.RelDistribution;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.type.RelDataType;
+import org.opensearch.analytics.planner.rel.OpenSearchExchangeReducer;
+import org.opensearch.analytics.planner.rel.OpenSearchExchangeWriter;
+import org.opensearch.analytics.planner.rel.OpenSearchShuffleReader;
+import org.opensearch.analytics.planner.rel.OpenSearchStageInputScan;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+
+/**
+ * Builds a {@link QueryDAG} from the CBO output by cutting at exchange boundaries.
+ *
+ * <p>SINGLETON: {@link OpenSearchExchangeReducer} is the boundary. Parent
+ * fragment is everything above the reducer (may be null for a pure gather).
+ * Child fragment is the reducer's input subtree.
+ *
+ * <p>HASH/RANGE: {@link OpenSearchShuffleReader} → {@link OpenSearchExchangeWriter}.
+ * ShuffleReader stays in parent as leaf (input severed). Writer + subtree
+ * below becomes the child fragment.
+ *
+ * <p>Stage IDs assigned bottom-up (leaf stages get lower IDs).
+ *
+ * @opensearch.internal
+ */
+public class DAGBuilder {
+
+    private DAGBuilder() {}
+
+    public static QueryDAG build(RelNode cboOutput) {
+        int[] counter = { 0 };
+        List<Stage> childStages = new ArrayList<>();
+
+        RelNode fragment;
+        if (cboOutput instanceof OpenSearchExchangeReducer reducer) {
+            // Root is an ExchangeReducer (e.g., shuffle case where coordinator
+            // is a pure gather). ExchangeReducer becomes the root stage fragment
+            // with its input severed. Child stage is the subtree below.
+            fragment = cutSingleton(reducer, counter, childStages);
+        } else {
+            fragment = sever(cboOutput, counter, childStages);
+        }
+
+        Stage rootStage = new Stage(counter[0]++, fragment, childStages, null);
+        return new QueryDAG(UUID.randomUUID().toString(), rootStage);
+    }
+
+    /**
+     * Walks top-down collecting operators into the current stage. When an
+     * exchange boundary is hit, severs the link: subtree below becomes a
+     * child stage, current stage continues with the boundary removed or
+     * replaced by a detached leaf.
+     *
+     * <p>Exchange boundaries are detected at the input level (not the node
+     * level) so the parent node never receives a null input. For SINGLETON
+     * cuts the input is simply dropped; for shuffle cuts the ShuffleReader
+     * replaces the input as a detached leaf.
+     *
+     * @param node        current node being visited
+     * @param counter     stage ID counter (bottom-up assignment)
+     * @param childStages accumulator for child stages found below exchanges
+     * @return the rewritten fragment root for the current stage
+     */
+    private static RelNode sever(RelNode node, int[] counter, List<Stage> childStages) {
+        // Check each input for exchange boundaries before recursing
+        List<RelNode> newInputs = new ArrayList<>();
+        for (RelNode input : node.getInputs()) {
+            if (input instanceof OpenSearchExchangeReducer reducer) {
+                // SINGLETON cut: ExchangeReducer stays in parent as leaf (input severed).
+                // Child stage is the reducer's input subtree. Analytics Core streams
+                // results from data nodes — no exchange operator needed in child fragment.
+                newInputs.add(cutSingleton(reducer, counter, childStages));
+            } else if (input instanceof OpenSearchShuffleReader reader) {
+                // Shuffle cut: ShuffleReader stays in parent as leaf (input severed).
+                // Child stage is ExchangeWriter + subtree below.
+                newInputs.add(cutShuffle(reader, counter, childStages));
+            } else {
+                newInputs.add(sever(input, counter, childStages));
+            }
+        }
+
+        // Leaf node (e.g., TableScan) — no inputs to process
+        if (node.getInputs().isEmpty()) {
+            return node;
+        }
+
+        // Rebuild only if inputs changed
+        boolean changed = false;
+        for (int idx = 0; idx < newInputs.size(); idx++) {
+            if (newInputs.get(idx) != node.getInputs().get(idx)) {
+                changed = true;
+                break;
+            }
+        }
+        return changed ? node.copy(node.getTraitSet(), newInputs) : node;
+    }
+
+    private static RelNode cutSingleton(OpenSearchExchangeReducer reducer,
+                                        int[] counter, List<Stage> parentChildStages) {
+        List<Stage> grandchildren = new ArrayList<>();
+        RelNode childFragment = sever(reducer.getInput(), counter, grandchildren);
+
+        int childStageId = counter[0]++;
+        parentChildStages.add(new Stage(
+            childStageId, childFragment, grandchildren,
+            new ExchangeInfo(RelDistribution.Type.SINGLETON, null, List.of())
+        ));
+
+        // Replace child subtree with StageInputScan, keep ExchangeReducer in parent
+        RelDataType childRowType = reducer.getInput().getRowType();
+        OpenSearchStageInputScan stageInput = new OpenSearchStageInputScan(
+            reducer.getCluster(), reducer.getTraitSet(), childStageId, childRowType
+        );
+        return new OpenSearchExchangeReducer(
+            reducer.getCluster(), reducer.getTraitSet(), stageInput, reducer.getViableBackends()
+        );
+    }
+
+    private static RelNode cutShuffle(OpenSearchShuffleReader reader,
+                                     int[] counter, List<Stage> parentChildStages) {
+        if (!(reader.getInput() instanceof OpenSearchExchangeWriter writer)) {
+            throw new IllegalStateException(
+                "ShuffleReader input must be ExchangeWriter, got: "
+                    + reader.getInput().getClass().getSimpleName());
+        }
+
+        List<Stage> grandchildren = new ArrayList<>();
+        RelNode belowWriter = sever(writer.getInput(), counter, grandchildren);
+        RelNode childFragment = writer.copy(writer.getTraitSet(), List.of(belowWriter));
+
+        int childStageId = counter[0]++;
+        parentChildStages.add(new Stage(
+            childStageId, childFragment, grandchildren,
+            new ExchangeInfo(RelDistribution.Type.HASH_DISTRIBUTED, writer.getShuffleImpl(), writer.getKeys())
+        ));
+
+        // Replace child subtree with StageInputScan, keep ShuffleReader in parent
+        RelDataType childRowType = writer.getInput().getRowType();
+        OpenSearchStageInputScan stageInput = new OpenSearchStageInputScan(
+            reader.getCluster(), reader.getTraitSet(), childStageId, childRowType
+        );
+        return new OpenSearchShuffleReader(
+            reader.getCluster(), reader.getTraitSet(), stageInput,
+            reader.getViableBackends(), reader.getShuffleImpl()
+        );
+    }
+}

sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/dag/ExchangeInfo.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.analytics.planner.dag;
+
+import org.apache.calcite.rel.RelDistribution;
+import org.opensearch.analytics.planner.rel.ShuffleImpl;
+
+import java.util.List;
+
+/**
+ * Exchange metadata extracted from exchange RelNodes during DAG construction.
+ * Describes how a child stage delivers data to its parent stage.
+ *
+ * @param distributionType distribution type from the exchange operator's trait
+ * @param shuffleImpl      shuffle implementation (null for SINGLETON)
+ * @param partitionKeyIndices field indices for hash/range partitioning (empty for SINGLETON)
+ * @opensearch.internal
+ */
+public record ExchangeInfo(
+    RelDistribution.Type distributionType,
+    ShuffleImpl shuffleImpl,
+    List<Integer> partitionKeyIndices
+) {
+    public ExchangeInfo {
+        partitionKeyIndices = List.copyOf(partitionKeyIndices);
+    }
+
+    public boolean isShuffle() {
+        return shuffleImpl != null;
+    }
+}

sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/dag/QueryDAG.java

@@ -0,0 +1,61 @@
+/*
+ * 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.planner.dag;
+
+import org.apache.calcite.plan.RelOptUtil;
+
+/**
+ * Root of the query execution DAG. Recursive tree of {@link Stage}s.
+ *
+ * @param queryId   unique identifier for this query. Currently a random UUID.
+ *                  Future: accept a user-provided ID or generate a cluster-wide
+ *                  unique ID (e.g., node-local counter + nodeId prefix).
+ * @param rootStage the coordinator/root stage
+ * @opensearch.internal
+ */
+public record QueryDAG(String queryId, Stage rootStage) {
+
+    /**
+     * Returns a human-readable representation of the entire DAG showing
+     * every stage with its fragment indented.
+     */
+    @Override
+    public String toString() {
+        StringBuilder builder = new StringBuilder();
+        builder.append("QueryDAG(queryId=").append(queryId).append(")\n");
+        appendStage(builder, rootStage, 0);
+        return builder.toString();
+    }
+
+    private static void appendStage(StringBuilder builder, Stage stage, int depth) {
+        String indent = "  ".repeat(depth);
+        builder.append(indent).append("Stage ").append(stage.getStageId());
+        if (stage.getExchangeInfo() != null) {
+            builder.append(" [exchange=").append(stage.getExchangeInfo()).append("]");
+        } else {
+            builder.append(" [root]");
+        }
+        builder.append("\n");
+
+        if (stage.getFragment() != null) {
+            String fragmentStr = RelOptUtil.toString(stage.getFragment());
+            for (String line : fragmentStr.split("\n")) {
+                if (!line.isEmpty()) {
+                    builder.append(indent).append("  ").append(line).append("\n");
+                }
+            }
+        } else {
+            builder.append(indent).append("  <gather>\n");
+        }
+
+        for (Stage child : stage.getChildStages()) {
+            appendStage(builder, child, depth + 1);
+        }
+    }
+}