fix: preserve statistics for safe casts without extrema

datafusion/datasource/src/file_scan_config/mod.rs

@@ -957,9 +957,11 @@ impl DataSource for FileScanConfig {
                 // Project the statistics based on the projection
                 let output_schema = self.projected_schema()?;
                 return if let Some(projection) = self.file_source.projection() {
-                    Ok(Arc::new(
-                        projection.project_statistics(stat.clone(), &output_schema)?,
-                    ))
+                    Ok(Arc::new(projection.project_statistics_with_input_schema(
+                        stat.clone(),
+                        self.file_source.table_schema().table_schema(),
+                        &output_schema,
+                    )?))
                 } else {
                     Ok(Arc::new(stat.clone()))
                 };
@@ -974,9 +976,11 @@ impl DataSource for FileScanConfig {
             let projection = self.file_source.projection();
             let output_schema = self.projected_schema()?;
             if let Some(projection) = &projection {
-                Ok(Arc::new(
-                    projection.project_statistics(statistics.clone(), &output_schema)?,
-                ))
+                Ok(Arc::new(projection.project_statistics_with_input_schema(
+                    statistics.clone(),
+                    self.file_source.table_schema().table_schema(),
+                    &output_schema,
+                )?))
             } else {
                 Ok(Arc::new(statistics))
             }

datafusion/physical-expr/src/projection.rs

@@ -710,8 +710,28 @@ impl ProjectionExprs {
     /// }
     /// ```
     pub fn project_statistics(
+        &self,
+        stats: Statistics,
+        output_schema: &Schema,
+    ) -> Result<Statistics> {
+        self.project_statistics_impl(stats, None, output_schema)
+    }
+
+    /// Projects `stats` using `input_schema` to identify safe casts even when
+    /// the input statistics do not contain typed minimum or maximum values.
+    pub fn project_statistics_with_input_schema(
+        &self,
+        stats: Statistics,
+        input_schema: &Schema,
+        output_schema: &Schema,
+    ) -> Result<Statistics> {
+        self.project_statistics_impl(stats, Some(input_schema), output_schema)
+    }
+
+    fn project_statistics_impl(
         &self,
         mut stats: Statistics,
+        input_schema: Option<&Schema>,
         output_schema: &Schema,
     ) -> Result<Statistics> {
         let mut column_statistics = Vec::with_capacity(self.exprs.len());
@@ -775,6 +795,7 @@ impl ProjectionExprs {
                 project_column_statistics_through_expr(
                     expr.as_ref(),
                     &stats.column_statistics,
+                    input_schema,
                 )
             };
             column_statistics.push(col_stats);
@@ -846,28 +867,37 @@ impl ProjectionExprs {
 fn project_column_statistics_through_expr(
     expr: &dyn PhysicalExpr,
     column_stats: &[ColumnStatistics],
+    input_schema: Option<&Schema>,
 ) -> ColumnStatistics {
     if let Some(col) = expr.downcast_ref::<Column>() {
         return column_statistics_at(column_stats, col.index());
     }
     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 inner_stats = project_column_statistics_through_expr(
+        cast_expr.expr.as_ref(),
+        column_stats,
+        input_schema,
+    );
     let target_type = cast_expr.cast_type();
+    let schema_source_type =
+        input_schema.and_then(|schema| cast_expr.expr.data_type(schema).ok());
 
     // A cast whose source values are already of the target `DataType` never
     // changes any value -- see `cast_array_by_name`'s same-type fast path in
     // `ColumnarValue::cast_to`. In that case every statistic, not just
     // min/max, carries over unchanged (this is what a cast that only
     // re-stamps a column's nullability, as `UnionExec`/`InterleaveExec`
     // insert, looks like here).
-    let already_target_type = matches!(
-        (inner_stats.min_value.get_value(), inner_stats.max_value.get_value()),
-        (Some(min), Some(max))
-            if min.data_type() == *target_type && max.data_type() == *target_type
-    );
+    let already_target_type = schema_source_type
+        .as_ref()
+        .is_some_and(|source_type| source_type == target_type)
+        || matches!(
+            (inner_stats.min_value.get_value(), inner_stats.max_value.get_value()),
+            (Some(min), Some(max))
+                if min.data_type() == *target_type && max.data_type() == *target_type
+        );
     if already_target_type {
         return inner_stats;
     }
@@ -880,11 +910,13 @@ fn project_column_statistics_through_expr(
         .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);
+    let source_type = schema_source_type.or_else(|| {
+        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
@@ -2290,6 +2322,36 @@ pub(crate) mod tests {
         Ok(())
     }
 
+    #[test]
+    fn test_project_statistics_safe_cast_without_extrema() {
+        let input_schema = Schema::new(vec![Field::new("a", DataType::Int32, true)]);
+        let mut stats = Statistics::new_unknown(&input_schema);
+        stats.num_rows = Precision::Exact(5);
+        stats.column_statistics[0].null_count = Precision::Exact(3);
+        stats.column_statistics[0].distinct_count = Precision::Exact(2);
+        let projection = ProjectionExprs::new(vec![ProjectionExpr::new(
+            Arc::new(CastExpr::new(
+                Arc::new(Column::new("a", 0)),
+                DataType::Int64,
+                None,
+            )),
+            "a",
+        )]);
+        let output_schema = projection
+            .project_schema(&input_schema)
+            .expect("valid projection schema");
+
+        let output = projection
+            .project_statistics_with_input_schema(stats, &input_schema, &output_schema)
+            .expect("statistics projection succeeds");
+
+        assert_eq!(output.column_statistics[0].null_count, Precision::Exact(3));
+        assert_eq!(
+            output.column_statistics[0].distinct_count,
+            Precision::Exact(2)
+        );
+    }
+
     #[test]
     fn test_project_statistics_non_monotonic_cast() {
         let input_schema = Schema::new(vec![Field::new("a", DataType::Utf8, false)]);

datafusion/physical-plan/src/operator_statistics/mod.rs

@@ -674,7 +674,11 @@ impl StatisticsProvider for ProjectionStatisticsProvider {
         // so expression-level NDV/min/max feeds into projected column stats.
         let stats = proj
             .projection_expr()
-            .project_statistics(input_stats, &output_schema)?;
+            .project_statistics_with_input_schema(
+                input_stats,
+                proj.input().schema().as_ref(),
+                &output_schema,
+            )?;
         Ok(StatisticsResult::Computed(ExtendedStatistics::new(stats)))
     }
 }

datafusion/physical-plan/src/projection.rs

@@ -492,7 +492,11 @@ impl ExecutionPlan for ProjectionExec {
         Ok(Arc::new(
             self.projector
                 .projection()
-                .project_statistics(input_stats, &output_schema)?,
+                .project_statistics_with_input_schema(
+                    input_stats,
+                    self.input.schema().as_ref(),
+                    &output_schema,
+                )?,
         ))
     }
 
@@ -1570,7 +1574,8 @@ mod tests {
     use datafusion_functions::core::arrow_metadata::ArrowMetadataFunc;
     use datafusion_physical_expr::ScalarFunctionExpr;
     use datafusion_physical_expr::expressions::{
-        BinaryExpr, Column, DynamicFilterPhysicalExpr, Literal, binary, col, is_null, lit,
+        BinaryExpr, CastExpr, Column, DynamicFilterPhysicalExpr, Literal, binary, col,
+        is_null, lit,
     };
 
     #[test]
@@ -2148,6 +2153,37 @@ mod tests {
         assert!(stats.total_byte_size.is_exact().unwrap_or(false));
     }
 
+    #[test]
+    fn test_projection_statistics_safe_cast_without_extrema() {
+        let input_schema = Schema::new(vec![Field::new("a", DataType::Int32, true)]);
+        let mut input_statistics = Statistics::new_unknown(&input_schema);
+        input_statistics.column_statistics[0].null_count = Precision::Exact(3);
+        input_statistics.column_statistics[0].distinct_count = Precision::Exact(2);
+        let input = Arc::new(StatisticsExec::new(input_statistics, input_schema));
+        let projection = ProjectionExec::try_new(
+            vec![ProjectionExpr::new(
+                Arc::new(CastExpr::new(
+                    Arc::new(Column::new("a", 0)),
+                    DataType::Int64,
+                    None,
+                )),
+                "a",
+            )],
+            input,
+        )
+        .unwrap();
+
+        let stats = StatisticsContext::new()
+            .compute(&projection, &StatisticsArgs::new())
+            .unwrap();
+
+        assert_eq!(stats.column_statistics[0].null_count, Precision::Exact(3));
+        assert_eq!(
+            stats.column_statistics[0].distinct_count,
+            Precision::Exact(2)
+        );
+    }
+
     #[test]
     fn test_filter_pushdown_with_alias() -> Result<()> {
         let input_schema = Schema::new(vec![Field::new("a", DataType::Int32, false)]);

datafusion/sqllogictest/test_files/explain.slt

@@ -283,6 +283,17 @@ set datafusion.explain.show_statistics = true;
 statement ok
 set datafusion.explain.physical_plan_only = true;
 
+# A safe cast preserves exact null counts even without min/max statistics.
+statement ok
+CREATE TABLE cast_statistics(a INT) AS VALUES (1), (NULL);
+
+query TT
+EXPLAIN SELECT CAST(a AS BIGINT) FROM cast_statistics;
+----
+physical_plan
+01)ProjectionExec: expr=[CAST(a@0 AS Int64) as cast_statistics.a], statistics=[Rows=Exact(2), Bytes=Exact(16), [(Col[0]: Null=Exact(1))]]
+02)--DataSourceExec: partitions=1, partition_sizes=[1], statistics=[Rows=Exact(2), Bytes=Exact(176), [(Col[0]: Null=Exact(1))]]
+
 # CSV scan with empty statistics
 query TT
 EXPLAIN SELECT a, b, c FROM simple_explain_test limit 10;