This is an automated email from the ASF dual-hosted git repository.

snuyanzin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git

commit 1b16259c81bd089439b20434e654a647e87fc8ad
Author: Sergey Nuyanzin <[email protected]>
AuthorDate: Mon Sep 28 12:02:12 2026 +0200

    [FLINK-40830][table] Reuse java's `getBytes` in `StringUtf8Utils`
---
 .../flink/table/data/binary/StringUtf8Utils.java   | 386 +++++++--------------
 .../table/data/binary/StringUtf8UtilsTest.java     |  11 +
 .../flink/table/runtime/util/StringUtf8Utils.java  |  95 +----
 .../flink/table/data/BinaryStringDataTest.java     |  28 ++
 4 files changed, 163 insertions(+), 357 deletions(-)

diff --git 
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/StringUtf8Utils.java
 
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/StringUtf8Utils.java
index a099b371973..e8c17fd796d 100644
--- 
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/StringUtf8Utils.java
+++ 
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/StringUtf8Utils.java
@@ -18,20 +18,24 @@
 package org.apache.flink.table.data.binary;
 
 import org.apache.flink.annotation.Internal;
-import org.apache.flink.core.memory.MemorySegment;
 
-import java.io.UnsupportedEncodingException;
+import java.lang.invoke.MethodHandles;
+import java.lang.invoke.VarHandle;
+import java.nio.ByteOrder;
 import java.nio.charset.StandardCharsets;
-import java.util.Arrays;
 
-import static 
org.apache.flink.table.data.binary.BinarySegmentUtils.allocateReuseBytes;
 import static 
