Unify bucketing contract between sort/join/group by algorithms #2543
Merged
Conversation
Flow\ETL\Bucketing namespace: BucketsStorage contract, Bucket/BucketStats value objects, Buckets repository - Memory, Filesystem, and PSR-16 cache storage backends - array-backed PSR-16 test fake and bucket ObjectMothers
- BucketingProcessor: barrier processor spilling rows into buckets, tagging _bucket_id - pluggable strategies: HashBucketing (hash-mod) and SortedRunBucketing (external-sort runs) - single-pass key extraction + value-based batch Hasher (KeyValues, NativeHasher) - per-bucket stats (rows/chunks/min/max/nullCount/capped-distinct) via BucketStatsCollector
… on the bucketing engin - BucketingStrategy spills to BucketsStorage itself and yields Bucket(id, totalRows) VOs; BucketChunk, BucketYield and the stats-collector machinery removed - BucketingProcessor reduced to an adapter: registers buckets, yields one metadata Rows per bucket (BucketShape: _bucket_id, _bucket_total_rows) - external sort rebuilt on the engine: SortingProcessor/ExternalSort family replaced by MemorySortProcessor + SortedRunBucketing + MergeSortProcessor, k-way merge internals moved to Sort\Merge - sorted runs written with set() (write-and-close), hash buckets with append() - storage verb owned by the strategy - external-sort storage and tuning moved from CacheConfig to SortConfig (public Config::builder() API unchanged)
…imization - grace hash join rebuilt on Buckets/BucketsStorage, config swapped from BucketsCache, BucketSpiller/BucketedHashJoin removed - SQL null semantics for Equal/Identical, null join keys dropped before spilling - hash table built from the smaller bucket of each pair (Bucket gains partition index) - equalities extracted from mixed join expressions, All(Equal, Any) runs as hash join instead of nested loop (~31x) - hashing consolidated: positional KeyValues + single NativeHasher, JoinKeys reduced to refs VO, HashTable to hash->rows multimap - BucketsStorage::get() yields batches (-16% joins), Rows::sortBy via type-specialized C-level sort (-10% external sort, also fixes never-working DateInterval ordering)
…n pruning - GroupBy runs on HashBucketing/BucketingProcessor; ExternalAggregation deleted - aggregate() registers pipeline segments via GroupBySteps (BucketingProcessor -> GroupByAggregationProcessor, or PivotProcessor) - AggregatingFunction::references() enables pruning spilled rows to grouping + aggregation columns (~-40% wall time); missing columns normalized to null entries - GroupingConfig swaps BucketsCache for BucketsStorage (groupingStorage()) - DataFrame::pivot() removed - pivot is declared between groupBy() and aggregate() - HashBucketing/KeyValues gain nullOnMissing flag
- ConfigBuilder::sort()/join()/groupBy() take algorithm builders
(memory_sort(), external_sort(), hash_join(), hash_group_by()); root
externalSort*/join*/grouping* delegators removed
- per-algorithm configs share one BucketingConfig{storage, bucketsCount,
batchSize}; Config::$sort is MemorySortConfig|ExternalSortConfig,
JoinConfig/GroupingConfig become HashJoinConfig/HashGroupByConfig
- SortAlgorithms enum replaced by SortSteps processor dispatch (mirrors
GroupBySteps); sort run knob renamed bucketSize -> runSize
- unused legacy Sort\ExternalSort\BucketsCache family deleted
- sort/join/group-by docs and upgrade notes rewritten to the new API
… join and group-by
- JoinSteps factory; DataFrame::join delegates like sortBy/aggregate
- bucket lifecycle unified: shared Buckets registry, ids from metadata
rows, plain finally cleanup; dead Buckets/ExternalSortConfig API removed
- RowsBuffer extracted for size-bounded batch emission; RowKey deleted,
HashBucketing pairs rows with hashes positionally
- group-by output batch size wired from config; GroupBy\Bucket renamed
to Group
- sort spill ids namespaced (sort-{run}-{index}); strategies take
list<Reference>
- int<1, max> guard convention applied to all bucketing constructors
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## 1.x #2543 +/- ##
============================================
+ Coverage 86.32% 86.45% +0.13%
- Complexity 23233 23329 +96
============================================
Files 1761 1772 +11
Lines 70803 71001 +198
============================================
+ Hits 61119 61384 +265
+ Misses 9684 9617 -67 🚀 New features to boost your workflow:
|
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Change Log
Added
flow-php/etl- StandaloneFlow\ETL\Bucketingstorage backend with Memory, Filesystem and PSR-16 implementations.flow-php/etl-BucketingProcessorwith pluggable strategies (HashBucketing,SortedRunBucketing).flow-php/etl- Algorithm builder DSL functions:memory_sort(),external_sort(),hash_join(),hash_group_by().flow-php/etl- FloeOptionsvalue object withfloe_options()DSL andvalidateDataoption.Fixed
flow-php/etl-DateIntervalordering in sort now works via type-specializedRows::sortBy.Changed
flow-php/etl- External sort rebuilt on the bucketing engine.flow-php/etl- Grace hash join rebuilt on the bucketing engine, building the hash table from the smaller bucket.flow-php/etl- Mixed join expressions with equalities now run as hash join instead of nested loop.flow-php/etl- Group-by runs on the bucketing engine with spill column pruning.flow-php/etl- Sort, join and group-by configuration is now algorithm-scoped viaConfigBuilder::sort()/join()/groupBy().flow-php/etl-Equal/Identicaljoin comparisons follow SQL null semantics.flow-php/etl-FloeWriter/FloeStreamWritertake schema in constructor and build the encoder once.Removed
flow-php/etl-DataFrame::pivot()- pivot is declared betweengroupBy()andaggregate().flow-php/etl- RootConfigBuilderexternalSort*/join*/grouping*delegators.flow-php/etl- LegacySort\ExternalSortandBucketsCachefamily.flow-php/etl-SortAlgorithmsenum replaced bySortStepsprocessor dispatch.Deprecated
Security
Benchmarks:
Need help on this PR? Tag
/codesmithwith what you need. Autofix is disabled.