#1390 - configurable backend for group by buckets

.github/workflows/job-mutation-tests.yml

@@ -28,6 +28,8 @@ jobs:
           dependencies: "locked"
           coverage: "pcov"
           cache-key-suffix: "-locked"
+          extensions: ':psr, apcu, bcmath, dom, hash, json, mbstring, xml, xmlwriter, xmlreader, zlib'
+          ini-values: 'memory_limit=-1, apc.enable_cli=1'
 
       - name: "Create cache directory"
         run: "mkdir -p var/infection/cache"

.github/workflows/job-tests.yml

@@ -96,8 +96,8 @@ jobs:
           php-version: "${{ matrix.php-version }}"
           dependencies: "${{ matrix.dependencies }}"
           coverage: ${{ matrix.php-version == '8.3' && 'pcov' || 'none' }}
-          extensions: ':psr, bcmath, dom, hash, json, mbstring, xml, xmlwriter, xmlreader, zlib, curl, pgsql, grpc, protobuf'
-          ini-values: 'memory_limit=-1, post_max_size=32M, upload_max_filesize=32M'
+          extensions: ':psr, apcu, bcmath, dom, hash, json, mbstring, xml, xmlwriter, xmlreader, zlib, curl, pgsql, grpc, protobuf'
+          ini-values: 'memory_limit=-1, post_max_size=32M, upload_max_filesize=32M, apc.enable_cli=1'
           apt-packages: "build-essential autoconf automake libtool protobuf-compiler libprotobuf-c-dev"
           pie-extensions: "flow-php/pg-query-ext:1.x-dev"
 

.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

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

src/core/etl/composer.json

