datafusion_physical_expr/expressions/
is_null.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//! IS NULL expression
19
20use crate::PhysicalExpr;
21use arrow::{
22    datatypes::{DataType, Schema},
23    record_batch::RecordBatch,
24};
25use datafusion_common::{Result, ScalarValue};
26use datafusion_expr::ColumnarValue;
27use std::hash::Hash;
28use std::sync::Arc;
29
30/// IS NULL expression
31#[derive(Debug, Eq)]
32pub struct IsNullExpr {
33    /// Input expression
34    arg: Arc<dyn PhysicalExpr>,
35}
36
37// Manually derive PartialEq and Hash to work around https://github.com/rust-lang/rust/issues/78808
38impl PartialEq for IsNullExpr {
39    fn eq(&self, other: &Self) -> bool {
40        self.arg.eq(&other.arg)
41    }
42}
43
44impl Hash for IsNullExpr {
45    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
46        self.arg.hash(state);
47    }
48}
49
50impl IsNullExpr {
51    /// Create new not expression
52    pub fn new(arg: Arc<dyn PhysicalExpr>) -> Self {
53        Self { arg }
54    }
55
56    /// Get the input expression
57    pub fn arg(&self) -> &Arc<dyn PhysicalExpr> {
58        &self.arg
59    }
60}
61
62impl std::fmt::Display for IsNullExpr {
63    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
64        write!(f, "{} IS NULL", self.arg)
65    }
66}
67
68impl PhysicalExpr for IsNullExpr {
69    fn data_type(&self, _input_schema: &Schema) -> Result<DataType> {
70        Ok(DataType::Boolean)
71    }
72
73    fn nullable(&self, _input_schema: &Schema) -> Result<bool> {
74        Ok(false)
75    }
76
77    fn evaluate(&self, batch: &RecordBatch) -> Result<ColumnarValue> {
78        let arg = self.arg.evaluate(batch)?;
79        match arg {
80            ColumnarValue::Array(array) => Ok(ColumnarValue::Array(Arc::new(
81                arrow::compute::is_null(&array)?,
82            ))),
83            ColumnarValue::Scalar(scalar) => Ok(ColumnarValue::Scalar(
84                ScalarValue::Boolean(Some(scalar.is_null())),
85            )),
86        }
87    }
88
89    fn children(&self) -> Vec<&Arc<dyn PhysicalExpr>> {
90        vec![&self.arg]
91    }
92
93    fn with_new_children(
94        self: Arc<Self>,
95        children: Vec<Arc<dyn PhysicalExpr>>,
96    ) -> Result<Arc<dyn PhysicalExpr>> {
97        Ok(Arc::new(IsNullExpr::new(Arc::clone(&children[0]))))
98    }
99
100    fn fmt_sql(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
101        self.arg.fmt_sql(f)?;
102        write!(f, " IS NULL")
103    }
104
105    #[cfg(feature = "proto")]
106    fn try_to_proto(
107        &self,
108        ctx: &datafusion_physical_expr_common::physical_expr::proto_encode::PhysicalExprEncodeCtx<'_>,
109    ) -> Result<Option<datafusion_proto_models::protobuf::PhysicalExprNode>> {
110        use datafusion_proto_models::protobuf;
111
112        Ok(Some(protobuf::PhysicalExprNode {
113            expr_id: None,
114            expr_type: Some(protobuf::physical_expr_node::ExprType::IsNullExpr(
115                Box::new(protobuf::PhysicalIsNull {
116                    expr: Some(Box::new(ctx.encode_child(&self.arg)?)),
117                }),
118            )),
119        }))
120    }
121}
122
123#[cfg(feature = "proto")]
124impl IsNullExpr {
125    /// Reconstruct an [`IsNullExpr`] from its protobuf representation.
126    pub fn try_from_proto(
127        node: &datafusion_proto_models::protobuf::PhysicalExprNode,
128        ctx: &datafusion_physical_expr_common::physical_expr::proto_decode::PhysicalExprDecodeCtx<'_>,
129    ) -> Result<Arc<dyn PhysicalExpr>> {
130        use datafusion_physical_expr_common::expect_expr_variant;
131        use datafusion_proto_models::protobuf;
132
133        let node = expect_expr_variant!(
134            node,
135            protobuf::physical_expr_node::ExprType::IsNullExpr,
136            "IsNullExpr",
137        );
138        let expr =
139            ctx.decode_required_expression(node.expr.as_deref(), "IsNullExpr", "expr")?;
140
141        Ok(Arc::new(IsNullExpr::new(expr)))
142    }
143}
144
145/// Create an IS NULL expression
146pub fn is_null(arg: Arc<dyn PhysicalExpr>) -> Result<Arc<dyn PhysicalExpr>> {
147    Ok(Arc::new(IsNullExpr::new(arg)))
148}
149
150#[cfg(test)]
151mod tests {
152    use super::*;
153    use crate::expressions::col;
154    use arrow::array::{
155        Array, BooleanArray, Float64Array, Int32Array, StringArray, UnionArray,
156    };
157    use arrow::buffer::ScalarBuffer;
158    use arrow::datatypes::*;
159    use datafusion_common::cast::as_boolean_array;
160    use datafusion_physical_expr_common::physical_expr::fmt_sql;
161
162    #[test]
163    fn is_null_op() -> Result<()> {
164        let schema = Schema::new(vec![Field::new("a", DataType::Utf8, true)]);
165        let a = StringArray::from(vec![Some("foo"), None]);
166
167        // expression: "a is null"
168        let expr = is_null(col("a", &schema)?).unwrap();
169        let batch = RecordBatch::try_new(Arc::new(schema), vec![Arc::new(a)])?;
170
171        let result = expr
172            .evaluate(&batch)?
173            .into_array(batch.num_rows())
174            .expect("Failed to convert to array");
175        let result =
176            as_boolean_array(&result).expect("failed to downcast to BooleanArray");
177
178        let expected = &BooleanArray::from(vec![false, true]);
179
180        assert_eq!(expected, result);
181
182        Ok(())
183    }
184
185    fn union_fields() -> UnionFields {
186        [
187            (0, Arc::new(Field::new("A", DataType::Int32, true))),
188            (1, Arc::new(Field::new("B", DataType::Float64, true))),
189            (2, Arc::new(Field::new("C", DataType::Utf8, true))),
190        ]
191        .into_iter()
192        .collect()
193    }
194
195    #[test]
196    fn sparse_union_is_null() {
197        // union of [{A=1}, {A=}, {B=1.1}, {B=1.2}, {B=}, {C=}, {C="a"}]
198        let int_array =
199            Int32Array::from(vec![Some(1), None, None, None, None, None, None]);
200        let float_array =
201            Float64Array::from(vec![None, None, Some(1.1), Some(1.2), None, None, None]);
202        let str_array =
203            StringArray::from(vec![None, None, None, None, None, None, Some("a")]);
204        let type_ids = [0, 0, 1, 1, 1, 2, 2]
205            .into_iter()
206            .collect::<ScalarBuffer<i8>>();
207
208        let children = vec![
209            Arc::new(int_array) as Arc<dyn Array>,
210            Arc::new(float_array),
211            Arc::new(str_array),
212        ];
213
214        let array =
215            UnionArray::try_new(union_fields(), type_ids, None, children).unwrap();
216
217        let result = arrow::compute::is_null(&array).unwrap();
218
219        let expected =
220            &BooleanArray::from(vec![false, true, false, false, true, true, false]);
221        assert_eq!(expected, &result);
222    }
223
224    #[test]
225    fn dense_union_is_null() {
226        // union of [{A=1}, {A=}, {B=3.2}, {B=}, {C="a"}, {C=}]
227        let int_array = Int32Array::from(vec![Some(1), None]);
228        let float_array = Float64Array::from(vec![Some(3.2), None]);
229        let str_array = StringArray::from(vec![Some("a"), None]);
230        let type_ids = [0, 0, 1, 1, 2, 2].into_iter().collect::<ScalarBuffer<i8>>();
231        let offsets = [0, 1, 0, 1, 0, 1]
232            .into_iter()
233            .collect::<ScalarBuffer<i32>>();
234
235        let children = vec![
236            Arc::new(int_array) as Arc<dyn Array>,
237            Arc::new(float_array),
238            Arc::new(str_array),
239        ];
240
241        let array =
242            UnionArray::try_new(union_fields(), type_ids, Some(offsets), children)
243                .unwrap();
244
245        let result = arrow::compute::is_null(&array).unwrap();
246
247        let expected = &BooleanArray::from(vec![false, true, false, true, false, true]);
248        assert_eq!(expected, &result);
249    }
250
251    #[test]
252    fn test_fmt_sql() -> Result<()> {
253        let schema = Schema::new(vec![Field::new("a", DataType::Utf8, true)]);
254
255        // expression: "a is null"
256        let expr = is_null(col("a", &schema)?).unwrap();
257        let display_string = expr.to_string();
258        assert_eq!(display_string, "a@0 IS NULL");
259        let sql_string = fmt_sql(expr.as_ref()).to_string();
260        assert_eq!(sql_string, "a IS NULL");
261
262        Ok(())
263    }
264}
265
266#[cfg(all(test, feature = "proto"))]
267mod proto_tests {
268    use super::*;
269    use crate::expressions::{Column, col};
270    use crate::proto_test_util::{
271        StubDecoder, StubEncoder, UnreachableDecoder, column_node,
272    };
273    use arrow::datatypes::Field;
274    use datafusion_common::DataFusionError;
275    use datafusion_physical_expr_common::physical_expr::proto_decode::PhysicalExprDecodeCtx;
276    use datafusion_physical_expr_common::physical_expr::proto_encode::PhysicalExprEncodeCtx;
277    use datafusion_proto_models::protobuf::{
278        PhysicalExprNode, PhysicalIsNull, physical_expr_node,
279    };
280
281    fn is_null_node(expr: Option<Box<PhysicalExprNode>>) -> PhysicalExprNode {
282        PhysicalExprNode {
283            expr_id: None,
284            expr_type: Some(physical_expr_node::ExprType::IsNullExpr(Box::new(
285                PhysicalIsNull { expr },
286            ))),
287        }
288    }
289
290    fn is_null_fixture() -> IsNullExpr {
291        let schema = Schema::new(vec![Field::new("a", DataType::Int64, true)]);
292        IsNullExpr::new(col("a", &schema).unwrap())
293    }
294
295    #[test]
296    fn try_to_proto_encodes_is_null_expr() {
297        let is_null = is_null_fixture();
298        let encoder = StubEncoder::ok();
299        let ctx = PhysicalExprEncodeCtx::new(&encoder);
300
301        let node = is_null
302            .try_to_proto(&ctx)
303            .unwrap()
304            .expect("IsNullExpr should encode to Some(node)");
305
306        assert!(node.expr_id.is_none());
307        let is_null_node = match node.expr_type {
308            Some(physical_expr_node::ExprType::IsNullExpr(boxed)) => *boxed,
309            other => panic!("expected an IsNullExpr node, got {other:?}"),
310        };
311        assert!(is_null_node.expr.is_some());
312    }
313
314    #[test]
315    fn try_to_proto_propagates_expr_encode_error() {
316        let is_null = is_null_fixture();
317        let encoder = StubEncoder::failing_on(1);
318        let ctx = PhysicalExprEncodeCtx::new(&encoder);
319        let err = is_null.try_to_proto(&ctx).unwrap_err();
320        assert!(matches!(err, DataFusionError::Internal(msg) if msg.contains("call 1")));
321    }
322
323    #[test]
324    fn try_from_proto_decodes_is_null_expr() {
325        let node = is_null_node(Some(Box::new(column_node("a"))));
326        let schema = Schema::empty();
327        let decoder = StubDecoder::ok();
328        let ctx = PhysicalExprDecodeCtx::new(&schema, &decoder);
329
330        let decoded = IsNullExpr::try_from_proto(&node, &ctx).unwrap();
331        let is_null = decoded
332            .downcast_ref::<IsNullExpr>()
333            .expect("decoded expr should be an IsNullExpr");
334        assert!(is_null.arg().downcast_ref::<Column>().is_some());
335    }
336
337    #[test]
338    fn try_from_proto_rejects_non_is_null_node() {
339        let node = column_node("a");
340        let schema = Schema::empty();
341        let decoder = UnreachableDecoder;
342        let ctx = PhysicalExprDecodeCtx::new(&schema, &decoder);
343        let err = IsNullExpr::try_from_proto(&node, &ctx).unwrap_err();
344        assert!(
345            matches!(err, DataFusionError::Internal(msg) if msg.contains("PhysicalExprNode is not a IsNullExpr"))
346        );
347    }
348
349    #[test]
350    fn try_from_proto_rejects_missing_expr() {
351        let node = is_null_node(None);
352        let schema = Schema::empty();
353        let decoder = UnreachableDecoder;
354        let ctx = PhysicalExprDecodeCtx::new(&schema, &decoder);
355        let err = IsNullExpr::try_from_proto(&node, &ctx).unwrap_err();
356        assert!(
357            matches!(err, DataFusionError::Internal(msg) if msg.contains("IsNullExpr is missing required field 'expr'"))
358        );
359    }
360
361    #[test]
362    fn try_from_proto_propagates_expr_decode_error() {
363        let node = is_null_node(Some(Box::new(column_node("a"))));
364        let schema = Schema::empty();
365        let decoder = StubDecoder::failing_on(1);
366        let ctx = PhysicalExprDecodeCtx::new(&schema, &decoder);
367        let err = IsNullExpr::try_from_proto(&node, &ctx).unwrap_err();
368        assert!(matches!(err, DataFusionError::Internal(msg) if msg.contains("call 1")));
369    }
370}
datafusion_physical_expr/expressions/is_null.rs

datafusion_physical_expr/expressions/
is_null.rs
