fix(plugin): keep system-catalog queries on default pipeline under cluster composite

When cluster.pluggable.dataformat=composite, isAnalyticsIndex routed every query to the analytics engine, which cannot serve the system catalog (*_ODFE_SYS_TABLE, .DATASOURCES) that SHOW/DESCRIBE resolve. Detect system-catalog queries and keep them on the default pipeline while data queries continue to the analytics engine. Also log query routing to the analytics engine at both call sites.

Signed-off-by: Chen Dai <daichen@amazon.com>

Author: dai-chen

plugin/src/main/java/org/opensearch/sql/plugin/SQLPlugin.java

@@ -63,6 +63,7 @@
 import org.opensearch.script.ScriptService;
 import org.opensearch.sql.ast.statement.ExplainMode;
 import org.opensearch.sql.common.response.ResponseListener;
+import org.opensearch.sql.common.utils.QueryContext;
 import org.opensearch.sql.datasource.DataSourceService;
 import org.opensearch.sql.datasources.auth.DataSourceUserAuthorizationHelper;
 import org.opensearch.sql.datasources.auth.DataSourceUserAuthorizationHelperImpl;
@@ -241,6 +242,7 @@ private BiFunction<SQLQueryRequest, RestChannel, Boolean> createSqlAnalyticsRout
           || !unifiedQueryHandler.isAnalyticsIndex(sqlRequest.getQuery(), QueryType.SQL)) {
         return false;
       }
