chore(query): fix hash join hang (#17389)

* chore(query): fix hash join hang

* chore(test): add sqllogictest

Author: Dousir9

src/query/service/src/pipelines/processors/transforms/hash_join/transform_hash_join_build.rs

@@ -168,6 +168,13 @@ impl TransformHashJoinBuild {
     }
 
     fn collect(&mut self) -> Result<Event> {
+        if self.input_port.has_data() {
+            self.add_data_block(self.input_port.pull_data().unwrap()?);
+            if self.need_spill() {
+                return self.next_step(Step::Async(AsyncStep::Spill));
+            }
+        }
+
         if self.input_port.is_finished() {
             if self.need_check_spill_happen() {
                 self.next_step(Step::Async(AsyncStep::CheckSpillHappen))
@@ -176,14 +183,6 @@ impl TransformHashJoinBuild {
             } else {
                 self.next_step(Step::Async(AsyncStep::WaitCollect))
             }
-        } else if self.input_port.has_data() {
-            self.add_data_block(self.input_port.pull_data().unwrap()?);
-            if self.need_spill() {
-                self.next_step(Step::Async(AsyncStep::Spill))
-            } else {
-                self.input_port.set_need_data();
-                Ok(Event::NeedData)
-            }
         } else {
             self.input_port.set_need_data();
             Ok(Event::NeedData)

tests/sqllogictests/suites/query/join/join.test

@@ -282,3 +282,6 @@ select * from (select number from numbers(5)) t2 full outer join (select a, 'A'
 
 statement ok
 drop table if exists t;
+
+statement ok
+with v2 as (SELECT 'xx' || cast(number as string) AS invoice_nr FROM numbers(135) group by invoice_nr order by invoice_nr) select v2.invoice_nr from v2 where EXISTS (SELECT cast(number as string) AS invoice_nr FROM numbers(800) where v2.invoice_nr = cast(number as string)) ignore_result;