This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion.git
The following commit(s) were added to refs/heads/main by this push:
new 993b3f3231 fix: preserve statistics for safe casts without extrema
(#25517)
993b3f3231 is described below
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]
