chore(HashJoinStream): rewrite to generators and simplify

datafusion/physical-plan/src/coalesce/mod.rs

@@ -138,6 +138,11 @@ impl LimitedBatchCoalescer {
         self.finished
     }
 
+    /// Return true if a completed batch is ready to be taken
+    pub fn has_completed_batch(&self) -> bool {
+        self.inner.has_completed_batch()
+    }
+
     /// Return the next completed batch, if any
     pub fn next_completed_batch(&mut self) -> Option<RecordBatch> {
         self.inner.next_completed_batch()

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

@@ -37,9 +37,7 @@ use crate::joins::hash_join::inlist_builder::build_struct_inlist_values;
 use crate::joins::hash_join::shared_bounds::{
     ColumnBounds, PartitionBounds, PushdownStrategy, SharedBuildAccumulator,
 };
-use crate::joins::hash_join::stream::{
-    BuildSide, BuildSideInitialState, HashJoinStream, HashJoinStreamState,
-};
+use crate::joins::hash_join::stream::HashJoinStream;
 use crate::joins::join_hash_map::{JoinHashMapU32, JoinHashMapU64};
 use crate::joins::utils::{
     OnceAsync, OnceFut, asymmetric_join_output_partitioning, reorder_output_after_swap,
@@ -1489,7 +1487,7 @@ impl ExecutionPlan for HashJoinExec {
             .map(|(_, right_expr)| Arc::clone(right_expr))
             .collect::<Vec<_>>();
 
-        Ok(Box::pin(HashJoinStream::new(
+        Ok(HashJoinStream::new(
             partition,
             self.schema(),
             on_right,
@@ -1500,16 +1498,15 @@ impl ExecutionPlan for HashJoinExec {
             join_metrics,
             column_indices_after_projection,
             self.null_equality,
-            HashJoinStreamState::WaitBuildSide,
-            BuildSide::Initial(BuildSideInitialState { left_fut }),
+            left_fut,
             batch_size,
             vec![],
             self.right.output_ordering().is_some(),
             build_accumulator,
             self.mode,
             self.null_aware,
             self.fetch,
-        )))
+        ))
     }
 
     fn metrics(&self) -> Option<MetricsSet> {