Add test on empty build side and test on multiple waiters

Author: LiaCastaneda

datafusion/physical-expr/src/expressions/dynamic_filters.rs

@@ -607,4 +607,32 @@ mod test {
         // wait_complete should return immediately
         dynamic_filter.wait_complete().await;
     }
+
+    #[tokio::test]
+    async fn test_multiple_waiters() {
+        let dynamic_filter = Arc::new(DynamicFilterPhysicalExpr::new(
+            vec![],
+            lit(42) as Arc<dyn PhysicalExpr>,
+        ));
+
+        // Create multiple tasks waiting for completion
+        let df1 = Arc::clone(&dynamic_filter);
+        let df2 = Arc::clone(&dynamic_filter);
+        let df3 = Arc::clone(&dynamic_filter);
+
+        let waiter1 = tokio::spawn(async move { df1.wait_complete().await });
+        let waiter2 = tokio::spawn(async move { df2.wait_complete().await });
+        let waiter3 = tokio::spawn(async move { df3.wait_complete().await });
+
+        // Small delay to ensure all waiters are subscribed
+        tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
+
+        // Mark as complete - all waiters should be notified
+        dynamic_filter.mark_complete();
+
+        // All waiters should complete successfully
+        waiter1.await.unwrap();
+        waiter2.await.unwrap();
+        waiter3.await.unwrap();
+    }
 }

datafusion/physical-plan/src/joins/hash_join/exec.rs

@@ -4545,4 +4545,52 @@ mod tests {
 
         Ok(())
     }
+
+    /// This test verifies that the dynamic filter is marked as complete even when the build side is empty.
+    #[tokio::test]
+    async fn test_hash_join_marks_filter_complete_empty_build_side() -> Result<()> {
+        let task_ctx = Arc::new(TaskContext::default());
+        // Empty left side (build side)
+        let left = build_table(("a1", &vec![]), ("b1", &vec![]), ("c1", &vec![]));
+        let right = build_table(
+            ("a2", &vec![10, 20, 30]),
+            ("b1", &vec![4, 5, 6]),
+            ("c2", &vec![70, 80, 90]),
+        );
+
+        let on = vec![(
+            Arc::new(Column::new_with_schema("b1", &left.schema())?) as _,
+            Arc::new(Column::new_with_schema("b1", &right.schema())?) as _,
+        )];
+
+        // Create a dynamic filter manually
+        let dynamic_filter = HashJoinExec::create_dynamic_filter(&on);
+        let dynamic_filter_clone = Arc::clone(&dynamic_filter);
+
+        // Create HashJoinExec with the dynamic filter
+        let mut join = HashJoinExec::try_new(
+            left,
+            right,
+            on,
+            None,
+            &JoinType::Inner,
+            None,
+            PartitionMode::CollectLeft,
+            NullEquality::NullEqualsNothing,
+        )?;
+        join.dynamic_filter = Some(HashJoinExecDynamicFilter {
+            filter: dynamic_filter,
+            bounds_accumulator: OnceLock::new(),
+        });
+
+        // Execute the join
+        let stream = join.execute(0, task_ctx)?;
+        let _batches = common::collect(stream).await?;
+
+        // Even with empty build side, the dynamic filter should be marked as complete
+        // wait_complete() should return immediately
+        dynamic_filter_clone.wait_complete().await;
+
+        Ok(())
+    }
 }