Uh oh!
There was an error while loading. Please reload this page.
perf: Coalesce batches before sorting in ExternalSorter to reduce merge fan-in - #21629
Closed
mbutrovich wants to merge 6 commits into
Closed
perf: Coalesce batches before sorting in ExternalSorter to reduce merge fan-in#21629mbutrovich wants to merge 6 commits into
mbutrovich wants to merge 6 commits into
Conversation
…erge fan-in. Subset of changes from apache#21600.
| /// | ||
| /// Larger values reduce merge fan-in by producing fewer, larger | ||
| /// sorted runs. | ||
| pub sort_coalesce_target_rows: usize, default = 32768 |
Contributor
There was a problem hiding this comment.
I wonder if we can make this somewhat adaptive: as we usually load everything in memory, it seems for very large sets larger batches would be even more favorable (e.g. use 10MiB "scratch space" for coalescing instead of 32KiB rows would make sense if our data is 1GiB and perhaps be even faster?)
ContributorAuthor
There was a problem hiding this comment.
Yeah I suspect we'd want to do a good sensitivity analysis on different types and batch sizes for lexsort_to_indices (and eventually the radix sort kernel). We might hit a point of diminishing returns/cache friendliness if our coalesced batches get too large.
This design also first spills from the sorted runs, so holding more unsorted rows in the coalescer may make it more likely for us to trigger spilling.
I'm definitely of the mind that we can and should tune this, but unclear what even a reasonable default right now would be. In Comet where we run TPC-H SF 1000, for example, I suspect we'll want longer sorted runs.
mbutrovich
commented
Apr 14, 2026
ContributorAuthor
run benchmark tpch10 |
adriangbot
commented
Apr 14, 2026
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing externalsorter (32a6882) to e6b32fe (merge-base) diff using: tpch10 File an issue against this benchmark runner |
mbutrovich
marked this pull request as ready for review
April 14, 2026 19:35
adriangbot
commented
Apr 14, 2026
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)DetailsResource Usagetpch10 — base (merge-base)
tpch10 — branch
File an issue against this benchmark runner |
Dandandan
commented
Apr 14, 2026
Contributor
Wooh, nice results |
Contributor
Lets compare to hash join (still 2x diff I think) |
Dandandan
commented
Apr 14, 2026
Contributor
run benchmark tpch10 |
adriangbot
commented
Apr 14, 2026
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing externalsorter (32a6882) to e6b32fe (merge-base) diff using: tpch10 File an issue against this benchmark runner |
mbutrovich
commented
Apr 14, 2026
ContributorAuthor
I also opened a new PR that scales our existing sort benchmark to larger numbers: #21630 In theory we should merge that first, then run the sort benchmark against |
adriangbot
commented
Apr 14, 2026
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)DetailsResource Usagetpch10 — base (merge-base)
tpch10 — branch
File an issue against this benchmark runner |
Dandandan
commented
Apr 14, 2026
Contributor
run benchmark tpcds tpch_sort |
adriangbot
commented
Apr 14, 2026
🤖 Criterion benchmark running (GKE) | trigger CPU Details (lscpu)Comparing externalsorter (32a6882) to e6b32fe (merge-base) diff File an issue against this benchmark runner |
adriangbot
commented
Apr 14, 2026
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing externalsorter (32a6882) to e6b32fe (merge-base) diff using: tpcds File an issue against this benchmark runner |
adriangbot
commented
Apr 14, 2026
Benchmark for this request failed. Last 20 lines of output: Click to expandFile an issue against this benchmark runner |
mbutrovich
commented
Apr 14, 2026
ContributorAuthor
run benchmark tpcds |
adriangbot
commented
Apr 14, 2026
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing externalsorter (32a6882) to e6b32fe (merge-base) diff using: tpcds File an issue against this benchmark runner |
adriangbot
commented
Apr 14, 2026
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)DetailsResource Usagetpcds — base (merge-base)
tpcds — branch
File an issue against this benchmark runner |
adriangbot
commented
Apr 14, 2026
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)DetailsResource Usagetpcds — base (merge-base)
tpcds — branch
File an issue against this benchmark runner |
github-merge-queueBot
pushed a commit
that referenced
this pull request
Apr 15, 2026
## Which issue does this PR close? - Partially addresses #21543. Also needed to properly evaluate the ExternalSorter refactor in #21629, which improves the merge path. ## Rationale for this change Current sort benchmarks use 100K rows across 8 partitions (~12.5K rows per partition, ~100KB for integers). This falls below the `sort_in_place_threshold_bytes` (1MB), so the "sort partitioned" benchmarks always take the concat-and-sort-in-place path and never exercise the sort-then-merge path that dominates real workloads. ## What changes are included in this PR? Parameterizes the sort benchmark on input size, running each case at both 100K rows (existing) and 1M rows (new). At 1M rows, each partition holds ~125K rows (~1MB for integers), which exercises the merge path. - `INPUT_SIZE` constant replaced with `INPUT_SIZES` array: `[(100_000, "100k"), (1_000_000, "1M")]` - `DataGenerator` takes `input_size` as a constructor parameter - All stream generator functions accept `input_size` - Benchmark names include size label (e.g. `sort partitioned i64 100k`, `sort partitioned i64 10M`) - Data distribution and cardinality ratios are preserved across sizes ## Are these changes tested? Benchmark compiles and runs. No functional test changes. ## Are there any user-facing changes? No.
mbutrovich
commented
Apr 15, 2026
ContributorAuthor
run benchmark sort |
adriangbot
commented
Apr 15, 2026
🤖 Criterion benchmark running (GKE) | trigger CPU Details (lscpu)Comparing externalsorter (689dfd2) to d0692b8 (merge-base) diff File an issue against this benchmark runner |
| /// `─────────────────' │ | ||
| /// spills | ||
| /// ┌──────────┐ ┌────────────┐ ┌──────┐ ┌────────────┐ | ||
| /// │ Incoming │────▶│ Batch │────▶│ Sort │────▶│ Sorted Run │ |
Contributor
There was a problem hiding this comment.
In the current approach, it would be worth it to "push down" the target batch size to the inner operation, so the output of the previous stage already is of the higher batch size instead of doing another copy.
ContributorAuthor
There was a problem hiding this comment.
@andygrove and I have been discussing this in the context of Comet, where for some Spark stages that get translated to native plans we'd possibly want to set different target batch sizes. What you're proposing sounds even more granular, where individual operators could potentially advertise an ideal input batch size, though this could get complicated fast based on schema. Maybe some sort of trait on operators to advertise when they want larger batch sizes, and the optimizer could figure out what that target should be?
Regardless, a general solution seems out of scope for this PR.
Contributor
There was a problem hiding this comment.
Yeah I think something like that could work.
The propagation doesn't seem too complex I think, just change the target batch size of te inner op based on the desired one.
I agree it is out of scope.
mbutrovich
commented
Apr 15, 2026
ContributorAuthor
I fear my sort benchmark scale up might have killed the benchmark bot :( |
| } | ||
| } | ||
| for batch in &completed { | ||
| self.sort_and_store_run(batch)?; |
Contributor
There was a problem hiding this comment.
Nice - I wonder if a part of the speedup comes from this, sorting batch immediately (while in CPU cache) instead of waiting to the end.
This was referenced Apr 15, 2026
mbutrovich
commented
Apr 15, 2026
ContributorAuthor
run benchmark sort |
adriangbot
commented
Apr 15, 2026
🤖 Criterion benchmark running (GKE) | trigger CPU Details (lscpu)Comparing externalsorter (0198302) to 5c653be (merge-base) diff File an issue against this benchmark runner |
github-merge-queueBot
pushed a commit
that referenced
this pull request
Apr 15, 2026
…21657) ## Which issue does this PR close? - Closes #. ## Rationale for this change PR #21620 (commit 5c653be) ported `test_aggregate_dynamic_filter_parquet_e2e` from Rust to sqllogictest using `analyze_categories = 'rows'`, which includes exact pushdown metrics. These metrics are nondeterministic under parallel execution — the order in which Partial aggregates publish dynamic filter updates races against when the scan reads each partition — so the expected output is flaky. Noticed on #21629 ([CI run](https://github.com/apache/datafusion/actions/runs/24472961090/job/71518239089?pr=21629)) and confirmed on main ([CI run](https://github.com/apache/datafusion/actions/runs/24467213913/job/71497866154)). ## What changes are included in this PR? Switch the `agg_dyn_e2e` test to `analyze_level = summary` + `analyze_categories = 'none'`, suppressing nondeterministic metrics. This matches the approach already used by the other aggregate dynamic filter tests in the same file. The original Rust test only asserted `matched < 4` (i.e. some pruning happened); the important invariant — that the `DynamicFilter [ column1@0 > 4 ]` text and pruning predicate are correct — is still verified. ## Are these changes tested? Yes — the test itself is what is being fixed. ## Are there any user-facing changes? No.
Dandandan
approved these changes
Apr 16, 2026
Dandandan
commented
Apr 16, 2026
Contributor
Would be nice to see results in |
mbutrovich
commented
Apr 16, 2026
ContributorAuthor
I’ll try to run locally with the scaled up |
mbutrovich
commented
Apr 16, 2026
ContributorAuthor
Benchmark Analysis: |
| Benchmark | externalsorter | main | Speedup |
|---|---|---|---|
| sort i64 100k | 2.40 ms | 3.88 ms | 1.61x |
| sort f64 100k | 2.61 ms | 3.94 ms | 1.51x |
| sort merge i64 1M | 26.6 ms | 38.5 ms | 1.45x |
| sort i64 1M | 33.8 ms | 45.5 ms | 1.35x |
| sort f64 1M | 35.8 ms | 46.4 ms | 1.30x |
| sort utf8 high cardinality 1M | 59.7 ms | 74.0 ms | 1.24x |
| sort utf8 view tuple 1M | 8.1 ms | 9.1 ms | 1.13x |
Regressions (multi-column StringArray and Dictionary)
| Benchmark | externalsorter | main | Slowdown |
|---|---|---|---|
| sort utf8 tuple 100k (3x StringArray) | 23.7 ms | 9.5 ms | 2.5x |
| sort utf8 tuple 1M (3x StringArray) | 272 ms | 111 ms | 2.5x |
| sort mixed dictionary tuple 1M (3x dict + i64) | 283 ms | 108 ms | 2.6x |
| sort utf8 dictionary tuple 1M (3x dict) | 92.4 ms | 78.5 ms | 1.18x |
Root cause
The coalesce-then-sort pipeline does: coalesce (copy all column data) -> lexsort_to_indices -> take (random-access scatter-gather) -> chunk back to batch_size. With multiple StringArray or Dictionary columns at 32K rows, the take scatter-gathers across ~1.9 MB of string heap data, exceeding L2 cache.
Schemas that don't regress help pinpoint the cause:
- Single StringArray (e.g.
sort utf8 high cardinality 1M, +1.24x): one ~640KB string buffer fits in L2 duringtake - StringViewArray (e.g.
sort utf8 view tuple 1M, +1.13x):takecopies fixed-size 16-byte view structs — no heap random access - Fixed-width (i64, f64): coalesce is memcpy,
takeis sequential — all cache-friendly
Benchmark batch size
The sort benchmark uses BATCH_SIZE = 1024, while DataFusion's default is 8192. This makes the coalesce expand 32x (1024 -> 32768) instead of 4x (8192 -> 32768), amplifying the copy cost. Small batches are a valid scenario though — filters, joins, and sources with pushdown can all produce sub-8192 batches.
Proposed fix
Arrow's BatchCoalescer has a with_biggest_coalesce_batch_size(Some(limit)) option: batches larger than limit bypass coalescing and pass through directly. For schemas with non-view variable-length columns (StringArray, BinaryArray, DictionaryArray), set this to batch_size. This way:
- Full-size batches (>= 8192 rows) pass through without string copying — sorted directly as individual runs
- Small batches still coalesce to reduce fan-in
- Fixed-width and StringView schemas keep the full 32K coalesce target unchanged
mbutrovich
commented
Apr 16, 2026
ContributorAuthor
Marking as draft while we explore performance trade-offs. I don't want to accidentally merge this yet. |
mbutrovich
marked this pull request as draft
April 16, 2026 15:14
mbutrovich
commented
Apr 16, 2026
ContributorAuthor
More brainstorming with Claude:
|
This was referenced Apr 16, 2026
Comment on lines
+656
to
+659
| // - This works because sort() only enters this path when | ||
| // merge_reservation > 0, guaranteeing pool headroom for cursors. | ||
| // When merge_reservation == 0, sort() takes the spill path instead. | ||
| let streams = all_runs |
There was a problem hiding this comment.
if i'm not mistaken merge_sorted_runs() is also being called on the non-spill path inside sort() when self.spilled_before() evaluates to false, where merge_reservation is never pre-reserved.
maybe we can make the comment more descriptive on the different headroom reliance of the different paths?
Comment on lines
-352
to
-358
| // Transfer the pre-reserved merge memory to the streaming merge | ||
| // using `take()` instead of `new_empty()`. This ensures the merge | ||
| // stream starts with `sort_spill_reservation_bytes` already | ||
| // allocated, preventing starvation when concurrent sort partitions | ||
| // compete for pool memory. `take()` moves the bytes atomically | ||
| // without releasing them back to the pool, so other partitions | ||
| // cannot race to consume the freed memory. |
There was a problem hiding this comment.
This comment is quite useful, maybe we should keep it, could help future developers know why take is being used!
| new_columns.push(Arc::clone(array)); | ||
| } | ||
| } | ||
| if self.merge_reservation.size() > 0 && self.sorted_runs.len() > 1 { |
There was a problem hiding this comment.
Hi, i'm tracing the spill cycles, and the first spill's merge_reservation value starts at 0 while others grow at the end of spill_sorted_runs
the second spill could be something small like 1024 and still take the merge path
is >0 just a good enough heuristic we are using without trying to guarantee actual headroom for the merge process?
feel free to correct me, i'm new and i'm trying to learn!!
Rich-T-kid pushed a commit
to Rich-T-kid/datafusion
that referenced
this pull request
Apr 21, 2026
…e#21630) ## Which issue does this PR close? - Partially addresses apache#21543. Also needed to properly evaluate the ExternalSorter refactor in apache#21629, which improves the merge path. ## Rationale for this change Current sort benchmarks use 100K rows across 8 partitions (~12.5K rows per partition, ~100KB for integers). This falls below the `sort_in_place_threshold_bytes` (1MB), so the "sort partitioned" benchmarks always take the concat-and-sort-in-place path and never exercise the sort-then-merge path that dominates real workloads. ## What changes are included in this PR? Parameterizes the sort benchmark on input size, running each case at both 100K rows (existing) and 1M rows (new). At 1M rows, each partition holds ~125K rows (~1MB for integers), which exercises the merge path. - `INPUT_SIZE` constant replaced with `INPUT_SIZES` array: `[(100_000, "100k"), (1_000_000, "1M")]` - `DataGenerator` takes `input_size` as a constructor parameter - All stream generator functions accept `input_size` - Benchmark names include size label (e.g. `sort partitioned i64 100k`, `sort partitioned i64 10M`) - Data distribution and cardinality ratios are preserved across sizes ## Are these changes tested? Benchmark compiles and runs. No functional test changes. ## Are there any user-facing changes? No.
Rich-T-kid pushed a commit
to Rich-T-kid/datafusion
that referenced
this pull request
Apr 21, 2026
…pache#21657) ## Which issue does this PR close? - Closes #. ## Rationale for this change PR apache#21620 (commit 5c653be) ported `test_aggregate_dynamic_filter_parquet_e2e` from Rust to sqllogictest using `analyze_categories = 'rows'`, which includes exact pushdown metrics. These metrics are nondeterministic under parallel execution — the order in which Partial aggregates publish dynamic filter updates races against when the scan reads each partition — so the expected output is flaky. Noticed on apache#21629 ([CI run](https://github.com/apache/datafusion/actions/runs/24472961090/job/71518239089?pr=21629)) and confirmed on main ([CI run](https://github.com/apache/datafusion/actions/runs/24467213913/job/71497866154)). ## What changes are included in this PR? Switch the `agg_dyn_e2e` test to `analyze_level = summary` + `analyze_categories = 'none'`, suppressing nondeterministic metrics. This matches the approach already used by the other aggregate dynamic filter tests in the same file. The original Rust test only asserted `matched < 4` (i.e. some pruning happened); the important invariant — that the `DynamicFilter [ column1@0 > 4 ]` text and pruning predicate are correct — is still verified. ## Are these changes tested? Yes — the test itself is what is being fixed. ## Are there any user-facing changes? No.
12 tasks
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 freeto join this conversation on GitHub.
Already have an account?
Sign in to comment
4 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.
Which issue does this PR close?
Partially addresses #21543. This is the ExternalSorter pipeline refactor from #21600, separated from the radix sort changes.
Rationale for this change
ExternalSorter's merge path sorts each incoming batch individually (typically 8192 rows), then k-way merges all of them. At scale (TPC-H SF10, ~60M rows in lineitem), this produces ~7300 individually-sorted batches feeding the k-way merge with high fan-in.
What changes are included in this PR?
Coalesce-then-sort pipeline
Replaces ExternalSorter's buffer-then-sort architecture with a coalesce-then-sort pipeline:
BatchCoalesceruntilsort_coalesce_target_rows(default 32768) is reachedbatch_size, producing fewer, larger sorted runsStreamingMergeBuilderThis reduces merge fan-in from ~7300 to ~1800 runs at SF10, which is the primary source of speedup.
Spill strategy
sort_spill_reservation_bytes > 0): merge all runs into a single sorted stream before spilling to one file. Fewer files = lower fan-in for the finalMultiLevelMerge.sort_spill_reservation_bytes == 0): spill each run as its own file. The multi-level merge handles low merge memory by reducing fan-in, so this no longer fails under tight memory budgets.Dead code removal
Sorted runs no longer require an in-memory merge before spilling. Removes
in_mem_sort_stream,sort_batch_stream,consume_and_spill_append,spill_finish,organize_stringview_arrays, andin_progress_spill_file.Config changes
sort_coalesce_target_rows(default 32768)sort_in_place_threshold_bytes(no longer read,warnattribute per API health policy)Are these changes tested?
information_schema.sltupdated for new configAre there any user-facing changes?
sort_coalesce_target_rows(default 32768) controls the coalesce target before sortingsort_in_place_threshold_bytesis deprecatedsort_spill_reservation_bytesnear zero) that previously failed now succeed via multi-level merge with reduced fan-in