ESQL: Block#deepCopy preserves constants (#133510)

This removes `BlockUtils#deepCopyOf` with and adds `Block#deepCopy`
which preserves the original `Block`'s constant block and ordinal
blocks. This should make testing the compute engine easier. And it might
be useful for merging blocks together one day.

Also! This fixed a CI failure that I needed to fix so I could test this
change.

muted-tests.yml

@@ -402,9 +402,6 @@ tests:
 - class: org.elasticsearch.packaging.test.DockerTests
   method: test010Install
   issue: https://github.com/elastic/elasticsearch/issues/131376
-- class: org.elasticsearch.compute.lucene.read.SortedSetOrdinalsBuilderTests
-  method: testReader
-  issue: https://github.com/elastic/elasticsearch/issues/131573
 - class: org.elasticsearch.packaging.test.DockerTests
   method: test151MachineDependentHeapWithSizeOverride
   issue: https://github.com/elastic/elasticsearch/issues/123437

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

@@ -47,6 +47,19 @@ public sealed interface BooleanBlock extends Block permits BooleanArrayBlock, Bo
     @Override
     BooleanBlock filter(int... positions);
 
+    /**
+     * Make a deep copy of this {@link Block} using the provided {@link BlockFactory},
+     * likely copying all data.
+     */
+    @Override
+    default BooleanBlock deepCopy(BlockFactory blockFactory) {
+        try (BooleanBlock.Builder builder = blockFactory.newBooleanBlockBuilder(getPositionCount())) {
+            builder.copyFrom(this, 0, getPositionCount());
+            builder.mvOrdering(mvOrdering());
+            return builder.build();
+        }
+    }
+
     @Override
     BooleanBlock keepMask(BooleanVector mask);
 

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

@@ -34,6 +34,19 @@ public sealed interface BooleanVector extends Vector permits ConstantBooleanVect
     @Override
     BooleanBlock keepMask(BooleanVector mask);
 
+    /**
+     * Make a deep copy of this {@link Vector} using the provided {@link BlockFactory},
+     * likely copying all data.
+     */
+    @Override
+    default BooleanVector deepCopy(BlockFactory blockFactory) {
+        try (BooleanBlock.Builder builder = blockFactory.newBooleanBlockBuilder(getPositionCount())) {
+            builder.copyFrom(asBlock(), 0, getPositionCount());
+            builder.mvOrdering(Block.MvOrdering.DEDUPLICATED_AND_SORTED_ASCENDING);
+            return builder.build().asVector();
+        }
+    }
+
     @Override
     ReleasableIterator<? extends BooleanBlock> lookup(IntBlock positions, ByteSizeValue targetBlockSize);
 

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

@@ -64,6 +64,11 @@ public final class BooleanVectorBlock extends AbstractVectorBlock implements Boo
         return vector.keepMask(mask);
     }
 
+    @Override
+    public BooleanBlock deepCopy(BlockFactory blockFactory) {
+        return vector.deepCopy(blockFactory).asBlock();
+    }
+
     @Override
     public ReleasableIterator<? extends BooleanBlock> lookup(IntBlock positions, ByteSizeValue targetBlockSize) {
         return vector.lookup(positions, targetBlockSize);

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

@@ -50,6 +50,19 @@ public sealed interface BytesRefBlock extends Block permits BytesRefArrayBlock,
     @Override
     BytesRefBlock filter(int... positions);
 
+    /**
+     * Make a deep copy of this {@link Block} using the provided {@link BlockFactory},
+     * likely copying all data.
+     */
+    @Override
+    default BytesRefBlock deepCopy(BlockFactory blockFactory) {
+        try (BytesRefBlock.Builder builder = blockFactory.newBytesRefBlockBuilder(getPositionCount())) {
+            builder.copyFrom(this, 0, getPositionCount());
+            builder.mvOrdering(mvOrdering());
+            return builder.build();
+        }
+    }
+
     @Override
     BytesRefBlock keepMask(BooleanVector mask);
 

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

@@ -41,6 +41,19 @@ public sealed interface BytesRefVector extends Vector permits ConstantBytesRefVe
     @Override
     BytesRefBlock keepMask(BooleanVector mask);
 
+    /**
+     * Make a deep copy of this {@link Vector} using the provided {@link BlockFactory},
+     * likely copying all data.
+     */
+    @Override
+    default BytesRefVector deepCopy(BlockFactory blockFactory) {
+        try (BytesRefBlock.Builder builder = blockFactory.newBytesRefBlockBuilder(getPositionCount())) {
+            builder.copyFrom(asBlock(), 0, getPositionCount());
+            builder.mvOrdering(Block.MvOrdering.DEDUPLICATED_AND_SORTED_ASCENDING);
+            return builder.build().asVector();
+        }
+    }
+
     @Override
     ReleasableIterator<? extends BytesRefBlock> lookup(IntBlock positions, ByteSizeValue targetBlockSize);
 

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

@@ -69,6 +69,11 @@ public final class BytesRefVectorBlock extends AbstractVectorBlock implements By
         return vector.keepMask(mask);
     }
 
+    @Override
+    public BytesRefBlock deepCopy(BlockFactory blockFactory) {
+        return vector.deepCopy(blockFactory).asBlock();
+    }
+
     @Override
     public ReleasableIterator<? extends BytesRefBlock> lookup(IntBlock positions, ByteSizeValue targetBlockSize) {
         return vector.lookup(positions, targetBlockSize);

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

@@ -119,6 +119,11 @@ final class ConstantBooleanVector extends AbstractVector implements BooleanVecto
         return true;
     }
 
