diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/AbstractTopBytesRefAggregatorFunctionTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/AbstractTopBytesRefAggregatorFunctionTests.java
index 2815dd70e8124..f02b0a8074d7d 100644
--- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/AbstractTopBytesRefAggregatorFunctionTests.java
+++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/AbstractTopBytesRefAggregatorFunctionTests.java
@@ -11,6 +11,7 @@
 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.operator.SequenceBytesRefBlockSourceOperator;
 import org.elasticsearch.compute.operator.SourceOperator;
 
@@ -30,8 +31,8 @@ protected final SourceOperator simpleInput(BlockFactory blockFactory, int size)
     protected abstract BytesRef randomValue();
 
     @Override
-    public final void assertSimpleOutput(List<Block> input, Block result) {
-        Object[] values = input.stream().flatMap(AggregatorFunctionTestCase::allBytesRefs).sorted().limit(LIMIT).toArray(Object[]::new);
+    public final void assertSimpleOutput(List<Page> input, Block result) {
+        Object[] values = input.stream().flatMap(p -> allBytesRefs(p.getBlock(0))).sorted().limit(LIMIT).toArray(Object[]::new);
         assertThat((List<?>) BlockUtils.toJavaObject(result, 0), contains(values));
     }
 }
diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/AggregatorFunctionTestCase.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/AggregatorFunctionTestCase.java
index ceb6978dc81e8..c1916b0df3304 100644
--- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/AggregatorFunctionTestCase.java
+++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/AggregatorFunctionTestCase.java
@@ -58,7 +58,12 @@ protected final int aggregatorIntermediateBlockCount() {
 
     protected abstract String expectedDescriptionOfAggregator();
 
-    protected abstract void assertSimpleOutput(List<Block> input, Block result);
+    /**
+     * Assert that the result is correct given the input.
+     * @param input the input pages build by {@link #simpleInput}
+     * @param result the result of running {@link #aggregatorFunction()}
+     */
+    protected abstract void assertSimpleOutput(List<Page> input, Block result);
 
     @Override
     protected Operator.OperatorFactory simpleWithMode(SimpleOptions options, AggregatorMode mode) {
@@ -99,7 +104,7 @@ protected final void assertSimpleOutput(List<Page> input, List<Page> results) {
         assertThat(results.get(0).getPositionCount(), equalTo(1));
 
         Block result = results.get(0).getBlock(0);
-        assertSimpleOutput(input.stream().map(p -> p.<Block>getBlock(0)).toList(), result);
+        assertSimpleOutput(input, result);
     }
 
     public final void testIgnoresNulls() {
diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/CountAggregatorFunctionTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/CountAggregatorFunctionTests.java
index e30082c843b19..8183b9c64baed 100644
--- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/CountAggregatorFunctionTests.java
+++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/CountAggregatorFunctionTests.java
@@ -10,6 +10,7 @@
 import org.elasticsearch.compute.data.Block;
 import org.elasticsearch.compute.data.BlockFactory;
 import org.elasticsearch.compute.data.LongBlock;
+import org.elasticsearch.compute.data.Page;
 import org.elasticsearch.compute.operator.SourceOperator;
 import org.elasticsearch.compute.test.SequenceLongBlockSourceOperator;
 
@@ -36,8 +37,8 @@ protected String expectedDescriptionOfAggregator() {
     }
 
     @Override
-    protected void assertSimpleOutput(List<Block> input, Block result) {
-        long count = input.stream().flatMapToLong(b -> allLongs(b)).count();
+    protected void assertSimpleOutput(List<Page> input, Block result) {
+        long count = input.stream().flatMapToLong(p -> allLongs(p.getBlock(0))).count();
         assertThat(((LongBlock) result).getLong(0), equalTo(count));
     }
 
diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/CountDistinctBooleanAggregatorFunctionTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/CountDistinctBooleanAggregatorFunctionTests.java
index d91fdce409835..03bd60644de63 100644
--- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/CountDistinctBooleanAggregatorFunctionTests.java
+++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/CountDistinctBooleanAggregatorFunctionTests.java
@@ -10,6 +10,7 @@
 import org.elasticsearch.compute.data.Block;
 import org.elasticsearch.compute.data.BlockFactory;
 import org.elasticsearch.compute.data.LongBlock;
+import org.elasticsearch.compute.data.Page;
 import org.elasticsearch.compute.operator.SequenceBooleanBlockSourceOperator;
 import org.elasticsearch.compute.operator.SourceOperator;
 
@@ -36,8 +37,8 @@ protected String expectedDescriptionOfAggregator() {
     }
 
     @Override
-    protected void assertSimpleOutput(List<Block> input, Block result) {
-        long expected = input.stream().flatMap(b -> allBooleans(b)).distinct().count();
+    protected void assertSimpleOutput(List<Page> input, Block result) {
+        long expected = input.stream().flatMap(p -> allBooleans(p.getBlock(0))).distinct().count();
         long count = ((LongBlock) result).getLong(0);
         assertThat(count, equalTo(expected));
     }
diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/CountDistinctBytesRefAggregatorFunctionTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/CountDistinctBytesRefAggregatorFunctionTests.java
index d01cbb39bf470..34b95cff77942 100644
--- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/CountDistinctBytesRefAggregatorFunctionTests.java
+++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/CountDistinctBytesRefAggregatorFunctionTests.java
@@ -11,6 +11,7 @@
 import org.elasticsearch.compute.data.Block;
 import org.elasticsearch.compute.data.BlockFactory;
 import org.elasticsearch.compute.data.LongBlock;
+import org.elasticsearch.compute.data.Page;
 import org.elasticsearch.compute.operator.BytesRefBlockSourceOperator;
 import org.elasticsearch.compute.operator.SourceOperator;
 
@@ -42,8 +43,8 @@ protected String expectedDescriptionOfAggregator() {
     }
 
     @Override
-    protected void assertSimpleOutput(List<Block> input, Block result) {
-        long expected = input.stream().flatMap(b -> allBytesRefs(b)).distinct().count();
+    protected void assertSimpleOutput(List<Page> input, Block result) {
+        long expected = input.stream().flatMap(p -> allBytesRefs(p.getBlock(0))).distinct().count();
         long count = ((LongBlock) result).getLong(0);
         // HLL is an approximation algorithm and precision depends on the number of values computed and the precision_threshold param
         // https://www.elastic.co/guide/en/elasticsearch/reference/current/search-aggregations-metrics-cardinality-aggregation.html
diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/CountDistinctDoubleAggregatorFunctionTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/CountDistinctDoubleAggregatorFunctionTests.java
index d1e845fba40ca..e6dae67ba6f10 100644
--- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/CountDistinctDoubleAggregatorFunctionTests.java
+++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/CountDistinctDoubleAggregatorFunctionTests.java
@@ -10,6 +10,7 @@
 import org.elasticsearch.compute.data.Block;
 import org.elasticsearch.compute.data.BlockFactory;
 import org.elasticsearch.compute.data.LongBlock;
+import org.elasticsearch.compute.data.Page;
 import org.elasticsearch.compute.operator.SequenceDoubleBlockSourceOperator;
 import org.elasticsearch.compute.operator.SourceOperator;
 import org.elasticsearch.test.ESTestCase;
@@ -38,8 +39,8 @@ protected String expectedDescriptionOfAggregator() {
     }
 
     @Override
-    protected void assertSimpleOutput(List<Block> input, Block result) {
-        long expected = input.stream().flatMapToDouble(b -> allDoubles(b)).distinct().count();
+    protected void assertSimpleOutput(List<Page> input, Block result) {
+        long expected = input.stream().flatMapToDouble(p -> allDoubles(p.getBlock(0))).distinct().count();
 
         long count = ((LongBlock) result).getLong(0);
         // HLL is an approximation algorithm and precision depends on the number of values computed and the precision_threshold param
diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/CountDistinctFloatAggregatorFunctionTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/CountDistinctFloatAggregatorFunctionTests.java
index 7c0d7c1e3d2fd..0ee3dfc9cd7b0 100644
--- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/CountDistinctFloatAggregatorFunctionTests.java
+++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/CountDistinctFloatAggregatorFunctionTests.java
@@ -10,6 +10,7 @@
 import org.elasticsearch.compute.data.Block;
 import org.elasticsearch.compute.data.BlockFactory;
 import org.elasticsearch.compute.data.LongBlock;
+import org.elasticsearch.compute.data.Page;
 import org.elasticsearch.compute.operator.SequenceFloatBlockSourceOperator;
 import org.elasticsearch.compute.operator.SourceOperator;
 import org.elasticsearch.test.ESTestCase;
@@ -38,8 +39,8 @@ protected String expectedDescriptionOfAggregator() {
     }
 
     @Override
-    protected void assertSimpleOutput(List<Block> input, Block result) {
-        long expected = input.stream().flatMap(AggregatorFunctionTestCase::allFloats).distinct().count();
+    protected void assertSimpleOutput(List<Page> input, Block result) {
+        long expected = input.stream().flatMap(p -> allFloats(p.getBlock(0))).distinct().count();
 
         long count = ((LongBlock) result).getLong(0);
         // HLL is an approximation algorithm and precision depends on the number of values computed and the precision_threshold param
diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/CountDistinctIntAggregatorFunctionTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/CountDistinctIntAggregatorFunctionTests.java
index 5259331a7a81f..f3bbf41ac66db 100644
--- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/CountDistinctIntAggregatorFunctionTests.java
+++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/CountDistinctIntAggregatorFunctionTests.java
@@ -45,8 +45,8 @@ protected String expectedDescriptionOfAggregator() {
     }
 
     @Override
-    protected void assertSimpleOutput(List<Block> input, Block result) {
-        long expected = input.stream().flatMapToInt(b -> allInts(b)).distinct().count();
+    protected void assertSimpleOutput(List<Page> input, Block result) {
+        long expected = input.stream().flatMapToInt(p -> allInts(p.getBlock(0))).distinct().count();
 
         long count = ((LongBlock) result).getLong(0);
         // HLL is an approximation algorithm and precision depends on the number of values computed and the precision_threshold param
diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/CountDistinctLongAggregatorFunctionTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/CountDistinctLongAggregatorFunctionTests.java
index a9aeeda7e04d6..4bfc8406c6c21 100644
--- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/CountDistinctLongAggregatorFunctionTests.java
+++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/CountDistinctLongAggregatorFunctionTests.java
@@ -46,8 +46,8 @@ protected String expectedDescriptionOfAggregator() {
     }
 
     @Override
-    protected void assertSimpleOutput(List<Block> input, Block result) {
-        long expected = input.stream().flatMapToLong(b -> allLongs(b)).distinct().count();
+    protected void assertSimpleOutput(List<Page> input, Block result) {
+        long expected = input.stream().flatMapToLong(p -> allLongs(p.getBlock(0))).distinct().count();
         long count = ((LongBlock) result).getLong(0);
 
         // HLL is an approximation algorithm and precision depends on the number of values computed and the precision_threshold param
diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/FilteredAggregatorFunctionTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/FilteredAggregatorFunctionTests.java
index a4411d92c6c29..bff4b8ea54fdc 100644
--- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/FilteredAggregatorFunctionTests.java
+++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/FilteredAggregatorFunctionTests.java
@@ -46,10 +46,10 @@ protected String expectedToStringOfSimpleAggregator() {
     }
 
     @Override
-    protected void assertSimpleOutput(List<Block> input, Block result) {
+    protected void assertSimpleOutput(List<Page> input, Block result) {
         long sum = 0;
-        for (Block block : input) {
-            IntBlock ints = (IntBlock) block;
+        for (Page page : input) {
+            IntBlock ints = page.getBlock(0);
             for (int p = 0; p < ints.getPositionCount(); p++) {
                 /*
                  * Perform the sum on the values *only* if any of the
diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MaxBooleanAggregatorFunctionTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MaxBooleanAggregatorFunctionTests.java
index a7164740af009..aeffaddc0d1ae 100644
--- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MaxBooleanAggregatorFunctionTests.java
+++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MaxBooleanAggregatorFunctionTests.java
@@ -10,6 +10,7 @@
 import org.elasticsearch.compute.data.Block;
 import org.elasticsearch.compute.data.BlockFactory;
 import org.elasticsearch.compute.data.BooleanBlock;
+import org.elasticsearch.compute.data.Page;
 import org.elasticsearch.compute.operator.SequenceBooleanBlockSourceOperator;
 import org.elasticsearch.compute.operator.SourceOperator;
 
@@ -36,8 +37,8 @@ protected String expectedDescriptionOfAggregator() {
     }
 
     @Override
-    public void assertSimpleOutput(List<Block> input, Block result) {
-        Boolean max = input.stream().flatMap(b -> allBooleans(b)).max(Comparator.naturalOrder()).get();
+    public void assertSimpleOutput(List<Page> input, Block result) {
+        Boolean max = input.stream().flatMap(p -> allBooleans(p.getBlock(0))).max(Comparator.naturalOrder()).get();
         assertThat(((BooleanBlock) result).getBoolean(0), equalTo(max));
     }
 }
diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MaxBytesRefAggregatorFunctionTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MaxBytesRefAggregatorFunctionTests.java
index 54b82dcbc5008..23568f8c47ec3 100644
--- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MaxBytesRefAggregatorFunctionTests.java
+++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MaxBytesRefAggregatorFunctionTests.java
@@ -11,6 +11,7 @@
 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.operator.SequenceBytesRefBlockSourceOperator;
 import org.elasticsearch.compute.operator.SourceOperator;
 
@@ -41,8 +42,8 @@ protected String expectedDescriptionOfAggregator() {
     }
 
     @Override
-    public void assertSimpleOutput(List<Block> input, Block result) {
-        Optional<BytesRef> max = input.stream().flatMap(b -> allBytesRefs(b)).max(Comparator.naturalOrder());
+    public void assertSimpleOutput(List<Page> input, Block result) {
+        Optional<BytesRef> max = input.stream().flatMap(p -> allBytesRefs(p.getBlock(0))).max(Comparator.naturalOrder());
         if (max.isEmpty()) {
             assertThat(result.isNull(0), equalTo(true));
             return;
diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MaxDoubleAggregatorFunctionTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MaxDoubleAggregatorFunctionTests.java
index fee5950c08257..d85b992e16af3 100644
--- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MaxDoubleAggregatorFunctionTests.java
+++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MaxDoubleAggregatorFunctionTests.java
@@ -10,6 +10,7 @@
 import org.elasticsearch.compute.data.Block;
 import org.elasticsearch.compute.data.BlockFactory;
 import org.elasticsearch.compute.data.DoubleBlock;
+import org.elasticsearch.compute.data.Page;
 import org.elasticsearch.compute.operator.SequenceDoubleBlockSourceOperator;
 import org.elasticsearch.compute.operator.SourceOperator;
 import org.elasticsearch.test.ESTestCase;
@@ -36,8 +37,8 @@ protected String expectedDescriptionOfAggregator() {
     }
 
     @Override
-    public void assertSimpleOutput(List<Block> input, Block result) {
-        double max = input.stream().flatMapToDouble(b -> allDoubles(b)).max().getAsDouble();
+    public void assertSimpleOutput(List<Page> input, Block result) {
+        double max = input.stream().flatMapToDouble(p -> allDoubles(p.getBlock(0))).max().getAsDouble();
         assertThat(((DoubleBlock) result).getDouble(0), equalTo(max));
     }
 }
diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MaxFloatAggregatorFunctionTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MaxFloatAggregatorFunctionTests.java
index a1f13566a069a..f66ea92f72cfa 100644
--- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MaxFloatAggregatorFunctionTests.java
+++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MaxFloatAggregatorFunctionTests.java
@@ -10,6 +10,7 @@
 import org.elasticsearch.compute.data.Block;
 import org.elasticsearch.compute.data.BlockFactory;
 import org.elasticsearch.compute.data.FloatBlock;
+import org.elasticsearch.compute.data.Page;
 import org.elasticsearch.compute.operator.SequenceFloatBlockSourceOperator;
 import org.elasticsearch.compute.operator.SourceOperator;
 import org.elasticsearch.test.ESTestCase;
@@ -36,8 +37,8 @@ protected String expectedDescriptionOfAggregator() {
     }
 
     @Override
-    public void assertSimpleOutput(List<Block> input, Block result) {
-        Float max = input.stream().flatMap(AggregatorFunctionTestCase::allFloats).max(floatComparator()).get();
+    public void assertSimpleOutput(List<Page> input, Block result) {
+        Float max = input.stream().flatMap(p -> allFloats(p.getBlock(0))).max(floatComparator()).get();
         assertThat(((FloatBlock) result).getFloat(0), equalTo(max));
     }
 }
diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MaxIntAggregatorFunctionTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MaxIntAggregatorFunctionTests.java
index 5507e2c261e97..52c28b080a624 100644
--- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MaxIntAggregatorFunctionTests.java
+++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MaxIntAggregatorFunctionTests.java
@@ -10,6 +10,7 @@
 import org.elasticsearch.compute.data.Block;
 import org.elasticsearch.compute.data.BlockFactory;
 import org.elasticsearch.compute.data.IntBlock;
+import org.elasticsearch.compute.data.Page;
 import org.elasticsearch.compute.operator.SequenceIntBlockSourceOperator;
 import org.elasticsearch.compute.operator.SourceOperator;
 
@@ -35,8 +36,8 @@ protected String expectedDescriptionOfAggregator() {
     }
 
     @Override
-    public void assertSimpleOutput(List<Block> input, Block result) {
-        int max = input.stream().flatMapToInt(b -> allInts(b)).max().getAsInt();
+    public void assertSimpleOutput(List<Page> input, Block result) {
+        int max = input.stream().flatMapToInt(p -> allInts(p.getBlock(0))).max().getAsInt();
         assertThat(((IntBlock) result).getInt(0), equalTo(max));
     }
 }
diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MaxIpAggregatorFunctionTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MaxIpAggregatorFunctionTests.java
index b39b5fe384961..5607bdd2f71e6 100644
--- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MaxIpAggregatorFunctionTests.java
+++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MaxIpAggregatorFunctionTests.java
@@ -12,6 +12,7 @@
 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.operator.SequenceBytesRefBlockSourceOperator;
 import org.elasticsearch.compute.operator.SourceOperator;
 
@@ -42,8 +43,8 @@ protected String expectedDescriptionOfAggregator() {
     }
 
     @Override
-    public void assertSimpleOutput(List<Block> input, Block result) {
-        Optional<BytesRef> max = input.stream().flatMap(b -> allBytesRefs(b)).max(Comparator.naturalOrder());
+    public void assertSimpleOutput(List<Page> input, Block result) {
+        Optional<BytesRef> max = input.stream().flatMap(p -> allBytesRefs(p.getBlock(0))).max(Comparator.naturalOrder());
         if (max.isEmpty()) {
             assertThat(result.isNull(0), equalTo(true));
             return;
diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MaxLongAggregatorFunctionTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MaxLongAggregatorFunctionTests.java
index 081ef44f37047..eb7be07b35bfd 100644
--- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MaxLongAggregatorFunctionTests.java
+++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MaxLongAggregatorFunctionTests.java
@@ -10,6 +10,7 @@
 import org.elasticsearch.compute.data.Block;
 import org.elasticsearch.compute.data.BlockFactory;
 import org.elasticsearch.compute.data.LongBlock;
+import org.elasticsearch.compute.data.Page;
 import org.elasticsearch.compute.operator.SourceOperator;
 import org.elasticsearch.compute.test.SequenceLongBlockSourceOperator;
 
@@ -36,8 +37,8 @@ protected String expectedDescriptionOfAggregator() {
     }
 
     @Override
-    public void assertSimpleOutput(List<Block> input, Block result) {
-        long max = input.stream().flatMapToLong(b -> allLongs(b)).max().getAsLong();
+    public void assertSimpleOutput(List<Page> input, Block result) {
+        long max = input.stream().flatMapToLong(p -> allLongs(p.getBlock(0))).max().getAsLong();
         assertThat(((LongBlock) result).getLong(0), equalTo(max));
     }
 }
diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MedianAbsoluteDeviationDoubleAggregatorFunctionTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MedianAbsoluteDeviationDoubleAggregatorFunctionTests.java
index db9ab2c998103..3da6961dba3db 100644
--- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MedianAbsoluteDeviationDoubleAggregatorFunctionTests.java
+++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MedianAbsoluteDeviationDoubleAggregatorFunctionTests.java
@@ -11,6 +11,7 @@
 import org.elasticsearch.compute.data.Block;
 import org.elasticsearch.compute.data.BlockFactory;
 import org.elasticsearch.compute.data.DoubleBlock;
+import org.elasticsearch.compute.data.Page;
 import org.elasticsearch.compute.operator.SequenceDoubleBlockSourceOperator;
 import org.elasticsearch.compute.operator.SourceOperator;
 
@@ -39,7 +40,7 @@ protected String expectedDescriptionOfAggregator() {
     }
 
     @Override
-    protected void assertSimpleOutput(List<Block> input, Block result) {
+    protected void assertSimpleOutput(List<Page> input, Block result) {
         assertThat(((DoubleBlock) result).getDouble(0), equalTo(0.8));
     }
 }
diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MedianAbsoluteDeviationFloatAggregatorFunctionTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MedianAbsoluteDeviationFloatAggregatorFunctionTests.java
index fb70fa6385d74..83465ea5b9379 100644
--- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MedianAbsoluteDeviationFloatAggregatorFunctionTests.java
+++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MedianAbsoluteDeviationFloatAggregatorFunctionTests.java
@@ -11,6 +11,7 @@
 import org.elasticsearch.compute.data.Block;
 import org.elasticsearch.compute.data.BlockFactory;
 import org.elasticsearch.compute.data.DoubleBlock;
+import org.elasticsearch.compute.data.Page;
 import org.elasticsearch.compute.operator.SequenceFloatBlockSourceOperator;
 import org.elasticsearch.compute.operator.SourceOperator;
 
@@ -39,7 +40,7 @@ protected String expectedDescriptionOfAggregator() {
     }
 
     @Override
-    protected void assertSimpleOutput(List<Block> input, Block result) {
+    protected void assertSimpleOutput(List<Page> input, Block result) {
         assertThat(((DoubleBlock) result).getDouble(0), closeTo(0.8, 0.001d));
     }
 }
diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MedianAbsoluteDeviationIntAggregatorFunctionTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MedianAbsoluteDeviationIntAggregatorFunctionTests.java
index fca7ec47b05a5..a0cd315a30fbf 100644
--- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MedianAbsoluteDeviationIntAggregatorFunctionTests.java
+++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MedianAbsoluteDeviationIntAggregatorFunctionTests.java
@@ -11,6 +11,7 @@
 import org.elasticsearch.compute.data.Block;
 import org.elasticsearch.compute.data.BlockFactory;
 import org.elasticsearch.compute.data.DoubleBlock;
+import org.elasticsearch.compute.data.Page;
 import org.elasticsearch.compute.operator.SequenceIntBlockSourceOperator;
 import org.elasticsearch.compute.operator.SourceOperator;
 
@@ -39,7 +40,7 @@ protected String expectedDescriptionOfAggregator() {
     }
 
     @Override
-    protected void assertSimpleOutput(List<Block> input, Block result) {
+    protected void assertSimpleOutput(List<Page> input, Block result) {
         assertThat(((DoubleBlock) result).getDouble(0), equalTo(23.0));
     }
 }
diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MedianAbsoluteDeviationLongAggregatorFunctionTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MedianAbsoluteDeviationLongAggregatorFunctionTests.java
index f700d4270f4d7..05a12001a4865 100644
--- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MedianAbsoluteDeviationLongAggregatorFunctionTests.java
+++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MedianAbsoluteDeviationLongAggregatorFunctionTests.java
@@ -11,6 +11,7 @@
 import org.elasticsearch.compute.data.Block;
 import org.elasticsearch.compute.data.BlockFactory;
 import org.elasticsearch.compute.data.DoubleBlock;
+import org.elasticsearch.compute.data.Page;
 import org.elasticsearch.compute.operator.SourceOperator;
 import org.elasticsearch.compute.test.SequenceLongBlockSourceOperator;
 
@@ -39,7 +40,7 @@ protected String expectedDescriptionOfAggregator() {
     }
 
     @Override
-    protected void assertSimpleOutput(List<Block> input, Block result) {
+    protected void assertSimpleOutput(List<Page> input, Block result) {
         assertThat(((DoubleBlock) result).getDouble(0), equalTo(23.0));
     }
 }
diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MinBooleanAggregatorFunctionTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MinBooleanAggregatorFunctionTests.java
index 186d9edf2a9e0..670c4943a5214 100644
--- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MinBooleanAggregatorFunctionTests.java
+++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MinBooleanAggregatorFunctionTests.java
@@ -10,6 +10,7 @@
 import org.elasticsearch.compute.data.Block;
 import org.elasticsearch.compute.data.BlockFactory;
 import org.elasticsearch.compute.data.BooleanBlock;
+import org.elasticsearch.compute.data.Page;
 import org.elasticsearch.compute.operator.SequenceBooleanBlockSourceOperator;
 import org.elasticsearch.compute.operator.SourceOperator;
 
@@ -36,8 +37,8 @@ protected String expectedDescriptionOfAggregator() {
     }
 
     @Override
-    public void assertSimpleOutput(List<Block> input, Block result) {
-        Boolean min = input.stream().flatMap(b -> allBooleans(b)).min(Comparator.naturalOrder()).get();
+    public void assertSimpleOutput(List<Page> input, Block result) {
+        Boolean min = input.stream().flatMap(p -> allBooleans(p.getBlock(0))).min(Comparator.naturalOrder()).get();
         assertThat(((BooleanBlock) result).getBoolean(0), equalTo(min));
     }
 }
diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MinBytesRefAggregatorFunctionTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MinBytesRefAggregatorFunctionTests.java
index 678f7259f7843..0f54c3d78aa7a 100644
--- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MinBytesRefAggregatorFunctionTests.java
+++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MinBytesRefAggregatorFunctionTests.java
@@ -11,6 +11,7 @@
 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.operator.SequenceBytesRefBlockSourceOperator;
 import org.elasticsearch.compute.operator.SourceOperator;
 
@@ -41,8 +42,8 @@ protected String expectedDescriptionOfAggregator() {
     }
 
     @Override
-    public void assertSimpleOutput(List<Block> input, Block result) {
-        Optional<BytesRef> max = input.stream().flatMap(b -> allBytesRefs(b)).min(Comparator.naturalOrder());
+    public void assertSimpleOutput(List<Page> input, Block result) {
+        Optional<BytesRef> max = input.stream().flatMap(p -> allBytesRefs(p.getBlock(0))).min(Comparator.naturalOrder());
         if (max.isEmpty()) {
             assertThat(result.isNull(0), equalTo(true));
             return;
diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MinDoubleAggregatorFunctionTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MinDoubleAggregatorFunctionTests.java
index 7f7095d13aa46..b16e1be7e97f9 100644
--- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MinDoubleAggregatorFunctionTests.java
+++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MinDoubleAggregatorFunctionTests.java
@@ -10,6 +10,7 @@
 import org.elasticsearch.compute.data.Block;
 import org.elasticsearch.compute.data.BlockFactory;
 import org.elasticsearch.compute.data.DoubleBlock;
+import org.elasticsearch.compute.data.Page;
 import org.elasticsearch.compute.operator.SequenceDoubleBlockSourceOperator;
 import org.elasticsearch.compute.operator.SourceOperator;
 import org.elasticsearch.test.ESTestCase;
@@ -36,8 +37,8 @@ protected String expectedDescriptionOfAggregator() {
     }
 
     @Override
-    protected void assertSimpleOutput(List<Block> input, Block result) {
-        double min = input.stream().flatMapToDouble(b -> allDoubles(b)).min().getAsDouble();
+    protected void assertSimpleOutput(List<Page> input, Block result) {
+        double min = input.stream().flatMapToDouble(p -> allDoubles(p.getBlock(0))).min().getAsDouble();
         assertThat(((DoubleBlock) result).getDouble(0), equalTo(min));
     }
 }
diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MinFloatAggregatorFunctionTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MinFloatAggregatorFunctionTests.java
index ef98a2dd7b954..303d9bb0958a0 100644
--- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MinFloatAggregatorFunctionTests.java
+++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MinFloatAggregatorFunctionTests.java
@@ -10,6 +10,7 @@
 import org.elasticsearch.compute.data.Block;
 import org.elasticsearch.compute.data.BlockFactory;
 import org.elasticsearch.compute.data.FloatBlock;
+import org.elasticsearch.compute.data.Page;
 import org.elasticsearch.compute.operator.SequenceFloatBlockSourceOperator;
 import org.elasticsearch.compute.operator.SourceOperator;
 import org.elasticsearch.test.ESTestCase;
@@ -36,8 +37,8 @@ protected String expectedDescriptionOfAggregator() {
     }
 
     @Override
-    protected void assertSimpleOutput(List<Block> input, Block result) {
-        Float min = input.stream().flatMap(b -> allFloats(b)).min(floatComparator()).get();
+    protected void assertSimpleOutput(List<Page> input, Block result) {
+        Float min = input.stream().flatMap(p -> allFloats(p.getBlock(0))).min(floatComparator()).get();
         assertThat(((FloatBlock) result).getFloat(0), equalTo(min));
     }
 }
diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MinIntAggregatorFunctionTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MinIntAggregatorFunctionTests.java
index e7296a5b08f4d..b75dab7793b99 100644
--- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MinIntAggregatorFunctionTests.java
+++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MinIntAggregatorFunctionTests.java
@@ -10,6 +10,7 @@
 import org.elasticsearch.compute.data.Block;
 import org.elasticsearch.compute.data.BlockFactory;
 import org.elasticsearch.compute.data.IntBlock;
+import org.elasticsearch.compute.data.Page;
 import org.elasticsearch.compute.operator.SequenceIntBlockSourceOperator;
 import org.elasticsearch.compute.operator.SourceOperator;
 
@@ -35,8 +36,8 @@ protected String expectedDescriptionOfAggregator() {
     }
 
     @Override
-    public void assertSimpleOutput(List<Block> input, Block result) {
-        int max = input.stream().flatMapToInt(b -> allInts(b)).min().getAsInt();
+    public void assertSimpleOutput(List<Page> input, Block result) {
+        int max = input.stream().flatMapToInt(p -> allInts(p.getBlock(0))).min().getAsInt();
         assertThat(((IntBlock) result).getInt(0), equalTo(max));
     }
 }
diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MinIpAggregatorFunctionTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MinIpAggregatorFunctionTests.java
index 9072702178316..bfb239fbeafb7 100644
--- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MinIpAggregatorFunctionTests.java
+++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MinIpAggregatorFunctionTests.java
@@ -12,6 +12,7 @@
 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.operator.SequenceBytesRefBlockSourceOperator;
 import org.elasticsearch.compute.operator.SourceOperator;
 
@@ -42,8 +43,8 @@ protected String expectedDescriptionOfAggregator() {
     }
 
     @Override
-    public void assertSimpleOutput(List<Block> input, Block result) {
-        Optional<BytesRef> max = input.stream().flatMap(b -> allBytesRefs(b)).min(Comparator.naturalOrder());
+    public void assertSimpleOutput(List<Page> input, Block result) {
+        Optional<BytesRef> max = input.stream().flatMap(p -> allBytesRefs(p.getBlock(0))).min(Comparator.naturalOrder());
         if (max.isEmpty()) {
             assertThat(result.isNull(0), equalTo(true));
             return;
diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MinLongAggregatorFunctionTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MinLongAggregatorFunctionTests.java
index 8bb82a149f45e..8120bfd7e67ab 100644
--- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MinLongAggregatorFunctionTests.java
+++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/MinLongAggregatorFunctionTests.java
@@ -10,6 +10,7 @@
 import org.elasticsearch.compute.data.Block;
 import org.elasticsearch.compute.data.BlockFactory;
 import org.elasticsearch.compute.data.LongBlock;
+import org.elasticsearch.compute.data.Page;
 import org.elasticsearch.compute.operator.SourceOperator;
 import org.elasticsearch.compute.test.SequenceLongBlockSourceOperator;
 
@@ -36,8 +37,8 @@ protected String expectedDescriptionOfAggregator() {
     }
 
     @Override
-    public void assertSimpleOutput(List<Block> input, Block result) {
-        long min = input.stream().flatMapToLong(b -> allLongs(b)).min().getAsLong();
+    public void assertSimpleOutput(List<Page> input, Block result) {
+        long min = input.stream().flatMapToLong(p -> allLongs(p.getBlock(0))).min().getAsLong();
         assertThat(((LongBlock) result).getLong(0), equalTo(min));
     }
 }
diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/PercentileDoubleAggregatorFunctionTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/PercentileDoubleAggregatorFunctionTests.java
index aa18c47733ff5..aac55af68d3f5 100644
--- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/PercentileDoubleAggregatorFunctionTests.java
+++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/PercentileDoubleAggregatorFunctionTests.java
@@ -11,6 +11,7 @@
 import org.elasticsearch.compute.data.Block;
 import org.elasticsearch.compute.data.BlockFactory;
 import org.elasticsearch.compute.data.DoubleBlock;
+import org.elasticsearch.compute.data.Page;
 import org.elasticsearch.compute.operator.SequenceDoubleBlockSourceOperator;
 import org.elasticsearch.compute.operator.SourceOperator;
 import org.elasticsearch.search.aggregations.metrics.TDigestState;
@@ -47,9 +48,9 @@ protected SourceOperator simpleInput(BlockFactory blockFactory, int size) {
     }
 
     @Override
-    protected void assertSimpleOutput(List<Block> input, Block result) {
+    protected void assertSimpleOutput(List<Page> input, Block result) {
         try (TDigestState td = TDigestState.create(newLimitedBreaker(ByteSizeValue.ofMb(100)), QuantileStates.DEFAULT_COMPRESSION)) {
-            input.stream().flatMapToDouble(b -> allDoubles(b)).forEach(td::add);
+            input.stream().flatMapToDouble(p -> allDoubles(p.getBlock(0))).forEach(td::add);
             double expected = td.quantile(percentile / 100);
             double value = ((DoubleBlock) result).getDouble(0);
             assertThat(value, closeTo(expected, expected * 0.1));
diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/PercentileFloatAggregatorFunctionTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/PercentileFloatAggregatorFunctionTests.java
index d23436310cff7..932c0879106af 100644
--- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/PercentileFloatAggregatorFunctionTests.java
+++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/PercentileFloatAggregatorFunctionTests.java
@@ -11,6 +11,7 @@
 import org.elasticsearch.compute.data.Block;
 import org.elasticsearch.compute.data.BlockFactory;
 import org.elasticsearch.compute.data.DoubleBlock;
+import org.elasticsearch.compute.data.Page;
 import org.elasticsearch.compute.operator.SequenceFloatBlockSourceOperator;
 import org.elasticsearch.compute.operator.SourceOperator;
 import org.elasticsearch.search.aggregations.metrics.TDigestState;
@@ -47,9 +48,9 @@ protected SourceOperator simpleInput(BlockFactory blockFactory, int size) {
     }
 
     @Override
-    protected void assertSimpleOutput(List<Block> input, Block result) {
+    protected void assertSimpleOutput(List<Page> input, Block result) {
         try (TDigestState td = TDigestState.create(newLimitedBreaker(ByteSizeValue.ofMb(100)), QuantileStates.DEFAULT_COMPRESSION)) {
-            input.stream().flatMap(AggregatorFunctionTestCase::allFloats).forEach(td::add);
+            input.stream().flatMap(p -> allFloats(p.getBlock(0))).forEach(td::add);
             double expected = td.quantile(percentile / 100);
             double value = ((DoubleBlock) result).getDouble(0);
             assertThat(value, closeTo(expected, expected * 0.1));
diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/PercentileIntAggregatorFunctionTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/PercentileIntAggregatorFunctionTests.java
index 278c37c02be35..ecee6de1a8364 100644
--- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/PercentileIntAggregatorFunctionTests.java
+++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/PercentileIntAggregatorFunctionTests.java
@@ -11,6 +11,7 @@
 import org.elasticsearch.compute.data.Block;
 import org.elasticsearch.compute.data.BlockFactory;
 import org.elasticsearch.compute.data.DoubleBlock;
+import org.elasticsearch.compute.data.Page;
 import org.elasticsearch.compute.operator.SequenceIntBlockSourceOperator;
 import org.elasticsearch.compute.operator.SourceOperator;
 import org.elasticsearch.search.aggregations.metrics.TDigestState;
@@ -47,9 +48,9 @@ protected SourceOperator simpleInput(BlockFactory blockFactory, int size) {
     }
 
     @Override
-    protected void assertSimpleOutput(List<Block> input, Block result) {
+    protected void assertSimpleOutput(List<Page> input, Block result) {
         try (TDigestState td = TDigestState.create(newLimitedBreaker(ByteSizeValue.ofMb(100)), QuantileStates.DEFAULT_COMPRESSION)) {
-            input.stream().flatMapToInt(b -> allInts(b)).forEach(td::add);
+            input.stream().flatMapToInt(p -> allInts(p.getBlock(0))).forEach(td::add);
             double expected = td.quantile(percentile / 100);
             double value = ((DoubleBlock) result).getDouble(0);
             assertThat(value, closeTo(expected, expected * 0.1));
diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/PercentileLongAggregatorFunctionTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/PercentileLongAggregatorFunctionTests.java
index b1896025c363e..520df9705434f 100644
--- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/PercentileLongAggregatorFunctionTests.java
+++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/PercentileLongAggregatorFunctionTests.java
@@ -11,6 +11,7 @@
 import org.elasticsearch.compute.data.Block;
 import org.elasticsearch.compute.data.BlockFactory;
 import org.elasticsearch.compute.data.DoubleBlock;
+import org.elasticsearch.compute.data.Page;
 import org.elasticsearch.compute.operator.SourceOperator;
 import org.elasticsearch.compute.test.SequenceLongBlockSourceOperator;
 import org.elasticsearch.search.aggregations.metrics.TDigestState;
@@ -47,9 +48,9 @@ protected SourceOperator simpleInput(BlockFactory blockFactory, int size) {
     }
 
     @Override
-    protected void assertSimpleOutput(List<Block> input, Block result) {
+    protected void assertSimpleOutput(List<Page> input, Block result) {
         try (TDigestState td = TDigestState.create(newLimitedBreaker(ByteSizeValue.ofMb(100)), QuantileStates.DEFAULT_COMPRESSION)) {
-            input.stream().flatMapToLong(p -> allLongs(p)).forEach(td::add);
+            input.stream().flatMapToLong(p -> allLongs(p.getBlock(0))).forEach(td::add);
             double expected = td.quantile(percentile / 100);
             double value = ((DoubleBlock) result).getDouble(0);
             assertThat(value, closeTo(expected, expected * 0.1));
diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/SampleBooleanAggregatorFunctionTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/SampleBooleanAggregatorFunctionTests.java
index 80a547cdb7774..771b001284564 100644
--- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/SampleBooleanAggregatorFunctionTests.java
+++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/SampleBooleanAggregatorFunctionTests.java
@@ -48,8 +48,8 @@ protected String expectedDescriptionOfAggregator() {
     }
 
     @Override
-    public void assertSimpleOutput(List<Block> input, Block result) {
-        List<Boolean> inputValues = input.stream().flatMap(AggregatorFunctionTestCase::allBooleans).collect(Collectors.toList());
+    public void assertSimpleOutput(List<Page> input, Block result) {
+        List<Boolean> inputValues = input.stream().flatMap(p -> allBooleans(p.getBlock(0))).collect(Collectors.toList());
         Boolean[] resultValues = AggregatorFunctionTestCase.allBooleans(result).toArray(Boolean[]::new);
         assertThat(resultValues, arrayWithSize(Math.min(inputValues.size(), LIMIT)));
     }
diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/SampleBytesRefAggregatorFunctionTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/SampleBytesRefAggregatorFunctionTests.java
index 17d20d83b7879..881c356fe9613 100644
--- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/SampleBytesRefAggregatorFunctionTests.java
+++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/SampleBytesRefAggregatorFunctionTests.java
@@ -54,8 +54,8 @@ protected String expectedDescriptionOfAggregator() {
     }
 
     @Override
-    public void assertSimpleOutput(List<Block> input, Block result) {
-        Set<BytesRef> inputValues = input.stream().flatMap(AggregatorFunctionTestCase::allBytesRefs).collect(Collectors.toSet());
+    public void assertSimpleOutput(List<Page> input, Block result) {
+        Set<BytesRef> inputValues = input.stream().flatMap(p -> allBytesRefs(p.getBlock(0))).collect(Collectors.toSet());
         BytesRef[] resultValues = AggregatorFunctionTestCase.allBytesRefs(result).toArray(BytesRef[]::new);
         assertThat(resultValues, arrayWithSize(Math.min(inputValues.size(), LIMIT)));
         assertThat(inputValues, hasItems(resultValues));
diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/SampleDoubleAggregatorFunctionTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/SampleDoubleAggregatorFunctionTests.java
index 73e7ed8eee5e1..bcb687d250122 100644
--- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/SampleDoubleAggregatorFunctionTests.java
+++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/SampleDoubleAggregatorFunctionTests.java
@@ -50,11 +50,8 @@ protected String expectedDescriptionOfAggregator() {
     }
 
     @Override
-    public void assertSimpleOutput(List<Block> input, Block result) {
-        Set<Double> inputValues = input.stream()
-            .flatMapToDouble(AggregatorFunctionTestCase::allDoubles)
-            .boxed()
-            .collect(Collectors.toSet());
+    public void assertSimpleOutput(List<Page> input, Block result) {
+        Set<Double> inputValues = input.stream().flatMapToDouble(p -> allDoubles(p.getBlock(0))).boxed().collect(Collectors.toSet());
         Double[] resultValues = AggregatorFunctionTestCase.allDoubles(result).boxed().toArray(Double[]::new);
         assertThat(resultValues, arrayWithSize(Math.min(inputValues.size(), LIMIT)));
         assertThat(inputValues, hasItems(resultValues));
diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/SampleIntAggregatorFunctionTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/SampleIntAggregatorFunctionTests.java
index 84b4fec44c289..6cf6318493cdc 100644
--- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/SampleIntAggregatorFunctionTests.java
+++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/SampleIntAggregatorFunctionTests.java
@@ -50,8 +50,8 @@ protected String expectedDescriptionOfAggregator() {
     }
 
     @Override
-    public void assertSimpleOutput(List<Block> input, Block result) {
-        Set<Integer> inputValues = input.stream().flatMapToInt(AggregatorFunctionTestCase::allInts).boxed().collect(Collectors.toSet());
+    public void assertSimpleOutput(List<Page> input, Block result) {
+        Set<Integer> inputValues = input.stream().flatMapToInt(p -> allInts(p.getBlock(0))).boxed().collect(Collectors.toSet());
         Integer[] resultValues = AggregatorFunctionTestCase.allInts(result).boxed().toArray(Integer[]::new);
         assertThat(resultValues, arrayWithSize(Math.min(inputValues.size(), LIMIT)));
         assertThat(inputValues, hasItems(resultValues));
diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/SampleLongAggregatorFunctionTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/SampleLongAggregatorFunctionTests.java
index 54c92dfb10cb7..49248bc30f314 100644
--- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/SampleLongAggregatorFunctionTests.java
+++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/SampleLongAggregatorFunctionTests.java
@@ -51,8 +51,8 @@ protected String expectedDescriptionOfAggregator() {
     }
 
     @Override
-    public void assertSimpleOutput(List<Block> input, Block result) {
-        Set<Long> inputValues = input.stream().flatMapToLong(AggregatorFunctionTestCase::allLongs).boxed().collect(Collectors.toSet());
+    public void assertSimpleOutput(List<Page> input, Block result) {
+        Set<Long> inputValues = input.stream().flatMapToLong(p -> allLongs(p.getBlock(0))).boxed().collect(Collectors.toSet());
         Long[] resultValues = AggregatorFunctionTestCase.allLongs(result).boxed().toArray(Long[]::new);
         assertThat(resultValues, arrayWithSize(Math.min(inputValues.size(), LIMIT)));
         assertThat(inputValues, hasItems(resultValues));
diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/SumDoubleAggregatorFunctionTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/SumDoubleAggregatorFunctionTests.java
index 9332c44de893d..b860301a1b9a2 100644
--- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/SumDoubleAggregatorFunctionTests.java
+++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/SumDoubleAggregatorFunctionTests.java
@@ -44,8 +44,8 @@ protected String expectedDescriptionOfAggregator() {
     }
 
     @Override
-    protected void assertSimpleOutput(List<Block> input, Block result) {
-        double sum = input.stream().flatMapToDouble(b -> allDoubles(b)).sum();
+    protected void assertSimpleOutput(List<Page> input, Block result) {
+        double sum = input.stream().flatMapToDouble(p -> allDoubles(p.getBlock(0))).sum();
         assertThat(((DoubleBlock) result).getDouble(0), closeTo(sum, .0001));
     }
 
diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/SumFloatAggregatorFunctionTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/SumFloatAggregatorFunctionTests.java
index 69187e40fd72f..5a2c3e40acd29 100644
--- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/SumFloatAggregatorFunctionTests.java
+++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/SumFloatAggregatorFunctionTests.java
@@ -44,8 +44,8 @@ protected String expectedDescriptionOfAggregator() {
     }
 
     @Override
-    protected void assertSimpleOutput(List<Block> input, Block result) {
-        double sum = input.stream().flatMap(AggregatorFunctionTestCase::allFloats).mapToDouble(f -> (double) f).sum();
+    protected void assertSimpleOutput(List<Page> input, Block result) {
+        double sum = input.stream().flatMap(p -> allFloats(p.getBlock(0))).mapToDouble(f -> (double) f).sum();
         assertThat(((DoubleBlock) result).getDouble(0), closeTo(sum, .0001));
     }
 
diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/SumIntAggregatorFunctionTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/SumIntAggregatorFunctionTests.java
index d7df7f2110081..27cc369fc3a7b 100644
--- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/SumIntAggregatorFunctionTests.java
+++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/SumIntAggregatorFunctionTests.java
@@ -43,8 +43,8 @@ protected String expectedDescriptionOfAggregator() {
     }
 
     @Override
-    protected void assertSimpleOutput(List<Block> input, Block result) {
-        long sum = input.stream().flatMapToInt(b -> allInts(b)).asLongStream().sum();
+    protected void assertSimpleOutput(List<Page> input, Block result) {
+        long sum = input.stream().flatMapToInt(p -> allInts(p.getBlock(0))).asLongStream().sum();
         assertThat(((LongBlock) result).getLong(0), equalTo(sum));
     }
 
diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/SumLongAggregatorFunctionTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/SumLongAggregatorFunctionTests.java
index 73f0837edabda..ab5d0fdb104e7 100644
--- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/SumLongAggregatorFunctionTests.java
+++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/SumLongAggregatorFunctionTests.java
@@ -43,8 +43,8 @@ protected String expectedDescriptionOfAggregator() {
     }
 
     @Override
-    public void assertSimpleOutput(List<Block> input, Block result) {
-        long sum = input.stream().flatMapToLong(b -> allLongs(b)).sum();
+    public void assertSimpleOutput(List<Page> input, Block result) {
+        long sum = input.stream().flatMapToLong(p -> allLongs(p.getBlock(0))).sum();
         assertThat(((LongBlock) result).getLong(0), equalTo(sum));
     }
 
diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/TopBooleanAggregatorFunctionTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/TopBooleanAggregatorFunctionTests.java
index cfb91acb1cf20..5a06468aa752a 100644
--- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/TopBooleanAggregatorFunctionTests.java
+++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/TopBooleanAggregatorFunctionTests.java
@@ -10,6 +10,7 @@
 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.operator.SequenceBooleanBlockSourceOperator;
 import org.elasticsearch.compute.operator.SourceOperator;
 
@@ -37,8 +38,8 @@ protected String expectedDescriptionOfAggregator() {
     }
 
     @Override
-    public void assertSimpleOutput(List<Block> input, Block result) {
-        Object[] values = input.stream().flatMap(b -> allBooleans(b)).sorted().limit(LIMIT).toArray(Object[]::new);
+    public void assertSimpleOutput(List<Page> input, Block result) {
+        Object[] values = input.stream().flatMap(p -> allBooleans(p.getBlock(0))).sorted().limit(LIMIT).toArray(Object[]::new);
         assertThat((List<?>) BlockUtils.toJavaObject(result, 0), contains(values));
     }
 }
diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/TopDoubleAggregatorFunctionTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/TopDoubleAggregatorFunctionTests.java
index 04c2000d2e2d7..3efb0c21c4ec4 100644
--- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/TopDoubleAggregatorFunctionTests.java
+++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/TopDoubleAggregatorFunctionTests.java
@@ -10,6 +10,7 @@
 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.operator.SequenceDoubleBlockSourceOperator;
 import org.elasticsearch.compute.operator.SourceOperator;
 
@@ -37,8 +38,13 @@ protected String expectedDescriptionOfAggregator() {
     }
 
     @Override
-    public void assertSimpleOutput(List<Block> input, Block result) {
-        Object[] values = input.stream().flatMapToDouble(b -> allDoubles(b)).sorted().limit(LIMIT).boxed().toArray(Object[]::new);
+    public void assertSimpleOutput(List<Page> input, Block result) {
+        Object[] values = input.stream()
+            .flatMapToDouble(p -> allDoubles(p.getBlock(0)))
+            .sorted()
+            .limit(LIMIT)
+            .boxed()
+            .toArray(Object[]::new);
         assertThat((List<?>) BlockUtils.toJavaObject(result, 0), contains(values));
     }
 }
diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/TopFloatAggregatorFunctionTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/TopFloatAggregatorFunctionTests.java
index 8dd2d5d82f815..4ba7730133b59 100644
--- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/TopFloatAggregatorFunctionTests.java
+++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/TopFloatAggregatorFunctionTests.java
@@ -10,6 +10,7 @@
 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.operator.SequenceFloatBlockSourceOperator;
 import org.elasticsearch.compute.operator.SourceOperator;
 
@@ -37,8 +38,8 @@ protected String expectedDescriptionOfAggregator() {
     }
 
     @Override
-    public void assertSimpleOutput(List<Block> input, Block result) {
-        Object[] values = input.stream().flatMap(b -> allFloats(b)).sorted().limit(LIMIT).toArray(Object[]::new);
+    public void assertSimpleOutput(List<Page> input, Block result) {
+        Object[] values = input.stream().flatMap(p -> allFloats(p.getBlock(0))).sorted().limit(LIMIT).toArray(Object[]::new);
         assertThat((List<?>) BlockUtils.toJavaObject(result, 0), contains(values));
     }
 }
diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/TopIntAggregatorFunctionTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/TopIntAggregatorFunctionTests.java
index b52439dc98263..d384f762c6639 100644
--- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/TopIntAggregatorFunctionTests.java
+++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/TopIntAggregatorFunctionTests.java
@@ -10,6 +10,7 @@
 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.operator.SequenceIntBlockSourceOperator;
 import org.elasticsearch.compute.operator.SourceOperator;
 
@@ -37,8 +38,8 @@ protected String expectedDescriptionOfAggregator() {
     }
 
     @Override
-    public void assertSimpleOutput(List<Block> input, Block result) {
-        Object[] values = input.stream().flatMapToInt(b -> allInts(b)).sorted().limit(LIMIT).boxed().toArray(Object[]::new);
+    public void assertSimpleOutput(List<Page> input, Block result) {
+        Object[] values = input.stream().flatMapToInt(p -> allInts(p.getBlock(0))).sorted().limit(LIMIT).boxed().toArray(Object[]::new);
         assertThat((List<?>) BlockUtils.toJavaObject(result, 0), contains(values));
     }
 }
diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/TopLongAggregatorFunctionTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/TopLongAggregatorFunctionTests.java
index 4ff27b092a183..fd37bc78a56e6 100644
--- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/TopLongAggregatorFunctionTests.java
+++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/TopLongAggregatorFunctionTests.java
@@ -10,6 +10,7 @@
 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.operator.SourceOperator;
 import org.elasticsearch.compute.test.SequenceLongBlockSourceOperator;
 
@@ -37,8 +38,8 @@ protected String expectedDescriptionOfAggregator() {
     }
 
     @Override
-    public void assertSimpleOutput(List<Block> input, Block result) {
-        Object[] values = input.stream().flatMapToLong(b -> allLongs(b)).sorted().limit(LIMIT).boxed().toArray(Object[]::new);
+    public void assertSimpleOutput(List<Page> input, Block result) {
+        Object[] values = input.stream().flatMapToLong(p -> allLongs(p.getBlock(0))).sorted().limit(LIMIT).boxed().toArray(Object[]::new);
         assertThat((List<?>) BlockUtils.toJavaObject(result, 0), contains(values));
     }
 }
diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/ValuesBytesRefAggregatorFunctionTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/ValuesBytesRefAggregatorFunctionTests.java
index 7c5e3f3861161..71fbc80110fe2 100644
--- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/ValuesBytesRefAggregatorFunctionTests.java
+++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/ValuesBytesRefAggregatorFunctionTests.java
@@ -11,6 +11,7 @@
 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.operator.SequenceBytesRefBlockSourceOperator;
 import org.elasticsearch.compute.operator.SourceOperator;
 
@@ -42,12 +43,9 @@ protected String expectedDescriptionOfAggregator() {
     }
 
     @Override
-    public void assertSimpleOutput(List<Block> input, Block result) {
+    public void assertSimpleOutput(List<Page> input, Block result) {
         TreeSet<?> set = new TreeSet<>((List<?>) BlockUtils.toJavaObject(result, 0));
-        Object[] values = input.stream()
-            .flatMap(AggregatorFunctionTestCase::allBytesRefs)
-            .collect(Collectors.toSet())
-            .toArray(Object[]::new);
+        Object[] values = input.stream().flatMap(p -> allBytesRefs(p.getBlock(0))).collect(Collectors.toSet()).toArray(Object[]::new);
         if (false == set.containsAll(Arrays.asList(values))) {
             assertThat(set, containsInAnyOrder(values));
         }
diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/ValuesDoubleAggregatorFunctionTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/ValuesDoubleAggregatorFunctionTests.java
index 497813e058e67..84c5308b410d8 100644
--- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/ValuesDoubleAggregatorFunctionTests.java
+++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/ValuesDoubleAggregatorFunctionTests.java
@@ -10,6 +10,7 @@
 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.operator.SequenceDoubleBlockSourceOperator;
 import org.elasticsearch.compute.operator.SourceOperator;
 
@@ -38,10 +39,10 @@ protected String expectedDescriptionOfAggregator() {
     }
 
     @Override
-    public void assertSimpleOutput(List<Block> input, Block result) {
+    public void assertSimpleOutput(List<Page> input, Block result) {
         TreeSet<?> set = new TreeSet<>((List<?>) BlockUtils.toJavaObject(result, 0));
         Object[] values = input.stream()
-            .flatMapToDouble(AggregatorFunctionTestCase::allDoubles)
+            .flatMapToDouble(p -> allDoubles(p.getBlock(0)))
             .boxed()
             .collect(Collectors.toSet())
             .toArray(Object[]::new);
diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/ValuesFloatAggregatorFunctionTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/ValuesFloatAggregatorFunctionTests.java
index 0cf536d3e0eca..a1a775c3eb129 100644
--- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/ValuesFloatAggregatorFunctionTests.java
+++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/ValuesFloatAggregatorFunctionTests.java
@@ -10,6 +10,7 @@
 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.operator.SequenceFloatBlockSourceOperator;
 import org.elasticsearch.compute.operator.SourceOperator;
 
@@ -38,9 +39,9 @@ protected String expectedDescriptionOfAggregator() {
     }
 
     @Override
-    public void assertSimpleOutput(List<Block> input, Block result) {
+    public void assertSimpleOutput(List<Page> input, Block result) {
         TreeSet<?> set = new TreeSet<>((List<?>) BlockUtils.toJavaObject(result, 0));
-        Object[] values = input.stream().flatMap(AggregatorFunctionTestCase::allFloats).collect(Collectors.toSet()).toArray(Object[]::new);
+        Object[] values = input.stream().flatMap(p -> allFloats(p.getBlock(0))).collect(Collectors.toSet()).toArray(Object[]::new);
         if (false == set.containsAll(Arrays.asList(values))) {
             assertThat(set, containsInAnyOrder(values));
         }
diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/ValuesIntAggregatorFunctionTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/ValuesIntAggregatorFunctionTests.java
index 9e4d56a962b2a..c35e81023517e 100644
--- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/ValuesIntAggregatorFunctionTests.java
+++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/ValuesIntAggregatorFunctionTests.java
@@ -10,6 +10,7 @@
 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.operator.SequenceIntBlockSourceOperator;
 import org.elasticsearch.compute.operator.SourceOperator;
 
@@ -38,10 +39,10 @@ protected String expectedDescriptionOfAggregator() {
     }
 
     @Override
-    public void assertSimpleOutput(List<Block> input, Block result) {
+    public void assertSimpleOutput(List<Page> input, Block result) {
         TreeSet<?> set = new TreeSet<>((List<?>) BlockUtils.toJavaObject(result, 0));
         Object[] values = input.stream()
-            .flatMapToInt(AggregatorFunctionTestCase::allInts)
+            .flatMapToInt(p -> allInts(p.getBlock(0)))
             .boxed()
             .collect(Collectors.toSet())
             .toArray(Object[]::new);
diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/ValuesLongAggregatorFunctionTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/ValuesLongAggregatorFunctionTests.java
index 32609edd2b8fe..d9b5066bd4059 100644
--- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/ValuesLongAggregatorFunctionTests.java
+++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/aggregation/ValuesLongAggregatorFunctionTests.java
@@ -10,6 +10,7 @@
 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.operator.SourceOperator;
 import org.elasticsearch.compute.test.SequenceLongBlockSourceOperator;
 
@@ -38,10 +39,10 @@ protected String expectedDescriptionOfAggregator() {
     }
 
     @Override
-    public void assertSimpleOutput(List<Block> input, Block result) {
+    public void assertSimpleOutput(List<Page> input, Block result) {
         TreeSet<?> set = new TreeSet<>((List<?>) BlockUtils.toJavaObject(result, 0));
         Object[] values = input.stream()
-            .flatMapToLong(AggregatorFunctionTestCase::allLongs)
+            .flatMapToLong(p -> allLongs(p.getBlock(0)))
             .boxed()
             .collect(Collectors.toSet())
             .toArray(Object[]::new);
diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/operator/AggregationOperatorTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/operator/AggregationOperatorTests.java
index 2408e758ec697..5ede747613fe2 100644
--- a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/operator/AggregationOperatorTests.java
+++ b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/operator/AggregationOperatorTests.java
@@ -80,7 +80,7 @@ protected void assertSimpleOutput(List<Page> input, List<Page> results) {
 
         Block sums = results.get(0).getBlock(0);
         Block maxs = results.get(0).getBlock(1);
-        sum.assertSimpleOutput(input.stream().map(p -> p.<Block>getBlock(0)).toList(), sums);
-        max.assertSimpleOutput(input.stream().map(p -> p.<Block>getBlock(0)).toList(), maxs);
+        sum.assertSimpleOutput(input, sums);
+        max.assertSimpleOutput(input, maxs);
     }
 }