fix(query): grace join and selectivity improvements

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

@@ -303,7 +303,7 @@ impl<T: GraceMemoryJoin> GraceHashJoin<T> {
         )?;
 
         for hash in hashes.iter_mut() {
-            *hash = ((*hash << self.shift_bits) >> 60) & 0b1111;
+            *hash = Self::get_partition_id(*hash, self.shift_bits);
         }
 
         Ok(self.build_partition_stream.partition(hashes, data, true))
@@ -324,7 +324,7 @@ impl<T: GraceMemoryJoin> GraceHashJoin<T> {
         )?;
 
         for hash in hashes.iter_mut() {
-            *hash = ((*hash << self.shift_bits) >> 60) & 0b1111;
+            *hash = Self::get_partition_id(*hash, self.shift_bits);
         }
 
         Ok(self.probe_partition_stream.partition(hashes, data, true))
@@ -385,6 +385,20 @@ impl<T: GraceMemoryJoin> GraceHashJoin<T> {
 
         Ok(())
     }
+
+    #[inline(always)]
+    #[cfg(target_feature = "sse4.2")]
+    fn get_partition_id(hash: u64, shift_bits: usize) -> u64 {
+        // On SSE4.2, _mm_crc32_u64 only sets the low 32 bits; high 32 bits are always 0.
+        // Extract partition bits from the low 32 bits to avoid all rows landing in partition 0.
+        (hash << shift_bits >> 28) & 0b1111
+    }
+
+    #[inline(always)]
+    #[cfg(not(target_feature = "sse4.2"))]
+    fn get_partition_id(hash: u64, shift_bits: usize) -> u64 {
+        (hash << shift_bits >> 60) & 0b1111
+    }
 }
 
 pub enum RestoreStage {

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

@@ -54,8 +54,15 @@ impl HybridHashJoinState {
         level: usize,
         factory: Arc<HashJoinFactory>,
     ) -> Result<Arc<HybridHashJoinState>> {
+        // On SSE4.2, fast_hash (_mm_crc32_u64) only sets the low 32 bits.
+        #[cfg(target_feature = "sse4.2")]
+        const HASH_JOIN_SPILL_MAX_LEVEL: usize = 7;
+        #[cfg(not(target_feature = "sse4.2"))]
+        const HASH_JOIN_SPILL_MAX_LEVEL: usize = 15;
+
         let settings = ctx.get_settings();
-        let max_level = settings.get_max_hash_join_spill_level()? as usize;
+        let max_spill_level = settings.get_max_hash_join_spill_level()? as usize;
+        let max_level = (max_spill_level).min(HASH_JOIN_SPILL_MAX_LEVEL);
 
         Ok(Arc::new(HybridHashJoinState {
             ctx,

src/query/sql/src/planner/optimizer/ir/stats/selectivity.rs

@@ -276,6 +276,7 @@ impl SelectivityVisitor<'_> {
                 let column_stat = self
                     .ensure_column_stat(column_index)
                     .expect("checked above");
+
                 return Self::compute_comparison_with_stat(
                     column_stat,
                     constant,
@@ -367,6 +368,11 @@ impl SelectivityVisitor<'_> {
                     let selectivity =
                         Self::compute_histogram_comparison(histogram, op, &const_datum)?;
                     if let Selectivity::N(n) = selectivity {
+                        // Too low selectivity in an unreliable histogram.
+                        if !histogram.accuracy && n < 0.05 {
+                            return Ok(Selectivity::LowerBound);
+                        }
+
                         let (new_min, new_max) = match op {
                             ComparisonOp::GT | ComparisonOp::GTE => {
                                 (const_datum.clone(), column_stat.max.clone())