@Override
    public Column convert(BasicTypeDefine typeDefine) {
        PhysicalColumn.PhysicalColumnBuilder builder =
                PhysicalColumn.builder()
                        .name(typeDefine.getName())
                        .sourceType(typeDefine.getColumnType())
                        .nullable(typeDefine.isNullable())
                        .defaultValue(typeDefine.getDefaultValue())
                        .comment(typeDefine.getComment());

        String mysqlDataType = typeDefine.getDataType().toUpperCase();
        if (mysqlDataType.endsWith("ZEROFILL")) {
            mysqlDataType =
                    mysqlDataType.substring(0, mysqlDataType.length() - "ZEROFILL".length()).trim();
        }
        if (typeDefine.isUnsigned() && !(mysqlDataType.endsWith(" UNSIGNED"))) {
            mysqlDataType = mysqlDataType + " UNSIGNED";
        }
        switch (mysqlDataType) {
            case MYSQL_NULL:
                builder.dataType(BasicType.VOID_TYPE);
                break;
            case MYSQL_BIT:
                if (typeDefine.getLength() == null || typeDefine.getLength() <= 0) {
                    builder.dataType(BasicType.BOOLEAN_TYPE);
                } else if (typeDefine.getLength() == 1) {
                    builder.dataType(BasicType.BOOLEAN_TYPE);
                } else {
                    builder.dataType(PrimitiveByteArrayType.INSTANCE);
                    // BIT(M) -> BYTE(M/8)
                    long byteLength = typeDefine.getLength() / 8;
                    byteLength += typeDefine.getLength() % 8 > 0 ? 1 : 0;
                    builder.columnLength(byteLength);
                }
                break;
            case MYSQL_TINYINT:
                if (typeDefine.getColumnType().equalsIgnoreCase("tinyint(1)")) {
                    builder.dataType(BasicType.BOOLEAN_TYPE);
                } else {
                    builder.dataType(BasicType.BYTE_TYPE);
                }
                break;
            case MYSQL_TINYINT_UNSIGNED:
            case MYSQL_SMALLINT:
                builder.dataType(BasicType.SHORT_TYPE);
                break;
            case MYSQL_SMALLINT_UNSIGNED:
            case MYSQL_MEDIUMINT:
            case MYSQL_MEDIUMINT_UNSIGNED:
            case MYSQL_INT:
            case MYSQL_INTEGER:
            case MYSQL_YEAR:
                builder.dataType(BasicType.INT_TYPE);
                break;
            case MYSQL_INT_UNSIGNED:
            case MYSQL_INTEGER_UNSIGNED:
            case MYSQL_BIGINT:
                builder.dataType(BasicType.LONG_TYPE);
                break;
            case MYSQL_BIGINT_UNSIGNED:
                DecimalType intDecimalType = new DecimalType(20, 0);
                builder.dataType(intDecimalType);
                builder.columnLength(Long.valueOf(intDecimalType.getPrecision()));
                builder.scale(intDecimalType.getScale());
                break;
            case MYSQL_FLOAT:
                builder.dataType(BasicType.FLOAT_TYPE);
                break;
            case MYSQL_FLOAT_UNSIGNED:
                log.warn("{} will probably cause value overflow.", MYSQL_FLOAT_UNSIGNED);
                builder.dataType(BasicType.FLOAT_TYPE);
                break;
            case MYSQL_DOUBLE:
                builder.dataType(BasicType.DOUBLE_TYPE);
                break;
            case MYSQL_DOUBLE_UNSIGNED:
                log.warn("{} will probably cause value overflow.", MYSQL_DOUBLE_UNSIGNED);
                builder.dataType(BasicType.DOUBLE_TYPE);
                break;
            case MYSQL_DECIMAL:
                Preconditions.checkArgument(typeDefine.getPrecision() > 0);

                DecimalType decimalType;
                if (typeDefine.getPrecision() > DEFAULT_PRECISION) {
                    log.warn("{} will probably cause value overflow.", MYSQL_DECIMAL);
                    decimalType = new DecimalType(DEFAULT_PRECISION, DEFAULT_SCALE);
                } else {
                    decimalType =
                            new DecimalType(
                                    typeDefine.getPrecision().intValue(),
                                    typeDefine.getScale() == null
                                            ? 0
                                            : typeDefine.getScale().intValue());
                }
                builder.dataType(decimalType);
                builder.columnLength(Long.valueOf(decimalType.getPrecision()));
                builder.scale(decimalType.getScale());
                break;
            case MYSQL_DECIMAL_UNSIGNED:
                Preconditions.checkArgument(typeDefine.getPrecision() > 0);

                log.warn("{} will probably cause value overflow.", MYSQL_DECIMAL_UNSIGNED);
                DecimalType decimalUnsignedType =
                        new DecimalType(
                                typeDefine.getPrecision().intValue() + 1,
                                typeDefine.getScale() == null
                                        ? 0
                                        : typeDefine.getScale().intValue());
                builder.dataType(decimalUnsignedType);
                builder.columnLength(Long.valueOf(decimalUnsignedType.getPrecision()));
                builder.scale(decimalUnsignedType.getScale());
                break;
            case MYSQL_ENUM:
                builder.dataType(BasicType.STRING_TYPE);
                if (typeDefine.getLength() == null || typeDefine.getLength() <= 0) {
                    builder.columnLength(100L);
                } else {
                    builder.columnLength(typeDefine.getLength());
                }
                break;
            case MYSQL_CHAR:
            case MYSQL_VARCHAR:
                if (typeDefine.getLength() == null || typeDefine.getLength() <= 0) {
                    builder.columnLength(TypeDefineUtils.charTo4ByteLength(1L));
                } else {
                    builder.columnLength(typeDefine.getLength());
                }
                builder.dataType(BasicType.STRING_TYPE);
                break;
            case MYSQL_TINYTEXT:
                builder.dataType(BasicType.STRING_TYPE);
                builder.columnLength(POWER_2_8 - 1);
                break;
            case MYSQL_TEXT:
                builder.dataType(BasicType.STRING_TYPE);
                builder.columnLength(POWER_2_16 - 1);
                break;
            case MYSQL_MEDIUMTEXT:
                builder.dataType(BasicType.STRING_TYPE);
                builder.columnLength(POWER_2_24 - 1);
                break;
            case MYSQL_LONGTEXT:
                builder.dataType(BasicType.STRING_TYPE);
                builder.columnLength(POWER_2_32 - 1);
                break;
            case MYSQL_JSON:
                builder.dataType(BasicType.STRING_TYPE);
                break;
            case MYSQL_BINARY:
            case MYSQL_VARBINARY:
                if (typeDefine.getLength() == null || typeDefine.getLength() <= 0) {
                    builder.columnLength(1L);
                } else {
                    builder.columnLength(typeDefine.getLength());
                }
                builder.dataType(PrimitiveByteArrayType.INSTANCE);
                break;
            case MYSQL_TINYBLOB:
                builder.dataType(PrimitiveByteArrayType.INSTANCE);
                builder.columnLength(POWER_2_8 - 1);
                break;
            case MYSQL_BLOB:
                builder.dataType(PrimitiveByteArrayType.INSTANCE);
                builder.columnLength(POWER_2_16 - 1);
                break;
            case MYSQL_MEDIUMBLOB:
                builder.dataType(PrimitiveByteArrayType.INSTANCE);
                builder.columnLength(POWER_2_24 - 1);
                break;
            case MYSQL_LONGBLOB:
                builder.dataType(PrimitiveByteArrayType.INSTANCE);
                builder.columnLength(POWER_2_32 - 1);
                break;
            case MYSQL_GEOMETRY:
                builder.dataType(PrimitiveByteArrayType.INSTANCE);
                break;
            case MYSQL_DATE:
                builder.dataType(LocalTimeType.LOCAL_DATE_TYPE);
                break;
            case MYSQL_TIME:
                builder.dataType(LocalTimeType.LOCAL_TIME_TYPE);
                builder.scale(typeDefine.getScale());
                break;
            case MYSQL_DATETIME:
            case MYSQL_TIMESTAMP:
                builder.dataType(LocalTimeType.LOCAL_DATE_TIME_TYPE);
                builder.scale(typeDefine.getScale());
                break;
            default:
                throw CommonError.convertToSeaTunnelTypeError(
                        DatabaseIdentifier.MYSQL, mysqlDataType, typeDefine.getName());
        }
        return builder.build();
    }
