test(integ-test): stabilize stream-order commands across shards

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

@@ -5,14 +5,18 @@
 
 package org.opensearch.sql.calcite.remote;
 
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
 import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_HDFS_LOGS;
-import static org.opensearch.sql.util.MatcherUtils.rows;
 import static org.opensearch.sql.util.MatcherUtils.schema;
-import static org.opensearch.sql.util.MatcherUtils.verifyDataRows;
 import static org.opensearch.sql.util.MatcherUtils.verifySchemaInOrder;
 
-import com.google.common.collect.ImmutableList;
 import java.io.IOException;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.json.JSONArray;
 import org.json.JSONObject;
 import org.junit.Test;
 import org.opensearch.sql.ppl.PPLIntegTestCase;
@@ -45,30 +49,52 @@ public void testDashboardBrainLabelStatsByPatternsField() throws IOException {
         schema("patterns_field", "string"),
         schema("pattern_count", "bigint"),
         schema("sample_logs", "array"));
-    verifyDataRows(
-        result,
-        rows(
-            "BLOCK* NameSystem.addStoredBlock: blockMap updated: <*IP*> is added to blk_<*> size"
-                + " <*>",
-            2,
-            ImmutableList.of(
-                "BLOCK* NameSystem.addStoredBlock: blockMap updated: 10.251.31.85:50010 is added"
-                    + " to blk_-7017553867379051457 size 67108864")),
-        rows(
-            "PacketResponder failed <*> blk_<*>",
-            2,
-            ImmutableList.of("PacketResponder failed for blk_6996194389878584395")),
-        rows(
-            "Verification succeeded <*> blk_<*>",
-            2,
-            ImmutableList.of("Verification succeeded for blk_-1547954353065580372")),
-        rows(
-            "<*> NameSystem.allocateBlock:"
-                + " /user/root/sortrand/_temporary/_task_<*>_<*>_r_<*>_<*>/part<*> blk_<*>",
-            2,
-            ImmutableList.of(
-                "BLOCK* NameSystem.allocateBlock:"
-                    + " /user/root/sortrand/_temporary/_task_200811092030_0002_r_000296_0/part-00296."
-                    + " blk_-6620182933895093708")));
+    // Every pattern's count is 2 (shard-invariant) but take(content, 1) samples one arbitrary
+    // matching document, which has no stable cross-shard tiebreaker (and all counts tie, so the
+    // sort - pattern_count row order is also unstable). Assert each pattern's count and that its
+    // single sampled log is one of the two documents that carry the pattern.
+    Map<String, List<String>> universe = new HashMap<>();
+    universe.put(
+        "BLOCK* NameSystem.addStoredBlock: blockMap updated: <*IP*> is added to blk_<*> size <*>",
+        Arrays.asList(
+            "BLOCK* NameSystem.addStoredBlock: blockMap updated: 10.251.31.85:50010 is added to"
+                + " blk_-7017553867379051457 size 67108864",
+            "BLOCK* NameSystem.addStoredBlock: blockMap updated: 10.251.107.19:50010 is added to"
+                + " blk_-3249711809227781266 size 67108864"));
+    universe.put(
+        "PacketResponder failed <*> blk_<*>",
+        Arrays.asList(
+            "PacketResponder failed for blk_6996194389878584395",
+            "PacketResponder failed for blk_-1547954353065580372"));
+    universe.put(
+        "Verification succeeded <*> blk_<*>",
+        Arrays.asList(
+            "Verification succeeded for blk_-1547954353065580372",
+            "Verification succeeded for blk_6996194389878584395"));
+    universe.put(
+        "<*> NameSystem.allocateBlock:"
+            + " /user/root/sortrand/_temporary/_task_<*>_<*>_r_<*>_<*>/part<*> blk_<*>",
+        Arrays.asList(
+            "BLOCK* NameSystem.allocateBlock:"
+                + " /user/root/sortrand/_temporary/_task_200811092030_0002_r_000296_0/part-00296."
+                + " blk_-6620182933895093708",
+            "BLOCK* NameSystem.allocateBlock:"
+                + " /user/root/sortrand/_temporary/_task_200811092030_0002_r_000318_0/part-00318."
+                + " blk_2096692261399680562"));
+
+    JSONArray datarows = result.getJSONArray("datarows");
+    assertEquals(4, datarows.length());
+    for (int i = 0; i < datarows.length(); i++) {
+      JSONArray row = datarows.getJSONArray(i);
+      String pattern = row.getString(0);
+      assertTrue("unexpected pattern: " + pattern, universe.containsKey(pattern));
+      assertEquals("count for " + pattern, 2L, row.getLong(1));
+      JSONArray samples = row.getJSONArray(2);
+      assertEquals(1, samples.length());
+      String sample = samples.getString(0);
+      assertTrue(
+          "sample not a member of pattern " + pattern + ": " + sample,
+          universe.get(pattern).contains(sample));
+    }
   }
 }

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

