Re-base PPL coverage bundle wiring onto main

Hand-port the analytics-engine compatibility test toggles from
the original ahkcs/feature/ppl-coverage-bundle branch (which was
based on the now-merged feature/mustang-ppl-integration). Most of the
bundle's commits were squash-merged into main via opensearch-project#5430; what remains
are the test-infra toggles for force-routing every IT through the
analytics-engine path with parquet-backed indices.

Changes:

- Add CALCITE_ANALYTICS_FORCE_ROUTING cluster setting key in
  common Settings + register it in OpenSearchSettings.
- RestUnifiedQueryAction.isAnalyticsIndex() short-circuits to true
  when the setting is on (the test-only knob; default false).
- PPLIntegTestCase.init() reads -Dtests.analytics.force_routing and
  flips the cluster setting on at setup. Provides static helpers
  enable/disableAnalyticsForceRouting().
- TestUtils.createIndexByRestClient honors -Dtests.analytics.parquet_indices
  by switching every test-created index to a single-shard composite/parquet
  store. Bulk loads on those indices use refresh=true (workaround for
  LuceneCommitter.getSafeCommitInfo TODO that hangs refresh=wait_for).

Run via:
  ./gradlew :integ-test:integTestRemote \
    -Dtests.rest.cluster=localhost:9200 \
    -Dtests.cluster=localhost:9300 \
    -Dtests.clustername=runTask \
    -Dtests.analytics.force_routing=true \
    -Dtests.analytics.parquet_indices=true \
    --tests "<class>"

Signed-off-by: Kai Huang <huangkaics@gmail.com>

Author: ahkcs

common/src/main/java/org/opensearch/sql/common/setting/Settings.java

@@ -45,6 +45,15 @@ public enum Key {
         "plugins.calcite.pushdown.rowcount.estimation.factor"),
     CALCITE_SUPPORT_ALL_JOIN_TYPES("plugins.calcite.all_join_types.allowed"),
 
+    /**
+     * Force-route every PPL/SQL query through the analytics engine path, regardless of the index
+     * name pattern. Test-only knob used by the analytics compatibility report; should remain {@code
+     * false} in production. When {@code true}, {@link
+     * org.opensearch.sql.plugin.rest.RestUnifiedQueryAction#isAnalyticsIndex} returns {@code true}
+     * unconditionally.
+     */
+    CALCITE_ANALYTICS_FORCE_ROUTING("plugins.calcite.analytics.force_routing"),
+
     /** Query Settings. */
     FIELD_TYPE_TOLERANCE("plugins.query.field_type_tolerance"),
 

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

