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

import org.apache.flink.annotation.PublicEvolving;

import java.io.Serializable;

/**
 * An interface that converts between the internal Flink data structure and the conversion class of a DataType.
*/
@PublicEvolving
public interface DataTypeConverter<I, E> extends Serializable {

   void open(ClassLoader classLoader);

   /**
    * Converts to internal data structure.
    *
    * <p>Note: Parameter must not be null. Output must not be null.
    */
   I toInternal(E external);

   /**
    * Converts to external data structure.
    *
    * <p>Note: Parameter must not be null. Output must not be null.
    */
   E toExternal(I internal);
}

public abstract class LogicalType implements Serializable {

    ...

    /**
     * Add a custom conversion to this type. Custom conversions are used to convert between
     * an external class and the internal representation of this logical type.
     *
     * @param clazz the external class for conversion.
     * @param customConverter the converter for the conversion.
     * @return a new LogicalType instance with the custom conversion added.
     */
    public final LogicalType withCustomConversion(
            Class<?> clazz, DataTypeConverter<Object, Object> customConverter) {
        Preconditions.checkNotNull(customConverter,
                "The custom converter should not be null.");
        // Make a copy to ensure the immutability of LogicalType.
        LogicalType copy = copy();
        copy.customConversions.put(clazz, customConverter);
        return copy;
    }

    /**
     * Add custom conversions to this type. Custom conversions are used to convert between
     * an external class and the internal representation of this logical type.
     *
     * @param conversions a map of external class to converter.
     * @return a new LogicalType instance with the custom conversions added.
     */
    public final LogicalType withCustomConversions(
            Map<Class<?>, DataTypeConverter<Object, Object>> conversions) {
        // make a copy to ensure the immutability of LogicalType
        LogicalType copy = copy();
        conversions.forEach((clazz, converter) -> {
            if (converter != null) {
                copy.customConversions.put(clazz, converter);
            }
        });
        return copy;
    }

    ...

}

public interface AbstractDataType<T extends AbstractDataType<T>> {
    ...

    /**
     * Adds a hint that data should be represented using the given class when entering or leaving
     * the table ecosystem.
     *
     * <p>A supported conversion class depends on the logical type and its nullability property.
     *
     * <p>Please see the implementation of {@link LogicalType#supportsInputConversion(Class)},
     * {@link LogicalType#supportsOutputConversion(Class)}, or the documentation for more
     * information about supported conversions.
     *
     * @param newConversionClass the class to convert the data to.
     * @param converter the converter to use for the conversion.
     * @return a new, reconfigured data type instance
     */
    default T bridgedTo(Class<?> newConversionClass, DataTypeConverter<Object, Object> converter) {
        throw new UnsupportedOperationException(
                "bridgedTo(Class<?>, DataTypeConverter) is not implemented");
    }
   
    ...
}