@Override
    public BasicTypeDefine reconvert(Column column) {
        BasicTypeDefine.BasicTypeDefineBuilder builder =
                BasicTypeDefine.builder()
                        .name(column.getName())
                        .nullable(column.isNullable())
                        .comment(column.getComment())
                        .defaultValue(column.getDefaultValue());
        switch (column.getDataType().getSqlType()) {
            case BOOLEAN:
                builder.columnType(PG_BOOLEAN);
                builder.dataType(PG_BOOLEAN);
                break;
            case TINYINT:
            case SMALLINT:
                builder.columnType(PG_SMALLINT);
                builder.dataType(PG_SMALLINT);
                break;
            case INT:
                builder.columnType(PG_INTEGER);
                builder.dataType(PG_INTEGER);
                break;
            case BIGINT:
                builder.columnType(PG_BIGINT);
                builder.dataType(PG_BIGINT);
                break;
            case FLOAT:
                builder.columnType(PG_REAL);
                builder.dataType(PG_REAL);
                break;
            case DOUBLE:
                builder.columnType(PG_DOUBLE_PRECISION);
                builder.dataType(PG_DOUBLE_PRECISION);
                break;
            case DECIMAL:
                if (column.getSourceType() != null
                        && column.getSourceType().equalsIgnoreCase(PG_MONEY)) {
                    builder.columnType(PG_MONEY);
                    builder.dataType(PG_MONEY);
                } else {
                    DecimalType decimalType = (DecimalType) column.getDataType();
                    long precision = decimalType.getPrecision();
                    int scale = decimalType.getScale();
                    if (precision <= 0) {
                        precision = DEFAULT_PRECISION;
                        scale = DEFAULT_SCALE;
                        log.warn(
                                "The decimal column {} type decimal({},{}) is out of range, "
                                        + "which is precision less than 0, "
                                        + "it will be converted to decimal({},{})",
                                column.getName(),
                                decimalType.getPrecision(),
                                decimalType.getScale(),
                                precision,
                                scale);
                    } else if (precision > MAX_PRECISION) {
                        scale = (int) Math.max(0, scale - (precision - MAX_PRECISION));
                        precision = MAX_PRECISION;
                        log.warn(
                                "The decimal column {} type decimal({},{}) is out of range, "
                                        + "which exceeds the maximum precision of {}, "
                                        + "it will be converted to decimal({},{})",
                                column.getName(),
                                decimalType.getPrecision(),
                                decimalType.getScale(),
                                MAX_PRECISION,
                                precision,
                                scale);
                    }
                    if (scale < 0) {
                        scale = 0;
                        log.warn(
                                "The decimal column {} type decimal({},{}) is out of range, "
                                        + "which is scale less than 0, "
                                        + "it will be converted to decimal({},{})",
                                column.getName(),
                                decimalType.getPrecision(),
                                decimalType.getScale(),
                                precision,
                                scale);
                    } else if (scale > MAX_SCALE) {
                        scale = MAX_SCALE;
                        log.warn(
                                "The decimal column {} type decimal({},{}) is out of range, "
                                        + "which exceeds the maximum scale of {}, "
                                        + "it will be converted to decimal({},{})",
                                column.getName(),
                                decimalType.getPrecision(),
                                decimalType.getScale(),
                                MAX_SCALE,
                                precision,
                                scale);
                    }

                    builder.columnType(String.format("%s(%s,%s)", PG_NUMERIC, precision, scale));
                    builder.dataType(PG_NUMERIC);
                    builder.precision(precision);
                    builder.scale(scale);
                }
                break;
            case BYTES:
                builder.columnType(PG_BYTEA);
                builder.dataType(PG_BYTEA);
                break;
            case STRING:
                if (column.getColumnLength() == null || column.getColumnLength() <= 0) {
                    builder.columnType(PG_TEXT);
                    builder.dataType(PG_TEXT);
                } else if (column.getColumnLength() <= MAX_VARCHAR_LENGTH) {
                    builder.columnType(
                            String.format("%s(%s)", PG_VARCHAR, column.getColumnLength()));
                    builder.dataType(PG_VARCHAR);
                } else {
                    builder.columnType(PG_TEXT);
                    builder.dataType(PG_TEXT);
                }
                break;
            case DATE:
                builder.columnType(PG_DATE);
                builder.dataType(PG_DATE);
                break;
            case TIME:
                Integer timeScale = column.getScale();
                if (timeScale != null && timeScale > MAX_TIME_SCALE) {
                    timeScale = MAX_TIME_SCALE;
                    log.warn(
                            "The time column {} type time({}) is out of range, "
                                    + "which exceeds the maximum scale of {}, "
                                    + "it will be converted to time({})",
                            column.getName(),
                            column.getScale(),
                            MAX_SCALE,
                            timeScale);
                }
                if (timeScale != null && timeScale > 0) {
                    builder.columnType(String.format("%s(%s)", PG_TIME, timeScale));
                } else {
                    builder.columnType(PG_TIME);
                }
                builder.dataType(PG_TIME);
                builder.scale(timeScale);
                break;
            case TIMESTAMP:
                Integer timestampScale = column.getScale();
                if (timestampScale != null && timestampScale > MAX_TIMESTAMP_SCALE) {
                    timestampScale = MAX_TIMESTAMP_SCALE;
                    log.warn(
                            "The timestamp column {} type timestamp({}) is out of range, "
                                    + "which exceeds the maximum scale of {}, "
                                    + "it will be converted to timestamp({})",
                            column.getName(),
                            column.getScale(),
                            MAX_TIMESTAMP_SCALE,
                            timestampScale);
                }
                if (timestampScale != null && timestampScale > 0) {
                    builder.columnType(String.format("%s(%s)", PG_TIMESTAMP, timestampScale));
                } else {
                    builder.columnType(PG_TIMESTAMP);
                }
                builder.dataType(PG_TIMESTAMP);
                builder.scale(timestampScale);
                break;
            case ARRAY:
                ArrayType arrayType = (ArrayType) column.getDataType();
                SeaTunnelDataType elementType = arrayType.getElementType();
                switch (elementType.getSqlType()) {
                    case BOOLEAN:
                        builder.columnType(PG_BOOLEAN_ARRAY);
                        builder.dataType(PG_BOOLEAN_ARRAY);
                        break;
                    case TINYINT:
                    case SMALLINT:
                        builder.columnType(PG_SMALLINT_ARRAY);
                        builder.dataType(PG_SMALLINT_ARRAY);
                        break;
                    case INT:
                        builder.columnType(PG_INTEGER_ARRAY);
                        builder.dataType(PG_INTEGER_ARRAY);
                        break;
                    case BIGINT:
                        builder.columnType(PG_BIGINT_ARRAY);
                        builder.dataType(PG_BIGINT_ARRAY);
                        break;
                    case FLOAT:
                        builder.columnType(PG_REAL_ARRAY);
                        builder.dataType(PG_REAL_ARRAY);
                        break;
                    case DOUBLE:
                        builder.columnType(PG_DOUBLE_PRECISION_ARRAY);
                        builder.dataType(PG_DOUBLE_PRECISION_ARRAY);
                        break;
                    case BYTES:
                        builder.columnType(PG_BYTEA);
                        builder.dataType(PG_BYTEA);
                        break;
                    case STRING:
                        builder.columnType(PG_TEXT_ARRAY);
                        builder.dataType(PG_TEXT_ARRAY);
                        break;
                    default:
                        throw CommonError.convertToConnectorTypeError(
                                DatabaseIdentifier.POSTGRESQL,
                                elementType.getSqlType().name(),
                                column.getName());
                }
                break;
            default:
                throw CommonError.convertToConnectorTypeError(
                        DatabaseIdentifier.POSTGRESQL,
                        column.getDataType().getSqlType().name(),
                        column.getName());
        }

        return builder.build();
    }
@Test
    public void testReconvertFloat() {
        Column column =
                PhysicalColumn.builder().name("test").dataType(BasicType.FLOAT_TYPE).build();

        BasicTypeDefine typeDefine = PostgresTypeConverter.INSTANCE.reconvert(column);
        Assertions.assertEquals(column.getName(), typeDefine.getName());
        Assertions.assertEquals(PostgresTypeConverter.PG_REAL, typeDefine.getColumnType());
        Assertions.assertEquals(PostgresTypeConverter.PG_REAL, typeDefine.getDataType());
    }