#1390 - configurable storage for aggregating functions#2473
Conversation
GroupBy aggregation can move state out of the PHP heap instead of holding every result in memory. A new AggregationStorage abstraction offers three backends: memory (default, unchanged behavior), filesystem spill+merge, and PSR-16 key-value (APCu/Redis). - MergeableAggregatingFunction + merge() on all built-in aggregators - AggregationStorage: Memory / External (filesystem spill+merge) / Kv (PSR-16) - KV storage fails loudly on eviction / write failure instead of returning a wrong aggregate - config: AggregationStorageStrategy + FLOW_AGGREGATION_MAX_MEMORY wired into ConfigBuilder - APCu enabled in the nix dev shell (for the PSR-16/APCu storage test)
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## 1.x #2473 +/- ##
============================================
+ Coverage 85.97% 85.99% +0.02%
- Complexity 0 22337 +22337
============================================
Files 1662 1670 +8
Lines 68370 68583 +213
============================================
+ Hits 58783 58981 +198
- Misses 9587 9602 +15 🚀 New features to boost your workflow:
|
This comment has been minimized.
This comment has been minimized.
| */ | ||
| public function __construct( | ||
| public array $values, | ||
| public array $aggregators, |
There was a problem hiding this comment.
I'm not really sure if I understand this data structure, it's keeping values for all aggregators?
| return $this; | ||
| } | ||
|
|
||
| public function aggregationStorage(AggregationStorageStrategy $strategy): self |
| return $this; | ||
| } | ||
|
|
||
| public function aggregationStore(AggregationStorage $storage): self |
There was a problem hiding this comment.
Those two methods, aggregationStorage and aggregationStore are a bit confusing
|
|
||
| public function flush(): void; | ||
|
|
||
| public function save(string $hash, GroupEntry $entry): void; |
There was a problem hiding this comment.
what's the difference between entry and save?
|
|
||
| namespace Flow\ETL\Function; | ||
|
|
||
| interface MergeableAggregatingFunction extends AggregatingFunction |
There was a problem hiding this comment.
could you walk me through this logic? Like when two AggregatingFunctions can be merged?
…torage Reworks the aggregation-storage approach following review. External aggregation is now sort-then-fold: sort rows by the group-by columns (reusing ExternalSort, which spills and merges in bounded memory) and fold adjacent equal keys. This makes the whole abstraction simpler and drops the concepts the review found confusing. - remove the AggregationStorage layer (Memory/External/Kv storage, GroupEntry, GroupBucket(sCache)) and the AggregationStorageStrategy enum - remove MergeableAggregatingFunction and merge() from aggregators - sort-fold never splits a group, so no merge contract is needed - add AggregatingAlgorithm with MemoryAggregation (default, hash map) and ExternalAggregation (sort + fold), selected via AggregationAlgorithms (mirrors SortAlgorithms) - GroupBy is now a spec (references/aggregations/hash/keyValues); GroupByProcessor routes pivot and global aggregation to memory - config: aggregationAlgorithm() replaces aggregationStorage()/aggregationStore(); output is streamed in batches instead of one Rows - external tags rows with an input ordinal and sorts by (group keys, ordinal), so first()/last()/collect() return the same result as the in-memory path even when a group spills across sort runs
Non-pivot aggregation now runs through hash aggregation with adaptive spilling: groups accumulate in an in-memory hash map, and once the configured memory limit is exceeded the map is spilled to a hash-sorted bucket and cleared. If nothing spills the map is the result; otherwise a streaming k-way merge folds partial aggregates by group key, so spilled state is O(groups), not O(rows). - merge() moves onto the base AggregatingFunction contract (no separate opt-in interface); every built-in aggregator implements it and self-validates by reference and result-affecting arguments - HashAggregation + GroupState / BucketGroup / GroupsMinHeap / GroupBucketsCache (FilesystemGroupBucketsCache), mirroring the external-sort building blocks - GroupBy becomes a spec (references/aggregations/hash/keyValues/newAggregators); pivot stays memory-only - config: aggregationMemoryLimit(Unit) (default unbounded, opt-in spilling) replaces the algorithm enum; output streamed in batches - fix multi-column group-by key hash collision (length-prefix the values)
- guard strpos()===false and drop unused mago pragmas flagged by CI static analysis
- annotate the bucket-cursor generators as Generator<int, array{...}> and narrow current() in the k-way merge
- cover merge() on all 10 aggregators (success, null branches, self-validation), the min-heap ordering/guard, the bucket cache round-trip, and the memory-limit env var
- drop the unused mago pragma in Sum::merge - hoist the pivot prototype behind an instanceof guard so clone is never applied to a possibly-false current() - drop over-specific @return docblocks that toArray() cannot satisfy
| private function hotCursor(array $hot): Generator | ||
| { | ||
| foreach ($hot as $hash => $state) { | ||
| yield [$hash, $state]; |
There was a problem hiding this comment.
please don't use tuples for return types, introduce a simple VO instead
| return new AggregationConfig($this->memoryLimit, $this->filesystemProtocol); | ||
| } | ||
|
|
||
| public function filesystemProtocol(string $protocol): self |
There was a problem hiding this comment.
we are not really using protocol, instead I started using mount as there can be multiple filesystems that share protocol but they might be registered under different mount names. For example when you want to move S3 to S3 through FSTab
| new FilesystemGroupBucketsCache( | ||
| $context->filesystem($config->filesystemProtocol), | ||
| $context->config->serializer(), | ||
| 100, |
Makes GroupBy aggregation memory-efficient and configurable (issue #1390): keep the fast in-memory path by default, or opt into an external, bounded-memory path.
The external path is sort-then-fold: sort the rows by the group-by columns (reusing
ExternalSort, which already spills to disk and merges in bounded memory), then fold adjacent equal keys in a single pass. Because a group is never split across independent partial accumulators, there is no merge contract — the whole thing is just "sort, then collapse equal keys", mirroring howExternalSortworks. Output is streamed in batches instead of one bigRows.aggregationAlgorithm(AggregationAlgorithms::MEMORY_AGGREGATION | EXTERNAL_AGGREGATION)selects the strategy (mirrorsSortConfigBuilder::algorithm()); the default stays in-memory, so existing pipelines are unchanged.Addressing the review
GroupEntry"keeps values for all aggregators?" — gone. There is no storage layer; the algorithms hold their accumulators directly.aggregationStorage(strategy)— "storage or strategy?" andaggregationStorage()vsaggregationStore()— replaced by a singleaggregationAlgorithm(AggregationAlgorithms); the "storage/store" naming is removed.AggregationStorage::entry()vssave()— gone with the storage interface; aggregation is one inline loop.MergeableAggregatingFunction"when can two functions merge?" — removed entirely. Sort-fold never merges partial aggregates, so the interface andmerge()on the built-ins are deleted.Correctness note: the external path tags each row with its input position and sorts by
(group keys, input ordinal), sofirst()/last()/ unsortedcollect()return the same result as the in-memory path even when a group spills across sort runs (covered by a test).Resolves: #1390
Change Log
Added
ConfigBuilder::aggregationAlgorithm()and theAggregationAlgorithmsenum select betweenMEMORY_AGGREGATION(default, in-memory hash map) andEXTERNAL_AGGREGATION(sorts by the group-by columns and folds, spilling to disk for bounded memory), withaggregationFilesystem()for the spill protocol.Rows.Fixed
Changed
Removed
Deprecated
Security