fix(query): handle empty LIKE ESCAPE in planner (#19595)

* fix(query): handle empty LIKE ESCAPE in planner (#19562)

* fix(query): validate LIKE ESCAPE literals (#19562)

* fix(query): preserve LIKE ESCAPE fallback semantics

* test(query): cover empty LIKE ESCAPE fallback (#19562)

* test(query): align empty LIKE ESCAPE regression output

* test(query): remove cluster-unstable explain assertion

Author: sundy-li

src/query/sql/src/planner/semantic/type_check.rs

@@ -5646,13 +5646,22 @@ impl<'a> TypeChecker<'a> {
         like_str: &str,
         escape: &Option<String>,
     ) -> Result<Box<(ScalarExpr, DataType)>> {
-        let new_like_str = if let Some(escape) = escape {
-            Cow::Owned(convert_escape_pattern(
-                like_str,
-                escape.chars().next().unwrap(),
-            ))
-        } else {
-            Cow::Borrowed(like_str)
+        let new_like_str = match escape.as_ref() {
+            Some(escape_literal) => {
+                let mut chars = escape_literal.chars();
+                let Some(escape_char) = chars.next() else {
+                    // Empty escape literals must stay on the builtin path to match runtime behavior.
+                    return self.resolve_like_escape(op, span, left, right, escape);
+                };
+
+                if chars.next().is_some() {
+                    // Preserve existing builtin behavior for non-single-character escape literals.
+                    return self.resolve_like_escape(op, span, left, right, escape);
+                }
+
+                Cow::Owned(convert_escape_pattern(like_str, escape_char))
+            }
+            None => Cow::Borrowed(like_str),
         };
         if check_percent(&new_like_str) {
             // Convert to `a is not null`

src/query/sql/tests/it/planner.rs

@@ -142,6 +142,24 @@ async fn test_lite_replay_service_optimizer_cases() -> Result<()> {
     Ok(())
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+async fn test_like_escape_preserves_existing_binding_semantics() -> Result<()> {
+    let ctx = LiteTableContext::create().await?;
+
+    for sql in [
+        "SELECT 'a' LIKE 'a' ESCAPE ''",
+        "SELECT 'a' LIKE concat('a') ESCAPE ''",
+        "SELECT '%' LIKE '\\\\%' ESCAPE ''",
+        "SELECT like_any('%', '\\\\%', '')",
+        "SELECT 'a' LIKE ANY ('a', 'b') ESCAPE ''",
+        "SELECT 'a' LIKE ANY (SELECT 'a') ESCAPE ''",
+    ] {
+        ctx.bind_sql(sql).await?;
+    }
+
+    Ok(())
+}
+
 async fn setup_tables(ctx: &Arc<LiteTableContext>, case: &TestCase) -> Result<()> {
     for sql in case.tables.values() {
         for statement in sql.split(';').filter(|s| !s.trim().is_empty()) {

tests/sqllogictests/suites/query/issues/issue_19562.test

@@ -0,0 +1,26 @@
+# GitHub issue: https://github.com/databendlabs/databend/issues/19562
+
+query B
+SELECT 'a' LIKE 'a' ESCAPE ''
+----
+1
+
+query B
+SELECT 'a' LIKE concat('a') ESCAPE ''
+----
+1
+
+query B
+SELECT '%' LIKE '\\\\%' ESCAPE ''
+----
+0
+
+query B
+SELECT like_any('%', '\\\\%', '')
+----
+0
+
+query B
+SELECT 'a' LIKE ANY (SELECT 'a') ESCAPE ''
+----
+1