diff --git a/datafusion/expr-common/src/interval_arithmetic.rs b/datafusion/expr-common/src/interval_arithmetic.rs index 9f1291353dc29..82577fb72a26c 100644 --- a/datafusion/expr-common/src/interval_arithmetic.rs +++ b/datafusion/expr-common/src/interval_arithmetic.rs @@ -422,9 +422,33 @@ impl Interval { data_type: &DataType, cast_options: &CastOptions, ) -> Result { + let source_type = self.data_type(); + // An unbounded integer endpoint still has a finite limit imposed by its + // type. Preserve that limit when widening so subsequent arithmetic can + // prove that it does not overflow the destination type. + use DataType::{Int8, Int16, Int32, Int64, UInt8, UInt16, UInt32, UInt64}; + let widening_integer_cast = matches!( + (&source_type, data_type), + (Int8, Int16 | Int32 | Int64) + | (Int16, Int32 | Int64) + | (Int32, Int64) + | (UInt8, UInt16 | UInt32 | UInt64 | Int16 | Int32 | Int64) + | (UInt16, UInt32 | UInt64 | Int32 | Int64) + | (UInt32, UInt64 | Int64) + ); + let lower = if widening_integer_cast && self.lower.is_null() { + ScalarValue::min(&source_type).expect("integer types have a minimum") + } else { + self.lower.clone() + }; + let upper = if widening_integer_cast && self.upper.is_null() { + ScalarValue::max(&source_type).expect("integer types have a maximum") + } else { + self.upper.clone() + }; Self::try_new( - cast_scalar_value(&self.lower, data_type, cast_options)?, - cast_scalar_value(&self.upper, data_type, cast_options)?, + cast_scalar_value(&lower, data_type, cast_options)?, + cast_scalar_value(&upper, data_type, cast_options)?, ) } @@ -2358,6 +2382,119 @@ mod tests { Ok(()) } + #[test] + fn test_widening_integer_cast_bounds() { + use DataType::{Int8, Int16, Int32, Int64, UInt8, UInt16, UInt32, UInt64}; + use arrow::compute::CastOptions; + + let types = [ + (Int8, i8::MIN as i128, i8::MAX as i128), + (Int16, i16::MIN as i128, i16::MAX as i128), + (Int32, i32::MIN as i128, i32::MAX as i128), + (Int64, i64::MIN as i128, i64::MAX as i128), + (UInt8, 0, u8::MAX as i128), + (UInt16, 0, u16::MAX as i128), + (UInt32, 0, u32::MAX as i128), + (UInt64, 0, u64::MAX as i128), + ]; + for (source, min, max) in &types { + for (target, target_min, target_max) in &types { + if source == target || min < target_min || max > target_max { + continue; + } + // Every source that can widen fits in Int64. + let lower = ScalarValue::Int64(Some(*min as i64)) + .cast_to(source) + .unwrap(); + let upper = ScalarValue::Int64(Some(*max as i64)) + .cast_to(source) + .unwrap(); + let zero = ScalarValue::new_zero(source).unwrap(); + let unbounded = ScalarValue::try_from(source).unwrap(); + for (lo, hi, expected_lo, expected_hi) in [ + ( + unbounded.clone(), + unbounded.clone(), + lower.clone(), + upper.clone(), + ), + (zero.clone(), unbounded.clone(), zero.clone(), upper.clone()), + (unbounded.clone(), zero.clone(), lower.clone(), zero.clone()), + (zero.clone(), zero.clone(), zero.clone(), zero.clone()), + ] { + let actual = Interval::try_new(lo, hi) + .unwrap() + .cast_to(target, &CastOptions::default()) + .unwrap(); + let expected = Interval::try_new( + expected_lo.cast_to(target).unwrap(), + expected_hi.cast_to(target).unwrap(), + ) + .unwrap(); + assert_eq!(actual, expected, "{source:?} -> {target:?}"); + } + } + } + // Same-type and narrowing casts retain their existing unbounded behavior. + let unbounded = Interval::make_unbounded(&Int64).unwrap(); + for target in [Int64, Int32] { + assert_eq!( + unbounded.cast_to(&target, &CastOptions::default()).unwrap(), + Interval::make_unbounded(&target).unwrap() + ); + } + } + + #[test] + fn test_widening_integer_coercion_arithmetic() -> Result<()> { + let unsigned = Interval::make_unbounded(&DataType::UInt8)?; + let two = Interval::make(Some(2_i16), Some(2_i16))?; + + // Mixed arithmetic must widen UInt8 to [0, 255] in Int16 before + // computing the result, including when it is the right operand. + let product = Interval::make(Some(0_i16), Some(510_i16))?; + assert_eq!(unsigned.mul(&two)?, product); + assert_eq!(two.mul(&unsigned)?, product); + assert_eq!( + unsigned.div(&two)?, + Interval::make(Some(0_i16), Some(127_i16))? + ); + Ok(()) + } + + #[test] + fn test_widening_integer_coercion_comparison() -> Result<()> { + let unsigned = Interval::make_unbounded(&DataType::UInt8)?; + let signed = Interval::make(Some(-1_i16), Some(256_i16))?; + let widened = Interval::make(Some(0_i16), Some(255_i16))?; + + // Comparison coercion must preserve both limits of the UInt8 domain. + assert_eq!(unsigned.intersect(&signed)?, Some(widened.clone())); + assert_eq!(signed.intersect(&unsigned)?, Some(widened.clone())); + assert_eq!(widened.contains(&unsigned)?, Interval::TRUE); + for value in [-1_i16, 256_i16] { + let outside = Interval::make(Some(value), Some(value))?; + assert_eq!(unsigned.contains(&outside)?, Interval::FALSE); + } + Ok(()) + } + + #[test] + fn test_integer_interval_cast_overflow() { + use arrow::compute::CastOptions; + + let options = CastOptions { + safe: false, + ..Default::default() + }; + // Check failures at either endpoint, including an upper endpoint that + // overflows after the lower endpoint has been successfully cast. + for (lower, upper) in [(-129i16, 0i16), (0, 128)] { + let interval = Interval::make(Some(lower), Some(upper)).unwrap(); + assert!(interval.cast_to(&DataType::Int8, &options).is_err()); + } + } + #[test] fn test_new_interval() -> Result<()> { use ScalarValue::*; diff --git a/datafusion/physical-expr/src/expressions/binary.rs b/datafusion/physical-expr/src/expressions/binary.rs index dfb1d136d0ff0..1117d3fdd5fc1 100644 --- a/datafusion/physical-expr/src/expressions/binary.rs +++ b/datafusion/physical-expr/src/expressions/binary.rs @@ -122,6 +122,23 @@ impl BinaryExpr { &self.op } + /// Mathematical intervals do not cover values produced by wrapping arithmetic. + fn integer_arithmetic_may_wrap( + &self, + left: &Interval, + right: &Interval, + result: &Interval, + ) -> bool { + !self.fail_on_overflow + && result.data_type().is_integer() + && matches!( + self.op, + Operator::Plus | Operator::Minus | Operator::Multiply + ) + && (result.is_unbounded() + || unsigned_subtraction_may_underflow(self.op, left, right, result)) + } + /// Wrapping on overflow breaks monotonicity (e.g. the sum of two /// ascending `UInt8` columns can wrap back to small values), so the /// derived ordering is kept only when overflow is impossible. `time ± @@ -700,7 +717,12 @@ impl PhysicalExpr for BinaryExpr { let left_interval = children[0]; let right_interval = children[1]; // Calculate current node's interval: - apply_operator(&self.op, left_interval, right_interval) + let result = apply_operator(&self.op, left_interval, right_interval)?; + if self.integer_arithmetic_may_wrap(left_interval, right_interval, &result) { + Interval::make_unbounded(&result.data_type()) + } else { + Ok(result) + } } fn propagate_constraints( @@ -712,6 +734,36 @@ impl PhysicalExpr for BinaryExpr { let left_interval = children[0]; let right_interval = children[1]; + if left_interval.data_type().is_integer() + && right_interval.data_type().is_integer() + && matches!( + self.op, + Operator::Plus | Operator::Minus | Operator::Multiply | Operator::Divide + ) + { + // Integer division truncates: a / 2 = 1 permits both 2 and 3. + // Wrapping arithmetic is likewise not invertible over mathematical + // intervals. Keep the input domains rather than exclude valid rows. + let contains_zero = |range: &Interval| -> Result { + range.contains_value(ScalarValue::new_zero(&range.data_type())?) + }; + // If an operand can be zero, a zero product does not constrain + // the other operand. Dividing the parent interval loses that case. + let zero_product = self.op == Operator::Multiply + && contains_zero(interval)? + && (contains_zero(left_interval)? || contains_zero(right_interval)?); + if self.op == Operator::Divide + || zero_product + || self.integer_arithmetic_may_wrap( + left_interval, + right_interval, + &apply_operator(&self.op, left_interval, right_interval)?, + ) + { + return Ok(Some(vec![])); + } + } + if self.op.eq(&Operator::And) { if interval.eq(&Interval::TRUE) { // A certainly true logical conjunction can only derive from possibly @@ -832,7 +884,7 @@ impl PhysicalExpr for BinaryExpr { let (r_order, r_range) = (children[1].sort_properties, &children[1].range); match self.op() { Operator::Plus => { - let range = l_range.add(r_range)?; + let range = self.evaluate_bounds(&[l_range, r_range])?; Ok(ExprProperties { sort_properties: self.arithmetic_sort_properties( l_order.add(&r_order), @@ -846,7 +898,7 @@ impl PhysicalExpr for BinaryExpr { }) } Operator::Minus => { - let range = l_range.sub(r_range)?; + let range = self.evaluate_bounds(&[l_range, r_range])?; Ok(ExprProperties { sort_properties: self.arithmetic_sort_properties( l_order.sub(&r_order), @@ -6353,6 +6405,246 @@ mod tests { } } + #[test] + fn test_integer_interval_propagation_covers_runtime_values() { + // Enumerate small domains and both ends of Int8, including zero divisors, + // truncation, signed overflow and checked arithmetic. Every successful + // runtime evaluation must remain possible after interval propagation. + let batch = RecordBatch::new_empty(Arc::new(Schema::empty())); + let domains = [ + (-3i8, 3i8), + (0, 1), + (1, 3), + (-3, -1), + (-128, -127), + (126, 127), + ]; + for checked in [false, true] { + for op in [ + Operator::Plus, + Operator::Minus, + Operator::Multiply, + Operator::Divide, + ] { + let expr = BinaryExpr::new(lit(0i8), op, lit(0i8)) + .with_fail_on_overflow(checked); + for (lo, hi) in domains { + for (rlo, rhi) in + domains.into_iter().chain([(-1, -1), (0, 0), (2, 2)]) + { + let left = Interval::make(Some(lo), Some(hi)).unwrap(); + let right = Interval::make(Some(rlo), Some(rhi)).unwrap(); + let bounds = expr.evaluate_bounds(&[&left, &right]).unwrap(); + if matches!(op, Operator::Plus | Operator::Minus) { + let children = [left.clone(), right.clone()].map(|range| { + ExprProperties { + range, + ..ExprProperties::new_unknown() + } + }); + assert_eq!( + expr.get_properties(&children).unwrap().range, + bounds + ); + } + for a in lo..=hi { + for b in rlo..=rhi { + // Use the execution kernel as the oracle rather than + // duplicating its checked and wrapping arithmetic. + let runtime = BinaryExpr::new(lit(a), op, lit(b)) + .with_fail_on_overflow(checked); + let result = match runtime.evaluate(&batch) { + Ok(value) => value.into_array(1).unwrap(), + Err(error) => { + assert!( + checked || op == Operator::Divide, + "wrapping {a} {op} {b} failed: {error}" + ); + continue; + } + }; + let result = result.as_primitive::().value(0); + let result = + Interval::make(Some(result), Some(result)).unwrap(); + assert_eq!( + bounds.contains(&result).unwrap(), + Interval::TRUE, + "forward {a} {op} {b}, checked={checked}, bounds={bounds:?}" + ); + let propagated = expr + .propagate_constraints(&result, &[&left, &right]) + .unwrap(); + let propagated = propagated + .expect("successful runtime result must be feasible"); + if !propagated.is_empty() { + assert_eq!( + propagated[0] + .contains( + Interval::make(Some(a), Some(a)).unwrap() + ) + .unwrap(), + Interval::TRUE, + "left input excluded for {a} {op} {b}, checked={checked}: {propagated:?}" + ); + assert_eq!( + propagated[1] + .contains( + Interval::make(Some(b), Some(b)).unwrap() + ) + .unwrap(), + Interval::TRUE, + "right input excluded for {a} {op} {b}, checked={checked}: {propagated:?}" + ); + } + } + } + } + } + } + } + } + + #[test] + fn test_integer_interval_error_propagation() { + // Compare error messages without backtraces, which differ by call path. + let integer = Interval::make(Some(1i32), Some(2i32)).unwrap(); + let boolean = Interval::TRUE; + for op in [Operator::Plus, Operator::Minus] { + let expr = BinaryExpr::new(lit(1i32), op, lit(true)); + let expected = apply_operator(&op, &integer, &boolean) + .unwrap_err() + .strip_backtrace(); + assert_eq!( + expr.evaluate_bounds(&[&integer, &boolean]) + .unwrap_err() + .strip_backtrace(), + expected + ); + let children = + [integer.clone(), boolean.clone()].map(|range| ExprProperties { + range, + ..ExprProperties::new_unknown() + }); + assert_eq!( + expr.get_properties(&children) + .unwrap_err() + .strip_backtrace(), + expected + ); + } + + // A nonnumeric parent cannot describe an integer product. Preserve + // the error from constructing zero for its unsupported type. + let parent = Interval::make_unbounded(&DataType::Utf8).unwrap(); + let expr = BinaryExpr::new(lit(1i32), Operator::Multiply, lit(2i32)); + assert_eq!( + expr.propagate_constraints(&parent, &[&integer, &integer]) + .unwrap_err() + .strip_backtrace(), + ScalarValue::new_zero(&DataType::Utf8) + .unwrap_err() + .strip_backtrace() + ); + } + + #[test] + fn test_unsigned_subtraction_interval_underflow() { + let expr = BinaryExpr::new(lit(0u8), Operator::Minus, lit(1u8)); + let left = Interval::make(Some(0u8), Some(2u8)).unwrap(); + let right = Interval::make(Some(1u8), Some(1u8)).unwrap(); + let wrapped = Interval::make(Some(255u8), Some(255u8)).unwrap(); + assert_eq!( + expr.evaluate_bounds(&[&left, &right]) + .unwrap() + .contains(&wrapped) + .unwrap(), + Interval::TRUE + ); + assert_eq!( + expr.propagate_constraints(&wrapped, &[&left, &right]) + .unwrap(), + Some(vec![]) + ); + } + + #[test] + fn test_nested_wrapping_arithmetic_properties() { + let difference = Arc::new(BinaryExpr::new(lit(0u8), Operator::Minus, lit(1u8))); + let singleton = |value: u8| ExprProperties { + sort_properties: SortProperties::Singleton, + range: Interval::make(Some(value), Some(value)).unwrap(), + ..ExprProperties::new_unknown() + }; + let difference_props = difference + .get_properties(&[singleton(0), singleton(1)]) + .unwrap(); + assert_eq!(difference_props.sort_properties, SortProperties::Singleton); + assert!( + difference_props + .range + .contains_value(ScalarValue::UInt8(Some(255))) + .unwrap() + ); + + // The inner constant wraps to 255. An incorrect range of [0, 0] + // would let the outer addition claim to preserve ascending order. + let expr = + BinaryExpr::new(Arc::new(Column::new("a", 0)), Operator::Plus, difference); + let properties = expr + .get_properties(&[ + ExprProperties { + sort_properties: SortProperties::Ordered(SortOptions::default()), + range: Interval::make(Some(0u8), Some(1u8)).unwrap(), + ..ExprProperties::new_unknown() + }, + difference_props, + ]) + .unwrap(); + let batch = RecordBatch::try_new( + Arc::new(Schema::new(vec![Field::new("a", DataType::UInt8, false)])), + vec![Arc::new(UInt8Array::from(vec![0, 1]))], + ) + .unwrap(); + let actual = expr.evaluate(&batch).unwrap().into_array(2).unwrap(); + assert_eq!(actual.as_ref(), &UInt8Array::from(vec![255, 0])); + assert_eq!(properties.sort_properties, SortProperties::Unordered); + for value in [0u8, 255] { + assert!( + properties + .range + .contains_value(ScalarValue::UInt8(Some(value))) + .unwrap() + ); + } + } + + #[test] + fn test_integer_comparison_still_propagates() { + // Integer comparisons must bypass the arithmetic overflow guard and + // still narrow their inputs: a = 5 restricts a in [0, 10] to [5, 5]. + let expr = BinaryExpr::new(lit(0i32), Operator::Eq, lit(5i32)); + let left = Interval::make(Some(0i32), Some(10i32)).unwrap(); + let right = Interval::make(Some(5i32), Some(5i32)).unwrap(); + assert_eq!( + expr.propagate_constraints(&Interval::TRUE, &[&left, &right]) + .unwrap(), + Some(vec![right.clone(), right]) + ); + } + + #[test] + fn test_safe_integer_multiplication_still_propagates() { + let expr = BinaryExpr::new(lit(0i32), Operator::Multiply, lit(2i32)); + let left = Interval::make(Some(0i32), Some(10i32)).unwrap(); + let right = Interval::make(Some(2i32), Some(2i32)).unwrap(); + let parent = Interval::make(Some(4i32), Some(4i32)).unwrap(); + assert_eq!( + expr.propagate_constraints(&parent, &[&left, &right]) + .unwrap(), + Some(vec![Interval::make(Some(2i32), Some(2i32)).unwrap(), right]) + ); + } + #[test] fn test_evaluate_bounds_int32() { let schema = Schema::new(vec![ diff --git a/datafusion/physical-plan/src/filter.rs b/datafusion/physical-plan/src/filter.rs index ae87168ec7598..7377ec893621c 100644 --- a/datafusion/physical-plan/src/filter.rs +++ b/datafusion/physical-plan/src/filter.rs @@ -2344,6 +2344,18 @@ mod tests { )), )), )); + // i32::MIN also satisfies this predicate because subtracting 5 wraps. + // A mathematical lower bound of 5 would exclude a valid input value. + let batch = RecordBatch::try_new( + input.schema(), + vec![Arc::new(arrow::array::Int32Array::from(vec![i32::MIN]))], + ) + .unwrap(); + let result = predicate.evaluate(&batch).unwrap().into_array(1).unwrap(); + assert_eq!( + result.as_ref(), + &arrow::array::BooleanArray::from(vec![true]) + ); let filter: Arc = Arc::new(FilterExec::try_new(predicate, input)?); let filter_statistics = @@ -2356,7 +2368,7 @@ mod tests { // `a <= 10` rejects nulls, so `a` has no surviving nulls even // though the input statistics are entirely unknown. null_count: Precision::Exact(0), - min_value: Precision::Inexact(ScalarValue::Int32(Some(5))), + min_value: Precision::Absent, max_value: Precision::Inexact(ScalarValue::Int32(Some(10))), sum_value: Precision::Absent, distinct_count: Precision::Absent, diff --git a/datafusion/sqllogictest/test_files/filter_without_sort_exec.slt b/datafusion/sqllogictest/test_files/filter_without_sort_exec.slt index 87e994f243183..aa17b82c67b9c 100644 --- a/datafusion/sqllogictest/test_files/filter_without_sort_exec.slt +++ b/datafusion/sqllogictest/test_files/filter_without_sort_exec.slt @@ -189,3 +189,98 @@ physical_plan 02)--FilterExec: b@0 > 1 03)----RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1, maintains_sort_order=true 04)------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/filter_without_sort_exec/cast_ordering.parquet]]}, projection=[b], output_ordering=[b@0 ASC NULLS LAST], file_type=parquet, predicate=b@0 > 1, pruning_predicate=b_null_count@1 != row_count@2 AND b_max@0 > 1, required_guarantees=[] + +statement ok +SET datafusion.execution.target_partitions = 1; + +# Wrapping multiplication has more than one preimage of 2. The filter +# must not mark a as constant and remove either requested sort. +statement ok +CREATE TABLE wrap_test (a INT) AS VALUES (-2147483647), (1); + +query II +SELECT a, a * 2::INT FROM wrap_test ORDER BY a; +---- +-2147483647 2 +1 2 + +query I +SELECT a FROM wrap_test WHERE a * 2::INT = 2::INT ORDER BY a DESC; +---- +1 +-2147483647 + +query I +SELECT a - 2::INT AS x FROM wrap_test WHERE a * 2::INT = 2::INT ORDER BY x; +---- +-1 +2147483647 + +query II +SELECT MIN(a), MAX(a) FROM wrap_test WHERE a * 2::INT = 2::INT; +---- +-2147483647 1 + +# Integer division truncates: both 2 and 3 satisfy a / 2 = 1. +statement ok +CREATE TABLE division_test (a INT) AS VALUES (2), (3); + +query II +SELECT a, a / 2::INT FROM division_test ORDER BY a; +---- +2 1 +3 1 + +query I +SELECT a FROM division_test WHERE a / 2::INT = 1::INT ORDER BY a DESC; +---- +3 +2 + +# Widening Int32 to Int64 preserves the source type's upper bound. +statement ok +SET datafusion.explain.physical_plan_only = true; + +statement ok +CREATE TABLE widening_input (a INT) AS +VALUES (2147483647), (1), (2147483646), (2147483647); + +# The predicate implies a = Int32::MAX, so ORDER BY a needs no sort. +query TT +EXPLAIN SELECT a FROM widening_input +WHERE CAST(a AS BIGINT) + 1 > 2147483647 ORDER BY a; +---- +physical_plan +01)FilterExec: CAST(a@0 AS Int64) + 1 > 2147483647 +02)--DataSourceExec: partitions=1, partition_sizes=[1] + +query I +SELECT a FROM widening_input +WHERE CAST(a AS BIGINT) + 1 > 2147483647 ORDER BY a; +---- +2147483647 +2147483647 + +# Two distinct values can pass this predicate; the sort must remain. +query TT +EXPLAIN SELECT a FROM widening_input +WHERE CAST(a AS BIGINT) + 1 > 2147483646 ORDER BY a; +---- +physical_plan +01)SortExec: expr=[a@0 ASC NULLS LAST], preserve_partitioning=[false] +02)--FilterExec: CAST(a@0 AS Int64) + 1 > 2147483646 +03)----DataSourceExec: partitions=1, partition_sizes=[1] + +query I +SELECT a FROM widening_input +WHERE CAST(a AS BIGINT) + 1 > 2147483646 ORDER BY a; +---- +2147483646 +2147483647 +2147483647 + +statement ok +SET datafusion.execution.target_partitions = 4; + +statement ok +RESET datafusion.explain.physical_plan_only;