feat: teach sqlx-cli about migrate.table-name

Author: abonander

sqlx-cli/src/migrate.rs

@@ -130,10 +130,10 @@ pub async fn info(config: &Config, migration_source: &MigrationSourceOpt, connec
     let migrator = Migrator::new(ResolveWith(Path::new(source), config.migrate.to_resolve_config())).await?;
     let mut conn = crate::connect(connect_opts).await?;
 
-    conn.ensure_migrations_table().await?;
+    conn.ensure_migrations_table(config.migrate.table_name()).await?;
 
     let applied_migrations: HashMap<_, _> = conn
-        .list_applied_migrations()
+        .list_applied_migrations(config.migrate.table_name())
         .await?
         .into_iter()
         .map(|m| (m.version, m))
@@ -224,14 +224,14 @@ pub async fn run(
 
     let mut conn = crate::connect(connect_opts).await?;
 
-    conn.ensure_migrations_table().await?;
+    conn.ensure_migrations_table(config.migrate.table_name()).await?;
 
-    let version = conn.dirty_version().await?;
+    let version = conn.dirty_version(config.migrate.table_name()).await?;
     if let Some(version) = version {
         bail!(MigrateError::Dirty(version));
     }
 
-    let applied_migrations = conn.list_applied_migrations().await?;
+    let applied_migrations = conn.list_applied_migrations(config.migrate.table_name()).await?;
     validate_applied_migrations(&applied_migrations, &migrator, ignore_missing)?;
 
     let latest_version = applied_migrations
@@ -269,7 +269,7 @@ pub async fn run(
                 let elapsed = if dry_run || skip {
                     Duration::new(0, 0)
                 } else {
-                    conn.apply(migration).await?
+                    conn.apply(config.migrate.table_name(), migration).await?
                 };
                 let text = if skip {
                     "Skipped"
@@ -319,14 +319,14 @@ pub async fn revert(
 
     let mut conn = crate::connect(connect_opts).await?;
 
-    conn.ensure_migrations_table().await?;
+    conn.ensure_migrations_table(config.migrate.table_name()).await?;
 
-    let version = conn.dirty_version().await?;
+    let version = conn.dirty_version(config.migrate.table_name()).await?;
     if let Some(version) = version {
         bail!(MigrateError::Dirty(version));
     }
 
-    let applied_migrations = conn.list_applied_migrations().await?;
+    let applied_migrations = conn.list_applied_migrations(config.migrate.table_name()).await?;
     validate_applied_migrations(&applied_migrations, &migrator, ignore_missing)?;
 
     let latest_version = applied_migrations
@@ -360,7 +360,7 @@ pub async fn revert(
             let elapsed = if dry_run || skip {
                 Duration::new(0, 0)
             } else {
-                conn.revert(migration).await?
+                conn.revert(config.migrate.table_name(), migration).await?
             };
             let text = if skip {
                 "Skipped"

sqlx-cli/tests/common/mod.rs

@@ -6,10 +6,12 @@ use std::{
     fs::remove_file,
     path::{Path, PathBuf},
 };
+use sqlx::_unstable::config::Config;
 
 pub struct TestDatabase {
     file_path: PathBuf,
     migrations: String,
+    config: &'static Config,
 }
 
 impl TestDatabase {
@@ -19,6 +21,7 @@ impl TestDatabase {
         let ret = Self {
             file_path,
             migrations: String::from(migrations_path.to_str().unwrap()),
+            config: Config::from_crate(),
         };
         Command::cargo_bin("cargo-sqlx")
             .unwrap()
@@ -77,7 +80,7 @@ impl TestDatabase {
         let mut conn = SqliteConnection::connect(&self.connection_string())
             .await
             .unwrap();
-        conn.list_applied_migrations()
+        conn.list_applied_migrations(self.config.migrate.table_name())
             .await
             .unwrap()
             .iter()

sqlx-core/src/any/migrate.rs

@@ -44,18 +44,16 @@ impl MigrateDatabase for Any {
 }
 
 impl Migrate for AnyConnection {
-    fn ensure_migrations_table(&mut self) -> BoxFuture<'_, Result<(), MigrateError>> {
-        Box::pin(async { self.get_migrate()?.ensure_migrations_table().await })
+    fn ensure_migrations_table<'e>(&'e mut self, table_name: &'e str) -> BoxFuture<'e, Result<(), MigrateError>> {
+        Box::pin(async { self.get_migrate()?.ensure_migrations_table(table_name).await })
     }
 
-    fn dirty_version(&mut self) -> BoxFuture<'_, Result<Option<i64>, MigrateError>> {
-        Box::pin(async { self.get_migrate()?.dirty_version().await })
+    fn dirty_version<'e>(&'e mut self, table_name: &'e str) -> BoxFuture<'e, Result<Option<i64>, MigrateError>> {
+        Box::pin(async { self.get_migrate()?.dirty_version(table_name).await })
     }
 
-    fn list_applied_migrations(
-        &mut self,
-    ) -> BoxFuture<'_, Result<Vec<AppliedMigration>, MigrateError>> {
-        Box::pin(async { self.get_migrate()?.list_applied_migrations().await })
+    fn list_applied_migrations<'e>(&'e mut self, table_name: &'e str) -> BoxFuture<'e, Result<Vec<AppliedMigration>, MigrateError>> {
+        Box::pin(async { self.get_migrate()?.list_applied_migrations(table_name).await })
     }
 
     fn lock(&mut self) -> BoxFuture<'_, Result<(), MigrateError>> {
@@ -66,17 +64,19 @@ impl Migrate for AnyConnection {
         Box::pin(async { self.get_migrate()?.unlock().await })
     }
 
-    fn apply<'e: 'm, 'm>(
+    fn apply<'e>(
         &'e mut self,
-        migration: &'m Migration,
-    ) -> BoxFuture<'m, Result<Duration, MigrateError>> {
-        Box::pin(async { self.get_migrate()?.apply(migration).await })
+        table_name: &'e str,
+        migration: &'e Migration,
+    ) -> BoxFuture<'e, Result<Duration, MigrateError>> {
+        Box::pin(async { self.get_migrate()?.apply(table_name, migration).await })
     }
 
-    fn revert<'e: 'm, 'm>(
+    fn revert<'e>(
         &'e mut self,
-        migration: &'m Migration,
-    ) -> BoxFuture<'m, Result<Duration, MigrateError>> {
-        Box::pin(async { self.get_migrate()?.revert(migration).await })
+        table_name: &'e str,
+        migration: &'e Migration,
+    ) -> BoxFuture<'e, Result<Duration, MigrateError>> {
+        Box::pin(async { self.get_migrate()?.revert(table_name, migration).await })
     }
 }

sqlx-core/src/config/migrate.rs

@@ -186,6 +186,10 @@ impl Config {
         self.migrations_dir.as_deref().unwrap_or("migrations")
     }
 
+    pub fn table_name(&self) -> &str {
+        self.table_name.as_deref().unwrap_or("_sqlx_migrations")
+    }
+    
     pub fn to_resolve_config(&self) -> crate::migrate::ResolveConfig {
         let mut config = crate::migrate::ResolveConfig::new();
         config.ignore_chars(self.ignored_chars.iter().copied());

sqlx-core/src/migrate/migrate.rs

@@ -27,16 +27,14 @@ pub trait MigrateDatabase {
 pub trait Migrate {
     // ensure migrations table exists
     // will create or migrate it if needed
-    fn ensure_migrations_table(&mut self) -> BoxFuture<'_, Result<(), MigrateError>>;
+    fn ensure_migrations_table<'e>(&'e mut self, table_name: &'e str) -> BoxFuture<'e, Result<(), MigrateError>>;
 
     // Return the version on which the database is dirty or None otherwise.
     // "dirty" means there is a partially applied migration that failed.
-    fn dirty_version(&mut self) -> BoxFuture<'_, Result<Option<i64>, MigrateError>>;
+    fn dirty_version<'e>(&'e mut self, table_name: &'e str) -> BoxFuture<'e, Result<Option<i64>, MigrateError>>;
 
     // Return the ordered list of applied migrations
-    fn list_applied_migrations(
-        &mut self,
-    ) -> BoxFuture<'_, Result<Vec<AppliedMigration>, MigrateError>>;
+    fn list_applied_migrations<'e>(&'e mut self, table_name: &'e str) -> BoxFuture<'e,  Result<Vec<AppliedMigration>, MigrateError>>;
 
     // Should acquire a database lock so that only one migration process
     // can run at a time. [`Migrate`] will call this function before applying
@@ -50,16 +48,18 @@ pub trait Migrate {
     // run SQL from migration in a DDL transaction
     // insert new row to [_migrations] table on completion (success or failure)
     // returns the time taking to run the migration SQL
-    fn apply<'e: 'm, 'm>(
+    fn apply<'e>(
         &'e mut self,
-        migration: &'m Migration,
-    ) -> BoxFuture<'m, Result<Duration, MigrateError>>;
+        table_name: &'e str,
+        migration: &'e Migration,
+    ) -> BoxFuture<'e, Result<Duration, MigrateError>>;
 
     // run a revert SQL from migration in a DDL transaction
     // deletes the row in [_migrations] table with specified migration version on completion (success or failure)
     // returns the time taking to run the migration SQL
-    fn revert<'e: 'm, 'm>(
+    fn revert<'e>(
         &'e mut self,
-        migration: &'m Migration,
-    ) -> BoxFuture<'m, Result<Duration, MigrateError>>;
+        table_name: &'e str,
+        migration: &'e Migration,
+    ) -> BoxFuture<'e, Result<Duration, MigrateError>>;
 }

sqlx-core/src/migrate/migrator.rs

@@ -27,25 +27,6 @@ pub struct Migrator {
     pub table_name: Cow<'static, str>,
 }
 
-fn validate_applied_migrations(
-    applied_migrations: &[AppliedMigration],
-    migrator: &Migrator,
-) -> Result<(), MigrateError> {
-    if migrator.ignore_missing {
-        return Ok(());
-    }
-
-    let migrations: HashSet<_> = migrator.iter().map(|m| m.version).collect();
-
-    for applied_migration in applied_migrations {
-        if !migrations.contains(&applied_migration.version) {
-            return Err(MigrateError::VersionMissing(applied_migration.version));
-        }
-    }
-
-    Ok(())
-}
-
 impl Migrator {
     #[doc(hidden)]
     pub const DEFAULT: Migrator = Migrator {
@@ -156,12 +137,21 @@ impl Migrator {
         <A::Connection as Deref>::Target: Migrate,
     {
         let mut conn = migrator.acquire().await?;
-        self.run_direct(&mut *conn).await
+        self.run_direct(None, &mut *conn).await
+    }
+
+    pub async fn run_to<'a, A>(&self, target: i64, migrator: A) -> Result<(), MigrateError>
+    where
+        A: Acquire<'a>,
+        <A::Connection as Deref>::Target: Migrate,
+    {
+        let mut conn = migrator.acquire().await?;
+        self.run_direct(Some(target), &mut *conn).await
     }
 
     // Getting around the annoying "implementation of `Acquire` is not general enough" error
     #[doc(hidden)]
-    pub async fn run_direct<C>(&self, conn: &mut C) -> Result<(), MigrateError>
+    pub async fn run_direct<C>(&self, target: Option<i64>, conn: &mut C) -> Result<(), MigrateError>
     where
         C: Migrate,
     {
@@ -172,14 +162,14 @@ impl Migrator {
 
         // creates [_migrations] table only if needed
         // eventually this will likely migrate previous versions of the table
-        conn.ensure_migrations_table().await?;
+        conn.ensure_migrations_table(&self.table_name).await?;
 
-        let version = conn.dirty_version().await?;
+        let version = conn.dirty_version(&self.table_name).await?;
         if let Some(version) = version {
             return Err(MigrateError::Dirty(version));
         }
 
-        let applied_migrations = conn.list_applied_migrations().await?;
+        let applied_migrations = conn.list_applied_migrations(&self.table_name).await?;
         validate_applied_migrations(&applied_migrations, self)?;
 
         let applied_migrations: HashMap<_, _> = applied_migrations
@@ -188,6 +178,11 @@ impl Migrator {
             .collect();
 
         for migration in self.iter() {
+            if target.is_some_and(|target| target < migration.version) {
+                // Target version reached
+                break;
+            }
+            
             if migration.migration_type.is_down_migration() {
                 continue;
             }
@@ -199,7 +194,7 @@ impl Migrator {
                     }
                 }
                 None => {
-                    conn.apply(migration).await?;
+                    conn.apply(&self.table_name, migration).await?;
                 }
             }
         }
@@ -244,14 +239,14 @@ impl Migrator {
 
         // creates [_migrations] table only if needed
         // eventually this will likely migrate previous versions of the table
-        conn.ensure_migrations_table().await?;
+        conn.ensure_migrations_table(&self.table_name).await?;
 
-        let version = conn.dirty_version().await?;
+        let version = conn.dirty_version(&self.table_name).await?;
         if let Some(version) = version {
             return Err(MigrateError::Dirty(version));
         }
 
-        let applied_migrations = conn.list_applied_migrations().await?;
+        let applied_migrations = conn.list_applied_migrations(&self.table_name).await?;
         validate_applied_migrations(&applied_migrations, self)?;
 
         let applied_migrations: HashMap<_, _> = applied_migrations
@@ -266,7 +261,7 @@ impl Migrator {
             .filter(|m| applied_migrations.contains_key(&m.version))
             .filter(|m| m.version > target)
         {
-            conn.revert(migration).await?;
+            conn.revert(&self.table_name, migration).await?;
         }
 
         // unlock the migrator to allow other migrators to run
@@ -278,3 +273,22 @@ impl Migrator {
         Ok(())
     }
 }
+
+fn validate_applied_migrations(
+    applied_migrations: &[AppliedMigration],
+    migrator: &Migrator,
+) -> Result<(), MigrateError> {
+    if migrator.ignore_missing {
+        return Ok(());
+    }
+
+    let migrations: HashSet<_> = migrator.iter().map(|m| m.version).collect();
+
+    for applied_migration in applied_migrations {
+        if !migrations.contains(&applied_migration.version) {
+            return Err(MigrateError::VersionMissing(applied_migration.version));
+        }
+    }
+
+    Ok(())
+}

sqlx-core/src/testing/mod.rs

@@ -243,7 +243,7 @@ async fn setup_test_db<DB: Database>(
 
     if let Some(migrator) = args.migrator {
         migrator
-            .run_direct(&mut conn)
+            .run_direct(None, &mut conn)
             .await
             .expect("failed to apply migrations");
     }