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::any::Any;
21use std::hash::Hash;
22use std::sync::Arc;
23
24use crate::PhysicalExpr;
25
26use arrow::{
27    compute::kernels::numeric::neg_wrapping,
28    datatypes::{DataType, Schema},
29    record_batch::RecordBatch,
30};
31use datafusion_common::{internal_err, plan_err, Result};
32use datafusion_expr::interval_arithmetic::Interval;
33use datafusion_expr::sort_properties::ExprProperties;
34use datafusion_expr::statistics::Distribution::{
35    self, Bernoulli, Exponential, Gaussian, Generic, Uniform,
36};
37use datafusion_expr::{
38    type_coercion::{is_interval, is_null, is_signed_numeric, is_timestamp},
39    ColumnarValue,
40};
41
42/// Negative expression
43#[derive(Debug, Eq)]
44pub struct NegativeExpr {
45    /// Input expression
46    arg: Arc<dyn PhysicalExpr>,
47}
48
49// Manually derive PartialEq and Hash to work around https://github.com/rust-lang/rust/issues/78808
50impl PartialEq for NegativeExpr {
51    fn eq(&self, other: &Self) -> bool {
52        self.arg.eq(&other.arg)
53    }
54}
55
56impl Hash for NegativeExpr {
57    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
58        self.arg.hash(state);
59    }
60}
61
62impl NegativeExpr {
63    /// Create new not expression
64    pub fn new(arg: Arc<dyn PhysicalExpr>) -> Self {
65        Self { arg }
66    }
67
68    /// Get the input expression
69    pub fn arg(&self) -> &Arc<dyn PhysicalExpr> {
70        &self.arg
71    }
72}
73
74impl std::fmt::Display for NegativeExpr {
75    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
76        write!(f, "(- {})", self.arg)
77    }
78}
79
80impl PhysicalExpr for NegativeExpr {
81    /// Return a reference to Any that can be used for downcasting
82    fn as_any(&self) -> &dyn Any {
83        self
84    }
85
86    fn data_type(&self, input_schema: &Schema) -> Result<DataType> {
87        self.arg.data_type(input_schema)
88    }
89
90    fn nullable(&self, input_schema: &Schema) -> Result<bool> {
91        self.arg.nullable(input_schema)
92    }
93
94    fn evaluate(&self, batch: &RecordBatch) -> Result<ColumnarValue> {
95        match self.arg.evaluate(batch)? {
96            ColumnarValue::Array(array) => {
97                let result = neg_wrapping(array.as_ref())?;
98                Ok(ColumnarValue::Array(result))
99            }
100            ColumnarValue::Scalar(scalar) => {
101                Ok(ColumnarValue::Scalar(scalar.arithmetic_negate()?))
102            }
103        }
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    fn evaluate_statistics(&self, children: &[&Distribution]) -> Result<Distribution> {
139        match children[0] {
140            Uniform(u) => Distribution::new_uniform(u.range().arithmetic_negate()?),
141            Exponential(e) => Distribution::new_exponential(
142                e.rate().clone(),
143                e.offset().arithmetic_negate()?,
144                !e.positive_tail(),
145            ),
146            Gaussian(g) => Distribution::new_gaussian(
147                g.mean().arithmetic_negate()?,
148                g.variance().clone(),
149            ),
150            Bernoulli(_) => {
151                internal_err!("NegativeExpr cannot operate on Boolean datatypes")
152            }
153            Generic(u) => Distribution::new_generic(
154                u.mean().arithmetic_negate()?,
155                u.median().arithmetic_negate()?,
156                u.variance().clone(),
157                u.range().arithmetic_negate()?,
158            ),
159        }
160    }
161
162    /// The ordering of a [`NegativeExpr`] is simply the reverse of its child.
163    fn get_properties(&self, children: &[ExprProperties]) -> Result<ExprProperties> {
164        Ok(ExprProperties {
165            sort_properties: -children[0].sort_properties,
166            range: children[0].range.clone().arithmetic_negate()?,
167            preserves_lex_ordering: false,
168        })
169    }
170}
171
172/// Creates a unary expression NEGATIVE
173///
174/// # Errors
175///
176/// This function errors when the argument's type is not signed numeric
177pub fn negative(
178    arg: Arc<dyn PhysicalExpr>,
179    input_schema: &Schema,
180) -> Result<Arc<dyn PhysicalExpr>> {
181    let data_type = arg.data_type(input_schema)?;
182    if is_null(&data_type) {
183        Ok(arg)
184    } else if !is_signed_numeric(&data_type)
185        && !is_interval(&data_type)
186        && !is_timestamp(&data_type)
187    {
188        plan_err!("Negation only supports numeric, interval and timestamp types")
189    } else {
190        Ok(Arc::new(NegativeExpr::new(arg)))
191    }
192}
193
194#[cfg(test)]
195mod tests {
196    use super::*;
197    use crate::expressions::{col, Column};
198
199    use arrow::array::*;
200    use arrow::datatypes::DataType::{Float32, Float64, Int16, Int32, Int64, Int8};
201    use arrow::datatypes::*;
202    use datafusion_common::cast::as_primitive_array;
203    use datafusion_common::{DataFusionError, ScalarValue};
204
205    use paste::paste;
206
207    macro_rules! test_array_negative_op {
208        ($DATA_TY:tt, $($VALUE:expr),*   ) => {
209            let schema = Schema::new(vec![Field::new("a", DataType::$DATA_TY, true)]);
210            let expr = negative(col("a", &schema)?, &schema)?;
211            assert_eq!(expr.data_type(&schema)?, DataType::$DATA_TY);
212            assert!(expr.nullable(&schema)?);
213            let mut arr = Vec::new();
214            let mut arr_expected = Vec::new();
215            $(
216                arr.push(Some($VALUE));
217                arr_expected.push(Some(-$VALUE));
218            )+
219            arr.push(None);
220            arr_expected.push(None);
221            let input = paste!{[<$DATA_TY Array>]::from(arr)};
222            let expected = &paste!{[<$DATA_TY Array>]::from(arr_expected)};
223            let batch =
224                RecordBatch::try_new(Arc::new(schema.clone()), vec![Arc::new(input)])?;
225            let result = expr.evaluate(&batch)?.into_array(batch.num_rows()).expect("Failed to convert to array");
226            let result =
227                as_primitive_array(&result).expect(format!("failed to downcast to {:?}Array", $DATA_TY).as_str());
228            assert_eq!(result, expected);
229        };
230    }
231
232    #[test]
233    fn array_negative_op() -> Result<()> {
234        test_array_negative_op!(Int8, 2i8, 1i8);
235        test_array_negative_op!(Int16, 234i16, 123i16);
236        test_array_negative_op!(Int32, 2345i32, 1234i32);
237        test_array_negative_op!(Int64, 23456i64, 12345i64);
238        test_array_negative_op!(Float32, 2345.0f32, 1234.0f32);
239        test_array_negative_op!(Float64, 23456.0f64, 12345.0f64);
240        Ok(())
241    }
242
243    #[test]
244    fn test_evaluate_bounds() -> Result<()> {
245        let negative_expr = NegativeExpr::new(Arc::new(Column::new("a", 0)));
246        let child_interval = Interval::make(Some(-2), Some(1))?;
247        let negative_expr_interval = Interval::make(Some(-1), Some(2))?;
248        assert_eq!(
249            negative_expr.evaluate_bounds(&[&child_interval])?,
250            negative_expr_interval
251        );
252        Ok(())
253    }
254
255    #[test]
256    fn test_evaluate_statistics() -> Result<()> {
257        let negative_expr = NegativeExpr::new(Arc::new(Column::new("a", 0)));
258
259        // Uniform
260        assert_eq!(
261            negative_expr.evaluate_statistics(&[&Distribution::new_uniform(
262                Interval::make(Some(-2.), Some(3.))?
263            )?])?,
264            Distribution::new_uniform(Interval::make(Some(-3.), Some(2.))?)?
265        );
266
267        // Bernoulli
268        assert!(negative_expr
269            .evaluate_statistics(&[&Distribution::new_bernoulli(ScalarValue::from(
270                0.75
271            ))?])
272            .is_err());
273
274        // Exponential
275        assert_eq!(
276            negative_expr.evaluate_statistics(&[&Distribution::new_exponential(
277                ScalarValue::from(1.),
278                ScalarValue::from(1.),
279                true
280            )?])?,
281            Distribution::new_exponential(
282                ScalarValue::from(1.),
283                ScalarValue::from(-1.),
284                false
285            )?
286        );
287
288        // Gaussian
289        assert_eq!(
290            negative_expr.evaluate_statistics(&[&Distribution::new_gaussian(
291                ScalarValue::from(15),
292                ScalarValue::from(225),
293            )?])?,
294            Distribution::new_gaussian(ScalarValue::from(-15), ScalarValue::from(225),)?
295        );
296
297        // Unknown
298        assert_eq!(
299            negative_expr.evaluate_statistics(&[&Distribution::new_generic(
300                ScalarValue::from(15),
301                ScalarValue::from(15),
302                ScalarValue::from(10),
303                Interval::make(Some(10), Some(20))?
304            )?])?,
305            Distribution::new_generic(
306                ScalarValue::from(-15),
307                ScalarValue::from(-15),
308                ScalarValue::from(10),
309                Interval::make(Some(-20), Some(-10))?
310            )?
311        );
312
313        Ok(())
314    }
315
316    #[test]
317    fn test_propagate_constraints() -> Result<()> {
318        let negative_expr = NegativeExpr::new(Arc::new(Column::new("a", 0)));
319        let original_child_interval = Interval::make(Some(-2), Some(3))?;
320        let negative_expr_interval = Interval::make(Some(0), Some(4))?;
321        let after_propagation = Some(vec![Interval::make(Some(-2), Some(0))?]);
322        assert_eq!(
323            negative_expr.propagate_constraints(
324                &negative_expr_interval,
325                &[&original_child_interval]
326            )?,
327            after_propagation
328        );
329        Ok(())
330    }
331
332    #[test]
333    fn test_propagate_statistics_range_holders() -> Result<()> {
334        let negative_expr = NegativeExpr::new(Arc::new(Column::new("a", 0)));
335        let original_child_interval = Interval::make(Some(-2), Some(3))?;
336        let after_propagation = Interval::make(Some(-2), Some(0))?;
337
338        let parent = Distribution::new_uniform(Interval::make(Some(0), Some(4))?)?;
339        let children: Vec<Vec<Distribution>> = vec![
340            vec![Distribution::new_uniform(original_child_interval.clone())?],
341            vec![Distribution::new_generic(
342                ScalarValue::from(0),
343                ScalarValue::from(0),
344                ScalarValue::Int32(None),
345                original_child_interval.clone(),
346            )?],
347        ];
348
349        for child_view in children {
350            let child_refs: Vec<_> = child_view.iter().collect();
351            let actual = negative_expr.propagate_statistics(&parent, &child_refs)?;
352            let expected = Some(vec![Distribution::new_from_interval(
353                after_propagation.clone(),
354            )?]);
355            assert_eq!(actual, expected);
356        }
357
358        Ok(())
359    }
360
361    #[test]
362    fn test_negation_valid_types() -> Result<()> {
363        let negatable_types = [
364            Int8,
365            DataType::Timestamp(TimeUnit::Second, None),
366            DataType::Interval(IntervalUnit::YearMonth),
367        ];
368        for negatable_type in negatable_types {
369            let schema = Schema::new(vec![Field::new("a", negatable_type, true)]);
370            let _expr = negative(col("a", &schema)?, &schema)?;
371        }
372        Ok(())
373    }
374
375    #[test]
376    fn test_negation_invalid_types() -> Result<()> {
377        let schema = Schema::new(vec![Field::new("a", DataType::Utf8, true)]);
378        let expr = negative(col("a", &schema)?, &schema).unwrap_err();
379        matches!(expr, DataFusionError::Plan(_));
380        Ok(())
381    }
382}
datafusion_physical_expr/expressions/negative.rs

datafusion_physical_expr/expressions/
negative.rs
