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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
