Skip to content

Aggs: Fix CB on reduction phase #133398

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 31 commits into from
Sep 12, 2025
Merged

Aggs: Fix CB on reduction phase #133398

Changes from 2 commits
Commits
Show all changes
31 commits
Select commit Hold shift + click to select a range
86f2d88
Fix wrong reduce memory estimate calculation
ivancea Aug 22, 2025
e798aa0
Add batched merge results size estimations to CB
ivancea Aug 22, 2025
7b3d88d
Merge branch 'main' into aggs-composite-memory
ivancea Aug 25, 2025
87ccf54
Merge branch 'main' into aggs-composite-memory
ivancea Aug 25, 2025
6fc8fa8
Merge branch 'main' into aggs-composite-memory
ivancea Aug 27, 2025
51fe8fc
Merge branch 'main' into aggs-composite-memory
ivancea Aug 27, 2025
41b86a0
Updated javadocs on estimates and replaced 1.5-size with 0.5
ivancea Aug 27, 2025
eb46c13
Merge branch 'main' into aggs-composite-memory
ivancea Aug 28, 2025
8270935
Initial CB tests for batched aggs
ivancea Aug 29, 2025
ba3748b
Merge branch 'main' into aggs-composite-memory
ivancea Aug 29, 2025
2140240
Merge branch 'main' into aggs-composite-memory
ivancea Sep 1, 2025
d2fed37
Randomize tests
ivancea Sep 1, 2025
449203f
Add estimate for field merge result
ivancea Sep 1, 2025
7c7d3eb
Merge branch 'main' into aggs-composite-memory
ivancea Sep 3, 2025
7402133
Merge branch 'main' into aggs-composite-memory
ivancea Sep 8, 2025
d344228
Added test cluster docs
ivancea Sep 9, 2025
f9b8379
Merge branch 'main' into aggs-composite-memory
ivancea Sep 9, 2025
0a89991
Added Integration test for the reduce phase CB
ivancea Sep 9, 2025
5f8a5cf
Add nullable
ivancea Sep 9, 2025
e8fda32
[CI] Auto commit changes from spotless
Sep 9, 2025
e173654
Fixed memory leak and added Repeat to test
ivancea Sep 10, 2025
3d0f864
Removed comment
ivancea Sep 10, 2025
c9101bb
Fixed comment
ivancea Sep 10, 2025
ac0929b
Decrement ref on search response in test
ivancea Sep 10, 2025
77b152b
Simplify test and add a random number of nodes to make it fail in bot…
ivancea Sep 10, 2025
e6d7beb
Supress repeat forbidden api error
ivancea Sep 10, 2025
09aff51
Merge branch 'main' into aggs-composite-memory
ivancea Sep 10, 2025
8e4fd41
Update docs/changelog/133398.yaml
ivancea Sep 10, 2025
a24eba1
Integrate test repetition in test case and remove unused setting
ivancea Sep 11, 2025
1581fc4
Merge branch 'main' into aggs-composite-memory
ivancea Sep 12, 2025
f10f1a5
Merge branch 'main' into aggs-composite-memory
ivancea Sep 12, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 3 additions & 2 deletions server/src/main/java/org/elasticsearch/action/search/QueryPhaseResultConsumer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -231,6 +231,7 @@ public SearchPhaseController.ReducedQueryPhase reduce() throws Exception {
while ((batchedResult = batchedResults.poll()) != null) {
topDocsStats.add(batchedResult.v1());
consumePartialMergeResult(batchedResult.v2(), topDocsList, aggsList);
addEstimateAndMaybeBreak(batchedResult.v2().estimatedSize);
Copy link
Contributor Author

@ivancea ivancea Aug 22, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

That MergeResult estimation was ignored and lost. I followed it since the deserialization, and it wasn't used anywhere else

Copy link
Contributor

@iverase iverase Sep 9, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I am wondering if we should add it to the breaker before consuming? or we only know the estimated size after consuming.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

That consumePartialMergeResult() simply adds the agg to the aggsList param, nothing else. They're deserialized later here:

elasticsearch/server/src/main/java/org/elasticsearch/action/search/QueryPhaseResultConsumer.java

Line 250 in 5a3c9e7

aggs = aggregate(buffer.iterator(), new Iterator<>() {

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ok, then it does not matter 👍

}
for (QuerySearchResult result : buffer) {
topDocsStats.add(result.topDocs(), result.searchTimedOut(), result.terminatedEarly());
Expand Down Expand Up @@ -392,7 +393,7 @@ private MergeResult partialReduce(
return new MergeResult(
processedShards,
newTopDocs,
newAggs == null ? null : DelayableWriteable.referencing(newAggs),
newAggs != null ? DelayableWriteable.referencing(newAggs) : null,
newAggs != null ? DelayableWriteable.getSerializedSize(newAggs) : 0
);
}
Expand Down Expand Up @@ -462,7 +463,7 @@ private long ramBytesUsedQueryResult(QuerySearchResult result) {
* the reduce completes.
*/
private static long estimateRamBytesUsedForReduce(long size) {
return Math.round(1.5d * size - size);
return Math.round(1.5d * size);
Copy link
Contributor Author

@ivancea ivancea Aug 22, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Check the javadoc for better understanding of this method. It's being used here:

elasticsearch/server/src/main/java/org/elasticsearch/action/search/QueryPhaseResultConsumer.java

Lines 245 to 264 in e798aa0

long breakerSize = circuitBreakerBytes;
final InternalAggregations aggs;
try {
if (aggsList != null) {
// Add an estimate of the final reduce size
breakerSize = addEstimateAndMaybeBreak(estimateRamBytesUsedForReduce(breakerSize));
aggs = aggregate(buffer.iterator(), new Iterator<>() {
@Override
public boolean hasNext() {
return aggsList.isEmpty() == false;
}
@Override
public DelayableWriteable<InternalAggregations> next() {
return aggsList.pollFirst();
}
},
resultSize,
performFinalReduce ? aggReduceContextBuilder.forFinalReduction() : aggReduceContextBuilder.forPartialReduction()
);

(circuitBreakerBytes is the total bytes added to the CB in the consumer; the batched results will be here)

It's interesting though that if we added the estimated MergeResult size before, we're adding it once again. I'll have to ensure the 1.5 thing wasn't actually "fixing" this problem here. It affects other cases however, so it makes things more complicated

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Update: I'll review this again, as maybe it was considering the original estimated size to be already accounted, so the 0.5 extra would be right.
I'll check the codepaths leading to this to see what's happening exactly, and maybe undo this or change the javadoc if required

}

private void consume(QuerySearchResult result, Runnable next) {
Expand Down