fix: Emit LIKE the way the Substrait extension defines it (#25442)

## Which issue does this PR close?

- Closes #25368.

## Rationale for this change

`functions_string.yaml` defines one function for this, with two
arguments and an option:

```yaml
    name: like
    impls:
      - args:
          - value: "varchar<L1>"
            name: "input"
          - value: "varchar<L2>"
            name: "match"
        options:
          case_sensitivity:
            values: [ CASE_SENSITIVE, CASE_INSENSITIVE, CASE_INSENSITIVE_ASCII ]
        return: "boolean"
```

The producer emitted three arguments, registered `ilike` for the case
insensitive form, and never set the option. Neither `ilike` nor an
escape argument appears in any extension file, in the pinned crate or
upstream.

## What changes are included in this PR?

- `make_substrait_like_expr` registers `like`, emits the two arguments
the definition takes, and sets `case_sensitivity` to `CASE_INSENSITIVE`
for `ILIKE`. A plain `LIKE` sets no option, which leaves the default.
- `LIKE ... ESCAPE` is now `not_impl_err`. There is no escape in the
definition, so the third argument only produced a call a consumer cannot
bind.
- The consumer reads `case_sensitivity`, taking the first value it
supports and refusing the ones it does not, as Substrait requires. It
still accepts the `ilike` name and the three argument form, so plans
written by an older DataFusion still load.

## What is the testing strategy for this PR?

- New producer tests: `LIKE` emits `like` with two arguments and no
option, `ILIKE` emits the `case_sensitivity` option, and an escape
character is rejected.
- `roundtrip_like` and `roundtrip_ilike` still pass, which is what
covers the consumer: with the option reading removed, `roundtrip_ilike`
fails because `ILIKE` comes back as `LIKE`. Removing the producer half
likewise fails the new producer test.
- `cargo test -p datafusion-substrait` passes and
`./ci/scripts/rust_clippy.sh` is clean.

Measured against substrait-java 0.103.0 and Spark 3.5.4, over `t(s)`
holding `abc` and `ABC`, with the #11545 URN patched and `output_type`
filled (#25049):

| plan | before | after |
| --- | --- | --- |
| `s LIKE 'a%'` | `AnalysisException: WRONG_NUM_ARGS ... requires 2
parameters but the actual number is 3` | runs, returns `abc` |
| `s ILIKE 'a%'` | `IllegalArgumentException: Unexpected scalar function
with key ilike:str_str` | runs, returns `abc` |

**The second row is a trade-off worth stating.** `ILIKE` used to fail
loudly there, and now it is accepted but answered case sensitively,
because substrait-spark does not read the option: `case_sensitivity`
appears nowhere in its sources. Substrait says a consumer that does not
recognise an option must reject the call, so that looks like a gap on
that side, and I am happy to report it there. If you would rather not
emit something a known consumer mishandles, the alternative is to reject
`ILIKE` as well until a consumer honours the option, and I will change
it.

## Are there any user-facing changes?

`LIKE` and `ILIKE` are emitted in the form the extension defines, so
consumers that bind arguments by the declaration can now read them.
`LIKE ... ESCAPE` is rejected instead of being emitted in a shape no
consumer can bind; DataFusion round trips it today only because its own
consumer reads the same private convention.

This touches `make_substrait_like_expr`, which #25367 folds into
`from_like`, so whichever lands second needs a small rebase.

---------

Co-authored-by: namanjain24-sudo <180642416+namanjain24-sudo@users.noreply.github.com>

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),
@@ -374,7 +403,7 @@ impl BuiltinExprBuilder {
 
 #[cfg(test)]
 mod tests {
-    use super::arg_list_to_binary_op_tree;
+    use super::{arg_list_to_binary_op_tree, case_insensitive_option};
     use crate::extensions::Extensions;
     use crate::logical_plan::consumer::tests::TEST_SESSION_STATE;
     use crate::logical_plan::consumer::{DefaultSubstraitConsumer, SubstraitConsumer};
@@ -583,4 +612,83 @@ mod tests {
 
         Ok(())
     }
+
+    fn scalar_function_with_case_sensitivity_preference(
+        preference: &[&str],
+    ) -> ScalarFunction {
+        ScalarFunction {
+            options: vec![substrait::proto::FunctionOption {
+                name: "case_sensitivity".to_string(),
+                preference: preference.iter().map(|s| s.to_string()).collect(),
+            }],
+            ..Default::default()
+        }
+    }
+
+    /// Substrait says a consumer must use the first value it supports in the
+    /// `case_sensitivity` option's preference list, and must reject the call
+    /// when it supports none of them.
+    #[test]
+    fn case_insensitive_option_cases() -> Result<()> {
+        // No `case_sensitivity` option at all defaults to case-sensitive.
+        assert!(!case_insensitive_option(&ScalarFunction::default())?);
+
+        // A single supported preference, in either sensitivity.
+        assert!(!case_insensitive_option(
+            &scalar_function_with_case_sensitivity_preference(&["CASE_SENSITIVE"])
+        )?);
+        assert!(case_insensitive_option(
+            &scalar_function_with_case_sensitivity_preference(&["CASE_INSENSITIVE"])
+        )?);
+
+        // The option name and its preference values are matched
+        // case-insensitively (ASCII).
+        assert!(case_insensitive_option(&ScalarFunction {
+            options: vec![substrait::proto::FunctionOption {
+                name: "Case_Sensitivity".to_string(),
+                preference: vec!["case_insensitive".to_string()],
+            }],
+            ..Default::default()
+        })?);
+
+        // Preference ordering: a consumer must use the first value it
+        // supports, not necessarily the first value in the list.
+        // `CASE_INSENSITIVE_ASCII` is not a value this consumer supports, so
+        // it is skipped in favor of the next, supported preference.
+        assert!(case_insensitive_option(
+            &scalar_function_with_case_sensitivity_preference(&[
+                "CASE_INSENSITIVE_ASCII",
+                "CASE_INSENSITIVE"
+            ])
+        )?);
+        assert!(!case_insensitive_option(
+            &scalar_function_with_case_sensitivity_preference(&[
+                "CASE_INSENSITIVE_ASCII",
+                "CASE_SENSITIVE"
+            ])
+        )?);
+
+        // Rejection: every preference is unsupported.
+        let err =
+            case_insensitive_option(&scalar_function_with_case_sensitivity_preference(
+                &["CASE_INSENSITIVE_ASCII"],
+            ))
+            .unwrap_err();
+        assert!(
+            err.to_string().contains("Unsupported case_sensitivity"),
+            "unexpected error: {err}"
+        );
+
+        // Rejection: an empty preference list supports nothing.
+        let err = case_insensitive_option(
+            &scalar_function_with_case_sensitivity_preference(&[]),
+        )
+        .unwrap_err();
+        assert!(
+            err.to_string().contains("Unsupported case_sensitivity"),
+            "unexpected error: {err}"
+        );
+
+        Ok(())
+    }
 }

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,42 @@ 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())
-    };
+    // `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`");
+    }
     // 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)?;
