diff --git a/datafusion/datasource/src/file_scan_config/mod.rs b/datafusion/datasource/src/file_scan_config/mod.rs index 4e72b5e83bd23..61b41fd1e95a9 100644 --- a/datafusion/datasource/src/file_scan_config/mod.rs +++ b/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)) } diff --git a/datafusion/physical-expr/src/projection.rs b/datafusion/physical-expr/src/projection.rs index 6a038b429b26c..b7196e3841e0c 100644 --- a/datafusion/physical-expr/src/projection.rs +++ b/datafusion/physical-expr/src/projection.rs @@ -710,8 +710,28 @@ impl ProjectionExprs { /// } /// ``` pub fn project_statistics( + &self, + stats: Statistics, + output_schema: &Schema, + ) -> Result { + 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 { + 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 { 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,6 +867,7 @@ 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::() { return column_statistics_at(column_stats, col.index()); @@ -853,9 +875,14 @@ fn project_column_statistics_through_expr( let Some(cast_expr) = expr.downcast_ref::() 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 @@ -863,11 +890,14 @@ fn project_column_statistics_through_expr( // 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)]); diff --git a/datafusion/physical-plan/src/operator_statistics/mod.rs b/datafusion/physical-plan/src/operator_statistics/mod.rs index b19f4e5fd4693..07a0fa49d64b3 100644 --- a/datafusion/physical-plan/src/operator_statistics/mod.rs +++ b/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))) } } diff --git a/datafusion/physical-plan/src/projection.rs b/datafusion/physical-plan/src/projection.rs index 9ec9b2ab1d786..89fe640f889b6 100644 --- a/datafusion/physical-plan/src/projection.rs +++ b/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)]); diff --git a/datafusion/sqllogictest/test_files/explain.slt b/datafusion/sqllogictest/test_files/explain.slt index e3490644068e1..de69f17085b1c 100644 --- a/datafusion/sqllogictest/test_files/explain.slt +++ b/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;