Route analytics queries by index setting, not table-name prefix

Today `RestUnifiedQueryAction.isAnalyticsIndex` dispatches to the analytics
engine when the source index name starts with `parquet_`. That's brittle —
it conflates naming convention with storage type. An index created without
the prefix but with pluggable dataformat enabled is silently sent to the
Lucene path; an index named `parquet_foo` without the setting is
mis-dispatched to analytics.

Use the authoritative signal instead: the `index.pluggable.dataformat.enabled`
setting on cluster-state metadata. This is the same setting integration tests
(`CoordinatorReduceIT`, `CompositeCommitDeletionIT`, etc.) already use to
create analytics-backed indices, and it's what `FieldStorageResolver` reads
to decide field-level storage.

Behavior:
- `index.pluggable.dataformat.enabled=true`  → analytics engine (DataFusion)
- flag absent / false / index missing         → Calcite→OpenSearch DSL path

Signed-off-by: bowenlan-amzn <bowenlan23@gmail.com>

Author: bowenlan-amzn

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

@@ -72,9 +72,14 @@ public RestUnifiedQueryAction(
     this.pluginSettings = pluginSettings;
   }
 
+  static final String PLUGGABLE_DATAFORMAT_ENABLED_SETTING = "index.pluggable.dataformat.enabled";
+
   /**
-   * Check if the query targets an analytics engine index (e.g., Parquet-backed). Uses the context's
-   * parser for index name extraction, supporting both PPL and SQL.
+   * Check if the query targets an analytics-engine-backed index. Parses the query to extract the
+   * source index name, then reads the cluster-state metadata for that index to see if {@code
+   * index.pluggable.dataformat.enabled} is set. When the flag is on, the index is backed by a
+   * pluggable columnar format (e.g. Parquet) and benefits from DataFusion execution; otherwise the
+   * query falls through to the sql-plugin's Calcite→OpenSearch DSL path.
    *
    * <p>Note: This creates a separate UnifiedQueryContext for parsing. The context cannot be shared
    * with doExecute/doExplain because UnifiedQueryContext holds a Calcite JDBC connection that fails
@@ -87,18 +92,27 @@ public boolean isAnalyticsIndex(String query, QueryType queryType) {
     }
     try (UnifiedQueryContext context = buildParsingContext(queryType)) {
       return extractIndexName(query, queryType, context)
-          .map(
-              indexName -> {
-                int lastDot = indexName.lastIndexOf('.');
-                return lastDot >= 0 ? indexName.substring(lastDot + 1) : indexName;
-              })
-          .map(tableName -> tableName.startsWith("parquet_"))
+          .map(this::stripSchemaPrefix)
+          .map(this::isPluggableDataformatIndex)
           .orElse(false);
     } catch (Exception e) {
       return false;
     }
   }
 
+  private String stripSchemaPrefix(String indexName) {
+    int lastDot = indexName.lastIndexOf('.');
+    return lastDot >= 0 ? indexName.substring(lastDot + 1) : indexName;
+  }
+
+  private boolean isPluggableDataformatIndex(String indexName) {
+    var indexMetadata = clusterService.state().metadata().index(indexName);
+    if (indexMetadata == null) {
+      return false;
+    }
+    return indexMetadata.getSettings().getAsBoolean(PLUGGABLE_DATAFORMAT_ENABLED_SETTING, false);
+  }
+
   /** Execute a query through the unified query pipeline on the sql-worker thread pool. */
   public void execute(
       String query,

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

@@ -8,44 +8,87 @@
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
 
 import org.apache.calcite.rel.RelNode;
 import org.junit.Before;
 import org.junit.Test;
 import org.opensearch.analytics.exec.QueryPlanExecutor;
+import org.opensearch.cluster.ClusterState;
+import org.opensearch.cluster.metadata.IndexMetadata;
+import org.opensearch.cluster.metadata.Metadata;
 import org.opensearch.cluster.service.ClusterService;
-import org.opensearch.sql.common.setting.Settings;
+import org.opensearch.common.settings.Settings;
 import org.opensearch.sql.executor.QueryType;
 import org.opensearch.transport.client.node.NodeClient;
 
 /**
- * Tests for analytics index routing in RestUnifiedQueryAction. Uses context parser for AST-based
- * index name extraction.
+ * Tests for analytics index routing in RestUnifiedQueryAction. Routing is driven by the {@code
+ * index.pluggable.dataformat.enabled} index setting, read from cluster state.
  */
 public class RestUnifiedQueryActionTest {
 
+  private ClusterService clusterService;
+  private Metadata metadata;
   private RestUnifiedQueryAction action;
 
   @Before
   public void setUp() {
+    clusterService = mock(ClusterService.class);
+    ClusterState clusterState = mock(ClusterState.class);
+    metadata = mock(Metadata.class);
+    when(clusterService.state()).thenReturn(clusterState);
+    when(clusterState.metadata()).thenReturn(metadata);
+
     @SuppressWarnings("unchecked")
     QueryPlanExecutor<RelNode, Iterable<Object[]>> executor = mock(QueryPlanExecutor.class);
     action =
         new RestUnifiedQueryAction(
-            mock(NodeClient.class), mock(ClusterService.class), executor, mock(Settings.class));
+            mock(NodeClient.class),
+            clusterService,
+            executor,
+            mock(org.opensearch.sql.common.setting.Settings.class));
   }
 
   @Test
-  public void parquetIndexRoutesToAnalytics() {
+  public void pluggableDataformatIndexRoutesToAnalytics() {
+    registerIndex(
+        "parquet_logs", Settings.builder().put("index.pluggable.dataformat.enabled", true).build());
+
     assertTrue(action.isAnalyticsIndex("source = parquet_logs | fields ts", QueryType.PPL));
     assertTrue(
         action.isAnalyticsIndex("source = opensearch.parquet_logs | fields ts", QueryType.PPL));
   }
 
   @Test
-  public void nonParquetIndexRoutesToLucene() {
-    assertFalse(action.isAnalyticsIndex("source = my_logs | fields ts", QueryType.PPL));
+  public void indexWithoutSettingRoutesToLucene() {
+    registerIndex("plain_logs", Settings.EMPTY);
+
+    assertFalse(action.isAnalyticsIndex("source = plain_logs | fields ts", QueryType.PPL));
+  }
+
+  @Test
+  public void indexWithSettingFalseRoutesToLucene() {
+    registerIndex(
+        "plain_logs", Settings.builder().put("index.pluggable.dataformat.enabled", false).build());
+
+    assertFalse(action.isAnalyticsIndex("source = plain_logs | fields ts", QueryType.PPL));
+  }
+
+  @Test
+  public void missingIndexRoutesToLucene() {
+    assertFalse(action.isAnalyticsIndex("source = does_not_exist | fields ts", QueryType.PPL));
+  }
+
+  @Test
+  public void nullAndEmptyQueriesRouteToLucene() {
     assertFalse(action.isAnalyticsIndex(null, QueryType.PPL));
     assertFalse(action.isAnalyticsIndex("", QueryType.PPL));
   }
+
+  private void registerIndex(String name, Settings settings) {
+    IndexMetadata indexMetadata = mock(IndexMetadata.class);
+    when(indexMetadata.getSettings()).thenReturn(settings);
+    when(metadata.index(name)).thenReturn(indexMetadata);
+  }
 }