Gate analytics routing on cluster.pluggable.dataformat.enabled

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

@@ -60,6 +60,15 @@ public class RestUnifiedQueryAction {
   private final org.opensearch.analytics.EngineContextProvider contextProvider;
   private final org.opensearch.sql.common.setting.Settings pluginSettings;
 
+  /**
+   * Cached value of {@link IndicesService#CLUSTER_PLUGGABLE_DATAFORMAT_ENABLED_SETTING}, kept
+   * current by a settings-update consumer. Reading {@code clusterService.getSettings()} would only
+   * ever return the node's startup settings and miss dynamic cluster-settings updates, so the live
+   * value is tracked here. Written from the cluster-state applier thread, read from the transport
+   * thread.
+   */
+  private volatile boolean clusterDataformatEnabled;
+
   public RestUnifiedQueryAction(
       NodeClient client,
       ClusterService clusterService,
@@ -71,6 +80,14 @@ public RestUnifiedQueryAction(
     this.analyticsEngine = new AnalyticsExecutionEngine(planExecutor);
     this.contextProvider = contextProvider;
     this.pluginSettings = pluginSettings;
+    this.clusterDataformatEnabled =
+        IndicesService.CLUSTER_PLUGGABLE_DATAFORMAT_ENABLED_SETTING.get(
+            clusterService.getSettings());
+    clusterService
+        .getClusterSettings()
+        .addSettingsUpdateConsumer(
+            IndicesService.CLUSTER_PLUGGABLE_DATAFORMAT_ENABLED_SETTING,
+            newValue -> this.clusterDataformatEnabled = newValue);
   }
 
   /**
@@ -88,15 +105,12 @@ public boolean isAnalyticsIndex(String query, QueryType queryType) {
     if (query == null || query.isEmpty()) {
       return false;
     }
-    // Cluster-level opt-in: when `cluster.pluggable.dataformat="composite"`, new indices
-    // inherit `index.pluggable.dataformat="composite"` at creation (see
-    // MetadataCreateIndexService), so every queryable target is analytics-eligible. Skip
-    // the per-index lookup — it doesn't work for aliases, wildcards, comma-lists, or data
-    // streams (Metadata#index() only resolves concrete names).
-    if ("composite"
-        .equals(
-            IndicesService.CLUSTER_PLUGGABLE_DATAFORMAT_VALUE_SETTING.get(
-                clusterService.getSettings()))) {
+    // Cluster-level opt-in: when `cluster.pluggable.dataformat.enabled=true`, new indices
+    // inherit the composite dataformat at creation (see MetadataCreateIndexService), so every
+    // queryable target is analytics-eligible. Skip the per-index lookup — it doesn't work for
+    // aliases, wildcards, comma-lists, or data streams (Metadata#index() only resolves concrete
+    // names).
+    if (clusterDataformatEnabled) {
       // Analytics engine can't serve system catalog; SHOW/DESCRIBE fall back to default pipeline
       try (UnifiedQueryContext context = buildParsingContext(queryType)) {
         boolean systemCatalog =

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

@@ -42,19 +42,23 @@ public void setUp() {
     metadata = mock(Metadata.class);
     when(clusterService.state()).thenReturn(clusterState);
     when(clusterState.metadata()).thenReturn(metadata);
-    // isAnalyticsIndex short-circuits on the cluster.pluggable.dataformat setting; the per-index
-    // path is only exercised when this returns something other than "composite".
+    // isAnalyticsIndex short-circuits on cluster.pluggable.dataformat.enabled; the per-index path
+    // is only exercised when the cluster-level flag is false. The flag is read once at construction
+    // from clusterService.getSettings(), so enabling it requires rebuilding the action (see
+    // enableClusterDataformat).
     when(clusterService.getSettings()).thenReturn(Settings.EMPTY);
+    action = buildAction();
+  }
 
+  private RestUnifiedQueryAction buildAction() {
     @SuppressWarnings("unchecked")
     QueryPlanExecutor<RelNode, Iterable<Object[]>> executor = mock(QueryPlanExecutor.class);
-    action =
-        new RestUnifiedQueryAction(
-            mock(NodeClient.class),
-            clusterService,
-            executor,
-            mock(EngineContextProvider.class),
-            mock(org.opensearch.sql.common.setting.Settings.class));
+    return new RestUnifiedQueryAction(
+        mock(NodeClient.class),
+        clusterService,
+        executor,
+        mock(EngineContextProvider.class),
+        mock(org.opensearch.sql.common.setting.Settings.class));
   }
 
   @Test
@@ -230,13 +234,17 @@ public void pplUnparseableQueryRoutesToAnalyticsUnderClusterComposite() {
     assertTrue(action.isAnalyticsIndex("source = parquet_logs | | fields ts", QueryType.PPL));
   }
 
+  /**
+   * Turns on the cluster-level opt-in by putting {@code cluster.pluggable.dataformat.enabled=true}
+   * into the node settings and rebuilding the action, since the flag is cached at construction.
+   */
   private void enableClusterComposite() {
     when(clusterService.getSettings())
         .thenReturn(
             Settings.builder()
-                .put(
-                    IndicesService.CLUSTER_PLUGGABLE_DATAFORMAT_VALUE_SETTING.getKey(), "composite")
+                .put(IndicesService.CLUSTER_PLUGGABLE_DATAFORMAT_ENABLED_SETTING.getKey(), true)
                 .build());
+    action = buildAction();
   }
 
   private void registerIndex(String name, Settings settings) {