Add AnalyticsExecutionEngine and query routing for Parquet-backed indices (opensearch-project#5247)

Implements the execution engine adapter and query routing infrastructure
for Project Mustang's unified query pipeline (Option B). PPL queries
targeting parquet_ prefixed indices are routed through UnifiedQueryPlanner
and AnalyticsExecutionEngine instead of the existing Lucene path.

Key changes:
- QueryPlanExecutor interface: contract for analytics engine execution
- AnalyticsExecutionEngine: converts QueryPlanExecutor results to QueryResponse
- RestUnifiedQueryAction: orchestrates schema building, planning, execution
- RestPPLQueryAction: routing branch for parquet_ indices
- StubQueryPlanExecutor: canned data for development/testing

Signed-off-by: Kai Huang <kaihuang@amazon.com>
Signed-off-by: Kai Huang <ahkcs@amazon.com>

Author: ahkcs

core/src/main/java/org/opensearch/sql/executor/analytics/AnalyticsExecutionEngine.java

@@ -0,0 +1,139 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.sql.executor.analytics;
+
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.calcite.plan.RelOptUtil;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeField;
+import org.apache.calcite.sql.SqlExplainLevel;
+import org.opensearch.sql.ast.statement.ExplainMode;
+import org.opensearch.sql.calcite.CalcitePlanContext;
+import org.opensearch.sql.calcite.utils.OpenSearchTypeFactory;
+import org.opensearch.sql.common.response.ResponseListener;
+import org.opensearch.sql.data.model.ExprTupleValue;
+import org.opensearch.sql.data.model.ExprValue;
+import org.opensearch.sql.data.model.ExprValueUtils;
+import org.opensearch.sql.data.type.ExprType;
+import org.opensearch.sql.executor.ExecutionContext;
+import org.opensearch.sql.executor.ExecutionEngine;
+import org.opensearch.sql.executor.pagination.Cursor;
+import org.opensearch.sql.planner.physical.PhysicalPlan;
+
+/**
+ * Execution engine adapter for the analytics engine (Project Mustang).
+ *
+ * <p>Bridges the analytics engine's {@link QueryPlanExecutor} with the SQL plugin's {@link
+ * ExecutionEngine} response pipeline. Takes a Calcite {@link RelNode}, delegates execution to the
+ * analytics engine, and converts the raw results into {@link QueryResponse}.
+ */
+public class AnalyticsExecutionEngine implements ExecutionEngine {
+
+  private final QueryPlanExecutor planExecutor;
+
+  public AnalyticsExecutionEngine(QueryPlanExecutor planExecutor) {
+    this.planExecutor = planExecutor;
+  }
+
+  /** Not supported. Analytics queries use the RelNode path exclusively. */
+  @Override
+  public void execute(PhysicalPlan plan, ResponseListener<QueryResponse> listener) {
+    listener.onFailure(
+        new UnsupportedOperationException("Analytics engine only supports RelNode execution"));
+  }
+
+  /** Not supported. Analytics queries use the RelNode path exclusively. */
+  @Override
+  public void execute(
+      PhysicalPlan plan, ExecutionContext context, ResponseListener<QueryResponse> listener) {
+    listener.onFailure(
+        new UnsupportedOperationException("Analytics engine only supports RelNode execution"));
+  }
+
+  /** Not supported. Analytics queries use the RelNode path exclusively. */
+  @Override
+  public void explain(PhysicalPlan plan, ResponseListener<ExplainResponse> listener) {
+    listener.onFailure(
+        new UnsupportedOperationException("Analytics engine only supports RelNode execution"));
+  }
+
+  @Override
+  public void execute(
+      RelNode plan, CalcitePlanContext context, ResponseListener<QueryResponse> listener) {
+    try {
+      Integer querySizeLimit = context.sysLimit.querySizeLimit();
+      Iterable<Object[]> rows = planExecutor.execute(plan, null);
+
+      List<RelDataTypeField> fields = plan.getRowType().getFieldList();
+      List<ExprValue> results = convertRows(rows, fields, querySizeLimit);
+      Schema schema = buildSchema(fields);
+
+      listener.onResponse(new QueryResponse(schema, results, Cursor.None));
+    } catch (Exception e) {
+      listener.onFailure(e);
+    }
+  }
+
+  @Override
+  public void explain(
+      RelNode plan,
+      ExplainMode mode,
+      CalcitePlanContext context,
+      ResponseListener<ExplainResponse> listener) {
+    try {
+      SqlExplainLevel level =
+          mode == ExplainMode.SIMPLE
+              ? SqlExplainLevel.NO_ATTRIBUTES
+              : mode == ExplainMode.COST
+                  ? SqlExplainLevel.ALL_ATTRIBUTES
+                  : SqlExplainLevel.EXPPLAN_ATTRIBUTES;
+      String logical = RelOptUtil.toString(plan, level);
+      ExplainResponseNodeV2 nodeV2 = new ExplainResponseNodeV2(logical, null, null);
+      listener.onResponse(new ExplainResponse(nodeV2));
+    } catch (Exception e) {
+      listener.onFailure(e);
+    }
+  }
+
+  private List<ExprValue> convertRows(
+      Iterable<Object[]> rows, List<RelDataTypeField> fields, Integer querySizeLimit) {
+    List<ExprValue> results = new ArrayList<>();
+    for (Object[] row : rows) {
+      if (querySizeLimit != null && results.size() >= querySizeLimit) {
+        break;
+      }
+      Map<String, ExprValue> valueMap = new LinkedHashMap<>();
+      for (int i = 0; i < fields.size(); i++) {
+        String columnName = fields.get(i).getName();
+        Object value = (i < row.length) ? row[i] : null;
+        valueMap.put(columnName, ExprValueUtils.fromObjectValue(value));
+      }
+      results.add(ExprTupleValue.fromExprValueMap(valueMap));
+    }
+    return results;
+  }
+
+  private Schema buildSchema(List<RelDataTypeField> fields) {
+    List<Schema.Column> columns = new ArrayList<>();
+    for (RelDataTypeField field : fields) {
+      ExprType exprType = convertType(field.getType());
+      columns.add(new Schema.Column(field.getName(), null, exprType));
+    }
+    return new Schema(columns);
+  }
+
+  private ExprType convertType(RelDataType type) {
+    try {
+      return OpenSearchTypeFactory.convertRelDataTypeToExprType(type);
+    } catch (IllegalArgumentException e) {
+      return org.opensearch.sql.data.type.ExprCoreType.UNKNOWN;
+    }
+  }
+}

core/src/main/java/org/opensearch/sql/executor/analytics/QueryPlanExecutor.java

@@ -0,0 +1,32 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.sql.executor.analytics;
+
+import org.apache.calcite.rel.RelNode;
+
+/**
+ * Executes a Calcite {@link RelNode} logical plan against the analytics engine.
+ *
+ * <p>This is a local equivalent of {@code org.opensearch.analytics.exec.QueryPlanExecutor} from the
+ * analytics-framework library. It will be replaced by the upstream interface once the
+ * analytics-framework JAR is published.
+ *
+ * @see <a
+ *     href="https://github.com/opensearch-project/OpenSearch/blob/9142d0e789c6a6c4708f1bc015745ed55202eefe/sandbox/libs/analytics-framework/src/main/java/org/opensearch/analytics/exec/QueryPlanExecutor.java">Upstream
+ *     QueryPlanExecutor</a>
+ */
+@FunctionalInterface
+public interface QueryPlanExecutor {
+
+  /**
+   * Executes the given logical plan and returns result rows.
+   *
+   * @param plan the Calcite RelNode subtree to execute
+   * @param context execution context (opaque to avoid server dependency)
+   * @return rows produced by the engine
+   */
+  Iterable<Object[]> execute(RelNode plan, Object context);
+}