fix: Emit LIKE the way the Substrait extension defines it

The extension defines one function, `like`, taking two arguments and
carrying case sensitivity as the `case_sensitivity` option. The producer
emitted three arguments, registered `ilike` for the case insensitive
form, which no extension defines, and never set the option. substrait-java
resolves `ilike` to nothing, and Spark rejects the three argument call
with WRONG_NUM_ARGS.

The producer now registers `like`, emits the two arguments, and sets the
option for ILIKE. An escape character has no place in that definition, so
LIKE ... ESCAPE is rejected rather than emitted as a third argument.

The consumer reads the option, taking the first value it supports, and
still accepts the `ilike` name and the three argument form so that plans
from an older DataFusion keep loading.

Author: namanjain24-sudo

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

@@ -16,17 +16,17 @@
 // under the License.
 
 use crate::logical_plan::producer::{
-    SubstraitProducer, to_substrait_literal_expr, to_substrait_type,
-    to_substrait_type_from_field,
+    SubstraitProducer, to_substrait_type, to_substrait_type_from_field,
 };
 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};
@@ -234,6 +234,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,
@@ -246,35 +251,65 @@ pub fn from_like(
         escape_char,
         case_insensitive,
     } = like;
-    let function_anchor = if *case_insensitive {
-        producer.register_function("ilike".to_string())
-    } else {
-        producer.register_function("like".to_string())
-    };
     // Substrait documents `output_type` as "Must be set to the return type of
     // the function, exactly as derived using the declaration in the extension",
     // and a consumer that reads it rejects the call when it is unset. The type
     // comes from the expression itself so that it matches what DataFusion
     // derives, rather than being restated here.
     let (_, output_field) = Expr::Like(like.clone()).to_field(schema)?;
     let output_type = to_substrait_type_from_field(producer, &output_field)?;
+
+    make_substrait_like_expr(
+        producer,
+        *case_insensitive,
+        *negated,
+        expr,
+        pattern,
+        *escape_char,
+        schema,
+        output_type,
+    )
+}
+
+#[expect(clippy::too_many_arguments)]
+fn make_substrait_like_expr(
+    producer: &mut impl SubstraitProducer,
+    ignore_case: bool,
+    negated: bool,
+    expr: &Expr,
+    pattern: &Expr,
+    escape_char: Option<char>,
+    schema: &DFSchemaRef,
+    output_type: Type,
+) -> datafusion::common::Result<Expression> {
+    // `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 {
@@ -283,11 +318,11 @@ pub fn from_like(
             arguments,
             output_type: Some(output_type.clone()),
             args: vec![],
-            options: vec![],
+            options,
         })),
     };
 
-    if *negated {
+    if negated {
         let function_anchor = producer.register_function("not".to_string());
 
         #[expect(deprecated)]
@@ -438,6 +473,7 @@ 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,
     };
@@ -447,9 +483,88 @@ mod tests {
     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};
     use substrait::proto::function_argument::ArgType;
 
+    /// `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();