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

        String dmType = typeDefine.getDataType().toUpperCase();
        switch (dmType) {
            case DM_BIT:
                builder.sourceType(DM_BIT);
                builder.dataType(BasicType.BOOLEAN_TYPE);
                break;
            case DM_TINYINT:
                builder.sourceType(DM_TINYINT);
                builder.dataType(BasicType.BYTE_TYPE);
                break;
            case DM_BYTE:
                builder.sourceType(DM_BYTE);
                builder.dataType(BasicType.BYTE_TYPE);
                break;
            case DM_SMALLINT:
                builder.sourceType(DM_SMALLINT);
                builder.dataType(BasicType.SHORT_TYPE);
                break;
            case DM_INT:
                builder.sourceType(DM_INT);
                builder.dataType(BasicType.INT_TYPE);
                break;
            case DM_INTEGER:
                builder.sourceType(DM_INTEGER);
                builder.dataType(BasicType.INT_TYPE);
                break;
            case DM_PLS_INTEGER:
                builder.sourceType(DM_PLS_INTEGER);
                builder.dataType(BasicType.INT_TYPE);
                break;
            case DM_BIGINT:
                builder.sourceType(DM_BIGINT);
                builder.dataType(BasicType.LONG_TYPE);
                break;
            case DM_REAL:
                builder.sourceType(DM_REAL);
                builder.dataType(BasicType.FLOAT_TYPE);
                break;
            case DM_FLOAT:
                builder.sourceType(DM_FLOAT);
                builder.dataType(BasicType.DOUBLE_TYPE);
                break;
            case DM_DOUBLE:
                builder.sourceType(DM_DOUBLE);
                builder.dataType(BasicType.DOUBLE_TYPE);
                break;
            case DM_DOUBLE_PRECISION:
                builder.sourceType(DM_DOUBLE_PRECISION);
                builder.dataType(BasicType.DOUBLE_TYPE);
                break;
            case DM_NUMERIC:
            case DM_NUMBER:
            case DM_DECIMAL:
            case DM_DEC:
                DecimalType decimalType;
                if (typeDefine.getPrecision() != null && typeDefine.getPrecision() > 0) {
                    decimalType =
                            new DecimalType(
                                    typeDefine.getPrecision().intValue(), typeDefine.getScale());
                } else {
                    decimalType = new DecimalType(DEFAULT_PRECISION, DEFAULT_SCALE);
                }
                builder.sourceType(
                        String.format(
                                "%s(%s,%s)",
                                DM_DECIMAL, decimalType.getPrecision(), decimalType.getScale()));
                builder.dataType(decimalType);
                builder.columnLength((long) decimalType.getPrecision());
                builder.scale(decimalType.getScale());
                break;
            case DM_CHAR:
            case DM_CHARACTER:
                builder.sourceType(String.format("%s(%s)", DM_CHAR, typeDefine.getLength()));
                builder.dataType(BasicType.STRING_TYPE);
                builder.columnLength(TypeDefineUtils.charTo4ByteLength(typeDefine.getLength()));
                break;
            case DM_VARCHAR:
            case DM_VARCHAR2:
                builder.sourceType(String.format("%s(%s)", DM_VARCHAR2, typeDefine.getLength()));
                builder.dataType(BasicType.STRING_TYPE);
                builder.columnLength(TypeDefineUtils.charTo4ByteLength(typeDefine.getLength()));
                break;
            case DM_TEXT:
                builder.sourceType(DM_TEXT);
                builder.dataType(BasicType.STRING_TYPE);
                // dm text max length is 2147483647
                builder.columnLength(typeDefine.getLength());
                break;
            case DM_LONG:
                builder.sourceType(DM_LONG);
                builder.dataType(BasicType.STRING_TYPE);
                // dm long max length is 2147483647
                builder.columnLength(typeDefine.getLength());
                break;
            case DM_LONGVARCHAR:
                builder.sourceType(DM_LONGVARCHAR);
                builder.dataType(BasicType.STRING_TYPE);
                // dm longvarchar max length is 2147483647
                builder.columnLength(typeDefine.getLength());
                break;
            case DM_CLOB:
                builder.sourceType(DM_CLOB);
                builder.dataType(BasicType.STRING_TYPE);
                // dm clob max length is 2147483647
                builder.columnLength(typeDefine.getLength());
                break;
            case DM_BINARY:
                builder.sourceType(String.format("%s(%s)", DM_BINARY, typeDefine.getLength()));
                builder.dataType(PrimitiveByteArrayType.INSTANCE);
                builder.columnLength(typeDefine.getLength());
                break;
            case DM_VARBINARY:
                builder.sourceType(String.format("%s(%s)", DM_VARBINARY, typeDefine.getLength()));
                builder.dataType(PrimitiveByteArrayType.INSTANCE);
                builder.columnLength(typeDefine.getLength());
                break;
            case DM_LONGVARBINARY:
                builder.sourceType(DM_LONGVARBINARY);
                builder.dataType(PrimitiveByteArrayType.INSTANCE);
                builder.columnLength(typeDefine.getLength());
                break;
            case DM_IMAGE:
                builder.sourceType(DM_IMAGE);
                builder.dataType(PrimitiveByteArrayType.INSTANCE);
                builder.columnLength(typeDefine.getLength());
                break;
            case DM_BLOB:
                builder.sourceType(DM_BLOB);
                builder.dataType(PrimitiveByteArrayType.INSTANCE);
                builder.columnLength(typeDefine.getLength());
                break;
            case DM_BFILE:
                builder.sourceType(DM_BFILE);
                builder.dataType(BasicType.STRING_TYPE);
                builder.columnLength(typeDefine.getLength());
                break;
            case DM_DATE:
                builder.sourceType(DM_DATE);
                builder.dataType(LocalTimeType.LOCAL_DATE_TYPE);
                break;
            case DM_TIME:
                if (typeDefine.getScale() == null) {
                    builder.sourceType(DM_TIME);
                } else {
                    builder.sourceType(String.format("%s(%s)", DM_TIME, typeDefine.getScale()));
                }
                builder.dataType(LocalTimeType.LOCAL_TIME_TYPE);
                builder.scale(typeDefine.getScale());
                break;
            case DM_TIME_WITH_TIME_ZONE:
                if (typeDefine.getScale() == null) {
                    builder.sourceType(DM_TIME_WITH_TIME_ZONE);
                } else {
                    builder.sourceType(
                            String.format("TIME(%s) WITH TIME ZONE", typeDefine.getScale()));
                }
                builder.dataType(LocalTimeType.LOCAL_TIME_TYPE);
                builder.scale(typeDefine.getScale());
                break;
            case DM_TIMESTAMP:
                if (typeDefine.getScale() == null) {
                    builder.sourceType(DM_TIMESTAMP);
                } else {
                    builder.sourceType(
                            String.format("%s(%s)", DM_TIMESTAMP, typeDefine.getScale()));
                }
                builder.dataType(LocalTimeType.LOCAL_DATE_TIME_TYPE);
                builder.scale(typeDefine.getScale());
                break;
            case DM_DATETIME:
                if (typeDefine.getScale() == null) {
                    builder.sourceType(DM_DATETIME);
                } else {
                    builder.sourceType(String.format("%s(%s)", DM_DATETIME, typeDefine.getScale()));
                }
                builder.dataType(LocalTimeType.LOCAL_DATE_TIME_TYPE);
                builder.scale(typeDefine.getScale());
                break;
            case DM_DATETIME_WITH_TIME_ZONE:
                if (typeDefine.getScale() == null) {
                    builder.sourceType(DM_DATETIME_WITH_TIME_ZONE);
                } else {
                    builder.sourceType(
                            String.format("DATETIME(%s) WITH TIME ZONE", typeDefine.getScale()));
                }
                builder.dataType(LocalTimeType.LOCAL_DATE_TIME_TYPE);
                builder.scale(typeDefine.getScale());
                break;
            default:
                throw CommonError.convertToSeaTunnelTypeError(
                        DatabaseIdentifier.DAMENG, typeDefine.getDataType(), typeDefine.getName());
        }
        return builder.build();
    }
@Test
    public void testConvertBfile() {
        BasicTypeDefine<Object> typeDefine =
                BasicTypeDefine.builder()
                        .name("test")
                        .columnType("bfile")
                        .dataType("bfile")
                        .length(2147483647L)
                        .build();
        Column column = DmdbTypeConverter.INSTANCE.convert(typeDefine);
        Assertions.assertEquals(typeDefine.getName(), column.getName());
        Assertions.assertEquals(BasicType.STRING_TYPE, column.getDataType());
        Assertions.assertEquals(2147483647L, column.getColumnLength());
        Assertions.assertEquals(typeDefine.getColumnType(), column.getSourceType().toLowerCase());
    }