Resolve MIN/MAX from Parquet metadata for Single-mode aggregates and CAST projections (#21651)

## Which issue does this PR close?

Related to improving ClickBench performance (metadata-only query
resolution)

## Rationale for this change

ClickBench Q6 (`SELECT MIN("EventDate"), MAX("EventDate") FROM hits`)
was doing a full column scan despite the answer being available in
Parquet row group statistics. Two issues prevented the
`AggregateStatistics` optimizer from firing:

1. **`take_optimizable` missed `Single` mode** — it only matched the
`Final → Partial` pair, not single-partition scans.
2. **Statistics lost through CAST projections** — `project_statistics`
returned `unknown` for any non-Column/Literal expression, discarding
Parquet min/max through casts like `CAST(CAST(EventDate AS Int32) AS
Date32)`.

This now avoids scanning any columns, going from ~6ms to ~1.5ms
```
│ QQuery 6  │           5.12 / 6.29 ±0.83 / 7.65 ms │           1.26 / 1.43 ±0.26 / 1.93 ms │ +4.39x faster │
```

## What changes are included in this PR?

- **`aggregate_statistics.rs`**: `take_optimizable` now also matches
`Single`/`SinglePartitioned` aggregates.
- **`projection.rs`**: Added `project_column_statistics_through_expr()`
which propagates min/max statistics through `CastExpr`.

Result: Q6 now resolves entirely from Parquet metadata (zero I/O).

## Are these changes tested?

Yes — existing tests pass, ClickBench sqllogictest updated with new
expected plan for Q6.

## Are there any user-facing changes?

Scalar `MIN`/`MAX` aggregates over CAST projections now resolve from
file metadata when statistics are available, avoiding unnecessary I/O.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

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;
 
@@ -714,9 +714,10 @@ impl ProjectionExprs {
                     }
                 }
             } else {
-                // TODO stats: estimate more statistics from expressions
-                // (expressions should compute their statistics themselves)
-                ColumnStatistics::new_unknown()
+                project_column_statistics_through_expr(
+                    expr.as_ref(),
+                    &stats.column_statistics,
+                )
             };
             column_statistics.push(col_stats);
         }
@@ -726,6 +727,39 @@ impl ProjectionExprs {
     }
 }
 
+/// Propagate column statistics through CAST projections. Other expressions
+/// return unknown — generalizing via [`PhysicalExpr::evaluate_bounds`] is
+/// unsafe for aggregate folding since many impls (e.g. `sin`) return a fixed
+/// envelope rather than tight bounds on the actual inputs.
+fn project_column_statistics_through_expr(
+    expr: &dyn PhysicalExpr,
+    column_stats: &[ColumnStatistics],
+) -> ColumnStatistics {
+    if let Some(col) = expr.downcast_ref::<Column>() {
+        return column_stats[col.index()].clone();
+    }
+    let Some(cast_expr) = expr.downcast_ref::<CastExpr>() else {
+        return ColumnStatistics::new_unknown();
+    };
+    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_value: Precision::Absent,
+        byte_size: Precision::Absent,
+    }
+}
+
 impl<'a> IntoIterator for &'a ProjectionExprs {
     type Item = &'a ProjectionExpr;
     type IntoIter = std::slice::Iter<'a, ProjectionExpr>;
@@ -1256,7 +1290,7 @@ pub(crate) mod tests {
 
     use super::*;
     use crate::equivalence::{EquivalenceProperties, convert_to_orderings};
-    use crate::expressions::{BinaryExpr, col};
+    use crate::expressions::{BinaryExpr, CastExpr, col};
     use crate::utils::tests::TestScalarUDF;
     use crate::{PhysicalExprRef, ScalarFunctionExpr};
 
@@ -2791,6 +2825,38 @@ pub(crate) mod tests {
         Ok(())
     }
 
+    #[test]
+    fn test_project_statistics_with_cast() -> Result<()> {
+        let input_stats = get_stats();
+        let input_schema = get_schema();
+
+        // SELECT CAST(col0 AS Int32) AS casted
+        let projection = ProjectionExprs::new(vec![ProjectionExpr {
+            expr: Arc::new(CastExpr::new(
+                Arc::new(Column::new("col0", 0)),
+                DataType::Int32,
+                None,
+            )),
+            alias: "casted".to_string(),
+        }]);
+
+        let output_stats = projection.project_statistics(
+            input_stats,
+            &projection.project_schema(&input_schema)?,
+        )?;
+
+        assert_eq!(
+            output_stats.column_statistics[0].min_value,
+            Precision::Exact(ScalarValue::Int32(Some(-4)))
+        );
+        assert_eq!(
+            output_stats.column_statistics[0].max_value,
+            Precision::Exact(ScalarValue::Int32(Some(21)))
+        );
+
+        Ok(())
+    }
+
     #[test]
     fn test_project_statistics_primitive_width_only() -> Result<()> {
         let input_stats = get_stats();

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,26 @@ 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
-///
-/// 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()
+/// Returns an `AggregateExec` whose statistics can replace the aggregate with
+/// literal values: either a `Single`/`SinglePartitioned` aggregate, or a
+/// `Final` aggregate wrapping a `Partial`. Must have no GROUP BY and no
+/// filters.
+fn take_optimizable(plan: &Arc<dyn ExecutionPlan>) -> Option<Arc<dyn ExecutionPlan>> {
+    let agg_exec = plan.downcast_ref::<AggregateExec>()?;
+
+    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));
+    }
+
+    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/aggregate.slt

@@ -8977,3 +8977,44 @@ GROUP BY id ORDER BY id;
 
 statement ok
 DROP TABLE first_last_value_str_tests;
+
+# MIN/MAX over a CAST projection folds from parquet statistics via the
+# aggregate-statistics optimizer. Non-CAST projections (multi-column or
+# self-referenced expressions) must keep returning correct results — guard
+# against future regressions if the optimizer is broadened.
+statement ok
+SET datafusion.execution.target_partitions = 1;
+
+statement ok
+CREATE EXTERNAL TABLE hits_raw
+STORED AS PARQUET
+LOCATION '../core/tests/data/clickbench_hits_10.parquet';
+
+query II
+SELECT MIN(CAST("EventDate" AS BIGINT)), MAX(CAST("EventDate" AS BIGINT)) FROM hits_raw;
+----
+15901 15901
+
+query II
+SELECT MIN(delta), MAX(delta)
+FROM (
+  SELECT "UserID" - CAST("ClientIP" AS BIGINT) AS delta
+  FROM hits_raw
+);
+----
+-2461439044872611287 7418527518698834918
+
+query II
+SELECT MIN(zero), MAX(zero)
+FROM (
+  SELECT "UserID" - "UserID" AS zero
+  FROM hits_raw
+);
+----
+0 0
+
+statement ok
+SET datafusion.execution.target_partitions = 4;
+
+statement ok
+DROP TABLE hits_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;