ESQL: Fix a breaker bug (elastic#136105) (elastic#136293)

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

Closes elastic#135224
Closes elastic#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

@@ -546,12 +546,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");
@@ -679,8 +679,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) {