feat: null aware RightAnti hash join execution + planning support - #23957
feat: null aware RightAnti hash join execution + planning support#23957saadtajwar wants to merge 9 commits into
Conversation
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #23957 +/- ##
=======================================
Coverage 80.85% 80.85%
=======================================
Files 1099 1099
Lines 374304 374352 +48
Branches 374304 374352 +48
=======================================
+ Hits 302642 302681 +39
- Misses 53607 53613 +6
- Partials 18055 18058 +3 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
|
cc @andygrove - looking forward to your feedback! I'll have the second PR with the planner support up shortly :) |
| if !matches!( | ||
| (join_type, partition_mode), | ||
| (JoinType::LeftAnti, _) | ||
| | (JoinType::RightAnti, PartitionMode::CollectLeft) // `PartitionMode::CollectLeft` is safe because `RightAnti` is probe-driven |
There was a problem hiding this comment.
Do we need the partition_mode check for CollectLeft here? Or should we just count on it being enforced/never chosen as anything but CollectLeft beforehand? I suppose it's possible to explicitly create a plan with a different partition mode if you bypass the SQL -> plan frontend and create directly? Or do we make the assumption that users will only create logical plans and not physical? 👀
|
cc @andygrove - just to update, this PR should now fully close out #23931 ! Just this one PR now with all changes instead of two :) |
Which issue does this PR close?
Rationale for this change (copied from issue):
DataFusion plans NOT IN (subquery) as a null-aware anti join, but HashJoinExec only supports null_aware = true for LeftAnti with a single join key (validated in datafusion-physical-plan/src/joins/hash_join/exec.rs). Since HashJoinExec always builds on the left input, the build side of a null-aware anti join is the outer table, not the subquery.
This has two costs:
Memory scales with the wrong side. For SELECT ... FROM big_fact WHERE key NOT IN (SELECT k FROM small_dim), the hash table is built over the entire fact table. Memory is O(outer) when it could be O(subquery).
The operator cannot be distributed. The null-aware logic coordinates three pieces of global state across probe partitions through in-process shared memory: probe_side_has_null: AtomicBool, probe_side_non_empty: AtomicBool, and the visited-build-row bitmap, with the last probe partition to finish emitting the unmatched build rows (hash_join/stream.rs). This is correct and cheap in one process, but engines that split probe partitions across processes get independent copies of all three and produce duplicated or incorrect rows. Ballista hit exactly this (apache/datafusion-ballista#2187) and currently has to force the join into a single task (apache/datafusion-ballista#2188), losing all parallelism.
What changes are included in this PR?
This PR adds physical execution support for null-aware
RightAntihash joins using theCollectLeftpartition mode.The build side now records whether it contains a NULL join key, and the join outputs no rows when the build side contains NULL, filters probe rows with NULL keys, and outputs all probe rows when the build side is empty.
For planner support,
JoinSelectionwas updated to swap a null-awareLeftAntijoin to a null-awareRightAntijoin when statistics show that the right side is smaller with the swapped join always using theCollectLeftpartition modeAre these changes tested?
Yes
Are there any user-facing changes?
NOT INqueries may now use a null-awareRightAntiphysical plan when the subquery is smaller. Query results remain unchanged, but these queries can use less memory and keep the outer table partitioned.