+    @Override
+    public BooleanVector deepCopy(BlockFactory blockFactory) {
+        return blockFactory.newConstantBooleanVector(value, getPositionCount());
+    }
+
     @Override
     public long ramBytesUsed() {
         return RAM_BYTES_USED;

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

@@ -112,6 +112,11 @@ final class ConstantBytesRefVector extends AbstractVector implements BytesRefVec
         return true;
     }
 
+    @Override
+    public BytesRefVector deepCopy(BlockFactory blockFactory) {
+        return blockFactory.newConstantBytesRefVector(value, getPositionCount());
+    }
+
     public static long ramBytesUsed(BytesRef value) {
         return BASE_RAM_BYTES_USED + RamUsageEstimator.sizeOf(value.bytes);
     }

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

@@ -103,6 +103,11 @@ final class ConstantDoubleVector extends AbstractVector implements DoubleVector
         return true;
     }
 
+    @Override
+    public DoubleVector deepCopy(BlockFactory blockFactory) {
+        return blockFactory.newConstantDoubleVector(value, getPositionCount());
+    }
+
     @Override
     public long ramBytesUsed() {
         return RAM_BYTES_USED;

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

@@ -103,6 +103,11 @@ final class ConstantFloatVector extends AbstractVector implements FloatVector {
         return true;
     }
 
+    @Override
+    public FloatVector deepCopy(BlockFactory blockFactory) {
+        return blockFactory.newConstantFloatVector(value, getPositionCount());
+    }
+
     @Override
     public long ramBytesUsed() {
         return RAM_BYTES_USED;

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

@@ -119,6 +119,11 @@ final class ConstantIntVector extends AbstractVector implements IntVector {
         return true;
     }
 
+    @Override
+    public IntVector deepCopy(BlockFactory blockFactory) {
+        return blockFactory.newConstantIntVector(value, getPositionCount());
+    }
+
     @Override
     public long ramBytesUsed() {
         return RAM_BYTES_USED;

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

@@ -103,6 +103,11 @@ final class ConstantLongVector extends AbstractVector implements LongVector {
         return true;
     }
 
+    @Override
+    public LongVector deepCopy(BlockFactory blockFactory) {
+        return blockFactory.newConstantLongVector(value, getPositionCount());
+    }
+
     @Override
     public long ramBytesUsed() {
         return RAM_BYTES_USED;

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

@@ -40,6 +40,19 @@ public sealed interface DoubleBlock extends Block permits DoubleArrayBlock, Doub
     @Override
     DoubleBlock filter(int... positions);
 
+    /**
+     * Make a deep copy of this {@link Block} using the provided {@link BlockFactory},
+     * likely copying all data.
+     */
+    @Override
+    default DoubleBlock deepCopy(BlockFactory blockFactory) {
+        try (DoubleBlock.Builder builder = blockFactory.newDoubleBlockBuilder(getPositionCount())) {
+            builder.copyFrom(this, 0, getPositionCount());
+            builder.mvOrdering(mvOrdering());
+            return builder.build();
+        }
+    }
+
     @Override
     DoubleBlock keepMask(BooleanVector mask);
 

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

@@ -34,6 +34,19 @@ public sealed interface DoubleVector extends Vector permits ConstantDoubleVector
     @Override
     DoubleBlock keepMask(BooleanVector mask);
 
+    /**
+     * Make a deep copy of this {@link Vector} using the provided {@link BlockFactory},
+     * likely copying all data.
+     */
+    @Override
+    default DoubleVector deepCopy(BlockFactory blockFactory) {
+        try (DoubleBlock.Builder builder = blockFactory.newDoubleBlockBuilder(getPositionCount())) {
+            builder.copyFrom(asBlock(), 0, getPositionCount());
+            builder.mvOrdering(Block.MvOrdering.DEDUPLICATED_AND_SORTED_ASCENDING);
+            return builder.build().asVector();
+        }
+    }
+
     @Override
     ReleasableIterator<? extends DoubleBlock> lookup(IntBlock positions, ByteSizeValue targetBlockSize);
 

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

@@ -58,6 +58,11 @@ public final class DoubleVectorBlock extends AbstractVectorBlock implements Doub
         return vector.keepMask(mask);
     }
 
+    @Override
+    public DoubleBlock deepCopy(BlockFactory blockFactory) {
+        return vector.deepCopy(blockFactory).asBlock();
+    }
+
     @Override
     public ReleasableIterator<? extends DoubleBlock> lookup(IntBlock positions, ByteSizeValue targetBlockSize) {
         return vector.lookup(positions, targetBlockSize);

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

@@ -40,6 +40,19 @@ public sealed interface FloatBlock extends Block permits FloatArrayBlock, FloatV
     @Override
     FloatBlock filter(int... positions);
 
+    /**
+     * Make a deep copy of this {@link Block} using the provided {@link BlockFactory},
+     * likely copying all data.
+     */
+    @Override
+    default FloatBlock deepCopy(BlockFactory blockFactory) {
+        try (FloatBlock.Builder builder = blockFactory.newFloatBlockBuilder(getPositionCount())) {
+            builder.copyFrom(this, 0, getPositionCount());
+            builder.mvOrdering(mvOrdering());
+            return builder.build();
+        }
+    }
+
     @Override
     FloatBlock keepMask(BooleanVector mask);
 

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

@@ -34,6 +34,19 @@ public sealed interface FloatVector extends Vector permits ConstantFloatVector,
     @Override
     FloatBlock keepMask(BooleanVector mask);
 
+    /**
+     * Make a deep copy of this {@link Vector} using the provided {@link BlockFactory},
+     * likely copying all data.
+     */
+    @Override
+    default FloatVector deepCopy(BlockFactory blockFactory) {
+        try (FloatBlock.Builder builder = blockFactory.newFloatBlockBuilder(getPositionCount())) {
+            builder.copyFrom(asBlock(), 0, getPositionCount());
+            builder.mvOrdering(Block.MvOrdering.DEDUPLICATED_AND_SORTED_ASCENDING);
+            return builder.build().asVector();
+        }
+    }
+
     @Override
     ReleasableIterator<? extends FloatBlock> lookup(IntBlock positions, ByteSizeValue targetBlockSize);
 

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

@@ -58,6 +58,11 @@ public final class FloatVectorBlock extends AbstractVectorBlock implements Float
         return vector.keepMask(mask);
     }
 
+    @Override
+    public FloatBlock deepCopy(BlockFactory blockFactory) {
+        return vector.deepCopy(blockFactory).asBlock();
+    }
+
     @Override
     public ReleasableIterator<? extends FloatBlock> lookup(IntBlock positions, ByteSizeValue targetBlockSize) {
         return vector.lookup(positions, targetBlockSize);

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

@@ -40,6 +40,19 @@ public sealed interface IntBlock extends Block permits IntArrayBlock, IntVectorB
     @Override
     IntBlock filter(int... positions);
 
+    /**
+     * Make a deep copy of this {@link Block} using the provided {@link BlockFactory},
+     * likely copying all data.
+     */
+    @Override
+    default IntBlock deepCopy(BlockFactory blockFactory) {
+        try (IntBlock.Builder builder = blockFactory.newIntBlockBuilder(getPositionCount())) {
+            builder.copyFrom(this, 0, getPositionCount());
+            builder.mvOrdering(mvOrdering());
+            return builder.build();
+        }
+    }
+
     @Override
     IntBlock keepMask(BooleanVector mask);
 

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

@@ -34,6 +34,19 @@ public sealed interface IntVector extends Vector permits ConstantIntVector, IntA
     @Override
     IntBlock keepMask(BooleanVector mask);
 
+    /**
+     * Make a deep copy of this {@link Vector} using the provided {@link BlockFactory},
+     * likely copying all data.
+     */
+    @Override
+    default IntVector deepCopy(BlockFactory blockFactory) {
+        try (IntBlock.Builder builder = blockFactory.newIntBlockBuilder(getPositionCount())) {
+            builder.copyFrom(asBlock(), 0, getPositionCount());
+            builder.mvOrdering(Block.MvOrdering.DEDUPLICATED_AND_SORTED_ASCENDING);
+            return builder.build().asVector();
+        }
+    }
+
     @Override
     ReleasableIterator<? extends IntBlock> lookup(IntBlock positions, ByteSizeValue targetBlockSize);
 

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

@@ -58,6 +58,11 @@ public final class IntVectorBlock extends AbstractVectorBlock implements IntBloc
         return vector.keepMask(mask);
     }
 
+    @Override
+    public IntBlock deepCopy(BlockFactory blockFactory) {
+        return vector.deepCopy(blockFactory).asBlock();
+    }
+
     @Override
     public ReleasableIterator<? extends IntBlock> lookup(IntBlock positions, ByteSizeValue targetBlockSize) {
         return vector.lookup(positions, targetBlockSize);

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

@@ -40,6 +40,19 @@ public sealed interface LongBlock extends Block permits LongArrayBlock, LongVect
     @Override
     LongBlock filter(int... positions);
 
+    /**
+     * Make a deep copy of this {@link Block} using the provided {@link BlockFactory},
+     * likely copying all data.
+     */
+    @Override
+    default LongBlock deepCopy(BlockFactory blockFactory) {
+        try (LongBlock.Builder builder = blockFactory.newLongBlockBuilder(getPositionCount())) {
+            builder.copyFrom(this, 0, getPositionCount());
+            builder.mvOrdering(mvOrdering());
+            return builder.build();
+        }
+    }
+
     @Override
     LongBlock keepMask(BooleanVector mask);
 

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

@@ -34,6 +34,19 @@ public sealed interface LongVector extends Vector permits ConstantLongVector, Lo
     @Override
     LongBlock keepMask(BooleanVector mask);
 
+    /**
+     * Make a deep copy of this {@link Vector} using the provided {@link BlockFactory},
+     * likely copying all data.
+     */
+    @Override
+    default LongVector deepCopy(BlockFactory blockFactory) {
+        try (LongBlock.Builder builder = blockFactory.newLongBlockBuilder(getPositionCount())) {
+            builder.copyFrom(asBlock(), 0, getPositionCount());
+            builder.mvOrdering(Block.MvOrdering.DEDUPLICATED_AND_SORTED_ASCENDING);
+            return builder.build().asVector();
+        }
+    }
+
     @Override
     ReleasableIterator<? extends LongBlock> lookup(IntBlock positions, ByteSizeValue targetBlockSize);
 

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

@@ -58,6 +58,11 @@ public final class LongVectorBlock extends AbstractVectorBlock implements LongBl
         return vector.keepMask(mask);
     }
 
+    @Override
+    public LongBlock deepCopy(BlockFactory blockFactory) {
+        return vector.deepCopy(blockFactory).asBlock();
+    }
+
     @Override
     public ReleasableIterator<? extends LongBlock> lookup(IntBlock positions, ByteSizeValue targetBlockSize) {
         return vector.lookup(positions, targetBlockSize);

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

@@ -193,6 +193,27 @@ public final class AggregateMetricDoubleArrayBlock extends AbstractNonThreadSafe
         }
     }
 
+    @Override
+    public Block deepCopy(BlockFactory blockFactory) {
+        AggregateMetricDoubleArrayBlock result = null;
+        DoubleBlock newMinBlock = null;
+        DoubleBlock newMaxBlock = null;
+        DoubleBlock newSumBlock = null;
+        IntBlock newCountBlock = null;
+        try {
+            newMinBlock = minBlock.deepCopy(blockFactory);
+            newMaxBlock = maxBlock.deepCopy(blockFactory);
+            newSumBlock = sumBlock.deepCopy(blockFactory);
+            newCountBlock = countBlock.deepCopy(blockFactory);
+            result = new AggregateMetricDoubleArrayBlock(newMinBlock, newMaxBlock, newSumBlock, newCountBlock);
+            return result;
+        } finally {
+            if (result == null) {
+                Releasables.close(newMinBlock, newMaxBlock, newSumBlock, newCountBlock);
+            }
+        }
+    }
+
     @Override
     public ReleasableIterator<? extends AggregateMetricDoubleBlock> lookup(IntBlock positions, ByteSizeValue targetBlockSize) {
         // TODO: support

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

@@ -257,6 +257,12 @@ public interface Block extends Accountable, BlockLoader.Block, Writeable, RefCou
         }
     }
 
+    /**
+     * Make a deep copy of this {@link Block} using the provided {@link BlockFactory},
+     * likely copying all data.
+     */
+    Block deepCopy(BlockFactory blockFactory);
+
     /**
      * Builds {@link Block}s. Typically, you use one of it's direct supinterfaces like {@link IntBlock.Builder}.
      * This is {@link Releasable} and should be released after building the block or if building the block fails.

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

@@ -167,6 +167,23 @@ public final class CompositeBlock extends AbstractNonThreadSafeRefCounted implem
         }
     }
 
+    @Override
+    public CompositeBlock deepCopy(BlockFactory blockFactory) {
+        CompositeBlock result = null;
+        final Block[] copied = new Block[blocks.length];
+        try {
+            for (int i = 0; i < blocks.length; i++) {
+                copied[i] = blocks[i].deepCopy(blockFactory);
+            }
+            result = new CompositeBlock(copied);
+            return result;
+        } finally {
+            if (result == null) {
+                Releasables.close(copied);
+            }
+        }
+    }
+
     @Override
     public Block keepMask(BooleanVector mask) {
         CompositeBlock result = null;

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

@@ -87,6 +87,11 @@ public final class ConstantNullBlock extends AbstractNonThreadSafeRefCounted
         return (ConstantNullBlock) blockFactory().newConstantNullBlock(positions.length);
     }
 
+    @Override
+    public ConstantNullBlock deepCopy(BlockFactory blockFactory) {
+        return (ConstantNullBlock) blockFactory.newConstantNullBlock(positionCount);
+    }
+
     @Override
     public ConstantNullBlock keepMask(BooleanVector mask) {
         return (ConstantNullBlock) blockFactory().newConstantNullBlock(getPositionCount());

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

@@ -54,6 +54,12 @@ public final class ConstantNullVector extends AbstractVector
         throw new UnsupportedOperationException("null vector");
     }
 
+    @Override
+    public ConstantNullVector deepCopy(BlockFactory blockFactory) {
+        assert false : "null vector";
+        throw new UnsupportedOperationException("null vector");
+    }
+
     @Override
     public ConstantNullBlock keepMask(BooleanVector mask) {
         assert false : "null vector";

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

@@ -44,7 +44,12 @@ public class DocBlock extends AbstractVectorBlock implements Block, RefCounted {
 
     @Override
     public Block filter(int... positions) {
-        return new DocBlock(asVector().filter(positions));
+        return new DocBlock(vector.filter(positions));
+    }
+
+    @Override
+    public Block deepCopy(BlockFactory blockFactory) {
+        return new DocBlock(vector.deepCopy(blockFactory));
     }
 
     @Override

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

@@ -278,6 +278,25 @@ public final class DocVector extends AbstractVector implements Vector {
         }
     }
 
+    @Override
+    public DocVector deepCopy(BlockFactory blockFactory) {
+        IntVector filteredShards = null;
+        IntVector filteredSegments = null;
+        IntVector filteredDocs = null;
+        DocVector result = null;
+        try {
+            filteredShards = shards.deepCopy(blockFactory);
+            filteredSegments = segments.deepCopy(blockFactory);
+            filteredDocs = docs.deepCopy(blockFactory);
+            result = new DocVector(shardRefCounters, filteredShards, filteredSegments, filteredDocs, null);
+            return result;
+        } finally {
+            if (result == null) {
+                Releasables.closeExpectNoException(filteredShards, filteredSegments, filteredDocs);
+            }
+        }
+    }
+
     @Override
     public DocBlock keepMask(BooleanVector mask) {
         throw new UnsupportedOperationException("can't mask DocVector because it can't contain nulls");

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

@@ -160,6 +160,22 @@ public final class OrdinalBytesRefBlock extends AbstractNonThreadSafeRefCounted
         return new BytesRefLookup(this, positions, targetBlockSize);
     }
 
+    @Override
+    public OrdinalBytesRefBlock deepCopy(BlockFactory blockFactory) {
+        IntBlock copiedOrdinals = null;
+        BytesRefVector copiedBytes = null;
+        try {
+            copiedOrdinals = ordinals.deepCopy(blockFactory);
+            copiedBytes = bytes.deepCopy(blockFactory);
+            OrdinalBytesRefBlock result = new OrdinalBytesRefBlock(copiedOrdinals, copiedBytes);
+            copiedOrdinals = null;
+            copiedBytes = null;
+            return result;
+        } finally {
+            Releasables.closeExpectNoException(copiedOrdinals, copiedBytes);
+        }
+    }
+
     @Override
     protected void closeInternal() {
         Releasables.close(ordinals, bytes);

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

@@ -131,6 +131,22 @@ public final class OrdinalBytesRefVector extends AbstractNonThreadSafeRefCounted
         return asBlock().keepMask(mask);
     }
 
+    @Override
+    public OrdinalBytesRefVector deepCopy(BlockFactory blockFactory) {
+        IntVector copiedOrdinals = null;
+        BytesRefVector copiedBytes = null;
+        try {
+            copiedOrdinals = ordinals.deepCopy(blockFactory);
+            copiedBytes = bytes.deepCopy(blockFactory);
+            OrdinalBytesRefVector result = new OrdinalBytesRefVector(copiedOrdinals, copiedBytes);
+            copiedOrdinals = null;
+            copiedBytes = null;
+            return result;
+        } finally {
+            Releasables.closeExpectNoException(copiedOrdinals, copiedBytes);
+        }
+    }
+
     @Override
     public ReleasableIterator<? extends BytesRefBlock> lookup(IntBlock positions, ByteSizeValue targetBlockSize) {
         return new BytesRefLookup(asBlock(), positions, targetBlockSize);

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

@@ -94,6 +94,12 @@ public interface Vector extends Accountable, RefCounted, Releasable {
      */
     void allowPassingToDifferentDriver();
 
+    /**
+     * Make a deep copy of this {@link Block} using the provided {@link BlockFactory},
+     * likely copying all data.
+     */
+    Vector deepCopy(BlockFactory blockFactory);
+
     /**
      * Builds {@link Vector}s. Typically, you use one of it's direct supinterfaces like {@link IntVector.Builder}.
      * This is {@link Releasable} and should be released after building the vector or if building the vector fails.

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

@@ -76,6 +76,19 @@ $endif$
     @Override
     $Type$Block filter(int... positions);
 
+    /**
+     * Make a deep copy of this {@link Block} using the provided {@link BlockFactory},
+     * likely copying all data.
+     */
+    @Override
+    default $Type$Block deepCopy(BlockFactory blockFactory) {
+        try ($Type$Block.Builder builder = blockFactory.new$Type$BlockBuilder(getPositionCount())) {
+            builder.copyFrom(this, 0, getPositionCount());
+            builder.mvOrdering(mvOrdering());
+            return builder.build();
+        }
+    }
+
     @Override
     $Type$Block keepMask(BooleanVector mask);
 

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

@@ -174,6 +174,11 @@ $endif$
         return true;
     }
 
