From f98e16ecc4bc12bdcbf2f866e1055650b118f26c Mon Sep 17 00:00:00 2001 From: Shrivas Shankar Date: Thu, 5 Nov 2026 17:43:41 -0600 Subject: [PATCH 1/3] feat: estimate Decimal filter selectivity with interval analysis Re-applies #25790, which was reverted in #27145 because of a TPC-H Q18 regression (#27108). The next commit fixes the regression. --- .../expr-common/src/interval_arithmetic.rs | 243 +++++++++++++++++- .../physical-expr/src/expressions/cast.rs | 97 +++++++ .../physical-expr/src/intervals/utils.rs | 112 +++++++- datafusion/physical-plan/src/filter.rs | 207 ++++++++++++++- .../test_files/filter_statistics_decimal.slt | 91 +++++++ 5 files changed, 731 insertions(+), 19 deletions(-) create mode 100644 datafusion/sqllogictest/test_files/filter_statistics_decimal.slt diff --git a/datafusion/expr-common/src/interval_arithmetic.rs b/datafusion/expr-common/src/interval_arithmetic.rs index 6aa59559b8178..23e2e2f6cab3c 100644 --- a/datafusion/expr-common/src/interval_arithmetic.rs +++ b/datafusion/expr-common/src/interval_arithmetic.rs @@ -14,8 +14,10 @@ use crate::type_coercion::binary::{BinaryTypeCoercer, comparison_coercion}; use arrow::compute::{CastOptions, cast_with_options}; use arrow::datatypes::{ DataType, IntervalDayTime, IntervalMonthDayNano, IntervalUnit, + MAX_DECIMAL32_FOR_EACH_PRECISION, MAX_DECIMAL64_FOR_EACH_PRECISION, MAX_DECIMAL128_FOR_EACH_PRECISION, MAX_DECIMAL256_FOR_EACH_PRECISION, - MIN_DECIMAL128_FOR_EACH_PRECISION, MIN_DECIMAL256_FOR_EACH_PRECISION, TimeUnit, + MIN_DECIMAL32_FOR_EACH_PRECISION, MIN_DECIMAL64_FOR_EACH_PRECISION, + MIN_DECIMAL128_FOR_EACH_PRECISION, MIN_DECIMAL256_FOR_EACH_PRECISION, TimeUnit, i256, }; use datafusion_common::rounding::{alter_fp_rounding_mode, next_down, next_up}; use datafusion_common::{ @@ -24,7 +26,14 @@ use datafusion_common::{ }; macro_rules! get_extreme_value { - ($extreme:ident, $DECIMAL128_ARRAY:ident, $DECIMAL256_ARRAY:ident, $value:expr) => { + ( + $extreme:ident, + $DECIMAL32_ARRAY:ident, + $DECIMAL64_ARRAY:ident, + $DECIMAL128_ARRAY:ident, + $DECIMAL256_ARRAY:ident, + $value:expr + ) => { match $value { DataType::UInt8 => ScalarValue::UInt8(Some(u8::$extreme)), DataType::UInt16 => ScalarValue::UInt16(Some(u16::$extreme)), @@ -71,6 +80,16 @@ macro_rules! get_extreme_value { DataType::Interval(IntervalUnit::MonthDayNano) => { ScalarValue::IntervalMonthDayNano(Some(IntervalMonthDayNano::$extreme)) } + DataType::Decimal32(precision, scale) => ScalarValue::Decimal32( + Some($DECIMAL32_ARRAY[*precision as usize]), + *precision, + *scale, + ), + DataType::Decimal64(precision, scale) => ScalarValue::Decimal64( + Some($DECIMAL64_ARRAY[*precision as usize]), + *precision, + *scale, + ), DataType::Decimal128(precision, scale) => ScalarValue::Decimal128( Some($DECIMAL128_ARRAY[*precision as usize]), *precision, @@ -87,7 +106,15 @@ macro_rules! get_extreme_value { } macro_rules! value_transition { - ($bound:ident, $direction:expr, $value:expr) => { + ( + $bound:ident, + $direction:expr, + $DECIMAL32_ARRAY:ident, + $DECIMAL64_ARRAY:ident, + $DECIMAL128_ARRAY:ident, + $DECIMAL256_ARRAY:ident, + $value:expr + ) => { match $value { UInt8(Some(value)) if value == u8::$bound => UInt8(None), UInt16(Some(value)) if value == u16::$bound => UInt16(None), @@ -134,6 +161,28 @@ macro_rules! value_transition { { IntervalMonthDayNano(None) } + // Decimals saturate at the extreme value of their precision, not of + // the underlying integer type: + Decimal32(Some(value), precision, scale) + if value == $DECIMAL32_ARRAY[precision as usize] => + { + Decimal32(None, precision, scale) + } + Decimal64(Some(value), precision, scale) + if value == $DECIMAL64_ARRAY[precision as usize] => + { + Decimal64(None, precision, scale) + } + Decimal128(Some(value), precision, scale) + if value == $DECIMAL128_ARRAY[precision as usize] => + { + Decimal128(None, precision, scale) + } + Decimal256(Some(value), precision, scale) + if value == $DECIMAL256_ARRAY[precision as usize] => + { + Decimal256(None, precision, scale) + } _ => next_value_helper::<$direction>($value), } }; @@ -1195,6 +1244,8 @@ fn handle_overflow( (true, false) => { get_extreme_value!( MIN, + MIN_DECIMAL32_FOR_EACH_PRECISION, + MIN_DECIMAL64_FOR_EACH_PRECISION, MIN_DECIMAL128_FOR_EACH_PRECISION, MIN_DECIMAL256_FOR_EACH_PRECISION, dt @@ -1203,6 +1254,8 @@ fn handle_overflow( (false, true) => { get_extreme_value!( MAX, + MAX_DECIMAL32_FOR_EACH_PRECISION, + MAX_DECIMAL64_FOR_EACH_PRECISION, MAX_DECIMAL128_FOR_EACH_PRECISION, MAX_DECIMAL256_FOR_EACH_PRECISION, dt @@ -1215,14 +1268,30 @@ fn handle_overflow( // used without caution. fn next_value(value: ScalarValue) -> ScalarValue { use ScalarValue::*; - value_transition!(MAX, true, value) + value_transition!( + MAX, + true, + MAX_DECIMAL32_FOR_EACH_PRECISION, + MAX_DECIMAL64_FOR_EACH_PRECISION, + MAX_DECIMAL128_FOR_EACH_PRECISION, + MAX_DECIMAL256_FOR_EACH_PRECISION, + value + ) } // This function should remain private since it may corrupt the an interval if // used without caution. fn prev_value(value: ScalarValue) -> ScalarValue { use ScalarValue::*; - value_transition!(MIN, false, value) + value_transition!( + MIN, + false, + MIN_DECIMAL32_FOR_EACH_PRECISION, + MIN_DECIMAL64_FOR_EACH_PRECISION, + MIN_DECIMAL128_FOR_EACH_PRECISION, + MIN_DECIMAL256_FOR_EACH_PRECISION, + value + ) } trait OneTrait: Sized + std::ops::Add + std::ops::Sub { @@ -1233,6 +1302,12 @@ macro_rules! impl_OneTrait{ } impl_OneTrait! {u8, u16, u32, u64, i8, i16, i32, i64, i128} +impl OneTrait for i256 { + fn one() -> Self { + i256::ONE + } +} + impl OneTrait for IntervalDayTime { fn one() -> Self { IntervalDayTime { @@ -1287,6 +1362,23 @@ fn next_value_helper(value: ScalarValue) -> ScalarValue { UInt16(Some(val)) => UInt16(Some(increment_decrement::(val))), UInt32(Some(val)) => UInt32(Some(increment_decrement::(val))), UInt64(Some(val)) => UInt64(Some(increment_decrement::(val))), + // Decimals step by one unit in the last place, keeping precision and scale: + Decimal32(Some(val), precision, scale) => { + Decimal32(Some(increment_decrement::(val)), precision, scale) + } + Decimal64(Some(val), precision, scale) => { + Decimal64(Some(increment_decrement::(val)), precision, scale) + } + Decimal128(Some(val), precision, scale) => Decimal128( + Some(increment_decrement::(val)), + precision, + scale, + ), + Decimal256(Some(val), precision, scale) => Decimal256( + Some(increment_decrement::(val)), + precision, + scale, + ), DurationSecond(Some(val)) => { DurationSecond(Some(increment_decrement::(val))) } @@ -2268,7 +2360,10 @@ mod tests { use crate::interval_arithmetic::NullableInterval; use arrow::compute::CastOptions; - use arrow::datatypes::DataType; + use arrow::datatypes::{ + DataType, MAX_DECIMAL128_FOR_EACH_PRECISION, MIN_DECIMAL128_FOR_EACH_PRECISION, + i256, + }; use datafusion_common::rounding::{next_down, next_up}; use datafusion_common::{Result, ScalarValue}; @@ -2581,6 +2676,124 @@ mod tests { Ok(()) } + #[test] + fn test_next_prev_value_decimal() { + use ScalarValue::*; + + // Decimals step by one unit in the last place, keeping precision and scale: + let cases = vec![ + (Decimal32(Some(2400), 9, 2), Decimal32(Some(2401), 9, 2)), + (Decimal64(Some(2400), 18, 2), Decimal64(Some(2401), 18, 2)), + (Decimal128(Some(2400), 15, 2), Decimal128(Some(2401), 15, 2)), + ( + Decimal256(Some(i256::from(2400)), 40, 2), + Decimal256(Some(i256::from(2401)), 40, 2), + ), + (Decimal128(Some(-1), 3, 2), Decimal128(Some(0), 3, 2)), + ]; + for (value, next) in cases { + assert_eq!(next_value(value.clone()), next); + assert_eq!(prev_value(next), value); + } + + // Values at the extremes of their precision (e.g. -9.99 and 9.99 for + // precision 3, scale 2) become unbounded: + let d128_max = MAX_DECIMAL128_FOR_EACH_PRECISION[38]; + let d128_min = MIN_DECIMAL128_FOR_EACH_PRECISION[38]; + let cases = vec![ + ( + Decimal32(Some(-999), 3, 2), + Decimal32(Some(999), 3, 2), + Decimal32(Some(-998), 3, 2), + Decimal32(Some(998), 3, 2), + Decimal32(None, 3, 2), + ), + ( + Decimal64(Some(-999), 3, 2), + Decimal64(Some(999), 3, 2), + Decimal64(Some(-998), 3, 2), + Decimal64(Some(998), 3, 2), + Decimal64(None, 3, 2), + ), + ( + Decimal128(Some(-999), 3, 2), + Decimal128(Some(999), 3, 2), + Decimal128(Some(-998), 3, 2), + Decimal128(Some(998), 3, 2), + Decimal128(None, 3, 2), + ), + ( + Decimal256(Some(i256::from(-999)), 3, 2), + Decimal256(Some(i256::from(999)), 3, 2), + Decimal256(Some(i256::from(-998)), 3, 2), + Decimal256(Some(i256::from(998)), 3, 2), + Decimal256(None, 3, 2), + ), + ( + Decimal128(Some(d128_min), 38, 0), + Decimal128(Some(d128_max), 38, 0), + Decimal128(Some(d128_min + 1), 38, 0), + Decimal128(Some(d128_max - 1), 38, 0), + Decimal128(None, 38, 0), + ), + ]; + for (min, max, after_min, before_max, inf) in cases { + assert_eq!(next_value(max.clone()), inf); + assert_eq!(prev_value(max), before_max); + + assert_eq!(prev_value(min.clone()), inf); + assert_eq!(next_value(min), after_min); + + assert_eq!(next_value(inf.clone()), inf); + assert_eq!(prev_value(inf.clone()), inf); + } + } + + #[test] + fn test_satisfy_strict_comparison_decimal() -> Result<()> { + use ScalarValue::*; + + // x in [0.00, 50.00], constraint 24.00 > x tightens x to [0.00, 23.99]: + let x = + Interval::try_new(Decimal128(Some(0), 15, 2), Decimal128(Some(5000), 15, 2))?; + let literal = Interval::try_new( + Decimal128(Some(2400), 15, 2), + Decimal128(Some(2400), 15, 2), + )?; + let (_, x_lt) = satisfy_greater(&literal, &x, true)?.unwrap(); + assert_eq!( + x_lt, + Interval::try_new(Decimal128(Some(0), 15, 2), Decimal128(Some(2399), 15, 2))? + ); + + // Constraint x > 24.00 tightens x to [24.01, 50.00]: + let (x_gt, _) = satisfy_greater(&x, &literal, true)?.unwrap(); + assert_eq!( + x_gt, + Interval::try_new( + Decimal128(Some(2401), 15, 2), + Decimal128(Some(5000), 15, 2), + )? + ); + + // Non-strict comparisons keep the closed bound: + let (_, x_le) = satisfy_greater(&literal, &x, false)?.unwrap(); + assert_eq!( + x_le, + Interval::try_new(Decimal128(Some(0), 15, 2), Decimal128(Some(2400), 15, 2))? + ); + + // A strict bound at the precision's extreme conservatively becomes + // unbounded instead of stepping past it (x > 9.99 for Decimal128(3, 2)): + let x = Interval::make_unbounded(&DataType::Decimal128(3, 2))?; + let max = + Interval::try_new(Decimal128(Some(999), 3, 2), Decimal128(Some(999), 3, 2))?; + let (x_gt, _) = satisfy_greater(&x, &max, true)?.unwrap(); + assert_eq!(x_gt, Interval::make_unbounded(&DataType::Decimal128(3, 2))?); + + Ok(()) + } + #[test] fn test_new_interval() -> Result<()> { use ScalarValue::*; @@ -4394,6 +4607,24 @@ mod tests { ScalarValue::Decimal128(Some(99999999999999999999999999999999999999), 38, 35) ); + // Test Decimal32/Decimal64 underflow handling: + let op = Operator::Minus; + let dt = DataType::Decimal32(3, 2); + let lhs = ScalarValue::Decimal32(Some(-999), 3, 2); + let rhs = ScalarValue::Decimal32(Some(1), 3, 2); + let result = handle_overflow::(&dt, op, &lhs, &rhs); + assert_eq!(result, ScalarValue::Decimal32(Some(-999), 3, 2)); + let result = handle_overflow::(&dt, op, &lhs, &rhs); + assert_eq!(result, ScalarValue::Decimal32(None, 3, 2)); + + let dt = DataType::Decimal64(3, 2); + let lhs = ScalarValue::Decimal64(Some(-999), 3, 2); + let rhs = ScalarValue::Decimal64(Some(1), 3, 2); + let result = handle_overflow::(&dt, op, &lhs, &rhs); + assert_eq!(result, ScalarValue::Decimal64(Some(-999), 3, 2)); + let result = handle_overflow::(&dt, op, &lhs, &rhs); + assert_eq!(result, ScalarValue::Decimal64(None, 3, 2)); + Ok(()) } diff --git a/datafusion/physical-expr/src/expressions/cast.rs b/datafusion/physical-expr/src/expressions/cast.rs index 7fbed9187b971..e56bf4dcbc8b9 100644 --- a/datafusion/physical-expr/src/expressions/cast.rs +++ b/datafusion/physical-expr/src/expressions/cast.rs @@ -699,6 +699,28 @@ fn can_propagate_cast_constraints(source: &DataType, target: &DataType) -> bool || (source.is_integer() && target.is_integer()) // NaN bounds are unbounded; finite Float32 values widen exactly. || (*source == Float32 && *target == Float64) + || is_decimal_widening(source, target) +} + +/// Whether a decimal-to-decimal cast keeps every value exactly, i.e. the target +/// has at least as many fractional and integer digits as the source. Casting +/// bounds back to the source then rounds to a neighboring representable value, +/// which never excludes a valid input. +fn is_decimal_widening(source: &DataType, target: &DataType) -> bool { + fn precision_scale(data_type: &DataType) -> Option<(i16, i16)> { + match data_type { + Decimal32(p, s) | Decimal64(p, s) | Decimal128(p, s) | Decimal256(p, s) => { + Some((*p as i16, *s as i16)) + } + _ => None, + } + } + match (precision_scale(source), precision_scale(target)) { + (Some((source_p, source_s)), Some((target_p, target_s))) => { + target_s >= source_s && target_p - target_s >= source_p - source_s + } + _ => false, + } } #[cfg(feature = "proto")] @@ -920,6 +942,11 @@ mod tests { (Int32, Float64, true), (Float32, Float64, true), (Decimal128(4, 1), Decimal128(4, 1), true), + (Decimal128(4, 1), Decimal128(6, 2), true), + (Decimal128(15, 2), Decimal128(30, 15), true), + (Decimal32(9, 2), Decimal128(20, 4), true), + (Decimal128(6, 2), Decimal128(6, 3), false), + (Decimal128(6, 2), Decimal128(8, 1), false), ] { let schema = Schema::new(vec![Field::new("x", source.clone(), true)]); let expr = CastExpr::new(col("x", &schema)?, target.clone(), None); @@ -940,6 +967,76 @@ mod tests { Ok(()) } + #[test] + fn test_decimal_widening_constraint_boundaries() -> Result<()> { + fn decimal(data_type: &DataType, value: i128) -> ScalarValue { + match data_type { + Decimal32(p, s) => ScalarValue::Decimal32(Some(value as i32), *p, *s), + Decimal64(p, s) => ScalarValue::Decimal64(Some(value as i64), *p, *s), + Decimal128(p, s) => ScalarValue::Decimal128(Some(value), *p, *s), + Decimal256(p, s) => { + ScalarValue::Decimal256(Some(i256::from_i128(value)), *p, *s) + } + _ => unreachable!(), + } + } + + // (source, target, scale factor, extreme unscaled source value) + for (source, target, factor, extreme) in [ + (Decimal128(3, 2), Decimal128(5, 3), 10, 999), + // Cross-width widening: + (Decimal32(4, 2), Decimal128(10, 4), 100, 9999), + (Decimal64(5, 1), Decimal256(10, 3), 100, 99999), + // Negative scale: + (Decimal128(3, -1), Decimal128(5, 0), 10, 999), + ] { + // Source values near zero and near the extremes of the precision: + let values: Vec = (-5..=5) + .chain(extreme - 5..=extreme) + .chain(-extreme..=-extreme + 5) + .collect(); + // Output bounds at, between, and just past the widened values, + // including past the source's extremes, so casting back rounds in + // both directions and overflows: + let mut bounds: Vec = values + .iter() + .flat_map(|v| { + let w = v * factor; + [w - factor, w - 1, w, w + 1, w + factor / 2, w + factor] + }) + .collect(); + bounds.sort_unstable(); + bounds.dedup(); + + let schema = + Arc::new(Schema::new(vec![Field::new("x", source.clone(), false)])); + let expr = CastExpr::new(col("x", &schema)?, target.clone(), None); + let input = Interval::make_unbounded(&source)?; + for (i, lower) in bounds.iter().enumerate() { + for upper in &bounds[i..] { + let output = Interval::try_new( + decimal(&target, *lower), + decimal(&target, *upper), + )?; + let propagated = + expr.propagate_constraints(&output, &[&input])?.unwrap(); + assert_eq!(propagated.len(), 1); + for value in &values { + if output.contains_value(decimal(&target, value * factor))? { + assert!( + propagated[0].contains_value(decimal(&source, *value))?, + "{source} -> {target}: input={value}, output={output}, \ + propagated={:?}", + propagated[0] + ); + } + } + } + } + } + Ok(()) + } + #[test] fn test_float_widening_constraint_boundaries() -> Result<()> { let mut values = vec![ diff --git a/datafusion/physical-expr/src/intervals/utils.rs b/datafusion/physical-expr/src/intervals/utils.rs index 35b1ee6ffd867..8f93e49c22abc 100644 --- a/datafusion/physical-expr/src/intervals/utils.rs +++ b/datafusion/physical-expr/src/intervals/utils.rs @@ -23,19 +23,57 @@ use datafusion_expr::interval_arithmetic::Interval; /// will relax as more types of `PhysicalExpr`s and `Operator`s are supported. /// Currently, [`CastExpr`], [`NegativeExpr`], [`BinaryExpr`], [`Column`] and [`Literal`] are supported. pub fn check_support(expr: &Arc, schema: &SchemaRef) -> bool { + check_support_with(expr, schema, is_datatype_supported) +} + +/// Like [`check_support`], but additionally accepts decimal types. This is +/// used when estimating filter statistics, where decimal predicates can be +/// analyzed with interval arithmetic. Other users of [`check_support`], such +/// as symmetric hash join pruning, do not support decimals yet. +pub fn check_statistics_support( + expr: &Arc, + schema: &SchemaRef, +) -> bool { + check_support_with(expr, schema, |data_type| { + is_datatype_supported(data_type) + || matches!( + data_type, + DataType::Decimal32(_, _) + | DataType::Decimal64(_, _) + | DataType::Decimal128(_, _) + | DataType::Decimal256(_, _) + ) + }) +} + +fn check_support_with( + expr: &Arc, + schema: &SchemaRef, + is_type_supported: fn(&DataType) -> bool, +) -> bool { if let Some(binary_expr) = expr.downcast_ref::() { + // Arithmetic is only analyzed over the types `check_support` accepts: + // decimal arithmetic changes the precision and scale of its result, and + // propagating constraints back to its operands would not restore their + // types. + let is_type_supported: fn(&DataType) -> bool = + if binary_expr.op().is_numerical_operators() { + is_datatype_supported + } else { + is_type_supported + }; is_operator_supported(binary_expr.op()) - && check_support(binary_expr.left(), schema) - && check_support(binary_expr.right(), schema) + && check_support_with(binary_expr.left(), schema, is_type_supported) + && check_support_with(binary_expr.right(), schema, is_type_supported) } else if let Some(column) = expr.downcast_ref::() { if let Ok(field) = schema.field_with_name(column.name()) { - is_datatype_supported(field.data_type()) + is_type_supported(field.data_type()) } else { false } } else if let Some(literal) = expr.downcast_ref::() { if let Ok(dt) = literal.data_type(schema) { - is_datatype_supported(&dt) + is_type_supported(&dt) } else { false } @@ -44,9 +82,10 @@ pub fn check_support(expr: &Arc, schema: &SchemaRef) -> bool { // interval endpoints to a type with a different ordering (e.g. integers // to strings) does not preserve the interval, so such expressions are // left unanalyzed rather than producing bogus bounds. - is_datatype_supported(cast.cast_type()) && check_support(cast.expr(), schema) + is_type_supported(cast.cast_type()) + && check_support_with(cast.expr(), schema, is_type_supported) } else if let Some(negative) = expr.downcast_ref::() { - check_support(negative.arg(), schema) + check_support_with(negative.arg(), schema, is_type_supported) } else { false } @@ -184,3 +223,64 @@ fn interval_dt_to_duration_ms(dt: &IntervalDayTime) -> Result { ) } } + +#[cfg(test)] +mod tests { + use super::*; + + use arrow::datatypes::{Field, Schema}; + + #[test] + fn check_statistics_support_decimal() { + for data_type in [ + DataType::Decimal32(9, 2), + DataType::Decimal64(18, 2), + DataType::Decimal128(15, 2), + DataType::Decimal256(40, 2), + ] { + let schema = + Arc::new(Schema::new(vec![Field::new("a", data_type.clone(), false)])); + let literal = ScalarValue::try_from(&data_type).unwrap(); + let expr = Arc::new(BinaryExpr::new( + Arc::new(Column::new("a", 0)), + Operator::Lt, + Arc::new(Literal::new(literal)), + )) as Arc; + + assert!(check_statistics_support(&expr, &schema)); + // Decimal support is limited to statistics estimation for now: + assert!(!check_support(&expr, &schema)); + + // Decimal arithmetic is not analyzed, since it changes the precision + // and scale of its operands: + let literal = ScalarValue::try_from(&data_type).unwrap(); + for op in [Operator::Plus, Operator::Minus, Operator::Multiply] { + let arithmetic = Arc::new(BinaryExpr::new( + Arc::new(Column::new("a", 0)), + op, + Arc::new(Literal::new(literal.clone())), + )) as Arc; + let expr = Arc::new(BinaryExpr::new( + arithmetic, + Operator::Gt, + Arc::new(Literal::new(literal.clone())), + )) as Arc; + assert!(!check_statistics_support(&expr, &schema), "{op}"); + } + } + + // Arithmetic over other supported types is unaffected: + let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int32, false)])); + let expr = Arc::new(BinaryExpr::new( + Arc::new(BinaryExpr::new( + Arc::new(Column::new("a", 0)), + Operator::Plus, + Arc::new(Literal::new(ScalarValue::Int32(Some(1)))), + )), + Operator::Gt, + Arc::new(Literal::new(ScalarValue::Int32(Some(5)))), + )) as Arc; + assert!(check_statistics_support(&expr, &schema)); + assert!(check_support(&expr, &schema)); + } +} diff --git a/datafusion/physical-plan/src/filter.rs b/datafusion/physical-plan/src/filter.rs index 0e8cf991cfaac..037cf65008e46 100644 --- a/datafusion/physical-plan/src/filter.rs +++ b/datafusion/physical-plan/src/filter.rs @@ -53,7 +53,7 @@ use datafusion_physical_expr::equivalence::ProjectionMapping; use datafusion_physical_expr::expressions::{ BinaryExpr, Column, InListExpr, IsNotNullExpr, IsNullExpr, LikeExpr, Literal, lit, }; -use datafusion_physical_expr::intervals::utils::check_support; +use datafusion_physical_expr::intervals::utils::check_statistics_support; use datafusion_physical_expr::utils::collect_columns; use datafusion_physical_expr::{ AcrossPartitions, AnalysisContext, ConstExpr, ExprBoundaries, PhysicalExpr, analyze, @@ -389,7 +389,7 @@ impl FilterExec { } else { let null_rejecting_columns = collect_null_rejecting_columns(predicate); - if check_support(predicate, schema) { + if check_statistics_support(predicate, schema) { let input_analysis_ctx = AnalysisContext::try_from_statistics( schema, &input_stats.column_statistics, @@ -2225,6 +2225,199 @@ mod tests { Ok(()) } + #[tokio::test] + async fn test_filter_statistics_decimal_expr() -> Result<()> { + // Table: + // a: Decimal128(15, 2), min=0.00, max=0.10 + let decimal = |v: i128| ScalarValue::Decimal128(Some(v), 15, 2); + let schema = + Schema::new(vec![Field::new("a", DataType::Decimal128(15, 2), false)]); + let input = Arc::new(StatisticsExec::new( + Statistics { + num_rows: Precision::Inexact(1100), + total_byte_size: Precision::Absent, + column_statistics: vec![ColumnStatistics { + min_value: Precision::Inexact(decimal(0)), + max_value: Precision::Inexact(decimal(10)), + ..Default::default() + }], + }, + schema.clone(), + )); + + // a >= 0.05 AND a < 0.08 + let predicate: Arc = binary( + binary(col("a", &schema)?, Operator::GtEq, lit(decimal(5)), &schema)?, + Operator::And, + binary(col("a", &schema)?, Operator::Lt, lit(decimal(8)), &schema)?, + &schema, + )?; + + // WHERE a >= 0.05 AND a < 0.08 + let filter: Arc = + Arc::new(FilterExec::try_new(predicate, input)?); + + // The strict bound tightens to 0.07, so [0.05, 0.07] keeps 3 of the 11 + // representable values in [0.00, 0.10]. + let statistics = + StatisticsContext::new().compute(filter.as_ref(), &StatisticsArgs::new())?; + assert_eq!(statistics.num_rows, Precision::Inexact(300)); + assert_eq!( + statistics.column_statistics, + vec![ColumnStatistics { + null_count: Precision::Exact(0), + min_value: Precision::Inexact(decimal(5)), + max_value: Precision::Inexact(decimal(7)), + ..Default::default() + }] + ); + + Ok(()) + } + + /// Arithmetic changes the precision and scale of decimal results, so it must + /// not narrow decimal columns: these predicates fall back to the default + /// selectivity and keep the input bounds in the column's own type. + #[tokio::test] + async fn test_filter_statistics_decimal_arithmetic_expr() -> Result<()> { + // Table: + // a: Decimal128(15, 2), min=1.00, max=10.00 + let decimal = |v: i128| ScalarValue::Decimal128(Some(v), 15, 2); + let schema = + Schema::new(vec![Field::new("a", DataType::Decimal128(15, 2), false)]); + let input_stats = Statistics { + num_rows: Precision::Inexact(1000), + total_byte_size: Precision::Absent, + column_statistics: vec![ColumnStatistics { + min_value: Precision::Inexact(decimal(100)), + max_value: Precision::Inexact(decimal(1000)), + ..Default::default() + }], + }; + + // a * 2.00 > 5.0000 and a - 2.00 > 5.00 + for (op, rhs) in [ + (Operator::Multiply, 50_000_i128), // 5.0000 at the product's scale 4 + (Operator::Minus, 500_i128), // 5.00 at the difference's scale 2 + ] { + let arithmetic = + Arc::new(BinaryExpr::new(col("a", &schema)?, op, lit(decimal(200)))) + as Arc; + let DataType::Decimal128(precision, scale) = arithmetic.data_type(&schema)? + else { + unreachable!() + }; + let predicate = Arc::new(BinaryExpr::new( + arithmetic, + Operator::Gt, + lit(ScalarValue::Decimal128(Some(rhs), precision, scale)), + )); + let input = + Arc::new(StatisticsExec::new(input_stats.clone(), schema.clone())); + let filter: Arc = + Arc::new(FilterExec::try_new(predicate, input)?); + + let statistics = StatisticsContext::new() + .compute(filter.as_ref(), &StatisticsArgs::new())?; + assert_eq!(statistics.num_rows, Precision::Inexact(200), "{op}"); + let column = &statistics.column_statistics[0]; + assert_eq!(column.min_value, Precision::Inexact(decimal(100)), "{op}"); + assert_eq!(column.max_value, Precision::Inexact(decimal(1000)), "{op}"); + for bound in [&column.min_value, &column.max_value] { + let value = bound.get_value().unwrap(); + assert_eq!(value.data_type(), DataType::Decimal128(15, 2), "{op}"); + } + } + + // Negation keeps the decimal type, so -a >= -5.00 still narrows `a` to + // [1.00, 5.00]: 401 of the 901 representable values in [1.00, 10.00], + // rounded up. + let predicate = Arc::new(BinaryExpr::new( + Arc::new(NegativeExpr::new(col("a", &schema)?)), + Operator::GtEq, + lit(decimal(-500)), + )); + let input = Arc::new(StatisticsExec::new(input_stats, schema.clone())); + let filter: Arc = + Arc::new(FilterExec::try_new(predicate, input)?); + let statistics = + StatisticsContext::new().compute(filter.as_ref(), &StatisticsArgs::new())?; + assert_eq!(statistics.num_rows, Precision::Inexact(446)); + let column = &statistics.column_statistics[0]; + assert_eq!(column.min_value, Precision::Inexact(decimal(100))); + assert_eq!(column.max_value, Precision::Inexact(decimal(500))); + + Ok(()) + } + + #[tokio::test] + async fn test_filter_statistics_decimal_mixed_scale_expr() -> Result<()> { + // Table: + // a: Decimal128(15, 2), min=0.00, max=0.10 + let schema = + Schema::new(vec![Field::new("a", DataType::Decimal128(15, 2), false)]); + let input = Arc::new(StatisticsExec::new( + Statistics { + num_rows: Precision::Inexact(1100), + total_byte_size: Precision::Absent, + column_statistics: vec![ColumnStatistics { + min_value: Precision::Inexact(ScalarValue::Decimal128( + Some(0), + 15, + 2, + )), + max_value: Precision::Inexact(ScalarValue::Decimal128( + Some(10), + 15, + 2, + )), + ..Default::default() + }], + }, + schema.clone(), + )); + + // Type coercion compares `a` with the float literals as Decimal128(30, 15), + // so the column is cast to the literals' scale: + // CAST(a AS Decimal128(30, 15)) >= 0.055 AND CAST(a AS Decimal128(30, 15)) < 0.085 + let coerced = DataType::Decimal128(30, 15); + let literal = |v: i128| lit(ScalarValue::Decimal128(Some(v), 30, 15)); + let predicate: Arc = binary( + binary( + cast(col("a", &schema)?, &schema, coerced.clone())?, + Operator::GtEq, + literal(55_000_000_000_000), + &schema, + )?, + Operator::And, + binary( + cast(col("a", &schema)?, &schema, coerced)?, + Operator::Lt, + literal(85_000_000_000_000), + &schema, + )?, + &schema, + )?; + let filter: Arc = + Arc::new(FilterExec::try_new(predicate, input)?); + + let statistics = + StatisticsContext::new().compute(filter.as_ref(), &StatisticsArgs::new())?; + // The bounds are cast back to scale 2 as [0.06, 0.08], keeping 3 of the + // 11 representable values in [0.00, 0.10]. + assert_eq!(statistics.num_rows, Precision::Inexact(300)); + assert_eq!( + statistics.column_statistics, + vec![ColumnStatistics { + min_value: Precision::Inexact(ScalarValue::Decimal128(Some(6), 15, 2)), + max_value: Precision::Inexact(ScalarValue::Decimal128(Some(8), 15, 2)), + ..Default::default() + }] + ); + + Ok(()) + } + #[tokio::test] async fn test_filter_statistics_column_level_nested() -> Result<()> { // Table: @@ -2868,9 +3061,9 @@ mod tests { #[tokio::test] async fn test_custom_filter_selectivity() -> Result<()> { - // Need a decimal to trigger inexact selectivity - let schema = - Schema::new(vec![Field::new("a", DataType::Decimal128(2, 3), false)]); + // Need a type unsupported by interval analysis to trigger the default + // selectivity + let schema = Schema::new(vec![Field::new("a", DataType::Utf8, false)]); let input = Arc::new(StatisticsExec::new( Statistics { num_rows: Precision::Inexact(1000), @@ -2879,11 +3072,11 @@ mod tests { }, schema, )); - // WHERE a = 10 + // WHERE a = '10' let predicate = Arc::new(BinaryExpr::new( Arc::new(Column::new("a", 0)), Operator::Eq, - Arc::new(Literal::new(ScalarValue::Decimal128(Some(10), 10, 10))), + Arc::new(Literal::new(ScalarValue::from("10"))), )); let filter = FilterExec::try_new(predicate, input)?; let statistics = diff --git a/datafusion/sqllogictest/test_files/filter_statistics_decimal.slt b/datafusion/sqllogictest/test_files/filter_statistics_decimal.slt new file mode 100644 index 0000000000000..09e525076f177 --- /dev/null +++ b/datafusion/sqllogictest/test_files/filter_statistics_decimal.slt @@ -0,0 +1,91 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: See the NOTICE file distributed with this work for additional information regarding copyright ownership +# SPDX-FileContributor: Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements + +# Tests for filter selectivity estimates on decimal columns, using min/max +# statistics from a parquet file. +# +# l_discount cycles through 0.00..0.10 and l_quantity through 1.00..50.00. Since +# lcm(11, 50) = 550, each (l_discount, l_quantity) pair appears exactly twice +# in the 1100 rows. + +query I +COPY ( + SELECT + CAST((value % 11) / 100.0 AS DECIMAL(15, 2)) AS l_discount, + CAST(1 + value % 50 AS DECIMAL(15, 2)) AS l_quantity + FROM range(1100) +) +TO 'test_files/scratch/filter_statistics_decimal/lineitem.parquet' +STORED AS PARQUET; +---- +1100 + +statement ok +set datafusion.explain.physical_plan_only = true; + +statement ok +set datafusion.explain.show_statistics = true; + +statement ok +set datafusion.execution.target_partitions = 1; + +statement ok +CREATE EXTERNAL TABLE lineitem +STORED AS PARQUET +LOCATION 'test_files/scratch/filter_statistics_decimal/lineitem.parquet'; + +# TPC-H Q6-style decimal ranges with fractional literals: the estimate uses +# 3 of 11 discount values and 2300 of 4901 quantity values (1.00..23.99). +query TT +EXPLAIN SELECT l_discount FROM lineitem +WHERE l_discount >= 0.05 AND l_discount <= 0.07 AND l_quantity < 24; +---- +physical_plan +01)FilterExec: l_discount@0 >= 0.05 AND l_discount@0 <= 0.07 AND l_quantity@1 < 24.00, projection=[l_discount@0], statistics=[Rows=Inexact(141), Bytes=Inexact(4506), [(Col[0]: Min=Inexact(Decimal128(0.05,15,2)) Max=Inexact(Decimal128(0.07,15,2)) Null=Exact(0) ScanBytes=Inexact(2253))]] +02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/filter_statistics_decimal/lineitem.parquet]]}, projection=[l_discount, l_quantity], file_type=parquet, predicate=l_discount@0 >= 0.05 AND l_discount@0 <= 0.07 AND l_quantity@1 < 24.00, pruning_predicate=l_discount_null_count@1 != row_count@2 AND l_discount_max@0 >= 0.05 AND l_discount_null_count@1 != row_count@2 AND l_discount_min@3 <= 0.07 AND l_quantity_null_count@5 != row_count@2 AND l_quantity_min@4 < 24.00, required_guarantees=[], statistics=[Rows=Inexact(1100), Bytes=Inexact(35200), [(Col[0]: Min=Inexact(Decimal128(0.00,15,2)) Max=Inexact(Decimal128(0.10,15,2)) Null=Inexact(0) ScanBytes=Inexact(17600)),(Col[1]: Min=Inexact(Decimal128(1.00,15,2)) Max=Inexact(Decimal128(50.00,15,2)) Null=Inexact(0) ScanBytes=Inexact(17600))]] + +query I +SELECT count(*) FROM lineitem +WHERE l_discount >= 0.05 AND l_discount <= 0.07 AND l_quantity < 24; +---- +138 + +# A literal with more digits than the column's scale is compared through a +# cast to a wider decimal; the column is still narrowed to [0.06, 0.10]. +query TT +EXPLAIN SELECT l_discount FROM lineitem WHERE l_discount >= 0.055; +---- +physical_plan +01)FilterExec: CAST(l_discount@0 AS Decimal128(30, 15)) >= 0.055000000000000, statistics=[Rows=Inexact(500), Bytes=Inexact(8000), [(Col[0]: Min=Inexact(Decimal128(0.06,15,2)) Max=Inexact(Decimal128(0.10,15,2)) Null=Inexact(0) ScanBytes=Inexact(8000))]] +02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/filter_statistics_decimal/lineitem.parquet]]}, projection=[l_discount], file_type=parquet, predicate=CAST(l_discount@0 AS Decimal128(30, 15)) >= 0.055000000000000, pruning_predicate=l_discount_null_count@1 != row_count@2 AND CAST(l_discount_max@0 AS Decimal128(30, 15);N) >= 0.055000000000000, required_guarantees=[], statistics=[Rows=Inexact(1100), Bytes=Inexact(17600), [(Col[0]: Min=Inexact(Decimal128(0.00,15,2)) Max=Inexact(Decimal128(0.10,15,2)) Null=Inexact(0) ScanBytes=Inexact(17600))]] + +query I +SELECT count(*) FROM lineitem WHERE l_discount >= 0.055; +---- +500 + +# Decimal arithmetic is not analyzed and uses the default selectivity. +query TT +EXPLAIN SELECT l_discount FROM lineitem WHERE l_discount * 2 > 0.1; +---- +physical_plan +01)FilterExec: l_discount@0 * 2 > 0.10, statistics=[Rows=Inexact(220), Bytes=Inexact(3520), [(Col[0]: Min=Inexact(Decimal128(0.00,15,2)) Max=Inexact(Decimal128(0.10,15,2)) Null=Inexact(0) ScanBytes=Inexact(3520))]] +02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/filter_statistics_decimal/lineitem.parquet]]}, projection=[l_discount], file_type=parquet, predicate=l_discount@0 * 2 > 0.10, statistics=[Rows=Inexact(1100), Bytes=Inexact(17600), [(Col[0]: Min=Inexact(Decimal128(0.00,15,2)) Max=Inexact(Decimal128(0.10,15,2)) Null=Inexact(0) ScanBytes=Inexact(17600))]] + +query I +SELECT count(*) FROM lineitem WHERE l_discount * 2 > 0.1; +---- +500 + +statement ok +DROP TABLE lineitem; + +statement ok +set datafusion.execution.target_partitions = 4; + +statement ok +RESET datafusion.explain.physical_plan_only; + +statement ok +RESET datafusion.explain.show_statistics; From 0b5fc9a4eb28e0abdb4c19d947a26614cd5574ee Mon Sep 17 00:00:00 2001 From: Shrivas Shankar Date: Thu, 5 Nov 2026 17:43:41 -0600 Subject: [PATCH 2/3] fix: keep the default selectivity for decimal filters without min/max stats Interval analysis estimates a selectivity of 1.0 for a column without bounds. In TPC-H Q18, `HAVING sum(l_quantity) > 300` filters an aggregate output, which has no min/max stats, so #25790 changed its estimate from the default 20% to 100% and the semi join built its hash table on the wrong side. Decimal predicates now use interval analysis only when every decimal column in them has min and max stats. Otherwise they keep the default selectivity, as before #25790. --- datafusion/physical-plan/src/filter.rs | 92 ++++++++++++++- .../test_files/filter_statistics_decimal.slt | 107 ++++++++++++++++++ 2 files changed, 198 insertions(+), 1 deletion(-) diff --git a/datafusion/physical-plan/src/filter.rs b/datafusion/physical-plan/src/filter.rs index 037cf65008e46..d7d2074ad67de 100644 --- a/datafusion/physical-plan/src/filter.rs +++ b/datafusion/physical-plan/src/filter.rs @@ -389,7 +389,13 @@ impl FilterExec { } else { let null_rejecting_columns = collect_null_rejecting_columns(predicate); - if check_statistics_support(predicate, schema) { + if check_statistics_support(predicate, schema) + && decimal_columns_have_bounds( + predicate, + schema, + &input_stats.column_statistics, + ) + { let input_analysis_ctx = AnalysisContext::try_from_statistics( schema, &input_stats.column_statistics, @@ -1249,6 +1255,33 @@ fn compute_fallback_selectivity( selectivity } +/// Returns true if every decimal column in `predicate` has both a min and a +/// max statistic. +/// +/// Interval analysis cannot narrow a column without bounds, so it estimates a +/// selectivity of 1.0 for it. Filters on aggregate outputs, such as +/// `HAVING sum(l_quantity) > 300` in TPC-H Q18, usually have no bounds. These +/// decimal predicates used the default selectivity before interval analysis +/// supported decimals, and join planning depends on that estimate, so they +/// keep using it. +fn decimal_columns_have_bounds( + predicate: &Arc, + schema: &SchemaRef, + column_statistics: &[ColumnStatistics], +) -> bool { + collect_columns(predicate).iter().all(|column| { + let is_decimal = schema + .fields() + .get(column.index()) + .is_some_and(|field| field.data_type().is_decimal()); + !is_decimal + || column_statistics.get(column.index()).is_some_and(|stats| { + stats.min_value.get_value().is_some() + && stats.max_value.get_value().is_some() + }) + }) +} + /// 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( @@ -2350,6 +2383,63 @@ mod tests { Ok(()) } + /// Interval analysis cannot narrow a decimal column without min/max + /// statistics, such as an aggregate output. Predicates on such columns keep + /// the default selectivity, as they did before decimals were analyzed. + #[tokio::test] + async fn test_filter_statistics_decimal_without_bounds() -> Result<()> { + // Table: + // a: Decimal128(15, 2), bounds vary per case + // b: Int32, min=0, max=99 + let decimal = |v: i128| ScalarValue::Decimal128(Some(v), 15, 2); + let schema = Schema::new(vec![ + Field::new("a", DataType::Decimal128(15, 2), false), + Field::new("b", DataType::Int32, false), + ]); + let b_stats = ColumnStatistics { + min_value: Precision::Inexact(ScalarValue::Int32(Some(0))), + max_value: Precision::Inexact(ScalarValue::Int32(Some(99))), + ..Default::default() + }; + + // a > 3.00 + let a_gt = binary(col("a", &schema)?, Operator::Gt, lit(decimal(300)), &schema)?; + // b < 50 AND a > 3.00 + let b_lt_and_a_gt = binary( + binary(col("b", &schema)?, Operator::Lt, lit(50i32), &schema)?, + Operator::And, + Arc::clone(&a_gt), + &schema, + )?; + + for a_stats in [ + ColumnStatistics::new_unknown(), + // Only a lower bound + ColumnStatistics { + min_value: Precision::Inexact(decimal(100)), + ..Default::default() + }, + ] { + for predicate in [&a_gt, &b_lt_and_a_gt] { + let input = Arc::new(StatisticsExec::new( + Statistics { + num_rows: Precision::Inexact(1000), + total_byte_size: Precision::Absent, + column_statistics: vec![a_stats.clone(), b_stats.clone()], + }, + schema.clone(), + )); + let filter: Arc = + Arc::new(FilterExec::try_new(Arc::clone(predicate), input)?); + let statistics = StatisticsContext::new() + .compute(filter.as_ref(), &StatisticsArgs::new())?; + assert_eq!(statistics.num_rows, Precision::Inexact(200), "{predicate}"); + } + } + + Ok(()) + } + #[tokio::test] async fn test_filter_statistics_decimal_mixed_scale_expr() -> Result<()> { // Table: diff --git a/datafusion/sqllogictest/test_files/filter_statistics_decimal.slt b/datafusion/sqllogictest/test_files/filter_statistics_decimal.slt index 09e525076f177..dff2586cd314c 100644 --- a/datafusion/sqllogictest/test_files/filter_statistics_decimal.slt +++ b/datafusion/sqllogictest/test_files/filter_statistics_decimal.slt @@ -81,6 +81,113 @@ SELECT count(*) FROM lineitem WHERE l_discount * 2 > 0.1; statement ok DROP TABLE lineitem; +# TPC-H Q18 plan check (https://github.com/apache/datafusion/issues/27108). +# +# The tables follow the TPC-H shape: 10 orders per customer and 4 lineitems +# per order. The HAVING filter runs on an aggregate output, which has no +# min/max statistics, so the filter keeps the default selectivity. If it were +# estimated to keep every row, the filtered subquery would look larger than the +# joined orders, and the semi join would build its hash table on the wrong side. + +query I +COPY (SELECT value AS c_custkey, 'Customer#' || value AS c_name FROM range(1, 1501)) +TO 'test_files/scratch/filter_statistics_decimal/q18/customer.parquet' +STORED AS PARQUET; +---- +1500 + +query I +COPY ( + SELECT + value AS o_orderkey, + 1 + value % 1500 AS o_custkey, + CAST(1000 + value % 9000 AS DECIMAL(15, 2)) AS o_totalprice, + arrow_cast(9000 + value % 2000, 'Date32') AS o_orderdate + FROM range(1, 15001) +) +TO 'test_files/scratch/filter_statistics_decimal/q18/orders.parquet' +STORED AS PARQUET; +---- +15000 + +query I +COPY ( + SELECT + 1 + value / 4 AS l_orderkey, + CAST(1 + value % 50 AS DECIMAL(15, 2)) AS l_quantity + FROM range(60000) +) +TO 'test_files/scratch/filter_statistics_decimal/q18/lineitem.parquet' +STORED AS PARQUET; +---- +60000 + +statement ok +CREATE EXTERNAL TABLE customer +STORED AS PARQUET +LOCATION 'test_files/scratch/filter_statistics_decimal/q18/customer.parquet'; + +statement ok +CREATE EXTERNAL TABLE orders +STORED AS PARQUET +LOCATION 'test_files/scratch/filter_statistics_decimal/q18/orders.parquet'; + +statement ok +CREATE EXTERNAL TABLE lineitem +STORED AS PARQUET +LOCATION 'test_files/scratch/filter_statistics_decimal/q18/lineitem.parquet'; + +# The filter on sum(l_quantity) keeps the default 20% of the aggregate's rows. +query TT +EXPLAIN SELECT l_orderkey FROM lineitem +GROUP BY l_orderkey HAVING sum(l_quantity) > 300; +---- +physical_plan +01)FilterExec: sum(lineitem.l_quantity)@1 > 300.00, projection=[l_orderkey@0], statistics=[Rows=Inexact(12000), Bytes=Inexact(288000), [(Col[0]: Min=Inexact(Int64(1)) Max=Inexact(Int64(15000)))]] +02)--AggregateExec: mode=Single, gby=[l_orderkey@0 as l_orderkey], aggr=[sum(lineitem.l_quantity)], statistics=[Rows=Inexact(60000), Bytes=Inexact(1440000), [(Col[0]: Min=Exact(Int64(1)) Max=Exact(Int64(15000))),(Col[1]:)]] +03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/filter_statistics_decimal/q18/lineitem.parquet]]}, projection=[l_orderkey, l_quantity], file_type=parquet, statistics=[Rows=Exact(60000), Bytes=Exact(1440000), [(Col[0]: Min=Exact(Int64(1)) Max=Exact(Int64(15000)) Null=Exact(0) ScanBytes=Exact(480000)),(Col[1]: Min=Exact(Decimal128(1.00,15,2)) Max=Exact(Decimal128(50.00,15,2)) Null=Exact(0) ScanBytes=Exact(960000))]] + +statement ok +set datafusion.explain.show_statistics = false; + +# The semi join builds on the filtered subquery (RightSemi), as in DataFusion +# 55, not on the joined customer, orders and lineitem rows (LeftSemi). +query TT +EXPLAIN SELECT + c_name, c_custkey, o_orderkey, o_orderdate, o_totalprice, sum(l_quantity) +FROM customer, orders, lineitem +WHERE o_orderkey IN ( + SELECT l_orderkey FROM lineitem + GROUP BY l_orderkey HAVING sum(l_quantity) > 300 + ) + AND c_custkey = o_custkey + AND o_orderkey = l_orderkey +GROUP BY c_name, c_custkey, o_orderkey, o_orderdate, o_totalprice +ORDER BY o_totalprice DESC, o_orderdate +LIMIT 100; +---- +physical_plan +01)SortExec: TopK(fetch=100), expr=[o_totalprice@4 DESC, o_orderdate@3 ASC NULLS LAST], preserve_partitioning=[false] +02)--AggregateExec: mode=Single, gby=[c_name@1 as c_name, c_custkey@0 as c_custkey, o_orderkey@2 as o_orderkey, o_orderdate@4 as o_orderdate, o_totalprice@3 as o_totalprice], aggr=[sum(lineitem.l_quantity)] +03)----HashJoinExec: mode=CollectLeft, join_type=RightSemi, on=[(l_orderkey@0, o_orderkey@2)] +04)------FilterExec: sum(lineitem.l_quantity)@1 > 300.00, projection=[l_orderkey@0] +05)--------AggregateExec: mode=Single, gby=[l_orderkey@0 as l_orderkey], aggr=[sum(lineitem.l_quantity)] +06)----------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/filter_statistics_decimal/q18/lineitem.parquet]]}, projection=[l_orderkey, l_quantity], file_type=parquet +07)------HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(o_orderkey@2, l_orderkey@0)], projection=[c_custkey@0, c_name@1, o_orderkey@2, o_totalprice@3, o_orderdate@4, l_quantity@6] +08)--------HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(c_custkey@0, o_custkey@1)], projection=[c_custkey@0, c_name@1, o_orderkey@2, o_totalprice@4, o_orderdate@5] +09)----------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/filter_statistics_decimal/q18/customer.parquet]]}, projection=[c_custkey, c_name], output_ordering=[c_custkey@0 ASC NULLS LAST], file_type=parquet +10)----------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/filter_statistics_decimal/q18/orders.parquet]]}, projection=[o_orderkey, o_custkey, o_totalprice, o_orderdate], output_ordering=[o_orderkey@0 ASC NULLS LAST], file_type=parquet, predicate=DynamicFilter [ empty ] AND DynamicFilter [ empty ] AND DynamicFilter [ empty ], dynamic_rg_pruning=eligible +11)--------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/filter_statistics_decimal/q18/lineitem.parquet]]}, projection=[l_orderkey, l_quantity], file_type=parquet, predicate=DynamicFilter [ empty ] AND DynamicFilter [ empty ], dynamic_rg_pruning=eligible + +statement ok +DROP TABLE customer; + +statement ok +DROP TABLE orders; + +statement ok +DROP TABLE lineitem; + statement ok set datafusion.execution.target_partitions = 4; From 68f472c86f03baaffaa06afa8913c7ce7b7a4d9b Mon Sep 17 00:00:00 2001 From: Shrivas Shankar Date: Thu, 5 Nov 2026 18:54:52 -0600 Subject: [PATCH 3/3] fix: keep the old estimate for equality filters on decimal columns Interval analysis estimates an equality from the number of values in the column's range, and for a decimal that counts every step of its scale. In TPC-DS, `ca_gmt_offset = -5` on a DECIMAL(5, 2) column holding whole numbers from -10 to -5 was estimated at 100 rows instead of 18,246, which flipped the semi joins in q33, q56, q60, q61 and q91. Decimal equality filters now keep the estimate they had before #25790. Range filters on decimal columns still use interval analysis. --- datafusion/physical-plan/src/filter.rs | 95 ++++++++++++++++++- .../test_files/filter_statistics_decimal.slt | 31 ++++++ 2 files changed, 125 insertions(+), 1 deletion(-) diff --git a/datafusion/physical-plan/src/filter.rs b/datafusion/physical-plan/src/filter.rs index d7d2074ad67de..52b7ccebbdf87 100644 --- a/datafusion/physical-plan/src/filter.rs +++ b/datafusion/physical-plan/src/filter.rs @@ -42,7 +42,7 @@ use arrow::record_batch::RecordBatch; use datafusion_common::cast::as_boolean_array; use datafusion_common::config::ConfigOptions; use datafusion_common::stats::Precision; -use datafusion_common::tree_node::TreeNodeRecursion; +use datafusion_common::tree_node::{TreeNode, TreeNodeRecursion}; use datafusion_common::{ DataFusionError, Result, ScalarValue, internal_datafusion_err, internal_err, plan_err, project_schema, @@ -395,6 +395,7 @@ impl FilterExec { schema, &input_stats.column_statistics, ) + && !has_decimal_equality(predicate, schema) { let input_analysis_ctx = AnalysisContext::try_from_statistics( schema, @@ -1282,6 +1283,29 @@ fn decimal_columns_have_bounds( }) } +/// Returns true if `predicate` compares a decimal value with `=`. +/// +/// Interval analysis estimates an equality from the number of values in the +/// column's range, and for a decimal that counts every step of its scale. +/// Decimal columns often hold far fewer distinct values. For example, TPC-DS +/// stores whole-hour time zone offsets in `ca_gmt_offset DECIMAL(5, 2)`: +/// `ca_gmt_offset = -5` matches over a third of the rows, but the range +/// [-10.00, -5.00] holds 501 values. These predicates keep the estimate they +/// had before interval analysis supported decimals. +fn has_decimal_equality(predicate: &Arc, schema: &SchemaRef) -> bool { + predicate + .exists(|expr| { + Ok(expr.downcast_ref::().is_some_and(|binary| { + *binary.op() == Operator::Eq + && [binary.left(), binary.right()].iter().any(|side| { + side.data_type(schema) + .is_ok_and(|data_type| data_type.is_decimal()) + }) + })) + }) + .unwrap_or(false) +} + /// 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( @@ -2440,6 +2464,75 @@ mod tests { Ok(()) } + /// Equality on a decimal column keeps the estimate it had before decimals + /// were analyzed: 1 / NDV when the distinct count is known, otherwise the + /// default selectivity. Ranges on the same column still use interval + /// analysis. + #[tokio::test] + async fn test_filter_statistics_decimal_equality() -> Result<()> { + // Table: + // a: Decimal128(5, 2), min=-10.00, max=-5.00 + let decimal = |v: i128| ScalarValue::Decimal128(Some(v), 5, 2); + let schema = + Schema::new(vec![Field::new("a", DataType::Decimal128(5, 2), false)]); + let a_stats = ColumnStatistics { + min_value: Precision::Inexact(decimal(-1000)), + max_value: Precision::Inexact(decimal(-500)), + ..Default::default() + }; + // a = -5.00 + let a_eq = binary( + col("a", &schema)?, + Operator::Eq, + lit(decimal(-500)), + &schema, + )?; + // a >= -6.00 + let a_gt_eq = binary( + col("a", &schema)?, + Operator::GtEq, + lit(decimal(-600)), + &schema, + )?; + + let cases = [ + // The default 20%, not 1 of the 501 values in [-10.00, -5.00] + (a_stats.clone(), Arc::clone(&a_eq), 200), + // 1 of 6 distinct values, rounded up + ( + ColumnStatistics { + distinct_count: Precision::Inexact(6), + ..a_stats.clone() + }, + Arc::clone(&a_eq), + 167, + ), + // 101 of the 501 values in the range, rounded up + (a_stats, a_gt_eq, 202), + ]; + for (a_stats, predicate, expected_rows) in cases { + let input = Arc::new(StatisticsExec::new( + Statistics { + num_rows: Precision::Inexact(1000), + total_byte_size: Precision::Absent, + column_statistics: vec![a_stats], + }, + schema.clone(), + )); + let filter: Arc = + Arc::new(FilterExec::try_new(Arc::clone(&predicate), input)?); + let statistics = StatisticsContext::new() + .compute(filter.as_ref(), &StatisticsArgs::new())?; + assert_eq!( + statistics.num_rows, + Precision::Inexact(expected_rows), + "{predicate}" + ); + } + + Ok(()) + } + #[tokio::test] async fn test_filter_statistics_decimal_mixed_scale_expr() -> Result<()> { // Table: diff --git a/datafusion/sqllogictest/test_files/filter_statistics_decimal.slt b/datafusion/sqllogictest/test_files/filter_statistics_decimal.slt index dff2586cd314c..77b1db2dee1bb 100644 --- a/datafusion/sqllogictest/test_files/filter_statistics_decimal.slt +++ b/datafusion/sqllogictest/test_files/filter_statistics_decimal.slt @@ -81,6 +81,37 @@ SELECT count(*) FROM lineitem WHERE l_discount * 2 > 0.1; statement ok DROP TABLE lineitem; +# Equality on a decimal column keeps the default selectivity. Like TPC-DS's +# ca_gmt_offset, this column holds whole numbers from -10 to -5, so counting +# the 501 values in [-10.00, -5.00] would estimate 2 rows for `= -5`. + +query I +COPY (SELECT CAST(-5 - value % 6 AS DECIMAL(5, 2)) AS ca_gmt_offset FROM range(600)) +TO 'test_files/scratch/filter_statistics_decimal/customer_address.parquet' +STORED AS PARQUET; +---- +600 + +statement ok +CREATE EXTERNAL TABLE customer_address +STORED AS PARQUET +LOCATION 'test_files/scratch/filter_statistics_decimal/customer_address.parquet'; + +query TT +EXPLAIN SELECT ca_gmt_offset FROM customer_address WHERE ca_gmt_offset = -5; +---- +physical_plan +01)FilterExec: ca_gmt_offset@0 = -5.00, statistics=[Rows=Inexact(120), Bytes=Inexact(1920), [(Col[0]: Min=Inexact(Decimal128(-10.00,5,2)) Max=Inexact(Decimal128(-5.00,5,2)) Null=Exact(0) Distinct=Inexact(1) ScanBytes=Inexact(1920))]] +02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/filter_statistics_decimal/customer_address.parquet]]}, projection=[ca_gmt_offset], file_type=parquet, predicate=ca_gmt_offset@0 = -5.00, pruning_predicate=ca_gmt_offset_null_count@2 != row_count@3 AND ca_gmt_offset_min@0 <= -5.00 AND -5.00 <= ca_gmt_offset_max@1, required_guarantees=[ca_gmt_offset in (-5.00)], statistics=[Rows=Inexact(600), Bytes=Inexact(9600), [(Col[0]: Min=Inexact(Decimal128(-10.00,5,2)) Max=Inexact(Decimal128(-5.00,5,2)) Null=Inexact(0) ScanBytes=Inexact(9600))]] + +query I +SELECT count(*) FROM customer_address WHERE ca_gmt_offset = -5; +---- +100 + +statement ok +DROP TABLE customer_address; + # TPC-H Q18 plan check (https://github.com/apache/datafusion/issues/27108). # # The tables follow the TPC-H shape: 10 orders per customer and 4 lineitems