@@ -11,6 +11,15 @@
 import static org.opensearch.sql.util.MatcherUtils.*;
 
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+import org.json.JSONArray;
 import org.json.JSONObject;
 import org.junit.jupiter.api.Test;
 import org.opensearch.sql.ppl.PPLIntegTestCase;
@@ -60,17 +69,25 @@ public void testDedupKeepEmpty() throws IOException {
             String.format(
                 "source=%s | dedup 1 name KEEPEMPTY=true | fields name, category",
                 TEST_INDEX_DUPLICATION_NULLABLE));
-    verifyDataRows(
-        actual,
-        rows("A", "X"),
-        rows("B", "Z"),
-        rows("C", "X"),
-        rows("D", "Z"),
-        rows("E", null),
-        rows(null, "Y"),
-        rows(null, "X"),
-        rows(null, "Z"),
-        rows(null, null));
+    // KEEPEMPTY keeps one row per distinct non-null name plus every null-name row. The null-name
+    // rows are fully determined (dedup keys on name only); the non-null representatives' category
+    // has no stable cross-shard tiebreaker, so assert one row per name with a valid pair.
+    List<List<Object>> rows = dataRows(actual);
+    assertEquals(9, rows.size());
+    Set<Object> nonNullNames = new HashSet<>();
+    Set<List<Object>> nullNameRows = new HashSet<>();
+    for (List<Object> row : rows) {
+      Object name = row.get(0);
+      Object category = row.get(1);
+      if (name == null) {
+        nullNameRows.add(Arrays.asList(name, category));
+      } else {
+        nonNullNames.add(name);
+        assertValidPair(name, category);
+      }
+    }
+    assertEquals(Set.of("A", "B", "C", "D", "E"), nonNullNames);
+    assertEquals(NULL_NAME_ROWS, nullNameRows);
   }
 
   @Test
@@ -96,6 +113,15 @@ public void testDedupMultipleFieldsKeepEmpty() throws IOException {
         rows(null, null));
   }
 
+  /**
+   * {@code CONSECUTIVE=true} collapses only <em>adjacent</em> duplicates, so its result depends
+   * entirely on the row-encounter order. A multi-shard index has no stable merge order (the counts
+   * observed on a single shard, 8/12/12/16, become 12/... on five shards), so this test cannot run
+   * against a sharded index. Drive it from a coordinator-only {@code makeresults} literal instead:
+   * the literal fixes the encounter order to the historical {@code duplication_nullable} insertion
+   * sequence (name column, with nulls preserved), making the consecutive-dedup counts deterministic
+   * on every route while still exercising real CONSECUTIVE semantics.
+   */
   @Test
   @RequiresCapability(
       value = DEDUP_NONDETERMINISTIC,
@@ -169,20 +195,25 @@ public void testDedupKeepEmpty2() throws IOException {
             String.format(
                 "source=%s | dedup 2 name KEEPEMPTY=true | fields name, category",
                 TEST_INDEX_DUPLICATION_NULLABLE));
-    verifyDataRows(
-        actual,
-        rows("A", "X"),
-        rows("A", "Y"),
-        rows("B", "Z"),
-        rows("B", "Z"),
-        rows("C", "X"),
-        rows("C", "X"),
-        rows("D", "Z"),
-        rows("E", null),
-        rows(null, "Y"),
-        rows(null, "X"),
-        rows(null, "Z"),
-        rows(null, null));
+    // dedup 2 keeps up to two rows per distinct non-null name (A/B/C have >=2, D/E have 1) plus
+    // every null-name row. Which two categories survive per name has no stable cross-shard
+    // tiebreaker, so assert the per-name kept-count, valid pairs, and the fixed null-name rows.
+    List<List<Object>> rows = dataRows(actual);
+    assertEquals(12, rows.size());
+    Map<Object, Integer> nameCounts = new HashMap<>();
+    Set<List<Object>> nullNameRows = new HashSet<>();
+    for (List<Object> row : rows) {
+      Object name = row.get(0);
+      Object category = row.get(1);
+      if (name == null) {
+        nullNameRows.add(Arrays.asList(name, category));
+      } else {
+        nameCounts.merge(name, 1, Integer::sum);
+        assertValidPair(name, category);
+      }
+    }
+    assertEquals(Map.of("A", 2, "B", 2, "C", 2, "D", 1, "E", 1), nameCounts);
+    assertEquals(NULL_NAME_ROWS, nullNameRows);
   }
 
   @Test
