async fn internal_batch_set(
        &self,
        records: Vec<DbRecord>,
        mut trans: mysql_async::Transaction<'a>,
    ) -> core::result::Result<mysql_async::Transaction<'a>, MySqlError> {
        if records.is_empty() {
            return Ok(trans);
        }

        *(self.num_writes.write().await) += records.len() as u64;
        self.record_call_stats('w', "internal_batch_set".to_string(), "".to_string())
            .await;

        debug!("BEGIN Computing mysql parameters");
        #[allow(clippy::needless_collect)]
        let chunked = records
            .chunks(self.tunable_insert_depth)
            .map(|batch| {
                if batch.is_empty() {
                    Ok(BatchMode::None)
                } else if batch.len() < self.tunable_insert_depth {
                    DbRecord::set_batch_params(batch)
                        .map(|out| BatchMode::Partial(out, batch.len()))
                } else {
                    DbRecord::set_batch_params(batch).map(BatchMode::Full)
                }
            })
            .collect::<Result<Vec<_>>>()?;
        debug!("END Computing mysql parameters");

        debug!("BEGIN MySQL set batch");
        let head = &records[0];
        let statement = |i: usize| -> String {
            match &head {
                DbRecord::Azks(_) => DbRecord::set_batch_statement::<akd::Azks>(i),
                DbRecord::HistoryNodeState(_) => {
                    DbRecord::set_batch_statement::<HistoryNodeState>(i)
                }
                DbRecord::HistoryTreeNode(_) => DbRecord::set_batch_statement::<HistoryTreeNode>(i),
                DbRecord::ValueState(_) => {
                    DbRecord::set_batch_statement::<akd::storage::types::ValueState>(i)
                }
            }
        };

        let mut params = vec![];
        let mut fallout: Option<(mysql_async::Params, usize)> = None;
        for item in chunked {
            match item {
                BatchMode::Full(part) => params.push(part),
                BatchMode::Partial(part, count) => fallout = Some((part, count)),
                _ => {}
            }
        }

        let tic = Instant::now();

        debug!("MySQL batch - {} full inserts", params.len());
        // insert the batches of size = MYSQL_EXTENDED_INSERT_DEPTH
        if !params.is_empty() {
            let fill_statement = statement(self.tunable_insert_depth);
            let out = trans.exec_batch(fill_statement, params).await;
            self.check_for_infra_error(out)?;
        }

        // insert the remainder as a final statement
        if let Some((remainder, count)) = fallout {
            debug!("MySQL batch - remainder {} insert", count);
            let remainder_stmt = statement(count);
            let out = trans.exec_drop(remainder_stmt, remainder).await;
            self.check_for_infra_error(out)?;
        }

        let toc = Instant::now() - tic;
        *(self.time_write.write().await) += toc;
        debug!("END MySQL set batch");
        Ok(trans)
    }
/// NOTE: This is assuming all of the DB records have been narrowed down to a single record type!
This is assuming all of the DB records have been narrowed down to a single record type!
This is assuming all of the DB records have been narrowed down to a single record type!
async fn internal_batch_set(
        &self,
        records: Vec<DbRecord>,
        mut trans: mysql_async::Transaction<'a>,
    ) -> core::result::Result<mysql_async::Transaction<'a>, MySqlError> {
        if records.is_empty() {
            return Ok(trans);
        }
        *(self.num_writes.write().await) += records.len() as u64;
        self.record_call_stats('w', "internal_batch_set".to_string(), "".to_string())
            .await;
        debug!("BEGIN Computing mysql parameters");
        #[allow(clippy::needless_collect)]
        let chunked = records
            .chunks(self.tunable_insert_depth)
            .map(|batch| {
                if batch.is_empty() {
                    Ok(BatchMode::None)
                } else if batch.len() < self.tunable_insert_depth {
                    DbRecord::set_batch_params(batch)
                        .map(|out| BatchMode::Partial(out, batch.len()))
                } else {
                    DbRecord::set_batch_params(batch).map(BatchMode::Full)
                }
            })
            .collect::<Result<Vec<_>>>()?;
        debug!("END Computing mysql parameters");
        debug!("BEGIN MySQL set batch");
        let head = &records[0];
        let statement = |i: usize| -> String {
            match &head {
                DbRecord::Azks(_) => DbRecord::set_batch_statement::<akd::Azks>(i),
                DbRecord::HistoryNodeState(_) => {
                    DbRecord::set_batch_statement::<HistoryNodeState>(i)
                }
                DbRecord::HistoryTreeNode(_) => DbRecord::set_batch_statement::<HistoryTreeNode>(i),
                DbRecord::ValueState(_) => {
                    DbRecord::set_batch_statement::<akd::storage::types::ValueState>(i)
                }
            }
        };
        let mut params = vec![];
        let mut fallout: Option<(mysql_async::Params, usize)> = None;
        for item in chunked {
            match item {
                BatchMode::Full(part) => params.push(part),
                BatchMode::Partial(part, count) => fallout = Some((part, count)),
                _ => {}
            }
        }
        let tic = Instant::now();
        debug!("MySQL batch - {} full inserts", params.len());
        // insert the batches of size = MYSQL_EXTENDED_INSERT_DEPTH
        if !params.is_empty() {
            let fill_statement = statement(self.tunable_insert_depth);
            let out = trans.exec_batch(fill_statement, params).await;
            self.check_for_infra_error(out)?;
        }
        // insert the remainder as a final statement
        if let Some((remainder, count)) = fallout {
            debug!("MySQL batch - remainder {} insert", count);
            let remainder_stmt = statement(count);
            let out = trans.exec_drop(remainder_stmt, remainder).await;
            self.check_for_infra_error(out)?;
        }
        let toc = Instant::now() - tic;
        *(self.time_write.write().await) += toc;
        debug!("END MySQL set batch");
        Ok(trans)
    }
