From 2f5a6e0e2357f45d3f7cda1bfa35ae8b298f5a28 Mon Sep 17 00:00:00 2001 From: Alessandro Solimando Date: Fri, 2 Oct 2026 12:31:04 +0200 Subject: [PATCH 1/6] refactor: deprecate statistics providers that duplicate operator estimates Rename `StatisticsRegistry::default_with_builtin_providers` to `with_experimental_providers` and deprecate the old name. The experimental set holds `FilterStatisticsProvider` and `JoinStatisticsProvider`. Both start from the operator's `statistics_from_inputs` and replace only the values their technique estimates. `JoinStatisticsProvider` only replaces the row count of multi-key inner equi-joins with the product of per-key distinct counts, and no longer falls back to the Cartesian product. `FilterStatisticsProvider` applies the survival model to the input distinct count, capped at the operator's. Deprecate the providers that duplicate operator estimates: Projection, Passthrough, Aggregate, Limit and Union. Closes #25571. --- benchmarks/src/statistics.rs | 2 +- .../examples/statistics/join_reorder.rs | 8 +- datafusion/ffi/src/physical_optimizer.rs | 2 +- .../src/operator_statistics/mod.rs | 742 ++++++++---------- datafusion/sqllogictest/src/test_context.rs | 32 +- .../test_files/statistics_registry.slt | 3 + .../library-user-guide/upgrading/56.0.0.md | 28 +- 7 files changed, 376 insertions(+), 441 deletions(-) diff --git a/benchmarks/src/statistics.rs b/benchmarks/src/statistics.rs index 6c14264356ef8..8981490919fc6 100644 --- a/benchmarks/src/statistics.rs +++ b/benchmarks/src/statistics.rs @@ -249,7 +249,7 @@ enum QError { } fn capture_statistics(plan: &dyn ExecutionPlan) -> Result> { - let statistics_context = StatisticsRegistry::default_with_builtin_providers(); + let statistics_context = StatisticsRegistry::with_experimental_providers(); let mut result = vec![]; capture_statistics_inner(plan, &statistics_context, "0", &mut result)?; Ok(result) diff --git a/datafusion-examples/examples/statistics/join_reorder.rs b/datafusion-examples/examples/statistics/join_reorder.rs index 0eccc93630af2..a825fd431977b 100644 --- a/datafusion-examples/examples/statistics/join_reorder.rs +++ b/datafusion-examples/examples/statistics/join_reorder.rs @@ -23,9 +23,9 @@ //! //! `(SELECT user_id FROM events WHERE amount < 50 GROUP BY user_id) JOIN dims`: a //! provider supplies the column stats a catalog knows (`amount` range, `user_id` -//! distinct count); the built-in `FilterStatisticsProvider` then refines the +//! distinct count); the experimental `FilterStatisticsProvider` then refines the //! post-filter distinct count with the survival formula -//! `NDV * (1 - (1 - selectivity)^(rows / NDV))` (Yao/Cardenas) to ~32, below `dims` +//! `NDV * (1 - (1 - selectivity)^(rows / NDV))` (Yao/Cardenas) to ~40, below `dims` //! (48), flipping the join build side. Core's simpler `min(NDV, rows)` cap would //! give 50 (> 48) and keep the other order; the refinement is the point. The //! ground-truth query prints the true surviving distinct count (below 48), @@ -111,7 +111,7 @@ fn build_ctx(with_registry: bool) -> Result { .with_config(config) .with_default_features(); if with_registry { - let mut registry = StatisticsRegistry::default_with_builtin_providers(); + let mut registry = StatisticsRegistry::with_experimental_providers(); registry.register(Arc::new(ClosureStatisticsProvider::with_matches( catalog_matches, catalog_stats, @@ -173,7 +173,7 @@ pub async fn join_reorder() -> Result<()> { ); println!("-- Without the registry (default estimation) --"); println!("{}\n", explain(&build_ctx(false)?).await?); - println!("-- With the registry (built-in refinement) --"); + println!("-- With the registry (experimental refinement) --"); println!("{}", explain(&build_ctx(true)?).await?); Ok(()) } diff --git a/datafusion/ffi/src/physical_optimizer.rs b/datafusion/ffi/src/physical_optimizer.rs index 87fe3314af9d6..8a01f916242a6 100644 --- a/datafusion/ffi/src/physical_optimizer.rs +++ b/datafusion/ffi/src/physical_optimizer.rs @@ -586,7 +586,7 @@ mod tests { let ctx = ContextWithRegistry { config: ConfigOptions::new(), - registry: StatisticsRegistry::default_with_builtin_providers(), + registry: StatisticsRegistry::with_experimental_providers(), }; let plan = create_test_plan(); diff --git a/datafusion/physical-plan/src/operator_statistics/mod.rs b/datafusion/physical-plan/src/operator_statistics/mod.rs index e09aa6b86a1a7..4b49a61b5eb74 100644 --- a/datafusion/physical-plan/src/operator_statistics/mod.rs +++ b/datafusion/physical-plan/src/operator_statistics/mod.rs @@ -37,17 +37,22 @@ //! - [`StatisticsRegistry`]: Chains providers, lives in SessionState //! - [`ExtendedStatistics`]: Statistics with type-safe custom extensions //! -//! # Built-in Providers +//! # Experimental Providers //! -//! The following providers are included and can be registered in this order: +//! These optional providers hold estimation techniques not yet ready to move +//! into the operator. A technique moves into the operator once it is shown to +//! improve estimates in general. Each provider starts from the operator's +//! [`ExecutionPlan::statistics_from_inputs`] and replaces only the values its +//! technique estimates, so improvements to the operator also reach it. Register +//! them with [`StatisticsRegistry::with_experimental_providers`]. //! -//! 1. [`FilterStatisticsProvider`] - selectivity-based filter estimation -//! 2. [`ProjectionStatisticsProvider`] - column mapping through projections -//! 3. [`PassthroughStatisticsProvider`] - passthrough for cardinality-preserving operators -//! 4. [`AggregateStatisticsProvider`] - NDV-based GROUP BY cardinality estimation -//! 5. [`JoinStatisticsProvider`] - NDV-based join output estimation (hash, sort-merge, cross) -//! 6. [`LimitStatisticsProvider`] - caps output at the fetch limit (local and global) -//! 7. [`UnionStatisticsProvider`] - sums input row counts +//! - [`FilterStatisticsProvider`] - adjusts the distinct count of each column +//! for the rows removed by the filter +//! - [`JoinStatisticsProvider`] - estimates multi-key inner equi-joins assuming +//! independent keys +//! +//! The remaining bundled providers are deprecated because they duplicate the +//! operators' own estimates. //! //! # Statistics walk //! @@ -81,13 +86,15 @@ use std::sync::Arc; use datafusion_common::extensions::Extensions; use datafusion_common::stats::Precision; -use datafusion_common::{Result, Statistics}; +use datafusion_common::{JoinType, Result, Statistics}; +use datafusion_physical_expr::PhysicalExpr; +use datafusion_physical_expr::expressions::Column; use crate::ExecutionPlan; use crate::aggregates::{AggregateExec, AggregateMode}; use crate::execution_plan::CardinalityEffect; use crate::filter::FilterExec; -use crate::joins::{CrossJoinExec, HashJoinExec, JoinOnRef, SortMergeJoinExec}; +use crate::joins::{HashJoinExec, SortMergeJoinExec}; use crate::limit::{GlobalLimitExec, LocalLimitExec}; use crate::projection::ProjectionExec; use crate::statistics::{ChildStats, StatisticsArgs, StatisticsContext}; @@ -392,7 +399,21 @@ impl StatisticsRegistry { Self { providers } } - /// Create a registry pre-loaded with the standard built-in providers. + /// Create a registry with the experimental providers (see the + /// [module documentation](crate::operator_statistics)). + /// + /// Provider order (first match wins): + /// 1. [`FilterStatisticsProvider`] + /// 2. [`JoinStatisticsProvider`] + pub fn with_experimental_providers() -> Self { + Self::with_providers(vec![ + Arc::new(FilterStatisticsProvider), + Arc::new(JoinStatisticsProvider), + ]) + } + + /// Create a registry with the experimental providers plus the deprecated + /// providers that duplicate the operators' own estimates. /// /// Provider order (first match wins): /// 1. [`FilterStatisticsProvider`] @@ -402,6 +423,11 @@ impl StatisticsRegistry { /// 5. [`JoinStatisticsProvider`] /// 6. [`LimitStatisticsProvider`] /// 7. [`UnionStatisticsProvider`] + #[deprecated( + since = "56.0.0", + note = "use `with_experimental_providers`, which registers only the providers that start from the operators' own estimates" + )] + #[expect(deprecated)] pub fn default_with_builtin_providers() -> Self { Self::with_providers(vec![ Arc::new(FilterStatisticsProvider), @@ -581,15 +607,13 @@ fn computed_with_row_count( Ok(StatisticsResult::Computed(ExtendedStatistics::new(base))) } -/// Statistics provider for [`FilterExec`] that uses -/// pre-computed enhanced child statistics from the registry walk. +/// Experimental statistics provider for [`FilterExec`]. /// -/// Unlike the built-in fallback (which calls `statistics_from_inputs` and gets raw -/// child stats), this provider receives enhanced child stats that may include -/// NDV overrides injected at the scan level. It applies the same selectivity -/// estimation logic as `FilterExec::statistics_helper`, then additionally -/// adjusts each column's `distinct_count` using [`ndv_after_selectivity`] based -/// on the computed selectivity ratio. +/// Starts from the operator's own [`ExecutionPlan::statistics_from_inputs`], +/// then lowers each column's `distinct_count` to [`ndv_after_selectivity`] of +/// the input column, using the ratio of output rows to input rows as the +/// selectivity. Columns with a distinct count of 0 or 1 keep the value the +/// operator returned. #[derive(Debug, Default)] pub struct FilterStatisticsProvider; @@ -610,35 +634,40 @@ impl StatisticsProvider for FilterStatisticsProvider { return Ok(StatisticsResult::Delegate); } - let input_stats = (*child_stats[0].base).clone(); - let input_rows = input_stats.num_rows; - let mut stats = FilterExec::statistics_helper( - &filter.input().schema(), - input_stats, - filter.predicate(), - filter.default_selectivity(), - // TODO: pass filter.expression_analyzer_registry() once #21122 lands - )?; + let input = child_stats[0].base(); + let input_rows = input.num_rows; + let child_base: Vec> = child_stats + .iter() + .map(|c| Arc::clone(c.base_arc())) + .collect(); + let mut stats = Arc::unwrap_or_clone( + plan.statistics_from_inputs(&child_base, &StatisticsArgs::new())?, + ); - // Adjust distinct_count for each column using the selectivity ratio - // via the probabilistic survival model from - // ndv_after_selectivity to account for rows removed by the filter. if let (Some(&orig_rows), Some(&filtered_rows)) = (input_rows.get_value(), stats.num_rows.get_value()) && orig_rows > 0 && filtered_rows < orig_rows { let selectivity = filtered_rows as f64 / orig_rows as f64; - for col_stat in &mut stats.column_statistics { - if let Some(&ndv) = col_stat.distinct_count.get_value() { - let adjusted = ndv_after_selectivity(ndv, orig_rows, selectivity); - col_stat.distinct_count = Precision::Inexact(adjusted); + for (i, col_stat) in stats.column_statistics.iter_mut().enumerate() { + let input_index = filter.projection().as_ref().map_or(i, |p| p[i]); + // The operator's distinct count is already capped at the + // filtered rows, so the survival model uses the input one + if let Some(&ndv) = col_stat.distinct_count.get_value() + && ndv > 1 + && let Some(&input_ndv) = input + .column_statistics + .get(input_index) + .and_then(|c| c.distinct_count.get_value()) + { + let adjusted = + ndv_after_selectivity(input_ndv, orig_rows, selectivity); + col_stat.distinct_count = Precision::Inexact(adjusted.min(ndv)); } } } - let stats = filter.statistics_with_fetch(stats, None)?; - let stats = stats.project(filter.projection().as_ref()); Ok(StatisticsResult::Computed(ExtendedStatistics::new(stats))) } } @@ -649,9 +678,14 @@ impl StatisticsProvider for FilterStatisticsProvider { /// Maps enhanced child column statistics to output columns based on the /// projection expressions, preserving NDV and other statistics through /// column references. +#[deprecated( + since = "56.0.0", + note = "duplicates the `statistics_from_inputs` of `ProjectionExec`; without a matching provider the statistics walk uses the operator's own estimate" +)] #[derive(Debug, Default)] pub struct ProjectionStatisticsProvider; +#[expect(deprecated)] impl StatisticsProvider for ProjectionStatisticsProvider { fn matches(&self, plan: &dyn ExecutionPlan) -> bool { plan.downcast_ref::().is_some() @@ -691,9 +725,14 @@ impl StatisticsProvider for ProjectionStatisticsProvider { /// transform statistics, so we pass through the enhanced child stats directly. /// This avoids the fallback calling `statistics_from_inputs` (overall) which would /// trigger a redundant internal recursion with raw (non-enhanced) stats. +#[deprecated( + since = "56.0.0", + note = "duplicates the `statistics_from_inputs` of the cardinality-preserving operators; without a matching provider the statistics walk uses the operator's own estimate" +)] #[derive(Debug, Default)] pub struct PassthroughStatisticsProvider; +#[expect(deprecated)] impl StatisticsProvider for PassthroughStatisticsProvider { fn matches(&self, plan: &dyn ExecutionPlan) -> bool { plan.children().len() == 1 @@ -742,9 +781,14 @@ impl StatisticsProvider for PassthroughStatisticsProvider { /// - GROUP BY is empty (scalar aggregate) /// - Any GROUP BY expression is not a simple column reference /// - Any GROUP BY column lacks NDV information +#[deprecated( + since = "56.0.0", + note = "duplicates the `statistics_from_inputs` of `AggregateExec`; without a matching provider the statistics walk uses the operator's own estimate" +)] #[derive(Debug, Default)] pub struct AggregateStatisticsProvider; +#[expect(deprecated)] impl StatisticsProvider for AggregateStatisticsProvider { fn matches(&self, plan: &dyn ExecutionPlan) -> bool { plan.downcast_ref::().is_some() @@ -817,24 +861,13 @@ impl StatisticsProvider for AggregateStatisticsProvider { } } -/// Statistics provider for equi-joins (hash join, sort-merge join) and cross joins. -/// -/// For equi-joins, estimates output cardinality as -/// `left_rows * right_rows / product(max(left_ndv_i, right_ndv_i))` -/// across all join key columns (assuming independence between keys), -/// falling back to the Cartesian product when any key lacks NDV on both sides. -/// For cross joins, uses the exact Cartesian product. +/// Experimental statistics provider for [`HashJoinExec`] and [`SortMergeJoinExec`]. /// -/// The base inner-join estimate is then adjusted for the join type: -/// - Semi joins: capped at the preserved-side row count -/// - Anti joins: preserved-side minus matched rows (clamped to 0) -/// - Left/Right outer: at least as many rows as the preserved side -/// - Full outer: at least `left + right - inner_estimate` -/// - Left mark: exactly `left_rows` (one output row per left row) -/// -/// Delegates when: -/// - The plan is not a supported join type -/// - Either input lacks row count information +/// For inner joins with two or more keys, all with a known distinct count on +/// both sides, replaces the operator's row count (based on the most selective +/// key) with `left_rows * right_rows / product(max(left_ndv_i, right_ndv_i))`, +/// assuming independent keys. Correlated keys make this estimate too low. +/// Other joins keep the operator's estimate. #[derive(Debug, Default)] pub struct JoinStatisticsProvider; @@ -842,7 +875,6 @@ impl StatisticsProvider for JoinStatisticsProvider { fn matches(&self, plan: &dyn ExecutionPlan) -> bool { plan.downcast_ref::().is_some() || plan.downcast_ref::().is_some() - || plan.downcast_ref::().is_some() } fn compute_statistics( @@ -850,119 +882,81 @@ impl StatisticsProvider for JoinStatisticsProvider { plan: &dyn ExecutionPlan, child_stats: &[ExtendedStatistics], ) -> Result { - use datafusion_common::JoinType; - use datafusion_physical_expr::expressions::Column; - if child_stats.len() < 2 { return Ok(StatisticsResult::Delegate); } + let (on, join_type) = if let Some(hash_join) = plan.downcast_ref::() + { + (hash_join.on(), *hash_join.join_type()) + } else if let Some(sort_merge_join) = plan.downcast_ref::() { + (sort_merge_join.on(), sort_merge_join.join_type()) + } else { + return Ok(StatisticsResult::Delegate); + }; + if join_type != JoinType::Inner || on.len() < 2 { + return Ok(StatisticsResult::Delegate); + } + let left = &child_stats[0].base; let right = &child_stats[1].base; - let (Some(&left_rows), Some(&right_rows)) = (left.num_rows.get_value(), right.num_rows.get_value()) else { return Ok(StatisticsResult::Delegate); }; - /// Estimate equi-join output using NDV of join key columns: - /// left_rows * right_rows / product(max(left_ndv_i, right_ndv_i)) - /// Falls back to Cartesian product if any key lacks NDV on both sides. - fn equi_join_estimate( - on: JoinOnRef, - left: &Statistics, - right: &Statistics, - left_rows: usize, - right_rows: usize, - ) -> usize { - if on.is_empty() { - return left_rows.saturating_mul(right_rows); - } - let mut ndv_divisor: usize = 1; - for (left_key, right_key) in on { - let left_ndv = left_key - .downcast_ref::() - .and_then(|c| left.column_statistics.get(c.index())) - .and_then(|s| s.distinct_count.get_value().copied()); - let right_ndv = right_key - .downcast_ref::() - .and_then(|c| right.column_statistics.get(c.index())) - .and_then(|s| s.distinct_count.get_value().copied()); - match (left_ndv, right_ndv) { - (Some(l), Some(r)) if l > 0 && r > 0 => { - ndv_divisor = ndv_divisor.saturating_mul(l.max(r)); - } - _ => return left_rows.saturating_mul(right_rows), + let mut ndv_divisor: usize = 1; + for (left_key, right_key) in on { + let left_ndv = key_distinct_count(left_key, left); + let right_ndv = key_distinct_count(right_key, right); + match (left_ndv, right_ndv) { + (Some(l), Some(r)) if l > 0 && r > 0 => { + ndv_divisor = ndv_divisor.saturating_mul(l.max(r)); } + _ => return Ok(StatisticsResult::Delegate), } - let max_rows = left_rows.saturating_mul(right_rows); - max_rows.checked_div(ndv_divisor).unwrap_or(max_rows) } - let (inner_estimate, is_exact_cartesian, join_type) = if let Some(hash_join) = - plan.downcast_ref::() - { - let est = - equi_join_estimate(hash_join.on(), left, right, left_rows, right_rows); - (est, false, *hash_join.join_type()) - } else if let Some(smj) = plan.downcast_ref::() { - let est = equi_join_estimate(smj.on(), left, right, left_rows, right_rows); - (est, false, smj.join_type()) - } else if plan.downcast_ref::().is_some() { - let both_exact = left.num_rows.is_exact().unwrap_or(false) - && right.num_rows.is_exact().unwrap_or(false); - ( - left_rows.saturating_mul(right_rows), - both_exact, - JoinType::Inner, - ) - } else { - return Ok(StatisticsResult::Delegate); - }; - - // Apply join-type-aware cardinality bounds - let estimated = match join_type { - JoinType::Inner => inner_estimate, - JoinType::Left => inner_estimate.max(left_rows), - JoinType::Right => inner_estimate.max(right_rows), - JoinType::Full => { - // At least left + right - matched, but never less than inner - let outer_bound = left_rows - .saturating_add(right_rows) - .saturating_sub(inner_estimate); - inner_estimate.max(outer_bound) - } - JoinType::LeftSemi => inner_estimate.min(left_rows), - JoinType::RightSemi => inner_estimate.min(right_rows), - JoinType::LeftAnti => left_rows.saturating_sub(inner_estimate.min(left_rows)), - JoinType::RightAnti => { - right_rows.saturating_sub(inner_estimate.min(right_rows)) - } - JoinType::LeftMark => left_rows, - JoinType::RightMark => right_rows, - }; - - // NL join inner with exact inputs is an exact Cartesian product; - // NDV-based estimates are inherently inexact. - let num_rows = if is_exact_cartesian && join_type == JoinType::Inner { - Precision::Exact(estimated) - } else { - Precision::Inexact(estimated) - }; + let mut estimate = left_rows + .saturating_mul(right_rows) + .checked_div(ndv_divisor) + .unwrap_or(usize::MAX); + if let Some(fetch) = plan.fetch() { + estimate = estimate.min(fetch); + } - computed_with_row_count(plan, num_rows, child_stats) + let child_base: Vec> = child_stats + .iter() + .map(|c| Arc::clone(c.base_arc())) + .collect(); + let mut stats = Arc::unwrap_or_clone( + plan.statistics_from_inputs(&child_base, &StatisticsArgs::new())?, + ); + rescale_byte_size(&mut stats, Precision::Inexact(estimate)); + Ok(StatisticsResult::Computed(ExtendedStatistics::new(stats))) } } +fn key_distinct_count(key: &Arc, stats: &Statistics) -> Option { + key.downcast_ref::() + .and_then(|c| stats.column_statistics.get(c.index())) + .and_then(|s| s.distinct_count.get_value().copied()) +} + /// Statistics provider for [`LocalLimitExec`] and /// [`GlobalLimitExec`]. /// /// Caps output row count at the limit value, accounting for any leading skip offset /// in `GlobalLimitExec`. +#[deprecated( + since = "56.0.0", + note = "duplicates the `statistics_from_inputs` of the limit operators; without a matching provider the statistics walk uses the operator's own estimate" +)] #[derive(Debug, Default)] pub struct LimitStatisticsProvider; +#[expect(deprecated)] impl StatisticsProvider for LimitStatisticsProvider { fn matches(&self, plan: &dyn ExecutionPlan) -> bool { plan.downcast_ref::().is_some() @@ -1011,9 +1005,14 @@ impl StatisticsProvider for LimitStatisticsProvider { /// Statistics provider for [`UnionExec`]. /// /// Sums row counts across all inputs. +#[deprecated( + since = "56.0.0", + note = "duplicates the `statistics_from_inputs` of `UnionExec`; without a matching provider the statistics walk uses the operator's own estimate" +)] #[derive(Debug, Default)] pub struct UnionStatisticsProvider; +#[expect(deprecated)] impl StatisticsProvider for UnionStatisticsProvider { fn matches(&self, plan: &dyn ExecutionPlan) -> bool { plan.downcast_ref::().is_some() @@ -1134,6 +1133,7 @@ impl StatisticsProvider for ClosureStatisticsProvider { } #[cfg(test)] +#[expect(deprecated)] mod tests { use super::*; use crate::filter::FilterExec; @@ -1810,6 +1810,84 @@ mod tests { Ok(()) } + #[test] + fn test_filter_provider_ndv_uses_post_fetch_rows() -> Result<()> { + use crate::filter::FilterExecBuilder; + + // 1000 rows with 100 distinct values; `fetch` keeps 200 rows + let source = make_source_with_ndvs(1000, Some(100), None); + let filter = FilterExecBuilder::new(lit(true), source) + .with_fetch(Some(200)) + .build()?; + let operator = compute(&StatisticsRegistry::new(), &filter)?; + let registry = + StatisticsRegistry::with_providers(vec![Arc::new(FilterStatisticsProvider)]); + let stats = compute(®istry, &filter)?; + + // The selectivity is the post-fetch row count over the input row count, + // and the survival model starts from the input distinct count + assert_eq!(stats.base.num_rows, Precision::Inexact(200)); + assert_eq!( + operator.base.column_statistics[0].distinct_count, + Precision::Inexact(100) + ); + let expected = ndv_after_selectivity(100, 1000, 200.0 / 1000.0); + assert_eq!(expected, 89); + assert_eq!( + stats.base.column_statistics[0].distinct_count, + Precision::Inexact(expected) + ); + Ok(()) + } + + #[test] + fn test_filter_provider_keeps_distinct_count_of_zero_or_one() -> Result<()> { + for ndv in [0, 1] { + // Column "a" has 1000 distinct values, column "b" has `ndv` + let mut a_stats = ColumnStatistics::new_unknown(); + a_stats.distinct_count = Precision::Exact(1000); + a_stats.min_value = Precision::Exact(ScalarValue::Int32(Some(1))); + a_stats.max_value = Precision::Exact(ScalarValue::Int32(Some(1000))); + let mut b_stats = ColumnStatistics::new_unknown(); + b_stats.distinct_count = Precision::Exact(ndv); + let source: Arc = + Arc::new(MockSourceExec::with_column_stats( + make_schema(), + Precision::Exact(1000), + vec![a_stats, b_stats], + )); + + // The predicate "a > 900" keeps about 10% of the rows + let predicate: Arc = Arc::new(BinaryExpr::new( + col("a", &make_schema())?, + Operator::Gt, + lit(900i32), + )); + let filter: Arc = + Arc::new(FilterExec::try_new(predicate, source)?); + + let operator = compute(&StatisticsRegistry::new(), filter.as_ref())?; + let registry = StatisticsRegistry::with_providers(vec![Arc::new( + FilterStatisticsProvider, + )]); + let stats = compute(®istry, filter.as_ref())?; + + assert!( + stats + .base + .num_rows + .get_value() + .is_some_and(|rows| *rows < 1000) + ); + assert_eq!( + stats.base.column_statistics[1].distinct_count, + operator.base.column_statistics[1].distinct_count, + "distinct count {ndv}" + ); + } + Ok(()) + } + #[test] fn test_filter_adjusts_ndv_by_selectivity() -> Result<()> { use datafusion_common::ScalarValue; @@ -2211,340 +2289,146 @@ mod tests { use crate::joins::{HashJoinExec, PartitionMode}; use datafusion_common::{JoinType, NullEquality}; - fn make_source_with_ndv_2col( + fn make_source_with_ndvs( num_rows: usize, ndv_a: Option, + ndv_b: Option, ) -> Arc { - let schema = make_schema(); // "a" Int32, "b" Int32 - let col_stats = vec![ - { + let column_statistics = [ndv_a, ndv_b] + .into_iter() + .map(|ndv| { let mut cs = ColumnStatistics::new_unknown(); - if let Some(n) = ndv_a { + if let Some(n) = ndv { cs.distinct_count = Precision::Exact(n); } cs - }, - ColumnStatistics::new_unknown(), - ]; + }) + .collect(); Arc::new(MockSourceExec::with_column_stats( - schema, + make_schema(), // "a" Int32, "b" Int32 Precision::Exact(num_rows), - col_stats, + column_statistics, )) } - fn make_hash_join( - left: Arc, - right: Arc, - ) -> Result> { - let _schema = make_schema(); - let on: JoinOn = vec![( - Arc::new(Column::new("a", 0)) as Arc, - Arc::new(Column::new("a", 0)) as Arc, - )]; - Ok(Arc::new(HashJoinExec::try_new( - left, - right, - on, - None, - &JoinType::Inner, - None, - PartitionMode::CollectLeft, - NullEquality::NullEqualsNull, - false, - )?)) - } - - #[test] - fn test_join_provider_with_ndv() -> Result<()> { - // left: 1000 rows, NDV(a)=100; right: 500 rows, NDV(a)=50 - // expected = 1000 * 500 / max(100, 50) = 5000 - let left = make_source_with_ndv_2col(1000, Some(100)); - let right = make_source_with_ndv_2col(500, Some(50)); - let join = make_hash_join(left, right)?; - - let registry = - StatisticsRegistry::with_providers(vec![Arc::new(JoinStatisticsProvider)]); - let stats = compute(®istry, join.as_ref())?; - assert_eq!(stats.base.num_rows, Precision::Inexact(5000)); - Ok(()) - } - - #[test] - fn test_join_provider_uses_actual_key_column_ndv() -> Result<()> { - // Join on column "b" (index 1), NDV only set on "b", not "a". - // Old first()-based code would look up column 0 (a), find no NDV, - // and fall back to Cartesian product. The fix looks up column 1 (b). - // left: 1000 rows, NDV(b)=50; right: 500 rows, NDV(b)=25 - // expected = 1000 * 500 / max(50, 25) = 10000 - let schema = make_schema(); // "a" Int32, "b" Int32 - let make_source_ndv_b = - |num_rows: usize, ndv_b: usize| -> Arc { - let col_stats = vec![ - ColumnStatistics::new_unknown(), // "a": no NDV - { - let mut cs = ColumnStatistics::new_unknown(); - cs.distinct_count = Precision::Exact(ndv_b); - cs - }, - ]; - Arc::new(MockSourceExec::with_column_stats( - Arc::clone(&schema), - Precision::Exact(num_rows), - col_stats, - )) - }; - - let left = make_source_ndv_b(1000, 50); - let right = make_source_ndv_b(500, 25); - - // Join on column "b" (index 1) - let on: JoinOn = vec![( - Arc::new(Column::new("b", 1)) as Arc, - Arc::new(Column::new("b", 1)) as Arc, - )]; - let join: Arc = Arc::new(HashJoinExec::try_new( - left, - right, - on, - None, - &JoinType::Inner, - None, - PartitionMode::CollectLeft, - NullEquality::NullEqualsNull, - false, - )?); - - let registry = - StatisticsRegistry::with_providers(vec![Arc::new(JoinStatisticsProvider)]); - let stats = compute(®istry, join.as_ref())?; - assert_eq!(stats.base.num_rows, Precision::Inexact(10_000)); - Ok(()) - } - - #[test] - fn test_join_provider_multi_key_ndv() -> Result<()> { - // Multi-key join: ON a.a = b.a AND a.b = b.b - // left: 1000 rows, NDV(a)=100, NDV(b)=20 - // right: 500 rows, NDV(a)=50, NDV(b)=10 - // expected = 1000 * 500 / (max(100,50) * max(20,10)) = 500000 / 2000 = 250 - let schema = make_schema(); // "a" Int32, "b" Int32 - let make_source_2ndv = - |num_rows: usize, ndv_a: usize, ndv_b: usize| -> Arc { - let col_stats = vec![ - { - let mut cs = ColumnStatistics::new_unknown(); - cs.distinct_count = Precision::Exact(ndv_a); - cs - }, - { - let mut cs = ColumnStatistics::new_unknown(); - cs.distinct_count = Precision::Exact(ndv_b); - cs - }, - ]; - Arc::new(MockSourceExec::with_column_stats( - Arc::clone(&schema), - Precision::Exact(num_rows), - col_stats, - )) - }; - - let left = make_source_2ndv(1000, 100, 20); - let right = make_source_2ndv(500, 50, 10); - - let on: JoinOn = vec![ - ( - Arc::new(Column::new("a", 0)) as Arc, - Arc::new(Column::new("a", 0)) as Arc, - ), - ( - Arc::new(Column::new("b", 1)) as Arc, - Arc::new(Column::new("b", 1)) as Arc, - ), - ]; - let join: Arc = Arc::new(HashJoinExec::try_new( - left, - right, - on, - None, - &JoinType::Inner, - None, - PartitionMode::CollectLeft, - NullEquality::NullEqualsNull, - false, - )?); - - let registry = - StatisticsRegistry::with_providers(vec![Arc::new(JoinStatisticsProvider)]); - let stats = compute(®istry, join.as_ref())?; - assert_eq!(stats.base.num_rows, Precision::Inexact(250)); - Ok(()) - } - - #[test] - fn test_join_provider_fallback_cartesian() -> Result<()> { - // No NDV available -> Cartesian product estimate - let left = make_source_with_ndv_2col(100, None); - let right = make_source_with_ndv_2col(200, None); - let join = make_hash_join(left, right)?; - - let registry = - StatisticsRegistry::with_providers(vec![Arc::new(JoinStatisticsProvider)]); - let stats = compute(®istry, join.as_ref())?; - assert_eq!(stats.base.num_rows, Precision::Inexact(20_000)); - Ok(()) - } - - #[test] - fn test_nl_join_delegates() -> Result<()> { - use crate::joins::NestedLoopJoinExec; - - // NL join delegates to the built-in (NestedLoopJoinExec may have an - // arbitrary JoinFilter, so the provider cannot safely assume Cartesian). - let left = make_source(100); - let right = make_source(200); - let join: Arc = Arc::new(NestedLoopJoinExec::try_new( - left, - right, - None, - &JoinType::Inner, - None, - )?); - - let registry = - StatisticsRegistry::with_providers(vec![Arc::new(JoinStatisticsProvider)]); - let stats = compute(®istry, join.as_ref())?; - // Provider delegates; result comes from built-in statistics_from_inputs. - assert!( - stats.base.num_rows.get_value().is_some() - || matches!(stats.base.num_rows, Precision::Absent) - ); - Ok(()) + fn join_on(num_keys: usize) -> JoinOn { + ["a", "b"] + .iter() + .enumerate() + .take(num_keys) + .map(|(index, name)| { + ( + Arc::new(Column::new(name, index)) as Arc, + Arc::new(Column::new(name, index)) as Arc, + ) + }) + .collect() } - fn make_hash_join_typed( + fn make_hash_join( left: Arc, right: Arc, + num_keys: usize, join_type: JoinType, + fetch: Option, ) -> Result> { - let on: JoinOn = vec![( - Arc::new(Column::new("a", 0)) as Arc, - Arc::new(Column::new("a", 0)) as Arc, - )]; - Ok(Arc::new(HashJoinExec::try_new( + let join: Arc = Arc::new(HashJoinExec::try_new( left, right, - on, + join_on(num_keys), None, &join_type, None, PartitionMode::CollectLeft, NullEquality::NullEqualsNull, false, - )?)) + )?); + match fetch { + Some(_) => Ok(join.with_fetch(fetch).expect("hash join supports fetch")), + None => Ok(join), + } } - fn compute_join_rows( - left_rows: usize, - left_ndv: Option, - right_rows: usize, - right_ndv: Option, + /// Two-key inner join: left has 1000 rows with NDV(a)=100 and NDV(b)=20, + /// right has 500 rows with NDV(a)=50 and NDV(b)=10. The product formula + /// gives 1000 * 500 / (max(100, 50) * max(20, 10)) = 250 rows. + fn make_two_key_join( join_type: JoinType, - ) -> Result> { - let left = make_source_with_ndv_2col(left_rows, left_ndv); - let right = make_source_with_ndv_2col(right_rows, right_ndv); - let join = make_hash_join_typed(left, right, join_type)?; - let registry = - StatisticsRegistry::with_providers(vec![Arc::new(JoinStatisticsProvider)]); - Ok(compute(®istry, join.as_ref())?.base.num_rows) + fetch: Option, + ) -> Result> { + make_hash_join( + make_source_with_ndvs(1000, Some(100), Some(20)), + make_source_with_ndvs(500, Some(50), Some(10)), + 2, + join_type, + fetch, + ) } - #[test] - fn test_join_provider_left_outer() -> Result<()> { - // left=1000, right=500, NDV(a)=100/50 - // inner estimate = 1000*500/100 = 5000, already >= left_rows - // Left outer: max(5000, 1000) = 5000 - assert_eq!( - compute_join_rows(1000, Some(100), 500, Some(50), JoinType::Left)?, - Precision::Inexact(5000) - ); - // Small inner estimate: left=1000, right=10, NDV=100/100 - // inner = 1000*10/100 = 100, left outer = max(100, 1000) = 1000 - assert_eq!( - compute_join_rows(1000, Some(100), 10, Some(100), JoinType::Left)?, - Precision::Inexact(1000) - ); - Ok(()) + fn join_provider_registry() -> StatisticsRegistry { + StatisticsRegistry::with_providers(vec![Arc::new(JoinStatisticsProvider)]) } #[test] - fn test_join_provider_right_outer() -> Result<()> { - // inner = 1000*10/100 = 100, right outer = max(100, 10) = 100 - assert_eq!( - compute_join_rows(1000, Some(100), 10, Some(100), JoinType::Right)?, - Precision::Inexact(100) - ); - // inner = 10*1000/100 = 100, right outer = max(100, 1000) = 1000 + fn test_join_provider_multi_key_inner_refines() -> Result<()> { + let join = make_two_key_join(JoinType::Inner, None)?; + let operator = compute(&StatisticsRegistry::new(), join.as_ref())?; + let refined = compute(&join_provider_registry(), join.as_ref())?; + assert_ne!(operator.base.num_rows, Precision::Inexact(250)); + assert_eq!(refined.base.num_rows, Precision::Inexact(250)); assert_eq!( - compute_join_rows(10, Some(100), 1000, Some(100), JoinType::Right)?, - Precision::Inexact(1000) + refined.base.column_statistics, + operator.base.column_statistics ); Ok(()) } #[test] - fn test_join_provider_semi_join() -> Result<()> { - // inner = 5000, left semi = min(5000, 1000) = 1000 - assert_eq!( - compute_join_rows(1000, Some(100), 500, Some(50), JoinType::LeftSemi)?, - Precision::Inexact(1000) - ); - // inner = 5000, right semi = min(5000, 500) = 500 - assert_eq!( - compute_join_rows(1000, Some(100), 500, Some(50), JoinType::RightSemi)?, - Precision::Inexact(500) - ); - // Cartesian fallback (no NDV): inner = 1000*500 = 500000, - // left semi = min(500000, 1000) = 1000 (selectivity = 1.0) - assert_eq!( - compute_join_rows(1000, None, 500, None, JoinType::LeftSemi)?, - Precision::Inexact(1000) - ); - Ok(()) - } + fn test_join_provider_caps_at_fetch() -> Result<()> { + let join = make_two_key_join(JoinType::Inner, Some(100))?; + let stats = compute(&join_provider_registry(), join.as_ref())?; + assert_eq!(stats.base.num_rows, Precision::Inexact(100)); - #[test] - fn test_join_provider_anti_join() -> Result<()> { - // inner = 1000*10/100 = 100, left anti = 1000 - min(100, 1000) = 900 - assert_eq!( - compute_join_rows(1000, Some(100), 10, Some(100), JoinType::LeftAnti)?, - Precision::Inexact(900) - ); - // inner = 5000, right anti = 500 - min(5000, 500) = 0 - assert_eq!( - compute_join_rows(1000, Some(100), 500, Some(50), JoinType::RightAnti)?, - Precision::Inexact(0) - ); + // A fetch above the estimate does not change it + let join = make_two_key_join(JoinType::Inner, Some(1000))?; + let stats = compute(&join_provider_registry(), join.as_ref())?; + assert_eq!(stats.base.num_rows, Precision::Inexact(250)); Ok(()) } - // ========================================================================= - // CrossJoinExec tests (handled by JoinStatisticsProvider) - // ========================================================================= - #[test] - fn test_cross_join_provider_exact() -> Result<()> { - use crate::joins::CrossJoinExec; - let left = make_source(100); - let right = make_source(200); - let join: Arc = Arc::new(CrossJoinExec::new(left, right)); + fn test_join_provider_delegates_to_operator() -> Result<()> { + let single_key = make_hash_join( + make_source_with_ndvs(1000, Some(100), None), + make_source_with_ndvs(500, Some(50), None), + 1, + JoinType::Inner, + None, + )?; + let missing_ndv = make_hash_join( + make_source_with_ndvs(1000, Some(100), Some(20)), + make_source_with_ndvs(500, Some(50), None), + 2, + JoinType::Inner, + None, + )?; + let zero_ndv = make_hash_join( + make_source_with_ndvs(1000, Some(100), Some(0)), + make_source_with_ndvs(500, Some(50), Some(10)), + 2, + JoinType::Inner, + None, + )?; + let mut joins = vec![single_key, missing_ndv, zero_ndv]; + for join_type in [JoinType::Left, JoinType::LeftSemi, JoinType::LeftAnti] { + joins.push(make_two_key_join(join_type, None)?); + } - let registry = - StatisticsRegistry::with_providers(vec![Arc::new(JoinStatisticsProvider)]); - let stats = compute(®istry, join.as_ref())?; - // Both inputs have Exact row counts -> result is also Exact - assert_eq!(stats.base.num_rows, Precision::Exact(20_000)); + for join in joins { + let operator = compute(&StatisticsRegistry::new(), join.as_ref())?; + let refined = compute(&join_provider_registry(), join.as_ref())?; + assert_eq!(refined.base(), operator.base()); + } Ok(()) } diff --git a/datafusion/sqllogictest/src/test_context.rs b/datafusion/sqllogictest/src/test_context.rs index bc2859aaa9453..e4870622a5c26 100644 --- a/datafusion/sqllogictest/src/test_context.rs +++ b/datafusion/sqllogictest/src/test_context.rs @@ -37,6 +37,7 @@ use datafusion::catalog::{ CatalogProvider, MemoryCatalogProvider, MemorySchemaProvider, SchemaProvider, Session, }; use datafusion::common::config::Dialect; +use datafusion::common::stats::Precision; use datafusion::common::{DataFusionError, Result, not_impl_err}; use datafusion::functions::math::abs; use datafusion::logical_expr::async_udf::{AsyncScalarUDF, AsyncScalarUDFImpl}; @@ -63,7 +64,11 @@ use datafusion::common::cast::as_float64_array; use datafusion::execution::SessionStateBuilder; use datafusion::execution::memory_pool::UnboundedMemoryPool; use datafusion::execution::runtime_env::RuntimeEnvBuilder; -use datafusion::physical_plan::operator_statistics::StatisticsRegistry; +use datafusion::physical_plan::joins::HashJoinExec; +use datafusion::physical_plan::operator_statistics::{ + ClosureStatisticsProvider, StatisticsRegistry, StatisticsResult, +}; +use datafusion::physical_plan::statistics::StatisticsArgs; use log::info; use sqlparser::ast; use tempfile::TempDir; @@ -144,9 +149,30 @@ impl TestContext { relative_path.file_name().and_then(|name| name.to_str()), Some("statistics_registry.slt") ) { - state_builder = state_builder.with_statistics_registry( - StatisticsRegistry::default_with_builtin_providers(), + // Replaces the join estimate with the Cartesian product + let join_provider = ClosureStatisticsProvider::with_matches( + |plan| plan.downcast_ref::().is_some(), + |plan, child_stats| { + let (Some(&left_rows), Some(&right_rows)) = ( + child_stats[0].base().num_rows.get_value(), + child_stats[1].base().num_rows.get_value(), + ) else { + return Ok(StatisticsResult::Delegate); + }; + let child_base = child_stats + .iter() + .map(|c| Arc::clone(c.base_arc())) + .collect::>(); + let mut stats = Arc::unwrap_or_clone( + plan.statistics_from_inputs(&child_base, &StatisticsArgs::new())?, + ); + stats.num_rows = Precision::Inexact(left_rows * right_rows); + Ok(StatisticsResult::Computed(stats.into())) + }, ); + let registry = + StatisticsRegistry::with_providers(vec![Arc::new(join_provider)]); + state_builder = state_builder.with_statistics_registry(registry); } let state = state_builder.build(); diff --git a/datafusion/sqllogictest/test_files/statistics_registry.slt b/datafusion/sqllogictest/test_files/statistics_registry.slt index 22209e5eb112f..7bf7dc1779c01 100644 --- a/datafusion/sqllogictest/test_files/statistics_registry.slt +++ b/datafusion/sqllogictest/test_files/statistics_registry.slt @@ -18,6 +18,9 @@ # StatisticsRegistry: demonstrates improved join ordering via more conservative # cardinality estimates on a skewed dataset. # +# The session registers a test-only provider in test_context.rs that replaces +# the join estimate with the Cartesian product of the input row counts. +# # customers (10 rows, 3 distinct customer_ids, skewed 8:1:1) # orders (10 rows, same distribution) # dim_small (50 rows) diff --git a/docs/source/library-user-guide/upgrading/56.0.0.md b/docs/source/library-user-guide/upgrading/56.0.0.md index 1f1855f76f149..d9e7dbb3ea164 100644 --- a/docs/source/library-user-guide/upgrading/56.0.0.md +++ b/docs/source/library-user-guide/upgrading/56.0.0.md @@ -317,9 +317,9 @@ previous default (`use_statistics_registry = false`). - Drop any `use_statistics_registry` setting. - To use statistics providers, register them on the session with - `SessionStateBuilder::with_statistics_registry(...)`. To keep the built-in - NDV-aware providers the flag previously enabled, register - `StatisticsRegistry::default_with_builtin_providers()`. + `SessionStateBuilder::with_statistics_registry(...)`. To use the + experimental providers, register + `StatisticsRegistry::with_experimental_providers()`. - To opt out, register nothing (the default). ### Generated protobuf `JsonWriterOptions` gained a `compression_level` field @@ -428,6 +428,28 @@ part of `StatisticsRegistry::default_with_builtin_providers()`. - Remove `DefaultStatisticsProvider` from any custom provider chain; register no terminal provider instead (the walk falls back on its own). +### `StatisticsRegistry::default_with_builtin_providers` is deprecated + +`StatisticsRegistry::default_with_builtin_providers()` is deprecated in favor of +`StatisticsRegistry::with_experimental_providers()`. Most of the providers it +registers reimplement the operator's own estimation. The experimental providers +start from the operator's estimate and replace only the values their technique +estimates, so improvements to the operators also reach them (see the +`datafusion_physical_plan::operator_statistics` module documentation). + +`ProjectionStatisticsProvider`, `PassthroughStatisticsProvider`, +`AggregateStatisticsProvider`, `LimitStatisticsProvider` and +`UnionStatisticsProvider` are deprecated because they duplicate the operator's +own estimate. + +**Migration guide:** + +- Replace `StatisticsRegistry::default_with_builtin_providers()` with + `StatisticsRegistry::with_experimental_providers()`. +- Remove the deprecated providers from custom provider chains. + +See [issue #25571](https://github.com/apache/datafusion/issues/25571) for details. + ### `StatisticsRegistry::compute` and `compute_base` are deprecated Use the walk instead: From 7a6782c3954b304867fb8583cfa37d7d8d7a880a Mon Sep 17 00:00:00 2001 From: Alessandro Solimando Date: Mon, 5 Oct 2026 14:01:35 +0200 Subject: [PATCH 2/6] refactor: deprecate all bundled statistics providers --- benchmarks/src/statistics.rs | 2 +- .../examples/statistics/join_reorder.rs | 64 ++++++++++++++++--- datafusion/ffi/src/physical_optimizer.rs | 2 +- .../src/operator_statistics/mod.rs | 52 ++++++--------- .../library-user-guide/upgrading/56.0.0.md | 33 +++++----- 5 files changed, 91 insertions(+), 62 deletions(-) diff --git a/benchmarks/src/statistics.rs b/benchmarks/src/statistics.rs index 8981490919fc6..f574add42bd40 100644 --- a/benchmarks/src/statistics.rs +++ b/benchmarks/src/statistics.rs @@ -249,7 +249,7 @@ enum QError { } fn capture_statistics(plan: &dyn ExecutionPlan) -> Result> { - let statistics_context = StatisticsRegistry::with_experimental_providers(); + let statistics_context = StatisticsRegistry::new(); let mut result = vec![]; capture_statistics_inner(plan, &statistics_context, "0", &mut result)?; Ok(result) diff --git a/datafusion-examples/examples/statistics/join_reorder.rs b/datafusion-examples/examples/statistics/join_reorder.rs index a825fd431977b..cdb8664d6da43 100644 --- a/datafusion-examples/examples/statistics/join_reorder.rs +++ b/datafusion-examples/examples/statistics/join_reorder.rs @@ -23,8 +23,8 @@ //! //! `(SELECT user_id FROM events WHERE amount < 50 GROUP BY user_id) JOIN dims`: a //! provider supplies the column stats a catalog knows (`amount` range, `user_id` -//! distinct count); the experimental `FilterStatisticsProvider` then refines the -//! post-filter distinct count with the survival formula +//! distinct count); a second provider then refines the post-filter distinct +//! count with the survival formula //! `NDV * (1 - (1 - selectivity)^(rows / NDV))` (Yao/Cardenas) to ~40, below `dims` //! (48), flipping the join build side. Core's simpler `min(NDV, rows)` cap would //! give 50 (> 48) and keep the other order; the refinement is the point. The @@ -43,8 +43,10 @@ use datafusion::common::ScalarValue; use datafusion::common::stats::Precision; use datafusion::execution::SessionStateBuilder; use datafusion::physical_plan::ExecutionPlan; +use datafusion::physical_plan::filter::FilterExec; use datafusion::physical_plan::operator_statistics::{ ClosureStatisticsProvider, ExtendedStatistics, StatisticsRegistry, StatisticsResult, + ndv_after_selectivity, }; use datafusion::physical_plan::statistics::StatisticsArgs; use datafusion::prelude::*; @@ -76,6 +78,45 @@ fn catalog_stats( Ok(StatisticsResult::Computed(ExtendedStatistics::new(stats))) } +fn filter_matches(plan: &dyn ExecutionPlan) -> bool { + plan.downcast_ref::().is_some() +} + +/// Starts from the filter's own estimate and lowers each column's distinct +/// count with the survival formula, applied to the input distinct count. +fn filter_stats( + plan: &dyn ExecutionPlan, + child_stats: &[ExtendedStatistics], +) -> Result { + let Some(filter) = plan.downcast_ref::() else { + return Ok(StatisticsResult::Delegate); + }; + let input = child_stats[0].base(); + let mut stats = (*plan.statistics_from_inputs( + &[Arc::clone(child_stats[0].base_arc())], + &StatisticsArgs::new(), + )?) + .clone(); + if let (Some(&input_rows), Some(&output_rows)) = + (input.num_rows.get_value(), stats.num_rows.get_value()) + && output_rows < input_rows + { + let selectivity = output_rows as f64 / input_rows as f64; + for (i, col) in stats.column_statistics.iter_mut().enumerate() { + let input_index = filter.projection().as_ref().map_or(i, |p| p[i]); + if let Some(&ndv) = col.distinct_count.get_value() + && let Some(&input_ndv) = input.column_statistics[input_index] + .distinct_count + .get_value() + { + let survived = ndv_after_selectivity(input_ndv, input_rows, selectivity); + col.distinct_count = Precision::Inexact(survived.min(ndv)); + } + } + } + Ok(StatisticsResult::Computed(ExtendedStatistics::new(stats))) +} + fn int_col(values: &[i32]) -> Arc { Arc::new(Int32Array::from_iter_values(values.iter().copied())) } @@ -111,11 +152,16 @@ fn build_ctx(with_registry: bool) -> Result { .with_config(config) .with_default_features(); if with_registry { - let mut registry = StatisticsRegistry::with_experimental_providers(); - registry.register(Arc::new(ClosureStatisticsProvider::with_matches( - catalog_matches, - catalog_stats, - ))); + let registry = StatisticsRegistry::with_providers(vec![ + Arc::new(ClosureStatisticsProvider::with_matches( + catalog_matches, + catalog_stats, + )), + Arc::new(ClosureStatisticsProvider::with_matches( + filter_matches, + filter_stats, + )), + ]); builder = builder.with_statistics_registry(registry); } let ctx = SessionContext::new_with_state(builder.build()); @@ -167,13 +213,13 @@ pub async fn join_reorder() -> Result<()> { "A hash join builds its in-memory hash table from one input and probes with\n\ the other, so the smaller input should be the build side. Default estimation\n\ sizes the grouped `events` at 1000 rows and builds from `dims`; the\n\ - registry's refined ~32 estimate is below `dims` (48 rows), so it flips the\n\ + registry's refined ~40 estimate is below `dims` (48 rows), so it flips the\n\ build side to `events`. The ground-truth count above (also below 48)\n\ confirms `events` really is the smaller, cheaper side.\n" ); println!("-- Without the registry (default estimation) --"); println!("{}\n", explain(&build_ctx(false)?).await?); - println!("-- With the registry (experimental refinement) --"); + println!("-- With the registry (refined estimation) --"); println!("{}", explain(&build_ctx(true)?).await?); Ok(()) } diff --git a/datafusion/ffi/src/physical_optimizer.rs b/datafusion/ffi/src/physical_optimizer.rs index 8a01f916242a6..e74a1cb9ff4d3 100644 --- a/datafusion/ffi/src/physical_optimizer.rs +++ b/datafusion/ffi/src/physical_optimizer.rs @@ -586,7 +586,7 @@ mod tests { let ctx = ContextWithRegistry { config: ConfigOptions::new(), - registry: StatisticsRegistry::with_experimental_providers(), + registry: StatisticsRegistry::new(), }; let plan = create_test_plan(); diff --git a/datafusion/physical-plan/src/operator_statistics/mod.rs b/datafusion/physical-plan/src/operator_statistics/mod.rs index 4b49a61b5eb74..c4dd0eddac25c 100644 --- a/datafusion/physical-plan/src/operator_statistics/mod.rs +++ b/datafusion/physical-plan/src/operator_statistics/mod.rs @@ -37,22 +37,12 @@ //! - [`StatisticsRegistry`]: Chains providers, lives in SessionState //! - [`ExtendedStatistics`]: Statistics with type-safe custom extensions //! -//! # Experimental Providers +//! # Bundled Providers //! -//! These optional providers hold estimation techniques not yet ready to move -//! into the operator. A technique moves into the operator once it is shown to -//! improve estimates in general. Each provider starts from the operator's -//! [`ExecutionPlan::statistics_from_inputs`] and replaces only the values its -//! technique estimates, so improvements to the operator also reach it. Register -//! them with [`StatisticsRegistry::with_experimental_providers`]. -//! -//! - [`FilterStatisticsProvider`] - adjusts the distinct count of each column -//! for the rows removed by the filter -//! - [`JoinStatisticsProvider`] - estimates multi-key inner equi-joins assuming -//! independent keys -//! -//! The remaining bundled providers are deprecated because they duplicate the -//! operators' own estimates. +//! The providers bundled in this module are deprecated: the default statistics +//! estimation is each operator's [`ExecutionPlan::statistics_from_inputs`], and +//! estimation improvements belong there. Register user-defined providers to +//! plug in other estimation. //! //! # Statistics walk //! @@ -399,21 +389,7 @@ impl StatisticsRegistry { Self { providers } } - /// Create a registry with the experimental providers (see the - /// [module documentation](crate::operator_statistics)). - /// - /// Provider order (first match wins): - /// 1. [`FilterStatisticsProvider`] - /// 2. [`JoinStatisticsProvider`] - pub fn with_experimental_providers() -> Self { - Self::with_providers(vec![ - Arc::new(FilterStatisticsProvider), - Arc::new(JoinStatisticsProvider), - ]) - } - - /// Create a registry with the experimental providers plus the deprecated - /// providers that duplicate the operators' own estimates. + /// Create a registry with the deprecated bundled providers. /// /// Provider order (first match wins): /// 1. [`FilterStatisticsProvider`] @@ -425,7 +401,7 @@ impl StatisticsRegistry { /// 7. [`UnionStatisticsProvider`] #[deprecated( since = "56.0.0", - note = "use `with_experimental_providers`, which registers only the providers that start from the operators' own estimates" + note = "the bundled providers are deprecated; the statistics walk uses each operator's `statistics_from_inputs` when no provider matches" )] #[expect(deprecated)] pub fn default_with_builtin_providers() -> Self { @@ -607,16 +583,21 @@ fn computed_with_row_count( Ok(StatisticsResult::Computed(ExtendedStatistics::new(base))) } -/// Experimental statistics provider for [`FilterExec`]. +/// Statistics provider for [`FilterExec`]. /// /// Starts from the operator's own [`ExecutionPlan::statistics_from_inputs`], /// then lowers each column's `distinct_count` to [`ndv_after_selectivity`] of /// the input column, using the ratio of output rows to input rows as the /// selectivity. Columns with a distinct count of 0 or 1 keep the value the /// operator returned. +#[deprecated( + since = "56.0.0", + note = "its distinct count adjustment belongs in `FilterExec::statistics_from_inputs`; without a matching provider the statistics walk uses the operator's own estimate" +)] #[derive(Debug, Default)] pub struct FilterStatisticsProvider; +#[expect(deprecated)] impl StatisticsProvider for FilterStatisticsProvider { fn matches(&self, plan: &dyn ExecutionPlan) -> bool { plan.downcast_ref::().is_some() @@ -861,16 +842,21 @@ impl StatisticsProvider for AggregateStatisticsProvider { } } -/// Experimental statistics provider for [`HashJoinExec`] and [`SortMergeJoinExec`]. +/// Statistics provider for [`HashJoinExec`] and [`SortMergeJoinExec`]. /// /// For inner joins with two or more keys, all with a known distinct count on /// both sides, replaces the operator's row count (based on the most selective /// key) with `left_rows * right_rows / product(max(left_ndv_i, right_ndv_i))`, /// assuming independent keys. Correlated keys make this estimate too low. /// Other joins keep the operator's estimate. +#[deprecated( + since = "56.0.0", + note = "multi-key join estimation belongs in the join operators, see https://github.com/apache/datafusion/issues/21583; without a matching provider the statistics walk uses the operator's own estimate" +)] #[derive(Debug, Default)] pub struct JoinStatisticsProvider; +#[expect(deprecated)] impl StatisticsProvider for JoinStatisticsProvider { fn matches(&self, plan: &dyn ExecutionPlan) -> bool { plan.downcast_ref::().is_some() diff --git a/docs/source/library-user-guide/upgrading/56.0.0.md b/docs/source/library-user-guide/upgrading/56.0.0.md index d9e7dbb3ea164..1306887988db9 100644 --- a/docs/source/library-user-guide/upgrading/56.0.0.md +++ b/docs/source/library-user-guide/upgrading/56.0.0.md @@ -317,9 +317,7 @@ previous default (`use_statistics_registry = false`). - Drop any `use_statistics_registry` setting. - To use statistics providers, register them on the session with - `SessionStateBuilder::with_statistics_registry(...)`. To use the - experimental providers, register - `StatisticsRegistry::with_experimental_providers()`. + `SessionStateBuilder::with_statistics_registry(...)`. - To opt out, register nothing (the default). ### Generated protobuf `JsonWriterOptions` gained a `compression_level` field @@ -428,25 +426,24 @@ part of `StatisticsRegistry::default_with_builtin_providers()`. - Remove `DefaultStatisticsProvider` from any custom provider chain; register no terminal provider instead (the walk falls back on its own). -### `StatisticsRegistry::default_with_builtin_providers` is deprecated +### The bundled statistics providers are deprecated -`StatisticsRegistry::default_with_builtin_providers()` is deprecated in favor of -`StatisticsRegistry::with_experimental_providers()`. Most of the providers it -registers reimplement the operator's own estimation. The experimental providers -start from the operator's estimate and replace only the values their technique -estimates, so improvements to the operators also reach them (see the -`datafusion_physical_plan::operator_statistics` module documentation). - -`ProjectionStatisticsProvider`, `PassthroughStatisticsProvider`, -`AggregateStatisticsProvider`, `LimitStatisticsProvider` and -`UnionStatisticsProvider` are deprecated because they duplicate the operator's -own estimate. +`StatisticsRegistry::default_with_builtin_providers()` and all the statistics +providers bundled in `datafusion_physical_plan::operator_statistics` +(`FilterStatisticsProvider`, `ProjectionStatisticsProvider`, +`PassthroughStatisticsProvider`, `AggregateStatisticsProvider`, +`JoinStatisticsProvider`, `LimitStatisticsProvider` and +`UnionStatisticsProvider`) are deprecated. They duplicate or replace the +estimation each operator already does in `statistics_from_inputs`, which is the +default. `StatisticsRegistry` remains the extension point for user-defined +providers. **Migration guide:** -- Replace `StatisticsRegistry::default_with_builtin_providers()` with - `StatisticsRegistry::with_experimental_providers()`. -- Remove the deprecated providers from custom provider chains. +- Stop registering the bundled providers; with no matching provider, the + statistics walk uses the operator's own estimate. +- To keep a bundled estimation technique, implement it in a user-defined + `StatisticsProvider`. See [issue #25571](https://github.com/apache/datafusion/issues/25571) for details. From 5c9b6b03a3604c557b46a4bdf2e043cdb456ba35 Mon Sep 17 00:00:00 2001 From: Alessandro Solimando Date: Mon, 5 Oct 2026 14:45:57 +0200 Subject: [PATCH 3/6] refactor: keep the deprecated Filter and Join providers unchanged --- .../src/operator_statistics/mod.rs | 676 +++++++++++------- 1 file changed, 421 insertions(+), 255 deletions(-) diff --git a/datafusion/physical-plan/src/operator_statistics/mod.rs b/datafusion/physical-plan/src/operator_statistics/mod.rs index c4dd0eddac25c..50124b2ac1bcf 100644 --- a/datafusion/physical-plan/src/operator_statistics/mod.rs +++ b/datafusion/physical-plan/src/operator_statistics/mod.rs @@ -76,15 +76,13 @@ use std::sync::Arc; use datafusion_common::extensions::Extensions; use datafusion_common::stats::Precision; -use datafusion_common::{JoinType, Result, Statistics}; -use datafusion_physical_expr::PhysicalExpr; -use datafusion_physical_expr::expressions::Column; +use datafusion_common::{Result, Statistics}; use crate::ExecutionPlan; use crate::aggregates::{AggregateExec, AggregateMode}; use crate::execution_plan::CardinalityEffect; use crate::filter::FilterExec; -use crate::joins::{HashJoinExec, SortMergeJoinExec}; +use crate::joins::{CrossJoinExec, HashJoinExec, JoinOnRef, SortMergeJoinExec}; use crate::limit::{GlobalLimitExec, LocalLimitExec}; use crate::projection::ProjectionExec; use crate::statistics::{ChildStats, StatisticsArgs, StatisticsContext}; @@ -583,16 +581,18 @@ fn computed_with_row_count( Ok(StatisticsResult::Computed(ExtendedStatistics::new(base))) } -/// Statistics provider for [`FilterExec`]. +/// Statistics provider for [`FilterExec`] that uses +/// pre-computed enhanced child statistics from the registry walk. /// -/// Starts from the operator's own [`ExecutionPlan::statistics_from_inputs`], -/// then lowers each column's `distinct_count` to [`ndv_after_selectivity`] of -/// the input column, using the ratio of output rows to input rows as the -/// selectivity. Columns with a distinct count of 0 or 1 keep the value the -/// operator returned. +/// Unlike the built-in fallback (which calls `statistics_from_inputs` and gets raw +/// child stats), this provider receives enhanced child stats that may include +/// NDV overrides injected at the scan level. It applies the same selectivity +/// estimation logic as `FilterExec::statistics_helper`, then additionally +/// adjusts each column's `distinct_count` using [`ndv_after_selectivity`] based +/// on the computed selectivity ratio. #[deprecated( since = "56.0.0", - note = "its distinct count adjustment belongs in `FilterExec::statistics_from_inputs`; without a matching provider the statistics walk uses the operator's own estimate" + note = "duplicates `FilterExec::statistics_from_inputs` and adds a distinct count adjustment that belongs in the operator, see https://github.com/apache/datafusion/issues/26052; without a matching provider the statistics walk uses the operator's own estimate" )] #[derive(Debug, Default)] pub struct FilterStatisticsProvider; @@ -615,40 +615,35 @@ impl StatisticsProvider for FilterStatisticsProvider { return Ok(StatisticsResult::Delegate); } - let input = child_stats[0].base(); - let input_rows = input.num_rows; - let child_base: Vec> = child_stats - .iter() - .map(|c| Arc::clone(c.base_arc())) - .collect(); - let mut stats = Arc::unwrap_or_clone( - plan.statistics_from_inputs(&child_base, &StatisticsArgs::new())?, - ); + let input_stats = (*child_stats[0].base).clone(); + let input_rows = input_stats.num_rows; + let mut stats = FilterExec::statistics_helper( + &filter.input().schema(), + input_stats, + filter.predicate(), + filter.default_selectivity(), + // TODO: pass filter.expression_analyzer_registry() once #21122 lands + )?; + // Adjust distinct_count for each column using the selectivity ratio + // via the probabilistic survival model from + // ndv_after_selectivity to account for rows removed by the filter. if let (Some(&orig_rows), Some(&filtered_rows)) = (input_rows.get_value(), stats.num_rows.get_value()) && orig_rows > 0 && filtered_rows < orig_rows { let selectivity = filtered_rows as f64 / orig_rows as f64; - for (i, col_stat) in stats.column_statistics.iter_mut().enumerate() { - let input_index = filter.projection().as_ref().map_or(i, |p| p[i]); - // The operator's distinct count is already capped at the - // filtered rows, so the survival model uses the input one - if let Some(&ndv) = col_stat.distinct_count.get_value() - && ndv > 1 - && let Some(&input_ndv) = input - .column_statistics - .get(input_index) - .and_then(|c| c.distinct_count.get_value()) - { - let adjusted = - ndv_after_selectivity(input_ndv, orig_rows, selectivity); - col_stat.distinct_count = Precision::Inexact(adjusted.min(ndv)); + for col_stat in &mut stats.column_statistics { + if let Some(&ndv) = col_stat.distinct_count.get_value() { + let adjusted = ndv_after_selectivity(ndv, orig_rows, selectivity); + col_stat.distinct_count = Precision::Inexact(adjusted); } } } + let stats = filter.statistics_with_fetch(stats, None)?; + let stats = stats.project(filter.projection().as_ref()); Ok(StatisticsResult::Computed(ExtendedStatistics::new(stats))) } } @@ -842,16 +837,27 @@ impl StatisticsProvider for AggregateStatisticsProvider { } } -/// Statistics provider for [`HashJoinExec`] and [`SortMergeJoinExec`]. +/// Statistics provider for equi-joins (hash join, sort-merge join) and cross joins. /// -/// For inner joins with two or more keys, all with a known distinct count on -/// both sides, replaces the operator's row count (based on the most selective -/// key) with `left_rows * right_rows / product(max(left_ndv_i, right_ndv_i))`, -/// assuming independent keys. Correlated keys make this estimate too low. -/// Other joins keep the operator's estimate. +/// For equi-joins, estimates output cardinality as +/// `left_rows * right_rows / product(max(left_ndv_i, right_ndv_i))` +/// across all join key columns (assuming independence between keys), +/// falling back to the Cartesian product when any key lacks NDV on both sides. +/// For cross joins, uses the exact Cartesian product. +/// +/// The base inner-join estimate is then adjusted for the join type: +/// - Semi joins: capped at the preserved-side row count +/// - Anti joins: preserved-side minus matched rows (clamped to 0) +/// - Left/Right outer: at least as many rows as the preserved side +/// - Full outer: at least `left + right - inner_estimate` +/// - Left mark: exactly `left_rows` (one output row per left row) +/// +/// Delegates when: +/// - The plan is not a supported join type +/// - Either input lacks row count information #[deprecated( since = "56.0.0", - note = "multi-key join estimation belongs in the join operators, see https://github.com/apache/datafusion/issues/21583; without a matching provider the statistics walk uses the operator's own estimate" + note = "replaces the estimates of `HashJoinExec`, `SortMergeJoinExec` and `CrossJoinExec` with a separate estimation logic; without a matching provider the statistics walk uses the operator's own estimate" )] #[derive(Debug, Default)] pub struct JoinStatisticsProvider; @@ -861,6 +867,7 @@ impl StatisticsProvider for JoinStatisticsProvider { fn matches(&self, plan: &dyn ExecutionPlan) -> bool { plan.downcast_ref::().is_some() || plan.downcast_ref::().is_some() + || plan.downcast_ref::().is_some() } fn compute_statistics( @@ -868,66 +875,109 @@ impl StatisticsProvider for JoinStatisticsProvider { plan: &dyn ExecutionPlan, child_stats: &[ExtendedStatistics], ) -> Result { - if child_stats.len() < 2 { - return Ok(StatisticsResult::Delegate); - } + use datafusion_common::JoinType; + use datafusion_physical_expr::expressions::Column; - let (on, join_type) = if let Some(hash_join) = plan.downcast_ref::() - { - (hash_join.on(), *hash_join.join_type()) - } else if let Some(sort_merge_join) = plan.downcast_ref::() { - (sort_merge_join.on(), sort_merge_join.join_type()) - } else { - return Ok(StatisticsResult::Delegate); - }; - if join_type != JoinType::Inner || on.len() < 2 { + if child_stats.len() < 2 { return Ok(StatisticsResult::Delegate); } let left = &child_stats[0].base; let right = &child_stats[1].base; + let (Some(&left_rows), Some(&right_rows)) = (left.num_rows.get_value(), right.num_rows.get_value()) else { return Ok(StatisticsResult::Delegate); }; - let mut ndv_divisor: usize = 1; - for (left_key, right_key) in on { - let left_ndv = key_distinct_count(left_key, left); - let right_ndv = key_distinct_count(right_key, right); - match (left_ndv, right_ndv) { - (Some(l), Some(r)) if l > 0 && r > 0 => { - ndv_divisor = ndv_divisor.saturating_mul(l.max(r)); + /// Estimate equi-join output using NDV of join key columns: + /// left_rows * right_rows / product(max(left_ndv_i, right_ndv_i)) + /// Falls back to Cartesian product if any key lacks NDV on both sides. + fn equi_join_estimate( + on: JoinOnRef, + left: &Statistics, + right: &Statistics, + left_rows: usize, + right_rows: usize, + ) -> usize { + if on.is_empty() { + return left_rows.saturating_mul(right_rows); + } + let mut ndv_divisor: usize = 1; + for (left_key, right_key) in on { + let left_ndv = left_key + .downcast_ref::() + .and_then(|c| left.column_statistics.get(c.index())) + .and_then(|s| s.distinct_count.get_value().copied()); + let right_ndv = right_key + .downcast_ref::() + .and_then(|c| right.column_statistics.get(c.index())) + .and_then(|s| s.distinct_count.get_value().copied()); + match (left_ndv, right_ndv) { + (Some(l), Some(r)) if l > 0 && r > 0 => { + ndv_divisor = ndv_divisor.saturating_mul(l.max(r)); + } + _ => return left_rows.saturating_mul(right_rows), } - _ => return Ok(StatisticsResult::Delegate), } + let max_rows = left_rows.saturating_mul(right_rows); + max_rows.checked_div(ndv_divisor).unwrap_or(max_rows) } - let mut estimate = left_rows - .saturating_mul(right_rows) - .checked_div(ndv_divisor) - .unwrap_or(usize::MAX); - if let Some(fetch) = plan.fetch() { - estimate = estimate.min(fetch); - } + let (inner_estimate, is_exact_cartesian, join_type) = if let Some(hash_join) = + plan.downcast_ref::() + { + let est = + equi_join_estimate(hash_join.on(), left, right, left_rows, right_rows); + (est, false, *hash_join.join_type()) + } else if let Some(smj) = plan.downcast_ref::() { + let est = equi_join_estimate(smj.on(), left, right, left_rows, right_rows); + (est, false, smj.join_type()) + } else if plan.downcast_ref::().is_some() { + let both_exact = left.num_rows.is_exact().unwrap_or(false) + && right.num_rows.is_exact().unwrap_or(false); + ( + left_rows.saturating_mul(right_rows), + both_exact, + JoinType::Inner, + ) + } else { + return Ok(StatisticsResult::Delegate); + }; - let child_base: Vec> = child_stats - .iter() - .map(|c| Arc::clone(c.base_arc())) - .collect(); - let mut stats = Arc::unwrap_or_clone( - plan.statistics_from_inputs(&child_base, &StatisticsArgs::new())?, - ); - rescale_byte_size(&mut stats, Precision::Inexact(estimate)); - Ok(StatisticsResult::Computed(ExtendedStatistics::new(stats))) - } -} + // Apply join-type-aware cardinality bounds + let estimated = match join_type { + JoinType::Inner => inner_estimate, + JoinType::Left => inner_estimate.max(left_rows), + JoinType::Right => inner_estimate.max(right_rows), + JoinType::Full => { + // At least left + right - matched, but never less than inner + let outer_bound = left_rows + .saturating_add(right_rows) + .saturating_sub(inner_estimate); + inner_estimate.max(outer_bound) + } + JoinType::LeftSemi => inner_estimate.min(left_rows), + JoinType::RightSemi => inner_estimate.min(right_rows), + JoinType::LeftAnti => left_rows.saturating_sub(inner_estimate.min(left_rows)), + JoinType::RightAnti => { + right_rows.saturating_sub(inner_estimate.min(right_rows)) + } + JoinType::LeftMark => left_rows, + JoinType::RightMark => right_rows, + }; -fn key_distinct_count(key: &Arc, stats: &Statistics) -> Option { - key.downcast_ref::() - .and_then(|c| stats.column_statistics.get(c.index())) - .and_then(|s| s.distinct_count.get_value().copied()) + // NL join inner with exact inputs is an exact Cartesian product; + // NDV-based estimates are inherently inexact. + let num_rows = if is_exact_cartesian && join_type == JoinType::Inner { + Precision::Exact(estimated) + } else { + Precision::Inexact(estimated) + }; + + computed_with_row_count(plan, num_rows, child_stats) + } } /// Statistics provider for [`LocalLimitExec`] and @@ -1796,84 +1846,6 @@ mod tests { Ok(()) } - #[test] - fn test_filter_provider_ndv_uses_post_fetch_rows() -> Result<()> { - use crate::filter::FilterExecBuilder; - - // 1000 rows with 100 distinct values; `fetch` keeps 200 rows - let source = make_source_with_ndvs(1000, Some(100), None); - let filter = FilterExecBuilder::new(lit(true), source) - .with_fetch(Some(200)) - .build()?; - let operator = compute(&StatisticsRegistry::new(), &filter)?; - let registry = - StatisticsRegistry::with_providers(vec![Arc::new(FilterStatisticsProvider)]); - let stats = compute(®istry, &filter)?; - - // The selectivity is the post-fetch row count over the input row count, - // and the survival model starts from the input distinct count - assert_eq!(stats.base.num_rows, Precision::Inexact(200)); - assert_eq!( - operator.base.column_statistics[0].distinct_count, - Precision::Inexact(100) - ); - let expected = ndv_after_selectivity(100, 1000, 200.0 / 1000.0); - assert_eq!(expected, 89); - assert_eq!( - stats.base.column_statistics[0].distinct_count, - Precision::Inexact(expected) - ); - Ok(()) - } - - #[test] - fn test_filter_provider_keeps_distinct_count_of_zero_or_one() -> Result<()> { - for ndv in [0, 1] { - // Column "a" has 1000 distinct values, column "b" has `ndv` - let mut a_stats = ColumnStatistics::new_unknown(); - a_stats.distinct_count = Precision::Exact(1000); - a_stats.min_value = Precision::Exact(ScalarValue::Int32(Some(1))); - a_stats.max_value = Precision::Exact(ScalarValue::Int32(Some(1000))); - let mut b_stats = ColumnStatistics::new_unknown(); - b_stats.distinct_count = Precision::Exact(ndv); - let source: Arc = - Arc::new(MockSourceExec::with_column_stats( - make_schema(), - Precision::Exact(1000), - vec![a_stats, b_stats], - )); - - // The predicate "a > 900" keeps about 10% of the rows - let predicate: Arc = Arc::new(BinaryExpr::new( - col("a", &make_schema())?, - Operator::Gt, - lit(900i32), - )); - let filter: Arc = - Arc::new(FilterExec::try_new(predicate, source)?); - - let operator = compute(&StatisticsRegistry::new(), filter.as_ref())?; - let registry = StatisticsRegistry::with_providers(vec![Arc::new( - FilterStatisticsProvider, - )]); - let stats = compute(®istry, filter.as_ref())?; - - assert!( - stats - .base - .num_rows - .get_value() - .is_some_and(|rows| *rows < 1000) - ); - assert_eq!( - stats.base.column_statistics[1].distinct_count, - operator.base.column_statistics[1].distinct_count, - "distinct count {ndv}" - ); - } - Ok(()) - } - #[test] fn test_filter_adjusts_ndv_by_selectivity() -> Result<()> { use datafusion_common::ScalarValue; @@ -2275,146 +2247,340 @@ mod tests { use crate::joins::{HashJoinExec, PartitionMode}; use datafusion_common::{JoinType, NullEquality}; - fn make_source_with_ndvs( + fn make_source_with_ndv_2col( num_rows: usize, ndv_a: Option, - ndv_b: Option, ) -> Arc { - let column_statistics = [ndv_a, ndv_b] - .into_iter() - .map(|ndv| { + let schema = make_schema(); // "a" Int32, "b" Int32 + let col_stats = vec![ + { let mut cs = ColumnStatistics::new_unknown(); - if let Some(n) = ndv { + if let Some(n) = ndv_a { cs.distinct_count = Precision::Exact(n); } cs - }) - .collect(); + }, + ColumnStatistics::new_unknown(), + ]; Arc::new(MockSourceExec::with_column_stats( - make_schema(), // "a" Int32, "b" Int32 + schema, Precision::Exact(num_rows), - column_statistics, + col_stats, )) } - fn join_on(num_keys: usize) -> JoinOn { - ["a", "b"] - .iter() - .enumerate() - .take(num_keys) - .map(|(index, name)| { - ( - Arc::new(Column::new(name, index)) as Arc, - Arc::new(Column::new(name, index)) as Arc, - ) - }) - .collect() - } - fn make_hash_join( left: Arc, right: Arc, - num_keys: usize, - join_type: JoinType, - fetch: Option, ) -> Result> { + let _schema = make_schema(); + let on: JoinOn = vec![( + Arc::new(Column::new("a", 0)) as Arc, + Arc::new(Column::new("a", 0)) as Arc, + )]; + Ok(Arc::new(HashJoinExec::try_new( + left, + right, + on, + None, + &JoinType::Inner, + None, + PartitionMode::CollectLeft, + NullEquality::NullEqualsNull, + false, + )?)) + } + + #[test] + fn test_join_provider_with_ndv() -> Result<()> { + // left: 1000 rows, NDV(a)=100; right: 500 rows, NDV(a)=50 + // expected = 1000 * 500 / max(100, 50) = 5000 + let left = make_source_with_ndv_2col(1000, Some(100)); + let right = make_source_with_ndv_2col(500, Some(50)); + let join = make_hash_join(left, right)?; + + let registry = + StatisticsRegistry::with_providers(vec![Arc::new(JoinStatisticsProvider)]); + let stats = compute(®istry, join.as_ref())?; + assert_eq!(stats.base.num_rows, Precision::Inexact(5000)); + Ok(()) + } + + #[test] + fn test_join_provider_uses_actual_key_column_ndv() -> Result<()> { + // Join on column "b" (index 1), NDV only set on "b", not "a". + // Old first()-based code would look up column 0 (a), find no NDV, + // and fall back to Cartesian product. The fix looks up column 1 (b). + // left: 1000 rows, NDV(b)=50; right: 500 rows, NDV(b)=25 + // expected = 1000 * 500 / max(50, 25) = 10000 + let schema = make_schema(); // "a" Int32, "b" Int32 + let make_source_ndv_b = + |num_rows: usize, ndv_b: usize| -> Arc { + let col_stats = vec![ + ColumnStatistics::new_unknown(), // "a": no NDV + { + let mut cs = ColumnStatistics::new_unknown(); + cs.distinct_count = Precision::Exact(ndv_b); + cs + }, + ]; + Arc::new(MockSourceExec::with_column_stats( + Arc::clone(&schema), + Precision::Exact(num_rows), + col_stats, + )) + }; + + let left = make_source_ndv_b(1000, 50); + let right = make_source_ndv_b(500, 25); + + // Join on column "b" (index 1) + let on: JoinOn = vec![( + Arc::new(Column::new("b", 1)) as Arc, + Arc::new(Column::new("b", 1)) as Arc, + )]; let join: Arc = Arc::new(HashJoinExec::try_new( left, right, - join_on(num_keys), + on, None, - &join_type, + &JoinType::Inner, None, PartitionMode::CollectLeft, NullEquality::NullEqualsNull, false, )?); - match fetch { - Some(_) => Ok(join.with_fetch(fetch).expect("hash join supports fetch")), - None => Ok(join), - } + + let registry = + StatisticsRegistry::with_providers(vec![Arc::new(JoinStatisticsProvider)]); + let stats = compute(®istry, join.as_ref())?; + assert_eq!(stats.base.num_rows, Precision::Inexact(10_000)); + Ok(()) + } + + #[test] + fn test_join_provider_multi_key_ndv() -> Result<()> { + // Multi-key join: ON a.a = b.a AND a.b = b.b + // left: 1000 rows, NDV(a)=100, NDV(b)=20 + // right: 500 rows, NDV(a)=50, NDV(b)=10 + // expected = 1000 * 500 / (max(100,50) * max(20,10)) = 500000 / 2000 = 250 + let schema = make_schema(); // "a" Int32, "b" Int32 + let make_source_2ndv = + |num_rows: usize, ndv_a: usize, ndv_b: usize| -> Arc { + let col_stats = vec![ + { + let mut cs = ColumnStatistics::new_unknown(); + cs.distinct_count = Precision::Exact(ndv_a); + cs + }, + { + let mut cs = ColumnStatistics::new_unknown(); + cs.distinct_count = Precision::Exact(ndv_b); + cs + }, + ]; + Arc::new(MockSourceExec::with_column_stats( + Arc::clone(&schema), + Precision::Exact(num_rows), + col_stats, + )) + }; + + let left = make_source_2ndv(1000, 100, 20); + let right = make_source_2ndv(500, 50, 10); + + let on: JoinOn = vec![ + ( + Arc::new(Column::new("a", 0)) as Arc, + Arc::new(Column::new("a", 0)) as Arc, + ), + ( + Arc::new(Column::new("b", 1)) as Arc, + Arc::new(Column::new("b", 1)) as Arc, + ), + ]; + let join: Arc = Arc::new(HashJoinExec::try_new( + left, + right, + on, + None, + &JoinType::Inner, + None, + PartitionMode::CollectLeft, + NullEquality::NullEqualsNull, + false, + )?); + + let registry = + StatisticsRegistry::with_providers(vec![Arc::new(JoinStatisticsProvider)]); + let stats = compute(®istry, join.as_ref())?; + assert_eq!(stats.base.num_rows, Precision::Inexact(250)); + Ok(()) + } + + #[test] + fn test_join_provider_fallback_cartesian() -> Result<()> { + // No NDV available -> Cartesian product estimate + let left = make_source_with_ndv_2col(100, None); + let right = make_source_with_ndv_2col(200, None); + let join = make_hash_join(left, right)?; + + let registry = + StatisticsRegistry::with_providers(vec![Arc::new(JoinStatisticsProvider)]); + let stats = compute(®istry, join.as_ref())?; + assert_eq!(stats.base.num_rows, Precision::Inexact(20_000)); + Ok(()) } - /// Two-key inner join: left has 1000 rows with NDV(a)=100 and NDV(b)=20, - /// right has 500 rows with NDV(a)=50 and NDV(b)=10. The product formula - /// gives 1000 * 500 / (max(100, 50) * max(20, 10)) = 250 rows. - fn make_two_key_join( + #[test] + fn test_nl_join_delegates() -> Result<()> { + use crate::joins::NestedLoopJoinExec; + + // NL join delegates to the built-in (NestedLoopJoinExec may have an + // arbitrary JoinFilter, so the provider cannot safely assume Cartesian). + let left = make_source(100); + let right = make_source(200); + let join: Arc = Arc::new(NestedLoopJoinExec::try_new( + left, + right, + None, + &JoinType::Inner, + None, + )?); + + let registry = + StatisticsRegistry::with_providers(vec![Arc::new(JoinStatisticsProvider)]); + let stats = compute(®istry, join.as_ref())?; + // Provider delegates; result comes from built-in statistics_from_inputs. + assert!( + stats.base.num_rows.get_value().is_some() + || matches!(stats.base.num_rows, Precision::Absent) + ); + Ok(()) + } + + fn make_hash_join_typed( + left: Arc, + right: Arc, join_type: JoinType, - fetch: Option, ) -> Result> { - make_hash_join( - make_source_with_ndvs(1000, Some(100), Some(20)), - make_source_with_ndvs(500, Some(50), Some(10)), - 2, - join_type, - fetch, - ) + let on: JoinOn = vec![( + Arc::new(Column::new("a", 0)) as Arc, + Arc::new(Column::new("a", 0)) as Arc, + )]; + Ok(Arc::new(HashJoinExec::try_new( + left, + right, + on, + None, + &join_type, + None, + PartitionMode::CollectLeft, + NullEquality::NullEqualsNull, + false, + )?)) } - fn join_provider_registry() -> StatisticsRegistry { - StatisticsRegistry::with_providers(vec![Arc::new(JoinStatisticsProvider)]) + fn compute_join_rows( + left_rows: usize, + left_ndv: Option, + right_rows: usize, + right_ndv: Option, + join_type: JoinType, + ) -> Result> { + let left = make_source_with_ndv_2col(left_rows, left_ndv); + let right = make_source_with_ndv_2col(right_rows, right_ndv); + let join = make_hash_join_typed(left, right, join_type)?; + let registry = + StatisticsRegistry::with_providers(vec![Arc::new(JoinStatisticsProvider)]); + Ok(compute(®istry, join.as_ref())?.base.num_rows) } #[test] - fn test_join_provider_multi_key_inner_refines() -> Result<()> { - let join = make_two_key_join(JoinType::Inner, None)?; - let operator = compute(&StatisticsRegistry::new(), join.as_ref())?; - let refined = compute(&join_provider_registry(), join.as_ref())?; - assert_ne!(operator.base.num_rows, Precision::Inexact(250)); - assert_eq!(refined.base.num_rows, Precision::Inexact(250)); + fn test_join_provider_left_outer() -> Result<()> { + // left=1000, right=500, NDV(a)=100/50 + // inner estimate = 1000*500/100 = 5000, already >= left_rows + // Left outer: max(5000, 1000) = 5000 + assert_eq!( + compute_join_rows(1000, Some(100), 500, Some(50), JoinType::Left)?, + Precision::Inexact(5000) + ); + // Small inner estimate: left=1000, right=10, NDV=100/100 + // inner = 1000*10/100 = 100, left outer = max(100, 1000) = 1000 assert_eq!( - refined.base.column_statistics, - operator.base.column_statistics + compute_join_rows(1000, Some(100), 10, Some(100), JoinType::Left)?, + Precision::Inexact(1000) ); Ok(()) } #[test] - fn test_join_provider_caps_at_fetch() -> Result<()> { - let join = make_two_key_join(JoinType::Inner, Some(100))?; - let stats = compute(&join_provider_registry(), join.as_ref())?; - assert_eq!(stats.base.num_rows, Precision::Inexact(100)); + fn test_join_provider_right_outer() -> Result<()> { + // inner = 1000*10/100 = 100, right outer = max(100, 10) = 100 + assert_eq!( + compute_join_rows(1000, Some(100), 10, Some(100), JoinType::Right)?, + Precision::Inexact(100) + ); + // inner = 10*1000/100 = 100, right outer = max(100, 1000) = 1000 + assert_eq!( + compute_join_rows(10, Some(100), 1000, Some(100), JoinType::Right)?, + Precision::Inexact(1000) + ); + Ok(()) + } - // A fetch above the estimate does not change it - let join = make_two_key_join(JoinType::Inner, Some(1000))?; - let stats = compute(&join_provider_registry(), join.as_ref())?; - assert_eq!(stats.base.num_rows, Precision::Inexact(250)); + #[test] + fn test_join_provider_semi_join() -> Result<()> { + // inner = 5000, left semi = min(5000, 1000) = 1000 + assert_eq!( + compute_join_rows(1000, Some(100), 500, Some(50), JoinType::LeftSemi)?, + Precision::Inexact(1000) + ); + // inner = 5000, right semi = min(5000, 500) = 500 + assert_eq!( + compute_join_rows(1000, Some(100), 500, Some(50), JoinType::RightSemi)?, + Precision::Inexact(500) + ); + // Cartesian fallback (no NDV): inner = 1000*500 = 500000, + // left semi = min(500000, 1000) = 1000 (selectivity = 1.0) + assert_eq!( + compute_join_rows(1000, None, 500, None, JoinType::LeftSemi)?, + Precision::Inexact(1000) + ); Ok(()) } #[test] - fn test_join_provider_delegates_to_operator() -> Result<()> { - let single_key = make_hash_join( - make_source_with_ndvs(1000, Some(100), None), - make_source_with_ndvs(500, Some(50), None), - 1, - JoinType::Inner, - None, - )?; - let missing_ndv = make_hash_join( - make_source_with_ndvs(1000, Some(100), Some(20)), - make_source_with_ndvs(500, Some(50), None), - 2, - JoinType::Inner, - None, - )?; - let zero_ndv = make_hash_join( - make_source_with_ndvs(1000, Some(100), Some(0)), - make_source_with_ndvs(500, Some(50), Some(10)), - 2, - JoinType::Inner, - None, - )?; - let mut joins = vec![single_key, missing_ndv, zero_ndv]; - for join_type in [JoinType::Left, JoinType::LeftSemi, JoinType::LeftAnti] { - joins.push(make_two_key_join(join_type, None)?); - } + fn test_join_provider_anti_join() -> Result<()> { + // inner = 1000*10/100 = 100, left anti = 1000 - min(100, 1000) = 900 + assert_eq!( + compute_join_rows(1000, Some(100), 10, Some(100), JoinType::LeftAnti)?, + Precision::Inexact(900) + ); + // inner = 5000, right anti = 500 - min(5000, 500) = 0 + assert_eq!( + compute_join_rows(1000, Some(100), 500, Some(50), JoinType::RightAnti)?, + Precision::Inexact(0) + ); + Ok(()) + } - for join in joins { - let operator = compute(&StatisticsRegistry::new(), join.as_ref())?; - let refined = compute(&join_provider_registry(), join.as_ref())?; - assert_eq!(refined.base(), operator.base()); - } + // ========================================================================= + // CrossJoinExec tests (handled by JoinStatisticsProvider) + // ========================================================================= + + #[test] + fn test_cross_join_provider_exact() -> Result<()> { + use crate::joins::CrossJoinExec; + let left = make_source(100); + let right = make_source(200); + let join: Arc = Arc::new(CrossJoinExec::new(left, right)); + + let registry = + StatisticsRegistry::with_providers(vec![Arc::new(JoinStatisticsProvider)]); + let stats = compute(®istry, join.as_ref())?; + // Both inputs have Exact row counts -> result is also Exact + assert_eq!(stats.base.num_rows, Precision::Exact(20_000)); Ok(()) } From 2b3c10d1c2d3c5d215e5961f99fb59ac4184361e Mon Sep 17 00:00:00 2001 From: Alessandro Solimando Date: Mon, 5 Oct 2026 14:54:08 +0200 Subject: [PATCH 4/6] docs: simplify the join_reorder statistics example to catalog statistics --- datafusion-examples/README.md | 6 +- .../examples/statistics/join_reorder.rs | 87 ++++--------------- .../examples/statistics/main.rs | 2 +- 3 files changed, 22 insertions(+), 73 deletions(-) diff --git a/datafusion-examples/README.md b/datafusion-examples/README.md index 6c7aad31a9233..42e3410025e98 100644 --- a/datafusion-examples/README.md +++ b/datafusion-examples/README.md @@ -220,9 +220,9 @@ cargo run --example dataframe -- dataframe #### Category: Single Process -| Subcommand | File Path | Description | -| ------------ | ------------------------------------------------------------------- | ----------------------------------------------------------------------- | -| join_reorder | [`statistics/join_reorder.rs`](examples/statistics/join_reorder.rs) | Supply and refine column statistics via a provider to flip a join order | +| Subcommand | File Path | Description | +| ------------ | ------------------------------------------------------------------- | -------------------------------------------------------------------- | +| join_reorder | [`statistics/join_reorder.rs`](examples/statistics/join_reorder.rs) | Supply catalog column statistics via a provider to flip a join order | ## UDF Examples diff --git a/datafusion-examples/examples/statistics/join_reorder.rs b/datafusion-examples/examples/statistics/join_reorder.rs index cdb8664d6da43..42ab2cfbcc4c7 100644 --- a/datafusion-examples/examples/statistics/join_reorder.rs +++ b/datafusion-examples/examples/statistics/join_reorder.rs @@ -15,21 +15,18 @@ // specific language governing permissions and limitations // under the License. -//! Plug refined cardinality estimation into the optimizer via the `StatisticsRegistry`. +//! Supply catalog statistics to the optimizer via the `StatisticsRegistry`. //! -//! DataFusion's built-in statistics are intentionally simple defaults; the registry -//! is the seam for plugging in more refined estimation (formulas or heuristics -//! suited to your system or data) without changing the core. +//! The registry is the extension point for statistics the core cannot derive, +//! such as column statistics a catalog knows. //! -//! `(SELECT user_id FROM events WHERE amount < 50 GROUP BY user_id) JOIN dims`: a -//! provider supplies the column stats a catalog knows (`amount` range, `user_id` -//! distinct count); a second provider then refines the post-filter distinct -//! count with the survival formula -//! `NDV * (1 - (1 - selectivity)^(rows / NDV))` (Yao/Cardenas) to ~40, below `dims` -//! (48), flipping the join build side. Core's simpler `min(NDV, rows)` cap would -//! give 50 (> 48) and keep the other order; the refinement is the point. The -//! ground-truth query prints the true surviving distinct count (below 48), -//! confirming the flip. +//! `(SELECT user_id FROM events WHERE amount < 50 GROUP BY user_id) JOIN dims`: +//! the in-memory `events` table has no column statistics, so the grouped side is +//! estimated at 1000 rows and `dims` (60 rows) becomes the build side. A provider +//! supplies the `amount` range and the `user_id` distinct count a catalog knows; +//! with them the operators estimate the grouped side at 50 rows, below `dims`, +//! which flips the join build side. The ground-truth query prints the true +//! distinct count (below 60), confirming the flip. use std::sync::Arc; @@ -43,10 +40,8 @@ use datafusion::common::ScalarValue; use datafusion::common::stats::Precision; use datafusion::execution::SessionStateBuilder; use datafusion::physical_plan::ExecutionPlan; -use datafusion::physical_plan::filter::FilterExec; use datafusion::physical_plan::operator_statistics::{ ClosureStatisticsProvider, ExtendedStatistics, StatisticsRegistry, StatisticsResult, - ndv_after_selectivity, }; use datafusion::physical_plan::statistics::StatisticsArgs; use datafusion::prelude::*; @@ -78,45 +73,6 @@ fn catalog_stats( Ok(StatisticsResult::Computed(ExtendedStatistics::new(stats))) } -fn filter_matches(plan: &dyn ExecutionPlan) -> bool { - plan.downcast_ref::().is_some() -} - -/// Starts from the filter's own estimate and lowers each column's distinct -/// count with the survival formula, applied to the input distinct count. -fn filter_stats( - plan: &dyn ExecutionPlan, - child_stats: &[ExtendedStatistics], -) -> Result { - let Some(filter) = plan.downcast_ref::() else { - return Ok(StatisticsResult::Delegate); - }; - let input = child_stats[0].base(); - let mut stats = (*plan.statistics_from_inputs( - &[Arc::clone(child_stats[0].base_arc())], - &StatisticsArgs::new(), - )?) - .clone(); - if let (Some(&input_rows), Some(&output_rows)) = - (input.num_rows.get_value(), stats.num_rows.get_value()) - && output_rows < input_rows - { - let selectivity = output_rows as f64 / input_rows as f64; - for (i, col) in stats.column_statistics.iter_mut().enumerate() { - let input_index = filter.projection().as_ref().map_or(i, |p| p[i]); - if let Some(&ndv) = col.distinct_count.get_value() - && let Some(&input_ndv) = input.column_statistics[input_index] - .distinct_count - .get_value() - { - let survived = ndv_after_selectivity(input_ndv, input_rows, selectivity); - col.distinct_count = Precision::Inexact(survived.min(ndv)); - } - } - } - Ok(StatisticsResult::Computed(ExtendedStatistics::new(stats))) -} - fn int_col(values: &[i32]) -> Arc { Arc::new(Int32Array::from_iter_values(values.iter().copied())) } @@ -152,16 +108,9 @@ fn build_ctx(with_registry: bool) -> Result { .with_config(config) .with_default_features(); if with_registry { - let registry = StatisticsRegistry::with_providers(vec![ - Arc::new(ClosureStatisticsProvider::with_matches( - catalog_matches, - catalog_stats, - )), - Arc::new(ClosureStatisticsProvider::with_matches( - filter_matches, - filter_stats, - )), - ]); + let registry = StatisticsRegistry::with_providers(vec![Arc::new( + ClosureStatisticsProvider::with_matches(catalog_matches, catalog_stats), + )]); builder = builder.with_statistics_registry(registry); } let ctx = SessionContext::new_with_state(builder.build()); @@ -181,8 +130,8 @@ fn build_ctx(with_registry: bool) -> Result { ctx.register_table( "dims", mem_table(&[ - ("user_id", int_col(&(0..48).collect::>())), - ("label", int_col(&(0..48).collect::>())), + ("user_id", int_col(&(0..60).collect::>())), + ("label", int_col(&(0..60).collect::>())), ])?, )?; Ok(ctx) @@ -213,13 +162,13 @@ pub async fn join_reorder() -> Result<()> { "A hash join builds its in-memory hash table from one input and probes with\n\ the other, so the smaller input should be the build side. Default estimation\n\ sizes the grouped `events` at 1000 rows and builds from `dims`; the\n\ - registry's refined ~40 estimate is below `dims` (48 rows), so it flips the\n\ - build side to `events`. The ground-truth count above (also below 48)\n\ + catalog statistics bring the estimate to 50, below `dims` (60 rows), so the\n\ + build side flips to `events`. The ground-truth count above (also below 60)\n\ confirms `events` really is the smaller, cheaper side.\n" ); println!("-- Without the registry (default estimation) --"); println!("{}\n", explain(&build_ctx(false)?).await?); - println!("-- With the registry (refined estimation) --"); + println!("-- With the registry (catalog statistics) --"); println!("{}", explain(&build_ctx(true)?).await?); Ok(()) } diff --git a/datafusion-examples/examples/statistics/main.rs b/datafusion-examples/examples/statistics/main.rs index 313e18645d50f..25e74fcf82612 100644 --- a/datafusion-examples/examples/statistics/main.rs +++ b/datafusion-examples/examples/statistics/main.rs @@ -30,7 +30,7 @@ //! - `all`: run all examples included in this module //! //! - `join_reorder` -//! (file: join_reorder.rs, desc: Supply and refine column statistics via a provider to flip a join order) +//! (file: join_reorder.rs, desc: Supply catalog column statistics via a provider to flip a join order) mod join_reorder; From b74c6b5c7c7e3d801c28cd5c0f61de35b3a1e21c Mon Sep 17 00:00:00 2001 From: Alessandro Solimando Date: Mon, 5 Oct 2026 15:04:53 +0200 Subject: [PATCH 5/6] docs: drop issue link from the statistics providers upgrade note --- docs/source/library-user-guide/upgrading/56.0.0.md | 2 -- 1 file changed, 2 deletions(-) diff --git a/docs/source/library-user-guide/upgrading/56.0.0.md b/docs/source/library-user-guide/upgrading/56.0.0.md index 1306887988db9..296d3baefed31 100644 --- a/docs/source/library-user-guide/upgrading/56.0.0.md +++ b/docs/source/library-user-guide/upgrading/56.0.0.md @@ -445,8 +445,6 @@ providers. - To keep a bundled estimation technique, implement it in a user-defined `StatisticsProvider`. -See [issue #25571](https://github.com/apache/datafusion/issues/25571) for details. - ### `StatisticsRegistry::compute` and `compute_base` are deprecated Use the walk instead: From 78117138d8b82141b9ed9043ab63bc239bedc9fe Mon Sep 17 00:00:00 2001 From: Alessandro Solimando Date: Mon, 5 Oct 2026 15:11:08 +0200 Subject: [PATCH 6/6] test: address review comments on the statistics registry test --- datafusion/physical-plan/src/operator_statistics/mod.rs | 2 +- datafusion/sqllogictest/src/test_context.rs | 3 ++- datafusion/sqllogictest/test_files/statistics_registry.slt | 4 +--- 3 files changed, 4 insertions(+), 5 deletions(-) diff --git a/datafusion/physical-plan/src/operator_statistics/mod.rs b/datafusion/physical-plan/src/operator_statistics/mod.rs index 50124b2ac1bcf..fd5c20350331b 100644 --- a/datafusion/physical-plan/src/operator_statistics/mod.rs +++ b/datafusion/physical-plan/src/operator_statistics/mod.rs @@ -857,7 +857,7 @@ impl StatisticsProvider for AggregateStatisticsProvider { /// - Either input lacks row count information #[deprecated( since = "56.0.0", - note = "replaces the estimates of `HashJoinExec`, `SortMergeJoinExec` and `CrossJoinExec` with a separate estimation logic; without a matching provider the statistics walk uses the operator's own estimate" + note = "replaces the estimates of `HashJoinExec`, `SortMergeJoinExec` and `CrossJoinExec` with separate estimation logic; without a matching provider the statistics walk uses the operator's own estimate" )] #[derive(Debug, Default)] pub struct JoinStatisticsProvider; diff --git a/datafusion/sqllogictest/src/test_context.rs b/datafusion/sqllogictest/src/test_context.rs index e4870622a5c26..0c8c2c5add363 100644 --- a/datafusion/sqllogictest/src/test_context.rs +++ b/datafusion/sqllogictest/src/test_context.rs @@ -166,7 +166,8 @@ impl TestContext { let mut stats = Arc::unwrap_or_clone( plan.statistics_from_inputs(&child_base, &StatisticsArgs::new())?, ); - stats.num_rows = Precision::Inexact(left_rows * right_rows); + stats.num_rows = + Precision::Inexact(left_rows.saturating_mul(right_rows)); Ok(StatisticsResult::Computed(stats.into())) }, ); diff --git a/datafusion/sqllogictest/test_files/statistics_registry.slt b/datafusion/sqllogictest/test_files/statistics_registry.slt index 7bf7dc1779c01..0314c0d9e544c 100644 --- a/datafusion/sqllogictest/test_files/statistics_registry.slt +++ b/datafusion/sqllogictest/test_files/statistics_registry.slt @@ -25,12 +25,10 @@ # orders (10 rows, same distribution) # dim_small (50 rows) # -# Built-in operator stats give 10*10 / NDV(3) = 33 < 50, which would keep the +# The operators' own estimates give 10*10 / NDV(3) = 33 < 50, which would keep the # inner join on the build side (wrong; actual output is 66). The registry's # conservative estimate 10*10 = 100 > 50 swaps dim_small to the build side. # -# Parquet files written by COPY TO carry min/max stats (NDV=3 via range) but no -# distinct_count, so the registry falls back to the cartesian product upper bound. # Threshold settings force Partitioned mode so statistics alone drive the swap. statement ok