fix: Emit LIKE the way the Substrait extension defines it

datafusion/substrait/src/logical_plan/consumer/expr/scalar_function.rs

@@ -187,6 +187,32 @@ fn arg_list_to_binary_op_tree_inner(
     }))
 }
 
+/// Reads the `case_sensitivity` option of a `like` call.
+///
+/// Substrait says a consumer must use the first value it supports, and must
+/// reject the call when it supports none of them.
+fn case_insensitive_option(f: &ScalarFunction) -> Result<bool> {
+    let Some(option) = f
+        .options
+        .iter()
+        .find(|option| option.name.eq_ignore_ascii_case("case_sensitivity"))
+    else {
+        return Ok(false);
+    };
+    for preference in &option.preference {
+        if preference.eq_ignore_ascii_case("CASE_SENSITIVE") {
+            return Ok(false);
+        }
+        if preference.eq_ignore_ascii_case("CASE_INSENSITIVE") {
+            return Ok(true);
+        }
+    }
+    not_impl_err!(
+        "Unsupported case_sensitivity for `like`: {:?}",
+        option.preference
+    )
+}
+
 /// Build [`Expr`] from its name and required inputs.
 struct BuiltinExprBuilder {
     expr_name: String,
@@ -213,7 +239,10 @@ impl BuiltinExprBuilder {
         args: Vec<Expr>,
     ) -> Result<Expr> {
         match self.expr_name.as_str() {
-            "like" => Self::build_like_expr(false, false, f, args),
+            // `like` carries case sensitivity as an option. `ilike` is not a
+            // Substrait function, but DataFusion used to emit it, so plans
+            // written by an older version are still read.
+            "like" => Self::build_like_expr(case_insensitive_option(f)?, false, f, args),
             "ilike" => Self::build_like_expr(true, false, f, args),
             "like_match" => Self::build_like_expr(false, false, f, args),
             "like_imatch" => Self::build_like_expr(true, false, f, args),

datafusion/substrait/src/logical_plan/producer/expr/scalar_function.rs

@@ -15,17 +15,16 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use crate::logical_plan::producer::{
-    SubstraitProducer, to_substrait_literal_expr, to_substrait_type,
-};
+use crate::logical_plan::producer::{SubstraitProducer, to_substrait_type};
 use datafusion::arrow::datatypes::DataType;
 use datafusion::common::datatype::FieldExt;
 use datafusion::common::{
-    DFSchemaRef, ScalarValue, internal_datafusion_err, not_impl_err, substrait_err,
+    DFSchemaRef, internal_datafusion_err, not_impl_err, substrait_err,
 };
 use datafusion::logical_expr::{
     Between, BinaryExpr, Expr, ExprSchemable, Like, Operator, expr,
 };
+use substrait::proto::FunctionOption;
 use substrait::proto::expression::{RexType, ScalarFunction};
 use substrait::proto::function_argument::ArgType;
 use substrait::proto::{Expression, FunctionArgument, Type};
@@ -233,6 +232,11 @@ pub fn from_binary_expr(
     ))
 }
 
+/// The option `like` uses to carry case sensitivity, and the value that asks
+/// for the case insensitive form. Defined in `functions_string.yaml`.
+pub(crate) const CASE_SENSITIVITY_OPTION: &str = "case_sensitivity";
+pub(crate) const CASE_INSENSITIVE: &str = "CASE_INSENSITIVE";
+
 pub fn from_like(
     producer: &mut impl SubstraitProducer,
     like: &Like,
@@ -265,28 +269,34 @@ fn make_substrait_like_expr(
     escape_char: Option<char>,
     schema: &DFSchemaRef,
 ) -> datafusion::common::Result<Expression> {
-    let function_anchor = if ignore_case {
-        producer.register_function("ilike".to_string())
-    } else {
-        producer.register_function("like".to_string())
-    };
+    // `like` takes two arguments and carries case sensitivity as an option;
+    // the extensions define no `ilike` and no escape character, so an escape
+    // has no encoding here and is rejected rather than emitted as a third
+    // argument that a consumer would bind to a parameter the function does
+    // not have.
+    if escape_char.is_some() {
+        return not_impl_err!("Substrait does not define an escape character for `like`");
+    }
+    let function_anchor = producer.register_function("like".to_string());
     let expr = producer.handle_expr(expr, schema)?;
     let pattern = producer.handle_expr(pattern, schema)?;
-    let escape_char = to_substrait_literal_expr(
-        producer,
-        &ScalarValue::Utf8(escape_char.map(|c| c.to_string())),
-    )?;
     let arguments = vec![
         FunctionArgument {
             arg_type: Some(ArgType::Value(expr)),
         },
         FunctionArgument {
             arg_type: Some(ArgType::Value(pattern)),
         },
-        FunctionArgument {
-            arg_type: Some(ArgType::Value(escape_char)),
-        },
     ];
+    // An unset option leaves the default, which is `CASE_SENSITIVE`.
+    let options = if ignore_case {
+        vec![FunctionOption {
+            name: CASE_SENSITIVITY_OPTION.to_string(),
+            preference: vec![CASE_INSENSITIVE.to_string()],
+        }]
+    } else {
+        vec![]
+    };
 
     #[expect(deprecated)]
     let substrait_like = Expression {
@@ -295,7 +305,7 @@ fn make_substrait_like_expr(
             arguments,
             output_type: None,
             args: vec![],
-            options: vec![],
+            options,
         })),
     };
 
@@ -449,16 +459,97 @@ pub fn operator_to_name(op: Operator) -> &'static str {
 
 #[cfg(test)]
 mod tests {
+    use super::{CASE_INSENSITIVE, CASE_SENSITIVITY_OPTION};
     use crate::logical_plan::producer::{
         DefaultSubstraitProducer, SubstraitProducer, to_substrait_type,
     };
-    use datafusion::arrow::datatypes::DataType;
+    use datafusion::arrow::datatypes::{DataType, Field, Schema};
     use datafusion::common::{DFSchema, DFSchemaRef};
     use datafusion::execution::SessionStateBuilder;
-    use datafusion::prelude::lit;
+    use datafusion::logical_expr::{Expr, Like};
+    use datafusion::prelude::{col, lit};
     use substrait::proto::Expression;
+    use substrait::proto::FunctionOption;
     use substrait::proto::expression::{RexType, ScalarFunction};
 
+    /// `like` takes two arguments and carries case sensitivity as an option,
+    /// so `ILIKE` is that option rather than a separate function.
+    #[tokio::test]
+    async fn like_emits_case_sensitivity_option() -> datafusion::common::Result<()> {
+        let state = SessionStateBuilder::default().build();
+        let schema =
+            DFSchemaRef::new(DFSchema::try_from(Schema::new(vec![Field::new(
+                "s",
+                DataType::Utf8,
+                true,
+            )]))?);
+
+        for (case_insensitive, expected_options) in [
+            (false, vec![]),
+            (
+                true,
+                vec![FunctionOption {
+                    name: CASE_SENSITIVITY_OPTION.to_string(),
+                    preference: vec![CASE_INSENSITIVE.to_string()],
+                }],
+            ),
+        ] {
+            let mut producer = DefaultSubstraitProducer::new(&state);
+            let like = Like::new(
+                false,
+                Box::new(col("s")),
+                Box::new(lit("a%")),
+                None,
+                case_insensitive,
+            );
+            let expr = producer.handle_expr(&Expr::Like(like), &schema)?;
+
+            let Some(RexType::ScalarFunction(call)) = expr.rex_type else {
+                panic!("Substrait ScalarFunction expected")
+            };
+            assert_eq!(
+                producer
+                    .get_extensions()
+                    .functions
+                    .get(&call.function_reference),
+                Some(&"like".to_string()),
+                "case_insensitive = {case_insensitive}"
+            );
+            assert_eq!(call.arguments.len(), 2, "no escape argument is emitted");
+            assert_eq!(call.options, expected_options);
+        }
+
+        Ok(())
+    }
+
+    /// The extensions define no escape character, so there is nothing to emit.
+    #[tokio::test]
+    async fn like_with_escape_is_rejected() -> datafusion::common::Result<()> {
+        let state = SessionStateBuilder::default().build();
+        let schema =
+            DFSchemaRef::new(DFSchema::try_from(Schema::new(vec![Field::new(
+                "s",
+                DataType::Utf8,
+                true,
+            )]))?);
+        let mut producer = DefaultSubstraitProducer::new(&state);
+
+        let like = Like::new(
+            false,
+            Box::new(col("s")),
+            Box::new(lit("a!%")),
+            Some('!'),
+            false,
+        );
+        let err = producer
+            .handle_expr(&Expr::Like(like), &schema)
+            .expect_err("an escape character must be rejected")
+            .to_string();
+        assert!(err.contains("escape character"), "unexpected error: {err}");
+
+        Ok(())
+    }
+
     #[tokio::test]
     async fn binary_expr_output_type() -> datafusion::common::Result<()> {
         let state = SessionStateBuilder::default().build();