diff --git a/docs/content.zh/docs/connectors/table/formats/raw.md b/docs/content.zh/docs/connectors/table/formats/raw.md index 69b4acfa1a0f49..cb1f2aea4f6bb3 100644 --- a/docs/content.zh/docs/connectors/table/formats/raw.md +++ b/docs/content.zh/docs/connectors/table/formats/raw.md @@ -178,6 +178,17 @@ Format 参数 RAW 通过 RAW 类型的底层 TypeSerializer 序列化的字节序列。 + + VARIANT + A UTF-8 (by default) encoded JSON document.
+ The encoding charset can be configured by 'raw.charset'.
+ On read, the decoded text is parsed like PARSE_JSON, so duplicate object keys are rejected and + malformed JSON fails the job. On write, the value is rendered by Variant#toJson. The round trip is + value-lossless but not byte-lossless: insignificant whitespace is dropped and object keys are ordered. + +Note: combining `VARIANT` with `raw.line-delimiter` gives you newline-delimited JSON, where each line of a message +becomes one row. + diff --git a/docs/content/docs/connectors/table/formats/raw.md b/docs/content/docs/connectors/table/formats/raw.md index 1efdd2b40ee4a4..3e2a2c60e8bde0 100644 --- a/docs/content/docs/connectors/table/formats/raw.md +++ b/docs/content/docs/connectors/table/formats/raw.md @@ -180,6 +180,17 @@ The table below details the SQL types the format supports, including details of RAW The sequence of bytes serialized by the underlying TypeSerializer of the RAW type. + + VARIANT + A UTF-8 (by default) encoded JSON document.
+ The encoding charset can be configured by 'raw.charset'.
+ On read, the decoded text is parsed like PARSE_JSON, so duplicate object keys are rejected and + malformed JSON fails the job. On write, the value is rendered by Variant#toJson. The round trip is + value-lossless but not byte-lossless: insignificant whitespace is dropped and object keys are ordered. + +Note: combining `VARIANT` with `raw.line-delimiter` gives you newline-delimited JSON, where each line of a message +becomes one row. + 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 4ac4f82f464475..10558ebad1ffc9 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 @@ -29,6 +29,8 @@ import org.apache.flink.table.data.StringData; import org.apache.flink.table.types.logical.LogicalType; import org.apache.flink.types.DeserializationException; +import org.apache.flink.types.variant.BinaryVariantInternalBuilder; +import org.apache.flink.types.variant.Variant; import org.apache.flink.util.Collector; import javax.annotation.Nullable; @@ -210,6 +212,9 @@ private static DeserializationRuntimeConverter createConverter( case RAW: return RawValueData::fromBytes; + case VARIANT: + return createVariantConverter(charsetName); + case BOOLEAN: return data -> data[0] != 0; @@ -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); + + 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); + } catch (Exception e) { + throw new DeserializationException( + "Failed to deserialize VARIANT type. " + + "The received data is not a valid JSON document.", + e); + } + } + }; + } + private static DeserializationRuntimeConverter createEndiannessAwareConverter( final boolean isBigEndian, final MemorySegmentConverter bigEndianConverter, @@ -302,6 +340,7 @@ private static DataLengthValidator createDataLengthValidator(LogicalType type) { case VARBINARY: case BINARY: case RAW: + case VARIANT: return data -> {}; case BOOLEAN: return createDataLengthValidator(1, "BOOLEAN"); 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 05b52cf2c94e4e..61adff04473ff7 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 @@ -150,7 +150,8 @@ public ChangelogMode getChangelogMode() { LogicalTypeRoot.INTEGER, LogicalTypeRoot.BIGINT, LogicalTypeRoot.FLOAT, - LogicalTypeRoot.DOUBLE); + LogicalTypeRoot.DOUBLE, + LogicalTypeRoot.VARIANT); /** Checks the given field type is supported. */ private static void checkFieldType(LogicalType fieldType) { diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/formats/raw/RawFormatSerializationSchema.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/formats/raw/RawFormatSerializationSchema.java index 2f16dee5caa210..bd27d4c2ad3948 100644 --- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/formats/raw/RawFormatSerializationSchema.java +++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/formats/raw/RawFormatSerializationSchema.java @@ -26,6 +26,8 @@ import org.apache.flink.table.data.RowData; import org.apache.flink.table.types.logical.LogicalType; import org.apache.flink.table.types.logical.RawType; +import org.apache.flink.types.variant.Variant; +import org.apache.flink.types.variant.VariantTypeException; import javax.annotation.Nullable; @@ -89,7 +91,7 @@ public byte[] serialize(RowData row) { byte[] result = Arrays.copyOf(valueBytes, valueBytes.length + delimiterBytes.length); System.arraycopy(delimiterBytes, 0, result, valueBytes.length, delimiterBytes.length); return result; - } catch (IOException e) { + } catch (IOException | VariantTypeException e) { throw new RuntimeException("Could not serialize row '" + row + "'. ", e); } } @@ -164,6 +166,9 @@ private SerializationRuntimeConverter createNotNullConverter( case RAW: return createRawValueConverter((RawType) type); + case VARIANT: + return createVariantConverter(charsetName); + case BOOLEAN: return row -> { byte b = (byte) (row.getBoolean(0) ? 1 : 0); @@ -220,6 +225,30 @@ public byte[] convert(RowData row) { }; } + /** + * Creates a converter that renders the {@link Variant} as a JSON document. The result is + * value-lossless but not byte-lossless: whitespace and object key order are normalized. + */ + private static SerializationRuntimeConverter createVariantConverter(final String charsetName) { + // this also checks the charsetName is valid + Charset.forName(charsetName); + + return new SerializationRuntimeConverter() { + private static final long serialVersionUID = 1L; + private transient Charset charset; + + @Override + public void open() { + charset = Charset.forName(charsetName); + } + + @Override + public byte[] convert(RowData row) { + return row.getVariant(0).toJson().getBytes(charset); + } + }; + } + @SuppressWarnings("unchecked") private static SerializationRuntimeConverter createRawValueConverter(RawType rawType) { final TypeSerializer serializer = 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 17ad29f92254b2..93a42c2a7090b8 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 @@ -175,6 +175,28 @@ void testInvalidFieldTypes() { .hasMessage("The 'raw' format doesn't supports 'MAP' as column type."); } + @Test + void testVariantSeDeSchema() { + final ResolvedSchema variantSchema = + ResolvedSchema.of(Column.physical("field1", DataTypes.VARIANT())); + final RowType variantRowType = + (RowType) variantSchema.toPhysicalRowDataType().getLogicalType(); + final Map tableOptions = getBasicOptions(); + + final RawFormatDeserializationSchema expectedDeser = + new RawFormatDeserializationSchema( + variantRowType.getTypeAt(0), + InternalTypeInfo.of(variantRowType), + "UTF-8", + true); + assertThat(createDeserializationSchema(variantSchema, tableOptions)) + .isEqualTo(expectedDeser); + + final RawFormatSerializationSchema expectedSer = + new RawFormatSerializationSchema(variantRowType.getTypeAt(0), "UTF-8", true); + assertThat(createSerializationSchema(variantSchema, tableOptions)).isEqualTo(expectedSer); + } + @Test void testLineDelimiterOption() { final Map 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 52652121698272..122f1f1d1d2805 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 @@ -28,14 +28,18 @@ 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.variant.BinaryVariantInternalBuilder; import org.apache.flink.util.Collector; import org.apache.flink.util.UserCodeClassLoader; import org.junit.jupiter.api.Test; +import java.io.IOException; import java.nio.charset.Charset; import java.nio.charset.StandardCharsets; import java.util.ArrayList; +import java.util.Arrays; import java.util.List; import static org.assertj.core.api.Assertions.assertThat; @@ -48,6 +52,8 @@ class RawFormatLineDelimiterTest { private static final VarCharType STRING_TYPE = VarCharType.STRING_TYPE; + private static final VariantType VARIANT_TYPE = new VariantType(); + // ----------------------------------------------------------------------- // Deserialization tests // ----------------------------------------------------------------------- @@ -239,10 +245,45 @@ void testRoundTrip_serializeThenDeserialize() throws Exception { assertThat(rows.get(0).getString(0)).hasToString("hello"); } + @Test + void testRoundTripNewlineDelimitedJsonAsVariant() throws Exception { + RawFormatSerializationSchema ser = + new RawFormatSerializationSchema( + VARIANT_TYPE, StandardCharsets.UTF_8.name(), true, "\n"); + openSer(ser); + + RawFormatDeserializationSchema deser = + new RawFormatDeserializationSchema( + VARIANT_TYPE, + TypeInformation.of(RowData.class), + StandardCharsets.UTF_8.name(), + true, + "\n"); + openDeser(deser); + + byte[] stream = + concat( + ser.serialize(buildVariantRow("{\"id\":1}")), + ser.serialize(buildVariantRow("{\"id\":2}"))); + assertThat(new String(stream, StandardCharsets.UTF_8)) + .isEqualTo("{\"id\":1}\n{\"id\":2}\n"); + + List rows = collectRows(deser, stream); + assertThat(rows).hasSize(2); + assertThat(rows.get(0).getVariant(0).getField("id").getByte()).isEqualTo((byte) 1); + assertThat(rows.get(1).getVariant(0).getField("id").getByte()).isEqualTo((byte) 2); + } + // ----------------------------------------------------------------------- // Helpers // ----------------------------------------------------------------------- + private static byte[] concat(byte[] first, byte[] second) { + byte[] result = Arrays.copyOf(first, first.length + second.length); + System.arraycopy(second, 0, result, first.length, second.length); + return result; + } + private void openDeser(RawFormatDeserializationSchema schema) throws Exception { schema.open( new DeserializationSchema.InitializationContext() { @@ -295,4 +336,10 @@ private RowData buildStringRow(String value) { row.setField(0, StringData.fromString(value)); return row; } + + private RowData buildVariantRow(String json) throws IOException { + GenericRowData row = new GenericRowData(1); + row.setField(0, BinaryVariantInternalBuilder.parseJson(json, false)); + return row; + } } 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 a4d7a6f67bbfbd..2edf54614b2adf 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 @@ -30,6 +30,8 @@ import org.apache.flink.table.data.conversion.DataStructureConverters; import org.apache.flink.table.types.DataType; import org.apache.flink.types.Row; +import org.apache.flink.types.variant.BinaryVariantInternalBuilder; +import org.apache.flink.types.variant.Variant; import org.apache.flink.util.StringUtils; import org.junit.jupiter.params.ParameterizedTest; @@ -56,6 +58,7 @@ import static org.apache.flink.table.api.DataTypes.STRING; import static org.apache.flink.table.api.DataTypes.TINYINT; import static org.apache.flink.table.api.DataTypes.VARCHAR; +import static org.apache.flink.table.api.DataTypes.VARIANT; import static org.apache.flink.util.StringUtils.hexStringToByte; import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.Mockito.mock; @@ -63,6 +66,9 @@ /** Tests for {@link RawFormatDeserializationSchema} {@link RawFormatSerializationSchema}. */ class RawFormatSerDeSchemaTest { + private static final String JSON_OBJECT = + "{\"a\":1,\"b\":\"x\",\"c\":[1,2,3],\"d\":null,\"e\":true}"; + static List testData() { return Arrays.asList( TestSpec.type(TINYINT()).values(Byte.MAX_VALUE).binary(new byte[] {Byte.MAX_VALUE}), @@ -125,6 +131,29 @@ static List testData() { serializeLocalDateTime( LocalDateTime.parse("2020-11-11T18:08:01.123"))), + // test variants, which are represented as JSON documents + TestSpec.type(VARIANT()) + .values(variant(JSON_OBJECT)) + .binary(JSON_OBJECT.getBytes(StandardCharsets.UTF_8)), + TestSpec.type(VARIANT()).values(variant("[1,2,3]")).binary("[1,2,3]".getBytes()), + TestSpec.type(VARIANT()) + .values(variant("\"hello\"")) + .binary("\"hello\"".getBytes()), + TestSpec.type(VARIANT()).values(variant("42")).binary("42".getBytes()), + TestSpec.type(VARIANT()).values(variant("3.5")).binary("3.5".getBytes()), + TestSpec.type(VARIANT()).values(variant("true")).binary("true".getBytes()), + TestSpec.type(VARIANT()).values(variant("null")).binary("null".getBytes()), + TestSpec.type(VARIANT()) + .values(variant("{\"id\":1}"), variant("{\"id\":2}"), variant("{\"id\":3}")) + .binary( + "{\"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), TestSpec.type(SMALLINT()).values((Object) null).binary((byte[]) null), @@ -137,7 +166,8 @@ static List testData() { TestSpec.type(BYTES()).values((Object) null).binary((byte[]) null), TestSpec.type(RAW(LocalDateTime.class, new LocalDateTimeSerializer())) .values((Object) null) - .binary((byte[]) null)); + .binary((byte[]) null), + TestSpec.type(VARIANT()).values((Object) null).binary((byte[]) null)); } @ParameterizedTest @@ -186,6 +216,14 @@ void testSerializationAndDeserialization(final TestSpec testSpec) throws Excepti } } + private static Variant variant(String json) { + try { + return BinaryVariantInternalBuilder.parseJson(json, false); + } catch (IOException e) { + throw new RuntimeException(e); + } + } + private static byte[] serializeLocalDateTime(LocalDateTime localDateTime) { DataOutputSerializer dos = new DataOutputSerializer(16); LocalDateTimeSerializer serializer = new LocalDateTimeSerializer();