HashJoin should return Err when the right side input stream produce Err, add more join UTs to cover different join types

datafusion/core/src/physical_plan/joins/hash_join.rs

@@ -1554,7 +1554,8 @@ impl HashJoinStream {
                     }
                     Some(result.map(|x| x.0))
                 }
-                other => {
+                Some(err) => Some(err),
+                None => {
                     let timer = self.join_metrics.join_time.timer();
                     // For the left join, produce rows for unmatched rows
                     match self.join_type {
@@ -1593,7 +1594,7 @@ impl HashJoinStream {
                         | JoinType::Right => {}
                     }
 
-                    other
+                    None
                 }
             })
     }
@@ -1618,9 +1619,11 @@ mod tests {
         physical_plan::{
             common, expressions::Column, memory::MemoryExec, repartition::RepartitionExec,
         },
+        test::exec::MockExec,
         test::{build_table_i32, columns},
     };
     use arrow::datatypes::Field;
+    use arrow::error::ArrowError;
     use datafusion_expr::Operator;
 
     use super::*;
@@ -3095,4 +3098,63 @@ mod tests {
 
         Ok(())
     }
+
+    #[tokio::test]
+    async fn join_with_error_right() {
+        let left = build_table(
+            ("a1", &vec![1, 2, 3]),
+            ("b1", &vec![4, 5, 7]),
+            ("c1", &vec![7, 8, 9]),
+        );
+
+        // right input stream returns one good batch and then one error.
+        // The error should be returned.
+        let err = Err(ArrowError::ComputeError("bad data error".to_string()));
+        let right = build_table_i32(("a2", &vec![]), ("b1", &vec![]), ("c2", &vec![]));
+
+        let on = vec![(
+            Column::new_with_schema("b1", &left.schema()).unwrap(),
+            Column::new_with_schema("b1", &right.schema()).unwrap(),
+        )];
+        let schema = right.schema();
+        let right = build_table_i32(("a2", &vec![]), ("b1", &vec![]), ("c2", &vec![]));
+        let right_input = Arc::new(MockExec::new(vec![Ok(right), err], schema));
+
+        let join_types = vec![
+            JoinType::Inner,
+            JoinType::Left,
+            JoinType::Right,
+            JoinType::Full,
+            JoinType::LeftSemi,
+            JoinType::LeftAnti,
+            JoinType::RightSemi,
+            JoinType::RightAnti,
+        ];
+
+        for join_type in join_types {
+            let join = join(
+                left.clone(),
+                right_input.clone(),
+                on.clone(),
+                &join_type,
+                false,
+            )
+            .unwrap();
+            let session_ctx = SessionContext::new();
+            let task_ctx = session_ctx.task_ctx();
+
+            let stream = join.execute(0, task_ctx).unwrap();
+
+            // Expect that an error is returned
+            let result_string = crate::physical_plan::common::collect(stream)
+                .await
+                .unwrap_err()
+                .to_string();
+            assert!(
+                result_string.contains("bad data error"),
+                "actual: {}",
+                result_string
+            );
+        }
+    }
 }