+    @Override
+    public $Type$Vector deepCopy(BlockFactory blockFactory) {
+        return blockFactory.newConstant$Type$Vector(value, getPositionCount());
+    }
+
 $if(BytesRef)$
     public static long ramBytesUsed(BytesRef value) {
         return BASE_RAM_BYTES_USED + RamUsageEstimator.sizeOf(value.bytes);

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

@@ -61,6 +61,19 @@ $endif$
     @Override
     $Type$Block keepMask(BooleanVector mask);
 
+    /**
+     * Make a deep copy of this {@link Vector} using the provided {@link BlockFactory},
+     * likely copying all data.
+     */
+    @Override
+    default $Type$Vector deepCopy(BlockFactory blockFactory) {
+        try ($Type$Block.Builder builder = blockFactory.new$Type$BlockBuilder(getPositionCount())) {
+            builder.copyFrom(asBlock(), 0, getPositionCount());
+            builder.mvOrdering(Block.MvOrdering.DEDUPLICATED_AND_SORTED_ASCENDING);
+            return builder.build().asVector();
+        }
+    }
+
     @Override
     ReleasableIterator<? extends $Type$Block> lookup(IntBlock positions, ByteSizeValue targetBlockSize);
 

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

@@ -85,6 +85,11 @@ $endif$
         return vector.keepMask(mask);
     }
 
