datafusion-physical-expr 55.0.0

// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

use crate::PhysicalExpr;
use arrow::datatypes::{DataType, Schema};
use arrow::record_batch::RecordBatch;
use datafusion_common::{Result, assert_or_internal_err};
use datafusion_expr::{ColumnarValue, Operator};
use datafusion_physical_expr_common::datum::apply_cmp;
use std::hash::Hash;
use std::sync::Arc;

// Like expression
#[derive(Debug, Eq)]
pub struct LikeExpr {
    negated: bool,
    case_insensitive: bool,
    expr: Arc<dyn PhysicalExpr>,
    pattern: Arc<dyn PhysicalExpr>,
}

// Manually derive PartialEq and Hash to work around https://github.com/rust-lang/rust/issues/78808
impl PartialEq for LikeExpr {
    fn eq(&self, other: &Self) -> bool {
        self.negated == other.negated
            && self.case_insensitive == other.case_insensitive
            && self.expr.eq(&other.expr)
            && self.pattern.eq(&other.pattern)
    }
}

impl Hash for LikeExpr {
    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
        self.negated.hash(state);
        self.case_insensitive.hash(state);
        self.expr.hash(state);
        self.pattern.hash(state);
    }
}

impl LikeExpr {
    pub fn new(
        negated: bool,
        case_insensitive: bool,
        expr: Arc<dyn PhysicalExpr>,
        pattern: Arc<dyn PhysicalExpr>,
    ) -> Self {
        Self {
            negated,
            case_insensitive,
            expr,
            pattern,
        }
    }

    /// Is negated
    pub fn negated(&self) -> bool {
        self.negated
    }

    /// Is case insensitive
    pub fn case_insensitive(&self) -> bool {
        self.case_insensitive
    }

    /// Input expression
    pub fn expr(&self) -> &Arc<dyn PhysicalExpr> {
        &self.expr
    }

    /// Pattern expression
    pub fn pattern(&self) -> &Arc<dyn PhysicalExpr> {
        &self.pattern
    }

    /// Operator name
    fn op_name(&self) -> &str {
        match (self.negated, self.case_insensitive) {
            (false, false) => "LIKE",
            (true, false) => "NOT LIKE",
            (false, true) => "ILIKE",
            (true, true) => "NOT ILIKE",
        }
    }
}

impl std::fmt::Display for LikeExpr {
    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
        write!(f, "{} {} {}", self.expr, self.op_name(), self.pattern)
    }
}

impl PhysicalExpr for LikeExpr {
    fn data_type(&self, _input_schema: &Schema) -> Result<DataType> {
        Ok(DataType::Boolean)
    }

    fn nullable(&self, input_schema: &Schema) -> Result<bool> {
        Ok(self.expr.nullable(input_schema)? || self.pattern.nullable(input_schema)?)
    }

    fn evaluate(&self, batch: &RecordBatch) -> Result<ColumnarValue> {
        let lhs = self.expr.evaluate(batch)?;
        let rhs = self.pattern.evaluate(batch)?;
        match (self.negated, self.case_insensitive) {
            (false, false) => apply_cmp(Operator::LikeMatch, &lhs, &rhs),
            (false, true) => apply_cmp(Operator::ILikeMatch, &lhs, &rhs),
            (true, false) => apply_cmp(Operator::NotLikeMatch, &lhs, &rhs),
            (true, true) => apply_cmp(Operator::NotILikeMatch, &lhs, &rhs),
        }
    }

    fn children(&self) -> Vec<&Arc<dyn PhysicalExpr>> {
        vec![&self.expr, &self.pattern]
    }

    fn with_new_children(
        self: Arc<Self>,
        children: Vec<Arc<dyn PhysicalExpr>>,
    ) -> Result<Arc<dyn PhysicalExpr>> {
        Ok(Arc::new(LikeExpr::new(
            self.negated,
            self.case_insensitive,
            Arc::clone(&children[0]),
            Arc::clone(&children[1]),
        )))
    }

    fn fmt_sql(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        self.expr.fmt_sql(f)?;
        write!(f, " {} ", self.op_name())?;
        self.pattern.fmt_sql(f)
    }

