From 75cb1a7efda3d8fd35b7bacf1f0cd2d9c38a002d Mon Sep 17 00:00:00 2001 From: ncordon Date: Fri, 27 Mar 2026 12:14:46 +0100 Subject: [PATCH 01/18] [ESQL] Adds HeapAttack tests for LIMIT BY and SORT | LIMIT BY --- .../esql-heap-attack/build.gradle | 5 + .../xpack/esql/heap_attack/HeapAttackIT.java | 51 --- .../esql/heap_attack/HeapAttackLimitByIT.java | 298 ++++++++++++++++++ .../esql/heap_attack/HeapAttackTestCase.java | 57 ++++ .../operator/GroupedLimitOperator.java | 2 +- .../compute/operator/topn/TopNQueue.java | 2 +- .../compute/operator/topn/TopNRow.java | 2 +- 7 files changed, 363 insertions(+), 54 deletions(-) create mode 100644 test/external-modules/esql-heap-attack/src/javaRestTest/java/org/elasticsearch/xpack/esql/heap_attack/HeapAttackLimitByIT.java 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..5d51a7a962591 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 @@ -270,16 +270,6 @@ private Response groupOnManyLongs(int count) throws IOException { query.append("\\n| STATS MAX(a)\"}"); 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); @@ -686,47 +676,6 @@ private Map queryDuplicatedHistograms(String index, String colum String queryStr = query.toString().replace("\n", "\\n"); 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..96fcde6da825e --- /dev/null +++ b/test/external-modules/esql-heap-attack/src/javaRestTest/java/org/elasticsearch/xpack/esql/heap_attack/HeapAttackLimitByIT.java @@ -0,0 +1,298 @@ +/* + * 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.client.ResponseException; +import org.elasticsearch.compute.operator.GroupKeyEncoder; +import org.elasticsearch.compute.operator.GroupedLimitOperator; +import org.elasticsearch.compute.operator.topn.GroupedTopNOperator; +import org.elasticsearch.compute.operator.topn.TopNQueue; +import org.elasticsearch.compute.operator.topn.TopNRow; +import org.elasticsearch.swisshash.BytesRefSwissHash; +import org.elasticsearch.test.ListMatcher; + +import java.io.IOException; +import java.util.Arrays; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; + +import static org.elasticsearch.test.ListMatcher.matchesList; +import static org.elasticsearch.test.MapMatcher.assertMap; +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 { + + // ------------------------------------------------------------------------- + // LIMIT BY — GroupedLimitOperator + // ------------------------------------------------------------------------- + + /** + * This limits by 100 computed columns which should succeed. The GroupedLimitOperator stores group + * keys in a hash table; 100K unique groups with 105 columns each is well within the circuit breaker. + */ + public void testLimitBySomeLongs() throws IOException { + assumeTrue("LIMIT BY requires snapshot builds", Build.current().isSnapshot()); + initManyLongs(10); + Map result = limitByManyLongs(100); + 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. Grouping by all 5 original + * fields (a,b,c,d,e) gives 100K unique groups; adding many computed columns widens each key + * until the hash table trips the circuit breaker. + */ + public void testLimitByTooMuchMemory() throws IOException { + assumeTrue("LIMIT BY requires snapshot builds", Build.current().isSnapshot()); + initManyLongs(10); + assertCircuitBreaksVia(attempt -> limitByManyLongs(attempt * 1000), GroupedLimitOperator.class, BytesRefSwissHash.class); + } + + private Map limitByManyLongs(int count) throws IOException { + logger.info("limit by {} group cols", count); + StringBuilder query = makeManyLongs(count); + // Group by all 5 original fields so there are 10^5 = 100K unique groups, + // then also include the computed fields to widen each group key. + query.append("| LIMIT 1 BY a, b, c, d, e, i0"); + for (int i = 1; i < count; i++) { + query.append(", i").append(i); + } + query.append("\\n| STATS MAX(a)\"}"); + return responseAsMap(query(query.toString(), null)); + } + + /** + * Grouping by 50 repeated 1MB strings should succeed. The {@code GroupKeyEncoder} scratch + * buffer grows to 50MB; combined with 50MB from REPEAT scratch buffers and 50MB of block + * storage, this stays comfortably within the circuit breaker. + */ + public void testLimitByKeyEncoderDoesNotCircuitBreak() throws IOException { + assumeTrue("LIMIT BY requires snapshot builds", Build.current().isSnapshot()); + initSingleDocIndex(); + Map result = limitByManyStrings(50); + 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 1MB each (one document): + *

    + *
  • N REPEAT scratch buffers (labeled {@code "repeat"}): N MB
    • + *
  • N keyword blocks: N MB
    • + *
  • keyEncoder scratch (accumulated during encoding): up to N MB
    • + *
+ * EVAL succeeds while 2N < limit, then the keyEncoder scratch pushes 2N + K over the limit + * on the K-th column encoded. The window 103 ≤ N ≤ 153 satisfies both constraints at once. + */ + public void testLimitByKeyEncoderTooMuchMemory() throws IOException { + assumeTrue("LIMIT BY requires snapshot builds", Build.current().isSnapshot()); + initSingleDocIndex(); + assertCircuitBreaksVia(attempt -> limitByManyStrings(attempt * 110), GroupedLimitOperator.class, GroupKeyEncoder.class); + } + + 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)); + } + + // ------------------------------------------------------------------------- + // SORT | LIMIT BY — GroupedTopNOperator + // ------------------------------------------------------------------------- + + /** + * This sorts by 500 columns then limits 1000 rows per group, which should succeed. The sort keys + * are stored per row in the GroupedTopNOperator queues; 500 columns × 10K rows is within limits. + */ + public void testTopNBySomeLongs() throws IOException { + assumeTrue("SORT | LIMIT BY requires snapshot builds", Build.current().isSnapshot()); + initManyLongs(10); + Map result = topNByManyLongs(500); + 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 stores full sort keys for every row in its per-group queues. Sorting by + * many columns means each stored row is very wide; with 1000 rows per group the total memory + * trips the circuit breaker. + */ + public void testTopNByTooMuchMemory() throws IOException { + assumeTrue("SORT | LIMIT BY requires snapshot builds", Build.current().isSnapshot()); + initManyLongs(10); + assertCircuitBreaksVia(attempt -> topNByManyLongs(attempt * 5000), GroupedTopNOperator.class, TopNRow.class); + } + + private Map topNByManyLongs(int count) throws IOException { + logger.info("topn by with {} sort cols", count); + StringBuilder query = makeManyLongs(count); + // Sort by all computed columns so they cannot be column-pruned and must be + // stored as sort keys in GroupedTopNOperator's per-group TopNQueues. + query.append("| SORT a, b, i0"); + for (int i = 1; i < count; i++) { + query.append(", i").append(i); + } + query.append("\\n| LIMIT 1000 BY a\\n| KEEP a, b\"}"); + return responseAsMap(query(query.toString(), null)); + } + + /** + * This sorts by 1 column then limits 1 row per group across 100K unique groups with 100 group + * key columns, which should succeed. The GroupedTopNOperator stores group keys in a hash table; + * 100K groups with 105-column keys is within the circuit breaker. + */ + public void testTopNByWideGroupKeySomeLongs() throws IOException { + assumeTrue("SORT | LIMIT BY requires snapshot builds", Build.current().isSnapshot()); + initManyLongs(10); + Map result = topNByWideGroupKey(100); + 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 stores group keys in a hash table ({@code keysHash}). Grouping by + * all 5 original fields plus many computed columns widens each encoded group key until the + * hash table trips the circuit breaker. Unlike {@link #testTopNByTooMuchMemory} which stresses + * sort key row storage, this stresses the group key hash table. + */ + public void testTopNByWideGroupKeyTooMuchMemory() throws IOException { + assumeTrue("SORT | LIMIT BY requires snapshot builds", Build.current().isSnapshot()); + initManyLongs(10); + assertCircuitBreaksVia(attempt -> topNByWideGroupKey(attempt * 1000), GroupedTopNOperator.class, BytesRefSwissHash.class); + } + + private Map topNByWideGroupKey(int count) throws IOException { + logger.info("topn by with {} group key cols", count); + StringBuilder query = makeManyLongs(count); + // Sort by a single column to keep TopNRow narrow; group by all 5 original fields + // plus computed columns to widen the composite group key stored in keysHash. + // 100K unique groups (a,b,c,d,e ∈ [0,9]) × wide encoded keys trips the circuit breaker. + query.append("| SORT a\\n"); + query.append("| LIMIT 1 BY a, b, c, d, e, i0"); + for (int i = 1; i < count; i++) { + query.append(", i").append(i); + } + query.append("\\n| KEEP a, b\"}"); + 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, TopNQueue.class); + } + + 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)); + } + + /** + * 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 the base {@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 {@code LIMIT} tripping before + * the operator under test). Each attempt scales the load, so a later attempt is more likely + * to reach the operator under test. + */ + private void assertCircuitBreaksVia(TryCircuitBreaking tryBreaking, Class... classes) throws IOException { + List classNames = Arrays.stream(classes).map(Class::getName).collect(Collectors.toList()); + int attempt = 1; + while (attempt <= MAX_ATTEMPTS) { + try { + Map response = tryBreaking.attempt(attempt); + logger.warn("{}: should 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 + && classNames.stream().allMatch(stackTrace::contains)) { + assertMap( + map, + matchesMap().entry("status", 429) + .entry("error", matchesMap().extraOk().entry("type", "circuit_breaking_exception")) + ); + return; + } + logger.warn("{}: circuit broke but not via expected classes {}: {}", attempt, classNames, map); + } + attempt++; + } + fail("giving up after " + attempt + " attempts waiting for circuit break via " + classNames); + } +} 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..c46b6a2a8e3f9 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 @@ -325,6 +325,63 @@ 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()); + } + + /** + * 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 static boolean isServerless() throws IOException { for (Map nodeInfo : getNodesInfo(adminClient()).values()) { for (Object module : (List) nodeInfo.get("modules")) { diff --git a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/GroupedLimitOperator.java b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/GroupedLimitOperator.java index ec63f969f2ec4..4087c81d5d0d6 100644 --- a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/GroupedLimitOperator.java +++ b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/GroupedLimitOperator.java @@ -99,7 +99,7 @@ public GroupedLimitOperator(int limitPerGroup, GroupKeyEncoder keyEncoder, Block success = true; } finally { if (success == false) { - Releasables.closeExpectNoException(keyEncoder, seenKeys); + Releasables.closeExpectNoException(keyEncoder, seenKeys, counts); } } } diff --git a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/topn/TopNQueue.java b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/topn/TopNQueue.java index 8bcda7d16e2c8..01164fd5a61bd 100644 --- a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/topn/TopNQueue.java +++ b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/topn/TopNQueue.java @@ -21,7 +21,7 @@ * Used both by {@link TopNOperator} (one global queue) and by {@link GroupedQueue} * (one queue per group). */ -class TopNQueue extends PriorityQueue implements Accountable, Releasable { +public class TopNQueue extends PriorityQueue implements Accountable, Releasable { private static final long SHALLOW_SIZE = RamUsageEstimator.shallowSizeOfInstance(TopNQueue.class); private final CircuitBreaker breaker; diff --git a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/topn/TopNRow.java b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/topn/TopNRow.java index 951d85145ed16..8cba8642cc985 100644 --- a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/topn/TopNRow.java +++ b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/topn/TopNRow.java @@ -22,7 +22,7 @@ * A single row in a top-N operation. Stores encoded sort keys and values. * Implements {@link Comparable} and {@link #equals} comparing the sort keys. */ -final class TopNRow implements Accountable, Comparable, Releasable { +public final class TopNRow implements Accountable, Comparable, Releasable { private static final long SHALLOW_SIZE = RamUsageEstimator.shallowSizeOfInstance(TopNRow.class); private final CircuitBreaker breaker; From be0b58a582672a8aee5c9feb3fd9be72e62b0ade Mon Sep 17 00:00:00 2001 From: elasticsearchmachine Date: Fri, 27 Mar 2026 17:10:39 +0000 Subject: [PATCH 02/18] [CI] Auto commit changes from spotless --- .../elasticsearch/xpack/esql/heap_attack/HeapAttackIT.java | 2 ++ .../xpack/esql/heap_attack/HeapAttackLimitByIT.java | 3 +-- .../xpack/esql/heap_attack/HeapAttackTestCase.java | 6 +----- 3 files changed, 4 insertions(+), 7 deletions(-) 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 5d51a7a962591..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 @@ -270,6 +270,7 @@ private Response groupOnManyLongs(int count) throws IOException { query.append("\\n| STATS MAX(a)\"}"); return query(query.toString(), null); } + public void testSmallConcat() throws IOException { initSingleDocIndex(); Response resp = concat(2); @@ -676,6 +677,7 @@ private Map queryDuplicatedHistograms(String index, String colum String queryStr = query.toString().replace("\n", "\\n"); return responseAsMap(query(queryStr, null)); } + 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 index 96fcde6da825e..2f966cdac544b 100644 --- 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 @@ -284,8 +284,7 @@ private void assertCircuitBreaksVia(TryCircuitBreaking tryBreaking, Class... && classNames.stream().allMatch(stackTrace::contains)) { assertMap( map, - matchesMap().entry("status", 429) - .entry("error", matchesMap().extraOk().entry("type", "circuit_breaking_exception")) + matchesMap().entry("status", 429).entry("error", matchesMap().extraOk().entry("type", "circuit_breaking_exception")) ); return; } 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 c46b6a2a8e3f9..79f4521c33102 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 @@ -347,11 +347,7 @@ protected void initManyLongs(int countPerLong) throws IOException { if (flush % 10_000 == 0) { bulk("manylongs", bulk.toString()); bulk.setLength(0); - logger.info( - "flushing {}/{} to manylongs", - flush, - numLongs - ); + logger.info("flushing {}/{} to manylongs", flush, numLongs); } } } From b4649114e44933f9a1080d066b75a7bca8cc03b4 Mon Sep 17 00:00:00 2001 From: ncordon Date: Fri, 27 Mar 2026 18:10:06 +0100 Subject: [PATCH 03/18] Adds test for GroupKeyEncoder and TopNBy --- .../esql/heap_attack/HeapAttackLimitByIT.java | 40 +++++++++++++++++++ 1 file changed, 40 insertions(+) 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 index 2f966cdac544b..742748eb62f98 100644 --- 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 @@ -220,6 +220,46 @@ private Map topNByWideGroupKey(int count) throws IOException { return responseAsMap(query(query.toString(), null)); } + /** + * Grouping by 50 repeated 1MB strings with a preceding sort should succeed. + * Same memory model as {@link #testLimitByKeyEncoderDoesNotCircuitBreak} 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(50); + 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 * 110), GroupedTopNOperator.class, GroupKeyEncoder.class); + } + + 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. */ From d0ab8625f052369361ab1781c74049f9024af6ff Mon Sep 17 00:00:00 2001 From: ncordon Date: Fri, 27 Mar 2026 18:20:40 +0100 Subject: [PATCH 04/18] Gates the hash table implementation --- .../esql/heap_attack/HeapAttackLimitByIT.java | 15 +++++++++++++-- 1 file changed, 13 insertions(+), 2 deletions(-) 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 index 742748eb62f98..0a46fd2fe09e1 100644 --- 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 @@ -12,6 +12,8 @@ import org.apache.lucene.tests.util.TimeUnits; import org.elasticsearch.Build; import org.elasticsearch.client.ResponseException; +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; @@ -64,7 +66,7 @@ public void testLimitBySomeLongs() throws IOException { public void testLimitByTooMuchMemory() throws IOException { assumeTrue("LIMIT BY requires snapshot builds", Build.current().isSnapshot()); initManyLongs(10); - assertCircuitBreaksVia(attempt -> limitByManyLongs(attempt * 1000), GroupedLimitOperator.class, BytesRefSwissHash.class); + assertCircuitBreaksVia(attempt -> limitByManyLongs(attempt * 1000), GroupedLimitOperator.class, bytesRefHashClass()); } private Map limitByManyLongs(int count) throws IOException { @@ -202,7 +204,7 @@ public void testTopNByWideGroupKeySomeLongs() throws IOException { public void testTopNByWideGroupKeyTooMuchMemory() throws IOException { assumeTrue("SORT | LIMIT BY requires snapshot builds", Build.current().isSnapshot()); initManyLongs(10); - assertCircuitBreaksVia(attempt -> topNByWideGroupKey(attempt * 1000), GroupedTopNOperator.class, BytesRefSwissHash.class); + assertCircuitBreaksVia(attempt -> topNByWideGroupKey(attempt * 1000), GroupedTopNOperator.class, bytesRefHashClass()); } private Map topNByWideGroupKey(int count) throws IOException { @@ -298,6 +300,15 @@ private Map topNByManyGroups(int topCount) throws IOException { return responseAsMap(query(query.toString(), null)); } + /** + * 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 Class bytesRefHashClass() { + return HashImplFactory.SWISS_TABLES_HASHING.isEnabled() ? BytesRefSwissHash.class : BytesRefHash.class; + } + /** * 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. From a25e1f52e5b4c5e931fa4aafcc45724b39ca6e82 Mon Sep 17 00:00:00 2001 From: ncordon Date: Mon, 30 Mar 2026 16:17:45 +0200 Subject: [PATCH 05/18] Tweaks tests --- .../esql/heap_attack/HeapAttackLimitByIT.java | 98 +++++++++++++------ .../esql/heap_attack/HeapAttackTestCase.java | 8 ++ 2 files changed, 76 insertions(+), 30 deletions(-) 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 index 0a46fd2fe09e1..6de46220f09ea 100644 --- 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 @@ -21,6 +21,8 @@ import org.elasticsearch.compute.operator.topn.TopNRow; 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.Arrays; @@ -41,18 +43,44 @@ @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: + *

    + *
  • We need to discount size of cached pages in {@code PageCacheRecycler}. + * Those cached pages are invisible to the breaker but still consume heap.
    • + *
  • ES classpath also takes up heap space
    • + *
  • Memory fragmentation might cause the circuit breaker to think we can allocate + * an array when it's not possible to find a big enough gap in 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 // ------------------------------------------------------------------------- /** - * This limits by 100 computed columns which should succeed. The GroupedLimitOperator stores group - * keys in a hash table; 100K unique groups with 105 columns each is well within the circuit breaker. + * This limits by 50 computed columns which should succeed. The GroupedLimitOperator stores group + * keys in a hash table; 100K unique groups with 55 columns each is well within the circuit breaker. */ - public void testLimitBySomeLongs() throws IOException { + public void testLimitByManyGroupingColumns() throws IOException { assumeTrue("LIMIT BY requires snapshot builds", Build.current().isSnapshot()); initManyLongs(10); - Map result = limitByManyLongs(100); + Map result = limitByManyLongs(50); ListMatcher columns = matchesList().item(matchesMap().entry("name", "MAX(a)").entry("type", "long")); ListMatcher values = matchesList().item(List.of(9)); assertResultMap(result, columns, values); @@ -63,10 +91,10 @@ public void testLimitBySomeLongs() throws IOException { * fields (a,b,c,d,e) gives 100K unique groups; adding many computed columns widens each key * until the hash table trips the circuit breaker. */ - public void testLimitByTooMuchMemory() throws IOException { + public void testLimitByManyGroupingColumnsTooMuchMemory() throws IOException { assumeTrue("LIMIT BY requires snapshot builds", Build.current().isSnapshot()); initManyLongs(10); - assertCircuitBreaksVia(attempt -> limitByManyLongs(attempt * 1000), GroupedLimitOperator.class, bytesRefHashClass()); + assertCircuitBreaksVia(attempt -> limitByManyLongs(attempt * 500), GroupedLimitOperator.class, bytesRefHashClass()); } private Map limitByManyLongs(int count) throws IOException { @@ -83,14 +111,19 @@ private Map limitByManyLongs(int count) throws IOException { } /** - * Grouping by 50 repeated 1MB strings should succeed. The {@code GroupKeyEncoder} scratch - * buffer grows to 50MB; combined with 50MB from REPEAT scratch buffers and 50MB of block - * storage, this stays comfortably within the circuit breaker. + * 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(50); + Map result = limitByManyStrings(30); ListMatcher columns = matchesList().item(matchesMap().entry("name", "a").entry("type", "long")); assertResultMap(result, columns, matchesList().item(List.of(1))); } @@ -101,19 +134,23 @@ public void testLimitByKeyEncoderDoesNotCircuitBreak() throws IOException { * 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 1MB each (one document): + * Memory model with N columns of 1 MB each (one document): *

    - *
  • N REPEAT scratch buffers (labeled {@code "repeat"}): N MB
    • + *
  • 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 N MB
    • + *
  • keyEncoder scratch (accumulated during encoding): up to ~1.125N MB (also oversized)
    • *
