Flaky fuzz tests for filtered outer SortMergeJoin

[korowa]

Describe the bug

After replacing the filter for join fuzz tests with the selective one (that doesn't return 100% of input rows), it turned out that following tests may periodically fail for HashJoin vs SortMergeJoin cases:

• test_left_join_1k_filtered
• test_right_join_1k_filtered
• test_full_join_1k_filtered

To Reproduce

To reproduce this one should add JoinTestType::HjSmj to mentioned above test cases and run them (likely a couple of times)

Expected behavior

SMJ and HJ should return equal resultsets for these test cases

Additional context

No response

[comphead]

left_semi_join was fixed sometime ago. Need to check whats happening with

test_left_join_1k_filtered
test_right_join_1k_filtered
test_full_join_1k_filtered

The repro case is to run the same test 1000 times and depending on data distribution between batches the problem can arise

[comphead]

fyi @viirya

[comphead]

I believe all of them flaky because of flaky LeftAnti join, all Left, Right, Full depends on LeftAnti starting from #10892 PR.
Checking if thats the case

[comphead]

Simple test case for LeftOuter

#[tokio::test]
async fn test_left_outer() {
    let left: Vec<RecordBatch> = make_staggered_batches(1);

    let left = vec![
        RecordBatch::try_new(
            left[0].schema().clone(),
            vec![
                Arc::new(Int32Array::from(vec![1])),
                Arc::new(Int32Array::from(vec![2])),
                Arc::new(Int32Array::from(vec![10])),
                Arc::new(Int32Array::from(vec![20])),
            ],
        ).unwrap()
    ];

    let right = vec![
        RecordBatch::try_new(
            left[0].schema().clone(),
            vec![
                Arc::new(Int32Array::from(vec![1, 1])),
                Arc::new(Int32Array::from(vec![2, 2])),
                Arc::new(Int32Array::from(vec![0, 30])),
                Arc::new(Int32Array::from(vec![0, 40])),
            ],
        ).unwrap()
    ];

    JoinFuzzTestCase::new(
        left,
        right,
        JoinType::Left,
        Some(Box::new(col_lt_col_filter)),
    )
        .run_test(&[JoinTestType::HjSmj], false)
        .await;
}

@viirya cc

[comphead]

For LeftOuter/RightOuter and FullJoin the rootcause is the same. SMJ emits a null row if the filtered match is not found.
But if the filtered match row is found it should emit the joined row instead of null row. And here we come to the same issue as for LeftAnti, so we need to know when all the right rows processed for the given left row.

Given that SMJ emits rows in freeze_streamed which can be called anytime once the output size is equal to batch size there is 2 possible options:

• Try to calculate if the SMJ is about to switch to ordering == Less which moves left pointer down, meaning everything is processed for the row
• Move emit to some other place, for example we can fill the buffer with rows and filtered flags until the ordering switch, and only then make a final emit based on the data in the buffer + filtered