+    @Override
+    public $Type$Block deepCopy(BlockFactory blockFactory) {
+        return vector.deepCopy(blockFactory).asBlock();
+    }
+
     @Override
     public ReleasableIterator<? extends $Type$Block> lookup(IntBlock positions, ByteSizeValue targetBlockSize) {
         return vector.lookup(positions, targetBlockSize);

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

@@ -193,6 +193,7 @@ public class BasicBlockTests extends ESTestCase {
         }
         assertKeepMask(initialBlock);
         assertKeepMask(initialBlock.asVector());
+        assertDeepCopy(initialBlock);
     }
 
     public void testIntBlock() {
@@ -223,6 +224,7 @@ public class BasicBlockTests extends ESTestCase {
             assertEmptyLookup(blockFactory, block);
             assertThat(block.asVector().min(), equalTo(0));
             assertThat(block.asVector().max(), equalTo(positionCount - 1));
+            assertDeepCopy(block);
 
             try (IntBlock.Builder blockBuilder = blockFactory.newIntBlockBuilder(1)) {
                 IntBlock copy = blockBuilder.copyFrom(block, 0, block.getPositionCount()).build();
@@ -255,6 +257,7 @@ public class BasicBlockTests extends ESTestCase {
                 assertThat(vector.min(), equalTo(0));
                 assertThat(vector.max(), equalTo(positionCount - 1));
                 assertInsertNulls(vector.asBlock());
+                assertDeepCopy(vector.asBlock());
                 releaseAndAssertBreaker(vector.asBlock());
             }
         }
@@ -278,6 +281,7 @@ public class BasicBlockTests extends ESTestCase {
             assertThat(block.asVector().min(), equalTo(Integer.MAX_VALUE));
             assertThat(block.asVector().max(), equalTo(Integer.MIN_VALUE));
             assertInsertNulls(block);
+            assertDeepCopy(block);
             releaseAndAssertBreaker(block);
 
             try (IntVector.Builder vectorBuilder = blockFactory.newIntVectorBuilder(0)) {
@@ -285,6 +289,7 @@ public class BasicBlockTests extends ESTestCase {
                 assertThat(vector.min(), equalTo(Integer.MAX_VALUE));
                 assertThat(vector.max(), equalTo(Integer.MIN_VALUE));
                 assertInsertNulls(vector.asBlock());
+                assertDeepCopy(vector.asBlock());
                 releaseAndAssertBreaker(vector.asBlock());
             }
         }
@@ -325,6 +330,7 @@ public class BasicBlockTests extends ESTestCase {
             assertThat(block.asVector().min(), equalTo(value));
             assertThat(block.asVector().max(), equalTo(value));
             assertInsertNulls(block);
+            assertDeepCopy(block);
             releaseAndAssertBreaker(block);
         }
     }
@@ -359,6 +365,7 @@ public class BasicBlockTests extends ESTestCase {
                 LongBlock copy = blockBuilder.copyFrom(block, 0, block.getPositionCount()).build();
                 assertThat(copy, equalTo(block));
                 assertInsertNulls(block);
+                assertDeepCopy(block);
                 releaseAndAssertBreaker(block, copy);
             }
 
@@ -382,6 +389,7 @@ public class BasicBlockTests extends ESTestCase {
             LongVector vector = vectorBuilder.build();
             assertSingleValueDenseBlock(vector.asBlock());
             assertInsertNulls(vector.asBlock());
+            assertDeepCopy(vector.asBlock());
             releaseAndAssertBreaker(vector.asBlock());
         }
     }
@@ -419,6 +427,7 @@ public class BasicBlockTests extends ESTestCase {
             );
             assertEmptyLookup(blockFactory, block);
             assertInsertNulls(block);
+            assertDeepCopy(block);
             releaseAndAssertBreaker(block);
         }
     }