org.apache.flink.table.data.binary.BinarySegmentUtils.allocateReuseChars;
 
 /** Utilities for String UTF-8. */
 @Internal
 public final class StringUtf8Utils {
 
-    private static final int MAX_BYTES_PER_CHAR = 3;
+    /** Reads 8 bytes at a time from a {@code byte[]} for the SWAR ASCII scan. 
*/
+    private static final VarHandle LONG_VIEW =
+            MethodHandles.byteArrayViewVarHandle(long[].class, 
ByteOrder.nativeOrder());
+
+    /** High bit of each byte in a 64-bit word; a set bit marks a non-ASCII 
byte. */
+    private static final long ASCII_MASK = 0x8080808080808080L;
 
     private StringUtf8Utils() {
         // do not instantiate
@@ -39,98 +43,137 @@ public final class StringUtf8Utils {
 
     /** This method must have the same result with JDK's String.getBytes. */
     public static byte[] encodeUTF8(String str) {
-        byte[] bytes = allocateReuseBytes(str.length() * MAX_BYTES_PER_CHAR);
-        int len = encodeUTF8(str, bytes);
-        return Arrays.copyOf(bytes, len);
+        return str.getBytes(StandardCharsets.UTF_8);
+    }
+
+    public static String decodeUTF8(byte[] input, int offset, int byteLen) {
+        // Most real text is ASCII: route it to the JDK's compact-string path, 
which is a large win
+        // for longer strings. Anything with a multibyte sequence goes through 
the hand-rolled
+        // decoder; it beats the CharsetDecoder path for non-ASCII input.
+        if (isAscii(input, offset, byteLen)) {
+            // Pure ASCII: bytes map 1:1 to chars; ISO-8859-1 is an intrinsic 
LATIN1 copy.
+            return new String(input, offset, byteLen, 
StandardCharsets.ISO_8859_1);
+        }
+        char[] chars = allocateReuseChars(byteLen);
+        int len = decodeUTF8Strict(input, offset, byteLen, chars);
+        if (len < 0) {
+            // Malformed input; map to U+FFFD via the JDK decoder (matches the 
previous fallback).
+            return new String(input, offset, byteLen, StandardCharsets.UTF_8);
+        }
+        return new String(chars, 0, len);
+    }
+
+    /**
+     * SWAR ASCII test: reads 8 bytes per step and checks the high bit of each 
via {@link
+     * #ASCII_MASK}, so a fully ASCII range is scanned ~8x faster than 
byte-by-byte and non-ASCII
+     * input bails at the first eight-byte block that contains a set high bit.
+     */
+    private static boolean isAscii(byte[] bytes, int offset, int len) {
+        int i = offset;
+        final int end = offset + len;
+        final int swarEnd = offset + (len & ~7);
+        while (i < swarEnd) {
+            if (((long) LONG_VIEW.get(bytes, i) & ASCII_MASK) != 0) {
+                return false;
+            }
+            i += 8;
+        }
+        while (i < end) {
+            if (bytes[i] < 0) {
+                return false;
+            }
+            i++;
+        }
+        return true;
     }
 
-    public static int encodeUTF8(String str, byte[] bytes) {
-        int offset = 0;
-        int len = str.length();
-        int sl = offset + len;
+    public static int decodeUTF8Strict(byte[] sa, int sp, int len, char[] da) {
+        final int sl = sp + len;
         int dp = 0;
-        int dlASCII = dp + Math.min(len, bytes.length);
+        int dlASCII = Math.min(len, da.length);
 
         // ASCII only optimized loop
-        while (dp < dlASCII && str.charAt(offset) < '\u0080') {
-            bytes[dp++] = (byte) str.charAt(offset++);
+        while (dp < dlASCII && sa[sp] >= 0) {
+            da[dp++] = (char) sa[sp++];
         }
 
-        while (offset < sl) {
-            char c = str.charAt(offset++);
-            if (c < 0x80) {
-                // Have at most seven bits
-                bytes[dp++] = (byte) c;
-            } else if (c < 0x800) {
-                // 2 bytes, 11 bits
-                bytes[dp++] = (byte) (0xc0 | (c >> 6));
-                bytes[dp++] = (byte) (0x80 | (c & 0x3f));
-            } else if (Character.isSurrogate(c)) {
-                final int uc;
-                int ip = offset - 1;
-                if (Character.isHighSurrogate(c)) {
-                    if (sl - ip < 2) {
-                        uc = -1;
+        while (sp < sl) {
+            int b1 = sa[sp++];
+            if (b1 >= 0) {
+                // 1 byte, 7 bits: 0xxxxxxx
+                da[dp++] = (char) b1;
+            } else if ((b1 >> 5) == -2 && (b1 & 0x1e) != 0) {
+                // 2 bytes, 11 bits: 110xxxxx 10xxxxxx
+                if (sp < sl) {
+                    int b2 = sa[sp++];
+                    if ((b2 & 0xc0) != 0x80) { // isNotContinuation(b2)
+                        return -1;
                     } else {
-                        char d = str.charAt(ip + 1);
-                        if (Character.isLowSurrogate(d)) {
-                            uc = Character.toCodePoint(c, d);
+                        da[dp++] = (char) (((b1 << 6) ^ b2) ^ (((byte) 0xC0 << 
6) ^ ((byte) 0x80)));
+                    }
+                    continue;
+                }
+                return -1;
+            } else if ((b1 >> 4) == -2) {
+                // 3 bytes, 16 bits: 1110xxxx 10xxxxxx 10xxxxxx
+                if (sp + 1 < sl) {
+                    int b2 = sa[sp++];
+                    int b3 = sa[sp++];
+                    if ((b1 == (byte) 0xe0 && (b2 & 0xe0) == 0x80)
+                            || (b2 & 0xc0) != 0x80
+                            || (b3 & 0xc0) != 0x80) { // isMalformed3(b1, b2, 
b3)
+                        return -1;
+                    } else {
+                        char c =
+                                (char)
+                                        ((b1 << 12)
+                                                ^ (b2 << 6)
+                                                ^ (b3
+                                                        ^ (((byte) 0xE0 << 12)
+                                                                ^ ((byte) 0x80 
<< 6)
+                                                                ^ ((byte) 
0x80))));
+                        if (Character.isSurrogate(c)) {
+                            return -1;
                         } else {
-                            // for some illegal character
-                            // the jdk will ignore the origin character and 
cast it to '?'
-                            // this acts the same with jdk
-                            return defaultEncodeUTF8(str, bytes);
+                            da[dp++] = c;
                         }
                     }
-                } else {
-                    if (Character.isLowSurrogate(c)) {
-                        // for some illegal character
-                        // the jdk will ignore the origin character and cast 
it to '?'
-                        // this acts the same with jdk
-                        return defaultEncodeUTF8(str, bytes);
+                    continue;
+                }
+                return -1;
+            } else if ((b1 >> 3) == -2) {
+                // 4 bytes, 21 bits: 11110xxx 10xxxxxx 10xxxxxx 10xxxxxx
+                if (sp + 2 < sl) {
+                    int b2 = sa[sp++];
+                    int b3 = sa[sp++];
+                    int b4 = sa[sp++];
+                    int uc =
+                            ((b1 << 18)
+                                    ^ (b2 << 12)
+                                    ^ (b3 << 6)
+                                    ^ (b4
+                                            ^ (((byte) 0xF0 << 18)
+                                                    ^ ((byte) 0x80 << 12)
+                                                    ^ ((byte) 0x80 << 6)
+                                                    ^ ((byte) 0x80))));
+                    // isMalformed4 and shortest form check
+                    if (((b2 & 0xc0) != 0x80 || (b3 & 0xc0) != 0x80 || (b4 & 
0xc0) != 0x80)
+                            || !Character.isSupplementaryCodePoint(uc)) {
+                        return -1;
                     } else {
-                        uc = c;
+                        da[dp++] = Character.highSurrogate(uc);
+                        da[dp++] = Character.lowSurrogate(uc);
                     }
+                    continue;
                 }
-
-                if (uc < 0) {
-                    bytes[dp++] = (byte) '?';
-                } else {
-                    bytes[dp++] = (byte) (0xf0 | ((uc >> 18)));
-                    bytes[dp++] = (byte) (0x80 | ((uc >> 12) & 0x3f));
-                    bytes[dp++] = (byte) (0x80 | ((uc >> 6) & 0x3f));
-                    bytes[dp++] = (byte) (0x80 | (uc & 0x3f));
-                    offset++; // 2 chars
-                }
+                return -1;
             } else {
-                // 3 bytes, 16 bits
-                bytes[dp++] = (byte) (0xe0 | ((c >> 12)));
-                bytes[dp++] = (byte) (0x80 | ((c >> 6) & 0x3f));
-                bytes[dp++] = (byte) (0x80 | (c & 0x3f));
+                return -1;
             }
         }
         return dp;
     }
 
-    public static int defaultEncodeUTF8(String str, byte[] bytes) {
-        try {
-            byte[] buffer = str.getBytes("UTF-8");
-            System.arraycopy(buffer, 0, bytes, 0, buffer.length);
-            return buffer.length;
-        } catch (UnsupportedEncodingException e) {
-            throw new RuntimeException("encodeUTF8 error", e);
-        }
-    }
-
-    public static String decodeUTF8(byte[] input, int offset, int byteLen) {
-        char[] chars = allocateReuseChars(byteLen);
-        int len = decodeUTF8Strict(input, offset, byteLen, chars);
-        if (len < 0) {
-            return defaultDecodeUTF8(input, offset, byteLen);
-        }
-        return new String(chars, 0, len);
-    }
-
     // Bit-pattern predicates for UTF-8 byte categorization. The JIT inlines 
these so they cost
     // nothing at runtime, but they make {@link #firstInvalidUtf8ByteIndex} 
read like prose.
     private static boolean isAsciiByte(int b) {
@@ -255,193 +298,4 @@ public final class StringUtf8Utils {
         }
         return -1;
     }
-
-    public static int decodeUTF8Strict(byte[] sa, int sp, int len, char[] da) {
-        final int sl = sp + len;
-        int dp = 0;
-        int dlASCII = Math.min(len, da.length);
-
-        // ASCII only optimized loop
-        while (dp < dlASCII && sa[sp] >= 0) {
-            da[dp++] = (char) sa[sp++];
-        }
-
-        while (sp < sl) {
-            int b1 = sa[sp++];
-            if (b1 >= 0) {
-                // 1 byte, 7 bits: 0xxxxxxx
-                da[dp++] = (char) b1;
-            } else if ((b1 >> 5) == -2 && (b1 & 0x1e) != 0) {
-                // 2 bytes, 11 bits: 110xxxxx 10xxxxxx
-                if (sp < sl) {
-                    int b2 = sa[sp++];
-                    if ((b2 & 0xc0) != 0x80) { // isNotContinuation(b2)
-                        return -1;
-                    } else {
-                        da[dp++] = (char) (((b1 << 6) ^ b2) ^ (((byte) 0xC0 << 
6) ^ ((byte) 0x80)));
-                    }
-                    continue;
-                }
-                return -1;
-            } else if ((b1 >> 4) == -2) {
-                // 3 bytes, 16 bits: 1110xxxx 10xxxxxx 10xxxxxx
-                if (sp + 1 < sl) {
-                    int b2 = sa[sp++];
-                    int b3 = sa[sp++];
-                    if ((b1 == (byte) 0xe0 && (b2 & 0xe0) == 0x80)
-                            || (b2 & 0xc0) != 0x80
-                            || (b3 & 0xc0) != 0x80) { // isMalformed3(b1, b2, 
b3)
-                        return -1;
-                    } else {
-                        char c =
-                                (char)
-                                        ((b1 << 12)
-                                                ^ (b2 << 6)
-                                                ^ (b3
-                                                        ^ (((byte) 0xE0 << 12)
-                                                                ^ ((byte) 0x80 
<< 6)
-                                                                ^ ((byte) 
0x80))));
-                        if (Character.isSurrogate(c)) {
-                            return -1;
-                        } else {
-                            da[dp++] = c;
-                        }
-                    }
-                    continue;
-                }
-                return -1;
-            } else if ((b1 >> 3) == -2) {
-                // 4 bytes, 21 bits: 11110xxx 10xxxxxx 10xxxxxx 10xxxxxx
-                if (sp + 2 < sl) {
-                    int b2 = sa[sp++];
-                    int b3 = sa[sp++];
-                    int b4 = sa[sp++];
-                    int uc =
-                            ((b1 << 18)
-                                    ^ (b2 << 12)
-                                    ^ (b3 << 6)
-                                    ^ (b4
-                                            ^ (((byte) 0xF0 << 18)
-                                                    ^ ((byte) 0x80 << 12)
-                                                    ^ ((byte) 0x80 << 6)
-                                                    ^ ((byte) 0x80))));
-                    // isMalformed4 and shortest form check
-                    if (((b2 & 0xc0) != 0x80 || (b3 & 0xc0) != 0x80 || (b4 & 
0xc0) != 0x80)
-                            || !Character.isSupplementaryCodePoint(uc)) {
-                        return -1;
-                    } else {
-                        da[dp++] = Character.highSurrogate(uc);
-                        da[dp++] = Character.lowSurrogate(uc);
-                    }
-                    continue;
-                }
-                return -1;
-            } else {
-                return -1;
-            }
-        }
-        return dp;
-    }
-
-    public static String decodeUTF8(MemorySegment input, int offset, int 
byteLen) {
-        char[] chars = allocateReuseChars(byteLen);
-        int len = decodeUTF8Strict(input, offset, byteLen, chars);
-        if (len < 0) {
-            byte[] bytes = allocateReuseBytes(byteLen);
-            input.get(offset, bytes, 0, byteLen);
-            return defaultDecodeUTF8(bytes, 0, byteLen);
-        }
-        return new String(chars, 0, len);
-    }
-
-    public static int decodeUTF8Strict(MemorySegment segment, int sp, int len, 
char[] da) {
-        final int sl = sp + len;
-        int dp = 0;
-        int dlASCII = Math.min(len, da.length);
-
-        // ASCII only optimized loop
-        while (dp < dlASCII && segment.get(sp) >= 0) {
-            da[dp++] = (char) segment.get(sp++);
-        }
-
-        while (sp < sl) {
-            int b1 = segment.get(sp++);
-            if (b1 >= 0) {
-                // 1 byte, 7 bits: 0xxxxxxx
-                da[dp++] = (char) b1;
-            } else if ((b1 >> 5) == -2 && (b1 & 0x1e) != 0) {
-                // 2 bytes, 11 bits: 110xxxxx 10xxxxxx
-                if (sp < sl) {
-                    int b2 = segment.get(sp++);
-                    if ((b2 & 0xc0) != 0x80) { // isNotContinuation(b2)
-                        return -1;
-                    } else {
-                        da[dp++] = (char) (((b1 << 6) ^ b2) ^ (((byte) 0xC0 << 
6) ^ ((byte) 0x80)));
-                    }
-                    continue;
-                }
-                return -1;
-            } else if ((b1 >> 4) == -2) {
-                // 3 bytes, 16 bits: 1110xxxx 10xxxxxx 10xxxxxx
-                if (sp + 1 < sl) {
-                    int b2 = segment.get(sp++);
-                    int b3 = segment.get(sp++);
-                    if ((b1 == (byte) 0xe0 && (b2 & 0xe0) == 0x80)
-                            || (b2 & 0xc0) != 0x80
-                            || (b3 & 0xc0) != 0x80) { // isMalformed3(b1, b2, 
b3)
-                        return -1;
-                    } else {
-                        char c =
-                                (char)
-                                        ((b1 << 12)
-                                                ^ (b2 << 6)
-                                                ^ (b3
-                                                        ^ (((byte) 0xE0 << 12)
-                                                                ^ ((byte) 0x80 
<< 6)
-                                                                ^ ((byte) 
0x80))));
-                        if (Character.isSurrogate(c)) {
-                            return -1;
-                        } else {
-                            da[dp++] = c;
-                        }
-                    }
-                    continue;
-                }
-                return -1;
-            } else if ((b1 >> 3) == -2) {
-                // 4 bytes, 21 bits: 11110xxx 10xxxxxx 10xxxxxx 10xxxxxx
-                if (sp + 2 < sl) {
-                    int b2 = segment.get(sp++);
-                    int b3 = segment.get(sp++);
-                    int b4 = segment.get(sp++);
-                    int uc =
-                            ((b1 << 18)
-                                    ^ (b2 << 12)
-                                    ^ (b3 << 6)
-                                    ^ (b4
-                                            ^ (((byte) 0xF0 << 18)
-                                                    ^ ((byte) 0x80 << 12)
-                                                    ^ ((byte) 0x80 << 6)
-                                                    ^ ((byte) 0x80))));
-                    // isMalformed4 and shortest form check
-                    if (((b2 & 0xc0) != 0x80 || (b3 & 0xc0) != 0x80 || (b4 & 
0xc0) != 0x80)
-                            || !Character.isSupplementaryCodePoint(uc)) {
-                        return -1;
-                    } else {
-                        da[dp++] = Character.highSurrogate(uc);
-                        da[dp++] = Character.lowSurrogate(uc);
-                    }
-                    continue;
-                }
-                return -1;
-            } else {
-                return -1;
-            }
-        }
-        return dp;
-    }
-
-    public static String defaultDecodeUTF8(byte[] bytes, int offset, int len) {
-        return new String(bytes, offset, len, StandardCharsets.UTF_8);
-    }
 }
diff --git 
a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/data/binary/StringUtf8UtilsTest.java
 
b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/data/binary/StringUtf8UtilsTest.java
index bb3af471491..f2fd53ba132 100644
--- 
a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/data/binary/StringUtf8UtilsTest.java
+++ 
b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/data/binary/StringUtf8UtilsTest.java
@@ -25,6 +25,7 @@ import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.Arguments;
 import org.junit.jupiter.params.provider.MethodSource;
+import org.junit.jupiter.params.provider.ValueSource;
 
 import java.nio.charset.StandardCharsets;
 import java.util.stream.Stream;
@@ -86,6 +87,16 @@ class StringUtf8UtilsTest {
                 .hasMessageContaining("Invalid UTF-8 byte at index 2 of 3");
     }
 
+    @ParameterizedTest(name = "{0}")
+    @ValueSource(strings = {"Hello, world", "Café au lait", "é", "a😀b"})
+    void testDecodeUtf8RoundTrip(String input) {
+        final byte[] b = input.getBytes(StandardCharsets.UTF_8);
+        assertThat(StringUtf8Utils.decodeUTF8(b, 0, 
b.length)).isEqualTo(input);
+        final byte[] padded = new byte[b.length + 1];
+        System.arraycopy(b, 0, padded, 1, b.length);
+        assertThat(StringUtf8Utils.decodeUTF8(padded, 1, 
b.length)).isEqualTo(input);
+    }
+
     private static byte[] bytes(int... values) {
         final byte[] result = new byte[values.length];
         for (int i = 0; i < values.length; i++) {
diff --git 
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/util/StringUtf8Utils.java
 
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/util/StringUtf8Utils.java
index a42d91b188a..f374a751adc 100644
--- 
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/util/StringUtf8Utils.java
+++ 
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/util/StringUtf8Utils.java
@@ -19,9 +19,7 @@ package org.apache.flink.table.runtime.util;
 
 import org.apache.flink.core.memory.MemorySegment;
 
-import java.io.UnsupportedEncodingException;
 import java.nio.charset.StandardCharsets;
-import java.util.Arrays;
 
 import static 
org.apache.flink.table.runtime.util.SegmentsUtil.allocateReuseBytes;
 import static 
org.apache.flink.table.runtime.util.SegmentsUtil.allocateReuseChars;
@@ -35,100 +33,15 @@ import static 
org.apache.flink.table.runtime.util.SegmentsUtil.allocateReuseChar
  */
 public class StringUtf8Utils {
 
-    private static final int MAX_BYTES_PER_CHAR = 3;
-
     /** This method must have the same result with JDK's String.getBytes. */
     public static byte[] encodeUTF8(String str) {
-        byte[] bytes = allocateReuseBytes(str.length() * MAX_BYTES_PER_CHAR);
-        int len = encodeUTF8(str, bytes);
-        return Arrays.copyOf(bytes, len);
-    }
-
-    public static int encodeUTF8(String str, byte[] bytes) {
-        int offset = 0;
-        int len = str.length();
-        int sl = offset + len;
-        int dp = 0;
-        int dlASCII = dp + Math.min(len, bytes.length);
-
-        // ASCII only optimized loop
-        while (dp < dlASCII && str.charAt(offset) < '\u0080') {
-            bytes[dp++] = (byte) str.charAt(offset++);
-        }
-
-        while (offset < sl) {
-            char c = str.charAt(offset++);
-            if (c < 0x80) {
-                // Have at most seven bits
-                bytes[dp++] = (byte) c;
-            } else if (c < 0x800) {
-                // 2 bytes, 11 bits
-                bytes[dp++] = (byte) (0xc0 | (c >> 6));
-                bytes[dp++] = (byte) (0x80 | (c & 0x3f));
-            } else if (Character.isSurrogate(c)) {
-                final int uc;
-                int ip = offset - 1;
-                if (Character.isHighSurrogate(c)) {
-                    if (sl - ip < 2) {
-                        uc = -1;
-                    } else {
-                        char d = str.charAt(ip + 1);
-                        if (Character.isLowSurrogate(d)) {
-                            uc = Character.toCodePoint(c, d);
-                        } else {
-                            // for some illegal character
-                            // the jdk will ignore the origin character and 
cast it to '?'
-                            // this acts the same with jdk
-                            return defaultEncodeUTF8(str, bytes);
-                        }
-                    }
-                } else {
-                    if (Character.isLowSurrogate(c)) {
-                        // for some illegal character
-                        // the jdk will ignore the origin character and cast 
it to '?'
-                        // this acts the same with jdk
-                        return defaultEncodeUTF8(str, bytes);
-                    } else {
-                        uc = c;
-                    }
-                }
-
-                if (uc < 0) {
-                    bytes[dp++] = (byte) '?';
-                } else {
-                    bytes[dp++] = (byte) (0xf0 | ((uc >> 18)));
-                    bytes[dp++] = (byte) (0x80 | ((uc >> 12) & 0x3f));
-                    bytes[dp++] = (byte) (0x80 | ((uc >> 6) & 0x3f));
-                    bytes[dp++] = (byte) (0x80 | (uc & 0x3f));
-                    offset++; // 2 chars
-                }
-            } else {
-                // 3 bytes, 16 bits
-                bytes[dp++] = (byte) (0xe0 | ((c >> 12)));
-                bytes[dp++] = (byte) (0x80 | ((c >> 6) & 0x3f));
-                bytes[dp++] = (byte) (0x80 | (c & 0x3f));
-            }
-        }
-        return dp;
-    }
-
-    public static int defaultEncodeUTF8(String str, byte[] bytes) {
-        try {
-            byte[] buffer = str.getBytes("UTF-8");
-            System.arraycopy(buffer, 0, bytes, 0, buffer.length);
-            return buffer.length;
-        } catch (UnsupportedEncodingException e) {
-            throw new RuntimeException("encodeUTF8 error", e);
-        }
+        return str.getBytes(StandardCharsets.UTF_8);
     }
 
     public static String decodeUTF8(byte[] input, int offset, int byteLen) {
-        char[] chars = allocateReuseChars(byteLen);
-        int len = decodeUTF8Strict(input, offset, byteLen, chars);
-        if (len < 0) {
-            return defaultDecodeUTF8(input, offset, byteLen);
-        }
-        return new String(chars, 0, len);
+        // Delegate to the optimized table-common impl so non-ASCII input 
doesn't regress.
+        return org.apache.flink.table.data.binary.StringUtf8Utils.decodeUTF8(
+                input, offset, byteLen);
     }
 
     public static int decodeUTF8Strict(byte[] sa, int sp, int len, char[] da) {
diff --git 
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/data/BinaryStringDataTest.java
 
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/data/BinaryStringDataTest.java
index 9f8c1db8629..4f2da4a4e76 100644
--- 
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/data/BinaryStringDataTest.java
+++ 
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/data/BinaryStringDataTest.java
@@ -896,6 +896,34 @@ class BinaryStringDataTest {
             assertThat(fromBytes(new byte[] {(byte) 
wrongFirstByte}).numChars()).isEqualTo(1);
         }
 
+        @Test
+        void testEncodeUnpairedSurrogateProducesQuestionMark() {
+            // Unpaired surrogates aren't representable in UTF-8; the JDK 
encoder substitutes '?'
+            // (0x3F).
+            assertThat(StringUtf8Utils.encodeUTF8("a\uD800b"))
+                    .containsExactly((byte) 'a', (byte) '?', (byte) 'b');
+            assertThat(StringUtf8Utils.encodeUTF8("a\uD800"))
+                    .containsExactly((byte) 'a', (byte) '?');
+            assertThat(StringUtf8Utils.encodeUTF8("a\uDC00b"))
+                    .containsExactly((byte) 'a', (byte) '?', (byte) 'b');
+        }
+
+        @Test
+        void testDecodeMalformedUtf8UsesReplacementChar() {
+            // Malformed UTF-8 (0xC3 lead + non-continuation byte; stray 
continuation 0x80) ->
+            // U+FFFD.
+            byte[] bytes = {(byte) 'a', (byte) 0xC3, (byte) 'b', (byte) 0x80, 
(byte) 'c'};
+            String expected = "a�b�c";
+            assertThat(new String(bytes, 
StandardCharsets.UTF_8)).isEqualTo(expected);
+
+            assertThat(StringUtf8Utils.decodeUTF8(bytes, 0, 
bytes.length)).isEqualTo(expected);
+            assertThat(
+                            StringUtf8Utils.decodeUTF8(
+                                    MemorySegmentFactory.wrap(bytes), 0, 
bytes.length))
+                    .isEqualTo(expected);
+            
assertThat(BinaryStringData.fromBytes(bytes).toString()).isEqualTo(expected);
+        }
+
         private Stream<Arguments> utf8Decoders() {
             return Stream.of(
                     arguments(

Reply via email to