@@ -264,20 +295,35 @@ public void testReorderDedupFieldsShouldNotAffectResult() throws IOException {
   public void testDedupComplex() throws IOException {
     JSONObject actual =
         executeQuery(String.format("source=%s | dedup 1 name", TEST_INDEX_DUPLICATION_NULLABLE));
-    verifyDataRows(
-        actual,
-        rows("X", "A", 1),
-        rows("Z", "B", 1),
-        rows("X", "C", 1),
-        rows("Z", "D", 1),
-        rows(null, "E", 1));
+    // dedup 1 name keeps one row per distinct non-null name. The surviving row's category (and any
+    // other non-key column) has no stable cross-shard tiebreaker, so assert the dedup invariant:
+    // exactly the five names, each once, and each surviving (name, category) is a real data pair.
+    List<List<Object>> byName = dataRows(actual);
+    assertEquals(5, byName.size());
+    Set<Object> names = new HashSet<>();
+    for (List<Object> row : byName) {
+      Object category = row.get(0);
+      Object name = row.get(1);
+      names.add(name);
+      assertEquals(1, ((Number) row.get(2)).intValue());
+      assertValidPair(name, category);
+    }
+    assertEquals(Set.of("A", "B", "C", "D", "E"), names);
     actual =
         executeQuery(
             String.format(
                 "source=%s | fields category, name | dedup 1 name",
                 TEST_INDEX_DUPLICATION_NULLABLE));
-    verifyDataRows(
-        actual, rows("X", "A"), rows("Z", "B"), rows("X", "C"), rows("Z", "D"), rows(null, "E"));
+    List<List<Object>> byNameProjected = dataRows(actual);
+    assertEquals(5, byNameProjected.size());
+    Set<Object> projectedNames = new HashSet<>();
+    for (List<Object> row : byNameProjected) {
+      Object category = row.get(0);
+      Object name = row.get(1);
+      projectedNames.add(name);
+      assertValidPair(name, category);
+    }
+    assertEquals(Set.of("A", "B", "C", "D", "E"), projectedNames);
     actual =
         executeQuery(
             String.format("source=%s | dedup 1 name, category", TEST_INDEX_DUPLICATION_NULLABLE));
@@ -361,9 +407,13 @@ public void testDedupWithRenamedField() throws IOException {
     JSONObject actual =
         executeQuery(
             String.format(
-                "source=%s | rename name as nm | dedup 1 category | fields category, nm",
+                "source=%s | where isnotnull(name) | rename name as nm | sort nm | dedup 1 category"
+                    + " | fields category, nm",
                 TEST_INDEX_DUPLICATION_NULLABLE));
-    // One representative row per category; nm must not be null
+    // Pin the surviving representative deterministically across shards: excluding null names and
+    // sorting by nm makes dedup keep the lexicographically-first name per category (X->A, Y->A,
+    // Z->B). This still exercises the #5150 fix (the renamed non-key field must resolve to a
+    // non-null value in the dedup top_hits response).
     verifyDataRows(actual, rows("X", "A"), rows("Z", "B"), rows("Y", "A"));
   }
 
@@ -380,10 +430,12 @@ public void testDedupWithRenamedFieldMappingCollision() throws IOException {
     JSONObject actual =
         executeQuery(
             String.format(
-                "source=%s | eval nm2 = name | rename name as nm | dedup 1 category"
-                    + " | fields category, nm, nm2",
+                "source=%s | where isnotnull(name) | eval nm2 = name | rename name as nm | sort nm"
+                    + " | dedup 1 category | fields category, nm, nm2",
                 TEST_INDEX_DUPLICATION_NULLABLE));
-    // Both nm (from rename) and nm2 (from eval col-ref) must carry the same non-null name value
+    // Pin the surviving representative deterministically across shards (see testDedupWithRenamed
+    // Field). Both nm (from rename) and nm2 (from the eval col-ref) must carry the same non-null
+    // name value, exercising the #5197 alias-collision fix.
     verifyDataRows(actual, rows("X", "A", "A"), rows("Z", "B", "B"), rows("Y", "A", "A"));
   }
 
@@ -422,13 +474,22 @@ public void testDedupExpr() throws IOException {
             String.format(
                 "source=%s | eval new_name = lower(name) | dedup 1 new_name",
                 TEST_INDEX_DUPLICATION_NULLABLE));
-    verifyDataRows(
-        actual,
-        rows("X", "A", 1, "a"),
-        rows("Z", "B", 1, "b"),
-        rows("X", "C", 1, "c"),
-        rows("Z", "D", 1, "d"),
-        rows(null, "E", 1, "e"));
+    // dedup 1 new_name keeps one row per distinct lower(name). name and new_name are fully
+    // determined (a<->A ...), but the surviving category has no stable cross-shard tiebreaker;
+    // assert the five keys, the derived-column relation, and a valid (name, category) pair.
+    List<List<Object>> byNewName = dataRows(actual);
+    assertEquals(5, byNewName.size());
+    Set<Object> newNames = new HashSet<>();
+    for (List<Object> row : byNewName) {
+      Object category = row.get(0);
+      Object name = row.get(1);
+      Object newName = row.get(3);
+      newNames.add(newName);
+      assertEquals(1, ((Number) row.get(2)).intValue());
+      assertEquals(((String) name).toLowerCase(Locale.ROOT), newName);
+      assertValidPair(name, category);
+    }
+    assertEquals(Set.of("a", "b", "c", "d", "e"), newNames);
     actual =
         executeQuery(
             String.format(
@@ -481,4 +542,54 @@ public void testDedupExpr() throws IOException {
         rows("Z", 1, "B", "b", "z"),
         rows("Z", 1, "B", "b", "z"));
   }
+
+  // ---- Multi-shard dedup helpers ------------------------------------------------------------
+
+  /**
+   * Every (name, category) pair with a non-null name present in the {@code duplication_nullable}
+   * dataset. dedup keeps one (or N) representative row(s) per key; because the merge has no stable
+   * cross-shard tiebreaker, tests assert the surviving row is a real pair rather than a fixed one.
+   */
+  private static final Set<List<Object>> VALID_NAME_CATEGORY =
+      Set.of(
+          Arrays.asList("A", "X"),
+          Arrays.asList("A", "Y"),
+          Arrays.asList("B", "Z"),
+          Arrays.asList("B", "Y"),
+          Arrays.asList("B", null),
+          Arrays.asList("C", "X"),
+          Arrays.asList("D", "Z"),
+          Arrays.asList("E", null));
+
+  /**
+   * The null-name rows kept by {@code KEEPEMPTY=true} are fully determined: dedup keys on name
+   * only, so every null-name document survives, one per distinct category those rows carry.
+   */
+  private static final Set<List<Object>> NULL_NAME_ROWS =
+      Set.of(
+          Arrays.asList(null, "Y"),
+          Arrays.asList(null, "X"),
+          Arrays.asList(null, "Z"),
+          Arrays.asList(null, null));
+
+  private static void assertValidPair(Object name, Object category) {
+    assertTrue(
+        "unexpected surviving (name, category) = (" + name + ", " + category + ")",
+        VALID_NAME_CATEGORY.contains(Arrays.asList(name, category)));
+  }
+
+  /** Materialize {@code datarows} into a list of rows, mapping JSON null to Java {@code null}. */
+  private static List<List<Object>> dataRows(JSONObject response) {
+    List<List<Object>> rows = new ArrayList<>();
+    JSONArray arr = response.getJSONArray("datarows");
+    for (int i = 0; i < arr.length(); i++) {
+      JSONArray r = arr.getJSONArray(i);
+      List<Object> row = new ArrayList<>();
+      for (int j = 0; j < r.length(); j++) {
+        row.add(r.isNull(j) ? null : r.get(j));
+      }
+      rows.add(row);
+    }
+    return rows;
+  }
 }