@@ -454,6 +463,7 @@ public class BasicBlockTests extends ESTestCase {
                 DoubleBlock copy = blockBuilder.copyFrom(block, 0, block.getPositionCount()).build();
                 assertThat(copy, equalTo(block));
                 assertInsertNulls(block);
+                assertDeepCopy(block);
                 releaseAndAssertBreaker(block, copy);
             }
 
@@ -479,6 +489,7 @@ public class BasicBlockTests extends ESTestCase {
                 DoubleVector vector = vectorBuilder.build();
                 assertSingleValueDenseBlock(vector.asBlock());
                 assertInsertNulls(vector.asBlock());
+                assertDeepCopy(vector.asBlock());
                 releaseAndAssertBreaker(vector.asBlock());
             }
         }
@@ -515,6 +526,7 @@ public class BasicBlockTests extends ESTestCase {
             );
             assertEmptyLookup(blockFactory, block);
             assertInsertNulls(block);
+            assertDeepCopy(block);
             releaseAndAssertBreaker(block);
         }
     }
@@ -551,6 +563,7 @@ public class BasicBlockTests extends ESTestCase {
                 FloatBlock copy = blockBuilder.copyFrom(block, 0, block.getPositionCount()).build();
                 assertThat(copy, equalTo(block));
                 assertInsertNulls(block);
+                assertDeepCopy(block);
                 releaseAndAssertBreaker(block, copy);
             }
 
