From 0900fc929f571eb3c75fa477e8b02bcffff4e623 Mon Sep 17 00:00:00 2001 From: Shrivas Shankar Date: Sat, 26 Sep 2026 23:02:05 -0500 Subject: [PATCH 1/2] feat: estimate Decimal filter selectivity with interval analysis Decimal columns were excluded from interval analysis, so any FilterExec predicate touching a decimal column fell back to the default selectivity. - Add `check_statistics_support`, which also accepts Decimal32/64/128/256, and use it for FilterExec statistics. `check_support` and `is_datatype_supported` are unchanged, so symmetric hash join pruning keeps its current behavior. - Step decimals by one unit in the last place in `next_value`/`prev_value` so strict comparisons tighten bounds, saturating to unbounded at the extremes of the decimal's precision. - Support Decimal32/64 in `get_extreme_value!` for overflow handling. - Allow reverse constraint propagation through lossless decimal widening casts, so predicates whose literal has a different scale than the column (e.g. `CAST(a AS Decimal128(30, 15)) >= 0.055`) still narrow the column. --- .../expr-common/src/interval_arithmetic.rs | 243 +++++++++++++++++- .../physical-expr/src/expressions/cast.rs | 65 +++++ .../physical-expr/src/intervals/utils.rs | 70 ++++- datafusion/physical-plan/src/filter.rs | 132 +++++++++- 4 files changed, 491 insertions(+), 19 deletions(-) diff --git a/datafusion/expr-common/src/interval_arithmetic.rs b/datafusion/expr-common/src/interval_arithmetic.rs index 9f1291353dc2..fcd9d0ea4e23 100644 --- a/datafusion/expr-common/src/interval_arithmetic.rs +++ b/datafusion/expr-common/src/interval_arithmetic.rs @@ -27,8 +27,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::{ @@ -37,7 +39,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)), @@ -84,6 +93,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, @@ -100,7 +119,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), @@ -147,6 +174,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), } }; @@ -1213,6 +1262,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 @@ -1221,6 +1272,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 @@ -1233,14 +1286,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 { @@ -1251,6 +1320,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 { @@ -1305,6 +1380,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))) } @@ -2267,7 +2359,10 @@ mod tests { }; use crate::interval_arithmetic::NullableInterval; - 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}; @@ -2358,6 +2453,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::*; @@ -4034,6 +4247,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 792687fe6aca..6ec48e4a79d1 100644 --- a/datafusion/physical-expr/src/expressions/cast.rs +++ b/datafusion/physical-expr/src/expressions/cast.rs @@ -477,6 +477,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")] @@ -675,6 +697,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); @@ -695,6 +722,44 @@ mod tests { Ok(()) } + #[test] + fn test_decimal_widening_constraint_boundaries() -> Result<()> { + // Every Decimal128(3, 2) value near zero, widened to Decimal128(5, 3): + let values: Vec = (-20..=20).collect(); + let schema = + Arc::new(Schema::new(vec![Field::new("x", Decimal128(3, 2), false)])); + let expr = CastExpr::new(col("x", &schema)?, Decimal128(5, 3), None); + let input = Interval::make_unbounded(&Decimal128(3, 2))?; + // Output bounds at every Decimal128(5, 3) value in range, including ones + // between two inputs, so casting back rounds in both directions: + let bounds: Vec = (-210..=210).collect(); + for (i, lower) in bounds.iter().enumerate() { + for upper in &bounds[i..] { + let output = Interval::try_new( + ScalarValue::Decimal128(Some(*lower), 5, 3), + ScalarValue::Decimal128(Some(*upper), 5, 3), + )?; + let propagated = expr.propagate_constraints(&output, &[&input])?.unwrap(); + assert_eq!(propagated.len(), 1); + for value in &values { + let widened = ScalarValue::Decimal128(Some(value * 10), 5, 3); + if output.contains_value(widened)? { + assert!( + propagated[0].contains_value(ScalarValue::Decimal128( + Some(*value), + 3, + 2 + ))?, + "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 1090660a6b5e..3041fb8d0142 100644 --- a/datafusion/physical-expr/src/intervals/utils.rs +++ b/datafusion/physical-expr/src/intervals/utils.rs @@ -36,26 +36,54 @@ 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::() { 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 } } else if let Some(cast) = expr.downcast_ref::() { - check_support(cast.expr(), schema) + 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 } @@ -193,3 +221,33 @@ 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)); + } + } +} diff --git a/datafusion/physical-plan/src/filter.rs b/datafusion/physical-plan/src/filter.rs index f71d67d04e39..4db1db7ac7e1 100644 --- a/datafusion/physical-plan/src/filter.rs +++ b/datafusion/physical-plan/src/filter.rs @@ -66,7 +66,7 @@ use datafusion_physical_expr::equivalence::ProjectionMapping; use datafusion_physical_expr::expressions::{ BinaryExpr, Column, InListExpr, IsNotNullExpr, 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, @@ -395,7 +395,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, @@ -1817,6 +1817,124 @@ 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(()) + } + + #[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: @@ -2460,9 +2578,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), @@ -2471,11 +2589,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 = From b4b5be30ecbb7a6ac6455c61a1ef8dcd734d4803 Mon Sep 17 00:00:00 2001 From: Shrivas Shankar Date: Wed, 21 Oct 2026 10:21:49 -0500 Subject: [PATCH 2/2] fix: exclude decimal arithmetic from filter statistics estimation Decimal arithmetic changes the precision and scale of its result, and propagating constraints back to its operands did not restore their types. This could make `FilterExec::try_new` fail (e.g. `a * 2.00 > 5.0000`) or publish column bounds with the wrong decimal type (e.g. `a - 2.00 > 5.00`). `check_statistics_support` now only accepts decimals outside arithmetic; operands of `+ - * / %` must pass `is_datatype_supported`, so decimal arithmetic predicates fall back to the default selectivity. Also adds: - regression tests for decimal multiplication and subtraction predicates - `filter_statistics_decimal.slt`, covering Q6-style decimal ranges with fractional literals through SQL coercion, EXPLAIN estimates and results - decimal widening boundary tests for cross-width casts, negative scales, and bounds near the precision limits --- .../physical-expr/src/expressions/cast.rs | 90 ++++++++++----- .../physical-expr/src/intervals/utils.rs | 41 +++++++ datafusion/physical-plan/src/filter.rs | 75 +++++++++++++ .../test_files/filter_statistics_decimal.slt | 104 ++++++++++++++++++ 4 files changed, 281 insertions(+), 29 deletions(-) create mode 100644 datafusion/sqllogictest/test_files/filter_statistics_decimal.slt diff --git a/datafusion/physical-expr/src/expressions/cast.rs b/datafusion/physical-expr/src/expressions/cast.rs index caee4b63a57d..0b228db8c98a 100644 --- a/datafusion/physical-expr/src/expressions/cast.rs +++ b/datafusion/physical-expr/src/expressions/cast.rs @@ -982,35 +982,67 @@ mod tests { #[test] fn test_decimal_widening_constraint_boundaries() -> Result<()> { - // Every Decimal128(3, 2) value near zero, widened to Decimal128(5, 3): - let values: Vec = (-20..=20).collect(); - let schema = - Arc::new(Schema::new(vec![Field::new("x", Decimal128(3, 2), false)])); - let expr = CastExpr::new(col("x", &schema)?, Decimal128(5, 3), None); - let input = Interval::make_unbounded(&Decimal128(3, 2))?; - // Output bounds at every Decimal128(5, 3) value in range, including ones - // between two inputs, so casting back rounds in both directions: - let bounds: Vec = (-210..=210).collect(); - for (i, lower) in bounds.iter().enumerate() { - for upper in &bounds[i..] { - let output = Interval::try_new( - ScalarValue::Decimal128(Some(*lower), 5, 3), - ScalarValue::Decimal128(Some(*upper), 5, 3), - )?; - let propagated = expr.propagate_constraints(&output, &[&input])?.unwrap(); - assert_eq!(propagated.len(), 1); - for value in &values { - let widened = ScalarValue::Decimal128(Some(value * 10), 5, 3); - if output.contains_value(widened)? { - assert!( - propagated[0].contains_value(ScalarValue::Decimal128( - Some(*value), - 3, - 2 - ))?, - "input={value}, output={output}, propagated={:?}", - propagated[0] - ); + 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] + ); + } } } } diff --git a/datafusion/physical-expr/src/intervals/utils.rs b/datafusion/physical-expr/src/intervals/utils.rs index 1f2d3672ce18..7554c60b03f9 100644 --- a/datafusion/physical-expr/src/intervals/utils.rs +++ b/datafusion/physical-expr/src/intervals/utils.rs @@ -65,6 +65,16 @@ fn check_support_with( 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_with(binary_expr.left(), schema, is_type_supported) && check_support_with(binary_expr.right(), schema, is_type_supported) @@ -253,6 +263,37 @@ mod tests { 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 54fa8374f465..c3dcb91635c9 100644 --- a/datafusion/physical-plan/src/filter.rs +++ b/datafusion/physical-plan/src/filter.rs @@ -2100,6 +2100,81 @@ mod tests { 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: 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 000000000000..dc4b5cd91545 --- /dev/null +++ b/datafusion/sqllogictest/test_files/filter_statistics_decimal.slt @@ -0,0 +1,104 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +# 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;