x

Author: forsaken628

src/common/statistics/src/datum.rs

@@ -20,6 +20,14 @@ use databend_common_exception::ErrorCode;
 
 pub type F64 = OrderedFloat<f64>;
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum DatumKind {
+    Int,
+    UInt,
+    Float,
+    Bytes,
+}
+
 #[derive(serde::Serialize, serde::Deserialize, Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
 pub enum Datum {
     Bool(bool),
@@ -114,6 +122,104 @@ impl Datum {
         matches!(self, Datum::Int(_) | Datum::UInt(_) | Datum::Float(_))
     }
 
+    pub fn kind(&self) -> Option<DatumKind> {
+        match self {
+            Datum::Int(_) => Some(DatumKind::Int),
+            Datum::UInt(_) => Some(DatumKind::UInt),
+            Datum::Float(_) => Some(DatumKind::Float),
+            Datum::Bytes(_) => Some(DatumKind::Bytes),
+            Datum::Bool(_) => None,
+        }
+    }
+
+    pub fn normalize_to_kind(&self, kind: DatumKind) -> Option<Datum> {
+        match kind {
+            DatumKind::Int => self.as_i64().map(Datum::Int),
+            DatumKind::UInt => self.as_u64().map(Datum::UInt),
+            DatumKind::Float => self.as_double().ok().map(F64::from).map(Datum::Float),
+            DatumKind::Bytes => match self {
+                Datum::Bytes(value) => Some(Datum::Bytes(value.clone())),
+                _ => None,
+            },
+        }
+    }
+
+    pub fn lower_bound_to_kind(&self, kind: DatumKind) -> Option<Datum> {
+        match kind {
+            DatumKind::Int => self.lower_bound_as_i64().map(Datum::Int),
+            DatumKind::UInt => self.lower_bound_as_u64().map(Datum::UInt),
+            _ => self.normalize_to_kind(kind),
+        }
+    }
+
+    pub fn upper_bound_to_kind(&self, kind: DatumKind) -> Option<Datum> {
+        match kind {
+            DatumKind::Int => self.upper_bound_as_i64().map(Datum::Int),
+            DatumKind::UInt => self.upper_bound_as_u64().map(Datum::UInt),
+            _ => self.normalize_to_kind(kind),
+        }
+    }
+
+    pub fn as_i64(&self) -> Option<i64> {
+        match self {
+            Datum::Int(value) => Some(*value),
+            Datum::UInt(value) => i64::try_from(*value).ok(),
+            Datum::Float(value) => {
+                let value = value.into_inner();
+                (value.is_finite()
+                    && value.fract() == 0.0
+                    && value >= i64::MIN as f64
+                    && value <= i64::MAX as f64)
+                    .then_some(value as i64)
+            }
+            _ => None,
+        }
+    }
+
+    pub fn as_u64(&self) -> Option<u64> {
+        match self {
+            Datum::UInt(value) => Some(*value),
+            Datum::Int(value) => u64::try_from(*value).ok(),
+            Datum::Float(value) => {
+                let value = value.into_inner();
+                (value.is_finite()
+                    && value.fract() == 0.0
+                    && value >= 0.0
+                    && value <= u64::MAX as f64)
+                    .then_some(value as u64)
+            }
+            _ => None,
+        }
+    }
+
+    fn lower_bound_as_i64(&self) -> Option<i64> {
+        match self {
+            Datum::Float(value) => float_to_i64_bound(value.into_inner(), f64::ceil),
+            _ => self.as_i64(),
+        }
+    }
+
+    fn upper_bound_as_i64(&self) -> Option<i64> {
+        match self {
+            Datum::Float(value) => float_to_i64_bound(value.into_inner(), f64::floor),
+            _ => self.as_i64(),
+        }
+    }
+
+    fn lower_bound_as_u64(&self) -> Option<u64> {
+        match self {
+            Datum::Float(value) => float_to_u64_bound(value.into_inner(), f64::ceil),
+            _ => self.as_u64(),
+        }
+    }
+
+    fn upper_bound_as_u64(&self) -> Option<u64> {
+        match self {
+            Datum::Float(value) => float_to_u64_bound(value.into_inner(), f64::floor),
+            _ => self.as_u64(),
+        }
+    }
+
     pub fn type_name(&self) -> &'static str {
         match self {
             Datum::Bool(_) => "Boolean",
@@ -151,3 +257,14 @@ impl Datum {
         }
     }
 }
+
+fn float_to_i64_bound(value: f64, round: fn(f64) -> f64) -> Option<i64> {
+    let value = round(value);
+    (value.is_finite() && value >= i64::MIN as f64 && value <= i64::MAX as f64)
+        .then_some(value as i64)
+}
+
+fn float_to_u64_bound(value: f64, round: fn(f64) -> f64) -> Option<u64> {
+    let value = round(value);
+    (value.is_finite() && value >= 0.0 && value <= u64::MAX as f64).then_some(value as u64)
+}

