#1390 - configurable storage for aggregating functions

.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
+opcache.enable_cli=0
+apc.enabled=1
+apc.enable_cli=1
+apc.shm_size=128M

.nix/pkgs/flow-php/package.nix

@@ -22,6 +22,7 @@ let
     with all;
     enabled
         ++ [
+            apcu
             bcmath
             dom
             mbstring

src/core/etl/src/Flow/ETL/Config.php

@@ -6,6 +6,7 @@
 
 use Flow\ETL\Config\Cache\CacheConfig;
 use Flow\ETL\Config\ConfigBuilder;
+use Flow\ETL\Config\Grouping\GroupingConfig;
 use Flow\ETL\Config\Sort\SortConfig;
 use Flow\ETL\Config\Telemetry\TelemetryConfig;
 use Flow\ETL\Filesystem\FilesystemStreams;
@@ -36,6 +37,7 @@ public function __construct(
         public SortConfig $sort,
         private ?Analyze $analyze,
         public TelemetryConfig $telemetry,
+        public GroupingConfig $grouping,
     ) {}
 
     public static function builder(): ConfigBuilder

src/core/etl/src/Flow/ETL/Config/ConfigBuilder.php

@@ -9,6 +9,7 @@
 use Flow\ETL\Cache;
 use Flow\ETL\Config;
 use Flow\ETL\Config\Cache\CacheConfigBuilder;
+use Flow\ETL\Config\Grouping\GroupingConfigBuilder;
 use Flow\ETL\Config\Sort\SortConfigBuilder;
 use Flow\ETL\Config\Telemetry\TelemetryConfig;
 use Flow\ETL\Config\Telemetry\TelemetryOptions;
@@ -36,6 +37,8 @@ final class ConfigBuilder
 {
     public readonly CacheConfigBuilder $cache;
 
+    public readonly GroupingConfigBuilder $grouping;
+
     public readonly SortConfigBuilder $sort;
 
     private ?Analyze $analyze;
@@ -70,6 +73,7 @@ public function __construct()
         $this->optimizer = null;
         $this->clock = null;
         $this->cache = new CacheConfigBuilder();
+        $this->grouping = new GroupingConfigBuilder();
         $this->sort = new SortConfigBuilder();
         $this->randomValueGenerator = new NativePHPRandomValueGenerator();
         $this->analyze = null;
@@ -111,6 +115,7 @@ public function build(EntryFactory $entryFactory = new EntryFactory()): Config
             $this->sort->build(),
             $this->analyze,
             $this->telemetryConfig ?? TelemetryConfig::default($this->getClock()),
+            $this->grouping->build(),
         );
     }
 
@@ -159,6 +164,23 @@ public function externalSortFilesystem(string $protocol): self
         return $this;
     }
 
+    public function groupingFilesystem(string $protocol = 'file'): self
+    {
+        $this->grouping->filesystemProtocol($protocol);
+
+        return $this;
+    }
+
+    /**
+     * @param int<1, max> $size
+     */
+    public function groupingRunSize(int $size): self
+    {
+        $this->grouping->runSize($size);
+
+        return $this;
+    }
+
     public function id(string $id): self
     {
         $this->id = $id;

src/core/etl/src/Flow/ETL/Config/Grouping/GroupingBackend.php

@@ -0,0 +1,11 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\ETL\Config\Grouping;
+
+enum GroupingBackend
+{
+    case Filesystem;
+    case Memory;
+}

src/core/etl/src/Flow/ETL/Config/Grouping/GroupingConfig.php

@@ -0,0 +1,19 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\ETL\Config\Grouping;
+
+final readonly class GroupingConfig
+{
+    /**
+     * @param int<1, max> $runSize
+     * @param int<1, max> $bucketsCount
+     */
+    public function __construct(
+        public GroupingBackend $backend,
+        public int $runSize = 10_000,
+        public int $bucketsCount = 10,
+        public string $filesystemProtocol = 'file',
+    ) {}
+}

src/core/etl/src/Flow/ETL/Config/Grouping/GroupingConfigBuilder.php

@@ -0,0 +1,58 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\ETL\Config\Grouping;
+
+final class GroupingConfigBuilder
+{
+    private GroupingBackend $backend = GroupingBackend::Memory;
+
+    /** @var int<1, max> */
+    private int $bucketsCount = 10;
+
+    private string $filesystemProtocol = 'file';
+
+    /** @var int<1, max> */
+    private int $runSize = 10_000;
+
+    public function backend(GroupingBackend $backend): self
+    {
+        $this->backend = $backend;
+
+        return $this;
+    }
+
+    /**
+     * @param int<1, max> $bucketsCount
+     */
+    public function bucketsCount(int $bucketsCount): self
+    {
+        $this->bucketsCount = $bucketsCount;
+
+        return $this;
+    }
+
+    public function build(): GroupingConfig
+    {
+        return new GroupingConfig($this->backend, $this->runSize, $this->bucketsCount, $this->filesystemProtocol);
+    }
+
+    public function filesystemProtocol(string $protocol): self
+    {
+        $this->backend = GroupingBackend::Filesystem;
+        $this->filesystemProtocol = $protocol;
+
+        return $this;
+    }
+
+    /**
+     * @param int<1, max> $runSize
+     */
+    public function runSize(int $runSize): self
+    {
+        $this->runSize = $runSize;
+
+        return $this;
+    }
+}

src/core/etl/src/Flow/ETL/DataFrame/GroupedDataFrame.php

@@ -21,15 +21,15 @@ public function aggregate(AggregatingFunction ...$aggregations): DataFrame
     {
         $this->groupBy->aggregate(...$aggregations);
 
-        $pipelineAdder = function (GroupBy $groupBy): void {
-            // @mago-ignore analysis:non-existent-property,method-access-on-null
-            $this->pipeline->add(new GroupByProcessor($groupBy));
-        };
-
-        // @mago-ignore analysis:invalid-method-access
-        $pipelineAdder->bindTo($this->df, $this->df)($this->groupBy);
+        return $this->addProcessor(new GroupByProcessor($this->groupBy));
+    }
 
-        return $this->df;
+    /**
+     * @param null|int<1, max> $size
+     */
+    public function getGroups(?int $size = null): DataFrame
+    {
+        return $this->addProcessor(new GroupByProcessor($this->groupBy, $size ?? GroupByProcessor::DEFAULT_BATCH_SIZE));
     }
 
     public function pivot(Reference $ref): self
@@ -38,4 +38,17 @@ public function pivot(Reference $ref): self
 
         return $this;
     }
+
+    private function addProcessor(GroupByProcessor $processor): DataFrame
+    {
+        $pipelineAdder = function (GroupByProcessor $processor): void {
+            // @mago-ignore analysis:non-existent-property,method-access-on-null
+            $this->pipeline->add($processor);
+        };
+
+        // @mago-ignore analysis:invalid-method-access
+        $pipelineAdder->bindTo($this->df, $this->df)($processor);
+
+        return $this->df;
+    }
 }

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>
      */

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;

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>
      */

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>
      */

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;

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>
      */

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>
      */

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>
      */