@@ -576,6 +589,7 @@ public class BasicBlockTests extends ESTestCase {
                 DoubleVector vector = vectorBuilder.build();
                 assertSingleValueDenseBlock(vector.asBlock());
                 assertInsertNulls(vector.asBlock());
+                assertDeepCopy(vector.asBlock());
                 releaseAndAssertBreaker(vector.asBlock());
             }
         }
@@ -612,6 +626,7 @@ public class BasicBlockTests extends ESTestCase {
             );
             assertEmptyLookup(blockFactory, block);
             assertInsertNulls(block);
+            assertDeepCopy(block);
             releaseAndAssertBreaker(block);
         }
     }
@@ -664,6 +679,7 @@ public class BasicBlockTests extends ESTestCase {
             BytesRefBlock copy = blockBuilder.copyFrom(block, 0, block.getPositionCount()).build();
             assertThat(copy, equalTo(block));
             assertInsertNulls(block);
+            assertDeepCopy(block);
             releaseAndAssertBreaker(block, copy);
         }
 
@@ -690,6 +706,7 @@ public class BasicBlockTests extends ESTestCase {
             BytesRefVector vector = vectorBuilder.build();
             assertSingleValueDenseBlock(vector.asBlock());
             assertInsertNulls(vector.asBlock());
+            assertDeepCopy(vector.asBlock());
             releaseAndAssertBreaker(vector.asBlock());
         }
     }
@@ -746,6 +763,7 @@ public class BasicBlockTests extends ESTestCase {
             }
             assertKeepMask(block);
             assertInsertNulls(block);
+            assertDeepCopy(block);
             releaseAndAssertBreaker(block);
         }
     }
@@ -786,6 +804,7 @@ public class BasicBlockTests extends ESTestCase {
             );
             assertEmptyLookup(blockFactory, block);
             assertInsertNulls(block);
+            assertDeepCopy(block);
             releaseAndAssertBreaker(block);
         }
     }
@@ -832,6 +851,7 @@ public class BasicBlockTests extends ESTestCase {
                 BooleanBlock copy = blockBuilder.copyFrom(block, 0, block.getPositionCount()).build();
                 assertThat(copy, equalTo(block));
                 assertInsertNulls(block);
+                assertDeepCopy(block);
                 releaseAndAssertBreaker(block, copy);
             }
 
