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