Stabilize order-dependent PPL ITs with explicit sort for multi-shard analytics runs

integ-test/build.gradle

@@ -1035,6 +1035,12 @@ task integTestRemote(type: RestIntegTestTask) {
         systemProperty 'tests.analytics.parquet_indices', System.getProperty("tests.analytics.parquet_indices")
     }
 
+    // Primary-shard count for analytics-backed test indices (default 1). Set to e.g. 3 to
+    // reproduce multi-shard coordination behavior on a single-node cluster.
+    if (System.getProperty("tests.analytics.num_shards") != null) {
+        systemProperty 'tests.analytics.num_shards', System.getProperty("tests.analytics.num_shards")
+    }
+
     // True only when the analytics-engine route is active (every test index parquet-backed). Matches
     // AnalyticsIndexConfig.isEnabled, which parses the value rather than checking mere presence, so a
     // `-Dtests.analytics.parquet_indices=false` run stays on the v2 path.

integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteBinCommandIT.java

@@ -85,10 +85,12 @@ public void testBinWithMinspan() throws IOException {
   public void testBinBasicFunctionality() throws IOException {
     JSONObject result =
         executeQuery(
-            String.format("source=%s | bin age span=5 | fields age | head 3", TEST_INDEX_ACCOUNT));
+            String.format(
+                "source=%s | bin age span=5 | sort account_number | fields age | head 3",
+                TEST_INDEX_ACCOUNT));
     verifySchema(result, schema("age", null, "string"));
 
-    verifyDataRows(result, rows("30-35"), rows("35-40"), rows("25-30"));
+    verifyDataRows(result, rows("25-30"), rows("30-35"), rows("20-25"));
   }
 
   @Test
@@ -195,7 +197,9 @@ public void testBinOnlyWithoutAggregation() throws IOException {
     JSONObject binOnlyResult =
         executeQuery(
             String.format(
-                "source=%s" + " | bin @timestamp span=4h" + " | fields `@timestamp` | head 3",
+                "source=%s"
+                    + " | bin @timestamp span=4h"
+                    + " | fields `@timestamp` | sort `@timestamp` | head 3",
                 TEST_INDEX_TIME_DATA));
 
     // Verify schema and that binning works correctly
@@ -235,7 +239,7 @@ public void testBinWithMonthlySpan() throws IOException {
         executeQuery(
             String.format(
                 "source=%s | bin @timestamp span=4mon as cate | fields"
-                    + " cate, @timestamp | head 5",
+                    + " cate, @timestamp | sort @timestamp | head 5",
                 TEST_INDEX_TIME_DATA));
     verifySchema(result, schema("cate", null, "string"), schema("@timestamp", null, "timestamp"));
 

integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteMVAppendFunctionIT.java

@@ -107,8 +107,8 @@ public void testMvappendWithRealFields() throws IOException {
         executeQuery(
             source(
                 TEST_INDEX_BANK,
-                "eval result = mvappend(firstname, lastname) | head 1 | fields firstname, lastname,"
-                    + " result"));
+                "eval result = mvappend(firstname, lastname) | sort account_number | head 1 |"
+                    + " fields firstname, lastname, result"));
 
     verifySchema(
         actual,
@@ -179,7 +179,7 @@ public void testMvappendInWhereClause() throws IOException {
             source(
                 TEST_INDEX_BANK,
                 "eval combined = mvappend(firstname, lastname) | where array_length(combined) = 2 |"
-                    + " head 1 | fields firstname, lastname, combined"));
+                    + " sort account_number | head 1 | fields firstname, lastname, combined"));
 
     verifySchema(
         actual,
@@ -198,8 +198,8 @@ public void testMvappendWithComplexExpression() throws IOException {
         executeQuery(
             source(
                 TEST_INDEX_BANK,
-                "eval result = mvappend(array(age), array(age * 2), age + 10) | head 1 | fields"
-                    + " age, result"));
+                "eval result = mvappend(array(age), array(age * 2), age + 10) | sort"
+                    + " account_number | head 1 | fields age, result"));
 
     verifySchema(actual, schema("age", "int"), schema("result", "array"));
     verifyDataRows(actual, rows(32, List.of(32, 64, 42)));

integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteMultisearchCommandIT.java

@@ -325,8 +325,10 @@ public void testMultisearchCrossIndexFieldSelection() throws IOException {
         executeQuery(
             String.format(
                 "| multisearch "
-                    + "[search source=%s | fields firstname, balance | head 2] "
-                    + "[search source=%s | fields description, place_id | head 2]",
+                    + "[search source=%s | sort account_number | fields firstname, balance"
+                    + " | head 2] "
+                    + "[search source=%s | sort place_id | fields description, place_id"
+                    + " | head 2]",
                 TEST_INDEX_ACCOUNT, TEST_INDEX_LOCATIONS_TYPE_CONFLICT));
 
     verifySchema(
@@ -338,8 +340,8 @@ public void testMultisearchCrossIndexFieldSelection() throws IOException {
 
     verifyDataRows(
         result,
+        rows("Bradshaw", 16623L, null, null),
         rows("Amber", 39225L, null, null),
-        rows("Hattie", 5686L, null, null),
         rows(null, null, "Central Park", 1001),
         rows(null, null, "Times Square", 1002));
   }

integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePPLAggregationIT.java

@@ -542,8 +542,8 @@ public void testCountByCustomTimeSpanWithDifferentUnits() throws IOException {
     JSONObject actual =
         executeQuery(
             String.format(
-                "source=%s | head 5 | stats count(datetime0) by span(datetime0, 15minute) as"
-                    + " datetime_span",
+                "source=%s | sort key | head 5 | stats count(datetime0) by span(datetime0,"
+                    + " 15minute) as datetime_span",
                 TEST_INDEX_CALCS));
     verifySchema(
         actual, schema("datetime_span", "timestamp"), schema("count(datetime0)", "bigint"));
@@ -558,8 +558,8 @@ public void testCountByCustomTimeSpanWithDifferentUnits() throws IOException {
     actual =
         executeQuery(
             String.format(
-                "source=%s | head 5 | stats count(datetime0) by span(datetime0, 5second) as"
-                    + " datetime_span",
+                "source=%s | sort key | head 5 | stats count(datetime0) by span(datetime0,"
+                    + " 5second) as datetime_span",
                 TEST_INDEX_CALCS));
     verifySchema(
         actual, schema("datetime_span", "timestamp"), schema("count(datetime0)", "bigint"));
@@ -574,8 +574,8 @@ public void testCountByCustomTimeSpanWithDifferentUnits() throws IOException {
     actual =
         executeQuery(
             String.format(
-                "source=%s | head 5 | stats count(datetime0) by span(datetime0, 3month) as"
-                    + " datetime_span",
+                "source=%s | sort key | head 5 | stats count(datetime0) by span(datetime0,"
+                    + " 3month) as datetime_span",
                 TEST_INDEX_CALCS));
     verifySchema(
         actual, schema("datetime_span", "timestamp"), schema("count(datetime0)", "bigint"));
@@ -587,8 +587,8 @@ public void testCountByNullableTimeSpan() throws IOException {
     JSONObject actual =
         executeQuery(
             String.format(
-                "source=%s | head 5 | stats count(datetime0), count(datetime1) by span(time1,"
-                    + " 15minute) as time_span",
+                "source=%s | sort key | head 5 | stats count(datetime0), count(datetime1) by"
+                    + " span(time1, 15minute) as time_span",
                 TEST_INDEX_CALCS));
     verifySchema(
         actual,

integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePPLEnhancedCoalesceIT.java

@@ -73,8 +73,8 @@ public void testCoalesceWithLiterals() throws IOException {
     JSONObject actual =
         executeQuery(
             String.format(
-                "source=%s | eval result = coalesce(name, 123, 'unknown') | fields name, result |"
-                    + " head 1",
+                "source=%s | eval result = coalesce(name, 123, 'unknown') | sort - age | fields"
+                    + " name, result | head 1",
                 TEST_INDEX_STATE_COUNTRY_WITH_NULL));
 
     verifySchema(actual, schema("name", "string"), schema("result", "string"));
@@ -100,8 +100,8 @@ public void testCoalesceWithMultipleFields() throws IOException {
     JSONObject actual =
         executeQuery(
             String.format(
-                "source=%s | eval result = coalesce(name, age, year, month) | fields name, age,"
-                    + " year, month, result | head 2",
+                "source=%s | eval result = coalesce(name, age, year, month) | sort - age | fields"
+                    + " name, age, year, month, result | head 2",
                 TEST_INDEX_STATE_COUNTRY_WITH_NULL));
 
     verifySchema(
@@ -139,8 +139,8 @@ public void testCoalesceWithNonExistentField() throws IOException {
     JSONObject actual =
         executeQuery(
             String.format(
-                "source=%s | eval result = coalesce(nonexistent_field, name) | fields name, result"
-                    + " | head 2",
+                "source=%s | eval result = coalesce(nonexistent_field, name) | sort - age"
+                    + " | fields name, result | head 2",
                 TEST_INDEX_STATE_COUNTRY_WITH_NULL));
 
     verifySchema(actual, schema("name", "string"), schema("result", "string"));
@@ -234,7 +234,8 @@ public void testCoalesceWithEmptyString() throws IOException {
     JSONObject actual =
         executeQuery(
             String.format(
-                "source=%s | eval result = coalesce('', name) | fields name, result | head 1",
+                "source=%s | eval result = coalesce('', name) | sort - age | fields name, result"
+                    + " | head 1",
                 TEST_INDEX_STATE_COUNTRY_WITH_NULL));
 
     verifySchema(actual, schema("name", "string"), schema("result", "string"));
@@ -247,7 +248,8 @@ public void testCoalesceWithSpaceString() throws IOException {
     JSONObject actual =
         executeQuery(
             String.format(
-                "source=%s | eval result = coalesce(' ', name) | fields name, result | head 1",
+                "source=%s | eval result = coalesce(' ', name) | sort - age | fields name, result"
+                    + " | head 1",
                 TEST_INDEX_STATE_COUNTRY_WITH_NULL));
 
     verifySchema(actual, schema("name", "string"), schema("result", "string"));
@@ -274,8 +276,8 @@ public void testCoalesceWithCompatibleNumericTypes() throws IOException {
     JSONObject actual =
         executeQuery(
             String.format(
-                "source=%s | eval result = coalesce(age, year, 999) | fields age, year, result |"
-                    + " head 2",
+                "source=%s | eval result = coalesce(age, year, 999) | sort - age | fields age,"
+                    + " year, result | head 2",
                 TEST_INDEX_STATE_COUNTRY_WITH_NULL));
 
     verifySchema(actual, schema("age", "int"), schema("year", "int"), schema("result", "int"));
@@ -288,7 +290,7 @@ public void testCoalesceTypeCoercionWithMixedTypes() throws IOException {
         executeQuery(
             String.format(
                 "source=%s | eval result = coalesce(nonexistent_field, age,"
-                    + " 'default') | fields age, result | head 2",
+                    + " 'default') | sort - age | fields age, result | head 2",
                 TEST_INDEX_STATE_COUNTRY_WITH_NULL));
 
     verifySchema(actual, schema("age", "int"), schema("result", "string"));
@@ -300,8 +302,8 @@ public void testCoalesceWithCompatibleNumericAndTemporalTypes() throws IOExcepti
     JSONObject actual =
         executeQuery(
             String.format(
-                "source=%s | eval result = coalesce(age, year, month) | fields age, year, month,"
-                    + " result | head 2",
+                "source=%s | eval result = coalesce(age, year, month) | sort - age | fields age,"
+                    + " year, month, result | head 2",
                 TEST_INDEX_STATE_COUNTRY_WITH_NULL));
 
     verifySchema(

integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePPLGrokIT.java

@@ -34,7 +34,8 @@ public void testGrokEmail() throws IOException {
         executeQuery(
             String.format(
                 Locale.ROOT,
-                "source = %s | grok email '%s' | head 3 | fields email, host",
+                "source = %s | grok email '%s' | sort account_number | head 3 | fields email,"
+                    + " host",
                 TEST_INDEX_BANK,
                 ".+@%{HOSTNAME:host}"));
     verifySchema(result, schema("email", "string"), schema("host", "string"));
@@ -49,7 +50,10 @@ public void testGrokEmail() throws IOException {
   public void testGrokAddressOverriding() throws IOException {
     JSONObject preGrokResult =
         executeQuery(
-            String.format(Locale.ROOT, "source = %s | head 3 | fields address", TEST_INDEX_BANK));
+            String.format(
+                Locale.ROOT,
+                "source = %s | sort account_number | head 3 | fields address",
+                TEST_INDEX_BANK));
     verifySchema(preGrokResult, schema("address", "string"));
     verifyDataRows(
         preGrokResult,
@@ -61,7 +65,7 @@ public void testGrokAddressOverriding() throws IOException {
         executeQuery(
             String.format(
                 Locale.ROOT,
-                "source = %s | grok address '%s' | head 3 | fields address",
+                "source = %s | grok address '%s' | sort account_number | head 3 | fields address",
                 TEST_INDEX_BANK,
                 "%{NUMBER} %{GREEDYDATA:address}"));
     verifySchema(result, schema("address", "string"));

integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePPLPatternsIT.java

@@ -37,7 +37,8 @@ public void testSimplePatternLabelMode_NotShowNumberedToken() throws IOException
     JSONObject result =
         executeQuery(
             String.format(
-                "source = %s | patterns email mode=label | head 1 | fields email, patterns_field",
+                "source = %s | patterns email mode=label | sort account_number | head 1 | fields"
+                    + " email, patterns_field",
                 TEST_INDEX_BANK));
     verifySchema(result, schema("email", "string"), schema("patterns_field", "string"));
     verifyDataRows(result, rows("amberduke@pyrami.com", "<*>@<*>.<*>"));
@@ -48,8 +49,8 @@ public void testSimplePatternLabelMode_ShowNumberedToken() throws IOException {
     JSONObject result =
         executeQuery(
             String.format(
-                "source = %s | patterns email mode=label show_numbered_token=true | head 1 | fields"
-                    + " email, patterns_field, tokens",
+                "source = %s | patterns email mode=label show_numbered_token=true | sort"
+                    + " account_number | head 1 | fields email, patterns_field, tokens",
                 TEST_INDEX_BANK));
     verifySchema(
         result,
@@ -100,7 +101,7 @@ public void testSimplePatternLabelModeWithCustomPattern_ShowNumberedToken() thro
         executeQuery(
             String.format(
                 "source = %s | patterns email mode=label show_numbered_token=true pattern='@.*' |"
-                    + " head 1 | fields email, patterns_field, tokens",
+                    + " sort account_number | head 1 | fields email, patterns_field, tokens",
                 TEST_INDEX_BANK));
     verifySchema(
         result,

integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePPLTrendlineIT.java

@@ -33,8 +33,8 @@ public void testTrendlineSma() throws IOException {
     JSONObject result =
         executeQuery(
             String.format(
-                "source=%s | where balance > 30000 | trendline sma(3, balance) as balance_trend |"
-                    + " fields balance_trend",
+                "source=%s | where balance > 30000 | sort account_number | trendline sma(3,"
+                    + " balance) as balance_trend | fields balance_trend",
                 TEST_INDEX_BANK));
     verifySchema(result, schema("balance_trend", "double"));
     verifyDataRows(
@@ -46,8 +46,8 @@ public void testTrendlineWma() throws IOException {
     JSONObject result =
         executeQuery(
             String.format(
-                "source=%s | where balance > 30000 | trendline wma(3, balance) as balance_trend |"
-                    + " fields balance_trend",
+                "source=%s | where balance > 30000 | sort account_number | trendline wma(3,"
+                    + " balance) as balance_trend | fields balance_trend",
                 TEST_INDEX_BANK));
     verifySchema(result, schema("balance_trend", "double"));
     verifyDataRows(
@@ -59,8 +59,8 @@ public void testTrendlineMultipleFields() throws Exception {
     JSONObject result =
         executeQuery(
             String.format(
-                "source=%s | where balance > 30000 | trendline sma(2, balance) as sma wma(3,"
-                    + " balance) as wma | fields balance, sma, wma",
+                "source=%s | where balance > 30000 | sort account_number | trendline sma(2,"
+                    + " balance) as sma wma(3, balance) as wma | fields balance, sma, wma",
                 TEST_INDEX_BANK));
     verifySchema(
         result, schema("balance", "bigint"), schema("sma", "double"), schema("wma", "double"));
@@ -77,8 +77,8 @@ public void testTrendlineNoAlias() throws Exception {
     JSONObject result =
         executeQuery(
             String.format(
-                "source=%s | where balance > 30000 | trendline sma(2, balance) | fields"
-                    + " balance, balance_trendline",
+                "source=%s | where balance > 30000 | sort account_number | trendline sma(2,"
+                    + " balance) | fields balance, balance_trendline",
                 TEST_INDEX_BANK));
     verifySchema(result, schema("balance", "bigint"), schema("balance_trendline", "double"));
     verifyDataRows(
@@ -90,8 +90,8 @@ public void testTrendlineOverwritesExisingField() throws Exception {
     JSONObject result =
         executeQuery(
             String.format(
-                "source=%s | where balance > 30000 | trendline sma(2, balance) as balance | fields"
-                    + " balance",
+                "source=%s | where balance > 30000 | sort account_number | trendline sma(2,"
+                    + " balance) as balance | fields balance",
                 TEST_INDEX_BANK));
     verifySchema(result, schema("balance", "double"));
     verifyDataRows(result, rows((Object) null), rows(36031.5), rows(36689), rows(44313));
@@ -115,7 +115,8 @@ public void testTrendlinePreFilterNullValues() throws Exception {
     JSONObject result =
         executeQuery(
             String.format(
-                "source=%s | trendline sma(2, balance) | fields" + " balance, balance_trendline",
+                "source=%s | sort account_number | trendline sma(2, balance) | fields"
+                    + " balance, balance_trendline",
                 TEST_INDEX_BANK_WITH_NULL_VALUES));
     verifySchema(result, schema("balance", "bigint"), schema("balance_trendline", "double"));
     verifyDataRows(

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

@@ -65,10 +65,20 @@ public static final class AnalyticsIndexConfig {
      */
     public static final String ENABLED_PROP = "tests.analytics.parquet_indices";
 
+    /**
+     * System property overriding the number of primary shards for analytics-backed test indices.
+     * Defaults to 1 (single-shard). Set to e.g. "3" for multi-shard coverage runs.
+     */
+    public static final String NUM_SHARDS_PROP = "tests.analytics.num_shards";
+
     public static boolean isEnabled() {
       return Boolean.parseBoolean(System.getProperty(ENABLED_PROP, "false"));
     }
 
+    public static int getNumShards() {
+      return Integer.parseInt(System.getProperty(NUM_SHARDS_PROP, "1"));
+    }
+
     // 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";
@@ -86,7 +96,7 @@ static void applyIndexCreationSettings(JSONObject jsonObject) {
           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("number_of_shards", getNumShards());
       indexSettings.put("pluggable.dataformat.enabled", true);
       indexSettings.put("pluggable.dataformat", DATAFORMAT_COMPOSITE);
       indexSettings.put("composite.primary_data_format", PRIMARY_FORMAT_PARQUET);