diff --git a/datafusion/physical-plan/src/filter.rs b/datafusion/physical-plan/src/filter.rs index cc5e0846d392d..6443784197d0f 100644 --- a/datafusion/physical-plan/src/filter.rs +++ b/datafusion/physical-plan/src/filter.rs @@ -1125,39 +1125,44 @@ fn unique_match_limit( predicate: &Arc, statistics: &Statistics, ) -> Option { - let mut limit: Option = None; - for expr in split_conjunction(predicate) { - let Some((index, values)) = restricted_column(expr) else { - continue; - }; - let holds_once = statistics + let holds_once_fn = |index: usize| { + statistics .column_statistics .get(index) - .is_some_and(|column| holds_each_value_once(column, &statistics.num_rows)); - if !holds_once { - continue; - } - limit = Some(limit.map_or(values, |limit: usize| limit.min(values))); - } - limit + .is_some_and(|column| holds_each_value_once(column, &statistics.num_rows)) + }; + split_conjunction(predicate) + .into_iter() + .filter_map(|expr| restricted_column(expr, holds_once_fn)) + .min() } -/// The column an expression restricts to a fixed set of values, and how many values -/// that is. NULL is never one of them: it matches nothing. -fn restricted_column(expr: &Arc) -> Option<(usize, usize)> { +/// How many values an expression restricts a column to, when it restricts a column +/// for which `holds_once_fn` is true to a fixed set of values. NULL is never one of +/// them: it matches nothing. +/// +/// The column is checked before the values are counted, so IN lists on columns that +/// may hold a value more than once are never scanned. +fn restricted_column( + expr: &Arc, + holds_once_fn: impl Fn(usize) -> bool, +) -> Option { if let Some(in_list) = expr.downcast_ref::() { if in_list.negated() { return None; } let column = in_list.expr().downcast_ref::()?; - let mut values: Vec<&ScalarValue> = vec![]; + if !holds_once_fn(column.index()) { + return None; + } + let mut values: HashSet<&ScalarValue> = HashSet::new(); for expr in in_list.list() { let value = expr.downcast_ref::()?.value(); - if !value.is_null() && !values.contains(&value) { - values.push(value); + if !value.is_null() { + values.insert(value); } } - return Some((column.index(), values.len())); + return Some(values.len()); } let binary = expr.downcast_ref::()?; @@ -1173,7 +1178,7 @@ fn restricted_column(expr: &Arc) -> Option<(usize, usize)> { _ => return None, }; let value = literal.downcast_ref::()?.value(); - (!value.is_null()).then_some((column.index(), 1)) + (!value.is_null() && holds_once_fn(column.index())).then_some(1) } /// Whether the column has as many distinct values as it has non-null rows, so each @@ -1921,11 +1926,11 @@ mod tests { } /// An equality on a column that holds each value once matches one row at most, - /// including on a type interval analysis cannot read, where the default - /// selectivity would otherwise apply. + /// including on a type neither interval analysis nor the string estimates can + /// read, where the default selectivity would otherwise apply. #[tokio::test] async fn test_filter_statistics_equality_on_a_unique_column() -> Result<()> { - let schema = Schema::new(vec![Field::new("id", DataType::Utf8, true)]); + let schema = Schema::new(vec![Field::new("id", DataType::Binary, true)]); let unique = ColumnStatistics { null_count: Precision::Exact(0), distinct_count: Precision::Exact(100), @@ -1940,8 +1945,12 @@ mod tests { }, schema.clone(), )); - let predicate = - binary(col("id", &schema)?, Operator::Eq, lit("seven"), &schema)?; + let predicate = binary( + col("id", &schema)?, + Operator::Eq, + lit(ScalarValue::Binary(Some(b"seven".to_vec()))), + &schema, + )?; let filter: Arc = Arc::new(FilterExec::try_new(predicate, input)?); Ok(StatisticsContext::new() @@ -1965,6 +1974,15 @@ mod tests { assert_eq!( rows(ColumnStatistics { distinct_count: Precision::Absent, + ..unique.clone() + })?, + Precision::Inexact(20) + ); + + // Nor without a null count. + assert_eq!( + rows(ColumnStatistics { + null_count: Precision::Absent, ..unique })?, Precision::Inexact(20) @@ -1973,31 +1991,30 @@ mod tests { Ok(()) } - /// Asking a unique column for three values matches three rows at most. + /// Asking a unique column for N distinct values matches N rows at most. #[tokio::test] async fn test_filter_statistics_in_list_on_a_unique_column() -> Result<()> { use datafusion_physical_expr::expressions::in_list; + let null = || lit(ScalarValue::Utf8(None)); + let schema = Schema::new(vec![Field::new("id", DataType::Utf8, true)]); - let rows = |list: Vec<&str>, negated: bool| -> Result> { + let rows = |list: Vec>, + negated: bool| + -> Result> { let input = Arc::new(StatisticsExec::new( Statistics { - num_rows: Precision::Exact(100), - total_byte_size: Precision::Exact(800), + num_rows: Precision::Exact(100_000), + total_byte_size: Precision::Absent, column_statistics: vec![ColumnStatistics { null_count: Precision::Exact(0), - distinct_count: Precision::Exact(100), + distinct_count: Precision::Exact(100_000), ..Default::default() }], }, schema.clone(), )); - let predicate = in_list( - col("id", &schema)?, - list.into_iter().map(|value| lit(value) as _).collect(), - &negated, - &schema, - )?; + let predicate = in_list(col("id", &schema)?, list, &negated, &schema)?; let filter: Arc = Arc::new(FilterExec::try_new(predicate, input)?); Ok(StatisticsContext::new() @@ -2005,12 +2022,101 @@ mod tests { .num_rows) }; - assert_eq!(rows(vec!["a", "b", "c"], false)?, Precision::Inexact(3)); + assert_eq!( + rows(vec![lit("a"), lit("b"), lit("c")], false)?, + Precision::Inexact(3) + ); // Repeats ask for the same row twice. - assert_eq!(rows(vec!["a", "b", "a"], false)?, Precision::Inexact(2)); + assert_eq!( + rows(vec![lit("a"), lit("b"), lit("a")], false)?, + Precision::Inexact(2) + ); + // NULL matches nothing. + assert_eq!(rows(vec![null(); 32], false)?, Precision::Inexact(0)); // `NOT IN` selects nearly everything, so the default applies. - assert_eq!(rows(vec!["a", "b", "c"], true)?, Precision::Inexact(20)); + assert_eq!( + rows(vec![lit("a"), lit("b"), lit("c")], true)?, + Precision::Inexact(20_000) + ); + // A list member that is not a literal leaves the set open. + assert_eq!( + rows(vec![lit("a"), col("id", &schema)?], false)?, + Precision::Inexact(20_000) + ); + + // A large list, with non-adjacent repeats and NULLs that must not raise the + // cap. + let distinct_values = 4096; + let values: Vec<_> = (0..distinct_values) + .map(|i| lit(format!("value_{i}"))) + .collect(); + let repeated_values = values + .iter() + .chain(values.iter().rev()) + .cloned() + .chain(std::iter::repeat_n(null(), 32)) + .collect(); + assert_eq!(rows(values, false)?, Precision::Inexact(distinct_values)); + assert_eq!( + rows(repeated_values, false)?, + Precision::Inexact(distinct_values) + ); + + Ok(()) + } + /// An IN list caps the estimate only when the column it restricts holds each + /// value once; a unique column elsewhere in the table does not count. + #[test] + fn test_filter_statistics_in_list_requires_unique_target_column() -> Result<()> { + use datafusion_physical_expr::expressions::in_list; + + let schema = Schema::new(vec![ + Field::new("other", DataType::Utf8, true), + Field::new("id", DataType::Utf8, true), + ]); + let unique = ColumnStatistics { + null_count: Precision::Exact(0), + distinct_count: Precision::Exact(100_000), + ..Default::default() + }; + let non_unique = ColumnStatistics { + distinct_count: Precision::Exact(50_000), + ..unique.clone() + }; + let cases = [ + ( + "only an unrelated column is unique", + vec![unique.clone(), non_unique.clone()], + 20_000, + ), + ("the second column is unique", vec![non_unique, unique], 3), + ]; + + for (description, column_statistics, expected_rows) in cases { + let input = Arc::new(StatisticsExec::new( + Statistics { + num_rows: Precision::Exact(100_000), + total_byte_size: Precision::Absent, + column_statistics, + }, + schema.clone(), + )); + let predicate = in_list( + col("id", &schema)?, + vec![lit("a"), lit("b"), lit("a"), lit("c")], + &false, + &schema, + )?; + let filter = FilterExec::try_new(predicate, input)?; + let statistics = + StatisticsContext::new().compute(&filter, &StatisticsArgs::new())?; + assert_eq!( + statistics.num_rows, + Precision::Inexact(expected_rows), + "{description}" + ); + } Ok(()) }