@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 dataType = typeDefine.getDataType().toUpperCase();
        switch (dataType) {
            case REDSHIFT_BOOLEAN:
                builder.sourceType(REDSHIFT_BOOLEAN);
                builder.dataType(BasicType.BOOLEAN_TYPE);
                break;
            case REDSHIFT_SMALLINT:
                builder.sourceType(REDSHIFT_SMALLINT);
                builder.dataType(BasicType.SHORT_TYPE);
                break;
            case REDSHIFT_INTEGER:
                builder.sourceType(REDSHIFT_INTEGER);
                builder.dataType(BasicType.INT_TYPE);
                break;
            case REDSHIFT_BIGINT:
                builder.sourceType(REDSHIFT_BIGINT);
                builder.dataType(BasicType.LONG_TYPE);
                break;
            case REDSHIFT_REAL:
                builder.sourceType(REDSHIFT_REAL);
                builder.dataType(BasicType.FLOAT_TYPE);
                break;
            case REDSHIFT_DOUBLE_PRECISION:
                builder.sourceType(REDSHIFT_DOUBLE_PRECISION);
                builder.dataType(BasicType.DOUBLE_TYPE);
                break;
            case REDSHIFT_NUMERIC:
                Long precision = typeDefine.getPrecision();
                Integer scale = typeDefine.getScale();
                if (precision == null || precision <= 0) {
                    precision = Long.valueOf(DEFAULT_PRECISION);
                    scale = DEFAULT_SCALE;
                } else if (precision > MAX_PRECISION) {
                    scale = scale - (int) (precision - MAX_PRECISION);
                    precision = Long.valueOf(MAX_PRECISION);
                }
                builder.sourceType(String.format("%s(%d,%d)", REDSHIFT_NUMERIC, precision, scale));
                builder.dataType(new DecimalType(Math.toIntExact(precision), scale));
                break;
            case REDSHIFT_CHARACTER:
                Long characterLength = typeDefine.getLength();
                if (characterLength == null || characterLength <= 0) {
                    characterLength = Long.valueOf(MAX_CHARACTER_LENGTH);
                }
                builder.sourceType(String.format("%s(%d)", REDSHIFT_CHARACTER, characterLength));
                builder.dataType(BasicType.STRING_TYPE);
                builder.columnLength(characterLength);
                break;
            case REDSHIFT_CHARACTER_VARYING:
                Long characterVaryingLength = typeDefine.getLength();
                if (characterVaryingLength == null || characterVaryingLength <= 0) {
                    characterVaryingLength = Long.valueOf(MAX_CHARACTER_VARYING_LENGTH);
                }
                builder.sourceType(
                        String.format(
                                "%s(%d)", REDSHIFT_CHARACTER_VARYING, characterVaryingLength));
                builder.dataType(BasicType.STRING_TYPE);
                builder.columnLength(characterVaryingLength);
                break;
            case REDSHIFT_HLLSKETCH:
                builder.sourceType(REDSHIFT_HLLSKETCH);
                builder.dataType(BasicType.STRING_TYPE);
                builder.columnLength(MAX_HLLSKETCH_LENGTH);
                break;
            case REDSHIFT_SUPER:
                builder.sourceType(REDSHIFT_SUPER);
                builder.dataType(BasicType.STRING_TYPE);
                builder.columnLength(MAX_SUPER_LENGTH);
                break;
            case REDSHIFT_VARBYTE:
            case REDSHIFT_BINARY_VARYING:
                builder.sourceType(
                        String.format(
                                "%s(%d)", typeDefine.getDataType(), MAX_BINARY_VARYING_LENGTH));
                builder.dataType(PrimitiveByteArrayType.INSTANCE);
                builder.columnLength(MAX_BINARY_VARYING_LENGTH);
                break;
            case REDSHIFT_TIME:
                builder.sourceType(REDSHIFT_TIME);
                builder.dataType(LocalTimeType.LOCAL_TIME_TYPE);
                builder.scale(MAX_TIME_SCALE);
                break;
            case REDSHIFT_TIMETZ:
                builder.sourceType(REDSHIFT_TIMETZ);
                builder.dataType(LocalTimeType.LOCAL_TIME_TYPE);
                builder.scale(MAX_TIME_SCALE);
                break;
            case REDSHIFT_TIMESTAMP:
                builder.sourceType(REDSHIFT_TIMESTAMP);
                builder.dataType(LocalTimeType.LOCAL_DATE_TIME_TYPE);
                builder.scale(MAX_TIMESTAMP_SCALE);
                break;
            case REDSHIFT_TIMESTAMPTZ:
                builder.sourceType(REDSHIFT_TIMESTAMPTZ);
                builder.dataType(LocalTimeType.LOCAL_DATE_TIME_TYPE);
                builder.scale(MAX_TIMESTAMP_SCALE);
                break;
            default:
                try {
                    return super.convert(typeDefine);
                } catch (SeaTunnelRuntimeException e) {
                    throw CommonError.convertToSeaTunnelTypeError(
                            DatabaseIdentifier.REDSHIFT,
                            typeDefine.getDataType(),
                            typeDefine.getName());
                }
        }
        return builder.build();
    }
@Test
    public void testConvertSmallint() {
        BasicTypeDefine<Object> typeDefine =
                BasicTypeDefine.builder()
                        .name("test")
                        .columnType("SMALLINT")
                        .dataType("SMALLINT")
                        .build();
        Column column = RedshiftTypeConverter.INSTANCE.convert(typeDefine);
        Assertions.assertEquals(typeDefine.getName(), column.getName());
        Assertions.assertEquals(BasicType.SHORT_TYPE, column.getDataType());
        Assertions.assertEquals(typeDefine.getColumnType(), column.getSourceType());
    }