Simplify the source sink interactions

sandbox/libs/analytics-framework/src/main/java/org/opensearch/analytics/backend/ExchangeSink.java

@@ -11,8 +11,14 @@
 import org.apache.arrow.vector.VectorSchemaRoot;
 
 /**
- * Per-stage result accumulator that collects Arrow {@link VectorSchemaRoot} batches
- * from shard executions and provides access to the accumulated result set.
+ * Write-only interface for feeding Arrow batches into a stage exchange.
+ * Producers (shard scan stages, local compute stages) call {@link #feed}
+ * to push data; they never read from the sink.
+ *
+ * <p>Implementations must be thread-safe — multiple shard response
+ * handlers may call {@link #feed} concurrently.
+ *
+ * @see ExchangeSource for the read-side counterpart
  */
 public interface ExchangeSink {
 
@@ -23,26 +29,7 @@ public interface ExchangeSink {
     void feed(VectorSchemaRoot batch);
 
     /**
-     * Signal that no more responses will be fed.
+     * Signal that no more batches will be fed. Releases resources.
      */
     void close();
-
-    /**
-     * Return all accumulated rows in insertion order.
-     */
-    Iterable<Object[]> readResult();
-
-    /**
-     * Return the total number of accumulated rows.
-     */
-    long getRowCount();
-
-    /**
-     * Look up a cell value by column name and row index.
-     *
-     * @param column   the column name
-     * @param rowIndex the zero-based row index
-     * @return the cell value, or {@code null} if the column is unknown or the row index is out of range
-     */
-    Object getValueAt(String column, int rowIndex);
 }

sandbox/libs/analytics-framework/src/main/java/org/opensearch/analytics/backend/ExchangeSource.java

@@ -0,0 +1,30 @@
+/*
+ * 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.backend;
+
+/**
+ * Read-only interface for consuming accumulated results from a stage exchange.
+ * Consumers (parent stages, the walker's completion listener) read from this;
+ * they never write to it.
+ *
+ * @see ExchangeSink for the write-side counterpart
+ */
+public interface ExchangeSource {
+
+    /**
+     * Return all accumulated rows in insertion order.
+     * Converts columnar Arrow batches to row-oriented {@code Object[]} arrays.
+     */
+    Iterable<Object[]> readResult();
+
+    /**
+     * Return the total number of accumulated rows across all batches.
+     */
+    long getRowCount();
+}

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

@@ -0,0 +1,102 @@
+/*
+ * 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.exec;
+
+import org.opensearch.analytics.exec.stage.StageExecution;
+import org.opensearch.analytics.exec.stage.StageMetrics;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Inspectable snapshot of a fully-wired execution graph. Built by
+ * {@link PlanWalker#build()}, consumed by {@link PlanWalker#start()}
+ * or by EXPLAIN without executing.
+ *
+ * <p>The graph holds all {@link StageExecution} instances with their
+ * state listeners already wired. No stage has been started yet — all
+ * are in {@link StageExecution.State#CREATED}.
+ *
+ * @opensearch.internal
+ */
+public class ExecutionGraph {
+
+    private final Map<Integer, StageExecution> executions;
+    private final StageExecution rootExecution;
+    private final List<StageExecution> leaves;
+    private final String queryId;
+
+    ExecutionGraph(
+        String queryId,
+        Map<Integer, StageExecution> executions,
+        StageExecution rootExecution,
+        List<StageExecution> leaves
+    ) {
+        this.queryId = queryId;
+        this.executions = executions;
+        this.rootExecution = rootExecution;
+        this.leaves = leaves;
+    }
+
+    /** The query this graph belongs to. */
+    public String queryId() {
+        return queryId;
+    }
+
+    /** The root stage execution. */
+    public StageExecution rootExecution() {
+        return rootExecution;
+    }
+
+    /** All leaf executions (stages with no children). */
+    public List<StageExecution> leaves() {
+        return Collections.unmodifiableList(leaves);
+    }
+
+    /** Lookup a stage execution by stage id. */
+    public StageExecution executionFor(int stageId) {
+        return executions.get(stageId);
+    }
+
+    /** All stage executions in the graph. */
+    public Collection<StageExecution> allExecutions() {
+        return Collections.unmodifiableCollection(executions.values());
+    }
+
+    /** Number of stages in the graph. */
+    public int stageCount() {
+        return executions.size();
+    }
+
+    /**
+     * Returns a human-readable summary of the execution graph for
+     * EXPLAIN output. Lists each stage with its type, state, and
+     * child dependencies.
+     */
+    public String explain() {
+        StringBuilder sb = new StringBuilder();
+        sb.append("ExecutionGraph[queryId=").append(queryId);
+        sb.append(", stages=").append(executions.size());
+        sb.append(", leaves=").append(leaves.size()).append("]\n");
+        for (StageExecution exec : executions.values()) {
+            sb.append("  Stage ").append(exec.getStageId());
+            sb.append(" [").append(exec.getClass().getSimpleName()).append("]");
+            sb.append(" state=").append(exec.getState());
+            StageMetrics m = exec.getMetrics();
+            if (m.getStartTimeMs() > 0) {
+                sb.append(" elapsed=").append(m.getEndTimeMs() - m.getStartTimeMs()).append("ms");
+                sb.append(" rows=").append(m.getRowsProcessed());
+            }
+            sb.append("\n");
+        }
+        return sb.toString();
+    }
+}