Parquet-back raw-PUT test indices on the analytics-engine route

With -Dtests.analytics.parquet_indices=true, indices created by a raw
document PUT (e.g. `PUT /test/_doc/1` in a test's init()) bypass
AnalyticsIndexConfig.applyIndexCreationSettings, so they inherit the
composite *value* — and are therefore routed to the analytics engine by
RestUnifiedQueryAction.isAnalyticsIndex — but not the
`pluggable.dataformat.enabled` flag. They are then stored as a
plain-Lucene EngineBackedIndexer whose acquireReader() is unimplemented,
and the query fails with
`StreamException[INTERNAL] Failed to start streaming fragment`.

Apply the cluster-level composite defaults in setUpIndices() so every
index — including raw-PUT ones — is stored as a parquet-backed
DataFormatAwareEngine that is actually scannable by the analytics engine
it routes to. No-op unless tests.analytics.parquet_indices=true, so
normal CI is unchanged.

Signed-off-by: Kai Huang <ahkcs@amazon.com>

Author: ahkcs

integ-test/src/test/java/org/opensearch/sql/legacy/SQLIntegTestCase.java

@@ -66,6 +66,12 @@ public void setUpIndices() throws Exception {
       initClient();
     }
 
+    // When -Dtests.analytics.parquet_indices=true, make every index (including ones a test
+    // auto-creates via a raw document PUT, which bypasses createIndexByRestClient) parquet-backed
+    // composite, so it is stored as a DataFormatAwareEngine and is actually scannable by the
+    // analytics engine it routes to. Must run before init() creates any index.
+    TestUtils.AnalyticsIndexConfig.applyClusterSettings(client());
+
     if (shouldResetQuerySizeLimit()) {
       resetQuerySizeLimit();
     }

integ-test/src/test/java/org/opensearch/sql/legacy/TestUtils.java

@@ -23,6 +23,7 @@
 import java.util.List;
 import java.util.Locale;
 import java.util.stream.Collectors;
+import org.json.JSONArray;
 import org.json.JSONObject;
 import org.opensearch.action.bulk.BulkRequest;
 import org.opensearch.action.bulk.BulkResponse;
@@ -68,6 +69,11 @@ public static boolean isEnabled() {
       return Boolean.parseBoolean(System.getProperty(ENABLED_PROP, "false"));
     }
 
+    // Composite-store format values shared by the index-level and cluster-level settings below.
+    private static final String DATAFORMAT_COMPOSITE = "composite";
+    private static final String PRIMARY_FORMAT_PARQUET = "parquet";
+    private static final String SECONDARY_FORMAT_LUCENE = "lucene";
+
     /**
      * Inject the parquet-backed composite-store index settings into {@code jsonObject}. No-op when
      * the config is disabled; idempotent — safe on any index-creation JSON shape.
@@ -82,13 +88,38 @@ static void applyIndexCreationSettings(JSONObject jsonObject) {
           settings.has("index") ? settings.getJSONObject("index") : new JSONObject();
       indexSettings.put("number_of_shards", 1);
       indexSettings.put("pluggable.dataformat.enabled", true);
-      indexSettings.put("pluggable.dataformat", "composite");
-      indexSettings.put("composite.primary_data_format", "parquet");
-      indexSettings.put("composite.secondary_data_formats", new org.json.JSONArray().put("lucene"));
+      indexSettings.put("pluggable.dataformat", DATAFORMAT_COMPOSITE);
+      indexSettings.put("composite.primary_data_format", PRIMARY_FORMAT_PARQUET);
+      indexSettings.put(
+          "composite.secondary_data_formats", new JSONArray().put(SECONDARY_FORMAT_LUCENE));
       settings.put("index", indexSettings);
       jsonObject.put("settings", settings);
     }
 
+    /**
+     * Set the composite-store defaults at the cluster level so even indices auto-created by a raw
+     * document {@code PUT} (which bypass {@link #applyIndexCreationSettings}) are parquet-backed.
+     * Otherwise such an index inherits only the composite value — so it routes to the analytics
+     * engine — but not the {@code .enabled} flag, leaving it stored as a plain-Lucene {@code
+     * EngineBackedIndexer} that fails at query time. No-op when disabled; idempotent.
+     */
+    public static void applyClusterSettings(RestClient client) {
+      if (!isEnabled()) {
+        return;
+      }
+      JSONObject persistent =
+          new JSONObject()
+              .put("cluster.pluggable.dataformat.enabled", true)
+              .put("cluster.pluggable.dataformat", DATAFORMAT_COMPOSITE)
+              .put("cluster.composite.primary_data_format", PRIMARY_FORMAT_PARQUET)
+              .put(
+                  "cluster.composite.secondary_data_formats",
+                  new JSONArray().put(SECONDARY_FORMAT_LUCENE));
+      Request request = new Request("PUT", "/_cluster/settings");
+      request.setJsonEntity(new JSONObject().put("persistent", persistent).toString());
+      performRequest(client, request);
+    }
+
     /**
      * Returns the {@code _bulk} refresh query string for the current index type. Parquet-backed
      * indices in the analytics-backend-lucene composite engine don't yet implement {@code