-    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 function_anchor = producer.register_function("like".to_string());
+    let substrait_expr = producer.handle_expr(expr, schema)?;
+    let substrait_pattern = producer.handle_expr(pattern, schema)?;
     let arguments = vec![
         FunctionArgument {
-            arg_type: Some(ArgType::Value(expr)),
-        },
-        FunctionArgument {
-            arg_type: Some(ArgType::Value(pattern)),
+            arg_type: Some(ArgType::Value(substrait_expr)),
         },
         FunctionArgument {
-            arg_type: Some(ArgType::Value(escape_char)),
+            arg_type: Some(ArgType::Value(substrait_pattern)),
         },
     ];
+    // An unset option leaves the default, which is `CASE_SENSITIVE`.
+    let options = if *case_insensitive {
+        vec![FunctionOption {
+            name: CASE_SENSITIVITY_OPTION.to_string(),
+            preference: vec![CASE_INSENSITIVE.to_string()],
+        }]
+    } else {
+        vec![]
+    };
 
     #[expect(deprecated)]
     let substrait_like = Expression {
@@ -283,7 +295,7 @@ pub fn from_like(
             arguments,
             output_type: Some(output_type.clone()),
             args: vec![],
-            options: vec![],
+            options,
         })),
     };
 
@@ -438,6 +450,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 +460,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();