diff --git a/datafusion/core/tests/sql/unparser.rs b/datafusion/core/tests/sql/unparser.rs index 3982c60dbc7d9..738cf0bd54943 100644 --- a/datafusion/core/tests/sql/unparser.rs +++ b/datafusion/core/tests/sql/unparser.rs @@ -56,6 +56,34 @@ use datafusion_sql::unparser::dialect::{DefaultDialect, DuckDBDialect}; use itertools::Itertools; use recursive::{set_minimum_stack_size, set_stack_allocation_size}; +#[tokio::test] +async fn optimized_unparse_parenthesizes_nested_negation() -> Result<()> { + let ctx = SessionContext::new(); + ctx.register_table( + "t", + Arc::new(EmptyTable::new(Arc::new(Schema::new(vec![Field::new( + "ts", + DataType::Timestamp(arrow::datatypes::TimeUnit::Nanosecond, None), + true, + )])))), + )?; + let plan = ctx + .sql("SELECT -(-ts) AS x FROM t") + .await? + .into_optimized_plan()?; + + for pretty in [false, true] { + let sql = Unparser::default() + .with_pretty(pretty) + .plan_to_sql(&plan)? + .to_string(); + assert_eq!(sql, "SELECT -(-t.ts) AS x FROM t"); + ctx.sql(&sql).await?.into_optimized_plan()?; + } + + Ok(()) +} + /// Paths to benchmark query files (supports running from repo root or different working directories). const BENCHMARK_PATHS: &[&str] = &["../../benchmarks/", "./benchmarks/"]; diff --git a/datafusion/optimizer/src/simplify_expressions/expr_simplifier.rs b/datafusion/optimizer/src/simplify_expressions/expr_simplifier.rs index dadea4784802a..176a00296a994 100644 --- a/datafusion/optimizer/src/simplify_expressions/expr_simplifier.rs +++ b/datafusion/optimizer/src/simplify_expressions/expr_simplifier.rs @@ -44,6 +44,7 @@ use datafusion_expr::physical_planning_context::PhysicalPlanningContext; use datafusion_expr::{ BinaryExpr, Case, ColumnarValue, Expr, ExprSchemable, Like, Operator, Volatility, and, binary::BinaryTypeCoercer, lit, or, preimage::PreimageResult, + type_coercion::is_signed_numeric, }; use datafusion_expr::{Cast, TryCast, simplify::ExprSimplifyResult}; use datafusion_expr::{expr::ScalarFunction, interval_arithmetic::NullableInterval}; @@ -760,6 +761,10 @@ struct Simplifier<'a> { info: &'a SimplifyContext, } +fn can_cancel_double_negation(data_type: &DataType) -> bool { + data_type.is_null() || is_signed_numeric(data_type) +} + impl<'a> Simplifier<'a> { pub fn new(info: &'a SimplifyContext) -> Self { Self { info } @@ -1178,30 +1183,6 @@ impl TreeNodeRewriter for Simplifier<'_> { right, }) if !info.nullable(&right)? && is_zero(&left) => Transformed::yes(*left), - // !A & A -> 0 (if A not nullable) - Expr::BinaryExpr(BinaryExpr { - left, - op: BitwiseAnd, - right, - }) if is_negative_of(&left, &right) && !info.nullable(&right)? => { - Transformed::yes(Expr::Literal( - ScalarValue::new_zero(&info.get_data_type(&left)?)?, - None, - )) - } - - // A & !A -> 0 (if A not nullable) - Expr::BinaryExpr(BinaryExpr { - left, - op: BitwiseAnd, - right, - }) if is_negative_of(&right, &left) && !info.nullable(&left)? => { - Transformed::yes(Expr::Literal( - ScalarValue::new_zero(&info.get_data_type(&left)?)?, - None, - )) - } - // (..A..) & A --> (..A..) Expr::BinaryExpr(BinaryExpr { left, @@ -1252,30 +1233,6 @@ impl TreeNodeRewriter for Simplifier<'_> { right, }) if is_zero(&left) => Transformed::yes(*right), - // !A | A -> -1 (if A not nullable) - Expr::BinaryExpr(BinaryExpr { - left, - op: BitwiseOr, - right, - }) if is_negative_of(&left, &right) && !info.nullable(&right)? => { - Transformed::yes(Expr::Literal( - ScalarValue::new_negative_one(&info.get_data_type(&left)?)?, - None, - )) - } - - // A | !A -> -1 (if A not nullable) - Expr::BinaryExpr(BinaryExpr { - left, - op: BitwiseOr, - right, - }) if is_negative_of(&right, &left) && !info.nullable(&left)? => { - Transformed::yes(Expr::Literal( - ScalarValue::new_negative_one(&info.get_data_type(&left)?)?, - None, - )) - } - // (..A..) | A --> (..A..) Expr::BinaryExpr(BinaryExpr { left, @@ -1326,30 +1283,6 @@ impl TreeNodeRewriter for Simplifier<'_> { right, }) if !info.nullable(&right)? && is_zero(&left) => Transformed::yes(*right), - // !A ^ A -> -1 (if A not nullable) - Expr::BinaryExpr(BinaryExpr { - left, - op: BitwiseXor, - right, - }) if is_negative_of(&left, &right) && !info.nullable(&right)? => { - Transformed::yes(Expr::Literal( - ScalarValue::new_negative_one(&info.get_data_type(&left)?)?, - None, - )) - } - - // A ^ !A -> -1 (if A not nullable) - Expr::BinaryExpr(BinaryExpr { - left, - op: BitwiseXor, - right, - }) if is_negative_of(&right, &left) && !info.nullable(&left)? => { - Transformed::yes(Expr::Literal( - ScalarValue::new_negative_one(&info.get_data_type(&left)?)?, - None, - )) - } - // (..A..) ^ A --> (the expression without A, if number of A is odd, otherwise one A) Expr::BinaryExpr(BinaryExpr { left, @@ -1411,10 +1344,21 @@ impl TreeNodeRewriter for Simplifier<'_> { // Expr::Not(inner) => Transformed::yes(negate_clause(*inner)), - // - // Rules for Negative - // - Expr::Negative(inner) => Transformed::yes(distribute_negation(*inner)), + // -(-A) -> A for types where negation is total and involutive. + // Signed integer negation wraps, floating-point negation is total, + // and valid decimal precision excludes the native signed minimum. + // Timestamp and interval negation remains explicit because it can + // overflow. + Expr::Negative(inner) => match *inner { + Expr::Negative(inner) + if can_cancel_double_negation( + &info.get_data_type(inner.as_ref())?, + ) => + { + Transformed::yes(*inner) + } + inner => Transformed::no(Expr::Negative(Box::new(inner))), + }, // // Rules for Case @@ -3115,47 +3059,91 @@ mod tests { } #[test] - fn test_simplify_negated_bitwise_and() { - // !c3 & c3 --> 0 - let expr = (-col("c3_non_null")) & col("c3_non_null"); - let expected = lit(0i64); - - assert_eq!(simplify(expr), expected); - // c3 & !c3 --> 0 - let expr = col("c3_non_null") & (-col("c3_non_null")); - let expected = lit(0i64); - - assert_eq!(simplify(expr), expected); - } - - #[test] - fn test_simplify_negated_bitwise_or() { - // !c3 | c3 --> -1 - let expr = (-col("c3_non_null")) | col("c3_non_null"); - let expected = lit(-1i64); + fn test_preserve_arithmetic_negation() { + let c3 = col("c3_non_null"); + let expressions = [ + (-c3.clone()) & c3.clone(), + c3.clone() & (-c3.clone()), + (-c3.clone()) | c3.clone(), + c3.clone() | (-c3.clone()), + (-c3.clone()) ^ c3.clone(), + c3.clone() ^ (-c3.clone()), + -bitwise_and(col("c3"), c3.clone()), + -bitwise_or(col("c3"), c3.clone()), + ]; - assert_eq!(simplify(expr), expected); + for expr in expressions { + assert_eq!(simplify(expr.clone()), expr); + } - // c3 | !c3 --> -1 - let expr = col("c3_non_null") | (-col("c3_non_null")); - let expected = lit(-1i64); + // Timestamp and interval negation are checked, so the inner expression + // can overflow and must remain visible. + let checked_types = [ + ( + "ts", + DataType::Timestamp(arrow::datatypes::TimeUnit::Nanosecond, None), + ), + ( + "year_month", + DataType::Interval(arrow::datatypes::IntervalUnit::YearMonth), + ), + ( + "day_time", + DataType::Interval(arrow::datatypes::IntervalUnit::DayTime), + ), + ( + "month_day_nano", + DataType::Interval(arrow::datatypes::IntervalUnit::MonthDayNano), + ), + ]; + let schema = Schema::new( + checked_types + .iter() + .map(|(name, data_type)| Field::new(*name, data_type.clone(), false)) + .collect::>(), + ) + .to_dfschema_ref() + .unwrap(); + let simplifier = + ExprSimplifier::new(SimplifyContext::builder().with_schema(schema).build()); - assert_eq!(simplify(expr), expected); + for (name, _) in checked_types { + let expr = -(-col(name)); + assert_eq!(simplifier.simplify(expr.clone()).unwrap(), expr); + } } #[test] - fn test_simplify_negated_bitwise_xor() { - // !c3 ^ c3 --> -1 - let expr = (-col("c3_non_null")) ^ col("c3_non_null"); - let expected = lit(-1i64); - - assert_eq!(simplify(expr), expected); - - // c3 ^ !c3 --> -1 - let expr = col("c3_non_null") ^ (-col("c3_non_null")); - let expected = lit(-1i64); + fn test_cancel_safe_double_negation() { + let safe_types = [ + ("i8", DataType::Int8), + ("i16", DataType::Int16), + ("i32", DataType::Int32), + ("i64", DataType::Int64), + ("f16", DataType::Float16), + ("f32", DataType::Float32), + ("f64", DataType::Float64), + ("d32", DataType::Decimal32(9, 0)), + ("d64", DataType::Decimal64(18, 0)), + ("d128", DataType::Decimal128(38, 0)), + ("d256", DataType::Decimal256(76, 0)), + ("null", DataType::Null), + ]; + let schema = Schema::new( + safe_types + .iter() + .map(|(name, data_type)| Field::new(*name, data_type.clone(), true)) + .collect::>(), + ) + .to_dfschema_ref() + .unwrap(); + let simplifier = + ExprSimplifier::new(SimplifyContext::builder().with_schema(schema).build()); - assert_eq!(simplify(expr), expected); + for (name, _) in safe_types { + assert_eq!(simplifier.simplify(-(-col(name))).unwrap(), col(name)); + assert_eq!(simplifier.simplify(-(-(-col(name)))).unwrap(), -col(name)); + } } #[test] @@ -3340,20 +3328,6 @@ mod tests { let expr = col("c3").not().not(); let expected = col("c3"); assert_eq!(simplify(expr), expected); - - // Laws with bitwise operations - // !(c3 & c4) --> !c3 | !c4 - let expr = -bitwise_and(col("c3"), col("c4")); - let expected = bitwise_or(-col("c3"), -col("c4")); - assert_eq!(simplify(expr), expected); - // !(c3 | c4) --> !c3 & !c4 - let expr = -bitwise_or(col("c3"), col("c4")); - let expected = bitwise_and(-col("c3"), -col("c4")); - assert_eq!(simplify(expr), expected); - // !(!c3) --> c3 - let expr = -(-col("c3")); - let expected = col("c3"); - assert_eq!(simplify(expr), expected); } #[test] diff --git a/datafusion/optimizer/src/simplify_expressions/utils.rs b/datafusion/optimizer/src/simplify_expressions/utils.rs index 89bb762d59ce2..dec0cfa8c1772 100644 --- a/datafusion/optimizer/src/simplify_expressions/utils.rs +++ b/datafusion/optimizer/src/simplify_expressions/utils.rs @@ -21,7 +21,7 @@ use datafusion_common::{Result, ScalarValue, internal_err}; use datafusion_expr::{ Case, Expr, Like, Operator, expr::{Between, BinaryExpr, InList}, - expr_fn::{and, bitwise_and, bitwise_or, or}, + expr_fn::{and, or}, }; /// returns true if `needle` is found in a chain of search_op @@ -181,11 +181,6 @@ pub fn is_not_of(not_expr: &Expr, expr: &Expr) -> bool { matches!(not_expr, Expr::Not(inner) if expr == inner.as_ref()) } -/// returns true if `not_expr` is !`expr` (bitwise not) -pub fn is_negative_of(not_expr: &Expr, expr: &Expr) -> bool { - matches!(not_expr, Expr::Negative(inner) if expr == inner.as_ref()) -} - /// returns the contained boolean value in `expr` as /// `Expr::Literal(ScalarValue::Boolean(v))`. pub fn as_bool_lit(expr: &Expr) -> Result> { @@ -343,45 +338,6 @@ pub fn negate_clause(expr: Expr) -> Expr { } } -/// bitwise negate a Negative clause -/// input is the clause to be bitwise negated.(args for Negative clause) -/// For BinaryExpr: -/// ~(A & B) ===> ~A | ~B -/// ~(A | B) ===> ~A & ~B -/// For Negative: -/// ~(~A) ===> A -/// For others, use Negative clause -pub fn distribute_negation(expr: Expr) -> Expr { - match expr { - Expr::BinaryExpr(BinaryExpr { left, op, right }) => { - match op { - // ~(A & B) ===> ~A | ~B - Operator::BitwiseAnd => { - let left = distribute_negation(*left); - let right = distribute_negation(*right); - - bitwise_or(left, right) - } - // ~(A | B) ===> ~A & ~B - Operator::BitwiseOr => { - let left = distribute_negation(*left); - let right = distribute_negation(*right); - - bitwise_and(left, right) - } - // use negative clause - _ => Expr::Negative(Box::new(Expr::BinaryExpr(BinaryExpr::new( - left, op, right, - )))), - } - } - // ~(~A) ===> A - Expr::Negative(expr) => *expr, - // use negative clause - _ => Expr::Negative(Box::new(expr)), - } -} - #[cfg(test)] mod tests { use super::{is_one, is_zero}; diff --git a/datafusion/physical-expr/src/expressions/negative.rs b/datafusion/physical-expr/src/expressions/negative.rs index c894c12784dc5..22458f3811fc6 100644 --- a/datafusion/physical-expr/src/expressions/negative.rs +++ b/datafusion/physical-expr/src/expressions/negative.rs @@ -22,13 +22,19 @@ use std::sync::Arc; use crate::PhysicalExpr; +use arrow::array::{Array, ArrayRef, AsArray}; use arrow::datatypes::FieldRef; use arrow::{ compute::kernels::numeric::neg_wrapping, - datatypes::{DataType, Schema}, + datatypes::{ + ArrowNativeTypeOp, ArrowTimestampType, DataType, Schema, TimeUnit, + TimestampMicrosecondType, TimestampMillisecondType, TimestampNanosecondType, + TimestampSecondType, + }, + error::ArrowError, record_batch::RecordBatch, }; -use datafusion_common::{Result, internal_err, plan_err}; +use datafusion_common::{DataFusionError, Result, ScalarValue, internal_err, plan_err}; use datafusion_expr::interval_arithmetic::Interval; use datafusion_expr::sort_properties::ExprProperties; #[expect(deprecated)] @@ -78,6 +84,46 @@ impl std::fmt::Display for NegativeExpr { } } +fn negate_timestamp_array(array: &dyn Array) -> Result { + let array = array.as_primitive::(); + let timezone = array.timezone().map(Arc::::from); + let result = array.try_unary::<_, T, _>(|value| value.neg_checked())?; + Ok(Arc::new(result.with_timezone_opt(timezone))) +} + +fn negate_scalar(scalar: ScalarValue) -> Result { + Ok(match scalar { + ScalarValue::Int8(value) => ScalarValue::Int8(value.map(i8::wrapping_neg)), + ScalarValue::Int16(value) => ScalarValue::Int16(value.map(i16::wrapping_neg)), + ScalarValue::Int32(value) => ScalarValue::Int32(value.map(i32::wrapping_neg)), + ScalarValue::Int64(value) => ScalarValue::Int64(value.map(i64::wrapping_neg)), + scalar => scalar.arithmetic_negate()?, + }) +} + +fn negate_interval(interval: &Interval) -> Result> { + // Signed integer negation wraps at the minimum. A lower-unbounded + // interval may contain that value, whose negation can make the result + // discontinuous and unsafe to represent as a single bounded interval. + if interval.data_type().is_signed_integer() && interval.lower().is_null() { + return Ok(None); + } + + match interval.arithmetic_negate() { + Ok(interval) => Ok(Some(interval)), + Err(error) + if matches!( + error.find_root(), + DataFusionError::ArrowError(error, _) + if matches!(error.as_ref(), ArrowError::ArithmeticOverflow(_)) + ) => + { + Ok(None) + } + Err(error) => Err(error), + } +} + impl PhysicalExpr for NegativeExpr { fn data_type(&self, input_schema: &Schema) -> Result { self.arg.data_type(input_schema) @@ -90,11 +136,29 @@ impl PhysicalExpr for NegativeExpr { fn evaluate(&self, batch: &RecordBatch) -> Result { match self.arg.evaluate(batch)? { ColumnarValue::Array(array) => { - let result = neg_wrapping(array.as_ref())?; + let result = match array.data_type() { + DataType::Timestamp(TimeUnit::Second, _) => { + negate_timestamp_array::(array.as_ref())? + } + DataType::Timestamp(TimeUnit::Millisecond, _) => { + negate_timestamp_array::( + array.as_ref(), + )? + } + DataType::Timestamp(TimeUnit::Microsecond, _) => { + negate_timestamp_array::( + array.as_ref(), + )? + } + DataType::Timestamp(TimeUnit::Nanosecond, _) => { + negate_timestamp_array::(array.as_ref())? + } + _ => neg_wrapping(array.as_ref())?, + }; Ok(ColumnarValue::Array(result)) } ColumnarValue::Scalar(scalar) => { - Ok(ColumnarValue::Scalar(scalar.arithmetic_negate()?)) + Ok(ColumnarValue::Scalar(negate_scalar(scalar)?)) } } } @@ -118,7 +182,10 @@ impl PhysicalExpr for NegativeExpr { /// It replaces the upper and lower bounds after multiplying them with -1. /// Ex: `(a, b]` => `[-b, -a)` fn evaluate_bounds(&self, children: &[&Interval]) -> Result { - children[0].arithmetic_negate() + match negate_interval(children[0])? { + Some(interval) => Ok(interval), + None => Interval::make_unbounded(&children[0].data_type()), + } } /// Returns a new [`Interval`] of a NegativeExpr that has the existing `interval` given that @@ -128,7 +195,9 @@ impl PhysicalExpr for NegativeExpr { interval: &Interval, children: &[&Interval], ) -> Result>> { - let negated_interval = interval.arithmetic_negate()?; + let Some(negated_interval) = negate_interval(interval)? else { + return Ok(Some(vec![])); + }; Ok(children[0] .intersect(negated_interval)? @@ -242,7 +311,7 @@ pub fn negative( #[cfg(test)] mod tests { use super::*; - use crate::expressions::{Column, col}; + use crate::expressions::{Column, Literal, col}; use arrow::array::*; use arrow::datatypes::DataType::{Float32, Float64, Int8, Int16, Int32, Int64}; @@ -288,6 +357,89 @@ mod tests { Ok(()) } + #[test] + fn scalar_integer_negative_op_wraps() -> Result<()> { + let batch = RecordBatch::new_empty(Arc::new(Schema::empty())); + let values = [ + ScalarValue::Int8(Some(i8::MIN)), + ScalarValue::Int16(Some(i16::MIN)), + ScalarValue::Int32(Some(i32::MIN)), + ScalarValue::Int64(Some(i64::MIN)), + ]; + + for value in values { + let expr = NegativeExpr::new(Arc::new(Literal::new(value.clone()))); + let ColumnarValue::Scalar(result) = expr.evaluate(&batch)? else { + panic!("signed integer literal should produce a scalar") + }; + assert_eq!( + result, value, + "signed integer scalar negation should match array wrapping semantics" + ); + } + Ok(()) + } + + macro_rules! test_timestamp_array_negative_op { + ($ARRAY_TY:ty, $UNIT:expr) => {{ + let data_type = DataType::Timestamp($UNIT, Some("America/New_York".into())); + let schema = Schema::new(vec![Field::new("a", data_type.clone(), true)]); + let expr = negative(col("a", &schema)?, &schema)?; + let input = <$ARRAY_TY>::from(vec![Some(2), None, Some(-1)]) + .with_timezone("America/New_York"); + let expected = <$ARRAY_TY>::from(vec![Some(-2), None, Some(1)]) + .with_timezone("America/New_York"); + let batch = RecordBatch::try_new(Arc::new(schema), vec![Arc::new(input)])?; + + let result = expr.evaluate(&batch)?.into_array(batch.num_rows())?; + assert_eq!(result.data_type(), &data_type); + assert_eq!( + result + .as_any() + .downcast_ref::<$ARRAY_TY>() + .expect("timestamp array"), + &expected + ); + }}; + } + + #[test] + fn timestamp_array_negative_op() -> Result<()> { + test_timestamp_array_negative_op!(TimestampSecondArray, TimeUnit::Second); + test_timestamp_array_negative_op!( + TimestampMillisecondArray, + TimeUnit::Millisecond + ); + test_timestamp_array_negative_op!( + TimestampMicrosecondArray, + TimeUnit::Microsecond + ); + test_timestamp_array_negative_op!(TimestampNanosecondArray, TimeUnit::Nanosecond); + Ok(()) + } + + #[test] + fn timestamp_array_negative_overflow() -> Result<()> { + let data_type = + DataType::Timestamp(TimeUnit::Nanosecond, Some("America/New_York".into())); + let schema = Schema::new(vec![Field::new("a", data_type, false)]); + let expr = negative(col("a", &schema)?, &schema)?; + let input = TimestampNanosecondArray::from(vec![i64::MIN]) + .with_timezone("America/New_York"); + let batch = RecordBatch::try_new(Arc::new(schema), vec![Arc::new(input)])?; + + let err = expr.evaluate(&batch).expect_err("negation should overflow"); + assert!( + matches!( + err.find_root(), + DataFusionError::ArrowError(err, _) + if matches!(err.as_ref(), ArrowError::ArithmeticOverflow(_)) + ), + "unexpected error: {err}" + ); + Ok(()) + } + #[test] fn test_evaluate_bounds() -> Result<()> { let negative_expr = NegativeExpr::new(Arc::new(Column::new("a", 0))); @@ -300,6 +452,33 @@ mod tests { Ok(()) } + #[test] + fn test_negation_bounds_are_conservative() -> Result<()> { + let negative_expr = NegativeExpr::new(Arc::new(Column::new("a", 0))); + let child_interval = Interval::make(Some(i8::MIN), Some(1_i8))?; + let unbounded = Interval::make_unbounded(&Int8)?; + + assert_eq!( + negative_expr.evaluate_bounds(&[&child_interval])?, + unbounded + ); + assert_eq!( + negative_expr.propagate_constraints(&child_interval, &[&child_interval])?, + Some(vec![]) + ); + + let lower_unbounded = Interval::make(None, Some(-1_i8))?; + assert_eq!( + negative_expr.evaluate_bounds(&[&lower_unbounded])?, + unbounded + ); + assert_eq!( + negative_expr.propagate_constraints(&lower_unbounded, &[&child_interval])?, + Some(vec![]) + ); + Ok(()) + } + #[test] #[expect(deprecated)] fn test_evaluate_statistics() -> Result<()> { diff --git a/datafusion/pruning/src/pruning_predicate.rs b/datafusion/pruning/src/pruning_predicate.rs index dff18173ae32a..398820547d78f 100644 --- a/datafusion/pruning/src/pruning_predicate.rs +++ b/datafusion/pruning/src/pruning_predicate.rs @@ -1059,6 +1059,7 @@ struct PruningExpressionBuilder<'a> { column_expr: Arc, op: Operator, scalar_expr: Arc, + signed_negative_min: Option, field: &'a Field, required_columns: &'a mut RequiredColumns, } @@ -1103,6 +1104,8 @@ impl<'a> PruningExpressionBuilder<'a> { } }; + let signed_negative_min = + signed_negative_min_guard(column_expr, correct_operator, schema.as_ref())?; let df_schema = DFSchema::try_from(Arc::clone(schema))?; let (column_expr, correct_operator, scalar_expr) = rewrite_expr_to_prunable( column_expr, @@ -1119,6 +1122,7 @@ impl<'a> PruningExpressionBuilder<'a> { column_expr, op: correct_operator, scalar_expr, + signed_negative_min, field, required_columns, }) @@ -1132,6 +1136,10 @@ impl<'a> PruningExpressionBuilder<'a> { &self.scalar_expr } + fn signed_negative_min(&self) -> Option<&ScalarValue> { + self.signed_negative_min.as_ref() + } + fn min_column_expr(&mut self) -> Result> { self.required_columns .min_column_expr(&self.column, &self.column_expr, self.field) @@ -1183,12 +1191,80 @@ impl<'a> PruningExpressionBuilder<'a> { } } +fn is_inequality_op(op: Operator) -> bool { + matches!( + op, + Operator::Lt | Operator::LtEq | Operator::Gt | Operator::GtEq + ) +} + +fn contains_signed_integer_negative( + expr: &PhysicalExprRef, + schema: &Schema, +) -> Result { + if let Some(negative) = expr.downcast_ref::() + && negative.data_type(schema)?.is_signed_integer() + { + return Ok(true); + } + + for child in expr.children() { + if contains_signed_integer_negative(child, schema)? { + return Ok(true); + } + } + Ok(false) +} + +/// Returns the wrapping point for a signed integer negation that can be +/// represented safely using column statistics. +/// +/// Wrapping negation is decreasing everywhere except at `MIN`. We can retain +/// precise inequality pruning for `-column` by guarding the normal rewrite +/// with checks for that value. More complex expressions fall back to the +/// unhandled-predicate path rather than assuming their statistics describe the +/// input to the negation. +fn signed_negative_min_guard( + column_expr: &PhysicalExprRef, + op: Operator, + schema: &Schema, +) -> Result> { + if let Some(negative) = column_expr.downcast_ref::() { + let data_type = negative.data_type(schema)?; + if data_type.is_signed_integer() { + if negative.arg().downcast_ref::().is_none() { + return plan_err!( + "Comparison pruning through signed integer negation only supports a direct column" + ); + } + if !is_inequality_op(op) { + return Ok(None); + } + let min = ScalarValue::min(&data_type).ok_or_else(|| { + internal_datafusion_err!( + "No minimum scalar value for signed integer type {data_type}" + ) + })?; + return Ok(Some(min)); + } + } + + if contains_signed_integer_negative(column_expr, schema)? { + return plan_err!( + "Comparison pruning through signed integer negation only supports a direct column" + ); + } + + Ok(None) +} + /// This function is designed to rewrite the column_expr to /// ensure the column_expr is monotonically increasing. /// /// For example, /// 1. `col > 10` -/// 2. `-col > 10` should be rewritten to `col < -10` +/// 2. `-col > 10` can be rewritten to `col < -10`; wrapping signed integers +/// also require guards at `MIN` /// 3. `!col = true` would be rewritten to `col = !true` /// 4. `abs(a - 10) > 0` not supported /// 5. `cast(can_prunable_expr) > 10` @@ -1753,7 +1829,7 @@ impl ColumnReferenceCount { fn build_statistics_expr( expr_builder: &mut PruningExpressionBuilder, ) -> Result> { - let statistics_expr: Arc = match expr_builder.op() { + let mut statistics_expr: Arc = match expr_builder.op() { Operator::NotEq => build_ne_statistics_expr(expr_builder)?, Operator::Eq => { // column = literal => (min, max) = literal => min <= literal && literal <= max @@ -1807,6 +1883,20 @@ fn build_statistics_expr( ); } }; + + if let Some(min) = expr_builder.signed_negative_min().cloned() { + let min = Arc::new(phys_expr::Literal::new(min)) as Arc; + let scalar_expr = Arc::clone(expr_builder.scalar_expr()); + let input_may_wrap = binary_expr( + expr_builder.min_column_expr()?, + Operator::Eq, + Arc::clone(&min), + ); + let scalar_may_wrap = binary_expr(scalar_expr, Operator::Eq, min); + statistics_expr = + or_expr(input_may_wrap, or_expr(scalar_may_wrap, statistics_expr)); + } + let statistics_expr = wrap_null_count_check_expr(statistics_expr, expr_builder)?; Ok(statistics_expr) } @@ -2182,7 +2272,10 @@ mod tests { use arrow::array::Decimal128Array; use arrow::{ - array::{BinaryArray, Int32Array, Int64Array, StringArray, UInt64Array}, + array::{ + BinaryArray, Int8Array, Int16Array, Int32Array, Int64Array, StringArray, + UInt64Array, + }, datatypes::TimeUnit, }; use datafusion_expr::expr::InList; @@ -4273,6 +4366,183 @@ mod tests { ); } + /// Creates statistics that exercise the discontinuity in wrapping signed + /// integer negation. Each container has the following `[min, max]` range: + /// + /// i [MIN, MIN] + /// i [MIN, -1] + /// i [1, 2] + /// i [NULL, 2] + /// i [-2, -1] + fn int32_wrapping_negation_setup() -> (SchemaRef, TestStatistics) { + let schema = Arc::new(Schema::new(vec![Field::new("i", DataType::Int32, true)])); + let statistics = TestStatistics::new().with( + "i", + ContainerStats::new_i32( + vec![Some(i32::MIN), Some(i32::MIN), Some(1), None, Some(-2)], + vec![Some(i32::MIN), Some(-1), Some(2), Some(2), Some(-1)], + ), + ); + (schema, statistics) + } + + #[test] + fn prune_signed_negative_min_guard_all_integer_widths() { + macro_rules! assert_guard { + ($data_type:expr, $array_type:ty, $native_min:expr) => {{ + let schema = + Arc::new(Schema::new(vec![Field::new("i", $data_type, false)])); + let statistics = TestStatistics::new().with( + "i", + ContainerStats::new() + .with_min(Arc::new(<$array_type>::from(vec![ + Some($native_min), + Some(1), + Some(-2), + ]))) + .with_max(Arc::new(<$array_type>::from(vec![ + Some($native_min), + Some(2), + Some(-1), + ]))), + ); + let min = ScalarValue::min(schema.field(0).data_type()).unwrap(); + let zero = ScalarValue::new_zero(schema.field(0).data_type()).unwrap(); + + // The first container exercises the input-MIN guard while the + // other two show that the normal monotone rewrite stays useful. + prune_with_expr( + (-col("i")).lt(lit(zero)), + &schema, + &statistics, + &[true, true, false], + ); + // Every container is kept when the scalar negation itself + // wraps at MIN. + prune_with_expr( + (-col("i")).gt(lit(min)), + &schema, + &statistics, + &[true, true, true], + ); + }}; + } + + assert_guard!(DataType::Int8, Int8Array, i8::MIN); + assert_guard!(DataType::Int16, Int16Array, i16::MIN); + assert_guard!(DataType::Int32, Int32Array, i32::MIN); + assert_guard!(DataType::Int64, Int64Array, i64::MIN); + } + + #[test] + fn prune_signed_negative_inequalities_guard_min() { + let (schema, statistics) = int32_wrapping_negation_setup(); + + // The MIN-containing and lower-unknown containers must be kept. The + // remaining containers retain the precision of the monotone rewrite. + let positive_range = &[true, true, true, true, false]; + let negative_range = &[true, true, false, true, true]; + prune_with_expr( + (-col("i")).lt(lit(0_i32)), + &schema, + &statistics, + positive_range, + ); + prune_with_expr( + (-col("i")).lt_eq(lit(0_i32)), + &schema, + &statistics, + positive_range, + ); + prune_with_expr( + (-col("i")).gt(lit(0_i32)), + &schema, + &statistics, + negative_range, + ); + prune_with_expr( + (-col("i")).gt_eq(lit(0_i32)), + &schema, + &statistics, + negative_range, + ); + + // Negating MIN wraps back to MIN. This guard is also required when the + // column-bearing expression is on the right of the comparison. + let keep_all = &[true, true, true, true, true]; + prune_with_expr( + (-col("i")).gt(lit(i32::MIN)), + &schema, + &statistics, + keep_all, + ); + prune_with_expr(lit(i32::MIN).lt(-col("i")), &schema, &statistics, keep_all); + } + + #[test] + fn prune_direct_signed_negative_equality_is_unchanged() { + let (schema, statistics) = int32_wrapping_negation_setup(); + + prune_with_expr( + (-col("i")).eq(lit(i32::MIN)), + &schema, + &statistics, + &[true, true, false, true, false], + ); + prune_with_expr( + (-col("i")).not_eq(lit(i32::MIN)), + &schema, + &statistics, + &[false, true, true, true, true], + ); + } + + #[test] + fn prune_complex_signed_negative_comparisons_fall_back() { + let (schema, statistics) = int32_wrapping_negation_setup(); + let keep_all = &[true, true, true, true, true]; + + let expressions = [ + (-cast(col("i"), DataType::Int64)).lt(lit(0_i64)), + cast(-col("i"), DataType::Int64).lt(lit(0_i64)), + try_cast(-col("i"), DataType::Int64).lt(lit(0_i64)), + (-(-(-col("i")))).lt(lit(0_i32)), + cast(-col("i"), DataType::Int64).eq(lit(i32::MIN as i64)), + is_not_distinct_from(cast(-col("i"), DataType::Int64), lit(i32::MIN as i64)), + ]; + for expr in expressions { + let physical = logical2physical(&expr, &schema); + let rewritten = PredicateRewriter::new() + .rewrite_predicate_to_statistics_predicate(&physical, &schema); + assert!(is_always_true(&rewritten), "rewritten: {rewritten}"); + prune_with_expr(expr, &schema, &statistics, keep_all); + } + + // Casting after Int8 negation changes the comparison type, not the + // type whose MIN wraps. All comparison variants must therefore fall + // back instead of pushing through the cast. + let schema = Arc::new(Schema::new(vec![Field::new("i", DataType::Int8, false)])); + let statistics = TestStatistics::new().with( + "i", + ContainerStats::new() + .with_min(Arc::new(Int8Array::from(vec![Some(i8::MIN), Some(1)]))) + .with_max(Arc::new(Int8Array::from(vec![Some(i8::MIN), Some(2)]))), + ); + let cast_negative = cast(-col("i"), DataType::Int64); + let expressions = [ + cast_negative.clone().eq(lit(-128_i64)), + is_distinct_from(cast_negative.clone(), lit(-128_i64)), + is_not_distinct_from(cast_negative, lit(-128_i64)), + ]; + for expr in expressions { + let physical = logical2physical(&expr, &schema); + let rewritten = PredicateRewriter::new() + .rewrite_predicate_to_statistics_predicate(&physical, &schema); + assert!(is_always_true(&rewritten), "rewritten: {rewritten}"); + prune_with_expr(expr, &schema, &statistics, &[true, true]); + } + } + #[test] fn prune_int32_col_lte_zero_cast() { let (schema, statistics) = int32_setup(); @@ -4649,7 +4919,7 @@ mod tests { .lt(lit(ScalarValue::Int64(Some(0)))), &schema, &statistics, - expected_ret, + &[true, true, true, true, true], ); } diff --git a/datafusion/sql/src/unparser/expr.rs b/datafusion/sql/src/unparser/expr.rs index 898330018c708..80c19f0f027e8 100644 --- a/datafusion/sql/src/unparser/expr.rs +++ b/datafusion/sql/src/unparser/expr.rs @@ -533,7 +533,23 @@ impl Unparser<'_> { }) } Expr::Negative(expr) => { - let sql_parser_expr = self.expr_to_sql_inner(expr)?; + let mut sql_parser_expr = self.expr_to_sql_inner(expr)?; + let needs_parentheses = matches!( + &sql_parser_expr, + AstExpr::UnaryOp { + op: UnaryOperator::Minus, + .. + } + ) || matches!( + &sql_parser_expr, + AstExpr::Value(ValueWithSpan { + value: ast::Value::Number(value, _), + .. + }) if value.starts_with('-') + ); + if needs_parentheses { + sql_parser_expr = AstExpr::Nested(Box::new(sql_parser_expr)); + } Ok(AstExpr::UnaryOp { op: UnaryOperator::Minus, expr: Box::new(sql_parser_expr), @@ -2349,6 +2365,11 @@ mod tests { r#"(a BETWEEN 1 AND 7)"#, ), (Expr::Negative(Box::new(col("a"))), r#"-a"#), + ( + Expr::Negative(Box::new(Expr::Negative(Box::new(col("a"))))), + r#"-(-a)"#, + ), + (Expr::Negative(Box::new(lit(-1_i64))), r#"-(-1)"#), ( exists(Arc::new(dummy_logical_plan.clone())), r#"EXISTS (SELECT * FROM t WHERE (t.a = 1))"#, diff --git a/datafusion/sqllogictest/test_files/filter_without_sort_exec.slt b/datafusion/sqllogictest/test_files/filter_without_sort_exec.slt index 633e19f7915db..99d01121665ad 100644 --- a/datafusion/sqllogictest/test_files/filter_without_sort_exec.slt +++ b/datafusion/sqllogictest/test_files/filter_without_sort_exec.slt @@ -25,6 +25,44 @@ CREATE UNBOUNDED EXTERNAL TABLE data ( WITH ORDER ("date", "ticker", "time") LOCATION './a.parquet'; +# Widening casts and double negation retain source ordering on unbounded inputs +statement ok +CREATE UNBOUNDED EXTERNAL TABLE neg_order_stream ( + i INTEGER NOT NULL, + j BIGINT NOT NULL +) STORED AS CSV +WITH ORDER (i ASC NULLS LAST, j ASC NULLS LAST) +LOCATION './a.parquet'; + +query TT +EXPLAIN SELECT i FROM neg_order_stream +ORDER BY -CAST(i AS BIGINT) DESC NULLS LAST; +---- +logical_plan +01)Sort: (- CAST(neg_order_stream.i AS Int64)) DESC NULLS LAST +02)--TableScan: neg_order_stream projection=[i] +physical_plan StreamingTableExec: partition_sizes=1, projection=[i], infinite_source=true, output_ordering=[i@0 ASC NULLS LAST] + +query TT +EXPLAIN SELECT i FROM neg_order_stream +ORDER BY -(-i) ASC NULLS LAST; +---- +logical_plan +01)Sort: neg_order_stream.i AS (- (- neg_order_stream.i)) ASC NULLS LAST +02)--TableScan: neg_order_stream projection=[i] +physical_plan StreamingTableExec: partition_sizes=1, projection=[i], infinite_source=true, output_ordering=[i@0 ASC NULLS LAST] + +# Double negation must not add a partial sort to an unbounded two-key input +query TT +EXPLAIN SELECT i, j FROM neg_order_stream +ORDER BY -(-i) ASC NULLS LAST, j ASC NULLS LAST +LIMIT 1; +---- +logical_plan +01)Sort: neg_order_stream.i AS (- (- neg_order_stream.i)) ASC NULLS LAST, neg_order_stream.j ASC NULLS LAST, fetch=1 +02)--TableScan: neg_order_stream projection=[i, j] +physical_plan StreamingTableExec: partition_sizes=1, projection=[i, j], infinite_source=true, fetch=1, output_ordering=[i@0 ASC NULLS LAST, j@1 ASC NULLS LAST] + # query query TT diff --git a/datafusion/sqllogictest/test_files/parquet_statistics.slt b/datafusion/sqllogictest/test_files/parquet_statistics.slt index 9cf6b1e0381d1..88393da530763 100644 --- a/datafusion/sqllogictest/test_files/parquet_statistics.slt +++ b/datafusion/sqllogictest/test_files/parquet_statistics.slt @@ -186,6 +186,179 @@ physical_plan statement ok DROP TABLE typed_table; +# Wrapping negation must remain safe in pruning and interval analysis +statement ok +SET datafusion.execution.collect_statistics = true; + +query I +COPY ( + SELECT CAST(i AS TINYINT) AS i + FROM (VALUES (1), (2)) AS t(i) +) +TO 'test_files/scratch/parquet_statistics/negative_bounds/positive.parquet' +STORED AS PARQUET; +---- +2 + +query I +COPY ( + SELECT CAST(i AS TINYINT) AS i + FROM (VALUES (-128), (-127)) AS t(i) +) +TO 'test_files/scratch/parquet_statistics/negative_bounds/wrap_point.parquet' +STORED AS PARQUET; +---- +2 + +statement ok +SET datafusion.execution.parquet.pushdown_filters = false; + +statement ok +SET datafusion.execution.parquet.pruning = true; + +statement ok +SET datafusion.execution.target_partitions = 1; + +statement ok +CREATE EXTERNAL TABLE negative_bounds ( + i TINYINT NOT NULL +) +STORED AS PARQUET +LOCATION 'test_files/scratch/parquet_statistics/negative_bounds'; + +# Pruning signed negation inequalities must account for the wrapping minimum +query I +SELECT i FROM negative_bounds +WHERE -i > CAST(-128 AS TINYINT) +ORDER BY i; +---- +-127 +1 +2 + +query I +SELECT i FROM negative_bounds +WHERE -i >= CAST(-128 AS TINYINT) +ORDER BY i; +---- +-128 +-127 +1 +2 + +query I +SELECT i FROM negative_bounds +WHERE -i < CAST(-1 AS TINYINT) +ORDER BY i; +---- +-128 +2 + +query I +SELECT i FROM negative_bounds +WHERE -i <= CAST(-1 AS TINYINT) +ORDER BY i; +---- +-128 +1 +2 + +statement ok +SET datafusion.execution.parquet.pruning = false; + +# Dynamic filter analysis must also retain the wrapping minimum +query I +SELECT i +FROM negative_bounds +WHERE -i < CAST(0 AS TINYINT) +ORDER BY i +LIMIT 1; +---- +-128 + +statement ok +DROP TABLE negative_bounds; + +# Constant column replacement must preserve signed integer negation semantics +query I +COPY ( + SELECT + CAST(-128 AS TINYINT) AS i, + CAST(64 AS TINYINT) AS half_i +) +TO 'test_files/scratch/parquet_statistics/constant_negative.parquet' +STORED AS PARQUET; +---- +1 + +statement ok +CREATE EXTERNAL TABLE constant_negative ( + i TINYINT NOT NULL, + half_i TINYINT NOT NULL +) +STORED AS PARQUET +LOCATION 'test_files/scratch/parquet_statistics/constant_negative.parquet'; + +query IIIIIII +SELECT + i, + -i, + -(-i), + (-i) & i, + i | (-i), + (-i) ^ i, + -(-(half_i * CAST(2 AS TINYINT))) +FROM constant_negative; +---- +-128 -128 -128 -128 -128 0 -128 + +query I +SELECT i +FROM constant_negative +WHERE -i = CAST(-128 AS TINYINT); +---- +-128 + +statement ok +DROP TABLE constant_negative; + +# Partition values use the same scalar replacement path +query I +COPY ( + SELECT CAST(1 AS INT) AS value, CAST(-128 AS TINYINT) AS i +) +TO 'test_files/scratch/parquet_statistics/negative_partition' +STORED AS PARQUET +PARTITIONED BY (i); +---- +1 + +statement ok +CREATE EXTERNAL TABLE negative_partition ( + value INT NOT NULL +) +STORED AS PARQUET +LOCATION 'test_files/scratch/parquet_statistics/negative_partition' +PARTITIONED BY (i TINYINT); + +query III +SELECT -i, -(-i), (-i) & i +FROM negative_partition; +---- +-128 -128 -128 + +statement ok +DROP TABLE negative_partition; + +statement ok +RESET datafusion.execution.parquet.pushdown_filters; + +statement ok +RESET datafusion.execution.parquet.pruning; + +statement ok +SET datafusion.execution.target_partitions = 4; + # Config reset statement ok RESET datafusion.execution.collect_statistics; diff --git a/datafusion/sqllogictest/test_files/scalar.slt b/datafusion/sqllogictest/test_files/scalar.slt index 175607b62089f..710f26bf2996c 100644 --- a/datafusion/sqllogictest/test_files/scalar.slt +++ b/datafusion/sqllogictest/test_files/scalar.slt @@ -1455,6 +1455,61 @@ from (values (NULL::INT, 7), (3, 7)) as t(a, b); 7 7 0 NULL NULL NULL +# arithmetic negation is not bitwise NOT +query IIIIII +select + i, + (-i) & i, + i | (-i), + (-i) ^ i, + -(i & j), + -(i | j) +from (values (5, 3), (6, 3)) as t(i, j) +order by i; +---- +5 1 -1 -2 -1 -7 +6 2 -2 -4 -2 -7 + +# double negation of timestamp columns remains executable +query P +select -(-ts) +from (values + (timestamp '2020-01-01 00:00:00'), + (timestamp '2020-01-02 00:00:00') +) as t(ts) +order by ts; +---- +2020-01-01T00:00:00 +2020-01-02T00:00:00 + +# interval analysis remains conservative at the signed minimum +query I +select i +from (values + (cast(-128 as tinyint)), + (cast(-1 as tinyint)), + (cast(1 as tinyint)) +) as t(i) +where -(-i) >= cast(-128 as tinyint) +order by i; +---- +-128 +-1 +1 + +# signed integer scalar negation matches array wrapping semantics +query I +select -(-cast(-128 as tinyint)); +---- +-128 + +# double negation does not hide checked timestamp overflow +query error Arithmetic overflow: Overflow happened on: - -9223372036854775808 +select -(-arrow_cast( + cast(-9223372036854775808 as bigint), + 'Timestamp(Nanosecond, None)' +)); + # bitwise xor with other operators query II rowsort select 2 * c - 1 ^ 856 + d + 3, d ^ 7 >> 4 from signed_integers;