[FLINK-39360][table] LIKE clause doesn't support some patterns

snuyanzin · web-flow · commit b9756414ba1e · 2026-03-31T16:35:56.000+02:00
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/functions/SqlLikeUtils.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/functions/SqlLikeUtils.java
@@ -18,6 +18,7 @@
 package org.apache.flink.table.functions;
 
 import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.ValidationException;
 
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
@@ -140,11 +141,11 @@ static String sqlToRegexLike(String sqlPattern, char escapeChar) {
     }
 
     public static RuntimeException invalidEscapeCharacter(String s) {
-        return new RuntimeException("Invalid escape character '" + s + "'");
+        return new ValidationException("Invalid escape character '" + s + "'");
     }
 
     public static RuntimeException invalidEscapeSequence(String s, int i) {
-        return new RuntimeException("Invalid escape sequence '" + s + "', " + i);
+        return new ValidationException("Invalid escape sequence '" + s + "', " + i);
     }
 
     private static void similarEscapeRuleChecking(String sqlPattern, char escapeChar) {
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/CodeGeneratorContext.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/CodeGeneratorContext.scala
@@ -32,7 +32,7 @@ import org.apache.flink.table.runtime.util.collections._
 import org.apache.flink.table.types.DataType
 import org.apache.flink.table.types.logical._
 import org.apache.flink.table.types.logical.LogicalTypeRoot._
-import org.apache.flink.table.utils.DateTimeUtils
+import org.apache.flink.table.utils.{DateTimeUtils, EncodingUtils}
 import org.apache.flink.util.InstantiationUtil
 
 import java.time.ZoneId
@@ -988,22 +988,21 @@ class CodeGeneratorContext(
   }
 
   /**
-   * Adds a reusable string constant to the member area of the generated class.
+   * Adds an already pre-escaped string constant to the reusable member area of the generated class.
    *
-   * The string must be already escaped with
-   * [[org.apache.flink.table.utils.EncodingUtils.escapeJava()]].
+   * The string must be already escaped with [[EncodingUtils.escapeJava()]].
    */
-  def addReusableEscapedStringConstant(value: String): String = {
-    reusableStringConstants.get(value) match {
+  def addReusablePreEscapedStringConstant(alreadyEscapedValue: String): String = {
+    reusableStringConstants.get(alreadyEscapedValue) match {
       case Some(field) => field
       case None =>
         val field = newName(this, "str")
         val stmt =
           s"""
-             |private final $BINARY_STRING $field = $BINARY_STRING.fromString("$value");
+             |private final $BINARY_STRING $field = $BINARY_STRING.fromString("$alreadyEscapedValue");
            """.stripMargin
         reusableMemberStatements.add(stmt)
-        reusableStringConstants(value) = field
+        reusableStringConstants(alreadyEscapedValue) = field
         field
     }
   }
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/GenerateUtils.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/GenerateUtils.scala
@@ -306,10 +306,12 @@ object GenerateUtils {
       // as they're not cheap to construct. For the other types, the return term is directly
       // the literal value
       case CHAR | VARCHAR =>
-        val escapedValue =
-          EncodingUtils.escapeJava(literalValue.asInstanceOf[BinaryStringData].toString)
-        val field = ctx.addReusableEscapedStringConstant(escapedValue)
-        generateNonNullLiteral(literalType, field, StringData.fromString(escapedValue))
+        val str = literalValue.asInstanceOf[BinaryStringData]
+        val field = ctx.addReusablePreEscapedStringConstant(EncodingUtils.escapeJava(str.toString))
+        // The original value should be passed as literalValue
+        // all required escaping should be done in corresponding code generation,
+        // so that the literalValue can be also used directly when needed
+        generateNonNullLiteral(literalType, field, str)
 
       case BINARY | VARBINARY =>
         val bytesVal = literalValue.asInstanceOf[Array[Byte]]
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/GeneratedExpression.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/GeneratedExpression.scala
@@ -33,8 +33,12 @@ import org.apache.flink.table.types.logical.LogicalType
  * @param resultType
  *   type of the resultTerm
  * @param literalValue
- *   None if the expression is not literal. Otherwise it represent the original object of the
- *   literal.
+ *   Contains the literal value (as internal data structure) for deep literal inspection if the
+ *   originating expression was a literal. Literal inspection is useful for performance
+ *   optimizations. For example, figuring out whether a time parsing function ever produces
+ *   sub-second data by inspecting the "format" literal string. NOTE: The literal value is not
+ *   intended to be used in generated code, use `resultTerm` for this purpose. The literal value is
+ *   NOT escaped.
  */
 case class GeneratedExpression(
     resultTerm: String,
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/LikeCallGen.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/LikeCallGen.scala
@@ -23,6 +23,7 @@ import org.apache.flink.table.planner.codegen.CodeGenUtils.{className, newName,
 import org.apache.flink.table.planner.codegen.GenerateUtils.generateCallIfArgsNotNull
 import org.apache.flink.table.runtime.functions.SqlLikeChainChecker
 import org.apache.flink.table.types.logical.{BooleanType, LogicalType}
+import org.apache.flink.table.utils.EncodingUtils
 
 import java.util.regex.Pattern
 
@@ -57,68 +58,73 @@ class LikeCallGen extends CallGenerator {
             !pattern.contains("_")
           } else {
             val escape = operands(2).literalValue.get.toString
-            if ((escape.length == 2 && escape.charAt(0) != '\\') || escape.length > 2) {
-              throw SqlLikeUtils.invalidEscapeCharacter(escape)
-            }
-            val escapeChar = escape.charAt(escape.length - 1)
-            var matched = true
-            var i = 0
-            val newBuilder = new StringBuilder
-            while (i < pattern.length && matched) {
-              var c = pattern.charAt(i)
-              if (c == '\\') {
-                i += 1
-                c = pattern.charAt(i)
+            if (escape.isEmpty) {
+              !pattern.contains("_")
+            } else {
+              if (escape.length > 1) {
+                throw SqlLikeUtils.invalidEscapeCharacter(escape)
               }
-              if (c == escapeChar) {
-                if (i == (pattern.length - 1)) {
-                  throw SqlLikeUtils.invalidEscapeSequence(pattern, i)
-                }
-                val nextChar = pattern.charAt(i + 1)
-                if (nextChar == '%') {
+              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 == '%') {
+                    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
+              if (matched) {
+                newPattern = newBuilder.toString
+              }
+              matched
             }
-            matched
           }
 
           if (allowQuick) {
             val noneMatcher = NONE_PATTERN.matcher(newPattern)
             val beginMatcher = BEGIN_PATTERN.matcher(newPattern)
             val endMatcher = END_PATTERN.matcher(newPattern)
             val middleMatcher = MIDDLE_PATTERN.matcher(newPattern)
+            val escapedNewPattern = EncodingUtils.escapeJava(newPattern)
 
             if (noneMatcher.matches()) {
-              val reusePattern = ctx.addReusableEscapedStringConstant(newPattern)
+              val reusePattern = ctx.addReusablePreEscapedStringConstant(escapedNewPattern)
               s"${terms.head}.equals($reusePattern)"
             } else if (beginMatcher.matches()) {
-              val field = ctx.addReusableEscapedStringConstant(beginMatcher.group(1))
+              val escapedStartValue = EncodingUtils.escapeJava(beginMatcher.group(1))
+              val field = ctx.addReusablePreEscapedStringConstant(escapedStartValue)
               s"${terms.head}.startsWith($field)"
             } else if (endMatcher.matches()) {
-              val field = ctx.addReusableEscapedStringConstant(endMatcher.group(1))
+              val escapedEndValue = EncodingUtils.escapeJava(endMatcher.group(1))
+              val field = ctx.addReusablePreEscapedStringConstant(escapedEndValue)
               s"${terms.head}.endsWith($field)"
             } else if (middleMatcher.matches()) {
-              val field = ctx.addReusableEscapedStringConstant(middleMatcher.group(1))
+              val escapedMiddleValue = EncodingUtils.escapeJava(middleMatcher.group(1))
+              val field = ctx.addReusablePreEscapedStringConstant(escapedMiddleValue)
               s"${terms.head}.contains($field)"
             } else {
               val field = className[SqlLikeChainChecker]
               val checker = newName(ctx, "likeChainChecker")
-              ctx.addReusableMember(s"$field $checker = new $field(${"\""}$newPattern${"\""});")
+              ctx.addReusableMember(
+                s"$field $checker = new $field(${"\""}$escapedNewPattern${"\""});")
               s"$checker.check(${terms.head})"
             }
           } else {
@@ -129,15 +135,18 @@ class LikeCallGen extends CallGenerator {
             val escape = if (operands.size == 2) {
               "null"
             } else {
+              val escapedEscapeLiteral =
+                EncodingUtils.escapeJava(operands(2).literalValue.get.toString)
               s"""
-                 |"${operands(2).literalValue.get}"
+                 |"$escapedEscapeLiteral"
                """.stripMargin
             }
+            val escapedPatternLiteral = EncodingUtils.escapeJava(pattern)
             ctx.addReusableMember(
               s"""
                  |$patternClass $patternName =
                  |  $patternClass.compile(
-                 |    $likeClass.sqlToRegexLike("${operands(1).literalValue.get}", $escape));
+                 |    $likeClass.sqlToRegexLike("$escapedPatternLiteral", $escape));
                  |""".stripMargin)
             s"$patternName.matcher(${terms.head}.toString()).matches()"
           }
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala
@@ -19,7 +19,7 @@ package org.apache.flink.table.planner.codegen.calls
 
 import org.apache.flink.table.api.ValidationException
 import org.apache.flink.table.api.config.ExecutionConfigOptions
-import org.apache.flink.table.data.binary.BinaryArrayData
+import org.apache.flink.table.data.binary.{BinaryArrayData, BinaryStringData}
 import org.apache.flink.table.data.util.MapDataUtil
 import org.apache.flink.table.data.utils.CastExecutor
 import org.apache.flink.table.data.writer.{BinaryArrayWriter, BinaryRowWriter}
@@ -41,6 +41,7 @@ import org.apache.flink.table.types.logical.utils.LogicalTypeChecks
 import org.apache.flink.table.types.logical.utils.LogicalTypeChecks.{getFieldTypes, getPrecision, getScale}
 import org.apache.flink.table.types.logical.utils.LogicalTypeMerging.findCommonType
 import org.apache.flink.table.utils.DateTimeUtils.MILLIS_PER_DAY
+import org.apache.flink.table.utils.EncodingUtils
 import org.apache.flink.types.ColumnList
 import org.apache.flink.util.Preconditions.checkArgument
 
@@ -1803,6 +1804,7 @@ object ScalarOperatorGens {
     }
 
     try {
+      // No escaping here as it will be done in the primitiveLiteralForType according to the type of the literal value.
       val result = castExecutor.cast(literalExpr.literalValue.get)
       val resultTerm = newName(ctx, "stringToTime")
 
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/LikeFunctionITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/LikeFunctionITCase.java
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/SqlLikeChainChecker.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/SqlLikeChainChecker.java

Original file line number	Diff line number	Diff line change
`@@ -18,6 +18,7 @@`
`18`	`18`	`package org.apache.flink.table.functions;`
`19`	`19`
`20`	`20`	`import org.apache.flink.annotation.Internal;`
	`21`	`+import org.apache.flink.table.api.ValidationException;`
`21`	`22`
`22`	`23`	`import java.util.regex.Matcher;`
`23`	`24`	`import java.util.regex.Pattern;`
`@@ -140,11 +141,11 @@ static String sqlToRegexLike(String sqlPattern, char escapeChar) {`
`140`	`141`	`}`
`141`	`142`
`142`	`143`	`public static RuntimeException invalidEscapeCharacter(String s) {`
`143`		`- return new RuntimeException("Invalid escape character '" + s + "'");`
	`144`	`+ return new ValidationException("Invalid escape character '" + s + "'");`
`144`	`145`	`}`
`145`	`146`
`146`	`147`	`public static RuntimeException invalidEscapeSequence(String s, int i) {`
`147`		`- return new RuntimeException("Invalid escape sequence '" + s + "', " + i);`
	`148`	`+ return new ValidationException("Invalid escape sequence '" + s + "', " + i);`
`148`	`149`	`}`
`149`	`150`
`150`	`151`	`private static void similarEscapeRuleChecking(String sqlPattern, char escapeChar) {`