feat: Add indexed hash join optimization #219
Conversation
This commit introduces an optimization to the hash join operator to avoid unnecessary data copying during repartitioning. The key changes are: - A new `Distribution::IndexedHashPartitioned` variant is introduced to signal that the output of a `RepartitionExec` is a stream of indexed batches. - A new `IndexedBatch` struct is introduced, which contains a shared reference to a `RecordBatch` and a set of indices. - `RepartitionExec` is modified to produce `IndexedBatch`es when the required output distribution is `IndexedHashPartitioned`. - `HashJoinExec` is modified to consume `IndexedBatch`es on the build side, building the hash table from the indexed data directly, thus avoiding the expensive `take` and `concat_batches` operations. This optimization significantly improves the performance of partitioned hash joins by reducing memory allocations and data copying.
|
👋 Jules, reporting for duty! I'm here to lend a hand with this pull request. When you start a review, I'll add a 👀 emoji to each comment to let you know I've read it. I'll focus on feedback directed at me and will do my best to stay out of conversations between you and other bots or reviewers to keep the noise down. I'll push a commit with your requested changes shortly after. Please note there might be a delay between these steps, but rest assured I'm on the job! For more direct control, you can switch me to Reactive Mode. When this mode is on, I will only act on comments where you specifically mention me with New to Jules? Learn more at jules.google/docs. For security, I will only act on instructions from the user who triggered this task. |
This commit introduces an optimization to the hash join operator to avoid unnecessary data copying during repartitioning.
The key changes are:
Distribution::IndexedHashPartitionedvariant is introduced to signal that the output of aRepartitionExecis a stream of indexed batches.IndexedBatchstruct is introduced, which contains a shared reference to aRecordBatchand a set of indices.RepartitionExecis modified to produceIndexedBatches when the required output distribution isIndexedHashPartitioned.HashJoinExecis modified to consumeIndexedBatches on the build side, building the hash table from the indexed data directly, thus avoiding the expensivetakeandconcat_batchesoperations.This optimization significantly improves the performance of partitioned hash joins by reducing memory allocations and data copying.
PR created automatically by Jules for task 8109793290170804910 started by @Dandandan