First, thanks for your efforts on rbatis. Conceptually I think you are really headed in the right direction. I wanted to inquire about expected behavior under a high volume of connections (in my case coming in facilitated by actix). I haven't spent much time investigating this particular issue, but, what I have observed is that under high connection load where those connections are issuing transactions against the database, rbatis quickly exhausts the connection pool and drives the CPU utilization up incredibly high and remains essentially frozen for an indeterminate amount of time.

I've been investigating the potential use of rbatis as a replacement for my direct interaction with sqlx as I like some of convenience of the insert/update processes. Performance is similar to sqlx in my testing(I realize you use sqlx-core), but I've ran into some issue with transactions. As an example if I do(in the example I setup in my project):

        let context = self.connection.get_context_rbatis().await?;
        let conn = self.connection.get_connection_rbatis().await?;
        conn.save(context.as_str(), registrant).await?;

And then run:

CLionProjects % wrk -c 50 -t 2 http://localhost:8080/v1/registrant/123
Running 10s test @ http://localhost:8080/v1/registrant/123
  2 threads and 50 connections
  Thread Stats   Avg      Stdev     Max   +/- Stdev
    Latency    21.80ms   12.50ms  99.78ms   81.08%
    Req/Sec     1.19k   243.06     1.59k    80.00%
  23788 requests in 10.01s, 27.72MB read
Requests/sec:   2375.27
Transfer/sec:      2.77MB

Performance is about 20% slower than my current implementation with sqlx, but fine. There is some abstraction in my own code, but what's happening above is passing a ""(blank) str to the context argument for save(). Ok no problem.

However, when I do:

        self.connection.start_transaction_rbatis().await?;
        let context = self.connection.get_context_rbatis().await?;
        let conn = self.connection.get_connection_rbatis().await?;
        conn.save(context.as_str(), registrant).await?;
        self.connection.commit_transaction_rbatis().await?;

And then run:

wrk -c 50 -t 2 http://localhost:8080/v1/registrant/123
Running 10s test @ http://localhost:8080/v1/registrant/123
  2 threads and 50 connections
  Thread Stats   Avg      Stdev     Max   +/- Stdev
    Latency    99.89ms   50.21ms 208.38ms   76.74%
    Req/Sec    71.33     52.97   140.00     66.67%
  43 requests in 10.04s, 51.31KB read
Requests/sec:      4.28
Transfer/sec:      5.11KB

rbatis quickly exhausts the connection pool(there are currently 100 available connections on postgres) and then proceeds to enter a state on my local workstation where it's occupying nearly 100% of available CPU where it stays indefinitely. The only difference is begin_tx() and commit() are being called across the same save() request.

What I'm currently doing with sqlx directly is to use acquire() to attach a dedicated connection, loading that into a Arc() and manually issuing BEGIN/COMMIT(and keeping track of things internally). I like your approach of generating a UUID for the context and loading that up into a dashmap of Transactions, but, something is awry with what's happening above. I should also add, if I issue requests at a lower velocity(say 15 connections across 2 threads) rbatis will keep up, and furthermore simply manual requests work flawlessly. This issue is related to the generation of a context, passing that context to save(), then calling commit() under a situation where many connections are simultaneously issuing requests.

When I run the same series of events(BEGIN/INSERT/COMMIT) on my direct integration with sqlx, I get(and still have available connections in the pool even with 100 connections requested by work over 2 threads):

CLionProjects % wrk -c 100 -t 2 http://localhost:8080/v1/registrant/123
Running 10s test @ http://localhost:8080/v1/registrant/123
  2 threads and 100 connections
  Thread Stats   Avg      Stdev     Max   +/- Stdev
    Latency    29.66ms   26.09ms 288.41ms   92.14%
    Req/Sec     1.94k   337.15     2.39k    86.50%
  38655 requests in 10.01s, 34.17MB read
Requests/sec:   3860.37
Transfer/sec:      3.41MB

And in my case I'm doing:

self.dedicated_connection = Some(Arc::new(Mutex::new(self.pool.acquire().await?)));

And then passing that around as necessary and then dropping the connection after commit/rollback manually.

Again I haven't done a lot of investigation on this in rbatis, but I would not expect rbatis to behave this way when generating a context, passing that to save(), then committing it. Which is essentially what my test is doing, except it's with significant load being generated by wrk.

