diff --git a/.nix/php/lib/php.ini.dist b/.nix/php/lib/php.ini.dist
index 488665798b..ba9900777c 100644
--- a/.nix/php/lib/php.ini.dist
+++ b/.nix/php/lib/php.ini.dist
@@ -9,4 +9,7 @@ file_uploads = On
 max_file_uploads = 20
 short_open_tag = off
 opcache.enable=1
-opcache.enable_cli=0
\ No newline at end of file
+opcache.enable_cli=0
+apc.enabled=1
+apc.enable_cli=1
+apc.shm_size=128M
\ No newline at end of file
diff --git a/.nix/pkgs/flow-php/package.nix b/.nix/pkgs/flow-php/package.nix
index 1d6bbd7406..384e33db14 100644
--- a/.nix/pkgs/flow-php/package.nix
+++ b/.nix/pkgs/flow-php/package.nix
@@ -22,6 +22,7 @@ let
     with all;
     enabled
         ++ [
+            apcu
             bcmath
             dom
             mbstring
diff --git a/src/core/etl/src/Flow/ETL/Config.php b/src/core/etl/src/Flow/ETL/Config.php
index 00ebf0a0c6..4cf96bd43b 100644
--- a/src/core/etl/src/Flow/ETL/Config.php
+++ b/src/core/etl/src/Flow/ETL/Config.php
@@ -4,6 +4,7 @@
 
 namespace Flow\ETL;
 
+use Flow\ETL\Config\Aggregation\AggregationConfig;
 use Flow\ETL\Config\Cache\CacheConfig;
 use Flow\ETL\Config\ConfigBuilder;
 use Flow\ETL\Config\Sort\SortConfig;
@@ -36,6 +37,7 @@ public function __construct(
         public SortConfig $sort,
         private ?Analyze $analyze,
         public TelemetryConfig $telemetry,
+        public AggregationConfig $aggregation,
     ) {}
 
     public static function builder(): ConfigBuilder
diff --git a/src/core/etl/src/Flow/ETL/Config/Aggregation/AggregationConfig.php b/src/core/etl/src/Flow/ETL/Config/Aggregation/AggregationConfig.php
new file mode 100644
index 0000000000..cfcb9bd5ea
--- /dev/null
+++ b/src/core/etl/src/Flow/ETL/Config/Aggregation/AggregationConfig.php
@@ -0,0 +1,17 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\ETL\Config\Aggregation;
+
+use Flow\ETL\Dataset\Memory\Unit;
+
+final readonly class AggregationConfig
+{
+    public const string AGGREGATION_MAX_MEMORY_ENV = 'FLOW_AGGREGATION_MAX_MEMORY';
+
+    public function __construct(
+        public Unit $memoryLimit,
+        public string $filesystemProtocol = 'file',
+    ) {}
+}
diff --git a/src/core/etl/src/Flow/ETL/Config/Aggregation/AggregationConfigBuilder.php b/src/core/etl/src/Flow/ETL/Config/Aggregation/AggregationConfigBuilder.php
new file mode 100644
index 0000000000..56446c2a6b
--- /dev/null
+++ b/src/core/etl/src/Flow/ETL/Config/Aggregation/AggregationConfigBuilder.php
@@ -0,0 +1,44 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\ETL\Config\Aggregation;
+
+use Flow\ETL\Dataset\Memory\Unit;
+
+use function getenv;
+use function is_string;
+
+use const PHP_INT_MAX;
+
+final class AggregationConfigBuilder
+{
+    private string $filesystemProtocol = 'file';
+
+    private ?Unit $memoryLimit = null;
+
+    public function build(): AggregationConfig
+    {
+        if ($this->memoryLimit === null) {
+            $envLimit = getenv(AggregationConfig::AGGREGATION_MAX_MEMORY_ENV);
+
+            $this->memoryLimit = is_string($envLimit) ? Unit::fromString($envLimit) : Unit::fromBytes(PHP_INT_MAX);
+        }
+
+        return new AggregationConfig($this->memoryLimit, $this->filesystemProtocol);
+    }
+
+    public function filesystemProtocol(string $protocol): self
+    {
+        $this->filesystemProtocol = $protocol;
+
+        return $this;
+    }
+
+    public function memoryLimit(Unit $memoryLimit): self
+    {
+        $this->memoryLimit = $memoryLimit;
+
+        return $this;
+    }
+}
diff --git a/src/core/etl/src/Flow/ETL/Config/ConfigBuilder.php b/src/core/etl/src/Flow/ETL/Config/ConfigBuilder.php
index 02606bbd83..609add4da6 100644
--- a/src/core/etl/src/Flow/ETL/Config/ConfigBuilder.php
+++ b/src/core/etl/src/Flow/ETL/Config/ConfigBuilder.php
@@ -8,6 +8,7 @@
 use Flow\ETL\Analyze;
 use Flow\ETL\Cache;
 use Flow\ETL\Config;
+use Flow\ETL\Config\Aggregation\AggregationConfigBuilder;
 use Flow\ETL\Config\Cache\CacheConfigBuilder;
 use Flow\ETL\Config\Sort\SortConfigBuilder;
 use Flow\ETL\Config\Telemetry\TelemetryConfig;
@@ -34,6 +35,8 @@
 
 final class ConfigBuilder
 {
+    public readonly AggregationConfigBuilder $aggregation;
+
     public readonly CacheConfigBuilder $cache;
 
     public readonly SortConfigBuilder $sort;
@@ -69,6 +72,7 @@ public function __construct()
         $this->putInputIntoRows = false;
         $this->optimizer = null;
         $this->clock = null;
+        $this->aggregation = new AggregationConfigBuilder();
         $this->cache = new CacheConfigBuilder();
         $this->sort = new SortConfigBuilder();
         $this->randomValueGenerator = new NativePHPRandomValueGenerator();
@@ -79,6 +83,20 @@ public function __construct()
             : PackageVersion::get('flow-php/etl');
     }
 
+    public function aggregationFilesystem(string $protocol): self
+    {
+        $this->aggregation->filesystemProtocol($protocol);
+
+        return $this;
+    }
+
+    public function aggregationMemoryLimit(Unit $unit): self
+    {
+        $this->aggregation->memoryLimit($unit);
+
+        return $this;
+    }
+
     public function analyze(Analyze $analyze): self
     {
         $this->analyze = $analyze;
@@ -111,6 +129,7 @@ public function build(EntryFactory $entryFactory = new EntryFactory()): Config
             $this->sort->build(),
             $this->analyze,
             $this->telemetryConfig ?? TelemetryConfig::default($this->getClock()),
+            $this->aggregation->build(),
         );
     }
 
diff --git a/src/core/etl/src/Flow/ETL/Function/AggregatingFunction.php b/src/core/etl/src/Flow/ETL/Function/AggregatingFunction.php
index 925e32da6b..b65743a9c3 100644
--- a/src/core/etl/src/Flow/ETL/Function/AggregatingFunction.php
+++ b/src/core/etl/src/Flow/ETL/Function/AggregatingFunction.php
@@ -13,6 +13,8 @@ interface AggregatingFunction
 {
     public function aggregate(Row $row, FlowContext $context): void;
 
+    public function merge(self $other): void;
+
     /**
      * @return Entry<mixed>
      */
diff --git a/src/core/etl/src/Flow/ETL/Function/Average.php b/src/core/etl/src/Flow/ETL/Function/Average.php
index d404adea22..00fe1a8775 100644
--- a/src/core/etl/src/Flow/ETL/Function/Average.php
+++ b/src/core/etl/src/Flow/ETL/Function/Average.php
@@ -82,6 +82,23 @@ public function apply(Row $row, Rows $partition, FlowContext $context): mixed
         return (new Calculator())->divide($sum, $count, $this->scale, $this->rounding);
     }
 
+    public function merge(AggregatingFunction $other): void
+    {
+        if (
+            !$other instanceof self
+            || !$this->ref->is($other->ref)
+            || $this->scale !== $other->scale
+            || $this->rounding !== $other->rounding
+        ) {
+            throw new InvalidArgumentException(
+                'Cannot merge ' . self::class . ' aggregating a different reference, scale or rounding',
+            );
+        }
+
+        $this->sum += $other->sum;
+        $this->count += $other->count;
+    }
+
     public function over(Window $window): WindowFunction
     {
         $this->window = $window;
diff --git a/src/core/etl/src/Flow/ETL/Function/Collect.php b/src/core/etl/src/Flow/ETL/Function/Collect.php
index 12209f5710..93614b95e4 100644
--- a/src/core/etl/src/Flow/ETL/Function/Collect.php
+++ b/src/core/etl/src/Flow/ETL/Function/Collect.php
@@ -11,6 +11,7 @@
 use Flow\ETL\Row\EntryFactory;
 use Flow\ETL\Row\Reference;
 
+use function array_merge;
 use function current;
 use function Flow\ETL\DSL\to_entry;
 
@@ -41,6 +42,15 @@ public function aggregate(Row $row, FlowContext $context): void
         }
     }
 
+    public function merge(AggregatingFunction $other): void
+    {
+        if (!$other instanceof self || !$this->ref->is($other->ref)) {
+            throw new InvalidArgumentException('Cannot merge ' . self::class . ' aggregating a different reference');
+        }
+
+        $this->collection = array_merge($this->collection, $other->collection);
+    }
+
     /**
      * @return Entry<mixed>
      */
