You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "raminqaf (via GitHub)" <gi...@apache.org> on 2026/09/30 11:37:22 UTC
[PR] [FLINK-40854][table] Parse UTF-8 bytes directly in PARSE_JSON and TRY_PARSE_JSON [flink]
raminqaf opened a new pull request, #29342:
URL: https://github.com/apache/flink/pull/29342
## What is the purpose of the change
`PARSE_JSON` and `TRY_PARSE_JSON` decoded every input `STRING` from UTF-8 into a Java `String`, and Jackson then parsed its chars. A Flink `STRING` is already UTF-8 bytes. This PR parses those bytes directly with Jackson's UTF-8 byte parser.
JMH, JDK 21. The input sits at a non-zero offset in a `MemorySegment`, like a field read from `BinaryRowData`.
| Document | Size | Before | After | Change |
|-------------------------------------------|--------|---------|---------|--------|
| Records: nested objects, numbers, strings | 1 KB | 3.51 µs | 3.01 µs | -14% |
| Records: nested objects, numbers, strings | 1 MB | 4.08 ms | 3.51 ms | -14% |
| One long string value | 100 KB | 160 µs | 63 µs | -61% |
| One long string value | 1 MB | 1.71 ms | 0.63 ms | -63% |
Allocation per call drops by 22 to 29% for documents of 100 KB and more.
Valid JSON parses to the same `VARIANT` as before. Only a `STRING` that holds invalid UTF-8 behaves differently:
| Input bytes | Function | Before | After |
|-------------|------------------|------------|--------------------------------------------------------------------------|
| `22 FF 22` | `PARSE_JSON` | `"\uFFFD"` | `Failed to parse json string`, caused by `Invalid UTF-8 start byte 0xff` |
| `22 FF 22` | `TRY_PARSE_JSON` | `"\uFFFD"` | `NULL` |
This matches FLINK-39623, which made `CAST(BYTES AS STRING)` fail on invalid UTF-8 by default.
The byte parser runs with `JsonFactory.Feature.CHARSET_DETECTION` disabled. With detection on, Jackson guesses UTF-16 or UTF-32 from NUL bytes or a BOM in the first four bytes. `TRY_PARSE_JSON(CONCAT('1', CHR(0)))` would then return `1` instead of `NULL`.
One open question for reviewers: Jackson decodes overlong UTF-8 such as `C0 AF` to `/` instead of rejecting it. Rejecting it too would need a separate UTF-8 validation pass. I left that out.
## Brief change log
- `BinaryVariantInternalBuilder#parseJson(byte[], boolean)` parses UTF-8 bytes. The shared `JsonFactory` disables `CHARSET_DETECTION`.
- `ParseJsonFunction` and `TryParseJsonFunction` pass `StringData.toBytes()` instead of `StringData.toString()`.
## Verifying this change
This change added tests and can be verified as follows:
- `BinaryVariantInternalBuilderTest`: the byte path yields the same `VARIANT` as the `String` path for non-ASCII input. Nine byte inputs must be rejected: a leading or trailing NUL, a UTF-8 BOM, three UTF-16 encodings, and three kinds of invalid UTF-8.
- `JsonFunctionsITCase`: `PARSE_JSON` fails and `TRY_PARSE_JSON` returns `NULL` for `1` followed by NUL, NUL followed by `1`, and a BOM followed by `1`. These fail without the `CHARSET_DETECTION` change.
- Existing `JsonFunctionsITCase`, `CastFunctionITCase`, `MapFunctionITCase`, `VariantSemanticTest` and `CalcRestoreTest` pass.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: no
- The serializers: no
- The runtime per-record code paths (performance sensitive): yes. `PARSE_JSON` and `TRY_PARSE_JSON` run per record. The benchmark above covers them.
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
- The S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? no
- If yes, how is the feature documented? not applicable. The invalid UTF-8 behavior change is in the release note on FLINK-40854.
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes (please specify the tool below)
Generated-by: Claude Opus 5.5
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40854][table] Parse UTF-8 bytes directly in PARSE_JSON and TRY_PARSE_JSON [flink]
Posted by "raminqaf (via GitHub)" <gi...@apache.org>.
raminqaf commented on code in PR #29342:
URL: https://github.com/apache/flink/pull/29342#discussion_r4153018266
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/ParseJsonFunction.java:
##########
@@ -43,9 +44,19 @@ public ParseJsonFunction(SpecializedFunction.SpecializedContext context) {
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.",
Review Comment:
Addressed
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/ParseJsonFunction.java:
##########
@@ -43,9 +44,19 @@ public ParseJsonFunction(SpecializedFunction.SpecializedContext context) {
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);
Review Comment:
Addressed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40854][table] Parse UTF-8 bytes directly in PARSE_JSON and TRY_PARSE_JSON [flink]
Posted by "raminqaf (via GitHub)" <gi...@apache.org>.
raminqaf commented on code in PR #29342:
URL: https://github.com/apache/flink/pull/29342#discussion_r4152584065
##########
flink-core/src/test/java/org/apache/flink/types/variant/BinaryVariantInternalBuilderTest.java:
##########
@@ -134,6 +141,37 @@ void testParseJsonWithNonAsciiStringsAndKeys() throws IOException {
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);
Review Comment:
No. `BinaryVariantInternalBuilder` is `@Internal`, and the public Variant and VariantBuilder APIs have no JSON parsing, so UDFs never reach this method. Its only callers are `PARSE_JSON`, which wraps any failure in a `TableRuntimeException`, and `TRY_PARSE_JSON`, which returns `NULL`. The `IOException` matches the existing `parseJson(String, boolean)` overload.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40854][table] Parse UTF-8 bytes directly in PARSE_JSON and TRY_PARSE_JSON [flink]
Posted by "twalthr (via GitHub)" <gi...@apache.org>.
twalthr merged PR #29342:
URL: https://github.com/apache/flink/pull/29342
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40854][table] Parse UTF-8 bytes directly in PARSE_JSON and TRY_PARSE_JSON [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #29342:
URL: https://github.com/apache/flink/pull/29342#issuecomment-5910519377
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0ba4933b046e840c6b580f78e77524255ba0f163",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "0ba4933b046e840c6b580f78e77524255ba0f163",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 0ba4933b046e840c6b580f78e77524255ba0f163 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40854][table] Parse UTF-8 bytes directly in PARSE_JSON and TRY_PARSE_JSON [flink]
Posted by "twalthr (via GitHub)" <gi...@apache.org>.
twalthr commented on code in PR #29342:
URL: https://github.com/apache/flink/pull/29342#discussion_r4145340995
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/JsonFunctionsITCase.java:
##########
@@ -1096,6 +1096,33 @@ private static List<TestSetSpec> parseJsonSpec() {
"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.
+ TestSetSpec.forFunction(
+ BuiltInFunctionDefinitions.PARSE_JSON,
+ "input bytes are always read as UTF-8")
+ .onFieldsWithData("1\u0000", "\u00001", "\uFEFF1")
+ .andDataTypes(STRING().notNull(), STRING().notNull(), STRING().notNull())
+ .testSqlRuntimeError(
+ "PARSE_JSON(f0)",
+ TableRuntimeException.class,
+ "Failed to parse json string")
+ .testSqlRuntimeError(
+ "PARSE_JSON(f1)",
+ TableRuntimeException.class,
+ "Failed to parse json string")
+ .testSqlRuntimeError(
+ "PARSE_JSON(f2)",
+ TableRuntimeException.class,
+ "Failed to parse json string"),
Review Comment:
What is the exact error message? This should be user friendly. Is there a hint to the charset?
##########
flink-core/src/test/java/org/apache/flink/types/variant/BinaryVariantInternalBuilderTest.java:
##########
@@ -134,6 +141,37 @@ void testParseJsonWithNonAsciiStringsAndKeys() throws IOException {
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);
Review Comment:
Will this be a VariantTypeException for users of Variant interface in UDFs?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40854][table] Parse UTF-8 bytes directly in PARSE_JSON and TRY_PARSE_JSON [flink]
Posted by "raminqaf (via GitHub)" <gi...@apache.org>.
raminqaf commented on code in PR #29342:
URL: https://github.com/apache/flink/pull/29342#discussion_r4152576629
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/JsonFunctionsITCase.java:
##########
@@ -1096,6 +1096,33 @@ private static List<TestSetSpec> parseJsonSpec() {
"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.
+ TestSetSpec.forFunction(
+ BuiltInFunctionDefinitions.PARSE_JSON,
+ "input bytes are always read as UTF-8")
+ .onFieldsWithData("1\u0000", "\u00001", "\uFEFF1")
+ .andDataTypes(STRING().notNull(), STRING().notNull(), STRING().notNull())
+ .testSqlRuntimeError(
+ "PARSE_JSON(f0)",
+ TableRuntimeException.class,
+ "Failed to parse json string")
+ .testSqlRuntimeError(
+ "PARSE_JSON(f1)",
+ TableRuntimeException.class,
+ "Failed to parse json string")
+ .testSqlRuntimeError(
+ "PARSE_JSON(f2)",
+ TableRuntimeException.class,
+ "Failed to parse json string"),
Review Comment:
When the input is not valid UTF-8, the error now names the charset and points at the byte: `Failed to parse json string: Invalid UTF-8 byte at index 1 of 3`. It uses the same wording as a strict `CAST(BYTES AS STRING)`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40854][table] Parse UTF-8 bytes directly in PARSE_JSON and TRY_PARSE_JSON [flink]
Posted by "twalthr (via GitHub)" <gi...@apache.org>.
twalthr commented on code in PR #29342:
URL: https://github.com/apache/flink/pull/29342#discussion_r4152984252
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/ParseJsonFunction.java:
##########
@@ -43,9 +44,19 @@ public ParseJsonFunction(SpecializedFunction.SpecializedContext context) {
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);
Review Comment:
```suggestion
String.format("Failed to parse JSON string: %s", jsonStr), e);
```
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/ParseJsonFunction.java:
##########
@@ -43,9 +44,19 @@ public ParseJsonFunction(SpecializedFunction.SpecializedContext context) {
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.",
Review Comment:
```suggestion
"Failed to parse JSON string: Invalid UTF-8 byte at index %d of %d.",
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org