diff --git a/datafusion/expr-common/src/interval_arithmetic.rs b/datafusion/expr-common/src/interval_arithmetic.rs index 797c48cbe0dd9..2df95cf673aaa 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), } }; @@ -1208,6 +1257,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 @@ -1216,6 +1267,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 @@ -1228,14 +1281,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 { @@ -1246,6 +1315,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 { @@ -1300,6 +1375,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))) } @@ -2281,7 +2373,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}; @@ -2594,6 +2689,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::*; @@ -4407,6 +4620,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 c99f9a68673cd..0b228db8c98a4 100644 --- a/datafusion/physical-expr/src/expressions/cast.rs +++ b/datafusion/physical-expr/src/expressions/cast.rs @@ -712,6 +712,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")] @@ -933,6 +955,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); @@ -953,6 +980,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 e85e7a5d26994..7554c60b03f94 100644 --- a/datafusion/physical-expr/src/intervals/utils.rs +++ b/datafusion/physical-expr/src/intervals/utils.rs @@ -36,19 +36,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 } @@ -57,9 +95,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 } @@ -197,3 +236,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 2500822c5db66..c3dcb91635c9b 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, 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, @@ -386,7 +386,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, @@ -2050,6 +2050,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: @@ -2693,9 +2886,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), @@ -2704,11 +2897,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..dc4b5cd915459 --- /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;