Aggs: Fix CB on reduction phase

[ivancea]

Fix an issue with aggs reduction phase where MergeResult estimations were ignored on batched results, leading to no CB changes before serialized aggs expansion

Notes

This comes from an OOM investigation on aggs. Investigating the CB, checking the estimation code, I noticed the coordinator QueryPhaseResultConsumer circuitBreakerBytes was at 0 before the actual reduction, apparently from those batched MergeResults being ignored.
Adding them to the accounting here, which automatically applies them a 1.5x later for the serialized->deserialized memory difference

[ivancea]

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

[iverase]

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

[ivancea]

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<>() {

[iverase]

ok, then it does not matter 👍

[ivancea]

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

[ivancea]

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

[nik9000]

This seems right. But it's tricky! I'd give it a sleep and another look before merging.

[elasticsearchmachine]

Hi @ivancea, I've created a changelog YAML for you.

[elasticsearchmachine]

💔 Backport failed

Status	Branch	Result
❌	8.19	Commit could not be cherrypicked due to conflicts
✅	9.1
❌	8.18	Commit could not be cherrypicked due to conflicts
❌	9.0	Commit could not be cherrypicked due to conflicts

You can use sqren/backport to manually backport by running backport --upstream elastic/elasticsearch --pr 133398

[ivancea]

Removing backports for 9.0 and before, as this change was made in 9.1 in #121885