src/common/statistics/src/histogram.rs

@@ -195,12 +195,200 @@ impl Histogram {
         }
     }
 
+    /// Estimate a join for matching histogram types, or for mixed numeric histograms
+    /// through a temporary float view. Non-numeric mixed types return `None`.
+    pub fn estimate_join_numeric_compatible(
+        &self,
+        other: &Histogram,
+    ) -> ExceptionResult<Option<JoinEstimation>> {
+        match (self, other) {
+            (Self::Int(left), Self::Int(right)) => Ok(Some(left.estimate_join(right))),
+            (Self::UInt(left), Self::UInt(right)) => Ok(Some(left.estimate_join(right))),
+            (Self::Float(left), Self::Float(right)) => Ok(Some(left.estimate_join(right))),
+            (Self::Bytes(left), Self::Bytes(right)) => Ok(Some(left.estimate_join(right))),
+            (Self::Bytes(_), _) | (_, Self::Bytes(_)) => Ok(None),
+            _ => estimate_mixed_numeric_histogram_join(self, other),
+        }
+    }
+
+    pub fn restrict_to_bounds(&self, min: &Datum, max: &Datum) -> ExceptionResult<Option<Self>> {
+        let buckets = self
+            .bucket_iter()
+            .map(|bucket| {
+                let bucket_min = bucket.lower_bound();
+                let bucket_max = bucket.upper_bound();
+                if bucket_min.compare(max)? == std::cmp::Ordering::Greater
+                    || bucket_max.compare(min)? == std::cmp::Ordering::Less
+                {
+                    return Ok(None);
+                }
+
+                let Some(lower_bound) = max_datum_as_bucket_kind(&bucket_min, min) else {
+                    return Ok(None);
+                };
+                let Some(upper_bound) = min_datum_as_bucket_kind(&bucket_max, max) else {
+                    return Ok(None);
+                };
+                if lower_bound.compare(&upper_bound)? == std::cmp::Ordering::Greater {
+                    return Ok(None);
+                }
+
+                let selectivity = bucket_overlap_selectivity(
+                    &bucket_min,
+                    &bucket_max,
+                    &lower_bound,
+                    &upper_bound,
+                );
+                HistogramBucket::try_from_bounds(
+                    lower_bound,
+                    upper_bound,
+                    bucket.num_values() * selectivity,
+                    bucket.num_distinct() * selectivity,
+                )
+                .map(Some)
+                .map_err(|err| ErrorCode::Internal(err.to_string()))
+            })
+            .collect::<ExceptionResult<Vec<_>>>()?
+            .into_iter()
+            .flatten()
+            .collect::<Vec<_>>();
+
+        if buckets.is_empty() {
+            return Ok(None);
+        }
+
+        Self::try_from_buckets(self.accuracy(), buckets, self.avg_spacing())
+            .map(Some)
+            .map_err(|err| ErrorCode::Internal(err.to_string()))
+    }
+
     pub fn is_range_distorted(&self) -> bool {
         self.avg_spacing()
             .is_some_and(|bucket_width| bucket_width > 1e12)
     }
 }
 
