public BaseRowTypeInfo getBaseRowTypeInfo(){
        String[] fieldNames = getFields();
        Class[] fieldClass = getFieldClasses();
        LogicalType[] logicalTypes = new LogicalType[fieldClass.length];
        for (int i = 0; i < fieldClass.length; i++) {
            if(fieldClass[i].getName().equals(BigDecimal.class.getName())){
                logicalTypes[i] = new DecimalType(DecimalType.MAX_PRECISION, 18);
                continue;
            }

            Optional<DataType> optionalDataType = ClassDataTypeConverter.extractDataType(fieldClass[i]);
            if(!optionalDataType.isPresent()){
                throw new RuntimeException(String.format("not support table % field %s type %s", getName(), fieldNames[i], fieldClass[i]));
            }

            logicalTypes[i] = optionalDataType.get().getLogicalType();
        }

        return new BaseRowTypeInfo(logicalTypes, fieldNames);
    }
async operator req outside conn pool size, egg rdb conn pool size
core/src/main/java/com/dtstack/flink/sql/side/AbstractSideTableInfo.java