andygrove opened a new pull request, #2108:
URL: https://github.com/apache/datafusion-ballista/pull/2108
# Which issue does this PR close?
Relates to #2025.
This does not `Closes` #2025: the operator introduced here is opt-in and
disabled
by default, so it does not change the default behaviour that OOMs. It
provides a
mechanism that resolves that OOM when enabled, and is intended as the first
step
toward addressing the issue.
# Rationale for this change
DataFusion's `HashJoinExec` does not spill. Its build side is collected fully
into memory; when a `Partitioned` build partition exceeds the memory
reservation
it returns `ResourcesExhausted` (or the process OOMs). This is the failure in
#2025 — a `Partitioned` build side on TPC-H Q18 that cannot fit and cannot
spill,
and does not shrink with `target_partitions`.
This PR adds a from-scratch spilling **hybrid** hash join operator for
Ballista.
It runs as a plain in-memory hash join when the build side fits, and degrades
gracefully to disk when it does not, so a large build side completes instead
of
failing. It is a **hybrid** join (as in Spark/Postgres), not pure Grace:
buckets
that fit stay resident and only the overflow is spilled, so the in-memory
fast
path pays no spill cost.
# What changes are included in this PR?
A new `SpillingHashJoinExec` operator (`ballista-core`) plus the scheduler
plumbing to substitute it for eligible joins.
**Operator** (`ballista/core/src/execution_plans/spilling_hash_join/`):
- Sub-partitions each task's build side into `P` buckets by a hash of the
join
keys (fixed seed, independent of the shuffle hash).
- Hybrid spill: buckets are kept resident against a `MemoryReservation`; when
`try_grow` fails, the largest resident bucket is spilled to disk (Arrow
IPC via
the runtime `DiskManager`). Resident buckets are probed streaming.
- Probe rows for spilled buckets are routed to a probe-side spill; after the
probe stream ends, spilled buckets are drained **one at a time** (build
table
built, probed, dropped) so peak memory is bounded to a single bucket.
- If a single spilled bucket's build side still does not fit, it is
recursively
re-partitioned with a depth-varied seed; an irreducible single-join-key
bucket
returns a clear error rather than OOMing, panicking, or hanging.
- Serialises scheduler → executor through `BallistaPhysicalExtensionCodec`.
**Scheduler**:
- `SpillingHashJoinRule` / `maybe_substitute_spilling_hash_join` substitutes
an
eligible `HashJoinExec` for the spilling operator. Eligibility is strict:
`Inner` join, `PartitionMode::Partitioned`, no projection, no residual
filter,
and `NullEquality::NullEqualsNothing`.
- Applied in both the default (`DefaultDistributedPlanner`) and AQE planners,
**after** broadcast promotion — so a small build side that can be promoted
to a
`CollectLeft` broadcast keeps that shuffle-avoiding plan (a `CollectLeft`
join
is not `Partitioned`, hence ineligible), and only the remaining large
`Partitioned` hash joins are made spillable.
**Config**: `ballista.execution.spilling_hash_join.enabled` (default
`false`) and
`ballista.execution.spilling_hash_join.partitions` (default `16`).
**Scope (v1)**: inner joins only. All other join types are left to the
existing
planning path unchanged. Outer/semi/anti and broadcast-mode support are
follow-ups.
**Testing**: the operator is validated against a DataFusion `HashJoinExec`
(`Inner`, `Partitioned`) oracle on hash-co-located inputs, with output
compared
under (a) fully resident, (b) forced build-side spill, (c) forced two-sided
spill, and (d) forced recursive re-partition — down to a ~1 KiB memory pool
— plus
a clean-error test for single-key skew. Codec round-trip and
planner-substitution
(including broadcast-wins and AQE dynamic-join-not-clobbered cases) are
covered by
plan-string tests. TPC-H plan-stability goldens are unchanged (the feature
is a
strict no-op when disabled).
> Note: this operator has **not yet been benchmarked** end-to-end on a
cluster.
> The next step is an SF10+ A/B run with the flag on vs. off under both AQE
on and
> AQE off, and an end-to-end run of the Q18 case from #2025 at a memory
setting
> that OOMs today. Opening as a draft for design feedback ahead of those
numbers.
# Are there any user-facing changes?
Two new configuration settings (above), both defaulting to off/existing
behaviour.
No public API changes and no change to default planning. With the feature
disabled
(the default), behaviour is unchanged.
--
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]
