Support zero copy hash repartitioning for Hash Join

[Dandandan]

Is your feature request related to a problem or challenge?

Currently RepartitionExec: partitioning=Hash will be added whenever a hash
The benefit is increased parallelism, but at the cost of copying the entire table (in a not-so efficient way).

We should consider lowering the cost of repartitioning by not having to copy the input.

Describe the solution you'd like

Instead of repartitioning the right side and left side input in RepartitionExec, support repartitioning the inputs based on a selection vector.

Instead of taking the RecordBatch, we can consider doing the following:

• Add a (boolean) selection vector as output column for each output partition. I.e. true means the row is selected for the partition.
• The rest of the RecordBatch remains unchanged (i.e. no copy).
• CoalesceBatchesExec is no longer needed for the output (reducing another copy)
• In the hash join algorith,, only try matching input indices for the selection vector for each partition. This seems not that hard as get_matched_indices already supports getting an input of input indices + hashes.

Dependencies

• Add selection vector repartitioning #15420

Describe alternatives you've considered

The partitioning could be done inside the hash join algorithm, however this would add more complexity to each operator (join / aggregates) to do hash-repartitioning inside

Additional context

No response

[zebsme]

take

[alamb]

@zebsme this is likely a fairly tricky thing to implement, just FYI

[zebsme]

If anyone's interested in this, feel free to try :)

[Dandandan]

Yes, I agree with @alamb I don't consider this super easy. However, feel free to try and implement some steps and let us know when you need help @zebsme .

Probably it makes sense to split this issue in some subtasks, these are the ones I can think of and I think the individually they might be doable for an "advanced beginner":

• Add a mode that outputs selection vectors (for now let's use dense boolean arrays so it can be added to RecordBatch) in RepartitionExec. The array outputs true for each row that has hash % partition == 0 (and false if not).
• Support using this selection vector in hash join (only matching indices in the selection vector)
• Support planning with the repartition mode (probably behind an option), run some benchmarks to decide on the default.

We probably want to think about some general way of representing / using selection vectors in RecordBatch (or as wrapped RecordBatch) as well but this requires some more design / discussion.

[goldmedal]

• Add a mode that outputs selection vectors (for now let's use dense boolean arrays so it can be added to RecordBatch) in RepartitionExec. The array outputs true for each row that has hash % partition == 0 (and false if not).

I'll work on #15383, but I guess there are some duplicate works here.

[zebsme]

• Add a mode that outputs selection vectors (for now let's use dense boolean arrays so it can be added to RecordBatch) in RepartitionExec. The array outputs true for each row that has hash % partition == 0 (and false if not).

I'll work on #15383, but I guess there are some duplicate works here.

That makes sense. Perhaps we should address #15420 first - I'd be happy to help review the code.