#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

@@ -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

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',
+    ) {}
+}

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;
+    }
+}

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(),
         );
     }
 

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

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

src/core/etl/src/Flow/ETL/Function/StringAggregate.php

@@ -4,13 +4,15 @@
 
 namespace Flow\ETL\Function;
 
+use Flow\ETL\Exception\InvalidArgumentException;
 use Flow\ETL\FlowContext;
 use Flow\ETL\Row;
 use Flow\ETL\Row\Entry;
 use Flow\ETL\Row\EntryFactory;
 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>
      */

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;