ESQL: Array blocks backed by vecs (#103623)

Use vectors consistently for storing blocks' values.

This allows for simple implementations of expand and filter by simply sharing the (refcounted) vector contained in a block.

benchmarks/src/main/java/org/elasticsearch/benchmark/compute/operator/BlockBenchmark.java

@@ -76,6 +76,7 @@ public class BlockBenchmark {
     public static final String[] RELEVANT_TYPE_BLOCK_COMBINATIONS = {
         "boolean/array",
         "boolean/array-multivalue-null",
+        "boolean/big-array",
         "boolean/big-array-multivalue-null",
         "boolean/vector",
         "boolean/vector-big-array",
@@ -86,18 +87,21 @@ public class BlockBenchmark {
         "BytesRef/vector-const",
         "double/array",
         "double/array-multivalue-null",
+        "double/big-array",
         "double/big-array-multivalue-null",
         "double/vector",
         "double/vector-big-array",
         "double/vector-const",
         "int/array",
         "int/array-multivalue-null",
+        "int/big-array",
         "int/big-array-multivalue-null",
         "int/vector",
         "int/vector-big-array",
         "int/vector-const",
         "long/array",
         "long/array-multivalue-null",
+        "long/big-array",
         "long/big-array-multivalue-null",
         "long/vector",
         "long/vector-big-array",
@@ -177,6 +181,23 @@ public class BlockBenchmark {
                                 Block.MvOrdering.UNORDERED
                             );
                         }
+                        case "big-array" -> {
+                            BitArray valuesBigArray = new BitArray(totalPositions, BigArrays.NON_RECYCLING_INSTANCE);
+                            for (int i = 0; i < values.length; i++) {
+                                if (values[i]) {
+                                    valuesBigArray.set(i);
+                                }
+                            }
+
+                            blocks[blockIndex] = new BooleanBigArrayBlock(
+                                valuesBigArray,
+                                totalPositions,
+                                null,
+                                null,
+                                Block.MvOrdering.DEDUPLICATED_AND_SORTED_ASCENDING,
+                                blockFactory
+                            );
+                        }
                         case "big-array-multivalue-null" -> {
                             int[] firstValueIndexes = randomFirstValueIndexes(totalPositions);
                             int positionCount = firstValueIndexes.length - 1;
@@ -315,6 +336,21 @@ public class BlockBenchmark {
                                 Block.MvOrdering.UNORDERED
                             );
                         }
+                        case "big-array" -> {
+                            DoubleArray valuesBigArray = blockFactory.bigArrays().newDoubleArray(totalPositions, false);
+                            for (int i = 0; i < values.length; i++) {
+                                valuesBigArray.set(i, values[i]);
+                            }
+
+                            blocks[blockIndex] = new DoubleBigArrayBlock(
+                                valuesBigArray,
+                                totalPositions,
+                                null,
+                                null,
+                                Block.MvOrdering.DEDUPLICATED_AND_SORTED_ASCENDING,
+                                blockFactory
+                            );
+                        }
                         case "big-array-multivalue-null" -> {
                             int[] firstValueIndexes = randomFirstValueIndexes(totalPositions);
                             int positionCount = firstValueIndexes.length - 1;
@@ -392,6 +428,21 @@ public class BlockBenchmark {
                                 Block.MvOrdering.UNORDERED
                             );
                         }
+                        case "big-array" -> {
+                            IntArray valuesBigArray = blockFactory.bigArrays().newIntArray(totalPositions, false);
+                            for (int i = 0; i < values.length; i++) {
+                                valuesBigArray.set(i, values[i]);
+                            }
+
+                            blocks[blockIndex] = new IntBigArrayBlock(
+                                valuesBigArray,
+                                totalPositions,
+                                null,
+                                null,
+                                Block.MvOrdering.DEDUPLICATED_AND_SORTED_ASCENDING,
+                                blockFactory
+                            );
+                        }
                         case "big-array-multivalue-null" -> {
                             int[] firstValueIndexes = randomFirstValueIndexes(totalPositions);
                             int positionCount = firstValueIndexes.length - 1;
@@ -469,6 +520,21 @@ public class BlockBenchmark {
                                 Block.MvOrdering.UNORDERED
                             );
                         }
+                        case "big-array" -> {
+                            LongArray valuesBigArray = blockFactory.bigArrays().newLongArray(totalPositions, false);
+                            for (int i = 0; i < values.length; i++) {
+                                valuesBigArray.set(i, values[i]);
+                            }
+
+                            blocks[blockIndex] = new LongBigArrayBlock(
+                                valuesBigArray,
+                                totalPositions,
+                                null,
+                                null,
+                                Block.MvOrdering.DEDUPLICATED_AND_SORTED_ASCENDING,
+                                blockFactory
+                            );
+                        }
                         case "big-array-multivalue-null" -> {
                             int[] firstValueIndexes = randomFirstValueIndexes(totalPositions);
                             int positionCount = firstValueIndexes.length - 1;
@@ -715,6 +781,7 @@ public class BlockBenchmark {
         {
             "boolean/array",
             "boolean/array-multivalue-null",
+            "boolean/big-array",
             "boolean/big-array-multivalue-null",
             "boolean/vector",
             "boolean/vector-big-array",
@@ -725,18 +792,21 @@ public class BlockBenchmark {
             "BytesRef/vector-const",
             "double/array",
             "double/array-multivalue-null",
+            "double/big-array",
             "double/big-array-multivalue-null",
             "double/vector",
             "double/vector-big-array",
             "double/vector-const",
             "int/array",
             "int/array-multivalue-null",
+            "int/big-array",
             "int/big-array-multivalue-null",
             "int/vector",
             "int/vector-big-array",
             "int/vector-const",
             "long/array",
             "long/array-multivalue-null",
+            "long/big-array",
             "long/big-array-multivalue-null",
             "long/vector",
             "long/vector-big-array",

server/src/main/java/org/elasticsearch/common/util/BitArray.java

@@ -129,6 +129,10 @@ public final class BitArray implements Accountable, Releasable {
         return (bits.get(wordNum) & bitmask) != 0;
     }
 
+    public long size() {
+        return bits.size() * (long) Long.BYTES * Byte.SIZE;
+    }
+
     private static long wordNum(long index) {
         return index >> 6;
     }

x-pack/plugin/esql/compute/src/main/generated-src/org/elasticsearch/compute/data/BooleanArrayBlock.java

@@ -8,19 +8,19 @@
 package org.elasticsearch.compute.data;
 
 import org.apache.lucene.util.RamUsageEstimator;
+import org.elasticsearch.core.Releasables;
 
-import java.util.Arrays;
 import java.util.BitSet;
 
 /**
- * Block implementation that stores an array of boolean.
+ * Block implementation that stores values in a {@link BooleanArrayVector}.
  * This class is generated. Do not edit it.
  */
 final class BooleanArrayBlock extends AbstractArrayBlock implements BooleanBlock {
 
     private static final long BASE_RAM_BYTES_USED = RamUsageEstimator.shallowSizeOfInstance(BooleanArrayBlock.class);
 
-    private final boolean[] values;
+    private final BooleanArrayVector vector;
 
     BooleanArrayBlock(
         boolean[] values,
@@ -31,7 +31,7 @@ final class BooleanArrayBlock extends AbstractArrayBlock implements BooleanBlock
         BlockFactory blockFactory
     ) {
         super(positionCount, firstValueIndexes, nulls, mvOrdering, blockFactory);
-        this.values = values;
+        this.vector = new BooleanArrayVector(values, values.length, blockFactory);
     }
 
     @Override
@@ -41,7 +41,7 @@ final class BooleanArrayBlock extends AbstractArrayBlock implements BooleanBlock
 
     @Override
     public boolean getBoolean(int valueIndex) {
-        return values[valueIndex];
+        return vector.getBoolean(valueIndex);
     }
 
     @Override
@@ -79,7 +79,7 @@ final class BooleanArrayBlock extends AbstractArrayBlock implements BooleanBlock
             incRef();
             return this;
         }
-        // TODO use reference counting to share the values
+        // TODO use reference counting to share the vector
         try (var builder = blockFactory().newBooleanBlockBuilder(firstValueIndexes[getPositionCount()])) {
             for (int pos = 0; pos < getPositionCount(); pos++) {
                 if (isNull(pos)) {
@@ -96,14 +96,13 @@ final class BooleanArrayBlock extends AbstractArrayBlock implements BooleanBlock
         }
     }
 
-    public static long ramBytesEstimated(boolean[] values, int[] firstValueIndexes, BitSet nullsMask) {
-        return BASE_RAM_BYTES_USED + RamUsageEstimator.sizeOf(values) + BlockRamUsageEstimator.sizeOf(firstValueIndexes)
-            + BlockRamUsageEstimator.sizeOfBitSet(nullsMask);
+    private long ramBytesUsedOnlyBlock() {
+        return BASE_RAM_BYTES_USED + BlockRamUsageEstimator.sizeOf(firstValueIndexes) + BlockRamUsageEstimator.sizeOfBitSet(nullsMask);
     }
 
     @Override
     public long ramBytesUsed() {
-        return ramBytesEstimated(values, firstValueIndexes, nullsMask);
+        return ramBytesUsedOnlyBlock() + vector.ramBytesUsed();
     }
 
     @Override
@@ -126,13 +125,20 @@ final class BooleanArrayBlock extends AbstractArrayBlock implements BooleanBlock
             + getPositionCount()
             + ", mvOrdering="
             + mvOrdering()
-            + ", values="
-            + Arrays.toString(values)
+            + ", vector="
+            + vector
             + ']';
     }
 
+    @Override
+    public void allowPassingToDifferentDriver() {
+        super.allowPassingToDifferentDriver();
+        vector.allowPassingToDifferentDriver();
+    }
+
     @Override
     public void closeInternal() {
-        blockFactory().adjustBreaker(-ramBytesUsed(), true);
+        blockFactory().adjustBreaker(-ramBytesUsedOnlyBlock(), true);
+        Releasables.closeExpectNoException(vector);
     }
 }

x-pack/plugin/esql/compute/src/main/generated-src/org/elasticsearch/compute/data/BooleanBigArrayBlock.java

@@ -14,13 +14,14 @@ import org.elasticsearch.core.Releasables;
 import java.util.BitSet;
 
 /**
- * Block implementation that stores values in a BooleanArray.
+ * Block implementation that stores values in a {@link BooleanBigArrayVector}. Does not take ownership of the given
+ * {@link BitArray} and does not adjust circuit breakers to account for it.
  * This class is generated. Do not edit it.
  */
 public final class BooleanBigArrayBlock extends AbstractArrayBlock implements BooleanBlock {
 
     private static final long BASE_RAM_BYTES_USED = 0; // TODO: fix this
-    private final BitArray values;
+    private final BooleanBigArrayVector vector;
 
     public BooleanBigArrayBlock(
         BitArray values,
@@ -31,7 +32,7 @@ public final class BooleanBigArrayBlock extends AbstractArrayBlock implements Bo
         BlockFactory blockFactory
     ) {
         super(positionCount, firstValueIndexes, nulls, mvOrdering, blockFactory);
-        this.values = values;
+        this.vector = new BooleanBigArrayVector(values, (int) values.size(), blockFactory);
     }
 
     @Override
@@ -41,7 +42,7 @@ public final class BooleanBigArrayBlock extends AbstractArrayBlock implements Bo
 
     @Override
     public boolean getBoolean(int valueIndex) {
-        return values.get(valueIndex);
+        return vector.getBoolean(valueIndex);
     }
 
     @Override
@@ -79,7 +80,7 @@ public final class BooleanBigArrayBlock extends AbstractArrayBlock implements Bo
             incRef();
             return this;
         }
-        // TODO use reference counting to share the values
+        // TODO use reference counting to share the vector
         try (var builder = blockFactory().newBooleanBlockBuilder(firstValueIndexes[getPositionCount()])) {
             for (int pos = 0; pos < getPositionCount(); pos++) {
                 if (isNull(pos)) {
@@ -96,10 +97,13 @@ public final class BooleanBigArrayBlock extends AbstractArrayBlock implements Bo
         }
     }
 
+    private long ramBytesUsedOnlyBlock() {
+        return BASE_RAM_BYTES_USED + BlockRamUsageEstimator.sizeOf(firstValueIndexes) + BlockRamUsageEstimator.sizeOfBitSet(nullsMask);
+    }
+
     @Override
     public long ramBytesUsed() {
-        return BASE_RAM_BYTES_USED + RamUsageEstimator.sizeOf(values) + BlockRamUsageEstimator.sizeOf(firstValueIndexes)
-            + BlockRamUsageEstimator.sizeOfBitSet(nullsMask);
+        return ramBytesUsedOnlyBlock() + RamUsageEstimator.sizeOf(vector);
     }
 
     @Override
@@ -123,13 +127,19 @@ public final class BooleanBigArrayBlock extends AbstractArrayBlock implements Bo
             + ", mvOrdering="
             + mvOrdering()
             + ", ramBytesUsed="
-            + values.ramBytesUsed()
+            + vector.ramBytesUsed()
             + ']';
     }
 
+    @Override
+    public void allowPassingToDifferentDriver() {
+        super.allowPassingToDifferentDriver();
+        vector.allowPassingToDifferentDriver();
+    }
+
     @Override
     public void closeInternal() {
-        blockFactory().adjustBreaker(-ramBytesUsed() + RamUsageEstimator.sizeOf(values), true);
-        Releasables.closeExpectNoException(values);
+        blockFactory().adjustBreaker(-ramBytesUsedOnlyBlock(), true);
+        Releasables.closeExpectNoException(vector);
     }
 }

x-pack/plugin/esql/compute/src/main/generated-src/org/elasticsearch/compute/data/BooleanBigArrayVector.java

@@ -12,7 +12,8 @@ import org.elasticsearch.common.util.BitArray;
 import org.elasticsearch.core.Releasable;
 
 /**
- * Vector implementation that defers to an enclosed BooleanArray.
+ * Vector implementation that defers to an enclosed {@link BitArray}.
+ * Does not take ownership of the array and does not adjust circuit breakers to account for it.
  * This class is generated. Do not edit it.
  */
 public final class BooleanBigArrayVector extends AbstractVector implements BooleanVector, Releasable {
@@ -65,6 +66,8 @@ public final class BooleanBigArrayVector extends AbstractVector implements Boole
 
     @Override
     public void closeInternal() {
+        // The circuit breaker that tracks the values {@link BitArray} is adjusted outside
+        // of this class.
         values.close();
     }
 

x-pack/plugin/esql/compute/src/main/generated-src/org/elasticsearch/compute/data/BooleanVectorBlock.java

@@ -10,7 +10,7 @@ package org.elasticsearch.compute.data;
 import org.elasticsearch.core.Releasables;
 
 /**
- * Block view of a BooleanVector.
+ * Block view of a {@link BooleanVector}. Cannot represent multi-values or nulls.
  * This class is generated. Do not edit it.
  */
 public final class BooleanVectorBlock extends AbstractVectorBlock implements BooleanBlock {

x-pack/plugin/esql/compute/src/main/generated-src/org/elasticsearch/compute/data/BytesRefArrayBlock.java

@@ -15,14 +15,15 @@ import org.elasticsearch.core.Releasables;
 import java.util.BitSet;
 
 /**
- * Block implementation that stores an array of BytesRef.
+ * Block implementation that stores values in a {@link BytesRefArrayVector}.
+ * Does not take ownership of the given {@link BytesRefArray} and does not adjust circuit breakers to account for it.
  * This class is generated. Do not edit it.
  */
 final class BytesRefArrayBlock extends AbstractArrayBlock implements BytesRefBlock {
 
     private static final long BASE_RAM_BYTES_USED = RamUsageEstimator.shallowSizeOfInstance(BytesRefArrayBlock.class);
 
-    private final BytesRefArray values;
+    private final BytesRefArrayVector vector;
 
     BytesRefArrayBlock(
         BytesRefArray values,
@@ -33,7 +34,7 @@ final class BytesRefArrayBlock extends AbstractArrayBlock implements BytesRefBlo
         BlockFactory blockFactory
     ) {
         super(positionCount, firstValueIndexes, nulls, mvOrdering, blockFactory);
-        this.values = values;
+        this.vector = new BytesRefArrayVector(values, (int) values.size(), blockFactory);
     }
 
     @Override
@@ -43,7 +44,7 @@ final class BytesRefArrayBlock extends AbstractArrayBlock implements BytesRefBlo
 
     @Override
     public BytesRef getBytesRef(int valueIndex, BytesRef dest) {
-        return values.get(valueIndex, dest);
+        return vector.getBytesRef(valueIndex, dest);
     }
 
     @Override
@@ -82,7 +83,7 @@ final class BytesRefArrayBlock extends AbstractArrayBlock implements BytesRefBlo
             incRef();
             return this;
         }
-        // TODO use reference counting to share the values
+        // TODO use reference counting to share the vector
         final BytesRef scratch = new BytesRef();
         try (var builder = blockFactory().newBytesRefBlockBuilder(firstValueIndexes[getPositionCount()])) {
             for (int pos = 0; pos < getPositionCount(); pos++) {
@@ -100,14 +101,13 @@ final class BytesRefArrayBlock extends AbstractArrayBlock implements BytesRefBlo
         }
     }
 
-    public static long ramBytesEstimated(BytesRefArray values, int[] firstValueIndexes, BitSet nullsMask) {
-        return BASE_RAM_BYTES_USED + RamUsageEstimator.sizeOf(values) + BlockRamUsageEstimator.sizeOf(firstValueIndexes)
-            + BlockRamUsageEstimator.sizeOfBitSet(nullsMask);
+    private long ramBytesUsedOnlyBlock() {
+        return BASE_RAM_BYTES_USED + BlockRamUsageEstimator.sizeOf(firstValueIndexes) + BlockRamUsageEstimator.sizeOfBitSet(nullsMask);
     }
 
     @Override
     public long ramBytesUsed() {
-        return ramBytesEstimated(values, firstValueIndexes, nullsMask);
+        return ramBytesUsedOnlyBlock() + vector.ramBytesUsed();
     }
 
     @Override
@@ -130,14 +130,20 @@ final class BytesRefArrayBlock extends AbstractArrayBlock implements BytesRefBlo
             + getPositionCount()
             + ", mvOrdering="
             + mvOrdering()
-            + ", values="
-            + values.size()
+            + ", vector="
+            + vector
             + ']';
     }
 
+    @Override
+    public void allowPassingToDifferentDriver() {
+        super.allowPassingToDifferentDriver();
+        vector.allowPassingToDifferentDriver();
+    }
+
     @Override
     public void closeInternal() {
-        blockFactory().adjustBreaker(-ramBytesUsed() + values.bigArraysRamBytesUsed(), true);
-        Releasables.closeExpectNoException(values);
+        blockFactory().adjustBreaker(-ramBytesUsedOnlyBlock(), true);
+        Releasables.closeExpectNoException(vector);
     }
 }

x-pack/plugin/esql/compute/src/main/generated-src/org/elasticsearch/compute/data/BytesRefArrayVector.java

@@ -14,6 +14,7 @@ import org.elasticsearch.core.Releasables;
 
 /**
  * Vector implementation that stores an array of BytesRef values.
+ * Does not take ownership of the given {@link BytesRefArray} and does not adjust circuit breakers to account for it.
  * This class is generated. Do not edit it.
  */
 final class BytesRefArrayVector extends AbstractVector implements BytesRefVector {
@@ -87,6 +88,8 @@ final class BytesRefArrayVector extends AbstractVector implements BytesRefVector
 
     @Override
     public void closeInternal() {
+        // The circuit breaker that tracks the values {@link BytesRefArray} is adjusted outside
+        // of this class.
         blockFactory().adjustBreaker(-ramBytesUsed() + values.bigArraysRamBytesUsed(), true);
         Releasables.closeExpectNoException(values);
     }

x-pack/plugin/esql/compute/src/main/generated-src/org/elasticsearch/compute/data/BytesRefVectorBlock.java

@@ -11,7 +11,7 @@ import org.apache.lucene.util.BytesRef;
 import org.elasticsearch.core.Releasables;
 
 /**
- * Block view of a BytesRefVector.
+ * Block view of a {@link BytesRefVector}. Cannot represent multi-values or nulls.
  * This class is generated. Do not edit it.
  */
 public final class BytesRefVectorBlock extends AbstractVectorBlock implements BytesRefBlock {

x-pack/plugin/esql/compute/src/main/generated-src/org/elasticsearch/compute/data/DoubleArrayBlock.java

@@ -8,19 +8,19 @@
 package org.elasticsearch.compute.data;
 
 import org.apache.lucene.util.RamUsageEstimator;
+import org.elasticsearch.core.Releasables;
 
-import java.util.Arrays;
 import java.util.BitSet;
 
 /**
- * Block implementation that stores an array of double.
+ * Block implementation that stores values in a {@link DoubleArrayVector}.
  * This class is generated. Do not edit it.
  */
 final class DoubleArrayBlock extends AbstractArrayBlock implements DoubleBlock {
 
     private static final long BASE_RAM_BYTES_USED = RamUsageEstimator.shallowSizeOfInstance(DoubleArrayBlock.class);
 
-    private final double[] values;
+    private final DoubleArrayVector vector;
 
     DoubleArrayBlock(
         double[] values,
@@ -31,7 +31,7 @@ final class DoubleArrayBlock extends AbstractArrayBlock implements DoubleBlock {
         BlockFactory blockFactory
     ) {
         super(positionCount, firstValueIndexes, nulls, mvOrdering, blockFactory);
-        this.values = values;
+        this.vector = new DoubleArrayVector(values, values.length, blockFactory);
     }
 
     @Override
@@ -41,7 +41,7 @@ final class DoubleArrayBlock extends AbstractArrayBlock implements DoubleBlock {
 
     @Override
     public double getDouble(int valueIndex) {
-        return values[valueIndex];
+        return vector.getDouble(valueIndex);
     }
 
     @Override
@@ -79,7 +79,7 @@ final class DoubleArrayBlock extends AbstractArrayBlock implements DoubleBlock {
             incRef();
             return this;
         }
-        // TODO use reference counting to share the values
+        // TODO use reference counting to share the vector
         try (var builder = blockFactory().newDoubleBlockBuilder(firstValueIndexes[getPositionCount()])) {
             for (int pos = 0; pos < getPositionCount(); pos++) {
                 if (isNull(pos)) {
@@ -96,14 +96,13 @@ final class DoubleArrayBlock extends AbstractArrayBlock implements DoubleBlock {
         }
     }
 
-    public static long ramBytesEstimated(double[] values, int[] firstValueIndexes, BitSet nullsMask) {
-        return BASE_RAM_BYTES_USED + RamUsageEstimator.sizeOf(values) + BlockRamUsageEstimator.sizeOf(firstValueIndexes)
-            + BlockRamUsageEstimator.sizeOfBitSet(nullsMask);
+    private long ramBytesUsedOnlyBlock() {
+        return BASE_RAM_BYTES_USED + BlockRamUsageEstimator.sizeOf(firstValueIndexes) + BlockRamUsageEstimator.sizeOfBitSet(nullsMask);
     }
 
     @Override
     public long ramBytesUsed() {
-        return ramBytesEstimated(values, firstValueIndexes, nullsMask);
+        return ramBytesUsedOnlyBlock() + vector.ramBytesUsed();
     }
 
     @Override
@@ -126,13 +125,20 @@ final class DoubleArrayBlock extends AbstractArrayBlock implements DoubleBlock {
             + getPositionCount()
             + ", mvOrdering="
             + mvOrdering()
-            + ", values="
-            + Arrays.toString(values)
+            + ", vector="
+            + vector
             + ']';
     }
 
+    @Override
+    public void allowPassingToDifferentDriver() {
+        super.allowPassingToDifferentDriver();
+        vector.allowPassingToDifferentDriver();
+    }
+
     @Override
     public void closeInternal() {
-        blockFactory().adjustBreaker(-ramBytesUsed(), true);
+        blockFactory().adjustBreaker(-ramBytesUsedOnlyBlock(), true);
+        Releasables.closeExpectNoException(vector);
     }
 }

x-pack/plugin/esql/compute/src/main/generated-src/org/elasticsearch/compute/data/DoubleBigArrayBlock.java

@@ -14,13 +14,14 @@ import org.elasticsearch.core.Releasables;
 import java.util.BitSet;
 
 /**
- * Block implementation that stores values in a DoubleArray.
+ * Block implementation that stores values in a {@link DoubleBigArrayVector}. Does not take ownership of the given
+ * {@link DoubleArray} and does not adjust circuit breakers to account for it.
  * This class is generated. Do not edit it.
  */
 public final class DoubleBigArrayBlock extends AbstractArrayBlock implements DoubleBlock {
 
     private static final long BASE_RAM_BYTES_USED = 0; // TODO: fix this
-    private final DoubleArray values;
+    private final DoubleBigArrayVector vector;
 
     public DoubleBigArrayBlock(
         DoubleArray values,
@@ -31,7 +32,7 @@ public final class DoubleBigArrayBlock extends AbstractArrayBlock implements Dou
         BlockFactory blockFactory
     ) {
         super(positionCount, firstValueIndexes, nulls, mvOrdering, blockFactory);
-        this.values = values;
+        this.vector = new DoubleBigArrayVector(values, (int) values.size(), blockFactory);
     }
 
     @Override
@@ -41,7 +42,7 @@ public final class DoubleBigArrayBlock extends AbstractArrayBlock implements Dou
 
     @Override
     public double getDouble(int valueIndex) {
-        return values.get(valueIndex);
+        return vector.getDouble(valueIndex);
     }
 
     @Override
@@ -79,7 +80,7 @@ public final class DoubleBigArrayBlock extends AbstractArrayBlock implements Dou
             incRef();
             return this;
         }
-        // TODO use reference counting to share the values
+        // TODO use reference counting to share the vector
         try (var builder = blockFactory().newDoubleBlockBuilder(firstValueIndexes[getPositionCount()])) {
             for (int pos = 0; pos < getPositionCount(); pos++) {
                 if (isNull(pos)) {
@@ -96,10 +97,13 @@ public final class DoubleBigArrayBlock extends AbstractArrayBlock implements Dou
         }
     }
 
+    private long ramBytesUsedOnlyBlock() {
+        return BASE_RAM_BYTES_USED + BlockRamUsageEstimator.sizeOf(firstValueIndexes) + BlockRamUsageEstimator.sizeOfBitSet(nullsMask);
+    }
+
     @Override
     public long ramBytesUsed() {
-        return BASE_RAM_BYTES_USED + RamUsageEstimator.sizeOf(values) + BlockRamUsageEstimator.sizeOf(firstValueIndexes)
-            + BlockRamUsageEstimator.sizeOfBitSet(nullsMask);
+        return ramBytesUsedOnlyBlock() + RamUsageEstimator.sizeOf(vector);
     }
 
     @Override
@@ -123,13 +127,19 @@ public final class DoubleBigArrayBlock extends AbstractArrayBlock implements Dou
             + ", mvOrdering="
             + mvOrdering()
             + ", ramBytesUsed="
-            + values.ramBytesUsed()
+            + vector.ramBytesUsed()
             + ']';
     }
 
+    @Override
+    public void allowPassingToDifferentDriver() {
+        super.allowPassingToDifferentDriver();
+        vector.allowPassingToDifferentDriver();
+    }
+
     @Override
     public void closeInternal() {
-        blockFactory().adjustBreaker(-ramBytesUsed() + RamUsageEstimator.sizeOf(values), true);
-        Releasables.closeExpectNoException(values);
+        blockFactory().adjustBreaker(-ramBytesUsedOnlyBlock(), true);
+        Releasables.closeExpectNoException(vector);
     }
 }

x-pack/plugin/esql/compute/src/main/generated-src/org/elasticsearch/compute/data/DoubleBigArrayVector.java

@@ -12,7 +12,8 @@ import org.elasticsearch.common.util.DoubleArray;
 import org.elasticsearch.core.Releasable;
 
 /**
- * Vector implementation that defers to an enclosed DoubleArray.
+ * Vector implementation that defers to an enclosed {@link DoubleArray}.
+ * Does not take ownership of the array and does not adjust circuit breakers to account for it.
  * This class is generated. Do not edit it.
  */
 public final class DoubleBigArrayVector extends AbstractVector implements DoubleVector, Releasable {
@@ -63,6 +64,8 @@ public final class DoubleBigArrayVector extends AbstractVector implements Double
 
     @Override
     public void closeInternal() {
+        // The circuit breaker that tracks the values {@link DoubleArray} is adjusted outside
+        // of this class.
         values.close();
     }
 

x-pack/plugin/esql/compute/src/main/generated-src/org/elasticsearch/compute/data/DoubleVectorBlock.java

@@ -10,7 +10,7 @@ package org.elasticsearch.compute.data;
 import org.elasticsearch.core.Releasables;
 
 /**
- * Block view of a DoubleVector.
+ * Block view of a {@link DoubleVector}. Cannot represent multi-values or nulls.
  * This class is generated. Do not edit it.
  */
 public final class DoubleVectorBlock extends AbstractVectorBlock implements DoubleBlock {

x-pack/plugin/esql/compute/src/main/generated-src/org/elasticsearch/compute/data/IntArrayBlock.java

@@ -8,19 +8,19 @@
 package org.elasticsearch.compute.data;
 
 import org.apache.lucene.util.RamUsageEstimator;
+import org.elasticsearch.core.Releasables;
 
-import java.util.Arrays;
 import java.util.BitSet;
 
 /**
- * Block implementation that stores an array of int.
+ * Block implementation that stores values in a {@link IntArrayVector}.
  * This class is generated. Do not edit it.
  */
 final class IntArrayBlock extends AbstractArrayBlock implements IntBlock {
 
     private static final long BASE_RAM_BYTES_USED = RamUsageEstimator.shallowSizeOfInstance(IntArrayBlock.class);
 
-    private final int[] values;
+    private final IntArrayVector vector;
 
     IntArrayBlock(
         int[] values,
@@ -31,7 +31,7 @@ final class IntArrayBlock extends AbstractArrayBlock implements IntBlock {
         BlockFactory blockFactory
     ) {
         super(positionCount, firstValueIndexes, nulls, mvOrdering, blockFactory);
-        this.values = values;
+        this.vector = new IntArrayVector(values, values.length, blockFactory);
     }
 
     @Override
@@ -41,7 +41,7 @@ final class IntArrayBlock extends AbstractArrayBlock implements IntBlock {
 
     @Override
     public int getInt(int valueIndex) {
-        return values[valueIndex];
+        return vector.getInt(valueIndex);
     }
 
     @Override
@@ -79,7 +79,7 @@ final class IntArrayBlock extends AbstractArrayBlock implements IntBlock {
             incRef();
             return this;
         }
-        // TODO use reference counting to share the values
+        // TODO use reference counting to share the vector
         try (var builder = blockFactory().newIntBlockBuilder(firstValueIndexes[getPositionCount()])) {
             for (int pos = 0; pos < getPositionCount(); pos++) {
                 if (isNull(pos)) {
@@ -96,14 +96,13 @@ final class IntArrayBlock extends AbstractArrayBlock implements IntBlock {
         }
     }
 
-    public static long ramBytesEstimated(int[] values, int[] firstValueIndexes, BitSet nullsMask) {
-        return BASE_RAM_BYTES_USED + RamUsageEstimator.sizeOf(values) + BlockRamUsageEstimator.sizeOf(firstValueIndexes)
-            + BlockRamUsageEstimator.sizeOfBitSet(nullsMask);
+    private long ramBytesUsedOnlyBlock() {
+        return BASE_RAM_BYTES_USED + BlockRamUsageEstimator.sizeOf(firstValueIndexes) + BlockRamUsageEstimator.sizeOfBitSet(nullsMask);
     }
 
     @Override
     public long ramBytesUsed() {
-        return ramBytesEstimated(values, firstValueIndexes, nullsMask);
+        return ramBytesUsedOnlyBlock() + vector.ramBytesUsed();
     }
 
     @Override
@@ -126,13 +125,20 @@ final class IntArrayBlock extends AbstractArrayBlock implements IntBlock {
             + getPositionCount()
             + ", mvOrdering="
             + mvOrdering()
-            + ", values="
-            + Arrays.toString(values)
+            + ", vector="
+            + vector
             + ']';
     }
 
+    @Override
+    public void allowPassingToDifferentDriver() {
+        super.allowPassingToDifferentDriver();
+        vector.allowPassingToDifferentDriver();
+    }
+
     @Override
     public void closeInternal() {
-        blockFactory().adjustBreaker(-ramBytesUsed(), true);
+        blockFactory().adjustBreaker(-ramBytesUsedOnlyBlock(), true);
+        Releasables.closeExpectNoException(vector);
     }
 }

x-pack/plugin/esql/compute/src/main/generated-src/org/elasticsearch/compute/data/IntBigArrayBlock.java

@@ -14,13 +14,14 @@ import org.elasticsearch.core.Releasables;
 import java.util.BitSet;
 
 /**
- * Block implementation that stores values in a IntArray.
+ * Block implementation that stores values in a {@link IntBigArrayVector}. Does not take ownership of the given
+ * {@link IntArray} and does not adjust circuit breakers to account for it.
  * This class is generated. Do not edit it.
  */
 public final class IntBigArrayBlock extends AbstractArrayBlock implements IntBlock {
 
     private static final long BASE_RAM_BYTES_USED = 0; // TODO: fix this
-    private final IntArray values;
+    private final IntBigArrayVector vector;
 
     public IntBigArrayBlock(
         IntArray values,
@@ -31,7 +32,7 @@ public final class IntBigArrayBlock extends AbstractArrayBlock implements IntBlo
         BlockFactory blockFactory
     ) {
         super(positionCount, firstValueIndexes, nulls, mvOrdering, blockFactory);
-        this.values = values;
+        this.vector = new IntBigArrayVector(values, (int) values.size(), blockFactory);
     }
 
     @Override
@@ -41,7 +42,7 @@ public final class IntBigArrayBlock extends AbstractArrayBlock implements IntBlo
 
     @Override
     public int getInt(int valueIndex) {
-        return values.get(valueIndex);
+        return vector.getInt(valueIndex);
     }
 
     @Override
@@ -79,7 +80,7 @@ public final class IntBigArrayBlock extends AbstractArrayBlock implements IntBlo
             incRef();
             return this;
         }
-        // TODO use reference counting to share the values
+        // TODO use reference counting to share the vector
         try (var builder = blockFactory().newIntBlockBuilder(firstValueIndexes[getPositionCount()])) {
             for (int pos = 0; pos < getPositionCount(); pos++) {
                 if (isNull(pos)) {
@@ -96,10 +97,13 @@ public final class IntBigArrayBlock extends AbstractArrayBlock implements IntBlo
         }
     }
 
+    private long ramBytesUsedOnlyBlock() {
+        return BASE_RAM_BYTES_USED + BlockRamUsageEstimator.sizeOf(firstValueIndexes) + BlockRamUsageEstimator.sizeOfBitSet(nullsMask);
+    }
+
     @Override
     public long ramBytesUsed() {
-        return BASE_RAM_BYTES_USED + RamUsageEstimator.sizeOf(values) + BlockRamUsageEstimator.sizeOf(firstValueIndexes)
-            + BlockRamUsageEstimator.sizeOfBitSet(nullsMask);
+        return ramBytesUsedOnlyBlock() + RamUsageEstimator.sizeOf(vector);
     }
 
     @Override
@@ -123,13 +127,19 @@ public final class IntBigArrayBlock extends AbstractArrayBlock implements IntBlo
             + ", mvOrdering="
             + mvOrdering()
             + ", ramBytesUsed="
-            + values.ramBytesUsed()
+            + vector.ramBytesUsed()
             + ']';
     }
 
+    @Override
+    public void allowPassingToDifferentDriver() {
+        super.allowPassingToDifferentDriver();
+        vector.allowPassingToDifferentDriver();
+    }
+
     @Override
     public void closeInternal() {
-        blockFactory().adjustBreaker(-ramBytesUsed() + RamUsageEstimator.sizeOf(values), true);
-        Releasables.closeExpectNoException(values);
+        blockFactory().adjustBreaker(-ramBytesUsedOnlyBlock(), true);
+        Releasables.closeExpectNoException(vector);
     }
 }

x-pack/plugin/esql/compute/src/main/generated-src/org/elasticsearch/compute/data/IntBigArrayVector.java

@@ -12,7 +12,8 @@ import org.elasticsearch.common.util.IntArray;
 import org.elasticsearch.core.Releasable;
 
 /**
- * Vector implementation that defers to an enclosed IntArray.
+ * Vector implementation that defers to an enclosed {@link IntArray}.
+ * Does not take ownership of the array and does not adjust circuit breakers to account for it.
  * This class is generated. Do not edit it.
  */
 public final class IntBigArrayVector extends AbstractVector implements IntVector, Releasable {
@@ -63,6 +64,8 @@ public final class IntBigArrayVector extends AbstractVector implements IntVector
 
     @Override
     public void closeInternal() {
+        // The circuit breaker that tracks the values {@link IntArray} is adjusted outside
+        // of this class.
         values.close();
     }
 

x-pack/plugin/esql/compute/src/main/generated-src/org/elasticsearch/compute/data/IntVectorBlock.java

@@ -10,7 +10,7 @@ package org.elasticsearch.compute.data;
 import org.elasticsearch.core.Releasables;
 
 /**
- * Block view of a IntVector.
+ * Block view of a {@link IntVector}. Cannot represent multi-values or nulls.
  * This class is generated. Do not edit it.
  */
 public final class IntVectorBlock extends AbstractVectorBlock implements IntBlock {

x-pack/plugin/esql/compute/src/main/generated-src/org/elasticsearch/compute/data/LongArrayBlock.java

@@ -8,19 +8,19 @@
 package org.elasticsearch.compute.data;
 
 import org.apache.lucene.util.RamUsageEstimator;
+import org.elasticsearch.core.Releasables;
 
-import java.util.Arrays;
 import java.util.BitSet;
 
 /**
- * Block implementation that stores an array of long.
+ * Block implementation that stores values in a {@link LongArrayVector}.
  * This class is generated. Do not edit it.
  */
 final class LongArrayBlock extends AbstractArrayBlock implements LongBlock {
 
     private static final long BASE_RAM_BYTES_USED = RamUsageEstimator.shallowSizeOfInstance(LongArrayBlock.class);
 
-    private final long[] values;
+    private final LongArrayVector vector;
 
     LongArrayBlock(
         long[] values,
@@ -31,7 +31,7 @@ final class LongArrayBlock extends AbstractArrayBlock implements LongBlock {
         BlockFactory blockFactory
     ) {
         super(positionCount, firstValueIndexes, nulls, mvOrdering, blockFactory);
-        this.values = values;
+        this.vector = new LongArrayVector(values, values.length, blockFactory);
     }
 
     @Override
@@ -41,7 +41,7 @@ final class LongArrayBlock extends AbstractArrayBlock implements LongBlock {
 
     @Override
     public long getLong(int valueIndex) {
-        return values[valueIndex];
+        return vector.getLong(valueIndex);
     }
 
     @Override
@@ -79,7 +79,7 @@ final class LongArrayBlock extends AbstractArrayBlock implements LongBlock {
             incRef();
             return this;
         }
-        // TODO use reference counting to share the values
+        // TODO use reference counting to share the vector
         try (var builder = blockFactory().newLongBlockBuilder(firstValueIndexes[getPositionCount()])) {
             for (int pos = 0; pos < getPositionCount(); pos++) {
                 if (isNull(pos)) {
@@ -96,14 +96,13 @@ final class LongArrayBlock extends AbstractArrayBlock implements LongBlock {
         }
     }
 
-    public static long ramBytesEstimated(long[] values, int[] firstValueIndexes, BitSet nullsMask) {
-        return BASE_RAM_BYTES_USED + RamUsageEstimator.sizeOf(values) + BlockRamUsageEstimator.sizeOf(firstValueIndexes)
-            + BlockRamUsageEstimator.sizeOfBitSet(nullsMask);
+    private long ramBytesUsedOnlyBlock() {
+        return BASE_RAM_BYTES_USED + BlockRamUsageEstimator.sizeOf(firstValueIndexes) + BlockRamUsageEstimator.sizeOfBitSet(nullsMask);
     }
 
     @Override
     public long ramBytesUsed() {
-        return ramBytesEstimated(values, firstValueIndexes, nullsMask);
+        return ramBytesUsedOnlyBlock() + vector.ramBytesUsed();
     }
 
     @Override
@@ -126,13 +125,20 @@ final class LongArrayBlock extends AbstractArrayBlock implements LongBlock {
             + getPositionCount()
             + ", mvOrdering="
             + mvOrdering()
-            + ", values="
-            + Arrays.toString(values)
+            + ", vector="
+            + vector
             + ']';
     }
 
+    @Override
+    public void allowPassingToDifferentDriver() {
+        super.allowPassingToDifferentDriver();
+        vector.allowPassingToDifferentDriver();
+    }
+
     @Override
     public void closeInternal() {
-        blockFactory().adjustBreaker(-ramBytesUsed(), true);
+        blockFactory().adjustBreaker(-ramBytesUsedOnlyBlock(), true);
+        Releasables.closeExpectNoException(vector);
     }
 }

x-pack/plugin/esql/compute/src/main/generated-src/org/elasticsearch/compute/data/LongBigArrayBlock.java

@@ -14,13 +14,14 @@ import org.elasticsearch.core.Releasables;
 import java.util.BitSet;
 
 /**
- * Block implementation that stores values in a LongArray.
+ * Block implementation that stores values in a {@link LongBigArrayVector}. Does not take ownership of the given
+ * {@link LongArray} and does not adjust circuit breakers to account for it.
  * This class is generated. Do not edit it.
  */
 public final class LongBigArrayBlock extends AbstractArrayBlock implements LongBlock {
 
     private static final long BASE_RAM_BYTES_USED = 0; // TODO: fix this
-    private final LongArray values;
+    private final LongBigArrayVector vector;
 
     public LongBigArrayBlock(
         LongArray values,
@@ -31,7 +32,7 @@ public final class LongBigArrayBlock extends AbstractArrayBlock implements LongB
         BlockFactory blockFactory
     ) {
         super(positionCount, firstValueIndexes, nulls, mvOrdering, blockFactory);
-        this.values = values;
+        this.vector = new LongBigArrayVector(values, (int) values.size(), blockFactory);
     }
 
     @Override
@@ -41,7 +42,7 @@ public final class LongBigArrayBlock extends AbstractArrayBlock implements LongB
 
     @Override
     public long getLong(int valueIndex) {
-        return values.get(valueIndex);
+        return vector.getLong(valueIndex);
     }
 
     @Override
@@ -79,7 +80,7 @@ public final class LongBigArrayBlock extends AbstractArrayBlock implements LongB
             incRef();
             return this;
         }
-        // TODO use reference counting to share the values
+        // TODO use reference counting to share the vector
         try (var builder = blockFactory().newLongBlockBuilder(firstValueIndexes[getPositionCount()])) {
             for (int pos = 0; pos < getPositionCount(); pos++) {
                 if (isNull(pos)) {
@@ -96,10 +97,13 @@ public final class LongBigArrayBlock extends AbstractArrayBlock implements LongB
         }
     }
 
+    private long ramBytesUsedOnlyBlock() {
+        return BASE_RAM_BYTES_USED + BlockRamUsageEstimator.sizeOf(firstValueIndexes) + BlockRamUsageEstimator.sizeOfBitSet(nullsMask);
+    }
+
     @Override
     public long ramBytesUsed() {
-        return BASE_RAM_BYTES_USED + RamUsageEstimator.sizeOf(values) + BlockRamUsageEstimator.sizeOf(firstValueIndexes)
-            + BlockRamUsageEstimator.sizeOfBitSet(nullsMask);
+        return ramBytesUsedOnlyBlock() + RamUsageEstimator.sizeOf(vector);
     }
 
     @Override
@@ -123,13 +127,19 @@ public final class LongBigArrayBlock extends AbstractArrayBlock implements LongB
             + ", mvOrdering="
             + mvOrdering()
             + ", ramBytesUsed="
-            + values.ramBytesUsed()
+            + vector.ramBytesUsed()
             + ']';
     }
 
+    @Override
+    public void allowPassingToDifferentDriver() {
+        super.allowPassingToDifferentDriver();
+        vector.allowPassingToDifferentDriver();
+    }
+
     @Override
     public void closeInternal() {
-        blockFactory().adjustBreaker(-ramBytesUsed() + RamUsageEstimator.sizeOf(values), true);
-        Releasables.closeExpectNoException(values);
+        blockFactory().adjustBreaker(-ramBytesUsedOnlyBlock(), true);
+        Releasables.closeExpectNoException(vector);
     }
 }

x-pack/plugin/esql/compute/src/main/generated-src/org/elasticsearch/compute/data/LongBigArrayVector.java

@@ -12,7 +12,8 @@ import org.elasticsearch.common.util.LongArray;
 import org.elasticsearch.core.Releasable;
 
 /**
- * Vector implementation that defers to an enclosed LongArray.
+ * Vector implementation that defers to an enclosed {@link LongArray}.
+ * Does not take ownership of the array and does not adjust circuit breakers to account for it.
  * This class is generated. Do not edit it.
  */
 public final class LongBigArrayVector extends AbstractVector implements LongVector, Releasable {
@@ -63,6 +64,8 @@ public final class LongBigArrayVector extends AbstractVector implements LongVect
 
     @Override
     public void closeInternal() {
+        // The circuit breaker that tracks the values {@link LongArray} is adjusted outside
+        // of this class.
         values.close();
     }
 

x-pack/plugin/esql/compute/src/main/generated-src/org/elasticsearch/compute/data/LongVectorBlock.java

@@ -10,7 +10,7 @@ package org.elasticsearch.compute.data;
 import org.elasticsearch.core.Releasables;
 
 /**
- * Block view of a LongVector.
+ * Block view of a {@link LongVector}. Cannot represent multi-values or nulls.
  * This class is generated. Do not edit it.
  */
 public final class LongVectorBlock extends AbstractVectorBlock implements LongBlock {

x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/data/X-ArrayBlock.java.st

@@ -15,25 +15,23 @@ import org.elasticsearch.core.Releasables;
 
 $else$
 import org.apache.lucene.util.RamUsageEstimator;
+import org.elasticsearch.core.Releasables;
 
-import java.util.Arrays;
 $endif$
 import java.util.BitSet;
 
 /**
- * Block implementation that stores an array of $type$.
+ * Block implementation that stores values in a {@link $Type$ArrayVector}.
+$if(BytesRef)$
+ * Does not take ownership of the given {@link BytesRefArray} and does not adjust circuit breakers to account for it.
+$endif$
  * This class is generated. Do not edit it.
  */
 final class $Type$ArrayBlock extends AbstractArrayBlock implements $Type$Block {
 
     private static final long BASE_RAM_BYTES_USED = RamUsageEstimator.shallowSizeOfInstance($Type$ArrayBlock.class);
 
-$if(BytesRef)$
-    private final BytesRefArray values;
-
-$else$
-    private final $type$[] values;
-$endif$
+    private final $Type$ArrayVector vector;
 
     $Type$ArrayBlock(
         $if(BytesRef)$BytesRefArray$else$$type$[]$endif$ values,
@@ -44,7 +42,11 @@ $endif$
         BlockFactory blockFactory
     ) {
         super(positionCount, firstValueIndexes, nulls, mvOrdering, blockFactory);
-        this.values = values;
+        $if(BytesRef)$
+        this.vector = new BytesRefArrayVector(values, (int) values.size(), blockFactory);
+        $else$
+        this.vector = new $Type$ArrayVector(values, values.length, blockFactory);
+        $endif$
     }
 
     @Override
@@ -55,10 +57,10 @@ $endif$
     @Override
 $if(BytesRef)$
     public BytesRef getBytesRef(int valueIndex, BytesRef dest) {
-        return values.get(valueIndex, dest);
+        return vector.getBytesRef(valueIndex, dest);
 $else$
     public $type$ get$Type$(int valueIndex) {
-        return values[valueIndex];
+        return vector.get$Type$(valueIndex);
 $endif$
     }
 
@@ -100,7 +102,7 @@ $endif$
             incRef();
             return this;
         }
-        // TODO use reference counting to share the values
+        // TODO use reference counting to share the vector
 $if(BytesRef)$
         final BytesRef scratch = new BytesRef();
 $endif$
@@ -124,14 +126,13 @@ $endif$
         }
     }
 
-    public static long ramBytesEstimated($if(BytesRef)$BytesRefArray$else$$type$[]$endif$ values, int[] firstValueIndexes, BitSet nullsMask) {
-        return BASE_RAM_BYTES_USED + RamUsageEstimator.sizeOf(values) + BlockRamUsageEstimator.sizeOf(firstValueIndexes)
-            + BlockRamUsageEstimator.sizeOfBitSet(nullsMask);
+    private long ramBytesUsedOnlyBlock() {
+        return BASE_RAM_BYTES_USED + BlockRamUsageEstimator.sizeOf(firstValueIndexes) + BlockRamUsageEstimator.sizeOfBitSet(nullsMask);
     }
 
     @Override
     public long ramBytesUsed() {
-        return ramBytesEstimated(values, firstValueIndexes, nullsMask);
+        return ramBytesUsedOnlyBlock() + vector.ramBytesUsed();
     }
 
     @Override
@@ -154,23 +155,20 @@ $endif$
             + getPositionCount()
             + ", mvOrdering="
             + mvOrdering()
-$if(BytesRef)$
-            + ", values="
-            + values.size()
-$else$
-            + ", values="
-            + Arrays.toString(values)
-$endif$
+            + ", vector="
+            + vector
             + ']';
     }
 
+    @Override
+    public void allowPassingToDifferentDriver() {
+        super.allowPassingToDifferentDriver();
+        vector.allowPassingToDifferentDriver();
+    }
+
     @Override
     public void closeInternal() {
-    $if(BytesRef)$
-        blockFactory().adjustBreaker(-ramBytesUsed() + values.bigArraysRamBytesUsed(), true);
-        Releasables.closeExpectNoException(values);
-    $else$
-        blockFactory().adjustBreaker(-ramBytesUsed(), true);
-    $endif$
+        blockFactory().adjustBreaker(-ramBytesUsedOnlyBlock(), true);
+        Releasables.closeExpectNoException(vector);
     }
 }

x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/data/X-ArrayVector.java.st

@@ -21,6 +21,9 @@ $endif$
 
 /**
  * Vector implementation that stores an array of $type$ values.
+$if(BytesRef)$
+ * Does not take ownership of the given {@link BytesRefArray} and does not adjust circuit breakers to account for it.
+$endif$
  * This class is generated. Do not edit it.
  */
 final class $Type$ArrayVector extends AbstractVector implements $Type$Vector {
@@ -118,6 +121,8 @@ $endif$
 $if(BytesRef)$
     @Override
     public void closeInternal() {
+        // The circuit breaker that tracks the values {@link BytesRefArray} is adjusted outside
+        // of this class.
         blockFactory().adjustBreaker(-ramBytesUsed() + values.bigArraysRamBytesUsed(), true);
         Releasables.closeExpectNoException(values);
     }

x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/data/X-BigArrayBlock.java.st

@@ -14,13 +14,14 @@ import org.elasticsearch.core.Releasables;
 import java.util.BitSet;
 
 /**
- * Block implementation that stores values in a $Type$Array.
+ * Block implementation that stores values in a {@link $Type$BigArrayVector}. Does not take ownership of the given
+ * {@link $if(boolean)$Bit$else$$Type$$endif$Array} and does not adjust circuit breakers to account for it.
  * This class is generated. Do not edit it.
  */
 public final class $Type$BigArrayBlock extends AbstractArrayBlock implements $Type$Block {
 
     private static final long BASE_RAM_BYTES_USED = 0; // TODO: fix this
-    private final $if(boolean)$Bit$else$$Type$$endif$Array values;
+    private final $Type$BigArrayVector vector;
 
     public $Type$BigArrayBlock(
         $if(boolean)$Bit$else$$Type$$endif$Array values,
@@ -31,7 +32,7 @@ public final class $Type$BigArrayBlock extends AbstractArrayBlock implements $Ty
         BlockFactory blockFactory
     ) {
         super(positionCount, firstValueIndexes, nulls, mvOrdering, blockFactory);
-        this.values = values;
+        this.vector = new $Type$BigArrayVector(values, (int) values.size(), blockFactory);
     }
 
     @Override
@@ -41,7 +42,7 @@ public final class $Type$BigArrayBlock extends AbstractArrayBlock implements $Ty
 
     @Override
     public $type$ get$Type$(int valueIndex) {
-        return values.get(valueIndex);
+        return vector.get$Type$(valueIndex);
     }
 
     @Override
@@ -79,7 +80,7 @@ public final class $Type$BigArrayBlock extends AbstractArrayBlock implements $Ty
             incRef();
             return this;
         }
-        // TODO use reference counting to share the values
+        // TODO use reference counting to share the vector
         try (var builder = blockFactory().new$Type$BlockBuilder(firstValueIndexes[getPositionCount()])) {
             for (int pos = 0; pos < getPositionCount(); pos++) {
                 if (isNull(pos)) {
@@ -96,10 +97,13 @@ public final class $Type$BigArrayBlock extends AbstractArrayBlock implements $Ty
         }
     }
 
+    private long ramBytesUsedOnlyBlock() {
+        return BASE_RAM_BYTES_USED + BlockRamUsageEstimator.sizeOf(firstValueIndexes) + BlockRamUsageEstimator.sizeOfBitSet(nullsMask);
+    }
+
     @Override
     public long ramBytesUsed() {
-        return BASE_RAM_BYTES_USED + RamUsageEstimator.sizeOf(values) + BlockRamUsageEstimator.sizeOf(firstValueIndexes)
-            + BlockRamUsageEstimator.sizeOfBitSet(nullsMask);
+        return ramBytesUsedOnlyBlock() + RamUsageEstimator.sizeOf(vector);
     }
 
     @Override
@@ -123,13 +127,19 @@ public final class $Type$BigArrayBlock extends AbstractArrayBlock implements $Ty
             + ", mvOrdering="
             + mvOrdering()
             + ", ramBytesUsed="
-            + values.ramBytesUsed()
+            + vector.ramBytesUsed()
             + ']';
     }
 
+    @Override
+    public void allowPassingToDifferentDriver() {
+        super.allowPassingToDifferentDriver();
+        vector.allowPassingToDifferentDriver();
+    }
+
     @Override
     public void closeInternal() {
-        blockFactory().adjustBreaker(-ramBytesUsed() + RamUsageEstimator.sizeOf(values), true);
-        Releasables.closeExpectNoException(values);
+        blockFactory().adjustBreaker(-ramBytesUsedOnlyBlock(), true);
+        Releasables.closeExpectNoException(vector);
     }
 }

x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/data/X-BigArrayVector.java.st

@@ -12,7 +12,8 @@ import org.elasticsearch.common.util.$if(boolean)$Bit$else$$Type$$endif$Array;
 import org.elasticsearch.core.Releasable;
 
 /**
- * Vector implementation that defers to an enclosed $Type$Array.
+ * Vector implementation that defers to an enclosed {@link $if(boolean)$Bit$else$$Type$$endif$Array}.
+ * Does not take ownership of the array and does not adjust circuit breakers to account for it.
  * This class is generated. Do not edit it.
  */
 public final class $Type$BigArrayVector extends AbstractVector implements $Type$Vector, Releasable {
@@ -72,6 +73,8 @@ public final class $Type$BigArrayVector extends AbstractVector implements $Type$
 
     @Override
     public void closeInternal() {
+        // The circuit breaker that tracks the values {@link $if(boolean)$Bit$else$$Type$$endif$Array} is adjusted outside
+        // of this class.
         values.close();
     }
 

x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/data/X-VectorBlock.java.st

@@ -13,7 +13,7 @@ $endif$
 import org.elasticsearch.core.Releasables;
 
 /**
- * Block view of a $Type$Vector.
+ * Block view of a {@link $Type$Vector}. Cannot represent multi-values or nulls.
  * This class is generated. Do not edit it.
  */
 public final class $Type$VectorBlock extends AbstractVectorBlock implements $Type$Block {

x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/data/FilteredBlockTests.java

@@ -288,7 +288,10 @@ public class FilteredBlockTests extends ESTestCase {
             var filter = block.filter(0, 1);
             assertThat(
                 filter.toString(),
-                containsString("BooleanArrayBlock[positions=2, mvOrdering=UNORDERED, values=[true, true, false, false]]")
+                containsString(
+                    "BooleanArrayBlock[positions=2, mvOrdering=UNORDERED, "
+                        + "vector=BooleanArrayVector[positions=4, values=[true, true, false, false]]]"
+                )
             );
             Releasables.close(builder, block);
             releaseAndAssertBreaker(filter);
@@ -300,7 +303,12 @@ public class FilteredBlockTests extends ESTestCase {
             builder.beginPositionEntry().appendInt(90).appendInt(1000).endPositionEntry();
             var block = builder.build();
             var filter = block.filter(0, 1);
-            assertThat(filter.toString(), containsString("IntArrayBlock[positions=2, mvOrdering=UNORDERED, values=[0, 10, 20, 50]]"));
+            assertThat(
+                filter.toString(),
+                containsString(
+                    "IntArrayBlock[positions=2, mvOrdering=UNORDERED, vector=IntArrayVector[positions=4, values=[0, 10, 20, 50]]]"
+                )
+            );
             Releasables.close(builder, block);
             releaseAndAssertBreaker(filter);
         }
@@ -311,7 +319,12 @@ public class FilteredBlockTests extends ESTestCase {
             builder.beginPositionEntry().appendLong(90).appendLong(1000).endPositionEntry();
             var block = builder.build();
             var filter = block.filter(0, 1);
-            assertThat(filter.toString(), containsString("LongArrayBlock[positions=2, mvOrdering=UNORDERED, values=[0, 10, 20, 50]]"));
+            assertThat(
+                filter.toString(),
+                containsString(
+                    "LongArrayBlock[positions=2, mvOrdering=UNORDERED, vector=LongArrayVector[positions=4, values=[0, 10, 20, 50]]]"
+                )
+            );
             Releasables.close(builder, block);
             releaseAndAssertBreaker(filter);
         }
@@ -324,7 +337,10 @@ public class FilteredBlockTests extends ESTestCase {
             var filter = block.filter(0, 1);
             assertThat(
                 filter.toString(),
-                containsString("DoubleArrayBlock[positions=2, mvOrdering=UNORDERED, values=[0.0, 10.0, 0.002, 1.0E9]]")
+                containsString(
+                    "DoubleArrayBlock[positions=2, mvOrdering=UNORDERED, "
+                        + "vector=DoubleArrayVector[positions=4, values=[0.0, 10.0, 0.002, 1.0E9]]]"
+                )
             );
             Releasables.close(builder, block);
             releaseAndAssertBreaker(filter);
@@ -338,7 +354,10 @@ public class FilteredBlockTests extends ESTestCase {
             builder.beginPositionEntry().appendBytesRef(new BytesRef("pig")).appendBytesRef(new BytesRef("chicken")).endPositionEntry();
             var block = builder.build();
             var filter = block.filter(0, 1);
-            assertThat(filter.toString(), containsString("BytesRefArrayBlock[positions=2, mvOrdering=UNORDERED, values=4]"));
+            assertThat(
+                filter.toString(),
+                containsString("BytesRefArrayBlock[positions=2, mvOrdering=UNORDERED, vector=BytesRefArrayVector[positions=4]]")
+            );
             assertThat(filter.getPositionCount(), equalTo(2));
             Releasables.close(builder, block);
             releaseAndAssertBreaker(filter);