Implement RightSemi join for SortMergeJoin

datafusion/core/tests/fuzz_cases/join_fuzz.rs

@@ -20,6 +20,7 @@ use arrow::compute::SortOptions;
 use arrow::record_batch::RecordBatch;
 use arrow::util::pretty::pretty_format_batches;
 use arrow_schema::Schema;
+use log::debug;
 use std::sync::Arc;
 use std::time::SystemTime;
 
@@ -186,7 +187,7 @@ async fn test_full_join_1k_filtered() {
 }
 
 #[tokio::test]
-async fn test_semi_join_1k() {
+async fn test_left_semi_join_1k() {
     JoinFuzzTestCase::new(
         make_staggered_batches(1000),
         make_staggered_batches(1000),
@@ -198,7 +199,7 @@ async fn test_semi_join_1k() {
 }
 
 #[tokio::test]
-async fn test_semi_join_1k_filtered() {
+async fn test_left_semi_join_1k_filtered() {
     JoinFuzzTestCase::new(
         make_staggered_batches(1000),
         make_staggered_batches(1000),
@@ -209,6 +210,30 @@ async fn test_semi_join_1k_filtered() {
     .await
 }
 
+#[tokio::test]
+async fn test_right_semi_join_1k() {
+    JoinFuzzTestCase::new(
+        make_staggered_batches(1000),
+        make_staggered_batches(1000),
+        JoinType::RightSemi,
+        None,
+    )
+    .run_test(&[HjSmj, NljHj], true)
+    .await
+}
+
+#[tokio::test]
+async fn test_right_semi_join_1k_filtered() {
+    JoinFuzzTestCase::new(
+        make_staggered_batches(1000),
+        make_staggered_batches(1000),
+        JoinType::RightSemi,
+        Some(Box::new(col_lt_col_filter)),
+    )
+    .run_test(&[HjSmj, NljHj], true)
+    .await
+}
+
 #[tokio::test]
 async fn test_anti_join_1k() {
     JoinFuzzTestCase::new(
@@ -478,9 +503,11 @@ impl JoinFuzzTestCase {
             let task_ctx = ctx.task_ctx();
 
             let hj = self.hash_join();
+            debug!("hj = {:?}", hj);
             let hj_collected = collect(hj, task_ctx.clone()).await.unwrap();
 
             let smj = self.sort_merge_join();
+            debug!("smj = {:#?}", smj);
             let smj_collected = collect(smj, task_ctx.clone()).await.unwrap();
 
             let nlj = self.nested_loop_join();