ESQL: Use blockFactory more in aggs (#100017)

This commit updates the hash grouping operator to close input pages, as well as use the block factory for internally created blocks.

Additionally:

 * Adds a MockBlockFactory to help with tracking block creation
 * Eagerly creates the block view of a vector, which helps with tracking since there can be only one block view instance per vector
 * Resolves an issue with Filter Blocks, whereby they previously tried to emit their contents in toString

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

@@ -144,7 +144,7 @@ private static Operator operator(String grouping, String op, String dataType) {
         DriverContext driverContext = driverContext();
         return new HashAggregationOperator(
             List.of(supplier(op, dataType, groups.size()).groupingAggregatorFactory(AggregatorMode.SINGLE)),
-            () -> BlockHash.build(groups, BIG_ARRAYS, 16 * 1024, false),
+            () -> BlockHash.build(groups, driverContext, 16 * 1024, false),
             driverContext
         );
     }

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

@@ -21,18 +21,21 @@ public final class BooleanArrayVector extends AbstractVector implements BooleanV
 
     private final boolean[] values;
 
+    private final BooleanBlock block;
+
     public BooleanArrayVector(boolean[] values, int positionCount) {
         this(values, positionCount, BlockFactory.getNonBreakingInstance());
     }
 
     public BooleanArrayVector(boolean[] values, int positionCount, BlockFactory blockFactory) {
         super(positionCount, blockFactory);
         this.values = values;
+        this.block = new BooleanVectorBlock(this);
     }
 
     @Override
     public BooleanBlock asBlock() {
-        return new BooleanVectorBlock(this);
+        return block;
     }
 
     @Override

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

@@ -21,18 +21,21 @@ public final class BooleanBigArrayVector extends AbstractVector implements Boole
 
     private final BitArray values;
 
+    private final BooleanBlock block;
+
     public BooleanBigArrayVector(BitArray values, int positionCount) {
         this(values, positionCount, BlockFactory.getNonBreakingInstance());
     }
 
     public BooleanBigArrayVector(BitArray values, int positionCount, BlockFactory blockFactory) {
         super(positionCount, blockFactory);
         this.values = values;
+        this.block = new BooleanVectorBlock(this);
     }
 
     @Override
     public BooleanBlock asBlock() {
-        return new BooleanVectorBlock(this);
+        return block;
     }
 
     @Override

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

@@ -55,10 +55,14 @@ public BooleanVector build() {
             throw new IllegalStateException("expected to write [" + values.length + "] entries but wrote [" + nextIndex + "]");
         }
         nextIndex = -1;
+        BooleanVector vector;
         if (values.length == 1) {
-            return blockFactory.newConstantBooleanBlockWith(values[0], 1, preAdjustedBytes).asVector();
+            vector = blockFactory.newConstantBooleanBlockWith(values[0], 1, preAdjustedBytes).asVector();
+        } else {
+            vector = blockFactory.newBooleanArrayVector(values, values.length, preAdjustedBytes);
         }
-        return blockFactory.newBooleanArrayVector(values, values.length, preAdjustedBytes);
+        assert vector.ramBytesUsed() == preAdjustedBytes : "fixed Builders should estimate the exact ram bytes used";
+        return vector;
     }
 
     @Override
@@ -68,4 +72,8 @@ public void close() {
             blockFactory.adjustBreaker(-preAdjustedBytes, false);
         }
     }
+
+    boolean isReleased() {
+        return nextIndex < 0;
+    }
 }

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

@@ -22,18 +22,21 @@ public final class BytesRefArrayVector extends AbstractVector implements BytesRe
 
     private final BytesRefArray values;
 
+    private final BytesRefBlock block;
+
     public BytesRefArrayVector(BytesRefArray values, int positionCount) {
         this(values, positionCount, BlockFactory.getNonBreakingInstance());
     }
 
     public BytesRefArrayVector(BytesRefArray values, int positionCount, BlockFactory blockFactory) {
         super(positionCount, blockFactory);
         this.values = values;
+        this.block = new BytesRefVectorBlock(this);
     }
 
     @Override
     public BytesRefBlock asBlock() {
-        return new BytesRefVectorBlock(this);
+        return block;
     }
 
     @Override

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

@@ -19,13 +19,16 @@ public final class ConstantBooleanVector extends AbstractVector implements Boole
 
     private final boolean value;
 
