diff --git a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePPLDashboardPatternsIT.java b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePPLDashboardPatternsIT.java index 1ab5c7fc9bc..b20c0d32293 100644 --- a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePPLDashboardPatternsIT.java +++ b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePPLDashboardPatternsIT.java @@ -11,7 +11,6 @@ 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 org.json.JSONObject; import org.junit.Test; @@ -35,7 +34,7 @@ public void testDashboardBrainLabelStatsByPatternsField() throws IOException { + " | patterns content method=BRAIN mode=label" + " max_sample_count=5 variable_count_threshold=5" + " frequency_threshold_percentage=0.2" - + " | stats count() as pattern_count, take(content, 1) as sample_logs" + + " | stats count() as pattern_count, max(content) as sample_logs" + " by patterns_field" + " | sort - pattern_count" + " | fields patterns_field, pattern_count, sample_logs", @@ -44,31 +43,34 @@ public void testDashboardBrainLabelStatsByPatternsField() throws IOException { result, schema("patterns_field", "string"), schema("pattern_count", "bigint"), - schema("sample_logs", "array")); + schema("sample_logs", "string")); + // Each of the four BRAIN patterns covers exactly two documents (count 2, shard-invariant). + // max(content) picks the lexicographically-largest of a pattern's two source lines, which is + // order-insensitive and therefore stable across shards and routes, so the sample is exact + // rather than an arbitrary take(content, 1). Row order is order-insensitive here (all counts + // tie at 2). 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")), + "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")), + "PacketResponder failed for blk_6996194389878584395"), rows( "Verification succeeded <*> blk_<*>", 2, - ImmutableList.of("Verification succeeded for blk_-1547954353065580372")), + "Verification succeeded for blk_6996194389878584395"), 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"))); + "BLOCK* NameSystem.allocateBlock:" + + " /user/root/sortrand/_temporary/_task_200811092030_0002_r_000318_0/part-00318." + + " blk_2096692261399680562")); } } diff --git a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePPLDedupIT.java b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePPLDedupIT.java index 4177d108440..db7333daaeb 100644 --- a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePPLDedupIT.java +++ b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePPLDedupIT.java @@ -7,10 +7,20 @@ import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_ACCOUNT; import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_DUPLICATION_NULLABLE; +import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_DUPLICATION_NULLABLE_ORDERED; import static org.opensearch.sql.util.Capability.DEDUP_NONDETERMINISTIC; 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; @@ -24,6 +34,7 @@ public void init() throws Exception { enableCalcite(); loadIndex(Index.DUPLICATION_NULLABLE); + loadIndex(Index.DUPLICATION_NULLABLE_ORDERED); loadIndex(Index.ACCOUNT); } @@ -55,15 +66,22 @@ public void testDedupMultipleFields() throws IOException { @Test public void testDedupKeepEmpty() throws IOException { + // dedup 1 name KEEPEMPTY=true keeps the first row per distinct non-null name plus every + // null-name row. An added `sort name, category` (PPL default ASC NULLS FIRST) pins each + // non-null name's surviving row to its smallest category on any shard layout and route -- the + // same sort-before-dedup determinism the #3922 regression tests rely on -- so the kept + // representative is exact rather than merely "some valid pair". Per name that is A->X, + // B->null, C->X, D->Z, E->null; the four null-name rows are always kept in full. JSONObject actual = executeQuery( String.format( - "source=%s | dedup 1 name KEEPEMPTY=true | fields name, category", + "source=%s | sort name, category | dedup 1 name KEEPEMPTY=true | fields name," + + " category", TEST_INDEX_DUPLICATION_NULLABLE)); verifyDataRows( actual, rows("A", "X"), - rows("B", "Z"), + rows("B", null), rows("C", "X"), rows("D", "Z"), rows("E", null), @@ -96,6 +114,17 @@ public void testDedupMultipleFieldsKeepEmpty() throws IOException { rows(null, null)); } + /** + * {@code CONSECUTIVE=true} collapses only adjacent 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). To keep the real index + * route while making the encounter order deterministic, this drives a seq-augmented fixture + * ({@code duplication_nullable_ordered}, same rows plus an explicit {@code seq}) and adds {@code + * | sort seq} before dedup, restoring the historical {@code duplication_nullable} insertion + * sequence on any shard layout while still exercising real CONSECUTIVE semantics over the index. + * The AE route has no stable per-fragment tiebreaker (DEDUP_NONDETERMINISTIC), so the assertion + * stays gated to the routes that produce a deterministic ordered stream. + */ @Test @RequiresCapability( value = DEDUP_NONDETERMINISTIC, @@ -104,29 +133,31 @@ public void testConsecutiveImplicitFallbackV2() throws IOException { JSONObject actual = executeQuery( String.format( - "source = %s | dedup 1 name CONSECUTIVE=true | fields name", - TEST_INDEX_DUPLICATION_NULLABLE)); + "source = %s | sort seq | dedup 1 name CONSECUTIVE=true | fields name", + TEST_INDEX_DUPLICATION_NULLABLE_ORDERED)); verifyNumOfRows(actual, 8); actual = executeQuery( String.format( - "source = %s | dedup 1 name KEEPEMPTY=true CONSECUTIVE=true | fields name", - TEST_INDEX_DUPLICATION_NULLABLE)); + "source = %s | sort seq | dedup 1 name KEEPEMPTY=true CONSECUTIVE=true | fields" + + " name", + TEST_INDEX_DUPLICATION_NULLABLE_ORDERED)); verifyNumOfRows(actual, 12); actual = executeQuery( String.format( - "source = %s | dedup 2 name CONSECUTIVE=true | fields name", - TEST_INDEX_DUPLICATION_NULLABLE)); + "source = %s | sort seq | dedup 2 name CONSECUTIVE=true | fields name", + TEST_INDEX_DUPLICATION_NULLABLE_ORDERED)); verifyNumOfRows(actual, 12); actual = executeQuery( String.format( - "source = %s | dedup 2 name KEEPEMPTY=true CONSECUTIVE=true | fields name", - TEST_INDEX_DUPLICATION_NULLABLE)); + "source = %s | sort seq | dedup 2 name KEEPEMPTY=true CONSECUTIVE=true | fields" + + " name", + TEST_INDEX_DUPLICATION_NULLABLE_ORDERED)); verifyNumOfRows(actual, 16); } @@ -169,20 +200,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> rows = dataRows(actual); + assertEquals(12, rows.size()); + Map nameCounts = new HashMap<>(); + Set> nullNameRows = new HashSet<>(); + for (List 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 +300,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> byName = dataRows(actual); + assertEquals(5, byName.size()); + Set names = new HashSet<>(); + for (List 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> byNameProjected = dataRows(actual); + assertEquals(5, byNameProjected.size()); + Set projectedNames = new HashSet<>(); + for (List 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 +412,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 +435,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 +479,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> byNewName = dataRows(actual); + assertEquals(5, byNewName.size()); + Set newNames = new HashSet<>(); + for (List 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 +547,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> 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> 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> dataRows(JSONObject response) { + List> rows = new ArrayList<>(); + JSONArray arr = response.getJSONArray("datarows"); + for (int i = 0; i < arr.length(); i++) { + JSONArray r = arr.getJSONArray(i); + List 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; + } } diff --git a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePPLPatternsIT.java b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePPLPatternsIT.java index fac69220fbf..f15d23d0c3b 100644 --- a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePPLPatternsIT.java +++ b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalcitePPLPatternsIT.java @@ -5,6 +5,8 @@ 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_BANK; import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_HDFS_LOGS; import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_WEBLOGS; @@ -17,6 +19,15 @@ import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableMap; import java.io.IOException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Comparator; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.regex.Pattern; +import org.json.JSONArray; import org.json.JSONObject; import org.junit.Test; import org.opensearch.sql.ppl.PPLIntegTestCase; @@ -127,13 +138,19 @@ public void testSimplePatternAggregationMode_NotShowNumberedToken() throws IOExc schema("pattern_count", "bigint"), schema("patterns_field", "string"), schema("sample_logs", "array")); - verifyDataRows( - result, - rows( - "<*>@<*>.<*>", - 7, - ImmutableList.of( - "amberduke@pyrami.com", "hattiebond@netagy.com", "nanettebates@quility.com"))); + // The single pattern and its total count are shard-invariant; which max_sample_count emails + // land in the sample is not, so assert the count exactly and that each sampled log is a valid + // instance of the pattern (an email) rather than pinning the specific three. + List> rows = dataRows(result); + assertEquals(1, rows.size()); + List row = rows.get(0); + assertEquals("<*>@<*>.<*>", row.get(0)); + assertEquals(7L, asLong(row.get(1))); + List samples = asStringList(row.get(2)); + assertEquals(3, samples.size()); + for (String s : samples) { + assertTrue("not an email: " + s, EMAIL.matcher(s).matches()); + } } @Test @@ -150,20 +167,29 @@ public void testSimplePatternAggregationMode_ShowNumberedToken() throws IOExcept schema("patterns_field", "string"), schema("tokens", "struct"), schema("sample_logs", "array")); - verifyDataRows( - result, - rows( - "@.", - 7, - ImmutableMap.of( - "", - ImmutableList.of("amberduke", "hattiebond", "nanettebates"), - "", - ImmutableList.of("pyrami", "netagy", "quility"), - "", - ImmutableList.of("com", "com", "com")), - ImmutableList.of( - "amberduke@pyrami.com", "hattiebond@netagy.com", "nanettebates@quility.com"))); + // Pattern, count and token layout are shard-invariant; the sampled emails are not. Assert the + // count and that, for every sampled email, the captured tokens reconstruct it exactly + // (@.) so token/pattern correctness is verified without pinning which + // three emails were sampled. + List> rows = dataRows(result); + assertEquals(1, rows.size()); + List row = rows.get(0); + assertEquals("@.", row.get(0)); + assertEquals(7L, asLong(row.get(1))); + Map> tokens = tokenMap(row.get(2)); + assertEquals(Set.of("", "", ""), tokens.keySet()); + List samples = asStringList(row.get(3)); + assertEquals(3, samples.size()); + List t1 = tokens.get(""); + List t2 = tokens.get(""); + List t3 = tokens.get(""); + assertEquals(3, t1.size()); + assertEquals(3, t2.size()); + assertEquals(3, t3.size()); + for (int i = 0; i < samples.size(); i++) { + assertEquals(samples.get(i), t1.get(i) + "@" + t2.get(i) + "." + t3.get(i)); + assertTrue("not an email: " + samples.get(i), EMAIL.matcher(samples.get(i)).matches()); + } } @Test @@ -181,32 +207,30 @@ public void testSimplePatternAggregationMode_WithGroupBy_ShowNumberedToken() thr schema("pattern_count", "bigint"), schema("tokens", "struct"), schema("sample_logs", "array")); - verifyDataRows( - result, - rows( - false, - "@.", - 3, - ImmutableMap.of( - "", - ImmutableList.of("nanettebates"), - "", - ImmutableList.of("quility"), - "", - ImmutableList.of("com")), - ImmutableList.of("nanettebates@quility.com")), - rows( - true, - "@.", - 4, - ImmutableMap.of( - "", - ImmutableList.of("amberduke"), - "", - ImmutableList.of("pyrami"), - "", - ImmutableList.of("com")), - ImmutableList.of("amberduke@pyrami.com"))); + // The two groups and their counts (male=false:3, male=true:4) are shard-invariant; the single + // sampled email per group is not. Assert the per-group count and that the sampled email is + // reconstructed by its captured tokens. + List> rows = dataRows(result); + assertEquals(2, rows.size()); + Map countsByMale = new HashMap<>(); + for (List row : rows) { + Object male = row.get(0); + assertEquals("@.", row.get(1)); + countsByMale.put(male, asLong(row.get(2))); + Map> tokens = tokenMap(row.get(3)); + assertEquals(Set.of("", "", ""), tokens.keySet()); + List samples = asStringList(row.get(4)); + assertEquals(1, samples.size()); + List t1 = tokens.get(""); + List t2 = tokens.get(""); + List t3 = tokens.get(""); + assertEquals(1, t1.size()); + assertEquals(1, t2.size()); + assertEquals(1, t3.size()); + assertEquals(samples.get(0), t1.get(0) + "@" + t2.get(0) + "." + t3.get(0)); + assertTrue("not an email: " + samples.get(0), EMAIL.matcher(samples.get(0)).matches()); + } + assertEquals(Map.of(false, 3L, true, 4L), countsByMale); } @Test @@ -216,23 +240,23 @@ public void testBrainLabelMode_NotShowNumberedToken() throws IOException { String.format( "source=%s | patterns content method=BRAIN mode=label" + " max_sample_count=5 variable_count_threshold=5" - + " frequency_threshold_percentage=0.2 | head 2 | fields content," + + " frequency_threshold_percentage=0.2 | sort pid | head 2 | fields content," + " patterns_field", TEST_INDEX_HDFS_LOGS)); verifySchema(result, schema("content", "string"), schema("patterns_field", "string")); + // `head 2` has no stable order across shards; sort on the unique pid first so the two returned + // documents are deterministic (pid 26 then pid 31). Pattern labeling is computed over the full + // input, so the sort only pins which two labeled rows surface. verifyDataRows( result, rows( - "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", "BLOCK* NameSystem.addStoredBlock: blockMap updated: <*IP*> is added to blk_<*>" + " size <*>"), rows( - "BLOCK* NameSystem.allocateBlock:" - + " /user/root/sortrand/_temporary/_task_200811092030_0002_r_000296_0/part-00296." - + " blk_-6620182933895093708", - "<*> NameSystem.allocateBlock:" - + " /user/root/sortrand/_temporary/_task_<*>_<*>_r_<*>_<*>/part<*> blk_<*>")); + "PacketResponder failed for blk_-1547954353065580372", + "PacketResponder failed <*> blk_<*>")); } @Test @@ -242,7 +266,7 @@ public void testBrainLabelMode_ShowNumberedToken() throws IOException { String.format( "source=%s | patterns content method=BRAIN mode=label" + " max_sample_count=5 show_numbered_token=true variable_count_threshold=5" - + " frequency_threshold_percentage=0.2 | head 2 | fields content," + + " frequency_threshold_percentage=0.2 | sort pid | head 2 | fields content," + " patterns_field, tokens", TEST_INDEX_HDFS_LOGS)); verifySchema( @@ -250,42 +274,30 @@ public void testBrainLabelMode_ShowNumberedToken() throws IOException { schema("content", "string"), schema("patterns_field", "string"), schema("tokens", "struct")); + // See testBrainLabelMode_NotShowNumberedToken: sort pid pins the two returned documents + // (pid 26 then pid 31) deterministically across shards. verifyDataRows( result, rows( - "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", "BLOCK* NameSystem.addStoredBlock: blockMap updated: is added to blk_" + " size ", ImmutableMap.of( "", - ImmutableList.of("10.251.31.85:50010"), + ImmutableList.of("10.251.107.19:50010"), "", - ImmutableList.of("-7017553867379051457"), + ImmutableList.of("-3249711809227781266"), "", ImmutableList.of("67108864"))), rows( - "BLOCK* NameSystem.allocateBlock:" - + " /user/root/sortrand/_temporary/_task_200811092030_0002_r_000296_0/part-00296." - + " blk_-6620182933895093708", - " NameSystem.allocateBlock:" - + " /user/root/sortrand/_temporary/_task___r__/part" - + " blk_", + "PacketResponder failed for blk_-1547954353065580372", + "PacketResponder failed blk_", ImmutableMap.of( "", - ImmutableList.of("BLOCK*"), + ImmutableList.of("for"), "", - ImmutableList.of("200811092030"), - "", - ImmutableList.of("0002"), - "", - ImmutableList.of("000296"), - "", - ImmutableList.of("0"), - "", - ImmutableList.of("-00296."), - "", - ImmutableList.of("-6620182933895093708")))); + ImmutableList.of("-1547954353065580372")))); } @Test @@ -301,41 +313,23 @@ public void testBrainAggregationMode_NotShowNumberedToken() throws IOException { schema("patterns_field", "string"), schema("pattern_count", "bigint"), schema("sample_logs", "array")); - verifyDataRows( - result, - rows( - "Verification succeeded <*> blk_<*>", - 2, - ImmutableList.of( - "Verification succeeded for blk_-1547954353065580372", - "Verification succeeded for blk_6996194389878584395")), - 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", - "BLOCK* NameSystem.addStoredBlock: blockMap updated: 10.251.107.19:50010 is added" - + " to blk_-3249711809227781266 size 67108864")), - 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", - "BLOCK* NameSystem.allocateBlock:" - + " /user/root/sortrand/_temporary/_task_200811092030_0002_r_000318_0/part-00318." - + " blk_2096692261399680562")), - rows( - "PacketResponder failed <*> blk_<*>", - 2, - ImmutableList.of( - "PacketResponder failed for blk_6996194389878584395", - "PacketResponder failed for blk_-1547954353065580372"))); + // Each pattern's count is 2 and its sample_logs covers both matching documents; only the order + // within sample_logs (and the order of the pattern rows) varies across shards. Assert the four + // patterns, their counts, and their full sample sets order-insensitively. + Map> expectedSamples = new HashMap<>(); + expectedSamples.put(PAT_VERIF, Arrays.asList(LOG_VERIF_4, LOG_VERIF_6)); + expectedSamples.put(PAT_ADDBLK, Arrays.asList(LOG_ADDBLK_1, LOG_ADDBLK_3)); + expectedSamples.put(PAT_ALLOC, Arrays.asList(LOG_ALLOC_2, LOG_ALLOC_5)); + expectedSamples.put(PAT_PACKET, Arrays.asList(LOG_PACKET_7, LOG_PACKET_8)); + + List> rows = dataRows(result); + assertEquals(4, rows.size()); + for (List row : rows) { + String pattern = (String) row.get(0); + assertTrue("unexpected pattern: " + pattern, expectedSamples.containsKey(pattern)); + assertEquals("count for " + pattern, 2L, asLong(row.get(1))); + assertListEqualsIgnoreOrder(expectedSamples.get(pattern), asStringList(row.get(2))); + } } @Test @@ -352,73 +346,51 @@ public void testBrainAggregationMode_ShowNumberedToken() throws IOException { schema("pattern_count", "bigint"), schema("tokens", "struct"), schema("sample_logs", "array")); - verifyDataRows( - result, - rows( - "Verification succeeded blk_", - 2, - ImmutableMap.of( - "", - ImmutableList.of("for", "for"), - "", - ImmutableList.of("-1547954353065580372", "6996194389878584395")), - ImmutableList.of( - "Verification succeeded for blk_-1547954353065580372", - "Verification succeeded for blk_6996194389878584395")), - rows( - "BLOCK* NameSystem.addStoredBlock: blockMap updated: is added to blk_" - + " size ", - 2, - ImmutableMap.of( - "", - ImmutableList.of("10.251.31.85:50010", "10.251.107.19:50010"), - "", - ImmutableList.of("67108864", "67108864"), - "", - ImmutableList.of("-7017553867379051457", "-3249711809227781266")), - ImmutableList.of( - "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")), - rows( - " NameSystem.allocateBlock:" - + " /user/root/sortrand/_temporary/_task___r__/part" - + " blk_", - 2, - ImmutableMap.of( - "", - ImmutableList.of("0", "0"), - "", - ImmutableList.of("000296", "000318"), - "", - ImmutableList.of("-6620182933895093708", "2096692261399680562"), - "", - ImmutableList.of("-00296.", "-00318."), - "", - ImmutableList.of("BLOCK*", "BLOCK*"), - "", - ImmutableList.of("0002", "0002"), - "", - ImmutableList.of("200811092030", "200811092030")), - ImmutableList.of( - "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")), - rows( - "PacketResponder failed blk_", - 2, - ImmutableMap.of( - "", - ImmutableList.of("for", "for"), - "", - ImmutableList.of("6996194389878584395", "-1547954353065580372")), - ImmutableList.of( - "PacketResponder failed for blk_6996194389878584395", - "PacketResponder failed for blk_-1547954353065580372"))); + // Count, token layout and sample content are shard-invariant; only the ordering within the + // token value lists and sample_logs varies. Assert both order-insensitively. + Map> expectedSamples = new HashMap<>(); + expectedSamples.put(PAT_NUMBERED_VERIF, Arrays.asList(LOG_VERIF_4, LOG_VERIF_6)); + expectedSamples.put(PAT_NUMBERED_ADDBLK, Arrays.asList(LOG_ADDBLK_1, LOG_ADDBLK_3)); + expectedSamples.put(PAT_NUMBERED_ALLOC, Arrays.asList(LOG_ALLOC_2, LOG_ALLOC_5)); + expectedSamples.put(PAT_NUMBERED_PACKET, Arrays.asList(LOG_PACKET_7, LOG_PACKET_8)); + + Map>> expectedTokens = new HashMap<>(); + expectedTokens.put( + PAT_NUMBERED_VERIF, + Map.of( + "", Arrays.asList("for", "for"), + "", Arrays.asList("-1547954353065580372", "6996194389878584395"))); + expectedTokens.put( + PAT_NUMBERED_ADDBLK, + Map.of( + "", Arrays.asList("10.251.31.85:50010", "10.251.107.19:50010"), + "", Arrays.asList("-7017553867379051457", "-3249711809227781266"), + "", Arrays.asList("67108864", "67108864"))); + expectedTokens.put( + PAT_NUMBERED_ALLOC, + Map.of( + "", Arrays.asList("BLOCK*", "BLOCK*"), + "", Arrays.asList("200811092030", "200811092030"), + "", Arrays.asList("0002", "0002"), + "", Arrays.asList("000296", "000318"), + "", Arrays.asList("0", "0"), + "", Arrays.asList("-00296.", "-00318."), + "", Arrays.asList("-6620182933895093708", "2096692261399680562"))); + expectedTokens.put( + PAT_NUMBERED_PACKET, + Map.of( + "", Arrays.asList("for", "for"), + "", Arrays.asList("6996194389878584395", "-1547954353065580372"))); + + List> rows = dataRows(result); + assertEquals(4, rows.size()); + for (List row : rows) { + String pattern = (String) row.get(0); + assertTrue("unexpected pattern: " + pattern, expectedSamples.containsKey(pattern)); + assertEquals("count for " + pattern, 2L, asLong(row.get(1))); + assertTokensEqualIgnoreOrder(expectedTokens.get(pattern), row.get(2)); + assertListEqualsIgnoreOrder(expectedSamples.get(pattern), asStringList(row.get(3))); + } } @Test @@ -437,69 +409,63 @@ public void testBrainAggregationModeWithGroupByClause_ShowNumberedToken() throws schema("pattern_count", "bigint"), schema("tokens", "struct"), schema("sample_logs", "array")); - verifyDataRows( - result, - rows( - "INFO", - "Verification succeeded for blk_", - 2, - ImmutableMap.of( - "", ImmutableList.of("-1547954353065580372", "6996194389878584395")), - ImmutableList.of( - "Verification succeeded for blk_-1547954353065580372", - "Verification succeeded for blk_6996194389878584395")), - rows( - "INFO", - "BLOCK* NameSystem.addStoredBlock: blockMap updated: is added to blk_" - + " size ", - 2, - ImmutableMap.of( - "", - ImmutableList.of("10.251.31.85:50010", "10.251.107.19:50010"), - "", - ImmutableList.of("67108864", "67108864"), - "", - ImmutableList.of("-7017553867379051457", "-3249711809227781266")), - ImmutableList.of( - "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")), - rows( - "INFO", - "BLOCK* NameSystem.allocateBlock:" - + " /user/root/sortrand/_temporary/_task___r__/part" - + " blk_", - 2, - ImmutableMap.of( - "", - ImmutableList.of("-00296.", "-00318."), - "", - ImmutableList.of("0", "0"), - "", - ImmutableList.of("-6620182933895093708", "2096692261399680562"), - "", - ImmutableList.of("200811092030", "200811092030"), - "", - ImmutableList.of("000296", "000318"), - "", - ImmutableList.of("0002", "0002")), - ImmutableList.of( - "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")), - rows( - "WARN", - "PacketResponder failed for blk_", - 2, - ImmutableMap.of( - "", ImmutableList.of("6996194389878584395", "-1547954353065580372")), - ImmutableList.of( - "PacketResponder failed for blk_6996194389878584395", - "PacketResponder failed for blk_-1547954353065580372"))); + + // variable_count_threshold=2 keeps more literals fixed, so the patterns differ from the + // ungrouped tests. level, pattern, count and content are shard-invariant; only the ordering + // inside token lists and sample_logs varies. + final String vPat = "Verification succeeded for blk_"; + final String aPat = + "BLOCK* NameSystem.addStoredBlock: blockMap updated: is added to blk_" + + " size "; + final String allocPat = + "BLOCK* NameSystem.allocateBlock:" + + " /user/root/sortrand/_temporary/_task___r__/part" + + " blk_"; + final String pPat = "PacketResponder failed for blk_"; + + Map expectedLevel = new HashMap<>(); + expectedLevel.put(vPat, "INFO"); + expectedLevel.put(aPat, "INFO"); + expectedLevel.put(allocPat, "INFO"); + expectedLevel.put(pPat, "WARN"); + + Map> expectedSamples = new HashMap<>(); + expectedSamples.put(vPat, Arrays.asList(LOG_VERIF_4, LOG_VERIF_6)); + expectedSamples.put(aPat, Arrays.asList(LOG_ADDBLK_1, LOG_ADDBLK_3)); + expectedSamples.put(allocPat, Arrays.asList(LOG_ALLOC_2, LOG_ALLOC_5)); + expectedSamples.put(pPat, Arrays.asList(LOG_PACKET_7, LOG_PACKET_8)); + + Map>> expectedTokens = new HashMap<>(); + expectedTokens.put( + vPat, Map.of("", Arrays.asList("-1547954353065580372", "6996194389878584395"))); + expectedTokens.put( + aPat, + Map.of( + "", Arrays.asList("10.251.31.85:50010", "10.251.107.19:50010"), + "", Arrays.asList("-7017553867379051457", "-3249711809227781266"), + "", Arrays.asList("67108864", "67108864"))); + expectedTokens.put( + allocPat, + Map.of( + "", Arrays.asList("200811092030", "200811092030"), + "", Arrays.asList("0002", "0002"), + "", Arrays.asList("000296", "000318"), + "", Arrays.asList("0", "0"), + "", Arrays.asList("-00296.", "-00318."), + "", Arrays.asList("-6620182933895093708", "2096692261399680562"))); + expectedTokens.put( + pPat, Map.of("", Arrays.asList("6996194389878584395", "-1547954353065580372"))); + + List> rows = dataRows(result); + assertEquals(4, rows.size()); + for (List row : rows) { + String pattern = (String) row.get(1); + assertTrue("unexpected pattern: " + pattern, expectedLevel.containsKey(pattern)); + assertEquals("level for " + pattern, expectedLevel.get(pattern), row.get(0)); + assertEquals("count for " + pattern, 2L, asLong(row.get(2))); + assertTokensEqualIgnoreOrder(expectedTokens.get(pattern), row.get(3)); + assertListEqualsIgnoreOrder(expectedSamples.get(pattern), asStringList(row.get(4))); + } } @Test @@ -531,4 +497,104 @@ public void testBrainParseWithUUID_ShowNumberedToken() throws IOException { "[PlaceOrder] user_id= user_currency=USD", ImmutableMap.of("", ImmutableList.of("d664d7be-77d8-11f0-8880-0242f00b101d")))); } + + // ---- Multi-shard patterns helpers --------------------------------------------------------- + + // hdfs_logs content lines (suffix = _id) and their BRAIN aggregation-mode pattern labels. + private static final String LOG_ADDBLK_1 = + "BLOCK* NameSystem.addStoredBlock: blockMap updated: 10.251.31.85:50010 is added to" + + " blk_-7017553867379051457 size 67108864"; + private static final String LOG_ADDBLK_3 = + "BLOCK* NameSystem.addStoredBlock: blockMap updated: 10.251.107.19:50010 is added to" + + " blk_-3249711809227781266 size 67108864"; + private static final String LOG_ALLOC_2 = + "BLOCK* NameSystem.allocateBlock:" + + " /user/root/sortrand/_temporary/_task_200811092030_0002_r_000296_0/part-00296." + + " blk_-6620182933895093708"; + private static final String LOG_ALLOC_5 = + "BLOCK* NameSystem.allocateBlock:" + + " /user/root/sortrand/_temporary/_task_200811092030_0002_r_000318_0/part-00318." + + " blk_2096692261399680562"; + private static final String LOG_VERIF_4 = "Verification succeeded for blk_-1547954353065580372"; + private static final String LOG_VERIF_6 = "Verification succeeded for blk_6996194389878584395"; + private static final String LOG_PACKET_7 = "PacketResponder failed for blk_6996194389878584395"; + private static final String LOG_PACKET_8 = "PacketResponder failed for blk_-1547954353065580372"; + + private static final String PAT_ADDBLK = + "BLOCK* NameSystem.addStoredBlock: blockMap updated: <*IP*> is added to blk_<*> size <*>"; + private static final String PAT_ALLOC = + "<*> NameSystem.allocateBlock:" + + " /user/root/sortrand/_temporary/_task_<*>_<*>_r_<*>_<*>/part<*> blk_<*>"; + private static final String PAT_VERIF = "Verification succeeded <*> blk_<*>"; + private static final String PAT_PACKET = "PacketResponder failed <*> blk_<*>"; + + private static final String PAT_NUMBERED_ADDBLK = + "BLOCK* NameSystem.addStoredBlock: blockMap updated: is added to blk_ size" + + " "; + private static final String PAT_NUMBERED_ALLOC = + " NameSystem.allocateBlock:" + + " /user/root/sortrand/_temporary/_task___r__/part" + + " blk_"; + private static final String PAT_NUMBERED_VERIF = "Verification succeeded blk_"; + private static final String PAT_NUMBERED_PACKET = "PacketResponder failed blk_"; + + private static final Pattern EMAIL = Pattern.compile("^[^@]+@[^.]+\\.[^.]+$"); + + /** Materialize {@code datarows} into a list of rows, mapping JSON null to Java {@code null}. */ + private static List> dataRows(JSONObject response) { + List> rows = new ArrayList<>(); + JSONArray arr = response.getJSONArray("datarows"); + for (int i = 0; i < arr.length(); i++) { + JSONArray r = arr.getJSONArray(i); + List 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; + } + + private static long asLong(Object value) { + return ((Number) value).longValue(); + } + + private static List asStringList(Object cell) { + List out = new ArrayList<>(); + JSONArray arr = (JSONArray) cell; + for (int i = 0; i < arr.length(); i++) { + out.add(arr.isNull(i) ? null : String.valueOf(arr.get(i))); + } + return out; + } + + /** A patterns {@code tokens} struct as a map of token name to its captured value list. */ + private static Map> tokenMap(Object cell) { + Map> out = new HashMap<>(); + JSONObject obj = (JSONObject) cell; + for (String key : obj.keySet()) { + out.put(key, asStringList(obj.get(key))); + } + return out; + } + + private static List sortedStr(List in) { + List copy = new ArrayList<>(in); + copy.sort(Comparator.nullsFirst(Comparator.naturalOrder())); + return copy; + } + + private static void assertListEqualsIgnoreOrder(List expected, List actual) { + assertEquals(sortedStr(expected), sortedStr(actual)); + } + + private static void assertTokensEqualIgnoreOrder( + Map> expected, Object tokensCell) { + Map> actual = tokenMap(tokensCell); + assertEquals(expected.keySet(), actual.keySet()); + for (Map.Entry> e : expected.entrySet()) { + assertEquals( + "token " + e.getKey(), sortedStr(e.getValue()), sortedStr(actual.get(e.getKey()))); + } + } } diff --git a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteReverseCommandIT.java b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteReverseCommandIT.java index fb88e9be73b..d84a47a3b14 100644 --- a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteReverseCommandIT.java +++ b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteReverseCommandIT.java @@ -6,7 +6,7 @@ package org.opensearch.sql.calcite.remote; import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_BANK; -import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_STATE_COUNTRY; +import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_STATE_COUNTRY_SINGLE_SHARD; import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_TIME_DATA; import static org.opensearch.sql.util.Capability.WILDCARD_COLUMN_ORDER; import static org.opensearch.sql.util.MatcherUtils.rows; @@ -31,9 +31,21 @@ public void init() throws Exception { loadIndex(Index.BANK); loadIndex(Index.TIME_TEST_DATA); loadIndex(Index.STATE_COUNTRY); + loadIndex(Index.STATE_COUNTRY_SINGLE_SHARD); loadIndex(Index.EVENTS); } + // See CalciteStreamstatsCommandIT: streamstats runs over the input stream in encounter order, + // which is non-deterministic on a multi-shard index. A deterministic in-memory makeresults + // stream (mirroring STATE_COUNTRY in the column order the index presents) keeps the streamstats + // values -- and therefore reverse's behavior over them -- stable across shard layouts. + private static final String SC = + "name:string,country:string,state:string,month:int,year:int,age:int\\n" + + "Jake,USA,California,4,2023,70\\n" + + "Hello,USA,New York,4,2023,30\\n" + + "John,Canada,Ontario,4,2023,25\\n" + + "Jane,Canada,Quebec,4,2023,20"; + @Test public void testReverse() throws IOException { JSONObject result = @@ -182,15 +194,18 @@ public void testDoubleReverseWithMixedSortDirections() throws IOException { @Test public void testReverseIgnoredWithoutSortOrTimestamp() throws IOException { - // Test that reverse is ignored when there's no explicit sort and no @timestamp field - // BANK index doesn't have @timestamp, so reverse should be ignored + // Purpose: reverse is a plan-level no-op when there is no explicit sort and no @timestamp + // field. This is value/plan semantics, not index-route behavior, and a real index makes + // `head 3` without a sort non-deterministic across shards. A coordinator-only makeresults + // stream has no @timestamp and no sort, so reverse is ignored and the rows stay in stream + // (insertion) order, exercising the same reverse-elimination path deterministically. JSONObject result = executeQuery( - String.format("source=%s | fields account_number | reverse | head 3", TEST_INDEX_BANK)); - verifySchema(result, schema("account_number", "bigint")); - // Without sort or @timestamp, reverse is ignored, so data comes in natural order - // The first 3 documents in natural order (ascending by account_number) - verifyDataRowsInOrder(result, rows(1), rows(6), rows(13)); + String.format("makeresults format=csv data='%s' | fields age | reverse | head 3", SC)); + verifySchema(result, schema("age", "int")); + // Without sort or @timestamp, reverse is ignored, so the rows stay in natural (stream) order: + // the first 3 rows of the stream. If reverse were honored, this would be 20, 25, 30. + verifyDataRowsInOrder(result, rows(70), rows(30), rows(25)); } @Test @@ -240,12 +255,16 @@ public void testReverseWithTimestampAndExplicitSort() throws IOException { public void testStreamstatsWithReverse() throws IOException { // Test that reverse is ignored when used directly after streamstats // streamstats maintains order via __stream_seq__, but this field is projected out - // and doesn't create a detectable collation, so reverse is ignored (no-op) + // and doesn't create a detectable collation, so reverse is ignored (no-op). + // The no-op depends on there being NO upstream sort, so we cannot use `sort seq` to make the + // encounter order deterministic here (a sort would change the behavior under test). Instead the + // fixture pins number_of_shards=1, so the encounter order is the deterministic insertion order + // on the index route without adding a sort. Values are unchanged from the shared fixture. JSONObject result = executeQuery( String.format( "source=%s | streamstats count() as cnt, avg(age) as avg | reverse", - TEST_INDEX_STATE_COUNTRY)); + TEST_INDEX_STATE_COUNTRY_SINGLE_SHARD)); verifySchema( result, schema("name", "string"), @@ -272,12 +291,14 @@ public void testStreamstatsWithReverse() throws IOException { "streamstats carries all source columns through; the AE route returns them in a different" + " order (WILDCARD_COLUMN_ORDER).") public void testStreamstatsWindowWithReverse() throws IOException { - // Test that reverse is ignored after streamstats with window + // Test that reverse is ignored after streamstats with window. + // Same rationale as testStreamstatsWithReverse: the no-op precludes an upstream sort, so the + // fixture pins number_of_shards=1 to get a deterministic encounter order on the index route. JSONObject result = executeQuery( String.format( "source=%s | streamstats window=2 avg(age) as avg | reverse", - TEST_INDEX_STATE_COUNTRY)); + TEST_INDEX_STATE_COUNTRY_SINGLE_SHARD)); verifySchema( result, schema("name", "string"), @@ -309,8 +330,9 @@ public void testStreamstatsByWithReverse() throws IOException { JSONObject result = executeQuery( String.format( - "source=%s | streamstats count() as cnt, avg(age) as avg by country | reverse", - TEST_INDEX_STATE_COUNTRY)); + "makeresults format=csv data='%s' | streamstats count() as cnt, avg(age) as avg by" + + " country | reverse", + SC)); verifySchema( result, schema("name", "string"), @@ -342,8 +364,9 @@ public void testStreamstatsWithSortThenReverse() throws IOException { JSONObject result = executeQuery( String.format( - "source=%s | streamstats count() as cnt | sort age | reverse | head 3", - TEST_INDEX_STATE_COUNTRY)); + "makeresults format=csv data='%s' | streamstats count() as cnt | sort age | reverse" + + " | head 3", + SC)); verifySchema( result, schema("name", "string"), diff --git a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteStreamstatsCommandIT.java b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteStreamstatsCommandIT.java index e70812e3c3b..ee34f6e3d3d 100644 --- a/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteStreamstatsCommandIT.java +++ b/integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteStreamstatsCommandIT.java @@ -12,7 +12,12 @@ import static org.opensearch.sql.util.MatcherUtils.*; import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; import java.util.List; +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.client.Request; @@ -26,19 +31,51 @@ public void init() throws Exception { enableCalcite(); loadIndex(Index.STATE_COUNTRY); loadIndex(Index.STATE_COUNTRY_WITH_NULL); + loadIndex(Index.STATE_COUNTRY_ORDERED); + loadIndex(Index.STATE_COUNTRY_WITH_NULL_ORDERED); + loadIndex(Index.STATE_COUNTRY_WITH_NULL_SINGLE_SHARD); loadIndex(Index.BANK_TWO); loadIndex(Index.LOGS); } + // streamstats computes running/window aggregates over the input stream in encounter order. On a + // multi-shard index that order is not deterministic, so the per-row cumulative values diverge + // between runs. To assert exact stream semantics we feed a deterministic in-memory stream via + // `makeresults format=csv` (typed inline rows, single relation, no shards) that mirrors the + // STATE_COUNTRY / STATE_COUNTRY_WITH_NULL fixtures row-for-row and in the same column order the + // index presents (name, country, state, month, year, age). The expected values are unchanged -- + // only the source is made deterministic. Real multi-shard index coverage is retained separately + // by testStreamstatsIndexMultiShardCoverage below, which asserts order-independent properties. + private static final String SC = + "name:string,country:string,state:string,month:int,year:int,age:int\\n" + + "Jake,USA,California,4,2023,70\\n" + + "Hello,USA,New York,4,2023,30\\n" + + "John,Canada,Ontario,4,2023,25\\n" + + "Jane,Canada,Quebec,4,2023,20"; + + // Doc-mutation fixtures: the original tests PUT an extra "Jay" doc as the last document. These + // deterministic streams preserve that order without mutating a shared index. + private static final String SC_JAY40 = SC + "\\nJay,USA,Quebec,4,2023,40"; + private static final String SC_JAY28 = SC + "\\nJay,USA,Quebec,4,2023,28"; + + private static final String LOGS_STREAM = + "created_at:string,server:string,@timestamp:string,message:string,level:string\\n" + + "2023-01-05T00:00:00.000Z,server1,2023-01-01T00:00:00.000Z,Database connection" + + " failed,ERROR\\n" + + "2023-01-04T00:00:00.000Z,server2,2023-01-02T00:00:00.000Z,Service started,INFO\\n" + + "2023-01-03T00:00:00.000Z,server1,2023-01-03T00:00:00.000Z,High memory usage,WARN\\n" + + "2023-01-02T00:00:00.000Z,server3,2023-01-04T00:00:00.000Z,Disk space low,ERROR\\n" + + "2023-01-01T00:00:00.000Z,server2,2023-01-05T00:00:00.000Z,Backup completed,INFO"; + @Test public void testStreamstats() throws IOException { JSONObject actual = executeQuery( String.format( - "source=%s | streamstats count() as cnt, avg(age) as avg, min(age) as min, max(age)" - + " as max | fields name, country, state, month, year, age, cnt, avg, min," - + " max", - TEST_INDEX_STATE_COUNTRY)); + "makeresults format=csv data='%s' | streamstats count() as cnt, avg(age) as avg," + + " min(age) as min, max(age) as max | fields name, country, state, month," + + " year, age, cnt, avg, min, max", + SC)); verifySchemaInOrder( actual, @@ -62,14 +99,20 @@ public void testStreamstats() throws IOException { } @Test + // Multi-shard determinism: streamstats orders by encounter order, which is non-deterministic + // across shards. Sourcing the seq-augmented fixture and adding `| sort seq` restores the + // single-shard encounter order on any shard layout; the AE route ignores that sort + // (STREAMSTATS_SORT_NOT_HONORED), so this exact-value assertion is gated to the routes that + // honor it. The expected rows are unchanged. + @RequiresCapability(STREAMSTATS_SORT_NOT_HONORED) public void testStreamstatsWithNull() throws IOException { JSONObject actual = executeQuery( String.format( - "source=%s | streamstats count() as cnt, avg(age) as avg, min(age) as min, max(age)" - + " as max | fields name, country, state, month, year, age, cnt, avg, min," - + " max", - TEST_INDEX_STATE_COUNTRY_WITH_NULL)); + "source=%s | sort seq | streamstats count() as cnt, avg(age) as avg, min(age) as" + + " min, max(age) as max | fields name, country, state, month, year, age, cnt," + + " avg, min, max", + TEST_INDEX_STATE_COUNTRY_WITH_NULL_ORDERED)); verifySchemaInOrder( actual, @@ -99,10 +142,10 @@ public void testStreamstatsBy() throws IOException { JSONObject actual = executeQuery( String.format( - "source=%s | streamstats count() as cnt, avg(age) as avg, min(age) as min, max(age)" - + " as max by country | fields name, country, state, month, year, age, cnt," - + " avg, min, max", - TEST_INDEX_STATE_COUNTRY)); + "makeresults format=csv data='%s' | streamstats count() as cnt, avg(age) as avg," + + " min(age) as min, max(age) as max by country | fields name, country, state," + + " month, year, age, cnt, avg, min, max", + SC)); verifySchemaInOrder( actual, @@ -126,14 +169,18 @@ public void testStreamstatsBy() throws IOException { } @Test + // See testStreamstatsWithNull: `| sort seq` on the seq-augmented fixture restores the + // single-shard encounter order across shards; gated to routes that honor a sort before + // streamstats. Expected rows are unchanged. + @RequiresCapability(STREAMSTATS_SORT_NOT_HONORED) public void testStreamstatsByWithNull() throws IOException { JSONObject actual = executeQuery( String.format( - "source=%s | streamstats count() as cnt, avg(age) as avg, min(age) as min, max(age)" - + " as max by country | fields name, country, state, month, year, age, cnt," - + " avg, min, max", - TEST_INDEX_STATE_COUNTRY_WITH_NULL)); + "source=%s | sort seq | streamstats count() as cnt, avg(age) as avg, min(age) as" + + " min, max(age) as max by country | fields name, country, state, month, year," + + " age, cnt, avg, min, max", + TEST_INDEX_STATE_COUNTRY_WITH_NULL_ORDERED)); verifySchemaInOrder( actual, @@ -160,10 +207,10 @@ public void testStreamstatsByWithNull() throws IOException { actual = executeQuery( String.format( - "source=%s | streamstats count() as cnt, avg(age) as avg, min(age) as min, max(age)" - + " as max by state | fields name, country, state, month, year, age, cnt, avg," - + " min, max", - TEST_INDEX_STATE_COUNTRY_WITH_NULL)); + "source=%s | sort seq | streamstats count() as cnt, avg(age) as avg, min(age) as" + + " min, max(age) as max by state | fields name, country, state, month, year," + + " age, cnt, avg, min, max", + TEST_INDEX_STATE_COUNTRY_WITH_NULL_ORDERED)); verifyDataRows( actual, rows("Jake", "USA", "California", 4, 2023, 70, 1, 70, 70, 70), @@ -175,14 +222,20 @@ public void testStreamstatsByWithNull() throws IOException { } @Test + // Multi-shard determinism: streamstats orders by encounter order, which is non-deterministic + // across shards. Sourcing the seq-augmented fixture and adding `| sort seq` restores the + // single-shard encounter order on any shard layout (the Calcite/Lucene route honors a sort + // before streamstats). The AE route ignores that sort (STREAMSTATS_SORT_NOT_HONORED), so this + // exact-value assertion is gated to the routes that honor it; the expected rows are unchanged. + @RequiresCapability(STREAMSTATS_SORT_NOT_HONORED) public void testStreamstatsByWithNullBucket() throws IOException { JSONObject actual = executeQuery( String.format( - "source=%s | streamstats bucket_nullable=false count() as cnt, avg(age) as avg," - + " min(age) as min, max(age) as max by country | fields name, country, state," - + " month, year, age, cnt, avg, min, max", - TEST_INDEX_STATE_COUNTRY_WITH_NULL)); + "source=%s | sort seq | streamstats bucket_nullable=false count() as cnt, avg(age)" + + " as avg, min(age) as min, max(age) as max by country | fields name, country," + + " state, month, year, age, cnt, avg, min, max", + TEST_INDEX_STATE_COUNTRY_WITH_NULL_ORDERED)); verifySchemaInOrder( actual, @@ -209,10 +262,10 @@ public void testStreamstatsByWithNullBucket() throws IOException { actual = executeQuery( String.format( - "source=%s | streamstats bucket_nullable=false count() as cnt, avg(age) as avg," - + " min(age) as min, max(age) as max by state | fields name, country, state," - + " month, year, age, cnt, avg, min, max", - TEST_INDEX_STATE_COUNTRY_WITH_NULL)); + "source=%s | sort seq | streamstats bucket_nullable=false count() as cnt, avg(age)" + + " as avg, min(age) as min, max(age) as max by state | fields name, country," + + " state, month, year, age, cnt, avg, min, max", + TEST_INDEX_STATE_COUNTRY_WITH_NULL_ORDERED)); verifyDataRows( actual, rows("Jake", "USA", "California", 4, 2023, 70, 1, 70, 70, 70), @@ -228,10 +281,10 @@ public void testStreamstatsBySpan() throws IOException { JSONObject actual = executeQuery( String.format( - "source=%s | streamstats count() as cnt, avg(age) as avg, min(age) as min, max(age)" - + " as max by span(age, 10) as age_span | fields name, country, state, month," - + " year, age, cnt, avg, min, max", - TEST_INDEX_STATE_COUNTRY)); + "makeresults format=csv data='%s' | streamstats count() as cnt, avg(age) as avg," + + " min(age) as min, max(age) as max by span(age, 10) as age_span | fields" + + " name, country, state, month, year, age, cnt, avg, min, max", + SC)); verifyDataRows( actual, @@ -266,10 +319,10 @@ public void testStreamstatsByMultiplePartitions1() throws IOException { JSONObject actual = executeQuery( String.format( - "source=%s | streamstats count() as cnt, avg(age) as avg, min(age) as min, max(age)" - + " as max by span(age, 10) as age_span, country | fields name, country, state," - + " month, year, age, cnt, avg, min, max", - TEST_INDEX_STATE_COUNTRY)); + "makeresults format=csv data='%s' | streamstats count() as cnt, avg(age) as avg," + + " min(age) as min, max(age) as max by span(age, 10) as age_span, country |" + + " fields name, country, state, month, year, age, cnt, avg, min, max", + SC)); verifyDataRows( actual, @@ -284,10 +337,10 @@ public void testStreamstatsByMultiplePartitions2() throws IOException { JSONObject actual = executeQuery( String.format( - "source=%s | streamstats count() as cnt, avg(age) as avg, min(age) as min, max(age)" - + " as max by span(age, 10) as age_span, state | fields name, country, state," - + " month, year, age, cnt, avg, min, max", - TEST_INDEX_STATE_COUNTRY)); + "makeresults format=csv data='%s' | streamstats count() as cnt, avg(age) as avg," + + " min(age) as min, max(age) as max by span(age, 10) as age_span, state |" + + " fields name, country, state, month, year, age, cnt, avg, min, max", + SC)); verifyDataRows( actual, @@ -376,9 +429,9 @@ public void testStreamstatsCurrent() throws IOException { JSONObject actual = executeQuery( String.format( - "source=%s | streamstats current=false avg(age) as prev_avg | fields name, country," - + " state, month, year, age, prev_avg", - TEST_INDEX_STATE_COUNTRY)); + "makeresults format=csv data='%s' | streamstats current=false avg(age) as prev_avg" + + " | fields name, country, state, month, year, age, prev_avg", + SC)); verifyDataRows( actual, @@ -389,13 +442,17 @@ public void testStreamstatsCurrent() throws IOException { } @Test + // See testStreamstatsWithNull: `| sort seq` on the seq-augmented fixture restores the + // single-shard encounter order across shards; gated to routes that honor a sort before + // streamstats. Expected rows are unchanged. + @RequiresCapability(STREAMSTATS_SORT_NOT_HONORED) public void testStreamstatsCurrentWithNUll() throws IOException { JSONObject actual = executeQuery( String.format( - "source=%s | streamstats current=false avg(age) as prev_avg | fields name, country," - + " state, month, year, age, prev_avg", - TEST_INDEX_STATE_COUNTRY_WITH_NULL)); + "source=%s | sort seq | streamstats current=false avg(age) as prev_avg | fields" + + " name, country, state, month, year, age, prev_avg", + TEST_INDEX_STATE_COUNTRY_WITH_NULL_ORDERED)); verifyDataRows( actual, @@ -412,9 +469,9 @@ public void testStreamstatsWindow() throws IOException { JSONObject actual = executeQuery( String.format( - "source=%s | streamstats window = 3 avg(age) as avg | fields name, country, state," - + " month, year, age, avg", - TEST_INDEX_STATE_COUNTRY)); + "makeresults format=csv data='%s' | streamstats window = 3 avg(age) as avg | fields" + + " name, country, state, month, year, age, avg", + SC)); verifyDataRows( actual, @@ -425,13 +482,17 @@ public void testStreamstatsWindow() throws IOException { } @Test + // See testStreamstatsWithNull: `| sort seq` on the seq-augmented fixture restores the + // single-shard encounter order across shards; gated to routes that honor a sort before + // streamstats. Expected rows are unchanged. + @RequiresCapability(STREAMSTATS_SORT_NOT_HONORED) public void testStreamstatsWindowWithNull() throws IOException { JSONObject actual = executeQuery( String.format( - "source=%s | streamstats window = 3 avg(age) as avg | fields name, country, state," - + " month, year, age, avg", - TEST_INDEX_STATE_COUNTRY_WITH_NULL)); + "source=%s | sort seq | streamstats window = 3 avg(age) as avg | fields name," + + " country, state, month, year, age, avg", + TEST_INDEX_STATE_COUNTRY_WITH_NULL_ORDERED)); verifyDataRows( actual, @@ -448,9 +509,9 @@ public void testStreamstatsBigWindow() throws IOException { JSONObject actual = executeQuery( String.format( - "source=%s | streamstats window = 10 avg(age) as avg | fields name, country, state," - + " month, year, age, avg", - TEST_INDEX_STATE_COUNTRY)); + "makeresults format=csv data='%s' | streamstats window = 10 avg(age) as avg |" + + " fields name, country, state, month, year, age, avg", + SC)); verifyDataRows( actual, @@ -478,9 +539,9 @@ public void testStreamstatsCurrentAndWindow() throws IOException { JSONObject actual = executeQuery( String.format( - "source=%s | streamstats current = false window = 2 avg(age) as avg | fields name," - + " country, state, month, year, age, avg", - TEST_INDEX_STATE_COUNTRY)); + "makeresults format=csv data='%s' | streamstats current = false window = 2 avg(age)" + + " as avg | fields name, country, state, month, year, age, avg", + SC)); verifyDataRows( actual, @@ -491,13 +552,17 @@ public void testStreamstatsCurrentAndWindow() throws IOException { } @Test + // See testStreamstatsWithNull: `| sort seq` on the seq-augmented fixture restores the + // single-shard encounter order across shards; gated to routes that honor a sort before + // streamstats. Expected rows are unchanged. + @RequiresCapability(STREAMSTATS_SORT_NOT_HONORED) public void testStreamstatsCurrentAndWindowWithNull() throws IOException { JSONObject actual = executeQuery( String.format( - "source=%s | streamstats current = false window = 2 avg(age) as avg | fields name," - + " country, state, month, year, age, avg", - TEST_INDEX_STATE_COUNTRY_WITH_NULL)); + "source=%s | sort seq | streamstats current = false window = 2 avg(age) as avg |" + + " fields name, country, state, month, year, age, avg", + TEST_INDEX_STATE_COUNTRY_WITH_NULL_ORDERED)); verifyDataRows( actual, @@ -510,75 +575,65 @@ public void testStreamstatsCurrentAndWindowWithNull() throws IOException { } @Test - @RequiresCapability(DOC_MUTATION) public void testStreamstatsGlobal() throws IOException { - final int docId = 5; - Request insertRequest = - new Request( - "PUT", String.format("/%s/_doc/%d?refresh=true", TEST_INDEX_STATE_COUNTRY, docId)); - insertRequest.setJsonEntity( - "{\"name\": \"Jay\",\"age\": 40,\"state\":" - + " \"Quebec\",\"country\": \"USA\",\"year\": 2023,\"month\":" - + " 4}\n"); - client().performRequest(insertRequest); - try { - JSONObject actual = - executeQuery( - String.format( - "source=%s | streamstats window=2 global=false avg(age) as avg by country |" - + " fields name, country, state, month, year, age, avg", - TEST_INDEX_STATE_COUNTRY)); + // Jay (age 40) is inlined as the trailing row of the deterministic stream, matching the + // original test which PUT it as the last document (encountered last) then DELETEd it. + JSONObject actual = + executeQuery( + String.format( + "makeresults format=csv data='%s' | streamstats window=2 global=false avg(age) as" + + " avg by country | fields name, country, state, month, year, age, avg", + SC_JAY40)); - verifyDataRows( - actual, - rows("Jake", "USA", "California", 4, 2023, 70, 70), - rows("Hello", "USA", "New York", 4, 2023, 30, 50), - rows("John", "Canada", "Ontario", 4, 2023, 25, 25), - rows("Jane", "Canada", "Quebec", 4, 2023, 20, 22.5), - rows("Jay", "USA", "Quebec", 4, 2023, 40, 35)); + verifyDataRows( + actual, + rows("Jake", "USA", "California", 4, 2023, 70, 70), + rows("Hello", "USA", "New York", 4, 2023, 30, 50), + rows("John", "Canada", "Ontario", 4, 2023, 25, 25), + rows("Jane", "Canada", "Quebec", 4, 2023, 20, 22.5), + rows("Jay", "USA", "Quebec", 4, 2023, 40, 35)); - JSONObject actual2 = - executeQuery( - String.format( - "source=%s | streamstats window=2 global=true avg(age) as avg by country |" - + " fields name, country, state, month, year, age, avg", - TEST_INDEX_STATE_COUNTRY)); + JSONObject actual2 = + executeQuery( + String.format( + "makeresults format=csv data='%s' | streamstats window=2 global=true avg(age) as" + + " avg by country | fields name, country, state, month, year, age, avg", + SC_JAY40)); - verifyDataRows( - actual2, - rows("Jake", "USA", "California", 4, 2023, 70, 70), - rows("Hello", "USA", "New York", 4, 2023, 30, 50), - rows("John", "Canada", "Ontario", 4, 2023, 25, 25), - rows("Jane", "Canada", "Quebec", 4, 2023, 20, 22.5), - rows("Jay", "USA", "Quebec", 4, 2023, 40, 40)); - } finally { - Request deleteRequest = - new Request( - "DELETE", String.format("/%s/_doc/%d?refresh=true", TEST_INDEX_STATE_COUNTRY, docId)); - client().performRequest(deleteRequest); - } + verifyDataRows( + actual2, + rows("Jake", "USA", "California", 4, 2023, 70, 70), + rows("Hello", "USA", "New York", 4, 2023, 30, 50), + rows("John", "Canada", "Ontario", 4, 2023, 25, 25), + rows("Jane", "Canada", "Quebec", 4, 2023, 20, 22.5), + rows("Jay", "USA", "Quebec", 4, 2023, 40, 40)); } @Test - @RequiresCapability(DOC_MUTATION) + @RequiresCapability({DOC_MUTATION, STREAMSTATS_SORT_NOT_HONORED}) public void testStreamstatsGlobalWithNull() throws IOException { + // Jay is PUT with seq 7 (one past the fixture's 6 rows) so `| sort seq` keeps it in the + // trailing encounter position the original single-shard test relied on, while the sort makes + // the order deterministic across shards. Mutation stays DOC_MUTATION-gated; the added sort is + // ignored on the AE route (STREAMSTATS_SORT_NOT_HONORED). Expected rows are unchanged. final int docId = 7; Request insertRequest = new Request( "PUT", - String.format("/%s/_doc/%d?refresh=true", TEST_INDEX_STATE_COUNTRY_WITH_NULL, docId)); + String.format( + "/%s/_doc/%d?refresh=true", TEST_INDEX_STATE_COUNTRY_WITH_NULL_ORDERED, docId)); insertRequest.setJsonEntity( "{\"name\": \"Jay\",\"age\": 40,\"state\":" + " \"Quebec\",\"country\": \"USA\",\"year\": 2023,\"month\":" - + " 4}\n"); + + " 4,\"seq\": 7}\n"); client().performRequest(insertRequest); try { JSONObject actual = executeQuery( String.format( - "source=%s | streamstats window=2 global=false avg(age) as avg by country |" - + " fields name, country, state, month, year, age, avg", - TEST_INDEX_STATE_COUNTRY_WITH_NULL)); + "source=%s | sort seq | streamstats window=2 global=false avg(age) as avg by" + + " country | fields name, country, state, month, year, age, avg", + TEST_INDEX_STATE_COUNTRY_WITH_NULL_ORDERED)); verifyDataRows( actual, @@ -593,9 +648,9 @@ public void testStreamstatsGlobalWithNull() throws IOException { JSONObject actual2 = executeQuery( String.format( - "source=%s | streamstats window=2 global=true avg(age) as avg by country |" - + " fields name, country, state, month, year, age, avg", - TEST_INDEX_STATE_COUNTRY_WITH_NULL)); + "source=%s | sort seq | streamstats window=2 global=true avg(age) as avg by" + + " country | fields name, country, state, month, year, age, avg", + TEST_INDEX_STATE_COUNTRY_WITH_NULL_ORDERED)); verifyDataRows( actual2, @@ -610,7 +665,8 @@ public void testStreamstatsGlobalWithNull() throws IOException { Request deleteRequest = new Request( "DELETE", - String.format("/%s/_doc/%d?refresh=true", TEST_INDEX_STATE_COUNTRY_WITH_NULL, docId)); + String.format( + "/%s/_doc/%d?refresh=true", TEST_INDEX_STATE_COUNTRY_WITH_NULL_ORDERED, docId)); client().performRequest(deleteRequest); } } @@ -672,63 +728,59 @@ public void testStreamstatsGlobalWithNullBucket() throws IOException { } @Test - @RequiresCapability(DOC_MUTATION) public void testStreamstatsReset() throws IOException { - final int docId = 5; - Request insertRequest = - new Request( - "PUT", String.format("/%s/_doc/%d?refresh=true", TEST_INDEX_STATE_COUNTRY, docId)); - insertRequest.setJsonEntity( - "{\"name\": \"Jay\",\"age\": 28,\"state\":" - + " \"Quebec\",\"country\": \"USA\",\"year\": 2023,\"month\":" - + " 4}\n"); - client().performRequest(insertRequest); - try { - JSONObject actual = - executeQuery( - String.format( - "source=%s | streamstats window=2 reset_before=age>29 avg(age) as avg by country" - + " | fields name, country, state, month, year, age, avg", - TEST_INDEX_STATE_COUNTRY)); + // Jay (age 28) is inlined as the trailing row of the deterministic stream, matching the + // original test which PUT it as the last document (encountered last) then DELETEd it. + JSONObject actual = + executeQuery( + String.format( + "makeresults format=csv data='%s' | streamstats window=2 reset_before=age>29" + + " avg(age) as avg by country | fields name, country, state, month, year, age," + + " avg", + SC_JAY28)); - verifyDataRows( - actual, - rows("Jake", "USA", "California", 4, 2023, 70, 70), - rows("Hello", "USA", "New York", 4, 2023, 30, 30), - rows("John", "Canada", "Ontario", 4, 2023, 25, 25), - rows("Jane", "Canada", "Quebec", 4, 2023, 20, 22.5), - rows("Jay", "USA", "Quebec", 4, 2023, 28, 28)); + verifyDataRows( + actual, + rows("Jake", "USA", "California", 4, 2023, 70, 70), + rows("Hello", "USA", "New York", 4, 2023, 30, 30), + rows("John", "Canada", "Ontario", 4, 2023, 25, 25), + rows("Jane", "Canada", "Quebec", 4, 2023, 20, 22.5), + rows("Jay", "USA", "Quebec", 4, 2023, 28, 28)); - JSONObject actual2 = - executeQuery( - String.format( - "source=%s | streamstats window=2 reset_after=age>22 avg(age) as avg by country" - + " | fields name, country, state, month, year, age, avg", - TEST_INDEX_STATE_COUNTRY)); + JSONObject actual2 = + executeQuery( + String.format( + "makeresults format=csv data='%s' | streamstats window=2 reset_after=age>22" + + " avg(age) as avg by country | fields name, country, state, month, year, age," + + " avg", + SC_JAY28)); - verifyDataRows( - actual2, - rows("Jake", "USA", "California", 4, 2023, 70, 70), - rows("Hello", "USA", "New York", 4, 2023, 30, 30), - rows("John", "Canada", "Ontario", 4, 2023, 25, 25), - rows("Jane", "Canada", "Quebec", 4, 2023, 20, 20), - rows("Jay", "USA", "Quebec", 4, 2023, 28, 28)); - } finally { - Request deleteRequest = - new Request( - "DELETE", String.format("/%s/_doc/%d?refresh=true", TEST_INDEX_STATE_COUNTRY, docId)); - client().performRequest(deleteRequest); - } + verifyDataRows( + actual2, + rows("Jake", "USA", "California", 4, 2023, 70, 70), + rows("Hello", "USA", "New York", 4, 2023, 30, 30), + rows("John", "Canada", "Ontario", 4, 2023, 25, 25), + rows("Jane", "Canada", "Quebec", 4, 2023, 20, 20), + rows("Jay", "USA", "Quebec", 4, 2023, 28, 28)); } @Test @RequiresCapability(DOC_MUTATION) public void testStreamstatsResetWithNull() throws IOException { + // reset_before/reset_after streamstats builds a self-correlated plan that the physical compiler + // cannot combine with an upstream `sort` (planner IndexOutOfBounds), so the seq-sort trick used + // by the other WithNull streamstats tests is unavailable here. Instead this drives a + // single-shard fixture whose encounter order is the deterministic insertion order on any run, + // reproducing the original single-shard behavior without injecting a sort. Jay is PUT as the + // last document (docId 7), matching the original trailing-encounter position. Expected rows are + // unchanged. final int docId = 7; Request insertRequest = new Request( "PUT", - String.format("/%s/_doc/%d?refresh=true", TEST_INDEX_STATE_COUNTRY_WITH_NULL, docId)); + String.format( + "/%s/_doc/%d?refresh=true", + TEST_INDEX_STATE_COUNTRY_WITH_NULL_SINGLE_SHARD, docId)); insertRequest.setJsonEntity( "{\"name\": \"Jay\",\"age\": 28,\"state\":" + " \"Quebec\",\"country\": \"USA\",\"year\": 2023,\"month\":" @@ -740,7 +792,7 @@ public void testStreamstatsResetWithNull() throws IOException { String.format( "source=%s | streamstats window=2 reset_before=age>29 avg(age) as avg by country" + " | fields name, country, state, month, year, age, avg", - TEST_INDEX_STATE_COUNTRY_WITH_NULL)); + TEST_INDEX_STATE_COUNTRY_WITH_NULL_SINGLE_SHARD)); verifyDataRows( actual, @@ -757,7 +809,7 @@ public void testStreamstatsResetWithNull() throws IOException { String.format( "source=%s | streamstats window=2 reset_after=age>22 avg(age) as avg by country" + " | fields name, country, state, month, year, age, avg", - TEST_INDEX_STATE_COUNTRY_WITH_NULL)); + TEST_INDEX_STATE_COUNTRY_WITH_NULL_SINGLE_SHARD)); verifyDataRows( actual2, @@ -772,7 +824,9 @@ public void testStreamstatsResetWithNull() throws IOException { Request deleteRequest = new Request( "DELETE", - String.format("/%s/_doc/%d?refresh=true", TEST_INDEX_STATE_COUNTRY_WITH_NULL, docId)); + String.format( + "/%s/_doc/%d?refresh=true", + TEST_INDEX_STATE_COUNTRY_WITH_NULL_SINGLE_SHARD, docId)); client().performRequest(deleteRequest); } } @@ -856,10 +910,10 @@ public void testMultipleStreamstats() throws IOException { JSONObject actual = executeQuery( String.format( - "source=%s | streamstats avg(age) as avg_age by state, country | streamstats" - + " avg(avg_age) as avg_state_age by country | fields name, country, state," - + " month, year, age, avg_age, avg_state_age", - TEST_INDEX_STATE_COUNTRY)); + "makeresults format=csv data='%s' | streamstats avg(age) as avg_age by state," + + " country | streamstats avg(avg_age) as avg_state_age by country | fields" + + " name, country, state, month, year, age, avg_age, avg_state_age", + SC)); verifyDataRows( actual, @@ -870,16 +924,18 @@ public void testMultipleStreamstats() throws IOException { } @Test - @RequiresCapability(CHAINED_STREAMSTATS_BY) + @RequiresCapability({CHAINED_STREAMSTATS_BY, STREAMSTATS_SORT_NOT_HONORED}) public void testMultipleStreamstatsWithWindow() throws IOException { - // Test case from GitHub issue #4800: chained streamstats with window=2 + // Test case from GitHub issue #4800: chained streamstats with window=2. + // `| sort seq` on the seq-augmented fixture makes the encounter order deterministic across + // shards; gated to routes that honor a sort before streamstats. Expected rows are unchanged. JSONObject actual = executeQuery( String.format( - "source=%s | streamstats window=2 avg(age) as avg_age by state, country" + "source=%s | sort seq | streamstats window=2 avg(age) as avg_age by state, country" + " | streamstats window=2 avg(avg_age) as avg_state_age by country | fields" + " name, country, state, month, year, age, avg_age, avg_state_age", - TEST_INDEX_STATE_COUNTRY_WITH_NULL)); + TEST_INDEX_STATE_COUNTRY_WITH_NULL_ORDERED)); verifySchemaInOrder( actual, @@ -907,15 +963,17 @@ public void testMultipleStreamstatsWithWindow() throws IOException { // causing Calcite's RelDecorrelator to fail on duplicate correlate references. @Test - @RequiresCapability(CHAINED_STREAMSTATS_BY) + @RequiresCapability({CHAINED_STREAMSTATS_BY, STREAMSTATS_SORT_NOT_HONORED}) public void testMultipleStreamstatsWithNull1() throws IOException { + // `| sort seq` on the seq-augmented fixture makes the encounter order deterministic across + // shards; gated to routes that honor a sort before streamstats. Expected rows are unchanged. JSONObject actual = executeQuery( String.format( - "source=%s | streamstats avg(age) as avg_age by state, country | streamstats" - + " avg(avg_age) as avg_state_age by country | fields name, country, state," - + " month, year, age, avg_age, avg_state_age", - TEST_INDEX_STATE_COUNTRY_WITH_NULL)); + "source=%s | sort seq | streamstats avg(age) as avg_age by state, country |" + + " streamstats avg(avg_age) as avg_state_age by country | fields name," + + " country, state, month, year, age, avg_age, avg_state_age", + TEST_INDEX_STATE_COUNTRY_WITH_NULL_ORDERED)); verifyDataRows( actual, @@ -929,11 +987,11 @@ public void testMultipleStreamstatsWithNull1() throws IOException { JSONObject actual2 = executeQuery( String.format( - "source=%s | streamstats bucket_nullable=false avg(age) as avg_age by state," - + " country | streamstats bucket_nullable=false avg(avg_age) as avg_state_age" - + " by country | fields name, country, state, month, year, age, avg_age," - + " avg_state_age", - TEST_INDEX_STATE_COUNTRY_WITH_NULL)); + "source=%s | sort seq | streamstats bucket_nullable=false avg(age) as avg_age by" + + " state, country | streamstats bucket_nullable=false avg(avg_age) as" + + " avg_state_age by country | fields name, country, state, month, year, age," + + " avg_age, avg_state_age", + TEST_INDEX_STATE_COUNTRY_WITH_NULL_ORDERED)); verifyDataRows( actual2, @@ -946,25 +1004,29 @@ public void testMultipleStreamstatsWithNull1() throws IOException { } @Test - @RequiresCapability(DOC_MUTATION) + @RequiresCapability({DOC_MUTATION, CHAINED_STREAMSTATS_BY, STREAMSTATS_SORT_NOT_HONORED}) public void testMultipleStreamstatsWithNull2() throws IOException { + // Jay is PUT as the last document (seq 5, one past the fixture's 4 rows) so `| sort seq` keeps + // it in the trailing encounter position the original single-shard test relied on, while the + // sort makes the order deterministic across shards. Mutation stays DOC_MUTATION-gated. final int docId = 5; Request insertRequest = new Request( - "PUT", String.format("/%s/_doc/%d?refresh=true", TEST_INDEX_STATE_COUNTRY, docId)); + "PUT", + String.format("/%s/_doc/%d?refresh=true", TEST_INDEX_STATE_COUNTRY_ORDERED, docId)); insertRequest.setJsonEntity( "{\"name\": \"Jay\",\"age\": 28," + " \"country\": \"USA\",\"year\": 2023,\"month\":" - + " 4}\n"); + + " 4,\"seq\": 5}\n"); client().performRequest(insertRequest); try { JSONObject actual = executeQuery( String.format( - "source=%s | streamstats avg(age) as avg_age by state, country | streamstats" - + " avg(avg_age) as avg_state_age by country | fields name, country, state," - + " month, year, age, avg_age, avg_state_age", - TEST_INDEX_STATE_COUNTRY)); + "source=%s | sort seq | streamstats avg(age) as avg_age by state, country |" + + " streamstats avg(avg_age) as avg_state_age by country | fields name," + + " country, state, month, year, age, avg_age, avg_state_age", + TEST_INDEX_STATE_COUNTRY_ORDERED)); verifyDataRows( actual, @@ -977,11 +1039,11 @@ public void testMultipleStreamstatsWithNull2() throws IOException { JSONObject actual2 = executeQuery( String.format( - "source=%s | streamstats bucket_nullable=false avg(age) as avg_age by state," - + " country | streamstats bucket_nullable=false avg(avg_age) as avg_state_age" - + " by country | fields name, country, state, month, year, age, avg_age," - + " avg_state_age", - TEST_INDEX_STATE_COUNTRY)); + "source=%s | sort seq | streamstats bucket_nullable=false avg(age) as avg_age by" + + " state, country | streamstats bucket_nullable=false avg(avg_age) as" + + " avg_state_age by country | fields name, country, state, month, year, age," + + " avg_age, avg_state_age", + TEST_INDEX_STATE_COUNTRY_ORDERED)); verifyDataRows( actual2, @@ -993,7 +1055,8 @@ public void testMultipleStreamstatsWithNull2() throws IOException { } finally { Request deleteRequest = new Request( - "DELETE", String.format("/%s/_doc/%d?refresh=true", TEST_INDEX_STATE_COUNTRY, docId)); + "DELETE", + String.format("/%s/_doc/%d?refresh=true", TEST_INDEX_STATE_COUNTRY_ORDERED, docId)); client().performRequest(deleteRequest); } } @@ -1003,10 +1066,10 @@ public void testStreamstatsAndEventstats() throws IOException { JSONObject actual = executeQuery( String.format( - "source=%s | eventstats avg(age) as avg_age| streamstats" + "makeresults format=csv data='%s' | eventstats avg(age) as avg_age| streamstats" + " avg(age) as avg_age_stream | fields name, country, state, month, year," + " age, avg_age, avg_age_stream", - TEST_INDEX_STATE_COUNTRY)); + SC)); verifyDataRows( actual, @@ -1035,15 +1098,22 @@ public void testStreamstatsAndSort() throws IOException { } @Test + // The streamstats lives in the JOIN right-subsearch, whose avg_age depends on the subsearch + // encounter order. Sourcing the seq-augmented fixture and adding `| sort seq` before streamstats + // restores the single-shard encounter order across shards; gated to routes that honor a sort + // before streamstats. The left source stays natural-order because verifyDataRows is + // order-insensitive and the left order does not change the joined row set. Expected rows are + // unchanged. + @RequiresCapability(STREAMSTATS_SORT_NOT_HONORED) public void testLeftJoinWithStreamstats() throws IOException { JSONObject actual = executeQuery( String.format( "source=%s as l | left join left=l right=r on l.country = r.country [ source=%s |" - + " streamstats window=2 avg(age) as avg_age] | fields l.name, l.country," - + " l.state, l.month, l.year, l.age, r.name, r.country, r.state, r.month," - + " r.year, r.age, avg_age", - TEST_INDEX_STATE_COUNTRY, TEST_INDEX_STATE_COUNTRY_WITH_NULL)); + + " sort seq | streamstats window=2 avg(age) as avg_age] | fields l.name," + + " l.country, l.state, l.month, l.year, l.age, r.name, r.country, r.state," + + " r.month, r.year, r.age, avg_age", + TEST_INDEX_STATE_COUNTRY, TEST_INDEX_STATE_COUNTRY_WITH_NULL_ORDERED)); verifyDataRows( actual, @@ -1089,13 +1159,14 @@ public void testMultipleStreamstatsWithEval() throws IOException { JSONObject actual = executeQuery( String.format( - "source=%s | streamstats avg(age) as avg_age by country, state, name | eval" - + " avg_age_divide_20 = avg_age - 20 | streamstats avg(avg_age_divide_20) as" - + " avg_state_age by country, state | where avg_state_age > 0 | streamstats" - + " count(avg_state_age) as count_country_age_greater_20 by country | fields" - + " name, country, state, month, year, age, avg_age, avg_age_divide_20," - + " avg_state_age, count_country_age_greater_20", - TEST_INDEX_STATE_COUNTRY)); + "makeresults format=csv data='%s' | streamstats avg(age) as avg_age by country," + + " state, name | eval avg_age_divide_20 = avg_age - 20 | streamstats" + + " avg(avg_age_divide_20) as avg_state_age by country, state | where" + + " avg_state_age > 0 | streamstats count(avg_state_age) as" + + " count_country_age_greater_20 by country | fields name, country, state," + + " month, year, age, avg_age, avg_age_divide_20, avg_state_age," + + " count_country_age_greater_20", + SC)); verifyDataRows( actual, @@ -1180,10 +1251,11 @@ public void testStreamstatsVariance() throws IOException { JSONObject actual = executeQuery( String.format( - "source=%s | streamstats stddev_pop(age), stddev_samp(age), var_pop(age)," - + " var_samp(age) | fields name, country, state, month, year, age," - + " `stddev_pop(age)`, `stddev_samp(age)`, `var_pop(age)`, `var_samp(age)`", - TEST_INDEX_STATE_COUNTRY)); + "makeresults format=csv data='%s' | streamstats stddev_pop(age), stddev_samp(age)," + + " var_pop(age), var_samp(age) | fields name, country, state, month, year," + + " age, `stddev_pop(age)`, `stddev_samp(age)`, `var_pop(age)`," + + " `var_samp(age)`", + SC)); verifySchemaInOrder( actual, @@ -1227,14 +1299,19 @@ public void testStreamstatsVariance() throws IOException { } @Test + // See testStreamstatsWithNull: `| sort seq` on the seq-augmented fixture restores the + // single-shard encounter order across shards; gated to routes that honor a sort before + // streamstats. Expected rows are unchanged. + @RequiresCapability(STREAMSTATS_SORT_NOT_HONORED) public void testStreamstatsVarianceWithNull() throws IOException { JSONObject actual = executeQuery( String.format( - "source=%s | streamstats stddev_pop(age), stddev_samp(age), var_pop(age)," - + " var_samp(age) | fields name, country, state, month, year, age," - + " `stddev_pop(age)`, `stddev_samp(age)`, `var_pop(age)`, `var_samp(age)`", - TEST_INDEX_STATE_COUNTRY_WITH_NULL)); + "source=%s | sort seq | streamstats stddev_pop(age), stddev_samp(age)," + + " var_pop(age), var_samp(age) | fields name, country, state, month, year," + + " age, `stddev_pop(age)`, `stddev_samp(age)`, `var_pop(age)`," + + " `var_samp(age)`", + TEST_INDEX_STATE_COUNTRY_WITH_NULL_ORDERED)); verifySchemaInOrder( actual, @@ -1284,10 +1361,11 @@ public void testStreamstatsVarianceBy() throws IOException { JSONObject actual = executeQuery( String.format( - "source=%s | streamstats stddev_pop(age), stddev_samp(age), var_pop(age)," - + " var_samp(age) by country | fields name, country, state, month, year, age," - + " `stddev_pop(age)`, `stddev_samp(age)`, `var_pop(age)`, `var_samp(age)`", - TEST_INDEX_STATE_COUNTRY)); + "makeresults format=csv data='%s' | streamstats stddev_pop(age), stddev_samp(age)," + + " var_pop(age), var_samp(age) by country | fields name, country, state," + + " month, year, age, `stddev_pop(age)`, `stddev_samp(age)`, `var_pop(age)`," + + " `var_samp(age)`", + SC)); verifyDataRows( actual, @@ -1302,9 +1380,10 @@ public void testStreamstatsVarianceBySpan() throws IOException { JSONObject actual = executeQuery( String.format( - "source=%s | where country != 'USA' | streamstats stddev_samp(age) by span(age," - + " 10) | fields name, country, state, month, year, age, `stddev_samp(age)`", - TEST_INDEX_STATE_COUNTRY)); + "makeresults format=csv data='%s' | where country != 'USA' | streamstats" + + " stddev_samp(age) by span(age, 10) | fields name, country, state, month," + + " year, age, `stddev_samp(age)`", + SC)); verifyDataRows( actual, @@ -1313,14 +1392,19 @@ public void testStreamstatsVarianceBySpan() throws IOException { } @Test + // See testStreamstatsWithNull: `| sort seq` on the seq-augmented fixture restores the + // single-shard encounter order across shards; gated to routes that honor a sort before + // streamstats. Expected rows are unchanged. + @RequiresCapability(STREAMSTATS_SORT_NOT_HONORED) public void testStreamstatsVarianceWithNullBy() throws IOException { JSONObject actual = executeQuery( String.format( - "source=%s | streamstats stddev_pop(age), stddev_samp(age), var_pop(age)," - + " var_samp(age) by country | fields name, country, state, month, year, age," - + " `stddev_pop(age)`, `stddev_samp(age)`, `var_pop(age)`, `var_samp(age)`", - TEST_INDEX_STATE_COUNTRY_WITH_NULL)); + "source=%s | sort seq | streamstats stddev_pop(age), stddev_samp(age)," + + " var_pop(age), var_samp(age) by country | fields name, country, state," + + " month, year, age, `stddev_pop(age)`, `stddev_samp(age)`, `var_pop(age)`," + + " `var_samp(age)`", + TEST_INDEX_STATE_COUNTRY_WITH_NULL_ORDERED)); verifyDataRows( actual, @@ -1347,9 +1431,9 @@ public void testStreamstatsDistinctCount() throws IOException { JSONObject actual = executeQuery( String.format( - "source=%s | streamstats dc(state) as dc_state | fields name, country, state," - + " month, year, age, dc_state", - TEST_INDEX_STATE_COUNTRY)); + "makeresults format=csv data='%s' | streamstats dc(state) as dc_state | fields" + + " name, country, state, month, year, age, dc_state", + SC)); verifySchemaInOrder( actual, @@ -1374,9 +1458,9 @@ public void testStreamstatsDistinctCountByCountry() throws IOException { JSONObject actual = executeQuery( String.format( - "source=%s | streamstats dc(state) as dc_state by country | fields name, country," - + " state, month, year, age, dc_state", - TEST_INDEX_STATE_COUNTRY)); + "makeresults format=csv data='%s' | streamstats dc(state) as dc_state by country |" + + " fields name, country, state, month, year, age, dc_state", + SC)); verifySchemaInOrder( actual, @@ -1401,9 +1485,9 @@ public void testStreamstatsDistinctCountFunction() throws IOException { JSONObject actual = executeQuery( String.format( - "source=%s | streamstats distinct_count(country) as dc_country | fields name," - + " country, state, month, year, age, dc_country", - TEST_INDEX_STATE_COUNTRY)); + "makeresults format=csv data='%s' | streamstats distinct_count(country) as" + + " dc_country | fields name, country, state, month, year, age, dc_country", + SC)); verifySchemaInOrder( actual, @@ -1424,13 +1508,17 @@ public void testStreamstatsDistinctCountFunction() throws IOException { } @Test + // See testStreamstatsWithNull: `| sort seq` on the seq-augmented fixture restores the + // single-shard encounter order across shards; gated to routes that honor a sort before + // streamstats. Expected rows are unchanged. + @RequiresCapability(STREAMSTATS_SORT_NOT_HONORED) public void testStreamstatsDistinctCountWithNull() throws IOException { JSONObject actual = executeQuery( String.format( - "source=%s | streamstats dc(state) as dc_state | fields name, country, state," - + " month, year, age, dc_state", - TEST_INDEX_STATE_COUNTRY_WITH_NULL)); + "source=%s | sort seq | streamstats dc(state) as dc_state | fields name, country," + + " state, month, year, age, dc_state", + TEST_INDEX_STATE_COUNTRY_WITH_NULL_ORDERED)); verifySchemaInOrder( actual, @@ -1457,10 +1545,11 @@ public void testStreamstatsEarliestAndLatest() throws IOException { JSONObject actual = executeQuery( String.format( - "source=%s | streamstats earliest(message), latest(message) by server | fields" - + " created_at, server, `@timestamp`, message, level, `earliest(message)`," - + " `latest(message)`", - TEST_INDEX_LOGS)); + "makeresults format=csv data='%s' | eval created_at=cast(created_at as timestamp)," + + " `@timestamp`=cast(`@timestamp` as timestamp) | streamstats" + + " earliest(message), latest(message) by server | fields created_at, server," + + " `@timestamp`, message, level, `earliest(message)`, `latest(message)`", + LOGS_STREAM)); verifySchema( actual, schema("created_at", "timestamp"), @@ -1513,4 +1602,58 @@ public void testStreamstatsEarliestAndLatest() throws IOException { "Service started", "Backup completed")); } + + @Test + public void testStreamstatsEarliestLatestIndexMultiShardCoverage() throws IOException { + JSONObject actual = + executeQuery( + String.format( + "source=%s | streamstats earliest(message), latest(message) by server | fields" + + " server, message, `earliest(message)`, `latest(message)`", + TEST_INDEX_LOGS)); + + Map> messagesByServer = + Map.of( + "server1", Set.of("Database connection failed", "High memory usage"), + "server2", Set.of("Service started", "Backup completed"), + "server3", Set.of("Disk space low")); + JSONArray rows = actual.getJSONArray("datarows"); + assertEquals(5, rows.length()); + for (int i = 0; i < rows.length(); i++) { + JSONArray row = rows.getJSONArray(i); + String server = row.getString(0); + Set validMessages = messagesByServer.get(server); + assertNotNull("unexpected server: " + server, validMessages); + assertTrue(validMessages.contains(row.getString(1))); + assertTrue(validMessages.contains(row.getString(2))); + assertTrue(validMessages.contains(row.getString(3))); + } + } + + @Test + public void testStreamstatsIndexMultiShardCoverage() throws IOException { + // Retains real multi-document, multi-shard index coverage for streamstats. streamstats' running + // count() visits every document exactly once, so regardless of shard encounter order the cnt + // column is always the multiset {1, 2, 3, 4} and the age column is the full fixture multiset. + // These are order-independent properties, so they are stable across shard layouts. + JSONObject result = + executeQuery( + String.format( + "source=%s | streamstats count() as cnt | fields age, cnt", + TEST_INDEX_STATE_COUNTRY)); + verifySchema(result, schema("age", "int"), schema("cnt", "bigint")); + + JSONArray datarows = result.getJSONArray("datarows"); + assertEquals(4, datarows.length()); + List ages = new ArrayList<>(); + List cnts = new ArrayList<>(); + for (int i = 0; i < datarows.length(); i++) { + ages.add(datarows.getJSONArray(i).getInt(0)); + cnts.add(datarows.getJSONArray(i).getInt(1)); + } + Collections.sort(ages); + Collections.sort(cnts); + assertEquals(List.of(20, 25, 30, 70), ages); + assertEquals(List.of(1, 2, 3, 4), cnts); + } } diff --git a/integ-test/src/test/java/org/opensearch/sql/legacy/SQLIntegTestCase.java b/integ-test/src/test/java/org/opensearch/sql/legacy/SQLIntegTestCase.java index fa75aa402dc..cc7b073bed6 100644 --- a/integ-test/src/test/java/org/opensearch/sql/legacy/SQLIntegTestCase.java +++ b/integ-test/src/test/java/org/opensearch/sql/legacy/SQLIntegTestCase.java @@ -821,6 +821,26 @@ public enum Index { "state_country_with_null", getStateCountryIndexMapping(), // with null index use the same schema "src/test/resources/state_country_with_null.json"), + STATE_COUNTRY_ORDERED( + TestsConstants.TEST_INDEX_STATE_COUNTRY_ORDERED, + "state_country_ordered", + getMappingFile("state_country_ordered_index_mapping.json"), + "src/test/resources/state_country_ordered.json"), + STATE_COUNTRY_WITH_NULL_ORDERED( + TestsConstants.TEST_INDEX_STATE_COUNTRY_WITH_NULL_ORDERED, + "state_country_with_null_ordered", + getMappingFile("state_country_ordered_index_mapping.json"), // same schema, plus seq + "src/test/resources/state_country_with_null_ordered.json"), + STATE_COUNTRY_SINGLE_SHARD( + TestsConstants.TEST_INDEX_STATE_COUNTRY_SINGLE_SHARD, + "state_country_single_shard", + getMappingFile("state_country_single_shard_index_mapping.json"), + "src/test/resources/state_country.json"), + STATE_COUNTRY_WITH_NULL_SINGLE_SHARD( + TestsConstants.TEST_INDEX_STATE_COUNTRY_WITH_NULL_SINGLE_SHARD, + "state_country_with_null_single_shard", + getMappingFile("state_country_single_shard_index_mapping.json"), // 1 shard, no seq needed + "src/test/resources/state_country_with_null.json"), OCCUPATION( TestsConstants.TEST_INDEX_OCCUPATION, "occupation", @@ -877,6 +897,11 @@ public enum Index { "duplication_nullable", getDuplicationNullableIndexMapping(), "src/test/resources/duplication_nullable.json"), + DUPLICATION_NULLABLE_ORDERED( + TestsConstants.TEST_INDEX_DUPLICATION_NULLABLE_ORDERED, + "duplication_nullable_ordered", + getMappingFile("duplication_nullable_ordered_index_mapping.json"), + "src/test/resources/duplication_nullable_ordered.json"), // Graph lookup test indices (inspired by MongoDB $graphLookup examples) GRAPH_EMPLOYEES( TestsConstants.TEST_INDEX_GRAPH_EMPLOYEES, diff --git a/integ-test/src/test/java/org/opensearch/sql/legacy/TestsConstants.java b/integ-test/src/test/java/org/opensearch/sql/legacy/TestsConstants.java index 5d7eeb328af..4dffcfa8f07 100644 --- a/integ-test/src/test/java/org/opensearch/sql/legacy/TestsConstants.java +++ b/integ-test/src/test/java/org/opensearch/sql/legacy/TestsConstants.java @@ -81,12 +81,35 @@ public class TestsConstants { public static final String TEST_INDEX_STATE_COUNTRY = TEST_INDEX + "_state_country"; public static final String TEST_INDEX_STATE_COUNTRY_WITH_NULL = TEST_INDEX + "_state_country_with_null"; + // Deterministic-order variants of the state_country fixtures for multi-shard streamstats tests: + // same documents plus an explicit `seq` field so a `sort seq` restores the single-shard encounter + // order on any shard layout. The single_shard variant pins number_of_shards=1 for the + // reverse-after-streamstats no-op tests, where injecting a sort would change the behavior under + // test. + public static final String TEST_INDEX_STATE_COUNTRY_ORDERED = + TEST_INDEX + "_state_country_ordered"; + public static final String TEST_INDEX_STATE_COUNTRY_WITH_NULL_ORDERED = + TEST_INDEX + "_state_country_with_null_ordered"; + public static final String TEST_INDEX_STATE_COUNTRY_SINGLE_SHARD = + TEST_INDEX + "_state_country_single_shard"; + // Single-shard (number_of_shards=1) variant of state_country_with_null. Used by the reset + // streamstats null test: `reset_before`/`reset_after` builds a self-correlated plan that the + // physical compiler cannot combine with an upstream `sort` (planner IndexOutOfBounds), so the + // seq-sort determinism trick used by the other streamstats tests is unavailable here. Pinning a + // single shard yields the deterministic insertion-order stream without injecting a sort. + public static final String TEST_INDEX_STATE_COUNTRY_WITH_NULL_SINGLE_SHARD = + TEST_INDEX + "_state_country_with_null_single_shard"; public static final String TEST_INDEX_OCCUPATION = TEST_INDEX + "_occupation"; public static final String TEST_INDEX_OCCUPATION_TOP_RARE = TEST_INDEX + "_occupation_top_rare"; public static final String TEST_INDEX_HOBBIES = TEST_INDEX + "_hobbies"; public static final String TEST_INDEX_WORKER = TEST_INDEX + "_worker"; public static final String TEST_INDEX_WORK_INFORMATION = TEST_INDEX + "_work_information"; public static final String TEST_INDEX_DUPLICATION_NULLABLE = TEST_INDEX + "_duplication_nullable"; + // Deterministic-order variant of duplication_nullable: same documents plus an explicit `seq` + // field so a `sort seq` restores the single-shard encounter order for CONSECUTIVE dedup on any + // shard layout. + public static final String TEST_INDEX_DUPLICATION_NULLABLE_ORDERED = + TEST_INDEX + "_duplication_nullable_ordered"; public static final String TEST_INDEX_GRAPH_EMPLOYEES = TEST_INDEX + "_graph_employees"; public static final String TEST_INDEX_GRAPH_TRAVELERS = TEST_INDEX + "_graph_travelers"; public static final String TEST_INDEX_GRAPH_AIRPORTS = TEST_INDEX + "_graph_airports"; diff --git a/integ-test/src/test/resources/duplication_nullable_ordered.json b/integ-test/src/test/resources/duplication_nullable_ordered.json new file mode 100644 index 00000000000..39177ece7e6 --- /dev/null +++ b/integ-test/src/test/resources/duplication_nullable_ordered.json @@ -0,0 +1,36 @@ +{"index":{"_id":"1"}} +{"id":1,"name":"A","category":"X","seq":1} +{"index":{"_id":"2"}} +{"id":1,"name":"A","category":"Y","seq":2} +{"index":{"_id":"3"}} +{"id":1,"name":"A","category":"Y","seq":3} +{"index":{"_id":"4"}} +{"id":1,"name":"B","category":"Z","seq":4} +{"index":{"_id":"5"}} +{"id":1,"name":"B","category":"Z","seq":5} +{"index":{"_id":"6"}} +{"id":1,"name":"B","category":"Z","seq":6} +{"index":{"_id":"7"}} +{"id":1,"name":"C","category":"X","seq":7} +{"index":{"_id":"8"}} +{"id":1,"name":null,"category":"Y","seq":8} +{"index":{"_id":"9"}} +{"id":1,"name":"D","category":"Z","seq":9} +{"index":{"_id":"10"}} +{"id":1,"name":"E","category":null,"seq":10} +{"index":{"_id":"11"}} +{"id":1,"name":"A","category":"X","seq":11} +{"index":{"_id":"12"}} +{"id":1,"name":"A","category":"Y","seq":12} +{"index":{"_id":"13"}} +{"id":1,"name":null,"category":"X","seq":13} +{"index":{"_id":"14"}} +{"id":1,"name":"B","category":null,"seq":14} +{"index":{"_id":"15"}} +{"id":1,"name":"B","category":"Y","seq":15} +{"index":{"_id":"16"}} +{"id":1,"name":null,"category":"Z","seq":16} +{"index":{"_id":"17"}} +{"id":1,"name":"C","category":"X","seq":17} +{"index":{"_id":"18"}} +{"id":1,"name":null,"category":null,"seq":18} diff --git a/integ-test/src/test/resources/indexDefinitions/duplication_nullable_ordered_index_mapping.json b/integ-test/src/test/resources/indexDefinitions/duplication_nullable_ordered_index_mapping.json new file mode 100644 index 00000000000..8592e35f027 --- /dev/null +++ b/integ-test/src/test/resources/indexDefinitions/duplication_nullable_ordered_index_mapping.json @@ -0,0 +1,18 @@ +{ + "mappings": { + "properties": { + "id": { + "type": "integer" + }, + "name": { + "type": "keyword" + }, + "category": { + "type": "keyword" + }, + "seq": { + "type": "integer" + } + } + } +} diff --git a/integ-test/src/test/resources/indexDefinitions/state_country_ordered_index_mapping.json b/integ-test/src/test/resources/indexDefinitions/state_country_ordered_index_mapping.json new file mode 100644 index 00000000000..9c06c7ce315 --- /dev/null +++ b/integ-test/src/test/resources/indexDefinitions/state_country_ordered_index_mapping.json @@ -0,0 +1,33 @@ +{ + "mappings": { + "properties": { + "name": { + "type": "keyword" + }, + "age": { + "type": "integer" + }, + "state": { + "type": "text", + "fields": { + "keyword": { + "type": "keyword", + "ignore_above": 256 + } + } + }, + "country": { + "type": "text" + }, + "year": { + "type": "integer" + }, + "month": { + "type": "integer" + }, + "seq": { + "type": "integer" + } + } + } +} diff --git a/integ-test/src/test/resources/indexDefinitions/state_country_single_shard_index_mapping.json b/integ-test/src/test/resources/indexDefinitions/state_country_single_shard_index_mapping.json new file mode 100644 index 00000000000..64341497fa8 --- /dev/null +++ b/integ-test/src/test/resources/indexDefinitions/state_country_single_shard_index_mapping.json @@ -0,0 +1,35 @@ +{ + "settings": { + "index": { + "number_of_shards": 1 + } + }, + "mappings": { + "properties": { + "name": { + "type": "keyword" + }, + "age": { + "type": "integer" + }, + "state": { + "type": "text", + "fields": { + "keyword": { + "type": "keyword", + "ignore_above": 256 + } + } + }, + "country": { + "type": "text" + }, + "year": { + "type": "integer" + }, + "month": { + "type": "integer" + } + } + } +} diff --git a/integ-test/src/test/resources/state_country_ordered.json b/integ-test/src/test/resources/state_country_ordered.json new file mode 100644 index 00000000000..9bcbc4c9645 --- /dev/null +++ b/integ-test/src/test/resources/state_country_ordered.json @@ -0,0 +1,8 @@ +{"index":{"_id":"1"}} +{"name":"Jake","age":70,"state":"California","country":"USA","year":2023,"month":4,"seq":1} +{"index":{"_id":"2"}} +{"name":"Hello","age":30,"state":"New York","country":"USA","year":2023,"month":4,"seq":2} +{"index":{"_id":"3"}} +{"name":"John","age":25,"state":"Ontario","country":"Canada","year":2023,"month":4,"seq":3} +{"index":{"_id":"4"}} +{"name":"Jane","age":20,"state":"Quebec","country":"Canada","year":2023,"month":4,"seq":4} diff --git a/integ-test/src/test/resources/state_country_with_null_ordered.json b/integ-test/src/test/resources/state_country_with_null_ordered.json new file mode 100644 index 00000000000..f2de2a00d80 --- /dev/null +++ b/integ-test/src/test/resources/state_country_with_null_ordered.json @@ -0,0 +1,12 @@ +{"index":{"_id":"1"}} +{"name":"Jake","age":70,"state":"California","country":"USA","year":2023,"month":4,"seq":1} +{"index":{"_id":"2"}} +{"name":"Hello","age":30,"state":"New York","country":"USA","year":2023,"month":4,"seq":2} +{"index":{"_id":"3"}} +{"name":"John","age":25,"state":"Ontario","country":"Canada","year":2023,"month":4,"seq":3} +{"index":{"_id":"4"}} +{"name":"Jane","age":20,"state":"Quebec","country":"Canada","year":2023,"month":4,"seq":4} +{"index":{"_id":"5"}} +{"name":null,"age":10,"state":null,"country":"Canada","year":2023,"month":4,"seq":5} +{"index":{"_id":"6"}} +{"name":"Kevin","year":2023,"month":4,"seq":6}