fix: only propagate cast statistics through safe conversions

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<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 {
@@ -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::*;

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<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_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),

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;