Add SQL planner support for Like, ILike and SimilarTo, with optional escape character

datafusion/core/src/physical_plan/planner.rs

@@ -1682,6 +1682,8 @@ mod tests {
     use datafusion_expr::expr::GroupingSet;
     use datafusion_expr::sum;
     use datafusion_expr::{col, lit};
+    use datafusion_optimizer::type_coercion::TypeCoercion;
+    use datafusion_optimizer::{OptimizerConfig, OptimizerRule};
     use fmt::Debug;
     use std::collections::HashMap;
     use std::convert::TryFrom;
@@ -1859,7 +1861,19 @@ mod tests {
             col("c1").like(col("c2")),
         ];
         for case in cases {
-            let logical_plan = test_csv_scan().await?.project(vec![case.clone()]);
+            let logical_plan = test_csv_scan()
+                .await?
+                .project(vec![case.clone()])
+                .and_then(|b| b.build())
+                .and_then(|plan| {
+                    // this test was expecting type coercion/validation errors before the optimizer
+                    // had run due to the legacy approach of type coercion but now we need to run the
+                    // optimizer here
+                    let type_coercion = TypeCoercion::default();
+                    let mut config = OptimizerConfig::new();
+                    type_coercion.optimize(&plan, &mut config)
+                });
+
             let message = format!(
                 "Expression {:?} expected to error due to impossible coercion",
                 case

datafusion/optimizer/src/type_coercion.rs

@@ -18,7 +18,8 @@
 //! Optimizer rule for type validation and coercion
 
 use crate::{OptimizerConfig, OptimizerRule};
-use datafusion_common::{DFSchema, DFSchemaRef, Result};
+use arrow::datatypes::DataType;
+use datafusion_common::{DFSchema, DFSchemaRef, DataFusionError, Result};
 use datafusion_expr::binary_rule::coerce_types;
 use datafusion_expr::expr_rewriter::{ExprRewritable, ExprRewriter, RewriteRecursion};
 use datafusion_expr::logical_plan::builder::build_join_schema;
@@ -86,29 +87,45 @@ impl ExprRewriter for TypeCoercionRewriter {
     }
 
     fn mutate(&mut self, expr: Expr) -> Result<Expr> {
-        match expr {
+        match &expr {
             Expr::BinaryExpr { left, op, right } => {
                 let left_type = left.get_type(&self.schema)?;
                 let right_type = right.get_type(&self.schema)?;
-                let coerced_type = coerce_types(&left_type, &op, &right_type)?;
+                let coerced_type = coerce_types(&left_type, op, &right_type)?;
                 Ok(Expr::BinaryExpr {
-                    left: Box::new(left.cast_to(&coerced_type, &self.schema)?),
-                    op,
-                    right: Box::new(right.cast_to(&coerced_type, &self.schema)?),
+                    left: Box::new(
+                        left.as_ref().clone().cast_to(&coerced_type, &self.schema)?,
+                    ),
+                    op: *op,
+                    right: Box::new(
+                        right
+                            .as_ref()
+                            .clone()
+                            .cast_to(&coerced_type, &self.schema)?,
+                    ),
                 })
             }
+            Expr::Like { pattern, .. }
+            | Expr::ILike { pattern, .. }
+            | Expr::SimilarTo { pattern, .. } => match pattern.get_type(&self.schema)? {
+                DataType::Utf8 => Ok(expr.clone()),
+                other => Err(DataFusionError::Plan(format!(
+                    "Expected pattern in Like, ILike, or SimilarTo to be Utf8 but was {}",
+                    other
+                ))),
+            },
             Expr::ScalarUDF { fun, args } => {
                 let new_expr = coerce_arguments_for_signature(
                     args.as_slice(),
                     &self.schema,
                     &fun.signature,
                 )?;
                 Ok(Expr::ScalarUDF {
-                    fun,
+                    fun: fun.clone(),
                     args: new_expr,
                 })
             }
-            expr => Ok(expr),
+            expr => Ok(expr.clone()),
         }
     }
 }

datafusion/physical-expr/src/planner.rs

@@ -203,6 +203,26 @@ pub fn create_physical_expr(
                 }
             }
         }
+        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::Case {
             expr,
             when_then_expr,

datafusion/proto/src/lib.rs

@@ -929,6 +929,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_count() {
         let test_expr = Expr::AggregateFunction {

datafusion/sql/src/planner.rs

@@ -1939,30 +1939,31 @@ 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, ctes)?),
-                            op: if negated { Operator::NotLike } else { Operator::Like },
-                            right: Box::new(self.sql_expr_to_logical_expr(*pattern, schema, ctes)?),
-                        })
-                    }
-                }
+                Ok(Expr::Like {
+                    negated,
+                    expr: Box::new(self.sql_expr_to_logical_expr(*expr, schema, ctes)?),
+                    pattern: Box::new(self.sql_expr_to_logical_expr(*pattern, schema, ctes)?),
+                    escape_char
+
+                })
             }
 
-            SQLExpr::ILike { .. } => {
-                // https://github.com/apache/arrow-datafusion/issues/3099
-                Err(DataFusionError::NotImplemented("ILIKE is not yet supported".to_string()))
+            SQLExpr::ILike { negated, expr, pattern, escape_char } => {
+                Ok(Expr::ILike {
+                    negated,
+                    expr: Box::new(self.sql_expr_to_logical_expr(*expr, schema, ctes)?),
+                    pattern: Box::new(self.sql_expr_to_logical_expr(*pattern, schema, ctes)?),
+                    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 } => {
+                Ok(Expr::SimilarTo {
+                    negated,
+                    expr: Box::new(self.sql_expr_to_logical_expr(*expr, schema, ctes)?),
+                    pattern: Box::new(self.sql_expr_to_logical_expr(*pattern, schema, ctes)?),
+                    escape_char
+                })
             }
 
             SQLExpr::BinaryOp {