StatisticsV2: statistics framework initial redesign for Datafusion

datafusion/common/src/spans.rs

@@ -140,7 +140,7 @@ impl Span {
 /// the column a that comes from SELECT 1 AS a UNION ALL SELECT 2 AS a you'll
 /// need two spans.
 #[derive(Debug, Clone)]
-// Store teh first [`Span`] on the stack because that is by far the most common
+// Store the first [`Span`] on the stack because that is by far the most common
 // case. More will spill onto the heap.
 pub struct Spans(pub Vec<Span>);
 

datafusion/expr-common/Cargo.toml

@@ -34,8 +34,6 @@ workspace = true
 name = "datafusion_expr_common"
 path = "src/lib.rs"
 
-[features]
-
 [dependencies]
 arrow = { workspace = true }
 datafusion-common = { workspace = true }

datafusion/expr-common/src/interval_arithmetic.rs

@@ -17,12 +17,13 @@
 
 //! Interval arithmetic library
 
-use crate::operator::Operator;
-use crate::type_coercion::binary::BinaryTypeCoercer;
 use std::borrow::Borrow;
 use std::fmt::{self, Display, Formatter};
 use std::ops::{AddAssign, SubAssign};
 
+use crate::operator::Operator;
+use crate::type_coercion::binary::{comparison_coercion_numeric, BinaryTypeCoercer};
+
 use arrow::compute::{cast_with_options, CastOptions};
 use arrow::datatypes::{
     DataType, IntervalDayTime, IntervalMonthDayNano, IntervalUnit, TimeUnit,
@@ -653,26 +654,31 @@ impl Interval {
         Ok(Some(Self { lower, upper }))
     }
 
-    /// Decide if this interval certainly contains, possibly contains, or can't
-    /// contain a [`ScalarValue`] (`other`) by returning `[true, true]`,
-    /// `[false, true]` or `[false, false]` respectively.
-    ///
-    /// NOTE: This function only works with intervals of the same data type.
-    ///       Attempting to compare intervals of different data types will lead
-    ///       to an error.
+    /// Decide if this interval contains a [`ScalarValue`] (`other`) by returning `true` or `false`.
     pub fn contains_value<T: Borrow<ScalarValue>>(&self, other: T) -> Result<bool> {
         let rhs = other.borrow();
-        if self.data_type().ne(&rhs.data_type()) {
+
+        let (lhs_lower, lhs_upper, rhs) = if self.data_type().eq(&rhs.data_type()) {
+            (&self.lower, &self.upper, rhs)
+        } else if let Some(common_type) =
+            comparison_coercion_numeric(&self.data_type(), &rhs.data_type())
+        {
+            (
+                &self.lower.cast_to(&common_type)?,
+                &self.upper.cast_to(&common_type)?,
+                &rhs.cast_to(&common_type)?,
+            )
+        } else {
             return internal_err!(
                 "Data types must be compatible for containment checks, lhs:{}, rhs:{}",
                 self.data_type(),
                 rhs.data_type()
             );
-        }
+        };
 
         // We only check the upper bound for a `None` value because `None`
         // values are less than `Some` values according to Rust.
-        Ok(&self.lower <= rhs && (self.upper.is_null() || rhs <= &self.upper))
+        Ok(lhs_lower <= rhs && (lhs_upper.is_null() || rhs <= lhs_upper))
     }
 
     /// Decide if this interval is a superset of, overlaps with, or
@@ -1177,7 +1183,7 @@ fn next_value_helper<const INC: bool>(value: ScalarValue) -> ScalarValue {
 
 /// Returns the greater of the given interval bounds. Assumes that a `NULL`
 /// value represents `NEG_INF`.
-fn max_of_bounds(first: &ScalarValue, second: &ScalarValue) -> ScalarValue {
+pub fn max_of_bounds(first: &ScalarValue, second: &ScalarValue) -> ScalarValue {
     if !first.is_null() && (second.is_null() || first >= second) {
         first.clone()
     } else {
@@ -1187,7 +1193,7 @@ fn max_of_bounds(first: &ScalarValue, second: &ScalarValue) -> ScalarValue {
 
 /// Returns the lesser of the given interval bounds. Assumes that a `NULL`
 /// value represents `INF`.
-fn min_of_bounds(first: &ScalarValue, second: &ScalarValue) -> ScalarValue {
+pub fn min_of_bounds(first: &ScalarValue, second: &ScalarValue) -> ScalarValue {
     if !first.is_null() && (second.is_null() || first <= second) {
         first.clone()
     } else {
@@ -1868,6 +1874,7 @@ mod tests {
     };
 
     use arrow::datatypes::DataType;
+    use datafusion_common::rounding::{next_down, next_up};
     use datafusion_common::{Result, ScalarValue};
 
     #[test]
@@ -2594,6 +2601,43 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn test_contains_value() -> Result<()> {
+        let possible_cases = vec![
+            (
+                Interval::make(Some(0), Some(100))?,
+                ScalarValue::Int32(Some(50)),
+                true,
+            ),
+            (
+                Interval::make(Some(0), Some(100))?,
+                ScalarValue::Int32(Some(150)),
+                false,
+            ),
+            (
+                Interval::make(Some(0), Some(100))?,
+                ScalarValue::Float64(Some(50.)),
+                true,
+            ),
+            (
+                Interval::make(Some(0), Some(100))?,
+                ScalarValue::Float64(Some(next_down(100.))),
+                true,
+            ),
+            (
+                Interval::make(Some(0), Some(100))?,
+                ScalarValue::Float64(Some(next_up(100.))),
+                false,
+            ),
+        ];
+
+        for (interval, value, expected) in possible_cases {
+            assert_eq!(interval.contains_value(value)?, expected)
+        }
+
+        Ok(())
+    }
+
     #[test]
     fn test_add() -> Result<()> {
         let cases = vec![

datafusion/physical-expr-common/src/lib.rs

@@ -28,5 +28,6 @@ pub mod binary_view_map;
 pub mod datum;
 pub mod physical_expr;
 pub mod sort_expr;
+pub mod stats_v2;
 pub mod tree_node;
 pub mod utils;

datafusion/physical-expr-common/src/physical_expr.rs

@@ -20,6 +20,7 @@ use std::fmt::{Debug, Display, Formatter};
 use std::hash::{Hash, Hasher};
 use std::sync::Arc;
 
+use crate::stats_v2::StatisticsV2;
 use crate::utils::scatter;
 
 use arrow::array::BooleanArray;
@@ -144,6 +145,27 @@ pub trait PhysicalExpr: Send + Sync + Display + Debug + DynEq + DynHash {
         Ok(Some(vec![]))
     }
 
+    /// Computes the final statistic of the expression by combining the input statistics
+    /// via a post-order, bottom-up traversal of the statistics graph.
+    fn evaluate_statistics(&self, _stats: &[&StatisticsV2]) -> Result<StatisticsV2> {
+        not_impl_err!("Not implemented for {self}")
+    }
+
+    /// Updates children statistics using the given parent statistic for this expression.
+    /// This is used to propagate constraints down through an expression tree.
+    ///
+    /// If statistics propagation reveals an infeasibility for any child, returns
+    /// [`None`]. If none of the children statistics change as a result of propagation,
+    /// may return an empty vector instead of cloning `children`. This is the default
+    /// (and conservative) return value.
+    fn propagate_statistics(
+        &self,
+        _parent_stat: &StatisticsV2,
+        _children_stat: &[&StatisticsV2],
+    ) -> Result<Option<Vec<StatisticsV2>>> {
+        Ok(Some(vec![]))
+    }
+
     /// Calculates the properties of this [`PhysicalExpr`] based on its
     /// children's properties (i.e. order and range), recursively aggregating
     /// the information from its children. In cases where the [`PhysicalExpr`]
@@ -155,7 +177,7 @@ pub trait PhysicalExpr: Send + Sync + Display + Debug + DynEq + DynHash {
 }
 
 /// [`PhysicalExpr`] can't be constrained by [`Eq`] directly because it must remain object
-/// safe. To ease implementation blanket implementation is provided for [`Eq`] types.
+/// safe. To ease implementation, blanket implementation is provided for [`Eq`] types.
 pub trait DynEq {
     fn dyn_eq(&self, other: &dyn Any) -> bool;
 }