Introduce Instructions for the Backend to use during execution

sandbox/libs/analytics-framework/src/main/java/org/opensearch/analytics/backend/ExecutionContext.java → sandbox/libs/analytics-framework/src/main/java/org/opensearch/analytics/backend/ShardScanExecutionContext.java

@@ -9,6 +9,7 @@
 package org.opensearch.analytics.backend;
 
 import org.apache.arrow.memory.BufferAllocator;
+import org.opensearch.analytics.spi.CommonExecutionContext;
 import org.opensearch.index.engine.exec.IndexReaderProvider.Reader;
 import org.opensearch.tasks.Task;
 
@@ -18,7 +19,7 @@
  *
  * @opensearch.internal
  */
-public class ExecutionContext {
+public class ShardScanExecutionContext implements CommonExecutionContext {
 
     private final String tableName;
     private final Reader reader;
@@ -32,7 +33,7 @@ public class ExecutionContext {
      * @param task the transport-created task for this fragment execution
      * @param reader the data-format aware reader
      */
-    public ExecutionContext(String tableName, Task task, Reader reader) {
+    public ShardScanExecutionContext(String tableName, Task task, Reader reader) {
         this.tableName = tableName;
         this.task = task;
         this.reader = reader;

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

@@ -70,4 +70,15 @@ default ExchangeSinkProvider getExchangeSinkProvider() {
         return null;
     }
 
+    /**
+     * Returns the instruction handler factory for this backend. Used at the coordinator
+     * to create instruction nodes (backend attaches custom config) and at the data node
+     * to create handlers that apply instructions to the execution context.
+     *
+     * <p>Backends that declare {@code supportedDelegations} or participate in multi-stage
+     * execution MUST implement this. Validation at startup ensures consistency.
+     */
+    default FragmentInstructionHandlerFactory getInstructionHandlerFactory() {
+        throw new UnsupportedOperationException("getInstructionHandlerFactory not implemented for [" + name() + "]");
+    }
 }

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

@@ -0,0 +1,22 @@
+/*
+ * 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;
+
+/**
+ * Marker interface for backend-specific execution context that flows between
+ * successive instruction handler calls. The first handler in the chain receives
+ * {@code null} and bootstraps the context; subsequent handlers receive and build
+ * upon the previous handler's output.
+ *
+ * <p>Each backend defines its own concrete implementation (e.g.,
+ * {@code DataFusionSessionState} holding a native SessionContext handle).
+ *
+ * @opensearch.internal
+ */
+public interface BackendExecutionContext {}

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

@@ -0,0 +1,22 @@
+/*
+ * 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;
+
+/**
+ * Marker interface for backend-specific execution state that flows between
+ * successive instruction handler calls. The first handler in the chain receives
+ * {@code null} and bootstraps the state; subsequent handlers receive and build
+ * upon the previous handler's output.
+ *
+ * <p>Each backend defines its own concrete implementation (e.g.,
+ * {@code DataFusionSessionState} holding a native SessionContext handle).
+ *
+ * @opensearch.internal
+ */
+public interface BackendExecutionState {}

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

@@ -0,0 +1,21 @@
+/*
+ * 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;
+
+/**
+ * Marker interface for execution contexts provided by Core to instruction handlers.
+ * Concrete implementations carry the information relevant to their execution path:
+ * <ul>
+ *   <li>{@code ShardScanExecutionContext} — shard fragment execution (reader, task, tableName)</li>
+ *   <li>{@code ExchangeSinkContext} — coordinator reduce execution (planBytes, allocator, schema)</li>
+ * </ul>
+ *
+ * @opensearch.internal
+ */
+public interface CommonExecutionContext {}

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

@@ -0,0 +1,60 @@
+/*
+ * 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.opensearch.core.common.io.stream.StreamInput;
+import org.opensearch.core.common.io.stream.StreamOutput;
+import org.opensearch.core.common.io.stream.Writeable;
+
+import java.io.IOException;
+
+/**
+ * A single delegated predicate — carries the annotation ID, the accepting backend,
+ * and the serialized bytes produced by the accepting backend's
+ * {@link DelegatedPredicateSerializer} or anything similar.
+ *
+ * @opensearch.internal
+ */
+public class DelegatedExpression implements Writeable {
+
+    private final int annotationId;
+    private final String acceptingBackendId;
+    private final byte[] expressionBytes;
+
+    public DelegatedExpression(int annotationId, String acceptingBackendId, byte[] expressionBytes) {
+        this.annotationId = annotationId;
+        this.acceptingBackendId = acceptingBackendId;
+        this.expressionBytes = expressionBytes;
+    }
+
+    public DelegatedExpression(StreamInput in) throws IOException {
+        this.annotationId = in.readInt();
+        this.acceptingBackendId = in.readString();
+        this.expressionBytes = in.readByteArray();
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        out.writeInt(annotationId);
+        out.writeString(acceptingBackendId);
+        out.writeByteArray(expressionBytes);
+    }
+
+    public int getAnnotationId() {
+        return annotationId;
+    }
+
+    public String getAcceptingBackendId() {
+        return acceptingBackendId;
+    }
+
+    public byte[] getExpressionBytes() {
+        return expressionBytes;
+    }
+}

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

@@ -43,7 +43,7 @@
  * @opensearch.internal
  */
 public record ExchangeSinkContext(String queryId, int stageId, byte[] fragmentBytes, BufferAllocator allocator, List<
-    ChildInput> childInputs, ExchangeSink downstream) {
+    ChildInput> childInputs, ExchangeSink downstream) implements CommonExecutionContext {
 
     /** Per-child input descriptor: the child stage id and the schema of its outgoing batches. */
     public record ChildInput(int childStageId, Schema schema) {

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

@@ -0,0 +1,68 @@
+/*
+ * 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.opensearch.core.common.io.stream.StreamInput;
+import org.opensearch.core.common.io.stream.StreamOutput;
+
+import java.io.IOException;
+import java.util.List;
+
+/**
+ * Instruction node for filter delegation to an index backend.
+ * Carries the tree shape, predicate count, and serialized delegated queries.
+ *
+ * @opensearch.internal
+ */
+public class FilterDelegationInstructionNode implements InstructionNode {
+
+    private final FilterTreeShape treeShape;
+    private final int delegatedPredicateCount;
+    private final List<DelegatedExpression> delegatedQueries;
+
+    public FilterDelegationInstructionNode(
+        FilterTreeShape treeShape,
+        int delegatedPredicateCount,
+        List<DelegatedExpression> delegatedQueries
+    ) {
+        this.treeShape = treeShape;
+        this.delegatedPredicateCount = delegatedPredicateCount;
+        this.delegatedQueries = delegatedQueries;
+    }
+
+    public FilterDelegationInstructionNode(StreamInput in) throws IOException {
+        this.treeShape = in.readEnum(FilterTreeShape.class);
+        this.delegatedPredicateCount = in.readInt();
+        this.delegatedQueries = in.readList(DelegatedExpression::new);
+    }
+
+    @Override
+    public InstructionType type() {
+        return InstructionType.SETUP_FILTER_DELEGATION_FOR_INDEX;
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        out.writeEnum(treeShape);
+        out.writeInt(delegatedPredicateCount);
+        out.writeCollection(delegatedQueries);
+    }
+
+    public FilterTreeShape getTreeShape() {
+        return treeShape;
+    }
+
+    public int getDelegatedPredicateCount() {
+        return delegatedPredicateCount;
+    }
+
+    public List<DelegatedExpression> getDelegatedQueries() {
+        return delegatedQueries;
+    }
+}

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

@@ -0,0 +1,32 @@
+/*
+ * 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;
+
+/**
+ * Backend-agnostic description of the boolean tree shape when filter delegation is active.
+ * Provided by the planner so backends can choose their execution strategy without
+ * re-inspecting the Substrait plan.
+ *
+ * @opensearch.internal
+ */
+public enum FilterTreeShape {
+    /** No delegation — all predicates handled natively by the driving backend. */
+    NO_DELEGATION,
+    /**
+     * All predicates (delegated + native) are under a single AND — no interleaving
+     * under OR/NOT. Backend can handle delegated bitsets and native predicates independently.
+     */
+    CONJUNCTIVE,
+    /**
+     * Delegated and native predicates are interleaved under OR/NOT — the boolean tree
+     * mixes predicates from different backends under non-AND operators. Backend needs a
+     * tree evaluator to combine bitsets from both backends per the boolean structure.
+     */
+    INTERLEAVED_BOOLEAN_EXPRESSION
+}

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

@@ -0,0 +1,41 @@
+/*
+ * 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.opensearch.core.common.io.stream.StreamInput;
+import org.opensearch.core.common.io.stream.StreamOutput;
+
+import java.io.IOException;
+
+/**
+ * Instruction node for final aggregate in coordinator reduce — ExchangeSink path,
+ * remove partial agg, preserve final-only for the driving backend's reduce execution.
+ *
+ * <p>TODO: add backend-specific config fields as final aggregate implementation is built out.
+ *
+ * @opensearch.internal
+ */
+public class FinalAggregateInstructionNode implements InstructionNode {
+
+    public FinalAggregateInstructionNode() {}
+
+    public FinalAggregateInstructionNode(StreamInput in) throws IOException {
+        // TODO: read config fields when added
+    }
+
+    @Override
+    public InstructionType type() {
+        return InstructionType.SETUP_FINAL_AGGREGATE;
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        // TODO: write config fields when added
+    }
+}

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

@@ -0,0 +1,31 @@
+/*
+ * 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;
+
+/**
+ * Applies an {@link InstructionNode} to the execution context at the data node.
+ * Each handler is created per-execution by the backend's
+ * {@link FragmentInstructionHandlerFactory#createHandler(InstructionNode)}.
+ *
+ * @param <N> the concrete instruction node type this handler processes
+ * @opensearch.internal
+ */
+public interface FragmentInstructionHandler<N extends InstructionNode> {
+
+    /**
+     * Applies the instruction, reading from Core's context and building upon the
+     * backend's accumulated execution context from previous handlers.
+     *
+     * @param node the instruction node
+     * @param commonContext Core-provided context (shard info or reduce info)
+     * @param backendContext backend state from previous handler, or {@code null} for the first handler
+     * @return updated backend execution context for the next handler or final consumer
+     */
+    BackendExecutionContext apply(N node, CommonExecutionContext commonContext, BackendExecutionContext backendContext);
+}