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-24000-bc8b1a7a9323f69970fe728c95f3930cfbd5ad41 in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 9eb31bfd1621cf9512c32dbf401467ff7ae32648 Author: Alessandro Solimando <[email protected]> AuthorDate: Tue Aug 4 22:12:01 2026 +0200 fix(physical-plan): preserve Exact(0) in FilterExec for null_count, distinct_count and total_byte_size upon empty input (#24000) ## Which issue does this PR close? - Part of #8227 ## Rationale for this change Follow-up to #23936 (suggested [here](https://github.com/apache/datafusion/pull/23936#discussion_r3672861490)). #23936 made `FilterExec` report `num_rows: Exact(0)` for a provably empty result. Two places in the same operator still turn the same proof into an estimate: - `cap_at_rows` demotes `null_count` and `distinct_count` unconditionally - `total_byte_size` is computed after the "infeasibility branch", so a contradictory predicate (`a = 1 AND a = 2`) gives `Exact(0)` for `num_rows` and per-column `byte_size` but an inexact `total_byte_size` ## What changes are included in this PR? - `cap_at_rows` now maps an `Exact(0)` row bound to `Exact(0)` - `statistics_helper` reports `total_byte_size: Exact(0)` for a contradictory predicate instead of routing it through the selectivity estimate ## Are these changes tested? Yes, both by extending `test_filter_statistics_preserves_exactly_empty_input` with two assertions. Both assertions were confirmed to fail without the respective fix. No sqllogictest baselines change. ## Are there any user-facing changes? No breaking changes and no signature changes. `FilterExec` reports exact rather than inexact zeros for these statistics, visible in `EXPLAIN` output that shows statistics. ---- Disclaimer: I used AI to assist in the code generation, I have manually reviewed the output and it matches my intention and understanding. --- datafusion/physical-plan/src/filter.rs | 159 +++++++++++++++++++++++++++++---- 1 file changed, 141 insertions(+), 18 deletions(-) diff --git a/datafusion/physical-plan/src/filter.rs b/datafusion/physical-plan/src/filter.rs index 511be0bbdd..50c8246b37 100644 --- a/datafusion/physical-plan/src/filter.rs +++ b/datafusion/physical-plan/src/filter.rs @@ -320,7 +320,8 @@ impl FilterExec { /// The estimated output row count is used to keep the per-column statistics /// consistent with it: /// - null and distinct counts are capped at the estimated row count; - /// - byte sizes (per column and total) are scaled by the selectivity; + /// - byte sizes (per column and total) are scaled by the selectivity, and + /// are an exact zero when the row count is an exact zero; /// - a column constrained to a single value (`col = literal`, or an /// interval that collapses to one point) gets a distinct count of 1; /// - a column in a null-rejecting conjunct gets a null count of 0. @@ -384,8 +385,11 @@ impl FilterExec { input_num_rows.with_estimated_selectivity(selectivity); let mut cs = input_stats.to_inexact().column_statistics; for (idx, col_stat) in cs.iter_mut().enumerate() { - col_stat.byte_size = - col_stat.byte_size.with_estimated_selectivity(selectivity); + col_stat.byte_size = scale_byte_size_at_rows( + col_stat.byte_size, + selectivity, + filtered_num_rows, + ); col_stat.null_count = if null_rejecting_columns.contains(&idx) { Precision::Exact(0) } else { @@ -402,7 +406,7 @@ impl FilterExec { }; let total_byte_size = - input_total_byte_size.with_estimated_selectivity(selectivity); + scale_byte_size_at_rows(input_total_byte_size, selectivity, num_rows); Ok(Statistics { num_rows, @@ -1032,19 +1036,35 @@ fn interval_bound_to_precision( } /// Caps a row-bounded column statistic (a null count or distinct count) at the -/// filtered row estimate, since a column cannot have more nulls or distinct -/// values than it has rows. Known counts are demoted to inexact because the -/// filtered row count is itself an estimate. +/// filtered row count, since a column cannot have more nulls or distinct values +/// than it has rows. Known counts are demoted to inexact because a +/// filter-derived row bound is normally an estimate, the exception being an +/// exact zero, which proves the column is empty. fn cap_at_rows( value: Precision<usize>, filtered_num_rows: Precision<usize>, ) -> Precision<usize> { match filtered_num_rows { Precision::Absent => value.to_inexact(), + Precision::Exact(0) => Precision::Exact(0), rows => value.to_inexact().min(&rows), } } +/// Scales a byte size by the filter selectivity. An exact zero row count means +/// the output is exactly empty, so the byte size is an exact zero too. +fn scale_byte_size_at_rows( + byte_size: Precision<usize>, + selectivity: f64, + filtered_num_rows: Precision<usize>, +) -> Precision<usize> { + if filtered_num_rows == Precision::Exact(0) { + Precision::Exact(0) + } else { + byte_size.with_estimated_selectivity(selectivity) + } +} + /// Returns the NDV for a column constrained to one non-null value (e.g. /// `column = literal` or a singleton interval), derived from the filtered row /// estimate: zero rows means zero distinct values, a known positive row count @@ -1124,9 +1144,11 @@ fn collect_new_statistics( } else { cap_at_rows(input_column_stats[idx].null_count, filtered_num_rows) }; - let byte_size = input_column_stats[idx] - .byte_size - .with_estimated_selectivity(selectivity); + let byte_size = scale_byte_size_at_rows( + input_column_stats[idx].byte_size, + selectivity, + filtered_num_rows, + ); ColumnStatistics { null_count: capped_null_count, max_value, @@ -2904,16 +2926,29 @@ mod tests { #[tokio::test] async fn test_filter_statistics_preserves_exactly_empty_input() -> Result<()> { // A satisfiable predicate over an exactly empty input: the filter cannot - // produce rows, so the whole estimate stays exact. - let schema = Schema::new(vec![Field::new("a", DataType::Int32, true)]); + // produce rows, so the whole estimate stays exact. Column `b` is not + // mentioned by the predicate, so its null and distinct counts go through + // the generic row cap. + let schema = Schema::new(vec![ + Field::new("a", DataType::Int32, true), + Field::new("b", DataType::Int32, true), + ]); let input_stats = Statistics { num_rows: Precision::Exact(0), total_byte_size: Precision::Exact(0), - column_statistics: vec![ColumnStatistics { - null_count: Precision::Exact(0), - byte_size: Precision::Exact(0), - ..Default::default() - }], + column_statistics: vec![ + ColumnStatistics { + null_count: Precision::Exact(0), + byte_size: Precision::Exact(0), + ..Default::default() + }, + ColumnStatistics { + null_count: Precision::Exact(3), + distinct_count: Precision::Exact(7), + byte_size: Precision::Exact(0), + ..Default::default() + }, + ], }; let predicate = Arc::new(BinaryExpr::new( Arc::new(Column::new("a", 0)), @@ -2921,7 +2956,7 @@ mod tests { Arc::new(Literal::new(ScalarValue::Int32(Some(5)))), )); - let input = Arc::new(StatisticsExec::new(input_stats, schema)); + let input = Arc::new(StatisticsExec::new(input_stats, schema.clone())); let filter: Arc<dyn ExecutionPlan> = Arc::new(FilterExec::try_new(predicate, input)?); let statistics = @@ -2933,6 +2968,94 @@ mod tests { statistics.column_statistics[0].byte_size, Precision::Exact(0) ); + assert_eq!( + statistics.column_statistics[1].null_count, + Precision::Exact(0) + ); + assert_eq!( + statistics.column_statistics[1].distinct_count, + Precision::Exact(0) + ); + + // A contradictory predicate (`a = 1 AND a = 2`) discards all rows, the + // output is empty independently of the input. + let input = Arc::new(StatisticsExec::new( + Statistics { + num_rows: Precision::Inexact(1000), + total_byte_size: Precision::Inexact(8000), + column_statistics: vec![ColumnStatistics::new_unknown(); 2], + }, + schema, + )); + let contradiction = Arc::new(BinaryExpr::new( + Arc::new(BinaryExpr::new( + Arc::new(Column::new("a", 0)), + Operator::Eq, + Arc::new(Literal::new(ScalarValue::Int32(Some(1)))), + )), + Operator::And, + Arc::new(BinaryExpr::new( + Arc::new(Column::new("a", 0)), + Operator::Eq, + Arc::new(Literal::new(ScalarValue::Int32(Some(2)))), + )), + )); + let filter: Arc<dyn ExecutionPlan> = + Arc::new(FilterExec::try_new(contradiction, input)?); + let statistics = + StatisticsContext::new().compute(filter.as_ref(), &StatisticsArgs::new())?; + + assert_eq!(statistics.num_rows, Precision::Exact(0)); + assert_eq!(statistics.total_byte_size, Precision::Exact(0)); + + Ok(()) + } + + #[tokio::test] + async fn test_filter_statistics_exact_empty_input_zeroes_byte_size() -> Result<()> { + let cases = [ + ("absent", Precision::Absent, Precision::Absent), + ("inexact", Precision::Inexact(8000), Precision::Inexact(400)), + ]; + + for (desc, input_total_byte_size, input_byte_size) in cases { + let schema = Schema::new(vec![Field::new("a", DataType::Int32, true)]); + let input_stats = Statistics { + num_rows: Precision::Exact(0), + total_byte_size: input_total_byte_size, + column_statistics: vec![ColumnStatistics { + byte_size: input_byte_size, + ..Default::default() + }], + }; + let predicate = Arc::new(BinaryExpr::new( + Arc::new(Column::new("a", 0)), + Operator::Gt, + Arc::new(Literal::new(ScalarValue::Int32(Some(5)))), + )); + + let input = Arc::new(StatisticsExec::new(input_stats, schema)); + let filter: Arc<dyn ExecutionPlan> = + Arc::new(FilterExec::try_new(predicate, input)?); + let statistics = StatisticsContext::new() + .compute(filter.as_ref(), &StatisticsArgs::new())?; + + assert_eq!( + statistics.num_rows, + Precision::Exact(0), + "case '{desc}': num_rows mismatch" + ); + assert_eq!( + statistics.total_byte_size, + Precision::Exact(0), + "case '{desc}': total_byte_size mismatch" + ); + assert_eq!( + statistics.column_statistics[0].byte_size, + Precision::Exact(0), + "case '{desc}': byte_size mismatch" + ); + } Ok(()) } --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
