datafusion_physical_expr/expressions/
like.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
18use crate::PhysicalExpr;
19use arrow::datatypes::{DataType, Schema};
20use arrow::record_batch::RecordBatch;
21use datafusion_common::{Result, assert_or_internal_err};
22use datafusion_expr::{ColumnarValue, Operator};
23use datafusion_physical_expr_common::datum::apply_cmp;
24use std::hash::Hash;
25use std::sync::Arc;
26
27// Like expression
28#[derive(Debug, Eq)]
29pub struct LikeExpr {
30    negated: bool,
31    case_insensitive: bool,
32    expr: Arc<dyn PhysicalExpr>,
33    pattern: Arc<dyn PhysicalExpr>,
34}
35
36// Manually derive PartialEq and Hash to work around https://github.com/rust-lang/rust/issues/78808
37impl PartialEq for LikeExpr {
38    fn eq(&self, other: &Self) -> bool {
39        self.negated == other.negated
40            && self.case_insensitive == other.case_insensitive
41            && self.expr.eq(&other.expr)
42            && self.pattern.eq(&other.pattern)
43    }
44}
45
46impl Hash for LikeExpr {
47    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
48        self.negated.hash(state);
49        self.case_insensitive.hash(state);
50        self.expr.hash(state);
51        self.pattern.hash(state);
52    }
53}
54
55impl LikeExpr {
56    pub fn new(
57        negated: bool,
58        case_insensitive: bool,
59        expr: Arc<dyn PhysicalExpr>,
60        pattern: Arc<dyn PhysicalExpr>,
61    ) -> Self {
62        Self {
63            negated,
64            case_insensitive,
65            expr,
66            pattern,
67        }
68    }
69
70    /// Is negated
71    pub fn negated(&self) -> bool {
72        self.negated
73    }
74
75    /// Is case insensitive
76    pub fn case_insensitive(&self) -> bool {
77        self.case_insensitive
78    }
79
80    /// Input expression
81    pub fn expr(&self) -> &Arc<dyn PhysicalExpr> {
82        &self.expr
83    }
84
85    /// Pattern expression
86    pub fn pattern(&self) -> &Arc<dyn PhysicalExpr> {
87        &self.pattern
88    }
89
90    /// Operator name
91    fn op_name(&self) -> &str {
92        match (self.negated, self.case_insensitive) {
93            (false, false) => "LIKE",
94            (true, false) => "NOT LIKE",
95            (false, true) => "ILIKE",
96            (true, true) => "NOT ILIKE",
97        }
98    }
99}
100
101impl std::fmt::Display for LikeExpr {
102    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
103        write!(f, "{} {} {}", self.expr, self.op_name(), self.pattern)
104    }
105}
106
107impl PhysicalExpr for LikeExpr {
108    fn data_type(&self, _input_schema: &Schema) -> Result<DataType> {
109        Ok(DataType::Boolean)
110    }
111
112    fn nullable(&self, input_schema: &Schema) -> Result<bool> {
113        Ok(self.expr.nullable(input_schema)? || self.pattern.nullable(input_schema)?)
114    }
115
116    fn evaluate(&self, batch: &RecordBatch) -> Result<ColumnarValue> {
117        let lhs = self.expr.evaluate(batch)?;
118        let rhs = self.pattern.evaluate(batch)?;
119        match (self.negated, self.case_insensitive) {
120            (false, false) => apply_cmp(Operator::LikeMatch, &lhs, &rhs),
121            (false, true) => apply_cmp(Operator::ILikeMatch, &lhs, &rhs),
122            (true, false) => apply_cmp(Operator::NotLikeMatch, &lhs, &rhs),
123            (true, true) => apply_cmp(Operator::NotILikeMatch, &lhs, &rhs),
124        }
125    }
126
127    fn children(&self) -> Vec<&Arc<dyn PhysicalExpr>> {
128        vec![&self.expr, &self.pattern]
129    }
130
131    fn with_new_children(
132        self: Arc<Self>,
133        children: Vec<Arc<dyn PhysicalExpr>>,
134    ) -> Result<Arc<dyn PhysicalExpr>> {
135        Ok(Arc::new(LikeExpr::new(
136            self.negated,
137            self.case_insensitive,
138            Arc::clone(&children[0]),
139            Arc::clone(&children[1]),
140        )))
141    }
142
143    fn fmt_sql(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
144        self.expr.fmt_sql(f)?;
145        write!(f, " {} ", self.op_name())?;
146        self.pattern.fmt_sql(f)
147    }
148
149    #[cfg(feature = "proto")]
150    fn try_to_proto(
151        &self,
152        ctx: &datafusion_physical_expr_common::physical_expr::proto_encode::PhysicalExprEncodeCtx<'_>,
153    ) -> Result<Option<datafusion_proto_models::protobuf::PhysicalExprNode>> {
154        use datafusion_proto_models::protobuf;
155
156        Ok(Some(protobuf::PhysicalExprNode {
157            expr_id: None,
158            expr_type: Some(protobuf::physical_expr_node::ExprType::LikeExpr(Box::new(
159                protobuf::PhysicalLikeExprNode {
160                    negated: self.negated,
161                    case_insensitive: self.case_insensitive,
162                    expr: Some(Box::new(ctx.encode_child(&self.expr)?)),
163                    pattern: Some(Box::new(ctx.encode_child(&self.pattern)?)),
164                },
165            ))),
166        }))
167    }
168}
169
170#[cfg(feature = "proto")]
171impl LikeExpr {
172    /// Reconstruct a [`LikeExpr`] from its protobuf representation.
173    ///
174    /// Takes the whole [`PhysicalExprNode`] so the decode signature matches
175    /// other migrated expressions and can inspect outer-node metadata if
176    /// needed in the future.
177    ///
178    /// [`PhysicalExprNode`]: datafusion_proto_models::protobuf::PhysicalExprNode
179    pub fn try_from_proto(
180        node: &datafusion_proto_models::protobuf::PhysicalExprNode,
181        ctx: &datafusion_physical_expr_common::physical_expr::proto_decode::PhysicalExprDecodeCtx<'_>,
182    ) -> Result<Arc<dyn PhysicalExpr>> {
183        use datafusion_physical_expr_common::expect_expr_variant;
184        use datafusion_proto_models::protobuf;
185
186        let like_expr = expect_expr_variant!(
187            node,
188            protobuf::physical_expr_node::ExprType::LikeExpr,
189            "LikeExpr",
190        );
191
192        Ok(Arc::new(LikeExpr::new(
193            like_expr.negated,
194            like_expr.case_insensitive,
195            ctx.decode_required_expression(
196                like_expr.expr.as_deref(),
197                "LikeExpr",
198                "expr",
199            )?,
200            ctx.decode_required_expression(
201                like_expr.pattern.as_deref(),
202                "LikeExpr",
203                "pattern",
204            )?,
205        )))
206    }
207}
208
209/// used for optimize Dictionary like
210fn can_like_type(from_type: &DataType) -> bool {
211    match from_type {
212        DataType::Dictionary(_, inner_type_from) => **inner_type_from == DataType::Utf8,
213        _ => false,
214    }
215}
216
217/// Create a like expression, erroring if the argument types are not compatible.
218pub fn like(
219    negated: bool,
220    case_insensitive: bool,
221    expr: Arc<dyn PhysicalExpr>,
222    pattern: Arc<dyn PhysicalExpr>,
223    input_schema: &Schema,
224) -> Result<Arc<dyn PhysicalExpr>> {
225    let expr_type = &expr.data_type(input_schema)?;
226    let pattern_type = &pattern.data_type(input_schema)?;
227    assert_or_internal_err!(
228        expr_type.eq(pattern_type) || can_like_type(expr_type),
229        "The type of {expr_type} AND {pattern_type} of like physical should be same"
230    );
231    Ok(Arc::new(LikeExpr::new(
232        negated,
233        case_insensitive,
234        expr,
235        pattern,
236    )))
237}
238
239#[cfg(test)]
240mod test {
241    use super::*;
242    use crate::expressions::col;
243    use arrow::array::*;
244    use arrow::datatypes::Field;
245    use datafusion_common::cast::as_boolean_array;
246    use datafusion_physical_expr_common::physical_expr::fmt_sql;
247
248    macro_rules! test_like {
249        ($A_VEC:expr, $B_VEC:expr, $VEC:expr, $NULLABLE: expr, $NEGATED:expr, $CASE_INSENSITIVE:expr,) => {{
250            let schema = Schema::new(vec![
251                Field::new("a", DataType::Utf8, $NULLABLE),
252                Field::new("b", DataType::Utf8, $NULLABLE),
253            ]);
254            let a = StringArray::from($A_VEC);
255            let b = StringArray::from($B_VEC);
256
257            let expression = like(
258                $NEGATED,
259                $CASE_INSENSITIVE,
260                col("a", &schema)?,
261                col("b", &schema)?,
262                &schema,
263            )?;
264            let batch = RecordBatch::try_new(
265                Arc::new(schema.clone()),
266                vec![Arc::new(a), Arc::new(b)],
267            )?;
268
269            // compute
270            let result = expression
271                .evaluate(&batch)?
272                .into_array(batch.num_rows())
273                .expect("Failed to convert to array");
274            let result =
275                as_boolean_array(&result).expect("failed to downcast to BooleanArray");
276            let expected = &BooleanArray::from($VEC);
277            assert_eq!(expected, result);
278        }};
279    }
280
281    #[test]
282    fn like_op() -> Result<()> {
283        test_like!(
284            vec!["hello world", "world"],
285            vec!["%hello%", "%hello%"],
286            vec![true, false],
287            false,
288            false,
289            false,
290        ); // like
291        test_like!(
292            vec![Some("hello world"), None, Some("world")],
293            vec![Some("%hello%"), None, Some("%hello%")],
294            vec![Some(false), None, Some(true)],
295            true,
296            true,
297            false,
298        ); // not like
299        test_like!(
300            vec!["hello world", "world"],
301            vec!["%helLo%", "%helLo%"],
302            vec![true, false],
303            false,
304            false,
305            true,
306        ); // ilike
307        test_like!(
308            vec![Some("hello world"), None, Some("world")],
309            vec![Some("%helLo%"), None, Some("%helLo%")],
310            vec![Some(false), None, Some(true)],
311            true,
312            true,
313            true,
314        ); // not ilike
315
316        Ok(())
317    }
318
319    #[test]
320    fn test_fmt_sql() -> Result<()> {
321        let schema = Schema::new(vec![
322            Field::new("a", DataType::Utf8, false),
323            Field::new("b", DataType::Utf8, false),
324        ]);
325
326        let expr = like(
327            false,
328            false,
329            col("a", &schema)?,
330            col("b", &schema)?,
331            &schema,
332        )?;
333
334        // Display format
335        let display_string = expr.to_string();
336        assert_eq!(display_string, "a@0 LIKE b@1");
337
338        // fmt_sql format
339        let sql_string = fmt_sql(expr.as_ref()).to_string();
340        assert_eq!(sql_string, "a LIKE b");
341
342        Ok(())
343    }
344}
345
346/// Tests for the `try_to_proto` / `try_from_proto` hooks.
347#[cfg(all(test, feature = "proto"))]
348mod proto_tests {
349    use super::*;
350    use crate::expressions::{Column, col};
351    use crate::proto_test_util::{
352        StubDecoder, StubEncoder, UnreachableDecoder, column_node,
353    };
354    use arrow::datatypes::Field;
355    use datafusion_common::DataFusionError;
356    use datafusion_physical_expr_common::physical_expr::proto_decode::PhysicalExprDecodeCtx;
357    use datafusion_physical_expr_common::physical_expr::proto_encode::PhysicalExprEncodeCtx;
358    use datafusion_proto_models::protobuf::{
359        PhysicalExprNode, PhysicalLikeExprNode, physical_expr_node,
360    };
361
362    /// Build a `LikeExpr` proto node with the given children.
363    fn like_node(
364        negated: bool,
365        case_insensitive: bool,
366        expr: Option<Box<PhysicalExprNode>>,
367        pattern: Option<Box<PhysicalExprNode>>,
368    ) -> PhysicalExprNode {
369        PhysicalExprNode {
370            expr_id: None,
371            expr_type: Some(physical_expr_node::ExprType::LikeExpr(Box::new(
372                PhysicalLikeExprNode {
373                    negated,
374                    case_insensitive,
375                    expr,
376                    pattern,
377                },
378            ))),
379        }
380    }
381
382    /// A `LikeExpr` over two `Utf8` columns with both flags set, so the
383    /// `negated` / `case_insensitive` wiring is actually exercised.
384    fn like_fixture() -> LikeExpr {
385        let schema = Schema::new(vec![
386            Field::new("a", DataType::Utf8, false),
387            Field::new("b", DataType::Utf8, false),
388        ]);
389        LikeExpr::new(
390            true,
391            true,
392            col("a", &schema).unwrap(),
393            col("b", &schema).unwrap(),
394        )
395    }
396
397    #[test]
398    fn try_to_proto_encodes_like_expr() {
399        let like = like_fixture();
400        let encoder = StubEncoder::ok();
401        let ctx = PhysicalExprEncodeCtx::new(&encoder);
402
403        let node = like
404            .try_to_proto(&ctx)
405            .unwrap()
406            .expect("LikeExpr should encode to Some(node)");
407
408        // Built-in exprs never set expr_id; only dynamic filters do.
409        assert!(node.expr_id.is_none());
410        let like_node = match node.expr_type {
411            Some(physical_expr_node::ExprType::LikeExpr(boxed)) => *boxed,
412            other => panic!("expected a LikeExpr node, got {other:?}"),
413        };
414        assert!(like_node.negated);
415        assert!(like_node.case_insensitive);
416        assert!(like_node.expr.is_some());
417        assert!(like_node.pattern.is_some());
418    }
419
420    #[test]
421    fn try_to_proto_propagates_expr_encode_error() {
422        let like = like_fixture();
423        let encoder = StubEncoder::failing_on(1);
424        let ctx = PhysicalExprEncodeCtx::new(&encoder);
425        let err = like.try_to_proto(&ctx).unwrap_err();
426        assert!(matches!(err, DataFusionError::Internal(msg) if msg.contains("call 1")));
427    }
428
429    #[test]
430    fn try_to_proto_propagates_pattern_encode_error() {
431        let like = like_fixture();
432        let encoder = StubEncoder::failing_on(2);
433        let ctx = PhysicalExprEncodeCtx::new(&encoder);
434        let err = like.try_to_proto(&ctx).unwrap_err();
435        assert!(matches!(err, DataFusionError::Internal(msg) if msg.contains("call 2")));
436    }
437
438    #[test]
439    fn try_from_proto_decodes_like_expr() {
440        let node = like_node(
441            true,
442            true,
443            Some(Box::new(column_node("a"))),
444            Some(Box::new(column_node("b"))),
445        );
446        let schema = Schema::empty();
447        let decoder = StubDecoder::ok();
448        let ctx = PhysicalExprDecodeCtx::new(&schema, &decoder);
449
450        let decoded = LikeExpr::try_from_proto(&node, &ctx).unwrap();
451        let like = decoded
452            .downcast_ref::<LikeExpr>()
453            .expect("decoded expr should be a LikeExpr");
454        assert!(like.negated());
455        assert!(like.case_insensitive());
456        assert!(like.expr().downcast_ref::<Column>().is_some());
457        assert!(like.pattern().downcast_ref::<Column>().is_some());
458    }
459
460    #[test]
461    fn try_from_proto_rejects_non_like_node() {
462        let node = column_node("a");
463        let schema = Schema::empty();
464        let decoder = UnreachableDecoder;
465        let ctx = PhysicalExprDecodeCtx::new(&schema, &decoder);
466        let err = LikeExpr::try_from_proto(&node, &ctx).unwrap_err();
467        assert!(matches!(
468            err,
469            DataFusionError::Internal(msg) if msg.contains("PhysicalExprNode is not a LikeExpr")
470        ));
471    }
472
473    #[test]
474    fn try_from_proto_rejects_missing_expr() {
475        let node = like_node(false, false, None, Some(Box::new(column_node("b"))));
476        let schema = Schema::empty();
477        let decoder = UnreachableDecoder;
478        let ctx = PhysicalExprDecodeCtx::new(&schema, &decoder);
479        let err = LikeExpr::try_from_proto(&node, &ctx).unwrap_err();
480        assert!(matches!(
481            err,
482            DataFusionError::Internal(msg) if msg.contains("LikeExpr is missing required field 'expr'")
483        ));
484    }
485
486    #[test]
487    fn try_from_proto_rejects_missing_pattern() {
488        let node = like_node(false, false, Some(Box::new(column_node("a"))), None);
489        let schema = Schema::empty();
490        // `expr` is present, so it is decoded before the missing-`pattern`
491        // check fires; use a decoder that succeeds for that first child.
492        let decoder = StubDecoder::ok();
493        let ctx = PhysicalExprDecodeCtx::new(&schema, &decoder);
494        let err = LikeExpr::try_from_proto(&node, &ctx).unwrap_err();
495        assert!(matches!(
496            err,
497            DataFusionError::Internal(msg) if msg.contains("LikeExpr is missing required field 'pattern'")
498        ));
499    }
500
501    #[test]
502    fn try_from_proto_propagates_expr_decode_error() {
503        let node = like_node(
504            false,
505            false,
506            Some(Box::new(column_node("a"))),
507            Some(Box::new(column_node("b"))),
508        );
509        let schema = Schema::empty();
510        let decoder = StubDecoder::failing_on(1);
511        let ctx = PhysicalExprDecodeCtx::new(&schema, &decoder);
512        let err = LikeExpr::try_from_proto(&node, &ctx).unwrap_err();
513        assert!(matches!(err, DataFusionError::Internal(msg) if msg.contains("call 1")));
514    }
515
516    #[test]
517    fn try_from_proto_propagates_pattern_decode_error() {
518        let node = like_node(
519            false,
520            false,
521            Some(Box::new(column_node("a"))),
522            Some(Box::new(column_node("b"))),
523        );
524        let schema = Schema::empty();
525        let decoder = StubDecoder::failing_on(2);
526        let ctx = PhysicalExprDecodeCtx::new(&schema, &decoder);
527        let err = LikeExpr::try_from_proto(&node, &ctx).unwrap_err();
528        assert!(matches!(err, DataFusionError::Internal(msg) if msg.contains("call 2")));
529    }
530}
datafusion_physical_expr/expressions/like.rs

datafusion_physical_expr/expressions/
like.rs