    #[cfg(feature = "proto")]
    fn try_to_proto(
        &self,
        ctx: &datafusion_physical_expr_common::physical_expr::proto_encode::PhysicalExprEncodeCtx<'_>,
    ) -> Result<Option<datafusion_proto_models::protobuf::PhysicalExprNode>> {
        use datafusion_proto_models::protobuf;

        Ok(Some(protobuf::PhysicalExprNode {
            expr_id: None,
            expr_type: Some(protobuf::physical_expr_node::ExprType::LikeExpr(Box::new(
                protobuf::PhysicalLikeExprNode {
                    negated: self.negated,
                    case_insensitive: self.case_insensitive,
                    expr: Some(Box::new(ctx.encode_child(&self.expr)?)),
                    pattern: Some(Box::new(ctx.encode_child(&self.pattern)?)),
                },
            ))),
        }))
    }
}

#[cfg(feature = "proto")]
impl LikeExpr {
    /// Reconstruct a [`LikeExpr`] from its protobuf representation.
    ///
    /// Takes the whole [`PhysicalExprNode`] so the decode signature matches
    /// other migrated expressions and can inspect outer-node metadata if
    /// needed in the future.
    ///
    /// [`PhysicalExprNode`]: datafusion_proto_models::protobuf::PhysicalExprNode
    pub fn try_from_proto(
        node: &datafusion_proto_models::protobuf::PhysicalExprNode,
        ctx: &datafusion_physical_expr_common::physical_expr::proto_decode::PhysicalExprDecodeCtx<'_>,
    ) -> Result<Arc<dyn PhysicalExpr>> {
        use datafusion_physical_expr_common::expect_expr_variant;
        use datafusion_proto_models::protobuf;

        let like_expr = expect_expr_variant!(
            node,
            protobuf::physical_expr_node::ExprType::LikeExpr,
            "LikeExpr",
        );

        Ok(Arc::new(LikeExpr::new(
            like_expr.negated,
            like_expr.case_insensitive,
            ctx.decode_required_expression(
                like_expr.expr.as_deref(),
                "LikeExpr",
                "expr",
            )?,
            ctx.decode_required_expression(
                like_expr.pattern.as_deref(),
                "LikeExpr",
                "pattern",
            )?,
        )))
    }
}

/// used for optimize Dictionary like
fn can_like_type(from_type: &DataType) -> bool {
    match from_type {
        DataType::Dictionary(_, inner_type_from) => **inner_type_from == DataType::Utf8,
        _ => false,
    }
}

/// Create a like expression, erroring if the argument types are not compatible.
pub fn like(
    negated: bool,
    case_insensitive: bool,
    expr: Arc<dyn PhysicalExpr>,
    pattern: Arc<dyn PhysicalExpr>,
    input_schema: &Schema,
) -> Result<Arc<dyn PhysicalExpr>> {
    let expr_type = &expr.data_type(input_schema)?;
    let pattern_type = &pattern.data_type(input_schema)?;
    assert_or_internal_err!(
        expr_type.eq(pattern_type) || can_like_type(expr_type),
        "The type of {expr_type} AND {pattern_type} of like physical should be same"
    );
    Ok(Arc::new(LikeExpr::new(
        negated,
        case_insensitive,
        expr,
        pattern,
    )))
}

#[cfg(test)]
mod test {
    use super::*;
    use crate::expressions::col;
    use arrow::array::*;
    use arrow::datatypes::Field;
    use datafusion_common::cast::as_boolean_array;
    use datafusion_physical_expr_common::physical_expr::fmt_sql;

    macro_rules! test_like {
        ($A_VEC:expr, $B_VEC:expr, $VEC:expr, $NULLABLE: expr, $NEGATED:expr, $CASE_INSENSITIVE:expr,) => {{
            let schema = Schema::new(vec![
                Field::new("a", DataType::Utf8, $NULLABLE),
                Field::new("b", DataType::Utf8, $NULLABLE),
            ]);
            let a = StringArray::from($A_VEC);
            let b = StringArray::from($B_VEC);

            let expression = like(
                $NEGATED,
                $CASE_INSENSITIVE,
                col("a", &schema)?,
                col("b", &schema)?,
                &schema,
            )?;
            let batch = RecordBatch::try_new(
                Arc::new(schema.clone()),
                vec![Arc::new(a), Arc::new(b)],
            )?;

            // compute
            let result = expression
                .evaluate(&batch)?
                .into_array(batch.num_rows())
                .expect("Failed to convert to array");
            let result =
                as_boolean_array(&result).expect("failed to downcast to BooleanArray");
            let expected = &BooleanArray::from($VEC);
            assert_eq!(expected, result);
        }};
    }

