test(integ-test): stabilize deterministic fixtures across shards

integ-test/src/test/java/org/opensearch/sql/api/UnifiedQueryOpenSearchIT.java

@@ -72,7 +72,8 @@ public void cleanUp() throws Exception {
   public void testSimplePPLQueryExecution() throws Exception {
     String pplQuery =
         String.format(
-            "source = opensearch.%s | fields firstname, age | where age > 30 | head 3",
+            "source = opensearch.%s | where age > 30 and account_number in (1, 6, 18) | fields"
+                + " firstname, age",
             TEST_INDEX_ACCOUNT);
 
     RelNode logicalPlan = planner.plan(pplQuery);

integ-test/src/test/java/org/opensearch/sql/calcite/CalciteComplexPoolIT.java

@@ -46,7 +46,7 @@ public void testParseCommandDispatchesToComplexPool() throws IOException {
         executeQuery(
             String.format(
                 "source=%s | parse address '(?<number>\\\\d+) (?<street>.*)'"
-                    + " | fields number, street | head 1",
+                    + " | sort account_number | fields number, street | head 1",
                 TEST_INDEX_BANK));
 
     verifyDataRows(result, rows("880", "Holmes Lane"));

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

@@ -955,8 +955,8 @@ public void testMvmapWithOtherFieldReference() throws IOException {
     JSONObject actual =
         executeQuery(
             String.format(
-                "source=%s | eval arr = array(1, 2, 3), result = mvmap(arr, arr * age) | head 1 |"
-                    + " fields age, result",
+                "source=%s | eval arr = array(1, 2, 3), result = mvmap(arr, arr * age) | sort"
+                    + " account_number | head 1 | fields age, result",
                 TEST_INDEX_BANK));
 
     verifySchema(actual, schema("age", "int"), schema("result", "array"));

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

@@ -441,8 +441,8 @@ public void testBinTimestampSpan7Days() throws IOException {
     JSONObject result =
         executeQuery(
             String.format(
-                "source=%s | bin @timestamp span=7day | fields"
-                    + " @timestamp, value | sort @timestamp | head 3",
+                "source=%s | eval original_timestamp = @timestamp | bin @timestamp span=7day |"
+                    + " sort original_timestamp | head 3 | fields @timestamp, value",
                 TEST_INDEX_TIME_DATA));
     verifySchema(result, schema("@timestamp", null, "timestamp"), schema("value", null, "int"));
     verifyDataRows(
@@ -457,8 +457,8 @@ public void testBinTimestampSpan6Days() throws IOException {
     JSONObject result =
         executeQuery(
             String.format(
-                "source=%s | bin @timestamp span=6day | fields"
-                    + " @timestamp, value | sort @timestamp | head 3",
+                "source=%s | eval original_timestamp = @timestamp | bin @timestamp span=6day |"
+                    + " sort original_timestamp | head 3 | fields @timestamp, value",
                 TEST_INDEX_TIME_DATA));
     verifySchema(result, schema("@timestamp", null, "timestamp"), schema("value", null, "int"));
     verifyDataRows(

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

@@ -573,10 +573,14 @@ public void testTableWithDuplicateWildcardMatches() throws IOException {
   public void testFieldsAndTableEquivalence() throws IOException {
     JSONObject fieldsResult =
         executeQuery(
-            String.format("source=%s | fields firstname, lastname | head 3", TEST_INDEX_ACCOUNT));
+            String.format(
+                "source=%s | where account_number in (1, 6, 13) | fields firstname, lastname",
+                TEST_INDEX_ACCOUNT));
     JSONObject tableResult =
         executeQuery(
-            String.format("source=%s | table firstname, lastname | head 3", TEST_INDEX_ACCOUNT));
+            String.format(
+                "source=%s | where account_number in (1, 6, 13) | table firstname, lastname",
+                TEST_INDEX_ACCOUNT));
 
     verifySchema(fieldsResult, schema("firstname", "string"), schema("lastname", "string"));
     verifySchema(tableResult, schema("firstname", "string"), schema("lastname", "string"));
@@ -592,11 +596,13 @@ public void testSpaceDelimitedEquivalentToCommaDelimited() throws IOException {
     JSONObject commaResult =
         executeQuery(
             String.format(
-                "source=%s | fields firstname, lastname, age | head 3", TEST_INDEX_ACCOUNT));
+                "source=%s | where account_number in (1, 6, 13) | fields firstname, lastname, age",
+                TEST_INDEX_ACCOUNT));
     JSONObject spaceResult =
         executeQuery(
             String.format(
-                "source=%s | fields firstname lastname age | head 3", TEST_INDEX_ACCOUNT));
+                "source=%s | where account_number in (1, 6, 13) | fields firstname lastname age",
+                TEST_INDEX_ACCOUNT));
 
     verifySchema(
         commaResult,

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

@@ -129,9 +129,10 @@ public void testMultisearchWithFieldsProjection() throws IOException {
     JSONObject result =
         executeQuery(
             String.format(
-                "| multisearch [search source=%s | where gender = \\\"M\\\" | fields"
-                    + " firstname, lastname, balance] [search source=%s | where gender = \\\"F\\\""
-                    + " | fields firstname, lastname, balance] | head 5",
+                "| multisearch [search source=%s | where gender = \\\"M\\\" and account_number in"
+                    + " (1, 6, 18) | fields firstname, lastname, balance] [search source=%s | where"
+                    + " gender = \\\"F\\\" and account_number in (13, 25) | fields firstname,"
+                    + " lastname, balance]",
                 TEST_INDEX_ACCOUNT, TEST_INDEX_ACCOUNT));
 
     verifySchema(
@@ -145,8 +146,8 @@ public void testMultisearchWithFieldsProjection() throws IOException {
         rows("Amber", "Duke", 39225L),
         rows("Hattie", "Bond", 5686L),
         rows("Dale", "Adams", 4180L),
-        rows("Elinor", "Ratliff", 16418L),
-        rows("Mcgee", "Mooney", 18612L));
+        rows("Nanette", "Bates", 32838L),
+        rows("Virginia", "Ayala", 40540L));
   }
 
   @Test

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

@@ -68,7 +68,14 @@ public void testMvCombine_basicGroupCollapsesToOneRow() throws IOException {
         schema("tags", null, "string"),
         schema("packets_str", null, "array"));
 
-    verifyDataRows(result, rows("10.0.0.1", 100, "t1", List.of("10", "20", "30")));
+    JSONArray row = result.getJSONArray("datarows").getJSONArray(0);
+    Assertions.assertEquals("10.0.0.1", row.getString(0));
+    Assertions.assertEquals(100, row.getLong(1));
+    Assertions.assertEquals("t1", row.getString(2));
+    List<String> packets = new ArrayList<>();
+    row.getJSONArray(3).forEach(value -> packets.add(value.toString()));
+    Collections.sort(packets);
+    Assertions.assertEquals(List.of("10", "20", "30"), packets);
   }
 
   @Test

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

@@ -44,8 +44,8 @@ public void testCoalesceBasic() throws IOException {
     JSONObject actual =
         executeQuery(
             String.format(
-                "source=%s | eval result = coalesce(name, age, 0) | fields name, age, result |"
-                    + " head 3",
+                "source=%s | eval result = coalesce(name, age, 0) | sort - age | fields name, age,"
+                    + " result | head 3",
                 TEST_INDEX_STATE_COUNTRY_WITH_NULL));
 
     verifySchema(
@@ -61,7 +61,7 @@ public void testCoalesceWithMixedTypes() throws IOException {
         executeQuery(
             String.format(
                 "source=%s | eval result = coalesce(name, age, 'fallback') |"
-                    + " fields name, age, result | head 3",
+                    + " sort - age | fields name, age, result | head 3",
                 TEST_INDEX_STATE_COUNTRY_WITH_NULL));
 
     verifySchema(
@@ -170,8 +170,8 @@ public void testCoalesceWithAllNonExistentFields() throws IOException {
     JSONObject actual =
         executeQuery(
             String.format(
-                "source=%s | eval result = coalesce(field1, field2, field3) | fields name, result |"
-                    + " head 1",
+                "source=%s | eval result = coalesce(field1, field2, field3) | sort - age | fields"
+                    + " name, result | head 1",
                 TEST_INDEX_STATE_COUNTRY_WITH_NULL));
 
     // When every COALESCE operand is missing/null, the result has no known type (see #5175).

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

@@ -5,6 +5,8 @@
 
 package org.opensearch.sql.calcite.remote;
 
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
 import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_GRAPH_AIRPORTS;
 import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_GRAPH_EMPLOYEES;
 import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_GRAPH_TRAVELERS;
@@ -14,10 +16,16 @@
 import static org.opensearch.sql.util.MatcherUtils.verifySchema;
 
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.Collections;
+import java.util.Iterator;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import org.hamcrest.Description;
+import org.hamcrest.Matcher;
+import org.hamcrest.TypeSafeMatcher;
+import org.json.JSONArray;
 import org.json.JSONObject;
 import org.junit.Test;
 import org.opensearch.sql.ppl.PPLIntegTestCase;
@@ -63,6 +71,159 @@ private static Map<String, Object> mapOf(Object... keysAndValues) {
     return map;
   }
 
+  /**
+   * Row matcher for graphLookup results whose collected arrays come back in a shard-dependent
+   * order.
+   *
+   * <p>graphLookup gathers traversal results into an array, and the order of elements within that
+   * array is not defined across shards. A multi-shard run therefore returns the same set of nodes
+   * (same values, same {@code depth}/{@code numConnections}, same cardinality) as a single-shard
+   * run, only in a different order. Unlike {@link
+   * org.opensearch.sql.util.MatcherUtils#rows(Object...)}, which compares row cells with
+   * order-sensitive {@link JSONArray#similar}, this matcher relaxes ordering <em>only within nested
+   * arrays and objects</em>. The top-level row cells are still compared positionally: cell {@code
+   * i} of the actual row must match cell {@code i} of the expected row. This preserves column
+   * identity, so a swap of two same-typed top-level columns is still rejected, while collected
+   * array order may vary across shards. It is scoped to this test class so the relaxed comparison
+   * never leaks into other suites.
+   */
+  private static TypeSafeMatcher<JSONArray> rowsUnordered(Object... expectedObjects) {
+    return new TypeSafeMatcher<>() {
+      @Override
+      protected boolean matchesSafely(JSONArray array) {
+        JSONArray expected = new JSONArray(expectedObjects);
+        if (array.length() != expected.length()) {
+          return false;
+        }
+        // Compare top-level cells positionally so column identity is preserved; only descend into
+        // nested arrays/objects with order-insensitive comparison.
+        for (int i = 0; i < expected.length(); i++) {
+          if (!jsonEqualsIgnoringOrder(array.get(i), expected.get(i))) {
+            return false;
+          }
+        }
+        return true;
+      }
+
+      @Override
+      public void describeTo(Description description) {
+        description.appendText(new JSONArray(expectedObjects).toString());
+      }
+    };
+  }
+
+  /** Recursively compares two JSON values, treating every array as an unordered multiset. */
+  private static boolean jsonEqualsIgnoringOrder(Object a, Object b) {
+    if (a instanceof JSONArray && b instanceof JSONArray) {
+      return jsonArrayEqualsIgnoringNestedOrder((JSONArray) a, (JSONArray) b);
+    }
+    if (a instanceof JSONObject && b instanceof JSONObject) {
+      JSONObject ao = (JSONObject) a;
+      JSONObject bo = (JSONObject) b;
+      if (ao.keySet().size() != bo.keySet().size()) {
+        return false;
+      }
+      for (String key : ao.keySet()) {
+        if (!bo.has(key) || !jsonEqualsIgnoringOrder(ao.get(key), bo.get(key))) {
+          return false;
+        }
+      }
+      return true;
+    }
+    // A container on exactly one side is a genuine type mismatch. Without this guard the
+    // fall-through to scalarEquals compares toString() forms, so a JSONArray/JSONObject whose
+    // compact serialization equals a scalar string (e.g. the array ["BOS","JFK"] vs the string
+    // "[\"BOS\",\"JFK\"]") would falsely match. Reject one-sided containers explicitly.
+    if (a instanceof JSONArray || b instanceof JSONArray) {
+      return false;
+    }
+    if (a instanceof JSONObject || b instanceof JSONObject) {
+      return false;
+    }
+    return scalarEquals(a, b);
+  }
+
+  private static boolean jsonArrayEqualsIgnoringNestedOrder(JSONArray actual, JSONArray expected) {
+    if (actual.length() != expected.length()) {
+      return false;
+    }
+    List<Object> remaining = new ArrayList<>();
+    for (int i = 0; i < expected.length(); i++) {
+      remaining.add(expected.get(i));
+    }
+    for (int i = 0; i < actual.length(); i++) {
+      Object actualElement = actual.get(i);
+      boolean matched = false;
+      for (Iterator<Object> it = remaining.iterator(); it.hasNext(); ) {
+        if (jsonEqualsIgnoringOrder(actualElement, it.next())) {
+          it.remove();
+          matched = true;
+          break;
+        }
+      }
+      if (!matched) {
+        return false;
+      }
+    }
+    return true;
+  }
+
+  private static boolean scalarEquals(Object a, Object b) {
+    boolean aNull = a == null || a == JSONObject.NULL;
+    boolean bNull = b == null || b == JSONObject.NULL;
+    if (aNull || bNull) {
+      return aNull && bNull;
+    }
+    if (a instanceof Number && b instanceof Number) {
+      return ((Number) a).doubleValue() == ((Number) b).doubleValue();
+    }
+    return a.toString().equals(b.toString());
+  }
+
+  /**
+   * Regression test for {@link #rowsUnordered(Object...)}. Proves the matcher relaxes ordering only
+   * inside nested arrays/objects while keeping top-level column positions and cardinality
+   * significant. This runs without a cluster so it always executes as part of the suite.
+   */
+  @Test
+  public void testRowsUnorderedMatcherPreservesColumnPositions() {
+    // 1. A shard-dependent reorder inside a nested collected array still matches.
+    Matcher<JSONArray> matcher = rowsUnordered("Jeff", List.of("BOS", "JFK", "LAX"));
+    JSONArray nestedReordered = new JSONArray(List.of("Jeff", List.of("LAX", "BOS", "JFK")));
+    assertTrue("nested array reorder should match", matcher.matches(nestedReordered));
+
+    // 2. A swap of two same-typed top-level columns must NOT match (column identity preserved).
+    Matcher<JSONArray> swapMatcher = rowsUnordered("BOS", "JFK");
+    JSONArray swapped = new JSONArray(List.of("JFK", "BOS"));
+    assertFalse("same-type top-level column swap must not match", swapMatcher.matches(swapped));
+
+    // 3. A missing value inside the nested array (different cardinality) must NOT match.
+    JSONArray missingNested = new JSONArray(List.of("Jeff", List.of("BOS", "JFK")));
+    assertFalse("missing nested element must not match", matcher.matches(missingNested));
+
+    // 4. A container versus a differently-shaped container (array vs object) must NOT match.
+    JSONArray arrCell = new JSONArray(List.of("BOS"));
+    JSONObject objCell = new JSONObject(Map.of("0", "BOS"));
+    assertFalse(
+        "array vs object must not match",
+        rowsUnordered((Object) arrCell).matches(new JSONArray(List.of(objCell))));
+
+    // 5. A JSONArray versus a scalar string with identical serialized text must NOT match. Guards
+    // against the scalarEquals toString() fall-through producing a false positive.
+    JSONArray arrExpected = new JSONArray(List.of("BOS", "JFK"));
+    assertFalse(
+        "array vs identically-serialized string must not match",
+        rowsUnordered((Object) arrExpected)
+            .matches(new JSONArray(List.of(arrExpected.toString()))));
+
+    // 6. A JSONObject versus a scalar string with identical serialized text must NOT match.
+    JSONObject objExpected = new JSONObject(Map.of("code", "BOS"));
+    assertFalse(
+        "object vs identically-serialized string must not match",
+        rowsUnordered((Object) objExpected)
+            .matches(new JSONArray(List.of(objExpected.toString()))));
+  }
+
   // ==================== Employee Hierarchy Tests ====================
 
   /** Test 1: Basic employee hierarchy traversal. Find all managers in the reporting chain. */
@@ -398,9 +559,11 @@ public void testTravelerReachableAirportsWithMaxDepth() throws IOException {
         schema("name", "string"),
         schema("nearestAirport", "string"),
         schema("reachableAirports", "array"));
+    // graphLookup collects reachable airports into an array whose element order is shard-dependent;
+    // compare values/connects/cardinality while ignoring nested array order.
     verifyDataRows(
         result,
-        rows(
+        rowsUnordered(
             "Jeff",
             "BOS",
             List.of(
@@ -708,9 +871,11 @@ public void testBatchModeEmployeeHierarchy() throws IOException {
                 TEST_INDEX_GRAPH_EMPLOYEES, TEST_INDEX_GRAPH_EMPLOYEES));
 
     verifySchema(result, schema("reportsTo", "array"), schema("reportingHierarchy", "array"));
+    // Batch mode returns a single row; the source-rows array and the hierarchy array both come
+    // back in a shard-dependent order, so compare them as unordered multisets.
     verifyDataRows(
         result,
-        rows(
+        rowsUnordered(
             List.of(
                 Map.of("name", "Dev", "reportsTo", "Eliot", "id", 1),
                 Map.of("name", "Asya", "reportsTo", "Ron", "id", 5)),
@@ -746,7 +911,7 @@ public void testBatchModeTravelersAirports() throws IOException {
     // - lookupResults: airports reachable from JFK and BOS within maxDepth=1
     verifyDataRows(
         result,
-        rows(
+        rowsUnordered(
             List.of(
                 Map.of("name", "Dev", "nearestAirport", "JFK"),
                 Map.of("name", "Eliot", "nearestAirport", "JFK"),
@@ -850,7 +1015,7 @@ public void testTopLevelGraphLookupLiteralList() throws IOException {
     //   because Ron.reportsTo=Andrew is in visited set
     verifyDataRows(
         result,
-        rows(
+        rowsUnordered(
             (Object)
                 List.of(
                     Map.of("name", "Eliot", "reportsTo", "Ron", "id", 2),

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

@@ -395,7 +395,10 @@ public void testSubsearchMaxOut() throws IOException {
             String.format(
                 "source = %s"
                     + "| where id in ["
-                    + "    source = %s | fields uid"
+                    // Sort the subsearch on a unique key before the maxout cap so the single
+                    // retained row (uid=1000) is deterministic across shard counts. Without this
+                    // the capped subsearch keeps an arbitrary uid on a multi-shard index.
+                    + "    source = %s | sort uid | fields uid"
                     + "  ]"
                     + "| sort  - salary"
                     + "| fields id, name, salary",