This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-21651-9a3d96acffdd3cd23cb484966984f8a794feb840 in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 2a2a060a93dfc0548415a68a7a192eed2e076f0e Author: Daniël Heres <[email protected]> AuthorDate: Mon May 18 21:19:40 2026 +0200 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) <[email protected]> --- datafusion/physical-expr/src/projection.rs | 76 ++++++++++++++++++++-- .../physical-optimizer/src/aggregate_statistics.rs | 37 +++++++---- datafusion/sqllogictest/test_files/aggregate.slt | 41 ++++++++++++ datafusion/sqllogictest/test_files/clickbench.slt | 4 +- 4 files changed, 137 insertions(+), 21 deletions(-) diff --git a/datafusion/physical-expr/src/projection.rs b/datafusion/physical-expr/src/projection.rs index 62c4d42570..ca999479fa 100644 --- a/datafusion/physical-expr/src/projection.rs +++ b/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(); diff --git a/datafusion/physical-optimizer/src/aggregate_statistics.rs b/datafusion/physical-optimizer/src/aggregate_statistics.rs index 75da187326..d0be53d59b 100644 --- a/datafusion/physical-optimizer/src/aggregate_statistics.rs +++ b/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 diff --git a/datafusion/sqllogictest/test_files/aggregate.slt b/datafusion/sqllogictest/test_files/aggregate.slt index e56eb9ef38..25b69d16dd 100644 --- a/datafusion/sqllogictest/test_files/aggregate.slt +++ b/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; diff --git a/datafusion/sqllogictest/test_files/clickbench.slt b/datafusion/sqllogictest/test_files/clickbench.slt index c79701e347..91463c9c2b 100644 --- a/datafusion/sqllogictest/test_files/clickbench.slt +++ b/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; --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