    #[test]
    fn like_op() -> Result<()> {
        test_like!(
            vec!["hello world", "world"],
            vec!["%hello%", "%hello%"],
            vec![true, false],
            false,
            false,
            false,
        ); // like
        test_like!(
            vec![Some("hello world"), None, Some("world")],
            vec![Some("%hello%"), None, Some("%hello%")],
            vec![Some(false), None, Some(true)],
            true,
            true,
            false,
        ); // not like
        test_like!(
            vec!["hello world", "world"],
            vec!["%helLo%", "%helLo%"],
            vec![true, false],
            false,
            false,
            true,
        ); // ilike
        test_like!(
            vec![Some("hello world"), None, Some("world")],
            vec![Some("%helLo%"), None, Some("%helLo%")],
            vec![Some(false), None, Some(true)],
            true,
            true,
            true,
        ); // not ilike

        Ok(())
    }

    #[test]
    fn test_fmt_sql() -> Result<()> {
        let schema = Schema::new(vec![
            Field::new("a", DataType::Utf8, false),
            Field::new("b", DataType::Utf8, false),
        ]);

        let expr = like(
            false,
            false,
            col("a", &schema)?,
            col("b", &schema)?,
            &schema,
        )?;

        // Display format
        let display_string = expr.to_string();
        assert_eq!(display_string, "a@0 LIKE b@1");

        // fmt_sql format
        let sql_string = fmt_sql(expr.as_ref()).to_string();
        assert_eq!(sql_string, "a LIKE b");

        Ok(())
    }
}

/// Tests for the `try_to_proto` / `try_from_proto` hooks.
#[cfg(all(test, feature = "proto"))]
mod proto_tests {
    use super::*;
    use crate::expressions::{Column, col};
    use crate::proto_test_util::{
        StubDecoder, StubEncoder, UnreachableDecoder, column_node,
    };
    use arrow::datatypes::Field;
    use datafusion_common::DataFusionError;
    use datafusion_physical_expr_common::physical_expr::proto_decode::PhysicalExprDecodeCtx;
    use datafusion_physical_expr_common::physical_expr::proto_encode::PhysicalExprEncodeCtx;
    use datafusion_proto_models::protobuf::{
        PhysicalExprNode, PhysicalLikeExprNode, physical_expr_node,
    };

    /// Build a `LikeExpr` proto node with the given children.
    fn like_node(
        negated: bool,
        case_insensitive: bool,
        expr: Option<Box<PhysicalExprNode>>,
        pattern: Option<Box<PhysicalExprNode>>,
    ) -> PhysicalExprNode {
        PhysicalExprNode {
            expr_id: None,
            expr_type: Some(physical_expr_node::ExprType::LikeExpr(Box::new(
                PhysicalLikeExprNode {
                    negated,
                    case_insensitive,
                    expr,
                    pattern,
                },
            ))),
        }
    }

    /// A `LikeExpr` over two `Utf8` columns with both flags set, so the
    /// `negated` / `case_insensitive` wiring is actually exercised.
    fn like_fixture() -> LikeExpr {
        let schema = Schema::new(vec![
            Field::new("a", DataType::Utf8, false),
            Field::new("b", DataType::Utf8, false),
        ]);
        LikeExpr::new(
            true,
            true,
            col("a", &schema).unwrap(),
            col("b", &schema).unwrap(),
        )
    }

    #[test]
    fn try_to_proto_encodes_like_expr() {
        let like = like_fixture();
        let encoder = StubEncoder::ok();
        let ctx = PhysicalExprEncodeCtx::new(&encoder);

        let node = like
            .try_to_proto(&ctx)
            .unwrap()
            .expect("LikeExpr should encode to Some(node)");

        // Built-in exprs never set expr_id; only dynamic filters do.
        assert!(node.expr_id.is_none());
        let like_node = match node.expr_type {
            Some(physical_expr_node::ExprType::LikeExpr(boxed)) => *boxed,
            other => panic!("expected a LikeExpr node, got {other:?}"),
        };
        assert!(like_node.negated);
        assert!(like_node.case_insensitive);
        assert!(like_node.expr.is_some());
        assert!(like_node.pattern.is_some());
    }

