jayzhan211 commented on issue #24768: URL: https://github.com/apache/datafusion/issues/24768#issuecomment-5552303741
I did some research on the SMJ fallback, and here is my conclusion. **Summary** - The simplest way to get spilling is a plan-time threshold: if the build side is estimated too big, plan a SortMergeJoin instead of a hash join. One small PR, no operator changes. - A runtime fallback (hash join starts, memory runs out, switch to SMJ) is also possible, for every partition mode and every join type that SMJ can represent. - Two cases cannot use SMJ: null-aware anti join, and joins that promise probe-side ordering. Reasons below. - Whatever we choose, five foundation PRs are needed first. They are also the groundwork of the hybrid design, so they are never wasted. ## Step 1: a plan-time threshold New config: `hash_join_max_build_size` (bytes, plus a `_rows` version like the CollectLeft thresholds; default unlimited). Rule in `JoinSelection`: if the build side's estimated size is above the threshold, plan a `SortMergeJoinExec` instead of a hash join. This is `prefer_hash_join = false` applied only to big joins. Today `prefer_hash_join = false` is the only escape hatch, and it slows down every join in the session by about 3.5×. With the threshold, only joins that are too big to build use sort-merge, and they spill through SMJ's existing code. Where it applies: | initial mode | with the threshold | |---|---| | Auto, build estimate ≥ threshold | SortMergeJoin (the enforcers add repartition and sort, same as `prefer_hash_join = false`) | | Auto, estimate below threshold, or unknown | unchanged: hash join. Unknown statistics stay on today's path | | CollectLeft because `target_partitions = 1` | SortMergeJoin on one partition. Fine: there is nothing to parallelise anyway | | CollectLeft because `repartition_joins = false` | no. SMJ needs both sides partitioned on the keys, and this setting forbids that repartition | | CollectLeft because null-aware anti | no. SMJ cannot represent a null-aware join (see below) | Limits: it needs statistics (no statistics, no effect); the estimate can be wrong in both directions; the user must choose the threshold, because the planner does not know the memory pool size. Why it is the right first step: it makes failing joins complete for the common Parquet case, at the lowest cost, and nothing in it is thrown away later. ## Step 2 (optional): a runtime fallback For the cases the threshold cannot catch (statistics missing or wrong), the hash join can start, run out of memory, and switch to sort-merge. This is possible for every partition mode and every join type SMJ can represent: | case | possible | difficulty | what it takes | |---|---|---|---| | Partitioned, any N, any join type | yes | easy | per partition: sort the batches already collected plus the rest of the left, sort the right, run `SortMergeJoinStream`. Both sides are already hash-partitioned on the keys, so nothing crosses partitions | | CollectLeft, N = 1 | yes | easy | same as above, one lane | | CollectLeft, N > 1, Right family (Inner / Right / RightSemi / RightAnti / RightMark) | yes | medium | sort the left once into a spill file. Each stream sorts its own right partition and merges it against that file (read back with `read_spill_as_stream`, at most N times). Each stream decides its own right rows and writes to its own `out_j`, so output partitioning is kept. The merge code is reused as is | | CollectLeft, N > 1, Left family (Left / Full / LeftSemi / LeftAnti / LeftMark) | yes | not trivial | same as above, but the merge must *mark* matched left rows in the existing `visited_indices_bitmap` instead of emitting them, and the existing `report_probe_completed()` lets the last stream emit. This is CollectLeft's own rule today, applied to a sorted left. SMJ has no "mark, don't emit" path, so this variant is new code | N = number of probe streams = `target_partitions`. ## What is not possible, and why **1. Null-aware anti join** (`x NOT IN (subquery)`). SMJ cannot represent it. `SortMergeJoinExec::try_new` has no `null_aware` parameter (`HashJoinExec::try_new` has one), and there is no `null_aware` anywhere under `sort_merge_join/`, in source or tests. This is not the `NullEquality` option that SMJ does have. `NullEquality` decides whether `NULL = NULL` matches for one key — a per-row question. Null-aware is the three-valued `NOT IN`: if there is a NULL *anywhere* on the right side, the whole result is empty, no matter which left rows matched. That is a property of the whole right side, and no key comparison can express it. This was a real bug, not theory: before #22810, `prefer_hash_join = false` with `target_partitions > 1` sent a null-aware anti join to SMJ, and it returned rows that should have been filtered out. The fix is the `!null_aware` guard on both partitioned branches in `physical_planner.rs`, with a regression test that keeps the plan on `HashJoinExec: mode=CollectLeft, join_type=LeftAnti, …, null_aware` even when `prefer_hash_join = false`. So an SMJ fallback here would mean writing the null-aware semantics into SMJ from scratch, where a mistake is silent wrong answers. And this is the case that needs spilling most: null-aware is pinned to CollectLeft with the size thresholds bypassed, so its build side has no size limit at all. It is fixed by the hybrid design's phase 2 (CollectLeft spilling keeps the hash join semantics), not by any SMJ path. **2. A join that promises probe-side ordering** — Inner / Right / RightSemi / RightAnti / RightMark with an ordered probe child. `maintains_input_order` is `[false, true]` for these join types, so the join declares the probe child's ordering as its own output ordering. The planner uses that promise at plan time: `EnforceSorting` removes the `SortExec` above the join, `SortPreservingMerge` is chosen instead of `CoalescePartitions`, TopK relies on it. An SMJ fallback sorts the probe side on the join keys, so its output comes out in key order. But `properties()` is computed once and cannot change at runtime, so the plan still says "probe order", and nothing downstream sorts again. The result is silently wrong ordering, not an error. The guard is one line and changes no plan: refuse the fallback (fail exactly as today) when `maintains_input_order()[1]` is true and the probe child has an ordering. Every join without an ordering promise — the whole Left family, and the Right family over an unordered probe — can fall back freely. **Not blockers.** Key types: `make_comparator` can order every type `create_hashes` can hash, including `Map` and `Union`. Equality: hash join and SMJ verify matching pairs with the same `JoinKeyComparator` (with `normalize_float_zero`), and `-0.0` is covered by tests on both paths, so a fallback never changes which rows match. Memory pressure during the probe phase is not a case: the hash join allocates nothing significant there, so there is no trigger. ## The PRs Each PR has the reason it exists, so the recommendation follows from the reason. **Recommended, in this order** | PR | what | why | |---|---|---| | P0 | `hash_join_max_build_size` → SMJ in `JoinSelection` | the smallest change that makes failing joins complete; no operator change; nothing thrown away later | | F1 | account for the `concat_batches` copy | `collect_left_input` reserves each build batch, then concatenates them into one batch for the hash table. That second copy is not reserved, and the originals are still alive, so the process briefly holds about 2× what the pool sees. The OS can kill the process before the pool refuses anything — so no runtime reaction can be trusted until this is fixed. Fix: `try_grow` before concatenating, `shrink` after the originals are dropped | | F2 | spill-ready build consumer: `with_can_spill(true)`; both refusal points (batch append, table build) become a reaction instead of an error; headroom reserved at stream construction | without `can_spill`, the fair pool cannot give the join a share. Without headroom reserved up front, the reaction fails at the moment it is needed (the sort needs `sort_spill_reservation_bytes` first; the grouped aggregate already hits this failure). This is the hybrid's T3 | | F3 | budgeted test harness + memory-limited join benchmark | records today's failure matrix so every later PR is measured against it. This is the hybrid's T0 | | F4 | dynamic filter coupling: the fallen-back partition still reports to `SharedBuildAccumulator`; `Map` filters are retracted or downgraded | a completed build publishes its hash map as an `Arc<Map>` inside `HashTableLookupExpr`, held by the probe scan. The table stays in memory after any fallback, so without this the fallback frees nothing. Without the report, the combined filter never completes. The hybrid has the same coupling | | F5 | ordering guard (case 2 above) | prevents the one silently-wrong case; changes no plan; one line. The same guard lets the hybrid's phase 1 keep the ordering declaration | **Only if a runtime reaction is wanted before the hybrid** | PR | what | why | |---|---|---| | S1 | Partitioned fallback: sort left + sort right + `SortMergeJoinStream`; projection after; spill cleanup; a `fallback_count` metric | covers what P0 cannot: statistics missing or wrong. Deleted when the hybrid lands | **Nice to have — I would not schedule them** | PR | what | why | |---|---|---| | S2 | CollectLeft, Right family | the multi-stream case only appears with `repartition_joins = false` or a wrong CollectLeft estimate — a small population | | S3 | CollectLeft, Left family (mark-mode merge) | same small population, more code | **Not recommended** | PR | what | why | |---|---|---| | ~~S4~~ | null-aware anti in SMJ | not a fallback: SMJ cannot represent the join, and the semantics would be written from scratch where a mistake is silent wrong answers (#22810 was exactly that). This case waits for the hybrid's phase 2 | ## What matters most - **Critical:** the dynamic filter keeping the hash table alive (F4); the unreserved `concat_batches` copy (F1); probe-side ordering (F5); headroom at the moment of refusal (F2). Each one either breaks correctness or breaks the purpose of the fallback. - **Critical and not fixable on this path:** null-aware anti. It needs spilling most and SMJ cannot do it. Only the hybrid's phase 2 covers it. - **Somewhat:** the cost when the fallback fires (two full sorts plus the merge, about the SMJ tax; up to N reads of the sorted left in CollectLeft) — paid only by joins that fail today; and S1–S3 being deleted when the hybrid lands. - **Not critical:** hash-vs-sort equality (already the same comparator); key type coverage (already complete); output partitioning (kept by construction). -- 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]
