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 fb82352b670d9b4cbc82b7556e2329dc2330f4c0 Author: Ramin Gharib <[email protected]> AuthorDate: Mon Oct 5 13:35:06 2026 +0200 [FLINK-40871][core] Append VARIANT strings as UTF-8 bytes and decimals as an unscaled long `BinaryVariantInternalBuilder` gets `appendString(byte[], int, int)`, which copies a range of UTF-8 bytes as they are. `appendString(byte[])` and `appendString(String)` delegate to it. `appendDecimal(long, int)` writes a `DECIMAL4` or a `DECIMAL8` straight from the unscaled value, choosing the width the same way as `appendDecimal(BigDecimal)`. Anything wider, or with a negative scale, goes through `appendDecimal(BigDecimal)`, so the result is the same for every input. --- .../variant/BinaryVariantInternalBuilder.java | 64 +++++++++++++++++++--- .../variant/BinaryVariantInternalBuilderTest.java | 63 +++++++++++++++++++++ 2 files changed, 120 insertions(+), 7 deletions(-) diff --git a/flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariantInternalBuilder.java b/flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariantInternalBuilder.java index 47a9bcd886f..0ad4fca01e5 100644 --- a/flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariantInternalBuilder.java +++ b/flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariantInternalBuilder.java @@ -101,6 +101,11 @@ public class BinaryVariantInternalBuilder { // UTF-32 hint and could turn input that is not valid JSON into a value. private static final JsonFactory JSON_FACTORY = new JsonFactoryBuilder().disable(JsonFactory.Feature.CHARSET_DETECTION).build(); + // An unscaled value fits DECIMAL4 or DECIMAL8 when its magnitude is below 10^precision. + private static final long DECIMAL4_UNSCALED_LIMIT = + BigInteger.TEN.pow(MAX_DECIMAL4_PRECISION).longValueExact(); + private static final long DECIMAL8_UNSCALED_LIMIT = + BigInteger.TEN.pow(MAX_DECIMAL8_PRECISION).longValueExact(); public BinaryVariantInternalBuilder(boolean allowDuplicateKeys) { this.allowDuplicateKeys = allowDuplicateKeys; @@ -185,18 +190,33 @@ public class BinaryVariantInternalBuilder { } public void appendString(String str) { - byte[] text = str.getBytes(StandardCharsets.UTF_8); - boolean longStr = text.length > MAX_SHORT_STR_SIZE; - checkCapacity((longStr ? 1 + U32_SIZE : 1) + text.length); + appendString(str.getBytes(StandardCharsets.UTF_8)); + } + + /** + * Appends a string given as UTF-8 bytes. The bytes are copied as they are, so the caller must + * make sure they are valid UTF-8, which the variant spec requires of every string. + */ + public void appendString(byte[] utf8) { + appendString(utf8, 0, utf8.length); + } + + /** + * Like {@link #appendString(byte[])}, for the UTF-8 bytes {@code utf8[offset, offset + + * length)}. + */ + public void appendString(byte[] utf8, int offset, int length) { + boolean longStr = length > MAX_SHORT_STR_SIZE; + checkCapacity(1 + (longStr ? U32_SIZE : 0) + length); if (longStr) { writeBuffer[writePos++] = primitiveHeader(LONG_STR); - writeLong(writeBuffer, writePos, text.length, U32_SIZE); + writeLong(writeBuffer, writePos, length, U32_SIZE); writePos += U32_SIZE; } else { - writeBuffer[writePos++] = shortStrHeader(text.length); + writeBuffer[writePos++] = shortStrHeader(length); } - System.arraycopy(text, 0, writeBuffer, writePos, text.length); - writePos += text.length; + System.arraycopy(utf8, offset, writeBuffer, writePos, length); + writePos += length; } public void appendNull() { @@ -295,6 +315,36 @@ public class BinaryVariantInternalBuilder { } } + /** + * Appends a decimal given as its unscaled value and scale. The result is the same as {@link + * #appendDecimal(BigDecimal)} for {@code BigDecimal.valueOf(unscaled, scale)}, but a decimal + * that fits {@link BinaryVariantUtil#DECIMAL4} or {@link BinaryVariantUtil#DECIMAL8} is written + * without building a {@link BigDecimal}. + */ + public void appendDecimal(long unscaled, int scale) { + if (scale < 0 + || scale > MAX_DECIMAL8_PRECISION + || unscaled <= -DECIMAL8_UNSCALED_LIMIT + || unscaled >= DECIMAL8_UNSCALED_LIMIT) { + appendDecimal(BigDecimal.valueOf(unscaled, scale)); + return; + } + checkCapacity(2 + 8); + if (scale <= MAX_DECIMAL4_PRECISION + && unscaled > -DECIMAL4_UNSCALED_LIMIT + && unscaled < DECIMAL4_UNSCALED_LIMIT) { + writeBuffer[writePos++] = primitiveHeader(DECIMAL4); + writeBuffer[writePos++] = (byte) scale; + writeLong(writeBuffer, writePos, unscaled, 4); + writePos += 4; + } else { + writeBuffer[writePos++] = primitiveHeader(DECIMAL8); + writeBuffer[writePos++] = (byte) scale; + writeLong(writeBuffer, writePos, unscaled, 8); + writePos += 8; + } + } + // The variant spec requires a scale in [0, 38] and a precision of at most 38. private static BigDecimal toVariantDecimal(BigDecimal d) { BigDecimal result = d; diff --git a/flink-core/src/test/java/org/apache/flink/types/variant/BinaryVariantInternalBuilderTest.java b/flink-core/src/test/java/org/apache/flink/types/variant/BinaryVariantInternalBuilderTest.java index 093be667a0c..b0e0090d7f9 100644 --- a/flink-core/src/test/java/org/apache/flink/types/variant/BinaryVariantInternalBuilderTest.java +++ b/flink-core/src/test/java/org/apache/flink/types/variant/BinaryVariantInternalBuilderTest.java @@ -27,6 +27,7 @@ import org.junit.jupiter.params.provider.ValueSource; import java.io.IOException; import java.math.BigDecimal; import java.util.ArrayList; +import java.util.Arrays; import java.util.Collections; import java.util.stream.Stream; @@ -34,6 +35,12 @@ import static java.nio.charset.StandardCharsets.UTF_16; import static java.nio.charset.StandardCharsets.UTF_16BE; import static java.nio.charset.StandardCharsets.UTF_16LE; import static java.nio.charset.StandardCharsets.UTF_8; +import static org.apache.flink.types.variant.BinaryVariantUtil.DECIMAL16; +import static org.apache.flink.types.variant.BinaryVariantUtil.DECIMAL4; +import static org.apache.flink.types.variant.BinaryVariantUtil.DECIMAL8; +import static org.apache.flink.types.variant.BinaryVariantUtil.MAX_SHORT_STR_SIZE; +import static org.apache.flink.types.variant.BinaryVariantUtil.U32_SIZE; +import static org.apache.flink.types.variant.BinaryVariantUtil.primitiveHeader; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatCode; import static org.assertj.core.api.Assertions.assertThatThrownBy; @@ -201,4 +208,60 @@ class BinaryVariantInternalBuilderTest { assertThatCode(() -> floatList.forEach(builder::appendFloat)).doesNotThrowAnyException(); } + + @ParameterizedTest + @ValueSource(ints = {0, MAX_SHORT_STR_SIZE, MAX_SHORT_STR_SIZE + 1}) + void testAppendStringStoresUtf8BytesAsTheyAre(final int length) { + final String str = "x".repeat(length); + final byte[] utf8 = str.getBytes(UTF_8); + // The bytes follow one other byte, so a range that ignores its offset reads the wrong ones. + final byte[] buffer = new byte[1 + length]; + System.arraycopy(utf8, 0, buffer, 1, length); + final BinaryVariantInternalBuilder fromRange = new BinaryVariantInternalBuilder(false); + fromRange.appendString(buffer, 1, length); + final BinaryVariantInternalBuilder fromArray = new BinaryVariantInternalBuilder(false); + fromArray.appendString(utf8); + + final BinaryVariant variant = fromRange.build(); + final byte[] value = variant.getValue(); + final int headerSize = 1 + (length > MAX_SHORT_STR_SIZE ? U32_SIZE : 0); + assertThat(Arrays.copyOfRange(value, headerSize, value.length)).isEqualTo(utf8); + assertThat(variant.getString()).isEqualTo(str); + assertThat(variant).isEqualTo(fromArray.build()); + } + + @ParameterizedTest(name = "unscaled={0}, scale={1}") + @MethodSource("unscaledDecimals") + void testAppendDecimalFromUnscaledLong( + final long unscaled, final int scale, final int decimalType) { + final BigDecimal decimal = BigDecimal.valueOf(unscaled, scale); + final BinaryVariantInternalBuilder fromLong = new BinaryVariantInternalBuilder(false); + fromLong.appendDecimal(unscaled, scale); + final BinaryVariantInternalBuilder fromBigDecimal = new BinaryVariantInternalBuilder(false); + fromBigDecimal.appendDecimal(decimal); + + final BinaryVariant variant = fromLong.build(); + assertThat(variant).isEqualTo(fromBigDecimal.build()); + assertThat(variant.getValue()[0]).isEqualTo(primitiveHeader(decimalType)); + assertThat(variant.getDecimal()).isEqualByComparingTo(decimal); + } + + private static Stream<Arguments> unscaledDecimals() { + return Stream.of( + Arguments.of(0L, 0, DECIMAL4), + Arguments.of(999_999_999L, 9, DECIMAL4), + Arguments.of(-999_999_999L, 9, DECIMAL4), + Arguments.of(1_000_000_000L, 0, DECIMAL8), + Arguments.of(-1_000_000_000L, 0, DECIMAL8), + Arguments.of(1L, 10, DECIMAL8), + Arguments.of(999_999_999_999_999_999L, 18, DECIMAL8), + Arguments.of(-999_999_999_999_999_999L, 18, DECIMAL8), + Arguments.of(1_000_000_000_000_000_000L, 0, DECIMAL16), + Arguments.of(-1_000_000_000_000_000_000L, 0, DECIMAL16), + Arguments.of(1L, 19, DECIMAL16), + Arguments.of(Long.MAX_VALUE, 38, DECIMAL16), + Arguments.of(Long.MIN_VALUE, 0, DECIMAL16), + // A negative scale is rescaled to 0. + Arguments.of(5L, -1, DECIMAL4)); + } }
