Resolve MIN/MAX aggregates from Parquet metadata for Single-mode and CAST projections

Two changes enable the AggregateStatistics optimizer to resolve
MIN/MAX (and COUNT) from file metadata instead of scanning data:

1. `take_optimizable` now handles `Single`/`SinglePartitioned`
   aggregate modes, not just `Final` wrapping `Partial`. This matters
   for single-partition scans where the planner skips the
   Partial→Final split.

2. `project_statistics` now propagates min/max statistics through
   CAST expressions (recursively), so that projections like
   `CAST(CAST(col AS Int32) AS Date32)` preserve the underlying
   column statistics instead of returning unknown.

Together these turn ClickBench Q6
(`SELECT MIN("EventDate"), MAX("EventDate") FROM hits`) from a full
column scan into a metadata-only lookup (PlaceholderRowExec).

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::{Column, Literal};
+use crate::expressions::{CastExpr, Column, Literal};
 use crate::scalar_function::ScalarFunctionExpr;
 use crate::utils::collect_columns;
 
@@ -715,9 +715,11 @@ impl ProjectionExprs {
                     }
                 }
             } else {
-                // TODO stats: estimate more statistics from expressions
-                // (expressions should compute their statistics themselves)
-                ColumnStatistics::new_unknown()
+                // Try to propagate statistics through expressions like CAST
+                project_column_statistics_through_expr(
+                    expr.as_ref(),
+                    &mut stats.column_statistics,
+                )
             };
             column_statistics.push(col_stats);
         }
@@ -727,6 +729,43 @@ impl ProjectionExprs {
     }
 }
 
+/// Propagate column statistics through expressions that preserve order.
+///
+/// Currently handles:
+/// - `Column` references (direct passthrough)
+/// - `CAST` expressions (casts min/max values to the target type)
+///
+/// For other expressions, 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();
+        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
+            sum_value: Precision::Absent,
+            byte_size: Precision::Absent,
+        }
+    } else {
+        ColumnStatistics::new_unknown()
+    }
+}
+
 impl<'a> IntoIterator for &'a ProjectionExprs {
     type Item = &'a ProjectionExpr;
     type IntoIter = std::slice::Iter<'a, ProjectionExpr>;

datafusion/physical-optimizer/src/aggregate_statistics.rs

@@ -20,7 +20,9 @@ use datafusion_common::Result;
 use datafusion_common::config::ConfigOptions;
 use datafusion_common::scalar::ScalarValue;
 use datafusion_common::tree_node::{Transformed, TransformedResult, TreeNode};
-use datafusion_physical_plan::aggregates::{AggregateExec, AggregateInputMode};
+use datafusion_physical_plan::aggregates::{
+    AggregateExec, AggregateInputMode, AggregateMode,
+};
 use datafusion_physical_plan::placeholder_row::PlaceholderRowExec;
 use datafusion_physical_plan::projection::{ProjectionExec, ProjectionExpr};
 use datafusion_physical_plan::udaf::{AggregateFunctionExpr, StatisticsArgs};
@@ -49,7 +51,7 @@ impl PhysicalOptimizerRule for AggregateStatistics {
         plan: Arc<dyn ExecutionPlan>,
         config: &ConfigOptions,
     ) -> Result<Arc<dyn ExecutionPlan>> {
-        if let Some(partial_agg_exec) = take_optimizable(&*plan) {
+        if let Some(partial_agg_exec) = take_optimizable(&plan) {
             let partial_agg_exec = partial_agg_exec
                 .downcast_ref::<AggregateExec>()
                 .expect("take_optimizable() ensures that this is a AggregateExec");
@@ -106,19 +108,37 @@ impl PhysicalOptimizerRule for AggregateStatistics {
     }
 }
 
-/// assert if the node passed as argument is a final `AggregateExec` node that can be optimized:
-/// - its child (with possible intermediate layers) is a partial `AggregateExec` node
-/// - they both have no grouping expression
+/// Returns an `AggregateExec` whose statistics can be used to replace the
+/// entire aggregate with literal values, if the plan is eligible.
 ///
-/// If this is the case, return a ref to the partial `AggregateExec`, else `None`.
-/// We would have preferred to return a casted ref to AggregateExec but the recursion requires
-/// the `ExecutionPlan.children()` method that returns an owned reference.
-fn take_optimizable(node: &dyn ExecutionPlan) -> Option<Arc<dyn ExecutionPlan>> {
-    if let Some(final_agg_exec) = node.downcast_ref::<AggregateExec>()
-        && final_agg_exec.mode().input_mode() == AggregateInputMode::Partial
-        && final_agg_exec.group_expr().is_empty()
+/// Two patterns are recognized:
+///
+/// 1. **Final wrapping Partial** (multi-partition): A final `AggregateExec`
+///    (input mode = `Partial`) with no GROUP BY whose descendant is a partial
+///    `AggregateExec` (input mode = `Raw`) with no GROUP BY and no filters.
+///    Returns the inner partial aggregate.
+///
+/// 2. **Single / SinglePartitioned** (single-partition): A `Single` or
+///    `SinglePartitioned` `AggregateExec` with no GROUP BY and no filters.
+///    Returns the aggregate itself.
+fn take_optimizable(plan: &Arc<dyn ExecutionPlan>) -> Option<Arc<dyn ExecutionPlan>> {
+    let agg_exec = plan.downcast_ref::<AggregateExec>()?;
+
+    // Case 1: Single-mode aggregate — processes raw input, produces final output
+    if matches!(
+        agg_exec.mode(),
+        AggregateMode::Single | AggregateMode::SinglePartitioned
+    ) && agg_exec.group_expr().is_empty()
+        && agg_exec.filter_expr().iter().all(|e| e.is_none())
+    {
+        return Some(Arc::clone(plan));
+    }
+
+    // Case 2: Final aggregate wrapping a Partial aggregate
+    if agg_exec.mode().input_mode() == AggregateInputMode::Partial
+        && agg_exec.group_expr().is_empty()
     {
-        let mut child = Arc::clone(final_agg_exec.input());
+        let mut child = Arc::clone(agg_exec.input());
         loop {
             if let Some(partial_agg_exec) = child.downcast_ref::<AggregateExec>()
                 && partial_agg_exec.mode().input_mode() == AggregateInputMode::Raw

datafusion/sqllogictest/test_files/clickbench.slt

@@ -192,8 +192,8 @@ logical_plan
 03)----Projection: CAST(CAST(hits_raw.EventDate AS Int32) AS Date32) AS EventDate
 04)------TableScan: hits_raw projection=[EventDate]
 physical_plan
-01)AggregateExec: mode=Single, gby=[], aggr=[min(hits.EventDate), max(hits.EventDate)]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/clickbench_hits_10.parquet]]}, projection=[CAST(CAST(EventDate@5 AS Int32) AS Date32) as EventDate], file_type=parquet
+01)ProjectionExec: expr=[2013-07-15 as min(hits.EventDate), 2013-07-15 as max(hits.EventDate)]
+02)--PlaceholderRowExec
 
 query DD
 SELECT MIN("EventDate"), MAX("EventDate") FROM hits;