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 031ff76cae511ae220e586c7c437e85c082b17a4 Author: Sergey Nuyanzin <[email protected]> AuthorDate: Fri Oct 2 14:20:50 2026 +0200 [FLINK-40830][table] Merge StringUtf8Utils into one class This closes #29313 --- .../serializers/python/BigDecSerializer.java | 2 +- .../serializers/python/DecimalDataSerializer.java | 2 +- .../serializers/python/StringSerializer.java | 2 +- .../flink/table/data/binary/StringUtf8Utils.java | 105 ++++++++- .../table/data/binary/BinaryStringDataUtil.java | 1 - .../flink/table/runtime/util/StringUtf8Utils.java | 235 --------------------- .../flink/table/data/BinaryStringDataTest.java | 2 +- 7 files changed, 108 insertions(+), 241 deletions(-) diff --git a/flink-python/src/main/java/org/apache/flink/table/runtime/typeutils/serializers/python/BigDecSerializer.java b/flink-python/src/main/java/org/apache/flink/table/runtime/typeutils/serializers/python/BigDecSerializer.java index d896944db9f..ce7a3a067f9 100644 --- a/flink-python/src/main/java/org/apache/flink/table/runtime/typeutils/serializers/python/BigDecSerializer.java +++ b/flink-python/src/main/java/org/apache/flink/table/runtime/typeutils/serializers/python/BigDecSerializer.java @@ -24,7 +24,7 @@ import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot; import org.apache.flink.api.common.typeutils.base.TypeSerializerSingleton; import org.apache.flink.core.memory.DataInputView; import org.apache.flink.core.memory.DataOutputView; -import org.apache.flink.table.runtime.util.StringUtf8Utils; +import org.apache.flink.table.data.binary.StringUtf8Utils; import java.io.IOException; import java.math.BigDecimal; diff --git a/flink-python/src/main/java/org/apache/flink/table/runtime/typeutils/serializers/python/DecimalDataSerializer.java b/flink-python/src/main/java/org/apache/flink/table/runtime/typeutils/serializers/python/DecimalDataSerializer.java index e1043e97741..5e39d43b591 100644 --- a/flink-python/src/main/java/org/apache/flink/table/runtime/typeutils/serializers/python/DecimalDataSerializer.java +++ b/flink-python/src/main/java/org/apache/flink/table/runtime/typeutils/serializers/python/DecimalDataSerializer.java @@ -25,7 +25,7 @@ import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot; import org.apache.flink.core.memory.DataInputView; import org.apache.flink.core.memory.DataOutputView; import org.apache.flink.table.data.DecimalData; -import org.apache.flink.table.runtime.util.StringUtf8Utils; +import org.apache.flink.table.data.binary.StringUtf8Utils; import java.io.IOException; import java.math.BigDecimal; diff --git a/flink-python/src/main/java/org/apache/flink/table/runtime/typeutils/serializers/python/StringSerializer.java b/flink-python/src/main/java/org/apache/flink/table/runtime/typeutils/serializers/python/StringSerializer.java index 3355b4327ea..c02359d5fe9 100644 --- a/flink-python/src/main/java/org/apache/flink/table/runtime/typeutils/serializers/python/StringSerializer.java +++ b/flink-python/src/main/java/org/apache/flink/table/runtime/typeutils/serializers/python/StringSerializer.java @@ -24,7 +24,7 @@ import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot; import org.apache.flink.api.common.typeutils.base.TypeSerializerSingleton; import org.apache.flink.core.memory.DataInputView; import org.apache.flink.core.memory.DataOutputView; -import org.apache.flink.table.runtime.util.StringUtf8Utils; +import org.apache.flink.table.data.binary.StringUtf8Utils; import java.io.IOException; 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 e8c17fd796d..4df4ebbe559 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,12 +18,14 @@ package org.apache.flink.table.data.binary; import org.apache.flink.annotation.Internal; +import org.apache.flink.core.memory.MemorySegment; import java.lang.invoke.MethodHandles; import java.lang.invoke.VarHandle; import java.nio.ByteOrder; import java.nio.charset.StandardCharsets; +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. */ @@ -41,7 +43,6 @@ public final class StringUtf8Utils { // do not instantiate } - /** This method must have the same result with JDK's String.getBytes. */ public static byte[] encodeUTF8(String str) { return str.getBytes(StandardCharsets.UTF_8); } @@ -298,4 +299,106 @@ public final class StringUtf8Utils { } return -1; } + + 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 String defaultDecodeUTF8(byte[] bytes, int offset, int len) { + return new String(bytes, offset, len, StandardCharsets.UTF_8); + } + + 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; + } } diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/data/binary/BinaryStringDataUtil.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/data/binary/BinaryStringDataUtil.java index 5c5844e673f..490dd3324aa 100644 --- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/data/binary/BinaryStringDataUtil.java +++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/data/binary/BinaryStringDataUtil.java @@ -24,7 +24,6 @@ import org.apache.flink.table.data.DecimalDataUtils; import org.apache.flink.table.data.StringData; import org.apache.flink.table.data.TimestampData; import org.apache.flink.table.runtime.util.SegmentsUtil; -import org.apache.flink.table.runtime.util.StringUtf8Utils; import org.apache.flink.table.utils.DateTimeUtils; import org.apache.flink.table.utils.EncodingUtils; 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 deleted file mode 100644 index f374a751adc..00000000000 --- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/util/StringUtf8Utils.java +++ /dev/null @@ -1,235 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.flink.table.runtime.util; - -import org.apache.flink.core.memory.MemorySegment; - -import java.nio.charset.StandardCharsets; - -import static org.apache.flink.table.runtime.util.SegmentsUtil.allocateReuseBytes; -import static org.apache.flink.table.runtime.util.SegmentsUtil.allocateReuseChars; - -/** - * String utf-8 utils. - * - * <p>{@code StringUtf8Utils} refers to the implementation from SerializeWriter and IOUtils of - * Alibaba fastjson. The difference is that StringUtf8Utils need to handle the wrong code, just like - * StringCoding.decode. - */ -public class StringUtf8Utils { - - /** This method must have the same result with JDK's String.getBytes. */ - public static byte[] encodeUTF8(String str) { - return str.getBytes(StandardCharsets.UTF_8); - } - - public static String decodeUTF8(byte[] input, int offset, int byteLen) { - // 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) { - 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-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 4f2da4a4e76..1341e0ed5c5 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 @@ -21,9 +21,9 @@ import org.apache.flink.core.memory.MemorySegment; import org.apache.flink.core.memory.MemorySegmentFactory; import org.apache.flink.table.data.binary.BinaryRowData; import org.apache.flink.table.data.binary.BinaryStringData; +import org.apache.flink.table.data.binary.StringUtf8Utils; import org.apache.flink.table.data.writer.BinaryRowWriter; import org.apache.flink.table.runtime.operators.sort.SortUtil; -import org.apache.flink.table.runtime.util.StringUtf8Utils; import org.junit.jupiter.api.DisplayName; import org.junit.jupiter.api.Nested;
