Add SQL planner support for Like, ILike and SimilarTo, with optional escape character (apache#3101)

* Make Like a top-level Expr

* revert some changes

* add type validation

* Revert physical plan changes and reduce scope of the PR

* Revert more changes

* Revert more changes

* clippy

* address feedback

* revert change to test

* revert more changes

Author: andygrove

datafusion/core/src/physical_plan/planner.rs

@@ -68,6 +68,7 @@ use arrow::datatypes::{Schema, SchemaRef};
 use arrow::{compute::can_cast_types, datatypes::DataType};
 use async_trait::async_trait;
 use datafusion_common::OuterQueryCursor;
+use datafusion_expr::expr_fn::binary_expr;
 use datafusion_physical_expr::expressions::{any, OuterColumn};
 use futures::future::BoxFuture;
 use futures::{FutureExt, StreamExt, TryStreamExt};
@@ -1080,6 +1081,46 @@ pub fn create_physical_expr(
             )?;
             binary(lhs, *op, rhs, input_schema)
         }
+        Expr::Like {
+            negated,
+            expr,
+            pattern,
+            escape_char,
+        } => {
+            if escape_char.is_some() {
+                return Err(DataFusionError::Execution(
+                    "LIKE does not support escape_char".to_string(),
+                ));
+            }
+            let op = if *negated {
+                Operator::NotLike
+            } else {
+                Operator::Like
+            };
+            let bin_expr =
+                binary_expr(expr.as_ref().clone(), op, pattern.as_ref().clone());
+            create_physical_expr(&bin_expr, input_dfschema, input_schema, execution_props)
+        }
+        Expr::ILike {
+            negated,
+            expr,
+            pattern,
+            escape_char,
+        } => {
+            if escape_char.is_some() {
+                return Err(DataFusionError::Execution(
+                    "ILIKE does not support escape_char".to_string(),
+                ));
+            }
+            let op = if *negated {
+                Operator::NotILike
+            } else {
+                Operator::ILike
+            };
+            let bin_expr =
+                binary_expr(expr.as_ref().clone(), op, pattern.as_ref().clone());
+            create_physical_expr(&bin_expr, input_dfschema, input_schema, execution_props)
+        }
         Expr::Case {
             expr,
             when_then_expr,

datafusion/core/src/sql/planner.rs

@@ -31,8 +31,8 @@ use crate::logical_plan::Expr::Alias;
 use crate::logical_plan::{
     and, builder::expand_qualified_wildcard, builder::expand_wildcard, col, lit,
     normalize_col, rewrite_udtfs_to_columns, Column, CreateMemoryTable, DFSchema,
-    DFSchemaRef, DropTable, Expr, LogicalPlan, LogicalPlanBuilder, Operator, PlanType,
-    ToDFSchema, ToStringifiedPlan,
+    DFSchemaRef, DropTable, Expr, ExprSchemable, LogicalPlan, LogicalPlanBuilder,
+    Operator, PlanType, ToDFSchema, ToStringifiedPlan,
 };
 use crate::optimizer::utils::exprlist_to_columns;
 use crate::prelude::JoinType;
@@ -2002,42 +2002,52 @@ impl<'a, S: ContextProvider> SqlToRel<'a, S> {
             }
 
             SQLExpr::Like { negated, expr, pattern, escape_char } => {
-                match escape_char {
-                    Some(_) => {
-                        // to support this we will need to introduce `Expr::Like` instead
-                        // of treating it like a binary expression
-                        Err(DataFusionError::NotImplemented("LIKE with ESCAPE is not yet supported".to_string()))
-                    },
-                    _ => {
-                        Ok(Expr::BinaryExpr {
-                            left: Box::new(self.sql_expr_to_logical_expr(*expr, schema,)?),
-                            op: if negated { Operator::NotLike } else { Operator::Like },
-                            right: Box::new(self.sql_expr_to_logical_expr(*pattern, schema)?),
-                        })
-                    }
+                let pattern = self.sql_expr_to_logical_expr(*pattern, schema)?;
+                let pattern_type = pattern.get_type(schema)?;
+                if pattern_type != DataType::Utf8 && pattern_type != DataType::Null {
+                    return Err(DataFusionError::Plan(
+                        "Invalid pattern in LIKE expression".to_string(),
+                    ));
                 }
+                Ok(Expr::Like {
+                    negated,
+                    expr: Box::new(self.sql_expr_to_logical_expr(*expr, schema)?),
+                    pattern: Box::new(pattern),
+                    escape_char
+
+                })
             }
 
             SQLExpr::ILike { negated, expr, pattern, escape_char } => {
-                match escape_char {
-                    Some(_) => {
-                        // to support this we will need to introduce `Expr::ILike` instead
-                        // of treating it like a binary expression
-                        Err(DataFusionError::NotImplemented("ILIKE with ESCAPE is not yet supported".to_string()))
-                    },
-                    _ => {
-                        Ok(Expr::BinaryExpr {
-                            left: Box::new(self.sql_expr_to_logical_expr(*expr, schema,)?),
-                            op: if negated { Operator::NotILike } else { Operator::ILike },
-                            right: Box::new(self.sql_expr_to_logical_expr(*pattern, schema)?),
-                        })
-                    }
+                let pattern = self.sql_expr_to_logical_expr(*pattern, schema)?;
+                let pattern_type = pattern.get_type(schema)?;
+                if pattern_type != DataType::Utf8 && pattern_type != DataType::Null {
+                    return Err(DataFusionError::Plan(
+                        "Invalid pattern in ILIKE expression".to_string(),
+                    ));
                 }
+                Ok(Expr::ILike {
+                    negated,
+                    expr: Box::new(self.sql_expr_to_logical_expr(*expr, schema)?),
+                    pattern: Box::new(pattern),
+                    escape_char
+                })
             }
 
-            SQLExpr::SimilarTo { .. } => {
-                // https://github.com/apache/arrow-datafusion/issues/3099
-                Err(DataFusionError::NotImplemented("SIMILAR TO is not yet supported".to_string()))
+            SQLExpr::SimilarTo { negated, expr, pattern, escape_char } => {
+                let pattern = self.sql_expr_to_logical_expr(*pattern, schema)?;
+                let pattern_type = pattern.get_type(schema)?;
+                if pattern_type != DataType::Utf8 && pattern_type != DataType::Null {
+                    return Err(DataFusionError::Plan(
+                        "Invalid pattern in SIMILAR TO expression".to_string(),
+                    ));
+                }
+                Ok(Expr::SimilarTo {
+                    negated,
+                    expr: Box::new(self.sql_expr_to_logical_expr(*expr, schema)?),
+                    pattern: Box::new(pattern),
+                    escape_char
+                })
             }
 
             SQLExpr::BinaryOp {

datafusion/proto/src/lib.rs

@@ -699,6 +699,60 @@ mod roundtrip_tests {
         roundtrip_expr_test!(test_expr, ctx);
     }
 
+    #[test]
+    fn roundtrip_like() {
+        fn like(negated: bool, escape_char: Option<char>) {
+            let test_expr = Expr::Like {
+                negated,
+                expr: Box::new(col("col")),
+                pattern: Box::new(lit("[0-9]+")),
+                escape_char,
+            };
+            let ctx = SessionContext::new();
+            roundtrip_expr_test!(test_expr, ctx);
+        }
+        like(true, Some('X'));
+        like(false, Some('\\'));
+        like(true, None);
+        like(false, None);
+    }
+
+    #[test]
+    fn roundtrip_ilike() {
+        fn ilike(negated: bool, escape_char: Option<char>) {
+            let test_expr = Expr::ILike {
+                negated,
+                expr: Box::new(col("col")),
+                pattern: Box::new(lit("[0-9]+")),
+                escape_char,
+            };
+            let ctx = SessionContext::new();
+            roundtrip_expr_test!(test_expr, ctx);
+        }
+        ilike(true, Some('X'));
+        ilike(false, Some('\\'));
+        ilike(true, None);
+        ilike(false, None);
+    }
+
+    #[test]
+    fn roundtrip_similar_to() {
+        fn similar_to(negated: bool, escape_char: Option<char>) {
+            let test_expr = Expr::SimilarTo {
+                negated,
+                expr: Box::new(col("col")),
+                pattern: Box::new(lit("[0-9]+")),
+                escape_char,
+            };
+            let ctx = SessionContext::new();
+            roundtrip_expr_test!(test_expr, ctx);
+        }
+        similar_to(true, Some('X'));
+        similar_to(false, Some('\\'));
+        similar_to(true, None);
+        similar_to(false, None);
+    }
+
     #[test]
     fn roundtrip_approx_percentile_cont() {
         let test_expr = Expr::AggregateFunction {