[SPARK-49207][SQL] Fix one-to-many case mapping in SplitPart and StringSplitSQL

### What changes were proposed in this pull request?
Fix the following string expressions to handle one-to-many case mapping properly:

- SplitPart
- StringSplitSQL

Examples of incorrect results (under `UTF8_LCASE` collation):

```
SplitPart("Ai\u0307B", "İ", 2) // returns: "\u0307B" (incorrect), instead of: "B" (correct)
SplitPart("AİB", "i\u0307", 1) // returns: "AİB", instead of: "A", "B" (correct)

StringSplitSQL("Ai\u0307B", "İ") // returns: ["A", "\u0307B"] (incorrect), instead of: ["A", "B"] (correct)
StringSplitSQL("AİB", "i\u0307") // returns: ["AİB"] (incorrect), instead of: ["A", "B"] (correct)
```

### Why are the changes needed?
Currently, some string expressions are giving wrong results when working with one-to-many case mapping.

### Does this PR introduce _any_ user-facing change?
Yes, this expression will now work properly with surrogate pairs: `split_part`.

### How was this patch tested?
New tests in `CollationSupportSuite`.

### Was this patch authored or co-authored using generative AI tooling?
Yes.

Closes #47715 from uros-db/fix-splitpart.

Authored-by: Uros Bojanic <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>

Author: uros-db

common/unsafe/src/main/java/org/apache/spark/sql/catalyst/util/CollationAwareUTF8String.java

@@ -36,7 +36,6 @@
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
-import java.util.regex.Pattern;
 
 /**
  * Utility class for collation-aware UTF8String operations.
@@ -1208,24 +1207,43 @@ public static UTF8String[] splitSQL(final UTF8String input, final UTF8String del
 
   public static UTF8String[] lowercaseSplitSQL(final UTF8String string, final UTF8String delimiter,
       final int limit) {
-      if (delimiter.numBytes() == 0) return new UTF8String[] { string };
-      if (string.numBytes() == 0) return new UTF8String[] { UTF8String.EMPTY_UTF8 };
-      Pattern pattern = Pattern.compile(Pattern.quote(delimiter.toString()),
-        CollationSupport.lowercaseRegexFlags);
-      String[] splits = pattern.split(string.toString(), limit);
-      UTF8String[] res = new UTF8String[splits.length];
-      for (int i = 0; i < res.length; i++) {
-        res[i] = UTF8String.fromString(splits[i]);
+    if (delimiter.numBytes() == 0) return new UTF8String[] { string };
+    if (string.numBytes() == 0) return new UTF8String[] { UTF8String.EMPTY_UTF8 };
+
+    List<UTF8String> strings = new ArrayList<>();
+    UTF8String lowercaseDelimiter = lowerCaseCodePoints(delimiter);
+    int startIndex = 0, nextMatch = 0, nextMatchLength;
+    while (nextMatch != MATCH_NOT_FOUND) {
+      if (limit > 0 && strings.size() == limit - 1) {
+        break;
+      }
+      nextMatch = lowercaseFind(string, lowercaseDelimiter, startIndex);
+      if (nextMatch != MATCH_NOT_FOUND) {
+        nextMatchLength = lowercaseMatchLengthFrom(string, lowercaseDelimiter, nextMatch);
+        strings.add(string.substring(startIndex, nextMatch));
+        startIndex = nextMatch + nextMatchLength;
       }
-      return res;
+    }
+    if (startIndex <= string.numChars()) {
+      strings.add(string.substring(startIndex, string.numChars()));
+    }
+    if (limit == 0) {
+      // Remove trailing empty strings
+      int i = strings.size() - 1;
+      while (i >= 0 && strings.get(i).numBytes() == 0) {
+        strings.remove(i);
+        i--;
+      }
+    }
+    return strings.toArray(new UTF8String[0]);
   }
 
   public static UTF8String[] icuSplitSQL(final UTF8String string, final UTF8String delimiter,
       final int limit, final int collationId) {
     if (delimiter.numBytes() == 0) return new UTF8String[] { string };
     if (string.numBytes() == 0) return new UTF8String[] { UTF8String.EMPTY_UTF8 };
     List<UTF8String> strings = new ArrayList<>();
-    String target = string.toString(), pattern = delimiter.toString();
+    String target = string.toValidString(), pattern = delimiter.toValidString();
     StringSearch stringSearch = CollationFactory.getStringSearch(target, pattern, collationId);
     int start = 0, end;
     while ((end = stringSearch.next()) != StringSearch.DONE) {

common/unsafe/src/test/java/org/apache/spark/unsafe/types/CollationSupportSuite.java

@@ -567,12 +567,17 @@ public void testEndsWith() throws SparkException {
     assertEndsWith("the İo", "İo", "UTF8_LCASE", true);
   }
 
+  /**
+   * Verify the behaviour of the `StringSplitSQL` collation support class.
+   */
+
   private void assertStringSplitSQL(String str, String delimiter, String collationName,
       UTF8String[] expected) throws SparkException {
     UTF8String s = UTF8String.fromString(str);
     UTF8String d = UTF8String.fromString(delimiter);
     int collationId = CollationFactory.collationNameToId(collationName);
-    assertArrayEquals(expected, CollationSupport.StringSplitSQL.exec(s, d, collationId));
+    UTF8String[] result = CollationSupport.StringSplitSQL.exec(s, d, collationId);
+    assertArrayEquals(expected, result);
   }
 
   @Test
