datafusion_physical_expr/expressions/
negative.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18//! Negation (-) expression
19
20use std::hash::Hash;
21use std::sync::Arc;
22
23use crate::PhysicalExpr;
24
25use arrow::datatypes::FieldRef;
26use arrow::{
27    compute::kernels::numeric::neg_wrapping,
28    datatypes::{DataType, Schema},
29    record_batch::RecordBatch,
30};
31use datafusion_common::{Result, internal_err, plan_err};
32use datafusion_expr::interval_arithmetic::Interval;
33use datafusion_expr::sort_properties::ExprProperties;
34#[expect(deprecated)]
35use datafusion_expr::statistics::Distribution::{
36    self, Bernoulli, Exponential, Gaussian, Generic, Uniform,
37};
38use datafusion_expr::{
39    ColumnarValue,
40    type_coercion::{is_interval, is_signed_numeric, is_timestamp},
41};
42
43/// Negative expression
44#[derive(Debug, Eq)]
45pub struct NegativeExpr {
46    /// Input expression
47    arg: Arc<dyn PhysicalExpr>,
48}
49
50// Manually derive PartialEq and Hash to work around https://github.com/rust-lang/rust/issues/78808
51impl PartialEq for NegativeExpr {
52    fn eq(&self, other: &Self) -> bool {
53        self.arg.eq(&other.arg)
54    }
55}
56
57impl Hash for NegativeExpr {
58    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
59        self.arg.hash(state);
60    }
61}
62
63impl NegativeExpr {
64    /// Create new not expression
65    pub fn new(arg: Arc<dyn PhysicalExpr>) -> Self {
66        Self { arg }
67    }
68
69    /// Get the input expression
70    pub fn arg(&self) -> &Arc<dyn PhysicalExpr> {
71        &self.arg
72    }
73}
74
75impl std::fmt::Display for NegativeExpr {
76    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
77        write!(f, "(- {})", self.arg)
78    }
79}
80
81impl PhysicalExpr for NegativeExpr {
82    fn data_type(&self, input_schema: &Schema) -> Result<DataType> {
83        self.arg.data_type(input_schema)
84    }
85
86    fn nullable(&self, input_schema: &Schema) -> Result<bool> {
87        self.arg.nullable(input_schema)
88    }
89
90    fn evaluate(&self, batch: &RecordBatch) -> Result<ColumnarValue> {
91        match self.arg.evaluate(batch)? {
92            ColumnarValue::Array(array) => {
93                let result = neg_wrapping(array.as_ref())?;
94                Ok(ColumnarValue::Array(result))
95            }
96            ColumnarValue::Scalar(scalar) => {
97                Ok(ColumnarValue::Scalar(scalar.arithmetic_negate()?))
98            }
99        }
100    }
101
102    fn return_field(&self, input_schema: &Schema) -> Result<FieldRef> {
103        self.arg.return_field(input_schema)
104    }
105
106    fn children(&self) -> Vec<&Arc<dyn PhysicalExpr>> {
107        vec![&self.arg]
108    }
109
110    fn with_new_children(
111        self: Arc<Self>,
112        children: Vec<Arc<dyn PhysicalExpr>>,
113    ) -> Result<Arc<dyn PhysicalExpr>> {
114        Ok(Arc::new(NegativeExpr::new(Arc::clone(&children[0]))))
115    }
116
117    /// Given the child interval of a NegativeExpr, it calculates the NegativeExpr's interval.
118    /// It replaces the upper and lower bounds after multiplying them with -1.
119    /// Ex: `(a, b]` => `[-b, -a)`
120    fn evaluate_bounds(&self, children: &[&Interval]) -> Result<Interval> {
121        children[0].arithmetic_negate()
122    }
123
124    /// Returns a new [`Interval`] of a NegativeExpr  that has the existing `interval` given that
125    /// given the input interval is known to be `children`.
126    fn propagate_constraints(
127        &self,
128        interval: &Interval,
129        children: &[&Interval],
130    ) -> Result<Option<Vec<Interval>>> {
131        let negated_interval = interval.arithmetic_negate()?;
132
133        Ok(children[0]
134            .intersect(negated_interval)?
135            .map(|result| vec![result]))
136    }
137
138    #[expect(deprecated)]
139    fn evaluate_statistics(&self, children: &[&Distribution]) -> Result<Distribution> {
140        match children[0] {
141            Uniform(u) => Distribution::new_uniform(u.range().arithmetic_negate()?),
142            Exponential(e) => Distribution::new_exponential(
143                e.rate().clone(),
144                e.offset().arithmetic_negate()?,
145                !e.positive_tail(),
146            ),
147            Gaussian(g) => Distribution::new_gaussian(
148                g.mean().arithmetic_negate()?,
149                g.variance().clone(),
150            ),
151            Bernoulli(_) => {
152                internal_err!("NegativeExpr cannot operate on Boolean datatypes")
153            }
154            Generic(u) => Distribution::new_generic(
155                u.mean().arithmetic_negate()?,
156                u.median().arithmetic_negate()?,
157                u.variance().clone(),
158                u.range().arithmetic_negate()?,
159            ),
160        }
161    }
162
163    /// The ordering of a [`NegativeExpr`] is simply the reverse of its child.
164    fn get_properties(&self, children: &[ExprProperties]) -> Result<ExprProperties> {
165        Ok(ExprProperties {
166            sort_properties: -children[0].sort_properties,
167            range: children[0].range.clone().arithmetic_negate()?,
168            preserves_lex_ordering: false,
169            // Negation is one-to-one but reverses the ordering direction.
170            strictly_order_preserving: false,
171        })
172    }
173
174    fn fmt_sql(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
175        write!(f, "(- ")?;
176        self.arg.fmt_sql(f)?;
177        write!(f, ")")
178    }
179
180    #[cfg(feature = "proto")]
181    fn try_to_proto(
182        &self,
183        ctx: &datafusion_physical_expr_common::physical_expr::proto_encode::PhysicalExprEncodeCtx<'_>,
184    ) -> Result<Option<datafusion_proto_models::protobuf::PhysicalExprNode>> {
185        use datafusion_proto_models::protobuf;
186
187        Ok(Some(protobuf::PhysicalExprNode {
188            expr_id: None,
189            expr_type: Some(protobuf::physical_expr_node::ExprType::Negative(Box::new(
190                protobuf::PhysicalNegativeNode {
191                    expr: Some(Box::new(ctx.encode_child(&self.arg)?)),
192                },
193            ))),
194        }))
195    }
196}
197
198#[cfg(feature = "proto")]
199impl NegativeExpr {
200    /// Reconstruct a [`NegativeExpr`] from its protobuf representation.
201    pub fn try_from_proto(
202        node: &datafusion_proto_models::protobuf::PhysicalExprNode,
203        ctx: &datafusion_physical_expr_common::physical_expr::proto_decode::PhysicalExprDecodeCtx<'_>,
204    ) -> Result<Arc<dyn PhysicalExpr>> {
205        use datafusion_physical_expr_common::expect_expr_variant;
206        use datafusion_proto_models::protobuf;
207
208        let n = expect_expr_variant!(
209            node,
210            protobuf::physical_expr_node::ExprType::Negative,
211            "Negative",
212        );
213        let expr =
214            ctx.decode_required_expression(n.expr.as_deref(), "NegativeExpr", "expr")?;
215
216        Ok(Arc::new(NegativeExpr::new(expr)))
217    }
218}
219
220/// Creates a unary expression NEGATIVE
221///
222/// # Errors
223///
224/// This function errors when the argument's type is not signed numeric
225pub fn negative(
226    arg: Arc<dyn PhysicalExpr>,
227    input_schema: &Schema,
228) -> Result<Arc<dyn PhysicalExpr>> {
229    let data_type = arg.data_type(input_schema)?;
230    if data_type.is_null() {
231        Ok(arg)
232    } else if !is_signed_numeric(&data_type)
233        && !is_interval(&data_type)
234        && !is_timestamp(&data_type)
235    {
236        plan_err!("Negation only supports numeric, interval and timestamp types")
237    } else {
238        Ok(Arc::new(NegativeExpr::new(arg)))
239    }
240}
241
242#[cfg(test)]
243mod tests {
244    use super::*;
245    use crate::expressions::{Column, col};
246
247    use arrow::array::*;
248    use arrow::datatypes::DataType::{Float32, Float64, Int8, Int16, Int32, Int64};
249    use arrow::datatypes::*;
250    use datafusion_common::cast::as_primitive_array;
251    use datafusion_common::{DataFusionError, ScalarValue};
252
253    use datafusion_physical_expr_common::physical_expr::fmt_sql;
254
255    macro_rules! test_array_negative_op {
256        ($DATA_TY:tt, $ARRAY_TY:ty, $($VALUE:expr),*   ) => {
257            let schema = Schema::new(vec![Field::new("a", DataType::$DATA_TY, true)]);
258            let expr = negative(col("a", &schema)?, &schema)?;
259            assert_eq!(expr.data_type(&schema)?, DataType::$DATA_TY);
260            assert!(expr.nullable(&schema)?);
261            let mut arr = Vec::new();
262            let mut arr_expected = Vec::new();
263            $(
264                arr.push(Some($VALUE));
265                arr_expected.push(Some(-$VALUE));
266            )+
267            arr.push(None);
268            arr_expected.push(None);
269            let input = <$ARRAY_TY>::from(arr);
270            let expected = &<$ARRAY_TY>::from(arr_expected);
271            let batch =
272                RecordBatch::try_new(Arc::new(schema.clone()), vec![Arc::new(input)])?;
273            let result = expr.evaluate(&batch)?.into_array(batch.num_rows()).expect("Failed to convert to array");
274            let result =
275                as_primitive_array(&result).expect(format!("failed to downcast to {:?}Array", $DATA_TY).as_str());
276            assert_eq!(result, expected);
277        };
278    }
279
280    #[test]
281    fn array_negative_op() -> Result<()> {
282        test_array_negative_op!(Int8, Int8Array, 2i8, 1i8);
283        test_array_negative_op!(Int16, Int16Array, 234i16, 123i16);
284        test_array_negative_op!(Int32, Int32Array, 2345i32, 1234i32);
285        test_array_negative_op!(Int64, Int64Array, 23456i64, 12345i64);
286        test_array_negative_op!(Float32, Float32Array, 2345.0f32, 1234.0f32);
287        test_array_negative_op!(Float64, Float64Array, 23456.0f64, 12345.0f64);
288        Ok(())
289    }
290
291    #[test]
292    fn test_evaluate_bounds() -> Result<()> {
293        let negative_expr = NegativeExpr::new(Arc::new(Column::new("a", 0)));
294        let child_interval = Interval::make(Some(-2), Some(1))?;
295        let negative_expr_interval = Interval::make(Some(-1), Some(2))?;
296        assert_eq!(
297            negative_expr.evaluate_bounds(&[&child_interval])?,
298            negative_expr_interval
299        );
300        Ok(())
301    }
302
303    #[test]
304    #[expect(deprecated)]
305    fn test_evaluate_statistics() -> Result<()> {
306        let negative_expr = NegativeExpr::new(Arc::new(Column::new("a", 0)));
307
308        // Uniform
309        assert_eq!(
310            negative_expr.evaluate_statistics(&[&Distribution::new_uniform(
311                Interval::make(Some(-2.), Some(3.))?
312            )?])?,
313            Distribution::new_uniform(Interval::make(Some(-3.), Some(2.))?)?
314        );
315
316        // Bernoulli
317        assert!(
318            negative_expr
319                .evaluate_statistics(&[&Distribution::new_bernoulli(ScalarValue::from(
320                    0.75
321                ))?])
322                .is_err()
323        );
324
325        // Exponential
326        assert_eq!(
327            negative_expr.evaluate_statistics(&[&Distribution::new_exponential(
328                ScalarValue::from(1.),
329                ScalarValue::from(1.),
330                true
331            )?])?,
332            Distribution::new_exponential(
333                ScalarValue::from(1.),
334                ScalarValue::from(-1.),
335                false
336            )?
337        );
338
339        // Gaussian
340        assert_eq!(
341            negative_expr.evaluate_statistics(&[&Distribution::new_gaussian(
342                ScalarValue::from(15),
343                ScalarValue::from(225),
344            )?])?,
345            Distribution::new_gaussian(ScalarValue::from(-15), ScalarValue::from(225),)?
346        );
347
348        // Unknown
349        assert_eq!(
350            negative_expr.evaluate_statistics(&[&Distribution::new_generic(
351                ScalarValue::from(15),
352                ScalarValue::from(15),
353                ScalarValue::from(10),
354                Interval::make(Some(10), Some(20))?
355            )?])?,
356            Distribution::new_generic(
357                ScalarValue::from(-15),
358                ScalarValue::from(-15),
359                ScalarValue::from(10),
360                Interval::make(Some(-20), Some(-10))?
361            )?
362        );
363
364        Ok(())
365    }
366
367    #[test]
368    fn test_propagate_constraints() -> Result<()> {
369        let negative_expr = NegativeExpr::new(Arc::new(Column::new("a", 0)));
370        let original_child_interval = Interval::make(Some(-2), Some(3))?;
371        let negative_expr_interval = Interval::make(Some(0), Some(4))?;
372        let after_propagation = Some(vec![Interval::make(Some(-2), Some(0))?]);
373        assert_eq!(
374            negative_expr.propagate_constraints(
375                &negative_expr_interval,
376                &[&original_child_interval]
377            )?,
378            after_propagation
379        );
380        Ok(())
381    }
382
383    #[test]
384    #[expect(deprecated)]
385    fn test_propagate_statistics_range_holders() -> Result<()> {
386        let negative_expr = NegativeExpr::new(Arc::new(Column::new("a", 0)));
387        let original_child_interval = Interval::make(Some(-2), Some(3))?;
388        let after_propagation = Interval::make(Some(-2), Some(0))?;
389
390        let parent = Distribution::new_uniform(Interval::make(Some(0), Some(4))?)?;
391        let children: Vec<Vec<Distribution>> = vec![
392            vec![Distribution::new_uniform(original_child_interval.clone())?],
393            vec![Distribution::new_generic(
394                ScalarValue::from(0),
395                ScalarValue::from(0),
396                ScalarValue::Int32(None),
397                original_child_interval.clone(),
398            )?],
399        ];
400
401        for child_view in children {
402            let child_refs: Vec<_> = child_view.iter().collect();
403            let actual = negative_expr.propagate_statistics(&parent, &child_refs)?;
404            let expected = Some(vec![Distribution::new_from_interval(
405                after_propagation.clone(),
406            )?]);
407            assert_eq!(actual, expected);
408        }
409
410        Ok(())
411    }
412
413    #[test]
414    fn test_negation_valid_types() -> Result<()> {
415        let negatable_types = [
416            Int8,
417            DataType::Timestamp(TimeUnit::Second, None),
418            DataType::Interval(IntervalUnit::YearMonth),
419        ];
420        for negatable_type in negatable_types {
421            let schema = Schema::new(vec![Field::new("a", negatable_type, true)]);
422            let _expr = negative(col("a", &schema)?, &schema)?;
423        }
424        Ok(())
425    }
426
427    #[test]
428    fn test_negation_invalid_types() -> Result<()> {
429        let schema = Schema::new(vec![Field::new("a", DataType::Utf8, true)]);
430        let expr = negative(col("a", &schema)?, &schema).unwrap_err();
431        matches!(expr, DataFusionError::Plan(_));
432        Ok(())
433    }
434
435    #[test]
436    fn test_fmt_sql() -> Result<()> {
437        let expr = NegativeExpr::new(Arc::new(Column::new("a", 0)));
438        let display_string = expr.to_string();
439        assert_eq!(display_string, "(- a@0)");
440        let sql_string = fmt_sql(&expr).to_string();
441        assert_eq!(sql_string, "(- a)");
442
443        Ok(())
444    }
445}
446
447#[cfg(all(test, feature = "proto"))]
448mod proto_tests {
449    use super::*;
450    use crate::expressions::{Column, col};
451    use crate::proto_test_util::{
452        StubDecoder, StubEncoder, UnreachableDecoder, column_node,
453    };
454    use arrow::datatypes::Field;
455    use datafusion_common::DataFusionError;
456    use datafusion_physical_expr_common::physical_expr::proto_decode::PhysicalExprDecodeCtx;
457    use datafusion_physical_expr_common::physical_expr::proto_encode::PhysicalExprEncodeCtx;
458    use datafusion_proto_models::protobuf::{
459        PhysicalExprNode, PhysicalNegativeNode, physical_expr_node,
460    };
461
462    /// Build a `NegativeExpr` proto node with the given children.
463    fn negative_node(expr: Option<Box<PhysicalExprNode>>) -> PhysicalExprNode {
464        PhysicalExprNode {
465            expr_id: None,
466            expr_type: Some(physical_expr_node::ExprType::Negative(Box::new(
467                PhysicalNegativeNode { expr },
468            ))),
469        }
470    }
471
472    /// A `NegativeExpr` over a column of type Int32.
473    fn negative_fixture() -> NegativeExpr {
474        let schema = Schema::new(vec![Field::new("a", DataType::Int32, true)]);
475        NegativeExpr::new(col("a", &schema).unwrap())
476    }
477
478    #[test]
479    fn try_to_proto_encodes_negative_expr() {
480        let negative = negative_fixture();
481        let encoder = StubEncoder::ok();
482        let ctx = PhysicalExprEncodeCtx::new(&encoder);
483
484        let node = negative
485            .try_to_proto(&ctx)
486            .unwrap()
487            .expect("NegativeExpr should encode to Some(node)");
488
489        assert!(node.expr_id.is_none());
490        let negative_node = match node.expr_type {
491            Some(physical_expr_node::ExprType::Negative(boxed)) => *boxed,
492            other => panic!("expected a NegativeExpr node, got {other:?}"),
493        };
494        assert!(negative_node.expr.is_some());
495    }
496
497    #[test]
498    fn try_to_proto_propagates_expr_encode_error() {
499        let negative = negative_fixture();
500        let encoder = StubEncoder::failing_on(1);
501        let ctx = PhysicalExprEncodeCtx::new(&encoder);
502        let err = negative.try_to_proto(&ctx).unwrap_err();
503        assert!(matches!(err, DataFusionError::Internal(msg) if msg.contains("call 1")));
504    }
505
506    #[test]
507    fn try_from_proto_decodes_negative_expr() {
508        let node = negative_node(Some(Box::new(column_node("a"))));
509        let schema = Schema::empty();
510        let decoder = StubDecoder::ok();
511        let ctx = PhysicalExprDecodeCtx::new(&schema, &decoder);
512
513        let decoded = NegativeExpr::try_from_proto(&node, &ctx).unwrap();
514        let negative = decoded
515            .downcast_ref::<NegativeExpr>()
516            .expect("decoded expr should be a NegativeExpr");
517        assert!(negative.arg().downcast_ref::<Column>().is_some());
518    }
519
520    #[test]
521    fn try_from_proto_rejects_non_negative_node() {
522        let node = column_node("a");
523        let schema = Schema::empty();
524        let decoder = UnreachableDecoder;
525        let ctx = PhysicalExprDecodeCtx::new(&schema, &decoder);
526        let err = NegativeExpr::try_from_proto(&node, &ctx).unwrap_err();
527        assert!(
528            matches!(err, DataFusionError::Internal(msg) if msg.contains("PhysicalExprNode is not a Negative"))
529        );
530    }
531
532    #[test]
533    fn try_from_proto_rejects_missing_expr() {
534        let node = negative_node(None);
535        let schema = Schema::empty();
536        let decoder = UnreachableDecoder;
537        let ctx = PhysicalExprDecodeCtx::new(&schema, &decoder);
538        let err = NegativeExpr::try_from_proto(&node, &ctx).unwrap_err();
539        assert!(
540            matches!(err, DataFusionError::Internal(msg) if msg.contains("NegativeExpr is missing required field 'expr'"))
541        );
542    }
543
544    #[test]
545    fn try_from_proto_propagates_expr_decode_error() {
546        let node = negative_node(Some(Box::new(column_node("a"))));
547        let schema = Schema::empty();
548        let decoder = StubDecoder::failing_on(1);
549        let ctx = PhysicalExprDecodeCtx::new(&schema, &decoder);
550        let err = NegativeExpr::try_from_proto(&node, &ctx).unwrap_err();
551        assert!(matches!(err, DataFusionError::Internal(msg) if msg.contains("call 1")));
552    }
553}
datafusion_physical_expr/expressions/negative.rs

datafusion_physical_expr/expressions/
negative.rs
