From acd2d6130ef3103f27f47aa4be724fbc2f4141ca Mon Sep 17 00:00:00 2001 From: mohitgurav20 Date: Fri, 25 Sep 2026 02:26:36 +0530 Subject: [PATCH 01/11] refactor: use NDV for unresolved scalar subquery selectivity instead of 20% fallback --- datafusion/physical-plan/src/filter.rs | 62 ++++++++++++++++++++++++-- 1 file changed, 58 insertions(+), 4 deletions(-) diff --git a/datafusion/physical-plan/src/filter.rs b/datafusion/physical-plan/src/filter.rs index f71d67d04e39c..107e2c6a678fc 100644 --- a/datafusion/physical-plan/src/filter.rs +++ b/datafusion/physical-plan/src/filter.rs @@ -414,10 +414,15 @@ impl FilterExec { ); (selectivity, filtered_num_rows, cs) } else { - // Without interval boundaries, use the default selectivity and - // apply the row-count constraints that still follow from the - // filter predicate. - let selectivity = default_selectivity as f64 / 100.0; + // Without interval boundaries, attempt a heuristic fallback for selectivities. + // For instance, an equality filter against an unresolved scalar subquery will + // fail `check_support`, but we can still estimate selectivity as `1.0 / NDV`. + let selectivity = compute_fallback_selectivity( + predicate, + &input_stats.column_statistics, + default_selectivity, + ); + let filtered_num_rows = input_num_rows.with_estimated_selectivity(selectivity); let mut cs = input_stats.to_inexact().column_statistics; @@ -455,7 +460,56 @@ impl FilterExec { column_statistics, }) } +} + +/// Fallback heuristic for when interval analysis is unsupported (e.g. scalar subqueries). +/// Iterates over conjunctions; uses `1.0 / NDV` for equalities involving a column with known NDV, +/// and applies `default_selectivity` for all other predicates. +fn compute_fallback_selectivity( + predicate: &Arc, + column_statistics: &[ColumnStatistics], + default_selectivity: u8, +) -> f64 { + let exprs = split_conjunction(predicate); + let mut selectivity = 1.0; + + for expr in exprs { + let mut handled = false; + if let Some(binary) = expr.downcast_ref::() { + if binary.op() == &Operator::Eq { + let col = if let Some(c) = binary.left().downcast_ref::() { + Some(c) + } else if let Some(c) = binary.right().downcast_ref::() { + Some(c) + } else { + None + }; + + if let Some(c) = col { + if let Some(stat) = column_statistics.get(c.index()) { + match stat.distinct_count { + Precision::Exact(ndv) | Precision::Inexact(ndv) + if ndv > 0 => + { + selectivity *= 1.0 / (ndv as f64); + handled = true; + } + _ => {} + } + } + } + } + } + + if !handled { + selectivity *= default_selectivity as f64 / 100.0; + } + } + selectivity +} + +impl FilterExec { /// This function creates the cache object that stores the plan properties such as schema, equivalence properties, ordering, partitioning, etc. fn compute_properties( input: &Arc, From bcc68fd190fe8eeb81cc4aa063fb025b04c0d90e Mon Sep 17 00:00:00 2001 From: mohitgurav20 Date: Sat, 3 Oct 2026 11:56:54 +0530 Subject: [PATCH 02/11] fix: address review feedback for fallback selectivity heuristic Rework compute_fallback_selectivity to resolve all issues raised by @asolimando: 1. Default selectivity is now applied at most once for all unhandled conjuncts combined, instead of multiplied per-conjunct. A predicate with three unhandled terms (e.g. LIKE, <>, IN) stays at 0.2, not 0.2^3 = 0.008. 2. For col_a = col_b equalities, use 1/max(NDV(left), NDV(right)) instead of only considering the left side's NDV. 3. Restore tests that were accidentally removed during the merge with main (fetch_preserves_singleton, fetch_statistics_match_execution, fetch_null_column_statistics, singleton_precision). 4. Add unit tests for compute_fallback_selectivity covering: - single handled equality (1/NDV) - several unhandled conjuncts (default applied once) - mixed handled and unhandled (1/NDV * default) - col = col with max(NDV) - non-equality predicate (default only) - CAST(col) = expr (not handled, shows current behavior) - utf8 col = literal (handled via NDV) 5. Add SLT plan test with scalar subquery equality showing the new row estimate. Note: utf8_col = 'literal' also fails check_support, so it now gets 1/NDV instead of 20%. This is generally more accurate since a Utf8 column with 60 distinct values should estimate ~1.7% selectivity for an equality match, not a blanket 20%. --- datafusion/physical-plan/src/filter.rs | 690 ++++++++++++++++-- .../sqllogictest/test_files/subquery.slt | 32 + 2 files changed, 645 insertions(+), 77 deletions(-) diff --git a/datafusion/physical-plan/src/filter.rs b/datafusion/physical-plan/src/filter.rs index 107e2c6a678fc..6d989d961966d 100644 --- a/datafusion/physical-plan/src/filter.rs +++ b/datafusion/physical-plan/src/filter.rs @@ -64,7 +64,7 @@ use datafusion_execution::TaskContext; use datafusion_expr::Operator; use datafusion_physical_expr::equivalence::ProjectionMapping; use datafusion_physical_expr::expressions::{ - BinaryExpr, Column, InListExpr, IsNotNullExpr, Literal, lit, + BinaryExpr, Column, InListExpr, IsNotNullExpr, IsNullExpr, Literal, lit, }; use datafusion_physical_expr::intervals::utils::check_support; use datafusion_physical_expr::utils::collect_columns; @@ -380,17 +380,8 @@ impl FilterExec { let match_limit = unique_match_limit(predicate, &input_stats); let (selectivity, num_rows, column_statistics) = if is_infeasible { - // Contradictory predicate: no rows survive. Row-bounded counts are - // zero; value statistics are undefined on an empty column. - let mut cs = input_stats.to_inexact().column_statistics; - for col_stat in &mut cs { - col_stat.distinct_count = Precision::Exact(0); - col_stat.null_count = Precision::Exact(0); - col_stat.min_value = Precision::Absent; - col_stat.max_value = Precision::Absent; - col_stat.sum_value = Precision::Absent; - col_stat.byte_size = Precision::Exact(0); - } + // Contradictory predicate: no rows survive. + let cs = vec![empty_column_statistics(); input_stats.column_statistics.len()]; (0.0, Precision::Exact(0), cs) } else { let null_rejecting_columns = collect_null_rejecting_columns(predicate); @@ -460,56 +451,88 @@ impl FilterExec { column_statistics, }) } -} -/// Fallback heuristic for when interval analysis is unsupported (e.g. scalar subqueries). -/// Iterates over conjunctions; uses `1.0 / NDV` for equalities involving a column with known NDV, -/// and applies `default_selectivity` for all other predicates. -fn compute_fallback_selectivity( - predicate: &Arc, - column_statistics: &[ColumnStatistics], - default_selectivity: u8, -) -> f64 { - let exprs = split_conjunction(predicate); - let mut selectivity = 1.0; + /// Applies the filter's `fetch` to its output statistics for `partition`, + /// or for all partitions when `partition` is `None`. The fetch stops each + /// partition separately. + #[inline] + pub(crate) fn statistics_with_fetch( + &self, + stats: Statistics, + partition: Option, + ) -> Result { + match self.fetch { + Some(fetch) => self.statistics_under_fetch(stats, fetch, partition), + None => Ok(stats), + } + } - for expr in exprs { - let mut handled = false; - if let Some(binary) = expr.downcast_ref::() { - if binary.op() == &Operator::Eq { - let col = if let Some(c) = binary.left().downcast_ref::() { - Some(c) - } else if let Some(c) = binary.right().downcast_ref::() { - Some(c) - } else { - None - }; + fn statistics_under_fetch( + &self, + stats: Statistics, + fetch: usize, + partition: Option, + ) -> Result { + if stats + .num_rows + .get_value() + .is_some_and(|rows| *rows <= fetch) + { + // No partition reaches the fetch, so no rows are dropped. + return Ok(stats); + } + let partitions = self.properties().partitioning.partition_count(); + let single = partition.is_some() || partitions <= 1; + let bound = if single { + fetch + } else { + fetch.saturating_mul(partitions) + }; + let input_columns = stats.column_statistics.clone(); + let stats = if single { + stats + } else { + // Any partition can hold more than `fetch` rows, so the total + // under the fetch is only an estimate. + stats.to_inexact() + }; - if let Some(c) = col { - if let Some(stat) = column_statistics.get(c.index()) { - match stat.distinct_count { - Precision::Exact(ndv) | Precision::Inexact(ndv) - if ndv > 0 => - { - selectivity *= 1.0 / (ndv as f64); - handled = true; - } - _ => {} - } + let mut stats = stats.with_fetch(Some(bound), 0, 1)?; + if stats.num_rows == Precision::Exact(0) { + stats.total_byte_size = Precision::Exact(0); + stats.column_statistics.fill(empty_column_statistics()); + return Ok(stats); + } + // A fetch only drops rows. A count that is exactly zero stays exact, and + // so does a column with one value in every row while rows remain. + for (column_stats, input) in stats.column_statistics.iter_mut().zip(input_columns) + { + if input.null_count == Precision::Exact(0) { + if fetch > 0 && input.is_singleton() { + column_stats.min_value = input.min_value; + column_stats.max_value = input.max_value; + if input.distinct_count == Precision::Exact(1) { + column_stats.distinct_count = input.distinct_count; } } + column_stats.null_count = input.null_count; + } else { + column_stats.null_count = + cap_at_rows(column_stats.null_count, stats.num_rows); + } + if input.distinct_count == Precision::Exact(0) { + column_stats.distinct_count = input.distinct_count; } } - - if !handled { - selectivity *= default_selectivity as f64 / 100.0; + if let Some((column, true)) = null_check_column(self.predicate()) + && let Some(column_stats) = stats.column_statistics.get_mut(column) + { + // Every surviving row is still null after the fetch. + column_stats.null_count = stats.num_rows; } + Ok(stats) } - selectivity -} - -impl FilterExec { /// This function creates the cache object that stores the plan properties such as schema, equivalence properties, ordering, partitioning, etc. fn compute_properties( input: &Arc, @@ -727,7 +750,7 @@ impl ExecutionPlan for FilterExec { fn statistics_from_inputs( &self, input_stats: &[Arc], - _args: &StatisticsArgs, + args: &StatisticsArgs, ) -> Result> { let input_stats = input_stats[0].as_ref().clone(); let stats = Self::statistics_helper( @@ -736,6 +759,7 @@ impl ExecutionPlan for FilterExec { self.predicate(), self.default_selectivity, )?; + let stats = self.statistics_with_fetch(stats, args.partition())?; Ok(Arc::new(stats.project(self.projection.as_ref()))) } @@ -1163,6 +1187,73 @@ fn holds_each_value_once(column: &ColumnStatistics, num_rows: &Precision) distinct.saturating_add(*nulls) >= *rows } +/// Heuristic selectivity for predicates that fail interval analysis. +/// +/// Splits the predicate into AND conjuncts. For each equality (`col = expr`) +/// where at least one side is a [`Column`] with a known NDV, the selectivity +/// contribution is `1 / NDV`. When both sides are columns with known NDV, +/// `1 / max(left_NDV, right_NDV)` is used. All remaining conjuncts that +/// cannot be estimated are covered by a single application of +/// `default_selectivity`, preserving the pre-existing estimate for predicates +/// that contain no recognizable equality. +fn compute_fallback_selectivity( + predicate: &Arc, + column_statistics: &[ColumnStatistics], + default_selectivity: u8, +) -> f64 { + let conjuncts = split_conjunction(predicate); + let mut selectivity = 1.0; + let mut has_unhandled = false; + + for expr in conjuncts { + let mut handled = false; + + if let Some(binary) = expr.downcast_ref::() + && *binary.op() == Operator::Eq + { + let left_ndv = column_ndv(binary.left(), column_statistics); + let right_ndv = column_ndv(binary.right(), column_statistics); + + let ndv = match (left_ndv, right_ndv) { + (Some(l), Some(r)) => Some(l.max(r)), + (Some(n), None) | (None, Some(n)) => Some(n), + (None, None) => None, + }; + + if let Some(n) = ndv { + selectivity *= 1.0 / (n as f64); + handled = true; + } + } + + if !handled { + has_unhandled = true; + } + } + + // Apply the default selectivity at most once for all unhandled conjuncts, + // so that a predicate with no handled equalities returns the default estimate. + if has_unhandled { + selectivity *= default_selectivity as f64 / 100.0; + } + + selectivity +} + +/// Extracts the NDV from a [`Column`] reference, if the expression is a bare +/// column and its statistics carry a positive distinct count. +fn column_ndv( + expr: &Arc, + column_statistics: &[ColumnStatistics], +) -> Option { + let col = expr.downcast_ref::()?; + let stat = column_statistics.get(col.index())?; + match stat.distinct_count { + Precision::Exact(ndv) | Precision::Inexact(ndv) if ndv > 0 => Some(ndv), + _ => None, + } +} + /// Collects column equality information from `col = literal` predicates in a /// conjunction. /// @@ -1261,6 +1352,31 @@ fn collect_null_rejecting_columns(predicate: &Arc) -> HashSet< columns } +/// Returns the checked column index, and whether the check is `IS NULL`, when +/// `predicate` is a bare `IS NULL` or `IS NOT NULL` check on a column. +pub(crate) fn null_check_column( + predicate: &Arc, +) -> Option<(usize, bool)> { + let (arg, is_null) = if let Some(expr) = predicate.downcast_ref::() { + (expr.arg(), true) + } else { + let expr = predicate.downcast_ref::()?; + (expr.arg(), false) + }; + Some((arg.downcast_ref::()?.index(), is_null)) +} + +/// Column statistics of an exactly empty output: no nulls, distinct values or +/// bytes, and no value bounds. +fn empty_column_statistics() -> ColumnStatistics { + ColumnStatistics { + null_count: Precision::Exact(0), + distinct_count: Precision::Exact(0), + byte_size: Precision::Exact(0), + ..ColumnStatistics::new_unknown() + } +} + /// Converts an interval bound to a [`Precision`] value. NULL bounds (which /// represent "unbounded" in the interval type) map to [`Precision::Absent`]. fn interval_bound_to_precision( @@ -1308,9 +1424,9 @@ fn scale_byte_size_at_rows( /// Returns the NDV for a column constrained to one non-null value (e.g. /// `column = literal` or a singleton interval), derived from the filtered row -/// estimate: zero rows means zero distinct values, a known positive row count -/// means exactly one, and an unknown row count means an inexact one (the column -/// could still be empty). +/// estimate: zero rows means zero distinct values, an exact positive row count +/// means exactly one, and an estimated or unknown row count means an inexact +/// one (the column could still be empty). /// /// The caller is responsible for proving the singleton domain. fn distinct_count_for_singleton_domain( @@ -1318,10 +1434,10 @@ fn distinct_count_for_singleton_domain( ) -> Precision { match filtered_num_rows { Precision::Exact(0) | Precision::Inexact(0) => filtered_num_rows, - // The row count is unknown, so the column could still be empty (zero - // distinct values); report an inexact one rather than overstating it. - Precision::Absent => Precision::Inexact(1), - _ => Precision::Exact(1), + Precision::Exact(_) => Precision::Exact(1), + // The row count is not known exactly, so the column could still be + // empty (zero distinct values); report an inexact one. + Precision::Inexact(_) | Precision::Absent => Precision::Inexact(1), } } @@ -3044,7 +3160,7 @@ mod tests { Operator::Eq, Arc::new(Literal::new(ScalarValue::Utf8(Some("hello".to_string())))), )), - vec![Precision::Exact(1)], + vec![Precision::Inexact(1)], ), ( "utf8view equality", @@ -3060,7 +3176,7 @@ mod tests { "hello".to_string(), )))), )), - vec![Precision::Exact(1)], + vec![Precision::Inexact(1)], ), ( "largeutf8 equality", @@ -3076,7 +3192,7 @@ mod tests { "hello".to_string(), )))), )), - vec![Precision::Exact(1)], + vec![Precision::Inexact(1)], ), ( "utf8 reversed (literal = column)", @@ -3090,7 +3206,7 @@ mod tests { Operator::Eq, Arc::new(Column::new("name", 0)), )), - vec![Precision::Exact(1)], + vec![Precision::Inexact(1)], ), ( "OR is not collapsed to NDV=1, but NDV is capped at filtered rows", @@ -3147,7 +3263,7 @@ mod tests { Arc::new(Literal::new(ScalarValue::Int32(Some(42)))), )), )), - vec![Precision::Exact(1), Precision::Exact(1)], + vec![Precision::Inexact(1), Precision::Inexact(1)], ), ( "numeric equality with min/max bounds (interval analysis path)", @@ -3163,7 +3279,7 @@ mod tests { Operator::Eq, Arc::new(Literal::new(ScalarValue::Int32(Some(42)))), )), - vec![Precision::Exact(1)], + vec![Precision::Inexact(1)], ), ( "timestamp equality", @@ -3184,7 +3300,7 @@ mod tests { None, ))), )), - vec![Precision::Exact(1)], + vec![Precision::Inexact(1)], ), ( "contradictory numeric equality (infeasible)", @@ -3220,7 +3336,7 @@ mod tests { Operator::Eq, Arc::new(Literal::new(ScalarValue::Utf8(Some("hello".to_string())))), )), - vec![Precision::Exact(1)], + vec![Precision::Inexact(1)], ), ( "contradictory utf8 equality (infeasible)", @@ -3285,7 +3401,7 @@ mod tests { Arc::new(Literal::new(ScalarValue::Int32(Some(2)))), )), )), - vec![Precision::Exact(1), Precision::Exact(1)], + vec![Precision::Inexact(1), Precision::Inexact(1)], ), ]; @@ -3583,7 +3699,7 @@ mod tests { // Equality predicates collapse NDV and reject nulls for their columns. assert_eq!( statistics.column_statistics[0].distinct_count, - Precision::Exact(1) + Precision::Inexact(1) ); assert_eq!( statistics.column_statistics[0].null_count, @@ -3598,7 +3714,7 @@ mod tests { ); assert_eq!( statistics.column_statistics[2].distinct_count, - Precision::Exact(1) + Precision::Inexact(1) ); assert_eq!( statistics.column_statistics[2].null_count, @@ -3636,7 +3752,7 @@ mod tests { StatisticsContext::new().compute(filter.as_ref(), &StatisticsArgs::new())?; assert_eq!( statistics.column_statistics[0].distinct_count, - Precision::Exact(1) + Precision::Inexact(1) ); Ok(()) } @@ -3670,7 +3786,7 @@ mod tests { StatisticsContext::new().compute(filter.as_ref(), &StatisticsArgs::new())?; assert_eq!( statistics.column_statistics[0].distinct_count, - Precision::Exact(1) + Precision::Inexact(1) ); Ok(()) } @@ -3704,7 +3820,7 @@ mod tests { StatisticsContext::new().compute(filter.as_ref(), &StatisticsArgs::new())?; assert_eq!( statistics.column_statistics[0].distinct_count, - Precision::Exact(1) + Precision::Inexact(1) ); Ok(()) } @@ -3738,7 +3854,7 @@ mod tests { StatisticsContext::new().compute(filter.as_ref(), &StatisticsArgs::new())?; assert_eq!( statistics.column_statistics[0].distinct_count, - Precision::Exact(1) + Precision::Inexact(1) ); Ok(()) } @@ -3773,7 +3889,7 @@ mod tests { StatisticsContext::new().compute(filter.as_ref(), &StatisticsArgs::new())?; assert_eq!( statistics.column_statistics[0].distinct_count, - Precision::Exact(1) + Precision::Inexact(1) ); Ok(()) } @@ -3820,7 +3936,7 @@ mod tests { StatisticsContext::new().compute(filter.as_ref(), &StatisticsArgs::new())?; assert_eq!( statistics.column_statistics[0].distinct_count, - Precision::Exact(1) + Precision::Inexact(1) ); Ok(()) } @@ -4228,6 +4344,158 @@ mod tests { Ok(()) } + #[test] + fn test_filter_statistics_fetch_preserves_singleton() -> Result<()> { + use Precision::{Absent, Exact, Inexact}; + + let schema = Schema::new(vec![Field::new("a", DataType::Int32, true)]); + let input = Arc::new(StatisticsExec::new( + Statistics::new_unknown(&schema), + schema, + )); + let filter = FilterExecBuilder::new(lit(true), input) + .with_fetch(Some(3)) + .build()?; + // A positive fetch keeps the singleton that its input reports, for + // both overall and partition stats. + for num_rows in [Exact(100), Inexact(100), Absent] { + for partition in [None, Some(0)] { + let stats = filter.statistics_with_fetch( + Statistics { + num_rows, + total_byte_size: Absent, + column_statistics: vec![ColumnStatistics { + null_count: Exact(0), + distinct_count: Exact(1), + min_value: Exact(ScalarValue::Int32(Some(5))), + max_value: Exact(ScalarValue::Int32(Some(5))), + ..Default::default() + }], + }, + partition, + )?; + let column = &stats.column_statistics[0]; + assert_eq!(column.null_count, Exact(0)); + assert!(column.is_singleton()); + assert_eq!(column.distinct_count, Exact(1)); + } + } + Ok(()) + } + + #[tokio::test] + async fn test_filter_fetch_statistics_match_execution() -> Result<()> { + let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int32, false)])); + let batch = RecordBatch::try_new( + Arc::clone(&schema), + vec![Arc::new(Int32Array::from_iter_values(0..100))], + )?; + for partitions in [1, 2] { + let input: Arc = test::TestMemoryExec::try_new_exec( + &vec![vec![batch.clone()]; partitions], + Arc::clone(&schema), + None, + )?; + for fetch in [0, 3] { + let filter: Arc = Arc::new( + FilterExecBuilder::new(lit(true), Arc::clone(&input)) + .with_fetch(Some(fetch)) + .build()?, + ); + let stats = StatisticsContext::new() + .compute(filter.as_ref(), &StatisticsArgs::new())?; + let partition_stats = StatisticsContext::new().compute( + filter.as_ref(), + &StatisticsArgs::new().with_partition(Some(0)), + )?; + assert_eq!(stats.num_rows.get_value(), Some(&(fetch * partitions))); + assert_eq!(partition_stats.num_rows.get_value(), Some(&fetch)); + let batches = crate::execution_plan::collect( + filter, + Arc::new(TaskContext::default()), + ) + .await?; + assert_eq!( + batches.iter().map(RecordBatch::num_rows).sum::(), + fetch * partitions + ); + } + } + Ok(()) + } + + #[test] + fn test_filter_fetch_null_column_statistics() -> Result<()> { + use Precision::{Exact, Inexact}; + + let schema = Schema::new(vec![Field::new("a", DataType::Int32, true)]); + let stats = Statistics { + num_rows: Exact(100), + total_byte_size: Exact(400), + column_statistics: vec![ColumnStatistics { + null_count: Exact(100), + distinct_count: Exact(0), + byte_size: Exact(400), + ..Default::default() + }], + }; + let input: Arc = Arc::new( + StatisticsExec::new(stats.clone(), schema.clone()) + .with_partition_statistics(vec![stats.clone()]), + ); + for fetch in [0, 3, 200] { + let filter = + FilterExecBuilder::new(is_null(col("a", &schema)?)?, Arc::clone(&input)) + .with_fetch(Some(fetch)) + .build()?; + let output = filter.statistics_with_fetch(stats.clone(), None)?; + assert_eq!(output.num_rows, Exact(fetch.min(100))); + assert_eq!(output.column_statistics[0].null_count, output.num_rows); + assert_eq!(output.column_statistics[0].distinct_count, Exact(0)); + let expected_bytes = match fetch { + 0 => Exact(0), + 3 => Inexact(12), + _ => Exact(400), + }; + assert_eq!(output.total_byte_size, expected_bytes); + assert_eq!(output.column_statistics[0].byte_size, expected_bytes); + } + Ok(()) + } + + #[tokio::test] + async fn test_filter_statistics_singleton_precision() -> Result<()> { + use Precision::{Exact, Inexact}; + + let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int32, true)])); + // The bounds include 5, but no row contains it. + let values = Int32Array::from_iter_values( + (0..1000).map(|i| if i % 2 == 0 { 1 } else { 100 }), + ); + let batch = RecordBatch::try_new(Arc::clone(&schema), vec![Arc::new(values)])?; + let input = test::TestMemoryExec::try_new_exec( + &[vec![batch]], + Arc::clone(&schema), + None, + )?; + let mut input_stats = StatisticsContext::new() + .compute(input.as_ref(), &StatisticsArgs::new())? + .as_ref() + .clone(); + // Supply the bounds a file scan could report for these rows. + input_stats.column_statistics[0].min_value = Exact(ScalarValue::Int32(Some(1))); + input_stats.column_statistics[0].max_value = Exact(ScalarValue::Int32(Some(100))); + let predicate = binary(col("a", &schema)?, Operator::Eq, lit(5i32), &schema)?; + let filter = FilterExecBuilder::new(predicate, input).build()?; + let stats = filter + .statistics_from_inputs(&[Arc::new(input_stats)], &StatisticsArgs::new())?; + let batches = + collect(filter.execute(0, Arc::new(TaskContext::default()))?).await?; + assert_eq!(batches.iter().map(RecordBatch::num_rows).sum::(), 0); + assert_eq!(stats.column_statistics[0].distinct_count, Inexact(1)); + Ok(()) + } + #[tokio::test] async fn test_filter_statistics_is_not_null_rejects_nulls() -> Result<()> { let schema = Schema::new(vec![Field::new("name", DataType::Utf8, true)]); @@ -4269,4 +4537,272 @@ mod tests { ); Ok(()) } + + // --------------------------------------------------------------- + // Unit tests for compute_fallback_selectivity + // --------------------------------------------------------------- + + #[test] + fn test_fallback_selectivity_single_handled_equality() { + // col_0 = , NDV(col_0) = 100 → selectivity = 1/100 + let schema = Schema::new(vec![Field::new("a", DataType::Int32, false)]); + let predicate: Arc = binary( + col("a", &schema).unwrap(), + Operator::Eq, + lit(42i32), + &schema, + ) + .unwrap(); + let col_stats = vec![ColumnStatistics { + distinct_count: Precision::Inexact(100), + ..Default::default() + }]; + let result = compute_fallback_selectivity(&predicate, &col_stats, 20); + assert!( + (result - 0.01).abs() < 1e-12, + "expected 1/100 = 0.01, got {result}" + ); + } + + #[test] + fn test_fallback_selectivity_multiple_unhandled_conjuncts() { + // s <> 'abc' AND t <> 'x' AND u <> 'p' + // None are handled equalities → selectivity = default once = 0.2 + let schema = Schema::new(vec![ + Field::new("s", DataType::Utf8, false), + Field::new("t", DataType::Utf8, false), + Field::new("u", DataType::Utf8, false), + ]); + // Simulate three non-equality conjuncts via NotEq operators + let pred1 = binary( + col("s", &schema).unwrap(), + Operator::NotEq, + lit("abc"), + &schema, + ) + .unwrap(); + let pred2 = binary( + col("t", &schema).unwrap(), + Operator::NotEq, + lit("x"), + &schema, + ) + .unwrap(); + let pred3 = binary( + col("u", &schema).unwrap(), + Operator::NotEq, + lit("p"), + &schema, + ) + .unwrap(); + let combined: Arc = conjunction(vec![pred1, pred2, pred3]); + let col_stats = vec![ + ColumnStatistics::new_unknown(), + ColumnStatistics::new_unknown(), + ColumnStatistics::new_unknown(), + ]; + let result = compute_fallback_selectivity(&combined, &col_stats, 20); + // default_selectivity applied exactly once: 0.2 + assert!((result - 0.2).abs() < 1e-12, "expected 0.2, got {result}"); + } + + #[test] + fn test_fallback_selectivity_mixed_handled_and_unhandled() { + // col_a = 42 AND col_b <> 'x' + // col_a has NDV=50, col_b is unhandled + // selectivity = (1/50) * 0.2 = 0.004 + let schema = Schema::new(vec![ + Field::new("a", DataType::Int32, false), + Field::new("b", DataType::Utf8, false), + ]); + let eq_pred = binary( + col("a", &schema).unwrap(), + Operator::Eq, + lit(42i32), + &schema, + ) + .unwrap(); + let neq_pred = binary( + col("b", &schema).unwrap(), + Operator::NotEq, + lit("x"), + &schema, + ) + .unwrap(); + let combined: Arc = conjunction(vec![eq_pred, neq_pred]); + let col_stats = vec![ + ColumnStatistics { + distinct_count: Precision::Inexact(50), + ..Default::default() + }, + ColumnStatistics::new_unknown(), + ]; + let result = compute_fallback_selectivity(&combined, &col_stats, 20); + let expected = (1.0 / 50.0) * 0.2; + assert!( + (result - expected).abs() < 1e-12, + "expected {expected}, got {result}" + ); + } + + #[test] + fn test_fallback_selectivity_col_eq_col_uses_max_ndv() { + // col_a = col_b, NDV(a)=100, NDV(b)=200 + // selectivity = 1 / max(100, 200) = 1/200 + let schema = Schema::new(vec![ + Field::new("a", DataType::Int32, false), + Field::new("b", DataType::Int32, false), + ]); + let predicate: Arc = binary( + col("a", &schema).unwrap(), + Operator::Eq, + col("b", &schema).unwrap(), + &schema, + ) + .unwrap(); + let col_stats = vec![ + ColumnStatistics { + distinct_count: Precision::Inexact(100), + ..Default::default() + }, + ColumnStatistics { + distinct_count: Precision::Inexact(200), + ..Default::default() + }, + ]; + let result = compute_fallback_selectivity(&predicate, &col_stats, 20); + assert!( + (result - 1.0 / 200.0).abs() < 1e-12, + "expected 1/200 = 0.005, got {result}" + ); + } + + #[test] + fn test_fallback_selectivity_non_equality_returns_default() { + // A single non-equality predicate (a > 10) → default_selectivity applied once + let schema = Schema::new(vec![Field::new("a", DataType::Int32, false)]); + let predicate: Arc = binary( + col("a", &schema).unwrap(), + Operator::Gt, + lit(10i32), + &schema, + ) + .unwrap(); + let col_stats = vec![ColumnStatistics { + distinct_count: Precision::Inexact(100), + ..Default::default() + }]; + let result = compute_fallback_selectivity(&predicate, &col_stats, 20); + assert!((result - 0.2).abs() < 1e-12, "expected 0.2, got {result}"); + } + + /// Verify that `FilterExec` uses NDV-based fallback selectivity for predicates + /// that `check_support` cannot handle. A Utf8 equality `name = 'alice'` fails + /// interval analysis, so the filter falls back to `1 / NDV` instead of the + /// flat 20% default. + #[tokio::test] + async fn test_filter_statistics_fallback_uses_ndv_for_unsupported_predicate() + -> Result<()> { + let schema = Schema::new(vec![Field::new("name", DataType::Utf8, false)]); + // 1000 rows, NDV = 200 for the `name` column. + let input = Arc::new(StatisticsExec::new( + Statistics { + num_rows: Precision::Inexact(1000), + total_byte_size: Precision::Absent, + column_statistics: vec![ColumnStatistics { + distinct_count: Precision::Inexact(200), + ..Default::default() + }], + }, + schema.clone(), + )); + + // Utf8 equality fails `check_support`, triggering the NDV fallback. + // Expected selectivity = 1 / 200 = 0.005 → num_rows = 1000 * 0.005 = 5. + let predicate: Arc = + binary(col("name", &schema)?, Operator::Eq, lit("alice"), &schema)?; + let filter = Arc::new(FilterExec::try_new(predicate, input)?); + let stats = + StatisticsContext::new().compute(filter.as_ref(), &StatisticsArgs::new())?; + assert_eq!(stats.num_rows, Precision::Inexact(5)); + Ok(()) + } + + /// Verify that a `CAST(a AS Int64) = ` predicate — which + /// `check_support` cannot resolve — still falls back gracefully to the + /// default selectivity (20%) rather than panicking or silently using 100%. + /// + /// In this case the left-hand side is a `CastExpr`, so `column_ndv` returns + /// `None` and no NDV estimate is available. The single unhandled conjunct + /// receives the flat default. + #[tokio::test] + async fn test_filter_statistics_fallback_cast_expr_uses_default_selectivity() + -> Result<()> { + let schema = Schema::new(vec![Field::new("a", DataType::Int32, false)]); + // 1000 rows, NDV = 500. + let input = Arc::new(StatisticsExec::new( + Statistics { + num_rows: Precision::Inexact(1000), + total_byte_size: Precision::Absent, + column_statistics: vec![ColumnStatistics { + distinct_count: Precision::Inexact(500), + ..Default::default() + }], + }, + schema.clone(), + )); + + // CAST(a AS Int64) = 42: the CastExpr wraps the column, so + // `column_ndv` cannot resolve it. The predicate also passes + // `check_support` (CastExpr + Int64 literal are both supported), + // so the interval-analysis path runs. This test documents the + // current boundary: once `check_support` accepts the predicate + // the fallback is not reached. + // + // We therefore use a plain Utf8 column compared to a literal, + // which is the concrete unsupported form this PR improves. + let predicate: Arc = Arc::new(BinaryExpr::new( + Arc::new(CastExpr::new(col("a", &schema)?, DataType::Int64, None)), + Operator::Eq, + Arc::new(Literal::new(ScalarValue::Int64(Some(42)))), + )); + let filter = Arc::new( + FilterExec::try_new(predicate, input)?.with_default_selectivity(20)?, + ); + let stats = + StatisticsContext::new().compute(filter.as_ref(), &StatisticsArgs::new())?; + // check_support accepts CAST + Int64 literal, so interval analysis + // runs and produces a result — num_rows will not be the full 1000. + assert!( + stats.num_rows != Precision::Inexact(1000), + "expected some rows to be filtered, got {:?}", + stats.num_rows + ); + Ok(()) + } + + #[test] + fn test_fallback_selectivity_utf8_equality_uses_ndv() { + // name = 'alice' on a Utf8 column with NDV=60. + // Utf8 equality fails `check_support`, so our fallback runs and + // returns 1/60 instead of the previous flat 20%. + let schema = Schema::new(vec![Field::new("name", DataType::Utf8, false)]); + let predicate: Arc = binary( + col("name", &schema).unwrap(), + Operator::Eq, + lit("alice"), + &schema, + ) + .unwrap(); + let col_stats = vec![ColumnStatistics { + distinct_count: Precision::Inexact(60), + ..Default::default() + }]; + let result = compute_fallback_selectivity(&predicate, &col_stats, 20); + let expected = 1.0 / 60.0; + assert!( + (result - expected).abs() < 1e-12, + "expected {expected}, got {result}" + ); + } } diff --git a/datafusion/sqllogictest/test_files/subquery.slt b/datafusion/sqllogictest/test_files/subquery.slt index d27c244e0d4df..c91bcdbe0d5b0 100644 --- a/datafusion/sqllogictest/test_files/subquery.slt +++ b/datafusion/sqllogictest/test_files/subquery.slt @@ -3022,3 +3022,35 @@ DROP TABLE oj_a; statement ok DROP TABLE oj_b; + +############# +## Scalar subquery equality: NDV-based row estimate +## When a filter contains `col = (SELECT ...)`, the interval solver +## cannot resolve the subquery, but the fallback heuristic uses +## `1 / NDV(col)` instead of the flat 20% default. +############# + +statement ok +CREATE TABLE ndv_main AS + SELECT column1 AS id, column2 AS val + FROM (VALUES (1, 'a'), (2, 'b'), (3, 'c'), (4, 'd'), (5, 'e'), + (6, 'f'), (7, 'g'), (8, 'h'), (9, 'i'), (10, 'j')); + +statement ok +CREATE TABLE ndv_lookup AS SELECT 5 AS v; + +# The filter `id = (SELECT v FROM ndv_lookup)` should use the NDV of `id` +# (10 distinct values) to estimate ~1 row instead of the 20% default (~2 rows). +query TT +EXPLAIN SELECT * FROM ndv_main WHERE id = (SELECT v FROM ndv_lookup); +---- +logical_plan + +physical_plan + + +statement ok +DROP TABLE ndv_main; + +statement ok +DROP TABLE ndv_lookup; From bb26702612e27045307e9103b21e69a90041c25e Mon Sep 17 00:00:00 2001 From: mohitgurav20 Date: Tue, 6 Oct 2026 01:25:56 +0530 Subject: [PATCH 03/11] test: add scalar subquery test in subquery.slt --- .../sqllogictest/test_files/subquery.slt | 39 +++++++++++++++++++ 1 file changed, 39 insertions(+) diff --git a/datafusion/sqllogictest/test_files/subquery.slt b/datafusion/sqllogictest/test_files/subquery.slt index 4e027431464be..b3afdf448fb76 100644 --- a/datafusion/sqllogictest/test_files/subquery.slt +++ b/datafusion/sqllogictest/test_files/subquery.slt @@ -3412,3 +3412,42 @@ DROP TABLE unnest_outer; statement ok DROP TABLE unnest_inner; +# --------------------------------------------------------- +# Test Scalar Subquery Selectivity Fallback (PR #25719) +# --------------------------------------------------------- + +statement ok +CREATE TABLE sq_fallback_table(name TEXT) AS VALUES +('a'), ('b'), ('c'), ('d'); + +statement ok +set datafusion.explain.logical_plan_only = false; + +statement ok +set datafusion.explain.show_statistics = true; + +# The physical plan for the Filter should use the NDV estimate for the scalar subquery +query TT +EXPLAIN SELECT * FROM sq_fallback_table WHERE name = (SELECT 'a'); +---- +logical_plan +01)Filter: sq_fallback_table.name = () +02)--Subquery: +03)----Projection: Utf8View("a") AS Utf8("a") +04)------EmptyRelation: rows=1 +05)--TableScan: sq_fallback_table projection=[name] +physical_plan +01)ScalarSubqueryExec: subqueries=1, statistics=[Rows=Inexact(1), Bytes=Inexact(36), [(Col[0]: Null=Exact(0))]] +02)--FilterExec: name@0 = scalar_subquery(), statistics=[Rows=Inexact(1), Bytes=Inexact(36), [(Col[0]: Null=Exact(0))]] +03)----DataSourceExec: partitions=1, partition_sizes=[1], statistics=[Rows=Exact(4), Bytes=Exact(176), [(Col[0]: Null=Exact(0))]] +04)--ProjectionExec: expr=[a as Utf8("a")], statistics=[Rows=Exact(1), Bytes=Inexact(0), [(Col[0]: Min=Exact(Utf8View("a")) Max=Exact(Utf8View("a")) Null=Exact(0) Distinct=Exact(1))]] +05)----PlaceholderRowExec, statistics=[Rows=Exact(1), Bytes=Exact(0), []] + +statement ok +RESET datafusion.explain.show_statistics; + +statement ok +RESET datafusion.explain.logical_plan_only; + +statement ok +DROP TABLE sq_fallback_table; From ef7b04e902a78481af2ff2382bf7a6d364493eb6 Mon Sep 17 00:00:00 2001 From: mohitgurav20 Date: Tue, 6 Oct 2026 02:00:11 +0530 Subject: [PATCH 04/11] chore: trigger CI From 64d9aaac62d5b00d8e7afc363e333aad69487ce2 Mon Sep 17 00:00:00 2001 From: mohitgurav20 Date: Tue, 6 Oct 2026 10:48:44 +0530 Subject: [PATCH 05/11] test: cover non-binary expr fallback selectivity --- datafusion/physical-plan/src/filter.rs | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/datafusion/physical-plan/src/filter.rs b/datafusion/physical-plan/src/filter.rs index 598996e10a536..3b9a0a41bb7c1 100644 --- a/datafusion/physical-plan/src/filter.rs +++ b/datafusion/physical-plan/src/filter.rs @@ -4696,6 +4696,23 @@ mod tests { assert!((result - 0.2).abs() < 1e-12, "expected 0.2, got {result}"); } + #[test] + fn test_fallback_selectivity_non_binary_expr_returns_default() { + // A non-BinaryExpr predicate (e.g. IsNotNull) → downcast to BinaryExpr fails, + // so it is unhandled and default selectivity is applied. + let schema = Schema::new(vec![Field::new("a", DataType::Int32, false)]); + let predicate: Arc = + Arc::new(datafusion_physical_expr::expressions::IsNotNullExpr::new( + col("a", &schema).unwrap(), + )); + let col_stats = vec![ColumnStatistics { + distinct_count: Precision::Inexact(100), + ..Default::default() + }]; + let result = compute_fallback_selectivity(&predicate, &col_stats, 20); + assert!((result - 0.2).abs() < 1e-12, "expected 0.2, got {result}"); + } + /// Verify that `FilterExec` uses NDV-based fallback selectivity for predicates /// that `check_support` cannot handle. A Utf8 equality `name = 'alice'` fails /// interval analysis, so the filter falls back to `1 / NDV` instead of the From bc0438a6a043e781c36c110d06b795477519ff22 Mon Sep 17 00:00:00 2001 From: mohitgurav20 Date: Tue, 6 Oct 2026 11:01:01 +0530 Subject: [PATCH 06/11] fix: resolve unused-qualifications lint in filter.rs --- datafusion/physical-plan/src/filter.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/datafusion/physical-plan/src/filter.rs b/datafusion/physical-plan/src/filter.rs index 3b9a0a41bb7c1..1a85b3bd68260 100644 --- a/datafusion/physical-plan/src/filter.rs +++ b/datafusion/physical-plan/src/filter.rs @@ -4702,7 +4702,7 @@ mod tests { // so it is unhandled and default selectivity is applied. let schema = Schema::new(vec![Field::new("a", DataType::Int32, false)]); let predicate: Arc = - Arc::new(datafusion_physical_expr::expressions::IsNotNullExpr::new( + Arc::new(IsNotNullExpr::new( col("a", &schema).unwrap(), )); let col_stats = vec![ColumnStatistics { From 008cb4dbe0f4a1fd420a88e5d1739000550020fc Mon Sep 17 00:00:00 2001 From: mohitgurav20 Date: Tue, 6 Oct 2026 11:05:35 +0530 Subject: [PATCH 07/11] style: apply cargo fmt to filter.rs --- datafusion/physical-plan/src/filter.rs | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/datafusion/physical-plan/src/filter.rs b/datafusion/physical-plan/src/filter.rs index 1a85b3bd68260..4f460231948d0 100644 --- a/datafusion/physical-plan/src/filter.rs +++ b/datafusion/physical-plan/src/filter.rs @@ -4702,9 +4702,7 @@ mod tests { // so it is unhandled and default selectivity is applied. let schema = Schema::new(vec![Field::new("a", DataType::Int32, false)]); let predicate: Arc = - Arc::new(IsNotNullExpr::new( - col("a", &schema).unwrap(), - )); + Arc::new(IsNotNullExpr::new(col("a", &schema).unwrap())); let col_stats = vec![ColumnStatistics { distinct_count: Precision::Inexact(100), ..Default::default() From a2218dc896ceaa16e25378d662be211b7bac2246 Mon Sep 17 00:00:00 2001 From: mohitgurav20 Date: Tue, 6 Oct 2026 15:02:30 +0530 Subject: [PATCH 08/11] test: cover column_ndv out-of-bounds path for 100% patch coverage --- datafusion/physical-plan/src/filter.rs | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/datafusion/physical-plan/src/filter.rs b/datafusion/physical-plan/src/filter.rs index 4f460231948d0..d692e78b8f253 100644 --- a/datafusion/physical-plan/src/filter.rs +++ b/datafusion/physical-plan/src/filter.rs @@ -4908,4 +4908,21 @@ mod tests { "expected 1/500 = 0.002 when left NDV dominates, got {result}" ); } + + #[test] + fn test_fallback_selectivity_column_index_out_of_bounds() { + // column_ndv: `column_statistics.get(col.index())` returns None when the + // column index exceeds the stats slice length. The conjunct is then + // unhandled → default selectivity is applied. + let schema = Schema::new(vec![Field::new("a", DataType::Int32, false)]); + let predicate: Arc = + binary(col("a", &schema).unwrap(), Operator::Eq, lit(1i32), &schema).unwrap(); + // Empty stats slice: col.index() == 0 is out of bounds. + let col_stats: Vec = vec![]; + let result = compute_fallback_selectivity(&predicate, &col_stats, 20); + assert!( + (result - 0.2).abs() < 1e-12, + "expected default 0.2 when column index is out of bounds, got {result}" + ); + } } From 283b90b45a7a65b32408bebb707dd8000a41a91f Mon Sep 17 00:00:00 2001 From: mohitgurav20 Date: Wed, 7 Oct 2026 17:05:58 +0530 Subject: [PATCH 09/11] test: add FilterExec scalar subquery tests and remove unneeded SLT tests - Add FilterExec statistics test with known NDV and  = , asserting 1.0 / NDV row estimate. - Add FilterExec statistics test with CAST(a) = , asserting fallback to default selectivity when cast is not unwrapped. - Add unit test for CAST expression in compute_fallback_selectivity. - Remove unneeded subquery.slt tests since table scans from VALUES lack NDV statistics. - Update test comment to describe current behavior. --- datafusion/physical-plan/src/filter.rs | 138 ++++++--- .../sqllogictest/test_files/subquery.slt | 268 +++++++++++++----- 2 files changed, 301 insertions(+), 105 deletions(-) diff --git a/datafusion/physical-plan/src/filter.rs b/datafusion/physical-plan/src/filter.rs index d692e78b8f253..bdf912e77b94d 100644 --- a/datafusion/physical-plan/src/filter.rs +++ b/datafusion/physical-plan/src/filter.rs @@ -4747,47 +4747,98 @@ mod tests { assert_eq!(stats.num_rows, Precision::Inexact(5)); } - /// Verify that a `CAST(a AS Int64) = ` predicate — which - /// `check_support` cannot resolve — still falls back gracefully to the - /// default selectivity (20%) rather than panicking or silently using 100%. + /// Verify that `FilterExec` uses NDV-based fallback selectivity for a scalar + /// subquery equality predicate `a = `. /// - /// In this case the left-hand side is a `CastExpr`, so `column_ndv` returns - /// `None` and no NDV estimate is available. The single unhandled conjunct - /// receives the flat default. + /// The scalar subquery is unresolved during planning and fails interval + /// analysis (`check_support`), but the fallback heuristic estimates + /// selectivity as `1.0 / NDV` rather than the flat 20% default. #[tokio::test] - async fn test_filter_statistics_fallback_cast_expr_uses_default_selectivity() { + async fn test_filter_statistics_fallback_scalar_subquery_uses_ndv() { + use datafusion_expr::physical_planning_context::{ + ScalarSubqueryResults, SubqueryIndex, + }; + use datafusion_physical_expr::scalar_subquery::ScalarSubqueryExpr; + let schema = Schema::new(vec![Field::new("a", DataType::Int32, false)]); - // 1000 rows, NDV = 500. + // 1000 rows, NDV = 200 for column `a`. let input = Arc::new(StatisticsExec::new( Statistics { num_rows: Precision::Inexact(1000), total_byte_size: Precision::Absent, column_statistics: vec![ColumnStatistics { - distinct_count: Precision::Inexact(500), + distinct_count: Precision::Inexact(200), ..Default::default() }], }, schema.clone(), )); - // CAST(a AS Int64) = 42: the CastExpr wraps the column, so - // `column_ndv` cannot resolve it. The predicate also passes - // `check_support` (CastExpr + Int64 literal are both supported), - // so the interval-analysis path runs. This test documents the - // current boundary: once `check_support` accepts the predicate - // the fallback is not reached. - // - // We therefore use a plain Utf8 column compared to a literal, - // which is the concrete unsupported form this PR improves. + let subquery_expr = Arc::new(ScalarSubqueryExpr::new( + DataType::Int32, + false, + SubqueryIndex::new(0), + ScalarSubqueryResults::new(1), + )); + + // a = let predicate: Arc = Arc::new(BinaryExpr::new( - Arc::new(CastExpr::new( - col("a", &schema).unwrap(), - DataType::Int64, - None, - )), + col("a", &schema).unwrap(), Operator::Eq, - Arc::new(Literal::new(ScalarValue::Int64(Some(42)))), + subquery_expr, + )); + + let filter = Arc::new(FilterExec::try_new(predicate, input).unwrap()); + let stats = StatisticsContext::new() + .compute(filter.as_ref(), &StatisticsArgs::new()) + .unwrap(); + + // Expected selectivity = 1 / 200 = 0.005 -> 1000 * 0.005 = 5 rows + // (rather than 20% default = 200 rows). + assert_eq!(stats.num_rows, Precision::Inexact(5)); + } + + /// Verify that a `CAST(a AS Int64) = ` predicate — where + /// the column is wrapped in a `CastExpr` and the subquery is unresolved — + /// falls back to the default selectivity (20%) because `column_ndv` does + /// not unwrap casts. + #[tokio::test] + async fn test_filter_statistics_fallback_cast_scalar_subquery_uses_default() { + use datafusion_expr::physical_planning_context::{ + ScalarSubqueryResults, SubqueryIndex, + }; + use datafusion_physical_expr::scalar_subquery::ScalarSubqueryExpr; + + let schema = Schema::new(vec![Field::new("a", DataType::Int32, false)]); + // 1000 rows, NDV = 200 for column `a`. + let input = Arc::new(StatisticsExec::new( + Statistics { + num_rows: Precision::Inexact(1000), + total_byte_size: Precision::Absent, + column_statistics: vec![ColumnStatistics { + distinct_count: Precision::Inexact(200), + ..Default::default() + }], + }, + schema.clone(), + )); + + let subquery_expr = Arc::new(ScalarSubqueryExpr::new( + DataType::Int64, + false, + SubqueryIndex::new(0), + ScalarSubqueryResults::new(1), + )); + + // CAST(a AS Int64) = + let cast_col: Arc = Arc::new(CastExpr::new( + col("a", &schema).unwrap(), + DataType::Int64, + None, )); + let predicate: Arc = + Arc::new(BinaryExpr::new(cast_col, Operator::Eq, subquery_expr)); + let filter = Arc::new( FilterExec::try_new(predicate, input) .unwrap() @@ -4797,20 +4848,41 @@ mod tests { let stats = StatisticsContext::new() .compute(filter.as_ref(), &StatisticsArgs::new()) .unwrap(); - // check_support accepts CAST + Int64 literal, so interval analysis - // runs and produces a result — num_rows will not be the full 1000. - assert_ne!( - stats.num_rows, - Precision::Inexact(1000), - "expected interval analysis to filter some rows" - ); + + // CAST wraps the column, so column_ndv returns None. + // Falls back to default selectivity 20% -> 1000 * 0.20 = 200 rows. + assert_eq!(stats.num_rows, Precision::Inexact(200)); + } + + #[test] + fn test_fallback_selectivity_cast_expr_not_handled() { + // CAST(a AS Int64) = 42: the left side is a CastExpr, not a bare Column, + // so column_ndv returns None. The conjunct is unhandled and falls through + // to default_selectivity. + let schema = Schema::new(vec![Field::new("a", DataType::Int32, false)]); + let cast_expr: Arc = Arc::new(CastExpr::new( + col("a", &schema).unwrap(), + DataType::Int64, + None, + )); + let predicate: Arc = Arc::new(BinaryExpr::new( + cast_expr, + Operator::Eq, + Arc::new(Literal::new(ScalarValue::Int64(Some(42)))), + )); + let col_stats = vec![ColumnStatistics { + distinct_count: Precision::Inexact(100), + ..Default::default() + }]; + let result = compute_fallback_selectivity(&predicate, &col_stats, 20); + assert!((result - 0.2).abs() < 1e-12, "expected 0.2, got {result}"); } #[test] fn test_fallback_selectivity_utf8_equality_uses_ndv() { // name = 'alice' on a Utf8 column with NDV=60. - // Utf8 equality fails `check_support`, so our fallback runs and - // returns 1/60 instead of the flat 20% default. + // Utf8 equality fails `check_support`, so fallback selectivity runs and + // returns 1/60 rather than the default 20% selectivity. let schema = Schema::new(vec![Field::new("name", DataType::Utf8, false)]); let predicate: Arc = binary( col("name", &schema).unwrap(), diff --git a/datafusion/sqllogictest/test_files/subquery.slt b/datafusion/sqllogictest/test_files/subquery.slt index b3afdf448fb76..09f648df27702 100644 --- a/datafusion/sqllogictest/test_files/subquery.slt +++ b/datafusion/sqllogictest/test_files/subquery.slt @@ -3014,6 +3014,202 @@ DROP TABLE gs_outer; statement ok DROP TABLE gs_inner; +############# +## Multi-column IN (LeftSemi) Tests +############# +## These tests verify that multi-column IN subqueries work correctly +## Multi-column IN uses LeftSemi join (not null-aware) + +############# +## Test 1: Basic two-column IN +############# + +statement ok +CREATE TABLE multi_in_left(a INT, b INT, value TEXT) AS VALUES +(1, 2, 'match'), +(3, 4, 'no_match'), +(5, 6, 'match'); + +statement ok +CREATE TABLE multi_in_right(x INT, y INT) AS VALUES +(1, 2), +(5, 6); + +# Should return rows where (a, b) matches (x, y) +query IIT rowsort +SELECT * FROM multi_in_left +WHERE (a, b) IN (SELECT x, y FROM multi_in_right); +---- +1 2 match +5 6 match + +############# +## Test 2: Multi-column IN with no matches +############# + +statement ok +CREATE TABLE multi_in_right_no_match(x INT, y INT) AS VALUES +(10, 20), +(30, 40); + +# Should return empty result +query IIT rowsort +SELECT * FROM multi_in_left +WHERE (a, b) IN (SELECT x, y FROM multi_in_right_no_match); +---- + +############# +## Test 3: Multi-column IN with NULL values +############# +## Note: Unlike NOT IN, regular IN does NOT use null-aware semantics +## NULL = NULL is always FALSE (not unknown) in regular semi joins + +statement ok +CREATE TABLE multi_in_left_null(a INT, b INT, value TEXT) AS VALUES +(1, 2, 'x'), +(3, NULL, 'y'), +(NULL, 6, 'z'); + +statement ok +CREATE TABLE multi_in_right_null(x INT, y INT) AS VALUES +(1, 2), +(NULL, 4); + +# Should return only (1, 2, 'x') +# (3, NULL, 'y') doesn't match because NULL doesn't equal anything +# (NULL, 6, 'z') doesn't match because NULL doesn't equal anything +query IIT rowsort +SELECT * FROM multi_in_left_null +WHERE (a, b) IN (SELECT x, y FROM multi_in_right_null); +---- +1 2 x + +############# +## Test 4: Three-column IN +############# + +statement ok +CREATE TABLE three_col_left(a INT, b INT, c INT, value TEXT) AS VALUES +(1, 2, 3, 'match1'), +(4, 5, 6, 'no_match'), +(7, 8, 9, 'match2'); + +statement ok +CREATE TABLE three_col_right(x INT, y INT, z INT) AS VALUES +(1, 2, 3), +(7, 8, 9); + +# Should return rows with matching three-column tuples +query IIIT rowsort +SELECT * FROM three_col_left +WHERE (a, b, c) IN (SELECT x, y, z FROM three_col_right); +---- +1 2 3 match1 +7 8 9 match2 + +############# +## Test 5: Correlated multi-column IN +############# + +statement ok +CREATE TABLE correlated_outer(id INT, a INT, b INT) AS VALUES +(1, 10, 20), +(2, 30, 40), +(3, 10, 20); + +statement ok +CREATE TABLE correlated_inner(id INT, x INT, y INT) AS VALUES +(1, 10, 20), +(1, 30, 40), +(2, 50, 60), +(3, 10, 20); + +# Should return outer rows where (a, b) matches (x, y) for the same id +query III rowsort +SELECT * FROM correlated_outer o +WHERE (a, b) IN ( + SELECT x, y FROM correlated_inner i WHERE i.id = o.id +); +---- +1 10 20 +3 10 20 + +############# +## Test 6: Verify logical plan shows LeftSemi join with multiple conditions +############# + +query TT +EXPLAIN SELECT * FROM multi_in_left +WHERE (a, b) IN (SELECT x, y FROM multi_in_right); +---- +logical_plan +01)LeftSemi Join: multi_in_left.a = __correlated_sq_1.x, multi_in_left.b = __correlated_sq_1.y +02)--TableScan: multi_in_left projection=[a, b, value] +03)--SubqueryAlias: __correlated_sq_1 +04)----TableScan: multi_in_right projection=[x, y] +physical_plan +01)HashJoinExec: mode=CollectLeft, join_type=RightSemi, on=[(x@0, a@0), (y@1, b@1)] +02)--DataSourceExec: partitions=1, partition_sizes=[1] +03)--DataSourceExec: partitions=1, partition_sizes=[1] + +############# +## Test 7: Multi-column IN with empty subquery +############# + +statement ok +CREATE TABLE multi_in_right_empty(x INT, y INT); + +# Should return empty result (empty subquery) +query IIT rowsort +SELECT * FROM multi_in_left +WHERE (a, b) IN (SELECT x, y FROM multi_in_right_empty); +---- + +############# +## Test 8: Multi-column IN with WHERE clause in subquery +############# + +# The filter drops (1, 2) from the subquery, so only (5, 6) matches +query IIT rowsort +SELECT * FROM multi_in_left +WHERE (a, b) IN (SELECT x, y FROM multi_in_right WHERE x > 1); +---- +5 6 match + +############# +## Cleanup +############# + +statement ok +DROP TABLE multi_in_left; + +statement ok +DROP TABLE multi_in_right; + +statement ok +DROP TABLE multi_in_right_no_match; + +statement ok +DROP TABLE multi_in_left_null; + +statement ok +DROP TABLE multi_in_right_null; + +statement ok +DROP TABLE three_col_left; + +statement ok +DROP TABLE three_col_right; + +statement ok +DROP TABLE correlated_outer; + +statement ok +DROP TABLE correlated_inner; + +statement ok +DROP TABLE multi_in_right_empty; + # Regression test for #25507: a correlated filter below the side of an outer # join that the join fills with NULLs must not be pulled above the join. The # filter decides which rows of the other side are unmatched, so once it is @@ -3188,38 +3384,6 @@ DROP TABLE oj_a; statement ok DROP TABLE oj_b; -############# -## Scalar subquery equality: NDV-based row estimate -## When a filter contains `col = (SELECT ...)`, the interval solver -## cannot resolve the subquery, but the fallback heuristic uses -## `1 / NDV(col)` instead of the flat 20% default. -############# - -statement ok -CREATE TABLE ndv_main AS - SELECT column1 AS id, column2 AS val - FROM (VALUES (1, 'a'), (2, 'b'), (3, 'c'), (4, 'd'), (5, 'e'), - (6, 'f'), (7, 'g'), (8, 'h'), (9, 'i'), (10, 'j')); - -statement ok -CREATE TABLE ndv_lookup AS SELECT 5 AS v; - -# The filter `id = (SELECT v FROM ndv_lookup)` should use the NDV of `id` -# (10 distinct values) to estimate ~1 row instead of the 20% default (~2 rows). -query TT -EXPLAIN SELECT * FROM ndv_main WHERE id = (SELECT v FROM ndv_lookup); ----- -logical_plan - -physical_plan - - -statement ok -DROP TABLE ndv_main; - -statement ok -DROP TABLE ndv_lookup; - # Regression test for #25792: a correlated filter below a window function must # not be pulled above the window. The filter selects the rows the window # function sees for each outer row, so once it is above the window the window @@ -3411,43 +3575,3 @@ DROP TABLE unnest_outer; statement ok DROP TABLE unnest_inner; - -# --------------------------------------------------------- -# Test Scalar Subquery Selectivity Fallback (PR #25719) -# --------------------------------------------------------- - -statement ok -CREATE TABLE sq_fallback_table(name TEXT) AS VALUES -('a'), ('b'), ('c'), ('d'); - -statement ok -set datafusion.explain.logical_plan_only = false; - -statement ok -set datafusion.explain.show_statistics = true; - -# The physical plan for the Filter should use the NDV estimate for the scalar subquery -query TT -EXPLAIN SELECT * FROM sq_fallback_table WHERE name = (SELECT 'a'); ----- -logical_plan -01)Filter: sq_fallback_table.name = () -02)--Subquery: -03)----Projection: Utf8View("a") AS Utf8("a") -04)------EmptyRelation: rows=1 -05)--TableScan: sq_fallback_table projection=[name] -physical_plan -01)ScalarSubqueryExec: subqueries=1, statistics=[Rows=Inexact(1), Bytes=Inexact(36), [(Col[0]: Null=Exact(0))]] -02)--FilterExec: name@0 = scalar_subquery(), statistics=[Rows=Inexact(1), Bytes=Inexact(36), [(Col[0]: Null=Exact(0))]] -03)----DataSourceExec: partitions=1, partition_sizes=[1], statistics=[Rows=Exact(4), Bytes=Exact(176), [(Col[0]: Null=Exact(0))]] -04)--ProjectionExec: expr=[a as Utf8("a")], statistics=[Rows=Exact(1), Bytes=Inexact(0), [(Col[0]: Min=Exact(Utf8View("a")) Max=Exact(Utf8View("a")) Null=Exact(0) Distinct=Exact(1))]] -05)----PlaceholderRowExec, statistics=[Rows=Exact(1), Bytes=Exact(0), []] - -statement ok -RESET datafusion.explain.show_statistics; - -statement ok -RESET datafusion.explain.logical_plan_only; - -statement ok -DROP TABLE sq_fallback_table; From 8777e4c2600d28008898b51a1cd32ce56e5d3557 Mon Sep 17 00:00:00 2001 From: mohitgurav20 Date: Thu, 8 Oct 2026 22:24:37 +0530 Subject: [PATCH 10/11] test: refine fallback selectivity unit tests per review feedback - Consolidate symmetric column equality tests into test_fallback_selectivity_col_eq_col_uses_max_ndv verifying both operand orders. - Add test_fallback_selectivity_right_col_uses_ndv covering equality with column on RHS. - Prune redundant unit tests already covered by FilterExec-level tests. --- datafusion/physical-plan/src/filter.rs | 155 ++++++------------------- 1 file changed, 37 insertions(+), 118 deletions(-) diff --git a/datafusion/physical-plan/src/filter.rs b/datafusion/physical-plan/src/filter.rs index bdf912e77b94d..a14758f7bd827 100644 --- a/datafusion/physical-plan/src/filter.rs +++ b/datafusion/physical-plan/src/filter.rs @@ -4647,20 +4647,22 @@ mod tests { #[test] fn test_fallback_selectivity_col_eq_col_uses_max_ndv() { - // col_a = col_b, NDV(a)=100, NDV(b)=200 - // selectivity = 1 / max(100, 200) = 1/200 + // col_a = col_b uses 1 / max(NDV_left, NDV_right). We verify both column + // orderings to ensure the max is taken symmetrically. let schema = Schema::new(vec![ Field::new("a", DataType::Int32, false), Field::new("b", DataType::Int32, false), ]); - let predicate: Arc = binary( + + // (a=100, b=200) → right NDV dominates → 1/200 + let pred_ab: Arc = binary( col("a", &schema).unwrap(), Operator::Eq, col("b", &schema).unwrap(), &schema, ) .unwrap(); - let col_stats = vec![ + let stats_ab = vec![ ColumnStatistics { distinct_count: Precision::Inexact(100), ..Default::default() @@ -4670,45 +4672,35 @@ mod tests { ..Default::default() }, ]; - let result = compute_fallback_selectivity(&predicate, &col_stats, 20); + let result_ab = compute_fallback_selectivity(&pred_ab, &stats_ab, 20); assert!( - (result - 1.0 / 200.0).abs() < 1e-12, - "expected 1/200 = 0.005, got {result}" + (result_ab - 1.0 / 200.0).abs() < 1e-12, + "expected 1/200 (right NDV dominates), got {result_ab}" ); - } - #[test] - fn test_fallback_selectivity_non_equality_returns_default() { - // A single non-equality predicate (a > 10) → default_selectivity applied once - let schema = Schema::new(vec![Field::new("a", DataType::Int32, false)]); - let predicate: Arc = binary( + // (a=500, b=100) → left NDV dominates → 1/500 + let pred_ba: Arc = binary( col("a", &schema).unwrap(), - Operator::Gt, - lit(10i32), + Operator::Eq, + col("b", &schema).unwrap(), &schema, ) .unwrap(); - let col_stats = vec![ColumnStatistics { - distinct_count: Precision::Inexact(100), - ..Default::default() - }]; - let result = compute_fallback_selectivity(&predicate, &col_stats, 20); - assert!((result - 0.2).abs() < 1e-12, "expected 0.2, got {result}"); - } - - #[test] - fn test_fallback_selectivity_non_binary_expr_returns_default() { - // A non-BinaryExpr predicate (e.g. IsNotNull) → downcast to BinaryExpr fails, - // so it is unhandled and default selectivity is applied. - let schema = Schema::new(vec![Field::new("a", DataType::Int32, false)]); - let predicate: Arc = - Arc::new(IsNotNullExpr::new(col("a", &schema).unwrap())); - let col_stats = vec![ColumnStatistics { - distinct_count: Precision::Inexact(100), - ..Default::default() - }]; - let result = compute_fallback_selectivity(&predicate, &col_stats, 20); - assert!((result - 0.2).abs() < 1e-12, "expected 0.2, got {result}"); + let stats_ba = vec![ + ColumnStatistics { + distinct_count: Precision::Exact(500), + ..Default::default() + }, + ColumnStatistics { + distinct_count: Precision::Exact(100), + ..Default::default() + }, + ]; + let result_ba = compute_fallback_selectivity(&pred_ba, &stats_ba, 20); + assert!( + (result_ba - 1.0 / 500.0).abs() < 1e-12, + "expected 1/500 (left NDV dominates), got {result_ba}" + ); } /// Verify that `FilterExec` uses NDV-based fallback selectivity for predicates @@ -4855,51 +4847,27 @@ mod tests { } #[test] - fn test_fallback_selectivity_cast_expr_not_handled() { - // CAST(a AS Int64) = 42: the left side is a CastExpr, not a bare Column, - // so column_ndv returns None. The conjunct is unhandled and falls through - // to default_selectivity. + fn test_fallback_selectivity_right_col_uses_ndv() { + // = col_a: column is on the right-hand side of the equality. + // Verifies that column_ndv identifies the NDV regardless of operand order, + // yielding 1 / NDV(col_a) rather than falling back to default selectivity. let schema = Schema::new(vec![Field::new("a", DataType::Int32, false)]); - let cast_expr: Arc = Arc::new(CastExpr::new( - col("a", &schema).unwrap(), - DataType::Int64, - None, - )); - let predicate: Arc = Arc::new(BinaryExpr::new( - cast_expr, - Operator::Eq, - Arc::new(Literal::new(ScalarValue::Int64(Some(42)))), - )); - let col_stats = vec![ColumnStatistics { - distinct_count: Precision::Inexact(100), - ..Default::default() - }]; - let result = compute_fallback_selectivity(&predicate, &col_stats, 20); - assert!((result - 0.2).abs() < 1e-12, "expected 0.2, got {result}"); - } - - #[test] - fn test_fallback_selectivity_utf8_equality_uses_ndv() { - // name = 'alice' on a Utf8 column with NDV=60. - // Utf8 equality fails `check_support`, so fallback selectivity runs and - // returns 1/60 rather than the default 20% selectivity. - let schema = Schema::new(vec![Field::new("name", DataType::Utf8, false)]); let predicate: Arc = binary( - col("name", &schema).unwrap(), + lit(42i32), Operator::Eq, - lit("alice"), + col("a", &schema).unwrap(), &schema, ) .unwrap(); let col_stats = vec![ColumnStatistics { - distinct_count: Precision::Inexact(60), + distinct_count: Precision::Inexact(80), ..Default::default() }]; let result = compute_fallback_selectivity(&predicate, &col_stats, 20); - let expected = 1.0 / 60.0; + let expected = 1.0 / 80.0; assert!( (result - expected).abs() < 1e-12, - "expected {expected}, got {result}" + "expected 1/80 = {expected} when column is on the right side, got {result}" ); } @@ -4948,53 +4916,4 @@ mod tests { "expected default 0.2 when both columns lack NDV, got {result}" ); } - - #[test] - fn test_fallback_selectivity_col_eq_col_left_ndv_dominates() { - // col_a = col_b, NDV(a)=500, NDV(b)=100 → max picks left → 1/500 - // This exercises the `l.max(r)` path where l > r. - let schema = Schema::new(vec![ - Field::new("a", DataType::Int32, false), - Field::new("b", DataType::Int32, false), - ]); - let predicate: Arc = binary( - col("a", &schema).unwrap(), - Operator::Eq, - col("b", &schema).unwrap(), - &schema, - ) - .unwrap(); - let col_stats = vec![ - ColumnStatistics { - distinct_count: Precision::Exact(500), - ..Default::default() - }, - ColumnStatistics { - distinct_count: Precision::Exact(100), - ..Default::default() - }, - ]; - let result = compute_fallback_selectivity(&predicate, &col_stats, 20); - assert!( - (result - 1.0 / 500.0).abs() < 1e-12, - "expected 1/500 = 0.002 when left NDV dominates, got {result}" - ); - } - - #[test] - fn test_fallback_selectivity_column_index_out_of_bounds() { - // column_ndv: `column_statistics.get(col.index())` returns None when the - // column index exceeds the stats slice length. The conjunct is then - // unhandled → default selectivity is applied. - let schema = Schema::new(vec![Field::new("a", DataType::Int32, false)]); - let predicate: Arc = - binary(col("a", &schema).unwrap(), Operator::Eq, lit(1i32), &schema).unwrap(); - // Empty stats slice: col.index() == 0 is out of bounds. - let col_stats: Vec = vec![]; - let result = compute_fallback_selectivity(&predicate, &col_stats, 20); - assert!( - (result - 0.2).abs() < 1e-12, - "expected default 0.2 when column index is out of bounds, got {result}" - ); - } } From ea1ab42e154937cb40ccc467d2412fd9af2cc7cc Mon Sep 17 00:00:00 2001 From: mohitgurav20 Date: Fri, 9 Oct 2026 01:16:12 +0530 Subject: [PATCH 11/11] test: cover missing column_statistics branch --- datafusion/physical-plan/src/filter.rs | 20 ++++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/datafusion/physical-plan/src/filter.rs b/datafusion/physical-plan/src/filter.rs index a14758f7bd827..4a704933174ad 100644 --- a/datafusion/physical-plan/src/filter.rs +++ b/datafusion/physical-plan/src/filter.rs @@ -4564,6 +4564,26 @@ mod tests { ); } + #[test] + fn test_fallback_selectivity_missing_column_statistics() { + // col_0 = , but col_stats is empty. -> NDV is unknown, default selectivity is applied. + let schema = Schema::new(vec![Field::new("a", DataType::Int32, false)]); + let predicate: Arc = binary( + col("a", &schema).unwrap(), + Operator::Eq, + lit(42i32), + &schema, + ) + .unwrap(); + // Empty column statistics, so getting index 0 will return None + let col_stats = vec![]; + let result = compute_fallback_selectivity(&predicate, &col_stats, 20); + assert!( + (result - 0.2).abs() < 1e-12, + "expected default selectivity 0.2, got {result}" + ); + } + #[test] fn test_fallback_selectivity_multiple_unhandled_conjuncts() { // s <> 'abc' AND t <> 'x' AND u <> 'p'