[Mustang] Add query routing and execution handoff for Parquet-backed indices (opensearch-project#5247)

Implements the query routing and AnalyticsExecutionEngine for Project
Mustang's unified query pipeline. PPL queries targeting parquet_ prefixed
indices are routed through UnifiedQueryPlanner and executed via a stub
QueryPlanExecutor, with results formatted through the existing JDBC
response pipeline.

New files:
- QueryPlanExecutor: @FunctionalInterface contract for analytics engine
- AnalyticsExecutionEngine: converts Iterable<Object[]> to QueryResponse
  with type mapping and query size limit enforcement
- RestUnifiedQueryAction: orchestrates schema building, planning,
  execution on sql-worker thread pool, with client/server error
  classification and metrics
- StubQueryPlanExecutor: canned data for parquet_logs and parquet_metrics
  tables for development and testing

Modified files:
- RestPPLQueryAction: routing branch for parquet_ indices
- SQLPlugin: passes ClusterService and NodeClient to RestPPLQueryAction
- plugin/build.gradle: adds :api dependency for UnifiedQueryPlanner

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,115 @@
+/*
+ * 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.rel.RelNode;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeField;
+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);
+    }
+  }
+
+  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);
+}