package org.apache.flink.types.variant;  

/** Variant represent a semi-structured data. */
public interface Variant extends Serializable {

    /** Returns true if the variant is a primitive typed value, such as INT, DOUBLE, STRING, etc. */
    boolean isPrimitive();

    /** Returns true if this variant is an Array, false otherwise. */
    boolean isArray();

    /** Returns true if this variant is an Object, false otherwise. */
    boolean isObject();

    /** Check If this variant is null. */
    boolean isNull();

    /** Get the type of variant. */
    Type getType();

    /**
     * Get the scalar value of variant as boolean, if the variant type is {@link Type#BOOLEAN}.
     *
     * @throws VariantTypeException If this variant is not a scalar value or is not {@link
     *     Type#BOOLEAN}.
     */
    boolean getBoolean() throws VariantTypeException;

    /**
     * Get the scalar value of variant as byte, if the variant type is {@link Type#TINYINT}.
     *
     * @throws VariantTypeException If this variant is not a scalar value or is not {@link
     *     Type#TINYINT}.
     */
    byte getByte() throws VariantTypeException;

    /**
     * Get the scalar value of variant as short, if the variant type is {@link Type#SMALLINT}.
     *
     * @throws VariantTypeException If this variant is not a scalar value or is not {@link
     *     Type#SMALLINT}.
     */
    short getShort() throws VariantTypeException;

    /**
     * Get the scalar value of variant as int, if the variant type is {@link Type#INT}.
     *
     * @throws VariantTypeException if this variant is not a scalar value or is not {@link
     *     Type#INT}.
     */
    int getInt() throws VariantTypeException;

    /**
     * Get the scalar value of variant as long, if the variant type is {@link Type#BIGINT}.
     *
     * @throws VariantTypeException If this variant is not a scalar value or is not {@link
     *     Type#BIGINT}.
     */
    long getLong() throws VariantTypeException;

    /**
     * Get the scalar value of variant as float, if the variant type is {@link Type#FLOAT}.
     *
     * @throws VariantTypeException If this variant is not a scalar value or is not {@link
     *     Type#FLOAT}.
     */
    float getFloat() throws VariantTypeException;

    /**
     * Get the scalar value of variant as BigDecimal, if the variant type is {@link Type#DECIMAL}.
     *
     * @throws VariantTypeException If this variant is not a scalar value or is not {@link
     *     Type#DECIMAL}.
     */
    BigDecimal getDecimal() throws VariantTypeException;

    /**
     * Get the scalar value of variant as double, if the variant type is {@link Type#DOUBLE}.
     *
     * @throws VariantTypeException If this variant is not a scalar value or is not {@link
     *     Type#DOUBLE}.
     */
    double getDouble() throws VariantTypeException;

    /**
     * Get the scalar value of variant as string, if the variant type is {@link Type#STRING}.
     *
     * @throws VariantTypeException If this variant is not a scalar value or is not {@link
     *     Type#STRING}.
     */
    String getString() throws VariantTypeException;

    /**
     * Get the scalar value of variant as LocalDate, if the variant type is {@link Type#DATE}.
     *
     * @throws VariantTypeException If this variant is not a scalar value or is not {@link
     *     Type#DATE}.
     */
    LocalDate getDate() throws VariantTypeException;

    /**
     * Get the scalar value of variant as LocalDateTime, if the variant type is {@link
     * Type#TIMESTAMP}.
     *
     * @throws VariantTypeException If this variant is not a scalar value or is not {@link
     *     Type#TIMESTAMP}.
     */
    LocalDateTime getTimestamp() throws VariantTypeException;

    /**
     * Get the scalar value of variant as Instant, if the variant type is {@link Type#TIMESTAMP}.
     *
     * @throws VariantTypeException If this variant is not a scalar value or is not {@link
     *     Type#TIMESTAMP}.
     */
    Instant getInstant() throws VariantTypeException;

    /**
     * Get the scalar value of variant as byte array, if the variant type is {@link Type#BINARY}.
     *
     * @throws VariantTypeException If this variant is not a scalar value or is not {@link
     *     Type#BINARY}.
     */
    byte[] getBytes() throws VariantTypeException;

    /**
     * Get the scalar value of variant.
     *
     * @throws VariantTypeException If this variant is not a scalar value.
     */
    Object get() throws VariantTypeException;

    /**
     * Get the scalar value of variant.
     *
     * @throws VariantTypeException If this variant is not a scalar value.
     */
    <T> T getAs() throws VariantTypeException;

    /**
     * Access value of the specified element of an array variant. If index is out of range, null is
     * returned.
     *
     * <p>NOTE: if the element value has been explicitly set as <code>null</code> (which is
     * different from removal!), a variant that @{@link Variant#isNull()} returns true will be
     * returned, not null.
     *
     * @throws VariantTypeException If this variant is not an array.
     */
    Variant getElement(int index) throws VariantTypeException;

    /**
     * Access value of the specified field of an object variant. If there is no field with the
     * specified name, null is returned.
     *
     * <p>NOTE: if the property value has been explicitly set as <code>null</code>, a variant
     * that @{@link Variant#isNull()} returns true will be returned, not null.
     *
     * @throws VariantTypeException If this variant is not an object.
     */
    Variant getField(String fieldName) throws VariantTypeException;

    /** Parses the variant to json. */
    String toJson();

    /** The type of variant. */
    enum Type {
        OBJECT,
        ARRAY,
        NULL,
        BOOLEAN,
        TINYINT,
        SMALLINT,
        INT,
        BIGINT,
        FLOAT,
        DOUBLE,
        DECIMAL,
        STRING,
        DATE,
        TIMESTAMP,
        TIMESTAMP_LTZ,
        BINARY
    }
}