+    private final BooleanBlock block;
+
     public ConstantBooleanVector(boolean value, int positionCount) {
         this(value, positionCount, BlockFactory.getNonBreakingInstance());
     }
 
     public ConstantBooleanVector(boolean value, int positionCount, BlockFactory blockFactory) {
         super(positionCount, blockFactory);
         this.value = value;
+        this.block = new BooleanVectorBlock(this);
     }
 
     @Override
@@ -35,7 +38,7 @@ public boolean getBoolean(int position) {
 
     @Override
     public BooleanBlock asBlock() {
-        return new BooleanVectorBlock(this);
+        return block;
     }
 
     @Override

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

@@ -20,13 +20,16 @@ public final class ConstantBytesRefVector extends AbstractVector implements Byte
         .shallowSizeOfInstance(BytesRef.class);
     private final BytesRef value;
 
+    private final BytesRefBlock block;
+
     public ConstantBytesRefVector(BytesRef value, int positionCount) {
         this(value, positionCount, BlockFactory.getNonBreakingInstance());
     }
 
     public ConstantBytesRefVector(BytesRef value, int positionCount, BlockFactory blockFactory) {
         super(positionCount, blockFactory);
         this.value = value;
+        this.block = new BytesRefVectorBlock(this);
     }
 
     @Override
@@ -36,7 +39,7 @@ public BytesRef getBytesRef(int position, BytesRef ignore) {
 
     @Override
     public BytesRefBlock asBlock() {
-        return new BytesRefVectorBlock(this);
+        return block;
     }
 
     @Override

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

@@ -19,13 +19,16 @@ public final class ConstantDoubleVector extends AbstractVector implements Double
 
     private final double value;
 
+    private final DoubleBlock block;
+
     public ConstantDoubleVector(double value, int positionCount) {
         this(value, positionCount, BlockFactory.getNonBreakingInstance());
     }
 
     public ConstantDoubleVector(double value, int positionCount, BlockFactory blockFactory) {
         super(positionCount, blockFactory);
         this.value = value;
+        this.block = new DoubleVectorBlock(this);
     }
 
     @Override
@@ -35,7 +38,7 @@ public double getDouble(int position) {
 
     @Override
     public DoubleBlock asBlock() {
-        return new DoubleVectorBlock(this);
+        return block;
     }
 
     @Override

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

@@ -19,13 +19,16 @@ public final class ConstantIntVector extends AbstractVector implements IntVector
 
     private final int value;
 
+    private final IntBlock block;
+
     public ConstantIntVector(int value, int positionCount) {
         this(value, positionCount, BlockFactory.getNonBreakingInstance());
     }
 
     public ConstantIntVector(int value, int positionCount, BlockFactory blockFactory) {
         super(positionCount, blockFactory);
         this.value = value;
+        this.block = new IntVectorBlock(this);
     }
 
     @Override
@@ -35,7 +38,7 @@ public int getInt(int position) {
 
     @Override
     public IntBlock asBlock() {
-        return new IntVectorBlock(this);
+        return block;
     }
 
     @Override

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

@@ -19,13 +19,16 @@ public final class ConstantLongVector extends AbstractVector implements LongVect
 
     private final long value;
 
+    private final LongBlock block;
+
     public ConstantLongVector(long value, int positionCount) {
         this(value, positionCount, BlockFactory.getNonBreakingInstance());
     }
 
     public ConstantLongVector(long value, int positionCount, BlockFactory blockFactory) {
         super(positionCount, blockFactory);
         this.value = value;
+        this.block = new LongVectorBlock(this);
     }
 
     @Override
@@ -35,7 +38,7 @@ public long getLong(int position) {
 
     @Override
     public LongBlock asBlock() {
-        return new LongVectorBlock(this);
+        return block;
     }
 
     @Override

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

@@ -21,18 +21,21 @@ public final class DoubleArrayVector extends AbstractVector implements DoubleVec
 
     private final double[] values;
 
+    private final DoubleBlock block;
+
     public DoubleArrayVector(double[] values, int positionCount) {
         this(values, positionCount, BlockFactory.getNonBreakingInstance());
     }
 
     public DoubleArrayVector(double[] values, int positionCount, BlockFactory blockFactory) {
         super(positionCount, blockFactory);
         this.values = values;
+        this.block = new DoubleVectorBlock(this);
     }
 
     @Override
     public DoubleBlock asBlock() {
-        return new DoubleVectorBlock(this);
+        return block;
     }
 
     @Override

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

@@ -21,18 +21,21 @@ public final class DoubleBigArrayVector extends AbstractVector implements Double
 
     private final DoubleArray values;
 
+    private final DoubleBlock block;
+
     public DoubleBigArrayVector(DoubleArray values, int positionCount) {
         this(values, positionCount, BlockFactory.getNonBreakingInstance());
     }
 
     public DoubleBigArrayVector(DoubleArray values, int positionCount, BlockFactory blockFactory) {
         super(positionCount, blockFactory);
         this.values = values;
+        this.block = new DoubleVectorBlock(this);
     }
 
     @Override
     public DoubleBlock asBlock() {
-        return new DoubleVectorBlock(this);
+        return block;
     }
 
     @Override

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

@@ -55,10 +55,14 @@ public DoubleVector build() {
             throw new IllegalStateException("expected to write [" + values.length + "] entries but wrote [" + nextIndex + "]");
         }
         nextIndex = -1;
+        DoubleVector vector;
         if (values.length == 1) {
-            return blockFactory.newConstantDoubleBlockWith(values[0], 1, preAdjustedBytes).asVector();
+            vector = blockFactory.newConstantDoubleBlockWith(values[0], 1, preAdjustedBytes).asVector();
+        } else {
+            vector = blockFactory.newDoubleArrayVector(values, values.length, preAdjustedBytes);
         }
-        return blockFactory.newDoubleArrayVector(values, values.length, preAdjustedBytes);
+        assert vector.ramBytesUsed() == preAdjustedBytes : "fixed Builders should estimate the exact ram bytes used";
+        return vector;
     }
 
     @Override
@@ -68,4 +72,8 @@ public void close() {
             blockFactory.adjustBreaker(-preAdjustedBytes, false);
         }
     }
+
+    boolean isReleased() {
+        return nextIndex < 0;
+    }
 }

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

@@ -94,9 +94,14 @@ public int hashCode() {
     public String toString() {
         StringBuilder sb = new StringBuilder();
         sb.append(this.getClass().getSimpleName());
-        sb.append("[positions=" + getPositionCount() + ", values=[");
-        appendValues(sb);
-        sb.append("]]");
+        sb.append("[positions=" + getPositionCount());
+        sb.append(", released=" + isReleased());
+        if (isReleased() == false) {
+            sb.append(", values=[");
+            appendValues(sb);
+            sb.append("]");
+        }
+        sb.append("]");
         return sb.toString();
     }
 

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

@@ -20,9 +20,12 @@ public final class FilterBooleanVector extends AbstractFilterVector implements B
 
     private final BooleanVector vector;
 
+    private final BooleanBlock block;
+
     FilterBooleanVector(BooleanVector vector, int... positions) {
         super(positions, vector.blockFactory());
         this.vector = vector;
+        this.block = new BooleanVectorBlock(this);
     }
 
     @Override
@@ -32,7 +35,7 @@ public boolean getBoolean(int position) {
 
     @Override
     public BooleanBlock asBlock() {
-        return new BooleanVectorBlock(this);
+        return block;
     }
 
     @Override

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

@@ -97,9 +97,14 @@ public int hashCode() {
     public String toString() {
         StringBuilder sb = new StringBuilder();
         sb.append(this.getClass().getSimpleName());
-        sb.append("[positions=" + getPositionCount() + ", values=[");
-        appendValues(sb);
-        sb.append("]]");
+        sb.append("[positions=" + getPositionCount());
+        sb.append(", released=" + isReleased());
+        if (isReleased() == false) {
+            sb.append(", values=[");
+            appendValues(sb);
+            sb.append("]");
+        }
+        sb.append("]");
         return sb.toString();
     }
 

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

@@ -21,9 +21,12 @@ public final class FilterBytesRefVector extends AbstractFilterVector implements
 
     private final BytesRefVector vector;
 
+    private final BytesRefBlock block;
+
     FilterBytesRefVector(BytesRefVector vector, int... positions) {
         super(positions, vector.blockFactory());
         this.vector = vector;
+        this.block = new BytesRefVectorBlock(this);
     }
 
     @Override
@@ -33,7 +36,7 @@ public BytesRef getBytesRef(int position, BytesRef dest) {
 
     @Override
     public BytesRefBlock asBlock() {
-        return new BytesRefVectorBlock(this);
+        return block;
     }
 
     @Override