@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 db2Type = typeDefine.getDataType().toUpperCase();
        switch (db2Type) {
            case DB2_BOOLEAN:
                builder.sourceType(DB2_BOOLEAN);
                builder.dataType(BasicType.BOOLEAN_TYPE);
                break;
            case DB2_SMALLINT:
                builder.sourceType(DB2_SMALLINT);
                builder.dataType(BasicType.SHORT_TYPE);
                break;
            case DB2_INT:
            case DB2_INTEGER:
                builder.sourceType(DB2_INT);
                builder.dataType(BasicType.INT_TYPE);
                break;
            case DB2_BIGINT:
                builder.sourceType(DB2_BIGINT);
                builder.dataType(BasicType.LONG_TYPE);
                break;
            case DB2_REAL:
                builder.sourceType(DB2_REAL);
                builder.dataType(BasicType.FLOAT_TYPE);
                break;
            case DB2_DOUBLE:
                builder.sourceType(DB2_DOUBLE);
                builder.dataType(BasicType.DOUBLE_TYPE);
                break;
            case DB2_DECFLOAT:
                builder.sourceType(DB2_DECFLOAT);
                builder.dataType(BasicType.DOUBLE_TYPE);
                break;
            case DB2_DECIMAL:
                builder.sourceType(
                        String.format(
                                "%s(%s,%s)",
                                DB2_DECIMAL, typeDefine.getPrecision(), typeDefine.getScale()));
                builder.dataType(
                        new DecimalType(
                                Math.toIntExact(typeDefine.getPrecision()), typeDefine.getScale()));
                builder.columnLength(typeDefine.getPrecision());
                builder.scale(typeDefine.getScale());
                break;
            case DB2_CHARACTER:
            case DB2_CHAR:
                builder.sourceType(String.format("%s(%d)", DB2_CHAR, typeDefine.getLength()));
                // For char/varchar this length is in bytes
                builder.columnLength(typeDefine.getLength());
                builder.dataType(BasicType.STRING_TYPE);
                break;
            case DB2_VARCHAR:
                builder.sourceType(String.format("%s(%d)", DB2_VARCHAR, typeDefine.getLength()));
                builder.columnLength(typeDefine.getLength());
                builder.dataType(BasicType.STRING_TYPE);
                break;
            case DB2_LONG_VARCHAR:
                builder.sourceType(DB2_LONG_VARCHAR);
                // default length is 32700
                builder.columnLength(typeDefine.getLength());
                builder.dataType(BasicType.STRING_TYPE);
                break;
            case DB2_CLOB:
                builder.sourceType(String.format("%s(%d)", DB2_CLOB, typeDefine.getLength()));
                builder.columnLength(typeDefine.getLength());
                builder.dataType(BasicType.STRING_TYPE);
                break;
            case DB2_GRAPHIC:
                builder.sourceType(String.format("%s(%d)", DB2_GRAPHIC, typeDefine.getLength()));
                builder.columnLength(TypeDefineUtils.charTo4ByteLength(typeDefine.getLength()));
                builder.dataType(BasicType.STRING_TYPE);
                break;
            case DB2_VARGRAPHIC:
                builder.sourceType(String.format("%s(%d)", DB2_VARGRAPHIC, typeDefine.getLength()));
                builder.columnLength(TypeDefineUtils.charTo4ByteLength(typeDefine.getLength()));
                builder.dataType(BasicType.STRING_TYPE);
                break;
            case DB2_DBCLOB:
                builder.sourceType(String.format("%s(%d)", DB2_DBCLOB, typeDefine.getLength()));
                builder.columnLength(TypeDefineUtils.charTo4ByteLength(typeDefine.getLength()));
                builder.dataType(BasicType.STRING_TYPE);
                break;
            case DB2_XML:
                builder.sourceType(DB2_XML);
                builder.columnLength((long) Integer.MAX_VALUE);
                builder.dataType(BasicType.STRING_TYPE);
                break;
            case DB2_BINARY:
                builder.sourceType(String.format("%s(%d)", DB2_BINARY, typeDefine.getLength()));
                builder.columnLength(typeDefine.getLength());
                builder.dataType(PrimitiveByteArrayType.INSTANCE);
                break;
            case DB2_VARBINARY:
                builder.sourceType(String.format("%s(%d)", DB2_VARBINARY, typeDefine.getLength()));
                builder.columnLength(typeDefine.getLength());
                builder.dataType(PrimitiveByteArrayType.INSTANCE);
                break;
            case DB2_BLOB:
                builder.sourceType(String.format("%s(%d)", DB2_BLOB, typeDefine.getLength()));
                builder.columnLength(typeDefine.getLength());
                builder.dataType(PrimitiveByteArrayType.INSTANCE);
                break;
            case DB2_DATE:
                builder.sourceType(DB2_DATE);
                builder.dataType(LocalTimeType.LOCAL_DATE_TYPE);
                break;
            case DB2_TIME:
                builder.sourceType(DB2_TIME);
                builder.dataType(LocalTimeType.LOCAL_TIME_TYPE);
                break;
            case DB2_TIMESTAMP:
                builder.sourceType(String.format("%s(%d)", DB2_TIMESTAMP, typeDefine.getScale()));
                builder.dataType(LocalTimeType.LOCAL_DATE_TIME_TYPE);
                builder.scale(typeDefine.getScale());
                break;
            default:
                throw CommonError.convertToSeaTunnelTypeError(
                        DatabaseIdentifier.DB_2, db2Type, typeDefine.getName());
        }
        return builder.build();
    }
@Test
    public void testConvertDecimal() {
        BasicTypeDefine<Object> typeDefine =
                BasicTypeDefine.builder()
                        .name("test")
                        .columnType("DECIMAL")
                        .dataType("DECIMAL")
                        .precision(31L)
                        .scale(1)
                        .build();
        Column column = DB2TypeConverter.INSTANCE.convert(typeDefine);
        Assertions.assertEquals(typeDefine.getName(), column.getName());
        Assertions.assertEquals(new DecimalType(31, 1), column.getDataType());
        Assertions.assertEquals(31, column.getColumnLength());
        Assertions.assertEquals(1, column.getScale());
        Assertions.assertEquals(
                String.format(
                        "%s(%s,%s)",
                        DB2TypeConverter.DB2_DECIMAL,
                        typeDefine.getPrecision(),
                        typeDefine.getScale()),
                column.getSourceType());
    }