[SPARK-55453][SQL] Fix LIKE pattern matching for supplementary Unicode characters

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/expressions.scala

@@ -828,8 +828,10 @@ object LikeSimplification extends Rule[LogicalPlan] with PredicateHelper {
           Some(EndsWith(input, Literal.create(postfix, input.dataType)))
         // 'a%a' pattern is basically same with 'a%' && '%a'.
         // However, the additional `Length` condition is required to prevent 'a' match 'a%a'.
-        case startsAndEndsWith(prefix, postfix) => Some(
-          And(GreaterThanOrEqual(Length(input), Literal.create(prefix.length + postfix.length)),
+        case startsAndEndsWith(prefix, postfix) =>
+          Some(And(GreaterThanOrEqual(Length(input),
+            Literal.create(prefix.codePointCount(0, prefix.length)
+              + postfix.codePointCount(0, postfix.length))),
           And(StartsWith(input, Literal.create(prefix, input.dataType)),
             EndsWith(input, Literal.create(postfix, input.dataType)))))
         case contains(infix) =>

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/StringUtils.scala

@@ -46,13 +46,13 @@ object StringUtils extends Logging {
    * @return the equivalent Java regular expression of the pattern
    */
   def escapeLikeRegex(pattern: String, escapeChar: Char): String = {
-    val in = pattern.iterator
+    val in = pattern.codePoints().iterator()
     val out = new StringBuilder()
 
     while (in.hasNext) {
-      in.next() match {
+      in.nextInt() match {
         case c1 if c1 == escapeChar && in.hasNext =>
-          val c = in.next()
+          val c = in.nextInt()
           c match {
             case '_' | '%' => out ++= Pattern.quote(Character.toString(c))
             case c if c == escapeChar => out ++= Pattern.quote(Character.toString(c))

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/RegexpExpressionsSuite.scala

@@ -153,6 +153,25 @@ class RegexpExpressionsSuite extends SparkFunSuite with ExpressionEvalHelper {
     checkLiteralRow("a\u20ACa" like _, "_€_", true)
     // scalastyle:on nonascii
 
+    // multi-byte UTF-8 characters
+    // scalastyle:off nonascii
+    checkLiteralRow("😀😇🥑" like _, "%🥑", true)
+    checkLiteralRow("😀😇🥑" like _, "😀%", true)
+    checkLiteralRow("😀😇🥑" like _, "😀_🥑", true)
+    checkLiteralRow("😀" like _, "😀", true)
+    checkLiteralRow("😀" like _, "_", true)
+    checkLiteralRow("😀😇🥑" like _, "___", true)
+    checkLiteralRow("😀😇🥑" like _, "__", false)
+    checkLiteralRow("😀😇🥑" like _, "____", false)
+    checkLiteralRow("a😀b" like _, "a_b", true)
+    checkLiteralRow("😀😇🥑" like _, "😀😇🥑", true)
+    checkLiteralRow("😀😇🥑" like _, "😀😇🥒", false)
+    checkLiteralRow("😀🥑" like _, "😀%🥑", true)
+    checkLiteralRow("😀abc🥑" like _, "😀%🥑", true)
+    checkLiteralRow("😀😇🥑" like _, "😀%🥑", true)
+    checkLiteralRow("🥑" like _, "😀%🥑", false)
+    // scalastyle:on nonascii
+
     // invalid escaping
     checkError(
       exception = intercept[AnalysisException] {

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/LikeSimplificationSuite.scala

@@ -309,4 +309,37 @@ class LikeSimplificationSuite extends PlanTest {
 
     comparePlans(Optimize.execute(originalQuery), originalQuery)
   }
+
+  // scalastyle:off nonascii
+  test("LikeSimplification with emojis") {
+    val originalQuery =
+      testRelation
+        .where($"a" like "😀%🥑")
+
+    val optimized = Optimize.execute(originalQuery.analyze)
+
+    val correctAnswer = testRelation
+      .where(Length($"a") >= 2 && (StartsWith($"a", "😀") && EndsWith($"a", "🥑")))
+      .analyze
+    comparePlans(optimized, correctAnswer)
+  }
+
+  test("LikeSimplification StartsWith/EndsWith/Contains with emojis") {
+    comparePlans(
+      Optimize.execute(testRelation.where($"a" like "😀%").analyze),
+      testRelation.where(StartsWith($"a", "😀")).analyze)
+
+    comparePlans(
+      Optimize.execute(testRelation.where($"a" like "%🥑").analyze),
+      testRelation.where(EndsWith($"a", "🥑")).analyze)
+
+    comparePlans(
+      Optimize.execute(testRelation.where($"a" like "%😇%").analyze),
+      testRelation.where(Contains($"a", "😇")).analyze)
+
+    comparePlans(
+      Optimize.execute(testRelation.where($"a" like "😀😇🥑").analyze),
+      testRelation.where($"a" === "😀😇🥑").analyze)
+  }
+  // scalastyle:on nonascii
 }

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/util/StringUtilsSuite.scala

@@ -53,6 +53,12 @@ class StringUtilsSuite extends SparkFunSuite with SQLHelper {
     assert(escapeLikeRegex("a_b", '\\') === expectedEscapedStrSeven)
     assert(escapeLikeRegex("a_b", '/') === expectedEscapedStrSeven)
     assert(escapeLikeRegex("a_b", '\"') === expectedEscapedStrSeven)
+
+    // scalastyle:off nonascii
+    assert(escapeLikeRegex("😀", '\\') === "(?s)\\Q😀\\E")
+    assert(escapeLikeRegex("_😀%", '\\') === "(?s).\\Q😀\\E.*")
+    assert(escapeLikeRegex("😀😇🥑", '\\') === "(?s)\\Q😀\\E\\Q😇\\E\\Q🥑\\E")
+    // scalastyle:on nonascii
   }
 
   test("filter pattern") {