[Analytics Engine] Eliminate string/utf8 views & timestamp field coercion across the data-node→coordinator pipeline

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

@@ -9,7 +9,6 @@
 package org.opensearch.analytics.spi;
 
 import org.apache.arrow.memory.BufferAllocator;
-import org.apache.arrow.vector.types.pojo.Schema;
 
 import java.util.List;
 
@@ -32,9 +31,10 @@
  *   <li>{@code childInputs} — one entry per child stage. Each entry carries
  *       the child's stage id (used by the backend to register a per-child
  *       input partition under a stable name like {@code "input-<stageId>"})
- *       and the Arrow schema of the batches the child will feed in. For
- *       single-input shapes this list has size 1; for {@code UNION}-style
- *       multi-input shapes it has one entry per Union branch.</li>
+ *       and the producer-side plan bytes (e.g. partial-aggregate substrait)
+ *       the backend lowers to derive the input schema. For single-input
+ *       shapes this list has size 1; for {@code UNION}-style multi-input
+ *       shapes it has one entry per Union branch.</li>
  *   <li>{@code downstream} — sink the backend drains its reduced output
  *       into. The backend owns {@code downstream}'s lifecycle: it must
  *       feed every produced batch and close it when draining is complete.</li>
@@ -45,21 +45,11 @@
 public record ExchangeSinkContext(String queryId, int stageId, byte[] fragmentBytes, BufferAllocator allocator, List<
     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) {
-    }
-
     /**
-     * Convenience for single-input back-compat. Returns the schema of the sole
-     * child input. Throws when {@link #childInputs} contains more than one entry —
-     * multi-input callers must inspect {@link #childInputs} directly.
+     * Per-child input descriptor: the child stage id and the producer-side plan bytes the
+     * backend lowers when it registers the child's input partition. The actual Arrow schema
+     * is learned at registration time, not declared here.
      */
-    public Schema inputSchema() {
-        if (childInputs.size() != 1) {
-            throw new IllegalStateException(
-                "inputSchema() requires exactly one child input; got " + childInputs.size() + " — use childInputs() instead"
-            );
-        }
-        return childInputs.get(0).schema();
+    public record ChildInput(int childStageId, byte[] producerPlanBytes) {
     }
 }

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

@@ -25,8 +25,13 @@ public interface ExchangeSinkProvider {
     /**
      * Creates a sink for coordinator-side execution. The backend implementation
      * uses {@link ExchangeSinkContext#fragmentBytes()} as the serialized plan
-     * (produced by {@link FragmentConvertor#convertFinalAggFragment}) and
-     * writes its reduced output into {@link ExchangeSinkContext#downstream()}.
+     * (produced by {@code FragmentConvertor#convertFragment}) and writes its
+     * reduced output into {@link ExchangeSinkContext#downstream()}.
+     *
+     * <p>The schema of each child's batches is learned at the backend boundary
+     * (not pre-declared on {@link ExchangeSinkContext.ChildInput}) — the backend
+     * derives it when it registers the child input on its native session, since
+     * the producer-side plan bytes already encode the producer schema.
      *
      * @param context core-provided context carrying plan bytes, allocator, child inputs, and downstream sink
      * @param backendContext backend-opaque state produced by instruction handlers (e.g.

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

@@ -21,9 +21,9 @@
  *
  * <p>Composable pipeline for multi-shard aggregate with sort at coordinator:
  * <ol>
- *   <li>{@code convertShardScanFragment(tableName, Filter(Scan))} → data node inner bytes</li>
+ *   <li>{@code convertFragment(Filter(Scan))} → data node inner bytes</li>
  *   <li>{@code attachPartialAggOnTop(PartialAgg, innerBytes)} → data node bytes</li>
- *   <li>{@code convertFinalAggFragment(FinalAgg(StageInputScan))} → reduce stage inner bytes</li>
+ *   <li>{@code convertFragment(FinalAgg(StageInputScan))} → reduce stage inner bytes</li>
  *   <li>{@code attachFragmentOnTop(Sort, innerBytes)} → reduce stage bytes</li>
  * </ol>
  *
@@ -35,16 +35,25 @@
 public interface FragmentConvertor {
 
     /**
-     * Converts a fragment whose leaf is a native physical shard scan, containing
-     * everything below a partial aggregate (e.g. Filter(Scan), Scan).
-     * The backend handles all operators natively — no delegation, no shuffle.
+     * Converts a resolved RelNode fragment (annotations stripped) into
+     * backend-specific serialized plan bytes. The fragment may be:
+     * <ul>
+     *   <li>A shard-scan subtree below a partial aggregate (e.g. Filter(Scan), Scan).</li>
+     *   <li>A reduce-stage final-aggregate fragment whose leaf is
+     *       {@code OpenSearchStageInputScan} — the backend rewrites these to
+     *       named-table reads pointing at the streaming input partition.</li>
+     *   <li>A coord-only literal source (e.g. {@code OpenSearchValues}).</li>
+     * </ul>
      *
-     * @param tableName named table the fragment's scan references
-     * @param fragment  resolved RelNode fragment (annotations stripped)
+     * <p>TODO: revisit placement of FragmentConvertor — it references Calcite RelNode
+     * and is called only by analytics-engine. Consider moving to analytics-engine and
+     * removing getFragmentConvertor() from AnalyticsSearchBackendPlugin SPI.
+     *
+     * @param fragment resolved RelNode fragment
      * @return backend-specific serialized plan bytes
      */
-    default byte[] convertShardScanFragment(String tableName, RelNode fragment) {
-        throw new UnsupportedOperationException("convertShardScanFragment not implemented for this backend");
+    default byte[] convertFragment(RelNode fragment) {
+        throw new UnsupportedOperationException("convertFragment not implemented for this backend");
     }
 
     /**
@@ -53,31 +62,13 @@ default byte[] convertShardScanFragment(String tableName, RelNode fragment) {
      * aggregate execution node.
      *
      * @param partialAggFragment the partial aggregate RelNode (annotations stripped, no children)
-     * @param innerBytes         serialized bytes from a prior {@code convert*} call
+     * @param innerBytes         serialized bytes from a prior {@link #convertFragment} call
      * @return serialized plan bytes with partial aggregate attached on top
      */
     default byte[] attachPartialAggOnTop(RelNode partialAggFragment, byte[] innerBytes) {
         throw new UnsupportedOperationException("attachPartialAggOnTop not implemented for this backend");
     }
 
-    /**
-     * Converts the final aggregate fragment at the reduce stage.
-     * The leaf is a StageInputScan placeholder used for schema inference —
-     * replaced at execution time with a streaming Arrow batch source
-     * (e.g. StreamingTableExec in DataFusion).
-     *
-     * <p>TODO: revisit placement of FragmentConvertor — it references Calcite RelNode
-     * and is called only by analytics-engine. Consider moving to analytics-engine and
-     * removing getFragmentConvertor() from AnalyticsSearchBackendPlugin SPI.
-     *
-     * @param fragment resolved final aggregate RelNode (annotations stripped,
-     *                 ExchangeReducer removed, StageInputScan as leaf)
-     * @return backend-specific serialized plan bytes
-     */
-    default byte[] convertFinalAggFragment(RelNode fragment) {
-        throw new UnsupportedOperationException("convertFinalAggFragment not implemented for this backend");
-    }
-
     /**
      * Attaches a generic fragment (Sort, Project, etc.) on top of already-converted
      * inner bytes. The backend deserializes the inner plan and wraps it with the