This is an automated email from the ASF dual-hosted git repository.
snuyanzin pushed a commit to branch release-2.4
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-2.4 by this push:
new 70d3df50eb4 [FLINK-40958][core]
`BinaryVariantInternalBuilder.appendVariantImpl` might lead to `StackOverflow`
70d3df50eb4 is described below
commit 70d3df50eb40a6941b7d631c9f47b16f0b8658db
Author: Sergey Nuyanzin <[email protected]>
AuthorDate: Fri Oct 30 19:48:01 2026 +0100
[FLINK-40958][core] `BinaryVariantInternalBuilder.appendVariantImpl` might
lead to `StackOverflow`
---
.../variant/BinaryVariantInternalBuilder.java | 141 +++++++++++++++------
.../variant/BinaryVariantInternalBuilderTest.java | 64 ++++++++++
.../flink/table/data/utils/ToVariantConverter.java | 21 +--
.../table/data/utils/ToVariantConverterTest.java | 35 -----
4 files changed, 173 insertions(+), 88 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 e922a2bdec8..4b72af72761 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
@@ -31,9 +31,11 @@ import java.io.IOException;
import java.math.BigDecimal;
import java.math.BigInteger;
import java.nio.charset.StandardCharsets;
+import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Comparator;
+import java.util.Deque;
import java.util.HashMap;
import java.util.UUID;
@@ -621,49 +623,78 @@ public class BinaryVariantInternalBuilder {
appendVariantImpl(v.rawValue(), v.getMetadata(), v.getPos());
}
+ // Uses an explicit stack rather than recursion, so deeply nested input
cannot overflow.
private void appendVariantImpl(byte[] value, byte[] metadata, int pos) {
checkIndex(pos, value.length);
- int basicType = value[pos] & BASIC_TYPE_MASK;
- switch (basicType) {
+ int rootType = value[pos] & BASIC_TYPE_MASK;
+ if (rootType != OBJECT && rootType != ARRAY) {
+ // Scalars are the common case on per-record paths, so skip
allocating the stack.
+ shallowAppendVariantImpl(value, pos);
+ return;
+ }
+ final Deque<AppendFrame> stack = new ArrayDeque<>();
+ appendOrPush(value, pos, stack);
+ while (!stack.isEmpty()) {
+ final AppendFrame frame = stack.peek();
+ if (frame.index == frame.size) {
+ stack.pop();
+ if (frame.fields != null) {
+ finishWritingObject(frame.start, frame.fields);
+ } else {
+ finishWritingArray(frame.start, frame.offsets);
+ }
+ continue;
+ }
+ int i = frame.index++;
+ int offset =
+ readUnsigned(value, frame.offsetStart + frame.offsetSize *
i, frame.offsetSize);
+ if (frame.fields != null) {
+ int id = readUnsigned(value, frame.idStart + frame.idSize * i,
frame.idSize);
+ addField(frame.fields, frame.start, getMetadataKey(metadata,
id));
+ } else {
+ frame.offsets.add(writePos - frame.start);
+ }
+ int elementPos = frame.dataStart + offset;
+ checkIndex(elementPos, value.length);
+ appendOrPush(value, elementPos, stack);
+ }
+ }
+
+ // Copies a scalar right away; for an object or array, pushes a frame
whose children are
+ // copied by the loop above.
+ private void appendOrPush(byte[] value, int pos, Deque<AppendFrame> stack)
{
+ switch (value[pos] & BASIC_TYPE_MASK) {
case OBJECT:
- handleObject(
- value,
- pos,
- (size, idSize, offsetSize, idStart, offsetStart,
dataStart) -> {
- ArrayList<FieldEntry> fields = new
ArrayList<>(size);
- int start = writePos;
- for (int i = 0; i < size; ++i) {
- int id = readUnsigned(value, idStart + idSize
* i, idSize);
- int offset =
- readUnsigned(
- value, offsetStart +
offsetSize * i, offsetSize);
- int elementPos = dataStart + offset;
- String key = getMetadataKey(metadata, id);
- addField(fields, start, key);
- appendVariantImpl(value, metadata, elementPos);
- }
- finishWritingObject(start, fields);
- return null;
- });
+ stack.push(
+ handleObject(
+ value,
+ pos,
+ (size, idSize, offsetSize, idStart,
offsetStart, dataStart) ->
+ new AppendFrame(
+ size,
+ idSize,
+ idStart,
+ offsetSize,
+ offsetStart,
+ dataStart,
+ writePos,
+ true)));
break;
case ARRAY:
- handleArray(
- value,
- pos,
- (size, offsetSize, offsetStart, dataStart) -> {
- ArrayList<Integer> offsets = new ArrayList<>(size);
- int start = writePos;
- for (int i = 0; i < size; ++i) {
- int offset =
- readUnsigned(
- value, offsetStart +
offsetSize * i, offsetSize);
- int elementPos = dataStart + offset;
- offsets.add(writePos - start);
- appendVariantImpl(value, metadata, elementPos);
- }
- finishWritingArray(start, offsets);
- return null;
- });
+ stack.push(
+ handleArray(
+ value,
+ pos,
+ (size, offsetSize, offsetStart, dataStart) ->
+ new AppendFrame(
+ size,
+ 0,
+ 0,
+ offsetSize,
+ offsetStart,
+ dataStart,
+ writePos,
+ false)));
break;
default:
shallowAppendVariantImpl(value, pos);
@@ -671,6 +702,40 @@ public class BinaryVariantInternalBuilder {
}
}
+ private static final class AppendFrame {
+ private final int size;
+ private final int idSize;
+ private final int idStart;
+ private final int offsetSize;
+ private final int offsetStart;
+ private final int dataStart;
+ private final int start;
+ // Exactly one is non-null: fields for an object, offsets for an array.
+ private final ArrayList<FieldEntry> fields;
+ private final ArrayList<Integer> offsets;
+ private int index;
+
+ private AppendFrame(
+ int size,
+ int idSize,
+ int idStart,
+ int offsetSize,
+ int offsetStart,
+ int dataStart,
+ int start,
+ boolean isObject) {
+ this.size = size;
+ this.idSize = idSize;
+ this.idStart = idStart;
+ this.offsetSize = offsetSize;
+ this.offsetStart = offsetStart;
+ this.dataStart = dataStart;
+ this.start = start;
+ this.fields = isObject ? new ArrayList<>(size) : null;
+ this.offsets = isObject ? null : new ArrayList<>(size);
+ }
+ }
+
private void shallowAppendVariantImpl(byte[] value, int pos) {
int size = valueSize(value, pos);
checkIndex(pos + size - 1, value.length);
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 a6b5d153a64..f9e242f476b 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
@@ -40,8 +40,13 @@ 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.U16_MAX;
+import static org.apache.flink.types.variant.BinaryVariantUtil.U24_MAX;
import static org.apache.flink.types.variant.BinaryVariantUtil.U32_SIZE;
+import static org.apache.flink.types.variant.BinaryVariantUtil.U8_MAX;
+import static org.apache.flink.types.variant.BinaryVariantUtil.arrayHeader;
import static org.apache.flink.types.variant.BinaryVariantUtil.primitiveHeader;
+import static org.apache.flink.types.variant.BinaryVariantUtil.writeLong;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatCode;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -247,6 +252,65 @@ class BinaryVariantInternalBuilderTest {
assertThat(object.toJson()).isEqualTo("{\"x\":true,\"y\":9}");
}
+ @Test
+ void testAppendDeeplyNestedVariant() {
+ // Deep enough that a recursive copy overflows any default stack with
StackOverflowError.
+ final int depth = 100_000;
+ final VariantBuilder builder = Variant.newBuilder();
+ final BinaryVariant leaf = (BinaryVariant) builder.of(1);
+ final byte[] leafValue = leaf.getValue();
+
+ // Encodes [[...[1]...]] directly, from the leaf outwards, since
building it level by
+ // level would copy the value once per level.
+ final byte[] buffer = new byte[leafValue.length + depth * (2 + 2 *
U32_SIZE)];
+ int start = buffer.length - leafValue.length;
+ System.arraycopy(leafValue, 0, buffer, start, leafValue.length);
+ for (int i = 0; i < depth; i++) {
+ final int childSize = buffer.length - start;
+ final int offsetSize =
+ childSize <= U8_MAX
+ ? 1
+ : childSize <= U16_MAX ? 2 : childSize <= U24_MAX
? 3 : U32_SIZE;
+ start -= 2 + 2 * offsetSize;
+ buffer[start] = arrayHeader(false, offsetSize);
+ buffer[start + 1] = 1;
+ writeLong(buffer, start + 2, 0, offsetSize);
+ writeLong(buffer, start + 2 + offsetSize, childSize, offsetSize);
+ }
+ final BinaryVariant nested =
+ new BinaryVariant(
+ Arrays.copyOfRange(buffer, start, buffer.length),
leaf.getMetadata());
+
+ final Variant wrapped = builder.array().add(nested).build();
+
+ assertThat(wrapped.toJson()).isEqualTo("[".repeat(depth + 1) + "1" +
"]".repeat(depth + 1));
+ }
+
+ @Test
+ void testAppendNestedObjectsAndArraysWithSiblings() {
+ final VariantBuilder builder = Variant.newBuilder();
+ Variant nested = builder.of(0);
+ String expected = "0";
+ // Each level copies the previous one, with siblings before and after
the nested value.
+ for (int i = 1; i <= 1000; i++) {
+ if (i % 2 == 0) {
+ // Keys out of order, so field order differs from data order.
+ nested = builder.object().add("v", builder.of(i)).add("k",
nested).build();
+ expected = "{\"k\":" + expected + ",\"v\":" + i + "}";
+ } else {
+ nested =
+ builder.array()
+ .add(builder.of(i))
+ .add(nested)
+ .add(builder.ofNull())
+ .build();
+ expected = "[" + i + "," + expected + ",null]";
+ }
+ }
+
+ assertThat(nested.toJson()).isEqualTo(expected);
+ }
+
@Test
void testAppendFloat() {
BinaryVariantInternalBuilder builder = new
BinaryVariantInternalBuilder(false);
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/utils/ToVariantConverter.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/utils/ToVariantConverter.java
index 8f32299806f..f4f94becc43 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/utils/ToVariantConverter.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/utils/ToVariantConverter.java
@@ -102,8 +102,8 @@ public final class ToVariantConverter implements
Serializable {
* Converts a value into a new {@code VARIANT}. A {@code null} value
becomes a variant null.
*
* @throws TableRuntimeException if the value has a {@code NULL} map key,
a time or timestamp
- * outside the range of its variant kind, does not fit into the 16 MiB
of a {@code VARIANT},
- * or is nested too deeply
+ * outside the range of its variant kind, or does not fit into the 16
MiB of a {@code
+ * VARIANT}
* @throws VariantTypeException if a nested {@code VARIANT} is malformed
*/
public Variant convert(Object value) {
@@ -122,15 +122,6 @@ public final class ToVariantConverter implements
Serializable {
"Cannot cast a value of type %s to VARIANT. A
VARIANT is limited to "
+ "16 MiB.",
sourceType));
- } catch (StackOverflowError e) {
- // The builder copies an embedded VARIANT with one call per
nesting level, so a deeply
- // nested one overflows the stack. A bare StackOverflowError would
not say which cast
- // failed or why.
- throw new TableRuntimeException(
- String.format(
- "Cannot cast a value of type %s to VARIANT because
it is nested too "
- + "deeply.",
- sourceType));
}
}
@@ -140,10 +131,10 @@ public final class ToVariantConverter implements
Serializable {
*
* <p>It fails like {@link #convert} for a {@code NULL} map key, a time or
timestamp out of
* range, and a string or binary value that can never fit. The caller
builds the result, so it
- * handles {@link
BinaryVariantInternalBuilder#VARIANT_SIZE_LIMIT_EXCEPTION} and a {@link
- * StackOverflowError} from a deeply nested {@code VARIANT} itself. A
repeated {@code MAP} key
- * keeps the last value if the caller's builder allows duplicate keys, and
fails otherwise. The
- * builder holds partial data after a failure, so the caller must not use
it any further.
+ * handles {@link
BinaryVariantInternalBuilder#VARIANT_SIZE_LIMIT_EXCEPTION} itself. A repeated
+ * {@code MAP} key keeps the last value if the caller's builder allows
duplicate keys, and fails
+ * otherwise. The builder holds partial data after a failure, so the
caller must not use it any
+ * further.
*/
public void writeTo(BinaryVariantInternalBuilder builder, Object value) {
writer.write(builder, value);
diff --git
a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/data/utils/ToVariantConverterTest.java
b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/data/utils/ToVariantConverterTest.java
index 91547c186d0..eef5ce60be1 100644
---
a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/data/utils/ToVariantConverterTest.java
+++
b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/data/utils/ToVariantConverterTest.java
@@ -63,7 +63,6 @@ import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.UUID;
-import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import java.util.stream.Stream;
@@ -336,40 +335,6 @@ class ToVariantConverterTest {
+ "limited to 16 MiB.");
}
- @Test
- void testDeeplyNestedVariantFails() throws InterruptedException {
- // [[[...[1]...]]], built without recursion
- final BinaryVariantInternalBuilder builder = new
BinaryVariantInternalBuilder(false);
- builder.appendInt(1);
- for (int i = 0; i < 10_000; i++) {
- builder.finishWritingArray(0, new ArrayList<>(List.of(0)));
- }
- final ArrayData array = new GenericArrayData(new Object[]
{builder.build()});
-
- // A small stack makes the overflow independent of the JVM's default
stack size.
- final AtomicReference<Throwable> error = new AtomicReference<>();
- final Thread thread =
- new Thread(
- null,
- () -> {
- try {
- convert(DataTypes.ARRAY(DataTypes.VARIANT()),
array);
- } catch (Throwable t) {
- error.set(t);
- }
- },
- "deeply-nested-variant",
- 256 * 1024);
- thread.start();
- thread.join();
-
- assertThat(error.get())
- .isInstanceOf(TableRuntimeException.class)
- .hasMessage(
- "Cannot cast a value of type ARRAY<VARIANT> to VARIANT
because it is "
- + "nested too deeply.");
- }
-
@Test
void testWriteIntoASharedBuilder() {
final ToVariantConverter converter =
