async fn batch_get<St: Storable>(
        &self,
        ids: &[St::Key],
    ) -> core::result::Result<Vec<DbRecord>, StorageError> {
        let mut map = Vec::new();

        if ids.is_empty() {
            // nothing to retrieve, save the cycles
            return Ok(map);
        }

        let mut key_set: HashSet<St::Key> = ids.iter().cloned().collect::<HashSet<_>>();

        let trans_active = self.is_transaction_active().await;
        // first check the transaction log & cache records
        for id in ids.iter() {
            if trans_active {
                // we're in a transaction, meaning the object _might_ be newer and therefore we should try and read if from the transaction
                // log instead of the raw storage layer
                if let Some(result) = self.trans.get::<St>(id).await {
                    map.push(result);
                    key_set.remove(id);
                    continue;
                }
            }

            // check if item is cached
            if let Some(cache) = &self.cache {
                if let Some(result) = cache.hit_test::<St>(id).await {
                    map.push(result);
                    key_set.remove(id);
                    continue;
                }
            }
        }

        if !key_set.is_empty() {
            // these are items to be retrieved from the backing database (not in pending transaction or in the object cache)
            let result = async {
                let tic = Instant::now();

                let key_set_vec: Vec<_> = key_set.into_iter().collect();

                debug!("BEGIN MySQL get batch");
                let mut conn = self.get_connection().await?;

                let results = if let Some(create_table_cmd) =
                    DbRecord::get_batch_create_temp_table::<St>()
                {
                    // Create the temp table of ids
                    let out = conn.query_drop(create_table_cmd).await;
                    self.check_for_infra_error(out)?;

                    // Fill temp table with the requested ids
                    let mut tx = conn.start_transaction(TxOpts::default()).await?;
                    tx.query_drop("SET autocommit=0").await?;
                    tx.query_drop("SET unique_checks=0").await?;
                    tx.query_drop("SET foreign_key_checks=0").await?;

                    let mut fallout: Option<Vec<_>> = None;
                    let mut params = vec![];
                    for batch in key_set_vec.chunks(self.tunable_insert_depth) {
                        if batch.len() < self.tunable_insert_depth {
                            fallout = Some(batch.to_vec());
                        } else if let Some(p) = DbRecord::get_multi_row_specific_params::<St>(batch)
                        {
                            params.push(p);
                        } else {
                            return Err(MySqlError::Other(
                                "Unable to generate type-specific MySQL parameters".into(),
                            ));
                        }
                    }

                    // insert the batches of size = MYSQL_EXTENDED_INSERT_DEPTH
                    if !params.is_empty() {
                        let fill_statement = DbRecord::get_batch_fill_temp_table::<St>(Some(
                            self.tunable_insert_depth,
                        ));
                        let out = tx.exec_batch(fill_statement, params).await;
                        self.check_for_infra_error(out)?;
                        // We would need the statement for it. (Possibly) No need for close here.
                        // See https://docs.rs/mysql_async/0.28.1/mysql_async/struct.Opts.html#caveats.
                        // tx.close().await?;
                    }

                    // insert the remainder as a final statement
                    if let Some(remainder) = fallout {
                        let remainder_stmt =
                            DbRecord::get_batch_fill_temp_table::<St>(Some(remainder.len()));
                        let params_batch =
                            DbRecord::get_multi_row_specific_params::<St>(&remainder);
                        if let Some(pb) = params_batch {
                            let out = tx.exec_drop(remainder_stmt, pb).await;
                            self.check_for_infra_error(out)?;
                        } else {
                            return Err(MySqlError::Other(
                                "Unable to generate type-specific MySQL parameters".into(),
                            ));
                        }
                    }

                    tx.query_drop("SET autocommit=1").await?;
                    tx.query_drop("SET unique_checks=1").await?;
                    tx.query_drop("SET foreign_key_checks=1").await?;
                    tx.commit().await?;

                    // Query the records which intersect (INNER JOIN) with the temp table of ids
                    let query = DbRecord::get_batch_statement::<St>();
                    let out = conn.query_iter(query).await;
                    let result = self.check_for_infra_error(out)?;

                    let out = result
                        .reduce_and_drop(vec![], |mut acc, mut row| {
                            if let Ok(result) = DbRecord::from_row::<St>(&mut row) {
                                acc.push(result);
                            }
                            acc
                        })
                        .await?;

                    // drop the temp table of ids
                    let t_out = conn
                        .query_drop(format!("DROP TEMPORARY TABLE `{}`", TEMP_IDS_TABLE))
                        .await;
                    self.check_for_infra_error(t_out)?;

                    out
                } else {
                    // no results (i.e. AZKS table doesn't support "get by batch ids")
                    vec![]
                };

                debug!("END MySQL get batch");
                let toc = Instant::now() - tic;
                *(self.time_read.write().await) += toc;

                if let Some(cache) = &self.cache {
                    // insert retrieved records into the cache for faster future access
                    for el in results.iter() {
                        cache.put(el).await;
                    }
                }

                Ok::<Vec<DbRecord>, mysql_async::Error>(results)
            };

            *(self.num_reads.write().await) += 1;
            self.record_call_stats(
                'r',
                "batch_get".to_string(),
                format!("{:?}", St::data_type()),
            )
            .await;

            match result.await {
                Ok(result_vec) => {
                    for item in result_vec.into_iter() {
                        map.push(item);
                    }
                }
                Err(error) => {
                    error!("MySQL error {}", error);
                    return Err(StorageError::Other(format!("MySQL Error {}", error)));
                }
            }
        }
        Ok(map)
    }
