Repository navigation
refactor: Rewrite Hash Join's regular (inner/left/right/full) join path - #26227
2010YOUY01 wants to merge 8 commits into
Conversation
|
run benchmark tpch10 hj |
github-actions
Bot
added
documentation
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing refactor-hj (c7d903b) to a23b89f (merge-base) diff Run configurationrun benchmark tpch10Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing refactor-hj (c7d903b) to a23b89f (merge-base) diff Run configurationrun benchmark hjResults will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing refactor-hj (c7d903b) to a23b89f (merge-base) diff Run configurationrun benchmark hjCPU Details (lscpu)Details
Resource Usagehj — base (merge-base)
hj — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing refactor-hj (c7d903b) to a23b89f (merge-base) diff Run configurationrun benchmark tpch10CPU Details (lscpu)Details
Resource Usagetpch10 — base (merge-base)
tpch10 — branch
File an issue against this benchmark runner |
|
Will resolve merge conflict after we agree on |
andygrove
left a comment
andygrove
left a comment
There was a problem hiding this comment.
Disclosure: I used an LLM to produce this review and only briefly reviewed its output myself, so please treat the points below as leads to check rather than verified findings. Apologies in advance for any that are wrong or already addressed.
The review read the whole diff against the legacy HashJoinStream and found no correctness bug in the new path. It traced these and found the results match the old code: empty-map and empty-build handling, chunk/offset bookkeeping, the right/full alignment range, the fetch cutoff, and the dynamic-filter report/cancel lifecycle. cargo test -p datafusion-physical-plan --lib hash_join passes (1022 tests). The sqllogictests and the rest of the workspace were not run. The remaining points are about robustness, coverage and maintainability, most important first.
Robustness and coverage
1. No tests for the new state machine, and the legacy path loses coverage (hash_join_table.rs)
The module documents a batch_size = 1 walkthrough (NeedInput, Output, Finished), but nothing pins it, and the PR adds no tests of its own. Chunk-boundary cases are covered only if the existing exec.rs and sqllogictest suites reach them: a probe row's matches spanning chunks plus a join filter, Right/Full alignment, ArrayMap buffer reclaim, and a fetch cutoff mid-chunk. Because the flag now defaults to true, the existing exec.rs tests no longer run against the legacy path for Inner/Left/Right/Full. That is the advertised enable_migration_hash_join=false workaround, so a regression there would go unnoticed. Suggest running the exec.rs join-type matrix under both flag values, and adding unit tests for HashJoinTable.
2. Unchecked decrement (build.rs:84)
unmatched.remaining_probes -= 1 underflows if finish_probe is called more times than probe_partitions, for example if the same probe partition of a Left/Full CollectLeft join runs twice against a shared regular_left_fut. Debug builds panic. Release builds wrap to usize::MAX, so the count never reaches 0 and the unmatched build rows are silently dropped, giving wrong Left/Full results. Suggest checked_sub with an internal error, or an assert.
3. A fetch cutoff skips finish_probe (hash_join_table.rs:404)
A LimitReached cutoff, or the can_skip_probe fast path, moves to DrainingOutput without calling build.finish_probe(). In a multi-partition CollectLeft or Partitioned Left/Full join with a pushed-down fetch, if one partition hits its limit while siblings run to EOF, no partition is ever "last" and none emits the unmatched build rows. This is fine only while the caller truncates the output anyway, and that contract is written only as a comment on finish_probe. The legacy path behaves the same way, so this is not a regression. Suggest encoding the invariant in the type, or having the cutoff partition still report completion.
4. Prepared builds hard-code visited = None (exec/prepared.rs:84)
regular_probe_data passes JoinBuildState::new(.., None, 0), which makes record_matches and finish_probe no-ops, relying on a comment that prepared builds are INNER-only. If prepared_key_indices is ever relaxed to allow Left or Full, the join silently loses all unmatched build rows with no error. The invariant is enforced in a different function from where it is relied on. Suggest a debug_assert or an explicit join-type check next to this constructor.
5. Cancel is reported even after a successful report (build.rs:207)
Drop for HashJoinBuild calls report_canceled_partition after a successful report too. It is safe today because store_canceled_partition ignores non-Pending partitions, but it takes the accumulator mutex once per partition for nothing, and any change to that no-op property would turn every normal completion into a spurious cancel. Suggest tracking a reported flag, as the legacy BuildReportHandle state enum does.
6. Child execution order changed (exec.rs:1925)
self.right.execute(...) now runs before self.left.execute(...) for the legacy Partitioned path as well; previously left ran first. This is probably harmless, but children whose execute has side effects or fails eagerly (mocks, RepartitionExec channel setup) will now see right-before-left. It is an unannounced behaviour change on a path the PR describes as unchanged.
Maintainability
7. Probe logic duplicated from the legacy stream (hash_join_table.rs:449)
probe_chunk and the unmatched-build emission copy process_probe_batch, count_new_matched_probe_rows and next_final_indices_chunk rather than reusing them. For example, line 521 re-implements the distinct count as 1 + values.windows(2).filter(..).count() instead of calling count_distinct_sorted_indices, which is also a single pass. While both paths are live behind the flag, a fix to one (chunk-boundary metric counting, the Right/Full alignment range) will not reach the other. Suggest making the shared helpers pub(super) and calling them from both.
8. Outputting and Probing repeat the same loop (regular_join.rs:125)
Outputting (lines 125-139) repeats the loop in Probing::handle_stage and turns NeedInput into an internal error. A single loop with a "probe exhausted" flag would remove a stage and an error path. The schema and metrics fields also duplicate config.schema and metrics.baseline already held in Building.
9. Timer and metric changes (hash_join_table.rs:344)
next_output_batch and finish_probe clone the join_time handle on every call and start a timer on every loop iteration, including the cheap next_completed_batch hand-off round trips. join_time now also covers coalescer flushing and unmatched-build emission, so EXPLAIN ANALYZE changes meaning versus the legacy stream. Suggest timing only the work sections, or calling out the change.
10. Placeholders and typos (exec.rs:1914)
// TODO: add issue link here and TODO(open the issue) in regular_join.rs are unresolved, and the PR checklist item "Update comments TODOs to link the Github issue" is unchecked. The config text for enable_migration_hash_join gives no tracking issue or removal timeline, unlike enable_migration_aggregate. There are also typos in build.rs ("syncrhonization", "paartitions", "coordiation"), and a doc comment in regular_join.rs that says "Outputting: Actually joining..." but describes the drain stage.
|
Thank you for the review @andygrove Note this PR depends on #26410 -- would love to hear your judgements also. Regarding AI review: Testing coverage (point 1)The testing coverage is a legit concern, I assume most test coverage already lives in existing operator/sql level tests, comprehensive UTs on newly refactored utilities are redundant. Now we got configuration
We did option 2 for rewriting aggregate, and I'm leaning more towards option 2, then we should do the above-mentioned feature freeze for one release, then delete the old implementation. (happy to do 1 if there are other concerns) Other pointsI'll let AI try to reproduce the potential vulnerabilities with behavior tests, and fix them if there are violations. UPDATE: 7: NA if we will remove the old implementation later |
|
Thank you for opening this pull request! Reviewer note: cargo-semver-checks reported the current version number is not SemVer-compatible with the changes in this pull request (compared against the base branch). Details |
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #26227 +/- ##
========================================
Coverage 82.92% 82.93%
========================================
Files 1154 1157 +3
Lines 457872 458352 +480
Branches 457872 458352 +480
========================================
+ Hits 379712 380135 +423
- Misses 55136 55174 +38
- Partials 23024 23043 +19 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
|
@2010YOUY01 is this ready for a comprehensive review? @andygrove posted a lot of notes, not sure if those are addressed before I dive in and up with possibly redundant feedback. |
|
@mbutrovich Yes it's ready for review. I've responded to all feedback in the above comment, and addressed the applicable ones. |
Sounds good. I’ll take a detailed look tomorrow. |
jayzhan211
left a comment
jayzhan211
left a comment
There was a problem hiding this comment.
@2010YOUY01 On the coverage question: the feature-freeze plan sounds good to me, but once this merges nothing runs with enable_migration_hash_join = false for inner/left/right/full joins, and that's the setting users would fall back to if the new path has a bug. #22729 handled the same problem by randomizing enable_migration_aggregate in the aggregation fuzzer. Doing the same in run_test in join_fuzz.rs is only a few lines, and it's fine as a follow-up:
for (i, batch_size) in self.batch_sizes.iter().enumerate() {
let mut session_config = SessionConfig::new().with_batch_size(*batch_size);
// Keep the legacy path covered until it is removed (#26361)
session_config
.options_mut()
.execution
.enable_migration_hash_join = i % 2 == 0;| /// Current execution phase. | ||
| phase: Phase, | ||
| /// Where to resume candidate lookup; `None` means this batch is exhausted. | ||
| next_offset: Option<ProbeOffset>, |
There was a problem hiding this comment.
nit: next_offset, joined_probe_idx, matched_probe_idx and key_comparator all describe progress through the current probe batch, and they're reset together by hand in push_batch, probe_chunk and release_probe. Moving them into ProbeBatch makes dropping the batch reset them, so those three blocks go away.
struct ProbeBatch {
input: RecordBatch,
keys: Vec<ArrayRef>,
valid_keys: Option<NullBuffer>,
+ next_offset: Option<ProbeOffset>,
+ joined_probe_idx: Option<usize>,
+ matched_probe_idx: Option<u32>,
+ key_comparator: Option<JoinKeyComparator>,
}- let Some(probe) = &self.probe else {
+ let Some(probe) = &mut self.probe else {
return Ok(None);
};
- let Some(offset) = self.next_offset.take() else {
+ let Some(offset) = probe.next_offset.take() else {
self.probe = None;
- self.key_comparator = None;
- self.joined_probe_idx = None;
- self.matched_probe_idx = None;
return Ok(None);
};The rest is renaming self.<field> to probe.<field> in probe_chunk, and deleting the same four resets from push_batch and release_probe.
| ) -> Self { | ||
| // An empty map may still have build rows with NULL keys. Left/full | ||
| // joins must retain those rows for final null extension. | ||
| let can_skip_probe = (build.data.batch.num_rows() == 0 |
There was a problem hiding this comment.
nit: these two conditions repeat what JoinType::empty_build_side_produces_empty_result and JoinType::empty_map_produces_empty_result already encode, and the legacy stream uses those helpers. Using them here keeps a single source of truth when existence joins move to this path:
let join_type = context.kind.join_type();
let can_skip_probe = (build.data.batch.num_rows() == 0
&& join_type.empty_build_side_produces_empty_result())
|| (build.data.map.is_empty() && join_type.empty_map_produces_empty_result());There was a problem hiding this comment.
addressed in baf4878, probably it's possible to further clean them up, it seems more complex than it have to be
|
|
||
| /// Collecting the build side data of hash join. | ||
| /// | ||
| /// It handles coordination/syncrhonization among all parallel paartitions for: |
There was a problem hiding this comment.
nit: a few typos are still here: "syncrhonization" and "paartitions" on this line, and "coordiation" on line 99.
-/// It handles coordination/syncrhonization among all parallel paartitions for:
+/// It handles coordination/synchronization among all parallel partitions for:- // Shared structures for coordiation among partitions.
+ // Shared structures for coordination among partitions.| /// see [`HashJoinBuild`] for different physical strategies | ||
| /// - Probing: Joining probe batch with build data incrementally, until probe | ||
| /// side input is exhausted. | ||
| /// - Outputting: Actually joining the probe batch and buffered build side, and |
There was a problem hiding this comment.
nit: by the time the stream reaches Outputting, probing is done. This stage only drains buffered output and emits the unmatched build rows for left/full joins, so the description reads like the Probing stage.
- /// - Outputting: Actually joining the probe batch and buffered build side, and
- /// produce final output.
+ /// - Outputting: Drain buffered output, then emit unmatched build rows for
+ /// left/full joins.| /// See <https://github.com/apache/datafusion/issues/22710> for details. | ||
| pub enable_migration_aggregate: bool, default = true | ||
|
|
||
| /// Whether inner, left outer, right outer, and full outer hash joins use |
There was a problem hiding this comment.
nit: enable_migration_aggregate just above says the fallback is only a workaround, says when it will be removed, and links its tracking issue. Matching that here tells users not to depend on the option. configs.md and information_schema.slt would need regenerating after this change.
/// Whether inner, left outer, right outer, and full outer hash joins use
/// the refactored implementation. When set to `false`, these joins use
/// the existing implementation. Other join types always use the existing
/// implementation.
+ ///
+ /// The fallback exists only as a workaround for bugs in the new
+ /// implementation and will be removed, together with this option, once
+ /// the migration is complete.
+ ///
+ /// See <https://github.com/apache/datafusion/issues/26361> for details.
pub enable_migration_hash_join: bool, default = true
|
@jayzhan211 Thank you for the review
Addressed in 4b20989, and also added several I agree we can keep them. It's only 20 LoC in total, and easy to add/revert. As explained above in the 'feature-freeze' approach, leaving the old impl untested is probably okay, since it should remain stable if we don't change it. However, I think changes in other components might propagate to the old HJ impl and cause some bugs to surface. I think this approach cheaply preserves most of the test coverage for extra assurance. Some UTs won't run on the old impl, but that should be acceptable given the 'feature-freeze' assumption. |
Which issue does this PR close?
Part of
HashJoinExecby splitting regular/existence join into dedicated streams #26361Rationale for this change
See #26225 for the rationale (terminology is explained in more detail there).
This PR rewrites Hash Join for maintainability. The two major design changes from the existing implementation are:
Point 2 justifies a holistic rewrite.
This PR incrementally rewrites regular joins, covering the existing functionality while keeping the legacy implementation unchanged. After a similar rewrite for existence joins, the legacy
HashJoinStreamcan be removed.Simplified pseudocode:
What changes are included in this PR?
enable_migration_hash_join, if set totrue, and join node is regular join, then use the new path; otherwise use the existing implementation as fallback (old implementation not changed). Defaulted totrue, because it has covered full functionality of existing implementation.To read the implementation, I recommend
datafusion/physical-plan/src/joins/hash_join/regular_join.rsstruct HashJoinTableindatafusion/physical-plan/src/joins/hash_join/hash_join_table.rs-- the struct's comment explains it at high-level.What is the testing strategy for this PR?
Existing tests
TODO for myself
Are there any user-facing changes?
No