jayzhan211 opened a new pull request, #25217:
URL: https://github.com/apache/datafusion/pull/25217
## Which issue does this PR close?
- Part of #24768. Related to #1599 and #12952.
This does not close them. It is the interim sort-based path, not the
dynamically
partitioned hybrid hash join that #24768 proposes, and it deliberately leaves
`CollectLeft` mode out of scope.
## Rationale for this change
A hash join whose build side exceeds the memory budget fails the query with
`Resources exhausted`, even though the same join completes when the planner
is
told to use a sort-merge join instead, because its sorts spill. Users are
left
choosing between two bad options: keep `prefer_hash_join = true` and lose
large
joins outright, or set it to `false` and pay for a sort on every join,
including
the ones that would have fit in memory comfortably.
Sort, aggregation, sort-merge join, repartition, and nested-loop join all
spill.
Hash join, the default join, is the last major operator that fails instead.
This makes the decision at runtime rather than at plan time. A join stays a
hash
join while its build side fits, and becomes a sort-merge join only in the
partitions where it does not.
On a 240k x 160k row join through a 512 KB pool, median of five runs each:
| variant | time |
|---|---|
| hash join, no memory limit | 107 ms |
| hash join under the limit, falls back | 376 ms |
| `prefer_hash_join = false`, planned sort-merge, same limit | 380 ms |
| planned sort-merge, no limit | 301 ms |
The fallback costs about what today's workaround costs, roughly 3.5x the
in-memory join, but it is paid only by the partitions that actually run out
of
memory, and without having to predict that in advance.
## Before and after, and what to set
`EXPLAIN ANALYZE` shows which case you are in: a partitioned hash join now
reports `sort_merge_fallback_count` next to its `spill_count`.
| Situation | Before | After | Action |
|---|---|---|---|
| Memory limit set, build side fits | works | identical | none |
| Memory limit set, build side too big, temp files enabled | `Resources
exhausted` | completes as a sort-merge join | none, automatic |
| Memory limit set, budget too small for the sorts | fails in the join |
still fails, now in the sort | raise the limit, or lower
`sort_spill_reservation_bytes` |
| No memory limit configured | build side grows unbounded | unchanged, no
reservation ever fails | `hash_join_max_build_size` |
| Join lands in `CollectLeft` mode over budget | `Resources exhausted` |
unchanged | keep `prefer_hash_join = false` |
## Which cases can fall back, and which are covered here
A hash join can be asked to switch to a sort-merge join in more situations
than
one PR should try to handle. The cases, and where each one stands:
| Case | In this PR | Why, or what it still needs |
|---|---|---|
| `PartitionMode::Partitioned`, all ten join types | ✅ | Each partition owns
a disjoint slice of the key space on both sides, so it sorts and merges alone
and every join type stays correct |
| Trigger: reservation fails while buffering build batches | ✅ | |
| Trigger: reservation fails building the hash table or bitmaps, on batches
that did fit | ✅ | |
| Trigger: build side grows past `hash_join_max_build_size` | ✅ | The only
trigger available under the default, unbounded memory pool |
| Join filter, embedded projection, `fetch` | ✅ | Applied to the merge
output exactly as the hash join applies them to its own |
| Dynamic filter pushdown active on the join | ✅ | The partition reports
bounds only, through the new `PushdownStrategy::Unknown` |
| `CollectLeft`, probe-preserving types: Inner, Right, RightSemi, RightAnti,
RightMark | ❌ | Follow-up 1. The single shared build side has to be sorted once
and materialized so every probe partition can re-read it |
| `CollectLeft`, build-preserving types: Left, Full, LeftSemi, LeftAnti,
LeftMark | ❌ | Follow-up 2. Unmatched build rows require all probe partitions
to agree on what matched, so restore ownership has to be coordinated |
| Join advertises a probe-side output ordering | ❌ | Follow-up 3. A merge
emits in join-key order, so the promise needs either an extra sort or to be
dropped when the fallback is reachable |
| Null-aware (`NOT IN`) joins | ❌ | Not planned. The sort-merge streams have
no three-valued `NOT IN` semantics |
| Join keys the row format cannot sort | ❌ | Not planned. Bounded by what
arrow-rs `RowConverter` supports |
| Spilling unavailable, `DiskManager` disabled | ❌ | By design. There is
nothing to fall back to, so today's behavior is kept |
## What changes are included in this PR?
Runtime fallback in `HashJoinExec`:
- Collecting the build side now yields either an in-memory hash table or a
sorted build side. When the join's own reservation fails with
`ResourcesExhausted`, that partition sorts its build side with
`ExternalSorter`, sorts the probe side, and finishes by merging them.
- A new `HashJoinStreamState::SortMergeFallback` forwards that output. Both
OOM
sites are covered: mid-stream while buffering batches, and afterwards when
the
hash table or a bitmap does not fit on top of batches that did.
- `SortMergeJoinExec::execute` is factored into a shared
`sort_merge_join_stream`, so the fallback runs the real sort-merge code
path
rather than a copy of it. Projection and `fetch` are applied to its output
exactly as the hash join applies them to its own.
- The fallback engages only when disk spilling is available, so
`DiskManagerMode::Disabled` keeps today's behavior.
New option `datafusion.execution.hash_join_max_build_size`
(`Option<usize>`, default `NULL` meaning no limit):
- Triggers the same switch once a partition's build side grows past the given
size, whether or not a memory limit is configured. Under the default
unbounded
memory pool no reservation ever fails, so this is the only available
trigger.
- It is not required for memory safety. With
`datafusion.runtime.memory_limit`
set, the fallback already happens on its own.
- Documented as a short-term measure that may be deprecated and removed once
hash joins spill natively.
Supporting changes:
- New metric `sort_merge_fallback_count`, plus the fallback's spills reported
through the join's own spill metrics, both visible in `EXPLAIN ANALYZE`.
- `PushdownStrategy::Unknown`, so a build side that holds rows but has no
hash
table to test membership against pushes down bounds only. Reporting `Empty`
there would give the partition a `false` branch in the routed dynamic
filter
and silently drop every match it owns.
- `ExternalSorter` and the sort-merge join stream constructors are now
`pub(crate)`. No public API changes.
Also fixes a latent sort-merge join bug found while building this:
`get_filter_columns` assembled filter columns as all-left-then-all-right,
which
is wrong for the column order `JoinFilter::swap` produces. Only this
fallback can
reach it today, because the planner never swaps a `SortMergeJoinExec`.
### Scope
Only `PartitionMode::Partitioned` joins fall back. There each partition owns
a
disjoint slice of the key space on both sides, so it can sort and merge
independently of its siblings and every join type stays correct.
`CollectLeft`
joins, null-aware joins, and joins that promise to preserve their probe-side
ordering are untouched and still fail under memory pressure.
### Known limitation
The fallback replaces a build side that must fit entirely in memory with a
much
smaller requirement, but not with none. Each sort pre-reserves
`sort_spill_reservation_bytes` so its merge phase can always run, and a
falling-back partition runs two sorts. A budget that cannot cover those
reservations plus the sorts' working set still fails, and it now fails
inside the
sort rather than in the hash join.
Measured on a join with roughly 12 MB of build side per partition, two
partitions: with the default 10 MB reservation no budget worked, because any
budget small enough to trigger the fallback was too small to pay for the
reservations. At a 1 MB reservation the same query succeeded from 16 MB
upward.
The fixed reservation matters proportionally less as the build side grows,
which
is the case this feature targets, but the floor is real. Sizing the
fallback's
sorts from the budget actually available, rather than inheriting the global
default, is follow-up 8.
Partitions also share one pool without coordinating, so a partition that
falls
back does not make a sibling that still fits release its hash table. That
narrows
the usable range further, though whether one or both partitions fall back is
not
by itself decisive: a single falling-back partition completes when the budget
covers its sorts.
## Follow-up PRs
Each is independent of the others and of #24768's hybrid hash join work.
1. **`CollectLeft`, probe-preserving join types.** Sort the shared build side
once and materialize it to a spill file that every probe partition reads.
No cross-partition coordination needed.
2. **`CollectLeft`, build-preserving join types.** Coordinate which partition
restores and merges, so unmatched build rows are emitted once. Carries the
same distributed-execution caveat as `enable_nlj_coordinated_fallback`.
3. **Keep the advertised probe-side ordering.** Re-sort the merge output on
the
promised ordering instead of declining the fallback for those joins.
4. **Make the fallback contagious within one join,** so a partition that
gives up
its hash table is not competing with a resident sibling.
5. **Residual sort-merge OOM.** The streamed side and the join-key arrays of
a
sort-merge join still never spill, so a very tight budget fails even after
falling back. See #17429 and #19013.
6. **`SMJStream` registers `can_spill = false`** although it spills. One-line
fix, affects `FairSpillPool` accounting.
7. **Memory-limited join benchmarks.** A profile that runs joins under fixed
budgets and reports spill metrics, so the behaviour above is tracked in CI
rather than measured by hand. Overlaps #24768's first task.
8. **Size the fallback's sorts from the available budget.** They currently
inherit `sort_spill_reservation_bytes`, so with the 10 MB default a small
pool
is consumed by reservations before the sorts can run.
## What is the testing strategy for this PR?
- `datafusion/physical-plan/src/joins/hash_join/exec.rs`: all ten join types,
with and without a join filter, under a 6 KB pool, each asserted to return
the
same rows as the in-memory join and to have actually fallen back and
spilled.
Plus projection and `fetch`, the disk-disabled path, a test that falls back
purely on `hash_join_max_build_size` with no memory limit, and a guard
that a
fallen-back partition leaves its pushed-down dynamic filter permissive.
That
last one was written after mutation testing showed that reporting `Empty`
instead of `Unknown` passed every other test; `shared_bounds` carries two
matching tests for the pruning semantics of the two reports.
- `datafusion/core/tests/memory_limit/hash_join_sort_merge_fallback.rs`: a
single table-driven `config_matrix` test covering the five configurations
that
decide between a plain hash join, the fallback, and outright failure,
asserting
both the answer and which algorithm ran. Alongside it, end-to-end SQL
coverage
of inner, filtered outer, semi, anti and `IN` subqueries, plus `LIMIT`. The
memory-limited budget and `sort_spill_reservation_bytes` were chosen by
measurement: the committed pair ran 12 consecutive times with both
partitions
falling back and correct results every time.
- `datafusion/core/tests/fuzz_cases/join_fuzz.rs`: a fuzz test comparing the
fallback against the in-memory hash join across join types, filters and
batch
sizes under a 16 KB pool.
- `datafusion/physical-plan/src/joins/sort_merge_join/tests.rs`: a regression
test for the filter column order fix.
Performance: benchmarked with `hash_join_semi_anti` (criterion) against the
parent commit, median +0.36% with a range of -1.43% to +3.08%. Re-running the
*parent* commit against its own baseline gives median +0.34% and a range of
-0.67% to +2.80%, so that spread is this machine's run-to-run variance rather
than an effect of the change. The three build-heavy cases, which cover the
rewritten build loop, were flat to slightly faster.
The full `sqllogictest` suite passes, all 516 files, including the one case
that
asserts partitioned hash join metrics.
## Are there any user-facing changes?
Yes, two.
Queries that previously failed with `Resources exhausted` in a partitioned
hash
join now complete by falling back, provided disk spilling is enabled and the
budget covers the sorts. Nothing changes for joins that fit in memory, or for
anyone running without a memory limit and without the new option set.
One new configuration option,
`datafusion.execution.hash_join_max_build_size`,
documented in the configuration guide and marked as possibly temporary. No
breaking changes to public APIs.
--
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]
