@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 hanaType = typeDefine.getDataType().toUpperCase();
        if (typeDefine.getColumnType().endsWith(" ARRAY")) {
            typeDefine.setColumnType(typeDefine.getColumnType().replace(" ARRAY", ""));
            typeDefine.setDataType(removeColumnSizeIfNeed(typeDefine.getColumnType()));
            Column arrayColumn = convert(typeDefine);
            SeaTunnelDataType<?> newType;
            switch (arrayColumn.getDataType().getSqlType()) {
                case STRING:
                    newType = ArrayType.STRING_ARRAY_TYPE;
                    break;
                case BOOLEAN:
                    newType = ArrayType.BOOLEAN_ARRAY_TYPE;
                    break;
                case TINYINT:
                    newType = ArrayType.BYTE_ARRAY_TYPE;
                    break;
                case SMALLINT:
                    newType = ArrayType.SHORT_ARRAY_TYPE;
                    break;
                case INT:
                    newType = ArrayType.INT_ARRAY_TYPE;
                    break;
                case BIGINT:
                    newType = ArrayType.LONG_ARRAY_TYPE;
                    break;
                case FLOAT:
                    newType = ArrayType.FLOAT_ARRAY_TYPE;
                    break;
                case DOUBLE:
                    newType = ArrayType.DOUBLE_ARRAY_TYPE;
                    break;
                case DATE:
                    newType = ArrayType.LOCAL_DATE_ARRAY_TYPE;
                    break;
                case TIME:
                    newType = ArrayType.LOCAL_TIME_ARRAY_TYPE;
                    break;
                case TIMESTAMP:
                    newType = ArrayType.LOCAL_DATE_TIME_ARRAY_TYPE;
                    break;
                default:
                    throw CommonError.unsupportedDataType(
                            "SeaTunnel",
                            arrayColumn.getDataType().getSqlType().toString(),
                            typeDefine.getName());
            }
            return new PhysicalColumn(
                    arrayColumn.getName(),
                    newType,
                    arrayColumn.getColumnLength(),
                    arrayColumn.getScale(),
                    arrayColumn.isNullable(),
                    arrayColumn.getDefaultValue(),
                    arrayColumn.getComment(),
                    arrayColumn.getSourceType() + " ARRAY",
                    arrayColumn.getOptions());
        }
        switch (hanaType) {
            case HANA_BINARY:
            case HANA_VARBINARY:
                builder.dataType(PrimitiveByteArrayType.INSTANCE);
                if (typeDefine.getLength() == null || typeDefine.getLength() == 0) {
                    builder.columnLength(MAX_BINARY_LENGTH);
                } else {
                    builder.columnLength(typeDefine.getLength());
                }
                break;
            case HANA_BOOLEAN:
                builder.dataType(BasicType.BOOLEAN_TYPE);
                break;
            case HANA_VARCHAR:
            case HANA_ALPHANUM:
            case HANA_CLOB:
            case HANA_NCLOB:
            case HANA_TEXT:
            case HANA_BINTEXT:
                builder.dataType(BasicType.STRING_TYPE);
                if (typeDefine.getLength() == null || typeDefine.getLength() == 0) {
                    builder.columnLength(MAX_LOB_LENGTH);
                } else {
                    builder.columnLength(typeDefine.getLength());
                }
                break;
            case HANA_NVARCHAR:
            case HANA_SHORTTEXT:
                builder.dataType(BasicType.STRING_TYPE);
                builder.columnLength(TypeDefineUtils.charTo4ByteLength(typeDefine.getLength()));
                break;
            case HANA_DATE:
                builder.dataType(LocalTimeType.LOCAL_DATE_TYPE);
                break;
            case HANA_TIME:
                builder.dataType(LocalTimeType.LOCAL_TIME_TYPE);
                builder.scale(0);
                break;
            case HANA_SECONDDATE:
                builder.dataType(LocalTimeType.LOCAL_DATE_TIME_TYPE);
                builder.scale(0);
                break;
            case HANA_TIMESTAMP:
                builder.dataType(LocalTimeType.LOCAL_DATE_TIME_TYPE);
                if (typeDefine.getScale() == null) {
                    builder.scale(TIMESTAMP_DEFAULT_SCALE);
                } else {
                    builder.scale(typeDefine.getScale());
                }
                break;
            case HANA_BLOB:
                builder.dataType(PrimitiveByteArrayType.INSTANCE);
                builder.columnLength(typeDefine.getLength());
                break;
            case HANA_TINYINT:
            case HANA_SMALLINT:
                builder.dataType(BasicType.SHORT_TYPE);
                break;
            case HANA_INTEGER:
                builder.dataType(BasicType.INT_TYPE);
                break;
            case HANA_BIGINT:
                builder.dataType(BasicType.LONG_TYPE);
                break;
            case HANA_DECIMAL:
                Integer scale = typeDefine.getScale();
                long precision =
                        typeDefine.getLength() != null
                                ? typeDefine.getLength().intValue()
                                : MAX_PRECISION - 4;
                if (scale == null) {
                    builder.dataType(new DecimalType((int) precision, MAX_SCALE));
                    builder.columnLength(precision);
                    builder.scale(MAX_SCALE);
                } else if (scale < 0) {
                    int newPrecision = (int) (precision - scale);
                    if (newPrecision == 1) {
                        builder.dataType(BasicType.SHORT_TYPE);
                    } else if (newPrecision <= 9) {
                        builder.dataType(BasicType.INT_TYPE);
                    } else if (newPrecision <= 18) {
                        builder.dataType(BasicType.LONG_TYPE);
                    } else if (newPrecision < 38) {
                        builder.dataType(new DecimalType(newPrecision, 0));
                        builder.columnLength((long) newPrecision);
                    } else {
                        builder.dataType(new DecimalType(DEFAULT_PRECISION, 0));
                        builder.columnLength((long) DEFAULT_PRECISION);
                    }
                } else {
                    builder.dataType(new DecimalType((int) precision, scale));
                    builder.columnLength(precision);
                    builder.scale(scale);
                }
                break;
            case HANA_SMALLDECIMAL:
                if (typeDefine.getPrecision() == null) {
                    builder.dataType(new DecimalType(DEFAULT_PRECISION, MAX_SMALL_DECIMAL_SCALE));
                    builder.columnLength((long) DEFAULT_PRECISION);
                    builder.scale(MAX_SMALL_DECIMAL_SCALE);
                } else {
                    builder.dataType(
                            new DecimalType(
                                    typeDefine.getPrecision().intValue(), MAX_SMALL_DECIMAL_SCALE));
                    builder.columnLength(typeDefine.getPrecision());
                    builder.scale(MAX_SMALL_DECIMAL_SCALE);
                }
                break;
            case HANA_REAL:
                builder.dataType(BasicType.FLOAT_TYPE);
                break;
            case HANA_DOUBLE:
                builder.dataType(BasicType.DOUBLE_TYPE);
                break;
            case HANA_ST_POINT:
            case HANA_ST_GEOMETRY:
                builder.dataType(PrimitiveByteArrayType.INSTANCE);
                break;
            default:
                throw CommonError.convertToSeaTunnelTypeError(
                        DatabaseIdentifier.SAP_HANA, hanaType, typeDefine.getName());
        }
        return builder.build();
    }
@Test
    public void testConvertInteger() {
        BasicTypeDefine<Object> typeDefine =
                BasicTypeDefine.builder()
                        .name("test")
                        .columnType("INTEGER")
                        .dataType("INTEGER")
                        .build();
        Column column = SapHanaTypeConverter.INSTANCE.convert(typeDefine);
        Assertions.assertEquals(typeDefine.getName(), column.getName());
        Assertions.assertEquals(BasicType.INT_TYPE, column.getDataType());
        Assertions.assertEquals(typeDefine.getColumnType(), column.getSourceType());
    }