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

        String sqlServerType = typeDefine.getDataType().toUpperCase();
        switch (sqlServerType) {
            case SQLSERVER_BIT:
                builder.sourceType(SQLSERVER_BIT);
                builder.dataType(BasicType.BOOLEAN_TYPE);
                break;
            case SQLSERVER_TINYINT:
            case SQLSERVER_TINYINT_IDENTITY:
                builder.sourceType(SQLSERVER_TINYINT);
                builder.dataType(BasicType.SHORT_TYPE);
                break;
            case SQLSERVER_SMALLINT:
            case SQLSERVER_SMALLINT_IDENTITY:
                builder.sourceType(SQLSERVER_SMALLINT);
                builder.dataType(BasicType.SHORT_TYPE);
                break;
            case SQLSERVER_INTEGER:
            case SQLSERVER_INTEGER_IDENTITY:
            case SQLSERVER_INT:
            case SQLSERVER_INT_IDENTITY:
                builder.sourceType(SQLSERVER_INT);
                builder.dataType(BasicType.INT_TYPE);
                break;
            case SQLSERVER_BIGINT:
            case SQLSERVER_BIGINT_IDENTITY:
                builder.sourceType(SQLSERVER_BIGINT);
                builder.dataType(BasicType.LONG_TYPE);
                break;
            case SQLSERVER_REAL:
                builder.sourceType(SQLSERVER_REAL);
                builder.dataType(BasicType.FLOAT_TYPE);
                break;
            case SQLSERVER_FLOAT:
                if (typeDefine.getPrecision() != null && typeDefine.getPrecision() <= 24) {
                    builder.sourceType(SQLSERVER_REAL);
                    builder.dataType(BasicType.FLOAT_TYPE);
                } else {
                    builder.sourceType(SQLSERVER_FLOAT);
                    builder.dataType(BasicType.DOUBLE_TYPE);
                }
                break;
            case SQLSERVER_DECIMAL:
            case SQLSERVER_NUMERIC:
                builder.sourceType(
                        String.format(
                                "%s(%s,%s)",
                                SQLSERVER_DECIMAL,
                                typeDefine.getPrecision(),
                                typeDefine.getScale()));
                builder.dataType(
                        new DecimalType(
                                typeDefine.getPrecision().intValue(), typeDefine.getScale()));
                builder.columnLength(typeDefine.getPrecision());
                builder.scale(typeDefine.getScale());
                break;
            case SQLSERVER_MONEY:
                builder.sourceType(SQLSERVER_MONEY);
                builder.dataType(
                        new DecimalType(
                                typeDefine.getPrecision().intValue(), typeDefine.getScale()));
                builder.columnLength(typeDefine.getPrecision());
                builder.scale(typeDefine.getScale());
                break;
            case SQLSERVER_SMALLMONEY:
                builder.sourceType(SQLSERVER_SMALLMONEY);
                builder.dataType(
                        new DecimalType(
                                typeDefine.getPrecision().intValue(), typeDefine.getScale()));
                builder.columnLength(typeDefine.getPrecision());
                builder.scale(typeDefine.getScale());
                break;
            case SQLSERVER_CHAR:
                builder.sourceType(String.format("%s(%s)", SQLSERVER_CHAR, typeDefine.getLength()));
                builder.dataType(BasicType.STRING_TYPE);
                builder.columnLength(
                        TypeDefineUtils.doubleByteTo4ByteLength(typeDefine.getLength()));
                break;
            case SQLSERVER_NCHAR:
                builder.sourceType(
                        String.format("%s(%s)", SQLSERVER_NCHAR, typeDefine.getLength()));
                builder.dataType(BasicType.STRING_TYPE);
                builder.columnLength(
                        TypeDefineUtils.doubleByteTo4ByteLength(typeDefine.getLength()));
                break;
            case SQLSERVER_VARCHAR:
                if (typeDefine.getLength() == -1) {
                    builder.sourceType(MAX_VARCHAR);
                    builder.columnLength(TypeDefineUtils.doubleByteTo4ByteLength(POWER_2_31 - 1));
                } else {
                    builder.sourceType(
                            String.format("%s(%s)", SQLSERVER_VARCHAR, typeDefine.getLength()));
                    builder.columnLength(
                            TypeDefineUtils.doubleByteTo4ByteLength(typeDefine.getLength()));
                }
                builder.dataType(BasicType.STRING_TYPE);
                break;
            case SQLSERVER_NVARCHAR:
                if (typeDefine.getLength() == -1) {
                    builder.sourceType(MAX_NVARCHAR);
                    builder.columnLength(TypeDefineUtils.doubleByteTo4ByteLength(POWER_2_31 - 1));
                } else {
                    builder.sourceType(
                            String.format("%s(%s)", SQLSERVER_NVARCHAR, typeDefine.getLength()));
                    builder.columnLength(
                            TypeDefineUtils.doubleByteTo4ByteLength(typeDefine.getLength()));
                }
                builder.dataType(BasicType.STRING_TYPE);
                break;
            case SQLSERVER_TEXT:
                builder.sourceType(SQLSERVER_TEXT);
                builder.dataType(BasicType.STRING_TYPE);
                builder.columnLength(POWER_2_31 - 1);
                break;
            case SQLSERVER_NTEXT:
                builder.sourceType(SQLSERVER_NTEXT);
                builder.dataType(BasicType.STRING_TYPE);
                builder.columnLength(POWER_2_30 - 1);
                break;
            case SQLSERVER_XML:
                builder.sourceType(SQLSERVER_XML);
                builder.dataType(BasicType.STRING_TYPE);
                builder.columnLength(POWER_2_31 - 1);
                break;
            case SQLSERVER_UNIQUEIDENTIFIER:
                builder.sourceType(SQLSERVER_UNIQUEIDENTIFIER);
                builder.dataType(BasicType.STRING_TYPE);
                builder.columnLength(TypeDefineUtils.charTo4ByteLength(typeDefine.getLength()));
                break;
            case SQLSERVER_SQLVARIANT:
                builder.sourceType(SQLSERVER_SQLVARIANT);
                builder.dataType(BasicType.STRING_TYPE);
                builder.columnLength(typeDefine.getLength());
                break;
            case SQLSERVER_BINARY:
                builder.sourceType(
                        String.format("%s(%s)", SQLSERVER_BINARY, typeDefine.getLength()));
                builder.dataType(PrimitiveByteArrayType.INSTANCE);
                builder.columnLength(typeDefine.getLength());
                break;
            case SQLSERVER_VARBINARY:
                if (typeDefine.getLength() == -1) {
                    builder.sourceType(MAX_VARBINARY);
                    builder.columnLength(POWER_2_31 - 1);
                } else {
                    builder.sourceType(
                            String.format("%s(%s)", SQLSERVER_VARBINARY, typeDefine.getLength()));
                    builder.columnLength(typeDefine.getLength());
                }
                builder.dataType(PrimitiveByteArrayType.INSTANCE);
                break;
            case SQLSERVER_IMAGE:
                builder.sourceType(SQLSERVER_IMAGE);
                builder.dataType(PrimitiveByteArrayType.INSTANCE);
                builder.columnLength(POWER_2_31 - 1);
                break;
            case SQLSERVER_TIMESTAMP:
                builder.sourceType(SQLSERVER_TIMESTAMP);
                builder.dataType(PrimitiveByteArrayType.INSTANCE);
                builder.columnLength(8L);
                break;
            case SQLSERVER_DATE:
                builder.sourceType(SQLSERVER_DATE);
                builder.dataType(LocalTimeType.LOCAL_DATE_TYPE);
                break;
            case SQLSERVER_TIME:
                builder.sourceType(String.format("%s(%s)", SQLSERVER_TIME, typeDefine.getScale()));
                builder.dataType(LocalTimeType.LOCAL_TIME_TYPE);
                builder.scale(typeDefine.getScale());
                break;
            case SQLSERVER_DATETIME:
                builder.sourceType(SQLSERVER_DATETIME);
                builder.dataType(LocalTimeType.LOCAL_DATE_TIME_TYPE);
                builder.scale(3);
                break;
            case SQLSERVER_DATETIME2:
                builder.sourceType(
                        String.format("%s(%s)", SQLSERVER_DATETIME2, typeDefine.getScale()));
                builder.dataType(LocalTimeType.LOCAL_DATE_TIME_TYPE);
                builder.scale(typeDefine.getScale());
                break;
            case SQLSERVER_DATETIMEOFFSET:
                builder.sourceType(
                        String.format("%s(%s)", SQLSERVER_DATETIMEOFFSET, typeDefine.getScale()));
                builder.dataType(LocalTimeType.LOCAL_DATE_TIME_TYPE);
                builder.scale(typeDefine.getScale());
                break;
            case SQLSERVER_SMALLDATETIME:
                builder.sourceType(SQLSERVER_SMALLDATETIME);
                builder.dataType(LocalTimeType.LOCAL_DATE_TIME_TYPE);
                break;
            default:
                throw CommonError.convertToSeaTunnelTypeError(
                        DatabaseIdentifier.SQLSERVER, sqlServerType, typeDefine.getName());
        }
        return builder.build();
    }
@Test
    public void testConvertBigint() {
        BasicTypeDefine<Object> typeDefine =
                BasicTypeDefine.builder()
                        .name("test")
                        .columnType("bigint")
                        .dataType("bigint")
                        .build();
        Column column = SqlServerTypeConverter.INSTANCE.convert(typeDefine);
        Assertions.assertEquals(typeDefine.getName(), column.getName());
        Assertions.assertEquals(BasicType.LONG_TYPE, column.getDataType());
        Assertions.assertEquals(typeDefine.getColumnType(), column.getSourceType().toLowerCase());
    }