From c784b25b98e913fb76dcca7b0f5ece50ca2127d6 Mon Sep 17 00:00:00 2001
From: Au_Miner <358671982@qq.com>
Date: Wed, 27 May 2026 15:34:12 +0800
Subject: [PATCH] [FLINK-36267][table] Fix SPLIT not preserving SMP characters
 when delimiter is empty

---
 .../planner/functions/CollectionFunctionsITCase.java     | 9 ++++++++-
 .../table/runtime/functions/scalar/SplitFunction.java    | 8 ++++++--
 2 files changed, 14 insertions(+), 3 deletions(-)

diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/CollectionFunctionsITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/CollectionFunctionsITCase.java
index f67d422c994a6..42bb85c814d4d 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/CollectionFunctionsITCase.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/CollectionFunctionsITCase.java
@@ -1813,7 +1813,8 @@ private Stream<TestSetSpec> splitTestCases() {
                                 ",123,123,",
                                 123,
                                 "12345",
-                                ",123,,,123,")
+                                ",123,,,123,",
+                                "123😊笑脸")
                         .andDataTypes(
                                 DataTypes.STRING().notNull(),
                                 DataTypes.STRING(),
@@ -1821,6 +1822,7 @@ private Stream<TestSetSpec> splitTestCases() {
                                 DataTypes.STRING().notNull(),
                                 DataTypes.INT().notNull(),
                                 DataTypes.STRING().notNull(),
+                                DataTypes.STRING().notNull(),
                                 DataTypes.STRING().notNull())
                         .testResult(
                                 $("f0").split(","),
@@ -1867,6 +1869,11 @@ private Stream<TestSetSpec> splitTestCases() {
                                 "SPLIT(f6, ',')",
                                 new String[] {"", "123", "", "", "123", ""},
                                 DataTypes.ARRAY(DataTypes.STRING()).notNull())
+                        .testResult(
+                                $("f7").split(""),
+                                "SPLIT(f7, '')",
+                                new String[] {"1", "2", "3", "😊", "笑", "脸"},
+                                DataTypes.ARRAY(DataTypes.STRING()).notNull())
                         .testTableApiValidationError(
                                 $("f4").split(","),
                                 "Invalid input arguments. Expected signatures are:\n"
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/SplitFunction.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/SplitFunction.java
index a1d75394dc190..5edc5412c7fde 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/SplitFunction.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/SplitFunction.java
@@ -49,8 +49,12 @@ public SplitFunction(SpecializedFunction.SpecializedContext context) {
             if (delimiter.toString().isEmpty()) {
                 String str = string.toString();
                 List<StringData> res = new ArrayList<>();
-                for (int i = 0; i < str.length(); i++) {
-                    res.add(StringData.fromString(String.valueOf(str.charAt(i))));
+                int i = 0;
+                while (i < str.length()) {
+                    int codePoint = str.codePointAt(i);
+                    int charCount = Character.charCount(codePoint);
+                    res.add(StringData.fromString(str.substring(i, i + charCount)));
+                    i += charCount;
                 }
                 return new GenericArrayData(res.toArray());
             }