From 0900fc929f571eb3c75fa477e8b02bcffff4e623 Mon Sep 17 00:00:00 2001 From: Shrivas Shankar Date: Sat, 26 Sep 2026 23:02:05 -0500 Subject: [PATCH] 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 9f1291353dc29..fcd9d0ea4e231 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 792687fe6acaa..6ec48e4a79d1c 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 1090660a6b5e6..3041fb8d01423 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 f71d67d04e39c..4db1db7ac7e1d 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 =