HashJoin with mode PartitionMode:CollectLeft has bug and can produce wrong result

[mingmwang]

Describe the bug
A clear and concise description of what the bug is.
The join result is wrong.

To Reproduce
Steps to reproduce the behavior:

#[tokio::test]
async fn equijoin_left_and_condition_from_right() -> Result<()> {
    let ctx = create_join_context("t1_id", "t2_id")?;
    let sql =
        "SELECT t1_id, t1_name, t2_name FROM t1 LEFT JOIN t2 ON t1_id = t2_id AND t2_name >= 'y' ORDER BY t1_id";
    let res = ctx.create_logical_plan(sql);
    assert!(res.is_ok());
    let actual = execute_to_batches(&ctx, sql).await;
    let expected = vec![
        "+-------+---------+---------+",
        "| t1_id | t1_name | t2_name |",
        "+-------+---------+---------+",
        "| 11    | a       | z       |",
        "| 22    | b       | y       |",
        "| 33    | c       |         |",
        "| 44    | d       |         |",
        "+-------+---------+---------+",
    ];
    assert_batches_eq!(expected, &actual);

    Ok(())
}

Run this test with PartitionMode:CollectLeft, the test will be failed.

Expected behavior
A clear and concise description of what you expected to happen.

Additional context
Add any other context about the problem here.

[mingmwang]

@alamb @tustvold

[mingmwang]

expected:

[
"+-------+---------+---------+",
"| t1_id | t1_name | t2_name |",
"+-------+---------+---------+",
"| 11 | a | z |",
"| 22 | b | y |",
"| 33 | c | |",
"| 44 | d | |",
"+-------+---------+---------+",
]
actual:

[
"+-------+---------+---------+",
"| t1_id | t1_name | t2_name |",
"+-------+---------+---------+",
"| 11 | a | |",
"| 11 | a | z |",
"| 11 | a | |",
"| 11 | a | |",
"| 11 | a | |",
"| 11 | a | |",
"| 11 | a | |",
"| 11 | a | |",
"| 22 | b | |",
"| 22 | b | |",
"| 22 | b | |",
"| 22 | b | |",
"| 22 | b | |",
"| 22 | b | |",
"| 22 | b | |",
"| 22 | b | y |",
"| 33 | c | |",
"| 33 | c | |",
"| 33 | c | |",
"| 33 | c | |",
"| 33 | c | |",
"| 33 | c | |",
"| 33 | c | |",
"| 33 | c | |",
"| 44 | d | |",
"| 44 | d | |",
"| 44 | d | |",
"| 44 | d | |",
"| 44 | d | |",
"| 44 | d | |",
"| 44 | d | |",
"| 44 | d | |",
"+-------+---------+---------+",
]

[mingmwang]

The root cause is that, for Left Out join, for each partition of the right side, they are running independently, each of them constructs the HashJoinStream and has the visited_left_side data structure, each of them will try to produce the unmatched right side and fill with null, which will cause duplications.

[mingmwang]

@yahoNanJing @liukun4515

[mingmwang]

For DataFusion, a possible fix is to maintain a global visited_left_side data structure.
And for Ballista, we can not rely on any global structure, because different partitions are different tasks and can run on different machines, we can not run Collect Left mode for LeftJoin/Left Anti/FullJoin.

[mingmwang]

I will fix the bug in the new JoinSelection Rule.