@@ -23,6 +23,7 @@
         "symfony/string": "^6.4 || ^7.4 || ^8.0"
     },
     "suggest": {
+        "ext-apcu": "Allows using ApcuCache as a pipeline cache implementation",
         "ext-bcmath": "BCMath extension for PHP for more precise calculations"
     },
     "require-dev": {

src/core/etl/src/Flow/ETL/Cache/Implementation/ApcuCache.php

@@ -0,0 +1,100 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\ETL\Cache\Implementation;
+
+use APCUIterator;
+use Flow\ETL\Cache;
+use Flow\ETL\Exception\KeyNotInCacheException;
+use Flow\ETL\Exception\RuntimeException;
+use Flow\ETL\Rows;
+use Generator;
+
+use function apcu_delete;
+use function apcu_enabled;
+use function apcu_exists;
+use function apcu_fetch;
+use function apcu_store;
+use function extension_loaded;
+use function preg_quote;
+use function sprintf;
+
+final readonly class ApcuCache implements Cache
+{
+    public function __construct(
+        private string $namespace = 'flow_php_cache',
+    ) {
+        if (!extension_loaded('apcu') || !apcu_enabled()) {
+            throw new RuntimeException(
+                'ApcuCache requires the APCu extension (ext-apcu) with apc.enabled=1'
+                . (PHP_SAPI === 'cli' ? ' and apc.enable_cli=1' : ''),
+            );
+        }
+    }
+
+    public function clear(): void
+    {
+        $iterator = new APCUIterator('/^' . preg_quote($this->namespace . ':', '/') . '/');
+
+        apcu_delete($iterator);
+    }
+
+    public function delete(string $key): void
+    {
+        apcu_delete($this->namespacedKey($key));
+    }
+
+    public function get(string $key): Rows
+    {
+        $success = false;
+
+        // @mago-ignore analysis:mixed-assignment
+        $value = apcu_fetch($this->namespacedKey($key), $success);
+
+        if (!$success) {
+            throw new KeyNotInCacheException($key);
+        }
+
+        if (!$value instanceof Rows) {
+            throw new RuntimeException(sprintf(
+                'Cache entry for key "%s" is corrupted or was not written by ApcuCache.',
+                $key,
+            ));
+        }
+
+        return $value;
+    }
+
+    public function has(string $key): bool
+    {
+        return (bool) apcu_exists($this->namespacedKey($key));
+    }
+
+    /**
+     * @throws KeyNotInCacheException
+     *
+     * @return Generator<int, Rows>
+     */
+    public function read(string $key): Generator
+    {
+        yield $this->get($key);
+    }
+
+    public function set(string $key, Rows $value): void
+    {
+        $result = apcu_store($this->namespacedKey($key), $value);
+
+        if ($result === false) {
+            throw new RuntimeException(sprintf(
+                'Failed to store cache entry for key "%s" in APCu, the cache segment might be full.',
+                $key,
+            ));
+        }
+    }
+
+    private function namespacedKey(string $key): string
+    {
+        return $this->namespace . ':' . $key;
+    }
+}

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

@@ -7,6 +7,7 @@
 use Flow\Calculator\Calculator;
 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;
@@ -37,6 +38,7 @@ public function __construct(
         public SortConfig $sort,
         private ?Analyze $analyze,
         public TelemetryConfig $telemetry,
+        public GroupingConfig $grouping,
         private Calculator $calculator = new Calculator(),
     ) {}
 

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;
@@ -19,6 +20,7 @@
 use Flow\ETL\Pipeline\Optimizer\LimitOptimization;
 use Flow\ETL\RandomValueGenerator;
 use Flow\ETL\Row\EntryFactory;
+use Flow\ETL\Sort\ExternalSort\BucketsCache;
 use Flow\Filesystem\Filesystem;
 use Flow\Filesystem\FilesystemTable;
 use Flow\Floe\FloeSerializer;
@@ -34,6 +36,8 @@ final class ConfigBuilder
 {
     public readonly CacheConfigBuilder $cache;
 
+    public readonly GroupingConfigBuilder $grouping;
+
     public readonly SortConfigBuilder $sort;
 
     private ?Analyze $analyze;
@@ -68,6 +72,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;
@@ -86,16 +91,18 @@ public function analyze(Analyze $analyze): self
 
     public function build(EntryFactory $entryFactory = new EntryFactory()): Config
     {
-        $this->id ??= 'flow-php-' . $this->randomValueGenerator->string(32);
+        $id = $this->id ??= 'flow-php-' . $this->randomValueGenerator->string(32);
         $this->serializer ??= new FloeSerializer();
         $this->optimizer ??= new Optimizer(new LimitOptimization(), new BatchSizeOptimization(batchSize: 1000));
 
         $serializer = $this->serializer;
         $optimizer = $this->optimizer;
         $dataframeName = $this->name ?? 'flow_dataframe';
 
+        $cacheConfig = $this->cache->build($this->fstab(), $this->telemetryConfig, $dataframeName);
+
         return new Config(
-            $this->id,
+            $id,
             $dataframeName,
             $this->version,
             $serializer,
@@ -105,10 +112,11 @@ public function build(EntryFactory $entryFactory = new EntryFactory()): Config
             $optimizer,
             $this->putInputIntoRows,
             $entryFactory,
-            $this->cache->build($this->fstab(), $this->telemetryConfig, $dataframeName),
+            $cacheConfig,
             $this->sort->build(),
             $this->analyze,
             $this->telemetryConfig ?? TelemetryConfig::default($this->getClock()),
+            $this->grouping->build($this->fstab(), $cacheConfig->localFilesystemCacheDir),
         );
     }
 
@@ -187,6 +195,33 @@ public function externalSortFilesystem(string $protocol): self
         return $this;
     }
 
+    /**
+     * @param int<1, max> $batchSize
+     */
+    public function groupingBatchSize(int $batchSize): self
+    {
+        $this->grouping->batchSize($batchSize);
+
+        return $this;
+    }
+
+    /**
+     * @param int<1, max> $bucketsCount
+     */
+    public function groupingBucketsCount(int $bucketsCount): self
+    {
+        $this->grouping->bucketsCount($bucketsCount);
+
+        return $this;
+    }
+
+    public function groupingCache(BucketsCache $cache): self
+    {
+        $this->grouping->cache($cache);
+
+        return $this;
+    }
+
     public function id(string $id): self
     {
         $this->id = $id;

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

@@ -0,0 +1,20 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\ETL\Config\Grouping;
+
+use Flow\ETL\Sort\ExternalSort\BucketsCache;
+
+final readonly class GroupingConfig
+{
+    /**
+     * @param int<1, max> $bucketsCount
+     * @param int<1, max> $batchSize
+     */
+    public function __construct(
+        public BucketsCache $cache,
+        public int $bucketsCount = 64,
+        public int $batchSize = 1000,
+    ) {}
+}

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

@@ -0,0 +1,76 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\ETL\Config\Grouping;
+
+use Flow\ETL\Exception\InvalidArgumentException;
+use Flow\ETL\Sort\ExternalSort\BucketsCache;
+use Flow\ETL\Sort\ExternalSort\BucketsCache\FilesystemBucketsCache;
+use Flow\Filesystem\FilesystemTable;
+use Flow\Filesystem\Path;
+
+final class GroupingConfigBuilder
+{
+    /**
+     * @var int<1, max>
+     */
+    private int $batchSize = 1000;
+
+    /**
+     * @var int<1, max>
+     */
+    private int $bucketsCount = 64;
+
+    private ?BucketsCache $cache = null;
+
+    private string $filesystemMount = 'file';
+
+    /**
+     * @param int<1, max> $batchSize
+     */
+    public function batchSize(int $batchSize): self
+    {
+        // @mago-ignore analysis:impossible-condition,redundant-comparison
+        if ($batchSize < 1) {
+            throw new InvalidArgumentException('Batch size must be at least 1');
+        }
+
+        $this->batchSize = $batchSize;
+
+        return $this;
+    }
+
+    public function build(FilesystemTable $filesystemTable, Path $localFilesystemCacheDir): GroupingConfig
+    {
+        $cache = $this->cache ?? new FilesystemBucketsCache(
+            $filesystemTable->for($this->filesystemMount),
+            $localFilesystemCacheDir->suffix('/flow-php-group-by/'),
+            $this->batchSize,
+        );
+
+        return new GroupingConfig($cache, $this->bucketsCount, $this->batchSize);
+    }
+
+    /**
+     * @param int<1, max> $bucketsCount
+     */
+    public function bucketsCount(int $bucketsCount): self
+    {
+        // @mago-ignore analysis:impossible-condition,redundant-comparison
+        if ($bucketsCount < 1) {
+            throw new InvalidArgumentException('Buckets count must be greater than 0');
+        }
+
+        $this->bucketsCount = $bucketsCount;
+
+        return $this;
+    }
+
+    public function cache(BucketsCache $cache): self
+    {
+        $this->cache = $cache;
+
+        return $this;
+    }
+}

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

@@ -21,15 +21,7 @@ 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->df;
+        return $this->addProcessor(new GroupByProcessor($this->groupBy));
     }
 
     public function pivot(Reference $ref): self
@@ -38,4 +30,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;
+    }
 }