ESQL: TopNOperator, release Row on failure (#130330)

Handles the case where the Row was released on failure, by moving the declaration to a try-with-resource clause.

Resolves #130215, #130222, #130270.

Author: GalLalouche

muted-tests.yml

@@ -560,12 +560,38 @@ tests:
 - class: org.elasticsearch.xpack.esql.qa.single_node.GenerativeIT
   method: test
   issue: https://github.com/elastic/elasticsearch/issues/130067
-- class: org.elasticsearch.compute.operator.topn.TopNOperatorTests
-  method: testSimpleWithCranky
-  issue: https://github.com/elastic/elasticsearch/issues/130215
-- class: org.elasticsearch.xpack.esql.action.EnrichIT
-  method: testProfile
-  issue: https://github.com/elastic/elasticsearch/issues/130270
+- class: org.elasticsearch.action.support.ThreadedActionListenerTests
+  method: testRejectionHandling
+  issue: https://github.com/elastic/elasticsearch/issues/130129
+- class: org.elasticsearch.compute.aggregation.TopIntAggregatorFunctionTests
+  method: testManyInitialManyPartialFinalRunnerThrowing
+  issue: https://github.com/elastic/elasticsearch/issues/130145
+- class: org.elasticsearch.xpack.logsdb.patternedtext.PatternedTextFieldMapperTests
+  issue: https://github.com/elastic/elasticsearch/issues/130162
+- class: org.elasticsearch.index.engine.ThreadPoolMergeExecutorServiceDiskSpaceTests
+  method: testUnavailableBudgetBlocksNewMergeTasksFromStartingExecution
+  issue: https://github.com/elastic/elasticsearch/issues/130205
+- class: org.elasticsearch.index.codec.vectors.cluster.KMeansLocalTests
+  method: testKMeansNeighbors
+  issue: https://github.com/elastic/elasticsearch/issues/130258
+- class: org.elasticsearch.xpack.test.rest.XPackRestIT
+  method: test {p0=esql/10_basic/basic with documents_found}
+  issue: https://github.com/elastic/elasticsearch/issues/130256
+- class: org.elasticsearch.xpack.searchablesnapshots.cache.shared.NodesCachesStatsIntegTests
+  method: testNodesCachesStats
+  issue: https://github.com/elastic/elasticsearch/issues/129863
+- class: org.elasticsearch.index.IndexingPressureIT
+  method: testWriteCanRejectOnPrimaryBasedOnMaxOperationSize
+  issue: https://github.com/elastic/elasticsearch/issues/130281
+- class: org.elasticsearch.xpack.esql.qa.multi_node.EsqlSpecIT
+  method: test {lookup-join.MvJoinKeyOnFrom SYNC}
+  issue: https://github.com/elastic/elasticsearch/issues/130296
+- class: org.elasticsearch.xpack.esql.inference.bulk.BulkInferenceExecutorTests
+  method: testSuccessfulExecution
+  issue: https://github.com/elastic/elasticsearch/issues/130306
+- class: org.elasticsearch.multiproject.test.CoreWithMultipleProjectsClientYamlTestSuiteIT
+  method: test {yaml=cluster.stats/10_basic/Dense vector stats}
+  issue: https://github.com/elastic/elasticsearch/issues/130307
 
 # Examples:
 #

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

@@ -469,51 +469,50 @@ private Iterator<Page> toPages() {
                     p = 0;
                 }
 
-                Row row = list.get(i);
-                BytesRef keys = row.keys.bytesRefView();
-                for (SortOrder so : sortOrders) {
-                    if (keys.bytes[keys.offset] == so.nul()) {
+                try (Row row = list.get(i)) {
+                    BytesRef keys = row.keys.bytesRefView();
+                    for (SortOrder so : sortOrders) {
+                        if (keys.bytes[keys.offset] == so.nul()) {
+                            keys.offset++;
+                            keys.length--;
+                            continue;
+                        }
                         keys.offset++;
                         keys.length--;
-                        continue;
+                        builders[so.channel].decodeKey(keys);
+                    }
+                    if (keys.length != 0) {
+                        throw new IllegalArgumentException("didn't read all keys");
                     }
-                    keys.offset++;
-                    keys.length--;
-                    builders[so.channel].decodeKey(keys);
-                }
-                if (keys.length != 0) {
-                    throw new IllegalArgumentException("didn't read all keys");
-                }
-
-                BytesRef values = row.values.bytesRefView();
-                for (ResultBuilder builder : builders) {
-                    builder.setNextRefCounted(row.shardRefCounter);
-                    builder.decodeValue(values);
-                }
-                if (values.length != 0) {
-                    throw new IllegalArgumentException("didn't read all values");
-                }
 
-                list.set(i, null);
+                    BytesRef values = row.values.bytesRefView();
+                    for (ResultBuilder builder : builders) {
+                        builder.setNextRefCounted(row.shardRefCounter);
+                        builder.decodeValue(values);
+                    }
+                    if (values.length != 0) {
+                        throw new IllegalArgumentException("didn't read all values");
+                    }
 
-                p++;
-                if (p == size) {
-                    Block[] blocks = new Block[builders.length];
-                    try {
-                        for (int b = 0; b < blocks.length; b++) {
-                            blocks[b] = builders[b].build();
-                        }
-                    } finally {
-                        if (blocks[blocks.length - 1] == null) {
-                            Releasables.closeExpectNoException(blocks);
+                    list.set(i, null);
+
+                    p++;
+                    if (p == size) {
+                        Block[] blocks = new Block[builders.length];
+                        try {
+                            for (int b = 0; b < blocks.length; b++) {
+                                blocks[b] = builders[b].build();
+                            }
+                        } finally {
+                            if (blocks[blocks.length - 1] == null) {
+                                Releasables.closeExpectNoException(blocks);
+                            }
                         }
+                        result.add(new Page(blocks));
+                        Releasables.closeExpectNoException(builders);
+                        builders = null;
                     }
-                    result.add(new Page(blocks));
-                    Releasables.closeExpectNoException(builders);
-                    builders = null;
                 }
-                // It's important to close the row only after we build the new block, so we don't pre-release any shard counter.
-                row.close();
             }
             assert builders == null;
             success = true;