package org.apache.flink.types.variant;

/** Builder for variants. */
@PublicEvolving
public interface VariantBuilder {

    /** Create a variant from a byte. */
    Variant of(byte b);

    /** Create a variant from a short. */
    Variant of(short s);

    /** Create a variant from a int. */
    Variant of(int i);

    /** Create a variant from a long. */
    Variant of(long l);

    /** Create a variant from a string. */
    Variant of(String s);

    /** Create a variant from a double. */
    Variant of(Double d);

    /** Create a variant from a float. */
    Variant of(Float f);

    /** Create a variant from a byte array. */
    Variant of(byte[] bytes);

    /** Create a variant from a boolean. */
    Variant of(boolean b);

    /** Create a variant from a BigDecimal. */
    Variant of(BigDecimal bigDecimal);

    /** Create a variant from an Instant. */
    Variant of(Instant instant);

    /** Create a variant from a LocalDate. */
    Variant of(LocalDate localDate);

    /** Create a variant from a LocalDateTime. */
    Variant of(LocalDateTime localDateTime);

    /** Create a variant of null. */
    Variant ofNull();

    /** Get the builder of a variant object. */
    VariantObjectBuilder object();

    /** Get the builder of a variant object. */
    VariantObjectBuilder object(boolean allowDuplicateKeys);

    /** Get the builder for a variant array. */
    VariantArrayBuilder array();

    /** Builder for a variant object. */
    interface VariantObjectBuilder {

        /** Add a field to the object. */
        VariantObjectBuilder add(String key, Variant value);

        /** Build the variant object. */
        Variant build();
    }

    /** Builder for a variant array. */
    interface VariantArrayBuilder {

        /** Add a value to the array. */
        VariantArrayBuilder add(Variant value);

        /** Build the variant array. */
        Variant build();
    }
}

package org.apache.flink.types.variant;  

/** Exception thrown when a variant value is not of the expected type. */
@PublicEvolving
public class VariantTypeException extends RuntimeException {
    private static final long serialVersionUID = 1L;

    public VariantTypeException(String message) {
        super(message);
    }
}

/**
 * A data structure that represents a semi-structured variant with two binary values: value
 * and metadata. The value encodes types and values, but not field names. The metadata currently
 * contains a version flag and a list of field names. We can extend/modify the detailed binary
 * format given the version flag.
 *
 * @see <a href="https://github.com/apache/parquet-format/blob/master/VariantEncoding.md">Variant
 *     Binary Encoding</a> for the detail layout of the data structure.
 */
@PublicEvolving
public final class BinaryVariant implements Variant {
    ...
}

@PublicEvolving
public class VariantUtils {

    /**
     * Parse a JSON string into a {@link Variant}.
     *
     * @param json The input json string.
     */
    public static Variant parseJson(String json) throws IOException {
        ...
    }

    /**
     * Parse a JSON string into a {@link Variant}.
     *
     * @param json The input json string.
     * @param allowDuplicateKeys If set to false, parsing variant from JSON will throw an error if
     *     there are duplicate keys in the input JSON string. When set to true, the parser will keep
     *     the last occurrence of all fields with the same key.
     */
    public static Variant parseJson(String json, boolean allowDuplicateKeys) throws IOException {
        ...
    }
}

/**
 * Type information for primitive types (int, long, double, byte, ...), String, Date, Void,
 * BigInteger, BigDecimal, and Variant.
 */
@Public
public class BasicTypeInfo<T> extends TypeInformation<T> implements AtomicType<T> {

      ...

      public static final BasicTypeInfo<Variant> VARIANT_TYPE_INFO =
            new BasicTypeInfo<Variant>(
                    Variant.class, new Class<?>[] {}, VariantSerializer.INSTANCE, null);
}

@PublicEvolving
public final class DataTypes {
    ...

    /**
     * Data type of semi-structured data.
     *
     * <p>The type supports storing any semi-structured data, including ARRAY, MAP, and scalar
     * types. VARIANT can only store MAP types with keys of type STRING.
     *
     * @see VariantType
     */
    public static DataType VARIANT() {
        return new AtomicDataType(new VariantType());
    }
}

package org.apache.flink.table.types.logical;

/**
 * Data type of semi-structured data.
 *
 * <p>The type supports storing any semi-structured data, including ARRAY, MAP, and scalar types.
 * VARIANT can only store MAP types with keys of type STRING.
 */
@PublicEvolving
public class VariantType extends LogicalType {
    ...
}

@PublicEvolving
public interface RowData {

    ...

    /** Returns the variant value at the given position. */
    Variant getVariant(int pos);
}

-- Data Ingestion from MySQL to Paimon

CREATE TABLE t1 (
  id INTEGER,
  v STRING -- a json string
) WITH (
  'connector' = 'mysql-cdc',
  ...
)

CREATE TABLE paimon_catalog.ods.t1 (
  id INTEGER,
  v VARIANT
) WITH (
  'connector' = 'paimon'
  ...
)

INSERT INTO paimon_catalog.ods.t1 SELECT id, PARSE_JSON(v) FROM t1
;

-- Processing with Varaint and write the Variant as JSON string to Starrocks

CREATE TABLE ads_t1 (
    id INTEGER,
    v STRING,
    v1 INTEGER,
    v2 INTEGER,
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'connector' = 'starrocks'
    ...
);


INSERT INTO ads_t1 SELECT 
  id, 
  TO_JSON(v),
  r.v1,
  r.v2
LEFT JOIN paimon_catalog.ods.r
ON VARIANT_VALUE(v, '$.a.key') = r.key
;