feat: Support u32 indices for HashJoinExec
#16434
Open
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
|
cc @Dandandan |
|
🤖 |
|
🤖: Benchmark completed Details
|
|
Those benchmarks make sense, just saves memory. |
There was a problem hiding this comment.
Pull Request Overview
Adds support for using 32-bit row indices in the hash join executor when the build side has fewer than u32::MAX rows to save memory.
- Introduce
JoinHashMapTypetrait withJoinHashMapU32andJoinHashMapU64implementations - Change all callers and constructors of the old
JoinHashMapto useBox<dyn JoinHashMapType> - Update memory estimation to pick
u32oru64variant based on row count and add corresponding tests
Reviewed Changes
Copilot reviewed 5 out of 5 changed files in this pull request and generated 2 comments.
Show a summary per file
| File | Description |
|---|---|
| joins/utils.rs | Remove obsolete JoinHashMap export, keep only JoinHashMapType |
| joins/symmetric_hash_join.rs | Update get_matched_indices calls to accept boxed iterator |
| joins/stream_join_utils.rs | Implement JoinHashMapType for PruningJoinHashMap using new helpers |
| joins/join_hash_map.rs | Define JoinHashMapType, JoinHashMapU32, JoinHashMapU64, and generic helpers |
| joins/hash_join.rs | Construct boxed hash map variant, update memory estimation, and adjust tests |
Comments suppressed due to low confidence (3)
datafusion/physical-plan/src/joins/join_hash_map.rs:38
- [nitpick] Clarify this comment to indicate that the chained list uses either
Vec<u32>orVec<u64>rather than both. E.g., “stored as eitherVec<u32>orVec<u64>based on size requirements.”
/// The indices (values) are stored in a separate chained list stored as `Vec<u32>` `Vec<u64>`.
datafusion/physical-plan/src/joins/hash_join.rs:3581
- [nitpick] The test name now differs from the u32 variant (which uses
collisionsplural). For consistency, consider renaming both tests to match a common pattern, e.g.,join_with_hash_collision_u32andjoin_with_hash_collision_u64.
fn join_with_hash_collisions_u64() -> Result<()> {
datafusion/physical-plan/src/joins/join_hash_map.rs:93
- [nitpick] Add a doc comment to
JoinHashMapTypeexplaining its purpose and when each method should be used; this will help maintainers understand the runtime‐selected index strategy.
pub trait JoinHashMapType: Send + Sync {
Comment on lines
+73
to
+75
| // Flatten the deque | ||
| let next: Vec<u64> = self.next.iter().copied().collect(); | ||
| get_matched_indices::<u64>(&self.map, &next, iter, deleted_offset) |
There was a problem hiding this comment.
[nitpick] Collecting the VecDeque into a new Vec<u64> on every call can be expensive. Consider reusing a buffer or accessing the deque’s contiguous slice when possible to reduce allocations.
Suggested change
| // Flatten the deque | |
| let next: Vec<u64> = self.next.iter().copied().collect(); | |
| get_matched_indices::<u64>(&self.map, &next, iter, deleted_offset) | |
| // Access the deque's slices directly | |
| let (head, tail) = self.next.as_slices(); | |
| get_matched_indices::<u64>(&self.map, head, tail, iter, deleted_offset) |
Copilot uses AI. Check for mistakes.
There was a problem hiding this comment.
This seems to be alittle bit difficult to do without changing function parameters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Which issue does this PR close?
Rationale for this change
We can use
u32indices instead ofu64indices when there are less thanu32::MAXrows when building the hashmap. This acts as a memory optimizationWhat changes are included in this PR?
During
HashJoinExecwe construct theJoinLeftDatawith aBox<dyn JoinHashMapType>choosing between a u32 indice or u64JoinHashMap.I changed the
JoinHashMapTypeto hold theupdate_from_iter,get_matched_indice, andget_matched_indices_with_limit_offsetand split theJoinHashMapintoJoinHashMapU32andJoinHashMapU64.I deliberately did not try to expose a generic in the trait, nor did I try to create a generic on the
JoinHashMapstruct as doing so would lead to many problems with having to call preceding functions with a generic; doing so is not possible since we are determining theJoinHashMapTypeduring runtime.Are these changes tested?
Yes I added a test for checking the hashmap created using u32 indices.