fix(query): data lost in new hash join caused by spill (#19415)

* fix(query): fail-close runtime filter on spill

revert

fix(query): disable runtime filter for correlated right semi/anti

fixup

change strategy

add log

fix: try fix lost data when new join spilled

test: revert enable_experimental_new_join

* fixup

* fixup

* fixup

* refine log

* revert reproduce ci

Author: dqhl76

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

@@ -256,16 +256,12 @@ pub fn build_runtime_filter_packet(
     is_spill_happened: bool,
 ) -> Result<JoinRuntimeFilterPacket> {
     if is_spill_happened {
-        return Ok(JoinRuntimeFilterPacket::disable_all(
-            runtime_filter_desc,
-            build_num_rows,
-        ));
+        return Ok(JoinRuntimeFilterPacket::disable_all(build_num_rows));
     }
     if build_num_rows == 0 {
-        return Ok(JoinRuntimeFilterPacket {
-            packets: None,
-            build_rows: build_num_rows,
-        });
+        return Ok(JoinRuntimeFilterPacket::complete_without_filters(
+            build_num_rows,
+        ));
     }
     let mut runtime_filters = HashMap::new();
     for rf in runtime_filter_desc {
@@ -283,8 +279,8 @@ pub fn build_runtime_filter_packet(
             .build(rf)?,
         );
     }
-    Ok(JoinRuntimeFilterPacket {
-        packets: Some(runtime_filters),
-        build_rows: build_num_rows,
-    })
+    Ok(JoinRuntimeFilterPacket::complete(
+        runtime_filters,
+        build_num_rows,
+    ))
 }

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

@@ -223,17 +223,11 @@ impl RuntimeFilterLocalBuilder {
         let total_rows = self.total_rows;
 
         if spill_happened {
-            return Ok(JoinRuntimeFilterPacket::disable_all(
-                &self.runtime_filters,
-                total_rows,
-            ));
+            return Ok(JoinRuntimeFilterPacket::disable_all(total_rows));
         }
 
         if total_rows == 0 {
-            return Ok(JoinRuntimeFilterPacket {
-                packets: None,
-                build_rows: 0,
-            });
+            return Ok(JoinRuntimeFilterPacket::complete_without_filters(0));
         }
 
         let packets: Vec<_> = self
@@ -245,10 +239,10 @@ impl RuntimeFilterLocalBuilder {
             })
             .collect::<Result<_>>()?;
 
-        Ok(JoinRuntimeFilterPacket {
-            packets: Some(packets.into_iter().collect()),
-            build_rows: total_rows,
-        })
+        Ok(JoinRuntimeFilterPacket::complete(
+            packets.into_iter().collect(),
+            total_rows,
+        ))
     }
 }
 

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

@@ -29,17 +29,22 @@ pub fn merge_join_runtime_filter_packets(
         packets
     );
     let total_build_rows: usize = packets.iter().map(|packet| packet.build_rows).sum();
-    // Skip packets that `JoinRuntimeFilterPacket::packets` is `None`
+
+    // If any packet is incomplete (disable_all_due_to_spill), the merged result is also incomplete
+    if packets.iter().any(|packet| packet.disable_all_due_to_spill) {
+        return Ok(JoinRuntimeFilterPacket::disable_all(total_build_rows));
+    }
+
     let packets = packets
         .into_iter()
         .filter_map(|packet| packet.packets)
         .collect::<Vec<_>>();
 
+    // Skip packets that `JoinRuntimeFilterPacket::packets` is `None`
     if packets.is_empty() {
-        return Ok(JoinRuntimeFilterPacket {
-            packets: None,
-            build_rows: total_build_rows,
-        });
+        return Ok(JoinRuntimeFilterPacket::complete_without_filters(
+            total_build_rows,
+        ));
     }
 
     let mut result = HashMap::new();
@@ -56,10 +61,14 @@ pub fn merge_join_runtime_filter_packets(
         "RUNTIME-FILTER: merge_join_runtime_filter_packets output: {:?}",
         result
     );
