This is an automated email from the ASF dual-hosted git repository. raminqaf pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit 2180171f1e2544f36f92cf694b05b6233f13ed15 Author: Ramin Gharib <[email protected]> AuthorDate: Tue Nov 3 11:44:38 2026 +0100 [FLINK-41305][table-runtime] Require UTF-8 for a VARIANT column in the raw format and parse it from bytes The raw format decoded a VARIANT message with 'raw.charset' into a String and parsed that text. Invalid UTF-8 became U+FFFD and was stored, while PARSE_JSON rejects the same bytes. A VARIANT column now requires 'raw.charset' to be UTF-8, which RFC 8259 requires for JSON exchanged between systems. Another charset fails with a ValidationException when a query reads or writes the table. The converter parses the message bytes the way PARSE_JSON does, so invalid UTF-8 now fails the record. JSON in another charset can still be read through a STRING column and PARSE_JSON. Generated-by: Claude Code 2.1.316 (Claude Opus 5.5) --- .../docs/connectors/table/formats/raw.md | 10 +++++--- docs/content/docs/connectors/table/formats/raw.md | 10 +++++--- .../raw/RawFormatDeserializationSchema.java | 18 ++++---------- .../apache/flink/formats/raw/RawFormatFactory.java | 20 +++++++++++++++ .../table/formats/raw/RawFormatFactoryTest.java | 18 ++++++++++++++ .../formats/raw/RawFormatLineDelimiterTest.java | 29 ++++++++++++++++++++++ .../formats/raw/RawFormatSerDeSchemaTest.java | 4 --- 7 files changed, 84 insertions(+), 25 deletions(-) diff --git a/docs/content.zh/docs/connectors/table/formats/raw.md b/docs/content.zh/docs/connectors/table/formats/raw.md index aa6ecfd6c0f..7bd20aba466 100644 --- a/docs/content.zh/docs/connectors/table/formats/raw.md +++ b/docs/content.zh/docs/connectors/table/formats/raw.md @@ -180,10 +180,12 @@ Format 参数 </tr> <tr> <td><code>VARIANT</code></td> - <td>A UTF-8 (by default) encoded JSON document.<br> - The encoding charset can be configured by 'raw.charset'.<br> - On read, the decoded text is parsed like <code>PARSE_JSON</code>, so duplicate object keys are rejected and - malformed JSON fails the job. On write, the value is rendered by <code>Variant#toJson</code>. The round trip is + <td>A UTF-8 encoded JSON document.<br> + 'raw.charset' must be UTF-8. To read or write JSON in another charset, declare the column as + <code>STRING</code> and convert it with <code>PARSE_JSON</code> when reading and <code>JSON_STRING</code> + when writing.<br> + On read, the bytes are parsed like <code>PARSE_JSON</code> with the same checks, so an invalid UTF-8 byte such as + <code>0xFF</code>, a duplicate object key or malformed JSON fails the job. On write, the value is rendered by <code>Variant#toJson</code>. The round trip is value-lossless but not byte-lossless: insignificant whitespace is dropped and object keys are ordered.</td> </tr> </tbody> diff --git a/docs/content/docs/connectors/table/formats/raw.md b/docs/content/docs/connectors/table/formats/raw.md index c7e44d25186..0886317085a 100644 --- a/docs/content/docs/connectors/table/formats/raw.md +++ b/docs/content/docs/connectors/table/formats/raw.md @@ -182,10 +182,12 @@ The table below details the SQL types the format supports, including details of </tr> <tr> <td><code>VARIANT</code></td> - <td>A UTF-8 (by default) encoded JSON document.<br> - The encoding charset can be configured by 'raw.charset'.<br> - On read, the decoded text is parsed like <code>PARSE_JSON</code>, so duplicate object keys are rejected and - malformed JSON fails the job. On write, the value is rendered by <code>Variant#toJson</code>. The round trip is + <td>A UTF-8 encoded JSON document.<br> + 'raw.charset' must be UTF-8. To read or write JSON in another charset, declare the column as + <code>STRING</code> and convert it with <code>PARSE_JSON</code> when reading and <code>JSON_STRING</code> + when writing.<br> + On read, the bytes are parsed like <code>PARSE_JSON</code> with the same checks, so an invalid UTF-8 byte such as + <code>0xFF</code>, a duplicate object key or malformed JSON fails the job. On write, the value is rendered by <code>Variant#toJson</code>. The round trip is value-lossless but not byte-lossless: insignificant whitespace is dropped and object keys are ordered.</td> </tr> </tbody> diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/formats/raw/RawFormatDeserializationSchema.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/formats/raw/RawFormatDeserializationSchema.java index 870d4d1a429..e6e894b5c61 100644 --- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/formats/raw/RawFormatDeserializationSchema.java +++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/formats/raw/RawFormatDeserializationSchema.java @@ -213,7 +213,7 @@ public class RawFormatDeserializationSchema implements DeserializationSchema<Row return RawValueData::fromBytes; case VARIANT: - return createVariantConverter(charsetName); + return createVariantConverter(); case BOOLEAN: return data -> data[0] != 0; @@ -284,25 +284,17 @@ public class RawFormatDeserializationSchema implements DeserializationSchema<Row } /** - * 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); - } @Override public Object convert(byte[] data) { try { - return BinaryVariantInternalBuilder.parseJson(new String(data, charset), false); + return BinaryVariantInternalBuilder.parseJson(data, false); } catch (Exception e) { throw new DeserializationException( "Failed to deserialize VARIANT type. " diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/formats/raw/RawFormatFactory.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/formats/raw/RawFormatFactory.java index 61adff04473..59fd275ba6c 100644 --- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/formats/raw/RawFormatFactory.java +++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/formats/raw/RawFormatFactory.java @@ -43,6 +43,7 @@ import org.apache.flink.table.types.logical.RowType; import org.apache.flink.shaded.guava33.com.google.common.collect.Sets; import java.nio.charset.Charset; +import java.nio.charset.StandardCharsets; import java.util.Collections; import java.util.HashSet; import java.util.Optional; @@ -92,6 +93,7 @@ public class RawFormatFactory implements DeserializationFormatFactory, Serializa DynamicTableSource.Context context, DataType producedDataType) { final RowType physicalRowType = (RowType) producedDataType.getLogicalType(); final LogicalType fieldType = validateAndExtractSingleField(physicalRowType); + checkVariantCharset(fieldType, charsetName); final TypeInformation<RowData> producedTypeInfo = context.createTypeInformation(producedDataType); return new RawFormatDeserializationSchema( @@ -124,6 +126,7 @@ public class RawFormatFactory implements DeserializationFormatFactory, Serializa DynamicTableSink.Context context, DataType consumedDataType) { final RowType physicalRowType = (RowType) consumedDataType.getLogicalType(); final LogicalType fieldType = validateAndExtractSingleField(physicalRowType); + checkVariantCharset(fieldType, charsetName); return new RawFormatSerializationSchema( fieldType, charsetName, isBigEndian, lineDelimiter.orElse(null)); } @@ -163,6 +166,23 @@ public class RawFormatFactory implements DeserializationFormatFactory, Serializa } } + /** + * Checks that a VARIANT column uses UTF-8, which RFC 8259 requires for JSON exchanged between + * systems and which {@code PARSE_JSON} reads. + */ + private static void checkVariantCharset(LogicalType fieldType, String charsetName) { + if (fieldType.is(LogicalTypeRoot.VARIANT) + && !StandardCharsets.UTF_8.equals(Charset.forName(charsetName))) { + throw new ValidationException( + String.format( + "The 'raw' format requires '%s.%s' to be UTF-8 for a VARIANT column, " + + "but it is '%s'. To use another charset, declare the column " + + "as STRING and convert it with PARSE_JSON when reading and " + + "JSON_STRING when writing.", + IDENTIFIER, RawFormatOptions.CHARSET.key(), charsetName)); + } + } + /** Validates and extract the single field type from the given physical row schema. */ private static LogicalType validateAndExtractSingleField(RowType physicalRowType) { if (physicalRowType.getFieldCount() != 1) { diff --git a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/formats/raw/RawFormatFactoryTest.java b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/formats/raw/RawFormatFactoryTest.java index c2e210afd67..11bf1b74d52 100644 --- a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/formats/raw/RawFormatFactoryTest.java +++ b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/formats/raw/RawFormatFactoryTest.java @@ -199,6 +199,24 @@ class RawFormatFactoryTest { assertThat(createSerializationSchema(variantSchema, tableOptions)).isEqualTo(expectedSer); } + @Test + void testVariantRequiresUtf8Charset() { + final ResolvedSchema variantSchema = + ResolvedSchema.of(Column.physical("field1", DataTypes.VARIANT())); + final Map<String, String> tableOptions = + getModifiedOptions(options -> options.put("raw.charset", UTF_16.name())); + final String expectedError = + "The 'raw' format requires 'raw.charset' to be UTF-8 for a VARIANT column, " + + "but it is 'UTF-16'. To use another charset, declare the column " + + "as STRING and convert it with PARSE_JSON when reading and " + + "JSON_STRING when writing."; + + assertThatThrownBy(() -> createDeserializationSchema(variantSchema, tableOptions)) + .hasMessage(expectedError); + assertThatThrownBy(() -> createSerializationSchema(variantSchema, tableOptions)) + .hasMessage(expectedError); + } + @Test void testLineDelimiterOption() { final Map<String, String> tableOptions = diff --git a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/formats/raw/RawFormatLineDelimiterTest.java b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/formats/raw/RawFormatLineDelimiterTest.java index 122f1f1d1d2..e12a28f958d 100644 --- a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/formats/raw/RawFormatLineDelimiterTest.java +++ b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/formats/raw/RawFormatLineDelimiterTest.java @@ -29,10 +29,13 @@ import org.apache.flink.table.data.RowData; import org.apache.flink.table.data.StringData; import org.apache.flink.table.types.logical.VarCharType; import org.apache.flink.table.types.logical.VariantType; +import org.apache.flink.types.DeserializationException; import org.apache.flink.types.variant.BinaryVariantInternalBuilder; import org.apache.flink.util.Collector; import org.apache.flink.util.UserCodeClassLoader; +import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonParseException; + import org.junit.jupiter.api.Test; import java.io.IOException; @@ -43,6 +46,7 @@ import java.util.Arrays; import java.util.List; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; /** * Tests for {@link RawFormatDeserializationSchema} and {@link RawFormatSerializationSchema} with @@ -274,6 +278,31 @@ class RawFormatLineDelimiterTest { assertThat(rows.get(1).getVariant(0).getField("id").getByte()).isEqualTo((byte) 2); } + @Test + void testDeserializeVariantRejectsInvalidUtf8() throws Exception { + RawFormatDeserializationSchema deser = + new RawFormatDeserializationSchema( + VARIANT_TYPE, + TypeInformation.of(RowData.class), + StandardCharsets.UTF_8.name(), + true, + null); + openDeser(deser); + + byte[] message = {'{', '"', 'a', '"', ':', '"', (byte) 0xFF, '"', '}'}; + assertThatThrownBy(() -> collectRows(deser, message)) + .isInstanceOf(DeserializationException.class) + .hasMessage( + "Failed to deserialize VARIANT type. " + + "The received data is not a valid JSON document.") + .cause() + .isInstanceOfSatisfying( + JsonParseException.class, + e -> + assertThat(e.getOriginalMessage()) + .isEqualTo("Invalid UTF-8 start byte 0xff")); + } + // ----------------------------------------------------------------------- // Helpers // ----------------------------------------------------------------------- diff --git a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/formats/raw/RawFormatSerDeSchemaTest.java b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/formats/raw/RawFormatSerDeSchemaTest.java index 83d0dc328d6..1f38c570af4 100644 --- a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/formats/raw/RawFormatSerDeSchemaTest.java +++ b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/formats/raw/RawFormatSerDeSchemaTest.java @@ -150,10 +150,6 @@ class RawFormatSerDeSchemaTest { "{\"id\":1}".getBytes(), "{\"id\":2}".getBytes(), "{\"id\":3}".getBytes()), - TestSpec.type(VARIANT()) - .values(variant("{\"greeting\":\"你好世界\"}")) - .withCharset("UTF-16") - .binary("{\"greeting\":\"你好世界\"}".getBytes(StandardCharsets.UTF_16)), // test nulls TestSpec.type(TINYINT()).values((Object) null).binary((byte[]) null),
