Fix data race with AbstractPageMappingToIteratorOperator (elastic#130963)

We need to release the blocks of the page in 
AbstractPageMappingToIteratorOperator immediately in single-iteration
cases, instead of delaying to the next iteration. This is because the
blocks of the page are now shared with the output page. The output page
can be passed to a separate driver, which may run concurrently with this
driver, leading to data races in AbstractNonThreadSafeRefCounted, which
is not thread-safe.

An alternative would be to make RefCounted for Vectors/Blocks 
thread-safe when they are about to be shared with other drivers via
#allowPassingToDifferentDriver.

Relates elastic#130573

Closes elastic#130959
Closes elastic#130958
Closes elastic#130950
Closes elastic#130925
Closes elastic#130881
Closes elastic#130796

Author: dnhatn

muted-tests.yml

@@ -555,9 +555,6 @@ tests:
 - class: org.elasticsearch.xpack.esql.action.EsqlRemoteErrorWrapIT
   method: testThatRemoteErrorsAreWrapped
   issue: https://github.com/elastic/elasticsearch/issues/130794
-- class: org.elasticsearch.xpack.esql.qa.multi_node.EsqlSpecIT
-  method: test {ints.InCast SYNC}
-  issue: https://github.com/elastic/elasticsearch/issues/130796
 - class: org.elasticsearch.backwards.MixedClusterClientYamlTestSuiteIT
   method: test {p0=mtermvectors/10_basic/Tests catching other exceptions per item}
   issue: https://github.com/elastic/elasticsearch/issues/122414

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

@@ -321,6 +321,7 @@ public TransportVersion getMinimalSupportedVersion() {
     private static class AppendBlocksIterator implements ReleasableIterator<Page> {
         private final Page page;
         private final ReleasableIterator<Block[]> next;
+        private boolean closed = false;
 
         private int positionOffset;
 
@@ -348,7 +349,15 @@ public final Page next() {
                 for (int b = 0; b < page.getBlockCount(); b++) {
                     page.getBlock(b).incRef();
                 }
-                return page.appendBlocks(read);
+                final Page result = page.appendBlocks(read);
+                // We need to release the blocks of the page in this iteration instead of delaying to the next,
+                // because the blocks of this page are now shared with the output page. The output page can be
+                // passed to a separate driver, which may run concurrently with this driver, leading to data races
+                // of references in AbstractNonThreadSafeRefCounted, which is not thread-safe.
+                // An alternative would be to make RefCounted for Vectors/Blocks thread-safe when they are about
+                // to be shared with other drivers via #allowPassingToDifferentDriver.
+                close();
+                return result;
             }
             Block[] newBlocks = new Block[page.getBlockCount() + read.length];
             System.arraycopy(read, 0, newBlocks, page.getBlockCount(), read.length);
@@ -368,7 +377,10 @@ public final Page next() {
 
         @Override
         public void close() {
-            Releasables.closeExpectNoException(page::releaseBlocks, next);
+            if (closed == false) {
+                closed = true;
+                Releasables.closeExpectNoException(page::releaseBlocks, next);
+            }
         }
     }
 }