Bug in HashJoin output_partition cause the input from left input not fully executed

[duongcongtoai]

Describe the bug

The HashJoinExec decides output_partition based on this function: https://github.com/apache/arrow-datafusion/blob/b7a33317c2abf265f4ab6b3fe636f87c4d01334c/datafusion/core/src/physical_plan/joins/utils.rs#L90

If PartitionMode is set to Partitioned, join_type is RIGHT, output_partition will depend on output_partition of the right child, this may cause missing execution on left child partitions, if left child has more partitions than right child partition: https://github.com/apache/arrow-datafusion/blob/e87754cfe3afa4c358a8ca9c21c3c4acd020dfe5/datafusion/core/src/physical_plan/joins/hash_join.rs#L413

To Reproduce

Code in this gist

Create 2 ExecutionPlan input from csv with only 1 field "id" and create a HashJoinExec from these inputs. Because during the execution, some parition from the left input is not executed on, they are never probed with associated rows in the right input, so result in a false join:

+----+----+
| id | id |
+----+----+
|    | 2  |
|    | 3  |
|    | 6  |
|    | 7  |
|    | 9  |
|    | 1  |
|    | 4  |
|    | 5  |
|    | 8  |
+----+----+

Expected behavior

HashJoin executes correctly

+----+----+
| id | id |
+----+----+
| 1  | 1  |
| 9  | 9  |
| 5  | 5  |
| 8  | 8  |
| 6  | 6  |
| 7  | 7  |
| 4  | 4  |
| 2  | 2  |
| 3  | 3  |
+----+----+

Additional context

No response

[mingmwang]

I will take a look.

[mingmwang]

@duongcongtoai
Regarding more partitions, do you mean the input partition count is different among left input and right input ?
Currently in DataFusion, we will insert a RepartitionExec for left input and right input, after the RepartitionExec
the partition count should be always the same.

[duongcongtoai]

Thank you. Shoud we have a small validation in new function to notify users about this constraint?

[mingmwang]

@duongcongtoai Agree. We should add a validation to enforce this constraint.

[duongcongtoai]

Okay, let me open my first PR :D

[mingmwang]

Okay, let me open my first PR :D

Sure, I think you can add the input partition count check during the real execution time(not plan time) in the execute()method.
And I think both the HashJoinExec and SortMergeJoinExec will need this check. I'm not sure for SymmetricHashJoinExec whether the check is required or not.

[duongcongtoai]

@mingmwang do we have special reason to validate at execution time instead of plan time? If the constraint is violated, we can avoid starting uncessary execution in partitioned tasks right?