public static org.apache.flink.table.types.DataType toFlinkDataType(DataType type) {
        // ordered by type root definition
        List<DataType> children = type.getChildren();
        int length = DataTypes.getLength(type).orElse(0);
        int precision = DataTypes.getPrecision(type).orElse(0);
        int scale = DataTypes.getScale(type).orElse(0);
        switch (type.getTypeRoot()) {
            case CHAR:
                return type.isNullable()
                        ? org.apache.flink.table.api.DataTypes.CHAR(length)
                        : org.apache.flink.table.api.DataTypes.CHAR(length).notNull();
            case VARCHAR:
                return type.isNullable()
                        ? org.apache.flink.table.api.DataTypes.VARCHAR(length)
                        : org.apache.flink.table.api.DataTypes.VARCHAR(length).notNull();
            case BOOLEAN:
                return type.isNullable()
                        ? org.apache.flink.table.api.DataTypes.BOOLEAN()
                        : org.apache.flink.table.api.DataTypes.BOOLEAN().notNull();
            case BINARY:
                return type.isNullable()
                        ? org.apache.flink.table.api.DataTypes.BINARY(length)
                        : org.apache.flink.table.api.DataTypes.BINARY(length).notNull();
            case VARBINARY:
                return type.isNullable()
                        ? org.apache.flink.table.api.DataTypes.VARBINARY(length)
                        : org.apache.flink.table.api.DataTypes.VARBINARY(length).notNull();
            case DECIMAL:
                return type.isNullable()
                        ? org.apache.flink.table.api.DataTypes.DECIMAL(precision, scale)
                        : org.apache.flink.table.api.DataTypes.DECIMAL(precision, scale).notNull();
            case TINYINT:
                return type.isNullable()
                        ? org.apache.flink.table.api.DataTypes.TINYINT()
                        : org.apache.flink.table.api.DataTypes.TINYINT().notNull();
            case SMALLINT:
                return type.isNullable()
                        ? org.apache.flink.table.api.DataTypes.SMALLINT()
                        : org.apache.flink.table.api.DataTypes.SMALLINT().notNull();
            case INTEGER:
                return type.isNullable()
                        ? org.apache.flink.table.api.DataTypes.INT()
                        : org.apache.flink.table.api.DataTypes.INT().notNull();
            case DATE:
                return type.isNullable()
                        ? org.apache.flink.table.api.DataTypes.DATE()
                        : org.apache.flink.table.api.DataTypes.DATE().notNull();
            case TIME_WITHOUT_TIME_ZONE:
                return type.isNullable()
                        ? org.apache.flink.table.api.DataTypes.TIME(precision)
                        : org.apache.flink.table.api.DataTypes.TIME(precision).notNull();
            case BIGINT:
                return type.isNullable()
                        ? org.apache.flink.table.api.DataTypes.BIGINT()
                        : org.apache.flink.table.api.DataTypes.BIGINT().notNull();
            case FLOAT:
                return type.isNullable()
                        ? org.apache.flink.table.api.DataTypes.FLOAT()
                        : org.apache.flink.table.api.DataTypes.FLOAT().notNull();
            case DOUBLE:
                return type.isNullable()
                        ? org.apache.flink.table.api.DataTypes.DOUBLE()
                        : org.apache.flink.table.api.DataTypes.DOUBLE().notNull();
            case TIMESTAMP_WITHOUT_TIME_ZONE:
                return type.isNullable()
                        ? org.apache.flink.table.api.DataTypes.TIMESTAMP(precision)
                        : org.apache.flink.table.api.DataTypes.TIMESTAMP(precision).notNull();
            case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
                return type.isNullable()
                        ? org.apache.flink.table.api.DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(
                                precision)
                        : org.apache.flink.table.api.DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(
                                        precision)
                                .notNull();
            case TIMESTAMP_WITH_TIME_ZONE:
                return type.isNullable()
                        ? org.apache.flink.table.api.DataTypes.TIMESTAMP_WITH_TIME_ZONE(precision)
                        : org.apache.flink.table.api.DataTypes.TIMESTAMP_WITH_TIME_ZONE(precision)
                                .notNull();
            case ARRAY:
                Preconditions.checkState(children != null && !children.isEmpty());
                return type.isNullable()
                        ? org.apache.flink.table.api.DataTypes.ARRAY(
                                toFlinkDataType(children.get(0)))
                        : org.apache.flink.table.api.DataTypes.ARRAY(
                                toFlinkDataType(children.get(0)).notNull());
            case MAP:
                Preconditions.checkState(children != null && children.size() > 1);
                return type.isNullable()
                        ? org.apache.flink.table.api.DataTypes.MAP(
                                toFlinkDataType(children.get(0)), toFlinkDataType(children.get(1)))
                        : org.apache.flink.table.api.DataTypes.MAP(
                                toFlinkDataType(children.get(0)),
                                toFlinkDataType(children.get(1)).notNull());
            case ROW:
                Preconditions.checkState(!CollectionUtil.isNullOrEmpty(children));
                RowType rowType = (RowType) type;
                List<org.apache.flink.table.api.DataTypes.Field> fields =
                        rowType.getFields().stream()
                                .map(DataField::toFlinkDataTypeField)
                                .collect(Collectors.toList());
                return type.isNullable()
                        ? org.apache.flink.table.api.DataTypes.ROW(fields)
                        : org.apache.flink.table.api.DataTypes.ROW(fields).notNull();
            default:
                throw new IllegalArgumentException("Illegal type: " + type);
        }
    }
@Test
    void testToFlinkDataType() {
        List<DataField> list =
                IntStream.range(0, ALL_TYPES.length)
                        .mapToObj(i -> DataTypes.FIELD("f" + i, ALL_TYPES[i]))
                        .collect(Collectors.toList());

        org.apache.flink.table.types.DataType dataType =
                DataTypeUtils.toFlinkDataType(new RowType(list));

        org.apache.flink.table.types.DataType expectedDataType =
                ROW(
                        FIELD("f0", BOOLEAN()),
                        FIELD("f1", BYTES()),
                        FIELD("f2", BINARY(10)),
                        FIELD("f3", VARBINARY(10)),
                        FIELD("f4", CHAR(10)),
                        FIELD("f5", VARCHAR(10)),
                        FIELD("f6", STRING()),
                        FIELD("f7", INT()),
                        FIELD("f8", TINYINT()),
                        FIELD("f9", SMALLINT()),
                        FIELD("f10", BIGINT()),
                        FIELD("f11", DOUBLE()),
                        FIELD("f12", FLOAT()),
                        FIELD("f13", DECIMAL(6, 3)),
                        FIELD("f14", DATE()),
                        FIELD("f15", TIME()),
                        FIELD("f16", TIME(6)),
                        FIELD("f17", TIMESTAMP()),
                        FIELD("f18", TIMESTAMP(6)),
                        FIELD("f19", TIMESTAMP_LTZ()),
                        FIELD("f20", TIMESTAMP_LTZ(6)),
                        FIELD("f21", TIMESTAMP_WITH_TIME_ZONE()),
                        FIELD("f22", TIMESTAMP_WITH_TIME_ZONE(6)),
                        FIELD("f23", ARRAY(BIGINT())),
                        FIELD("f24", MAP(SMALLINT(), STRING())),
                        FIELD("f25", ROW(FIELD("f1", STRING()), FIELD("f2", STRING(), "desc"))),
                        FIELD("f26", ROW(SMALLINT(), STRING())));

        assertThat(dataType).isEqualTo(expectedDataType);
    }