fix: make block data type be consistent with hash method (#15089)

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

@@ -57,7 +57,7 @@ pub fn get_hashes(
     }
 
     let evaluator = Evaluator::new(&block, func_ctx, &BUILTIN_FUNCTIONS);
-    let columns: Vec<(Column, DataType)> = keys
+    let mut columns: Vec<(Column, DataType)> = keys
         .iter()
         .map(|expr| {
             let return_type = expr.data_type();
@@ -69,6 +69,11 @@ pub fn get_hashes(
             ))
         })
         .collect::<Result<_>>()?;
+    // When chose hash method, the keys are removed nullable, so we need to remove nullable here.
+    for (col, ty) in columns.iter_mut() {
+        *col = col.remove_nullable();
+        *ty = ty.remove_nullable();
+    }
     hash_by_method(method, &columns, block.num_rows(), hashes)?;
     Ok(())
 }

tests/sqllogictests/suites/query/spill/basic_spill.test

@@ -9,6 +9,36 @@ set disable_join_reorder = 1;
 statement ok
 set join_spilling_bytes_threshold_per_proc = 1;
 
+statement ok
+drop table if exists t1;
+
+statement ok
+drop table if exists t2;
+
+statement ok
+create table t1(a varchar not null);
+
+statement ok
+create table t2(b varchar not null);
+
+statement ok
+insert into t1 values('a'),('b');
+
+statement ok
+insert into t2 values('a'),('b');
+
+query TT
+select a from t2 right join t1 on t1.a = t2.b order by a;
+----
+a
+b
+
+statement ok
+drop table t1;
+
+statement ok
+drop table t2;
+
 statement ok
 drop table if exists t3;