Can I transfer Arc<Vec<Row>> between 2 tokio thread?

[tc19901016]

My rust program has a oracle-read thread and a consume thread.
1)oracle-read thread use [ResultSet.next] to get Row and send a Arc<Vec> to tokio::sync::mpcs::unbounded queue.
2)consume thread receive Arc<Vec> and do something.
but Rust compiler outut some error:*mut oracle::binding::binding::dpiVector cannot be shared between threads safely
thank you

the oracle-read code is below:

fn read_data_to_queue(&mut self, tx:UnboundedSender<(Arc<Vec<Row>>, i32, i32)>){
        let start_time = Instant::now();
        println!("start read rows");
        let mut index = 0;
        'iter_rows:loop{
            let mut row_buffer:Vec<Row> = Vec::with_capacity(ORACLE_ARRAY_SIZE as usize);
            let mut is_finish = false;
            for i in 0..ORACLE_ARRAY_SIZE{
                if let Some(r) = (*self.rows).next(){//r: Result<Row, Error>
                    row_buffer.push(r.unwrap());
                }else{
                    is_finish = true;
                    break;
                }
            }
            let temp = Arc::new(row_buffer);
            // print!("sender{};", index);
            tx.send((Arc::clone(&temp), 1, index));
            index+=1;
            if is_finish{
                break 'iter_rows;
            }
        }
        tx.send((Arc::new(Vec::new()), 0, -1));
        println!("[{:?}]read data time:{:?}",thread::current().id(), start_time.elapsed());
    }

the consume thread code is below:

'consumer:loop{
                            match rx.recv().await {
                                Some((row_buffer, is_finish, batch_no)) =>{
                                    let timestamp = Instant::now();
                                    if is_finish==0 {
                                        break 'consumer;
                                    }
                                    let (n, is_last) = parser.fetch_next_producer_thread(row_buffer).unwrap();
                                    duration_pre+=timestamp.elapsed().as_micros();
                                   #[allow(clippy::needless_range_loop)]
                                    for col in 0..dst.ncols(){
                                        let (duration_parse_for, duration_convert_for, duration_write_for) = f[col](&mut parser, &mut dst).unwrap();
                                        duration_parse+=duration_parse_for;
                                        duration_convert+=duration_convert_for;
                                        duration_write+=duration_write_for;
                                    }
                                },
                                None =>{break 'consumer;},
                            }
                        }

queue code:let (tx, mut rx):(UnboundedSender<(Arc<Vec>, i32, i32)>, UnboundedReceiver<(Arc<Vec>, i32, i32)>)=mpsc::unbounded_channel();

