diff --git a/datafusion/physical-expr/src/expressions/cast.rs b/datafusion/physical-expr/src/expressions/cast.rs index 845b9608d94f..0d87ec0900bd 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,55 @@ mod tests { Ok(()) } + #[test] + fn test_int32_date32_cast_preserves_values_and_ordering() { + 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).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 0e8876f01737..d041bb67dd4c 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,100 @@ 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 = 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_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 ade96e128ce3..d4933ee8de45 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;