jayzhan211 commented on issue #24768:
URL: https://github.com/apache/datafusion/issues/24768#issuecomment-5468538792
@alamb Thanks — all four questions hit places where the write-up compressed
too much. Long answers below.
*Notation, so this comment stands alone:* each output partition's build data
is split into 16 **buckets** by the join hash. A spilled bucket *i* is a
**pair** of files — `build.Bi` (its build rows) and `probe.Bi` (the probe rows
that might match them). The **share** is one partition's slice of the memory
budget — 75 MB in the running example (300 MB pool, `target_partitions=4`, as
in the symptoms table). **Destage** = move a bucket's buffered rows to disk
mid-build; **restore** = read a spilled file back into memory during cleanup.
---
## 1. CollectLeft: will it still OOM? Yes, in phase 1 — here's why that's
deliberate, and what the follow-up looks like
**Direct answer:** with the flag on, a join the planner places in
`CollectLeft` mode whose build side exceeds the budget fails exactly as it does
today. Phase 1 adds spilling to `Partitioned` mode only, and T8's planner rule
steers large-build joins to `Partitioned` — so the situation mostly doesn't
arise. "Mostly" is spelled out below, along with why CollectLeft waits.
**The two modes in one line each** (full definitions in the epic body): in
`Partitioned` mode, every output partition builds a *private* hash table over
its own disjoint slice of keys — nothing is shared, so each partition can spill
on its own. In `CollectLeft` mode, the build side is collected once into a
*single* hash table that all probe streams share read-only — the cheap plan for
small builds, which is why the planner picks it for them.
**Where the coordination problem actually lives.** My earlier blanket
statement — "CollectLeft is blocked on cross-stream coordination" — was too
coarse. Split it by phase and most of it disappears:
| phase | what spilling adds | cross-stream coordination needed? |
|---|---|---|
| build | one task collects the build side and destages buckets under
pressure | **none** — there is only one builder (`OnceAsync`), so eviction is a
private decision; the T3 buffer drops in unchanged |
| end of build | shared state is now: resident map + per-bucket spill files
+ Bloom filters — complete, and never modified again | **none** — "many streams
share immutable build state" is already CollectLeft's contract today |
| probe | each stream routes its rows, Bloom-tests against spilled buckets,
writes its **own** probe spill files (`probe.Bi.j`, j = stream id) | **none** —
the files are private |
| cleanup | spilled bucket pairs are restored and drained; unmatched build
rows emitted | **yes — the one new problem** |
**The cleanup problem, concretely.** Two decisions:
- *Who restores a spilled bucket?* If each stream restores `build.Bi` for
itself and drains only its own probe file, no coordination is needed — but up
to N copies of one bucket can sit in memory at once (staggering the streams,
stream *j* starting at bucket *j* mod 16, removes most of the overlap). If
instead one cleanup task restores each bucket once and drains every stream's
probe file through it, memory is minimal — but its output rows belong to
different output partitions and must be routed back across streams, which is
exactly the channel `Partitioned` mode never needs.
- *When can unmatched build rows be emitted?*
Left/Full/LeftSemi/LeftAnti/LeftMark emit build rows that no probe row matched.
Today's CollectLeft already coordinates this: every stream marks one shared
bitmap, and the last stream to finish emits. Spilling keeps the same pattern
per bucket, plus one rule — bucket *i*'s unmatched rows emit only after *every*
stream has drained its probe file against the restore.
Neither decision is research-grade — the shared-bitmap pattern is in-tree
today — but each one interacts with the full join-type matrix, which is where
this epic's risk is already concentrated (T5).
**Why defer it, then? Economics, not difficulty:**
1. **Hot path.** CollectLeft serves *small* builds — the most common joins
in a typical workload. Phase 1 keeps every byte of spill machinery out of that
path; `Partitioned` is where large builds already live, so it's where spilling
pays.
2. **Small exposed population.** With T8 steering, an over-budget
CollectLeft build arises only where steering can't reach: `target_partitions =
1` (single-thread and embedded use, where every hash join is CollectLeft),
`repartition_joins = false`, null-aware anti joins (`x NOT IN (subquery)` —
always planned as CollectLeft), and statistics underestimates against the
CollectLeft thresholds (4 MiB / 131 072 rows — estimates, so they can be
wrong). A correction to my own earlier line in this thread: "with the flag on,
the joins that can hit the limit are exactly the joins that can spill" holds on
the default multi-partition path — these four are the exceptions, and they keep
today's failure behavior in phase 1.
3. **Sequencing.** T3–T7 — bucketing, destaging, spill formats, the
join-type matrix, recursion, Blooms — are ~90% of a CollectLeft implementation
too. Landing them first in the coordination-free mode makes the follow-up a
small delta over proven code, reviewable against something real.
4. **Review size.** T4/T5 are already the epic's largest steps; stacking
cleanup-ownership × join-type matrix on top works against the
one-reviewable-PR-per-step goal.
**The follow-up issue, so this isn't a vague promise.** Once T5 lands, I'd
file roughly:
> **CollectLeft spilling.** Reuse the T3 build buffer under CollectLeft's
single build task (no build-phase coordination). Freeze bucket state at build
completion. Per-stream probe spill files. Extend the existing shared-bitmap +
last-stream-emission pattern to per-bucket bitmaps. Decide cleanup ownership —
staggered per-stream restores vs. one cleanup task per bucket with output
routing — by benchmark. Null-aware anti, the join type that always uses
CollectLeft, needs only a single "build side saw a NULL key" boolean
accumulated during build regardless of spilling, so its semantics are
unaffected by the hard parts.
---
## 2. Will an oversized bucket be split too? Yes — at restore time, on the
next 4 bits of the same hash
**Direct answer:** yes. A spilled bucket that turns out too big to process
is split into 16 children, recursively, up to 4 levels deep. The mechanism from
scratch, because it also explains where the scheme stops working (question 4):
Everything rests on one decision: each row's join key is hashed **once**, at
build time, to 64 bits, and that hash is stored with the row in every spill
file (the ~10–15% spill-size overhead in the cost section). Routing never looks
at the key again — at every level it reads *digits* of the stored hash. Read
the hash as 16 hex digits:
```
h("user_42") = 0x5C31 9A40 … → digits 5, C, 3, 1, …
h("user_99") = 0x51B7 2211 … → digits 5, 1, B, 7, …
```
- **Level 0** groups rows by digit 1 → the initial sixteen buckets B0…B15.
Both keys above land in B5.
- **Level 1** — only if B5 proves too big: group *B5's rows alone* by digit
2 → `user_42` goes to child B5.C, `user_99` to B5.1. The two keys that collided
at level 0 are now apart.
- **Levels 2, 3** — digit 3, then digit 4. Same move again.
Like sorting mail by the first letter of the surname, then pigeonholing an
overfull "S" drawer by the second letter.
**Why matches are never lost.** Equal keys have equal hashes, so equal digit
strings — on both sides. A probe row for `user_42` reads 5,C,3,1… exactly like
every build row for `user_42`. Split `build.B5` by digit 2 and split `probe.B5`
by the same digit, and every match stays inside the same child pair — no key
comparison ever happens. A pair stays a pair at every depth.
**Why a split shrinks the bucket.** Rows in B5 agree on digit 1, but
*distinct* keys have effectively random remaining digits, so digit 2 spreads
them ~evenly across 16 children. Each level multiplies capacity by 16: from the
75 MB share, level 0 handles ~1.2 GB of build data per partition, level 1 ~19
GB, level 2 ~300 GB, level 3 ~4.9 TB. That is why the depth cap is 4 — past it,
size is no longer what's wrong (see question 4).
**When splitting happens — not during the build.** Under pressure the build
just destages the largest bucket wholesale, the cheapest possible reaction, and
keeps going. Re-splitting waits for cleanup, when the bucket's true size is
known: restore `build.Bi` and try to build its hash table; if that fails, one
pass over `build.Bi` writes 16 child files by the next digit, one pass over
`probe.Bi` does the same, the parent pair is deleted, and the children are
processed one at a time. Each level costs one extra read+write of just the
offending pair, so the depth cap also caps I/O amplification (≤ ~5× on the data
that overflowed — still linear).
**The built-in limit — and the bridge to question 4:** digits can separate
*different keys*; they can never separate *one key's rows*, which share all 16
digits. A bucket dominated by a single key re-splits into one child ≈ its
parent, at every level, forever. That case needs different tools — see 4.
---
## 3. Ordering: we stop advertising it — no re-sort. And DataFusion is the
outlier for advertising it at all
Background, since this corner of DataFusion is niche: operators *declare* to
the planner which input orderings survive them (`maintains_input_order`), and
the planner uses the declarations to skip `SortExec`s it would otherwise
insert. `HashJoinExec` today declares, one entry per child:
```rust
fn maintains_input_order(join_type: JoinType) -> Vec<bool> {
vec![
false, // left child = build side: never
preserved
matches!(join_type, // right child = probe side: preserved
for…
JoinType::Inner | JoinType::Right | JoinType::RightAnti
| JoinType::RightSemi | JoinType::RightMark),
]
}
```
The build side is always `false` — a hash join regroups build rows by what
they matched. The probe side is `true` for the five listed types because probe
batches are processed one at a time, in order, and each output row stays
anchored to its probe row. It is *already* `false` for
Left/Full/LeftSemi/LeftAnti/LeftMark, because their output includes build rows
— Left and Full, for instance, emit unmatched build rows in a trailing pass
after the probe stream ends.
**Spilling breaks the property in a familiar way.** Cleanup emits the
spilled buckets' results *after* the in-memory pass — a trailing emission
phase. That is precisely the shape that already makes Left and Full `false`
today. With spilling, Inner/Right acquire the same shape, so the property goes
away for the same reason it was never claimed for Left/Full.
**Dropping it cannot change any query's results.** The declaration is an
optimizer contract, not a SQL guarantee. Concretely: `… JOIN … ORDER BY t.ts`
with the probe side already sorted on `ts` — today the planner sees the
declaration and skips the sort; with the declaration gone, it inserts a
`SortExec`. Same rows out either way. A query without an `ORDER BY` was never
promised an order at all. The entire cost is: some plans gain a sort.
**Peer engines: the question barely exists elsewhere.** DuckDB's docs list
`JOIN` among the operations that "do not guarantee that the row order is
preserved" ([order
preservation](https://duckdb.org/docs/current/sql/dialect/order_preservation)).
Spark advertises sorted output only from its sort-merge join. As far as I can
tell, none of Spark, Trino, Velox, or ClickHouse exposes hash-join output
ordering to its optimizer at all — and, tellingly, none of them re-sorts inside
a spilling hash join; an explicit sort serves ordered consumers. We would be
giving up a property none of our peers ever offered, in exchange for the one
none of them lack: hash joins that finish under a memory budget.
**Why not recover the order instead?** Doable — tag each probe row with a
sequence number, carry it through the spill files, merge cleanup output with
resident output — but the tag and the merge would be paid on **every** query
with the flag on, spilled or not. That is the same shape as the 3.5× SMJ tax
this epic exists to delete. Rejected.
**Keeping the damage contained (part of T8).** Plan properties are static —
fixed when the plan is built, before anyone knows whether the query will spill
— so the drop cannot be conditioned on "did we spill." It *can* be conditioned
on what the planner already knows: make the declaration a function of
`(join_type, partition_mode, flag)`, so CollectLeft plans keep the property
even with the flag on and only `Partitioned` plans — the ones that can spill —
lose it. The residual, a `Partitioned` join that never spills still triggering
a downstream sort, is what T8's TPC-H plan-diff audit measures, and it is a
stated gate for default-on in T11. Escape hatch if the audit finds an expensive
regression: keep the property and refuse to spill for the order-preserving join
types. Flag off ⇒ nothing changes anywhere.
---
## 4. One bucket very large while the rest are fine — e.g. one key holding
90% of the rows
**Direct answer:** the hot bucket is simply the first one destaged
(destage-largest picks it), the build then completes without further pressure,
and at cleanup the design *detects* that re-splitting won't help and switches
to fallbacks that don't need the bucket to shrink. Nothing global happens; the
other 15 buckets never notice.
An oversized bucket has exactly three possible causes, and they get
different treatment:
| cause | what's inside | does re-split help? | handled by |
|---|---|---|---|
| **volume** — build ≫ budget, all 16 buckets big | many distinct keys | yes
— ×16 capacity per level | recursion (T6) |
| **hash skew** — unlucky digit-1 distribution | many distinct keys that
collided on one prefix | yes — digit 2 re-randomizes them | recursion (T6) |
| **value skew** — one hot key (the 90% case) | one key's rows — inseparable
by any digit (§2) | **no — provably** | detect, then: role reversal (T7) →
chunked build + replay (T6) |
The first two are what recursion exists for. The third is **detected rather
than predicted**: after a re-split pass, if the largest child is still ≈ the
parent, no further digit can help — stop recursing. Detection costs one wasted
pass, bounded by the depth cap.
The 90%-one-key build, end to end:
- **Build.** All the hot key's rows carry one hash, so they all route to one
bucket — say B5. The first refused allocation destages B5 (destage-largest
picks it by construction — and this case is the argument *for* that victim
policy: any other choice evicts cold buckets while the hot one keeps growing in
memory). From then on, ~90% of incoming rows stream straight into `build.B5`'s
file: pressure stops rising, and the other 15 buckets stay in memory.
- **Probe.** B5's Bloom filter answers "might this row match?" — for the hot
key, truthfully yes — so the hot probe rows all land in `probe.B5`. Bloom
filters only eliminate provably-absent keys; here they are useless by design.
- **Cleanup.** Restore fails → one re-split pass → largest child ≈ parent →
value skew confirmed. Now the fallback ladder, tried in order:
**Rung 1 — role reversal (T7): the cheap exit.** Inside one bucket pair, the
join is just "match these two bags of rows on key," and a hash join needs only
*one* side in memory — nothing says it must be the build side. If `probe.B5` is
40 MB against a 3 GB `build.B5`, build the table from the probe file and stream
the build file past it: one pass over each, normal cost, done. A huge repeated
key on the build side met by few probe rows is the most common real-world skew,
and this ends it outright. (Mechanical for Inner/Semi; join types with sided
semantics constrain it — hence T7's scoping.)
**Rung 2 — chunked build + probe replay (the pattern DF 54's nested-loop
join already uses).** Both sides too big: build a table from the first 75 MB of
`build.B5`, stream *all* of `probe.B5` past it, drop the table, take the next
75 MB chunk, replay the probe file again. `ceil(build_bytes / share)` passes —
a 3 GB bucket ≈ 40 passes over the probe file. Slow, but bounded, correct, and
never out of memory.
Two correctness traps in rung 2 — and the reason T6 builds on T5's join-type
matrix. A *build* row lives in exactly one chunk, so build-side unmatched
tracking works per chunk. A *probe* row, though, may find its match in *any*
chunk — so for the join types that emit unmatched probe rows
(Right/Full/RightAnti/RightSemi), a per-bucket "matched" bitmap must survive
across all chunks, and unmatched rows can only be emitted after the last one.
Get that wrong and outer joins return duplicated or missing rows *only under
skew* — exactly the kind of bug the spill-forcing fuzzer in T5 exists to catch.
**Worth a follow-up issue afterwards:** a bucket that defeated re-split has
one dominant key by definition, and every probe row with that key matches
*every* build row with it — a hash table adds nothing there. Peel the dominant
key off as a streamed cross-product (O(1) build-side memory) and run only the
leftover keys through the normal path: rung 2's ~40 passes become one.
And the floor, stated plainly: if the same key is hot on *both* sides, the
join's own *output* is |build_K| × |probe_K| rows — quadratic in any engine,
spilled or not. For two-sided skew the goal is "completes within the budget";
the case worth optimizing is one-sided skew, and rung 1 targets 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: [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]
