This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-25517-923d64fd5448121f538612a7173c99108c108ae7 in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 993b3f3231abc5a8fc96e34656bbd881cca18e26 Author: Liang-Chi Hsieh <[email protected]> AuthorDate: Sun Sep 20 20:26:45 2026 +0000 fix: preserve statistics for safe casts without extrema (#25517) ## Which issue does this PR close? N/A ## Rationale for this change Projection statistics currently infer a CAST's source type from its minimum or maximum value. When both extrema are absent, even a lossless widening cast such as `Int32` to `Int64` is treated as unsafe. This discards otherwise valid statistics, including exact null counts and distinct counts, and can reduce the quality of optimizer estimates. ## What changes are included in this PR? - Add `ProjectionExprs::project_statistics_with_input_schema`, which uses the input schema to identify lossless casts when typed extrema are unavailable. - Preserve the existing conservative behavior for casts that may fail, discard values, or change ordering. - Use the schema-aware method in `ProjectionExec`, extended statistics projection, and file scan statistics projection. - Add helper-level and `ProjectionExec` regression tests covering an `Int32` to `Int64` cast with absent extrema but exact null and distinct counts. The existing `project_statistics` API remains available, so this is an additive, non-breaking API change. ## What is the testing strategy for this PR? Added: - `test_project_statistics_safe_cast_without_extrema` - `test_projection_statistics_safe_cast_without_extrema` Ablation testing confirmed that the old implementation changes `null_count = Exact(3)` to `Absent`. Both the helper-level and production `ProjectionExec` paths preserve the exact null and distinct counts after the fix. Also verified: - Targeted regression tests - `cargo fmt --all -- --check` - `git diff --check` The full test suite is covered by CI. ## Are there any user-facing changes? Optimizer statistics are now preserved through known-lossless casts even when minimum and maximum statistics are unavailable. This PR adds a public method but does not remove or change the existing API. --- datafusion/datasource/src/file_scan_config/mod.rs | 16 ++-- datafusion/physical-expr/src/projection.rs | 86 +++++++++++++++++++--- .../physical-plan/src/operator_statistics/mod.rs | 6 +- datafusion/physical-plan/src/projection.rs | 40 +++++++++- datafusion/sqllogictest/test_files/explain.slt | 11 +++ 5 files changed, 138 insertions(+), 21 deletions(-) diff --git a/datafusion/datasource/src/file_scan_config/mod.rs b/datafusion/datasource/src/file_scan_config/mod.rs index 4e72b5e83b..61b41fd1e9 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 6a038b429b..b7196e3841 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<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,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::<Column>() { 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::<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 @@ -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 b19f4e5fd4..07a0fa49d6 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 9ec9b2ab1d..89fe640f88 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 e349064406..de69f17085 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; --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
