Integrate SQL REST endpoint with analytics engine path

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

@@ -99,7 +99,9 @@ public void explain(
       ResponseListener<ExplainResponse> listener) {
     try {
       String logical = RelOptUtil.toString(plan, mode.toExplainLevel());
-      listener.onResponse(new ExplainResponse(new ExplainResponseNodeV2(logical, null, null)));
+      ExplainResponse response =
+          new ExplainResponse(new ExplainResponseNodeV2(logical, null, null));
+      listener.onResponse(ExplainResponse.normalizeLf(response));
     } catch (Exception e) {
       listener.onFailure(e);
     }

integ-test/src/test/java/org/opensearch/sql/sql/AnalyticsSQLExplainIT.java

@@ -0,0 +1,82 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.sql.sql;
+
+import static org.opensearch.sql.legacy.TestUtils.isIndexExist;
+import static org.opensearch.sql.util.MatcherUtils.assertJsonEqualsIgnoreId;
+
+import com.google.common.io.Resources;
+import java.io.IOException;
+import java.net.URI;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import org.junit.Test;
+import org.opensearch.client.Request;
+import org.opensearch.sql.legacy.SQLIntegTestCase;
+
+/**
+ * Explain integration tests for SQL queries routed through the analytics engine path (Project
+ * Analytics engine). Validates that SQL queries targeting "parquet_*" indices produce correct
+ * logical plans via the _plugins/_sql/_explain endpoint.
+ *
+ * <p>Expected output files are in resources/expectedOutput/analytics_sql/. Each test compares the
+ * explain JSON output against its expected file.
+ */
+@SuppressWarnings("deprecation") // assertJsonEqualsIgnoreId is correct for JSON explain response
+public class AnalyticsSQLExplainIT extends SQLIntegTestCase {
+
+  @Override
+  protected void init() throws Exception {
+    if (!isIndexExist(client(), "parquet_logs")) {
+      Request request = new Request("PUT", "/parquet_logs");
+      request.setJsonEntity(
+          "{"
+              + "\"mappings\": {"
+              + "  \"properties\": {"
+              + "    \"ts\": {\"type\": \"date\"},"
+              + "    \"status\": {\"type\": \"integer\"},"
+              + "    \"message\": {\"type\": \"keyword\"},"
+              + "    \"ip_addr\": {\"type\": \"keyword\"}"
+              + "  }"
+              + "}"
+              + "}");
+      client().performRequest(request);
+    }
+  }
+
+  private static String loadExpectedJson(String fileName) {
+    return loadFromFile("expectedOutput/analytics_sql/" + fileName);
+  }
+
+  private static String loadFromFile(String filename) {
+    try {
+      URI uri = Resources.getResource(filename).toURI();
+      return new String(Files.readAllBytes(Paths.get(uri)));
+    } catch (Exception e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  @Test
+  public void testExplainSelectStar() throws IOException {
+    assertJsonEqualsIgnoreId(
+        loadExpectedJson("explain_select_star.json"), explainQuery("SELECT * FROM parquet_logs"));
+  }
+
+  @Test
+  public void testExplainSelectColumns() throws IOException {
+    assertJsonEqualsIgnoreId(
+        loadExpectedJson("explain_select_columns.json"),
+        explainQuery("SELECT ts, status FROM parquet_logs"));
+  }
+
+  @Test
+  public void testExplainSelectWithWhere() throws IOException {
+    assertJsonEqualsIgnoreId(
+        loadExpectedJson("explain_select_where.json"),
+        explainQuery("SELECT ts, message FROM parquet_logs WHERE status = 200"));
+  }
+}

integ-test/src/test/java/org/opensearch/sql/sql/AnalyticsSQLIT.java

@@ -0,0 +1,90 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.sql.sql;
+
+import static org.opensearch.sql.legacy.TestUtils.isIndexExist;
+import static org.opensearch.sql.util.MatcherUtils.rows;
+import static org.opensearch.sql.util.MatcherUtils.schema;
+import static org.opensearch.sql.util.MatcherUtils.verifyDataRows;
+import static org.opensearch.sql.util.MatcherUtils.verifySchema;
+
+import java.io.IOException;
+import org.json.JSONObject;
+import org.junit.Test;
+import org.opensearch.client.Request;
+import org.opensearch.client.ResponseException;
+import org.opensearch.sql.legacy.SQLIntegTestCase;
+
+/**
+ * Integration tests for SQL queries routed through the analytics engine path. Queries targeting
+ * "parquet_*" indices are routed to {@code RestUnifiedQueryAction} which uses {@code
+ * AnalyticsExecutionEngine} with a stub {@code QueryPlanExecutor}.
+ *
+ * <p>The stub executor returns rows in a fixed order [ts, status, message, ip_addr] regardless of
+ * the plan. The schema from OpenSearchSchemaBuilder is alphabetical [ip_addr, message, status, ts].
+ * AnalyticsExecutionEngine maps values by position, so the data values appear mismatched. This is
+ * expected; the real analytics engine will evaluate the plan correctly.
+ */
+public class AnalyticsSQLIT extends SQLIntegTestCase {
+
+  @Override
+  protected void init() throws Exception {
+    createParquetLogsIndex();
+  }
+
+  private void createParquetLogsIndex() throws IOException {
+    if (isIndexExist(client(), "parquet_logs")) {
+      return;
+    }
+    Request request = new Request("PUT", "/parquet_logs");
+    request.setJsonEntity(
+        "{"
+            + "\"mappings\": {"
+            + "  \"properties\": {"
+            + "    \"ts\": {\"type\": \"date\"},"
+            + "    \"status\": {\"type\": \"integer\"},"
+            + "    \"message\": {\"type\": \"keyword\"},"
+            + "    \"ip_addr\": {\"type\": \"keyword\"}"
+            + "  }"
+            + "}"
+            + "}");
+    client().performRequest(request);
+  }
+
+  @Test
+  public void testSelectStarSchemaAndData() throws IOException {
+    JSONObject result = executeQuery("SELECT * FROM parquet_logs");
+    verifySchema(
+        result,
+        schema("ip_addr", "string"),
+        schema("message", "string"),
+        schema("status", "integer"),
+        schema("ts", "timestamp"));
+    // Stub returns [ts, status, message, ip_addr] per row, mapped by position to
+    // [ip_addr, message, status, ts] schema. Values appear mismatched — expected with stub.
+    verifyDataRows(
+        result,
+        rows("2024-01-15 10:30:00", 200, "Request completed", "192.168.1.1"),
+        rows("2024-01-15 10:31:00", 200, "Health check OK", "192.168.1.2"),
+        rows("2024-01-15 10:32:00", 500, "Internal server error", "192.168.1.3"));
+  }
+
+  @Test
+  public void testSelectSpecificColumns() throws IOException {
+    JSONObject result = executeQuery("SELECT status, message FROM parquet_logs");
+    verifySchema(result, schema("status", "integer"), schema("message", "string"));
+    verifyDataRows(
+        result,
+        rows("2024-01-15 10:30:00", 200),
+        rows("2024-01-15 10:31:00", 200),
+        rows("2024-01-15 10:32:00", 500));
+  }
+
+  @Test(expected = ResponseException.class)
+  public void testSyntaxError() throws IOException {
+    executeQuery("SELEC * FROM parquet_logs");
+  }
+}

integ-test/src/test/resources/expectedOutput/analytics_sql/explain_select_columns.json

@@ -0,0 +1,5 @@
+{
+  "calcite": {
+    "logical": "LogicalSystemLimit(fetch=[10000], type=[QUERY_SIZE_LIMIT])\n  LogicalProject(ts=[$3], status=[$2])\n    LogicalTableScan(table=[[opensearch, parquet_logs]])\n"
+  }
+}

integ-test/src/test/resources/expectedOutput/analytics_sql/explain_select_star.json

@@ -0,0 +1,5 @@
+{
+  "calcite": {
+    "logical": "LogicalSystemLimit(fetch=[10000], type=[QUERY_SIZE_LIMIT])\n  LogicalProject(ip_addr=[$0], message=[$1], status=[$2], ts=[$3])\n    LogicalTableScan(table=[[opensearch, parquet_logs]])\n"
+  }
+}

integ-test/src/test/resources/expectedOutput/analytics_sql/explain_select_where.json

@@ -0,0 +1,5 @@
+{
+  "calcite": {
+    "logical": "LogicalSystemLimit(fetch=[10000], type=[QUERY_SIZE_LIMIT])\n  LogicalProject(ts=[$3], message=[$1])\n    LogicalFilter(condition=[=($2, 200)])\n      LogicalTableScan(table=[[opensearch, parquet_logs]])\n"
+  }
+}

legacy/src/main/java/org/opensearch/sql/legacy/plugin/RestSqlAction.java

@@ -19,6 +19,7 @@
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
+import java.util.function.BiFunction;
 import java.util.function.Predicate;
 import java.util.regex.Pattern;
 import org.apache.logging.log4j.LogManager;
@@ -83,10 +84,25 @@ public class RestSqlAction extends BaseRestHandler {
   /** New SQL query request handler. */
   private final RestSQLQueryAction newSqlQueryHandler;
 
+  /**
+   * Optional analytics router. If set, it's called before the normal SQL engine. Accepts the
+   * request and channel, returns {@code true} if it handled the request, {@code false} to fall
+   * through to normal SQL engine.
+   */
+  private final BiFunction<SQLQueryRequest, RestChannel, Boolean> analyticsRouter;
+
   public RestSqlAction(Settings settings, Injector injector) {
+    this(settings, injector, null);
+  }
+
+  public RestSqlAction(
+      Settings settings,
+      Injector injector,
+      BiFunction<SQLQueryRequest, RestChannel, Boolean> analyticsRouter) {
     super();
     this.allowExplicitIndex = MULTI_ALLOW_EXPLICIT_INDEX.get(settings);
     this.newSqlQueryHandler = new RestSQLQueryAction(injector);
+    this.analyticsRouter = analyticsRouter;
   }
 
   @Override
@@ -134,14 +150,44 @@ protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient cli
 
       Format format = SqlRequestParam.getFormat(request.params());
 
-      // Route request to new query engine if it's supported already
       SQLQueryRequest newSqlRequest =
           new SQLQueryRequest(
               sqlRequest.getJsonContent(),
               sqlRequest.getSql(),
               request.path(),
               request.params(),
               sqlRequest.cursor());
+
+      // Route to analytics engine for non-Lucene (e.g., Parquet-backed) indices.
+      // The router returns true and sends the response directly if it handled the request.
+      if (analyticsRouter != null) {
+        final SQLQueryRequest finalRequest = newSqlRequest;
+        return channel -> {
+          if (!analyticsRouter.apply(finalRequest, channel)) {
+            // Not an analytics query — delegate to normal SQL engine
+            try {
+              newSqlQueryHandler
+                  .prepareRequest(
+                      finalRequest,
+                      (ch, ex) -> {
+                        try {
+                          Format fmt = SqlRequestParam.getFormat(request.params());
+                          QueryAction qa = explainRequest(client, sqlRequest, fmt);
+                          executeSqlRequest(request, qa, client, ch);
+                        } catch (Exception e) {
+                          handleException(ch, e);
+                        }
+                      },
+                      this::handleException)
+                  .accept(channel);
+            } catch (Exception e) {
+              handleException(channel, e);
+            }
+          }
+        };
+      }
+
+      // Route request to new query engine if it's supported already
       return newSqlQueryHandler.prepareRequest(
           newSqlRequest,
           (restChannel, exception) -> {

plugin/build.gradle

@@ -348,6 +348,7 @@ def getJobSchedulerPlugin() {
 
 testClusters.integTest {
     plugin(getJobSchedulerPlugin())
+    plugin provider { (RegularFile) (() -> file("${rootDir}/libs/analytics-engine-3.6.0-SNAPSHOT.zip")) }
     plugin(project.tasks.bundlePlugin.archiveFile)
     testDistribution = "ARCHIVE"