This is an automated email from the ASF dual-hosted git repository. twalthr pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit 374289707dc40a056f0a97914da65a20b19a26da Author: Ramin Gharib <[email protected]> AuthorDate: Mon Oct 5 15:08:25 2026 +0200 [FLINK-40911][core] Fix two bugs in the variant builder BinaryVariantInternalBuilder.appendVariant read a variant through getValue(), which copies a nested value, such as a field or an element of another variant, to a new array starting at position 0. It then read that copy at the original position, so adding such a value to an object or an array failed with MALFORMED_VARIANT or silently wrote a wrong value. It now reads the shared buffer at the variant's own position. The builder also asserted that every size, offset and id fits into 3 bytes. An array or object whose content fills exactly 16 MiB broke that assertion, so it failed with a bare AssertionError when assertions are enabled, and with VARIANT_SIZE_LIMIT_EXCEPTION only when they are not. A value that large is always over the size limit, so the builder now throws VARIANT_SIZE_LIMIT_EXCEPTION in both cases. --- .../variant/BinaryVariantInternalBuilder.java | 13 ++++-- .../variant/BinaryVariantInternalBuilderTest.java | 46 ++++++++++++++++++++++ 2 files changed, 55 insertions(+), 4 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 0ad4fca01e5..0143f81da41 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 @@ -587,7 +587,9 @@ public class BinaryVariantInternalBuilder { // the // input variant, we can directly copy the binary slice. public void appendVariant(BinaryVariant v) { - appendVariantImpl(v.getValue(), v.getMetadata(), v.getPos()); + // A nested variant, such as a field or an element of another variant, starts at its own + // position in the shared buffer. getValue() would copy it to position 0 instead. + appendVariantImpl(v.rawValue(), v.getMetadata(), v.getPos()); } private void appendVariantImpl(byte[] value, byte[] metadata, int pos) { @@ -748,10 +750,13 @@ public class BinaryVariantInternalBuilder { } } - // Choose the smallest unsigned integer type that can store `value`. It must be within - // `[0, U24_MAX]`. + // Choose the smallest unsigned integer type that can store `value`. A size, offset or id above + // `U24_MAX` only occurs in a variant over the size limit. private int getIntegerSize(int value) { - assert value >= 0 && value <= U24_MAX; + assert value >= 0; + if (value > U24_MAX) { + throw VARIANT_SIZE_LIMIT_EXCEPTION; + } if (value <= U8_MAX) { return 1; } 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 b0e0090d7f9..5a8e2c07254 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 @@ -39,6 +39,7 @@ 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.SIZE_LIMIT; 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; @@ -201,6 +202,51 @@ class BinaryVariantInternalBuilderTest { assertThat(variant.getDouble()).isEqualTo(Double.parseDouble(number)); } + @Test + void testContainerOfExactlyTheSizeLimitFails() { + // A binary value of this length fills the 16 MiB alone, so no header fits around it. + final byte[] payload = new byte[SIZE_LIMIT - 1 - U32_SIZE]; + + final BinaryVariantInternalBuilder arrayBuilder = new BinaryVariantInternalBuilder(false); + arrayBuilder.appendBinary(payload); + final ArrayList<Integer> offsets = new ArrayList<>(Collections.singletonList(0)); + assertThatThrownBy(() -> arrayBuilder.finishWritingArray(0, offsets)) + .isSameAs(BinaryVariantInternalBuilder.VARIANT_SIZE_LIMIT_EXCEPTION); + + final BinaryVariantInternalBuilder objectBuilder = new BinaryVariantInternalBuilder(false); + final int id = objectBuilder.addKey("a"); + objectBuilder.appendBinary(payload); + final ArrayList<BinaryVariantInternalBuilder.FieldEntry> fields = + new ArrayList<>( + Collections.singletonList( + new BinaryVariantInternalBuilder.FieldEntry("a", id, 0))); + assertThatThrownBy(() -> objectBuilder.finishWritingObject(0, fields)) + .isSameAs(BinaryVariantInternalBuilder.VARIANT_SIZE_LIMIT_EXCEPTION); + } + + @Test + void testAppendNestedVariant() throws IOException { + final BinaryVariant source = + BinaryVariantInternalBuilder.parseJson( + "{\"a\":[7,8,9],\"b\":\"hello\",\"c\":{\"d\":true}}", false); + final VariantBuilder builder = Variant.newBuilder(); + + final Variant array = + builder.array() + .add(source.getField("a")) + .add(source.getField("b")) + .add(source.getField("c")) + .build(); + assertThat(array.toJson()).isEqualTo("[[7,8,9],\"hello\",{\"d\":true}]"); + + final Variant object = + builder.object() + .add("x", source.getField("c").getField("d")) + .add("y", source.getField("a").getElement(2)) + .build(); + assertThat(object.toJson()).isEqualTo("{\"x\":true,\"y\":9}"); + } + @Test void testAppendFloat() { BinaryVariantInternalBuilder builder = new BinaryVariantInternalBuilder(false);
