[core] Introduce VariantType

[Zouxxyy]

Purpose

Subtask of #4471

• Add a new data type VariantType

/**
 * Data type of {@link Variant}.
 *
 * @since 1.1.0
 */
@Public
public class VariantType extends DataType {

    private static final String FORMAT = "VARIANT";
}

• Support reading and writing Variant into parquet with java api

FileFormat format = new ParquetFileFormat(new FileFormatFactory.FormatContext(new Options(), 1024, 1024));

RowType writeType = DataTypes.ROW(DataTypes.FIELD(0, "v", DataTypes.VARIANT()));

try (PositionOutputStream out = fileIO.newOutputStream(file, false);
        FormatWriter writer = format.createWriterFactory(writeType).create(out, "zstd")) {
    writer.addElement(
            GenericRow.of(GenericVariant.fromJson("{\"age\":35,\"city\":\"Chicago\"}")));
}

List<InternalRow> result = new ArrayList<>();
try (RecordReader<InternalRow> reader =  format.createReaderFactory(writeType) 
         .createReader( new FormatReaderContext(fileIO, file, fileIO.getFileSize(file)))) {
    InternalRowSerializer serializer = new InternalRowSerializer(writeType);
    reader.forEachRemaining(row -> result.add(serializer.copy(row)));
}

assertThat(result.get(0).getVariant(0).toJson()) .isEqualTo("{\"age\":35,\"city\":\"Chicago\"}");

Tests

API and Format

Documentation

[JingsongLi]

+1

[raphaelauv]

before hudi,delta,iceberg :)