MartijnVisser commented on code in PR #29313:
URL: https://github.com/apache/flink/pull/29313#discussion_r4135273729
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/StringUtf8Utils.java:
##########
@@ -18,117 +18,25 @@
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.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;
-
private StringUtf8Utils() {
// do not instantiate
}
/** 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);
+ // JDK-intrinsified decode; maps malformed input to U+FFFD (matches
the previous fallback).
+ return new String(input, offset, byteLen, StandardCharsets.UTF_8);
Review Comment:
Thanks, verified: same output as the old decoder on 22M inputs, and
non-ASCII is back at 0.8x to 1.1x of master on 11, 17 and 21.
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/util/StringUtf8Utils.java:
##########
@@ -35,100 +33,14 @@
*/
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);
+ // JDK-intrinsified decode; maps malformed input to U+FFFD (matches
the previous fallback).
+ return new String(input, offset, byteLen, StandardCharsets.UTF_8);
Review Comment:
This copy still uses `new String(.., UTF_8)`, and PyFlink's
`StringSerializer` calls it per record, so non-ASCII still regresses there. Can
it delegate to the table-common `decodeUTF8`?
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/binary/StringUtf8Utils.java:
##########
@@ -18,119 +18,162 @@
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
}
/** 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;
Review Comment:
No test reaches this `return false`: with the check disabled,
`StringUtf8UtilsTest` and `BinaryStringDataTest` still pass. Decoding `Café au
lait` from bytes catches it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