+      LOGGER.info("[{}] Routing SQL query to analytics engine", QueryContext.getRequestId());
       if (sqlRequest.isExplainRequest()) {
         unifiedQueryHandler.explain(
             sqlRequest.getQuery(),

plugin/src/main/java/org/opensearch/sql/plugin/rest/RestUnifiedQueryAction.java

@@ -40,6 +40,7 @@
 import org.opensearch.sql.protocol.response.QueryResult;
 import org.opensearch.sql.protocol.response.format.ResponseFormatter;
 import org.opensearch.sql.protocol.response.format.SimpleJsonResponseFormatter;
+import org.opensearch.sql.utils.SystemIndexUtils;
 import org.opensearch.transport.client.node.NodeClient;
 
 /**
@@ -95,7 +96,17 @@ public boolean isAnalyticsIndex(String query, QueryType queryType) {
         .equals(
             IndicesService.CLUSTER_PLUGGABLE_DATAFORMAT_VALUE_SETTING.get(
                 clusterService.getSettings()))) {
-      return true;
+      // Analytics engine can't serve system catalog; SHOW/DESCRIBE fall back to default pipeline
+      try (UnifiedQueryContext context = buildParsingContext(queryType)) {
+        boolean systemCatalog =
+            extractIndexName(query, queryType, context)
+                .map(RestUnifiedQueryAction::isSystemCatalog)
+                .orElse(false);
+        return !systemCatalog;
+      } catch (Exception e) {
+        // Let the analytics engine re-parse and surface the syntax error
+        return true;
+      }
     }
     try (UnifiedQueryContext context = buildParsingContext(queryType)) {
       return extractIndexName(query, queryType, context)
@@ -107,6 +118,11 @@ public boolean isAnalyticsIndex(String query, QueryType queryType) {
     }
   }
 
+  private static boolean isSystemCatalog(String name) {
+    return SystemIndexUtils.isSystemIndex(name)
+        || SystemIndexUtils.DATASOURCES_TABLE_NAME.equals(name);
+  }
+
   private String stripSchemaPrefix(String indexName) {
     int lastDot = indexName.lastIndexOf('.');
     return lastDot >= 0 ? indexName.substring(lastDot + 1) : indexName;

plugin/src/main/java/org/opensearch/sql/plugin/transport/TransportPPLQueryAction.java

@@ -14,6 +14,8 @@
 import java.util.Optional;
 import java.util.function.Supplier;
 import org.apache.calcite.rel.RelNode;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 import org.opensearch.action.ActionRequest;
 import org.opensearch.action.support.ActionFilters;
 import org.opensearch.action.support.HandledTransportAction;
@@ -59,6 +61,8 @@
 public class TransportPPLQueryAction
     extends HandledTransportAction<ActionRequest, TransportPPLQueryResponse> {
 
+  private static final Logger LOG = LogManager.getLogger(TransportPPLQueryAction.class);
+
   private final Injector injector;
 
   private final Supplier<Boolean> pplEnabled;
@@ -171,6 +175,7 @@ protected void doExecute(
     // Route to analytics engine for non-Lucene (e.g., Parquet-backed) indices.
     if (unifiedQueryHandler != null
         && unifiedQueryHandler.isAnalyticsIndex(transformedRequest.getRequest(), QueryType.PPL)) {
+      LOG.info("[{}] Routing PPL query to analytics engine", QueryContext.getRequestId());
       if (transformedRequest.isExplainRequest()) {
         unifiedQueryHandler.explain(
             transformedRequest.getRequest(),

plugin/src/test/java/org/opensearch/sql/plugin/rest/RestUnifiedQueryActionTest.java

@@ -21,6 +21,7 @@
 import org.opensearch.cluster.service.ClusterService;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.index.IndexSettings;
+import org.opensearch.indices.IndicesService;
 import org.opensearch.sql.executor.QueryType;
 import org.opensearch.transport.client.node.NodeClient;
 
@@ -142,7 +143,7 @@ public void nullAndEmptyQueriesRouteToLucene() {
   }
 
   @Test
-  public void showStatementRoutesToLucene() {
+  public void showStatementNotRoutedToAnalyticsEngine() {
     registerIndex(
         "parquet_logs",
         Settings.builder()
@@ -154,7 +155,7 @@ public void showStatementRoutesToLucene() {
   }
 
   @Test
-  public void describeStatementRoutesToLucene() {
+  public void describeStatementNotRoutedToAnalyticsEngine() {
     registerIndex(
         "parquet_logs",
         Settings.builder()
@@ -165,6 +166,65 @@ public void describeStatementRoutesToLucene() {
     assertFalse(action.isAnalyticsIndex("DESCRIBE TABLES LIKE 'parquet_logs'", QueryType.SQL));
   }
 
+  @Test
+  public void showStatementNotRoutedToAnalyticsEngineUnderClusterComposite() {
+    enableClusterComposite();
+    assertFalse(action.isAnalyticsIndex("SHOW TABLES LIKE 'parquet_logs'", QueryType.SQL));
+  }
+
+  @Test
+  public void describeStatementNotRoutedToAnalyticsEngineUnderClusterComposite() {
+    enableClusterComposite();
+    assertFalse(action.isAnalyticsIndex("DESCRIBE TABLES LIKE 'parquet_logs'", QueryType.SQL));
+  }
+
+  @Test
+  public void dataQueryStillRoutesToAnalyticsUnderClusterComposite() {
+    enableClusterComposite();
+    assertTrue(action.isAnalyticsIndex("SELECT * FROM parquet_logs", QueryType.SQL));
+  }
+
+  @Test
+  public void unparseableQueryRoutesToAnalyticsUnderClusterComposite() {
+    enableClusterComposite();
+    // malformed -> AE re-parses & reports
+    assertTrue(action.isAnalyticsIndex("SELECT FROM WHERE", QueryType.SQL));
+  }
+
+  @Test
+  public void pplDescribeNotRoutedToAnalyticsEngineUnderClusterComposite() {
+    enableClusterComposite();
+    assertFalse(action.isAnalyticsIndex("describe parquet_logs", QueryType.PPL));
+  }
+
+  @Test
+  public void pplShowDatasourcesNotRoutedToAnalyticsEngineUnderClusterComposite() {
+    enableClusterComposite();
+    assertFalse(action.isAnalyticsIndex("show datasources", QueryType.PPL));
+  }
+
+  @Test
+  public void pplDataQueryStillRoutesToAnalyticsUnderClusterComposite() {
+    enableClusterComposite();
+    assertTrue(action.isAnalyticsIndex("source = parquet_logs | fields ts", QueryType.PPL));
+  }
+
+  @Test
+  public void pplUnparseableQueryRoutesToAnalyticsUnderClusterComposite() {
+    enableClusterComposite();
+    // malformed -> AE re-parses & reports
+    assertTrue(action.isAnalyticsIndex("source = parquet_logs | | fields ts", QueryType.PPL));
+  }
+
+  private void enableClusterComposite() {
+    when(clusterService.getSettings())
+        .thenReturn(
+            Settings.builder()
+                .put(
+                    IndicesService.CLUSTER_PLUGGABLE_DATAFORMAT_VALUE_SETTING.getKey(), "composite")
+                .build());
+  }
+
   private void registerIndex(String name, Settings settings) {
     IndexMetadata indexMetadata = mock(IndexMetadata.class);
     when(indexMetadata.getSettings()).thenReturn(settings);