[FLINK-40241][table-runtime] Support VARIANT as column type in the 'raw' format

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.
+

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.
+

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");

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) {

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<Object> serializer =

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<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",
+                        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<String, String> tableOptions =

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<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 @@ 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;
+    }
 }

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,13 +58,17 @@
 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;
 
 /** 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 +131,29 @@ static List<TestSpec> 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<TestSpec> 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();