perf(flow-php/etl): linear sort accumulation, working memory-to-exter…#2521
Closed
MrHDOLEK wants to merge 2 commits into
Closed
perf(flow-php/etl): linear sort accumulation, working memory-to-exter…#2521MrHDOLEK wants to merge 2 commits into
MrHDOLEK wants to merge 2 commits into
Conversation
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## 1.x #2521 +/- ##
============================================
+ Coverage 86.27% 86.38% +0.10%
+ Complexity 22924 22901 -23
============================================
Files 1739 1740 +1
Lines 70415 70395 -20
============================================
+ Hits 60753 60811 +58
+ Misses 9662 9584 -78 🚀 New features to boost your workflow:
|
norberttech
added
the
hackaton-07-2026
…nal fallback and precomputed sort keys Sorting a 1M-row CSV with default settings took hours and a configured sort memory limit crashed the pipeline instead of falling back to external sort. - Rows::mergeAll(iterable) merges batches without the quadratic cost of calling merge() in a loop, preserving merge() partition semantics (pinned by a test comparing both against 10 batch combinations). MemorySort, CollectingProcessor, DataFrame::fetch() and ExternalSort::extractSortedBuckets() no longer accumulate by re-copying every already-collected row per batch. - MEMORY_FALLBACK_EXTERNAL_SORT never actually fell back: OutOfMemoryException was uncaught. MemorySort wraps its input in a memory-guard generator, mergeAll attaches the rows collected so far to the rethrown exception, and SortingProcessor resumes external sort from those rows plus the remaining generator. MEMORY_SORT keeps throwing. - Rows::sortBy/sortAscending/sortDescending extract entry values once per row instead of resolving Row::valueOf inside every comparison. The one-stable-pass-per-reference structure is kept deliberately: the comparator treats mixed-type pairs as equal, which is not transitive, so a single lexicographic pass would change observable ordering (verified with a randomized differential test against the previous implementation). - RowsMinHeap caches sort key values on BucketRow at insert instead of re-resolving them per heap comparison; ExternalSort run size raised from hardcoded 500 to 10_000 and both sorters emit output batches of at least 1000 rows instead of inheriting 1-row input batches. - New tests: fallback completes under a 2MB sort memory limit with no dropped or duplicated rows, MEMORY_SORT still throws; both run in separate processes because real-usage memory accounting is unreliable after allocator page reuse. Measured on the 1M-row benchmark dataset: sortBy with default batching 66s -> 1.3s at 100k rows and ~2h (extrapolated) -> 26s at 1M rows; FLOW_SORT_MAX_MEMORY=256M now completes at 432MB peak instead of a fatal error.
MrHDOLEK
force-pushed
the
perf/sort-linear-accumulation-and-fallback
branch
from
bb66ba1 to
aaa2c7b
Compare
…d sort value comparator External sort no longer re-encodes the whole dataset through identity merge passes: single-bucket chunks pass through unchanged and the final merge level streams straight to the consumer instead of being spilled and re-read. At 1M rows with default settings this cuts Floe encode/decode passes from 3+3 to 1+1 per row. - ExternalSort output batch size comes from the existing externalSortBatchSize config knob instead of a new const plus max-input-batch tracking; run size becomes a constructor parameter - SortingProcessor fallback yields collected rows directly, dropping the FALLBACK_BATCH_SIZE re-chunking pass; MemorySort output batching uses the same config knob - Rows and RowsMinHeap share one ValueComparator, removing the second subtly different comparison ladder for the same sort - CollectingExtractor uses Rows::mergeAll, the last quadratic accumulate-by-merge loop in core; dead SortBucketsExtractor removed
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
2 participants
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.
Summary
Profiling against a 1M-row CSV showed that
sortBy()with default settings is quadratic — and that the documented memory→external fallback does not exist:MemorySortaccumulated input with$mergedRows = $mergedRows->merge($batch);Rows::merge()re-copies every already-collected row on each call. File extractors emitone-row
Rowsbatches, so sorting 1M rows meant ~1M merges and ~5·10¹¹ row-reference copies. Measured scaling: 25k → 3.5 s, 50k → 12.8 s, 100k → 66 s (~4–5× per doubling); 1Mextrapolates to ~2 hours.
MEMORY_FALLBACK_EXTERNAL_SORT, but whenMemorySortexceeded its limit it threwOutOfMemoryExceptionthat nothing caught — settingFLOW_SORT_MAX_MEMORYfatally crashed the pipeline instead of switching to external sort (verified empirically). Withmemory_limit=-1the limit resolved toPHP_INT_MAX,so the external path effectively never ran.
Row::valueOf()(a multi-level entry lookup) twice inside every one of the n·log n comparisons — 44% of sort wall time in cachegrindprofiles.
sortBy()paid per-batch overhead onceper row.
Changes
MemorySortcollects rows into a plain buffer and builds a singleRowsat the end, preserving partition semantics (partitions kept only when allbatches share one partition id, empty batches transparent, empty input yields no batches). The same accumulate-by-merge pattern is fixed in
CollectingProcessor,DataFrame::fetch()andExternalSort::extractSortedBuckets().OutOfMemoryExceptionnow carries the rows collected before the limit was hit;SortingProcessorcatches it, skips the already-buffered batch of thepartially-consumed generator, and finishes the job with external sort (buffered rows + remaining stream).
MEMORY_SORTkeeps throwing, as before.Rows::sortBy()/sortAscending()/sortDescending()extract entry values once per row and sort an index array. The one-stable-pass-per-referencestructure is kept deliberately: the comparator treats mixed-type pairs (e.g. null vs string) as equal, which is not transitive, so collapsing to a single lexicographic
pass would change observable ordering — proven with a randomized differential test against the previous implementation (200 trials × 5 reference combos, orders identical).
RowsMinHeapextracts sort key values once at insert and caches them onBucketRow(@internal) instead of resolvingvalueOf()per referenceper heap comparison — the same precomputed-key pattern the external grouping work (Storage for Aggregating Functions #1390) uses in its
BucketsMinHeap.run-size policy),
batchSizetracks the maximum input batch, and both sorters emit output batches of at least 1,000 rows.SortingProcessor::externalSort()is intentionally left untouched to avoid conflicting with #1390, which rewrites that method.Benchmarks
1M-row CSV (57 MB, 2 string columns), PHP 8.3 + opcache, fresh process per run:
1.xsortBy(), default batching, 100k rowssortBy(), default batching, 1M rowssortBy(),batchSize(10000), 1M rowssortBy(), 1M rows,FLOW_SORT_MAX_MEMORY=256MTests & verification
range(0, 2499)asserted (catches dropped or duplicated rows from the generator-resume logic);MEMORY_SORTstill throws. Both#[RunInSeparateProcess]— real-usage memory accounting is unreliable in a shared PHPUnit process after the allocator reuses freed pages.Related historical issues (both closed): #1116 (wrong ordering when external sort memory was exceeded), #1142 (external sort cache memory consumption). Not addressed here on
purpose: the spill format still serializes full
Rowobject graphs (~43× size amplification) — that belongs in a shared codec with #1390'sBucketRunCacheafter #1390 lands.Resolves: no open roadmap item — standalone performance/correctness fix from a profiling audit; happy to open an issue if preferred
Change Log
Added
Fixed
Changed
Removed
Deprecated
Security