- * EVAL succeeds while 2N < limit, then the keyEncoder scratch pushes 2N + K over the limit - * on the K-th column encoded. The window 103 ≤ N ≤ 153 satisfies both constraints at once. + * 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 * 110), GroupedLimitOperator.class, GroupKeyEncoder.class); + assertCircuitBreaksVia(attempt -> limitByManyStrings(attempt * 80), GroupedLimitOperator.class, GroupKeyEncoder.class); } private Map limitByManyStrings(int count) throws IOException { @@ -148,7 +185,7 @@ private Map limitByManyStrings(int count) throws IOException { * This sorts by 500 columns then limits 1000 rows per group, which should succeed. The sort keys * are stored per row in the GroupedTopNOperator queues; 500 columns × 10K rows is within limits. */ - public void testTopNBySomeLongs() throws IOException { + public void testTopNByManySortColumns() throws IOException { assumeTrue("SORT | LIMIT BY requires snapshot builds", Build.current().isSnapshot()); initManyLongs(10); Map result = topNByManyLongs(500); @@ -162,10 +199,10 @@ public void testTopNBySomeLongs() throws IOException { * many columns means each stored row is very wide; with 1000 rows per group the total memory * trips the circuit breaker. */ - public void testTopNByTooMuchMemory() throws IOException { + public void testTopNByManySortColumnsTooMuchMemory() throws IOException { assumeTrue("SORT | LIMIT BY requires snapshot builds", Build.current().isSnapshot()); initManyLongs(10); - assertCircuitBreaksVia(attempt -> topNByManyLongs(attempt * 5000), GroupedTopNOperator.class, TopNRow.class); + assertCircuitBreaksVia(attempt -> topNByManyLongs(attempt * 10000), GroupedTopNOperator.class, TopNRow.class); } private Map topNByManyLongs(int count) throws IOException { @@ -182,14 +219,14 @@ private Map topNByManyLongs(int count) throws IOException { } /** - * This sorts by 1 column then limits 1 row per group across 100K unique groups with 100 group + * This sorts by 1 column then limits 1 row per group across 50K unique groups with 100 group * key columns, which should succeed. The GroupedTopNOperator stores group keys in a hash table; - * 100K groups with 105-column keys is within the circuit breaker. + * 50K groups with 105-column keys is within the circuit breaker. */ - public void testTopNByWideGroupKeySomeLongs() throws IOException { + public void testTopNByManyGroupingColumns() throws IOException { assumeTrue("SORT | LIMIT BY requires snapshot builds", Build.current().isSnapshot()); initManyLongs(10); - Map result = topNByWideGroupKey(100); + Map result = topNByWideGroupKey(50); 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)); @@ -198,10 +235,10 @@ public void testTopNByWideGroupKeySomeLongs() throws IOException { /** * The GroupedTopNOperator stores group keys in a hash table ({@code keysHash}). Grouping by * all 5 original fields plus many computed columns widens each encoded group key until the - * hash table trips the circuit breaker. Unlike {@link #testTopNByTooMuchMemory} which stresses + * hash table trips the circuit breaker. Unlike {@link #testTopNByManySortColumns} which stresses * sort key row storage, this stresses the group key hash table. */ - public void testTopNByWideGroupKeyTooMuchMemory() throws IOException { + public void testTopNByManyGroupingColumnsTooMuchMemory() throws IOException { assumeTrue("SORT | LIMIT BY requires snapshot builds", Build.current().isSnapshot()); initManyLongs(10); assertCircuitBreaksVia(attempt -> topNByWideGroupKey(attempt * 1000), GroupedTopNOperator.class, bytesRefHashClass()); @@ -223,14 +260,14 @@ private Map topNByWideGroupKey(int count) throws IOException { } /** - * Grouping by 50 repeated 1MB strings with a preceding sort should succeed. - * Same memory model as {@link #testLimitByKeyEncoderDoesNotCircuitBreak} but with - * {@code GroupedTopNOperator} rather than {@code GroupedLimitOperator}. + * 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(50); + Map result = topNByManyStrings(30); ListMatcher columns = matchesList().item(matchesMap().entry("name", "a").entry("type", "long")); assertResultMap(result, columns, matchesList().item(List.of(1))); } @@ -243,7 +280,7 @@ public void testTopNByKeyEncoderDoesNotCircuitBreak() throws IOException { public void testTopNByKeyEncoderTooMuchMemory() throws IOException { assumeTrue("SORT | LIMIT BY requires snapshot builds", Build.current().isSnapshot()); initSingleDocIndex(); - assertCircuitBreaksVia(attempt -> topNByManyStrings(attempt * 110), GroupedTopNOperator.class, GroupKeyEncoder.class); + assertCircuitBreaksVia(attempt -> topNByManyStrings(attempt * 80), GroupedTopNOperator.class, GroupKeyEncoder.class); } private Map topNByManyStrings(int count) throws IOException { @@ -323,6 +360,7 @@ private void assertCircuitBreaksVia(TryCircuitBreaking tryBreaking, Class... List classNames = Arrays.stream(classes).map(Class::getName).collect(Collectors.toList()); int attempt = 1; while (attempt <= MAX_ATTEMPTS) { + logger.info("Attempt {} to circuit break via {}", attempt, classNames); try { Map response = tryBreaking.attempt(attempt); logger.warn("{}: should circuit broken but got {}", attempt, response); 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 79f4521c33102..8735c8a89082f 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 @@ -378,6 +378,14 @@ protected void initSingleDocIndex() throws IOException { """); } + 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")) { From 91a4b1daca6711a754ca41a9155577e2b249b735 Mon Sep 17 00:00:00 2001 From: ncordon Date: Mon, 30 Mar 2026 16:45:10 +0200 Subject: [PATCH 06/18] Moves helper to common class for it to be reused --- .../esql/heap_attack/HeapAttackLimitByIT.java | 42 ------------------- .../esql/heap_attack/HeapAttackTestCase.java | 40 ++++++++++++++++++ 2 files changed, 40 insertions(+), 42 deletions(-) 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 index 6de46220f09ea..be40a333aa8cc 100644 --- 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 @@ -11,7 +11,6 @@ import org.apache.lucene.tests.util.TimeUnits; import org.elasticsearch.Build; -import org.elasticsearch.client.ResponseException; import org.elasticsearch.common.util.BytesRefHash; import org.elasticsearch.compute.aggregation.blockhash.HashImplFactory; import org.elasticsearch.compute.operator.GroupKeyEncoder; @@ -25,13 +24,10 @@ import org.junit.Before; import java.io.IOException; -import java.util.Arrays; import java.util.List; import java.util.Map; -import java.util.stream.Collectors; import static org.elasticsearch.test.ListMatcher.matchesList; -import static org.elasticsearch.test.MapMatcher.assertMap; import static org.elasticsearch.test.MapMatcher.matchesMap; import static org.hamcrest.Matchers.any; @@ -345,42 +341,4 @@ private Map topNByManyGroups(int topCount) throws IOException { private static Class bytesRefHashClass() { return HashImplFactory.SWISS_TABLES_HASHING.isEnabled() ? BytesRefSwissHash.class : BytesRefHash.class; } - - /** - * 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 the base {@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 {@code LIMIT} tripping before - * the operator under test). Each attempt scales the load, so a later attempt is more likely - * to reach the operator under test. - */ - private void assertCircuitBreaksVia(TryCircuitBreaking tryBreaking, Class... classes) throws IOException { - List classNames = Arrays.stream(classes).map(Class::getName).collect(Collectors.toList()); - int attempt = 1; - while (attempt <= MAX_ATTEMPTS) { - logger.info("Attempt {} to circuit break via {}", attempt, classNames); - try { - Map response = tryBreaking.attempt(attempt); - logger.warn("{}: should 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 - && classNames.stream().allMatch(stackTrace::contains)) { - assertMap( - map, - matchesMap().entry("status", 429).entry("error", matchesMap().extraOk().entry("type", "circuit_breaking_exception")) - ); - return; - } - logger.warn("{}: circuit broke but not via expected classes {}: {}", attempt, classNames, map); - } - attempt++; - } - fail("giving up after " + attempt + " attempts waiting for circuit break via " + classNames); - } } 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 8735c8a89082f..c1e9f66d7ae57 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,10 +36,12 @@ import org.junit.ClassRule; import java.io.IOException; +import java.util.Arrays; import java.util.List; import java.util.Locale; import java.util.Map; import java.util.function.IntFunction; +import java.util.stream.Collectors; import static org.elasticsearch.common.Strings.hasText; import static org.elasticsearch.test.MapMatcher.assertMap; @@ -98,6 +100,44 @@ 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, Class... classes) throws IOException { + List classNames = Arrays.stream(classes).map(Class::getName).collect(Collectors.toList()); + int attempt = 1; + while (attempt <= MAX_ATTEMPTS) { + logger.info("Attempt {} to circuit break via {}", attempt, classNames); + try { + Map response = tryBreaking.attempt(attempt); + logger.warn("{}: should 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 + && classNames.stream().allMatch(stackTrace::contains)) { + assertMap( + map, + matchesMap().entry("status", 429) + ); + return; + } + logger.warn("{}: circuit broke but not via expected classes {}: {}", attempt, classNames, map); + } + attempt++; + } + fail("giving up after " + attempt + " attempts waiting for circuit break via " + classNames); + } + protected Response query(String query, String filterPath) throws IOException { Request request = new Request("POST", "/_query"); request.addParameter("error_trace", ""); From a3148c63d217219058755ff73b63356cad32ec56 Mon Sep 17 00:00:00 2001 From: elasticsearchmachine Date: Mon, 30 Mar 2026 15:21:00 +0000 Subject: [PATCH 07/18] [CI] Auto commit changes from spotless --- .../xpack/esql/heap_attack/HeapAttackTestCase.java | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) 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 c1e9f66d7ae57..3cc52a6f694fc 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 @@ -125,10 +125,7 @@ protected void assertCircuitBreaksVia(TryCircuitBreaking tryBreaking, Class.. && "circuit_breaking_exception".equals(errorMap.get("type")) && errorMap.get("stack_trace") instanceof String stackTrace && classNames.stream().allMatch(stackTrace::contains)) { - assertMap( - map, - matchesMap().entry("status", 429) - ); + assertMap(map, matchesMap().entry("status", 429)); return; } logger.warn("{}: circuit broke but not via expected classes {}: {}", attempt, classNames, map); From a66b5cc6ff883fc3bae55e0bbd11023832c6b107 Mon Sep 17 00:00:00 2001 From: ncordon Date: Mon, 30 Mar 2026 18:55:49 +0200 Subject: [PATCH 08/18] Fix test --- .../xpack/esql/heap_attack/HeapAttackTestCase.java | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) 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 3cc52a6f694fc..40ae3810adeb6 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 @@ -125,7 +125,10 @@ protected void assertCircuitBreaksVia(TryCircuitBreaking tryBreaking, Class.. && "circuit_breaking_exception".equals(errorMap.get("type")) && errorMap.get("stack_trace") instanceof String stackTrace && classNames.stream().allMatch(stackTrace::contains)) { - assertMap(map, matchesMap().entry("status", 429)); + assertMap( + map, + matchesMap().entry("status", 429).entry("error", matchesMap().extraOk().entry("type", "circuit_breaking_exception")) + ); return; } logger.warn("{}: circuit broke but not via expected classes {}: {}", attempt, classNames, map); From 4c8c23fc9241438588757cd2c7653fe84f1e3cd0 Mon Sep 17 00:00:00 2001 From: ncordon Date: Tue, 31 Mar 2026 11:47:33 +0200 Subject: [PATCH 09/18] Addresses pr review comments --- .../esql/heap_attack/HeapAttackLimitByIT.java | 37 +++++++++++++++++-- .../esql/heap_attack/HeapAttackTestCase.java | 6 +-- 2 files changed, 37 insertions(+), 6 deletions(-) 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 index be40a333aa8cc..b28ed2672c130 100644 --- 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 @@ -173,6 +173,22 @@ private Map limitByManyStrings(int count) throws IOException { 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 // ------------------------------------------------------------------------- @@ -215,9 +231,9 @@ private Map topNByManyLongs(int count) throws IOException { } /** - * This sorts by 1 column then limits 1 row per group across 50K unique groups with 100 group - * key columns, which should succeed. The GroupedTopNOperator stores group keys in a hash table; - * 50K groups with 105-column keys is within the circuit breaker. + * This sorts by 1 column then limits 1 row per group across 100K unique groups with 55 group + * key computed columns, which should succeed. The GroupedTopNOperator stores group + * keys in a hash table. 100K groups with 55-column keys is within the circuit breaker. */ public void testTopNByManyGroupingColumns() throws IOException { assumeTrue("SORT | LIMIT BY requires snapshot builds", Build.current().isSnapshot()); @@ -333,6 +349,21 @@ private Map topNByManyGroups(int topCount) throws IOException { 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, TopNQueue.class); + } + /** * Returns the concrete {@link BytesRefSwissHash} class when swiss-table hashing is enabled, * or {@link BytesRefHash} otherwise. Use this when asserting circuit breaks that originate 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 40ae3810adeb6..0ad798fe3874d 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 @@ -89,7 +89,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()); @@ -117,7 +117,7 @@ protected void assertCircuitBreaksVia(TryCircuitBreaking tryBreaking, Class.. logger.info("Attempt {} to circuit break via {}", attempt, classNames); 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); } catch (ResponseException e) { Map map = responseAsMap(e.getResponse()); Object error = map.get("error"); @@ -135,7 +135,7 @@ protected void assertCircuitBreaksVia(TryCircuitBreaking tryBreaking, Class.. } attempt++; } - fail("giving up after " + attempt + " attempts waiting for circuit break via " + classNames); + fail("giving up after " + (attempt - 1) + " attempts waiting for circuit break via " + classNames); } protected Response query(String query, String filterPath) throws IOException { From 29f54cc1d41066e30e42106493ba5ae750bead3a Mon Sep 17 00:00:00 2001 From: ncordon Date: Tue, 31 Mar 2026 13:35:45 +0200 Subject: [PATCH 10/18] Reshapes a test, removes setting circuit breaker threshold lower, tracks more memory --- .../esql/heap_attack/HeapAttackLimitByIT.java | 73 ++++++------------- .../compute/operator/topn/GroupedQueue.java | 32 +++++--- .../operator/topn/GroupedTopNOperator.java | 54 ++++++++++++-- .../compute/operator/topn/TopNOperator.java | 2 +- .../operator/topn/GroupedQueueTests.java | 4 +- 5 files changed, 97 insertions(+), 68 deletions(-) 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 index b28ed2672c130..2908f19faaa92 100644 --- 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 @@ -16,12 +16,10 @@ import org.elasticsearch.compute.operator.GroupKeyEncoder; import org.elasticsearch.compute.operator.GroupedLimitOperator; import org.elasticsearch.compute.operator.topn.GroupedTopNOperator; +import org.elasticsearch.compute.operator.topn.TopNOperator; import org.elasticsearch.compute.operator.topn.TopNQueue; -import org.elasticsearch.compute.operator.topn.TopNRow; 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; @@ -39,32 +37,6 @@ @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: - *

    - *
  • We need to discount size of cached pages in {@code PageCacheRecycler}. - * Those cached pages are invisible to the breaker but still consume heap.
    • - *
  • ES classpath also takes up heap space
    • - *
  • Memory fragmentation might cause the circuit breaker to think we can allocate - * an array when it's not possible to find a big enough gap in 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 // ------------------------------------------------------------------------- @@ -194,39 +166,42 @@ public void testLimitByLargePerGroupCount() throws IOException { // ------------------------------------------------------------------------- /** - * This sorts by 500 columns then limits 1000 rows per group, which should succeed. The sort keys - * are stored per row in the GroupedTopNOperator queues; 500 columns × 10K rows is within limits. + * 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()); - initManyLongs(10); - Map result = topNByManyLongs(500); - 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)); + 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 columns means each stored row is very wide; with 1000 rows per group the total memory - * trips the circuit breaker. + * 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()); - initManyLongs(10); - assertCircuitBreaksVia(attempt -> topNByManyLongs(attempt * 10000), GroupedTopNOperator.class, TopNRow.class); + initSingleDocIndex(); + assertCircuitBreaksVia(attempt -> topNByManyStringSortCols(attempt * 80), GroupedTopNOperator.class, TopNOperator.RowFiller.class); } - private Map topNByManyLongs(int count) throws IOException { - logger.info("topn by with {} sort cols", count); - StringBuilder query = makeManyLongs(count); - // Sort by all computed columns so they cannot be column-pruned and must be - // stored as sort keys in GroupedTopNOperator's per-group TopNQueues. - query.append("| SORT a, b, i0"); + 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(", i").append(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 1000 BY a\\n| KEEP a, b\"}"); + query.append("\\n| LIMIT 1 BY a\\n| KEEP a\"}"); return responseAsMap(query(query.toString(), null)); } diff --git a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/topn/GroupedQueue.java b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/topn/GroupedQueue.java index 5f44e948e9ce6..d06a2fa5725b9 100644 --- a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/topn/GroupedQueue.java +++ b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/topn/GroupedQueue.java @@ -14,9 +14,11 @@ import org.elasticsearch.core.Releasable; import org.elasticsearch.core.Releasables; -import java.util.ArrayList; import java.util.List; +import static org.apache.lucene.util.RamUsageEstimator.NUM_BYTES_ARRAY_HEADER; +import static org.apache.lucene.util.RamUsageEstimator.NUM_BYTES_OBJECT_REF; +import static org.apache.lucene.util.RamUsageEstimator.alignObjectSize; import static org.apache.lucene.util.RamUsageEstimator.shallowSizeOfInstance; /** @@ -66,22 +68,34 @@ TopNQueue getOrCreateQueue(long groupId) { } /** - * Removes and returns all rows from all per-group queues. - * For an ascending order, the first element will be the min element (or last in the - * priority queue), and vice versa. + * Drains all rows from all per-group queues into the given list, closing each queue + * as it is emptied. The list is then sorted in descending key order so that for an + * ascending sort the first element is the minimum. */ - List popAll() { - List allRows = new ArrayList<>(size()); + void popAllInto(List target) { for (long i = 0; i < queues.size(); i++) { TopNQueue queue = queues.get(i); if (queue != null) { - queue.popAllInto(allRows); + queue.popAllInto(target); queue.close(); queues.set(i, null); } } - allRows.sort((r1, r2) -> -r1.compareTo(r2)); - return allRows; + // sort allocates an internal array for sorting + long sortBytes = estimateArrayBytes(target.size()); + breaker.addEstimateBytesAndMaybeBreak(sortBytes, "grouped_queue"); + try { + target.sort((r1, r2) -> -r1.compareTo(r2)); + } finally { + breaker.addWithoutBreaking(-sortBytes); + } + } + + /** + * Estimates the bytes used by an {@code Object[]} of the given capacity. + */ + static long estimateArrayBytes(int size) { + return alignObjectSize(NUM_BYTES_ARRAY_HEADER + (long) size * NUM_BYTES_OBJECT_REF); } @Override diff --git a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/topn/GroupedTopNOperator.java b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/topn/GroupedTopNOperator.java index 3b13ee5172544..1c1cc6a362abb 100644 --- a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/topn/GroupedTopNOperator.java +++ b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/topn/GroupedTopNOperator.java @@ -24,6 +24,7 @@ import org.elasticsearch.core.ReleasableIterator; import org.elasticsearch.core.Releasables; +import java.util.ArrayList; import java.util.Arrays; import java.util.List; @@ -312,26 +313,53 @@ private ReleasableIterator buildResult() { spare = null; } - if (inputQueue.size() == 0) { + var resultSize = inputQueue.size(); + if (resultSize == 0) { return ReleasableIterator.empty(); } - List rows = inputQueue.popAll(); - inputQueue.close(); - keysHash.close(); - inputQueue = null; - keysHash = null; - return new Result(rows); + Result result = null; + var success = false; + try { + result = new Result(breaker, resultSize); + inputQueue.popAllInto(result.getRows()); + success = true; + } finally { + if (success == false) { + Releasables.closeExpectNoException(result, inputQueue, keysHash); + inputQueue = null; + keysHash = null; + } + } + return result; } private class Result implements ReleasableIterator { + private static final long SHALLOW_SIZE = RamUsageEstimator.shallowSizeOfInstance(Result.class); private final List rows; + private final CircuitBreaker breaker; private int r; - private Result(List rows) { + Result(CircuitBreaker breaker, int resultSize) { + breaker.addEstimateBytesAndMaybeBreak(sizeOf(resultSize), "grouped topn result"); + var success = false; + List rows; + try { + rows = new ArrayList<>(resultSize); + success = true; + } finally { + if (success == false) { + this.close(); + } + } + this.breaker = breaker; this.rows = rows; } + public List getRows() { + return rows; + } + @Override public boolean hasNext() { return r < rows.size(); @@ -377,7 +405,17 @@ private long totalSize(ResultBuilder[] builders) { @Override public void close() { + var resultSize = rows.size(); Releasables.close(rows); + breaker.addWithoutBreaking(-sizeOf(resultSize)); + } + + static long sizeOf(int resultSize) { + long total = SHALLOW_SIZE; + total += RamUsageEstimator.alignObjectSize( + RamUsageEstimator.NUM_BYTES_ARRAY_HEADER + RamUsageEstimator.NUM_BYTES_OBJECT_REF * ((long) resultSize) + ); + return total; } private void readKeys(ResultBuilder[] builders, BytesRef keys) { diff --git a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/topn/TopNOperator.java b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/topn/TopNOperator.java index 3e3409e8e1adc..0c104b8d0d06b 100644 --- a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/topn/TopNOperator.java +++ b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/topn/TopNOperator.java @@ -49,7 +49,7 @@ public enum InputOrdering { * Fills {@link TopNRow}s from page data. Handles both sort-key encoding and value * extraction, and tracks pre-allocation sizes for key and value buffers. */ - static final class RowFiller { + public static final class RowFiller { private final ValueExtractor[] valueExtractors; private final KeyExtractor[] keyExtractors; diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/operator/topn/GroupedQueueTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/operator/topn/GroupedQueueTests.java index e08f817a9c4fc..4999bec0773a9 100644 --- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/operator/topn/GroupedQueueTests.java +++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/operator/topn/GroupedQueueTests.java @@ -21,6 +21,7 @@ import org.elasticsearch.indices.breaker.CircuitBreakerService; import org.elasticsearch.test.ESTestCase; +import java.util.ArrayList; import java.util.List; import static org.hamcrest.Matchers.equalTo; @@ -100,7 +101,8 @@ private void addRow(GroupedQueue queue, int groupKey, int value) { private void assertQueueContents(GroupedQueue queue, List expectedSortKeys) { assertThat(queue.size(), equalTo(expectedSortKeys.size())); - List actual = queue.popAll(); + List actual = new ArrayList<>(queue.size()); + queue.popAllInto(actual); for (int i = 0; i < expectedSortKeys.size(); i++) { int sortKey = expectedSortKeys.get(i); assertRowValues(actual.get(i), sortKey, sortKey * 2); From dac0c91fc030e42820872717c06238c44246f0e7 Mon Sep 17 00:00:00 2001 From: ncordon Date: Tue, 31 Mar 2026 14:37:39 +0200 Subject: [PATCH 11/18] Avoids making some clases public --- .../esql/heap_attack/HeapAttackLimitByIT.java | 36 +++++++++++++------ .../esql/heap_attack/HeapAttackTestCase.java | 33 ++++++++++------- .../compute/operator/topn/TopNOperator.java | 2 +- .../compute/operator/topn/TopNQueue.java | 2 +- .../compute/operator/topn/TopNRow.java | 2 +- 5 files changed, 48 insertions(+), 27 deletions(-) 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 index 2908f19faaa92..30672306c32ce 100644 --- 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 @@ -16,8 +16,6 @@ import org.elasticsearch.compute.operator.GroupKeyEncoder; import org.elasticsearch.compute.operator.GroupedLimitOperator; import org.elasticsearch.compute.operator.topn.GroupedTopNOperator; -import org.elasticsearch.compute.operator.topn.TopNOperator; -import org.elasticsearch.compute.operator.topn.TopNQueue; import org.elasticsearch.swisshash.BytesRefSwissHash; import org.elasticsearch.test.ListMatcher; @@ -62,7 +60,7 @@ public void testLimitByManyGroupingColumns() throws IOException { public void testLimitByManyGroupingColumnsTooMuchMemory() throws IOException { assumeTrue("LIMIT BY requires snapshot builds", Build.current().isSnapshot()); initManyLongs(10); - assertCircuitBreaksVia(attempt -> limitByManyLongs(attempt * 500), GroupedLimitOperator.class, bytesRefHashClass()); + assertCircuitBreaksVia(attempt -> limitByManyLongs(attempt * 500), GroupedLimitOperator.class.getName(), bytesRefHashClassName()); } private Map limitByManyLongs(int count) throws IOException { @@ -118,7 +116,11 @@ public void testLimitByKeyEncoderDoesNotCircuitBreak() throws IOException { public void testLimitByKeyEncoderTooMuchMemory() throws IOException { assumeTrue("LIMIT BY requires snapshot builds", Build.current().isSnapshot()); initSingleDocIndex(); - assertCircuitBreaksVia(attempt -> limitByManyStrings(attempt * 80), GroupedLimitOperator.class, GroupKeyEncoder.class); + assertCircuitBreaksVia( + attempt -> limitByManyStrings(attempt * 80), + GroupedLimitOperator.class.getName(), + GroupKeyEncoder.class.getName() + ); } private Map limitByManyStrings(int count) throws IOException { @@ -186,7 +188,11 @@ public void testTopNByManySortColumns() throws IOException { public void testTopNByManySortColumnsTooMuchMemory() throws IOException { assumeTrue("SORT | LIMIT BY requires snapshot builds", Build.current().isSnapshot()); initSingleDocIndex(); - assertCircuitBreaksVia(attempt -> topNByManyStringSortCols(attempt * 80), GroupedTopNOperator.class, TopNOperator.RowFiller.class); + assertCircuitBreaksVia( + attempt -> topNByManyStringSortCols(attempt * 80), + GroupedTopNOperator.class.getName(), + "TopNOperator$RowFiller" + ); } private Map topNByManyStringSortCols(int count) throws IOException { @@ -228,7 +234,7 @@ public void testTopNByManyGroupingColumns() throws IOException { public void testTopNByManyGroupingColumnsTooMuchMemory() throws IOException { assumeTrue("SORT | LIMIT BY requires snapshot builds", Build.current().isSnapshot()); initManyLongs(10); - assertCircuitBreaksVia(attempt -> topNByWideGroupKey(attempt * 1000), GroupedTopNOperator.class, bytesRefHashClass()); + assertCircuitBreaksVia(attempt -> topNByWideGroupKey(attempt * 1000), GroupedTopNOperator.class.getName(), bytesRefHashClassName()); } private Map topNByWideGroupKey(int count) throws IOException { @@ -267,7 +273,11 @@ public void testTopNByKeyEncoderDoesNotCircuitBreak() throws IOException { public void testTopNByKeyEncoderTooMuchMemory() throws IOException { assumeTrue("SORT | LIMIT BY requires snapshot builds", Build.current().isSnapshot()); initSingleDocIndex(); - assertCircuitBreaksVia(attempt -> topNByManyStrings(attempt * 80), GroupedTopNOperator.class, GroupKeyEncoder.class); + assertCircuitBreaksVia( + attempt -> topNByManyStrings(attempt * 80), + GroupedTopNOperator.class.getName(), + GroupKeyEncoder.class.getName() + ); } private Map topNByManyStrings(int count) throws IOException { @@ -308,7 +318,11 @@ public void testTopNByManyGroupsSmallTopCount() throws IOException { public void testTopNByManyGroupsLargeTopCountTooMuchMemory() throws IOException { assumeTrue("SORT | LIMIT BY requires snapshot builds", Build.current().isSnapshot()); initManyLongs(10); - assertCircuitBreaksVia(attempt -> topNByManyGroups(attempt * 400), GroupedTopNOperator.class, TopNQueue.class); + assertCircuitBreaksVia( + attempt -> topNByManyGroups(attempt * 400), + GroupedTopNOperator.class.getName(), + "org.elasticsearch.compute.operator.topn.TopNQueue" + ); } private Map topNByManyGroups(int topCount) throws IOException { @@ -336,7 +350,7 @@ public void testTopNByLargePerGroupCountTooMuchMemory() throws IOException { StringBuilder query = startQuery(); query.append("FROM manylongs | SORT a | LIMIT 2147483630 BY a | KEEP a\"}"); return responseAsMap(query(query.toString(), null)); - }, GroupedTopNOperator.class, TopNQueue.class); + }, GroupedTopNOperator.class.getName(), "org.elasticsearch.compute.operator.topn.TopNQueue"); } /** @@ -344,7 +358,7 @@ public void testTopNByLargePerGroupCountTooMuchMemory() throws IOException { * or {@link BytesRefHash} otherwise. Use this when asserting circuit breaks that originate * from the group-key hash table. */ - private static Class bytesRefHashClass() { - return HashImplFactory.SWISS_TABLES_HASHING.isEnabled() ? BytesRefSwissHash.class : BytesRefHash.class; + 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 0ad798fe3874d..67829268edb8f 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 @@ -41,7 +41,6 @@ import java.util.Locale; import java.util.Map; import java.util.function.IntFunction; -import java.util.stream.Collectors; import static org.elasticsearch.common.Strings.hasText; import static org.elasticsearch.test.MapMatcher.assertMap; @@ -110,11 +109,11 @@ protected void assertCircuitBreaks(TryCircuitBreaking tryBreaking, MapMatcher re * 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, Class... classes) throws IOException { - List classNames = Arrays.stream(classes).map(Class::getName).collect(Collectors.toList()); + 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, classNames); + 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); @@ -123,19 +122,27 @@ protected void assertCircuitBreaksVia(TryCircuitBreaking tryBreaking, Class.. Object error = map.get("error"); if (error instanceof Map errorMap && "circuit_breaking_exception".equals(errorMap.get("type")) - && errorMap.get("stack_trace") instanceof String stackTrace - && classNames.stream().allMatch(stackTrace::contains)) { - assertMap( - map, - matchesMap().entry("status", 429).entry("error", matchesMap().extraOk().entry("type", "circuit_breaking_exception")) - ); - return; + && 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 + ); + } } - logger.warn("{}: circuit broke but not via expected classes {}: {}", attempt, classNames, map); } attempt++; } - fail("giving up after " + (attempt - 1) + " attempts waiting for circuit break via " + classNames); + fail("giving up after " + (attempt - 1) + " attempts waiting for circuit break via " + expected); } protected Response query(String query, String filterPath) throws IOException { diff --git a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/topn/TopNOperator.java b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/topn/TopNOperator.java index 0c104b8d0d06b..3e3409e8e1adc 100644 --- a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/topn/TopNOperator.java +++ b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/topn/TopNOperator.java @@ -49,7 +49,7 @@ public enum InputOrdering { * Fills {@link TopNRow}s from page data. Handles both sort-key encoding and value * extraction, and tracks pre-allocation sizes for key and value buffers. */ - public static final class RowFiller { + static final class RowFiller { private final ValueExtractor[] valueExtractors; private final KeyExtractor[] keyExtractors; diff --git a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/topn/TopNQueue.java b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/topn/TopNQueue.java index 01164fd5a61bd..8bcda7d16e2c8 100644 --- a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/topn/TopNQueue.java +++ b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/topn/TopNQueue.java @@ -21,7 +21,7 @@ * Used both by {@link TopNOperator} (one global queue) and by {@link GroupedQueue} * (one queue per group). */ -public class TopNQueue extends PriorityQueue implements Accountable, Releasable { +class TopNQueue extends PriorityQueue implements Accountable, Releasable { private static final long SHALLOW_SIZE = RamUsageEstimator.shallowSizeOfInstance(TopNQueue.class); private final CircuitBreaker breaker; diff --git a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/topn/TopNRow.java b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/topn/TopNRow.java index 8cba8642cc985..951d85145ed16 100644 --- a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/topn/TopNRow.java +++ b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/topn/TopNRow.java @@ -22,7 +22,7 @@ * A single row in a top-N operation. Stores encoded sort keys and values. * Implements {@link Comparable} and {@link #equals} comparing the sort keys. */ -public final class TopNRow implements Accountable, Comparable, Releasable { +final class TopNRow implements Accountable, Comparable, Releasable { private static final long SHALLOW_SIZE = RamUsageEstimator.shallowSizeOfInstance(TopNRow.class); private final CircuitBreaker breaker; From eb1d71a003adce96969af0f7a56e614fa5c1cd45 Mon Sep 17 00:00:00 2001 From: ncordon Date: Tue, 31 Mar 2026 14:54:51 +0200 Subject: [PATCH 12/18] Cleans up non needed method --- .../xpack/esql/heap_attack/HeapAttackTestCase.java | 8 -------- 1 file changed, 8 deletions(-) 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 67829268edb8f..1c451c88598e7 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 @@ -425,14 +425,6 @@ protected void initSingleDocIndex() throws IOException { """); } - 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")) { From e89560783cc0992075cccbf61e35572fb0b70741 Mon Sep 17 00:00:00 2001 From: ncordon Date: Tue, 31 Mar 2026 16:26:15 +0200 Subject: [PATCH 13/18] Revert "Cleans up non needed method" This reverts commit eb1d71a003adce96969af0f7a56e614fa5c1cd45. --- .../xpack/esql/heap_attack/HeapAttackTestCase.java | 8 ++++++++ 1 file changed, 8 insertions(+) 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 1c451c88598e7..67829268edb8f 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 @@ -425,6 +425,14 @@ protected void initSingleDocIndex() throws IOException { """); } + 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")) { From 48bb8e096d9c0a5b84fedd6b4e49f7d56c684a1a Mon Sep 17 00:00:00 2001 From: ncordon Date: Tue, 31 Mar 2026 16:37:23 +0200 Subject: [PATCH 14/18] Lowers the circuit breaker threshold again :__ --- .../esql/heap_attack/HeapAttackLimitByIT.java | 27 +++++++++++++++++++ 1 file changed, 27 insertions(+) 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 index 30672306c32ce..6de16f8de0d9e 100644 --- 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 @@ -18,6 +18,8 @@ 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; @@ -35,6 +37,31 @@ @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: + *
    + *
  • We need to discount size of cached pages in {@code PageCacheRecycler}. + * Those cached pages are invisible to the breaker but still consume heap.
    • + *
  • ES classpath also takes up heap space
    • + *
  • Memory fragmentation might cause the circuit breaker to think we can allocate + * an array when it's not possible to find a big enough gap in 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 // ------------------------------------------------------------------------- From 5ba10994dcb09a3a2ae70633276ca6888c5d9749 Mon Sep 17 00:00:00 2001 From: ncordon Date: Tue, 31 Mar 2026 18:46:14 +0200 Subject: [PATCH 15/18] Adjusts threshold for test --- .../xpack/esql/heap_attack/HeapAttackLimitByIT.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 index 6de16f8de0d9e..05eec01bc7027 100644 --- 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 @@ -87,7 +87,7 @@ public void testLimitByManyGroupingColumns() throws IOException { public void testLimitByManyGroupingColumnsTooMuchMemory() throws IOException { assumeTrue("LIMIT BY requires snapshot builds", Build.current().isSnapshot()); initManyLongs(10); - assertCircuitBreaksVia(attempt -> limitByManyLongs(attempt * 500), GroupedLimitOperator.class.getName(), bytesRefHashClassName()); + assertCircuitBreaksVia(attempt -> limitByManyLongs(attempt * 1000), GroupedLimitOperator.class.getName(), bytesRefHashClassName()); } private Map limitByManyLongs(int count) throws IOException { From 4a5df79be21e185d626ea7fc324c020b34cc3302 Mon Sep 17 00:00:00 2001 From: ncordon Date: Tue, 31 Mar 2026 19:40:11 +0200 Subject: [PATCH 16/18] Undoes changes to GroupedTopNOperator --- .../operator/GroupedLimitOperator.java | 2 +- .../compute/operator/topn/GroupedQueue.java | 32 ++++------- .../operator/topn/GroupedTopNOperator.java | 54 +++---------------- .../operator/topn/GroupedQueueTests.java | 4 +- 4 files changed, 19 insertions(+), 73 deletions(-) diff --git a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/GroupedLimitOperator.java b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/GroupedLimitOperator.java index 4087c81d5d0d6..ec63f969f2ec4 100644 --- a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/GroupedLimitOperator.java +++ b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/GroupedLimitOperator.java @@ -99,7 +99,7 @@ public GroupedLimitOperator(int limitPerGroup, GroupKeyEncoder keyEncoder, Block success = true; } finally { if (success == false) { - Releasables.closeExpectNoException(keyEncoder, seenKeys, counts); + Releasables.closeExpectNoException(keyEncoder, seenKeys); } } } diff --git a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/topn/GroupedQueue.java b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/topn/GroupedQueue.java index d06a2fa5725b9..5f44e948e9ce6 100644 --- a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/topn/GroupedQueue.java +++ b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/topn/GroupedQueue.java @@ -14,11 +14,9 @@ import org.elasticsearch.core.Releasable; import org.elasticsearch.core.Releasables; +import java.util.ArrayList; import java.util.List; -import static org.apache.lucene.util.RamUsageEstimator.NUM_BYTES_ARRAY_HEADER; -import static org.apache.lucene.util.RamUsageEstimator.NUM_BYTES_OBJECT_REF; -import static org.apache.lucene.util.RamUsageEstimator.alignObjectSize; import static org.apache.lucene.util.RamUsageEstimator.shallowSizeOfInstance; /** @@ -68,34 +66,22 @@ TopNQueue getOrCreateQueue(long groupId) { } /** - * Drains all rows from all per-group queues into the given list, closing each queue - * as it is emptied. The list is then sorted in descending key order so that for an - * ascending sort the first element is the minimum. + * Removes and returns all rows from all per-group queues. + * For an ascending order, the first element will be the min element (or last in the + * priority queue), and vice versa. */ - void popAllInto(List target) { + List popAll() { + List allRows = new ArrayList<>(size()); for (long i = 0; i < queues.size(); i++) { TopNQueue queue = queues.get(i); if (queue != null) { - queue.popAllInto(target); + queue.popAllInto(allRows); queue.close(); queues.set(i, null); } } - // sort allocates an internal array for sorting - long sortBytes = estimateArrayBytes(target.size()); - breaker.addEstimateBytesAndMaybeBreak(sortBytes, "grouped_queue"); - try { - target.sort((r1, r2) -> -r1.compareTo(r2)); - } finally { - breaker.addWithoutBreaking(-sortBytes); - } - } - - /** - * Estimates the bytes used by an {@code Object[]} of the given capacity. - */ - static long estimateArrayBytes(int size) { - return alignObjectSize(NUM_BYTES_ARRAY_HEADER + (long) size * NUM_BYTES_OBJECT_REF); + allRows.sort((r1, r2) -> -r1.compareTo(r2)); + return allRows; } @Override diff --git a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/topn/GroupedTopNOperator.java b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/topn/GroupedTopNOperator.java index 1c1cc6a362abb..3b13ee5172544 100644 --- a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/topn/GroupedTopNOperator.java +++ b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/topn/GroupedTopNOperator.java @@ -24,7 +24,6 @@ import org.elasticsearch.core.ReleasableIterator; import org.elasticsearch.core.Releasables; -import java.util.ArrayList; import java.util.Arrays; import java.util.List; @@ -313,53 +312,26 @@ private ReleasableIterator buildResult() { spare = null; } - var resultSize = inputQueue.size(); - if (resultSize == 0) { + if (inputQueue.size() == 0) { return ReleasableIterator.empty(); } - Result result = null; - var success = false; - try { - result = new Result(breaker, resultSize); - inputQueue.popAllInto(result.getRows()); - success = true; - } finally { - if (success == false) { - Releasables.closeExpectNoException(result, inputQueue, keysHash); - inputQueue = null; - keysHash = null; - } - } - return result; + List rows = inputQueue.popAll(); + inputQueue.close(); + keysHash.close(); + inputQueue = null; + keysHash = null; + return new Result(rows); } private class Result implements ReleasableIterator { - private static final long SHALLOW_SIZE = RamUsageEstimator.shallowSizeOfInstance(Result.class); private final List rows; - private final CircuitBreaker breaker; private int r; - Result(CircuitBreaker breaker, int resultSize) { - breaker.addEstimateBytesAndMaybeBreak(sizeOf(resultSize), "grouped topn result"); - var success = false; - List rows; - try { - rows = new ArrayList<>(resultSize); - success = true; - } finally { - if (success == false) { - this.close(); - } - } - this.breaker = breaker; + private Result(List rows) { this.rows = rows; } - public List getRows() { - return rows; - } - @Override public boolean hasNext() { return r < rows.size(); @@ -405,17 +377,7 @@ private long totalSize(ResultBuilder[] builders) { @Override public void close() { - var resultSize = rows.size(); Releasables.close(rows); - breaker.addWithoutBreaking(-sizeOf(resultSize)); - } - - static long sizeOf(int resultSize) { - long total = SHALLOW_SIZE; - total += RamUsageEstimator.alignObjectSize( - RamUsageEstimator.NUM_BYTES_ARRAY_HEADER + RamUsageEstimator.NUM_BYTES_OBJECT_REF * ((long) resultSize) - ); - return total; } private void readKeys(ResultBuilder[] builders, BytesRef keys) { diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/operator/topn/GroupedQueueTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/operator/topn/GroupedQueueTests.java index 4999bec0773a9..e08f817a9c4fc 100644 --- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/operator/topn/GroupedQueueTests.java +++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/operator/topn/GroupedQueueTests.java @@ -21,7 +21,6 @@ import org.elasticsearch.indices.breaker.CircuitBreakerService; import org.elasticsearch.test.ESTestCase; -import java.util.ArrayList; import java.util.List; import static org.hamcrest.Matchers.equalTo; @@ -101,8 +100,7 @@ private void addRow(GroupedQueue queue, int groupKey, int value) { private void assertQueueContents(GroupedQueue queue, List expectedSortKeys) { assertThat(queue.size(), equalTo(expectedSortKeys.size())); - List actual = new ArrayList<>(queue.size()); - queue.popAllInto(actual); + List actual = queue.popAll(); for (int i = 0; i < expectedSortKeys.size(); i++) { int sortKey = expectedSortKeys.get(i); assertRowValues(actual.get(i), sortKey, sortKey * 2); From a359c4809ff03f750365cb805d08e6162891b947 Mon Sep 17 00:00:00 2001 From: ncordon Date: Tue, 31 Mar 2026 21:24:36 +0200 Subject: [PATCH 17/18] Tries to stabilize tests --- .../esql/heap_attack/HeapAttackLimitByIT.java | 103 ++++++++++-------- .../esql/heap_attack/HeapAttackTestCase.java | 33 ++++++ 2 files changed, 89 insertions(+), 47 deletions(-) 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 index 05eec01bc7027..fc5554f54b010 100644 --- 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 @@ -67,39 +67,45 @@ public void resetRequestBreakerLimit() throws IOException { // ------------------------------------------------------------------------- /** - * This limits by 50 computed columns which should succeed. The GroupedLimitOperator stores group - * keys in a hash table; 100K unique groups with 55 columns each is well within the circuit breaker. + * 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 testLimitByManyGroupingColumns() throws IOException { + public void testLimitByHighCardinality() throws IOException { assumeTrue("LIMIT BY requires snapshot builds", Build.current().isSnapshot()); - initManyLongs(10); - Map result = limitByManyLongs(50); + 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. Grouping by all 5 original - * fields (a,b,c,d,e) gives 100K unique groups; adding many computed columns widens each key - * until the hash table trips the circuit breaker. + * 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 testLimitByManyGroupingColumnsTooMuchMemory() throws IOException { + public void testLimitByHighCardinalityTooMuchMemory() throws IOException { assumeTrue("LIMIT BY requires snapshot builds", Build.current().isSnapshot()); - initManyLongs(10); - assertCircuitBreaksVia(attempt -> limitByManyLongs(attempt * 1000), GroupedLimitOperator.class.getName(), bytesRefHashClassName()); + initManyLongsAndString(10, 100); + assertCircuitBreaksVia( + attempt -> limitByWideStringKey(attempt * 40), + GroupedLimitOperator.class.getName(), + bytesRefHashClassName() + ); } - private Map limitByManyLongs(int count) throws IOException { - logger.info("limit by {} group cols", count); - StringBuilder query = makeManyLongs(count); - // Group by all 5 original fields so there are 10^5 = 100K unique groups, - // then also include the computed fields to widen each group key. - query.append("| LIMIT 1 BY a, b, c, d, e, i0"); - for (int i = 1; i < count; i++) { - query.append(", i").append(i); + 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("\\n| STATS MAX(a)\"}"); + query.append("| STATS MAX(a)\"}"); return responseAsMap(query(query.toString(), null)); } @@ -239,43 +245,46 @@ private Map topNByManyStringSortCols(int count) throws IOExcepti } /** - * This sorts by 1 column then limits 1 row per group across 100K unique groups with 55 group - * key computed columns, which should succeed. The GroupedTopNOperator stores group - * keys in a hash table. 100K groups with 55-column keys is within the circuit breaker. + * 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 testTopNByManyGroupingColumns() throws IOException { + public void testTopNByHighCardinality() throws IOException { assumeTrue("SORT | LIMIT BY requires snapshot builds", Build.current().isSnapshot()); - initManyLongs(10); - Map result = topNByWideGroupKey(50); - 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)); + 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); } /** - * The GroupedTopNOperator stores group keys in a hash table ({@code keysHash}). Grouping by - * all 5 original fields plus many computed columns widens each encoded group key until the - * hash table trips the circuit breaker. Unlike {@link #testTopNByManySortColumns} which stresses - * sort key row storage, this stresses the group key hash table. + * 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 testTopNByManyGroupingColumnsTooMuchMemory() throws IOException { + public void testTopNByHighCardinalityTooMuchMemory() throws IOException { assumeTrue("SORT | LIMIT BY requires snapshot builds", Build.current().isSnapshot()); - initManyLongs(10); - assertCircuitBreaksVia(attempt -> topNByWideGroupKey(attempt * 1000), GroupedTopNOperator.class.getName(), bytesRefHashClassName()); + initManyLongsAndString(10, 100); + assertCircuitBreaksVia( + attempt -> topNByWideStringKey(attempt * 40), + GroupedTopNOperator.class.getName(), + bytesRefHashClassName() + ); } - private Map topNByWideGroupKey(int count) throws IOException { - logger.info("topn by with {} group key cols", count); - StringBuilder query = makeManyLongs(count); - // Sort by a single column to keep TopNRow narrow; group by all 5 original fields - // plus computed columns to widen the composite group key stored in keysHash. - // 100K unique groups (a,b,c,d,e ∈ [0,9]) × wide encoded keys trips the circuit breaker. - query.append("| SORT a\\n"); - query.append("| LIMIT 1 BY a, b, c, d, e, i0"); - for (int i = 1; i < count; i++) { - query.append(", i").append(i); + 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("\\n| KEEP a, b\"}"); + query.append("| STATS MAX(a)\"}"); return responseAsMap(query(query.toString(), null)); } 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 67829268edb8f..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 @@ -404,6 +404,39 @@ protected void initManyLongs(int countPerLong) throws IOException { 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}. From 137963c2ea0cdb3e0c224222a65eac208568c517 Mon Sep 17 00:00:00 2001 From: elasticsearchmachine Date: Tue, 31 Mar 2026 19:33:02 +0000 Subject: [PATCH 18/18] [CI] Auto commit changes from spotless --- .../xpack/esql/heap_attack/HeapAttackLimitByIT.java | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) 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 index fc5554f54b010..c3d2a190656a9 100644 --- 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 @@ -267,11 +267,7 @@ public void testTopNByHighCardinality() throws IOException { 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() - ); + assertCircuitBreaksVia(attempt -> topNByWideStringKey(attempt * 40), GroupedTopNOperator.class.getName(), bytesRefHashClassName()); } private Map topNByWideStringKey(int repeatScale) throws IOException {