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
The following commit(s) were added to refs/heads/master by this push:
new b2454792477 [FLINK-40958][core] Do not use recurcion in `VARIANT`
`toJson`
b2454792477 is described below
commit b245479247729620e2c4c4128b843ba56f2f42a2
Author: Sergey Nuyanzin <[email protected]>
AuthorDate: Thu Oct 8 16:23:47 2026 +0200
[FLINK-40958][core] Do not use recurcion in `VARIANT` `toJson`
---
.../flink/types/variant/JsonVariantFormatter.java | 93 +++++++++++++++-------
.../types/variant/JsonVariantFormatterTest.java | 14 ++++
2 files changed, 80 insertions(+), 27 deletions(-)
diff --git
a/flink-core/src/main/java/org/apache/flink/types/variant/JsonVariantFormatter.java
b/flink-core/src/main/java/org/apache/flink/types/variant/JsonVariantFormatter.java
index 534e3723a3b..5ad25ac3cd4 100644
---
a/flink-core/src/main/java/org/apache/flink/types/variant/JsonVariantFormatter.java
+++
b/flink-core/src/main/java/org/apache/flink/types/variant/JsonVariantFormatter.java
@@ -28,7 +28,9 @@ import java.io.IOException;
import java.time.LocalDate;
import java.time.LocalTime;
import java.time.ZoneOffset;
+import java.util.ArrayDeque;
import java.util.Base64;
+import java.util.Deque;
import static
org.apache.flink.types.variant.BinaryVariantUtil.TIMESTAMP_FORMATTER;
import static
org.apache.flink.types.variant.BinaryVariantUtil.TIMESTAMP_LTZ_FORMATTER;
@@ -70,43 +72,69 @@ public final class JsonVariantFormatter {
return sb.toString();
}
+ // Renders the node and its descendants via an explicit work stack, so
deep nesting cannot
+ // overflow.
private void append(
- final byte[] value, final byte[] metadata, final int pos, final
StringBuilder sb) {
- final int start = sb.length();
- try {
- appendNode(value, metadata, pos, sb);
- } catch (VariantTypeException e) {
- if (!lenient) {
- throw e;
+ final byte[] value, final byte[] metadata, final int startPos,
final StringBuilder sb) {
+ final Deque<Object> stack = new ArrayDeque<>();
+ stack.push(startPos);
+ while (!stack.isEmpty()) {
+ final Object top = stack.peek();
+ if (top instanceof ContainerFrame) {
+ final ContainerFrame frame = (ContainerFrame) top;
+ if (frame.index == frame.childPositions.length) {
+ sb.append(frame.closing);
+ stack.pop();
+ continue;
+ }
+ if (frame.index != 0) {
+ sb.append(',');
+ }
+ if (frame.keys != null) {
+ sb.append(frame.keys[frame.index]);
+ sb.append(':');
+ }
+ stack.push(frame.childPositions[frame.index++]);
+ continue;
+ }
+ final int pos = (Integer) top;
+ stack.pop();
+ final int start = sb.length();
+ try {
+ appendNode(value, metadata, pos, sb, stack);
+ } catch (VariantTypeException e) {
+ if (!lenient) {
+ throw e;
+ }
+ // Drop the node's partial output, such as a dangling key.
+ sb.setLength(start);
+ appendQuoted(sb, BinaryVariantUtil.undecodableNode(value,
pos));
}
- // Drop the node's partial output, such as a dangling key.
- sb.setLength(start);
- appendQuoted(sb, BinaryVariantUtil.undecodableNode(value, pos));
}
}
- private void appendNode(byte[] value, byte[] metadata, int pos,
StringBuilder sb) {
+ // Decodes a container's header before writing anything so a malformed one
leaves sb untouched
+ // for the lenient reset path.
+ private void appendNode(
+ byte[] value, byte[] metadata, int pos, StringBuilder sb,
Deque<Object> stack) {
switch (BinaryVariantUtil.getType(value, pos)) {
case OBJECT:
handleObject(
value,
pos,
(size, idSize, offsetSize, idStart, offsetStart,
dataStart) -> {
- sb.append('{');
+ final int[] childPositions = new int[size];
+ final String[] keys = new String[size];
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;
- if (i != 0) {
- sb.append(',');
- }
- sb.append(escapeJson(getMetadataKey(metadata,
id)));
- sb.append(':');
- append(value, metadata, elementPos, sb);
+ childPositions[i] = dataStart + offset;
+ keys[i] = escapeJson(getMetadataKey(metadata,
id));
}
- sb.append('}');
+ sb.append('{');
+ stack.push(new ContainerFrame(childPositions,
keys, '}'));
return null;
});
break;
@@ -115,18 +143,15 @@ public final class JsonVariantFormatter {
value,
pos,
(size, offsetSize, offsetStart, dataStart) -> {
- sb.append('[');
+ final int[] childPositions = new int[size];
for (int i = 0; i < size; ++i) {
int offset =
readUnsigned(
value, offsetStart +
offsetSize * i, offsetSize);
- int elementPos = dataStart + offset;
- if (i != 0) {
- sb.append(',');
- }
- append(value, metadata, elementPos, sb);
+ childPositions[i] = dataStart + offset;
}
- sb.append(']');
+ sb.append('[');
+ stack.push(new ContainerFrame(childPositions,
null, ']'));
return null;
});
break;
@@ -224,6 +249,20 @@ public final class JsonVariantFormatter {
}
}
+ private static final class ContainerFrame {
+ private final int[] childPositions;
+ // Quoted "key" tokens for an object, or null for an array.
+ private final String[] keys;
+ private final char closing;
+ private int index;
+
+ private ContainerFrame(int[] childPositions, String[] keys, char
closing) {
+ this.childPositions = childPositions;
+ this.keys = keys;
+ this.closing = closing;
+ }
+ }
+
/** JSON has no literal for NaN or infinity. */
private void appendNonFinite(final StringBuilder sb, final String text,
final double number) {
if (!lenient) {
diff --git
a/flink-core/src/test/java/org/apache/flink/types/variant/JsonVariantFormatterTest.java
b/flink-core/src/test/java/org/apache/flink/types/variant/JsonVariantFormatterTest.java
index e47878dfcf6..3a2aac1b95d 100644
---
a/flink-core/src/test/java/org/apache/flink/types/variant/JsonVariantFormatterTest.java
+++
b/flink-core/src/test/java/org/apache/flink/types/variant/JsonVariantFormatterTest.java
@@ -94,6 +94,20 @@ class JsonVariantFormatterTest {
assertThat(variant.toString()).isEqualTo("[1,\"<INVALID>\"]");
}
+ @Test
+ void testDeeplyNestedValueRenders() {
+ final int depth = 5000;
+ Variant nested = BUILDER.of(1);
+ for (int i = 0; i < depth; i++) {
+ nested = BUILDER.array().add(nested).build();
+ }
+ final Variant variant = nested;
+
+ final String expected = "[".repeat(depth) + "1" + "]".repeat(depth);
+ assertThat(variant.toJson()).isEqualTo(expected);
+ assertThat(variant.toString()).isEqualTo(expected);
+ }
+
/** 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();
