You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "jonathanc-n (via GitHub)" <gi...@apache.org> on 2025/08/20 23:05:56 UTC
[I] Hash Join Spilling Proposal [datafusion]
jonathanc-n opened a new issue, #17267:
URL: https://github.com/apache/datafusion/issues/17267
### Is your feature request related to a problem or challenge?
I wanted to share the idea for hash join spilling here as I would like to get input on this as hash join spilling is one of the core functionalities for query engines.
## Hybrid Hash Join
The core idea follows a hybrid hash join -> track memory for partitions + spill the largest batches until memory is back under a threshold.
## Build Phase
During build phase we collect each batch we use each batch to update the hash map before iterating to the next one.
The order would be as such:
1. Partition the build side values. Every partition is to have its memory tracked to check for spilling (partitioned hash join problems are discussed below).
2. If there is a spill, we calculate the partitions that will be spilled greedily until we are under *~80%* (configurable) of the threshold. The partitions that are spilled will be recorded to later specify which partitions need to be spilled at the end of the build process process + to know which partitioned probe batches need to be spilled (keeping one spill file per partition).
3. Resume as normal -> use the remaining values to update the hash table.
Memory threshold should also be made to be *around 90%* of memory for transient buffers when spilling.
## Probe phase
As each probe-side batch arrives, compute its hash values on the join key(s) using the same hash function/partitioning logic as the build. For each probe row, determine its partition. These are the two scenarios occur:
If the partition was *not spilled* in the build phase (meaning the build side’s corresponding partition hash table is in memory), the in-memory hash table is probed to find matches and produce joined output rows.
If the partition was spilled on the build side, then the build side’s rows for that partition are on disk (not in the hash table). In this case, the probe row is spilled into a probe-side spill file for that partition. This defers processing this partition’s matches until later, so we write all probe-side rows for that partition to disk.
If no partitions were spilled during build, then we skip any partitioning overhead on the probe side altogether and just do a normal hash join in memory for all batches.
## Recursive hash join spilling
Do the same thing in the probe phase, check if there is spilled data again.
## Problems
### 1. Splitting `RecordBatch`
We would need to do two `take()` operations on the `RecordBatch`, one for spilled and another for the batch kept in memory. I might need someone with more expertise on Arrow to offer some insight on this.
### 2. Partitioned Hash Join
Partitioned hash join execution is a little weird, due to having a shared memory pool + each partition having their own `MemoryReservation`. Datafusion is different to other engines due to other engines being able to spill entire partitions but Datafusion has no way of coordinating that. We could add a coordinator between partitions which is a similar idea to [this PR](https://github.com/apache/datafusion/pull/17197) @adriangb added, however syncing the spilling between the partitions may be nontrivial. It is difficult to sync when record batches are being spilled due to the construction of hash tables not consistently happening at the same time.
Another solution I was thinking of was isolating a single coordinator per partition, and having it keep track of memory per partition within the partition + using the `FairSpillPool` for should be a fair solution. This would be quite limiting for memory therefore we could use the idea of a dynamic memory distributor to pass more memory back to the pool if a partition doesn't seem to be using it.
*Some discussion on this part would be much appreciated*.
### 3. Dynamic filter pushdown
Dynamic filter pushdown will conflict with this feature. The coordinator that is added keeps track of the filters per spilled/unspilled partitions.
## Edge cases
- For skewed join keys that cause a single partition to collect an overwhelming amount of inputs, we fall back to a sort merge join for that individual partition.
## Follow up improvements
- In cases where there is guaranteed to be spills in a partitioned hash join, we pass the computed hash values and take parts of the hash to use to avoid rehashing. The same thing be done if recursive partitioning is done.
- This implementation will be done for regular hash join first before symmetric hash join.
- tbd... (there are many things that can be done)
If more context is needed for the coordinator and refer to `HashBuild` portion here:
https://facebookincubator.github.io/velox/develop/spilling.html - However Velox is quite different as Datafusion needs to stream batches
### Describe the solution you'd like
_No response_
### Describe alternatives you've considered
_No response_
### Additional context
_No response_
--
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.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: [I] PROPOSAL Hash Join Spilling Proposal [datafusion]
Posted by "jonathanc-n (via GitHub)" <gi...@apache.org>.
jonathanc-n commented on issue #17267:
URL: https://github.com/apache/datafusion/issues/17267#issuecomment-3290043065
Yes the hybrid hash join described above is very similar to the grace hash join. The only difference with the hybrid hash join is that it keeps one of the partitions in memory and tries to build the hash table. Most modern query engines use the hybrid hash join.
--
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: [I] Hash Join Spilling Proposal [datafusion]
Posted by "2010YOUY01 (via GitHub)" <gi...@apache.org>.
2010YOUY01 commented on issue #17267:
URL: https://github.com/apache/datafusion/issues/17267#issuecomment-3209708754
> Yes I think it would be a good idea to work on a POC for this and the SMJ improvements in parallel. Which parts of SMJ has problems currently? nothing seems to stand out to me. cc [@comphead](https://github.com/comphead)
I tried it a couple month ago, it's failing several tpch queries if it's under some memory limit. Perhaps it just requires several fixes
--
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: [I] PROPOSAL Hash Join Spilling Proposal [datafusion]
Posted by "tomz (via GitHub)" <gi...@apache.org>.
tomz commented on issue #17267:
URL: https://github.com/apache/datafusion/issues/17267#issuecomment-4410356794
# RFC: External Hash Join — refining the partitioned-build direction
**Status:** draft, intended as a **comment on
[#17267](https://github.com/apache/datafusion/issues/17267)**
(the active proposal thread), not a fresh issue.
**Posted date:** 2026-05-08
## TL;DR
A refinement of the hybrid-hash-join direction discussed in
[#17267](https://github.com/apache/datafusion/issues/17267):
1. Code-level pointers (file:line in the current `HashJoinExec`)
2. A staged 4-PR plan
3. PR 1 in progress at
[`tomz/datafusion:external-hash-join-pr1`](https://github.com/tomz/datafusion/tree/external-hash-join-pr1)
— `partitioned_build` foundation module, 270 LoC + 7 tests.
`cargo test -p datafusion-physical-plan --lib joins::hash_join`:
386/386 pass, zero diff in existing tests.
4. A proposed answer to the architectural question that
stalled [#18589](https://github.com/apache/datafusion/pull/18589)
("separate operator vs. refactor HashJoinExec")
Prior art: #1599 (2022), #12952 (2024), #17267 (2025), #18589
(2025, closed stale 2026-01).
## Why now
DataFusion sessions configured with `with_memory_limit()` bound
HashAggregate, ExternalSort, and external-merge-join. Hash joins
register as `can_spill: false` and return `Resources exhausted`
on overflow. DuckDB, PostgreSQL, Spark, and Photon implement
partitioned external hash join.
This is observable in:
* Memory-bounded execution for hash-join-heavy workloads
(TPC-DS — see "Real-world impact" below)
* Embedded / WASM / edge deployments where RAM is bounded
* Multi-tenant deployments with per-task memory limits
* DataFusion-in-Spark integrations (datafusion-comet) that need
external reclaim hooks
## Real-world impact (measured)
We hit this while running TPC-DS sf=100 single-node in a dev
environment on a VM with 8 cores, 62 GiB RAM, 223 GiB disk.
**Concrete bench across 16 probes** captured 2026-05-08.
### Headline data — Q14 sf=100 budget sweep
| Budget | Wall | Result |
|---|---:|---|
| 4 GiB | — | ❌ OOM (HashJoinInput) |
| **8 GiB** | — | ❌ **OOM (HashJoinInput)** |
| 16 GiB | 54.7 s | ✅ |
| 32 GiB | 56.2 s | ✅ |
| unlimited | 54.0 s | ✅ |
Q14 fails at any budget below ~16 GiB. No spill path; the
allocation returns `Resources exhausted`. Verbatim error:
```
Resources exhausted: Additional allocation failed for HashJoinInput[1]
HashJoinInput[3]#108 (can spill: false) consumed 220.1 MB
HashJoinInput[3]#226 (can spill: false) consumed 220.1 MB
HashJoinInput[3]#361 (can spill: false) consumed 220.1 MB
HashJoinInput[7]#386 (can spill: false) consumed 220.0 MB
HashJoinInput[7]#249 (can spill: false) consumed 220.0 MB.
Failed to allocate additional 99.0 KB for HashJoinInput[1] —
30.5 KB remain available for the total pool
```
5 concurrent HashJoinInputs at 220 MB each = 1.1 GB of
build-side state, plus DataFusion's other reservations, exceed
the 8 GiB pool by 30 KB at the failure point.
### Multi-query sf=100 picture — query shape matters
| Query | 8 GiB | 16 GiB | unlimited | Notes |
|---|---|---|---:|---|
| Q14 | ❌ OOM | ✅ 54.7 s | ✅ 54.0 s | HashJoinInput memory-cap |
| Q23 | ❌ disk-cap | (n/a) | ✅ 65.0 s | spilled aggregate hit DiskManager 100 GiB cap |
| Q4 | ✅ 77.4 s | (n/a) | ✅ 77.0 s | hash-heavy, but build sizes fit |
| Q11 | ✅ 58.9 s | (n/a) | ✅ 59.4 s | hash-heavy, but build sizes fit |
| Q72 | (n/a) | (n/a) | ⏱️ **>30min** | 9-way join, single-node-CPU bound |
The OOM surface is wider than "build side too big". Q14 hits
HashJoinInput memory exhaustion. Q23 spills via the existing
aggregate path but exceeds DataFusion's default 100 GiB
DiskManager temp cap — partition-aware spill writes only what's
needed per partition, not unbounded per-batch.
Q72 does not finish in 30 minutes single-node even with
unlimited memory. Distributed execution completes Q72 sf=100 in
~61 s; spill is not the relevant fix for that shape.
### SortMergeJoin-fallback workaround data
We carry a downstream `SortMergeJoinFallbackRule` to mitigate
the OOM by swapping HashJoin → SortMergeJoin when build-side
estimates exceed a threshold. The bench tested it explicitly:
| Run | Wall | Result |
|---|---:|---|
| Q14 8 GiB no SMJ | — | ❌ OOM (HashJoinInput) |
| Q14 8 GiB + SMJ thresh 1 GiB | — | ❌ **same OOM** |
| Q14 15 GiB + SMJ thresh 1.5 GiB | 55.9 s | ✅ |
| Q14 16 GiB + SMJ thresh 2 GiB | 56.1 s | ✅ |
| Q14 16 GiB no SMJ | 54.7 s | ✅ |
Two findings:
1. At 8 GiB the SMJ fallback cannot fire — each individual
build is below threshold; the OOM is from concurrent
reservations summing past budget. The threshold-based SMJ
fallback does not address the multi-medium-builds case.
2. At budgets where HJ fits, SMJ is ~2.5 % slower (56.1 vs
54.7 s).
The SMJ fallback is not equivalent to HashJoinExec spill in
this regime.
### Implications for the design's perf target
The 1.5× target the RFC sets (vs memory-fitting hash join)
maps to:
| Query | Target wall at tight budget | vs unlimited baseline |
|---|---:|---:|
| Q14 | ≤ 80 s at 8 GiB | (unlimited 54 s) |
| Q23 | ≤ 100 s at 8 GiB | (unlimited 65 s) |
| Q4 | ≤ 80 s at 8 GiB | (unlimited 77 s — already fits) |
| Q11 | ≤ 70 s at 8 GiB | (unlimited 59 s — already fits) |
These anchor the perf expectations for the spillable-HashJoin
PR's benchmark suite.
[Full bench writeup with raw logs.](https://github.com/your-org/spark-rust-jaaicode/blob/main/docs/datafusion-rfc-bench-results.md)
## Existing community work this builds on
| Issue / PR | Status | Relationship |
|---|---|---|
| [#1599](https://github.com/apache/datafusion/issues/1599) *Memory Limited Joins (Externalized / Spill)* (2022-01) | OPEN | The original 4-year-old ask. SortMergeJoin half landed (#9359); HashJoin half is what this closes. |
| [#12952](https://github.com/apache/datafusion/issues/12952) *Add spilling support for HashJoin* (2024-10) | OPEN | Direct restatement of the ask. |
| [#17267](https://github.com/apache/datafusion/issues/17267) *PROPOSAL Hash Join Spilling Proposal* (2025-08, **21 comments**) | OPEN | **The active design thread.** This RFC is intended to land here as a comment. Same hybrid-hash-join target; this refines with code-level concreteness + a staged plan + an in-flight PR. |
| [PR #18589](https://github.com/apache/datafusion/pull/18589) *Draft GraceHashJoin* by @osipovartem (2025-11) | CLOSED (stale, 2026-01) | **1,825 LoC working draft.** Closed as stale — not for technical reasons, but because reviewers couldn't agree on whether to do it as a separate operator or fold into HashJoinExec. We address that below. |
| [PR #17444](https://github.com/apache/datafusion/pull/17444) by @adriangb (2025-09) | **MERGED** | Refactored HashJoinExec to progressively accumulate dynamic filter bounds during build "to unblock spilling". The dynamic-filter compatibility question is **already resolved upstream**. |
| [PR #21882](https://github.com/apache/datafusion/pull/21882) *pluggable SpillFile trait* | OPEN | Introduces async-streaming spill backends (S3, BufFile, …). If we write spill code against this, we get backend portability for free. |
| [PR #22035](https://github.com/apache/datafusion/pull/22035) *SubPool* | OPEN | Per-operator memory sub-pool. Designed exactly for the partitioned-hash-join's per-partition reservations. |
| [PR #22043](https://github.com/apache/datafusion/pull/22043) *MemoryReclaimer* | OPEN | Cooperative async spill on OOM. Our partitioned hash join should register as one. |
| [PR #21425](https://github.com/apache/datafusion/pull/21425) + [#21422](https://github.com/apache/datafusion/issues/21422) *external reclaim hooks* | OPEN | Spark/Comet integration; we should make our spill compatible. |
| [#16065](https://github.com/apache/datafusion/issues/16065) *Epic: GSoC 2025 Spilling* | OPEN | Companion epic. Hash join spill fits as a natural follow-up. |
| [#13123](https://github.com/apache/datafusion/issues/13123) *EPIC Improved Externalized Hash Aggregation* | OPEN | Aggregation-side companion. Establishes spill-pool patterns we reuse. |
| [#18840](https://github.com/apache/datafusion/issues/18840) *Dynamic hash join order switching* | OPEN | Composes with this RFC: a partitioned build can also be left/right-swapped after Phase 1 completes. |
| [#20847](https://github.com/apache/datafusion/issues/20847) *Dynamic switching to partitioned mode* | OPEN | Adjacent; this RFC's partitioning is finer-grained (per-build hash partitioning, distinct from `PartitionMode::Partitioned`'s shuffle-partition). |
**Net assessment:** the feature has been requested in #1599
(2022), #12952 (2024), and #17267 (2025). #18589 reached
1,825 LoC of working draft before being closed stale. Open
threads: operator architecture (separate vs. refactor),
benchmarking, PR sequencing.
## Addressing the design questions that stalled #17267 / #18589
Two reviewer concerns surfaced that this RFC explicitly addresses:
### Concern 1: "SortMergeJoin spill is good enough" (@2010YOUY01)
Quote from #17267:
> Memory-limited equal join is a solved problem through sort
> merge join, though there are some outstanding work to further
> improve it. Grace Hash Join is trickier to configure because
> we have to decide partition number up front, and it's not
> obvious to me it's more performant.
**Response:** at memory-fitting scale, hash join is faster
than SMJ. SortMergeJoin spill helps when the build does not
fit. When the build fits but is large, hash join is the
faster path. Data points:
* DuckDB ran TPC-H sf=300 on a 16 GB Raspberry Pi 5 using
hybrid hash join.
* Per @Omega359 on #18589: "I don't care how fast an algo is
if I need a 512GB+ machine to run it."
**Mitigation for the partition-count tuning concern** (this
RFC's Open Question 1): default to 8 partitions, expose as
`SessionConfig::hash_join_partitions`, recursively split on
overflow (PR 4 in this RFC). Same approach DuckDB uses; their
default is 8.
### Concern 2: "separate operator vs. refactor" (the #18589 stall)
Quote from @osipovartem's PR description:
> We should add a physical plan optimizer rule to replace
> HashJoin by GraceHashJoin for inner join type only for now.
> Also maybe add some config flag to enable/disable …
The stall was "should this be `GraceHashJoinExec` (sibling) or
should we fold spill into `HashJoinExec`?"
**Resolution proposed here: fold into `HashJoinExec`, gated by
config.** Concretely:
```rust
// SessionConfig
pub struct ExecutionOptions {
/// When the build side exceeds this fraction of the memory
/// pool, switch to partitioned execution with on-disk spill.
/// Default 0.0 (disabled, current behavior).
pub hash_join_spill_threshold: f64,
/// Number of partitions for partitioned-build mode.
/// Default 8.
pub hash_join_partitions: usize,
}
```
This avoids the duplication concern while still being opt-in
for stability. Existing users see no behavior change unless
they set `hash_join_spill_threshold > 0` or
`with_memory_limit()` (under which the threshold auto-derives,
similar to how aggregation spill works today).
The 4-PR plan below makes this incremental:
* PR 1 (already up): partition utility module — pure addition
* PR 2: BuildSideState becomes per-partition (in-memory only)
* PR 3: spill to disk on overflow
* PR 4: skew + outer joins + dynamic filter integration
Each PR is independently reviewable; behavior change is
gated by config from PR 3 onward.
## What's already at `tomz/datafusion:external-hash-join-pr1`
A 270-LoC `partitioned_build` module
(`datafusion/physical-plan/src/joins/hash_join/partitioned_build.rs`)
with:
```rust
pub const DEFAULT_NUM_PARTITIONS: usize = 8;
pub fn partition_batch_by_hash(
batch: &RecordBatch,
on_keys: &[Arc<dyn PhysicalExpr>],
num_partitions: usize,
random_state: &RandomState,
) -> Result<Vec<Option<RecordBatch>>>;
pub fn partition_with_hashes(
batch: &RecordBatch,
hashes: &[u64],
num_partitions: usize,
) -> Result<Vec<Option<RecordBatch>>>;
```
7 unit tests covering hash distribution, multi-column compound
keys, determinism, edge cases. Module is registered in `mod.rs`
but not yet wired into `HashJoinExec`, so existing behavior is
unchanged. `cargo test -p datafusion-physical-plan --lib
joins::hash_join`: 386/386 pass.
The module is reusable across both architectural paths
(refactor vs. sibling operator) and across both partitioning
sources (build keys vs. probe keys with caller-provided
hashes).
### The hash-aggregate contrast (Q67)
Q67 is a rollup with grouping sets — the workload DuckDB
highlights in their March 2026
[cheap-MacBook-can-do-big-data post](https://duckdb.org/2026/03/11/big-data-on-the-cheapest-macbook).
At TPC-DS sf=100 single-node, varying the memory budget on
Q67:
| Budget | Q67 wall |
|---|---:|
| unlimited | 48.8 s |
| 8 GiB | 48.7 s |
| 4 GiB | 49.9 s |
| 2 GiB | 50.0 s |
**2.5 % wall delta at the tightest budget.** Q67 spills via
DataFusion's hash-aggregate spill path (#7400 / 2023).
The contrast: hash-aggregate is memory-bounded with disk
spill; hash-join is not. Q67 and Q14 are both
memory-pressure-sensitive; Q67 spills, Q14 returns
`Resources exhausted`.
### Note on distributed-scale data
The single-node Q14 / Q23 / Q67 bench above demonstrates the
HashJoin OOM, hash-agg spill, and SMJ-fallback behavior.
Distributed sf=1000 runs would test whether sharding moves
the boundary, but per-executor working sets still hit
HashJoinInput memory caps. Distributed validation can be
added post-RFC on top of the existing
[Q72 sf=1000 cluster validation](https://github.com/your-org/spark-rust-jaaicode/blob/main/docs/bench-tpcds-sf1000-q72-utf8view-revalidation.md)
that ran end-to-end at default (unlimited) executor memory.
## Workarounds tried
* **Larger memory budget**: works at sf=100/single-node;
doesn't scale to larger inputs.
* **`SortMergeJoinExec` fallback** via a physical-optimizer
rule that swaps based on stats
([`SortMergeJoinFallbackRule`](https://github.com/your-org/spark-rust-jaaicode/blob/main/crates/spark-dataframe/src/sort_merge_fallback.rs)).
Functions as a correctness floor; SortMergeJoin is 1.5-3×
slower than memory-fitting hash join (and per the bench
above, does not fire in the multi-medium-build case).
* **Distribute the query** across cluster nodes: works at
scale; does not apply to single-node, embedded, or
per-task-bounded deployments.
## Design — partitioned build with disk spill
The standard textbook approach. Break the build side into N
partitions by hash modulo N. Each partition gets `pool / N` of
the budget. When a partition exceeds its share, spill it to
disk. On the probe side, partition rows the same way; only one
partition's build state is in memory at a time during probe.
### Phase 1 — partition-then-build
```
build_input ──► HashPartitioner(N=8, hash on join_key)
│
├── partition_0 → in-mem hash table (or spill if overflow)
├── partition_1 → in-mem hash table (or spill)
├── ...
└── partition_7 → in-mem hash table (or spill)
```
When `MemoryReservation::try_grow` fails, victim-pick the
largest in-memory partition, write it to disk via the
`DiskManager`, free its memory, retry.
### Phase 2 — probe-then-merge
```
probe_input ──► HashPartitioner(N=8, same hash function)
│
└── For each partition i:
if build[i] is in memory: probe directly
else: load from disk,
probe, free memory
```
**Memory invariant**: at most one build partition's hash map is
materialized at any moment during probe. Total memory =
`pool / N` instead of full pool.
### Algorithm pseudocode
```rust
// Build phase
for batch in build_input {
let p = hash_partition(batch, on_left, N);
match reservation.try_grow(batch.size()) {
Ok(_) => state.partitions[p].push(batch),
Err(_) => {
spill_largest_partition(&mut state)?;
reservation.try_grow(batch.size())?; // should fit now
state.partitions[p].push(batch);
}
}
}
// Probe phase
for probe_batch in probe_input {
let probed_partitions = partition_by_hash(probe_batch, on_right, N);
for (p, p_batch) in probed_partitions {
let build_data = state.partitions[p].load_for_probe(disk_manager)?;
run_probe(p_batch, build_data.hash_map);
// Optionally free build_data after probe completes.
}
}
```
## Where this lands in the source tree
Concrete file-level mapping (relative to
`datafusion/physical-plan/src/joins/hash_join/`):
| File | Change | LoC |
|---|---|---:|
| `exec.rs::BuildSideState` | replace `batches: Vec<RecordBatch>` with `partitions: [PartitionSlot; N]` | ~30 |
| `exec.rs::collect_left_input` (~L1836) | partition each incoming batch via existing `create_hashes`; spill on `try_grow` failure | ~80 |
| `exec.rs::HashJoinStream::poll_next_impl` | partition probe batches by hash; per-partition probe loop | ~150 |
| `exec.rs` (line 1303 + 1323) | switch `MemoryConsumer::new("HashJoinInput")` to `.with_can_spill(true)` | ~5 |
| `spill.rs` (NEW module) | `PartitionSlot { InMemory(Vec<RecordBatch>) \| Spilled { path, ... } }` + spill IPC writer/reader. Mirror `aggregates/group_values/spilling.rs`. | ~600 |
| Tests in `core/tests/sql/joins.rs` | spill-fixture parquet builds + integration tests under tight memory budget | ~400 |
**Total estimate: ~1,200-1,500 LoC + tests.**
### Concrete code skeletons
**`BuildSideState` becomes partitioned:**
```rust
const NUM_PARTITIONS: usize = 8; // tunable via SessionConfig
struct BuildSideState {
metrics: BuildProbeJoinMetrics,
reservation: MemoryReservation,
on_left: Vec<PhysicalExprRef>,
schema: SchemaRef,
/// One slot per partition. In-memory until try_grow fails;
/// then individual partitions get spilled to disk.
partitions: [PartitionSlot; NUM_PARTITIONS],
num_rows: usize,
bounds_accumulators: Option<Vec<CollectLeftAccumulator>>,
}
enum PartitionSlot {
InMemory(Vec<RecordBatch>),
Spilled {
path: RefCountedTempFile, // managed by DiskManager
bytes_written: usize,
rows: usize,
},
}
```
**Spill trigger** — replaces today's `try_grow?` line at L1873:
```rust
let batch_size = get_record_batch_memory_size(&batch);
let p = hash_partition_idx(&batch, &state.on_left, NUM_PARTITIONS)?;
if let Err(_) = state.reservation.try_grow(batch_size) {
spill_largest_partition(&mut state, disk_manager)?;
state.reservation.try_grow(batch_size)?; // should fit
}
state.partitions[p].push(batch);
```
**Probe-side** — replaces single-pass probe with per-partition:
```rust
let probe_partitioned = partition_batch_by_hash(
&probe_batch, &self.on_right, NUM_PARTITIONS,
)?;
let mut output = Vec::new();
for (p_idx, p_batch) in probe_partitioned {
let build = self.load_build_partition(p_idx, disk_manager)?;
output.push(self.run_probe(p_batch, &build.hash_map)?);
// build dropped here, memory reclaimed
}
concat_batches(&self.schema, &output)
```
**DiskManager integration** mirrors existing
`group_values/spilling.rs`:
```rust
fn spill_partition(
state: &mut BuildSideState,
p_idx: usize,
disk_manager: &DiskManager,
) -> Result<()> {
let path = disk_manager.create_tmp_file("HashJoin_BuildSpill")?;
let writer = StreamWriter::try_new(File::create(&path)?, &state.schema)?;
let batches = std::mem::take(state.partitions[p_idx].as_in_memory_mut());
let bytes: usize = batches.iter().map(get_record_batch_memory_size).sum();
for b in &batches { writer.write(b)?; }
writer.finish()?;
state.reservation.shrink(bytes);
state.partitions[p_idx] = PartitionSlot::Spilled {
path, bytes_written: bytes,
rows: batches.iter().map(|b| b.num_rows()).sum(),
};
Ok(())
}
```
## Edge cases
1. **Skewed builds.** All rows hash to one partition →
recursive partitioning. When `spill_largest_partition` is
called twice on the same partition, double NUM_PARTITIONS
for that partition's children.
2. **Single-large-key.** All rows have the same key. Recursive
partitioning doesn't help. Options:
- Fall back to NestedLoopJoin for this build (slow but
bounded);
- Emit warning + use SortMergeJoinExec via the existing
fallback rule (mirroring DuckDB's behavior).
3. **Outer joins.** The "matched" bitmap (visited-indices for
FULL/LEFT joins) needs care. Two options:
- Per-partition bitmap, merged at end. Memory cost: 1 bit per
build row, fine even at sf=1000 (a few hundred MB max).
- Disable spill for outer joins in v1; document as future work.
4. **Dynamic filter pushdown.** HashJoinExec creates a dynamic
filter from build-side bounds and pushes to probe scan. With
partitioned build:
- Compute filter ONCE at end of build phase (after all
partitions including spilled ones contribute their min/max);
- OR skip dynamic filter for spilling joins (acceptable v1).
5. **`PartitionMode::CollectLeft` vs `Partitioned`.** Today's
code has separate paths. The spill rewrite naturally
consolidates: both modes become "partition the build by hash
modulo N". CollectLeft becomes "1 input partition × N hash
partitions"; Partitioned becomes "M input partitions × N hash
partitions, distributed by input partition".
## Performance target
| Workload | Today (memory-fitting) | This RFC (memory-bounded) |
|---|---|---|
| Build fits in pool (common case) | Hash-build O(n), hash-probe O(m) | Same — partition is cheap (O(n) hash, no I/O), in-memory probe identical |
| Build exceeds pool by 2-4× | OOM | Spill 1-3 partitions, probe-time disk reads ≈ 1.2-1.5× wall vs memory-fitting |
| Build exceeds pool by 10×+ | OOM | Most partitions spill, probe-time disk reads ≈ 2-3× wall |
Target perf for spilling case: within 1.5× of memory-fitting
hash join (SortMergeJoin is 1.5-3× per the bench above).
## Implementation plan / PR breakdown
For upstream review-ability, suggest 4 PRs:
| PR | Scope | LoC | Time |
|---|---|---:|---:|
| 1 | Partitioned BuildSide (in-mem only, no spill yet). Refactor `collect_left_input` to always partition. Probe pipeline changes. Tests verify behavioral equivalence. | ~400 + 200 | ~1 week |
| 2 | Add spill via DiskManager. Victim-pick + spill on `try_grow` failure. Probe-time deserialization. Tests verify spill triggers + correct results under tight budget. | ~600 + 400 | ~1 week |
| 3 | Recursive partitioning for skew. | ~200 + 150 | ~3 days |
| 4 | Outer join handling + dynamic filter integration. | ~150 + 100 | ~2 days |
Total: 3-4 weeks elapsed including review cycles.
PR 1 is the architecture-only change — no behavior change.
Subsequent PRs add features incrementally.
## Open questions for community
1. **Default `NUM_PARTITIONS` value** (raised by @2010YOUY01 on
#17267). Proposal: default 8 (DuckDB's default), exposed as
`datafusion.execution.hash_join.partitions`. Recursive split
on overflow (PR 4) handles undersized cases.
2. **Recursive partitioning depth.** Cap at 3 levels (8 → 64 →
512 partitions max)? Beyond that, fall through to
NestedLoopJoin or error?
3. **Probe-side partition reuse.** If the probe side is already
`Partitioning::Hash(on_right, N')`, can we reuse that
partitioning for the join (avoid the re-hash)? Requires
matching N == N' and identical hash function. Optimizer-side
integration; could be follow-up.
4. **Spill format / backend.** Arrow IPC stream matches existing
`group_values/spilling.rs`. The pluggable `SpillFile` trait
in #21882 would provide backend portability (S3, BufFile,
etc.) at the cost of waiting for #21882 to merge or landing
both PRs in lock-step.
5. **Dynamic filter integration** (raised on #17267). Resolved
upstream by #17444 (merged Sep 2025): dynamic filter bounds
accumulate progressively during build, which is
spill-compatible.
6. **Memory pool integration.** Should partitioned reservations
register through `SubPool` (PR #22035)? If `SubPool` lands,
per-partition build reservations match its target use case:
one `SubPool` per `HashJoinExec`, sub-reservation per
partition. Addresses the "FairSpillPool penalises the active
operator" concern from #22036.
7. **External reclaim.** Register spill as a `MemoryReclaimer`
(PRs #21425 / #22043) so external Spark/Comet drivers can
request spill on demand. Covered in PR 4.
8. **Operator architecture: refactor vs. sibling** (the #18589
stall point). Proposal: refactor `HashJoinExec`, gated by
config. `hash_join_spill_threshold` defaults to 0.0
(disabled, current behavior); auto-derives under
`with_memory_limit()`.
9. **Backwards compatibility.** The `PartitionMode` enum on
`HashJoinExec` is public API. Behavior change: when spill
is enabled, `CollectLeft` no longer collects to a single
Vec; it partitions. Should be transparent to most
downstream consumers, but worth flagging.
## What this enables downstream
- DataFusion sessions with `with_memory_limit()` become
memory-bounded for hash-join workloads (today they bound
hash-aggregate, sort, and external-merge-join only).
- TPC-DS at sf=1000 single-node on 64 GiB-class hardware.
- Cloud-deployed DataFusion (AWS Lambda, Cloudflare Workers,
Fly.io) with datasets larger than allocated RAM.
- Embedded use cases (DataFusion-WASM, edge devices).
## Reference: similar implementations elsewhere
- **DuckDB**: `src/execution/operator/join/perfect_hash_join_executor.cpp`,
`physical_hash_join.cpp` — partitioned hash join with spill.
- **PostgreSQL**: Hash join in `src/backend/executor/nodeHashjoin.c` +
`nodeHash.c` — batch-based external hash join (very similar
partitioning approach).
- **DataBricks Photon**: closed-source, but their published papers
describe a similar partitioned approach.
## Next steps
This RFC is meant to land as a **comment on #17267**, not a new
issue, to converge that thread toward implementation:
1. **Comment on #17267** with this body. Tag the participants
(@jonathanc-n @2010YOUY01 @adriangb @alamb @comphead
@osipovartem @Omega359 @Dandandan @camuel) and propose
resolution: refactor HashJoinExec, opt-in via config.
2. **Reach out to @osipovartem** specifically about #18589.
Their 1,825-LoC draft has substantial reusable code
(`PartitionWriter`, `SpillLocation` enum, hybrid
in-memory/disk fallback). Two paths:
- Revive #18589 with the architectural change (refactor vs
sibling) plus our `partitioned_build` foundation
- Cherry-pick relevant pieces into a fresh PR series
3. **Open Draft PR 1** for the `partitioned_build` foundation
([already up](https://github.com/tomz/datafusion/tree/external-hash-join-pr1))
once the design direction is acknowledged on #17267.
4. **Iterate PRs 2-4** in sequence, each with its own RFC
comment if the design changes.
If significant design changes emerge from the #17267
discussion, the pseudocode and file pointers here are
adjustable.
## Acknowledgments
- @jonathanc-n for the #17267 proposal and hybrid-hash-join
framing
- @osipovartem for the #18589 draft (reusable code:
`PartitionWriter`, `SpillLocation`, hybrid in-memory/disk
fallback)
- @adriangb for the dynamic-filter spill-compat refactor
(#17444)
- @2010YOUY01, @comphead, @alamb, @Omega359, @camuel,
@Dandandan for the #17267 discussion
--
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: [I] PROPOSAL Hash Join Spilling Proposal [datafusion]
Posted by "osipovartem (via GitHub)" <gi...@apache.org>.
osipovartem commented on issue #17267:
URL: https://github.com/apache/datafusion/issues/17267#issuecomment-3467735359
Here are several ideas and directions for improving the HashJoin implementation during DF code research
### Hybrid spilling with memory awareness
- We could extend SpillManager with an automatic hybrid spilling mechanism, something like:
```rust
fn spill_batch_auto(&self, batch: &RecordBatch, request_msg: &str) -> Result<SpillLocation> { ... }
```
Where
```rust
pub enum SpillLocation {
Memory(Arc<InMemorySpillBuffer>),
Disk(RefCountedTempFile),
}
pub struct InMemorySpillBuffer {
batches: Vec<RecordBatch>,
total_bytes: usize,
}
```
This would allow DataFusion to decide dynamically whether to keep a batch in memory or spill it to disk based on current memory pool availability.
### Index-based left-side storage
- Since the left (build-side) RecordBatch is only required later during build_batch_from_indices,
we could store only references (metadata) to the actual spilled or in-memory chunks:
```rust
pub struct LeftChunkIndex {
/// Spill file or memory handle
pub location: SpillLocation,
/// Global start offset of this chunk (in build-side concatenation)
pub start_row: usize,
/// Number of rows in this chunk
pub num_rows: usize,
}
```
- This allows lazy reloading of only the necessary build-side chunks during probing.
When **lookup_join_hashmap** func filters out 0..N rows, a significant portion of data may never need to be reloaded from disk or memory at all.
### Avoid duplicate evaluation during collect_left_input
- Currently we evaluate an expression against a RecordBatch twice (not sure if it is havy operation)
```rust
let keys_values = on
.iter()
.map(|c| c.evaluate(batch)?.into_array(batch.num_rows()))
.collect::<Result<Vec<_>>>()?;
```
- once inside update_hash, and again later inside collect_left_input.
It would be more efficient to return these values directly from update_hash so they can be reused, avoiding redundant array computation.
### Reduced in-memory footprint per partition
- By combining the above ideas, we could avoid keeping the entire left-side RecordBatch in memory for each partition.
Instead, only the hash map and chunk indices would remain resident, while the actual build-side data could be reloaded on demand (from either memory buffers or spill files).
- For TPCH Q18 with scale factor 10 we store (lineitem table ~ 66KK records , 32 partitions). Hashmap looks over-allocated
```JoinLeftData for partition 31 with total 182.74 MB:
• Hashmap 68.00 MB
• Batch: 114.50 MB, count 2069975
• Bitmap: 252.68 KB
```
--
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: [I] Hash Join Spilling Proposal [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on issue #17267:
URL: https://github.com/apache/datafusion/issues/17267#issuecomment-3211526631
TLDR is I agree that the [hybrid hash join](https://en.wikipedia.org/wiki/Hash_join#Hybrid_hash_join) is basically state of the art for hash join implementations and I think would be a great end goal
> I recommend to get sort merge join working reliably before experimenting HJ spilling (i.e. benchmarks should be able to finish under a modest memory limit, perhaps also more tests),
I think this is a really wise suggestion from @2010YOUY01
> Yes I think it would be a good idea to work on a POC for this and the SMJ improvements in parallel. Which parts of SMJ has problems currently? nothing seems to stand out to me. cc [@comphead](https://github.com/comphead)
One major limitation today is that the the user has to pick between HashJoin and MergeJoin at *Plan* time.
From the [config settings](https://datafusion.apache.org/user-guide/configs.html)
datafusion.optimizer.prefer_hash_join | true | When set to true, the physical plan optimizer will prefer HashJoin over SortMergeJoin. HashJoin can work more efficiently than SortMergeJoin but consumes more memory
-- | -- | --
## Proposed Course of Action
1. Phase 1: Implement HashJoin spilling as a "runtime switch to MergeJoin" -- basically dump the hash table to spill files and switch to using MergeJoin for the rest of the query
2. Phase 2: Implement the more sophisticated HybridHashJoin
Rationale:
1. As @2010YOUY01 says, I think the runtime switch to MergeJoin is simpler and can build on our work with the spilled sorts.
2. While we implement the simpler case, we will gain experience (and write tests) the HashJoin code
3. The major downside of the "runtime switch to MergeJoin" approach is that it has a "performance cliff" -- if the operator switches from Hash to MergeJoin it will be much slower
The major benefit of a Hybrid Hash Join type approach is that it degrades more gracefully (as you only need to spill some partitions, for example). While an important feature, I think we should first focus on basic functionality
I have some more thoughts about the hybrid hash join approach, but I don't want to confuse this conversation now
--
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: [I] Hash Join Spilling Proposal [datafusion]
Posted by "comphead (via GitHub)" <gi...@apache.org>.
comphead commented on issue #17267:
URL: https://github.com/apache/datafusion/issues/17267#issuecomment-3211423217
Thanks for taking look on that, HJ spilling is challenging task, we started to explore it in https://github.com/apache/datafusion/issues/12952 also there is some implementation details from Meta
--
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: [I] Hash Join Spilling Proposal [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on issue #17267:
URL: https://github.com/apache/datafusion/issues/17267#issuecomment-3208341077
Filter pushdown is pretty flexible. Very little information needs to be kept in memory and building it can be deferred. I think we can start by saying that things that fit in the memory budget should push down as they do now (no perf loss) and things that don't fit in memory can be less efficient about pushdown and we can improve that later.
--
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: [I] Hash Join Spilling Proposal [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on issue #17267:
URL: https://github.com/apache/datafusion/issues/17267#issuecomment-3210062449
One thing that might help stability is having a feature flag for spilling / not spilling and running it under https://github.com/apache/datafusion/pull/17177 or similar
--
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: [I] PROPOSAL Hash Join Spilling Proposal [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on issue #17267:
URL: https://github.com/apache/datafusion/issues/17267#issuecomment-5399783691
> I'd like to help with this topic, but I've noticed several related PRs have gone stale and several discussions ended without consensus. Is anyone interested in moving this forward or reviewing the code?
>
> What's the status of [#22043](https://github.com/apache/datafusion/pull/22043) and GraceHashJoin? Are they stale due to lack of reviewers, or because of unresolved design challenges?
I think lack of reviewers
But in general, externalized hash join is likely one of the most complex operators in a real database system, and I don't think we have had anyone who has the combination of:
1. Knowledge (or ability to get the knowledge)
2. System skill to implement / explain it
3. Familiarity with DataFusion (and can help get people to comment on the deisgn and review)
> Since I'd rather put effort into something that will actually get reviewed, one area I think could be worked on first is benchmarks/tests for external joins. Does anyone who follows this topic closely have a suggestion on where to start — or better yet, would be willing to commit to reviewing it?
I think this would be a great place to start. I think @mbutrovich and @2010YOUY01 have been interested recently. @Dandandan also seems to have gotten the join bug recently 🥳 🦜
With that data I would personally recommend writing up a proposal / design. This would include classic things like
* problem being solved
* Current implementation status (e.g. how large is the largest hash join today)
* Proposed design
* Breakdown of steps needed to get there
* Alternatives considered (and maybe pointers to other systems / what they implement -- e.g. DuckDB, Spark and ClickHouse)
I think if you kept the proposal reasonably short (a few pages) you would likely get quite a lot of feedback
--
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: [I] PROPOSAL Hash Join Spilling Proposal [datafusion]
Posted by "comphead (via GitHub)" <gi...@apache.org>.
comphead commented on issue #17267:
URL: https://github.com/apache/datafusion/issues/17267#issuecomment-3366176816
SMJ may require more memory and additional expenses on sorting, so the HJ would be faster in most cases. But SMJ is more robust on limited memory if implemented correctly.
Optimized SMJ can be awesome, some notes on optimization here https://cs186berkeley.net/notes/note9/#an-important-refinement
DF SMJ still missing some optimizations so it is possible to end up with what @camuel faced.
--
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: [I] PROPOSAL Hash Join Spilling Proposal [datafusion]
Posted by "jonathanc-n (via GitHub)" <gi...@apache.org>.
jonathanc-n commented on issue #17267:
URL: https://github.com/apache/datafusion/issues/17267#issuecomment-5400041513
Currently finishing up school + internship and will be completely free in the fall to be able to write something up. If there is a good amount of reviewers, I'd be happy to try to attempt this.
--
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: [I] PROPOSAL Hash Join Spilling Proposal [datafusion]
Posted by "jayzhan211 (via GitHub)" <gi...@apache.org>.
jayzhan211 commented on issue #17267:
URL: https://github.com/apache/datafusion/issues/17267#issuecomment-5461759886
Start with https://github.com/apache/datafusion/issues/24768
--
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: [I] PROPOSAL Hash Join Spilling Proposal [datafusion]
Posted by "jonathanc-n (via GitHub)" <gi...@apache.org>.
jonathanc-n commented on issue #17267:
URL: https://github.com/apache/datafusion/issues/17267#issuecomment-3290346862
Yes that sounds like a good idea. I will try to do this, avoiding the config options would be ideal in this case.
--
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: [I] PROPOSAL Hash Join Spilling Proposal [datafusion]
Posted by "camuel (via GitHub)" <gi...@apache.org>.
camuel commented on issue #17267:
URL: https://github.com/apache/datafusion/issues/17267#issuecomment-3364176121
From my tests SMJ requires more memory than HJ with a large number of partitions (like 32), so switching from HJ to SMJ makes memory situation worse not better in that case.
--
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: [I] PROPOSAL Hash Join Spilling Proposal [datafusion]
Posted by "jayzhan211 (via GitHub)" <gi...@apache.org>.
jayzhan211 commented on issue #17267:
URL: https://github.com/apache/datafusion/issues/17267#issuecomment-5386029991
I'd like to help with this topic, but I've noticed several related PRs have gone stale and several discussions ended without consensus. Is anyone interested in moving this forward or reviewing the code?
What's the status of [#22043](https://github.com/apache/datafusion/pull/22043) and GraceHashJoin? Are they stale due to lack of reviewers, or because of unresolved design challenges?
Since I'd rather put effort into something that will actually get reviewed, one area I think could be worked on first is benchmarks/tests for external joins. Does anyone who follows this topic closely have a suggestion on where to start — or better yet, would be willing to commit to reviewing it?
--
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: [I] PROPOSAL Hash Join Spilling Proposal [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on issue #17267:
URL: https://github.com/apache/datafusion/issues/17267#issuecomment-3259218092
I've opened a PR which should unblock / make dynamic filters compatible with join spilling: https://github.com/apache/datafusion/pull/17444
--
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: [I] Hash Join Spilling Proposal [datafusion]
Posted by "2010YOUY01 (via GitHub)" <gi...@apache.org>.
2010YOUY01 commented on issue #17267:
URL: https://github.com/apache/datafusion/issues/17267#issuecomment-3209685859
> Yes that is a bit difficult as well, we don't know size up front and usually the number of partitions is based on the memory size and memory limit. It will just be a configurable variable.
>
> > I recommend to get sort merge join working reliably before experimenting HJ spilling (i.e. benchmarks should be able to finish under a modest memory limit, perhaps also more tests), the existing solution is not production ready yet, but I think SMJ should have lower maintenance overhead -- It's core is reusing the external sort implementation.
>
> Is there a reason we shouldn't do it in parallel? I will only start after [#17260](https://github.com/apache/datafusion/issues/17260) is finished.
I don’t think we’re ready to merge another external join implementation right now. The main reason is that the existing SMJ is still flaky. Adding another potentially flaky implementation (given the lack of tests to ensure reliability) won’t really help users, while it will increase maintenance overhead.
To make this implementation mergeable, I think it would be better to:
1. Strengthen external join tests, e.g., allow the TPC-H benchmark to run under fuzzed memory limit configurations.
2. Fix issues in SMJ and improve its performance. Letting HJ fall back to SMJ when the memory limit is hit could be both simpler and avoid introducing additional configuration (like partition numbers).
3. Do a PoC with this approach. If it provides significant performance improvements, then it would make sense to include it.
So at this stage, I think it’s a good idea to implement it as a PoC, but there’s still quite a bit of work to do before it’s mergeable.
--
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: [I] PROPOSAL Hash Join Spilling Proposal [datafusion]
Posted by "comphead (via GitHub)" <gi...@apache.org>.
comphead commented on issue #17267:
URL: https://github.com/apache/datafusion/issues/17267#issuecomment-3290029337
HJ Spilling is complicated as mentioned above, however we can consider a new type of join which is Grace hash Join https://en.wikipedia.org/wiki/Hash_join#Grace_hash_join which designed specifically for processing tables if build hash table is too large for available memory pages. the approach is recursive using Divide and Conquer algorithm
The algorithm
### Partition Phase (Hash Partitioning)
Choose a hash function h1 on the join key.
For each tuple r ∈ R, compute bucket h1(r.key) mod p and write tuple into that bucket on disk.
Do the same for S.
Now both R and S are split into p partitions: R1…Rp, S1…Sp.
Important: corresponding partitions Ri and Si contain all possible join partners (since hashing guarantees collocation).
If partitions are still too large to fit in memory, recursively re-partition them with a different hash function h2, h3, … until partitions are small enough.
This recursive spilling is sometimes called hybrid grace hash join.
### Build Phase (In-Memory Hash Table)
For each partition pair (Ri, Si):
Load the smaller partition (say Ri) into memory and build a hash table on its join key.
### Probe Phase
Stream tuples from the other partition (Si), probing the in-memory hash table to find matches.
Output joined results.
--
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: [I] PROPOSAL Hash Join Spilling Proposal [datafusion]
Posted by "Dandandan (via GitHub)" <gi...@apache.org>.
Dandandan commented on issue #17267:
URL: https://github.com/apache/datafusion/issues/17267#issuecomment-3364471092
> From my tests SMJ requires more memory than HJ with a large number of partitions (like 32), so switching from HJ to SMJ makes memory situation worse not better in that case.
I think that makes sense when all memory can be in memory, but is that also the case when memory is constrained (e.g. limiting the amount of memory so it _has_ to spill to disk)? I think that's what is proposed here.
--
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: [I] PROPOSAL Hash Join Spilling Proposal [datafusion]
Posted by "Omega359 (via GitHub)" <gi...@apache.org>.
Omega359 commented on issue #17267:
URL: https://github.com/apache/datafusion/issues/17267#issuecomment-3493620648
I see a PR for grace hash join was mentioned above. Still draft but promising :)
--
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: [I] PROPOSAL Hash Join Spilling Proposal [datafusion]
Posted by "2010YOUY01 (via GitHub)" <gi...@apache.org>.
2010YOUY01 commented on issue #17267:
URL: https://github.com/apache/datafusion/issues/17267#issuecomment-3290338857
> [@alamb](https://github.com/alamb) [@2010YOUY01](https://github.com/2010YOUY01) There is a problem with the HJ -> SMJ runtime switch. I might be unaware but there isn't a way to enforce sort and repartition at runtime.
>
> The solution I was thinking of was to:
>
> * During physical planner, pass in a flag for whether to allow for the switch from HJ to SMJ during runtime.
> * Hash join will have required input ordering enforce a sort if the flag above is true
> * If hash join detects a spill during `collect_left_input`, it will spill and let SMJ pick up the work. If no spill is detected, then it will continue as usual.
>
> This solution doesn't make too much sense since we would need to pay the cost of sorting + the regular hash join execution if there is no need to spill. We should just specify to users that the prefer_hash_join flag should be false if the user needs spilling for joins.
Perhaps we can extract the sorting logic from SortExec into a utility function, like `fn with_sort(input: SendableRecordBatchStream, sort_order: SortExpr) -> Result<SendableRecordBatchStream>`
https://github.com/apache/datafusion/blob/e2c2d3840e010cdf4b3104d4ceb21174d9afeeb3/datafusion/physical-plan/src/sorts/sort.rs#L1221-L1282
When we hit memory limit during execution, the HJ->SMJ transition includes a step that converts the original input stream to a sorted stream. I think a important advantage (at this moment) of this online transition approach is it won't require any config options for different workloads
--
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: [I] Hash Join Spilling Proposal [datafusion]
Posted by "2010YOUY01 (via GitHub)" <gi...@apache.org>.
2010YOUY01 commented on issue #17267:
URL: https://github.com/apache/datafusion/issues/17267#issuecomment-3208947844
One challenge is how to decide the partition number up front, looks like it's a static configuration, and should be configured differently from run to run?
I recommend to get sort merge join working reliably before experimenting HJ spilling (i.e. benchmarks should be able to finish under a modest memory limit, perhaps also more tests), the existing solution is not production ready yet, but I think SMJ should have lower maintenance overhead -- It's core is reusing the external sort implementation.
--
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: [I] Hash Join Spilling Proposal [datafusion]
Posted by "jonathanc-n (via GitHub)" <gi...@apache.org>.
jonathanc-n commented on issue #17267:
URL: https://github.com/apache/datafusion/issues/17267#issuecomment-3209074353
Yes that is a bit difficult as well, we don't know size up front and usually the number of partitions is based on the memory size and memory limit. It will just be a configurable variable.
> I recommend to get sort merge join working reliably before experimenting HJ spilling (i.e. benchmarks should be able to finish under a modest memory limit, perhaps also more tests), the existing solution is not production ready yet, but I think SMJ should have lower maintenance overhead -- It's core is reusing the external sort implementation.
Is there a reason we shouldn't do it in parallel? I will only start after #17260 is finished.
--
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: [I] PROPOSAL Hash Join Spilling Proposal [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on issue #17267:
URL: https://github.com/apache/datafusion/issues/17267#issuecomment-5462704891
@jayzhan211 consolidated this project / plan in
- https://github.com/apache/datafusion/issues/24768
Let's focus our discussion there
--
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: [I] Hash Join Spilling Proposal [datafusion]
Posted by "jonathanc-n (via GitHub)" <gi...@apache.org>.
jonathanc-n commented on issue #17267:
URL: https://github.com/apache/datafusion/issues/17267#issuecomment-3209701419
Yes I think it would be a good idea to work on a POC for this and the SMJ improvements in parallel. Which parts of SMJ has problems currently? nothing seems to stand out to me. cc @comphead
--
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: [I] Hash Join Spilling Proposal [datafusion]
Posted by "jonathanc-n (via GitHub)" <gi...@apache.org>.
jonathanc-n commented on issue #17267:
URL: https://github.com/apache/datafusion/issues/17267#issuecomment-3208332071
@adriangb If possible, I'd like some input for the filter pushdown side of things.
--
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: [I] PROPOSAL Hash Join Spilling Proposal [datafusion]
Posted by "jonathanc-n (via GitHub)" <gi...@apache.org>.
jonathanc-n commented on issue #17267:
URL: https://github.com/apache/datafusion/issues/17267#issuecomment-3290012901
@alamb @2010YOUY01 There is a problem with the HJ -> SMJ runtime switch. I might be unaware but there isn't a way to enforce sort and repartition at runtime.
The solution I was thinking of was to:
- During physical planner, pass in a flag for whether to allow for the switch from HJ to SMJ during runtime.
- Hash join will have required input ordering enforce a sort if the flag above is true
- If hash join detects a spill during `collect_left_input`, it will spill and let SMJ pick up the work. If no spill is detected, then it will continue as usual.
This solution doesn't make too much sense since we would need to pay the cost of sorting + the regular hash join execution if there is no need to spill. We should just specify to users that the prefer_hash_join flag should be false if the user needs spilling for joins.
--
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: [I] PROPOSAL Hash Join Spilling Proposal [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb closed issue #17267: PROPOSAL Hash Join Spilling Proposal
URL: https://github.com/apache/datafusion/issues/17267
--
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