raminqaf commented on code in PR #29760:
URL: https://github.com/apache/flink/pull/29760#discussion_r4467049359
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/formats/raw/RawFormatDeserializationSchema.java:
##########
@@ -284,29 +369,47 @@ 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}.
+ * Creates a converter that parses UTF-8 JSON bytes into a {@link Variant}
like {@code
+ * PARSE_JSON}, so invalid UTF-8 and duplicate object keys are rejected.
*/
- private static DeserializationRuntimeConverter createVariantConverter(
- final String charsetName) {
+ private static DeserializationRuntimeConverter createVariantConverter() {
return new DeserializationRuntimeConverter() {
private static final long serialVersionUID = 1L;
- private transient Charset charset;
@Override
- public void open() {
- charset = Charset.forName(charsetName);
+ public Object convert(byte[] data) {
+ try {
+ return BinaryVariantInternalBuilder.parseJson(data, false);
+ } catch (Exception e) {
+ throw new DeserializationException(
+ "Failed to deserialize VARIANT type. "
+ + "The received data is not a valid JSON
document.",
+ e);
+ }
}
+ };
+ }
+
+ /**
+ * Creates a converter that reads the Variant binary encoding, the
metadata followed by the
+ * value. The bytes are only checked as far as needed to split them. A
malformed value fails
+ * when it is read.
+ */
+ private static DeserializationRuntimeConverter
createBinaryVariantConverter() {
+ return new DeserializationRuntimeConverter() {
+ private static final long serialVersionUID = 1L;
@Override
public Object convert(byte[] data) {
try {
- return BinaryVariantInternalBuilder.parseJson(new
String(data, charset), false);
+ final int metadataSize =
BinaryVariantUtil.metadataSize(data);
+ return new BinaryVariant(
+ Arrays.copyOfRange(data, metadataSize,
data.length),
+ Arrays.copyOf(data, metadataSize));
} catch (Exception e) {
throw new DeserializationException(
"Failed to deserialize VARIANT type. "
- + "The received data is not a valid JSON
document.",
+ + "The received data is not a valid
Variant binary encoding.",
Review Comment:
done
--
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]
