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
The following commit(s) were added to refs/heads/master by this push:
new f18d273d48b [FLINK-40854][table] Parse UTF-8 bytes directly in
PARSE_JSON and TRY_PARSE_JSON
f18d273d48b is described below
commit f18d273d48bb5d4d75e3827177a5437ebba0a3b9
Author: Ramin Gharib <[email protected]>
AuthorDate: Fri Oct 2 09:34:50 2026 +0200
[FLINK-40854][table] Parse UTF-8 bytes directly in PARSE_JSON and
TRY_PARSE_JSON
This closes #29342.
---
.../variant/BinaryVariantInternalBuilder.java | 20 +++++++-
.../variant/BinaryVariantInternalBuilderTest.java | 38 +++++++++++++++
.../planner/functions/JsonFunctionsITCase.java | 55 ++++++++++++++++++++--
.../nodes/exec/stream/VariantSemanticTest.java | 2 +-
.../functions/scalar/ParseJsonFunction.java | 15 +++++-
.../functions/scalar/TryParseJsonFunction.java | 2 +-
6 files changed, 123 insertions(+), 9 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 5e3520985d3..47a9bcd886f 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
@@ -21,6 +21,7 @@ package org.apache.flink.types.variant;
import org.apache.flink.annotation.Internal;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonFactory;
+import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonFactoryBuilder;
import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonParseException;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonParser;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonToken;
@@ -96,7 +97,10 @@ public class BinaryVariantInternalBuilder {
new VariantTypeException("VARIANT_SIZE_LIMIT");
public static final VariantTypeException VARIANT_DUPLICATE_KEY_EXCEPTION =
new VariantTypeException("VARIANT_DUPLICATE_KEY");
- private static final JsonFactory JSON_FACTORY = new JsonFactory();
+ // Byte input is always UTF-8. Charset detection would take NUL bytes or a
BOM as a UTF-16 or
+ // 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();
public BinaryVariantInternalBuilder(boolean allowDuplicateKeys) {
this.allowDuplicateKeys = allowDuplicateKeys;
@@ -115,6 +119,20 @@ public class BinaryVariantInternalBuilder {
}
}
+ /**
+ * Parse UTF-8 encoded JSON bytes as a Variant value. Invalid UTF-8 is
rejected rather than
+ * replaced with U+FFFD.
+ *
+ * @throws IOException if any JSON parsing error happens.
+ */
+ public static BinaryVariant parseJson(byte[] bytes, boolean
allowDuplicateKeys)
+ throws IOException {
+ try (JsonParser parser = JSON_FACTORY.createParser(bytes)) {
+ parser.nextToken();
+ return parseJson(parser, allowDuplicateKeys);
+ }
+ }
+
/**
* Similar {@link #parseJson(String, boolean)}, but takes a JSON parser
instead of string input.
*/
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 a1b3eb5ef06..093be667a0c 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
@@ -20,13 +20,20 @@ package org.apache.flink.types.variant;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
import org.junit.jupiter.params.provider.ValueSource;
import java.io.IOException;
import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.stream.Stream;
+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.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatCode;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -134,6 +141,37 @@ class BinaryVariantInternalBuilderTest {
assertThat(variant.toJson()).isEqualTo(json);
}
+ @Test
+ void testParseJsonFromUtf8Bytes() throws IOException {
+ final String json = "{\"schlüssel\":\"Grüße, 世界 🚀\",\"キー\":[\"äöü\"]}";
+
+
assertThat(BinaryVariantInternalBuilder.parseJson(json.getBytes(UTF_8), false))
+ .isEqualTo(BinaryVariantInternalBuilder.parseJson(json,
false));
+ }
+
+ private static Stream<Arguments> nonUtf8JsonBytes() {
+ return Stream.of(
+ // Charset detection would read these as UTF-16 or UTF-32, or
skip the BOM.
+ Arguments.of("trailing NUL", "1\u0000".getBytes(UTF_8)),
+ Arguments.of("leading NUL", "\u00001".getBytes(UTF_8)),
+ Arguments.of("UTF-8 BOM", "\uFEFF1".getBytes(UTF_8)),
+ Arguments.of("UTF-16BE", "{\"a\":1}".getBytes(UTF_16BE)),
+ Arguments.of("UTF-16LE", "{\"a\":1}".getBytes(UTF_16LE)),
+ Arguments.of("UTF-16 with BOM", "{\"a\":1}".getBytes(UTF_16)),
+ Arguments.of("invalid start byte", new byte[] {'"', (byte)
0xFF, '"'}),
+ Arguments.of("truncated sequence", new byte[] {'"', 'a',
(byte) 0xC3, '"'}),
+ Arguments.of(
+ "surrogate code point",
+ new byte[] {'"', (byte) 0xED, (byte) 0xA0, (byte)
0x80, '"'}));
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("nonUtf8JsonBytes")
+ void testParseJsonRejectsBytesThatAreNotUtf8Json(final String name, final
byte[] bytes) {
+ assertThatThrownBy(() -> BinaryVariantInternalBuilder.parseJson(bytes,
false))
+ .isInstanceOf(IOException.class);
+ }
+
@ParameterizedTest
@ValueSource(strings = {"NaN", "Infinity", "-Infinity", "1e400", "-1e400"})
void testParseJsonRejectsNonFiniteNumbers(final String nonFiniteNumber) {
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/JsonFunctionsITCase.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/JsonFunctionsITCase.java
index 358e5dd0bd0..3d8d6eefa4e 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/JsonFunctionsITCase.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/JsonFunctionsITCase.java
@@ -1051,20 +1051,20 @@ class JsonFunctionsITCase extends
BuiltInFunctionTestBase {
.testSqlRuntimeError(
"PARSE_JSON(f1)",
TableRuntimeException.class,
- "Failed to parse json string")
+ "Failed to parse JSON string")
.testTableApiRuntimeError(
$("f1").parseJson(),
TableRuntimeException.class,
- "Failed to parse json string")
+ "Failed to parse JSON string")
// allowDuplicateKeys: false (the default) rejects
duplicate keys
.testSqlRuntimeError(
"PARSE_JSON(f2, false)",
TableRuntimeException.class,
- "Failed to parse json string")
+ "Failed to parse JSON string")
.testTableApiRuntimeError(
$("f2").parseJson(false),
TableRuntimeException.class,
- "Failed to parse json string")
+ "Failed to parse JSON string")
// allowDuplicateKeys: true keeps the last occurrence
of the duplicated key
.testResult(
jsonString($("f2").parseJson(true)),
@@ -1096,6 +1096,53 @@ class JsonFunctionsITCase extends
BuiltInFunctionTestBase {
"JSON_STRING(TRY_PARSE_JSON(f2, true))",
"{\"a\":2}",
STRING()),
+ // Input is parsed from its UTF-8 bytes. A NUL or a BOM must
not be read as a
+ // UTF-16 or UTF-32 hint that turns invalid JSON into a value.
Only input that is
+ // not valid UTF-8 reports the index of the invalid byte.
+ // FROM_BASE64 wraps the decoded bytes without validating
them, so 'Iv8i' yields
+ // the invalid UTF-8 bytes 22 FF 22.
+ TestSetSpec.forFunction(
+ BuiltInFunctionDefinitions.PARSE_JSON,
+ "input bytes are always read as UTF-8")
+ .onFieldsWithData("1\u0000", "\u00001", "\uFEFF1",
"[\u65E5]", "Iv8i")
+ .andDataTypes(
+ STRING().notNull(),
+ STRING().notNull(),
+ STRING().notNull(),
+ STRING().notNull(),
+ STRING().notNull())
+ .testSqlRuntimeError(
+ "PARSE_JSON(f0)",
+ "Unexpected character ((CTRL-CHAR, code 0)):
Expected space separating root-level values")
+ .testSqlRuntimeError(
+ "PARSE_JSON(f1)",
+ "Illegal character ((CTRL-CHAR, code 0)): only
regular white space")
+ .testSqlRuntimeError(
+ "PARSE_JSON(f2)",
+ TableRuntimeException.class,
+ "Failed to parse JSON string: \uFEFF1")
+ .testSqlRuntimeError(
+ "PARSE_JSON(f3)",
+ TableRuntimeException.class,
+ "Failed to parse JSON string: [\u65E5]")
+ .testSqlRuntimeError(
+ "PARSE_JSON(FROM_BASE64(f4))",
+ TableRuntimeException.class,
+ "Failed to parse JSON string: Invalid UTF-8
byte at index 1 of 3."),
+ TestSetSpec.forFunction(
+ BuiltInFunctionDefinitions.TRY_PARSE_JSON,
+ "input bytes are always read as UTF-8")
+ .onFieldsWithData("1\u0000", "\u00001", "\uFEFF1",
"Iv8i")
+ .andDataTypes(
+ STRING().notNull(),
+ STRING().notNull(),
+ STRING().notNull(),
+ STRING().notNull())
+ .testSqlResult("JSON_STRING(TRY_PARSE_JSON(f0))",
null, STRING())
+ .testSqlResult("JSON_STRING(TRY_PARSE_JSON(f1))",
null, STRING())
+ .testSqlResult("JSON_STRING(TRY_PARSE_JSON(f2))",
null, STRING())
+ .testSqlResult(
+
"JSON_STRING(TRY_PARSE_JSON(FROM_BASE64(f3)))", null, STRING()),
TestSetSpec.forFunction(
BuiltInFunctionDefinitions.PARSE_JSON,
"VARIANT expression preceding another
expression in a"
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/VariantSemanticTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/VariantSemanticTest.java
index c3653a2e194..86e5b660210 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/VariantSemanticTest.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/VariantSemanticTest.java
@@ -91,7 +91,7 @@ public class VariantSemanticTest extends SemanticTestBase {
.runFailingSql(
"INSERT INTO sink_t SELECT PARSE_JSON(v) FROM t",
TableRuntimeException.class,
- "Failed to parse json string")
+ "Failed to parse JSON string")
.build();
static final TableTestProgram TRY_PARSE_JSON_HANDLE_MALFORMED_JSON =
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/ParseJsonFunction.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/ParseJsonFunction.java
index 94ed8c980e2..50f758e9b4d 100644
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/ParseJsonFunction.java
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/ParseJsonFunction.java
@@ -20,6 +20,7 @@ package org.apache.flink.table.runtime.functions.scalar;
import org.apache.flink.table.api.TableRuntimeException;
import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.binary.StringUtf8Utils;
import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
import org.apache.flink.table.functions.SpecializedFunction;
import org.apache.flink.types.variant.BinaryVariantInternalBuilder;
@@ -43,11 +44,21 @@ public class ParseJsonFunction extends
BuiltInScalarFunction {
return null;
}
+ final byte[] bytes = jsonStr.toBytes();
try {
- return BinaryVariantInternalBuilder.parseJson(jsonStr.toString(),
allowDuplicateKeys);
+ return BinaryVariantInternalBuilder.parseJson(bytes,
allowDuplicateKeys);
} catch (Throwable e) {
+ final int invalidIndex =
+ StringUtf8Utils.firstInvalidUtf8ByteIndex(bytes, 0,
bytes.length);
+ if (invalidIndex >= 0) {
+ throw new TableRuntimeException(
+ String.format(
+ "Failed to parse JSON string: Invalid UTF-8
byte at index %d of %d.",
+ invalidIndex, bytes.length),
+ e);
+ }
throw new TableRuntimeException(
- String.format("Failed to parse json string: %s", jsonStr),
e);
+ String.format("Failed to parse JSON string: %s", jsonStr),
e);
}
}
}
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/TryParseJsonFunction.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/TryParseJsonFunction.java
index 9cdde087ad7..161e1c65131 100644
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/TryParseJsonFunction.java
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/TryParseJsonFunction.java
@@ -43,7 +43,7 @@ public class TryParseJsonFunction extends
BuiltInScalarFunction {
}
try {
- return BinaryVariantInternalBuilder.parseJson(jsonStr.toString(),
allowDuplicateKeys);
+ return BinaryVariantInternalBuilder.parseJson(jsonStr.toBytes(),
allowDuplicateKeys);
} catch (Throwable e) {
return null;
}