diff --git a/src/core/etl/src/Flow/ETL/Function/CollectUnique.php b/src/core/etl/src/Flow/ETL/Function/CollectUnique.php
index 3f87b0ff09..e4c9173c94 100644
--- a/src/core/etl/src/Flow/ETL/Function/CollectUnique.php
+++ b/src/core/etl/src/Flow/ETL/Function/CollectUnique.php
@@ -49,6 +49,20 @@ public function aggregate(Row $row, FlowContext $context): void
         }
     }
 
+    public function merge(AggregatingFunction $other): void
+    {
+        if (!$other instanceof self || !$this->ref->is($other->ref)) {
+            throw new InvalidArgumentException('Cannot merge ' . self::class . ' aggregating a different reference');
+        }
+
+        /** @var mixed $value */
+        foreach ($other->collection as $value) {
+            if (!in_array($value, $this->collection, true)) {
+                $this->collection[] = $value;
+            }
+        }
+    }
+
     /**
      * @return Entry<mixed>
      */
diff --git a/src/core/etl/src/Flow/ETL/Function/Count.php b/src/core/etl/src/Flow/ETL/Function/Count.php
index 52e585ef3c..8fc456391a 100644
--- a/src/core/etl/src/Flow/ETL/Function/Count.php
+++ b/src/core/etl/src/Flow/ETL/Function/Count.php
@@ -80,6 +80,19 @@ public function apply(Row $row, Rows $partition, FlowContext $context): mixed
         return $count;
     }
 
+    public function merge(AggregatingFunction $other): void
+    {
+        if (
+            !$other instanceof self
+            || ($this->ref === null) !== ($other->ref === null)
+            || $this->ref !== null && $other->ref !== null && !$this->ref->is($other->ref)
+        ) {
+            throw new InvalidArgumentException('Cannot merge ' . self::class . ' aggregating a different reference');
+        }
+
+        $this->count += $other->count;
+    }
+
     public function over(Window $window): WindowFunction
     {
         $this->window = $window;
diff --git a/src/core/etl/src/Flow/ETL/Function/First.php b/src/core/etl/src/Flow/ETL/Function/First.php
index 36126f0176..7be217b5a6 100644
--- a/src/core/etl/src/Flow/ETL/Function/First.php
+++ b/src/core/etl/src/Flow/ETL/Function/First.php
@@ -37,6 +37,15 @@ public function aggregate(Row $row, FlowContext $context): void
         }
     }
 
+    public function merge(AggregatingFunction $other): void
+    {
+        if (!$other instanceof self || !$this->ref->is($other->ref)) {
+            throw new InvalidArgumentException('Cannot merge ' . self::class . ' aggregating a different reference');
+        }
+
+        $this->first ??= $other->first;
+    }
+
     /**
      * @return Entry<mixed>
      */
diff --git a/src/core/etl/src/Flow/ETL/Function/Last.php b/src/core/etl/src/Flow/ETL/Function/Last.php
index d5c1e76f76..6dd8f36b8d 100644
--- a/src/core/etl/src/Flow/ETL/Function/Last.php
+++ b/src/core/etl/src/Flow/ETL/Function/Last.php
@@ -35,6 +35,17 @@ public function aggregate(Row $row, FlowContext $context): void
         }
     }
 
+    public function merge(AggregatingFunction $other): void
+    {
+        if (!$other instanceof self || !$this->ref->is($other->ref)) {
+            throw new InvalidArgumentException('Cannot merge ' . self::class . ' aggregating a different reference');
+        }
+
+        if ($other->last !== null) {
+            $this->last = $other->last;
+        }
+    }
+
     /**
      * @return Entry<mixed>
      */
diff --git a/src/core/etl/src/Flow/ETL/Function/Max.php b/src/core/etl/src/Flow/ETL/Function/Max.php
index 6794783b8c..36b88e69f4 100644
--- a/src/core/etl/src/Flow/ETL/Function/Max.php
+++ b/src/core/etl/src/Flow/ETL/Function/Max.php
@@ -52,6 +52,25 @@ public function aggregate(Row $row, FlowContext $context): void
         }
     }
 
+    public function merge(AggregatingFunction $other): void
+    {
+        if (!$other instanceof self || !$this->ref->is($other->ref)) {
+            throw new InvalidArgumentException('Cannot merge ' . self::class . ' aggregating a different reference');
+        }
+
+        if ($other->max === null) {
+            return;
+        }
+
+        if ($this->max === null) {
+            $this->max = $other->max;
+
+            return;
+        }
+
+        $this->max = max($this->max, $other->max);
+    }
+
     /**
      * @return Entry<?\DateTimeInterface>|Entry<?float>|Entry<?int>
      */
diff --git a/src/core/etl/src/Flow/ETL/Function/Min.php b/src/core/etl/src/Flow/ETL/Function/Min.php
index 4960439cae..c3d5578fd5 100644
--- a/src/core/etl/src/Flow/ETL/Function/Min.php
+++ b/src/core/etl/src/Flow/ETL/Function/Min.php
@@ -52,6 +52,25 @@ public function aggregate(Row $row, FlowContext $context): void
         }
     }
 
+    public function merge(AggregatingFunction $other): void
+    {
+        if (!$other instanceof self || !$this->ref->is($other->ref)) {
+            throw new InvalidArgumentException('Cannot merge ' . self::class . ' aggregating a different reference');
+        }
+
+        if ($other->min === null) {
+            return;
+        }
+
+        if ($this->min === null) {
+            $this->min = $other->min;
+
+            return;
+        }
+
+        $this->min = min($this->min, $other->min);
+    }
+
     /**
      * @return Entry<?\DateTimeInterface>|Entry<?float>|Entry<?int>
      */
diff --git a/src/core/etl/src/Flow/ETL/Function/StringAggregate.php b/src/core/etl/src/Flow/ETL/Function/StringAggregate.php
index 13f1e468a3..f190a7a203 100644
--- a/src/core/etl/src/Flow/ETL/Function/StringAggregate.php
+++ b/src/core/etl/src/Flow/ETL/Function/StringAggregate.php
@@ -4,6 +4,7 @@
 
 namespace Flow\ETL\Function;
 
+use Flow\ETL\Exception\InvalidArgumentException;
 use Flow\ETL\FlowContext;
 use Flow\ETL\Row;
 use Flow\ETL\Row\Entry;
@@ -11,6 +12,7 @@
 use Flow\ETL\Row\Reference;
 use Flow\ETL\Row\SortOrder;
 
+use function array_merge;
 use function count;
 use function Flow\ETL\DSL\str_entry;
 use function implode;
@@ -40,6 +42,22 @@ public function aggregate(Row $row, FlowContext $context): void
         }
     }
 
+    public function merge(AggregatingFunction $other): void
+    {
+        if (
+            !$other instanceof self
+            || !$this->ref->is($other->ref)
+            || $this->separator !== $other->separator
+            || $this->sort !== $other->sort
+        ) {
+            throw new InvalidArgumentException(
+                'Cannot merge ' . self::class . ' aggregating a different reference, separator or sort',
+            );
+        }
+
+        $this->values = array_merge($this->values, $other->values);
+    }
+
     /**
      * @return Row\Entry<?string>
      */
diff --git a/src/core/etl/src/Flow/ETL/Function/Sum.php b/src/core/etl/src/Flow/ETL/Function/Sum.php
index 77415475c3..9addb4f814 100644
--- a/src/core/etl/src/Flow/ETL/Function/Sum.php
+++ b/src/core/etl/src/Flow/ETL/Function/Sum.php
@@ -68,6 +68,15 @@ public function apply(Row $row, Rows $partition, FlowContext $context): mixed
         return $sum;
     }
 
