diff --git a/test/external-modules/esql-heap-attack/build.gradle b/test/external-modules/esql-heap-attack/build.gradle index 52837133d55f4..20d56c939e11c 100644 --- a/test/external-modules/esql-heap-attack/build.gradle +++ b/test/external-modules/esql-heap-attack/build.gradle @@ -16,6 +16,11 @@ esplugin { classname ='org.elasticsearch.test.esql.heap_attack.HeapAttackPlugin' } +dependencies { + javaRestTestImplementation project(':x-pack:plugin:esql:compute') + javaRestTestImplementation project(':libs:swisshash') +} + tasks.named('javaRestTest') { usesDefaultDistribution("to be triaged") it.onlyIf("snapshot build") { buildParams.snapshotBuild } diff --git a/test/external-modules/esql-heap-attack/src/javaRestTest/java/org/elasticsearch/xpack/esql/heap_attack/HeapAttackIT.java b/test/external-modules/esql-heap-attack/src/javaRestTest/java/org/elasticsearch/xpack/esql/heap_attack/HeapAttackIT.java index 665eb9eda83dd..43ab1b7619af6 100644 --- a/test/external-modules/esql-heap-attack/src/javaRestTest/java/org/elasticsearch/xpack/esql/heap_attack/HeapAttackIT.java +++ b/test/external-modules/esql-heap-attack/src/javaRestTest/java/org/elasticsearch/xpack/esql/heap_attack/HeapAttackIT.java @@ -271,15 +271,6 @@ private Response groupOnManyLongs(int count) throws IOException { return query(query.toString(), null); } - private StringBuilder makeManyLongs(int count) { - StringBuilder query = startQuery(); - query.append("FROM manylongs\\n| EVAL i0 = a + b, i1 = b + i0"); - for (int i = 2; i < count; i++) { - query.append(", i").append(i).append(" = i").append(i - 2).append(" + ").append(i - 1); - } - return query.append("\\n"); - } - public void testSmallConcat() throws IOException { initSingleDocIndex(); Response resp = concat(2); @@ -687,46 +678,6 @@ private Map queryDuplicatedHistograms(String index, String colum return responseAsMap(query(queryStr, null)); } - private void initManyLongs(int countPerLong) throws IOException { - logger.info("loading many documents with longs"); - StringBuilder bulk = new StringBuilder(); - int flush = 0; - for (int a = 0; a < countPerLong; a++) { - for (int b = 0; b < countPerLong; b++) { - for (int c = 0; c < countPerLong; c++) { - for (int d = 0; d < countPerLong; d++) { - for (int e = 0; e < countPerLong; e++) { - bulk.append(String.format(Locale.ROOT, """ - {"create":{}} - {"a":%d,"b":%d,"c":%d,"d":%d,"e":%d} - """, a, b, c, d, e)); - flush++; - if (flush % 10_000 == 0) { - bulk("manylongs", bulk.toString()); - bulk.setLength(0); - logger.info( - "flushing {}/{} to manylongs", - flush, - countPerLong * countPerLong * countPerLong * countPerLong * countPerLong - ); - - } - } - } - } - } - } - initIndex("manylongs", bulk.toString()); - } - - private void initSingleDocIndex() throws IOException { - logger.info("loading a single document"); - initIndex("single", """ - {"create":{}} - {"a":1} - """); - } - void initManyBigFieldsIndex(int docs, String type, boolean random, int fields) throws IOException { logger.info("loading many documents with many big fields"); int docsPerBulk = 5; diff --git a/test/external-modules/esql-heap-attack/src/javaRestTest/java/org/elasticsearch/xpack/esql/heap_attack/HeapAttackLimitByIT.java b/test/external-modules/esql-heap-attack/src/javaRestTest/java/org/elasticsearch/xpack/esql/heap_attack/HeapAttackLimitByIT.java new file mode 100644 index 0000000000000..c3d2a190656a9 --- /dev/null +++ b/test/external-modules/esql-heap-attack/src/javaRestTest/java/org/elasticsearch/xpack/esql/heap_attack/HeapAttackLimitByIT.java @@ -0,0 +1,396 @@ +/* + * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one + * or more contributor license agreements. Licensed under the Elastic License + * 2.0; you may not use this file except in compliance with the Elastic License + * 2.0. + */ + +package org.elasticsearch.xpack.esql.heap_attack; + +import com.carrotsearch.randomizedtesting.annotations.TimeoutSuite; + +import org.apache.lucene.tests.util.TimeUnits; +import org.elasticsearch.Build; +import org.elasticsearch.common.util.BytesRefHash; +import org.elasticsearch.compute.aggregation.blockhash.HashImplFactory; +import org.elasticsearch.compute.operator.GroupKeyEncoder; +import org.elasticsearch.compute.operator.GroupedLimitOperator; +import org.elasticsearch.compute.operator.topn.GroupedTopNOperator; +import org.elasticsearch.swisshash.BytesRefSwissHash; +import org.elasticsearch.test.ListMatcher; +import org.junit.After; +import org.junit.Before; + +import java.io.IOException; +import java.util.List; +import java.util.Map; + +import static org.elasticsearch.test.ListMatcher.matchesList; +import static org.elasticsearch.test.MapMatcher.matchesMap; +import static org.hamcrest.Matchers.any; + +/** + * Heap-attack tests for {@code LIMIT BY} ({@code GroupedLimitOperator}) + * and {@code SORT | LIMIT BY} ({@code GroupedTopNOperator}). + * Each test exercises a distinct circuit-breaking path inside its respective operator. + */ +@TimeoutSuite(millis = 5 * TimeUnits.MINUTE) +public class HeapAttackLimitByIT extends HeapAttackTestCase { + + /** + * Lower the request circuit breaker to 40% of JVM heap (default is 60%). A tighter limit ensures + * the breaker fires well before the JVM runs out of memory. + * + * We want to have some headroom between what the circuit breaker sees and the real free JVM heap: + * + */ + @Before + public void lowerRequestBreakerLimit() throws IOException { + setRequestBreakerLimit("40%"); + } + + /** + * Restores circuit breaker default limit + */ + @After + public void resetRequestBreakerLimit() throws IOException { + setRequestBreakerLimit(null); + } + // ------------------------------------------------------------------------- + // LIMIT BY — GroupedLimitOperator + // ------------------------------------------------------------------------- + + /** + * Grouping by a,b,c,d,e plus a 100-char string should succeed. 100K unique groups with + * ~140-byte keys ≈ 10 MB in the hash table, well within the circuit breaker. + */ + public void testLimitByHighCardinality() throws IOException { + assumeTrue("LIMIT BY requires snapshot builds", Build.current().isSnapshot()); + initManyLongsAndString(10, 100); + Map result = limitByWideStringKey(1); + ListMatcher columns = matchesList().item(matchesMap().entry("name", "MAX(a)").entry("type", "long")); + ListMatcher values = matchesList().item(List.of(9)); + assertResultMap(result, columns, values); + } + + /** + * The GroupedLimitOperator stores all group keys in a hash table. With 100K unique groups + * (a,b,c,d,e ∈ [0,9]) and a wide string column in the key, the hash table trips the circuit + * breaker. {@code REPEAT(f, scale)} widens the string per attempt without needing many EVAL + * columns: at scale=10 each key is ~4KB, so 100K groups ≈ 400 MB. + */ + public void testLimitByHighCardinalityTooMuchMemory() throws IOException { + assumeTrue("LIMIT BY requires snapshot builds", Build.current().isSnapshot()); + initManyLongsAndString(10, 100); + assertCircuitBreaksVia( + attempt -> limitByWideStringKey(attempt * 40), + GroupedLimitOperator.class.getName(), + bytesRefHashClassName() + ); + } + + private Map limitByWideStringKey(int repeatScale) throws IOException { + logger.info("limit by a,b,c,d,e + string repeated {}x", repeatScale); + StringBuilder query = startQuery(); + query.append("FROM manylongsandstring\\n"); + if (repeatScale > 1) { + query.append("| EVAL g = REPEAT(f, ").append(repeatScale).append(")\\n"); + query.append("| LIMIT 1 BY a, b, c, d, e, g\\n"); + } else { + query.append("| LIMIT 1 BY a, b, c, d, e, f\\n"); + } + query.append("| STATS MAX(a)\"}"); + return responseAsMap(query(query.toString(), null)); + } + + /** + * Grouping by 30 repeated 1 MB strings should succeed. With the breaker at 40 % of 512 MB + * (≈ 205 MB) and N = 30: + *
    + *
  • REPEAT scratch buffers: ~34 MB (30 × 1.125 MB, oversized by 12.5 %)
    • + *
  • keyword blocks: ~30 MB
    • + *
  • keyEncoder scratch at peak grow: old ≈ 29 MB + new ≈ 34 MB
    • + *
+ * The peak tracked total (~127 MB) stays well within the 205 MB limit. + */ + public void testLimitByKeyEncoderDoesNotCircuitBreak() throws IOException { + assumeTrue("LIMIT BY requires snapshot builds", Build.current().isSnapshot()); + initSingleDocIndex(); + Map result = limitByManyStrings(30); + ListMatcher columns = matchesList().item(matchesMap().entry("name", "a").entry("type", "long")); + assertResultMap(result, columns, matchesList().item(List.of(1))); + } + + /** + * The {@code GroupKeyEncoder} encodes group key columns into a {@code BreakingBytesRefBuilder} + * scratch buffer (labeled {@code "group-key-encoder"}) which is charged to the circuit breaker + * and never released between rows. By grouping by many wide strings, the scratch accumulates + * past the circuit breaker limit independently of how many rows or unique groups there are. + *

+ * Memory model with N columns of 1 MB each (one document): + *

    + *
  • N REPEAT scratch buffers (labeled {@code "repeat"}): ~1.125N MB (oversized by 12.5 % + * due to {@link org.apache.lucene.util.ArrayUtil#oversize})
    • + *
  • N keyword blocks: N MB
    • + *
  • keyEncoder scratch (accumulated during encoding): up to ~1.125N MB (also oversized)
    • + *
+ * EVAL succeeds while 2.125N < limit (≈ 205 MB at 40 %), i.e. N ≤ 96. + * Then the keyEncoder scratch pushes the total over the limit on the K-th column encoded. + * During each grow the breaker sees both old and new array capacities (~2.25K), so the trip + * condition is 2.125N + 2.25K > limit. We use N = k = 80 which leaves ample JVM headroom: + * at the trip point (~341 MB live heap) there are ~171 MB free in the 512 MB heap. + */ + public void testLimitByKeyEncoderTooMuchMemory() throws IOException { + assumeTrue("LIMIT BY requires snapshot builds", Build.current().isSnapshot()); + initSingleDocIndex(); + assertCircuitBreaksVia( + attempt -> limitByManyStrings(attempt * 80), + GroupedLimitOperator.class.getName(), + GroupKeyEncoder.class.getName() + ); + } + + private Map limitByManyStrings(int count) throws IOException { + logger.info("limit by {} repeated 1MB string cols", count); + // Each REPEAT(TO_STRING(a), 1_000_000) produces a 1MB string at execution time (a=1, so + // TO_STRING(a)="1"). Using TO_STRING(a) rather than a literal prevents the optimizer from + // constant-folding the REPEAT at plan time. The resulting 1MB string per column is kept in + // a thread-local scratch buffer (labeled "repeat") for the duration of the query. Using them + // as group keys forces the GroupKeyEncoder to copy each string into its own scratch buffer, + // accumulating the charge column by column. + StringBuilder query = startQuery(); + // REPEAT(TO_STRING(a), ...) instead of REPEAT("x", ...) prevents constant folding at plan + // time: the string argument depends on a field so the optimizer cannot precompute the result. + // At execution time TO_STRING(1) = "1" (1 byte), giving the same 1MB string per column. + query.append("FROM single\\n| EVAL s0 = REPEAT(TO_STRING(a), 1000000)"); + for (int i = 1; i < count; i++) { + query.append(", s").append(i).append(" = REPEAT(TO_STRING(a), 1000000)"); + } + query.append("\\n| LIMIT 1 BY s0"); + for (int i = 1; i < count; i++) { + query.append(", s").append(i); + } + query.append("\\n| KEEP a\"}"); + return responseAsMap(query(query.toString(), null)); + } + + /** + * Unlike {@code SORT | LIMIT BY}, a near-{@code Integer.MAX_VALUE} per-group limit in + * {@code LIMIT BY} does not cause excessive memory usage. {@code GroupedLimitOperator} is + * a streaming filter that only stores one counter per unique group — the limit value itself + * has no effect on memory. + */ + public void testLimitByLargePerGroupCount() throws IOException { + assumeTrue("LIMIT BY requires snapshot builds", Build.current().isSnapshot()); + initManyLongs(1); + StringBuilder query = startQuery(); + query.append("FROM manylongs | LIMIT 2147483630 BY a | STATS MAX(a)\"}"); + Map result = responseAsMap(query(query.toString(), null)); + ListMatcher columns = matchesList().item(matchesMap().entry("name", "MAX(a)").entry("type", "long")); + assertResultMap(result, columns, any(List.class)); + } + + // ------------------------------------------------------------------------- + // SORT | LIMIT BY — GroupedTopNOperator + // ------------------------------------------------------------------------- + + /** + * Sorting by 30 repeated 1 MB string columns should succeed. Each {@code TopNRow} stores + * ~30 MB of encoded sort keys, which stays within the circuit breaker for a single row. + */ + public void testTopNByManySortColumns() throws IOException { + assumeTrue("SORT | LIMIT BY requires snapshot builds", Build.current().isSnapshot()); + initSingleDocIndex(); + Map result = topNByManyStringSortCols(30); + ListMatcher columns = matchesList().item(matchesMap().entry("name", "a").entry("type", "long")); + assertResultMap(result, columns, matchesList().item(List.of(1))); + } + + /** + * The GroupedTopNOperator stores full sort keys for every row in its per-group queues. + * Sorting by many 1 MB string columns means each stored {@code TopNRow}'s encoded keys + * buffer is very wide (encoding gets done by {@code TopNOperator.RowFiller}), tripping the + * circuit breaker even with a single row. + */ + public void testTopNByManySortColumnsTooMuchMemory() throws IOException { + assumeTrue("SORT | LIMIT BY requires snapshot builds", Build.current().isSnapshot()); + initSingleDocIndex(); + assertCircuitBreaksVia( + attempt -> topNByManyStringSortCols(attempt * 80), + GroupedTopNOperator.class.getName(), + "TopNOperator$RowFiller" + ); + } + + private Map topNByManyStringSortCols(int count) throws IOException { + logger.info("topn by with {} string sort cols", count); + StringBuilder query = startQuery(); + query.append("FROM single\\n| EVAL s0 = REPEAT(TO_STRING(a), 1000000)"); + for (int i = 1; i < count; i++) { + query.append(", s").append(i).append(" = REPEAT(TO_STRING(a), 1000000)"); + } + // Sort by all string columns so they are stored as sort keys in each TopNRow. + query.append("\\n| SORT s0"); + for (int i = 1; i < count; i++) { + query.append(", s").append(i); + } + query.append("\\n| LIMIT 1 BY a\\n| KEEP a\"}"); + return responseAsMap(query(query.toString(), null)); + } + + /** + * Same as {@link #testLimitByHighCardinality} but for {@code SORT | LIMIT BY}. + * Sorting by a single column keeps {@code TopNRow} narrow; the hash table stores the wide + * group keys. 100K groups with ~140-byte keys ≈ 10 MB, well within the circuit breaker. + */ + public void testTopNByHighCardinality() throws IOException { + assumeTrue("SORT | LIMIT BY requires snapshot builds", Build.current().isSnapshot()); + initManyLongsAndString(10, 100); + Map result = topNByWideStringKey(1); + ListMatcher columns = matchesList().item(matchesMap().entry("name", "MAX(a)").entry("type", "long")); + ListMatcher values = matchesList().item(List.of(9)); + assertResultMap(result, columns, values); + } + + /** + * Same as {@link #testLimitByHighCardinalityTooMuchMemory} but for {@code SORT | LIMIT BY}. + * The {@code GroupedTopNOperator} stores group keys in a hash table ({@code keysHash}). + * {@code REPEAT(f, scale)} widens the string per attempt: at scale=40 each key is ~4 KB, + * so 100K groups ≈ 400 MB, tripping the circuit breaker. + */ + public void testTopNByHighCardinalityTooMuchMemory() throws IOException { + assumeTrue("SORT | LIMIT BY requires snapshot builds", Build.current().isSnapshot()); + initManyLongsAndString(10, 100); + assertCircuitBreaksVia(attempt -> topNByWideStringKey(attempt * 40), GroupedTopNOperator.class.getName(), bytesRefHashClassName()); + } + + private Map topNByWideStringKey(int repeatScale) throws IOException { + logger.info("topn by a,b,c,d,e + string repeated {}x", repeatScale); + StringBuilder query = startQuery(); + query.append("FROM manylongsandstring\\n"); + if (repeatScale > 1) { + query.append("| EVAL g = REPEAT(f, ").append(repeatScale).append(")\\n"); + query.append("| SORT a\\n| LIMIT 1 BY a, b, c, d, e, g\\n"); + } else { + query.append("| SORT a\\n| LIMIT 1 BY a, b, c, d, e, f\\n"); + } + query.append("| STATS MAX(a)\"}"); + return responseAsMap(query(query.toString(), null)); + } + + /** + * Grouping by 30 repeated 1 MB strings with a preceding sort should succeed. + * Same memory model as {@link #testLimitByKeyEncoderDoesNotCircuitBreak} (~127 MB peak) + * but with {@code GroupedTopNOperator} rather than {@code GroupedLimitOperator}. + */ + public void testTopNByKeyEncoderDoesNotCircuitBreak() throws IOException { + assumeTrue("SORT | LIMIT BY requires snapshot builds", Build.current().isSnapshot()); + initSingleDocIndex(); + Map result = topNByManyStrings(30); + ListMatcher columns = matchesList().item(matchesMap().entry("name", "a").entry("type", "long")); + assertResultMap(result, columns, matchesList().item(List.of(1))); + } + + /** + * Same circuit-breaking path as {@link #testLimitByKeyEncoderTooMuchMemory} but triggered + * inside {@code GroupedTopNOperator}: grouping by many wide strings causes the + * {@code GroupKeyEncoder} scratch buffer to accumulate past the circuit-breaker limit. + */ + public void testTopNByKeyEncoderTooMuchMemory() throws IOException { + assumeTrue("SORT | LIMIT BY requires snapshot builds", Build.current().isSnapshot()); + initSingleDocIndex(); + assertCircuitBreaksVia( + attempt -> topNByManyStrings(attempt * 80), + GroupedTopNOperator.class.getName(), + GroupKeyEncoder.class.getName() + ); + } + + private Map topNByManyStrings(int count) throws IOException { + logger.info("topn by with {} repeated 1MB string group key cols", count); + StringBuilder query = startQuery(); + query.append("FROM single\\n| EVAL s0 = REPEAT(TO_STRING(a), 1000000)"); + for (int i = 1; i < count; i++) { + query.append(", s").append(i).append(" = REPEAT(TO_STRING(a), 1000000)"); + } + query.append("\\n| SORT a\\n"); + query.append("| LIMIT 1 BY s0"); + for (int i = 1; i < count; i++) { + query.append(", s").append(i); + } + query.append("\\n| KEEP a\"}"); + return responseAsMap(query(query.toString(), null)); + } + + /** + * This limits 10 rows per group across 100K unique groups, which should succeed. + */ + public void testTopNByManyGroupsSmallTopCount() throws IOException { + assumeTrue("SORT | LIMIT BY requires snapshot builds", Build.current().isSnapshot()); + initManyLongs(10); + Map result = topNByManyGroups(10); + ListMatcher columns = matchesList().item(matchesMap().entry("name", "a").entry("type", "long")) + .item(matchesMap().entry("name", "b").entry("type", "long")); + assertResultMap(result, columns, any(List.class)); + } + + /** + * The GroupedTopNOperator creates a {@code TopNQueue} + * for every group, and each queue pre-allocates a heap array of {@code topCount} slots. + * With 100K unique groups (all combinations of a,b,c,d,e ∈ [0,9]) and a large per-group + * limit, the combined queue allocations trip the circuit breaker even when the queues are + * never filled. + */ + public void testTopNByManyGroupsLargeTopCountTooMuchMemory() throws IOException { + assumeTrue("SORT | LIMIT BY requires snapshot builds", Build.current().isSnapshot()); + initManyLongs(10); + assertCircuitBreaksVia( + attempt -> topNByManyGroups(attempt * 400), + GroupedTopNOperator.class.getName(), + "org.elasticsearch.compute.operator.topn.TopNQueue" + ); + } + + private Map topNByManyGroups(int topCount) throws IOException { + logger.info("topn by {} rows per group across 100K unique groups", topCount); + // No EVAL needed: (a,b,c,d,e) ∈ [0,9]^5 gives 10^5 = 100K unique groups. + // Sort by `a` to keep rows narrow; each group gets a TopNQueue pre-allocating + // a heap array of topCount slots charged to the circuit breaker. + StringBuilder query = startQuery(); + query.append("FROM manylongs\\n"); + query.append("| SORT a\\n"); + query.append("| LIMIT ").append(topCount).append(" BY a, b, c, d, e\\n"); + query.append("| KEEP a, b\"}"); + return responseAsMap(query(query.toString(), null)); + } + + /** + * Like {@code testStupidTopN} in {@link HeapAttackIT}, but for {@code SORT | LIMIT BY}. + * A near-{@code Integer.MAX_VALUE} per-group limit forces the first {@code TopNQueue} + * to try to pre-allocate a ~16 GB heap array, which must trip the circuit breaker immediately. + */ + public void testTopNByLargePerGroupCountTooMuchMemory() throws IOException { + assumeTrue("SORT | LIMIT BY requires snapshot builds", Build.current().isSnapshot()); + initManyLongs(1); + assertCircuitBreaksVia(attempt -> { + StringBuilder query = startQuery(); + query.append("FROM manylongs | SORT a | LIMIT 2147483630 BY a | KEEP a\"}"); + return responseAsMap(query(query.toString(), null)); + }, GroupedTopNOperator.class.getName(), "org.elasticsearch.compute.operator.topn.TopNQueue"); + } + + /** + * Returns the concrete {@link BytesRefSwissHash} class when swiss-table hashing is enabled, + * or {@link BytesRefHash} otherwise. Use this when asserting circuit breaks that originate + * from the group-key hash table. + */ + private static String bytesRefHashClassName() { + return HashImplFactory.SWISS_TABLES_HASHING.isEnabled() ? BytesRefSwissHash.class.getName() : BytesRefHash.class.getName(); + } +} diff --git a/test/external-modules/esql-heap-attack/src/javaRestTest/java/org/elasticsearch/xpack/esql/heap_attack/HeapAttackTestCase.java b/test/external-modules/esql-heap-attack/src/javaRestTest/java/org/elasticsearch/xpack/esql/heap_attack/HeapAttackTestCase.java index 04e740c362f9b..3b9567861e04a 100644 --- a/test/external-modules/esql-heap-attack/src/javaRestTest/java/org/elasticsearch/xpack/esql/heap_attack/HeapAttackTestCase.java +++ b/test/external-modules/esql-heap-attack/src/javaRestTest/java/org/elasticsearch/xpack/esql/heap_attack/HeapAttackTestCase.java @@ -36,6 +36,7 @@ import org.junit.ClassRule; import java.io.IOException; +import java.util.Arrays; import java.util.List; import java.util.Locale; import java.util.Map; @@ -87,7 +88,7 @@ protected void assertCircuitBreaks(TryCircuitBreaking tryBreaking, MapMatcher re while (attempt <= MAX_ATTEMPTS) { try { Map response = tryBreaking.attempt(attempt); - logger.warn("{}: should circuit broken but got {}", attempt, response); + logger.warn("{}: should have circuit broken but got {}", attempt, response); attempt++; } catch (ResponseException e) { Map map = responseAsMap(e.getResponse()); @@ -98,6 +99,52 @@ protected void assertCircuitBreaks(TryCircuitBreaking tryBreaking, MapMatcher re fail("giving up circuit breaking after " + attempt + " attempts"); } + /** + * Asserts that the given operation eventually trips the circuit breaker via the expected code + * path, as confirmed by all {@code classes} appearing in the exception's stack trace. + *

+ * Unlike {@link #assertCircuitBreaks} which stops on the first circuit-breaking exception + * regardless of origin, this method continues to the next attempt if the exception came from + * a different part of the pipeline (e.g. an upstream operator tripping before the operator + * under test). Each attempt scales the load, so a later attempt is more likely to reach the + * operator under test. + */ + protected void assertCircuitBreaksVia(TryCircuitBreaking tryBreaking, String... classNames) throws IOException { + List expected = Arrays.asList(classNames); + int attempt = 1; + while (attempt <= MAX_ATTEMPTS) { + logger.info("Attempt {} to circuit break via {}", attempt, expected); + try { + Map response = tryBreaking.attempt(attempt); + logger.warn("{}: should have circuit broken but got {}", attempt, response); + } catch (ResponseException e) { + Map map = responseAsMap(e.getResponse()); + Object error = map.get("error"); + if (error instanceof Map errorMap + && "circuit_breaking_exception".equals(errorMap.get("type")) + && errorMap.get("stack_trace") instanceof String stackTrace) { + if (expected.stream().allMatch(stackTrace::contains)) { + assertMap( + map, + matchesMap().entry("status", 429) + .entry("error", matchesMap().extraOk().entry("type", "circuit_breaking_exception")) + ); + return; + } else { + logger.warn( + "{}: circuit broke but not via expected classes {}. The stacktrace was: {}", + attempt, + expected, + stackTrace + ); + } + } + } + attempt++; + } + fail("giving up after " + (attempt - 1) + " attempts waiting for circuit break via " + expected); + } + protected Response query(String query, String filterPath) throws IOException { Request request = new Request("POST", "/_query"); request.addParameter("error_trace", ""); @@ -325,6 +372,100 @@ protected static StringBuilder startQuery() { return query; } + /** + * Loads {@code countPerLong^5} documents into the {@code manylongs} index with fields + * {@code a, b, c, d, e ∈ [0, countPerLong-1]}, one document per unique combination. + */ + protected void initManyLongs(int countPerLong) throws IOException { + logger.info("loading many documents with longs"); + StringBuilder bulk = new StringBuilder(); + int flush = 0; + long numLongs = (long) countPerLong * countPerLong * countPerLong * countPerLong * countPerLong; + for (int a = 0; a < countPerLong; a++) { + for (int b = 0; b < countPerLong; b++) { + for (int c = 0; c < countPerLong; c++) { + for (int d = 0; d < countPerLong; d++) { + for (int e = 0; e < countPerLong; e++) { + bulk.append(String.format(Locale.ROOT, """ + {"create":{}} + {"a":%d,"b":%d,"c":%d,"d":%d,"e":%d} + """, a, b, c, d, e)); + flush++; + if (flush % 10_000 == 0) { + bulk("manylongs", bulk.toString()); + bulk.setLength(0); + logger.info("flushing {}/{} to manylongs", flush, numLongs); + } + } + } + } + } + } + initIndex("manylongs", bulk.toString()); + } + + /** + * Like {@link #initManyLongs} but also adds a keyword field {@code f} containing a string + * of the given length. This produces wide group keys without needing many EVAL columns. + */ + protected void initManyLongsAndString(int countPerLong, int stringLength) throws IOException { + logger.info("loading many documents with longs and a {}-char string", stringLength); + String f = "x".repeat(stringLength); + StringBuilder bulk = new StringBuilder(); + int flush = 0; + long numDocs = (long) countPerLong * countPerLong * countPerLong * countPerLong * countPerLong; + for (int a = 0; a < countPerLong; a++) { + for (int b = 0; b < countPerLong; b++) { + for (int c = 0; c < countPerLong; c++) { + for (int d = 0; d < countPerLong; d++) { + for (int e = 0; e < countPerLong; e++) { + bulk.append(String.format(Locale.ROOT, """ + {"create":{}} + {"a":%d,"b":%d,"c":%d,"d":%d,"e":%d,"f":"%s"} + """, a, b, c, d, e, f)); + flush++; + if (flush % 10_000 == 0) { + bulk("manylongsandstring", bulk.toString()); + bulk.setLength(0); + logger.info("flushing {}/{} to manylongsandstring", flush, numDocs); + } + } + } + } + } + } + initIndex("manylongsandstring", bulk.toString()); + } + + /** + * Builds a query preamble that EVALs {@code count} computed long columns + * ({@code i0, i1, ..., i(count-1)}) as running sums over {@code a} and {@code b}. + */ + protected static StringBuilder makeManyLongs(int count) { + StringBuilder query = startQuery(); + query.append("FROM manylongs\\n| EVAL i0 = a + b, i1 = b + i0"); + for (int i = 2; i < count; i++) { + query.append(", i").append(i).append(" = i").append(i - 2).append(" + ").append(i - 1); + } + return query.append("\\n"); + } + + protected void initSingleDocIndex() throws IOException { + logger.info("loading a single document"); + initIndex("single", """ + {"create":{}} + {"a":1} + """); + } + + protected void setRequestBreakerLimit(String limit) throws IOException { + Request request = new Request("PUT", "/_cluster/settings"); + request.setJsonEntity( + "{\"persistent\": {\"indices.breaker.request.limit\": " + (limit == null ? "null" : "\"" + limit + "\"") + "}}" + ); + adminClient().performRequest(request); + } + protected static boolean isServerless() throws IOException { for (Map nodeInfo : getNodesInfo(adminClient()).values()) { for (Object module : (List) nodeInfo.get("modules")) {