fix: prevent hash join deadlock when dynamic filtering is enabled

Empty partitions were short-circuiting to Completed state without reporting
to the shared accumulator. This caused other partitions to wait indefinitely
at the barrier.

This fix ensures that even empty partitions proceed to report their data
if a build accumulator is present.

Made-with: Cursor

Author: RatulDawar

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

@@ -408,12 +408,9 @@ impl HashJoinStream {
 
     /// Returns the next state after the build side has been fully collected
     /// and any required build-side coordination has completed.
-    fn state_after_build_ready(
-        join_type: JoinType,
-        left_data: &JoinLeftData,
-    ) -> HashJoinStreamState {
+    fn state_after_build_ready(&self, left_data: &JoinLeftData) -> HashJoinStreamState {
         if left_data.map().is_empty()
-            && join_type.empty_build_side_produces_empty_result()
+            && self.join_type.empty_build_side_produces_empty_result()
         {
             HashJoinStreamState::Completed
         } else {
@@ -485,8 +482,7 @@ impl HashJoinStream {
             ready!(fut.get_shared(cx))?;
         }
         let build_side = self.build_side.try_as_ready()?;
-        self.state =
-            Self::state_after_build_ready(self.join_type, build_side.left_data.as_ref());
+        self.state = self.state_after_build_ready(build_side.left_data.as_ref());
         Poll::Ready(Ok(StatefulStreamResult::Continue))
     }
 
@@ -557,8 +553,7 @@ impl HashJoinStream {
             }));
             self.state = HashJoinStreamState::WaitPartitionBoundsReport;
         } else {
-            self.state =
-                Self::state_after_build_ready(self.join_type, left_data.as_ref());
+            self.state = self.state_after_build_ready(left_data.as_ref());
         }
 
         self.build_side = BuildSide::Ready(BuildSideReadyState { left_data });
@@ -668,7 +663,21 @@ impl HashJoinStream {
         if is_empty {
             // Invariant: state_after_build_ready should have already completed
             // join types whose result is fixed to empty when the build side is empty.
-            debug_assert!(!self.join_type.empty_build_side_produces_empty_result());
+            //
+            // However, when dynamic filtering is enabled, we skip the short-circuit
+            // in state_after_build_ready to ensure all partitions report their
+            // build-side data. In this case, we might reach here with an empty
+            // build side even for join types that produce empty results.
+            if self.build_accumulator.is_none() {
+                debug_assert!(!self.join_type.empty_build_side_produces_empty_result());
+            }
+
+            if self.join_type.empty_build_side_produces_empty_result() {
+                timer.done();
+                self.state = HashJoinStreamState::FetchProbeBatch;
+                return Ok(StatefulStreamResult::Continue);
+            }
+
             let result = build_batch_empty_build_side(
                 &self.schema,
                 build_side.left_data.batch(),