@@ -590,7 +595,21 @@ public void testStringSplitSQL() throws SparkException {
     var array_A_B = new UTF8String[] { UTF8String.fromString("A"), UTF8String.fromString("B") };
     var array_a_e = new UTF8String[] { UTF8String.fromString("ä"), UTF8String.fromString("e") };
     var array_Aa_bB = new UTF8String[] { UTF8String.fromString("Aa"), UTF8String.fromString("bB") };
-    // Edge cases
+    var array_Turkish_uppercase_dotted_I = new UTF8String[] { UTF8String.fromString("İ") };
+    var array_Turkish_lowercase_dotted_i = new UTF8String[] { UTF8String.fromString("i\u0307") };
+    var array_i = new UTF8String[] { UTF8String.fromString("i"), UTF8String.fromString("") };
+    var array_dot = new UTF8String[] { UTF8String.fromString(""), UTF8String.fromString("\u0307") };
+    var array_AiB = new UTF8String[] { UTF8String.fromString("Ai\u0307B") };
+    var array_AIB = new UTF8String[] { UTF8String.fromString("AİB") };
+    var array_small_nonfinal_sigma = new UTF8String[] { UTF8String.fromString("σ") };
+    var array_small_final_sigma = new UTF8String[] { UTF8String.fromString("ς") };
+    var array_capital_sigma = new UTF8String[] { UTF8String.fromString("Σ") };
+    var array_a_b_c = new UTF8String[] { UTF8String.fromString("a"), UTF8String.fromString("b"),
+      UTF8String.fromString("c") };
+    var array_emojis = new UTF8String[] { UTF8String.fromString("😀"), UTF8String.fromString("😄") };
+    var array_AOB = new UTF8String[] { UTF8String.fromString("A𐐅B") };
+    var array_AoB = new UTF8String[] { UTF8String.fromString("A𐐭B") };
+    // Empty strings.
     assertStringSplitSQL("", "", "UTF8_BINARY", empty_match);
     assertStringSplitSQL("abc", "", "UTF8_BINARY", array_abc);
     assertStringSplitSQL("", "abc", "UTF8_BINARY", empty_match);
@@ -603,7 +622,7 @@ public void testStringSplitSQL() throws SparkException {
     assertStringSplitSQL("", "", "UNICODE_CI", empty_match);
     assertStringSplitSQL("abc", "", "UNICODE_CI", array_abc);
     assertStringSplitSQL("", "abc", "UNICODE_CI", empty_match);
-    // Basic tests
+    // Basic tests.
     assertStringSplitSQL("1a2", "a", "UTF8_BINARY", array_1_2);
     assertStringSplitSQL("1a2", "A", "UTF8_BINARY", array_1a2);
     assertStringSplitSQL("1a2", "b", "UTF8_BINARY", array_1a2);
@@ -617,25 +636,7 @@ public void testStringSplitSQL() throws SparkException {
     assertStringSplitSQL("1a2", "A", "UNICODE_CI", array_1_2);
     assertStringSplitSQL("1a2", "1A2", "UNICODE_CI", full_match);
     assertStringSplitSQL("1a2", "123", "UNICODE_CI", array_1a2);
-    // Case variation
-    assertStringSplitSQL("AaXbB", "x", "UTF8_BINARY", array_AaXbB);
-    assertStringSplitSQL("AaXbB", "X", "UTF8_BINARY", array_Aa_bB);
-    assertStringSplitSQL("AaXbB", "axb", "UNICODE", array_AaXbB);
-    assertStringSplitSQL("AaXbB", "aXb", "UNICODE", array_A_B);
-    assertStringSplitSQL("AaXbB", "axb", "UTF8_LCASE", array_A_B);
-    assertStringSplitSQL("AaXbB", "AXB", "UTF8_LCASE", array_A_B);
-    assertStringSplitSQL("AaXbB", "axb", "UNICODE_CI", array_A_B);
-    assertStringSplitSQL("AaXbB", "AxB", "UNICODE_CI", array_A_B);
-    // Accent variation
-    assertStringSplitSQL("aBcDe", "bćd", "UTF8_BINARY", array_aBcDe);
-    assertStringSplitSQL("aBcDe", "BćD", "UTF8_BINARY", array_aBcDe);
-    assertStringSplitSQL("aBcDe", "abćde", "UNICODE", array_aBcDe);
-    assertStringSplitSQL("aBcDe", "aBćDe", "UNICODE", array_aBcDe);
-    assertStringSplitSQL("aBcDe", "bćd", "UTF8_LCASE", array_aBcDe);
-    assertStringSplitSQL("aBcDe", "BĆD", "UTF8_LCASE", array_aBcDe);
-    assertStringSplitSQL("aBcDe", "abćde", "UNICODE_CI", array_aBcDe);
-    assertStringSplitSQL("aBcDe", "AbĆdE", "UNICODE_CI", array_aBcDe);
-    // Variable byte length characters
+    // Advanced tests.
     assertStringSplitSQL("äb世De", "b世D", "UTF8_BINARY", array_a_e);
     assertStringSplitSQL("äb世De", "B世d", "UTF8_BINARY", array_special);
     assertStringSplitSQL("äbćδe", "bćδ", "UTF8_BINARY", array_a_e);
@@ -652,6 +653,115 @@ public void testStringSplitSQL() throws SparkException {
     assertStringSplitSQL("äb世De", "AB世dE", "UNICODE_CI", array_special);
     assertStringSplitSQL("äbćδe", "ÄbćδE", "UNICODE_CI", full_match);
     assertStringSplitSQL("äbćδe", "ÄBcΔÉ", "UNICODE_CI", array_abcde);
+    // Case variation.
+    assertStringSplitSQL("AaXbB", "x", "UTF8_BINARY", array_AaXbB);
+    assertStringSplitSQL("AaXbB", "X", "UTF8_BINARY", array_Aa_bB);
+    assertStringSplitSQL("AaXbB", "axb", "UNICODE", array_AaXbB);
+    assertStringSplitSQL("AaXbB", "aXb", "UNICODE", array_A_B);
+    assertStringSplitSQL("AaXbB", "axb", "UTF8_LCASE", array_A_B);
+    assertStringSplitSQL("AaXbB", "AXB", "UTF8_LCASE", array_A_B);
+    assertStringSplitSQL("AaXbB", "axb", "UNICODE_CI", array_A_B);
+    assertStringSplitSQL("AaXbB", "AxB", "UNICODE_CI", array_A_B);
+    // Accent variation.
+    assertStringSplitSQL("aBcDe", "bćd", "UTF8_BINARY", array_aBcDe);
+    assertStringSplitSQL("aBcDe", "BćD", "UTF8_BINARY", array_aBcDe);
+    assertStringSplitSQL("aBcDe", "abćde", "UNICODE", array_aBcDe);
+    assertStringSplitSQL("aBcDe", "aBćDe", "UNICODE", array_aBcDe);
+    assertStringSplitSQL("aBcDe", "bćd", "UTF8_LCASE", array_aBcDe);
+    assertStringSplitSQL("aBcDe", "BĆD", "UTF8_LCASE", array_aBcDe);
+    assertStringSplitSQL("aBcDe", "abćde", "UNICODE_CI", array_aBcDe);
+    assertStringSplitSQL("aBcDe", "AbĆdE", "UNICODE_CI", array_aBcDe);
+    // One-to-many case mapping (e.g. Turkish dotted I).
+    assertStringSplitSQL("İ", "i", "UTF8_BINARY", array_Turkish_uppercase_dotted_I);
+    assertStringSplitSQL("İ", "i", "UTF8_LCASE", array_Turkish_uppercase_dotted_I);
+    assertStringSplitSQL("İ", "i", "UNICODE", array_Turkish_uppercase_dotted_I);
+    assertStringSplitSQL("İ", "i", "UNICODE_CI", array_Turkish_uppercase_dotted_I);
+    assertStringSplitSQL("İ", "\u0307", "UTF8_BINARY", array_Turkish_uppercase_dotted_I);
+    assertStringSplitSQL("İ", "\u0307", "UTF8_LCASE", array_Turkish_uppercase_dotted_I);
+    assertStringSplitSQL("İ", "\u0307", "UNICODE", array_Turkish_uppercase_dotted_I);
+    assertStringSplitSQL("İ", "\u0307", "UNICODE_CI", array_Turkish_uppercase_dotted_I);
+    assertStringSplitSQL("i\u0307", "i", "UTF8_BINARY", array_dot);
+    assertStringSplitSQL("i\u0307", "i", "UTF8_LCASE", array_dot);
+    assertStringSplitSQL("i\u0307", "i", "UNICODE", array_Turkish_lowercase_dotted_i);
+    assertStringSplitSQL("i\u0307", "i", "UNICODE_CI", array_Turkish_lowercase_dotted_i);
+    assertStringSplitSQL("i\u0307", "\u0307", "UTF8_BINARY", array_i);
+    assertStringSplitSQL("i\u0307", "\u0307", "UTF8_LCASE", array_i);
+    assertStringSplitSQL("i\u0307", "\u0307", "UNICODE", array_Turkish_lowercase_dotted_i);
+    assertStringSplitSQL("i\u0307", "\u0307", "UNICODE_CI", array_Turkish_lowercase_dotted_i);
+    assertStringSplitSQL("AİB", "İ", "UTF8_BINARY", array_A_B);
+    assertStringSplitSQL("AİB", "İ", "UTF8_LCASE", array_A_B);
+    assertStringSplitSQL("AİB", "İ", "UNICODE", array_A_B);
+    assertStringSplitSQL("AİB", "İ", "UNICODE_CI", array_A_B);
+    assertStringSplitSQL("AİB", "i\u0307", "UTF8_BINARY", array_AIB);
+    assertStringSplitSQL("AİB", "i\u0307", "UTF8_LCASE", array_A_B);
+    assertStringSplitSQL("AİB", "i\u0307", "UNICODE", array_AIB);
+    assertStringSplitSQL("AİB", "i\u0307", "UNICODE_CI", array_A_B);
+    assertStringSplitSQL("Ai\u0307B", "İ", "UTF8_BINARY", array_AiB);
+    assertStringSplitSQL("Ai\u0307B", "İ", "UTF8_LCASE", array_A_B);
+    assertStringSplitSQL("Ai\u0307B", "İ", "UNICODE", array_AiB);
+    assertStringSplitSQL("Ai\u0307B", "İ", "UNICODE_CI", array_A_B);
+    assertStringSplitSQL("Ai\u0307B", "i\u0307", "UTF8_BINARY", array_A_B);
+    assertStringSplitSQL("Ai\u0307B", "i\u0307", "UTF8_LCASE", array_A_B);
+    assertStringSplitSQL("Ai\u0307B", "i\u0307", "UNICODE", array_A_B);
+    assertStringSplitSQL("Ai\u0307B", "i\u0307", "UNICODE_CI", array_A_B);
+    // Conditional case mapping (e.g. Greek sigmas).
+    assertStringSplitSQL("σ", "σ", "UTF8_BINARY", full_match);
+    assertStringSplitSQL("σ", "σ", "UTF8_LCASE", full_match);
+    assertStringSplitSQL("σ", "σ", "UNICODE", full_match);
+    assertStringSplitSQL("σ", "σ", "UNICODE_CI", full_match);
+    assertStringSplitSQL("σ", "ς", "UTF8_BINARY", array_small_nonfinal_sigma);
+    assertStringSplitSQL("σ", "ς", "UTF8_LCASE", full_match);
+    assertStringSplitSQL("σ", "ς", "UNICODE", array_small_nonfinal_sigma);
+    assertStringSplitSQL("σ", "ς", "UNICODE_CI", full_match);
+    assertStringSplitSQL("σ", "Σ", "UTF8_BINARY", array_small_nonfinal_sigma);
+    assertStringSplitSQL("σ", "Σ", "UTF8_LCASE", full_match);
+    assertStringSplitSQL("σ", "Σ", "UNICODE", array_small_nonfinal_sigma);
+    assertStringSplitSQL("σ", "Σ", "UNICODE_CI", full_match);
+    assertStringSplitSQL("ς", "σ", "UTF8_BINARY", array_small_final_sigma);
+    assertStringSplitSQL("ς", "σ", "UTF8_LCASE", full_match);
+    assertStringSplitSQL("ς", "σ", "UNICODE", array_small_final_sigma);
+    assertStringSplitSQL("ς", "σ", "UNICODE_CI", full_match);
+    assertStringSplitSQL("ς", "ς", "UTF8_BINARY", full_match);
+    assertStringSplitSQL("ς", "ς", "UTF8_LCASE", full_match);
+    assertStringSplitSQL("ς", "ς", "UNICODE", full_match);
+    assertStringSplitSQL("ς", "ς", "UNICODE_CI", full_match);
+    assertStringSplitSQL("ς", "Σ", "UTF8_BINARY", array_small_final_sigma);
+    assertStringSplitSQL("ς", "Σ", "UTF8_LCASE", full_match);
+    assertStringSplitSQL("ς", "Σ", "UNICODE", array_small_final_sigma);
+    assertStringSplitSQL("ς", "Σ", "UNICODE_CI", full_match);
+    assertStringSplitSQL("Σ", "σ", "UTF8_BINARY", array_capital_sigma);
+    assertStringSplitSQL("Σ", "σ", "UTF8_LCASE", full_match);
+    assertStringSplitSQL("Σ", "σ", "UNICODE", array_capital_sigma);
+    assertStringSplitSQL("Σ", "σ", "UNICODE_CI", full_match);
+    assertStringSplitSQL("Σ", "ς", "UTF8_BINARY", array_capital_sigma);
+    assertStringSplitSQL("Σ", "ς", "UTF8_LCASE", full_match);
+    assertStringSplitSQL("Σ", "ς", "UNICODE", array_capital_sigma);
+    assertStringSplitSQL("Σ", "ς", "UNICODE_CI", full_match);
+    assertStringSplitSQL("Σ", "Σ", "UTF8_BINARY", full_match);
+    assertStringSplitSQL("Σ", "Σ", "UTF8_LCASE", full_match);
+    assertStringSplitSQL("Σ", "Σ", "UNICODE", full_match);
+    assertStringSplitSQL("Σ", "Σ", "UNICODE_CI", full_match);
+    // Surrogate pairs.
+    assertStringSplitSQL("a🙃b🙃c", "🙃", "UTF8_BINARY", array_a_b_c);
+    assertStringSplitSQL("a🙃b🙃c", "🙃", "UTF8_LCASE", array_a_b_c);
+    assertStringSplitSQL("a🙃b🙃c", "🙃", "UNICODE", array_a_b_c);
+    assertStringSplitSQL("a🙃b🙃c", "🙃", "UNICODE_CI", array_a_b_c);
+    assertStringSplitSQL("😀😆😃😄", "😆😃", "UTF8_BINARY", array_emojis);
+    assertStringSplitSQL("😀😆😃😄", "😆😃", "UTF8_LCASE", array_emojis);
+    assertStringSplitSQL("😀😆😃😄", "😆😃", "UNICODE", array_emojis);
+    assertStringSplitSQL("😀😆😃😄", "😆😃", "UNICODE_CI", array_emojis);
+    assertStringSplitSQL("A𐐅B", "𐐅", "UTF8_BINARY", array_A_B);
+    assertStringSplitSQL("A𐐅B", "𐐅", "UTF8_LCASE", array_A_B);
+    assertStringSplitSQL("A𐐅B", "𐐅", "UNICODE", array_A_B);
+    assertStringSplitSQL("A𐐅B", "𐐅", "UNICODE_CI", array_A_B);
+    assertStringSplitSQL("A𐐅B", "𐐭", "UTF8_BINARY", array_AOB);
+    assertStringSplitSQL("A𐐅B", "𐐭", "UTF8_LCASE", array_A_B);
+    assertStringSplitSQL("A𐐅B", "𐐭", "UNICODE", array_AOB);
+    assertStringSplitSQL("A𐐅B", "𐐭", "UNICODE_CI", array_A_B);
+    assertStringSplitSQL("A𐐭B", "𐐅", "UTF8_BINARY", array_AoB);
+    assertStringSplitSQL("A𐐭B", "𐐅", "UTF8_LCASE", array_A_B);
+    assertStringSplitSQL("A𐐭B", "𐐅", "UNICODE", array_AoB);
+    assertStringSplitSQL("A𐐭B", "𐐅", "UNICODE_CI", array_A_B);
   }
 
   private void assertUpper(String target, String collationName, String expected)