Any insights on this? I'd like to use rbatis, but would need to resolve this behavior.

Thanks again for any insights.

比如: #[query(sql="select* from xxx where name = $1")] Option<Vec> selectXxxByName(name: String) {None}

然后宏会改造这个函数为类似： Option<Vec> selectXxxByName(name:String, pool:Rbatis) { Some(pool.select(....)) }

Hi,

I try to write a test on database and want TxManager to rollback the transaction at the end of the test.

It seems, that rollback does not work.

Tried:

• begin_defer()
• begin_tx_defer()
• Rbatis.rollback()

Database is PostgreSQL

Can you please advise?

1. 使用like 的时候, 生成的语句在?的左右会有空格，导致找不到数据。 SELECT count(1) FROM t_test WHERE del = 0 AND TITLE LIKE '% ? %' ORDER BY create_time DESC
2. or 是怎么用的？ 会无端端多了一个or，其他的用了and. RB.new_wrapper() .eq("1","1") .or() .like("TITLE",&str) .or() .like("ORIGINAL_NAME",&str) 生成的语句： SELECT count(1) FROM t_test WHERE del = 0 AND 1 = ? OR OR TITLE LIKE '% ? %' OR OR ORIGINAL_NAME LIKE '% ? %'

fail to insert on postgres

DB Schema ordinal_position | column_name | data_type | character_maximum_length | modifier | notnull | hasdefault ------------------+----------------+-----------+--------------------------+----------+---------+------------ 1 | id | uuid | 16 | -1 | t | t 2 | create_date | timestamp | 8 | -1 | t | t 3 | update_date | timestamp | 8 | -1 | t | t 4 | version | int4 | 4 | -1 | t | f 5 | is_active | bool | 1 | -1 | t | t 6 | dateOfDelivery | timestamp | 8 | -1 | t | f 7 | tokenValue | int4 | 4 | -1 | t | t 8 | maxQuantity | int4 | 4 | -1 | t | t 9 | status | varchar | -1 | 54 | t | t 10 | coordinates | point | 16 | -1 | f | f 11 | chefId | uuid | 16 | -1 | f | f 12 | dishId | uuid | 16 | -1 | t | f (12 rows)

RUST CODE

let off = Offer {
        id: Some(Uuid::new_v4()),
        create_date: Some(Utc::now().naive_local()),
        update_date: Some(Utc::now().naive_local()),
        version: Some(1),
        is_active: Some(true),
        date_of_delivery: Some(Utc::now().naive_local()),
        token_value: Some(2),
        max_quantity: Some(2),
        status: Some("OUT_OF_DELIVERY".to_string()),
        // coordinates: None,
        chefId: Some(Uuid::parse_str("ebb787b4-d90c-4b0f-9bbe-7324f3d90efe").unwrap()),
        dishId: Some(Uuid::parse_str("85357370-8aef-4262-bf68-f6d38522cf6b").unwrap()),
    };
    Json(RB.save_batch("", &vec![off]).await)

log

[2020-09-26T12:22:06Z INFO  rbatis::rbatis] [rbatis] [] Exec ==> INSERT INTO Offer (chefId,create_date,date_of_delivery,dishId,id,is_active,max_quantity,status,token_value,update_date,version) VALUES ( $1 ,cast( $2  as timestamp),cast( $3  as timestamp), $4 , $5 , $6 , $7 , $8 , $9 ,cast( $10  as timestamp), $11 )
[2020-09-26T12:22:06Z INFO  rbatis::rbatis] [rbatis] [] Args ==> ["ebb787b4-d90c-4b0f-9bbe-7324f3d90efe","2020-09-26T12:22:06.537112100","2020-09-26T12:22:06.537112100","85357370-8aef-4262-bf68-f6d38522cf6b","3d30b1fa-4d41-4b54-88e4-9e9065904b9f",true,2,"OUT_OF_DELIVERY",2,"2020-09-26T12:22:06.537112100",1]
[2020-09-26T12:22:06Z INFO  rbatis::rbatis] [rbatis] [] RowsAffected <== 0

• my struct:

impl_select!(User{select_by_user_role(user:String, roles: Vec<String>) -> Option => "`where user = #{user} and role IN (#{roles}) limit 1`"});
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct User {
    pub username: String,
}

• log:

[rbatis] [428143780235972608] ==> select * from user where user = ? and role IN (?) limit 1
[rbatis]               Args   ==> ["test",["service_admin"]]

• I would expect generated query to be: select * from upis_role_ejb_realm net_id where net_id = "test" and role IN ("service_admin");

code:

#![allow(unused_must_use)]
#![allow(non_snake_case)]
#[macro_use]
extern crate rbatis;

use actix_web::{web, App, HttpResponse, HttpServer, Responder};
// use chrono::NaiveDateTime;
use rbatis::crud::{CRUD};
use rbatis::rbatis::Rbatis;
use rbatis::core::runtime::sync::Arc;

// #[crud_table(table_name:"task" | table_columns:"id,status,sequence_id,functor_id,total,current,complete,startTime,completeTime,create_at,update_at")]
#[crud_table]
#[derive(Clone, Debug)]
pub struct Task {
    pub id: Option<String>,
    pub status: Option<String>,
    pub sequence_id: Option<String>,
    pub functor_id: Option<String>,
    pub total: Option<i32>,
    pub current: Option<i32>,
    pub complete: Option<i32>,
    pub startTime: Option<i64>,
    pub completeTime: Option<i64>,
    pub created_at: Option<String>,
    pub updated_at: Option<String>,
}

impl Default for Task {
    fn default() -> Self {
        Task {
            id: None,
            status: None,
            sequence_id: None,
            functor_id: None,
            total: None,
            current: None,
            complete: None,
            startTime: None,
            completeTime: None,
            created_at: None,
            updated_at: None,
        }
    }
}

pub const PG_URL: &'static str = "postgresql://postgres:123456@abyssii:35432/ac";
pub const MYSQL_URL: &'static str = "mysql://root:123456@abyssii:33060/ma";

async fn postgres(rb: web::Data<Arc<RbatisConnection>>) -> impl Responder {
    let v = rb.postgres.fetch_list::<Task>().await.unwrap();
    HttpResponse::Ok().json(serde_json::json!(v).to_string())
}

async fn mysql(rb: web::Data<Arc<RbatisConnection>>) -> impl Responder {
    let v = rb.mysql.fetch_list::<Task>().await.unwrap();
    HttpResponse::Ok().json(serde_json::json!(v).to_string())
}

struct RbatisConnection {
    postgres: Rbatis,
    mysql: Rbatis,
}

#[actix_web::main]
async fn main() -> std::io::Result<()> {
    //log
    fast_log::init_log("requests.log", 1000, log::Level::Info, None, true);
    //init rbatis . also you can use  lazy_static! { static ref RB: Rbatis = Rbatis::new(); } replace this
    log::info!("linking database...");
    let pgrb = Rbatis::new();
    pgrb.link(PG_URL).await.expect("rbatis link postgres fail");
    let myrb = Rbatis::new();
    myrb.link(MYSQL_URL).await.expect("rbatis link mysql fail");

    let rb: RbatisConnection = RbatisConnection {
        postgres: pgrb,
        mysql: myrb,
    };

    let rb = Arc::new(rb);

    log::info!("linking database successful!");
    //router
    HttpServer::new(move || {
        App::new()
            //add into actix-web data
            .data(rb.to_owned())
            .route("/pg", web::get().to(postgres))
            .route("/my", web::get().to(mysql))
    })
        .bind("0.0.0.0:8000")?
        .run()
        .await
}

debug_mode output:

............gen impl CRUDTable:
 #[derive(serde :: Serialize, serde :: Deserialize)] #[derive(Clone, Debug)]