@@ -37,6 +37,16 @@ public class TestUtils {
 
   private static final String MAPPING_FILE_PATH = "src/test/resources/indexDefinitions/";
 
+  /**
+   * System property that makes every test-created index parquet-backed (composite primary data
+   * format = parquet) with a single shard. Set by the analytics-engine compatibility test path so
+   * {@link org.opensearch.sql.ppl.PPLIntegTestCase}'s force-routing flag (which routes every query
+   * through the analytics-engine planner) can actually find a backend that supports the underlying
+   * storage. Without parquet, the analytics planner fails with "No backend can scan all requested
+   * fields..." regardless of the query.
+   */
+  public static final String ANALYTICS_PARQUET_INDICES_PROP = "tests.analytics.parquet_indices";
+
   /**
    * Create test index by REST client.
    *
@@ -48,6 +58,9 @@ public static void createIndexByRestClient(RestClient client, String indexName,
     Request request = new Request("PUT", "/" + indexName);
     JSONObject jsonObject = isNullOrEmpty(mapping) ? new JSONObject() : new JSONObject(mapping);
     setZeroReplicas(jsonObject);
+    if (Boolean.parseBoolean(System.getProperty(ANALYTICS_PARQUET_INDICES_PROP, "false"))) {
+      makeParquetBacked(jsonObject);
+    }
     request.setJsonEntity(jsonObject.toString());
     performRequest(client, request);
   }
@@ -69,6 +82,28 @@ private static void setZeroReplicas(JSONObject jsonObject) {
     jsonObject.put("settings", settings);
   }
 
+  /**
+   * Switches the test index to a parquet-backed composite store with a single shard so the
+   * analytics-engine path has a backend that can scan it. Mirrors the per-IT setup pattern in
+   * {@code
+   * sandbox/plugins/analytics-backend-datafusion/src/internalClusterTest/.../CoordinatorReduceIT.java}
+   * but applied at the REST-test layer. No-op for tests that go through the legacy / Calcite path
+   * against Lucene indices.
+   */
+  private static void makeParquetBacked(JSONObject jsonObject) {
+    JSONObject settings =
+        jsonObject.has("settings") ? jsonObject.getJSONObject("settings") : new JSONObject();
+    JSONObject indexSettings =
+        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());
+    settings.put("index", indexSettings);
+    jsonObject.put("settings", settings);
+  }
+
   /**
    * https://github.com/elastic/elasticsearch/pull/49959<br>
    * Deprecate creation of dot-prefixed index names except for hidden and system indices. Create
@@ -116,8 +151,17 @@ public static boolean isIndexExist(RestClient client, String indexName) {
   public static void loadDataByRestClient(
       RestClient client, String indexName, String dataSetFilePath) throws IOException {
     Path path = Paths.get(getResourceFilePath(dataSetFilePath));
-    Request request =
-        new Request("POST", "/" + indexName + "/_bulk?refresh=wait_for&wait_for_active_shards=all");
+    // Workaround: parquet-backed indices in the analytics-backend-lucene composite engine
+    // do not yet implement LuceneCommitter.getSafeCommitInfo (UnsupportedOperationException
+    // "TODO:: with index deleter"), which hangs refresh=wait_for until the test framework
+    // request timeout (~60s). Force-refresh sidesteps the safe-commit-info path while still
+    // making the bulk-loaded docs immediately searchable. Drop this branch once
+    // LuceneCommitter.getSafeCommitInfo is implemented.
+    String refreshParam =
+        Boolean.parseBoolean(System.getProperty(ANALYTICS_PARQUET_INDICES_PROP, "false"))
+            ? "refresh=true"
+            : "refresh=wait_for&wait_for_active_shards=all";
+    Request request = new Request("POST", "/" + indexName + "/_bulk?" + refreshParam);
     request.setJsonEntity(new String(Files.readAllBytes(path)));
     performRequest(client, request);
   }

integ-test/src/test/java/org/opensearch/sql/ppl/PPLIntegTestCase.java

@@ -42,11 +42,38 @@ public abstract class PPLIntegTestCase extends SQLIntegTestCase {
   public static final Integer DEFAULT_SUBSEARCH_MAXOUT = 10000;
   public static final Integer DEFAULT_JOIN_SUBSEARCH_MAXOUT = 50000;
 
+  /**
+   * System property toggle for the analytics compatibility report. When set, every PPL IT base
+   * setup forces the analytics routing flag on so all queries traverse the analytics-engine path.
+   * Off by default — normal CI runs untouched.
+   */
+  public static final String ANALYTICS_FORCE_ROUTING_PROP = "tests.analytics.force_routing";
+
   @Override
   protected void init() throws Exception {
     super.init();
     updatePushdownSettings();
     disableCalcite(); // calcite is enabled by default from 3.3.0
+    if (Boolean.parseBoolean(System.getProperty(ANALYTICS_FORCE_ROUTING_PROP, "false"))) {
+      enableCalcite();
+      enableAnalyticsForceRouting();
+    }
+  }
+
+  /**
+   * Force every PPL/SQL query to route through the analytics-engine path regardless of the index
+   * name pattern. Used by the analytics compatibility report.
+   */
+  public static void enableAnalyticsForceRouting() throws IOException {
+    updateClusterSettings(
+        new SQLIntegTestCase.ClusterSetting(
+            "persistent", Settings.Key.CALCITE_ANALYTICS_FORCE_ROUTING.getKeyValue(), "true"));
+  }
+
+  public static void disableAnalyticsForceRouting() throws IOException {
+    updateClusterSettings(
+        new SQLIntegTestCase.ClusterSetting(
+            "persistent", Settings.Key.CALCITE_ANALYTICS_FORCE_ROUTING.getKeyValue(), "false"));
   }
 
   protected JSONObject executeQuery(String query) throws IOException {

opensearch/src/main/java/org/opensearch/sql/opensearch/setting/OpenSearchSettings.java

@@ -186,6 +186,13 @@ public class OpenSearchSettings extends Settings {
           Setting.Property.NodeScope,
           Setting.Property.Dynamic);
 
+  public static final Setting<Boolean> CALCITE_ANALYTICS_FORCE_ROUTING_SETTING =
+      Setting.boolSetting(
+          Key.CALCITE_ANALYTICS_FORCE_ROUTING.getKeyValue(),
+          false,
+          Setting.Property.NodeScope,
+          Setting.Property.Dynamic);
+
   public static final Setting<?> QUERY_MEMORY_LIMIT_SETTING =
       Setting.memorySizeSetting(
           Key.QUERY_MEMORY_LIMIT.getKeyValue(),
@@ -467,6 +474,12 @@ public OpenSearchSettings(ClusterSettings clusterSettings) {
         Key.CALCITE_SUPPORT_ALL_JOIN_TYPES,
         CALCITE_SUPPORT_ALL_JOIN_TYPES_SETTING,
         new Updater(Key.CALCITE_SUPPORT_ALL_JOIN_TYPES));
+    register(
+        settingBuilder,
+        clusterSettings,
+        Key.CALCITE_ANALYTICS_FORCE_ROUTING,
+        CALCITE_ANALYTICS_FORCE_ROUTING_SETTING,
+        new Updater(Key.CALCITE_ANALYTICS_FORCE_ROUTING));
     register(
         settingBuilder,
         clusterSettings,
@@ -658,6 +671,7 @@ public static List<Setting<?>> pluginSettings() {
         .add(CALCITE_PUSHDOWN_ENABLED_SETTING)
         .add(CALCITE_PUSHDOWN_ROWCOUNT_ESTIMATION_FACTOR_SETTING)
         .add(CALCITE_SUPPORT_ALL_JOIN_TYPES_SETTING)
+        .add(CALCITE_ANALYTICS_FORCE_ROUTING_SETTING)
         .add(DEFAULT_PATTERN_METHOD_SETTING)
         .add(DEFAULT_PATTERN_MODE_SETTING)
         .add(DEFAULT_PATTERN_MAX_SAMPLE_COUNT_SETTING)

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

@@ -37,6 +37,7 @@
 import org.opensearch.sql.calcite.CalcitePlanContext;
 import org.opensearch.sql.calcite.plan.rel.LogicalSystemLimit;
 import org.opensearch.sql.common.response.ResponseListener;
+import org.opensearch.sql.common.setting.Settings;
 import org.opensearch.sql.executor.ExecutionEngine.QueryResponse;
 import org.opensearch.sql.executor.QueryType;
 import org.opensearch.sql.executor.analytics.AnalyticsExecutionEngine;
@@ -88,6 +89,10 @@ public boolean isAnalyticsIndex(String query, QueryType queryType) {
     if (query == null || query.isEmpty()) {
       return false;
     }
+    if (Boolean.TRUE.equals(
+        pluginSettings.getSettingValue(Settings.Key.CALCITE_ANALYTICS_FORCE_ROUTING))) {
+      return true;
+    }
     try (UnifiedQueryContext context = buildParsingContext(queryType)) {
       return extractIndexName(query, queryType, context)
           .map(this::stripSchemaPrefix)