+    public function merge(AggregatingFunction $other): void
+    {
+        if (!$other instanceof self || !$this->ref->is($other->ref)) {
+            throw new InvalidArgumentException('Cannot merge ' . self::class . ' aggregating a different reference');
+        }
+
+        $this->sum = (new Calculator())->add($this->sum, $other->sum);
+    }
+
     public function over(Window $window): WindowFunction
     {
         $this->window = $window;
diff --git a/src/core/etl/src/Flow/ETL/GroupBy.php b/src/core/etl/src/Flow/ETL/GroupBy.php
index d606797cd4..264fbb886e 100644
--- a/src/core/etl/src/Flow/ETL/GroupBy.php
+++ b/src/core/etl/src/Flow/ETL/GroupBy.php
@@ -10,12 +10,15 @@
 use Flow\ETL\Exception\RuntimeException;
 use Flow\ETL\Function\AggregatingFunction;
 use Flow\ETL\Hash\NativePHPHash;
+use Flow\ETL\Row\EntryFactory;
 use Flow\ETL\Row\Reference;
 use Flow\ETL\Row\References;
+use Generator;
 use Stringable;
 
 use function array_filter;
 use function array_key_exists;
+use function array_map;
 use function array_unique;
 use function array_values;
 use function count;
@@ -24,45 +27,27 @@
 use function Flow\Types\DSL\type_integer;
 use function Flow\Types\DSL\type_string;
 use function Flow\Types\DSL\type_union;
-use function implode;
 use function is_array;
 use function is_scalar;
 use function serialize;
+use function strlen;
 
 final class GroupBy
 {
-    /**
-     * @var array<AggregatingFunction>
-     */
-    private array $aggregations;
+    private const int RESULT_BATCH_SIZE = 1000;
 
     /**
-     * @var array<string, array{values?: array<string, mixed>, aggregators: array<AggregatingFunction>}>
-     */
-    private array $groupedTable;
-
-    private ?Reference $pivot;
-
-    /**
-     * @var array<int, null|array<array-key, mixed>|bool|float|int|object|string>
+     * @var array<AggregatingFunction>
      */
-    private array $pivotColumns;
+    private array $aggregations = [];
 
-    /**
-     * @var array<string, array<string, AggregatingFunction|null|array<array-key, mixed>|bool|float|int|object|string>>
-     */
-    private array $pivotedTable;
+    private ?Reference $pivot = null;
 
     private readonly References $refs;
 
     public function __construct(string|Reference ...$entries)
     {
         $this->refs = References::init(...array_unique($entries));
-        $this->aggregations = [];
-        $this->groupedTable = [];
-        $this->pivotedTable = [];
-        $this->pivotColumns = [];
-        $this->pivot = null;
     }
 
     public function aggregate(AggregatingFunction ...$aggregator): void
@@ -80,22 +65,143 @@ public function aggregate(AggregatingFunction ...$aggregator): void
         $this->aggregations = $aggregator;
     }
 
-    public function group(Rows $rows, FlowContext $context): void
+    /**
+     * @param array<string, mixed> $keyValues
+     * @param array<AggregatingFunction> $aggregators
+     */
+    public function aggregatedRow(array $keyValues, array $aggregators, EntryFactory $entryFactory): Row
+    {
+        $entries = [];
+
+        /** @var mixed $value */
+        foreach ($keyValues as $name => $value) {
+            $entries[] = $entryFactory->create($name, $value);
+        }
+
+        foreach ($aggregators as $aggregator) {
+            $entries[] = $aggregator->result($entryFactory);
+        }
+
+        return Row::create(...$entries);
+    }
+
+    /**
+     * @return array<AggregatingFunction>
+     */
+    public function aggregations(): array
+    {
+        return $this->aggregations;
+    }
+
+    /**
+     * @param array<array-key, mixed> $values
+     */
+    public function hash(array $values): string
+    {
+        /** @var array<string> $stringValues */
+        $stringValues = [];
+
+        /** @var mixed $value */
+        foreach ($values as $value) {
+            if ($value === null) {
+                $stringValues[] = 'null';
+            } elseif (is_scalar($value)) {
+                $stringValues[] = (string) $value;
+            } else {
+                if ($value instanceof Stringable) {
+                    $stringValues[] = $value->__toString();
+                } elseif ($value instanceof DateTimeInterface) {
+                    $stringValues[] = $value->format(DateTimeImmutable::ATOM);
+                } elseif (is_array($value)) {
+                    $stringValues[] = $this->hash($value);
+                } else {
+                    $stringValues[] = serialize($value);
+                }
+            }
+        }
+
+        $key = '';
+
+        foreach ($stringValues as $stringValue) {
+            $key .= strlen($stringValue) . ':' . $stringValue;
+        }
+
+        return NativePHPHash::xxh128($key);
+    }
+
+    public function isPivot(): bool
+    {
+        return $this->pivot !== null;
+    }
+
+    /**
+     * @return array<string, mixed>
+     */
+    public function keyValues(Row $row): array
+    {
+        $values = [];
+
+        foreach ($this->refs as $ref) {
+            try {
+                $values[$ref->name()] = $row->valueOf($ref);
+            } catch (InvalidArgumentException) {
+                $values[$ref->name()] = null;
+            }
+        }
+
+        return $values;
+    }
+
+    /**
+     * @return array<AggregatingFunction>
+     */
+    public function newAggregators(): array
+    {
+        return array_map(
+            static fn(AggregatingFunction $aggregation): AggregatingFunction => clone $aggregation,
+            $this->aggregations,
+        );
+    }
+
+    public function pivot(Reference $ref): void
+    {
+        $this->pivot = $ref;
+    }
+
+    /**
+     * @param Generator<Rows> $rows
+     *
+     * @return Generator<Rows>
+     */
+    public function pivotResult(Generator $rows, FlowContext $context): Generator
     {
         $pivot = $this->pivot;
 
-        if ($pivot !== null) {
-            foreach ($rows as $row) {
+        if ($pivot === null) {
+            throw new RuntimeException('pivotResult() called without a pivot reference');
+        }
+
+        $aggregation = current($this->aggregations);
+
+        if (!$aggregation instanceof AggregatingFunction) {
+            throw new RuntimeException('Pivot requires exactly one aggregation');
+        }
+
+        /** @var array<int, null|array<array-key, mixed>|bool|float|int|object|string> $pivotColumns */
+        $pivotColumns = [];
+        /** @var array<string, array<string, AggregatingFunction|null|array<array-key, mixed>|bool|float|int|object|string>> $pivotedTable */
+        $pivotedTable = [];
+
+        foreach ($rows as $batch) {
+            foreach ($batch as $row) {
                 try {
-                    $this->pivotColumns[] = $row->valueOf($pivot);
+                    $pivotColumns[] = $row->valueOf($pivot);
                 } catch (InvalidArgumentException) {
-                    $this->pivotColumns[] = null;
+                    $pivotColumns[] = null;
                 }
             }
 
-            $this->pivotColumns = array_values(array_filter(array_unique($this->pivotColumns)));
-
-            foreach ($rows as $row) {
+            foreach ($batch as $row) {
                 $values = [];
 
                 foreach ($this->refs as $ref) {
@@ -103,15 +209,14 @@ public function group(Rows $rows, FlowContext $context): void
                 }
 
                 $indexValue = $this->hash($values);
-
                 $pivotValue = $row->valueOf($pivot);
 
-                if (!array_key_exists($indexValue, $this->pivotedTable)) {
-                    $this->pivotedTable[$indexValue] = [];
+                if (!array_key_exists($indexValue, $pivotedTable)) {
+                    $pivotedTable[$indexValue] = [];
                 }
 
                 foreach ($this->refs as $ref) {
-                    $this->pivotedTable[$indexValue][$ref->name()] = $row->valueOf($ref);
+                    $pivotedTable[$indexValue][$ref->name()] = $row->valueOf($ref);
                 }
 
                 if ($pivotValue === null) {
@@ -120,132 +225,54 @@ public function group(Rows $rows, FlowContext $context): void
 
                 $pivotValue = type_union(type_string(), type_integer())->assert($pivotValue);
 
-                if (!array_key_exists($pivotValue, $this->pivotedTable[$indexValue])) {
-                    // @mago-ignore analysis:invalid-property-assignment-value,possibly-invalid-clone
-                    $this->pivotedTable[$indexValue][$pivotValue] = clone current($this->aggregations);
+                if (!array_key_exists($pivotValue, $pivotedTable[$indexValue])) {
+                    $pivotedTable[$indexValue][$pivotValue] = clone $aggregation;
                 }
 
-                $aggregator = $this->pivotedTable[$indexValue][$pivotValue];
+                $aggregator = $pivotedTable[$indexValue][$pivotValue];
 
                 if ($aggregator instanceof AggregatingFunction) {
                     $aggregator->aggregate($row, $context);
                 }
             }
-        } else {
-            foreach ($rows as $row) {
-                /** @var array<string, null|mixed> $values */
-                $values = [];
-
-                foreach ($this->refs as $ref) {
-                    try {
-                        $values[$ref->name()] = $row->valueOf($ref);
-                    } catch (InvalidArgumentException) {
-                        $values[$ref->name()] = null;
-                    }
-                }
-
-                $valuesHash = $this->hash($values);
-
-                if (!array_key_exists($valuesHash, $this->groupedTable)) {
-                    $aggregators = [];
-
-                    foreach ($this->aggregations as $aggregator) {
-                        $aggregators[] = clone $aggregator;
-                    }
-
-                    $this->groupedTable[$valuesHash] = [
-                        'values' => $values,
-                        'aggregators' => $aggregators,
-                    ];
-                }
-
-                foreach ($this->groupedTable[$valuesHash]['aggregators'] as $aggregator) {
-                    $aggregator->aggregate($row, $context);
-                }
-            }
         }
-    }
 
-    public function pivot(Reference $ref): void
-    {
-        $this->pivot = $ref;
-    }
+        $pivotColumns = array_values(array_filter(array_unique($pivotColumns)));
 
-    public function result(FlowContext $context): Rows
-    {
-        $rows = [];
+        $buffer = [];
 
-        if ($this->pivot) {
-            foreach ($this->pivotedTable as $index => $columns) {
-                $row = [$this->refs->first()->name() => $index];
+        foreach ($pivotedTable as $index => $columns) {
+            $row = [$this->refs->first()->name() => $index];
 
-                foreach ($columns as $rowIndex => $values) {
-                    $row[$rowIndex] = $values instanceof AggregatingFunction
-                        ? $values->result($context->entryFactory())->value()
-                        : $values;
-                }
-
-                foreach ($this->pivotColumns as $column) {
-                    $column = type_union(type_string(), type_integer())->assert($column);
-
-                    if (!array_key_exists($column, $row)) {
-                        $row[$column] = null;
-                    }
-                }
-
-                $rows[] = $row;
+            foreach ($columns as $rowIndex => $value) {
+                $row[$rowIndex] = $value instanceof AggregatingFunction
+                    ? $value->result($context->entryFactory())->value()
+                    : $value;
             }
 
-            return array_to_rows($rows, $context->entryFactory());
-        }
-
-        foreach ($this->groupedTable as $group) {
-            $entries = [];
+            foreach ($pivotColumns as $column) {
+                $column = type_union(type_string(), type_integer())->assert($column);
 
-            /** @var mixed $value */
-            foreach ($group['values'] ?? [] as $entry => $value) {
-                $entries[] = $context->entryFactory()->create($entry, $value);
+                if (!array_key_exists($column, $row)) {
+                    $row[$column] = null;
+                }
             }
 
-            foreach ($group['aggregators'] as $aggregator) {
-                $entries[] = $aggregator->result($context->entryFactory());
-            }
+            $buffer[] = $row;
 
-            if (count($entries)) {
-                $rows[] = Row::create(...$entries);
+            if (count($buffer) >= self::RESULT_BATCH_SIZE) {
+                yield array_to_rows($buffer, $context->entryFactory());
+                $buffer = [];
             }
         }
 
-        return new Rows(...$rows);
+        if ($buffer !== []) {
+            yield array_to_rows($buffer, $context->entryFactory());
+        }
     }
 
-    /**
-     * @param array<array-key, mixed> $values
-     */
-    private function hash(array $values): string
+    public function references(): References
     {
-        /** @var array<string> $stringValues */
-        $stringValues = [];
-
-        /** @var mixed $value */
-        foreach ($values as $value) {
-            if ($value === null) {
-                $stringValues[] = 'null';
-            } elseif (is_scalar($value)) {
-                $stringValues[] = (string) $value;
-            } else {
-                if ($value instanceof Stringable) {
-                    $stringValues[] = $value->__toString();
-                } elseif ($value instanceof DateTimeInterface) {
-                    $stringValues[] = $value->format(DateTimeImmutable::ATOM);
-                } elseif (is_array($value)) {
-                    $stringValues[] = $this->hash($value);
-                } else {
-                    $stringValues[] = serialize($value);
-                }
-            }
-        }
-
-        return NativePHPHash::xxh128(implode('', $stringValues));
+        return $this->refs;
     }
 }
diff --git a/src/core/etl/src/Flow/ETL/GroupBy/BucketGroup.php b/src/core/etl/src/Flow/ETL/GroupBy/BucketGroup.php
new file mode 100644
index 0000000000..1737d13297
--- /dev/null
+++ b/src/core/etl/src/Flow/ETL/GroupBy/BucketGroup.php
@@ -0,0 +1,17 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\ETL\GroupBy;
+
+/**
+ * @internal
+ */
+final readonly class BucketGroup
+{
+    public function __construct(
+        public string $hash,
+        public int $bucketIndex,
+        public GroupState $state,
+    ) {}
+}
diff --git a/src/core/etl/src/Flow/ETL/GroupBy/BucketsCache/FilesystemGroupBucketsCache.php b/src/core/etl/src/Flow/ETL/GroupBy/BucketsCache/FilesystemGroupBucketsCache.php
new file mode 100644
index 0000000000..3e35c3897e
--- /dev/null
+++ b/src/core/etl/src/Flow/ETL/GroupBy/BucketsCache/FilesystemGroupBucketsCache.php
@@ -0,0 +1,99 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\ETL\GroupBy\BucketsCache;
+
+use Flow\ETL\GroupBy\GroupBucketsCache;
+use Flow\ETL\GroupBy\GroupState;
+use Flow\ETL\GroupBy\HashedGroup;
+use Flow\ETL\Hash\NativePHPHash;
+use Flow\Filesystem\Filesystem;
+use Flow\Filesystem\Path;
+use Flow\Serializer\NativePHPSerializer;
+use Flow\Serializer\Serializer;
+use Generator;
+
+use function strpos;
+use function substr;
+
+final readonly class FilesystemGroupBucketsCache implements GroupBucketsCache
+{
+    private Path $cacheDir;
+
+    /**
+     * @param int<1, max> $chunkSize
+     */
+    public function __construct(
+        private Filesystem $filesystem,
+        private Serializer $serializer = new NativePHPSerializer(),
+        private int $chunkSize = 100,
+        ?Path $cacheDir = null,
+    ) {
+        $this->cacheDir = ($cacheDir ?? $this->filesystem->getSystemTmpDir())->suffix('/flow-php-group-by/');
+    }
+
+    /**
+     * @return Generator<int, HashedGroup>
+     */
+    public function get(string $bucketId): Generator
+    {
+        $path = $this->keyPath($bucketId);
+
+        if (!$this->filesystem->status($path)) {
+            return;
+        }
+
+        $stream = $this->filesystem->readFrom($path);
+
+        foreach ($stream->readLines() as $line) {
+            $separator = strpos($line, ' ');
+
+            if ($separator === false) {
+                continue;
+            }
+
+            $hash = substr($line, 0, $separator);
+            $serialized = substr($line, $separator + 1);
+
+            yield new HashedGroup($hash, $this->serializer->unserialize($serialized, [GroupState::class]));
+        }
+
+        $stream->close();
+    }
+
+    public function remove(string $bucketId): void
+    {
+        $this->filesystem->rm($this->keyPath($bucketId)->parentDirectory());
+    }
+
+    public function set(string $bucketId, iterable $groups): void
+    {
+        $stream = $this->filesystem->writeTo($this->keyPath($bucketId));
+
+        $buffer = '';
+        $counter = 0;
+
+        foreach ($groups as $group) {
+            $buffer .= $group->hash . ' ' . $this->serializer->serialize($group->state) . "\n";
+            $counter++;
+
+            if ($counter >= $this->chunkSize) {
+                $stream->append($buffer);
+                $buffer = '';
+                $counter = 0;
+            }
+        }
+
+        if ($counter > 0) {
+            $stream->append($buffer);
+        }
+
+        $stream->close();
+    }
+
+    private function keyPath(string $key): Path
+    {
+        return $this->cacheDir->suffix(NativePHPHash::xxh128($key) . '/' . $key . '.php.cache');
+    }
+}
diff --git a/src/core/etl/src/Flow/ETL/GroupBy/GroupBucketsCache.php b/src/core/etl/src/Flow/ETL/GroupBy/GroupBucketsCache.php
new file mode 100644
index 0000000000..0db68173c1
--- /dev/null
+++ b/src/core/etl/src/Flow/ETL/GroupBy/GroupBucketsCache.php
@@ -0,0 +1,25 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\ETL\GroupBy;
+
+use Generator;
+
+/**
+ * @internal
+ */
+interface GroupBucketsCache
+{
+    /**
+     * @return Generator<int, HashedGroup>
+     */
+    public function get(string $bucketId): Generator;
+
+    public function remove(string $bucketId): void;
+
+    /**
+     * @param iterable<HashedGroup> $groups
+     */
+    public function set(string $bucketId, iterable $groups): void;
+}
diff --git a/src/core/etl/src/Flow/ETL/GroupBy/GroupState.php b/src/core/etl/src/Flow/ETL/GroupBy/GroupState.php
new file mode 100644
index 0000000000..ded39a6a25
--- /dev/null
+++ b/src/core/etl/src/Flow/ETL/GroupBy/GroupState.php
@@ -0,0 +1,22 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\ETL\GroupBy;
+
+use Flow\ETL\Function\AggregatingFunction;
+
+/**
+ * @internal
+ */
+final readonly class GroupState
+{
+    /**
+     * @param array<string, mixed> $keyValues
+     * @param array<AggregatingFunction> $aggregators
+     */
+    public function __construct(
+        public array $keyValues,
+        public array $aggregators,
+    ) {}
+}
diff --git a/src/core/etl/src/Flow/ETL/GroupBy/GroupsMinHeap.php b/src/core/etl/src/Flow/ETL/GroupBy/GroupsMinHeap.php
new file mode 100644
index 0000000000..8bd8180101
--- /dev/null
+++ b/src/core/etl/src/Flow/ETL/GroupBy/GroupsMinHeap.php
@@ -0,0 +1,36 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\ETL\GroupBy;
+
+use Flow\ETL\Exception\InvalidArgumentException;
+use SplMinHeap;
+
+/**
+ * @internal
+ *
+ * @extends SplMinHeap<BucketGroup>
+ */
+final class GroupsMinHeap extends SplMinHeap
+{
+    public function insert(mixed $value): true
+    {
+        if (!$value instanceof BucketGroup) {
+            throw new InvalidArgumentException('Value inserted into GroupsMinHeap must be a ' . BucketGroup::class);
+        }
+
+        parent::insert($value);
+
+        return true;
+    }
+
+    /**
+     * @param BucketGroup $value1
+     * @param BucketGroup $value2
+     */
+    protected function compare($value1, $value2): int
+    {
+        return [$value2->hash, $value2->bucketIndex] <=> [$value1->hash, $value1->bucketIndex];
+    }
+}
diff --git a/src/core/etl/src/Flow/ETL/GroupBy/HashAggregation.php b/src/core/etl/src/Flow/ETL/GroupBy/HashAggregation.php
new file mode 100644
index 0000000000..1519c1799c
--- /dev/null
+++ b/src/core/etl/src/Flow/ETL/GroupBy/HashAggregation.php
@@ -0,0 +1,194 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\ETL\GroupBy;
+
+use Flow\ETL\Dataset\Memory\Consumption;
+use Flow\ETL\Dataset\Memory\Unit;
+use Flow\ETL\FlowContext;
+use Flow\ETL\GroupBy;
+use Flow\ETL\Rows;
+use Generator;
+
+use function array_key_exists;
+use function bin2hex;
+use function count;
+use function ksort;
+use function random_bytes;
+
+/**
+ * @internal
+ */
+final class HashAggregation
+{
+    private const int BATCH_SIZE = 1000;
+
+    public function __construct(
+        private readonly Unit $memoryLimit,
+        private readonly GroupBucketsCache $cache,
+    ) {}
+
+    /**
+     * @param Generator<Rows> $rows
+     *
+     * @return Generator<Rows>
+     */
+    public function aggregate(Generator $rows, FlowContext $context, GroupBy $groupBy): Generator
+    {
+        $consumption = new Consumption(false);
+        /** @var array<string, GroupState> $hot */
+        $hot = [];
+        /** @var list<string> $buckets */
+        $buckets = [];
+
+        foreach ($rows as $batch) {
+            foreach ($batch as $row) {
+                $keyValues = $groupBy->keyValues($row);
+                $hash = $groupBy->hash($keyValues);
+
+                if (!array_key_exists($hash, $hot)) {
+                    $hot[$hash] = new GroupState($keyValues, $groupBy->newAggregators());
+                }
+
+                foreach ($hot[$hash]->aggregators as $aggregator) {
+                    $aggregator->aggregate($row, $context);
+                }
+            }
+
+            if ($consumption->currentDiff()->isGreaterThan($this->memoryLimit)) {
+                $buckets[] = $this->spill($hot);
+                $hot = [];
+                $consumption = new Consumption(false);
+            }
+        }
+
+        $groups = $buckets === [] ? $this->iterate($hot) : $this->merge($buckets, $hot);
+
+        yield from $this->emit($groups, $groupBy, $context);
+    }
+
+    /**
+     * @param Generator<GroupState> $groups
+     *
+     * @return Generator<Rows>
+     */
+    private function emit(Generator $groups, GroupBy $groupBy, FlowContext $context): Generator
+    {
+        $buffer = [];
+
+        foreach ($groups as $state) {
+            $buffer[] = $groupBy->aggregatedRow($state->keyValues, $state->aggregators, $context->entryFactory());
+
+            if (count($buffer) >= self::BATCH_SIZE) {
+                yield new Rows(...$buffer);
+                $buffer = [];
+            }
+        }
+
+        if ($buffer !== []) {
+            yield new Rows(...$buffer);
+        }
+    }
+
+    /**
+     * @param array<string, GroupState> $hot
+     *
+     * @return Generator<GroupState>
+     */
+    private function iterate(array $hot): Generator
+    {
+        foreach ($hot as $state) {
+            yield $state;
+        }
+    }
+
+    /**
+     * @param list<string> $buckets
+     * @param array<string, GroupState> $hot
+     *
+     * @return Generator<GroupState>
+     */
+    private function merge(array $buckets, array $hot): Generator
+    {
+        ksort($hot);
+
+        /** @var array<int, Generator<int, HashedGroup>> $cursors */
+        $cursors = [];
+
+        foreach ($buckets as $index => $bucketId) {
+            $cursors[$index] = $this->cache->get($bucketId);
+        }
+
+        $cursors[count($buckets)] = $this->hotCursor($hot);
+
+        $heap = new GroupsMinHeap();
+
+        foreach ($cursors as $index => $cursor) {
+            $this->advance($cursor, $index, $heap);
+        }
+
+        while (!$heap->isEmpty()) {
+            $current = $heap->extract();
+            $merged = $current->state;
+            $this->advance($cursors[$current->bucketIndex], $current->bucketIndex, $heap);
+
+            while (!$heap->isEmpty() && $heap->top()->hash === $current->hash) {
+                $next = $heap->extract();
+
+                foreach ($merged->aggregators as $index => $aggregator) {
+                    $aggregator->merge($next->state->aggregators[$index]);
+                }
+
+                $this->advance($cursors[$next->bucketIndex], $next->bucketIndex, $heap);
+            }
+
+            yield $merged;
+        }
+
+        foreach ($buckets as $bucketId) {
+            $this->cache->remove($bucketId);
+        }
+    }
+
+    /**
+     * @param Generator<int, HashedGroup> $cursor
+     */
+    private function advance(Generator $cursor, int $bucketIndex, GroupsMinHeap $heap): void
+    {
+        if (!$cursor->valid()) {
+            return;
+        }
+
+        /** @var HashedGroup $group */
+        $group = $cursor->current();
+        $heap->insert(new BucketGroup($group->hash, $bucketIndex, $group->state));
+        $cursor->next();
+    }
+
+    /**
+     * @param array<string, GroupState> $hot
+     *
+     * @return Generator<int, HashedGroup>
+     */
+    private function hotCursor(array $hot): Generator
+    {
+        foreach ($hot as $hash => $state) {
+            yield new HashedGroup($hash, $state);
+        }
+    }
+
+    /**
+     * @param array<string, GroupState> $hot
+     */
+    private function spill(array $hot): string
+    {
+        ksort($hot);
+
+        $bucketId = bin2hex(random_bytes(16));
+
+        $this->cache->set($bucketId, $this->hotCursor($hot));
+
+        return $bucketId;
+    }
+}
diff --git a/src/core/etl/src/Flow/ETL/GroupBy/HashedGroup.php b/src/core/etl/src/Flow/ETL/GroupBy/HashedGroup.php
new file mode 100644
index 0000000000..15830f06aa
--- /dev/null
+++ b/src/core/etl/src/Flow/ETL/GroupBy/HashedGroup.php
@@ -0,0 +1,16 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\ETL\GroupBy;
+
+/**
+ * @internal
+ */
+final readonly class HashedGroup
+{
+    public function __construct(
+        public string $hash,
+        public GroupState $state,
+    ) {}
+}
diff --git a/src/core/etl/src/Flow/ETL/Processor/GroupByProcessor.php b/src/core/etl/src/Flow/ETL/Processor/GroupByProcessor.php
index 46f0050ad6..ccc4eb59e5 100644
--- a/src/core/etl/src/Flow/ETL/Processor/GroupByProcessor.php
+++ b/src/core/etl/src/Flow/ETL/Processor/GroupByProcessor.php
@@ -6,6 +6,8 @@
 
 use Flow\ETL\FlowContext;
 use Flow\ETL\GroupBy;
+use Flow\ETL\GroupBy\BucketsCache\FilesystemGroupBucketsCache;
+use Flow\ETL\GroupBy\HashAggregation;
 use Flow\ETL\Processor;
 use Generator;
 
@@ -22,10 +24,24 @@ public function __construct(
 
     public function process(Generator $rows, FlowContext $context): Generator
     {
-        foreach ($rows as $batch) {
-            $this->groupBy->group($batch, $context);
+        if ($this->groupBy->isPivot()) {
+            yield from $this->groupBy->pivotResult($rows, $context);
+
+            return;
         }
 
-        yield $this->groupBy->result($context);
+        $config = $context->config->aggregation;
+
+        $aggregation = new HashAggregation(
+            $config->memoryLimit,
+            new FilesystemGroupBucketsCache(
+                $context->filesystem($config->filesystemProtocol),
+                $context->config->serializer(),
+                100,
+                $context->config->cache->localFilesystemCacheDir->suffix('/flow-php-group-by/'),
+            ),
+        );
+
+        yield from $aggregation->aggregate($rows, $context, $this->groupBy);
     }
 }
diff --git a/src/core/etl/tests/Flow/ETL/Tests/Integration/DataFrame/GroupByAggregationTest.php b/src/core/etl/tests/Flow/ETL/Tests/Integration/DataFrame/GroupByAggregationTest.php
new file mode 100644
index 0000000000..87ec052ccb
--- /dev/null
+++ b/src/core/etl/tests/Flow/ETL/Tests/Integration/DataFrame/GroupByAggregationTest.php
@@ -0,0 +1,145 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\ETL\Tests\Integration\DataFrame;
+
+use Flow\ETL\Config\ConfigBuilder;
+use Flow\ETL\Dataset\Memory\Unit;
+use Flow\ETL\Tests\Double\FakeRandomOrdersExtractor;
+use Flow\ETL\Tests\FlowIntegrationTestCase;
+
+use function Flow\ETL\DSL\collect;
+use function Flow\ETL\DSL\config_builder;
+use function Flow\ETL\DSL\count;
+use function Flow\ETL\DSL\df;
+use function Flow\ETL\DSL\first;
+use function Flow\ETL\DSL\from_array;
+use function Flow\ETL\DSL\last;
+use function Flow\ETL\DSL\ref;
+use function Flow\ETL\DSL\sum;
+use function is_numeric;
+
+final class GroupByAggregationTest extends FlowIntegrationTestCase
+{
+    private const int ORDERS = 20_000;
+
+    /**
+     * @var null|list<array{seller_id: string, email: string, customer: string, discount: null|float}>
+     */
+    private static ?array $orders = null;
+
+    public function test_first_and_last_preserve_input_order_across_spilled_buckets(): void
+    {
+        $data = [];
+
+        for ($i = 0; $i < 400; $i++) {
+            $data[] = ['g' => ($i % 2) === 0 ? 'a' : 'b', 'v' => $i];
+        }
+
+        $run = static fn(ConfigBuilder $config): array => df($config)
+            ->read(from_array($data))
+            ->groupBy('g')
+            ->aggregate(first(ref('v')), last(ref('v')))
+            ->sortBy(ref('g'))
+            ->fetch()
+            ->toArray();
+
+        $memory = $run(config_builder());
+        $spilled = $run(config_builder()->aggregationMemoryLimit(Unit::fromBytes(1)));
+
+        static::assertSame($memory, $spilled);
+        static::assertSame(
+            [
+                ['g' => 'a', 'v_first' => 0, 'v_last' => 398],
+                ['g' => 'b', 'v_first' => 1, 'v_last' => 399],
+            ],
+            $spilled,
+        );
+    }
+
+    public function test_high_cardinality_group_by_is_identical_when_spilling(): void
+    {
+        $memory = $this->aggregateByEmail(config_builder());
+        $spilled = $this->aggregateByEmail(config_builder()->aggregationMemoryLimit(Unit::fromKb(256)));
+
+        static::assertEqualsCanonicalizing($memory, $spilled);
+    }
+
+    public function test_multi_column_keys_do_not_collide(): void
+    {
+        $data = [
+            ['x' => 'a', 'y' => 'bc'],
+            ['x' => 'ab', 'y' => 'c'],
+            ['x' => 'a', 'y' => 'bc'],
+        ];
+
+        $run = static fn(ConfigBuilder $config): array => df($config)
+            ->read(from_array($data))
+            ->groupBy('x', 'y')
+            ->aggregate(count())
+            ->sortBy(ref('x'))
+            ->fetch()
+            ->toArray();
+
+        $expected = [
+            ['x' => 'a', 'y' => 'bc', '_count' => 2],
+            ['x' => 'ab', 'y' => 'c', '_count' => 1],
+        ];
+
+        static::assertSame($expected, $run(config_builder()));
+        static::assertSame($expected, $run(config_builder()->aggregationMemoryLimit(Unit::fromBytes(1))));
+    }
+
+    public function test_seller_aggregation_is_identical_when_spilling(): void
+    {
+        $memory = $this->aggregateBySeller(config_builder());
+        $spilled = $this->aggregateBySeller(config_builder()->aggregationMemoryLimit(Unit::fromKb(256)));
+
+        static::assertEqualsCanonicalizing($memory, $spilled);
+        static::assertCount(5, $memory);
+    }
+
+    private function aggregateByEmail(ConfigBuilder $config): array
+    {
+        return df($config)->read(from_array($this->orders()))->groupBy('email')->aggregate(count())->fetch()->toArray();
+    }
+
+    private function aggregateBySeller(ConfigBuilder $config): array
+    {
+        return df($config)
+            ->read(from_array($this->orders()))
+            ->groupBy('seller_id')
+            ->aggregate(count(), sum('discount'), collect('customer'))
+            ->fetch()
+            ->toArray();
+    }
+
+    /**
+     * @return list<array{seller_id: string, email: string, customer: string, discount: null|float}>
+     */
+    private function orders(): array
+    {
+        if (self::$orders !== null) {
+            return self::$orders;
+        }
+
+        $orders = [];
+
+        foreach ((new FakeRandomOrdersExtractor(self::ORDERS))->rawData() as $order) {
+            /** @var mixed $discount */
+            $discount = $order['discount'];
+
+            $orders[] = [
+                'seller_id' => (string) $order['seller_id'],
+                'email' => (string) $order['email'],
+                'customer' => (string) $order['customer'],
+                'discount' => is_numeric($discount) ? (float) $discount : null,
+            ];
+        }
+
+        self::$orders = $orders;
+
+        return $orders;
+    }
+}
diff --git a/src/core/etl/tests/Flow/ETL/Tests/Integration/GroupBy/FilesystemGroupBucketsCacheTest.php b/src/core/etl/tests/Flow/ETL/Tests/Integration/GroupBy/FilesystemGroupBucketsCacheTest.php
new file mode 100644
index 0000000000..a042c3ebfc
--- /dev/null
+++ b/src/core/etl/tests/Flow/ETL/Tests/Integration/GroupBy/FilesystemGroupBucketsCacheTest.php
@@ -0,0 +1,54 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\ETL\Tests\Integration\GroupBy;
+
+use Flow\ETL\GroupBy\BucketsCache\FilesystemGroupBucketsCache;
+use Flow\ETL\GroupBy\GroupState;
+use Flow\ETL\GroupBy\HashedGroup;
+use Flow\ETL\Tests\FlowIntegrationTestCase;
+
+use function Flow\ETL\DSL\ref;
+use function Flow\ETL\DSL\sum;
+use function iterator_to_array;
+
+final class FilesystemGroupBucketsCacheTest extends FlowIntegrationTestCase
+{
+    public function test_missing_bucket_yields_nothing(): void
+    {
+        $cache = new FilesystemGroupBucketsCache($this->fs(), $this->serializer(), 100, $this->cacheDir);
+
+        static::assertSame([], iterator_to_array($cache->get('does-not-exist')));
+    }
+
+    public function test_round_trip_and_remove(): void
+    {
+        $cache = new FilesystemGroupBucketsCache($this->fs(), $this->serializer(), 2, $this->cacheDir);
+
+        $groups = [
+            new HashedGroup('aaa', new GroupState(['type' => 'a'], [sum(ref('v'))])),
+            new HashedGroup('bbb', new GroupState(['type' => 'b'], [sum(ref('v'))])),
+            new HashedGroup('ccc', new GroupState(['type' => 'c'], [sum(ref('v'))])),
+        ];
+
+        $cache->set(
+            'bucket-1',
+            (static function () use ($groups) {
+                yield from $groups;
+            })(),
+        );
+
+        $read = [];
+
+        foreach ($cache->get('bucket-1') as $group) {
+            $read[$group->hash] = $group->state->keyValues;
+        }
+
+        static::assertSame(['aaa' => ['type' => 'a'], 'bbb' => ['type' => 'b'], 'ccc' => ['type' => 'c']], $read);
+
+        $cache->remove('bucket-1');
+
+        static::assertSame([], iterator_to_array($cache->get('bucket-1')));
+    }
+}
diff --git a/src/core/etl/tests/Flow/ETL/Tests/Unit/Config/Aggregation/AggregationConfigBuilderTest.php b/src/core/etl/tests/Flow/ETL/Tests/Unit/Config/Aggregation/AggregationConfigBuilderTest.php
new file mode 100644
index 0000000000..3dd6226210
--- /dev/null
+++ b/src/core/etl/tests/Flow/ETL/Tests/Unit/Config/Aggregation/AggregationConfigBuilderTest.php
@@ -0,0 +1,49 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\ETL\Tests\Unit\Config\Aggregation;
+
+use Flow\ETL\Config\Aggregation\AggregationConfig;
+use Flow\ETL\Config\Aggregation\AggregationConfigBuilder;
+use Flow\ETL\Dataset\Memory\Unit;
+use Flow\ETL\Tests\FlowTestCase;
+
+use function putenv;
+
+use const PHP_INT_MAX;
+
+final class AggregationConfigBuilderTest extends FlowTestCase
+{
+    public function test_default_memory_limit_is_unbounded(): void
+    {
+        $config = (new AggregationConfigBuilder())->build();
+
+        static::assertSame(PHP_INT_MAX, $config->memoryLimit->inBytes());
+        static::assertSame('file', $config->filesystemProtocol);
+    }
+
+    public function test_memory_limit_is_read_from_the_environment(): void
+    {
+        putenv(AggregationConfig::AGGREGATION_MAX_MEMORY_ENV . '=32M');
+
+        try {
+            $config = (new AggregationConfigBuilder())->build();
+
+            static::assertSame(Unit::fromMb(32)->inBytes(), $config->memoryLimit->inBytes());
+        } finally {
+            putenv(AggregationConfig::AGGREGATION_MAX_MEMORY_ENV);
+        }
+    }
+
+    public function test_setting_a_memory_limit_and_filesystem(): void
+    {
+        $config = (new AggregationConfigBuilder())
+            ->memoryLimit(Unit::fromMb(16))
+            ->filesystemProtocol('memory')
+            ->build();
+
+        static::assertSame(Unit::fromMb(16)->inBytes(), $config->memoryLimit->inBytes());
+        static::assertSame('memory', $config->filesystemProtocol);
+    }
+}
diff --git a/src/core/etl/tests/Flow/ETL/Tests/Unit/Function/AggregatingFunctionMergeTest.php b/src/core/etl/tests/Flow/ETL/Tests/Unit/Function/AggregatingFunctionMergeTest.php
new file mode 100644
index 0000000000..1c3b07defe
--- /dev/null
+++ b/src/core/etl/tests/Flow/ETL/Tests/Unit/Function/AggregatingFunctionMergeTest.php
@@ -0,0 +1,313 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\ETL\Tests\Unit\Function;
+
+use Flow\ETL\Exception\InvalidArgumentException;
+use Flow\ETL\Function\StringAggregate;
+use Flow\ETL\Row\SortOrder;
+use Flow\ETL\Tests\FlowTestCase;
+
+use function Flow\ETL\DSL\average;
+use function Flow\ETL\DSL\collect;
+use function Flow\ETL\DSL\collect_unique;
+use function Flow\ETL\DSL\count;
+use function Flow\ETL\DSL\first;
+use function Flow\ETL\DSL\flow_context;
+use function Flow\ETL\DSL\int_entry;
+use function Flow\ETL\DSL\last;
+use function Flow\ETL\DSL\max;
+use function Flow\ETL\DSL\min;
+use function Flow\ETL\DSL\ref;
+use function Flow\ETL\DSL\row;
+use function Flow\ETL\DSL\str_entry;
+use function Flow\ETL\DSL\sum;
+
+final class AggregatingFunctionMergeTest extends FlowTestCase
+{
+    public function test_average_merges_partial_sums_and_counts(): void
+    {
+        $context = flow_context();
+
+        $left = average(ref('v'));
+        $left->aggregate(row(int_entry('v', 1)), $context);
+        $left->aggregate(row(int_entry('v', 2)), $context);
+
+        $right = average(ref('v'));
+        $right->aggregate(row(int_entry('v', 6)), $context);
+
+        $left->merge($right);
+
+        static::assertSame(3, $left->result($context->entryFactory())->value());
+    }
+
+    public function test_average_rejects_a_different_scale(): void
+    {
+        $this->expectException(InvalidArgumentException::class);
+
+        average(ref('v'), 2)->merge(average(ref('v'), 4));
+    }
+
+    public function test_collect_merges_in_order(): void
+    {
+        $context = flow_context();
+
+        $left = collect(ref('v'));
+        $left->aggregate(row(int_entry('v', 1)), $context);
+
+        $right = collect(ref('v'));
+        $right->aggregate(row(int_entry('v', 2)), $context);
+        $right->aggregate(row(int_entry('v', 3)), $context);
+
+        $left->merge($right);
+
+        static::assertSame([1, 2, 3], $left->result($context->entryFactory())->value());
+    }
+
+    public function test_collect_unique_merges_as_a_union(): void
+    {
+        $context = flow_context();
+
+        $left = collect_unique(ref('v'));
+        $left->aggregate(row(int_entry('v', 1)), $context);
+        $left->aggregate(row(int_entry('v', 2)), $context);
+
+        $right = collect_unique(ref('v'));
+        $right->aggregate(row(int_entry('v', 2)), $context);
+        $right->aggregate(row(int_entry('v', 3)), $context);
+
+        $left->merge($right);
+
+        static::assertSame([1, 2, 3], $left->result($context->entryFactory())->value());
+    }
+
+    public function test_count_merges_counts(): void
+    {
+        $context = flow_context();
+
+        $left = count(ref('v'));
+        $left->aggregate(row(int_entry('v', 1)), $context);
+        $left->aggregate(row(int_entry('v', 2)), $context);
+
+        $right = count(ref('v'));
+        $right->aggregate(row(int_entry('v', 3)), $context);
+
+        $left->merge($right);
+
+        static::assertSame(3, $left->result($context->entryFactory())->value());
+    }
+
+    public function test_count_without_reference_merges(): void
+    {
+        $context = flow_context();
+
+        $left = count();
+        $left->aggregate(row(int_entry('v', 1)), $context);
+
+        $right = count();
+        $right->aggregate(row(int_entry('v', 2)), $context);
+
+        $left->merge($right);
+
+        static::assertSame(2, $left->result($context->entryFactory())->value());
+    }
+
+    public function test_count_rejects_a_referenced_and_unreferenced_mix(): void
+    {
+        $this->expectException(InvalidArgumentException::class);
+
+        count(ref('v'))->merge(count());
+    }
+
+    public function test_first_keeps_the_earlier_partial(): void
+    {
+        $context = flow_context();
+
+        $left = first(ref('v'));
+        $left->aggregate(row(int_entry('v', 10)), $context);
+
+        $right = first(ref('v'));
+        $right->aggregate(row(int_entry('v', 20)), $context);
+
+        $left->merge($right);
+
+        static::assertSame(10, $left->result($context->entryFactory())->value());
+    }
+
+    public function test_first_takes_the_other_when_empty(): void
+    {
+        $context = flow_context();
+
+        $left = first(ref('v'));
+
+        $right = first(ref('v'));
+        $right->aggregate(row(int_entry('v', 20)), $context);
+
+        $left->merge($right);
+
+        static::assertSame(20, $left->result($context->entryFactory())->value());
+    }
+
+    public function test_last_keeps_the_later_partial(): void
+    {
+        $context = flow_context();
+
+        $left = last(ref('v'));
+        $left->aggregate(row(int_entry('v', 10)), $context);
+
+        $right = last(ref('v'));
+        $right->aggregate(row(int_entry('v', 20)), $context);
+
+        $left->merge($right);
+
+        static::assertSame(20, $left->result($context->entryFactory())->value());
+    }
+
+    public function test_last_keeps_current_when_other_is_empty(): void
+    {
+        $context = flow_context();
+
+        $left = last(ref('v'));
+        $left->aggregate(row(int_entry('v', 10)), $context);
+
+        $left->merge(last(ref('v')));
+
+        static::assertSame(10, $left->result($context->entryFactory())->value());
+    }
+
+    public function test_max_keeps_the_greatest(): void
+    {
+        $context = flow_context();
+
+        $left = max(ref('v'));
+        $left->aggregate(row(int_entry('v', 5)), $context);
+
+        $right = max(ref('v'));
+        $right->aggregate(row(int_entry('v', 9)), $context);
+
+        $left->merge($right);
+
+        static::assertSame(9, $left->result($context->entryFactory())->value());
+    }
+
+    public function test_max_takes_the_other_when_empty(): void
+    {
+        $context = flow_context();
+
+        $left = max(ref('v'));
+
+        $right = max(ref('v'));
+        $right->aggregate(row(int_entry('v', 9)), $context);
+
+        $left->merge($right);
+
+        static::assertSame(9, $left->result($context->entryFactory())->value());
+    }
+
+    public function test_max_keeps_current_when_other_is_empty(): void
+    {
+        $context = flow_context();
+
+        $left = max(ref('v'));
+        $left->aggregate(row(int_entry('v', 5)), $context);
+
+        $left->merge(max(ref('v')));
+
+        static::assertSame(5, $left->result($context->entryFactory())->value());
+    }
+
+    public function test_merging_a_different_function_type_throws(): void
+    {
+        $this->expectException(InvalidArgumentException::class);
+
+        sum(ref('v'))->merge(count());
+    }
+
+    public function test_merging_the_same_function_on_a_different_reference_throws(): void
+    {
+        $this->expectException(InvalidArgumentException::class);
+
+        sum(ref('a'))->merge(sum(ref('b')));
+    }
+
+    public function test_min_keeps_the_smallest(): void
+    {
+        $context = flow_context();
+
+        $left = min(ref('v'));
+        $left->aggregate(row(int_entry('v', 5)), $context);
+
+        $right = min(ref('v'));
+        $right->aggregate(row(int_entry('v', 2)), $context);
+
+        $left->merge($right);
+
+        static::assertSame(2, $left->result($context->entryFactory())->value());
+    }
+
+    public function test_min_takes_the_other_when_empty(): void
+    {
+        $context = flow_context();
+
+        $left = min(ref('v'));
+
+        $right = min(ref('v'));
+        $right->aggregate(row(int_entry('v', 2)), $context);
+
+        $left->merge($right);
+
+        static::assertSame(2, $left->result($context->entryFactory())->value());
+    }
+
+    public function test_min_keeps_current_when_other_is_empty(): void
+    {
+        $context = flow_context();
+
+        $left = min(ref('v'));
+        $left->aggregate(row(int_entry('v', 5)), $context);
+
+        $left->merge(min(ref('v')));
+
+        static::assertSame(5, $left->result($context->entryFactory())->value());
+    }
+
+    public function test_string_aggregate_merges_values(): void
+    {
+        $context = flow_context();
+
+        $left = new StringAggregate(ref('v'), ',', SortOrder::ASC);
+        $left->aggregate(row(str_entry('v', 'b')), $context);
+
+        $right = new StringAggregate(ref('v'), ',', SortOrder::ASC);
+        $right->aggregate(row(str_entry('v', 'a')), $context);
+        $right->aggregate(row(str_entry('v', 'c')), $context);
+
+        $left->merge($right);
+
+        static::assertSame('a,b,c', $left->result($context->entryFactory())->value());
+    }
+
+    public function test_string_aggregate_rejects_a_different_separator(): void
+    {
+        $this->expectException(InvalidArgumentException::class);
+
+        (new StringAggregate(ref('v'), ','))->merge(new StringAggregate(ref('v'), ';'));
+    }
+
+    public function test_sum_merges_partial_sums(): void
+    {
+        $context = flow_context();
+
+        $left = sum(ref('v'));
+        $left->aggregate(row(int_entry('v', 1)), $context);
+        $left->aggregate(row(int_entry('v', 2)), $context);
+
+        $right = sum(ref('v'));
+        $right->aggregate(row(int_entry('v', 3)), $context);
+
+        $left->merge($right);
+
+        static::assertSame(6, $left->result($context->entryFactory())->value());
+    }
+}
diff --git a/src/core/etl/tests/Flow/ETL/Tests/Unit/GroupBy/GroupsMinHeapTest.php b/src/core/etl/tests/Flow/ETL/Tests/Unit/GroupBy/GroupsMinHeapTest.php
new file mode 100644
index 0000000000..5fe062c8cc
--- /dev/null
+++ b/src/core/etl/tests/Flow/ETL/Tests/Unit/GroupBy/GroupsMinHeapTest.php
@@ -0,0 +1,38 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\ETL\Tests\Unit\GroupBy;
+
+use Flow\ETL\Exception\InvalidArgumentException;
+use Flow\ETL\GroupBy\BucketGroup;
+use Flow\ETL\GroupBy\GroupsMinHeap;
+use Flow\ETL\GroupBy\GroupState;
+use Flow\ETL\Tests\FlowTestCase;
+
+final class GroupsMinHeapTest extends FlowTestCase
+{
+    public function test_orders_by_hash_then_bucket_index(): void
+    {
+        $heap = new GroupsMinHeap();
+        $heap->insert(new BucketGroup('b', 0, new GroupState([], [])));
+        $heap->insert(new BucketGroup('a', 2, new GroupState([], [])));
+        $heap->insert(new BucketGroup('a', 1, new GroupState([], [])));
+
+        $order = [];
+
+        while (!$heap->isEmpty()) {
+            $group = $heap->extract();
+            $order[] = $group->hash . $group->bucketIndex;
+        }
+
+        static::assertSame(['a1', 'a2', 'b0'], $order);
+    }
+
+    public function test_rejects_a_non_bucket_group(): void
+    {
+        $this->expectException(InvalidArgumentException::class);
+
+        (new GroupsMinHeap())->insert('not-a-bucket-group');
+    }
+}
diff --git a/src/core/etl/tests/Flow/ETL/Tests/Unit/GroupByTest.php b/src/core/etl/tests/Flow/ETL/Tests/Unit/GroupByTest.php
index 2d97fc5e58..89926efe6f 100644
--- a/src/core/etl/tests/Flow/ETL/Tests/Unit/GroupByTest.php
+++ b/src/core/etl/tests/Flow/ETL/Tests/Unit/GroupByTest.php
@@ -6,6 +6,8 @@
 
 use Flow\ETL\Exception\InvalidArgumentException;
 use Flow\ETL\GroupBy;
+use Flow\ETL\Processor\GroupByProcessor;
+use Flow\ETL\Rows;
 use Flow\ETL\Tests\FlowTestCase;
 
 use function Flow\ETL\DSL\config;
@@ -24,32 +26,20 @@ public function test_group_by_missing_entry(): void
     {
         $groupBy = new GroupBy('type');
 
-        $groupBy->group(
-            rows(row(str_entry('type', 'a')), row(str_entry('not-type', 'b')), row(str_entry('type', 'c'))),
-            flow_context(),
-        );
-
         static::assertEquals(
             rows(row(str_entry('type', 'a')), row(null_entry('type')), row(str_entry('type', 'c'))),
-            $groupBy->result(flow_context(config())),
+            $this->aggregate($groupBy, rows(
+                row(str_entry('type', 'a')),
+                row(str_entry('not-type', 'b')),
+                row(str_entry('type', 'c')),
+            )),
         );
     }
 
     public function test_group_by_with_aggregation(): void
     {
         $group = new GroupBy('type');
-
         $group->aggregate(sum(ref('id')));
-        $group->group(
-            rows(
-                row(int_entry('id', 1), str_entry('type', 'a')),
-                row(int_entry('id', 2), str_entry('type', 'b')),
-                row(int_entry('id', 3), str_entry('type', 'c')),
-                row(int_entry('id', 4), str_entry('type', 'a')),
-                row(int_entry('id', 5), str_entry('type', 'd')),
-            ),
-            flow_context(),
-        );
 
         static::assertEquals(
             rows(
@@ -58,7 +48,13 @@ public function test_group_by_with_aggregation(): void
                 row(int_entry('id_sum', 3), str_entry('type', 'c')),
                 row(int_entry('id_sum', 5), str_entry('type', 'd')),
             ),
-            $group->result(flow_context(config())),
+            $this->aggregate($group, rows(
+                row(int_entry('id', 1), str_entry('type', 'a')),
+                row(int_entry('id', 2), str_entry('type', 'b')),
+                row(int_entry('id', 3), str_entry('type', 'c')),
+                row(int_entry('id', 4), str_entry('type', 'a')),
+                row(int_entry('id', 5), str_entry('type', 'd')),
+            )),
         );
     }
 
@@ -72,27 +68,10 @@ public function test_group_by_with_empty_aggregations(): void
 
     public function test_group_by_with_pivoting(): void
     {
-        $rows = rows(
-            row(str_entry('product', 'Banana'), int_entry('amount', 1000), str_entry('country', 'USA')),
-            row(str_entry('product', 'Carrots'), int_entry('amount', 1500), str_entry('country', 'USA')),
-            row(str_entry('product', 'Beans'), int_entry('amount', 1600), str_entry('country', 'USA')),
-            row(str_entry('product', 'Orange'), int_entry('amount', 2000), str_entry('country', 'USA')),
-            row(str_entry('product', 'Orange'), int_entry('amount', 2000), str_entry('country', 'USA')),
-            row(str_entry('product', 'Banana'), int_entry('amount', 400), str_entry('country', 'China')),
-            row(str_entry('product', 'Carrots'), int_entry('amount', 1200), str_entry('country', 'China')),
-            row(str_entry('product', 'Beans'), int_entry('amount', 1500), str_entry('country', 'China')),
-            row(str_entry('product', 'Orange'), int_entry('amount', 4000), str_entry('country', 'China')),
-            row(str_entry('product', 'Banana'), int_entry('amount', 2000), str_entry('country', 'Canada')),
-            row(str_entry('product', 'Carrots'), int_entry('amount', 2000), str_entry('country', 'Canada')),
-            row(str_entry('product', 'Beans'), int_entry('amount', 2000), str_entry('country', 'Mexico')),
-        );
-
         $group = new GroupBy(ref('product'));
         $group->aggregate(sum(ref('amount')));
         $group->pivot(ref('country'));
 
-        $group->group($rows, flow_context());
-
         static::assertEquals(
             rows(
                 row(
@@ -124,29 +103,52 @@ public function test_group_by_with_pivoting(): void
                     int_entry('USA', 4000),
                 ),
             ),
-            $group->result(flow_context(config()))->sortBy(ref('product')),
+            $this->aggregate($group, rows(
+                row(str_entry('product', 'Banana'), int_entry('amount', 1000), str_entry('country', 'USA')),
+                row(str_entry('product', 'Carrots'), int_entry('amount', 1500), str_entry('country', 'USA')),
+                row(str_entry('product', 'Beans'), int_entry('amount', 1600), str_entry('country', 'USA')),
+                row(str_entry('product', 'Orange'), int_entry('amount', 2000), str_entry('country', 'USA')),
+                row(str_entry('product', 'Orange'), int_entry('amount', 2000), str_entry('country', 'USA')),
+                row(str_entry('product', 'Banana'), int_entry('amount', 400), str_entry('country', 'China')),
+                row(str_entry('product', 'Carrots'), int_entry('amount', 1200), str_entry('country', 'China')),
+                row(str_entry('product', 'Beans'), int_entry('amount', 1500), str_entry('country', 'China')),
+                row(str_entry('product', 'Orange'), int_entry('amount', 4000), str_entry('country', 'China')),
+                row(str_entry('product', 'Banana'), int_entry('amount', 2000), str_entry('country', 'Canada')),
+                row(str_entry('product', 'Carrots'), int_entry('amount', 2000), str_entry('country', 'Canada')),
+                row(str_entry('product', 'Beans'), int_entry('amount', 2000), str_entry('country', 'Mexico')),
+            ))->sortBy(ref('product')),
         );
     }
 
     public function test_group_by_with_pivoting_with_null_pivot_column(): void
     {
-        $rows = rows(
-            row(str_entry('product', 'Banana'), str_entry('country', 'USA'), int_entry('amount', 1000)),
-            row(str_entry('product', 'Apple'), str_entry('country', null), int_entry('amount', 400)),
-        );
-
         $group = new GroupBy(ref('product'));
         $group->aggregate(sum(ref('amount')));
         $group->pivot(ref('country'));
 
-        $group->group($rows, flow_context());
-
         static::assertEquals(
             rows(
                 row(str_entry('product', 'Apple'), null_entry('USA')),
                 row(str_entry('product', 'Banana'), int_entry('USA', 1000)),
             ),
-            $group->result(flow_context(config()))->sortBy(ref('product')),
+            $this->aggregate($group, rows(
+                row(str_entry('product', 'Banana'), str_entry('country', 'USA'), int_entry('amount', 1000)),
+                row(str_entry('product', 'Apple'), str_entry('country', null), int_entry('amount', 400)),
+            ))->sortBy(ref('product')),
         );
     }
+
+    private function aggregate(GroupBy $groupBy, Rows $input): Rows
+    {
+        $result = new Rows();
+
+        $output = (new GroupByProcessor($groupBy))->process((static fn() => yield $input)(), flow_context(config()));
+
+        /** @var Rows $batch */
+        foreach ($output as $batch) {
+            $result = $result->merge($batch);
+        }
+
+        return $result;
+    }
 }
diff --git a/src/core/etl/tests/Flow/ETL/Tests/Unit/Processor/GroupByProcessorTest.php b/src/core/etl/tests/Flow/ETL/Tests/Unit/Processor/GroupByProcessorTest.php
index fa42188bb7..0659e97f71 100644
--- a/src/core/etl/tests/Flow/ETL/Tests/Unit/Processor/GroupByProcessorTest.php
+++ b/src/core/etl/tests/Flow/ETL/Tests/Unit/Processor/GroupByProcessorTest.php
@@ -89,7 +89,12 @@ public function test_handles_empty_input(): void
         /** @var list<Rows> $result */
         $result = iterator_to_array($processor->process($generator, flow_context()));
 
-        static::assertCount(1, $result);
-        static::assertCount(0, $result[0]);
+        $total = 0;
+
+        foreach ($result as $batch) {
+            $total += $batch->count();
+        }
+
+        static::assertSame(0, $total);
     }
 }