Skip to content

Commit 08a73a3

Browse files
jxsjxs
committed
update migrate Transaction and AsyncTransaction
execute functions, to avoid double iteration.
1 parent d3fdf34 commit 08a73a3

File tree

9 files changed

+73
-35
lines changed

9 files changed

+73
-35
lines changed

‎refinery_core/src/drivers/config.rs

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,10 @@ use std::convert::Infallible;
1818
impl Transaction for Config {
1919
type Error = Infallible;
2020

21-
fn execute(&mut self, _queries: &[&str]) -> Result<usize, Self::Error> {
21+
fn execute<'a, T: Iterator<Item = &'a str>>(
22+
&mut self,
23+
_queries: T,
24+
) -> Result<usize, Self::Error> {
2225
Ok(0)
2326
}
2427
}
@@ -33,7 +36,10 @@ impl Query<Vec<Migration>> for Config {
3336
impl AsyncTransaction for Config {
3437
type Error = Infallible;
3538

36-
async fn execute(&mut self, _queries: &[&str]) -> Result<usize, Self::Error> {
39+
async fn execute<'a, T: Iterator<Item = &'a str> + Send>(
40+
&mut self,
41+
_queries: T,
42+
) -> Result<usize, Self::Error> {
3743
Ok(0)
3844
}
3945
}

‎refinery_core/src/drivers/mysql.rs

Lines changed: 10 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -43,10 +43,13 @@ fn query_applied_migrations(
4343
impl Transaction for Conn {
4444
type Error = MError;
4545

46-
fn execute(&mut self, queries: &[&str]) -> Result<usize, Self::Error> {
46+
fn execute<'a, T: Iterator<Item = &'a str>>(
47+
&mut self,
48+
queries: T,
49+
) -> Result<usize, Self::Error> {
4750
let mut transaction = self.start_transaction(get_tx_opts())?;
4851
let mut count = 0;
49-
for query in queries.iter() {
52+
for query in queries {
5053
transaction.query_iter(query)?;
5154
count += 1;
5255
}
@@ -58,11 +61,14 @@ impl Transaction for Conn {
5861
impl Transaction for PooledConn {
5962
type Error = MError;
6063

61-
fn execute(&mut self, queries: &[&str]) -> Result<usize, Self::Error> {
64+
fn execute<'a, T: Iterator<Item = &'a str>>(
65+
&mut self,
66+
queries: T,
67+
) -> Result<usize, Self::Error> {
6268
let mut transaction = self.start_transaction(get_tx_opts())?;
6369
let mut count = 0;
6470

65-
for query in queries.iter() {
71+
for query in queries {
6672
transaction.query_iter(query)?;
6773
count += 1;
6874
}

‎refinery_core/src/drivers/mysql_async.rs

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -40,15 +40,18 @@ async fn query_applied_migrations<'a>(
4040
impl AsyncTransaction for Pool {
4141
type Error = MError;
4242

43-
async fn execute(&mut self, queries: &[&str]) -> Result<usize, Self::Error> {
43+
async fn execute<'a, T: Iterator<Item = &'a str> + Send>(
44+
&mut self,
45+
queries: T,
46+
) -> Result<usize, Self::Error> {
4447
let mut conn = self.get_conn().await?;
4548
let mut options = TxOpts::new();
4649
options.with_isolation_level(Some(IsolationLevel::ReadCommitted));
4750

4851
let mut transaction = conn.start_transaction(options).await?;
4952
let mut count = 0;
5053
for query in queries {
51-
transaction.query_drop(*query).await?;
54+
transaction.query_drop(query).await?;
5255
count += 1;
5356
}
5457
transaction.commit().await?;

‎refinery_core/src/drivers/postgres.rs

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -33,10 +33,13 @@ fn query_applied_migrations(
3333
impl Transaction for PgClient {
3434
type Error = PgError;
3535

36-
fn execute(&mut self, queries: &[&str]) -> Result<usize, Self::Error> {
36+
fn execute<'a, T: Iterator<Item = &'a str>>(
37+
&mut self,
38+
queries: T,
39+
) -> Result<usize, Self::Error> {
3740
let mut transaction = PgClient::transaction(self)?;
3841
let mut count = 0;
39-
for query in queries.iter() {
42+
for query in queries {
4043
PgTransaction::batch_execute(&mut transaction, query)?;
4144
count += 1;
4245
}

‎refinery_core/src/drivers/rusqlite.rs

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -32,10 +32,13 @@ fn query_applied_migrations(
3232

3333
impl Transaction for RqlConnection {
3434
type Error = RqlError;
35-
fn execute(&mut self, queries: &[&str]) -> Result<usize, Self::Error> {
35+
fn execute<'a, T: Iterator<Item = &'a str>>(
36+
&mut self,
37+
queries: T,
38+
) -> Result<usize, Self::Error> {
3639
let transaction = self.transaction()?;
3740
let mut count = 0;
38-
for query in queries.iter() {
41+
for query in queries {
3942
transaction.execute_batch(query)?;
4043
count += 1;
4144
}

‎refinery_core/src/drivers/tiberius.rs

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -47,13 +47,16 @@ where
4747
{
4848
type Error = Error;
4949

50-
async fn execute(&mut self, queries: &[&str]) -> Result<usize, Self::Error> {
50+
async fn execute<'a, T: Iterator<Item = &'a str> + Send>(
51+
&mut self,
52+
queries: T,
53+
) -> Result<usize, Self::Error> {
5154
// Tiberius doesn't support transactions, see https://github.com/prisma/tiberius/issues/28
5255
self.simple_query("BEGIN TRAN T1;").await?;
5356
let mut count = 0;
5457
for query in queries {
5558
// Drop the returning `QueryStream<'a>` to avoid compiler complaning regarding lifetimes
56-
if let Err(err) = self.simple_query(*query).await.map(drop) {
59+
if let Err(err) = self.simple_query(query).await.map(drop) {
5760
if let Err(err) = self.simple_query("ROLLBACK TRAN T1").await {
5861
log::error!("could not ROLLBACK transaction, {}", err);
5962
}

‎refinery_core/src/drivers/tokio_postgres.rs

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -35,7 +35,10 @@ async fn query_applied_migrations(
3535
impl AsyncTransaction for Client {
3636
type Error = PgError;
3737

38-
async fn execute(&mut self, queries: &[&str]) -> Result<usize, Self::Error> {
38+
async fn execute<'a, T: Iterator<Item = &'a str> + Send>(
39+
&mut self,
40+
queries: T,
41+
) -> Result<usize, Self::Error> {
3942
let transaction = self.transaction().await?;
4043
let mut count = 0;
4144
for query in queries {

‎refinery_core/src/traits/async.rs

Lines changed: 19 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,10 @@ use std::string::ToString;
1212
pub trait AsyncTransaction {
1313
type Error: std::error::Error + Send + Sync + 'static;
1414

15-
async fn execute(&mut self, query: &[&str]) -> Result<usize, Self::Error>;
15+
async fn execute<'a, T: Iterator<Item = &'a str> + Send>(
16+
&mut self,
17+
queries: T,
18+
) -> Result<usize, Self::Error>;
1619
}
1720

1821
#[async_trait]
@@ -43,10 +46,13 @@ async fn migrate<T: AsyncTransaction>(
4346
migration.set_applied();
4447
let update_query = insert_migration_query(&migration, migration_table_name);
4548
transaction
46-
.execute(&[
47-
migration.sql().as_ref().expect("sql must be Some!"),
48-
&update_query,
49-
])
49+
.execute(
50+
[
51+
migration.sql().as_ref().expect("sql must be Some!"),
52+
update_query.as_str(),
53+
]
54+
.into_iter(),
55+
)
5056
.await
5157
.migration_err(
5258
&format!("error applying migration {migration}"),
@@ -109,10 +115,8 @@ async fn migrate_grouped<T: AsyncTransaction>(
109115
);
110116
}
111117

112-
let refs: Vec<&str> = grouped_migrations.iter().map(AsRef::as_ref).collect();
113-
114118
transaction
115-
.execute(refs.as_ref())
119+
.execute(grouped_migrations.iter().map(AsRef::as_ref))
116120
.await
117121
.migration_err("error applying migrations", None)?;
118122

@@ -142,7 +146,7 @@ where
142146
migration_table_name: &str,
143147
) -> Result<Option<Migration>, Error> {
144148
let mut migrations = self
145-
.query(Self::get_last_applied_migration_query(migration_table_name).as_str())
149+
.query(Self::get_last_applied_migration_query(migration_table_name).as_ref())
146150
.await
147151
.migration_err("error getting last applied migration", None)?;
148152

@@ -154,7 +158,7 @@ where
154158
migration_table_name: &str,
155159
) -> Result<Vec<Migration>, Error> {
156160
let migrations = self
157-
.query(Self::get_applied_migrations_query(migration_table_name).as_str())
161+
.query(Self::get_applied_migrations_query(migration_table_name).as_ref())
158162
.await
159163
.migration_err("error getting applied migrations", None)?;
160164

@@ -170,9 +174,11 @@ where
170174
target: Target,
171175
migration_table_name: &str,
172176
) -> Result<Report, Error> {
173-
self.execute(&[&Self::assert_migrations_table_query(migration_table_name)])
174-
.await
175-
.migration_err("error asserting migrations table", None)?;
177+
self.execute(
178+
[Self::assert_migrations_table_query(migration_table_name).as_ref()].into_iter(),
179+
)
180+
.await
181+
.migration_err("error asserting migrations table", None)?;
176182

177183
let applied_migrations = self
178184
.get_applied_migrations(migration_table_name)

‎refinery_core/src/traits/sync.rs

Lines changed: 12 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,10 @@ use crate::{Error, Migration, Report, Target};
88
pub trait Transaction {
99
type Error: std::error::Error + Send + Sync + 'static;
1010

11-
fn execute(&mut self, queries: &[&str]) -> Result<usize, Self::Error>;
11+
fn execute<'a, T: Iterator<Item = &'a str>>(
12+
&mut self,
13+
queries: T,
14+
) -> Result<usize, Self::Error>;
1215
}
1316

1417
pub trait Query<T>: Transaction {
@@ -66,7 +69,7 @@ pub fn migrate<T: Transaction>(
6669
}
6770
};
6871

69-
let refs: Vec<&str> = migration_batch.iter().map(AsRef::as_ref).collect();
72+
let refs = migration_batch.iter().map(AsRef::as_ref);
7073

7174
if batched {
7275
let migrations_display = applied_migrations
@@ -76,10 +79,10 @@ pub fn migrate<T: Transaction>(
7679
.join("\n");
7780
log::info!("going to apply batch migrations in single transaction:\n{migrations_display}");
7881
transaction
79-
.execute(refs.as_ref())
82+
.execute(refs)
8083
.migration_err("error applying migrations", None)?;
8184
} else {
82-
for (i, update) in refs.iter().enumerate() {
85+
for (i, update) in refs.enumerate() {
8386
// first iteration is pair so we know the following even in the iteration index
8487
// marks the previous (pair) migration as completed.
8588
let applying_migration = i % 2 == 0;
@@ -91,7 +94,7 @@ pub fn migrate<T: Transaction>(
9194
log::debug!("applied migration: {current_migration} writing state to db.");
9295
}
9396
transaction
94-
.execute(&[update])
97+
.execute([update].into_iter())
9598
.migration_err("error applying update", Some(&applied_migrations[0..i / 2]))?;
9699
}
97100
}
@@ -119,8 +122,10 @@ where
119122
fn assert_migrations_table(&mut self, migration_table_name: &str) -> Result<usize, Error> {
120123
// Needed cause some database vendors like Mssql have a non sql standard way of checking the migrations table,
121124
// though on this case it's just to be consistent with the async trait `AsyncMigrate`
122-
self.execute(&[Self::assert_migrations_table_query(migration_table_name).as_str()])
123-
.migration_err("error asserting migrations table", None)
125+
self.execute(
126+
[Self::assert_migrations_table_query(migration_table_name).as_str()].into_iter(),
127+
)
128+
.migration_err("error asserting migrations table", None)
124129
}
125130

126131
fn get_last_applied_migration(

0 commit comments

Comments
 (0)