Support bounds evaluation for temporal data types

datafusion-examples/examples/expr_api.rs

@@ -24,7 +24,7 @@ use arrow::record_batch::RecordBatch;
 use datafusion::arrow::datatypes::{DataType, Field, Schema, TimeUnit};
 use datafusion::common::stats::Precision;
 use datafusion::common::tree_node::{Transformed, TreeNode};
-use datafusion::common::{ColumnStatistics, DFSchema};
+use datafusion::common::{internal_datafusion_err, ColumnStatistics, DFSchema};
 use datafusion::common::{ScalarValue, ToDFSchema};
 use datafusion::error::Result;
 use datafusion::functions_aggregate::first_last::first_value_udaf;
@@ -264,6 +264,8 @@ fn range_analysis_demo() -> Result<()> {
     // Now, we invoke the analysis code to perform the range analysis
     let df_schema = DFSchema::try_from(schema)?;
     let physical_expr = SessionContext::new().create_physical_expr(expr, &df_schema)?;
+
+    assert!(physical_expr.supports_bounds_evaluation(df_schema.as_ref()));
     let analysis_result = analyze(
         &physical_expr,
         AnalysisContext::new(boundaries),
@@ -305,10 +307,17 @@ fn boundary_analysis_and_selectivity_demo() -> Result<()> {
         distinct_count: Precision::Absent,
     };
 
+    let field = schema.fields().first().ok_or_else(|| {
+        internal_datafusion_err!("schema does not have a field at index 0")
+    })?;
+
     // We can then build our expression boundaries from the column statistics
     // allowing the analysis to be more precise.
-    let initial_boundaries =
-        vec![ExprBoundaries::try_from_column(&schema, &column_stats, 0)?];
+    let initial_boundaries = vec![ExprBoundaries::try_from_column(
+        field.as_ref(),
+        &column_stats,
+        0,
+    )?];
 
     // With the above we can perform the boundary analysis similar to the previous
     // example.
@@ -317,6 +326,8 @@ fn boundary_analysis_and_selectivity_demo() -> Result<()> {
     // Analysis case id >= 5000
     let physical_expr =
         SessionContext::new().create_physical_expr(id_greater_5000, &df_schema)?;
+    assert!(physical_expr.supports_bounds_evaluation(df_schema.as_ref()));
+
     let analysis = analyze(
         &physical_expr,
         AnalysisContext::new(initial_boundaries.clone()),
@@ -359,7 +370,8 @@ fn boundary_analysis_in_conjuctions_demo() -> Result<()> {
     let age_between_18_25 = col("age").gt(lit(18i64)).and(col("age").lt_eq(lit(25)));
 
     // As always we need to tell DataFusion the type of the column.
-    let schema = Arc::new(Schema::new(vec![make_field("age", DataType::Int64)]));
+    let age_field = make_field("age", DataType::Int64);
+    let schema = Arc::new(Schema::new(vec![age_field.clone()]));
 
     // Similarly to the example in `boundary_analysis_and_selectivity_demo` we
     // can establish column statistics that can be used to describe certain
@@ -372,8 +384,11 @@ fn boundary_analysis_in_conjuctions_demo() -> Result<()> {
         distinct_count: Precision::Absent,
     };
 
-    let initial_boundaries =
-        vec![ExprBoundaries::try_from_column(&schema, &column_stats, 0)?];
+    let initial_boundaries = vec![ExprBoundaries::try_from_column(
+        &age_field,
+        &column_stats,
+        0,
+    )?];
 
     // Before we run the analysis pass; let us describe what we can infer from
     // the initial information.
@@ -394,6 +409,9 @@ fn boundary_analysis_in_conjuctions_demo() -> Result<()> {
 
     let physical_expr =
         SessionContext::new().create_physical_expr(age_between_18_25, &df_schema)?;
+
+    assert!(physical_expr.supports_bounds_evaluation(df_schema.as_ref()));
+
     let analysis = analyze(
         &physical_expr,
         // We re-use initial_boundaries elsewhere so we must clone it.
@@ -446,7 +464,11 @@ fn boundary_analysis_in_conjuctions_demo() -> Result<()> {
     let physical_expr = SessionContext::new()
         .create_physical_expr(age_greater_than_60_less_than_18, &df_schema)?;
 
-    // Since we don't handle interval arithmetic for `OR` operator this will error out.
+    // This check will return false since we don't handle interval arithmetic
+    // for `OR` operator.
+    assert!(!physical_expr.supports_bounds_evaluation(df_schema.inner()));
+
+    // In consequence, this will error out.
     let analysis = analyze(
         &physical_expr,
         AnalysisContext::new(initial_boundaries),

datafusion/common/src/scalar/mod.rs

@@ -1471,7 +1471,7 @@ impl ScalarValue {
     }
 
     /// Calculate arithmetic negation for a scalar value
-    pub fn arithmetic_negate(&self) -> Result<Self> {
+    pub fn negate(&self) -> Result<Self> {
         fn neg_checked_with_ctx<T: ArrowNativeTypeOp>(
             v: T,
             ctx: impl Fn() -> String,
@@ -1578,6 +1578,8 @@ impl ScalarValue {
                     tz.clone(),
                 ))
             }
+            ScalarValue::Boolean(None) => Ok(self.clone()),
+            ScalarValue::Boolean(Some(value)) => Ok(ScalarValue::Boolean(Some(!value))),
             value => _internal_err!(
                 "Can not run arithmetic negative on scalar value {value:?}"
             ),
@@ -4598,7 +4600,7 @@ mod tests {
         let try_into_value: i128 = decimal_value.clone().try_into().unwrap();
         assert_eq!(123_i128, try_into_value);
         assert!(!decimal_value.is_null());
-        let neg_decimal_value = decimal_value.arithmetic_negate()?;
+        let neg_decimal_value = decimal_value.negate()?;
         match neg_decimal_value {
             ScalarValue::Decimal128(v, _, _) => {
                 assert_eq!(-123, v.unwrap());
@@ -6334,15 +6336,17 @@ mod tests {
     fn test_scalar_negative() -> Result<()> {
         // positive test
         let value = ScalarValue::Int32(Some(12));
-        assert_eq!(ScalarValue::Int32(Some(-12)), value.arithmetic_negate()?);
+        assert_eq!(ScalarValue::Int32(Some(-12)), value.negate()?);
         let value = ScalarValue::Int32(None);
-        assert_eq!(ScalarValue::Int32(None), value.arithmetic_negate()?);
+        assert_eq!(ScalarValue::Int32(None), value.negate()?);
+        let value = ScalarValue::Boolean(Some(false));
+        assert_eq!(ScalarValue::Boolean(Some(true)), value.negate()?);
 
         // negative test
         let value = ScalarValue::UInt8(Some(12));
-        assert!(value.arithmetic_negate().is_err());
-        let value = ScalarValue::Boolean(None);
-        assert!(value.arithmetic_negate().is_err());
+        assert!(value.negate().is_err());
+        let value = ScalarValue::Binary(None);
+        assert!(value.negate().is_err());
         Ok(())
     }
 
@@ -6353,7 +6357,7 @@ mod tests {
             ($($val:expr),* $(,)?) => {$(
                 {
                     let value: ScalarValue = $val;
-                    let err = value.arithmetic_negate().expect_err("Should receive overflow error on negating {value:?}");
+                    let err = value.negate().expect_err("Should receive overflow error on negating {value:?}");
                     let root_err = err.find_root();
                     match  root_err{
                         DataFusionError::ArrowError(
@@ -6404,7 +6408,7 @@ mod tests {
         ];
         // skip float 16 because they aren't supported
         for (test, expected) in float_cases.into_iter().skip(2) {
-            assert_eq!(test.arithmetic_negate()?, expected);
+            assert_eq!(test.negate()?, expected);
         }
         Ok(())
     }
@@ -6424,7 +6428,7 @@ mod tests {
         ];
 
         for (test, expected) in cases {
-            assert_eq!(test.arithmetic_negate().unwrap(), expected);
+            assert_eq!(test.negate().unwrap(), expected);
         }
     }
 
@@ -6721,7 +6725,7 @@ mod tests {
             ),
         ];
         for (expr, expected) in cases.iter() {
-            let result = expr.arithmetic_negate().unwrap();
+            let result = expr.negate().unwrap();
             assert_eq!(*expected, result, "-expr:{expr:?}");
         }
     }

datafusion/common/src/stats.rs

@@ -158,7 +158,7 @@ impl Precision<usize> {
     /// rows are selected. A selectivity of `0.5` means half the rows are
     /// selected. Will always return inexact statistics.
     pub fn with_estimated_selectivity(self, selectivity: f64) -> Self {
-        self.map(|v| ((v as f64 * selectivity).ceil()) as usize)
+        self.map(|v| (v as f64 * selectivity).ceil() as usize)
             .to_inexact()
     }
 }

datafusion/core/tests/physical_optimizer/join_selection.rs

@@ -32,7 +32,6 @@ use datafusion_execution::{RecordBatchStream, SendableRecordBatchStream, TaskCon
 use datafusion_expr::Operator;
 use datafusion_physical_expr::expressions::col;
 use datafusion_physical_expr::expressions::{BinaryExpr, Column, NegativeExpr};
-use datafusion_physical_expr::intervals::utils::check_support;
 use datafusion_physical_expr::PhysicalExprRef;
 use datafusion_physical_expr::{EquivalenceProperties, Partitioning, PhysicalExpr};
 use datafusion_physical_optimizer::join_selection::{
@@ -1080,21 +1079,21 @@ fn check_expr_supported() {
         Operator::Plus,
         Arc::new(Column::new("a", 0)),
     )) as Arc<dyn PhysicalExpr>;
-    assert!(check_support(&supported_expr, &schema));
+    assert!(&supported_expr.supports_bounds_evaluation(&schema));
     let supported_expr_2 = Arc::new(Column::new("a", 0)) as Arc<dyn PhysicalExpr>;
-    assert!(check_support(&supported_expr_2, &schema));
+    assert!(&supported_expr_2.supports_bounds_evaluation(&schema));
     let unsupported_expr = Arc::new(BinaryExpr::new(
         Arc::new(Column::new("a", 0)),
         Operator::Or,
         Arc::new(Column::new("a", 0)),
     )) as Arc<dyn PhysicalExpr>;
-    assert!(!check_support(&unsupported_expr, &schema));
+    assert!(!&unsupported_expr.supports_bounds_evaluation(&schema));
     let unsupported_expr_2 = Arc::new(BinaryExpr::new(
         Arc::new(Column::new("a", 0)),
         Operator::Or,
         Arc::new(NegativeExpr::new(Arc::new(Column::new("a", 0)))),
     )) as Arc<dyn PhysicalExpr>;
-    assert!(!check_support(&unsupported_expr_2, &schema));
+    assert!(!&unsupported_expr_2.supports_bounds_evaluation(&schema));
 }
 
 struct TestCase {

datafusion/expr-common/Cargo.toml

@@ -42,3 +42,6 @@ datafusion-common = { workspace = true }
 indexmap = { workspace = true }
 itertools = { workspace = true }
 paste = "^1.0"
+
+[dev-dependencies]
+arrow-buffer = { workspace = true }