fix

Author: SkyFan2002

src/query/service/src/pipelines/processors/transforms/hash_join/runtime_filter/builder.rs

@@ -42,6 +42,7 @@ struct JoinRuntimeFilterPacketBuilder<'a> {
     inlist_threshold: usize,
     bloom_threshold: usize,
     min_max_threshold: usize,
+    is_spill_happened: bool,
     selectivity_threshold: u64,
 }
 
@@ -54,6 +55,7 @@ impl<'a> JoinRuntimeFilterPacketBuilder<'a> {
         bloom_threshold: usize,
         min_max_threshold: usize,
         selectivity_threshold: u64,
+        is_spill_happened: bool,
     ) -> Result<Self> {
         let build_key_column = Self::eval_build_key_column(data_blocks, func_ctx, build_key)?;
         Ok(Self {
@@ -63,14 +65,17 @@ impl<'a> JoinRuntimeFilterPacketBuilder<'a> {
             bloom_threshold,
             min_max_threshold,
             selectivity_threshold,
+            is_spill_happened,
         })
     }
     fn build(&self, desc: &RuntimeFilterDesc) -> Result<RuntimeFilterPacket> {
-        if !should_enable_runtime_filter(
-            desc,
-            self.build_key_column.len(),
-            self.selectivity_threshold,
-        ) {
+        if self.is_spill_happened
+            || !should_enable_runtime_filter(
+                desc,
+                self.build_key_column.len(),
+                self.selectivity_threshold,
+            )
+        {
             return Ok(RuntimeFilterPacket {
                 id: desc.id,
                 inlist: None,
@@ -249,6 +254,7 @@ pub fn build_runtime_filter_packet(
     bloom_threshold: usize,
     min_max_threshold: usize,
     selectivity_threshold: u64,
+    is_spill_happened: bool,
 ) -> Result<JoinRuntimeFilterPacket> {
     if build_num_rows == 0 {
         return Ok(JoinRuntimeFilterPacket {
@@ -268,6 +274,7 @@ pub fn build_runtime_filter_packet(
                 bloom_threshold,
                 min_max_threshold,
                 selectivity_threshold,
+                is_spill_happened,
             )?
             .build(rf)?,
         );

src/query/service/src/pipelines/processors/transforms/hash_join/runtime_filter/interface.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::sync::atomic::Ordering;
 use std::time::Instant;
 
 use databend_common_exception::Result;
@@ -48,6 +49,10 @@ pub async fn build_and_push_down_runtime_filter(
         .get_join_runtime_filter_selectivity_threshold()?;
 
     let build_start = Instant::now();
+    let is_spill_happened = join
+        .hash_join_state
+        .is_spill_happened
+        .load(Ordering::Acquire);
     let mut packet = build_runtime_filter_packet(
         build_chunks,
         build_num_rows,
@@ -57,6 +62,7 @@ pub async fn build_and_push_down_runtime_filter(
         bloom_threshold,
         min_max_threshold,
         selectivity_threshold,
+        is_spill_happened,
     )?;
     let build_time = build_start.elapsed();
 

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

@@ -300,6 +300,7 @@ impl Processor for TransformHashJoinBuild {
                             .chunks
                             .clone()
                     };
+
                     let build_num_rows = unsafe {
                         (*self.build_state.hash_join_state.build_state.get())
                             .generation_state

src/query/service/src/pipelines/processors/transforms/new_hash_join/memory/inner_join.rs

@@ -103,6 +103,7 @@ impl Join for InnerHashJoin {
             desc.bloom_threshold,
             desc.min_max_threshold,
             desc.selectivity_threshold,
+            false,
         )
     }