This is an automated email from the ASF dual-hosted git repository.
snuyanzin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new cc633c644e5 [FLINK-40241][table-runtime] Support `VARIANT` as column
type in the 'raw' format
cc633c644e5 is described below
commit cc633c644e510eac74b06ab470e467b43a1987b2
Author: Ramin Gharib <[email protected]>
AuthorDate: Fri Aug 28 16:44:53 2026 +0200
[FLINK-40241][table-runtime] Support `VARIANT` as column type in the 'raw'
format
The 'raw' format now accepts a single VARIANT column, treating the bytes as
a JSON document. On read the bytes are decoded with 'raw.charset' and parsed
like PARSE_JSON with duplicate keys rejected; on write the value is rendered
with Variant#toJson and encoded with the same charset. The round trip is
value-lossless but not byte-lossless: insignificant whitespace is dropped and
object keys are ordered. 'raw.endianness' does not apply to VARIANT.
Combined with 'raw.line-delimiter' this reads and writes newline-delimited
JSON, one row per line.
---
.../docs/connectors/table/formats/raw.md | 11 ++++
docs/content/docs/connectors/table/formats/raw.md | 11 ++++
.../raw/RawFormatDeserializationSchema.java | 36 +++++++++++++
.../apache/flink/formats/raw/RawFormatFactory.java | 3 +-
.../formats/raw/RawFormatSerializationSchema.java | 28 +++++++++-
.../table/formats/raw/RawFormatFactoryTest.java | 42 ++++++++++++---
.../formats/raw/RawFormatLineDelimiterTest.java | 47 +++++++++++++++++
.../formats/raw/RawFormatSerDeSchemaTest.java | 61 +++++++++++++++++++---
8 files changed, 222 insertions(+), 17 deletions(-)
diff --git a/docs/content.zh/docs/connectors/table/formats/raw.md
b/docs/content.zh/docs/connectors/table/formats/raw.md
index 69b4acfa1a0..cb1f2aea4f6 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 参数
<td><code>RAW</code></td>
<td>通过 RAW 类型的底层 TypeSerializer 序列化的字节序列。</td>
</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
+ value-lossless but not byte-lossless: insignificant whitespace is
dropped and object keys are ordered.</td>
+ </tr>
</tbody>
</table>
+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 1efdd2b40ee..3e2a2c60e8b 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
<td><code>RAW</code></td>
<td>The sequence of bytes serialized by the underlying TypeSerializer of
the RAW type.</td>
</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
+ value-lossless but not byte-lossless: insignificant whitespace is
dropped and object keys are ordered.</td>
+ </tr>
</tbody>
</table>
+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 4ac4f82f464..870d4d1a429 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.RowData;
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 @@ public class RawFormatDeserializationSchema implements
DeserializationSchema<Row
case RAW:
return RawValueData::fromBytes;
+ case VARIANT:
+ return createVariantConverter(charsetName);
+
case BOOLEAN:
return data -> data[0] != 0;
@@ -278,6 +283,36 @@ 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}.
+ */
+ private static DeserializationRuntimeConverter createVariantConverter(
+ final String 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 +337,7 @@ public class RawFormatDeserializationSchema implements
DeserializationSchema<Row
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 05b52cf2c94..61adff04473 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 class RawFormatFactory implements
DeserializationFormatFactory, Serializa
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 2f16dee5caa..1661b154bbd 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.core.memory.MemorySegmentFactory;
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 class RawFormatSerializationSchema implements
SerializationSchema<RowData
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 @@ public class RawFormatSerializationSchema implements
SerializationSchema<RowData
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,27 @@ public class RawFormatSerializationSchema implements
SerializationSchema<RowData
};
}
+ /**
+ * 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) {
+ 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<Object> 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 17ad29f9225..c2e210afd67 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
@@ -41,6 +41,8 @@ import java.util.HashMap;
import java.util.Map;
import java.util.function.Consumer;
+import static java.nio.charset.StandardCharsets.UTF_16;
+import static java.nio.charset.StandardCharsets.UTF_8;
import static org.apache.flink.core.testutils.FlinkAssertions.anyCauseMatches;
import static
org.apache.flink.table.factories.utils.FactoryMocks.createTableSink;
import static
org.apache.flink.table.factories.utils.FactoryMocks.createTableSource;
@@ -63,14 +65,14 @@ class RawFormatFactoryTest {
// test deserialization
final RawFormatDeserializationSchema expectedDeser =
new RawFormatDeserializationSchema(
- ROW_TYPE.getTypeAt(0), InternalTypeInfo.of(ROW_TYPE),
"UTF-8", true);
+ ROW_TYPE.getTypeAt(0), InternalTypeInfo.of(ROW_TYPE),
UTF_8.name(), true);
DeserializationSchema<RowData> actualDeser =
createDeserializationSchema(SCHEMA, tableOptions);
assertThat(actualDeser).isEqualTo(expectedDeser);
// test serialization
final RawFormatSerializationSchema expectedSer =
- new RawFormatSerializationSchema(ROW_TYPE.getTypeAt(0),
"UTF-8", true);
+ new RawFormatSerializationSchema(ROW_TYPE.getTypeAt(0),
UTF_8.name(), true);
SerializationSchema<RowData> actualSer =
createSerializationSchema(SCHEMA, tableOptions);
assertThat(actualSer).isEqualTo(expectedSer);
}
@@ -80,21 +82,21 @@ class RawFormatFactoryTest {
final Map<String, String> tableOptions =
getModifiedOptions(
options -> {
- options.put("raw.charset", "UTF-16");
+ options.put("raw.charset", UTF_16.name());
options.put("raw.endianness", "little-endian");
});
// test deserialization
final RawFormatDeserializationSchema expectedDeser =
new RawFormatDeserializationSchema(
- ROW_TYPE.getTypeAt(0), InternalTypeInfo.of(ROW_TYPE),
"UTF-16", false);
+ ROW_TYPE.getTypeAt(0), InternalTypeInfo.of(ROW_TYPE),
UTF_16.name(), false);
DeserializationSchema<RowData> actualDeser =
createDeserializationSchema(SCHEMA, tableOptions);
assertThat(actualDeser).isEqualTo(expectedDeser);
// test serialization
final RawFormatSerializationSchema expectedSer =
- new RawFormatSerializationSchema(ROW_TYPE.getTypeAt(0),
"UTF-16", false);
+ new RawFormatSerializationSchema(ROW_TYPE.getTypeAt(0),
UTF_16.name(), false);
SerializationSchema<RowData> actualSer =
createSerializationSchema(SCHEMA, tableOptions);
assertThat(actualSer).isEqualTo(expectedSer);
}
@@ -175,6 +177,28 @@ class RawFormatFactoryTest {
.hasMessage("The 'raw' format doesn't supports 'MAP<INT,
STRING>' 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<String, String> tableOptions = getBasicOptions();
+
+ final RawFormatDeserializationSchema expectedDeser =
+ new RawFormatDeserializationSchema(
+ variantRowType.getTypeAt(0),
+ InternalTypeInfo.of(variantRowType),
+ UTF_8.name(),
+ true);
+ assertThat(createDeserializationSchema(variantSchema, tableOptions))
+ .isEqualTo(expectedDeser);
+
+ final RawFormatSerializationSchema expectedSer =
+ new RawFormatSerializationSchema(variantRowType.getTypeAt(0),
UTF_8.name(), true);
+ assertThat(createSerializationSchema(variantSchema,
tableOptions)).isEqualTo(expectedSer);
+ }
+
@Test
void testLineDelimiterOption() {
final Map<String, String> tableOptions =
@@ -186,14 +210,18 @@ class RawFormatFactoryTest {
// test deserialization schema contains line delimiter
final RawFormatDeserializationSchema expectedDeser =
new RawFormatDeserializationSchema(
- ROW_TYPE.getTypeAt(0), InternalTypeInfo.of(ROW_TYPE),
"UTF-8", true, "\n");
+ ROW_TYPE.getTypeAt(0),
+ InternalTypeInfo.of(ROW_TYPE),
+ UTF_8.name(),
+ true,
+ "\n");
DeserializationSchema<RowData> actualDeser =
createDeserializationSchema(SCHEMA, tableOptions);
assertThat(actualDeser).isEqualTo(expectedDeser);
// test serialization schema contains line delimiter
final RawFormatSerializationSchema expectedSer =
- new RawFormatSerializationSchema(ROW_TYPE.getTypeAt(0),
"UTF-8", true, "\n");
+ new RawFormatSerializationSchema(ROW_TYPE.getTypeAt(0),
UTF_8.name(), true, "\n");
SerializationSchema<RowData> actualSer =
createSerializationSchema(SCHEMA, tableOptions);
assertThat(actualSer).isEqualTo(expectedSer);
}
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 52652121698..122f1f1d1d2 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.GenericRowData;
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 @@ class RawFormatLineDelimiterTest {
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<RowData> 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 @@ class RawFormatLineDelimiterTest {
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 a4d7a6f67bb..83d0dc328d6 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,9 @@ import
org.apache.flink.table.data.conversion.DataStructureConverter;
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.InstantiationUtil;
import org.apache.flink.util.StringUtils;
import org.junit.jupiter.params.ParameterizedTest;
@@ -56,6 +59,7 @@ import static org.apache.flink.table.api.DataTypes.SMALLINT;
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 +67,9 @@ import static org.mockito.Mockito.mock;
/** 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<TestSpec> testData() {
return Arrays.asList(
TestSpec.type(TINYINT()).values(Byte.MAX_VALUE).binary(new
byte[] {Byte.MAX_VALUE}),
@@ -125,6 +132,29 @@ class RawFormatSerDeSchemaTest {
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,21 +167,28 @@ class RawFormatSerDeSchemaTest {
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
@MethodSource("testData")
void testSerializationAndDeserialization(final TestSpec testSpec) throws
Exception {
+ // Clone through serialization, as Flink does when shipping the schema
to a task manager.
+ // This also guards that the schema stays serializable for every
supported type.
RawFormatDeserializationSchema deserializationSchema =
- new RawFormatDeserializationSchema(
- testSpec.type.getLogicalType(),
- TypeInformation.of(RowData.class),
- testSpec.charsetName,
- testSpec.isBigEndian);
+ InstantiationUtil.clone(
+ new RawFormatDeserializationSchema(
+ testSpec.type.getLogicalType(),
+ TypeInformation.of(RowData.class),
+ testSpec.charsetName,
+ testSpec.isBigEndian));
RawFormatSerializationSchema serializationSchema =
- new RawFormatSerializationSchema(
- testSpec.type.getLogicalType(), testSpec.charsetName,
testSpec.isBigEndian);
+ InstantiationUtil.clone(
+ new RawFormatSerializationSchema(
+ testSpec.type.getLogicalType(),
+ testSpec.charsetName,
+ testSpec.isBigEndian));
deserializationSchema.open(mock(DeserializationSchema.InitializationContext.class));
serializationSchema.open(mock(SerializationSchema.InitializationContext.class));
@@ -186,6 +223,14 @@ class RawFormatSerDeSchemaTest {
}
}
+ 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();
