You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "jayzhan211 (via GitHub)" <gi...@apache.org> on 2026/10/04 04:57:54 UTC
[PR] feat: sort-merge fallback for partitioned hash joins under memory pressure [datafusion]
jayzhan211 opened a new pull request, #26015:
URL: https://github.com/apache/datafusion/pull/26015
## Which issue does this PR close?
- Part of #24768. Related to #1599 and #12952.
This is an interim, sort-based path, not the hybrid hash join #24768 proposes.
Supersedes #25217, split into smaller PRs for review: this is the core, and the
dynamic-filter bounds and the `hash_join_max_build_size` option follow
separately (linked below). The review comment there about probe-side ordering
is handled here (the fallback is declined when the join advertises an ordering;
covered by `sort_merge_fallback_honors_a_promised_probe_ordering`).
## Rationale for this change
A partitioned hash join whose build side does not fit in the memory pool fails
the query with `Resources exhausted`, even though the same join completes as a
sort-merge join, because sorts spill. Today the only workaround is
`prefer_hash_join = false`, which pays for a sort on every join, including the
ones that fit.
With this PR the choice is made at runtime, per partition: a join stays a hash
join while its build side fits, and only a partition that runs out of memory
sorts both inputs (spilling) and finishes as a sort-merge join.
Reproducer in `datafusion-cli` (12-core laptop, `release-nonlto` builds of this
PR and of its base commit, median of 5-10 interleaved runs):
```sql
-- datafusion-cli --mem-pool-type fair -m 256m
SET datafusion.execution.target_partitions = 4;
SELECT count(*), sum(l.v), sum(r.w)
FROM (SELECT value AS k, value AS v FROM generate_series(1, 20000000)) l
JOIN (SELECT value AS k, value * 2 AS w FROM generate_series(1, 20000000)) r
ON l.k = r.k;
```
| | main | this PR |
|---|---|---|
| `--mem-pool-type fair -m 256m` | `Resources exhausted ... HashJoinInput[3]` (8/8 runs) | 0.50 s (8/8 runs), `sort_merge_fallback_count=4`, 621 MB spilled |
| same, `prefer_hash_join = false` (today's workaround) | 0.30 s | 0.30 s |
| no memory limit | 0.144 s | 0.143 s |
Main also fails at `-m 768m`. With the default greedy pool this PR completes
4-5 runs out of 6 at 256-512 MB: the failures are in a sort's merge phase,
which the greedy pool cannot make sibling sorters spill for, as with any
multi-partition sort; 768 MB completes 6/6.
## What changes are included in this PR?
- `HashJoinExec` (`PartitionMode::Partitioned` only): when the join's own
reservation fails with `ResourcesExhausted`, either while collecting build
batches or while building the hash table over batches that fit, the partition
sorts its build side with `ExternalSorter`, sorts its probe side, and finishes
through the existing sort-merge join stream (`sort_merge_join_stream`, already
shared since #25490). New module `hash_join/sort_merge_fallback.rs`, new stream
state `SortMergeFallback`.
- The table-building half of `collect_left_input` moves into `build_in_memory`
(body unchanged) so the build rows are still available when it fails.
- New metric `sort_merge_fallback_count`; the fallback's spills are reported in
the join's spill metrics.
- A partition that falls back reports itself to the dynamic filter like a
canceled partition: its probe rows are not filtered (no hash table to test
against). Keeping bounds pruning for it is a follow-up.
- `ExternalSorter` and a few of its methods become `pub(crate)`. No public API
change.
The fallback is declined (today's behavior is kept) when: disk spilling is
disabled, the join is `CollectLeft`, it is a null-aware `NOT IN` join (#25546),
it advertises a probe-side output ordering (a merge emits key order), or a key
type is not supported by the row format.
Known limitation: each fallback sort pre-reserves `sort_spill_reservation_bytes`
(10 MB default) and a falling-back partition runs two, so a budget smaller than
that floor still fails, now inside the sort. See the module docs.
Follow-ups (stacked on this PR, each small):
- #TODO keep bounds pruning for fallen-back partitions (`PushdownStrategy::Unknown`)
- #TODO `datafusion.execution.hash_join_max_build_size`, a trigger that works
without a memory limit
## What is the testing strategy for this PR?
- `hash_join/exec.rs`: all ten join types, with and without a join filter, at
two budgets: one that runs out while collecting the build batches and one that
only runs out building the hash table (each verified by disabling the other
trigger). Each asserts the same rows as the in-memory join, that it fell back,
and that it spilled. Plus projection and `fetch`, the dynamic filter staying
permissive, and the probe-ordering gate.
- `core/tests/memory_limit/hash_join_sort_merge_fallback.rs`: `config_matrix`
runs one SQL join under four configurations (no limit, limit with spilling,
limit without spilling, large limit) and checks both the answer and which
algorithm ran.
- Two existing tests that expect a memory error now disable the disk manager.
## Are there any user-facing changes?
Queries that failed with `Resources exhausted` in a partitioned hash join now
complete when disk spilling is enabled and the budget covers the sorts. Nothing
changes for joins that fit in memory. No new configuration options.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: sort-merge fallback for partitioned hash joins under memory pressure [datafusion]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #26015:
URL: https://github.com/apache/datafusion/pull/26015#issuecomment-5976863947
## [Codecov](https://app.codecov.io/gh/apache/datafusion/pull/26015?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `87.43363%` with `71 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 82.65%. Comparing base ([`6047791`](https://app.codecov.io/gh/apache/datafusion/commit/60477917bd9eadd1a299185db6a8a08ccae7a4c7?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`c41a6b3`](https://app.codecov.io/gh/apache/datafusion/commit/c41a6b3b50bb9b36ed1b528953c98d115bfea1f2?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
| [Files with missing lines](https://app.codecov.io/gh/apache/datafusion/pull/26015?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...tafusion/physical-plan/src/joins/hash\_join/exec.rs](https://app.codecov.io/gh/apache/datafusion/pull/26015?src=pr&el=tree&filepath=datafusion%2Fphysical-plan%2Fsrc%2Fjoins%2Fhash_join%2Fexec.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-ZGF0YWZ1c2lvbi9waHlzaWNhbC1wbGFuL3NyYy9qb2lucy9oYXNoX2pvaW4vZXhlYy5ycw==) | 88.07% | [6 Missing and 38 partials :warning: ](https://app.codecov.io/gh/apache/datafusion/pull/26015?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...al-plan/src/joins/hash\_join/sort\_merge\_fallback.rs](https://app.codecov.io/gh/apache/datafusion/pull/26015?src=pr&el=tree&filepath=datafusion%2Fphysical-plan%2Fsrc%2Fjoins%2Fhash_join%2Fsort_merge_fallback.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-ZGF0YWZ1c2lvbi9waHlzaWNhbC1wbGFuL3NyYy9qb2lucy9oYXNoX2pvaW4vc29ydF9tZXJnZV9mYWxsYmFjay5ycw==) | 86.71% | [9 Missing and 10 partials :warning: ](https://app.codecov.io/gh/apache/datafusion/pull/26015?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...fusion/physical-plan/src/joins/hash\_join/stream.rs](https://app.codecov.io/gh/apache/datafusion/pull/26015?src=pr&el=tree&filepath=datafusion%2Fphysical-plan%2Fsrc%2Fjoins%2Fhash_join%2Fstream.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-ZGF0YWZ1c2lvbi9waHlzaWNhbC1wbGFuL3NyYy9qb2lucy9oYXNoX2pvaW4vc3RyZWFtLnJz) | 83.33% | [6 Missing and 1 partial :warning: ](https://app.codecov.io/gh/apache/datafusion/pull/26015?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...physical-plan/src/joins/hash\_join/exec/prepared.rs](https://app.codecov.io/gh/apache/datafusion/pull/26015?src=pr&el=tree&filepath=datafusion%2Fphysical-plan%2Fsrc%2Fjoins%2Fhash_join%2Fexec%2Fprepared.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-ZGF0YWZ1c2lvbi9waHlzaWNhbC1wbGFuL3NyYy9qb2lucy9oYXNoX2pvaW4vZXhlYy9wcmVwYXJlZC5ycw==) | 80.00% | [1 Missing :warning: ](https://app.codecov.io/gh/apache/datafusion/pull/26015?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## main #26015 +/- ##
========================================
Coverage 82.64% 82.65%
========================================
Files 1147 1148 +1
Lines 445883 446407 +524
Branches 445883 446407 +524
========================================
+ Hits 368513 368985 +472
- Misses 54982 54997 +15
- Partials 22388 22425 +37
```
</details>
[:umbrella: View full report in Codecov by Harness](https://app.codecov.io/gh/apache/datafusion/pull/26015?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org