@@ -875,6 +895,7 @@ public class BasicBlockTests extends ESTestCase {
                 }
             }
             assertInsertNulls(vector.asBlock());
+            assertDeepCopy(vector.asBlock());
             releaseAndAssertBreaker(vector.asBlock());
         }
     }
@@ -917,6 +938,7 @@ public class BasicBlockTests extends ESTestCase {
                 assertTrue(block.asVector().allFalse());
             }
             assertInsertNulls(block);
+            assertDeepCopy(block);
             releaseAndAssertBreaker(block);
         }
     }
@@ -960,6 +982,7 @@ public class BasicBlockTests extends ESTestCase {
                 b -> assertThat(b, instanceOf(ConstantNullBlock.class))
             );
             assertInsertNulls(block);
+            assertDeepCopy(block);
             releaseAndAssertBreaker(block);
         }
     }
@@ -1328,18 +1351,46 @@ public class BasicBlockTests extends ESTestCase {
     void releaseAndAssertBreaker(Block... blocks) {
         assertThat(breaker.getUsed(), greaterThan(0L));
         Page[] pages = Arrays.stream(blocks).map(Page::new).toArray(Page[]::new);
+
+        /*
+         * Deep copy the block into the non-breaking instance to make
+         * sure that works and that we can read from the deep copy after
+         * this has been released.
+         */
+        Block[] deepCopies = new Block[blocks.length];
+        for (int b = 0; b < blocks.length; b++) {
+            Block copiedOutOfBreaker = blocks[b].deepCopy(TestBlockFactory.getNonBreakingInstance());
+            assertThat(copiedOutOfBreaker, equalTo(blocks[b]));
+            deepCopies[b] = copiedOutOfBreaker;
+        }
+
         Releasables.closeExpectNoException(blocks);
         Arrays.stream(blocks).forEach(block -> assertThat(block.isReleased(), is(true)));
         Arrays.stream(blocks).forEach(BasicBlockTests::assertCannotDoubleRelease);
         Arrays.stream(pages).forEach(BasicBlockTests::assertCannotReadFromPage);
         Arrays.stream(blocks).forEach(BasicBlockTests::assertCannotAddToPage);
         assertThat(breaker.getUsed(), is(0L));
+
+        for (int b = 0; b < deepCopies.length; b++) {
+            BlockTestUtils.readInto(new ArrayList<>(), deepCopies[b]);
+        }
     }
 
     void releaseAndAssertBreaker(Vector vector) {
         assertThat(breaker.getUsed(), greaterThan(0L));
+
+        /*
+         * Deep copy the vector into the non-breaking instance to make
+         * sure that works and that we can read from the deep copy after
+         * this has been released.
+         */
+        Vector copiedOutOfBreaker = vector.deepCopy(TestBlockFactory.getNonBreakingInstance());
+        assertThat(copiedOutOfBreaker, equalTo(vector));
+
         Releasables.closeExpectNoException(vector);
         assertThat(breaker.getUsed(), is(0L));
+
+        BlockTestUtils.readInto(new ArrayList<>(), copiedOutOfBreaker.asBlock());
     }
 
     static void assertCannotDoubleRelease(Block block) {
@@ -1714,6 +1765,24 @@ public class BasicBlockTests extends ESTestCase {
         }
     }
 
+    public static Block assertDeepCopy(Block block) {
+        CircuitBreaker breaker = new MockBigArrays.LimitedBreaker("esql-test-into", ByteSizeValue.ofGb(1));
+        BigArrays bigArrays = new MockBigArrays(PageCacheRecycler.NON_RECYCLING_INSTANCE, mockBreakerService(breaker));
+        BlockFactory into = new BlockFactory(bigArrays.breakerService().getBreaker(CircuitBreaker.REQUEST), bigArrays);
+        try (Block deepCopy = block.deepCopy(into)) {
+            assertThat(deepCopy, equalTo(block));
+
+            assertThat(
+                deepCopy.asVector() != null && deepCopy.asVector().isConstant(),
+                equalTo(block.asVector() != null && block.asVector().isConstant())
+            );
+        }
+        Block untracked = block.deepCopy(TestBlockFactory.getNonBreakingInstance());
+        assertThat(untracked, equalTo(block));
+        // untracked doesn't need to be released
+        return untracked;
+    }
+
     static void assertKeepMask(Vector vector) {
         int maskPositions = vector.getPositionCount();
         if (randomBoolean()) {

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

@@ -33,6 +33,7 @@ import java.util.Objects;
 import java.util.function.IntUnaryOperator;
 import java.util.stream.IntStream;
 
+import static org.elasticsearch.compute.data.BasicBlockTests.assertDeepCopy;
 import static org.elasticsearch.compute.data.BasicBlockTests.assertInsertNulls;
 import static org.elasticsearch.compute.test.BlockTestUtils.valuesAtPositions;
 import static org.hamcrest.Matchers.equalTo;
@@ -194,6 +195,16 @@ public class BlockMultiValuedTests extends ESTestCase {
         }
     }
 
+    public void testDeepCopy() {
+        int positionCount = randomIntBetween(1, 16 * 1024);
+        var b = RandomBlock.randomBlock(blockFactory(), elementType, positionCount, nullAllowed, 2, 10, 0, 0);
+        try {
+            assertDeepCopy(b.block());
+        } finally {
+            b.block().close();
+        }
+    }
+
     private void assertFiltered(boolean all, boolean shuffled) {
         int positionCount = randomIntBetween(1, 16 * 1024);
         var b = RandomBlock.randomBlock(blockFactory(), elementType, positionCount, nullAllowed, 0, 10, 0, 0);

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

@@ -14,6 +14,7 @@ import java.util.Arrays;
 import java.util.List;
 import java.util.function.Supplier;
 
+import static org.elasticsearch.compute.data.BasicBlockTests.assertDeepCopy;
 import static org.hamcrest.Matchers.equalTo;
 
 public class CompositeBlockTests extends ComputeTestCase {
@@ -85,6 +86,28 @@ public class CompositeBlockTests extends ComputeTestCase {
         }
     }
 
+    public void testDeepCopy() {
+        final BlockFactory blockFactory1 = blockFactory();
+        final BlockFactory blockFactory2 = blockFactory();
+        int numBlocks = randomIntBetween(1, 1000);
+        int positionCount = randomIntBetween(1, 1000);
+        try (
+            CompositeBlock origComposite = randomCompositeBlock(
+                blockFactory1,
+                () -> randomFrom(supportedSubElementTypes),
+                true,
+                numBlocks,
+                positionCount,
+                0,
+                between(1, 2),
+                0,
+                between(1, 2)
+            )
+        ) {
+            assertDeepCopy(origComposite);
+        }
+    }
+
     public void testTotalValueCount() {
         final BlockFactory blockFactory = blockFactory();
         int numBlocks = randomIntBetween(1, 1000);

x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/lucene/read/SortedSetOrdinalsBuilderTests.java

@@ -30,6 +30,7 @@ import java.util.Locale;
 import java.util.Map;
 import java.util.stream.IntStream;
 
+import static org.elasticsearch.compute.data.BasicBlockTests.assertDeepCopy;
 import static org.hamcrest.Matchers.equalTo;
 
 public class SortedSetOrdinalsBuilderTests extends ComputeTestCase {
@@ -56,7 +57,7 @@ public class SortedSetOrdinalsBuilderTests extends ComputeTestCase {
             Map<Integer, List<String>> expectedValues = new HashMap<>();
             List<String> allKeys = IntStream.range(0, between(1, 20)).mapToObj(n -> String.format(Locale.ROOT, "v%02d", n)).toList();
             for (int i = 0; i < numDocs; i++) {
-                List<String> subs = randomSubsetOf(allKeys).stream().sorted().toList();
+                List<String> subs = randomNonEmptySubsetOf(allKeys).stream().sorted().toList();
                 expectedValues.put(i, subs);
                 Document doc = new Document();
                 for (String v : subs) {
@@ -108,6 +109,8 @@ public class SortedSetOrdinalsBuilderTests extends ComputeTestCase {
                                     subs.add(val);
                                 }
                             }
+                            BytesRefBlock deepCopy = (BytesRefBlock) assertDeepCopy(valuesBlock);
+                            assertThat(deepCopy.asOrdinals(), equalTo(valuesBlock.asOrdinals()));
                         }
                     }
                 }

x-pack/plugin/esql/compute/test/src/main/java/org/elasticsearch/compute/test/BlockTestUtils.java

@@ -245,7 +245,7 @@ public class BlockTestUtils {
     public static Page deepCopyOf(Page page, BlockFactory blockFactory) {
         Block[] blockCopies = new Block[page.getBlockCount()];
         for (int i = 0; i < blockCopies.length; i++) {
-            blockCopies[i] = BlockUtils.deepCopyOf(page.getBlock(i), blockFactory);
+            blockCopies[i] = page.getBlock(i).deepCopy(blockFactory);
         }
         return new Page(blockCopies);
     }

x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/EsqlAsyncActionIT.java

@@ -13,7 +13,6 @@ import org.elasticsearch.action.support.master.AcknowledgedResponse;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.compute.data.Block;
 import org.elasticsearch.compute.data.BlockFactory;
-import org.elasticsearch.compute.data.BlockUtils;
 import org.elasticsearch.compute.data.Page;
 import org.elasticsearch.compute.test.TestBlockFactory;
 import org.elasticsearch.core.TimeValue;
@@ -131,7 +130,7 @@ public class EsqlAsyncActionIT extends EsqlActionIT {
     public static Page deepCopyOf(Page page, BlockFactory blockFactory) {
         Block[] blockCopies = new Block[page.getBlockCount()];
         for (int i = 0; i < blockCopies.length; i++) {
-            blockCopies[i] = BlockUtils.deepCopyOf(page.getBlock(i), blockFactory);
+            blockCopies[i] = page.getBlock(i).deepCopy(blockFactory);
         }
         return new Page(blockCopies);
     }

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plan/logical/local/CopyingLocalSupplier.java

@@ -11,7 +11,6 @@ import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.compute.data.Block;
-import org.elasticsearch.compute.data.BlockUtils;
 import org.elasticsearch.xpack.esql.optimizer.rules.logical.ReplaceRowAsLocalRelation;
 import org.elasticsearch.xpack.esql.plan.logical.InlineStats;
 import org.elasticsearch.xpack.esql.planner.PlannerUtils;
@@ -74,7 +73,7 @@ public class CopyingLocalSupplier implements LocalSupplier {
     public Block[] get() {
         Block[] blockCopies = new Block[delegate.blocks.length];
         for (int i = 0; i < blockCopies.length; i++) {
-            blockCopies[i] = BlockUtils.deepCopyOf(delegate.blocks[i], PlannerUtils.NON_BREAKING_BLOCK_FACTORY);
+            blockCopies[i] = delegate.blocks[i].deepCopy(PlannerUtils.NON_BREAKING_BLOCK_FACTORY);
         }
         return blockCopies;
     }