You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/05/18 19:20:03 UTC
(datafusion) 01/01: Resolve MIN/MAX from Parquet metadata for Single-mode aggregates and CAST projections (#21651)
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 <da...@gmail.com>
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) <no...@anthropic.com>
---
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: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org