[FLINK-39424][table] Setting LIKE does not support default escape characters

---------

Co-authored-by: Au-Miner <75485809+Au-Miner@users.noreply.github.com>
Co-authored-by: Timo Walther <twalthr@apache.org>

Author: 3 people

docs/data/sql_functions.yml

@@ -50,9 +50,9 @@ comparison:
     description: By default (or with the ASYMMETRIC keyword), returns TRUE if value1 is less than value2 or greater than value3. With the SYMMETRIC keyword, returns TRUE if value1 is not inclusively between value2 and value3. When either value2 or value3 is NULL, returns TRUE or UNKNOWN. E.g., 12 NOT BETWEEN 15 AND 12 returns TRUE; 12 NOT BETWEEN SYMMETRIC 15 AND 12 returns FALSE; 12 NOT BETWEEN NULL AND 15 returns UNKNOWN; 12 NOT BETWEEN 15 AND NULL returns TRUE; 12 NOT BETWEEN SYMMETRIC 12 AND NULL returns UNKNOWN.
   - sql: string1 LIKE string2 [ ESCAPE char ]
     table: string1.like(string2[, char])
-    description: Returns TRUE if string1 matches pattern string2; returns UNKNOWN if string1 or string2 is NULL. An escape character consisting of a single char can be defined if necessary, '\' by default.
+    description: Returns TRUE if string1 matches pattern string2; returns UNKNOWN if string1 or string2 is NULL. An escape character consisting of a single char can be defined using the ESCAPE clause if necessary. There is no default escape character.
   - sql: string1 NOT LIKE string2 [ ESCAPE char ]
-    description: Returns TRUE if string1 does not match pattern string2; returns UNKNOWN if string1 or string2 is NULL. An escape character consisting of a single char can be defined if necessary, '\' by default.
+    description: Returns TRUE if string1 does not match pattern string2; returns UNKNOWN if string1 or string2 is NULL. An escape character consisting of a single char can be defined using the ESCAPE clause if necessary. There is no default escape character.
   - sql: string1 SIMILAR TO string2 [ ESCAPE char ]
     table: string1.similar(string2)
     description: Returns TRUE if string1 matches SQL regular expression string2; returns UNKNOWN if string1 or string2 is NULL. An escape character can be defined if necessary. The escape character has not been supported yet.

docs/data/sql_functions_zh.yml

@@ -72,11 +72,11 @@ comparison:
     table: string1.like(string2[, char])
     description: |
       如果 string1 匹配 string2 返回 `TRUE`；如果 string1 或 string2 为 `NULL` 返回 `UNKNOWN`。
-      如果需要可以定义包含单个字符的转义字符，默认为 '\'。
+      如果需要，可以通过 ESCAPE 子句定义包含单个字符的转义字符。默认没有转义字符。
   - sql: string1 NOT LIKE string2 [ ESCAPE char ]
     description: |
       如果 string1 与 string2 不匹配返回 `TRUE`；如果 string1 或 string2 为 `NULL` 返回 `UNKNOWN`。
-      如果需要可以定义包含单个字符的转义字符，默认为 '\'。
+      如果需要，可以通过 ESCAPE 子句定义包含单个字符的转义字符。默认没有转义字符。
   - sql: string1 SIMILAR TO string2 [ ESCAPE char ]
     table: string1.similar(string2)
     description: |

flink-python/pyflink/table/expression.py

@@ -1192,10 +1192,10 @@ def like(self,
              pattern: Union[str, 'Expression[str]'] = None,
              escape=None) -> 'Expression[bool]':
         """
-        Returns true, if a string matches the specified LIKE pattern
+        Returns true, if a string matches the specified LIKE pattern.
         e.g. 'Jo_n%' matches all strings that start with 'Jo(arbitrary letter)n'.
-        An escape character consisting of a single char can be defined if necessary,
-        '\\' by default.
+        An escape character consisting of a single char can be defined if necessary.
+        There is no default escape character.
         """
         if escape is None:
             return _binary_op("like")(self, pattern)

flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java

@@ -1056,8 +1056,8 @@ public OutType initCap() {
     }
 
     /**
-     * Returns true, if a string matches the specified LIKE pattern with default escape character
-     * '/'.
+     * Returns true, if a string matches the specified LIKE pattern. There is no default escape
+     * character.
      *
      * <p>e.g. "Jo_n%" matches all strings that start with "Jo(arbitrary letter)n"
      */

flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/functions/SqlLikeUtils.java

@@ -96,8 +96,11 @@ public static String sqlToRegexLike(String sqlPattern, CharSequence escapeStr) {
                 throw invalidEscapeCharacter(escapeStr.toString());
             }
             escapeChar = escapeStr.charAt(0);
+            if (escapeChar == 0) {
+                throw invalidEscapeCharacter(escapeStr.toString());
+            }
         } else {
-            escapeChar = '\\';
+            escapeChar = 0;
         }
         return sqlToRegexLike(sqlPattern, escapeChar);
     }
@@ -109,7 +112,7 @@ static String sqlToRegexLike(String sqlPattern, char escapeChar) {
         final StringBuilder javaPattern = new StringBuilder(len + len);
         for (i = 0; i < len; i++) {
             char c = sqlPattern.charAt(i);
-            if (c == escapeChar) {
+            if (c == escapeChar && escapeChar != 0) {
                 if (i == (sqlPattern.length() - 1)) {
                     throw invalidEscapeSequence(sqlPattern, i);
                 }
@@ -192,7 +195,7 @@ private static int sqlSimilarRewriteCharEnumeration(
             char c = sqlPattern.charAt(i);
             if (c == ']') {
                 return i - 1;
-            } else if (c == escapeChar) {
+            } else if (c == escapeChar && escapeChar != 0) {
                 i++;
                 char nextChar = sqlPattern.charAt(i);
                 if (SQL_SIMILAR_SPECIALS.indexOf(nextChar) >= 0) {
@@ -241,6 +244,9 @@ public static String sqlToRegexSimilar(String sqlPattern, CharSequence escapeStr
                 throw invalidEscapeCharacter(escapeStr.toString());
             }
             escapeChar = escapeStr.charAt(0);
+            if (escapeChar == 0) {
+                throw invalidEscapeCharacter(escapeStr.toString());
+            }
         } else {
             escapeChar = 0;
         }
@@ -256,7 +262,7 @@ public static String sqlToRegexSimilar(String sqlPattern, char escapeChar) {
         final int len = sqlPattern.length();
         for (int i = 0; i < len; i++) {
             char c = sqlPattern.charAt(i);
-            if (c == escapeChar) {
+            if (c == escapeChar && escapeChar != 0) {
                 if (i == (len - 1)) {
                     // It should never reach here after the escape rule
                     // checking.

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

@@ -65,6 +65,9 @@ class LikeCallGen extends CallGenerator {
                 throw SqlLikeUtils.invalidEscapeCharacter(escape)
               }
               val escapeChar = escape.charAt(escape.length - 1)
+              if (escapeChar == 0) {
+                throw SqlLikeUtils.invalidEscapeCharacter(escape)
+              }
               var matched = true
               var i = 0
               val newBuilder = new StringBuilder

flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/calcite/FlinkSqlLikeUtilsTest.java

@@ -36,9 +36,14 @@ void testSqlLike() {
         assertThat(SqlLikeUtils.like("abcd", "a.*d", "\\")).isEqualTo(false);
         assertThat(SqlLikeUtils.like("abcde", "%c.e", "\\")).isEqualTo(false);
 
-        // default escape character
+        // no default escape character - backslash is treated as a literal character
         assertThat(SqlLikeUtils.like("a-c", "a\\_c")).isEqualTo(false);
-        assertThat(SqlLikeUtils.like("a_c", "a\\_c")).isEqualTo(true);
+        assertThat(SqlLikeUtils.like("a_c", "a\\_c")).isEqualTo(false);
+        assertThat(SqlLikeUtils.like("a\\_c", "a\\_c")).isEqualTo(true);
+
+        // default escape also excludes \u0000
+        assertThat(SqlLikeUtils.like("_", "\u0000_", null)).isEqualTo(false);
+        assertThat(SqlLikeUtils.like("\u0000x", "\u0000_", null)).isEqualTo(true);
 
         // -------------------------------- sqlToRegexLike ----------------------------------------
 
@@ -66,5 +71,8 @@ void testSqlLike() {
         assertThat(SqlLikeUtils.similar("abc", "a.c", "\\")).isEqualTo(true);
         assertThat(SqlLikeUtils.similar("a.c", "a.c", "\\")).isEqualTo(true);
         assertThat(SqlLikeUtils.similar("abcd", "a.*d", "\\")).isEqualTo(true);
+        // default escape also excludes \u0000
+        assertThat(SqlLikeUtils.similar("_", "\u0000_", null)).isEqualTo(false);
+        assertThat(SqlLikeUtils.similar("\u0000x", "\u0000_", null)).isEqualTo(true);
     }
 }

flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/LikeFunctionITCase.java

@@ -204,6 +204,10 @@ private Stream<TestSetSpec> withEscape() {
                         .testSqlResult(
                                 "'\btest\ne\\nd\f' LIKE '\btest\ne\\nd\f' ESCAPE '!'",
                                 true,
-                                DataTypes.BOOLEAN().notNull()));
+                                DataTypes.BOOLEAN().notNull())
+                        // Invalid escape character
+                        .testSqlValidationError(
+                                "f0 LIKE 'test' ESCAPE '\u0000'",
+                                "Invalid escape character '\u0000'"));
     }
 }

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

@@ -408,22 +408,24 @@ 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 - without ESCAPE clause, '\' is a literal character
     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("%E\\_S%"), "'TE-ST' LIKE '%E\\_S%'", "FALSE")
+    testAllApis("\u0000".like("\u0000"), "'\u0000' LIKE '\u0000'", "TRUE")
+    testAllApis("a".like("\u0000_"), "'a' LIKE '\u0000_'", "FALSE")
   }
 
   @Test
   def testNotLike(): Unit = {
     testAllApis(!'f0.like("Th_s%"), "f0 NOT LIKE 'Th_s%'", "FALSE")
     testAllApis(!'f0.like("%is a%"), "f0 NOT LIKE '%is a%'", "FALSE")
 
-    // reported in FLINK-36100
+    // reported in FLINK-36100 - without ESCAPE clause, '\' is a literal character
     testSqlApi("'TE_ST' NOT LIKE '%E_S%'", "FALSE")
     testSqlApi("'TE-ST' NOT LIKE '%E_S%'", "FALSE")
-    testSqlApi("'TE_ST' NOT LIKE '%E\\_S%'", "FALSE")
+    testSqlApi("'TE_ST' NOT LIKE '%E\\_S%'", "TRUE")
     testSqlApi("'TE-ST' NOT LIKE '%E\\_S%'", "TRUE")
   }
 
@@ -443,6 +445,8 @@ class ScalarFunctionsTest extends ScalarTypesTestBase {
     testAllApis("TE_ST".like("%E__S%", "_"), "'TE_ST' LIKE '%E__S%' ESCAPE '_'", "TRUE")
     testAllApis("TE-ST".like("TE%_ST", "%"), "'TE-ST' LIKE 'TE%_ST' ESCAPE '%'", "FALSE")
     testAllApis("TE_ST".like("TE%_ST", "%"), "'TE_ST' LIKE 'TE%_ST' ESCAPE '%'", "TRUE")
+    testAllApis("TE-ST".like("%E*_S%", "*"), "'TE-ST' LIKE '%E*_S%' ESCAPE '*'", "FALSE")
+    testAllApis("TE_ST".like("%E*_S%", "*"), "'TE_ST' LIKE '%E*_S%' ESCAPE '*'", "TRUE")
 
     // special character in Java Regex
     testAllApis("TE-ST".like("%E\\_S%", "\\"), "'TE-ST' LIKE '%E\\_S%' ESCAPE '\\'", "FALSE")
@@ -498,6 +502,8 @@ class ScalarFunctionsTest extends ScalarTypesTestBase {
     testSqlApi("'TE_ST' NOT LIKE '%E__S%' ESCAPE '_'", "FALSE")
     testSqlApi("'TE-ST' NOT LIKE 'TE%_ST' ESCAPE '%'", "TRUE")
     testSqlApi("'TE_ST' NOT LIKE 'TE%_ST' ESCAPE '%'", "FALSE")
+    testSqlApi("'TE-ST' NOT LIKE '%E*_S%' ESCAPE '*'", "TRUE")
+    testSqlApi("'TE_ST' NOT LIKE '%E*_S%' ESCAPE '*'", "FALSE")
 
     // special character in Java Regex
     testSqlApi("'TE-ST' NOT LIKE '%E\\_S%' ESCAPE '\\'", "TRUE")