pub struct Task
{
    pub id : Option < String >, pub status : Option < String >, pub
    sequence_id : Option < String >, pub functor_id : Option < String >, pub
    total : Option < i32 >, pub current : Option < i32 >, pub complete :
    Option < i32 >, pub startTime : Option < i64 >, pub completeTime : Option
    < i64 >, pub created_at : Option < String >, pub updated_at : Option <
    String >,
} impl rbatis :: crud :: CRUDTable for Task
{
    fn get(& self, column : & str) -> serde_json :: Value
    {
        return match column
        {
            "id" => { return serde_json :: json! (& self.id) ; } "status" =>
            { return serde_json :: json! (& self.status) ; } "sequence_id" =>
            { return serde_json :: json! (& self.sequence_id) ; } "functor_id"
            => { return serde_json :: json! (& self.functor_id) ; } "total" =>
            { return serde_json :: json! (& self.total) ; } "current" =>
            { return serde_json :: json! (& self.current) ; } "complete" =>
            { return serde_json :: json! (& self.complete) ; } "startTime" =>
            { return serde_json :: json! (& self.startTime) ; } "completeTime"
            => { return serde_json :: json! (& self.completeTime) ; }
            "created_at" =>
            { return serde_json :: json! (& self.created_at) ; } "updated_at"
            => { return serde_json :: json! (& self.updated_at) ; } _ =>
            { serde_json :: Value :: Null }
        }
    } fn table_name() -> String { "task".to_string() } fn table_columns() ->
    String
    {
        "id,status,sequence_id,functor_id,total,current,complete,startTime,completeTime,created_at,updated_at".to_string()
    } fn formats(driver_type : & rbatis :: core :: db :: DriverType) -> std ::
    collections :: HashMap < String, fn(arg : & str) -> String >
    {
        let mut m : std :: collections :: HashMap < String, fn(arg : & str) ->
        String > = std :: collections :: HashMap :: new() ; match driver_type
        {
            rbatis :: core :: db :: DriverType :: Mysql => { return m ; },
            rbatis :: core :: db :: DriverType :: Postgres => { return m ; },
            rbatis :: core :: db :: DriverType :: Sqlite => { return m ; },
            rbatis :: core :: db :: DriverType :: Mssql => { return m ; },
            rbatis :: core :: db :: DriverType :: None => { return m ; },
        }
    }
}
............gen impl CRUDTable end............

mysql table:

-- auto-generated definition
create table task
(
    id           char(36)     not null
        primary key,
    status       varchar(100) null,
    sequence_id  char(36)     null,
    functor_id   varchar(100) null,
    total        int          null,
    current      int          null,
    complete     int          null,
    startTime    bigint       null,
    completeTime bigint       null,
    created_at   timestamp    not null,
    updated_at   timestamp    not null
);

postgres table:

-- auto-generated definition
create table task
(
    id             uuid                     not null
        constraint task_pkey
            primary key,
    status         varchar(100),
    sequence_id    uuid,
    functor_id     varchar(100),
    total          integer,
    current        integer,
    complete       integer,
    "startTime"    bigint,
    "completeTime" bigint,
    created_at     timestamp with time zone not null,
    updated_at     timestamp with time zone not null
);

alter table task
    owner to postgres;

work with mysql, perfect!

work with postgres: postgres log:

lazy_static!{ pub static ref Rb: Rbatis<'static>={ let rb = Rbatis::new(); async_std::task::block_on(async{ rb.link(&CONFIG.database.url).await; }); return rb; }; }

Rb.fetch_by_wrapper()===>success

Rb.remove_by_wrapper() ===>卡主一直不会动

============================== 我改成 pub async fn connect()->Rbatis<'static>{ let rb = Rbatis::new(); rb.link(&CONFIG.database.url).await; return rb; }

connecti().await.fetch_by_wrapper()===>success connecti().await.remove_by_wrapper()===>success 这个就正常了。请问这个是什么原因

db design

CREATE TABLE IF NOT EXISTS categories
(
    id              UUID      DEFAULT uuid_generate_v4(),
    created_at      TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at      TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    uuid            VARCHAR NOT NULL UNIQUE,
    name            VARCHAR NOT NULL,
    default_setting jsonb   NOT NULL,
    PRIMARY KEY (id)
);

model

#[crud_table(table_name:categories| formats_pg: "id:{}::uuid,created_at:{}::timestamp,updated_at:{}::timestamp,default_setting:{}::jsonb")]
#[derive(Debug, Default, Serialize, Deserialize, Clone)]
#[serde(rename_all = "camelCase")]
pub struct Category {
    pub id: Option<Uuid>,
    pub created_at: Option<DateTimeNative>,
    pub updated_at: Option<DateTimeNative>,
    pub uuid: String,
    pub name: String,
    pub default_setting: Option<String>,
}

run save

 let category = Category {
            uuid: input.uuid,
            name: input.name,
            default_setting: Some(input.default_setting.to_string()),
            ..Default::default()
        };
        DB.save(
            &category,
            &[
                Skip::Column("id"),
                Skip::Column("created_at"),
                Skip::Column("updated_at"),
            ],
        )
        .await?;

and got

2021-11-03 12:19:47.140014700 UTC    INFO rbatis::plugin::log - [rbatis] [] Exec   ==> insert into categories (uuid,name,default_setting) values ($1,$2,$3::jsonb)
                                                                [rbatis] [] Args   ==> ["234", "34636", null]

2021-11-03 12:19:47.155176800 UTC    ERROR rbatis::plugin::log - [rbatis] [] ReturnErr  <== error returned from database: 无法把类型 integer 转换为 jsonb  (D:\rustup\.cargo\registry\src\mirrors.sjtug.sjtu.edu.cn-7a04d2510079875b\rbatis-3.0.1\src\plugin\log.rs:43)

我建立了一个struct和trait如下：

#[derive(Serialize, Deserialize, Clone, Debug)]
pub struct IpInfo {
    pub id: Option<u32>,
    pub net_group_id: Option<u32>,
    pub ip_pool_id: Option<u32>,
    pub ip: Option<String>,
    pub mask: Option<String>,
    pub gateway: Option<String>,
    pub status: Option<u8>, //0-> unused 1-> using 2-> reserved 3->abandoned
    pub pod: Option<String>,
    pub create_time: Option<NaiveDateTime>,

    //none table column
    pub ip_pool: Option<String>
}

impl CRUDTable for IpInfo {
    type IdType = u32; //默认提供IdType类型即可，接口里其他的method默认使用json序列化实现
    fn get_id(&self) -> Option<&Self::IdType> {
        return self.id.as_ref();
    } // 必须实现获取id值
    //fn table_name() -> String {} //可重写,默认ip_pool
    fn table_columns() -> String {
        "id,net_group_id,ip_pool_id,ip,mask,gateway,status,pod,create_time".to_string()
    } //可重写
    //fn format_chain() -> Vec<Box<dyn ColumnFormat>>{} //可重写
}

其中ip_pool字段不是数据库字段，只是后续给前端返回需要填上去的。所以我在impl CRUDTable里声明了table_columns()，把它去掉了，增删查没什么问题，但是 db.update_by_wrapper::<IpInfo>("", &mut update_ip, &w, true)的时候,发现log里面生成的sql，如下：

2021-04-08 02:48:40.399983300 +00:00 INFO rbatis::plugin::log - [rbatis] [] Exec  ==> update ip_info set  create_time = ?, gateway = ?, ip = ?, ip_pool = ?, ip_pool_id = ?, mask = ?, net_group_id = ?, pod = ?, status = ?  where id = ?
                                                                [rbatis] [] Args  ==> ["2021-04-07T12:38:50","10.85.243.1","10.85.243.20",null,3,"255.255.255.0",3,null,0,277]
thread 'actix-rt:worker:1' panicked at 'called `Result::unwrap()` on an `Err` value: E("error returned from database: 1054 (42S22): Unknown column \'ip_pool\' in \'field list\'")', src/ip_manage.rs:125:10
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace

居然update了ip_pool，这个非数据库字段，导致报错。。。 我排查了一下。发现这个方法是直接json！打包然后更新，并未读取table_columns()然后做更新，这是个问题。盼及时修复！，我在生产使用，太难了。。。

let req = PageRequest::new(1, 1); //分页请求，页码，条数
    let wraper = RB
        .new_wrapper()
        .eq("1", 1)
        .order_by(false, &["create_date"])
        .check()
        .unwrap();
    let r: rbatis_core::Result<Page<CyCustomZtzts>> =
        RB.fetch_page_by_wrapper("", &wraper, &req).await;

生成了sql：

SELECT count(1) FROM cy_custom_ztzts WHERE 1 = $1 ORDER BY create_date DESC

含有ORDER BY 无法正常运行

hi

使用mysql show 语句只打印字段名，无内容输出，代码如下

    let sql_show_ssl_cipher = "SHOW STATUS LIKE 'Ssl_cipher'";
    let cipher_rbatis = rb
        .fetch_decode::<Vec<HashMap<String, String>>>(sql_show_ssl_cipher, vec![])
        .await;
    println!(">>>>> Cipher in use from rbatis: {:?}", cipher_rbatis);

运行结果输出如下

>>>>> Cipher in use from rbatis: Ok([{"Variable_name": "Ssl_cipher", "Value": ""}])

是哪里用错了吗？谢谢