@Override
    public Column convert(BasicTypeDefine typeDefine) {
        try {
            return super.convert(typeDefine);
        } catch (SeaTunnelRuntimeException e) {
            PhysicalColumn.PhysicalColumnBuilder builder =
                    PhysicalColumn.builder()
                            .name(typeDefine.getName())
                            .sourceType(typeDefine.getColumnType())
                            .nullable(typeDefine.isNullable())
                            .defaultValue(typeDefine.getDefaultValue())
                            .comment(typeDefine.getComment());

            String kingbaseDataType = typeDefine.getDataType().toUpperCase();
            switch (kingbaseDataType) {
                case KB_TINYINT:
                    builder.dataType(BasicType.BYTE_TYPE);
                    break;
                case KB_MONEY:
                    builder.dataType(new DecimalType(38, 18));
                    builder.columnLength(38L);
                    builder.scale(18);
                    break;
                case KB_BLOB:
                    builder.dataType(PrimitiveByteArrayType.INSTANCE);
                    builder.columnLength((long) (1024 * 1024 * 1024));
                    break;
                case KB_CLOB:
                    builder.dataType(BasicType.STRING_TYPE);
                    builder.columnLength(typeDefine.getLength());
                    builder.columnLength((long) (1024 * 1024 * 1024));
                    break;
                case KB_BIT:
                    builder.dataType(PrimitiveByteArrayType.INSTANCE);
                    // BIT(M) -> BYTE(M/8)
                    long byteLength = typeDefine.getLength() / 8;
                    byteLength += typeDefine.getLength() % 8 > 0 ? 1 : 0;
                    builder.columnLength(byteLength);
                    break;
                default:
                    throw CommonError.convertToSeaTunnelTypeError(
                            DatabaseIdentifier.KINGBASE,
                            typeDefine.getDataType(),
                            typeDefine.getName());
            }
            return builder.build();
        }
    }
@Test
    public void testConvertOtherString() {
        BasicTypeDefine<Object> typeDefine =
                BasicTypeDefine.builder().name("test").columnType("text").dataType("text").build();
        Column column = KingbaseTypeConverter.INSTANCE.convert(typeDefine);
        Assertions.assertEquals(typeDefine.getName(), column.getName());
        Assertions.assertEquals(BasicType.STRING_TYPE, column.getDataType());
        Assertions.assertEquals(null, column.getColumnLength());
        Assertions.assertEquals(typeDefine.getColumnType(), column.getSourceType());

        typeDefine =
                BasicTypeDefine.builder().name("test").columnType("json").dataType("json").build();
        column = KingbaseTypeConverter.INSTANCE.convert(typeDefine);
        Assertions.assertEquals(typeDefine.getName(), column.getName());
        Assertions.assertEquals(BasicType.STRING_TYPE, column.getDataType());
        Assertions.assertEquals(null, column.getColumnLength());
        Assertions.assertEquals(typeDefine.getColumnType(), column.getSourceType());

        typeDefine =
                BasicTypeDefine.builder()
                        .name("test")
                        .columnType("jsonb")
                        .dataType("jsonb")
                        .build();
        column = KingbaseTypeConverter.INSTANCE.convert(typeDefine);
        Assertions.assertEquals(typeDefine.getName(), column.getName());
        Assertions.assertEquals(BasicType.STRING_TYPE, column.getDataType());
        Assertions.assertEquals(null, column.getColumnLength());
        Assertions.assertEquals(typeDefine.getColumnType(), column.getSourceType());

        typeDefine =
                BasicTypeDefine.builder().name("test").columnType("xml").dataType("xml").build();
        column = KingbaseTypeConverter.INSTANCE.convert(typeDefine);
        Assertions.assertEquals(typeDefine.getName(), column.getName());
        Assertions.assertEquals(BasicType.STRING_TYPE, column.getDataType());
        Assertions.assertEquals(null, column.getColumnLength());
        Assertions.assertEquals(typeDefine.getColumnType(), column.getSourceType());
    }