raminqaf commented on code in PR #29010:
URL: https://github.com/apache/flink/pull/29010#discussion_r3880111277
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/formats/raw/RawFormatDeserializationSchema.java:
##########
@@ -278,6 +283,39 @@ public Object convert(byte[] data) {
};
}
+ /**
+ * Creates a converter that decodes the bytes with the configured charset
and parses the text as
+ * a JSON document into a {@link Variant}. Duplicate object keys are
rejected, matching the
+ * default of {@code PARSE_JSON}.
+ */
+ private static DeserializationRuntimeConverter createVariantConverter(
+ final String charsetName) {
+ // this also checks the charsetName is valid
+ Charset.forName(charsetName);
Review Comment:
The charset can't be resolved once up front and reused, because `Charset`
isn't `Serializable` and this schema is serialized and shipped to the
TaskManagers. If the converter holds a resolved `Charset`, serializing the
schema fails with `NotSerializableException: sun.nio.cs.UTF_8`. I have varified
this with the updated `testSerializationAndDeserialization`. The test now
clones the schema via `InstantiationUtil.clone, the way Flink does at deploy
time, so this fails fast if a converter ever captures a non-serializable field
again.
Only the `String charsetName` survives serialization, so the `Charset` has
to be rebuilt on the task side, which is what `open()` does into a `transient`
field.
That call runs once per subtask, not per record, and `Charset.forName` is
cached, so it isn't on the hot path; every record reuses the resolved
`charset`. This is the same pattern the existing `createStringConverter` in
this file already uses.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
