This is an automated email from the ASF dual-hosted git repository. AHeise pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit bd7e555ad480b65bce10ad3993a6e31d03c64333 Author: Ramin Gharib <[email protected]> AuthorDate: Mon Oct 5 13:35:06 2026 +0200 [FLINK-40871][table] Avoid the UTF-8 round trip when casting strings and decimals to VARIANT `CAST(s AS VARIANT)` decoded the UTF-8 bytes of the string into a Java String and encoded it back. It now checks the bytes in one pass and copies them, and it reads a string that lies in the first heap segment in place. A `StringData` can hold invalid UTF-8, which the Variant spec does not allow, so invalid input is still decoded and every malformed sequence is stored as U+FFFD, as before. A string that only exists as a Java object, such as a literal or a function result, is encoded s [...] `CAST(d AS VARIANT)` writes a compact decimal from its unscaled long instead of building a `BigDecimal`. The encoding lives in `VariantCastUtils#appendString` and `#appendDecimal`, which write into a shared builder, so a cast of a nested value can use the same encoding. The stored VARIANT is unchanged for every input. --- .../table/runtime/functions/VariantCastUtils.java | 54 +++++++- .../runtime/functions/VariantCastUtilsTest.java | 142 +++++++++++++++++++++ 2 files changed, 193 insertions(+), 3 deletions(-) diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/VariantCastUtils.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/VariantCastUtils.java index b834a157c76..12a4fdd1fda 100644 --- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/VariantCastUtils.java +++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/VariantCastUtils.java @@ -19,6 +19,7 @@ package org.apache.flink.table.runtime.functions; import org.apache.flink.annotation.Internal; +import org.apache.flink.core.memory.MemorySegment; import org.apache.flink.table.api.TableRuntimeException; import org.apache.flink.table.data.DecimalData; import org.apache.flink.table.data.StringData; @@ -593,14 +594,61 @@ public final class VariantCastUtils { } public static Variant fromDecimal(DecimalData value) { - return BUILDER.of(value.toBigDecimal()); + final BinaryVariantInternalBuilder builder = new BinaryVariantInternalBuilder(false); + appendDecimal(builder, value); + return builder.build(); + } + + /** Like {@link #fromDecimal(DecimalData)}, but writes into a shared builder. */ + static void appendDecimal(BinaryVariantInternalBuilder builder, DecimalData value) { + if (value.isCompact()) { + builder.appendDecimal(value.toUnscaledLong(), value.scale()); + } else { + builder.appendDecimal(value.toBigDecimal()); + } } + /** + * Stores the UTF-8 bytes of the string as they are. A {@link StringData} may hold invalid + * UTF-8, which the variant spec does not allow, so such a value is decoded first and every + * malformed sequence is stored as the U+FFFD replacement character. + */ public static Variant fromString(StringData value) { + final BinaryVariantInternalBuilder builder = new BinaryVariantInternalBuilder(false); try { - return BUILDER.of(value.toString()); + appendString(builder, value); + return builder.build(); } catch (VariantTypeException e) { - throw sizeLimitExceeded(e, "string", value.toBytes().length); + throw sizeLimitExceeded(e, "string", ((BinaryStringData) value).getSizeInBytes()); + } + } + + /** Like {@link #fromString(StringData)}, but writes into a shared builder. */ + static void appendString(BinaryVariantInternalBuilder builder, StringData value) { + final BinaryStringData string = (BinaryStringData) value; + if (string.getBinarySection() == null) { + // A string that only exists as a Java object, such as a literal or a function result, + // is valid UTF-8 once encoded. Encoding it here skips the binary form and the check. + builder.appendString(string.getJavaObject()); + return; + } + final MemorySegment[] segments = string.getSegments(); + final int length = string.getSizeInBytes(); + final byte[] utf8; + final int offset; + // Read a string that lies in the first heap segment in place, which saves copying it out. + if (!segments[0].isOffHeap() && string.getOffset() + length <= segments[0].size()) { + utf8 = segments[0].getHeapMemory(); + offset = string.getOffset(); + } else { + utf8 = string.toBytes(); + offset = 0; + } + if (StringUtf8Utils.firstInvalidUtf8ByteIndex(utf8, offset, length) >= 0) { + // The same decoding StringData#toString falls back to for invalid UTF-8. + builder.appendString(new String(utf8, offset, length, StandardCharsets.UTF_8)); + } else { + builder.appendString(utf8, offset, length); } } diff --git a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/functions/VariantCastUtilsTest.java b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/functions/VariantCastUtilsTest.java index 4492a94b234..c4419de2365 100644 --- a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/functions/VariantCastUtilsTest.java +++ b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/functions/VariantCastUtilsTest.java @@ -18,19 +18,35 @@ package org.apache.flink.table.runtime.functions; +import org.apache.flink.core.memory.MemorySegment; +import org.apache.flink.core.memory.MemorySegmentFactory; import org.apache.flink.table.api.TableRuntimeException; +import org.apache.flink.table.data.DecimalData; import org.apache.flink.table.data.StringData; +import org.apache.flink.table.data.binary.BinaryStringData; import org.apache.flink.types.variant.BinaryVariant; import org.apache.flink.types.variant.Variant; import org.apache.flink.types.variant.VariantBuilder; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; +import org.junit.jupiter.params.provider.EnumSource; +import org.junit.jupiter.params.provider.MethodSource; +import org.junit.jupiter.params.provider.ValueSource; +import java.math.BigDecimal; +import java.nio.ByteBuffer; import java.time.Instant; +import java.util.Arrays; import java.util.TimeZone; +import java.util.function.Function; +import java.util.stream.Stream; +import static java.nio.charset.StandardCharsets.UTF_8; import static org.apache.flink.table.runtime.functions.VariantCastUtils.MAX_PAYLOAD_BYTES; import static org.apache.flink.table.runtime.functions.VariantCastUtils.fromBytes; +import static org.apache.flink.table.runtime.functions.VariantCastUtils.fromDecimal; import static org.apache.flink.table.runtime.functions.VariantCastUtils.fromString; import static org.apache.flink.table.runtime.functions.VariantCastUtils.toPrintString; import static org.apache.flink.types.variant.BinaryVariantUtil.primitiveHeader; @@ -44,6 +60,8 @@ class VariantCastUtilsTest { private static final VariantBuilder BUILDER = Variant.newBuilder(); + private static final String REPLACEMENT_CHARACTER = "\uFFFD"; + @Test void testCastToVariantHoldsUpToTheSizeLimit() { assertThat(fromBytes(new byte[MAX_PAYLOAD_BYTES]).getBytes()).hasSize(MAX_PAYLOAD_BYTES); @@ -63,6 +81,93 @@ class VariantCastUtilsTest { () -> fromString(StringData.fromString("x".repeat(MAX_PAYLOAD_BYTES + 1)))) .isInstanceOf(TableRuntimeException.class) .hasMessageStartingWith("Cannot cast a string value of 16777212 bytes to VARIANT."); + final byte[] tooLong = new byte[MAX_PAYLOAD_BYTES + 1]; + Arrays.fill(tooLong, (byte) 'x'); + assertThatThrownBy(() -> fromString(binaryString(tooLong))) + .isInstanceOf(TableRuntimeException.class) + .hasMessageStartingWith("Cannot cast a string value of 16777212 bytes to VARIANT."); + } + + @ParameterizedTest + @ValueSource(strings = {"", "hello", "Grüße, 世界 🚀"}) + void testCastStringToVariantStoresItsUtf8Bytes(final String str) { + assertThat(fromString(binaryString(str.getBytes(UTF_8)))).isEqualTo(BUILDER.of(str)); + } + + @ParameterizedTest + @MethodSource("invalidUtf8") + void testCastStringToVariantReplacesInvalidUtf8(final byte[] invalid) { + final Variant variant = fromString(binaryString(invalid)); + + assertThat(variant.getString()).contains(REPLACEMENT_CHARACTER); + assertThat(variant).isEqualTo(BUILDER.of(binaryString(invalid).toString())); + } + + @Test + void testCastStringToVariantEncodesAJavaStringWithoutItsBinaryForm() { + // An unpaired surrogate has no UTF-8 form, so encoding stores '?' in its place. + final StringData value = StringData.fromString("a\uD800b"); + + final Variant variant = fromString(value); + + assertThat(variant).isEqualTo(BUILDER.of(value.toString())); + assertThat(variant.getString()).isEqualTo("a?b"); + assertThat(((BinaryStringData) value).getBinarySection()).isNull(); + } + + private static Stream<byte[]> invalidUtf8() { + return Stream.of( + new byte[] {'a', (byte) 0xFF, 'b'}, + new byte[] {'a', (byte) 0xC3}, + new byte[] {(byte) 0xC0, (byte) 0xAF}, + new byte[] {(byte) 0xED, (byte) 0xA0, (byte) 0x80}); + } + + @ParameterizedTest + @EnumSource(Layout.class) + void testCastStringToVariantReadsEverySegmentLayout(final Layout layout) { + final String valid = "Grüße, 世界 🚀"; + final byte[] invalid = {'a', (byte) 0xFF, 'b'}; + + assertThat(fromString(layout.place.apply(valid.getBytes(UTF_8)))) + .isEqualTo(BUILDER.of(layout.place.apply(valid.getBytes(UTF_8)).toString())); + assertThat(fromString(layout.place.apply(invalid))) + .isEqualTo(BUILDER.of(layout.place.apply(invalid).toString())); + } + + /** Where the bytes of a {@link BinaryStringData} live. */ + private enum Layout { + HEAP_SEGMENT(VariantCastUtilsTest::binaryString), + OFF_HEAP_SEGMENT(VariantCastUtilsTest::offHeapString), + FIRST_OF_TWO_SEGMENTS(VariantCastUtilsTest::firstOfTwoSegments), + ACROSS_TWO_SEGMENTS(VariantCastUtilsTest::splitString), + JAVA_OBJECT(utf8 -> StringData.fromString(new String(utf8, UTF_8))); + + private final Function<byte[], StringData> place; + + Layout(final Function<byte[], StringData> place) { + this.place = place; + } + } + + @ParameterizedTest(name = "{0} as DECIMAL({1}, {2})") + @CsvSource({ + "0, 1, 0", + "1.50, 10, 2", + "999999999, 9, 0", + "-0.999999999, 9, 9", + "1000000000, 10, 0", + "0.0000000001, 10, 10", + "-999999999999999999, 18, 0", + "0.999999999999999999, 18, 18", + "1000000000000000000, 19, 0", + "-99999999999999999999999999999999999999, 38, 0", + "0.99999999999999999999999999999999999999, 38, 38" + }) + void testCastDecimalToVariantKeepsUnscaledValueAndScale( + final BigDecimal decimal, final int precision, final int scale) { + assertThat(fromDecimal(DecimalData.fromBigDecimal(decimal, precision, scale))) + .isEqualTo(BUILDER.of(decimal)); } @Test @@ -111,6 +216,43 @@ class VariantCastUtilsTest { .isEqualTo("2021-09-24 14:34:56.123456"); } + /** Places the bytes inside a larger segment, the way a row field points into its row. */ + private static StringData binaryString(final byte[] utf8) { + final byte[] row = new byte[utf8.length + 2]; + System.arraycopy(utf8, 0, row, 1, utf8.length); + return BinaryStringData.fromAddress( + new MemorySegment[] {MemorySegmentFactory.wrap(row)}, 1, utf8.length); + } + + private static StringData offHeapString(final byte[] utf8) { + final MemorySegment segment = + MemorySegmentFactory.wrapOffHeapMemory(ByteBuffer.allocateDirect(utf8.length + 2)); + segment.put(1, utf8); + return BinaryStringData.fromAddress(new MemorySegment[] {segment}, 1, utf8.length); + } + + /** Places the bytes in the first of two segments, the way a short field sits in a large row. */ + private static StringData firstOfTwoSegments(final byte[] utf8) { + final byte[] first = new byte[utf8.length + 2]; + System.arraycopy(utf8, 0, first, 1, utf8.length); + final MemorySegment[] segments = { + MemorySegmentFactory.wrap(first), MemorySegmentFactory.wrap(new byte[first.length]) + }; + return BinaryStringData.fromAddress(segments, 1, utf8.length); + } + + /** Splits the bytes over two segments of the same size, the way a large row spans pages. */ + private static StringData splitString(final byte[] utf8) { + final int segmentSize = utf8.length / 2 + 1; + final byte[] row = new byte[2 * segmentSize]; + System.arraycopy(utf8, 0, row, 1, utf8.length); + final MemorySegment[] segments = { + MemorySegmentFactory.wrap(Arrays.copyOfRange(row, 0, segmentSize)), + MemorySegmentFactory.wrap(Arrays.copyOfRange(row, segmentSize, row.length)) + }; + return BinaryStringData.fromAddress(segments, 1, utf8.length); + } + /** Replaces the header of the array's second element. */ private static Variant withHeader(final BinaryVariant array, final byte header) { final byte[] value = array.getValue().clone();