/// Retrieve a batch of records by id
Retrieve a batch of records by id
Retrieve a batch of records by id
async fn batch_get<St: Storable>(
        &self,
        ids: &[St::Key],
    ) -> core::result::Result<Vec<DbRecord>, StorageError> {
        let mut map = Vec::new();
        if ids.is_empty() {
            // nothing to retrieve, save the cycles
            return Ok(map);
        }
        let mut key_set: HashSet<St::Key> = ids.iter().cloned().collect::<HashSet<_>>();
        let trans_active = self.is_transaction_active().await;
        // first check the transaction log & cache records
        for id in ids.iter() {
            if trans_active {
                // we're in a transaction, meaning the object _might_ be newer and therefore we should try and read if from the transaction
                // log instead of the raw storage layer
                if let Some(result) = self.trans.get::<St>(id).await {
                    map.push(result);
                    key_set.remove(id);
                    continue;
                }
            }
            // check if item is cached
            if let Some(cache) = &self.cache {
                if let Some(result) = cache.hit_test::<St>(id).await {
                    map.push(result);
                    key_set.remove(id);
                    continue;
                }
            }
        }
        if !key_set.is_empty() {
            // these are items to be retrieved from the backing database (not in pending transaction or in the object cache)
            let result = async {
                let tic = Instant::now();
                let key_set_vec: Vec<_> = key_set.into_iter().collect();
                debug!("BEGIN MySQL get batch");
                let mut conn = self.get_connection().await?;
                let results = if let Some(create_table_cmd) =
                    DbRecord::get_batch_create_temp_table::<St>()
                {
                    // Create the temp table of ids
                    let out = conn.query_drop(create_table_cmd).await;
                    self.check_for_infra_error(out)?;
                    // Fill temp table with the requested ids
                    let mut tx = conn.start_transaction(TxOpts::default()).await?;
                    tx.query_drop("SET autocommit=0").await?;
                    tx.query_drop("SET unique_checks=0").await?;
                    tx.query_drop("SET foreign_key_checks=0").await?;
                    let mut fallout: Option<Vec<_>> = None;
                    let mut params = vec![];
                    for batch in key_set_vec.chunks(self.tunable_insert_depth) {
                        if batch.len() < self.tunable_insert_depth {
                            fallout = Some(batch.to_vec());
                        } else if let Some(p) = DbRecord::get_multi_row_specific_params::<St>(batch)
                        {
                            params.push(p);
                        } else {
                            return Err(MySqlError::Other(
                                "Unable to generate type-specific MySQL parameters".into(),
                            ));
                        }
                    }
                    // insert the batches of size = MYSQL_EXTENDED_INSERT_DEPTH
                    if !params.is_empty() {
                        let fill_statement = DbRecord::get_batch_fill_temp_table::<St>(Some(
                            self.tunable_insert_depth,
                        ));
                        let out = tx.exec_batch(fill_statement, params).await;
                        self.check_for_infra_error(out)?;
                        // We would need the statement for it. (Possibly) No need for close here.
                        // See https://docs.rs/mysql_async/0.28.1/mysql_async/struct.Opts.html#caveats.
                        // tx.close().await?;
                    }
                    // insert the remainder as a final statement
                    if let Some(remainder) = fallout {
                        let remainder_stmt =
                            DbRecord::get_batch_fill_temp_table::<St>(Some(remainder.len()));
                        let params_batch =
                            DbRecord::get_multi_row_specific_params::<St>(&remainder);
                        if let Some(pb) = params_batch {
                            let out = tx.exec_drop(remainder_stmt, pb).await;
                            self.check_for_infra_error(out)?;
                        } else {
                            return Err(MySqlError::Other(
                                "Unable to generate type-specific MySQL parameters".into(),
                            ));
                        }
                    }
                    tx.query_drop("SET autocommit=1").await?;
                    tx.query_drop("SET unique_checks=1").await?;
                    tx.query_drop("SET foreign_key_checks=1").await?;
                    tx.commit().await?;
                    // Query the records which intersect (INNER JOIN) with the temp table of ids
                    let query = DbRecord::get_batch_statement::<St>();
                    let out = conn.query_iter(query).await;
                    let result = self.check_for_infra_error(out)?;
                    let out = result
                        .reduce_and_drop(vec![], |mut acc, mut row| {
                            if let Ok(result) = DbRecord::from_row::<St>(&mut row) {
                                acc.push(result);
                            }
                            acc
                        })
                        .await?;
                    // drop the temp table of ids
                    let t_out = conn
                        .query_drop(format!("DROP TEMPORARY TABLE `{}`", TEMP_IDS_TABLE))
                        .await;
                    self.check_for_infra_error(t_out)?;
                    out
                } else {
                    // no results (i.e. AZKS table doesn't support "get by batch ids")
                    vec![]
                };
                debug!("END MySQL get batch");
                let toc = Instant::now() - tic;
                *(self.time_read.write().await) += toc;
                if let Some(cache) = &self.cache {
                    // insert retrieved records into the cache for faster future access
                    for el in results.iter() {
                        cache.put(el).await;
                    }
                }
                Ok::<Vec<DbRecord>, mysql_async::Error>(results)
            };
            *(self.num_reads.write().await) += 1;
            self.record_call_stats(
                'r',
                "batch_get".to_string(),
                format!("{:?}", St::data_type()),
            )
            .await;
            match result.await {
                Ok(result_vec) => {
                    for item in result_vec.into_iter() {
                        map.push(item);
                    }
                }
                Err(error) => {
                    error!("MySQL error {}", error);
                    return Err(StorageError::Other(format!("MySQL Error {}", error)));
                }
            }
        }
        Ok(map)
    }