    #[test]
    fn try_to_proto_propagates_expr_encode_error() {
        let like = like_fixture();
        let encoder = StubEncoder::failing_on(1);
        let ctx = PhysicalExprEncodeCtx::new(&encoder);
        let err = like.try_to_proto(&ctx).unwrap_err();
        assert!(matches!(err, DataFusionError::Internal(msg) if msg.contains("call 1")));
    }

    #[test]
    fn try_to_proto_propagates_pattern_encode_error() {
        let like = like_fixture();
        let encoder = StubEncoder::failing_on(2);
        let ctx = PhysicalExprEncodeCtx::new(&encoder);
        let err = like.try_to_proto(&ctx).unwrap_err();
        assert!(matches!(err, DataFusionError::Internal(msg) if msg.contains("call 2")));
    }

    #[test]
    fn try_from_proto_decodes_like_expr() {
        let node = like_node(
            true,
            true,
            Some(Box::new(column_node("a"))),
            Some(Box::new(column_node("b"))),
        );
        let schema = Schema::empty();
        let decoder = StubDecoder::ok();
        let ctx = PhysicalExprDecodeCtx::new(&schema, &decoder);

        let decoded = LikeExpr::try_from_proto(&node, &ctx).unwrap();
        let like = decoded
            .downcast_ref::<LikeExpr>()
            .expect("decoded expr should be a LikeExpr");
        assert!(like.negated());
        assert!(like.case_insensitive());
        assert!(like.expr().downcast_ref::<Column>().is_some());
        assert!(like.pattern().downcast_ref::<Column>().is_some());
    }

    #[test]
    fn try_from_proto_rejects_non_like_node() {
        let node = column_node("a");
        let schema = Schema::empty();
        let decoder = UnreachableDecoder;
        let ctx = PhysicalExprDecodeCtx::new(&schema, &decoder);
        let err = LikeExpr::try_from_proto(&node, &ctx).unwrap_err();
        assert!(matches!(
            err,
            DataFusionError::Internal(msg) if msg.contains("PhysicalExprNode is not a LikeExpr")
        ));
    }

    #[test]
    fn try_from_proto_rejects_missing_expr() {
        let node = like_node(false, false, None, Some(Box::new(column_node("b"))));
        let schema = Schema::empty();
        let decoder = UnreachableDecoder;
        let ctx = PhysicalExprDecodeCtx::new(&schema, &decoder);
        let err = LikeExpr::try_from_proto(&node, &ctx).unwrap_err();
        assert!(matches!(
            err,
            DataFusionError::Internal(msg) if msg.contains("LikeExpr is missing required field 'expr'")
        ));
    }

    #[test]
    fn try_from_proto_rejects_missing_pattern() {
        let node = like_node(false, false, Some(Box::new(column_node("a"))), None);
        let schema = Schema::empty();
        // `expr` is present, so it is decoded before the missing-`pattern`
        // check fires; use a decoder that succeeds for that first child.
        let decoder = StubDecoder::ok();
        let ctx = PhysicalExprDecodeCtx::new(&schema, &decoder);
        let err = LikeExpr::try_from_proto(&node, &ctx).unwrap_err();
        assert!(matches!(
            err,
            DataFusionError::Internal(msg) if msg.contains("LikeExpr is missing required field 'pattern'")
        ));
    }

    #[test]
    fn try_from_proto_propagates_expr_decode_error() {
        let node = like_node(
            false,
            false,
            Some(Box::new(column_node("a"))),
            Some(Box::new(column_node("b"))),
        );
        let schema = Schema::empty();
        let decoder = StubDecoder::failing_on(1);
        let ctx = PhysicalExprDecodeCtx::new(&schema, &decoder);
        let err = LikeExpr::try_from_proto(&node, &ctx).unwrap_err();
        assert!(matches!(err, DataFusionError::Internal(msg) if msg.contains("call 1")));
    }

    #[test]
    fn try_from_proto_propagates_pattern_decode_error() {
        let node = like_node(
            false,
            false,
            Some(Box::new(column_node("a"))),
            Some(Box::new(column_node("b"))),
        );
        let schema = Schema::empty();
        let decoder = StubDecoder::failing_on(2);
        let ctx = PhysicalExprDecodeCtx::new(&schema, &decoder);
        let err = LikeExpr::try_from_proto(&node, &ctx).unwrap_err();
        assert!(matches!(err, DataFusionError::Internal(msg) if msg.contains("call 2")));
    }
}