diff --git a/datafusion/physical-plan/src/filter.rs b/datafusion/physical-plan/src/filter.rs index cc5e0846d392d..8225dcc9bd8a0 100644 --- a/datafusion/physical-plan/src/filter.rs +++ b/datafusion/physical-plan/src/filter.rs @@ -408,14 +408,24 @@ impl FilterExec { // String predicates cannot use interval analysis. Estimate them // from distinct counts and pattern form when possible, then // apply the row-count constraints implied by the predicate. - let default_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 default_selectivity_f64 = default_selectivity as f64 / 100.0; let selectivity = string_selectivity( predicate, schema, &input_stats, - default_selectivity, + default_selectivity_f64, ) - .unwrap_or(default_selectivity); + .unwrap_or_else(|| { + 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; @@ -1189,6 +1199,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. /// @@ -4779,4 +4856,403 @@ 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_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' + // 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 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), + ]); + + // (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 stats_ab = vec![ + ColumnStatistics { + distinct_count: Precision::Inexact(100), + ..Default::default() + }, + ColumnStatistics { + distinct_count: Precision::Inexact(200), + ..Default::default() + }, + ]; + let result_ab = compute_fallback_selectivity(&pred_ab, &stats_ab, 20); + assert!( + (result_ab - 1.0 / 200.0).abs() < 1e-12, + "expected 1/200 (right NDV dominates), got {result_ab}" + ); + + // (a=500, b=100) → left NDV dominates → 1/500 + let pred_ba: Arc = binary( + col("a", &schema).unwrap(), + Operator::Eq, + col("b", &schema).unwrap(), + &schema, + ) + .unwrap(); + 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 + /// 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() { + 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).unwrap(), + Operator::Eq, + lit("alice"), + &schema, + ) + .unwrap(); + let filter = Arc::new(FilterExec::try_new(predicate, input).unwrap()); + let stats = StatisticsContext::new() + .compute(filter.as_ref(), &StatisticsArgs::new()) + .unwrap(); + assert_eq!(stats.num_rows, Precision::Inexact(5)); + } + + /// Verify that `FilterExec` uses NDV-based fallback selectivity for a scalar + /// subquery equality predicate `a = `. + /// + /// 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_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 = 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::Int32, + false, + SubqueryIndex::new(0), + ScalarSubqueryResults::new(1), + )); + + // a = + let predicate: Arc = Arc::new(BinaryExpr::new( + col("a", &schema).unwrap(), + Operator::Eq, + 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() + .with_default_selectivity(20) + .unwrap(), + ); + let stats = StatisticsContext::new() + .compute(filter.as_ref(), &StatisticsArgs::new()) + .unwrap(); + + // 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_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 predicate: Arc = binary( + lit(42i32), + Operator::Eq, + col("a", &schema).unwrap(), + &schema, + ) + .unwrap(); + let col_stats = vec![ColumnStatistics { + distinct_count: Precision::Inexact(80), + ..Default::default() + }]; + let result = compute_fallback_selectivity(&predicate, &col_stats, 20); + let expected = 1.0 / 80.0; + assert!( + (result - expected).abs() < 1e-12, + "expected 1/80 = {expected} when column is on the right side, got {result}" + ); + } + + #[test] + fn test_fallback_selectivity_ndv_zero_falls_back_to_default() { + // col = lit where the column statistics report NDV = 0 (empty table or + // unknown). `column_ndv` returns None for ndv == 0, so the conjunct is + // unhandled and the 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(); + let col_stats = vec![ColumnStatistics { + distinct_count: Precision::Exact(0), // ndv == 0 → column_ndv returns None + ..Default::default() + }]; + let result = compute_fallback_selectivity(&predicate, &col_stats, 20); + assert!( + (result - 0.2).abs() < 1e-12, + "expected default 0.2 when ndv=0, got {result}" + ); + } + + #[test] + fn test_fallback_selectivity_col_eq_col_no_ndv_falls_back_to_default() { + // col_a = col_b where neither column has NDV statistics. + // Both sides return None from column_ndv, so (None, None) matches and + // the conjunct is unhandled → default selectivity applied once. + 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::new_unknown(), // no NDV + ColumnStatistics::new_unknown(), // no NDV + ]; + let result = compute_fallback_selectivity(&predicate, &col_stats, 20); + assert!( + (result - 0.2).abs() < 1e-12, + "expected default 0.2 when both columns lack NDV, got {result}" + ); + } }