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/09/19 09:38:35 UTC
(datafusion) branch gh-readonly-queue/main/pr-25227-d477a2b038730647bef258cb98e58978b0c74699 created (now ccfe704806)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-25227-d477a2b038730647bef258cb98e58978b0c74699
in repository https://gitbox.apache.org/repos/asf/datafusion.git
at ccfe704806 fix: only propagate cast statistics through safe conversions (#25227)
This branch includes the following new commits:
new ccfe704806 fix: only propagate cast statistics through safe conversions (#25227)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion) 01/01: fix: only propagate cast statistics through safe conversions (#25227)
Posted by gi...@apache.org.
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-25227-d477a2b038730647bef258cb98e58978b0c74699
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit ccfe70480630762da96c2698bceaad9d919b786c
Author: Huaijin <ha...@gmail.com>
AuthorDate: Sat Sep 19 09:38:11 2026 +0000
fix: only propagate cast statistics through safe conversions (#25227)
## Which issue does this PR close?
- Closes #25221.
## Rationale for this change
`MIN`/`MAX` over a cast can return incorrect results when aggregate
optimization replaces the scan with converted column statistics. Casting
the original endpoints is only valid if they remain extrema in the
target domain.
For example, a Parquet string column containing `('1', '100', '2')` has
string extrema `'1'` and `'2'`. Previously, `MIN(CAST(a AS INT)),
MAX(CAST(a AS INT))` could return `1, 2` from those statistics instead
of the correct `1, 100`. The same issue affects `BIGINT`.
## What changes are included in this PR?
- Restrict CAST statistics propagation to conversions proven safe for
the source type or its exact integer bounds. Unsupported or unsafe
conversions return unknown column statistics, so aggregates evaluate the
data.
- Preserve the existing same-type passthrough and supported lossless
widening conversions.
- Retain integer narrowing and signed/unsigned conversions when both
original bounds are exact and non-null, and both converted endpoints are
non-null. Successful endpoint conversions establish that the entire
integer range fits in the target type; overflow errors or null endpoints
do not qualify.
- Extend `CastExpr::check_bigger_cast` to recognize `Int32 ↔ Date32`,
and clarify that its contract includes lossless, strictly
order-preserving conversions that preserve nulls. Arrow reinterprets the
same `i32` values as days since the epoch, so these conversions preserve
both statistics and ordering properties. This also retains the
statistics-based `MIN`/`MAX` optimization for ClickBench's `UInt16 →
Int32 → Date32` projection.
## What is the testing strategy for this PR?
- Add projection unit tests for non-monotonic string-to-integer casts
and integer narrowing with safe, overflowing, and inexact bounds.
Removing the statistics guard makes the new string-cast regression test
fail.
- Add end-to-end Parquet regression cases in `parquet_statistics.slt`
for `INT` and `BIGINT`, both expecting `1, 100`.
- Add a bidirectional `Int32 ↔ Date32` test covering `i32::MIN`,
`i32::MAX`, negative values, zero, NULL, and strict ordering properties.
- Run the related projection and cast unit tests, plus the ClickBench
and Parquet statistics SLT files. The existing ClickBench plan
expectation passes unchanged, confirming that its aggregate still folds
to constants.
Local formatting, Clippy, license, spelling, workflow, and documentation
checks also pass.
## Are there any user-facing changes?
Affected casts now produce correct `MIN`/`MAX` results. Conversions
whose safety is not established may require a scan instead of using
statistics-based aggregate optimization. Supported safe integer
conversions and `Int32 ↔ Date32` retain that optimization.
---
datafusion/physical-expr/src/expressions/cast.rs | 73 +++++-
datafusion/physical-expr/src/projection.rs | 248 ++++++++++++++++++++-
.../sqllogictest/test_files/parquet_statistics.slt | 87 ++++++++
3 files changed, 395 insertions(+), 13 deletions(-)
diff --git a/datafusion/physical-expr/src/expressions/cast.rs b/datafusion/physical-expr/src/expressions/cast.rs
index 8e49b6ff48..9a2e4ab648 100644
--- a/datafusion/physical-expr/src/expressions/cast.rs
+++ b/datafusion/physical-expr/src/expressions/cast.rs
@@ -248,8 +248,12 @@ impl CastExpr {
})
}
- /// Check if casting from the specified source type to the target type is a
- /// widening cast (e.g. from `Int8` to `Int16`).
+ /// Check if casting from the source type to the target type is known to be
+ /// lossless and strictly order-preserving for all source values, preserving nulls.
+ /// This includes widening casts (e.g. `Int8` to `Int16`) and representation
+ /// conversions such as `Int32` to `Date32`, which interprets the same integer
+ /// as days since the epoch, or `Int64` to `Date64`, which interprets the same
+ /// integer as milliseconds since the epoch.
pub fn check_bigger_cast(cast_type: &DataType, src: &DataType) -> bool {
if cast_type.eq(src) {
return true;
@@ -259,6 +263,10 @@ impl CastExpr {
(Int8, Int16 | Int32 | Int64)
| (Int16, Int32 | Int64)
| (Int32, Int64)
+ | (Int32, Date32)
+ | (Date32, Int32)
+ | (Int64, Date64)
+ | (Date64, Int64)
| (UInt8, UInt16 | UInt32 | UInt64)
| (UInt16, UInt32 | UInt64)
| (UInt32, UInt64)
@@ -268,7 +276,8 @@ impl CastExpr {
)
}
- /// Check if the cast is a widening cast (e.g. from `Int8` to `Int16`).
+ /// Check if the cast is lossless and strictly order-preserving for all source
+ /// values, preserving nulls. See [`Self::check_bigger_cast`].
pub fn is_bigger_cast(&self, src: &DataType) -> bool {
Self::check_bigger_cast(self.cast_type(), src)
}
@@ -289,8 +298,8 @@ pub(crate) fn cast_expr_properties(
) -> Result<ExprProperties> {
let unbounded = Interval::make_unbounded(target_type)?;
let source_type = child.range.data_type();
- // A widening cast is additionally one-to-one, so it is strictly
- // order-preserving; a narrowing cast may collapse distinct values,
+ // A lossless cast recognized by check_bigger_cast is one-to-one, so it is
+ // strictly order-preserving; a narrowing cast may collapse distinct values,
// breaking the ordering of subsequent sort keys.
let bigger_cast = CastExpr::check_bigger_cast(target_type, &source_type);
if is_order_preserving_cast_family(&source_type, target_type) || bigger_cast {
@@ -1505,6 +1514,60 @@ mod tests {
Ok(())
}
+ #[test]
+ fn test_integer_date_cast_preserves_values_and_ordering() {
+ use arrow::array::{Date32Array, Date64Array};
+ use arrow::compute::SortOptions;
+ use datafusion_expr_common::sort_properties::SortProperties;
+
+ let values = vec![None, Some(i32::MIN), Some(-1), Some(0), Some(i32::MAX)];
+ let integers: ArrayRef = Arc::new(Int32Array::from(values.clone()));
+ let dates: ArrayRef = Arc::new(Date32Array::from(values));
+ let values = vec![None, Some(i64::MIN), Some(-1), Some(0), Some(i64::MAX)];
+ let integers64: ArrayRef = Arc::new(Int64Array::from(values.clone()));
+ let dates64: ArrayRef = Arc::new(Date64Array::from(values));
+ for (input, expected) in [
+ (Arc::clone(&integers), Arc::clone(&dates)),
+ (dates, integers),
+ (Arc::clone(&integers64), Arc::clone(&dates64)),
+ (dates64, integers64),
+ ] {
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "a",
+ input.data_type().clone(),
+ true,
+ )]));
+ let expr = CastExpr::new(
+ col("a", &schema).expect("column exists"),
+ expected.data_type().clone(),
+ None,
+ );
+ assert!(expr.is_bigger_cast(input.data_type()));
+ let child = ExprProperties::new_unknown()
+ .with_range(
+ Interval::make_unbounded(input.data_type())
+ .expect("supported interval type"),
+ )
+ .with_order(SortProperties::Ordered(SortOptions::default()))
+ .with_strictly_order_preserving(true);
+ let properties = expr
+ .get_properties(std::slice::from_ref(&child))
+ .expect("cast properties");
+ assert_eq!(properties.sort_properties, child.sort_properties);
+ assert!(properties.strictly_order_preserving);
+ assert_eq!(properties.range.data_type(), *expected.data_type());
+
+ let batch =
+ RecordBatch::try_new(schema, vec![input]).expect("valid input batch");
+ let actual = expr
+ .evaluate(&batch)
+ .expect("cast succeeds")
+ .into_array(batch.num_rows())
+ .expect("array result");
+ assert_eq!(actual.as_ref(), expected.as_ref());
+ }
+ }
+
#[test]
fn test_check_bigger_cast_precision_loss() {
use DataType::*;
diff --git a/datafusion/physical-expr/src/projection.rs b/datafusion/physical-expr/src/projection.rs
index 0e8876f017..6a038b429b 100644
--- a/datafusion/physical-expr/src/projection.rs
+++ b/datafusion/physical-expr/src/projection.rs
@@ -872,15 +872,39 @@ fn project_column_statistics_through_expr(
return inner_stats;
}
+ let min_value = inner_stats
+ .min_value
+ .cast_to(target_type)
+ .unwrap_or(Precision::Absent);
+ let max_value = inner_stats
+ .max_value
+ .cast_to(target_type)
+ .unwrap_or(Precision::Absent);
+ let source_type = inner_stats
+ .min_value
+ .get_value()
+ .or_else(|| inner_stats.max_value.get_value())
+ .map(ScalarValue::data_type);
+ // Copy extrema only for casts that preserve order and cannot discard values
+ // or fail within the input domain. Copying string endpoints into a numeric
+ // domain, for example, does not bound the converted column. Merely casting
+ // a failing endpoint to NULL also cannot establish the remaining extrema.
+ let preserves_values = source_type.is_some_and(|source_type| {
+ CastExpr::check_bigger_cast(target_type, &source_type)
+ || is_within_extrema(
+ &inner_stats.min_value,
+ &inner_stats.max_value,
+ &min_value,
+ &max_value,
+ )
+ });
+ if !preserves_values {
+ return ColumnStatistics::new_unknown();
+ }
+
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),
+ min_value,
+ max_value,
null_count: inner_stats.null_count,
distinct_count: inner_stats.distinct_count,
sum_value: Precision::Absent,
@@ -888,6 +912,23 @@ fn project_column_statistics_through_expr(
}
}
+/// Whether an integer cast preserves every value between the source extrema.
+/// The converted extrema must come from casting the corresponding source bounds.
+/// Exact, non-null integer bounds that both cast successfully prove the entire
+/// range fits in the target type. For example, Int64 [-128, 127] fits in Int8,
+/// whereas [-129, 127] does not. Inexact bounds cannot establish exact extrema.
+fn is_within_extrema(
+ lower: &Precision<ScalarValue>,
+ upper: &Precision<ScalarValue>,
+ min: &Precision<ScalarValue>,
+ max: &Precision<ScalarValue>,
+) -> bool {
+ [lower, upper, min, max].into_iter().all(|bound| {
+ matches!(bound, Precision::Exact(value)
+ if value.data_type().is_integer() && !value.is_null())
+ })
+}
+
fn column_statistics_at(
column_stats: &[ColumnStatistics],
index: usize,
@@ -2249,6 +2290,197 @@ pub(crate) mod tests {
Ok(())
}
+ #[test]
+ fn test_project_statistics_non_monotonic_cast() {
+ let input_schema = Schema::new(vec![Field::new("a", DataType::Utf8, false)]);
+ let mut stats = Statistics::new_unknown(&input_schema);
+ stats.num_rows = Precision::Exact(3);
+ stats.column_statistics[0].min_value = Precision::Exact(ScalarValue::from("1"));
+ stats.column_statistics[0].max_value = Precision::Exact(ScalarValue::from("2"));
+ let expr: Arc<dyn PhysicalExpr> = Arc::new(CastExpr::new(
+ Arc::new(Column::new("a", 0)),
+ DataType::Int32,
+ None,
+ ));
+ let projection = ProjectionExprs::new(vec![ProjectionExpr {
+ expr: Arc::clone(&expr),
+ alias: "x".to_string(),
+ }]);
+ let out = projection
+ .project_statistics(
+ stats,
+ &projection
+ .project_schema(&input_schema)
+ .expect("valid projection schema"),
+ )
+ .expect("statistics projection succeeds");
+ let batch = RecordBatch::try_new(
+ Arc::new(input_schema),
+ vec![Arc::new(arrow::array::StringArray::from(vec![
+ "1", "100", "2",
+ ]))],
+ )
+ .expect("valid input batch");
+ let actual = expr
+ .evaluate(&batch)
+ .expect("cast succeeds")
+ .into_array(3)
+ .expect("array result");
+ assert_eq!(out.column_statistics[0].max_value, Precision::Absent);
+ assert_eq!(out.column_statistics[0].min_value, Precision::Absent);
+ assert_eq!(
+ ScalarValue::try_from_array(&actual, 1).expect("valid scalar value"),
+ ScalarValue::Int32(Some(100))
+ );
+ }
+
+ #[test]
+ fn test_is_within_extrema() {
+ use Precision::{Absent, Exact, Inexact};
+ use ScalarValue::{Int8, Int64, UInt8};
+
+ for (lower, upper, target_type, expected) in [
+ (
+ Exact(Int64(Some(-128))),
+ Exact(Int64(Some(127))),
+ DataType::Int8,
+ true,
+ ),
+ (
+ Exact(Int64(Some(-129))),
+ Exact(Int64(Some(127))),
+ DataType::Int8,
+ false,
+ ),
+ (
+ Exact(Int64(Some(-128))),
+ Exact(Int64(Some(128))),
+ DataType::Int8,
+ false,
+ ),
+ (
+ Exact(Int64(Some(0))),
+ Exact(Int64(Some(255))),
+ DataType::UInt8,
+ true,
+ ),
+ (
+ Exact(Int64(Some(-1))),
+ Exact(Int64(Some(255))),
+ DataType::UInt8,
+ false,
+ ),
+ (
+ Exact(UInt8(Some(0))),
+ Exact(UInt8(Some(127))),
+ DataType::Int8,
+ true,
+ ),
+ (
+ Exact(UInt8(Some(0))),
+ Exact(UInt8(Some(128))),
+ DataType::Int8,
+ false,
+ ),
+ (
+ Exact(ScalarValue::from("1")),
+ Exact(ScalarValue::from("2")),
+ DataType::Int8,
+ false,
+ ),
+ (
+ Exact(Int64(Some(1))),
+ Exact(Int64(Some(2))),
+ DataType::Utf8,
+ false,
+ ),
+ ] {
+ let min = lower.cast_to(&target_type).unwrap_or(Absent);
+ let max = upper.cast_to(&target_type).unwrap_or(Absent);
+ assert_eq!(
+ is_within_extrema(&lower, &upper, &min, &max),
+ expected,
+ "{lower:?}..{upper:?} -> {target_type:?}",
+ );
+ }
+
+ // Every bound must be exact and non-null, including converted endpoints.
+ let bounds = [
+ Exact(Int64(Some(-128))),
+ Exact(Int64(Some(127))),
+ Exact(Int8(Some(-128))),
+ Exact(Int8(Some(127))),
+ ];
+ for index in 0..bounds.len() {
+ for invalid in [
+ Absent,
+ Inexact(bounds[index].get_value().unwrap().clone()),
+ Exact(
+ ScalarValue::try_from(
+ &bounds[index].get_value().unwrap().data_type(),
+ )
+ .unwrap(),
+ ),
+ ] {
+ let mut bounds = bounds.clone();
+ bounds[index] = invalid;
+ assert!(!is_within_extrema(
+ &bounds[0], &bounds[1], &bounds[2], &bounds[3],
+ ));
+ }
+ }
+ }
+
+ #[test]
+ fn test_project_statistics_narrowing_cast_requires_safe_bounds() {
+ let schema = Schema::new(vec![Field::new("a", DataType::Int32, true)]);
+ for (lower, upper, exact, safe) in [
+ (-100, 100, true, true),
+ (-200, 100, true, false),
+ (-100, 200, true, false),
+ (-100, 100, false, false),
+ ] {
+ let precision = |v| {
+ if exact {
+ Precision::Exact(ScalarValue::Int32(Some(v)))
+ } else {
+ Precision::Inexact(ScalarValue::Int32(Some(v)))
+ }
+ };
+ let mut stats = Statistics::new_unknown(&schema);
+ stats.column_statistics[0].min_value = precision(lower);
+ stats.column_statistics[0].max_value = precision(upper);
+ let projection = ProjectionExprs::new(vec![ProjectionExpr::new(
+ Arc::new(CastExpr::new(
+ Arc::new(Column::new("a", 0)),
+ DataType::Int8,
+ None,
+ )),
+ "x",
+ )]);
+ let output = projection
+ .project_statistics(
+ stats,
+ &projection
+ .project_schema(&schema)
+ .expect("valid projection schema"),
+ )
+ .expect("statistics projection succeeds");
+ if safe {
+ assert_eq!(
+ output.column_statistics[0].min_value,
+ Precision::Exact(ScalarValue::Int8(Some(-100)))
+ );
+ assert_eq!(
+ output.column_statistics[0].max_value,
+ Precision::Exact(ScalarValue::Int8(Some(100)))
+ );
+ } else {
+ assert_eq!(output.column_statistics[0], ColumnStatistics::new_unknown());
+ }
+ }
+ }
+
fn get_stats() -> Statistics {
Statistics {
num_rows: Precision::Exact(5),
diff --git a/datafusion/sqllogictest/test_files/parquet_statistics.slt b/datafusion/sqllogictest/test_files/parquet_statistics.slt
index ade96e128c..2c9a29c74b 100644
--- a/datafusion/sqllogictest/test_files/parquet_statistics.slt
+++ b/datafusion/sqllogictest/test_files/parquet_statistics.slt
@@ -186,6 +186,93 @@ physical_plan
statement ok
DROP TABLE typed_table;
+######
+# Aggregate statistics through casts
+######
+
+# String extrema do not bound the numeric values after a CAST. These aggregates
+# must not be replaced with casts of the Parquet string min/max statistics.
+statement ok
+SET datafusion.execution.target_partitions = 1;
+
+statement ok
+COPY (SELECT * FROM (VALUES ('1'), ('100'), ('2')) AS t(a))
+TO 'test_files/scratch/parquet_statistics/cast_strings.parquet' STORED AS PARQUET;
+
+statement ok
+CREATE EXTERNAL TABLE cast_stats_strings STORED AS PARQUET
+LOCATION 'test_files/scratch/parquet_statistics/cast_strings.parquet';
+
+query II
+SELECT MIN(CAST(a AS INT)), MAX(CAST(a AS INT)) FROM cast_stats_strings;
+----
+1 100
+
+query II
+SELECT MIN(CAST(a AS BIGINT)), MAX(CAST(a AS BIGINT)) FROM cast_stats_strings;
+----
+1 100
+
+statement ok
+DROP TABLE cast_stats_strings;
+
+# Exact Int64 extrema that fit in Int8 preserve the aggregate optimization.
+statement ok
+COPY (SELECT CAST(a AS BIGINT) AS a FROM (VALUES (-128), (0), (127), (NULL)) AS t(a))
+TO 'test_files/scratch/parquet_statistics/cast_ints.parquet' STORED AS PARQUET;
+
+statement ok
+CREATE EXTERNAL TABLE cast_stats_ints STORED AS PARQUET
+LOCATION 'test_files/scratch/parquet_statistics/cast_ints.parquet';
+
+query II
+SELECT MIN(CAST(a AS TINYINT)), MAX(CAST(a AS TINYINT)) FROM cast_stats_ints;
+----
+-128 127
+
+statement ok
+SET datafusion.explain.show_statistics = false;
+
+query TT
+EXPLAIN SELECT MIN(CAST(a AS TINYINT)), MAX(CAST(a AS TINYINT)) FROM cast_stats_ints;
+----
+physical_plan
+01)ProjectionExec: expr=[-128 as min(cast_stats_ints.a), 127 as max(cast_stats_ints.a)]
+02)--PlaceholderRowExec
+
+statement ok
+DROP TABLE cast_stats_ints;
+
+# An out-of-range minimum or maximum must prevent aggregate folding and preserve
+# the runtime cast error, even when the other endpoint fits in Int8.
+statement ok
+COPY (SELECT CAST(a AS BIGINT) AS a, CAST(b AS BIGINT) AS b
+ FROM (VALUES (-129, -128), (0, 0), (127, 128), (NULL, NULL)) AS t(a, b))
+TO 'test_files/scratch/parquet_statistics/cast_ints_overflow.parquet' STORED AS PARQUET;
+
+statement ok
+CREATE EXTERNAL TABLE cast_stats_ints_overflow STORED AS PARQUET
+LOCATION 'test_files/scratch/parquet_statistics/cast_ints_overflow.parquet';
+
+query error Cast error: Can't cast value -129 to type Int8
+SELECT MIN(CAST(a AS TINYINT)), MAX(CAST(a AS TINYINT)) FROM cast_stats_ints_overflow;
+
+query error Cast error: Can't cast value 128 to type Int8
+SELECT MIN(CAST(b AS TINYINT)), MAX(CAST(b AS TINYINT)) FROM cast_stats_ints_overflow;
+
+query TT
+EXPLAIN SELECT MIN(CAST(a AS TINYINT)), MAX(CAST(a AS TINYINT)) FROM cast_stats_ints_overflow;
+----
+physical_plan
+01)AggregateExec: mode=Single, gby=[], aggr=[min(cast_stats_ints_overflow.a), max(cast_stats_ints_overflow.a)]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_statistics/cast_ints_overflow.parquet]]}, projection=[CAST(a@0 AS Int8) as __common_expr_1], file_type=parquet
+
+statement ok
+DROP TABLE cast_stats_ints_overflow;
+
+statement ok
+SET datafusion.execution.target_partitions = 4;
+
# Config reset
statement ok
RESET datafusion.execution.collect_statistics;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org