Fix data race in InlineStats (#131370)

Although blocks/vectors are immutable and safe to share between threads, 
their references are currently not thread-safe, which can lead to data 
races. Previously, blocks/vectors were exclusively owned by a single
thread, but this is no longer always the case with InlineJoin. We should
consider switching to AbstractRefCounted, which is thread-safe, and
benchmark it with many-fields use cases to ensure there is no
performance regression. As a temporary solution, this change clones the
values block in InlineJoin until thread-safe blocks/vectors are
available.

Author: dnhatn

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

@@ -11,6 +11,7 @@
 import org.elasticsearch.compute.data.Block;
 import org.elasticsearch.compute.data.Page;
 import org.elasticsearch.core.ReleasableIterator;
+import org.elasticsearch.core.Releasables;
 
 /**
  * {@link Block#lookup Looks up} values from a provided {@link Block} and
@@ -44,8 +45,19 @@ public String describe() {
     private final int positionsOrd;
 
     public ColumnLoadOperator(Values values, int positionsOrd) {
-        this.values = values;
         this.positionsOrd = positionsOrd;
+        this.values = clone(values);
+    }
+
+    // FIXME: Since we don't have a thread-safe RefCounted for blocks/vectors, we have to clone the values block to avoid
+    // data races of reference when sharing blocks/vectors across threads. Remove this when we have a thread-safe RefCounted
+    // for blocks/vectors.
+    static Values clone(Values values) {
+        final Block block = values.block;
+        try (var builder = block.elementType().newBlockBuilder(block.getPositionCount(), block.blockFactory())) {
+            builder.copyFrom(block, 0, block.getPositionCount());
+            return new Values(values.name, builder.build());
+        }
     }
 
     /**
@@ -67,6 +79,11 @@ protected ReleasableIterator<Page> receive(Page page) {
         return appendBlocks(page, values.block.lookup(page.getBlock(positionsOrd), TARGET_BLOCK_SIZE));
     }
 
+    @Override
+    public void close() {
+        Releasables.closeExpectNoException(values.block, super::close);
+    }
+
     @Override
     public String toString() {
         return "ColumnLoad[values=" + values + ", positions=" + positionsOrd + "]";