[FLINK-39402][table] Fix LIKE quick path ignoring default escape character '\'

Author: Au-Miner

flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/LikeCallGen.scala

@@ -54,47 +54,58 @@ class LikeCallGen extends CallGenerator {
         terms =>
           val pattern = operands(1).literalValue.get.toString
           var newPattern: String = pattern
-          val allowQuick = if (operands.length == 2) {
-            !pattern.contains("_")
-          } else {
-            val escape = operands(2).literalValue.get.toString
-            if (escape.isEmpty) {
-              !pattern.contains("_")
+          val allowQuick = {
+            // Determine escape character:
+            // - operands.length == 2 (no ESCAPE clause): default escape is '\'
+            // - operands.length == 3 with empty escape: no escape character
+            // - operands.length == 3 with non-empty escape: use specified escape
+            val escapeCharOpt: Option[Char] = if (operands.length == 2) {
+              Some('\\')
             } else {
-              if (escape.length > 1) {
-                throw SqlLikeUtils.invalidEscapeCharacter(escape)
+              val escape = operands(2).literalValue.get.toString
+              if (escape.isEmpty) {
+                None
+              } else {
+                if (escape.length > 1) {
+                  throw SqlLikeUtils.invalidEscapeCharacter(escape)
+                }
+                Some(escape.charAt(escape.length - 1))
               }
-              val escapeChar = escape.charAt(escape.length - 1)
-              var matched = true
-              var i = 0
-              val newBuilder = new StringBuilder
-              while (i < pattern.length && matched) {
-                val c = pattern.charAt(i)
-                if (c == escapeChar) {
-                  if (i == (pattern.length - 1)) {
-                    throw SqlLikeUtils.invalidEscapeSequence(pattern, i)
-                  }
-                  val nextChar = pattern.charAt(i + 1)
-                  if (nextChar == '%') {
+            }
+
+            escapeCharOpt match {
+              case None => !pattern.contains("_")
+              case Some(escapeChar) =>
+                var matched = true
+                var i = 0
+                val newBuilder = new StringBuilder
+                while (i < pattern.length && matched) {
+                  val c = pattern.charAt(i)
+                  if (c == escapeChar) {
+                    if (i == (pattern.length - 1)) {
+                      throw SqlLikeUtils.invalidEscapeSequence(pattern, i)
+                    }
+                    val nextChar = pattern.charAt(i + 1)
+                    if (nextChar == '%') {
+                      matched = false
+                    } else if ((nextChar == '_') || (nextChar == escapeChar)) {
+                      newBuilder.append(nextChar)
+                      i += 1
+                    } else {
+                      throw SqlLikeUtils.invalidEscapeSequence(pattern, i)
+                    }
+                  } else if (c == '_') {
                     matched = false
-                  } else if ((nextChar == '_') || (nextChar == escapeChar)) {
-                    newBuilder.append(nextChar)
-                    i += 1
                   } else {
-                    throw SqlLikeUtils.invalidEscapeSequence(pattern, i)
+                    newBuilder.append(c)
                   }
-                } else if (c == '_') {
-                  matched = false
-                } else {
-                  newBuilder.append(c)
+                  i += 1
                 }
-                i += 1
-              }
 
-              if (matched) {
-                newPattern = newBuilder.toString
-              }
-              matched
+                if (matched) {
+                  newPattern = newBuilder.toString
+                }
+                matched
             }
           }
 

flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/expressions/ScalarFunctionsTest.scala

@@ -408,11 +408,28 @@ class ScalarFunctionsTest extends ScalarTypesTestBase {
     testAllApis("abcxxxdef".like("%abc%qef%"), "'abcxxxdef' LIKE '%abc%qef%'", "FALSE")
     testAllApis("abcxxxdef".like("abc%qef"), "'abcxxxdef' LIKE 'abc%qef'", "FALSE")
 
-    // reported in FLINK-36100
+    // reported in FLINK-36100: verify default escape char '\' behavior
+    // \_ means literal '_', \% means literal '%'
     testAllApis("TE_ST".like("%E_S%"), "'TE_ST' LIKE '%E_S%'", "TRUE")
     testAllApis("TE-ST".like("%E_S%"), "'TE-ST' LIKE '%E_S%'", "TRUE")
     testAllApis("TE_ST".like("%E\\_S%"), "'TE_ST' LIKE '%E\\_S%'", "TRUE")
     testAllApis("TE-ST".like("%E\\_S%"), "'TE-ST' LIKE '%E\\_S%'", "FALSE")
+    testAllApis("TE%ST".like("TE\\%ST"), "'TE%ST' LIKE 'TE\\%ST'", "TRUE")
+    testAllApis("TExST".like("TE\\%ST"), "'TExST' LIKE 'TE\\%ST'", "FALSE")
+
+    // escape character at the end
+    testExpectedAllApisException(
+      "TE-ST".like("%E-S%\\"),
+      "'TE-ST' LIKE '%E-S%\\'",
+      "",
+      classOf[RuntimeException])
+
+    // invalid character after escape character
+    testExpectedAllApisException(
+      "TEST".like("%E\\S%"),
+      "'TEST' LIKE '%E\\S%'",
+      "Invalid escape",
+      classOf[RuntimeException])
   }
 
   @Test
@@ -429,6 +446,13 @@ class ScalarFunctionsTest extends ScalarTypesTestBase {
 
   @Test
   def testLikeWithEscape(): Unit = {
+    // empty ESCAPE string: no escape character, _ and % retain wildcard semantics
+    testAllApis("abc".like("abc", ""), "'abc' LIKE 'abc' ESCAPE ''", "TRUE")
+    testAllApis("abd".like("abc", ""), "'abd' LIKE 'abc' ESCAPE ''", "FALSE")
+    testAllApis("abcdef".like("abc%", ""), "'abcdef' LIKE 'abc%' ESCAPE ''", "TRUE")
+    testAllApis("abcdef".like("%def", ""), "'abcdef' LIKE '%def' ESCAPE ''", "TRUE")
+    testAllApis("abcdef".like("%cd%", ""), "'abcdef' LIKE '%cd%' ESCAPE ''", "TRUE")
+
     testAllApis('f23.like("&%Th_s%", "&"), "f23 LIKE '&%Th_s%' ESCAPE '&'", "TRUE")
     testAllApis('f23.like("&%%is a%", "&"), "f23 LIKE '&%%is a%' ESCAPE '&'", "TRUE")
     testAllApis('f0.like("Th_s%", "&"), "f0 LIKE 'Th_s%' ESCAPE '&'", "TRUE")