async fn batch_get < St : Storable > ( & self , ids : & [ St :: Key ] , ) -> core :: result :: Result < Vec < DbRecord > , StorageError > { let mut map = Vec :: new ( ) ; if ids . is_empty ( ) { // nothing to retrieve, save the cycles return Ok ( map ) ; } let mut key_set : HashSet < St :: Key > = ids . iter ( ) . cloned ( ) . collect :: < HashSet < _ > > ( ) ; let trans_active = self . is_transaction_active ( ) . await ; // first check the transaction log & cache records for id in ids . iter ( ) { if trans_active { // we're in a transaction, meaning the object _might_ be newer and therefore we should try and read if from the transaction // log instead of the raw storage layer if let Some ( result ) = self . trans . get :: < St > ( id ) . await { map . push ( result ) ; key_set . remove ( id ) ; continue ; } } // check if item is cached if let Some ( cache ) = & self . cache { if let Some ( result ) = cache . hit_test :: < St > ( id ) . await { map . push ( result ) ; key_set . remove ( id ) ; continue ; } } } if ! key_set . is_empty ( ) { // these are items to be retrieved from the backing database (not in pending transaction or in the object cache) let result = async { let tic = Instant :: now ( ) ; let key_set_vec : Vec < _ > = key_set . into_iter ( ) . collect ( ) ; debug ! ( " " ) ; let mut conn = self . get_connection ( ) . await ? ; let results = if let Some ( create_table_cmd ) = DbRecord :: get_batch_create_temp_table :: < St > ( ) { // Create the temp table of ids let out = conn . query_drop ( create_table_cmd ) . await ; self . check_for_infra_error ( out ) ? ; // Fill temp table with the requested ids let mut tx = conn . start_transaction ( TxOpts :: default ( ) ) . await ? ; tx . query_drop ( " " ) . await ? ; tx . query_drop ( " " ) . await ? ; tx . query_drop ( " " ) . await ? ; let mut fallout : Option < Vec < _ > > = None ; let mut params = vec ! [ ] ; for batch in key_set_vec . chunks ( self . tunable_insert_depth ) { if batch . len ( ) < self . tunable_insert_depth { fallout = Some ( batch . to_vec ( ) ) ; } else if let Some ( p ) = DbRecord :: get_multi_row_specific_params :: < St > ( batch ) { params . push ( p ) ; } else { return Err ( MySqlError :: Other ( " " . into ( ) , ) ) ; } } // insert the batches of size = MYSQL_EXTENDED_INSERT_DEPTH if ! params . is_empty ( ) { let fill_statement = DbRecord :: get_batch_fill_temp_table :: < St > ( Some ( self . tunable_insert_depth , ) ) ; let out = tx . exec_batch ( fill_statement , params ) . await ; self . check_for_infra_error ( out ) ? ; // We would need the statement for it. (Possibly) No need for close here. // See https://docs.rs/mysql_async/0.28.1/mysql_async/struct.Opts.html#caveats. // tx.close().await?; } // insert the remainder as a final statement if let Some ( remainder ) = fallout { let remainder_stmt = DbRecord :: get_batch_fill_temp_table :: < St > ( Some ( remainder . len ( ) ) ) ; let params_batch = DbRecord :: get_multi_row_specific_params :: < St > ( & remainder ) ; if let Some ( pb ) = params_batch { let out = tx . exec_drop ( remainder_stmt , pb ) . await ; self . check_for_infra_error ( out ) ? ; } else { return Err ( MySqlError :: Other ( " " . into ( ) , ) ) ; } } tx . query_drop ( " " ) . await ? ; tx . query_drop ( " " ) . await ? ; tx . query_drop ( " " ) . await ? ; tx . commit ( ) . await ? ; // Query the records which intersect (INNER JOIN) with the temp table of ids let query = DbRecord :: get_batch_statement :: < St > ( ) ; let out = conn . query_iter ( query ) . await ; let result = self . check_for_infra_error ( out ) ? ; let out = result . reduce_and_drop ( vec ! [ ] , | mut acc , mut row | { if let Ok ( result ) = DbRecord :: from_row :: < St > ( & mut row ) { acc . push ( result ) ; } acc } ) . await ? ; // drop the temp table of ids let t_out = conn . query_drop ( format ! ( " " TEMP_IDS_TABLE ) ) . await ; self . check_for_infra_error ( t_out ) ? ; out } else { // no results (i.e. AZKS table doesn't support "get by batch ids") vec ! [ ] } ; debug ! ( " " ) ; let toc = Instant :: now ( ) - tic ; * ( self . time_read . write ( ) . await ) += toc ; if let Some ( cache ) = & self . cache { // insert retrieved records into the cache for faster future access for el in results . iter ( ) { cache . put ( el ) . await ; } } Ok :: < Vec < DbRecord > , mysql_async :: Error > ( results ) } ; * ( self . num_reads . write ( ) . await ) += 1 ; self . record_call_stats ( 'r' , " " . to_string ( ) , format ! ( " " St data_type ( ) ) , ) . await ; match result . await { Ok ( result_vec ) => { for item in result_vec . into_iter ( ) { map . push ( item ) ; } } Err ( error ) => { error ! ( " " error ) ; return Err ( StorageError :: Other ( format ! ( " " error ) ) ) ; } } } Ok ( map ) }
Retrieve a batch of records by id
Retrieve a batch of records by id
&self
ids
&[St::Key]
&self
ids
&[St::Key]