Move SMJ join filtered part out of join_output stage. LeftOuter, Left…

…Semi (apache#12764) * WIP: move filtered join out of join_output stage * WIP: move filtered join out of join_output stage * WIP: move filtered join out of join_output stage * cleanup * cleanup * Move Left/LeftAnti filtered SMJ join out of join partial stage * Move Left/LeftAnti filtered SMJ join out of join partial stage * Address comments
jonathanc-n · Oct 18, 2024 · 3405234 · 3405234
1 parent 42f9060
commit 3405234
Show file tree

Hide file tree

Showing 3 changed files with 1,061 additions and 524 deletions.
diff --git a/datafusion/core/tests/fuzz_cases/join_fuzz.rs b/datafusion/core/tests/fuzz_cases/join_fuzz.rs
@@ -125,16 +125,14 @@ async fn test_left_join_1k() {
 }
 
 #[tokio::test]
-// flaky for HjSmj case
-// https://github.com/apache/datafusion/issues/12359
 async fn test_left_join_1k_filtered() {
     JoinFuzzTestCase::new(
         make_staggered_batches(1000),
         make_staggered_batches(1000),
         JoinType::Left,
         Some(Box::new(col_lt_col_filter)),
     )
-    .run_test(&[JoinTestType::NljHj], false)
+    .run_test(&[JoinTestType::HjSmj, JoinTestType::NljHj], false)
     .await
 }
 
@@ -229,6 +227,7 @@ async fn test_anti_join_1k() {
 #[tokio::test]
 // flaky for HjSmj case, giving 1 rows difference sometimes
 // https://github.com/apache/datafusion/issues/11555
+#[ignore]
 async fn test_anti_join_1k_filtered() {
     JoinFuzzTestCase::new(
         make_staggered_batches(1000),
@@ -515,14 +514,11 @@ impl JoinFuzzTestCase {
                     "input2",
                 );
 
-                if join_tests.contains(&JoinTestType::NljHj)
-                    && join_tests.contains(&JoinTestType::NljHj)
-                    && nlj_rows != hj_rows
-                {
+                if join_tests.contains(&JoinTestType::NljHj) && nlj_rows != hj_rows {
                     println!("=============== HashJoinExec ==================");
                     hj_formatted_sorted.iter().for_each(|s| println!("{}", s));
                     println!("=============== NestedLoopJoinExec ==================");
-                    smj_formatted_sorted.iter().for_each(|s| println!("{}", s));
+                    nlj_formatted_sorted.iter().for_each(|s| println!("{}", s));
 
                     Self::save_partitioned_batches_as_parquet(
                         &nlj_collected,