+fn estimate_mixed_numeric_histogram_join(
+    left: &Histogram,
+    right: &Histogram,
+) -> ExceptionResult<Option<JoinEstimation>> {
+    let Some(left) = numeric_histogram_as_float(left)? else {
+        return Ok(None);
+    };
+    let Some(right) = numeric_histogram_as_float(right)? else {
+        return Ok(None);
+    };
+
+    Ok(Some(left.estimate_join(&right)))
+}
+
+fn numeric_histogram_as_float(
+    histogram: &Histogram,
+) -> ExceptionResult<Option<TypedHistogram<F64>>> {
+    if matches!(histogram, Histogram::Bytes(_)) {
+        return Ok(None);
+    }
+
+    let buckets = histogram
+        .bucket_iter()
+        .map(|bucket| {
+            let lower_bound = F64::from(bucket.lower_bound().as_double()?);
+            let upper_bound = F64::from(bucket.upper_bound().as_double()?);
+            Ok(TypedHistogramBucket::new(
+                lower_bound,
+                upper_bound,
+                bucket.num_values(),
+                bucket.num_distinct(),
+            ))
+        })
+        .collect::<ExceptionResult<Vec<_>>>()?;
+
+    Ok(Some(TypedHistogram {
+        accuracy: histogram.accuracy(),
+        buckets,
+        avg_spacing: histogram.avg_spacing(),
+    }))
+}
+
+fn max_datum_as_bucket_kind(bucket_value: &Datum, stat_value: &Datum) -> Option<Datum> {
+    let selected = if bucket_value.compare(stat_value).ok()? == std::cmp::Ordering::Less {
+        stat_value
+    } else {
+        bucket_value
+    };
+    selected.normalize_to_kind(bucket_value.kind()?)
+}
+
+fn min_datum_as_bucket_kind(bucket_value: &Datum, stat_value: &Datum) -> Option<Datum> {
+    let selected = if bucket_value.compare(stat_value).ok()? == std::cmp::Ordering::Greater {
+        stat_value
+    } else {
+        bucket_value
+    };
+    selected.normalize_to_kind(bucket_value.kind()?)
+}
+
+fn bucket_overlap_selectivity(
+    bucket_min: &Datum,
+    bucket_max: &Datum,
+    new_min: &Datum,
+    new_max: &Datum,
+) -> f64 {
+    match (bucket_min, bucket_max, new_min, new_max) {
+        (
+            Datum::Int(bucket_min),
+            Datum::Int(bucket_max),
+            Datum::Int(new_min),
+            Datum::Int(new_max),
+        ) => discrete_overlap_selectivity(
+            *bucket_min as i128,
+            *bucket_max as i128,
+            *new_min as i128,
+            *new_max as i128,
+        ),
+        (
+            Datum::UInt(bucket_min),
+            Datum::UInt(bucket_max),
+            Datum::UInt(new_min),
+            Datum::UInt(new_max),
+        ) => discrete_overlap_selectivity(
+            *bucket_min as i128,
+            *bucket_max as i128,
+            *new_min as i128,
+            *new_max as i128,
+        ),
+        (
+            Datum::Float(bucket_min),
+            Datum::Float(bucket_max),
+            Datum::Float(new_min),
+            Datum::Float(new_max),
+        ) => {
+            let bucket_width = bucket_max.into_inner() - bucket_min.into_inner();
+            if bucket_width <= 0.0 {
+                return 1.0;
+            }
+            let overlap_width = new_max.into_inner() - new_min.into_inner();
+            (overlap_width / bucket_width).clamp(0.0, 1.0)
+        }
+        (Datum::Bytes(_), Datum::Bytes(_), Datum::Bytes(_), Datum::Bytes(_)) => 1.0,
+        _ => 1.0,
+    }
+}
+
+fn discrete_overlap_selectivity(
+    bucket_min: i128,
+    bucket_max: i128,
+    new_min: i128,
+    new_max: i128,
+) -> f64 {
+    let bucket_count = bucket_max - bucket_min + 1;
+    if bucket_count <= 0 {
+        return 1.0;
+    }
+    let overlap_count = new_max - new_min + 1;
+    (overlap_count as f64 / bucket_count as f64).clamp(0.0, 1.0)
+}
+
 pub enum HistogramBucketIter<'a> {
     Int(std::slice::Iter<'a, TypedHistogramBucket<i64>>),
     UInt(std::slice::Iter<'a, TypedHistogramBucket<u64>>),
@@ -393,3 +581,58 @@ impl fmt::Display for Histogram {
         Ok(())
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_restrict_to_bounds_uses_existing_buckets() -> ExceptionResult<()> {
+        let histogram = Histogram::UInt(TypedHistogram {
+            accuracy: true,
+            buckets: vec![
+                TypedHistogramBucket::new(0, 4, 5.0, 5.0),
+                TypedHistogramBucket::new(5, 9, 5.0, 5.0),
+            ],
+            avg_spacing: None,
+        });
+
+        let restricted = histogram
+            .restrict_to_bounds(&Datum::UInt(2), &Datum::UInt(6))?
+            .expect("histogram should keep intersecting buckets");
+        let buckets = restricted.bucket_iter().collect::<Vec<_>>();
+
+        assert_eq!(buckets.len(), 2);
+        assert_eq!(buckets[0].lower_bound(), Datum::UInt(2));
+        assert_eq!(buckets[0].upper_bound(), Datum::UInt(4));
+        assert_eq!(buckets[0].num_values(), 3.0);
+        assert_eq!(buckets[0].num_distinct(), 3.0);
+        assert_eq!(buckets[1].lower_bound(), Datum::UInt(5));
+        assert_eq!(buckets[1].upper_bound(), Datum::UInt(6));
+        assert_eq!(buckets[1].num_values(), 2.0);
+        assert_eq!(buckets[1].num_distinct(), 2.0);
+        Ok(())
+    }
+
+    #[test]
+    fn test_mixed_numeric_join_uses_float_view_of_existing_buckets() -> ExceptionResult<()> {
+        let left = Histogram::Int(TypedHistogram {
+            accuracy: true,
+            buckets: vec![TypedHistogramBucket::new(1, 1, 3.0, 1.0)],
+            avg_spacing: None,
+        });
+        let right = Histogram::UInt(TypedHistogram {
+            accuracy: true,
+            buckets: vec![TypedHistogramBucket::new(1, 1, 2.0, 1.0)],
+            avg_spacing: None,
+        });
+
+        let estimation = left
+            .estimate_join_numeric_compatible(&right)?
+            .expect("mixed numeric histograms should use a float view");
+
+        assert_eq!(estimation.cardinality.expected, 6.0);
+        assert_eq!(estimation.ndv.expected, 1.0);
+        Ok(())
+    }
+}