ESQL: Fix a breaker bug (#136105)

Fixes a case where we release memory we never acquired, which breaker the circuit breaker accounting. Closes #135224 Closes #135260

Author: nik9000

docs/changelog/136105.yaml

@@ -0,0 +1,7 @@
+pr: 136105
+summary: Fix a breaker bug
+area: ES|QL
+type: bug
+issues:
+ - 135224
+ - 135260

muted-tests.yml

@@ -525,12 +525,6 @@ tests:
 - class: org.elasticsearch.xpack.test.rest.XPackRestIT
  method: test {p0=analytics/nested_top_metrics_sort/terms order by top metrics numeric not null integer values}
  issue: https://github.com/elastic/elasticsearch/issues/135162
-- class: org.elasticsearch.compute.operator.topn.TopNOperatorTests
- method: testSimpleWithCranky
- issue: https://github.com/elastic/elasticsearch/issues/135224
-- class: org.elasticsearch.compute.operator.topn.TopNOperatorTests
- method: testSimpleCircuitBreaking
- issue: https://github.com/elastic/elasticsearch/issues/135260
 - class: org.elasticsearch.xpack.esql.qa.single_node.PushQueriesIT
  method: testEqualityAndOther {SEMANTIC_TEXT_WITH_KEYWORD}
  issue: https://github.com/elastic/elasticsearch/issues/135333

x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/topn/TopNOperator.java

@@ -84,10 +84,10 @@ static final class Row implements Accountable, Releasable {
  RefCounted shardRefCounter;
 
  Row(CircuitBreaker breaker, List<SortOrder> sortOrders, int preAllocatedKeysSize, int preAllocatedValueSize) {
+ breaker.addEstimateBytesAndMaybeBreak(SHALLOW_SIZE, "topn");
  this.breaker = breaker;
  boolean success = false;
  try {
- breaker.addEstimateBytesAndMaybeBreak(SHALLOW_SIZE, "topn");
  keys = new BreakingBytesRefBuilder(breaker, "topn", preAllocatedKeysSize);
  values = new BreakingBytesRefBuilder(breaker, "topn", preAllocatedValueSize);
  bytesOrder = new BytesOrder(sortOrders, breaker, "topn");
@@ -684,8 +684,12 @@ public long ramBytesUsed() {
 
  @Override
  public void close() {
- Releasables.close(Releasables.wrap(this), () -> breaker.addWithoutBreaking(-Queue.sizeOf(topCount)));
-
+ Releasables.close(
+ // Release all entries in the topn
+ Releasables.wrap(this),
+ // Release the array itself
+ () -> breaker.addWithoutBreaking(-Queue.sizeOf(topCount))
+ );
  }
 
  public static long sizeOf(int topCount) {