From 5a356803c65d12b07a3c7bef1a1db95be6545b0a Mon Sep 17 00:00:00 2001 From: Huaijin Date: Sat, 12 Sep 2026 16:54:12 +0800 Subject: [PATCH 1/4] fix: only propagate cast statistics through safe conversions --- datafusion/physical-expr/src/projection.rs | 119 ++++++++++++++++-- .../test_files/cast_statistics.slt | 45 +++++++ 2 files changed, 156 insertions(+), 8 deletions(-) create mode 100644 datafusion/sqllogictest/test_files/cast_statistics.slt diff --git a/datafusion/physical-expr/src/projection.rs b/datafusion/physical-expr/src/projection.rs index 0e8876f017379..8a6296ecd8be0 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) + || (source_type.is_integer() && target_type.is_integer() + && matches!( + (&inner_stats.min_value, &inner_stats.max_value, &min_value, &max_value), + (Precision::Exact(lower), Precision::Exact(upper), Precision::Exact(min), Precision::Exact(max)) + if !lower.is_null() && !upper.is_null() && !min.is_null() && !max.is_null() + )) + }); + 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, @@ -2249,6 +2273,85 @@ pub(crate) mod tests { Ok(()) } + #[test] + fn test_project_statistics_non_monotonic_cast() -> Result<()> { + 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 = 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)?)?; + let batch = RecordBatch::try_new( + Arc::new(input_schema), + vec![Arc::new(arrow::array::StringArray::from(vec![ + "1", "100", "2", + ]))], + )?; + let actual = expr.evaluate(&batch)?.into_array(3)?; + 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)?, + ScalarValue::Int32(Some(100)) + ); + Ok(()) + } + + #[test] + fn test_project_statistics_narrowing_cast_requires_safe_bounds() -> Result<()> { + 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)?)?; + 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()); + } + } + Ok(()) + } + fn get_stats() -> Statistics { Statistics { num_rows: Precision::Exact(5), diff --git a/datafusion/sqllogictest/test_files/cast_statistics.slt b/datafusion/sqllogictest/test_files/cast_statistics.slt new file mode 100644 index 0000000000000..09bc58e7dffdc --- /dev/null +++ b/datafusion/sqllogictest/test_files/cast_statistics.slt @@ -0,0 +1,45 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +# 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/cast_statistics/strings.parquet' STORED AS PARQUET; + +statement ok +CREATE EXTERNAL TABLE cast_stats_strings STORED AS PARQUET +LOCATION 'test_files/scratch/cast_statistics/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; + +statement ok +SET datafusion.execution.target_partitions = 4; From 6f5c5e873cfacb0b328662a3582afbc2fd9bc978 Mon Sep 17 00:00:00 2001 From: Huaijin Date: Sat, 12 Sep 2026 20:39:39 +0800 Subject: [PATCH 2/4] test: consolidate cast statistics cases into parquet statistics suite --- .../test_files/cast_statistics.slt | 45 ------------------- .../test_files/parquet_statistics.slt | 33 ++++++++++++++ 2 files changed, 33 insertions(+), 45 deletions(-) delete mode 100644 datafusion/sqllogictest/test_files/cast_statistics.slt diff --git a/datafusion/sqllogictest/test_files/cast_statistics.slt b/datafusion/sqllogictest/test_files/cast_statistics.slt deleted file mode 100644 index 09bc58e7dffdc..0000000000000 --- a/datafusion/sqllogictest/test_files/cast_statistics.slt +++ /dev/null @@ -1,45 +0,0 @@ -# Licensed to the Apache Software Foundation (ASF) under one -# or more contributor license agreements. See the NOTICE file -# distributed with this work for additional information -# regarding copyright ownership. The ASF licenses this file -# to you under the Apache License, Version 2.0 (the -# "License"); you may not use this file except in compliance -# with the License. You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, -# software distributed under the License is distributed on an -# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY -# KIND, either express or implied. See the License for the -# specific language governing permissions and limitations -# under the License. - -# 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/cast_statistics/strings.parquet' STORED AS PARQUET; - -statement ok -CREATE EXTERNAL TABLE cast_stats_strings STORED AS PARQUET -LOCATION 'test_files/scratch/cast_statistics/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; - -statement ok -SET datafusion.execution.target_partitions = 4; diff --git a/datafusion/sqllogictest/test_files/parquet_statistics.slt b/datafusion/sqllogictest/test_files/parquet_statistics.slt index ade96e128ce3b..d4933ee8de457 100644 --- a/datafusion/sqllogictest/test_files/parquet_statistics.slt +++ b/datafusion/sqllogictest/test_files/parquet_statistics.slt @@ -186,6 +186,39 @@ 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; + +statement ok +SET datafusion.execution.target_partitions = 4; + # Config reset statement ok RESET datafusion.execution.collect_statistics; From b94929a87f74dbe194db8ccd528592491d96606f Mon Sep 17 00:00:00 2001 From: Huaijin Date: Sat, 12 Sep 2026 21:16:59 +0800 Subject: [PATCH 3/4] fix: preserve statistics and ordering for Int32 Date32 casts --- .../physical-expr/src/expressions/cast.rs | 53 +++++++++++++++++-- 1 file changed, 48 insertions(+), 5 deletions(-) diff --git a/datafusion/physical-expr/src/expressions/cast.rs b/datafusion/physical-expr/src/expressions/cast.rs index 845b9608d94f3..bee97634d7d16 100644 --- a/datafusion/physical-expr/src/expressions/cast.rs +++ b/datafusion/physical-expr/src/expressions/cast.rs @@ -249,8 +249,11 @@ 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. pub fn check_bigger_cast(cast_type: &DataType, src: &DataType) -> bool { if cast_type.eq(src) { return true; @@ -260,6 +263,8 @@ impl CastExpr { (Int8, Int16 | Int32 | Int64) | (Int16, Int32 | Int64) | (Int32, Int64) + | (Int32, Date32) + | (Date32, Int32) | (UInt8, UInt16 | UInt32 | UInt64) | (UInt16, UInt32 | UInt64) | (UInt32, UInt64) @@ -269,7 +274,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) } @@ -290,8 +296,8 @@ pub(crate) fn cast_expr_properties( ) -> Result { 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 { @@ -1506,6 +1512,43 @@ mod tests { Ok(()) } + #[test] + fn test_int32_date32_cast_preserves_values_and_ordering() -> Result<()> { + use arrow::array::Date32Array; + 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)); + for (input, expected) in [ + (Arc::clone(&integers), Arc::clone(&dates)), + (dates, integers), + ] { + let schema = Arc::new(Schema::new(vec![Field::new( + "a", + input.data_type().clone(), + true, + )])); + let expr = + CastExpr::new(col("a", &schema)?, 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())?) + .with_order(SortProperties::Ordered(SortOptions::default())) + .with_strictly_order_preserving(true); + let properties = expr.get_properties(std::slice::from_ref(&child))?; + 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])?; + let actual = expr.evaluate(&batch)?.into_array(batch.num_rows())?; + assert_eq!(actual.as_ref(), expected.as_ref()); + } + Ok(()) + } + #[test] fn test_check_bigger_cast_precision_loss() { use DataType::*; From 3cca46b038847391c22f56ef18f2482d739e3845 Mon Sep 17 00:00:00 2001 From: Huaijin Date: Sun, 13 Sep 2026 11:10:20 +0800 Subject: [PATCH 4/4] test: avoid uncovered error returns in cast statistics tests --- .../physical-expr/src/expressions/cast.rs | 28 +++++++++++----- datafusion/physical-expr/src/projection.rs | 33 ++++++++++++++----- 2 files changed, 44 insertions(+), 17 deletions(-) diff --git a/datafusion/physical-expr/src/expressions/cast.rs b/datafusion/physical-expr/src/expressions/cast.rs index bee97634d7d16..0d87ec0900bd7 100644 --- a/datafusion/physical-expr/src/expressions/cast.rs +++ b/datafusion/physical-expr/src/expressions/cast.rs @@ -1513,7 +1513,7 @@ mod tests { } #[test] - fn test_int32_date32_cast_preserves_values_and_ordering() -> Result<()> { + fn test_int32_date32_cast_preserves_values_and_ordering() { use arrow::array::Date32Array; use arrow::compute::SortOptions; use datafusion_expr_common::sort_properties::SortProperties; @@ -1530,23 +1530,35 @@ mod tests { input.data_type().clone(), true, )])); - let expr = - CastExpr::new(col("a", &schema)?, expected.data_type().clone(), None); + 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())?) + .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))?; + 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])?; - let actual = expr.evaluate(&batch)?.into_array(batch.num_rows())?; + 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()); } - Ok(()) } #[test] diff --git a/datafusion/physical-expr/src/projection.rs b/datafusion/physical-expr/src/projection.rs index 8a6296ecd8be0..d041bb67dd4cf 100644 --- a/datafusion/physical-expr/src/projection.rs +++ b/datafusion/physical-expr/src/projection.rs @@ -2274,7 +2274,7 @@ pub(crate) mod tests { } #[test] - fn test_project_statistics_non_monotonic_cast() -> Result<()> { + 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); @@ -2290,25 +2290,35 @@ pub(crate) mod tests { alias: "x".to_string(), }]); let out = projection - .project_statistics(stats, &projection.project_schema(&input_schema)?)?; + .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", ]))], - )?; - let actual = expr.evaluate(&batch)?.into_array(3)?; + ) + .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)?, + ScalarValue::try_from_array(&actual, 1).expect("valid scalar value"), ScalarValue::Int32(Some(100)) ); - Ok(()) } #[test] - fn test_project_statistics_narrowing_cast_requires_safe_bounds() -> Result<()> { + 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), @@ -2335,7 +2345,13 @@ pub(crate) mod tests { "x", )]); let output = projection - .project_statistics(stats, &projection.project_schema(&schema)?)?; + .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, @@ -2349,7 +2365,6 @@ pub(crate) mod tests { assert_eq!(output.column_statistics[0], ColumnStatistics::new_unknown()); } } - Ok(()) } fn get_stats() -> Statistics {