detail error is below:
error[E0277]: *mut oracle::binding::binding::dpiVector cannot be shared between threads safely
--> connectorsdm\src\dispatcher.rs:191:48
|
191 | let read_task=tokio::spawn(async move {
| _______________------------^
| | |
| | required by a bound introduced by this call
192 | | let mut data_reader = crate::sources::oracle::OracleTextSourceDataReader::new(conn_url.unwrap().as_str(), sql.as_str());
193 | | data_reader.read_data_to_queue(tx);
194 | | });
| |^ *mut oracle::binding::binding::dpiVector cannot be shared between threads safely
|
= help: within SqlValue<'static>, the trait Sync is not implemented for *mut oracle::binding::binding::dpiVector, which is required by {async block@connectorsdm\src\dispatcher.rs:191:48: 194:22}: Send
note: required because it appears within the type oracle::binding::binding::dpiDataBuffer
--> C:\Users\tianchuan01.cargo\registry\src\mirrors.ustc.edu.cn-12df342d903acd47\oracle-0.6.2\src\binding\binding.rs:846:11
|
846 | pub union dpiDataBuffer {
| ^^^^^^^^^^^^^
note: required because it appears within the type oracle::binding::binding::dpiData
--> C:\Users\tianchuan01.cargo\registry\src\mirrors.ustc.edu.cn-12df342d903acd47\oracle-0.6.2\src\binding\binding.rs:1797:12
|
1797 | pub struct dpiData {
| ^^^^^^^
= note: required because it appears within the type &'static mut oracle::binding::binding::dpiData
note: required because it appears within the type oracle::sql_value::DpiData<'static>
--> C:\Users\tianchuan01.cargo\registry\src\mirrors.ustc.edu.cn-12df342d903acd47\oracle-0.6.2\src\sql_value.rs:128:6
|
128 | enum DpiData<'a> {
| ^^^^^^^
note: required because it appears within the type SqlValue<'static>
--> C:\Users\tianchuan01.cargo\registry\src\mirrors.ustc.edu.cn-12df342d903acd47\oracle-0.6.2\src\sql_value.rs:143:12
|
143 | pub struct SqlValue<'a> {
| ^^^^^^^^
= note: required for std::ptr::Unique<SqlValue<'static>> to implement Sync
note: required because it appears within the type alloc::raw_vec::RawVec<SqlValue<'static>>
--> /rustc/129f3b9964af4d4a709d1383930ade12dfe7c081\library\alloc\src\raw_vec.rs:69:19
note: required because it appears within the type Vec<SqlValue<'static>>
--> /rustc/129f3b9964af4d4a709d1383930ade12dfe7c081\library\alloc\src\vec\mod.rs:398:12
note: required because it appears within the type oracle::Row
--> C:\Users\tianchuan01.cargo\registry\src\mirrors.ustc.edu.cn-12df342d903acd47\oracle-0.6.2\src\row.rs:34:12
|
34 | pub struct Row {
| ^^^
= note: required for std::ptr::Uniqueoracle::Row to implement Sync
note: required because it appears within the type alloc::raw_vec::RawVecoracle::Row
--> /rustc/129f3b9964af4d4a709d1383930ade12dfe7c081\library\alloc\src\raw_vec.rs:69:19
note: required because it appears within the type Vecoracle::Row
--> /rustc/129f3b9964af4d4a709d1383930ade12dfe7c081\library\alloc\src\vec\mod.rs:398:12
= note: required for Arc<Vecoracle::Row> to implement Send
= note: required because it appears within the type (Arc<Vecoracle::Row>, i32, i32)
= note: required for tokio::sync::mpsc::chan::Chan<(Arc<Vecoracle::Row>, i32, i32), tokio::sync::mpsc::unbounded::Semaphore> to implement Sync
= note: 1 redundant requirement hidden
= note: required for Arc<tokio::sync::mpsc::chan::Chan<(Arc<Vecoracle::Row>, i32, i32), tokio::sync::mpsc::unbounded::Semaphore>> to implement Send
note: required because it appears within the type tokio::sync::mpsc::chan::Tx<(Arc<Vecoracle::Row>, i32, i32), tokio::sync::mpsc::unbounded::Semaphore>
--> C:\Users\tianchuan01.cargo\registry\src\mirrors.ustc.edu.cn-12df342d903acd47\tokio-1.40.0\src\sync\mpsc\chan.rs:19:19
|
19 | pub(crate) struct Tx<T, S> {
| ^^
note: required because it appears within the type UnboundedSender<(Arc<Vecoracle::Row>, i32, i32)>
--> C:\Users\tianchuan01.cargo\registry\src\mirrors.ustc.edu.cn-12df342d903acd47\tokio-1.40.0\src\sync\mpsc\unbounded.rs:11:12
|
11 | pub struct UnboundedSender {
| ^^^^^^^^^^^^^^^
note: required because it's used within this async block
--> connectorsdm\src\dispatcher.rs:191:48
|
191 | let read_task=tokio::spawn(async move {
| ___________________________^
192 | | let mut data_reader = crate::sources::oracle::OracleTextSourceDataReader::new(conn_url.unwrap().as_str(), sql.as_str());
193 | | data_reader.read_data_to_queue(tx);
194 | | });
| |^
note: required by a bound in tokio::spawn
--> C:\Users\tianchuan01.cargo\registry\src\mirrors.ustc.edu.cn-12df342d903acd47\tokio-1.40.0\src\task\spawn.rs:167:21
|
165 | pub fn spawn(future: F) -> JoinHandle<F::Output>
| ----- required by a bound in this function
166 | where
167 | F: Future + Send + 'static,
| ^^^^ required by this bound in spawn

[kubo]

Use Arc<Mutex<Vec<Row>>> instead. If you have no need to wrap Vec with Arc, use Vec<Row> instead of Arc<Vec<Row>>.

In order to transfer type across threads, Send must be implemented. Both Arc<Mutex<Vec<Row>>> and Vec<Row> implement Send but Arc<Vec<Row>> doesn't. That's because Arc<T> implements Send when T implements both Send and Sync but Vec<Row> doesn't implements Sync.

[tc19901016]

Thank you. however, I have another question.
I want to convert Vec to Vec. So, I will use rayon to parse every column by thread pool. The code like this: (0..column_len).into_par_iter().enumerate().(|(i, _)| rows.iter().map(|r| r.get(i)?).collect()).collect();
but the compiler show some error, like this code can't run in thread pool, error line: self.data = DpiData::Var(Rc::new(DpiVar::new(handle, data)));

can I update this line by using Arc::new....?
thank you.

[kubo]

You cannot. Though Sync must be implemented to share references between threads, Row doesn't.
Well, to be precise, you can wrap Row in Mutex to share references between threads but using it with threading will be slower than simple code without threading.

Though I'm not sure what is your case, it is better to use query_as instead of query in general. See https://github.com/kubo/rust-oracle/blob/master/docs/query-methods.md#with-and-without-_as

Another option is https://quietboil.github.io/sibyl/. It uses Oracle Call Interface directly. On the other hand rust-oracle uses the interface via ODPI-C. I'm not sure which is faster.

[tc19901016]

I tested millions rows data in oracle by using jdbc and rust-oracle.In different data type, some queries show rust is faster and the other show jdbc is faster. the record is below:

1. 30 columns(NUMBER(28.18)) * 5670,0000 rows

Rust-oracle:row.next()+row.get()(sec)	Rust-oracle:Just row.next()(sec)	jdbc:rs.next()+rs.getFloat()(sec)	jdbc:rs.next()(sec)
44	26	21	10.9

2. 30 columns(NUMBER(10.0)) * 5670,0000 rows

Rust-oracle:row.next()+row.get()(sec)	Rust-oracle:Just row.next()(sec)	jdbc:rs.next()+rs.getInt()(sec)	jdbc:rs.next()(sec)
24	21	20	10.4

3. 30 columns(Date) * 5670,0000 rows

Rust-oracle:row.next()+row.get()(sec)	Rust-oracle:Just row.next()(sec)	jdbc:rs.next()+rs.getDate()(sec)	jdbc:rs.next()(sec)
25	23	82	13

my boss believe there must be a way that rust read all type data is faster than jdbc. So, I must explain:

1. row.next() is slower than jdbc. (I need to test more, I think this is faster than jdbc but in fact, it is not.)
2. number is slower than jdbc.
   maybe other type(like varchar) is also slower, I will test tomorrow

all sql like:
select T_DATE AS A0,T_DATE AS A1,T_DATE AS A2,T_DATE AS A3,T_DATE AS A4,T_DATE AS A5,T_DATE AS A6,T_DATE AS A7,T_DATE AS A8,T_DATE AS A9,T_DATE AS A10,T_DATE AS A11,T_DATE AS A12,T_DATE AS A13,T_DATE AS A14,T_DATE AS A15,T_DATE AS A16,T_DATE AS A17,T_DATE AS A18,T_DATE AS A19,T_DATE AS A20,T_DATE AS A21,T_DATE AS A22,T_DATE AS A23,T_DATE AS A24,T_DATE AS A25,T_DATE AS A26,T_DATE AS A27,T_DATE AS A28,T_DATE AS A29 from tableA

test case is below:

1. Rust-oracle:row.next()+row.get()

fn oracle_test_iter(sql:&str) ->std::result::Result<(), Box<dyn Error>>{
    println!("oracle_test_iter_col sql:{}", sql);
    let manager = OracleConnectionManager::new("xxx", "xxxx", "1.1.1.1:1521/xxxx");
    let pool = r2d2::Pool::builder().max_size(1).build(manager).unwrap();
    let conn = pool.get().unwrap();
    for _ in 0..4{
        let start_time = Local::now();
        // let col_len = conn.query(sql, &[]).unwrap().column_info().len();
        let mut stmt = conn.statement(sql).prefetch_rows(1024).fetch_array_size(1024).build().unwrap();
        let rows = stmt.query(&[]).unwrap();
        let col_len = rows.column_info().len();
        
        let mut count=0;
        for row in rows{
            let r = row?;
            for i in 0..col_len {
                let x = r.get::<_,String>(i); //according column data type to update: get::<_, Timestamp>, get::<_, i64>, get::<_,f64>
            }
        }
        let duration = Local::now()-start_time;
        println!("duration：{}", duration);
    }
    Ok(())
}

2. Rust-oracle:row.next()

fn oracle_test_only_read(sql:&str) ->std::result::Result<(), Box<dyn Error>>{
    println!("oracle_test_only_read sql:{}", sql);
    let manager = OracleConnectionManager::new("xxx", "xxxx", "1.1.1.1:1521/xxxx");
    let pool = r2d2::Pool::builder().max_size(1).build(manager).unwrap();
    let conn = pool.get().unwrap();
    for _ in 0..10{
        let start_time = Local::now();
        // let col_len = conn.query(sql, &[]).unwrap().column_info().len();
        let mut stmt = conn.statement(sql).prefetch_rows(1024).fetch_array_size(1024).build().unwrap();
        let rows = stmt.query(&[]).unwrap();
        let col_len = rows.column_info().len();
        
        let mut count=0;
        for row in rows{
            // let r = row?;
            // for i in 0..col_len {
            //     let x = r.get::<_,String>(i);
            // }
            // count+=1;
        }
        let duration = Local::now()-start_time;
        println!("duraction：{}", duration);
    }
    println!();
    Ok(())
}

3. jdbc: rs.next()+rs.getFloat()

    private static void test_iter(String sql, int testNum, boolean onlineFlag) throws Exception{
        Connection conn = null;
        if(onlineFlag){
            conn = DriverManager.getConnection("jdbc:oracle:thin:@1.1.1.1:1521/xxx", "xxx", "xxx");
        }else{
            conn = DriverManager.getConnection("jdbc:oracle:thin:@1.1.1.1:1521/xxx", "xxx", "xxx");
        }
       System.out.println("test_iter"+sql);
        List<List<Object>> list = new ArrayList<>();
        for(int k=0;k<testNum;k++){
            long timestamp = System.currentTimeMillis();
            try (PreparedStatement ps = conn.prepareStatement(sql)) {
                ps.setFetchSize(1024);
                try (ResultSet rs = ps.executeQuery()) {
                    ResultSetMetaData metaData = rs.getMetaData();
                    int columnCount = metaData.getColumnCount();
                    while (rs.next()) {
                        for (int i = 0; i < columnCount; i++) {
                            float columnValue = rs.getFloat(i+1);//rs.getInt;rs.getDate
                        }
                    }
                }
            } catch (SQLException e) {
                e.printStackTrace();
            }
            long duration = System.currentTimeMillis() - timestamp;
            System.out.println("Finished:"+duration);
        }
    }

4)jdbc: rs.next()

private static void test_only_read(String sql, int testNum, boolean onlineFlag) throws Exception{
        Connection conn = null;
        if(onlineFlag){
            conn = DriverManager.getConnection("jdbc:oracle:thin:@1.1.1.1:1521/xxx", "xxx", "xxx");
        }else{
            conn = DriverManager.getConnection("jdbc:oracle:thin:@1.1.1.1:1521/xxx", "xxx", "xxx");
        }
        System.out.println("test_only_read"+sql);
        List<List<Object>> list = new ArrayList<>();
        for(int k=0;k<testNum;k++){
            long timestamp = System.currentTimeMillis();
            try (PreparedStatement ps = conn.prepareStatement(sql)) {
                ps.setFetchSize(1024);
                try (ResultSet rs = ps.executeQuery()) {
                    ResultSetMetaData metaData = rs.getMetaData();
                    int columnCount = metaData.getColumnCount();
                    while (rs.next()) {
//                        for (int i = 0; i < columnCount; i++) {
//                            String columnValue = rs.getString(i+1);
//                        }
                    }
                }
            } catch (SQLException e) {
                e.printStackTrace();
            }
            long duration = System.currentTimeMillis() - timestamp;
            System.out.println("Finished:"+duration);
        }
    }

[kubo]

In the flame graph of version 0.6.1-dev in #44 (comment), upirtrc, which is an internal function inside of an Oracle client library, takes 47.77% of samples. Could you make flame graph of your code? I guess that more than half of time is consumed in the Oracle library in the case of "Rust-oracle:Just row.next()"

As for the difference of seconds between NUMBER(28.18) and NUMBER(10.0), I guess that is caused by here.
When the column definition is NUMBER(10.0), the column values are converted from NUMBER to int64 inside of Oracle Call Interface (OCI). row.get<_, i64> just copies the int64 value.
On the other hand when the definition is NUMBER(28.18), the values are passed through the OCI, formatted as string representation in ODPI-C and then converted to requested type such f64 inside of rust-oracle. In the current implementation, the string values owned by ODPI-C are converted to f64 via String. I may improve it by using &str referring to the buffer in OCPI-C instead of temporal String owning data copied from the buffer.

[cjbj]

@tc19901016 Did you try slightly bigger arraysize values? For such a large number of rows I would have left prefetch size at its default.

@kubo ping us if you need help, or to nudge us about requests like oracle/odpi#172

[tc19901016]

@kubo You are right.
1)In rust-readonly test(test case 2), dpiStmt_fetchRows cost 56%, RowValue::get and drop cost the others. I think that's why rust-readonly test slower than jdbc(maybe jdbc only run like stmt.next())

2)In rust-itercol test(test case 1), Number(28,18) increase get_string_unchecked cost than Number(10).
f64 flamegraph:

i64 flamegraph:

Though flamegraph is created, I have no idea to increase reading speed. I will try to modify this project. if I have some results, I will write a comment .

[kubo]

@tc19901016
I committed https://github.com/kubo/rust-oracle-private/commit/0c495534a0c454ead9f7d270032666995a6a02c3, which doesn't use temporary strings to fetch number where scale isn't zero.
Could you use the master branch on github?

[tc19901016]

No,I can't open rust-oracle-private page:

By the way, I tried read data through odpi directly. But, some step I don't know how to write.

dpiConn_create();
dpiConn_prepareStmt();
dpiStmt_setFetchArraySize();
for i in 0..column_number {
  dpiConn_newVar();
  dpiStmt_define();
}
loop{
  dpiStmt_fetchRows();
  for column_indxe in 0..column_number{
    //I don't know how to query Vector of column data
    //.......
  }
}

this issue(oracle/odpi#70) tell me the preparing steps before dpiStmt_fetchRows(), does not tell me how to get data.

[kubo]

No,I can't open rust-oracle-private page:

Sorry, I posted wrong URL. It was pushed to public repository also. 0c49553

[cjbj]

@kubo ODPI-C 5.4 will be tagged very soon

[tc19901016]

I have tested Number(28,18) columns case(30 col * 5670,0000 rows). Commit<0c49553> can improve about 40% get::<, f64> speed. According the below table, old version cost 44-27=17 seconds in get::<, f64>, and new version cost 37-27=10 seconds.

version old: 0.6.2	version new: 0c49553	only-read
44	37	27

old version flamegraph:

new version flamegraph:

[tc19901016]

I tried to use odpi interface derictly to fetch rows. But, I only get the first row's data. I don't know how to iterate the dpiVar(every column has a dpiVar) to get the list of column data.

pub fn test() -> Result<(), Box<dyn std::error::Error>> {
        unsafe {
            //0. params
            let query_string = OdpiStr::new("select * from (select * from (select * from TEST_TABLE) where rownum<100) order by id1");
            let col_num =9u32;

            //1. init conn
            let ctxt = Context::new()?;
            let common_params = ctxt.common_create_params();
            let mut conn_params = ctxt.conn_create_params();
            let username = OdpiStr::new("test");
            let password = OdpiStr::new("****");
            let connect_string = OdpiStr::new("192.168.1.56:1521/ORCL");
            let mut handle = ptr::null_mut();
            Self::check_error(dpiConn_create(ctxt.context, username.ptr, username.len, password.ptr, password.len, connect_string.ptr, connect_string.len,&common_params,
                                       &mut conn_params,&mut handle) as i32);


            //2.init stmt
            let tag = OdpiStr::new("");
            let mut stmt_handle = ptr::null_mut();
            let mut num_query_columns=0;
            Self::check_error(dpiConn_prepareStmt(
                handle,
                i32::from(false),
                query_string.ptr,
                query_string.len,
                tag.ptr,
                tag.len,
                &mut stmt_handle
            ));
            let fetch_array_size = 1024u32;
            Self::check_error(dpiStmt_setFetchArraySize(stmt_handle, fetch_array_size));
            Self::check_error(dpiStmt_execute(stmt_handle, DPI_MODE_EXEC_DEFAULT, &mut num_query_columns));
            //2.1 define column
            let mut newVarHandles = vec![];
            for i in 0..col_num{
                let mut var_handle = ptr::null_mut();
                let mut var_data = ptr::null_mut();
                Self::check_error(dpiConn_newVar(
                    handle,
                    DPI_ORACLE_TYPE_NUMBER,
                    DPI_NATIVE_TYPE_INT64,
                    1024,
                    0,
                    0,
                    0,
                    ptr::null_mut(), 
                    &mut var_handle,
                    &mut var_data,
                ));
                Self::check_error(dpiStmt_define(stmt_handle, i+1, var_handle));
                newVarHandles.push((i, var_handle, var_data));
            }

            //3.fetch rows
            let mut new_index = 0;
            let mut num_rows = 0;
            let mut more_rows = 0;
            loop {
                Self::check_error(dpiStmt_fetchRows(stmt_handle,fetch_array_size,
                                                    &mut new_index,&mut num_rows,&mut more_rows
                ));
                for n in 0..100{
                    for col_index in 0..col_num{
                        let (index, var_handle, var_data) = newVarHandles[col_index as usize];
                        print!("{},",dpiData_getInt64(var_data));
                    }
                    println!("");
                    //todo how to fetch next row's data. thank you. 
                }
                if more_rows == 0 {break;}
            }
        }
        Ok(())
    }

thank you.
I have tried many times, but all are failed. There are too few example about DpiStmt_fetchRows on the Internet.

[kubo]

I've not tested the following code.

                for offset in new_index..(new_index + num_rows) {
                    for col_index in 0..col_num {
                        let (index, var_handle, var_data) = newVarHandles[col_index as usize];
                        let var_data = var_data.offset(offset as isize);
                        print!("{},",dpiData_getInt64(var_data));
                    }
                    println!("");
                    //todo how to fetch next row's data. thank you. 
                }

[kubo]

I committed e64754f.

It skips the following UTF-8 validation.

[kubo]

Could you use query_as<TYPE>() instead of query()?
The following functions consumed in the flame graph of 0c49553 are not used by query_as<TYPE>().

• oracle::sql_value::SqlValue::clone_except_fetch_array_buffer (964 samples, 8.20%)
• core::ptr::drop_in_place<oracle::sql_value::SqlValue> (636 samples, 5.41%)

The following code fetch row as a tuple containing 30 f64 values.

        let rows = stmt.query_as<(f64, f64, f64, f64, f64, f64, f64, f64, f64, f64, f64, f64, f64, f64, f64, f64, f64, f64, f64, f64, f64, f64, f64, f64, f64, f64, f64, f64, f64, f64)>(&[]).unwrap();
        let col_len = rows.column_info().len();
        
        let mut count=0;
        for row in rows{
            let r = row?;
            count+=1;
        }

[tc19901016]

I tested f6430columns5670,0000 in e64754f, result is below:

Rust-oracle:query_as(sec)	Rust-oracle:row.next()+row.get()(sec)	Rust-oracle:Just row.next()(sec)	jdbc:rs.next()+rs.getBigDecimal()(sec)	jdbc:rs.next()(sec)
27	34	27	22	10.5

test summary:

1. dpiStmt_fetchRows cost 65% in query_as case, almost 16 seconds. It's slower than jdbc only_read case;
   However, the efficiency of dpiStmt_fetchRows is odpi's business.
2. query_as is 20% faster than query.
3. In dpiSmt_fetchRows:
   1-step: stmtFetch2, cost 11sec
   2-step: getValue(run fromOracleNumberAsText) cost 4.8sec(...AsText cost 2.7sec)
   maybe, 2-step is the reason why dpiStmt_fetchRows is slower than jdbc, and rust-oracle has to convert str to num again.

Thank you, I think query_as is very efficient. I will try to use query_as to read data from oracle.

query_as:

only_read:

query: