Use generic evaluate_bounds for statistics propagation instead of CAST-specific code

Replace the CAST-specific statistics propagation with a generic approach
using PhysicalExpr::evaluate_bounds(). This works for any expression
that implements evaluate_bounds — including CAST, negation, and
arithmetic with literals.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: Dandandan

datafusion/physical-expr/src/projection.rs

@@ -21,7 +21,7 @@ use std::ops::Deref;
 use std::sync::Arc;
 
 use crate::PhysicalExpr;
-use crate::expressions::{CastExpr, Column, Literal};
+use crate::expressions::{Column, Literal};
 use crate::scalar_function::ScalarFunctionExpr;
 use crate::utils::collect_columns;
 
@@ -33,6 +33,7 @@ use datafusion_common::{
     Result, ScalarValue, Statistics, assert_or_internal_err, internal_datafusion_err,
     plan_err,
 };
+use datafusion_expr_common::interval_arithmetic::Interval;
 
 use datafusion_physical_expr_common::metrics::ExecutionPlanMetricsSet;
 use datafusion_physical_expr_common::metrics::ExpressionEvaluatorMetrics;
@@ -729,35 +730,41 @@ impl ProjectionExprs {
     }
 }
 
-/// Propagate column statistics through expressions that preserve order.
+/// Propagate min/max statistics through an expression using the
+/// [`PhysicalExpr::evaluate_bounds`] interval arithmetic system.
 ///
-/// Currently handles:
-/// - `Column` references (direct passthrough)
-/// - `CAST` expressions (casts min/max values to the target type)
-///
-/// For other expressions, returns unknown statistics.
+/// This works for any expression that implements `evaluate_bounds`,
+/// including `CAST`, negation, and arithmetic with literals.
+/// For expressions that don't implement it, returns unknown statistics.
 fn project_column_statistics_through_expr(
     expr: &dyn PhysicalExpr,
     column_stats: &mut [ColumnStatistics],
 ) -> ColumnStatistics {
     if let Some(col) = expr.as_any().downcast_ref::<Column>() {
-        std::mem::take(&mut column_stats[col.index()])
-    } else if let Some(cast_expr) = expr.as_any().downcast_ref::<CastExpr>() {
-        let inner_stats =
-            project_column_statistics_through_expr(cast_expr.expr.as_ref(), column_stats);
-        let target_type = cast_expr.cast_type();
+        return std::mem::take(&mut column_stats[col.index()]);
+    }
+
+    if let Some(interval) = compute_bounds_from_stats(expr, column_stats) {
+        let all_exact = leaf_stats_all_exact(expr, column_stats);
+        let lower = interval.lower().clone();
+        let upper = interval.upper().clone();
         ColumnStatistics {
-            min_value: inner_stats
-                .min_value
-                .cast_to(target_type)
-                .unwrap_or(Precision::Absent),
-            max_value: inner_stats
-                .max_value
-                .cast_to(target_type)
-                .unwrap_or(Precision::Absent),
-            null_count: inner_stats.null_count,
-            distinct_count: inner_stats.distinct_count,
-            // Sum and byte size change under CAST, don't propagate
+            min_value: if lower.is_null() {
+                Precision::Absent
+            } else if all_exact {
+                Precision::Exact(lower)
+            } else {
+                Precision::Inexact(lower)
+            },
+            max_value: if upper.is_null() {
+                Precision::Absent
+            } else if all_exact {
+                Precision::Exact(upper)
+            } else {
+                Precision::Inexact(upper)
+            },
+            null_count: Precision::Absent,
+            distinct_count: Precision::Absent,
             sum_value: Precision::Absent,
             byte_size: Precision::Absent,
         }
@@ -766,6 +773,61 @@ fn project_column_statistics_through_expr(
     }
 }
 
+/// Recursively compute the output [`Interval`] for an expression from column
+/// statistics, using [`PhysicalExpr::evaluate_bounds`].
+fn compute_bounds_from_stats(
+    expr: &dyn PhysicalExpr,
+    column_stats: &[ColumnStatistics],
+) -> Option<Interval> {
+    if let Some(col) = expr.as_any().downcast_ref::<Column>() {
+        let stats = &column_stats[col.index()];
+        let min = stats.min_value.get_value()?;
+        let max = stats.max_value.get_value()?;
+        return Interval::try_new(min.clone(), max.clone()).ok();
+    }
+
+    if let Some(lit) = expr.as_any().downcast_ref::<Literal>() {
+        let val = lit.value();
+        return Interval::try_new(val.clone(), val.clone()).ok();
+    }
+
+    let children = expr.children();
+    let child_intervals: Option<Vec<Interval>> = children
+        .iter()
+        .map(|c| compute_bounds_from_stats(c.as_ref(), column_stats))
+        .collect();
+    let child_intervals = child_intervals?;
+    let child_refs: Vec<&Interval> = child_intervals.iter().collect();
+
+    expr.evaluate_bounds(&child_refs).ok()
+}
+
+/// Returns true if all leaf column statistics referenced by the expression
+/// have exact min/max values.
+fn leaf_stats_all_exact(
+    expr: &dyn PhysicalExpr,
+    column_stats: &[ColumnStatistics],
+) -> bool {
+    if let Some(col) = expr.as_any().downcast_ref::<Column>() {
+        return column_stats[col.index()]
+            .min_value
+            .is_exact()
+            .unwrap_or(false)
+            && column_stats[col.index()]
+                .max_value
+                .is_exact()
+                .unwrap_or(false);
+    }
+
+    if expr.as_any().downcast_ref::<Literal>().is_some() {
+        return true;
+    }
+
+    expr.children()
+        .iter()
+        .all(|c| leaf_stats_all_exact(c.as_ref(), column_stats))
+}
+
 impl<'a> IntoIterator for &'a ProjectionExprs {
     type Item = &'a ProjectionExpr;
     type IntoIter = std::slice::Iter<'a, ProjectionExpr>;
@@ -2824,13 +2886,17 @@ pub(crate) mod tests {
         // Should have 2 column statistics
         assert_eq!(output_stats.column_statistics.len(), 2);
 
-        // First column (expression) should have unknown statistics
+        // First column (col0 + 1) should have propagated min/max via evaluate_bounds
         assert_eq!(
-            output_stats.column_statistics[0].distinct_count,
-            Precision::Absent
+            output_stats.column_statistics[0].min_value,
+            Precision::Exact(ScalarValue::Int64(Some(-3)))
         );
         assert_eq!(
             output_stats.column_statistics[0].max_value,
+            Precision::Exact(ScalarValue::Int64(Some(22)))
+        );
+        assert_eq!(
+            output_stats.column_statistics[0].distinct_count,
             Precision::Absent
         );