-    Ok(JoinRuntimeFilterPacket {
-        packets: Some(result),
-        build_rows: total_build_rows,
-    })
+
+    if result.is_empty() {
+        return Ok(JoinRuntimeFilterPacket::complete_without_filters(
+            total_build_rows,
+        ));
+    }
+
+    Ok(JoinRuntimeFilterPacket::complete(result, total_build_rows))
 }
 
 fn merge_inlist(

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

@@ -31,8 +31,6 @@ use databend_common_expression::types::NumberColumn;
 use databend_common_expression::types::NumberColumnBuilder;
 use databend_common_expression::types::array::ArrayColumnBuilder;
 
-use crate::pipelines::processors::transforms::RuntimeFilterDesc;
-
 /// Represents a runtime filter that can be transmitted and merged.
 ///
 /// # Fields
@@ -66,30 +64,41 @@ impl Debug for RuntimeFilterPacket {
 ///
 /// # Fields
 ///
-/// * `packets` - A map of runtime filter packets, keyed by their unique identifier `RuntimeFilterPacket::id`. When `packets` is `None`, it means that `build_num_rows` is zero.
+/// * `packets` - A map of runtime filter packets, keyed by their unique identifier `RuntimeFilterPacket::id`.
 /// * `build_rows` - Total number of rows used when building the runtime filters.
+/// * `disable_all_due_to_spill` - Indicates if this packet comes from a spilled build and should disable all runtime filters globally.
 #[derive(serde::Serialize, serde::Deserialize, Clone, Debug, Default, PartialEq)]
 pub struct JoinRuntimeFilterPacket {
     #[serde(default)]
     pub packets: Option<HashMap<usize, RuntimeFilterPacket>>,
     #[serde(default)]
     pub build_rows: usize,
+    #[serde(default)]
+    pub disable_all_due_to_spill: bool,
 }
 
 impl JoinRuntimeFilterPacket {
-    pub fn disable_all(descs: &[RuntimeFilterDesc], build_rows: usize) -> Self {
-        let mut packets = HashMap::new();
-        for desc in descs {
-            packets.insert(desc.id, RuntimeFilterPacket {
-                id: desc.id,
-                inlist: None,
-                min_max: None,
-                bloom: None,
-            });
+    pub fn complete_without_filters(build_rows: usize) -> Self {
+        Self {
+            packets: None,
+            build_rows,
+            disable_all_due_to_spill: false,
         }
+    }
+
+    pub fn complete(packets: HashMap<usize, RuntimeFilterPacket>, build_rows: usize) -> Self {
         Self {
             packets: Some(packets),
             build_rows,
+            disable_all_due_to_spill: false,
+        }
+    }
+
+    pub fn disable_all(build_rows: usize) -> Self {
+        Self {
+            packets: None,
+            build_rows,
+            disable_all_due_to_spill: true,
         }
     }
 }
@@ -108,6 +117,8 @@ struct FlightJoinRuntimeFilterPacket {
     pub build_rows: usize,
     #[serde(default)]
     pub packets: Option<HashMap<usize, FlightRuntimeFilterPacket>>,
+    #[serde(default)]
+    pub disable_all_due_to_spill: bool,
 
     pub schema: DataSchemaRef,
 }
@@ -166,6 +177,7 @@ impl TryInto<DataBlock> for JoinRuntimeFilterPacket {
         data_block.add_meta(Some(Box::new(FlightJoinRuntimeFilterPacket {
             build_rows: self.build_rows,
             packets: join_flight_packets,
+            disable_all_due_to_spill: self.disable_all_due_to_spill,
             schema,
         })))
     }
@@ -183,6 +195,7 @@ impl TryFrom<DataBlock> for JoinRuntimeFilterPacket {
                 return Ok(JoinRuntimeFilterPacket {
                     packets: None,
                     build_rows: flight_join_rf.build_rows,
+                    disable_all_due_to_spill: flight_join_rf.disable_all_due_to_spill,
                 });
             };
 
@@ -219,6 +232,7 @@ impl TryFrom<DataBlock> for JoinRuntimeFilterPacket {
             return Ok(JoinRuntimeFilterPacket {
                 packets: Some(flight_packets),
                 build_rows: flight_join_rf.build_rows,
+                disable_all_due_to_spill: flight_join_rf.disable_all_due_to_spill,
             });
         }
 

src/query/service/src/pipelines/processors/transforms/new_hash_join/grace/grace_join.rs

@@ -31,6 +31,7 @@ use databend_common_storages_parquet::ReadSettings;
 
 use crate::pipelines::processors::HashJoinDesc;
 use crate::pipelines::processors::transforms::Join;
+use crate::pipelines::processors::transforms::JoinRuntimeFilterPacket;
 use crate::pipelines::processors::transforms::get_hashes;
 use crate::pipelines::processors::transforms::new_hash_join::grace::grace_memory::GraceMemoryJoin;
 use crate::pipelines::processors::transforms::new_hash_join::grace::grace_state::GraceHashJoinState;
@@ -68,6 +69,11 @@ unsafe impl<T: GraceMemoryJoin> Send for GraceHashJoin<T> {}
 unsafe impl<T: GraceMemoryJoin> Sync for GraceHashJoin<T> {}
 
 impl<T: GraceMemoryJoin> Join for GraceHashJoin<T> {
+    fn build_runtime_filter(&self) -> Result<JoinRuntimeFilterPacket> {
+        // TODO: this is hacked to mark it as disabled, we may need look back latter
+        Ok(JoinRuntimeFilterPacket::disable_all(0))
+    }
+
     fn add_block(&mut self, data: Option<DataBlock>) -> Result<()> {
         let ready_partitions = match data {
             None => self.finalize_build_data(),

src/query/service/src/pipelines/processors/transforms/new_hash_join/hybrid/hybrid_join.rs

@@ -199,10 +199,11 @@ impl Join for HybridHashJoin {
     }
 
     fn final_build(&mut self) -> Result<Option<ProgressValues>> {
-        // Only the state needs to be transitioned, as all data migration must have been completed in the previous stage.
+        // A processor can reach final_build while still in Memory mode even if another processor
+        // has already triggered spill. In that case we must still perform full transition work
+        // to avoid leaving late-arrived chunks in memory state.
         if self.state.check_spilled() && matches!(self.mode, HybridJoinMode::Memory(_)) {
-            let grace_join = self.create_grace_join()?;
-            self.mode = HybridJoinMode::Grace(Box::new(grace_join));
+            self.switch_to_grace_mode(true)?;
         }
 
         match &mut self.mode {

src/query/service/src/pipelines/processors/transforms/new_hash_join/transform_hash_join.rs

@@ -28,6 +28,7 @@ use databend_common_pipeline::core::OutputPort;
 use databend_common_pipeline::core::Processor;
 use databend_common_pipeline::core::ProcessorPtr;
 use databend_common_sql::ColumnSet;
+use log::info;
 
 use crate::pipelines::processors::transforms::RuntimeFilterLocalBuilder;
 use crate::pipelines::processors::transforms::new_hash_join::join::Join;
@@ -227,7 +228,17 @@ impl Processor for TransformHashJoin {
                 let before_wait = self.instant.elapsed();
 
                 if wait_res.is_leader() {
+                    let spilled = self.join.is_spill_happened();
                     let packet = self.join.build_runtime_filter()?;
+                    if let Some(packets) = &packet.packets {
+                        info!(
+                            "spilled: {}, globalize runtime filter: total {}, disable_all_due_to_spill: {}",
+                            spilled,
+                            packets.len(),
+                            packet.disable_all_due_to_spill
+                        );
+                    };
+
                     self.rf_desc.globalization(packet).await?;
                 }