@Test
    public void testConvertBlob() {
        BasicTypeDefine<Object> typeDefine =
                BasicTypeDefine.builder()
                        .name("test")
                        .columnType("tinyblob")
                        .dataType("tinyblob")
                        .build();
        Column column = MySqlTypeConverter.DEFAULT_INSTANCE.convert(typeDefine);
        Assertions.assertEquals(typeDefine.getName(), column.getName());
        Assertions.assertEquals(PrimitiveByteArrayType.INSTANCE, column.getDataType());
        Assertions.assertEquals(255, column.getColumnLength());
        Assertions.assertEquals(typeDefine.getColumnType(), column.getSourceType());

        typeDefine =
                BasicTypeDefine.builder().name("test").columnType("blob").dataType("blob").build();
        column = MySqlTypeConverter.DEFAULT_INSTANCE.convert(typeDefine);
        Assertions.assertEquals(typeDefine.getName(), column.getName());
        Assertions.assertEquals(PrimitiveByteArrayType.INSTANCE, column.getDataType());
        Assertions.assertEquals(65535, column.getColumnLength());
        Assertions.assertEquals(typeDefine.getColumnType(), column.getSourceType());

        typeDefine =
                BasicTypeDefine.builder()
                        .name("test")
                        .columnType("mediumblob")
                        .dataType("mediumblob")
                        .build();
        column = MySqlTypeConverter.DEFAULT_INSTANCE.convert(typeDefine);
        Assertions.assertEquals(typeDefine.getName(), column.getName());
        Assertions.assertEquals(PrimitiveByteArrayType.INSTANCE, column.getDataType());
        Assertions.assertEquals(16777215, column.getColumnLength());
        Assertions.assertEquals(typeDefine.getColumnType(), column.getSourceType());

        typeDefine =
                BasicTypeDefine.builder()
                        .name("test")
                        .columnType("longblob")
                        .dataType("longblob")
                        .build();
        column = MySqlTypeConverter.DEFAULT_INSTANCE.convert(typeDefine);
        Assertions.assertEquals(typeDefine.getName(), column.getName());
        Assertions.assertEquals(PrimitiveByteArrayType.INSTANCE, column.getDataType());
        Assertions.assertEquals(4294967295L, column.getColumnLength());
        Assertions.assertEquals(typeDefine.getColumnType(), column.getSourceType());
    }