post-merge fixes

Author: korowa

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

@@ -1159,10 +1159,7 @@ impl HashJoinStream {
     fn poll_next_impl(
         &mut self,
         cx: &mut std::task::Context<'_>,
-    ) -> Poll<Option<Result<RecordBatch>>>
-    where
-        Self: Send,
-    {
+    ) -> Poll<Option<Result<RecordBatch>>> {
         loop {
             return match self.state {
                 HashJoinStreamState::WaitBuildSide => {

datafusion/physical-plan/src/joins/stream_join_utils.rs

@@ -759,7 +759,7 @@ pub trait EagerJoinStream {
         match self.right_stream().next().await {
             Some(Ok(batch)) => {
                 if batch.num_rows() == 0 {
-                    return Ok(StreamJoinStateResult::Continue);
+                    return Ok(StatefulStreamResult::Continue);
                 }
 
                 self.set_state(EagerJoinStreamState::PullLeft);
@@ -788,7 +788,7 @@ pub trait EagerJoinStream {
         match self.left_stream().next().await {
             Some(Ok(batch)) => {
                 if batch.num_rows() == 0 {
-                    return Ok(StreamJoinStateResult::Continue);
+                    return Ok(StatefulStreamResult::Continue);
                 }
                 self.set_state(EagerJoinStreamState::PullRight);
                 self.process_batch_from_left(batch)
@@ -817,7 +817,7 @@ pub trait EagerJoinStream {
         match self.left_stream().next().await {
             Some(Ok(batch)) => {
                 if batch.num_rows() == 0 {
-                    return Ok(StreamJoinStateResult::Continue);
+                    return Ok(StatefulStreamResult::Continue);
                 }
                 self.process_batch_after_right_end(batch)
             }
@@ -847,7 +847,7 @@ pub trait EagerJoinStream {
         match self.right_stream().next().await {
             Some(Ok(batch)) => {
                 if batch.num_rows() == 0 {
-                    return Ok(StreamJoinStateResult::Continue);
+                    return Ok(StatefulStreamResult::Continue);
                 }
                 self.process_batch_after_left_end(batch)
             }