async fn internal_batch_set ( & self , records : Vec < DbRecord > , mut trans : mysql_async :: Transaction < ' a > , ) -> core :: result :: Result < mysql_async :: Transaction < ' a > , MySqlError > { if records . is_empty ( ) { return Ok ( trans ) ; } * ( self . num_writes . write ( ) . await ) += records . len ( ) as u64 ; self . record_call_stats ( 'w' , " " . to_string ( ) , " " . to_string ( ) ) . await ; debug ! ( " " ) ; # [ allow ( clippy needless_collect ) ] let chunked = records . chunks ( self . tunable_insert_depth ) . map ( | batch | { if batch . is_empty ( ) { Ok ( BatchMode :: None ) } else if batch . len ( ) < self . tunable_insert_depth { DbRecord :: set_batch_params ( batch ) . map ( | out | BatchMode :: Partial ( out , batch . len ( ) ) ) } else { DbRecord :: set_batch_params ( batch ) . map ( BatchMode :: Full ) } } ) . collect :: < Result < Vec < _ > > > ( ) ? ; debug ! ( " " ) ; debug ! ( " " ) ; let head = & records [ 0 ] ; let statement = | i : usize | -> String { match & head { DbRecord :: Azks ( _ ) => DbRecord :: set_batch_statement :: < akd :: Azks > ( i ) , DbRecord :: HistoryNodeState ( _ ) => { DbRecord :: set_batch_statement :: < HistoryNodeState > ( i ) } DbRecord :: HistoryTreeNode ( _ ) => DbRecord :: set_batch_statement :: < HistoryTreeNode > ( i ) , DbRecord :: ValueState ( _ ) => { DbRecord :: set_batch_statement :: < akd :: storage :: types :: ValueState > ( i ) } } } ; let mut params = vec ! [ ] ; let mut fallout : Option < ( mysql_async :: Params , usize ) > = None ; for item in chunked { match item { BatchMode :: Full ( part ) => params . push ( part ) , BatchMode :: Partial ( part , count ) => fallout = Some ( ( part , count ) ) , _ => { } } } let tic = Instant :: now ( ) ; debug ! ( " " params len ( ) ) ; // insert the batches of size = MYSQL_EXTENDED_INSERT_DEPTH if ! params . is_empty ( ) { let fill_statement = statement ( self . tunable_insert_depth ) ; let out = trans . exec_batch ( fill_statement , params ) . await ; self . check_for_infra_error ( out ) ? ; } // insert the remainder as a final statement if let Some ( ( remainder , count ) ) = fallout { debug ! ( " " count ) ; let remainder_stmt = statement ( count ) ; let out = trans . exec_drop ( remainder_stmt , remainder ) . await ; self . check_for_infra_error ( out ) ? ; } let toc = Instant :: now ( ) - tic ; * ( self . time_write . write ( ) . await ) += toc ; debug ! ( " " ) ; Ok ( trans ) }
NOTE: This is assuming all of the DB records have been narrowed down to a single record type!
NOTE : This is assuming all of the DB records have been narrowed down to a single record type!
&self
records
Vec<DbRecord>
trans
mysql_async::Transaction<'a>
&self
records
Vec<DbRecord>
trans
mysql_async::Transaction<'a>