Repository: diesel-rs/diesel (language: rust)

Issue:
Add any_pending_migrations to `embed_migrations!`
Implements #1845

It turned out the logic already exists, but hasn't been exposed for the `embed_migrations!` macro yet.

```rust
if embedded_migrations::any_pending_migrations()? {
    println!("executing migrations);
    embedded_migrations::run(&db)?;
}
```


Discussion / hints:
@kpcyrd doesn't look like your `rustfmt` patch succeeded:

```
$ cargo fmt --all -- --write-mode=diff
Diff in /home/travis/build/diesel-rs/diesel/diesel_migrations/migrations_internals/src/lib.rs at line 183:
 /// structured, and where Diesel will look for them by default.
 pub fn any_pending_migrations_from<Conn, List>(
     conn: &Conn,
-    migrations: List
+    migrations: List,
 ) -> Result<bool, RunMigrationsError>
 where
     Conn: MigrationConnection,
The command "cargo fmt --all -- --write-mode=diff" exited with 4.
```

I'm assuming that's why there are no takers on reviewing it.
@weiznich can you confirm that a review isn't picked up until the build shows passing?
@dmcguire81 I somehow missed that, thanks for pointing it out. rustfmt should pass with the patch I just pushed.
Any update on this? :)
I'm not sure I understand why this is necessary. `embedded_migrations::run(&db)?;` internally checks already if any migration needs to run. 
My example, was probably confusing, I'm more interested in the `println!` part:
```rust
if embedded_migrations::any_pending_migrations()? {
    println!("executing migrations);
}
embedded_migrations::run(&db)?;
```
Having access to these things allows me to write a better user interface. @Boscop may have a similar usecase.
This would be great to see in the library, as it enables several use-cases. 

- Confirmation of running migrations
- Running migrations in a separate CLI, and verifying they have been run by a daemon process.

The latter case supports the use-case where migrations can be run via a sysadmin/dba with permissions to create/destroy tables, but the application runs with a user that only has read/write permissions for the tables .
This functionality is indeed useful as a standalone method for `embedded_migrations`, mostly due to the fact that I would like to deny further processing in my application if the database has yet to be (manually) migrated.
Looks like there are some conflicts that need to be resolved - any possibility of more action on this @kpcyrd ? 
> I'm not sure I understand why this is necessary. `embedded_migrations::run(&db)?;` internally checks already if any migration needs to run.

I've just been dealing with a power loss during a migration scenario using Sqlite which resulted in database corruption, to avoid this in the future we run migrations on a working copy and then replace the old database on success.

Without the ability to check whether we need to run migrations, we need to create the working copy each time which increases startup times and decreases NAND durability unnecessarily.
I'm probably willing to accept such a change if there is at least:
* A complete documentation why this can be useful
* A changelog entry
* A test case 
* Passing CI
* No merge conflict
Can you please expand upon exactly what you'd like for "A complete documentation why this can be useful"?
@weiznich would you accept a patch that would instead expose functions necessary to build a custom progress bar? Something like:

```rust
let n = embedded_migrations::number_of_missing_transactions(&db)?;
if n > 0 {
    for i in 0..n {
        println!("Applying migration ({}/{})", i+1, n);
        embedded_migrations::run_next(&db)?;
    }
} else {
    println!("Database already up-to-date!");
}
```
@disconsented 
> Can you please expand upon exactly what you'd like for "A complete documentation why this can be useful"?

The current PR description does only briefly mention potential use cases. Also I feel that all use cases listed in the comments, are quite incomplete. I can understand why this feature would be interesting but I would like to have a concrete written down example why this must be implemented that way and not some other way.

@kpcyrd Having tough a bit more about this, I came to the conclusion that the probably best solution for this and similar problems would be to just make the `ALL_MIGRATIONS` constant part of our public API. Given that all this functionality could be implemented in `diesel_migrations` itself as function not specific to `embed_migration!`, which also would make all that much more discoverable in my opinion. I think I would be open for a PR doing that instead.
I'm working with @disconsented on this. While I respect the need to maintain a tidy API, why can this sort of refactoring not be done later as an additional PR, instead of requiring all of the work up-front and further delaying a change that would satisfy affected users in the meantime?
@genbattle There are multiple reasons for that:

* Removing an API which is added "just" as quick fix is really hard. That does require releasing a new major diesel version, which is something I rather would not like to do often.
* Documenting the currently proposed extensions in a discoverable manner is not really possible in my opinion, as all of those methods would not be part of any diesel crate, but rather generated by some opaque proc-macro. That makes it hard to discover them easily and at least for me that feels like to much magic. If possible we should try to hide as less information behind a proc-macro as possible.
* I feel that the solution proposed I've proposed above is easier to implement and understand than adding a bunch of semi-hidden functions to the output generated by `embed_migration!`

Additionally I do my work on diesel in my free time and share that for free with you. So maybe requesting "requiring all of the work up-front and further delaying a change that would satisfy affected users in the meantime?" is not the right way to convince me that that's the right way to implement such an API, because for me that sounds like: You need that for your potential commercial use case, you want to have a quick fix for that and rather then spending the time yourself to implement a sustainable solution yourself you want to outsource that for free to the diesel maintainers. That's not something I'm willing to accept here.
I think there has been a bit of a miscommunication here, I have _already_ spent a couple of hours working on updating the original work (with credit to kpcyrd), so we're not trying to "outsource that for free to the diesel maintainers". Which is why I reached out to you on gitter with a question about how much leeway I'd have with tests because I wanted to write something a bit more comprehensive. 

Your reply makes it seem like you didn't read the question either.

I won’t comment on the first however I disagree on the second, proc macros in general make discovery harder but that cost was already willingly paid about 7 months ago adding in an example in the docs isn't any worse.

As for the third, I want to reiterate that the code in this PR still works rewriting code is harder than not rewriting code.

Not sure how this could be easier than using the existing 27 lines, we're happy to spend the time doing this, but we're hesitant to spend a couple of weeks going through the entire process. 
@disconsented I think I should clarify a few points about the general process here:

1. At least till last week I would consider this PR have going stale. It was not something there was active work on + there was no indication that someone from the contributor team would be willing to merge it as it was or even that the API was something we would like to have.
2. You comment 6 days ago put that PR back to my notification. You've given a good example why something like this can be useful, so that's a totally fine way to do this.
3. I had another look at the change and figured out that this would be OK given the points I wrote up above.
4. You've asked a question about some detail of my points which should be clarified. That's also totally fine.
5. Then the original author of this PR stepped in and asked for yet another slightly different API, at this point it is clear for me that there is more than one uncovered use case here and we should have a discussion about the exact API. In my understanding it's my job as maintainer to sit down an think about such cases then and propose a solution that is maintainable, opens up the minimal required API and is generally discoverable. In this case that seems quite easy for me, so I've proposed a different API quite soon. At that point I consider that proposal open for discussion and wait for arguments made that this is the acceptable approach or not.
6. At that point you've asked questions about how to test the old proposed solution in our gitter channel, where I've answered that I would rather like to go with my new proposal for the reasons listed in the corresponding comment here. It's fine to ask that question, but after pointing out that there is a new proposal I would at least expect that you put in some argument why you would like to go with the old variant. (Note: I do not really consider: "But I've already started with implementing that" a valid argument here, because at this point that was not clear for me or anyone else in this thread.)
7. Someone unrelated starts to comment in this thread requesting in quite hard words that the original solution is implemented as "quick way" to "satisfy users". That's probably not the right way to give a technical argument.


May main points here are:

* The discussion on this feature restarted less then a week before. 
* The original author expressed that they are interested in a other slightly different API. That means the proposed API is not set in stone yet.
* You basically expect that I'm willing to decide on an API in that time, while I'm only working on and of in my free time on this project. That means at least for me it's totally OK to not comment for a few days on something, because I'm occupied with other things like weekend, my paid work and so on. If that's not a acceptable level of support for you, you need find a way to pay someone for providing a support in a more timely manner.

Additionally I do not see any concrete plans to put out a release as soon as some potential PR is merged here. Therefore I do not see any reason to rush the API design, to have this done in less then a week or so.

Fix patch:
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 4acd9e26572e..fbef00417d80 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -383,13 +383,13 @@ jobs:
           args: --manifest-path diesel_cli/Cargo.toml --no-default-features --features "sqlite-bundled"
 
   minimal_rust_version:
-    name: Check Minimal supported rust version (1.40.0)
+    name: Check Minimal supported rust version (1.48.0)
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v2
       - uses: actions-rs/toolchain@v1
         with:
-          toolchain: 1.40.0
+          toolchain: 1.48.0
           profile: minimal
           override: true
       - name: Cache cargo registry
@@ -407,7 +407,7 @@ jobs:
         run: |
           RUSTC_BOOTSTRAP=1 cargo update -Z minimal-versions
 
-      - name: Check building with rust 1.40.0
+      - name: Check building with rust 1.48.0
         uses: actions-rs/cargo@v1
         with:
           command: check
diff --git a/CHANGELOG.md b/CHANGELOG.md
index ebb18b2e36f8..3807c50aa241 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -65,11 +65,12 @@ for Rust libraries in [RFC #1105](https://github.com/rust-lang/rfcs/blob/master/
 * Support for `mysqlclient-sys` < 0.2.0 has been removed.
 * Support for `time` types has been removed.
 * Support for `chrono` < 0.4.19 has been removed.
-* The `NonNull` for sql types has been removed in favour of the new `SqlType` trait.
-
+* The `NonNull` trait for sql types has been removed in favour of the new `SqlType` trait.
 * `no_arg_sql_function!` has been deprecated without replacement.
   [`sql_function!`][sql-function-2-0-0] can now be used for functions with zero
   arguments. See [the migration guide][2-0-migration] for more details.
+* Support for `barrel` based migrations has been removed for now. We are happy to
+  add this support back as soon as `barrel` integrates with our new migration framework.
 
 ### Changed
 
@@ -86,7 +87,7 @@ for Rust libraries in [RFC #1105](https://github.com/rust-lang/rfcs/blob/master/
   you are implementing `HasSqlType` for `Mysql` manually, you may need to adjust
   your implementation to fully use the new unsigned variants in `MysqlType`
 
-* The minimal officially supported rustc version is now 1.40.0
+* The minimal officially supported rustc version is now 1.48.0
 
 * The `RawValue` types for the `Mysql` and `Postgresql` backend where changed
   from `[u8]` to distinct opaque types. If you used the concrete `RawValue` type
@@ -155,6 +156,10 @@ for Rust libraries in [RFC #1105](https://github.com/rust-lang/rfcs/blob/master/
 
 * For the `Pg` backend, `TypeMetadata::MetadataLookup` has changed to `dyn PgMetadataLookup`.
 
+* Diesel's migration framework was rewritten from the ground. Existing migrations continue to 
+  be compatible with the rewrite, but code calling into `diesel_migrations` requires an update.
+  See the [migration guide](#2-0-0-upgrade-migrations) for details.
+
 ### Fixed
 
 * Many types were incorrectly considered non-aggregate when they should not
@@ -242,6 +247,37 @@ Key points:
 - In diesel v1.4 sql functions without arguments used the `no_arg_sql_function!` macro,
   which has since been deprecated. The new `sql_function!` macro supports functions without
   arguments.
+  
+#### `diesel_migration` rewrite
+<a name = "2-0-0-upgrade-migrations"></a>
+
+Key points:
+
+- Functions for interacting with migrations are not any more free standing 
+  functions in `diesel_migration`. They are provided by anything that implements
+  `MigrationHarness` now. Out of the box implementations are provided by any 
+  connection type (for applying migrations without output) and `HarnessWithOutput`
+  (for applying migrations with output)
+- A set of migrations is now provided by a `MigrationSource`. Diesel brings implementations 
+  `FileBasedMigrations` (for migration scripts in a `migration` folder) and `EmbededMigrations`
+  (for migrations embedded at compile time based on an existing `migration` folder)
+- `embed_migrations!` now creates a value of the type `EmbededMigrations` instead of a module. 
+That means code using `embed_migrations!()` needs to be changed from
+```rust
+embed_migrations!()
+
+fn run_migration(conn: &PgConnection) {
+    embedded_migrations::run(conn).unwrap()
+}
+```
+to 
+```rust
+pub const MIGRATIONS: EmbededMigrations = embed_migrations!();
+
+fn run_migration(conn: &PgConnection) {
+    conn.run_pending_migrations(MIGRATIONS).unwrap();
+}
+```
 
 [2-0-migration]: FIXME write a migration guide
 
diff --git a/diesel/Cargo.toml b/diesel/Cargo.toml
index 0cff7ef670d6..6ac0022f29b8 100644
--- a/diesel/Cargo.toml
+++ b/diesel/Cargo.toml
@@ -38,7 +38,7 @@ version = "~2.0.0"
 path = "../diesel_derives"
 
 [dev-dependencies]
-cfg-if = "0.1.10"
+cfg-if = "1"
 dotenv = "0.15"
 ipnetwork = ">=0.12.2, <0.18.0"
 quickcheck = "0.9"
diff --git a/diesel/src/connection/mod.rs b/diesel/src/connection/mod.rs
index be921456f9da..d7b11f39cd94 100644
--- a/diesel/src/connection/mod.rs
+++ b/diesel/src/connection/mod.rs
@@ -27,18 +27,18 @@ pub trait SimpleConnection {
 }
 
 /// A connection to a database
-pub trait Connection: SimpleConnection + Sized + Send {
+pub trait Connection: SimpleConnection + Send {
     /// The backend this type connects to
     type Backend: Backend;
-    #[doc(hidden)]
-    type TransactionManager: TransactionManager<Self>;
 
     /// Establishes a new connection to the database
     ///
     /// The argument to this method and the method's behavior varies by backend.
     /// See the documentation for that backend's connection class
     /// for details about what it accepts and how it behaves.
-    fn establish(database_url: &str) -> ConnectionResult<Self>;
+    fn establish(database_url: &str) -> ConnectionResult<Self>
+    where
+        Self: Sized;
 
     /// Executes the given function inside of a database transaction
     ///
@@ -100,6 +100,7 @@ pub trait Connection: SimpleConnection + Sized + Send {
     /// ```
     fn transaction<T, E, F>(&self, f: F) -> Result<T, E>
     where
+        Self: Sized,
         F: FnOnce() -> Result<T, E>,
         E: From<Error>,
     {
@@ -119,7 +120,10 @@ pub trait Connection: SimpleConnection + Sized + Send {
 
     /// Creates a transaction that will never be committed. This is useful for
     /// tests. Panics if called while inside of a transaction.
-    fn begin_test_transaction(&self) -> QueryResult<()> {
+    fn begin_test_transaction(&self) -> QueryResult<()>
+    where
+        Self: Sized,
+    {
         let transaction_manager = self.transaction_manager();
         assert_eq!(transaction_manager.get_transaction_depth(), 0);
         transaction_manager.begin_transaction(self)
@@ -162,6 +166,7 @@ pub trait Connection: SimpleConnection + Sized + Send {
     where
         F: FnOnce() -> Result<T, E>,
         E: Debug,
+        Self: Sized,
     {
         let mut user_result = None;
         let _ = self.transaction::<(), _, _>(|| {
@@ -177,6 +182,7 @@ pub trait Connection: SimpleConnection + Sized + Send {
     #[doc(hidden)]
     fn load<T, U>(&self, source: T) -> QueryResult<Vec<U>>
     where
+        Self: Sized,
         T: AsQuery,
         T::Query: QueryFragment<Self::Backend> + QueryId,
         U: FromSqlRow<T::SqlType, Self::Backend>,
@@ -185,8 +191,57 @@ pub trait Connection: SimpleConnection + Sized + Send {
     #[doc(hidden)]
     fn execute_returning_count<T>(&self, source: &T) -> QueryResult<usize>
     where
+        Self: Sized,
         T: QueryFragment<Self::Backend> + QueryId;
 
     #[doc(hidden)]
-    fn transaction_manager(&self) -> &Self::TransactionManager;
+    fn transaction_manager(&self) -> &dyn TransactionManager<Self>
+    where
+        Self: Sized;
+}
+
+/// A variant of the [`Connection`](trait.Connection.html) trait that is
+/// usable with dynamic dispatch
+///
+/// If you are looking for a way to use pass database connections
+/// for different database backends around in your application
+/// this trait won't help you much. Normally you should only
+/// need to use this trait if you are interacting with a connection
+/// passed to a [`Migration`](../migration/trait.Migration.html)
+pub trait BoxableConnection<DB: Backend>: Connection<Backend = DB> + std::any::Any {
+    #[doc(hidden)]
+    fn as_any(&self) -> &dyn std::any::Any;
+}
+
+impl<C> BoxableConnection<C::Backend> for C
+where
+    C: Connection + std::any::Any,
+{
+    fn as_any(&self) -> &dyn std::any::Any {
+        self
+    }
+}
+
+impl<DB: Backend> dyn BoxableConnection<DB> {
+    /// Downcast the current connection to a specific connection
+    /// type.
+    ///
+    /// This will return `None` if the underlying
+    /// connection does not match the corresponding
+    /// type, otherwise a reference to the underlying connection is returned
+    pub fn downcast_ref<T>(&self) -> Option<&T>
+    where
+        T: Connection<Backend = DB> + 'static,
+    {
+        self.as_any().downcast_ref::<T>()
+    }
+
+    /// Check if the current connection is
+    /// a specific connection type
+    pub fn is<T>(&self) -> bool
+    where
+        T: Connection<Backend = DB> + 'static,
+    {
+        self.as_any().is::<T>()
+    }
 }
diff --git a/diesel/src/migration/mod.rs b/diesel/src/migration/mod.rs
index a37068d03529..9ce3e697563f 100644
--- a/diesel/src/migration/mod.rs
+++ b/diesel/src/migration/mod.rs
@@ -1,57 +1,178 @@
 //! Representation of migrations
 
-mod errors;
-pub use self::errors::{MigrationError, RunMigrationsError};
-
-use crate::connection::{Connection, SimpleConnection};
+use crate::backend::Backend;
+use crate::connection::{BoxableConnection, Connection};
+use crate::deserialize::{FromSql, FromSqlRow};
+use crate::expression::AsExpression;
 use crate::result::QueryResult;
-use std::path::Path;
+use crate::serialize::ToSql;
+use crate::sql_types::Text;
+use std::borrow::Cow;
+use std::error::Error;
+use std::fmt::Display;
+
+/// A specialized result type representing the result of
+/// a migration operation
+pub type Result<T> = std::result::Result<T, Box<dyn Error + Send + Sync>>;
+
+/// A migration version identifier
+///
+/// This is used by the migration harness to place migrations
+/// in order, therefore two different instances of this type
+/// must be sortable
+#[derive(Debug, Hash, PartialEq, Eq, PartialOrd, Ord, FromSqlRow, AsExpression)]
+#[sql_type = "Text"]
+pub struct MigrationVersion<'a>(Cow<'a, str>);
+
+impl<'a> MigrationVersion<'a> {
+    /// Convert the current migration version into
+    /// an owned variant with static life time
+    pub fn as_owned(&self) -> MigrationVersion<'static> {
+        MigrationVersion(Cow::Owned(self.0.as_ref().to_owned()))
+    }
+}
+
+impl<'a, DB> FromSql<Text, DB> for MigrationVersion<'a>
+where
+    String: FromSql<Text, DB>,
+    DB: Backend,
+{
+    fn from_sql(bytes: crate::backend::RawValue<DB>) -> crate::deserialize::Result<Self> {
+        let s = String::from_sql(bytes)?;
+        Ok(Self(Cow::Owned(s)))
+    }
+}
+
+impl<'a, DB> ToSql<Text, DB> for MigrationVersion<'a>
+where
+    Cow<'a, str>: ToSql<Text, DB>,
+    DB: Backend,
+{
+    fn to_sql<W: std::io::Write>(
+        &self,
+        out: &mut crate::serialize::Output<W, DB>,
+    ) -> crate::serialize::Result {
+        self.0.to_sql(out)
+    }
+}
+
+impl<'a> From<String> for MigrationVersion<'a> {
+    fn from(s: String) -> Self {
+        MigrationVersion(Cow::Owned(s))
+    }
+}
+
+impl<'a> From<&'a str> for MigrationVersion<'a> {
+    fn from(s: &'a str) -> Self {
+        MigrationVersion(Cow::Borrowed(s))
+    }
+}
+
+impl<'a> From<&'a String> for MigrationVersion<'a> {
+    fn from(s: &'a String) -> Self {
+        MigrationVersion(Cow::Borrowed(s))
+    }
+}
+
+impl<'a> Display for MigrationVersion<'a> {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str(self.0.as_ref())
+    }
+}
+
+/// Represents the name of a migration
+///
+/// Users should threat this as `impl Display` type,
+/// for implementors of custom migration types
+/// this opens the possibility to roll out their own versioning
+/// schema.
+pub trait MigrationName: Display {
+    /// The version corresponding to the current migration name
+    fn version(&self) -> MigrationVersion;
+}
 
 /// Represents a migration that interacts with diesel
-pub trait Migration {
-    /// Get the migration version
-    fn version(&self) -> &str;
+pub trait Migration<DB: Backend> {
     /// Apply this migration
-    fn run(&self, conn: &dyn SimpleConnection) -> Result<(), RunMigrationsError>;
+    fn run(&self, conn: &dyn BoxableConnection<DB>) -> Result<()>;
+
     /// Revert this migration
-    fn revert(&self, conn: &dyn SimpleConnection) -> Result<(), RunMigrationsError>;
-    /// Get the migration file path
-    fn file_path(&self) -> Option<&Path> {
-        None
-    }
+    fn revert(&self, conn: &dyn BoxableConnection<DB>) -> Result<()>;
+
+    /// Get a the attached metadata for this migration
+    fn metadata(&self) -> &dyn MigrationMetadata;
+
+    /// Get the name of the current migration
+    ///
+    /// The provided name is used by migration harness
+    /// to get the version of a migration and to
+    /// as something to that is displayed and allows
+    /// user to identify a specific migration
+    fn name(&self) -> &dyn MigrationName;
 }
 
-impl<'a> Migration for Box<dyn Migration + 'a> {
-    fn version(&self) -> &str {
-        (&**self).version()
+/// This trait is designed to customize the behaviour
+/// of the default migration harness of diesel
+///
+/// Any new customization option will be added
+/// as new function here. Each new function
+/// will have a default implementation
+/// returning the old a value corresponding
+/// to the old uncustomized behaviour
+pub trait MigrationMetadata {
+    /// Whether the current migration is executed in a transaction or not
+    ///
+    /// By default each migration is executed in a own transaction, but
+    /// certain operations (like creating an index on an existing column)
+    /// requires running the migration without transaction.
+    ///
+    /// By default this function returns true
+    fn run_in_transaction(&self) -> bool {
+        true
     }
+}
 
-    fn run(&self, conn: &dyn SimpleConnection) -> Result<(), RunMigrationsError> {
+/// A migration source is an entity that can be used
+/// to receive a number of migrations from.
+pub trait MigrationSource<DB: Backend> {
+    /// Get a list of migrations associated with this
+    /// migration soucre.
+    fn migrations(&self) -> Result<Vec<Box<dyn Migration<DB>>>>;
+}
+
+impl<'a, DB: Backend> Migration<DB> for Box<dyn Migration<DB> + 'a> {
+    fn run(&self, conn: &dyn BoxableConnection<DB>) -> Result<()> {
         (&**self).run(conn)
     }
 
-    fn revert(&self, conn: &dyn SimpleConnection) -> Result<(), RunMigrationsError> {
+    fn revert(&self, conn: &dyn BoxableConnection<DB>) -> Result<()> {
         (&**self).revert(conn)
     }
-    fn file_path(&self) -> Option<&Path> {
-        (&**self).file_path()
+
+    fn metadata(&self) -> &dyn MigrationMetadata {
+        (&**self).metadata()
     }
-}
 
-impl<'a> Migration for &'a dyn Migration {
-    fn version(&self) -> &str {
-        (&**self).version()
+    fn name(&self) -> &dyn MigrationName {
+        (&**self).name()
     }
+}
 
-    fn run(&self, conn: &dyn SimpleConnection) -> Result<(), RunMigrationsError> {
+impl<'a, DB: Backend> Migration<DB> for &'a dyn Migration<DB> {
+    fn run(&self, conn: &dyn BoxableConnection<DB>) -> Result<()> {
         (&**self).run(conn)
     }
 
-    fn revert(&self, conn: &dyn SimpleConnection) -> Result<(), RunMigrationsError> {
+    fn revert(&self, conn: &dyn BoxableConnection<DB>) -> Result<()> {
         (&**self).revert(conn)
     }
-    fn file_path(&self) -> Option<&Path> {
-        (&**self).file_path()
+
+    fn metadata(&self) -> &dyn MigrationMetadata {
+        (&**self).metadata()
+    }
+
+    fn name(&self) -> &dyn MigrationName {
+        (&**self).name()
     }
 }
 
diff --git a/diesel/src/mysql/connection/mod.rs b/diesel/src/mysql/connection/mod.rs
index e1e40c22d600..a6e59416b3b6 100644
--- a/diesel/src/mysql/connection/mod.rs
+++ b/diesel/src/mysql/connection/mod.rs
@@ -34,7 +34,6 @@ impl SimpleConnection for MysqlConnection {
 
 impl Connection for MysqlConnection {
     type Backend = Mysql;
-    type TransactionManager = AnsiTransactionManager;
 
     fn establish(database_url: &str) -> ConnectionResult<Self> {
         use crate::result::ConnectionError::CouldntSetupConfiguration;
@@ -89,7 +88,7 @@ impl Connection for MysqlConnection {
     }
 
     #[doc(hidden)]
-    fn transaction_manager(&self) -> &Self::TransactionManager {
+    fn transaction_manager(&self) -> &dyn TransactionManager<Self> {
         &self.transaction_manager
     }
 }
diff --git a/diesel/src/pg/connection/mod.rs b/diesel/src/pg/connection/mod.rs
index acd2562694a2..148fc667be08 100644
--- a/diesel/src/pg/connection/mod.rs
+++ b/diesel/src/pg/connection/mod.rs
@@ -47,7 +47,6 @@ impl SimpleConnection for PgConnection {
 
 impl Connection for PgConnection {
     type Backend = Pg;
-    type TransactionManager = AnsiTransactionManager;
 
     fn establish(database_url: &str) -> ConnectionResult<PgConnection> {
         RawConnection::establish(database_url).and_then(|raw_conn| {
@@ -95,7 +94,7 @@ impl Connection for PgConnection {
     }
 
     #[doc(hidden)]
-    fn transaction_manager(&self) -> &Self::TransactionManager {
+    fn transaction_manager(&self) -> &dyn TransactionManager<Self> {
         &self.transaction_manager
     }
 }
diff --git a/diesel/src/pg/transaction.rs b/diesel/src/pg/transaction.rs
index fc49a2f06fd1..27dd1b40d560 100644
--- a/diesel/src/pg/transaction.rs
+++ b/diesel/src/pg/transaction.rs
@@ -275,7 +275,7 @@ impl<'a> TransactionBuilder<'a> {
         let mut query_builder = <Pg as Backend>::QueryBuilder::default();
         self.to_sql(&mut query_builder)?;
         let sql = query_builder.finish();
-        let transaction_manager = self.connection.transaction_manager();
+        let transaction_manager = &self.connection.transaction_manager;
 
         transaction_manager.begin_transaction_sql(self.connection, &sql)?;
         match f() {
diff --git a/diesel/src/r2d2.rs b/diesel/src/r2d2.rs
index 93863c3df215..20b26382b44b 100644
--- a/diesel/src/r2d2.rs
+++ b/diesel/src/r2d2.rs
@@ -136,7 +136,6 @@ where
     M::Connection: Connection + R2D2Connection + Send + 'static,
 {
     type Backend = <M::Connection as Connection>::Backend;
-    type TransactionManager = PooledConnectionTransactionManager<M>;
 
     fn establish(_: &str) -> ConnectionResult<Self> {
         Err(ConnectionError::BadConnection(String::from(
@@ -165,48 +164,34 @@ where
         (&**self).execute_returning_count(source)
     }
 
-    fn transaction_manager(&self) -> &Self::TransactionManager {
-        // This is actually fine because we have an #[repr(transparent)]
-        // on LoggingTransactionManager, which means the layout is the same
-        // as the inner type
-        // See the ref-cast crate for a longer version: https://github.com/dtolnay/ref-cast
-        unsafe {
-            &*((&**self).transaction_manager() as *const _ as *const Self::TransactionManager)
-        }
+    fn transaction_manager(&self) -> &dyn TransactionManager<Self> {
+        self
     }
 }
 
-#[doc(hidden)]
-#[repr(transparent)]
-#[allow(missing_debug_implementations)]
-pub struct PooledConnectionTransactionManager<M>
-where
-    M: ManageConnection,
-    M::Connection: Connection,
-{
-    inner: <M::Connection as Connection>::TransactionManager,
-}
-
-impl<M> TransactionManager<PooledConnection<M>> for PooledConnectionTransactionManager<M>
+impl<M> TransactionManager<PooledConnection<M>> for PooledConnection<M>
 where
     M: ManageConnection,
-    PooledConnection<M>: Connection,
-    M::Connection: Connection,
+    M::Connection: Connection + R2D2Connection,
 {
-    fn begin_transaction(&self, conn: &PooledConnection<M>) -> QueryResult<()> {
-        self.inner.begin_transaction(&**conn)
+    fn begin_transaction(&self, _conn: &PooledConnection<M>) -> QueryResult<()> {
+        let conn = &**self;
+        conn.transaction_manager().begin_transaction(conn)
     }
 
-    fn rollback_transaction(&self, conn: &PooledConnection<M>) -> QueryResult<()> {
-        self.inner.rollback_transaction(&**conn)
+    fn rollback_transaction(&self, _conn: &PooledConnection<M>) -> QueryResult<()> {
+        let conn = &**self;
+        conn.transaction_manager().rollback_transaction(conn)
     }
 
-    fn commit_transaction(&self, conn: &PooledConnection<M>) -> QueryResult<()> {
-        self.inner.commit_transaction(&**conn)
+    fn commit_transaction(&self, _conn: &PooledConnection<M>) -> QueryResult<()> {
+        let conn = &**self;
+        conn.transaction_manager().commit_transaction(conn)
     }
 
     fn get_transaction_depth(&self) -> u32 {
-        self.inner.get_transaction_depth()
+        let conn = &**self;
+        conn.transaction_manager().get_transaction_depth()
     }
 }
 
diff --git a/diesel/src/sqlite/connection/mod.rs b/diesel/src/sqlite/connection/mod.rs
index 35c604dd725a..810533b8d4c6 100644
--- a/diesel/src/sqlite/connection/mod.rs
+++ b/diesel/src/sqlite/connection/mod.rs
@@ -52,7 +52,6 @@ impl SimpleConnection for SqliteConnection {
 
 impl Connection for SqliteConnection {
     type Backend = Sqlite;
-    type TransactionManager = AnsiTransactionManager;
 
     /// Establish a connection to the database specified by `database_url`.
     ///
@@ -106,7 +105,7 @@ impl Connection for SqliteConnection {
     }
 
     #[doc(hidden)]
-    fn transaction_manager(&self) -> &Self::TransactionManager {
+    fn transaction_manager(&self) -> &dyn TransactionManager<Self> {
         &self.transaction_manager
     }
 }
@@ -175,7 +174,7 @@ impl SqliteConnection {
         F: FnOnce() -> Result<T, E>,
         E: From<Error>,
     {
-        let transaction_manager = self.transaction_manager();
+        let transaction_manager = &self.transaction_manager;
 
         transaction_manager.begin_transaction_sql(self, sql)?;
         match f() {
diff --git a/diesel_bench/build.rs b/diesel_bench/build.rs
index 43131df3f718..43b8ec711334 100644
--- a/diesel_bench/build.rs
+++ b/diesel_bench/build.rs
@@ -52,14 +52,14 @@ fn database_url_from_env(backend_specific_env_var: &str) -> String {
 }
 
 fn main() {
-    let migrations_dir = migrations::find_migrations_directory()
-        .unwrap()
-        .join(MIGRATION_SUBDIR);
+    use migrations::MigrationHarness;
+    use std::path::PathBuf;
+
+    let manifest_dir = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
+    let migrations_dir = manifest_dir.join("migrations").join(MIGRATION_SUBDIR);
+    let migrations = migrations::FileBasedMigrations::from_path(&migrations_dir).unwrap();
+
     println!("cargo:rerun-if-changed={}", migrations_dir.display());
-    migrations::run_pending_migrations_in_directory(
-        &connection(),
-        &migrations_dir,
-        &mut io::sink(),
-    )
-    .unwrap();
+
+    connection().run_pending_migrations(migrations).unwrap();
 }
diff --git a/diesel_cli/Cargo.toml b/diesel_cli/Cargo.toml
index 59d648590acb..1544d43c76bc 100644
--- a/diesel_cli/Cargo.toml
+++ b/diesel_cli/Cargo.toml
@@ -24,9 +24,8 @@ clap = "2.27"
 dotenv = "0.15"
 heck = "0.3.1"
 serde = { version = "1.0.0", features = ["derive"] }
-toml = "0.4.6"
+toml = "0.5"
 url = { version = "2.1.0", optional = true }
-barrel = { version = ">= 0.5.0", optional = true, features = ["diesel"] }
 libsqlite3-sys = { version = ">=0.8.0, <0.21.0", optional = true, features = ["min_sqlite_version_3_7_16"] }
 diffy = "0.2.0"
 regex = "1.0.6"
@@ -37,9 +36,9 @@ version = "~2.0.0"
 path = "../diesel"
 default-features = false
 
-[dependencies.migrations_internals]
-version = "~1.4.0"
-path = "../diesel_migrations/migrations_internals"
+[dependencies.diesel_migrations]
+version = "~2.0.0"
+path = "../diesel_migrations/"
 
 [dev-dependencies]
 difference = "2.0"
@@ -52,7 +51,6 @@ default = ["postgres", "sqlite", "mysql"]
 postgres = ["diesel/postgres", "url", "uses_information_schema"]
 sqlite = ["diesel/sqlite"]
 mysql = ["diesel/mysql", "url", "uses_information_schema"]
-barrel-migrations = ["migrations_internals/barrel", "barrel"]
 sqlite-bundled = ["sqlite", "libsqlite3-sys/bundled"]
 uses_information_schema = []
 
diff --git a/diesel_cli/src/config.rs b/diesel_cli/src/config.rs
index 5e90b18432d2..e915b32a3159 100644
--- a/diesel_cli/src/config.rs
+++ b/diesel_cli/src/config.rs
@@ -28,7 +28,7 @@ impl Config {
             .unwrap_or_else(|| find_project_root().unwrap_or_default().join("diesel.toml"))
     }
 
-    pub fn read(matches: &ArgMatches) -> Result<Self, Box<dyn Error>> {
+    pub fn read(matches: &ArgMatches) -> Result<Self, Box<dyn Error + Send + Sync + 'static>> {
         let path = Self::file_path(matches);
 
         if path.exists() {
diff --git a/diesel_cli/src/database.rs b/diesel_cli/src/database.rs
index cf976cabd38d..610031595598 100644
--- a/diesel_cli/src/database.rs
+++ b/diesel_cli/src/database.rs
@@ -4,7 +4,7 @@ use clap::ArgMatches;
 use diesel::dsl::sql;
 use diesel::sql_types::Bool;
 use diesel::*;
-use migrations_internals as migrations;
+use diesel_migrations::FileBasedMigrations;
 
 use crate::database_error::{DatabaseError, DatabaseResult};
 
@@ -12,7 +12,6 @@ use std::env;
 use std::error::Error;
 #[cfg(feature = "postgres")]
 use std::fs::{self, File};
-use std::io::stdout;
 #[cfg(feature = "postgres")]
 use std::io::Write;
 use std::path::Path;
@@ -234,12 +233,9 @@ fn create_schema_table_and_run_migrations_if_needed(
     migrations_dir: &Path,
 ) -> DatabaseResult<()> {
     if !schema_table_exists(database_url).unwrap_or_else(handle_error) {
-        call_with_conn!(database_url, migrations::setup_database())?;
-        call_with_conn!(
-            database_url,
-            migrations::run_pending_migrations_in_directory(migrations_dir, &mut stdout())
-        )
-        .unwrap_or_else(handle_error);
+        let migrations =
+            FileBasedMigrations::from_path(migrations_dir).unwrap_or_else(handle_error);
+        call_with_conn!(database_url, super::run_migrations_with_output(migrations))?;
     };
     Ok(())
 }
diff --git a/diesel_cli/src/database_error.rs b/diesel_cli/src/database_error.rs
index 2c467ea597fb..3b98c7e9b8f2 100644
--- a/diesel_cli/src/database_error.rs
+++ b/diesel_cli/src/database_error.rs
@@ -16,6 +16,7 @@ pub enum DatabaseError {
     IoError(io::Error),
     QueryError(result::Error),
     ConnectionError(result::ConnectionError),
+    MigrationError(Box<dyn Error + Send + Sync + 'static>),
 }
 
 impl From<io::Error> for DatabaseError {
@@ -36,6 +37,12 @@ impl From<result::ConnectionError> for DatabaseError {
     }
 }
 
+impl From<Box<dyn Error + Send + Sync + 'static>> for DatabaseError {
+    fn from(e: Box<dyn Error + Send + Sync + 'static>) -> Self {
+        MigrationError(e)
+    }
+}
+
 impl Error for DatabaseError {}
 
 impl fmt::Display for DatabaseError {
@@ -59,6 +66,9 @@ impl fmt::Display for DatabaseError {
                 .source()
                 .map(ToString::to_string)
                 .unwrap_or_else(|| error.to_string())),
+            MigrationError(ref error) => {
+                write!(f, "Failed to run migrations: {}", error)
+            }
         }
     }
 }
diff --git a/diesel_cli/src/infer_schema_internals/inference.rs b/diesel_cli/src/infer_schema_internals/inference.rs
index 6b4ee7af61ad..a72c87046cc6 100644
--- a/diesel_cli/src/infer_schema_internals/inference.rs
+++ b/diesel_cli/src/infer_schema_internals/inference.rs
@@ -60,7 +60,7 @@ pub fn rust_name_for_sql_name(sql_name: &str) -> String {
 pub fn load_table_names(
     database_url: &str,
     schema_name: Option<&str>,
-) -> Result<Vec<TableName>, Box<dyn Error>> {
+) -> Result<Vec<TableName>, Box<dyn Error + Send + Sync + 'static>> {
     let connection = InferConnection::establish(database_url)?;
 
     match connection {
@@ -77,7 +77,7 @@ fn get_column_information(
     conn: &InferConnection,
     table: &TableName,
     column_sorting: &ColumnSorting,
-) -> Result<Vec<ColumnInformation>, Box<dyn Error>> {
+) -> Result<Vec<ColumnInformation>, Box<dyn Error + Send + Sync + 'static>> {
     let column_info = match *conn {
         #[cfg(feature = "sqlite")]
         InferConnection::Sqlite(ref c) => super::sqlite::get_table_data(c, table, column_sorting),
@@ -100,7 +100,7 @@ fn get_column_information(
 fn determine_column_type(
     attr: &ColumnInformation,
     conn: &InferConnection,
-) -> Result<ColumnType, Box<dyn Error>> {
+) -> Result<ColumnType, Box<dyn Error + Send + Sync + 'static>> {
     match *conn {
         #[cfg(feature = "sqlite")]
         InferConnection::Sqlite(_) => super::sqlite::determine_column_type(attr),
@@ -114,7 +114,7 @@ fn determine_column_type(
 pub(crate) fn get_primary_keys(
     conn: &InferConnection,
     table: &TableName,
-) -> Result<Vec<String>, Box<dyn Error>> {
+) -> Result<Vec<String>, Box<dyn Error + Send + Sync + 'static>> {
     let primary_keys: Vec<String> = match *conn {
         #[cfg(feature = "sqlite")]
         InferConnection::Sqlite(ref c) => super::sqlite::get_primary_keys(c, table),
@@ -138,7 +138,7 @@ pub(crate) fn get_primary_keys(
 pub fn load_foreign_key_constraints(
     database_url: &str,
     schema_name: Option<&str>,
-) -> Result<Vec<ForeignKeyConstraint>, Box<dyn Error>> {
+) -> Result<Vec<ForeignKeyConstraint>, Box<dyn Error + Send + Sync + 'static>> {
     let connection = InferConnection::establish(database_url)?;
 
     let constraints = match connection {
@@ -181,7 +181,7 @@ pub fn load_table_data(
     database_url: &str,
     name: TableName,
     column_sorting: &ColumnSorting,
-) -> Result<TableData, Box<dyn Error>> {
+) -> Result<TableData, Box<dyn Error + Send + Sync + 'static>> {
     let connection = InferConnection::establish(database_url)?;
     let docs = doc_comment!(
         "Representation of the `{}` table.
@@ -217,7 +217,7 @@ pub fn load_table_data(
                 rust_name,
             })
         })
-        .collect::<Result<_, Box<dyn Error>>>()?;
+        .collect::<Result<_, Box<dyn Error + Send + Sync + 'static>>>()?;
 
     Ok(TableData {
         name,
diff --git a/diesel_cli/src/infer_schema_internals/information_schema.rs b/diesel_cli/src/infer_schema_internals/information_schema.rs
index 2cf2813bd704..d82fd2ddcffb 100644
--- a/diesel_cli/src/infer_schema_internals/information_schema.rs
+++ b/diesel_cli/src/infer_schema_internals/information_schema.rs
@@ -244,7 +244,7 @@ where
 pub fn load_table_names<'a, Conn>(
     connection: &Conn,
     schema_name: Option<&'a str>,
-) -> Result<Vec<TableName>, Box<dyn Error>>
+) -> Result<Vec<TableName>, Box<dyn Error + Send + Sync + 'static>>
 where
     Conn: Connection,
     Conn::Backend: UsesInformationSchema,
diff --git a/diesel_cli/src/infer_schema_internals/mysql.rs b/diesel_cli/src/infer_schema_internals/mysql.rs
index 0b69762dce72..73c0fcdc6004 100644
--- a/diesel_cli/src/infer_schema_internals/mysql.rs
+++ b/diesel_cli/src/infer_schema_internals/mysql.rs
@@ -85,7 +85,9 @@ pub fn load_foreign_key_constraints(
     Ok(constraints)
 }
 
-pub fn determine_column_type(attr: &ColumnInformation) -> Result<ColumnType, Box<dyn Error>> {
+pub fn determine_column_type(
+    attr: &ColumnInformation,
+) -> Result<ColumnType, Box<dyn Error + Send + Sync + 'static>> {
     let tpe = determine_type_name(&attr.type_name)?;
     let unsigned = determine_unsigned(&attr.type_name);
 
@@ -97,7 +99,9 @@ pub fn determine_column_type(attr: &ColumnInformation) -> Result<ColumnType, Box
     })
 }
 
-fn determine_type_name(sql_type_name: &str) -> Result<String, Box<dyn Error>> {
+fn determine_type_name(
+    sql_type_name: &str,
+) -> Result<String, Box<dyn Error + Send + Sync + 'static>> {
     let result = if sql_type_name == "tinyint(1)" {
         "bool"
     } else if sql_type_name.starts_with("int") {
diff --git a/diesel_cli/src/infer_schema_internals/pg.rs b/diesel_cli/src/infer_schema_internals/pg.rs
index 8507bccc4240..0fcbc6e71659 100644
--- a/diesel_cli/src/infer_schema_internals/pg.rs
+++ b/diesel_cli/src/infer_schema_internals/pg.rs
@@ -3,7 +3,9 @@ use heck::CamelCase;
 use std::error::Error;
 use std::io::{stderr, Write};
 
-pub fn determine_column_type(attr: &ColumnInformation) -> Result<ColumnType, Box<dyn Error>> {
+pub fn determine_column_type(
+    attr: &ColumnInformation,
+) -> Result<ColumnType, Box<dyn Error + Send + Sync + 'static>> {
     let is_array = attr.type_name.starts_with('_');
     let tpe = if is_array {
         &attr.type_name[1..]
diff --git a/diesel_cli/src/infer_schema_internals/sqlite.rs b/diesel_cli/src/infer_schema_internals/sqlite.rs
index 1c5a7d24cbf3..ed4adca465a8 100644
--- a/diesel_cli/src/infer_schema_internals/sqlite.rs
+++ b/diesel_cli/src/infer_schema_internals/sqlite.rs
@@ -40,7 +40,7 @@ table! {
 pub fn load_table_names(
     connection: &SqliteConnection,
     schema_name: Option<&str>,
-) -> Result<Vec<TableName>, Box<dyn Error>> {
+) -> Result<Vec<TableName>, Box<dyn Error + Send + Sync + 'static>> {
     use self::sqlite_master::dsl::*;
 
     if schema_name.is_some() {
@@ -64,7 +64,7 @@ pub fn load_table_names(
 pub fn load_foreign_key_constraints(
     connection: &SqliteConnection,
     schema_name: Option<&str>,
-) -> Result<Vec<ForeignKeyConstraint>, Box<dyn Error>> {
+) -> Result<Vec<ForeignKeyConstraint>, Box<dyn Error + Send + Sync + 'static>> {
     let tables = load_table_names(connection, schema_name)?;
     let rows = tables
         .into_iter()
@@ -138,7 +138,9 @@ pub fn get_primary_keys(conn: &SqliteConnection, table: &TableName) -> QueryResu
         .collect())
 }
 
-pub fn determine_column_type(attr: &ColumnInformation) -> Result<ColumnType, Box<dyn Error>> {
+pub fn determine_column_type(
+    attr: &ColumnInformation,
+) -> Result<ColumnType, Box<dyn Error + Send + Sync + 'static>> {
     let type_name = attr.type_name.to_lowercase();
     let path = if is_bool(&type_name) {
         String::from("Bool")
diff --git a/diesel_cli/src/main.rs b/diesel_cli/src/main.rs
index 1d50ec9c23dc..4ab5fa3ed6da 100644
--- a/diesel_cli/src/main.rs
+++ b/diesel_cli/src/main.rs
@@ -28,9 +28,13 @@ mod validators;
 
 use chrono::*;
 use clap::{ArgMatches, Shell};
-use migrations_internals::{self as migrations, MigrationConnection};
+use diesel::backend::Backend;
+use diesel::migration::MigrationSource;
+use diesel::Connection;
+use diesel_migrations::{FileBasedMigrations, HarnessWithOutput, MigrationError, MigrationHarness};
 use regex::Regex;
 use std::any::Any;
+use std::collections::{HashMap, HashSet};
 use std::error::Error;
 use std::fmt::Display;
 use std::io::stdout;
@@ -39,8 +43,7 @@ use std::{env, fs};
 
 use self::config::Config;
 use self::database_error::{DatabaseError, DatabaseResult};
-use crate::migrations::MigrationError;
-use migrations_internals::TIMESTAMP_FORMAT;
+pub static TIMESTAMP_FORMAT: &str = "%Y-%m-%d-%H%M%S";
 
 fn main() {
     use dotenv::dotenv;
@@ -61,26 +64,23 @@ fn main() {
 
 // https://github.com/rust-lang-nursery/rust-clippy/issues/2927#issuecomment-405705595
 #[allow(clippy::similar_names)]
-fn run_migration_command(matches: &ArgMatches) -> Result<(), Box<dyn Error>> {
+fn run_migration_command(
+    matches: &ArgMatches,
+) -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
     match matches.subcommand() {
         ("run", Some(_)) => {
             let database_url = database::database_url(matches);
             let dir = migrations_dir(matches).unwrap_or_else(handle_error);
-            call_with_conn!(
-                database_url,
-                migrations::run_pending_migrations_in_directory(&dir, &mut stdout())
-            )?;
+            let dir = FileBasedMigrations::from_path(dir).unwrap_or_else(handle_error);
+            call_with_conn!(database_url, run_migrations_with_output(dir))?;
             regenerate_schema_if_file_specified(matches)?;
         }
         ("revert", Some(args)) => {
             let database_url = database::database_url(matches);
             let dir = migrations_dir(matches).unwrap_or_else(handle_error);
-
+            let dir = FileBasedMigrations::from_path(dir).unwrap_or_else(handle_error);
             if args.is_present("REVERT_ALL") {
-                call_with_conn!(
-                    database_url,
-                    migrations::revert_all_migrations_in_directory(&dir)
-                )?;
+                call_with_conn!(database_url, revert_all_migrations_with_output(dir))?;
             } else {
                 // TODO : remove this logic when upgrading to clap 3.0.
                 // We handle the default_value here instead of doing it
@@ -91,14 +91,23 @@ fn run_migration_command(matches: &ArgMatches) -> Result<(), Box<dyn Error>> {
                     None => "1",
                     Some(number) => number,
                 };
-
-                call_with_conn!(
-                    database_url,
-                    migrations::revert_latest_migrations_in_directory(
-                        &dir,
-                        number.parse::<u64>().expect("Unable to parse the value of the --number argument. A positive integer is expected.")
-                    )
-                )?;
+                for _ in 0..number.parse::<u64>().expect("Unable to parse the value of the --number argument. A positive integer is expected.") {
+                        match call_with_conn!(
+                            database_url,
+                            revert_migration_with_output(dir.clone())
+                        ) {
+                            Ok(_) => {}
+                            Err(e) if e.is::<MigrationError>() => {
+                                match e.downcast_ref::<MigrationError>() {
+                                    // If n is larger then the actual number of migrations,
+                                    // just stop reverting them
+                                    Some(MigrationError::NoMigrationRun) => break,
+                                    _ => return Err(e),
+                                }
+                            }
+                            Err(e) => return Err(e),
+                        }
+                    }
             }
 
             regenerate_schema_if_file_specified(matches)?;
@@ -106,37 +115,22 @@ fn run_migration_command(matches: &ArgMatches) -> Result<(), Box<dyn Error>> {
         ("redo", Some(args)) => {
             let database_url = database::database_url(matches);
             let dir = migrations_dir(matches).unwrap_or_else(handle_error);
-
-            call_with_conn!(database_url, redo_migrations(&dir, args));
+            let dir = FileBasedMigrations::from_path(dir).unwrap_or_else(handle_error);
+            call_with_conn!(database_url, redo_migrations(dir, args));
             regenerate_schema_if_file_specified(matches)?;
         }
         ("list", Some(_)) => {
             let database_url = database::database_url(matches);
             let dir = migrations_dir(matches).unwrap_or_else(handle_error);
-            let mut migrations =
-                call_with_conn!(database_url, migrations::mark_migrations_in_directory(&dir))?;
-
-            migrations.sort_by_key(|&(ref m, _)| m.version().to_string());
-
-            println!("Migrations:");
-            for (migration, applied) in migrations {
-                let name = migration
-                    .file_path()
-                    .unwrap()
-                    .file_name()
-                    .unwrap()
-                    .to_string_lossy();
-                let x = if applied { 'X' } else { ' ' };
-                println!("  [{}] {}", x, name);
-            }
+            let dir = FileBasedMigrations::from_path(dir).unwrap_or_else(handle_error);
+            call_with_conn!(database_url, list_migrations(dir))?;
         }
         ("pending", Some(_)) => {
             let database_url = database::database_url(matches);
             let dir = migrations_dir(matches).unwrap_or_else(handle_error);
-            let result = call_with_conn!(
-                database_url,
-                migrations::any_pending_migrations_in_directory(&dir)
-            )?;
+            let dir = FileBasedMigrations::from_path(dir).unwrap_or_else(handle_error);
+            let result =
+                call_with_conn!(database_url, MigrationHarness::has_pending_migration(dir))?;
             println!("{:?}", result);
         }
         ("generate", Some(args)) => {
@@ -149,8 +143,6 @@ fn run_migration_command(matches: &ArgMatches) -> Result<(), Box<dyn Error>> {
             fs::create_dir(&migration_dir).unwrap();
 
             match args.value_of("MIGRATION_FORMAT") {
-                #[cfg(feature = "barrel-migrations")]
-                Some("barrel") => ::barrel::integrations::diesel::generate_initial(&migration_dir),
                 Some("sql") => generate_sql_migration(&migration_dir),
                 Some(x) => return Err(format!("Unrecognized migration format `{}`", x).into()),
                 None => unreachable!("MIGRATION_FORMAT has a default value"),
@@ -205,6 +197,71 @@ fn migrations_dir_from_cli(matches: &ArgMatches) -> Option<PathBuf> {
         })
 }
 
+fn run_migrations_with_output<Conn, DB>(
+    conn: &Conn,
+    migrations: FileBasedMigrations,
+) -> Result<(), Box<dyn Error + Send + Sync + 'static>>
+where
+    Conn: MigrationHarness<DB> + Connection<Backend = DB> + 'static,
+    DB: Backend,
+{
+    HarnessWithOutput::write_to_stdout(conn)
+        .run_pending_migrations(migrations)
+        .map(|_| ())
+}
+
+fn revert_all_migrations_with_output<Conn, DB>(
+    conn: &Conn,
+    migrations: FileBasedMigrations,
+) -> Result<(), Box<dyn Error + Send + Sync + 'static>>
+where
+    Conn: MigrationHarness<DB> + Connection<Backend = DB> + 'static,
+    DB: Backend,
+{
+    HarnessWithOutput::write_to_stdout(conn)
+        .revert_all_migrations(migrations)
+        .map(|_| ())
+}
+
+fn revert_migration_with_output<Conn, DB>(
+    conn: &Conn,
+    migrations: FileBasedMigrations,
+) -> Result<(), Box<dyn Error + Send + Sync + 'static>>
+where
+    Conn: MigrationHarness<DB> + Connection<Backend = DB> + 'static,
+    DB: Backend,
+{
+    HarnessWithOutput::write_to_stdout(conn)
+        .revert_last_migration(migrations)
+        .map(|_| ())
+}
+
+fn list_migrations<Conn, DB>(
+    conn: &Conn,
+    migrations: FileBasedMigrations,
+) -> Result<(), Box<dyn Error + Send + Sync + 'static>>
+where
+    Conn: MigrationHarness<DB> + Connection<Backend = DB> + 'static,
+    DB: Backend,
+{
+    let applied_migrations = conn
+        .applied_migrations()?
+        .into_iter()
+        .collect::<HashSet<_>>();
+
+    let mut migrations = MigrationSource::<DB>::migrations(&migrations)?;
+    migrations.sort_unstable_by(|a, b| a.name().version().cmp(&b.name().version()));
+    println!("Migrations:");
+    for migration in migrations {
+        let applied = applied_migrations.contains(&migration.name().version());
+        let name = migration.name();
+        let x = if applied { 'X' } else { ' ' };
+        println!("  [{}] {}", x, name);
+    }
+
+    Ok(())
+}
+
 /// Checks for a migrations folder in the following order :
 /// 1. From the CLI arguments
 /// 2. From the MIGRATION_DIRECTORY environment variable
@@ -229,7 +286,7 @@ fn migrations_dir(matches: &ArgMatches) -> Result<PathBuf, MigrationError> {
 
     match migrations_dir {
         Some(dir) => Ok(dir),
-        None => migrations::find_migrations_directory(),
+        None => FileBasedMigrations::find_migrations_directory().map(|p| p.path().to_path_buf()),
     }
 }
 
@@ -289,11 +346,14 @@ fn create_config_file(matches: &ArgMatches) -> DatabaseResult<()> {
     Ok(())
 }
 
-fn run_database_command(matches: &ArgMatches) -> Result<(), Box<dyn Error>> {
+fn run_database_command(
+    matches: &ArgMatches,
+) -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
     match matches.subcommand() {
         ("setup", Some(args)) => {
             let migrations_dir = migrations_dir(matches).unwrap_or_else(handle_error);
             database::setup_database(args, &migrations_dir)?;
+            regenerate_schema_if_file_specified(matches)?;
         }
         ("reset", Some(args)) => {
             let migrations_dir = migrations_dir(matches).unwrap_or_else(handle_error);
@@ -356,13 +416,16 @@ fn search_for_directory_containing_file(path: &Path, file: &str) -> DatabaseResu
 /// if the `--number` argument is used.
 /// Migrations are performed in a transaction. If either part fails,
 /// the transaction is not committed.
-fn redo_migrations<Conn>(conn: &Conn, migrations_dir: &Path, args: &ArgMatches)
+fn redo_migrations<Conn, DB>(conn: &Conn, migrations_dir: FileBasedMigrations, args: &ArgMatches)
 where
-    Conn: MigrationConnection + Any,
+    DB: Backend,
+    Conn: MigrationHarness<DB> + Connection<Backend = DB> + 'static,
 {
-    let migrations_inner = || -> Result<(), diesel::migration::RunMigrationsError> {
+    let harness = HarnessWithOutput::write_to_stdout(conn);
+
+    let migrations_inner = || -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
         let reverted_versions = if args.is_present("REDO_ALL") {
-            migrations::revert_all_migrations_in_directory(conn, migrations_dir)?
+            harness.revert_all_migrations(migrations_dir.clone())?
         } else {
             // TODO : remove this logic when upgrading to clap 3.0.
             // We handle the default_value here instead of doing it
@@ -374,21 +437,43 @@ where
                 Some(number) => number,
             };
 
-            migrations::revert_latest_migrations_in_directory(
-                conn,
-                migrations_dir,
-                number.parse::<u64>().expect("Unable to parse the value of the --number argument. A positive integer is expected."),
-            )?
+            (0..number.parse::<u64>().expect("Unable to parse the value of the --number argument. A positive integer is expected."))
+                .filter_map(|_|{
+                    match harness.revert_last_migration(migrations_dir.clone()) {
+                        Ok(v) => {
+                            Some(Ok(v))
+                        }
+                        Err(e) if e.is::<MigrationError>() => {
+                            match e.downcast_ref::<MigrationError>() {
+                                // If n is larger then the actual number of migrations,
+                                // just stop reverting them
+                                Some(MigrationError::NoMigrationRun) => None,
+                                _ => Some(Err(e)),
+                            }
+                        }
+                        Err(e) => {
+                            Some(Err(e))
+                        }
+                    }
+                })
+                .collect::<Result<Vec<_>, _>>()?
         };
 
-        for reverted_version in reverted_versions {
-            migrations::run_migration_with_version(
-                conn,
-                migrations_dir,
-                &reverted_version,
-                &mut stdout(),
-            )?;
-        }
+        let mut migrations = MigrationSource::<DB>::migrations(&migrations_dir)?
+            .into_iter()
+            .map(|m| (m.name().version().as_owned(), m))
+            .collect::<HashMap<_, _>>();
+
+        let migrations = reverted_versions
+            .into_iter()
+            .map(|v| {
+                migrations
+                    .remove(&v)
+                    .ok_or_else(|| MigrationError::UnknownMigrationVersion(v.as_owned()))
+            })
+            .collect::<Result<Vec<_>, _>>()?;
+
+        harness.run_migrations(&migrations)?;
 
         Ok(())
     };
@@ -433,7 +518,7 @@ fn convert_absolute_path_to_relative(target_path: &Path, mut current_path: &Path
     result.join(target_path.strip_prefix(current_path).unwrap())
 }
 
-fn run_infer_schema(matches: &ArgMatches) -> Result<(), Box<dyn Error>> {
+fn run_infer_schema(matches: &ArgMatches) -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
     use crate::print_schema::*;
 
     let database_url = database::database_url(matches);
@@ -481,7 +566,9 @@ fn run_infer_schema(matches: &ArgMatches) -> Result<(), Box<dyn Error>> {
     Ok(())
 }
 
-fn regenerate_schema_if_file_specified(matches: &ArgMatches) -> Result<(), Box<dyn Error>> {
+fn regenerate_schema_if_file_specified(
+    matches: &ArgMatches,
+) -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
     use std::io::Read;
 
     let config = Config::read(matches)?;
diff --git a/diesel_cli/src/print_schema.rs b/diesel_cli/src/print_schema.rs
index f5dfce0ae464..384d0ef778f3 100644
--- a/diesel_cli/src/print_schema.rs
+++ b/diesel_cli/src/print_schema.rs
@@ -57,7 +57,7 @@ pub fn run_print_schema<W: IoWrite>(
     database_url: &str,
     config: &config::PrintSchema,
     output: &mut W,
-) -> Result<(), Box<dyn Error>> {
+) -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
     let schema = output_schema(database_url, config)?;
 
     output.write_all(schema.as_bytes())?;
@@ -67,7 +67,7 @@ pub fn run_print_schema<W: IoWrite>(
 pub fn output_schema(
     database_url: &str,
     config: &config::PrintSchema,
-) -> Result<String, Box<dyn Error>> {
+) -> Result<String, Box<dyn Error + Send + Sync + 'static>> {
     let table_names = load_table_names(database_url, config.schema_name())?
         .into_iter()
         .filter(|t| !config.filter.should_ignore_table(t))
@@ -78,7 +78,7 @@ pub fn output_schema(
     let table_data = table_names
         .into_iter()
         .map(|t| load_table_data(database_url, t, &config.column_sorting))
-        .collect::<Result<_, Box<dyn Error>>>()?;
+        .collect::<Result<_, Box<dyn Error + Send + Sync + 'static>>>()?;
     let definitions = TableDefinitions {
         tables: table_data,
         fk_constraints: foreign_keys,
diff --git a/diesel_derives/Cargo.toml b/diesel_derives/Cargo.toml
index 601aefbf4ba8..7df9b103d8c8 100644
--- a/diesel_derives/Cargo.toml
+++ b/diesel_derives/Cargo.toml
@@ -16,7 +16,7 @@ quote = "1"
 proc-macro2 = "1"
 
 [dev-dependencies]
-cfg-if = "0.1.10"
+cfg-if = "1"
 dotenv = "0.15"
 
 [dev-dependencies.diesel]
diff --git a/diesel_migrations/Cargo.toml b/diesel_migrations/Cargo.toml
index cfaa58c4710d..9b05447c4f51 100644
--- a/diesel_migrations/Cargo.toml
+++ b/diesel_migrations/Cargo.toml
@@ -1,25 +1,27 @@
 [package]
 name = "diesel_migrations"
-version = "1.4.0"
+version = "2.0.0"
 authors = ["Sean Griffin <sean@seantheprogrammer.com>"]
 license = "MIT OR Apache-2.0"
 description = "Migration management for diesel"
 documentation = "https://docs.rs/crate/diesel_migrations"
 homepage = "https://diesel.rs"
+edition = "2018"
 
 [dependencies.migrations_internals]
-version = "~1.4.0"
+version = "~2.0.0"
 path = "migrations_internals"
 
 [dependencies.migrations_macros]
-version = "~1.4.0"
+version = "~2.0.0"
 path = "migrations_macros"
 
 [dev-dependencies]
 dotenv = "0.15"
 cfg-if = "0.1.10"
+tempfile = "3.2"
 
-[dev-dependencies.diesel]
+[dependencies.diesel]
 version = "~2.0.0"
 path = "../diesel"
 default-features = false
diff --git a/diesel_migrations/migrations_internals/Cargo.toml b/diesel_migrations/migrations_internals/Cargo.toml
index 50e479648073..30ab02ee9c79 100644
--- a/diesel_migrations/migrations_internals/Cargo.toml
+++ b/diesel_migrations/migrations_internals/Cargo.toml
@@ -1,21 +1,12 @@
 [package]
 name = "migrations_internals"
-version = "1.4.1"
+version = "2.0.0"
 authors = ["Sean Griffin <sean@seantheprogrammer.com>"]
 license = "MIT OR Apache-2.0"
 description = "Internal implementation of diesels migration mechanism"
 homepage = "https://diesel.rs"
 
 [dependencies]
-barrel = { version = ">= 0.5.0", optional = true, features = ["diesel"] }
+serde = {version = "1", features = ["derive"]}
+toml = "0.5"
 
-[dependencies.diesel]
-version = "~2.0.0"
-path = "../../diesel"
-default-features = false
-
-[dev-dependencies]
-tempfile = "3.1.0"
-
-[features]
-default = []
diff --git a/diesel_migrations/migrations_internals/src/connection.rs b/diesel_migrations/migrations_internals/src/connection.rs
deleted file mode 100644
index 20f679d80d83..000000000000
--- a/diesel_migrations/migrations_internals/src/connection.rs
+++ /dev/null
@@ -1,71 +0,0 @@
-use diesel::deserialize::FromSql;
-use diesel::expression::bound::Bound;
-use diesel::expression::QueryMetadata;
-use diesel::helper_types::{max, Desc, Limit, Order, Select};
-use diesel::insertable::ColumnInsertValue;
-use diesel::prelude::*;
-use diesel::query_builder::{InsertStatement, QueryFragment, ValuesClause};
-use diesel::query_dsl::methods::{self, ExecuteDsl, LoadQuery};
-use diesel::sql_types::{Nullable, VarChar};
-use std::collections::HashSet;
-use std::iter::FromIterator;
-
-use super::schema::__diesel_schema_migrations::dsl::*;
-
-/// A connection which can be passed to the migration methods. This exists only
-/// to wrap up some constraints which are meant to hold for *all* connections.
-/// This trait will go away at some point in the future. Any Diesel connection
-/// should be useable where this trait is required.
-pub trait MigrationConnection: diesel::migration::MigrationConnection {
-    fn previously_run_migration_versions(&self) -> QueryResult<HashSet<String>>;
-    fn latest_run_migration_version(&self) -> QueryResult<Option<String>>;
-    fn latest_run_migration_versions(&self, number: u64) -> QueryResult<Vec<String>>;
-    fn insert_new_migration(&self, version: &str) -> QueryResult<()>;
-}
-
-impl<T> MigrationConnection for T
-where
-    T: diesel::migration::MigrationConnection,
-    String: FromSql<VarChar, T::Backend>,
-    // FIXME: HRTB is preventing projecting on any associated types here
-    for<'a> InsertStatement<
-        __diesel_schema_migrations,
-        ValuesClause<
-            ColumnInsertValue<version, &'a Bound<VarChar, &'a str>>,
-            __diesel_schema_migrations,
-        >,
-    >: ExecuteDsl<T>,
-    __diesel_schema_migrations: methods::SelectDsl<version>,
-    Select<__diesel_schema_migrations, version>: LoadQuery<T, String>,
-    Limit<Select<__diesel_schema_migrations, max<version>>>: QueryFragment<T::Backend>,
-    T::Backend: QueryMetadata<Nullable<VarChar>>,
-    Order<Limit<Select<__diesel_schema_migrations, version>>, Desc<version>>:
-        QueryFragment<T::Backend>,
-{
-    fn previously_run_migration_versions(&self) -> QueryResult<HashSet<String>> {
-        __diesel_schema_migrations
-            .select(version)
-            .load(self)
-            .map(FromIterator::from_iter)
-    }
-
-    fn latest_run_migration_version(&self) -> QueryResult<Option<String>> {
-        use diesel::dsl::max;
-        __diesel_schema_migrations.select(max(version)).first(self)
-    }
-
-    fn latest_run_migration_versions(&self, number: u64) -> QueryResult<Vec<String>> {
-        __diesel_schema_migrations
-            .select(version)
-            .order(version.desc())
-            .limit(number as i64)
-            .load(self)
-    }
-
-    fn insert_new_migration(&self, ver: &str) -> QueryResult<()> {
-        ::diesel::insert_into(__diesel_schema_migrations)
-            .values(&version.eq(ver))
-            .execute(self)?;
-        Ok(())
-    }
-}
diff --git a/diesel_migrations/migrations_internals/src/lib.rs b/diesel_migrations/migrations_internals/src/lib.rs
index 4927cfb9499a..da49cd4469ff 100644
--- a/diesel_migrations/migrations_internals/src/lib.rs
+++ b/diesel_migrations/migrations_internals/src/lib.rs
@@ -1,4 +1,3 @@
-// Built-in Lints
 #![deny(warnings, missing_debug_implementations, missing_copy_implementations)]
 // Clippy lints
 #![allow(
@@ -21,307 +20,85 @@
     clippy::items_after_statements,
     clippy::used_underscore_binding
 )]
-#![cfg_attr(test, allow(clippy::option_unwrap_used, clippy::result_unwrap_used))]
-//! Provides functions for maintaining database schema.
-//!
-//! A database migration always provides procedures to update the schema, as well as to revert
-//! itself. Diesel's migrations are versioned, and run in order. Diesel also takes care of tracking
-//! which migrations have already been run automatically. Your migrations don't need to be
-//! idempotent, as Diesel will ensure no migration is run twice unless it has been reverted.
-//!
-//! Migrations should be placed in a `/migrations` directory at the root of your project (the same
-//! directory as `Cargo.toml`). When any of these functions are run, Diesel will search for the
-//! migrations directory in the current directory and its parents, stopping when it finds the
-//! directory containing `Cargo.toml`.
-//!
-//! Individual migrations should be a folder containing exactly two files, `up.sql` and `down.sql`.
-//! `up.sql` will be used to run the migration, while `down.sql` will be used for reverting it. The
-//! folder itself should have the structure `{version}_{migration_name}`. It is recommended that
-//! you use the timestamp of creation for the version.
-//!
-//! Migrations can either be run with the CLI or embedded into the compiled application
-//! and executed with code, for example right after establishing a database connection.
-//! For more information, consult the [`embed_migrations!`](../macro.embed_migrations.html) macro.
-//!
-//! ## Example
-//!
-//! ```text
-//! # Directory Structure
-//! - 20151219180527_create_users
-//!     - up.sql
-//!     - down.sql
-//! - 20160107082941_create_posts
-//!     - up.sql
-//!     - down.sql
-//! ```
-//!
-//! ```sql
-//! -- 20151219180527_create_users/up.sql
-//! CREATE TABLE users (
-//!   id SERIAL PRIMARY KEY,
-//!   name VARCHAR NOT NULL,
-//!   hair_color VARCHAR
-//! );
-//! ```
-//!
-//! ```sql
-//! -- 20151219180527_create_users/down.sql
-//! DROP TABLE users;
-//! ```
-//!
-//! ```sql
-//! -- 20160107082941_create_posts/up.sql
-//! CREATE TABLE posts (
-//!   id SERIAL PRIMARY KEY,
-//!   user_id INTEGER NOT NULL,
-//!   title VARCHAR NOT NULL,
-//!   body TEXT
-//! );
-//! ```
-//!
-//! ```sql
-//! -- 20160107082941_create_posts/down.sql
-//! DROP TABLE posts;
-//! ```
-#[macro_use]
-extern crate diesel;
 
-#[cfg(feature = "barrel")]
-extern crate barrel;
-
-#[doc(hidden)]
-pub mod connection;
-pub mod migration;
-#[doc(hidden)]
-pub mod schema;
-
-#[doc(inline)]
-pub use self::connection::MigrationConnection;
-#[doc(inline)]
-pub use self::migration::*;
-pub use diesel::migration::*;
-
-use std::fs::DirEntry;
-use std::io::{stdout, Write};
-
-use self::schema::__diesel_schema_migrations::dsl::*;
-use diesel::expression_methods::*;
-use diesel::{Connection, QueryDsl, QueryResult, RunQueryDsl};
-
-use std::env;
+use std::fs::{DirEntry, File};
+use std::io::Read;
 use std::path::{Path, PathBuf};
 
-pub static TIMESTAMP_FORMAT: &str = "%Y-%m-%d-%H%M%S";
-
-/// Runs all migrations that have not yet been run. This function will print all progress to
-/// stdout. This function will return an `Err` if some error occurs reading the migrations, or if
-/// any migration fails to run. Each migration is run in its own transaction, so some migrations
-/// may be committed, even if a later migration fails to run.
-///
-/// It should be noted that this runs all migrations that have not already been run, regardless of
-/// whether or not their version is later than the latest run migration. This is generally not a
-/// problem, and eases the more common case of two developers generating independent migrations on
-/// a branch. Whoever created the second one will eventually need to run the first when both
-/// branches are merged.
-///
-/// See the [module level documentation](index.html) for information on how migrations should be
-/// structured, and where Diesel will look for them by default.
-pub fn run_pending_migrations<Conn>(conn: &Conn) -> Result<(), RunMigrationsError>
-where
-    Conn: MigrationConnection,
-{
-    let migrations_dir = find_migrations_directory()?;
-    run_pending_migrations_in_directory(conn, &migrations_dir, &mut stdout())
-}
-
 #[doc(hidden)]
-pub fn run_pending_migrations_in_directory<Conn>(
-    conn: &Conn,
-    migrations_dir: &Path,
-    output: &mut dyn Write,
-) -> Result<(), RunMigrationsError>
-where
-    Conn: MigrationConnection,
-{
-    let all_migrations = migrations_in_directory(migrations_dir)?;
-    run_migrations(conn, all_migrations, output)
-}
-
-/// Compares migrations found in `migrations_dir` to those that have been applied.
-/// Returns a list of pathbufs and whether they have been applied.
-pub fn mark_migrations_in_directory<Conn>(
-    conn: &Conn,
-    migrations_dir: &Path,
-) -> Result<Vec<(Box<dyn Migration>, bool)>, RunMigrationsError>
-where
-    Conn: MigrationConnection,
-{
-    let migrations = migrations_in_directory(migrations_dir)?;
-    setup_database(conn)?;
-    let already_run = conn.previously_run_migration_versions()?;
-    let migrations = migrations
-        .into_iter()
-        .map(|m| {
-            let applied = already_run.contains(&m.version().to_string());
-            (m, applied)
-        })
-        .collect();
-    Ok(migrations)
-}
-
-// Returns true if there are outstanding migrations in the migrations directory, otherwise
-// returns false. Returns an `Err` if there are problems with migration setup.
-///
-/// See the [module level documentation](index.html) for information on how migrations should be
-/// structured, and where Diesel will look for them by default.
-pub fn any_pending_migrations<Conn>(conn: &Conn) -> Result<bool, RunMigrationsError>
-where
-    Conn: MigrationConnection,
-{
-    let migrations_dir = find_migrations_directory()?;
-    any_pending_migrations_in_directory(conn, &migrations_dir)
-}
-
-pub fn any_pending_migrations_in_directory<Conn>(
-    conn: &Conn,
-    migrations_dir: &Path,
-) -> Result<bool, RunMigrationsError>
-where
-    Conn: MigrationConnection,
-{
-    let all_migrations = migrations_in_directory(&migrations_dir)?;
-    setup_database(conn)?;
-    let already_run = conn.previously_run_migration_versions()?;
-
-    let pending = all_migrations
-        .into_iter()
-        .any(|m| !already_run.contains(&m.version().to_string()));
-
-    Ok(pending)
-}
-
-pub fn revert_all_migrations_in_directory<Conn>(
-    conn: &Conn,
-    path: &Path,
-) -> Result<Vec<String>, RunMigrationsError>
-where
-    Conn: MigrationConnection,
-{
-    let number = migrations_in_directory(path)?.len() as u64;
-
-    revert_latest_migrations_in_directory(conn, path, number)
+#[derive(Debug, serde::Deserialize)]
+#[allow(missing_copy_implementations)]
+pub struct TomlMetadata {
+    #[serde(default)]
+    pub run_in_transaction: bool,
 }
 
-pub fn revert_latest_migrations_in_directory<Conn>(
-    conn: &Conn,
-    path: &Path,
-    number: u64,
-) -> Result<Vec<String>, RunMigrationsError>
-where
-    Conn: MigrationConnection,
-{
-    setup_database(conn)?;
-
-    let latest_migration_versions = conn
-        .latest_run_migration_versions(number)
-        .map_err(|_| RunMigrationsError::MigrationError(MigrationError::NoMigrationRun))?;
-
-    if latest_migration_versions.is_empty() {
-        return Err(RunMigrationsError::MigrationError(
-            MigrationError::NoMigrationRun,
-        ));
+impl Default for TomlMetadata {
+    fn default() -> Self {
+        Self {
+            run_in_transaction: true,
+        }
     }
+}
 
-    for migration_version in &latest_migration_versions {
-        revert_migration_with_version(conn, path, &migration_version, &mut stdout())?;
+impl TomlMetadata {
+    pub const fn new(run_in_transaction: bool) -> Self {
+        Self { run_in_transaction }
     }
 
-    Ok(latest_migration_versions)
-}
-
-/// Reverts the last migration that was run. Returns the version that was reverted. Returns an
-/// `Err` if no migrations have ever been run.
-///
-/// See the [module level documentation](index.html) for information on how migrations should be
-/// structured, and where Diesel will look for them by default.
-pub fn revert_latest_migration<Conn>(conn: &Conn) -> Result<String, RunMigrationsError>
-where
-    Conn: MigrationConnection,
-{
-    let migrations_dir = find_migrations_directory()?;
-    revert_latest_migration_in_directory(conn, &migrations_dir)
-}
+    pub fn read_from_file(path: &Path) -> Result<Self, Box<dyn std::error::Error>> {
+        let mut toml = String::new();
+        let mut file = File::open(path)?;
+        file.read_to_string(&mut toml)?;
 
-pub fn revert_latest_migration_in_directory<Conn>(
-    conn: &Conn,
-    path: &Path,
-) -> Result<String, RunMigrationsError>
-where
-    Conn: MigrationConnection,
-{
-    setup_database(conn)?;
-    let latest_migration_version = conn
-        .latest_run_migration_version()?
-        .ok_or_else(|| RunMigrationsError::MigrationError(MigrationError::NoMigrationRun))?;
-    revert_migration_with_version(conn, path, &latest_migration_version, &mut stdout())
-        .map(|_| latest_migration_version)
+        Ok(toml::from_str(&toml)?)
+    }
 }
 
-#[doc(hidden)]
-pub fn revert_migration_with_version<Conn: Connection>(
-    conn: &Conn,
-    migrations_dir: &Path,
-    ver: &str,
-    output: &mut dyn Write,
-) -> Result<(), RunMigrationsError> {
-    migration_with_version(migrations_dir, ver)
-        .map_err(Into::into)
-        .and_then(|m| revert_migration(conn, &m, output))
+pub fn search_for_migrations_directory(path: &Path) -> Option<PathBuf> {
+    let migration_path = path.join("migrations");
+    if migration_path.is_dir() {
+        Some(migration_path)
+    } else {
+        path.parent()
+            .and_then(|p| search_for_migrations_directory(p))
+    }
 }
 
-#[doc(hidden)]
-pub fn run_migration_with_version<Conn>(
-    conn: &Conn,
-    migrations_dir: &Path,
-    ver: &str,
-    output: &mut dyn Write,
-) -> Result<(), RunMigrationsError>
-where
-    Conn: MigrationConnection,
-{
-    migration_with_version(migrations_dir, ver)
-        .map_err(Into::into)
-        .and_then(|m| run_migration(conn, &*m, output))
+pub fn valid_sql_migration_directory(path: &Path) -> bool {
+    file_names(path)
+        .map(|files| files.contains(&"down.sql".into()) && files.contains(&"up.sql".into()))
+        .unwrap_or(false)
 }
 
-fn migration_with_version(
-    migrations_dir: &Path,
-    ver: &str,
-) -> Result<Box<dyn Migration>, MigrationError> {
-    let all_migrations = migrations_in_directory(migrations_dir)?;
-    let migration = all_migrations.into_iter().find(|m| m.version() == ver);
-    match migration {
-        Some(m) => Ok(m),
-        None => Err(MigrationError::UnknownMigrationVersion(ver.into())),
-    }
+pub fn version_from_string(path: &str) -> Option<String> {
+    path.split('_').next().map(|s| s.replace('-', ""))
 }
 
-#[doc(hidden)]
-pub fn setup_database<Conn: MigrationConnection>(conn: &Conn) -> QueryResult<usize> {
-    conn.setup()
+pub fn file_names(path: &Path) -> Result<Vec<String>, std::io::Error> {
+    path.read_dir()?
+        .filter_map(|entry| match entry {
+            Ok(entry) if entry.file_name().to_string_lossy().starts_with('.') => None,
+            Ok(entry) => Some(Ok(entry.file_name().to_string_lossy().into_owned())),
+            Err(e) => Some(Err(e)),
+        })
+        .collect::<Result<Vec<_>, _>>()
 }
 
-#[doc(hidden)]
-pub fn migration_paths_in_directory(path: &Path) -> Result<Vec<DirEntry>, MigrationError> {
-    path.read_dir()?
-        .filter_map(|entry| {
+pub fn migrations_directories<'a>(
+    path: &'a Path,
+) -> impl Iterator<Item = Result<DirEntry, std::io::Error>> + 'a {
+    path.read_dir()
+        .into_iter()
+        .flatten()
+        .filter_map(move |entry| {
             let entry = match entry {
                 Ok(e) => e,
-                Err(e) => return Some(Err(e.into())),
+                Err(e) => return Some(Err(e)),
             };
             let metadata = match entry.metadata() {
                 Ok(m) => m,
-                Err(e) => return Some(Err(e.into())),
+                Err(e) => return Some(Err(e)),
             };
             if metadata.is_file() {
                 return None;
@@ -332,197 +109,4 @@ pub fn migration_paths_in_directory(path: &Path) -> Result<Vec<DirEntry>, Migrat
                 Some(Ok(entry))
             }
         })
-        .collect()
-}
-
-fn migrations_in_directory(path: &Path) -> Result<Vec<Box<dyn Migration>>, MigrationError> {
-    migration_paths_in_directory(path)?
-        .iter()
-        .map(|e| migration_from(e.path()))
-        .collect()
-}
-
-/// Run all pending migrations in the given list. Apps should likely be calling
-/// `run_pending_migrations` or `run_pending_migrations_in_directory` instead.
-pub fn run_migrations<Conn, List>(
-    conn: &Conn,
-    migrations: List,
-    output: &mut dyn Write,
-) -> Result<(), RunMigrationsError>
-where
-    Conn: MigrationConnection,
-    List: IntoIterator,
-    List::Item: Migration,
-{
-    setup_database(conn)?;
-    let already_run = conn.previously_run_migration_versions()?;
-    let mut pending_migrations: Vec<_> = migrations
-        .into_iter()
-        .filter(|m| !already_run.contains(&m.version().to_string()))
-        .collect();
-
-    pending_migrations.sort_by(|a, b| a.version().cmp(b.version()));
-    for migration in pending_migrations {
-        run_migration(conn, &migration, output)?;
-    }
-    Ok(())
-}
-
-fn run_migration<Conn>(
-    conn: &Conn,
-    migration: &dyn Migration,
-    output: &mut dyn Write,
-) -> Result<(), RunMigrationsError>
-where
-    Conn: MigrationConnection,
-{
-    conn.transaction(|| {
-        if migration.version() != "00000000000000" {
-            writeln!(output, "Running migration {}", name(&migration))?;
-        }
-        if let Err(e) = migration.run(conn) {
-            writeln!(
-                output,
-                "Executing migration script {}",
-                file_name(&migration, "up.sql")
-            )?;
-            return Err(e);
-        }
-        conn.insert_new_migration(migration.version())?;
-        Ok(())
-    })
-}
-
-fn revert_migration<Conn: Connection>(
-    conn: &Conn,
-    migration: &dyn Migration,
-    output: &mut dyn Write,
-) -> Result<(), RunMigrationsError> {
-    conn.transaction(|| {
-        if migration.version() != "00000000000000" {
-            writeln!(output, "Rolling back migration {}", name(&migration))?;
-        }
-
-        if let Err(e) = migration.revert(conn) {
-            writeln!(
-                output,
-                "Executing migration script {}",
-                file_name(&migration, "down.sql")
-            )?;
-            return Err(e);
-        }
-        let target = __diesel_schema_migrations.filter(version.eq(migration.version()));
-        ::diesel::delete(target).execute(conn)?;
-        Ok(())
-    })
-}
-
-/// Returns the directory containing migrations. Will look at for
-/// $PWD/migrations. If it is not found, it will search the parents of the
-/// current directory, until it reaches the root directory.  Returns
-/// `MigrationError::MigrationDirectoryNotFound` if no directory is found.
-pub fn find_migrations_directory() -> Result<PathBuf, MigrationError> {
-    search_for_migrations_directory(&env::current_dir()?)
-}
-
-/// Searches for the migrations directory relative to the given path. See
-/// `find_migrations_directory` for more details.
-pub fn search_for_migrations_directory(path: &Path) -> Result<PathBuf, MigrationError> {
-    let migration_path = path.join("migrations");
-    if migration_path.is_dir() {
-        Ok(migration_path)
-    } else {
-        path.parent()
-            .map(|p| search_for_migrations_directory(p))
-            .unwrap_or_else(|| Err(MigrationError::MigrationDirectoryNotFound(path.into())))
-            .map_err(|_| MigrationError::MigrationDirectoryNotFound(path.into()))
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    extern crate tempfile;
-
-    use super::*;
-
-    use self::tempfile::Builder;
-    use std::fs;
-
-    #[test]
-    fn migration_directory_not_found_if_no_migration_dir_exists() {
-        let dir = Builder::new().prefix("diesel").tempdir().unwrap();
-
-        assert_eq!(
-            Err(MigrationError::MigrationDirectoryNotFound(
-                dir.path().into()
-            )),
-            search_for_migrations_directory(dir.path())
-        );
-    }
-
-    #[test]
-    fn migration_directory_defaults_to_pwd_slash_migrations() {
-        let dir = Builder::new().prefix("diesel").tempdir().unwrap();
-        let temp_path = dir.path().canonicalize().unwrap();
-        let migrations_path = temp_path.join("migrations");
-
-        fs::create_dir(&migrations_path).unwrap();
-
-        assert_eq!(
-            Ok(migrations_path),
-            search_for_migrations_directory(&temp_path)
-        );
-    }
-
-    #[test]
-    fn migration_directory_checks_parents() {
-        let dir = Builder::new().prefix("diesel").tempdir().unwrap();
-        let temp_path = dir.path().canonicalize().unwrap();
-        let migrations_path = temp_path.join("migrations");
-        let child_path = temp_path.join("child");
-
-        fs::create_dir(&child_path).unwrap();
-        fs::create_dir(&migrations_path).unwrap();
-
-        assert_eq!(
-            Ok(migrations_path),
-            search_for_migrations_directory(&child_path)
-        );
-    }
-
-    #[test]
-    fn migration_paths_in_directory_ignores_files() {
-        let dir = Builder::new().prefix("diesel").tempdir().unwrap();
-        let temp_path = dir.path().canonicalize().unwrap();
-        let migrations_path = temp_path.join("migrations");
-        let file_path = migrations_path.join("README.md");
-
-        fs::create_dir(&migrations_path).unwrap();
-        fs::File::create(&file_path).unwrap();
-
-        assert_eq!(
-            0,
-            migration_paths_in_directory(&migrations_path)
-                .unwrap()
-                .len()
-        );
-    }
-
-    #[test]
-    fn migration_paths_in_directory_ignores_dot_directories() {
-        let dir = Builder::new().prefix("diesel").tempdir().unwrap();
-        let temp_path = dir.path().canonicalize().unwrap();
-        let migrations_path = temp_path.join("migrations");
-        let dot_path = migrations_path.join(".hidden_dir");
-
-        fs::create_dir(&migrations_path).unwrap();
-        fs::create_dir(&dot_path).unwrap();
-
-        assert_eq!(
-            0,
-            migration_paths_in_directory(&migrations_path)
-                .unwrap()
-                .len()
-        );
-    }
 }
diff --git a/diesel_migrations/migrations_internals/src/migration.rs b/diesel_migrations/migrations_internals/src/migration.rs
deleted file mode 100644
index e03dbe07c493..000000000000
--- a/diesel_migrations/migrations_internals/src/migration.rs
+++ /dev/null
@@ -1,245 +0,0 @@
-use diesel::connection::SimpleConnection;
-use diesel::migration::*;
-
-use std::fmt;
-use std::path::{Path, PathBuf};
-
-#[allow(missing_debug_implementations)]
-#[derive(Clone, Copy)]
-pub struct MigrationName<'a> {
-    pub migration: &'a dyn Migration,
-}
-
-pub fn name(migration: &dyn Migration) -> MigrationName {
-    MigrationName { migration }
-}
-
-impl<'a> fmt::Display for MigrationName<'a> {
-    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        let file_name = self
-            .migration
-            .file_path()
-            .and_then(|file_path| file_path.file_name()?.to_str());
-        if let Some(name) = file_name {
-            f.write_str(name)?;
-        } else {
-            f.write_str(self.migration.version())?;
-        }
-        Ok(())
-    }
-}
-
-#[allow(missing_debug_implementations)]
-#[derive(Clone, Copy)]
-pub struct MigrationFileName<'a> {
-    pub migration: &'a dyn Migration,
-    pub sql_file: &'a str,
-}
-
-pub fn file_name<'a>(migration: &'a dyn Migration, sql_file: &'a str) -> MigrationFileName<'a> {
-    MigrationFileName {
-        migration,
-        sql_file,
-    }
-}
-
-impl<'a> fmt::Display for MigrationFileName<'a> {
-    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        if let Some(path) = self.migration.file_path() {
-            let fpath = path.join(self.sql_file);
-            f.write_str(fpath.to_str().unwrap_or("Invalid utf8 in filename"))
-        } else {
-            write!(f, "{}/{}", self.migration.version(), self.sql_file)
-        }
-    }
-}
-
-pub fn migration_from(path: PathBuf) -> Result<Box<dyn Migration>, MigrationError> {
-    #[cfg(feature = "barrel")]
-    match ::barrel::integrations::diesel::migration_from(&path) {
-        Some(migration) => return Ok(migration),
-        None => {}
-    }
-
-    if valid_sql_migration_directory(&path) {
-        let version = version_from_path(&path)?;
-        Ok(Box::new(SqlFileMigration(path, version)))
-    } else {
-        Err(MigrationError::UnknownMigrationFormat(path))
-    }
-}
-
-fn valid_sql_migration_directory(path: &Path) -> bool {
-    file_names(path)
-        .map(|files| files.contains(&"down.sql".into()) && files.contains(&"up.sql".into()))
-        .unwrap_or(false)
-}
-
-fn file_names(path: &Path) -> Result<Vec<String>, MigrationError> {
-    path.read_dir()?
-        .map(|entry| {
-            let file_name = entry?.file_name();
-
-            // FIXME(killercup): Decide whether to add MigrationError variant for this
-            match file_name.into_string() {
-                Ok(utf8_file_name) => Ok(utf8_file_name),
-                Err(original_os_string) => panic!(
-                    "Can't convert file name `{:?}` into UTF8 string",
-                    original_os_string
-                ),
-            }
-        })
-        .filter(|file_name| match *file_name {
-            Ok(ref name) => !name.starts_with('.'),
-            _ => true,
-        })
-        .collect()
-}
-
-#[doc(hidden)]
-pub fn version_from_path(path: &Path) -> Result<String, MigrationError> {
-    path.file_name()
-        .unwrap_or_else(|| panic!("Can't get file name from path `{:?}`", path))
-        .to_string_lossy()
-        .split('_')
-        .next()
-        .map(|s| Ok(s.replace('-', "")))
-        .unwrap_or_else(|| Err(MigrationError::UnknownMigrationFormat(path.to_path_buf())))
-}
-
-use std::fs::File;
-use std::io::Read;
-
-struct SqlFileMigration(PathBuf, String);
-
-impl Migration for SqlFileMigration {
-    fn file_path(&self) -> Option<&Path> {
-        Some(self.0.as_path())
-    }
-
-    fn version(&self) -> &str {
-        &self.1
-    }
-
-    fn run(&self, conn: &dyn SimpleConnection) -> Result<(), RunMigrationsError> {
-        run_sql_from_file(conn, &self.0.join("up.sql"))
-    }
-
-    fn revert(&self, conn: &dyn SimpleConnection) -> Result<(), RunMigrationsError> {
-        run_sql_from_file(conn, &self.0.join("down.sql"))
-    }
-}
-
-fn run_sql_from_file(conn: &dyn SimpleConnection, path: &Path) -> Result<(), RunMigrationsError> {
-    let mut sql = String::new();
-    let mut file = File::open(path)?;
-    file.read_to_string(&mut sql)?;
-
-    if sql.is_empty() {
-        return Err(RunMigrationsError::EmptyMigration);
-    }
-
-    conn.batch_execute(&sql)?;
-    Ok(())
-}
-
-#[cfg(test)]
-mod tests {
-    extern crate tempfile;
-
-    use super::{valid_sql_migration_directory, version_from_path};
-
-    use self::tempfile::Builder;
-    use std::fs;
-    use std::path::PathBuf;
-
-    #[test]
-    fn files_are_not_valid_sql_file_migrations() {
-        let dir = Builder::new().prefix("diesel").tempdir().unwrap();
-        let file_path = dir.path().join("12345");
-
-        fs::File::create(&file_path).unwrap();
-
-        assert!(!valid_sql_migration_directory(&file_path));
-    }
-
-    #[test]
-    fn directory_containing_exactly_up_sql_and_down_sql_is_valid_migration_dir() {
-        let tempdir = Builder::new().prefix("diesel").tempdir().unwrap();
-        let folder = tempdir.path().join("12345");
-
-        fs::create_dir(&folder).unwrap();
-        fs::File::create(folder.join("up.sql")).unwrap();
-        fs::File::create(folder.join("down.sql")).unwrap();
-
-        assert!(valid_sql_migration_directory(&folder));
-    }
-
-    #[test]
-    fn directory_containing_unknown_files_is_valid_migration_dir() {
-        let tempdir = Builder::new().prefix("diesel").tempdir().unwrap();
-        let folder = tempdir.path().join("12345");
-
-        fs::create_dir(&folder).unwrap();
-        fs::File::create(folder.join("up.sql")).unwrap();
-        fs::File::create(folder.join("down.sql")).unwrap();
-        fs::File::create(folder.join("foo")).unwrap();
-
-        assert!(valid_sql_migration_directory(&folder));
-    }
-
-    #[test]
-    fn files_beginning_with_dot_are_allowed() {
-        let tempdir = Builder::new().prefix("diesel").tempdir().unwrap();
-        let folder = tempdir.path().join("12345");
-
-        fs::create_dir(&folder).unwrap();
-        fs::File::create(folder.join("up.sql")).unwrap();
-        fs::File::create(folder.join("down.sql")).unwrap();
-        fs::File::create(folder.join(".foo")).unwrap();
-
-        assert!(valid_sql_migration_directory(&folder));
-    }
-
-    #[test]
-    fn empty_directory_is_not_valid_migration_dir() {
-        let tempdir = Builder::new().prefix("diesel").tempdir().unwrap();
-        let folder = tempdir.path().join("12345");
-
-        fs::create_dir(&folder).unwrap();
-
-        assert!(!valid_sql_migration_directory(&folder));
-    }
-
-    #[test]
-    fn directory_with_only_up_sql_is_not_valid_migration_dir() {
-        let tempdir = Builder::new().prefix("diesel").tempdir().unwrap();
-        let folder = tempdir.path().join("12345");
-
-        fs::create_dir(&folder).unwrap();
-        fs::File::create(folder.join("up.sql")).unwrap();
-
-        assert!(!valid_sql_migration_directory(&folder));
-    }
-
-    #[test]
-    fn migration_version_is_based_on_folder_name() {
-        let path = PathBuf::new().join("migrations").join("12345");
-
-        assert_eq!(Ok("12345".into()), version_from_path(&path));
-    }
-
-    #[test]
-    fn migration_version_allows_additional_naming() {
-        let path = PathBuf::new().join("migrations").join("54321_create_stuff");
-
-        assert_eq!(Ok("54321".into()), version_from_path(&path));
-    }
-
-    #[test]
-    fn migration_version_when_dir_doesnt_start_with_num_is_allowed() {
-        let path = PathBuf::new().join("migrations").join("create_stuff_12345");
-
-        assert_eq!(Ok("create".into()), version_from_path(&path));
-    }
-}
diff --git a/diesel_migrations/migrations_internals/src/schema.rs b/diesel_migrations/migrations_internals/src/schema.rs
deleted file mode 100644
index 36f6b47862d2..000000000000
--- a/diesel_migrations/migrations_internals/src/schema.rs
+++ /dev/null
@@ -1,6 +0,0 @@
-table! {
-    __diesel_schema_migrations (version) {
-        version -> VarChar,
-        run_on -> Timestamp,
-    }
-}
diff --git a/diesel_migrations/migrations_macros/Cargo.toml b/diesel_migrations/migrations_macros/Cargo.toml
index 9c93b44b42ce..cfbc28dc1e93 100644
--- a/diesel_migrations/migrations_macros/Cargo.toml
+++ b/diesel_migrations/migrations_macros/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "migrations_macros"
-version = "1.4.2"
+version = "2.0.0"
 authors = ["Sean Griffin <sean@seantheprogrammer.com>"]
 license = "MIT OR Apache-2.0"
 description = "Codegeneration macros for diesels embedded migrations"
@@ -12,7 +12,7 @@ quote = "1"
 proc-macro2 = "1"
 
 [dependencies.migrations_internals]
-version = "~1.4.0"
+version = "~2.0.0"
 path = "../migrations_internals"
 
 [dev-dependencies]
@@ -26,7 +26,7 @@ path = "../../diesel"
 default-features = false
 
 [dev-dependencies.diesel_migrations]
-version = "~1.4.0"
+version = "~2.0.0"
 path = "../"
 default-features = false
 
diff --git a/diesel_migrations/migrations_macros/src/embed_migrations.rs b/diesel_migrations/migrations_macros/src/embed_migrations.rs
index 3fff7e375372..72a052ecae3f 100644
--- a/diesel_migrations/migrations_macros/src/embed_migrations.rs
+++ b/diesel_migrations/migrations_macros/src/embed_migrations.rs
@@ -1,5 +1,5 @@
 use crate::migrations::migration_directory_from_given_path;
-use migrations_internals::{migration_paths_in_directory, version_from_path};
+use migrations_internals::{migrations_directories, version_from_string, TomlMetadata};
 use quote::quote;
 use std::error::Error;
 use std::fs::DirEntry;
@@ -13,61 +13,18 @@ pub fn expand(path: String) -> proc_macro2::TokenStream {
     };
     let migrations_expr =
         migration_directory_from_given_path(migrations_path_opt.as_ref().map(String::as_str))
-            .and_then(|path| migration_literals_from_path(&path));
-    let migrations_expr = match migrations_expr {
-        Ok(v) => v,
-        Err(e) => panic!("Error reading migrations: {}", e),
-    };
+            .unwrap();
+    let embeded_migrations = migration_literals_from_path(&migrations_expr).unwrap();
 
     quote! {
-        #[allow(dead_code)]
-        mod embedded_migrations {
-            extern crate diesel;
-            extern crate diesel_migrations;
-
-            use self::diesel_migrations::*;
-            use self::diesel::connection::SimpleConnection;
-            use std::io;
-
-            const ALL_MIGRATIONS: &[&Migration] = &[#(#migrations_expr),*];
-
-            struct EmbeddedMigration {
-                version: &'static str,
-                up_sql: &'static str,
-            }
-
-            impl Migration for EmbeddedMigration {
-                fn version(&self) -> &str {
-                    self.version
-                }
-
-                fn run(&self, conn: &SimpleConnection) -> Result<(), RunMigrationsError> {
-                    conn.batch_execute(self.up_sql).map_err(Into::into)
-                }
-
-                fn revert(&self, _conn: &SimpleConnection) -> Result<(), RunMigrationsError> {
-                    unreachable!()
-                }
-            }
-
-            pub fn run<C: MigrationConnection>(conn: &C) -> Result<(), RunMigrationsError> {
-                run_with_output(conn, &mut io::sink())
-            }
-
-            pub fn run_with_output<C: MigrationConnection>(
-                conn: &C,
-                out: &mut io::Write,
-            ) -> Result<(), RunMigrationsError> {
-                run_migrations(conn, ALL_MIGRATIONS.iter().map(|v| *v), out)
-            }
-        }
+        diesel_migrations::EmbeddedMigrations::new(&[#(#embeded_migrations,)*])
     }
 }
 
 fn migration_literals_from_path(
     path: &Path,
 ) -> Result<Vec<proc_macro2::TokenStream>, Box<dyn Error>> {
-    let mut migrations = migration_paths_in_directory(path)?;
+    let mut migrations = migrations_directories(path).collect::<Result<Vec<_>, _>>()?;
 
     migrations.sort_by_key(DirEntry::path);
 
@@ -78,12 +35,28 @@ fn migration_literals_from_path(
 }
 
 fn migration_literal_from_path(path: &Path) -> Result<proc_macro2::TokenStream, Box<dyn Error>> {
-    let version = version_from_path(path)?;
-    let sql_file = path.join("up.sql");
-    let sql_file_path = sql_file.to_str();
-
-    Ok(quote!(&EmbeddedMigration {
-        version: #version,
-        up_sql: include_str!(#sql_file_path),
-    }))
+    let name = path
+        .file_name()
+        .unwrap_or_else(|| panic!("Can't get file name from path `{:?}`", path))
+        .to_string_lossy();
+    if version_from_string(&name).is_none() {
+        panic!(
+            "Invalid migration directory, the directory's name should be \
+             <timestamp>_<name_of_migration>, and it should only contain \
+             up.sql and down.sql."
+        );
+    }
+    let up_sql = path.join("up.sql");
+    let up_sql_path = up_sql.to_str();
+    let down_sql = path.join("down.sql");
+    let down_sql_path = down_sql.to_str();
+    let metadata = TomlMetadata::read_from_file(&path.join("metadata.toml")).unwrap_or_default();
+    let run_in_transaction = metadata.run_in_transaction;
+
+    Ok(quote!(diesel_migrations::EmbeddedMigration::new(
+        include_str!(#up_sql_path),
+        include_str!(#down_sql_path),
+        diesel_migrations::EmbeddedName::new(#name),
+        diesel_migrations::TomlMetadataWrapper::new(#run_in_transaction)
+    )))
 }
diff --git a/diesel_migrations/migrations_macros/src/lib.rs b/diesel_migrations/migrations_macros/src/lib.rs
index eb07ac4c9b73..304df149fe73 100644
--- a/diesel_migrations/migrations_macros/src/lib.rs
+++ b/diesel_migrations/migrations_macros/src/lib.rs
@@ -29,45 +29,71 @@ mod migrations;
 
 use proc_macro::TokenStream;
 
-/// This macro will read your migrations at compile time, and embed a module you can use to execute
-/// them at runtime without the migration files being present on the file system. This is useful if
-/// you would like to use Diesel's migration infrastructure, but want to ship a single executable
+/// This macro will read your migrations at compile time, and create a constant value containing
+/// an embedded list of all your migrations as available at compile time.
+/// This is useful if you would like to use Diesel's migration infrastructure, but want to ship a single executable
 /// file (such as for embedded applications). It can also be used to apply migrations to an in
 /// memory database (Diesel does this for its own test suite).
 ///
 /// You can optionally pass the path to the migrations directory to this macro. When left
-/// unspecified, Diesel Codegen will search for the migrations directory in the same way that
+/// unspecified, Diesel will search for the migrations directory in the same way that
 /// Diesel CLI does. If specified, the path should be relative to the directory where `Cargo.toml`
 /// resides.
 ///
 /// # Examples
 ///
 /// ```rust
-/// # use diesel_migrations::embed_migrations;
+/// use diesel_migrations::{embed_migrations, EmbeddedMigrations, MigrationHarness};
+/// # use std::error::Error;
 /// # include!("../../../diesel/src/doctest_setup.rs");
-/// # table! {
-/// #   users {
-/// #       id -> Integer,
-/// #       name -> VarChar,
-/// #   }
+/// #
+/// # #[cfg(feature = "postgres")]
+/// # fn migration_connection() -> diesel::PgConnection {
+/// #    let connection_url = database_url_from_env("PG_DATABASE_URL");
+/// #    let conn = diesel::PgConnection::establish(&connection_url).unwrap();
+/// #    conn.begin_test_transaction().unwrap();
+/// #    conn
 /// # }
 /// #
+/// # #[cfg(feature = "sqlite")]
+/// # fn migration_connection() -> diesel::SqliteConnection {
+/// #    let connection_url = database_url_from_env("SQLITE_DATABASE_URL");
+/// #    let conn = diesel::SqliteConnection::establish(&connection_url).unwrap();
+/// #    conn.begin_test_transaction().unwrap();
+/// #    conn
+/// # }
+/// #
+/// # #[cfg(feature = "mysql")]
+/// # fn migration_connection() -> diesel::MysqlConnection {
+/// #    let connection_url = database_url_from_env("MYSQL_DATABASE_URL");
+/// #    let conn = diesel::MysqlConnection::establish(&connection_url).unwrap();
+/// #    conn
+/// # }
+/// #
+/// #
 /// # #[cfg(feature = "postgres")]
-/// # embed_migrations!("../../migrations/postgresql");
+/// pub const MIGRATIONS: EmbeddedMigrations = embed_migrations!("../../migrations/postgresql");
 /// # #[cfg(all(feature = "mysql", not(feature = "postgres")))]
-/// # embed_migrations!("../../migrations/mysql");
+/// # pub const MIGRATIONS: EmbeddedMigrations = embed_migrations!("../../migrations/mysql");
 /// # #[cfg(all(feature = "sqlite", not(any(feature = "postgres", feature = "mysql"))))]
-/// embed_migrations!("../../migrations/sqlite");
+/// # pub const MIGRATIONS: EmbeddedMigrations = embed_migrations!("../../migrations/sqlite");
+///
+/// # fn main() {
+/// #     let connection = migration_connection();
+/// #     run_migrations(&connection).unwrap();
+/// # }
 ///
-/// fn main() {
-///     let connection = establish_connection();
+/// fn run_migrations(connection: &impl MigrationHarness<DB>) -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
+/// #   #[cfg(feature = "mysql")]
+/// #   connection.revert_all_migrations(MIGRATIONS)?;
 ///
 ///     // This will run the necessary migrations.
-///     embedded_migrations::run(&connection);
+///     //
+///     // See the documentation for `MigrationHarness` for
+///     // all available methods.
+///     connection.run_pending_migrations(MIGRATIONS)?;
 ///
-///     // By default the output is thrown out. If you want to redirect it to stdout, you
-///     // should call embedded_migrations::run_with_output.
-///     embedded_migrations::run_with_output(&connection, &mut std::io::stdout());
+///     Ok(())
 /// }
 /// ```
 #[proc_macro]
diff --git a/diesel_migrations/migrations_macros/src/migrations.rs b/diesel_migrations/migrations_macros/src/migrations.rs
index 70ac4dbf1700..67c25ad83d32 100644
--- a/diesel_migrations/migrations_macros/src/migrations.rs
+++ b/diesel_migrations/migrations_macros/src/migrations.rs
@@ -6,7 +6,7 @@ use std::path::{Path, PathBuf};
 
 pub fn migration_directory_from_given_path(
     given_path: Option<&str>,
-) -> Result<PathBuf, Box<dyn Error>> {
+) -> Result<PathBuf, Box<dyn Error + Send + Sync>> {
     let cargo_toml_directory = env::var("CARGO_MANIFEST_DIR")?;
     let cargo_manifest_path = Path::new(&cargo_toml_directory);
     let migrations_path = given_path.as_ref().map(Path::new);
@@ -16,14 +16,19 @@ pub fn migration_directory_from_given_path(
 fn resolve_migrations_directory(
     cargo_manifest_dir: &Path,
     relative_path_to_migrations: Option<&Path>,
-) -> Result<PathBuf, Box<dyn Error>> {
+) -> Result<PathBuf, Box<dyn Error + Send + Sync>> {
     let result = match relative_path_to_migrations {
         Some(dir) => cargo_manifest_dir.join(dir),
         None => {
             // People commonly put their migrations in src/migrations
             // so start the search there rather than the project root
             let src_dir = cargo_manifest_dir.join("src");
-            search_for_migrations_directory(&src_dir)?
+            search_for_migrations_directory(&src_dir).ok_or_else(|| {
+                format!(
+                    "Failed to find migration directory in {}",
+                    src_dir.display()
+                )
+            })?
         }
     };
 
diff --git a/diesel_migrations/src/embedded_migrations.rs b/diesel_migrations/src/embedded_migrations.rs
new file mode 100644
index 000000000000..da11dd4d5ef4
--- /dev/null
+++ b/diesel_migrations/src/embedded_migrations.rs
@@ -0,0 +1,112 @@
+use std::fmt::Display;
+
+use crate::errors::RunMigrationsError;
+use crate::file_based_migrations::{DieselMigrationName, TomlMetadataWrapper};
+use diesel::backend::Backend;
+use diesel::migration::{Migration, MigrationName, MigrationSource, MigrationVersion, Result};
+
+/// A migration source that embeds migrations into the final binary
+///
+/// This source can be create via the [`embed_migrations!`](crate::embed_migrations!)
+/// at compile time.
+#[allow(missing_copy_implementations)]
+pub struct EmbeddedMigrations {
+    migrations: &'static [EmbeddedMigration],
+}
+
+impl EmbeddedMigrations {
+    #[doc(hidden)]
+    pub const fn new(migrations: &'static [EmbeddedMigration]) -> Self {
+        Self { migrations }
+    }
+}
+
+impl<DB: Backend> MigrationSource<DB> for EmbeddedMigrations {
+    fn migrations(&self) -> Result<Vec<Box<dyn Migration<DB>>>> {
+        Ok(self
+            .migrations
+            .iter()
+            .map(|m| Box::new(m) as Box<dyn Migration<DB>>)
+            .collect())
+    }
+}
+
+#[doc(hidden)]
+pub struct EmbeddedMigration {
+    up: &'static str,
+    down: &'static str,
+    name: EmbeddedName,
+    metadata: TomlMetadataWrapper,
+}
+
+impl EmbeddedMigration {
+    #[doc(hidden)]
+    pub const fn new(
+        up: &'static str,
+        down: &'static str,
+        name: EmbeddedName,
+        metadata: TomlMetadataWrapper,
+    ) -> Self {
+        Self {
+            up,
+            down,
+            name,
+            metadata,
+        }
+    }
+}
+
+#[derive(Clone, Copy)]
+#[doc(hidden)]
+pub struct EmbeddedName {
+    name: &'static str,
+}
+
+impl EmbeddedName {
+    #[doc(hidden)]
+    pub const fn new(name: &'static str) -> Self {
+        Self { name }
+    }
+}
+
+impl MigrationName for EmbeddedName {
+    fn version(&self) -> MigrationVersion {
+        migrations_internals::version_from_string(self.name)
+            .expect(
+                "This name contains a valid version. We checked this at compile time by our macro",
+            )
+            .into()
+    }
+}
+
+impl Display for EmbeddedName {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{}", self.name)
+    }
+}
+
+impl<'a, DB: Backend> Migration<DB> for &'a EmbeddedMigration {
+    fn run(&self, conn: &dyn diesel::connection::BoxableConnection<DB>) -> Result<()> {
+        Ok(conn.batch_execute(self.up).map_err(|e| {
+            let name = DieselMigrationName::from_name(self.name.name)
+                .expect("We have a vaild name here, we checked this in `embed_migration!`");
+            RunMigrationsError::QueryError(name, e)
+        })?)
+    }
+
+    fn revert(&self, conn: &dyn diesel::connection::BoxableConnection<DB>) -> Result<()> {
+        Ok(conn.batch_execute(self.down).map_err(|e| {
+            let name = DieselMigrationName::from_name(self.name.name)
+                .expect("We have a vaild name here, we checked this in `embed_migration!`");
+            RunMigrationsError::QueryError(name, e)
+        })?)
+    }
+
+    fn metadata(&self) -> &dyn diesel::migration::MigrationMetadata {
+        &self.metadata as &dyn diesel::migration::MigrationMetadata
+    }
+
+    fn name(&self) -> &dyn MigrationName {
+        &self.name
+    }
+}
diff --git a/diesel/src/migration/errors.rs b/diesel_migrations/src/errors.rs
similarity index 76%
rename from diesel/src/migration/errors.rs
rename to diesel_migrations/src/errors.rs
index c0983a71e0d5..512ff49afcf9 100644
--- a/diesel/src/migration/errors.rs
+++ b/diesel_migrations/src/errors.rs
@@ -7,7 +7,9 @@ use std::error::Error;
 use std::path::PathBuf;
 use std::{fmt, io};
 
-use crate::result;
+use diesel::migration::MigrationVersion;
+
+use crate::file_based_migrations::DieselMigrationName;
 
 /// Errors that occur while preparing to run migrations
 #[derive(Debug)]
@@ -20,7 +22,7 @@ pub enum MigrationError {
     /// General system IO error
     IoError(io::Error),
     /// Provided migration had an incompatible version number
-    UnknownMigrationVersion(String),
+    UnknownMigrationVersion(MigrationVersion<'static>),
     /// No migrations had to be/ could be run
     NoMigrationRun,
 }
@@ -81,41 +83,29 @@ impl From<io::Error> for MigrationError {
 #[non_exhaustive]
 pub enum RunMigrationsError {
     /// A general migration error occured
-    MigrationError(MigrationError),
+    MigrationError(DieselMigrationName, MigrationError),
     /// The provided migration included an invalid query
-    QueryError(result::Error),
+    QueryError(DieselMigrationName, diesel::result::Error),
     /// The provided migration was empty
-    EmptyMigration,
+    EmptyMigration(DieselMigrationName),
 }
 
 impl Error for RunMigrationsError {}
 
 impl fmt::Display for RunMigrationsError {
     fn fmt(&self, f: &mut fmt::Formatter) -> Result<(), fmt::Error> {
-        match *self {
-            RunMigrationsError::MigrationError(ref error) => write!(f, "Failed with: {}", error),
-            RunMigrationsError::QueryError(ref error) => write!(f, "Failed with: {}", error),
-            RunMigrationsError::EmptyMigration => {
-                write!(f, "Failed with: Attempted to run an empty migration.")
+        match self {
+            RunMigrationsError::MigrationError(v, err) => {
+                write!(f, "Failed to run {} with: {}", v, err)
             }
+            RunMigrationsError::QueryError(v, err) => {
+                write!(f, "Failed to run {} with: {}", v, err)
+            }
+            RunMigrationsError::EmptyMigration(v) => write!(
+                f,
+                "Failed to run {} with: Attempted to run an empty migration.",
+                v
+            ),
         }
     }
 }
-
-impl From<MigrationError> for RunMigrationsError {
-    fn from(e: MigrationError) -> Self {
-        RunMigrationsError::MigrationError(e)
-    }
-}
-
-impl From<result::Error> for RunMigrationsError {
-    fn from(e: result::Error) -> Self {
-        RunMigrationsError::QueryError(e)
-    }
-}
-
-impl From<io::Error> for RunMigrationsError {
-    fn from(e: io::Error) -> Self {
-        RunMigrationsError::MigrationError(e.into())
-    }
-}
diff --git a/diesel_migrations/src/file_based_migrations.rs b/diesel_migrations/src/file_based_migrations.rs
new file mode 100644
index 000000000000..d8d38b54623a
--- /dev/null
+++ b/diesel_migrations/src/file_based_migrations.rs
@@ -0,0 +1,380 @@
+use std::fmt::Display;
+use std::fs::{DirEntry, File};
+use std::io::Read;
+use std::path::{Path, PathBuf};
+
+use diesel::backend::Backend;
+use diesel::connection::BoxableConnection;
+use diesel::migration::{
+    self, Migration, MigrationMetadata, MigrationName, MigrationSource, MigrationVersion,
+};
+use migrations_internals::TomlMetadata;
+
+use crate::errors::{MigrationError, RunMigrationsError};
+
+/// A migration source based on a migration directory in the file system
+///
+/// A valid migration directroy contains a sub folder per migration.
+/// Each migration folder contains a `up.sql` file containing the migration itself
+/// and a `down.sql` file containing the nessesary SQL to revert the migration.
+/// Additionaly each folder can contain a `metadata.toml` file controling how the
+/// individual migration should be handled by the migration harness.
+///
+/// To embed an existing migration folder into the final binary see
+/// [`embed_migrations!`](crate::embed_migrations!).
+///
+/// ## Example
+///
+/// ```text
+/// # Directory Structure
+/// - 20151219180527_create_users
+///     - up.sql
+///     - down.sql
+/// - 20160107082941_create_posts
+///     - up.sql
+///     - down.sql
+/// ```
+///
+/// ```sql
+/// -- 20151219180527_create_users/up.sql
+/// CREATE TABLE users (
+///   id SERIAL PRIMARY KEY,
+///   name VARCHAR NOT NULL,
+///   hair_color VARCHAR
+/// );
+/// ```
+///
+/// ```sql
+/// -- 20151219180527_create_users/down.sql
+/// DROP TABLE users;
+/// ```
+///
+/// ```sql
+/// -- 20160107082941_create_posts/up.sql
+/// CREATE TABLE posts (
+///   id SERIAL PRIMARY KEY,
+///   user_id INTEGER NOT NULL,
+///   title VARCHAR NOT NULL,
+///   body TEXT
+/// );
+/// ```
+///
+/// ```sql
+/// -- 20160107082941_create_posts/down.sql
+/// DROP TABLE posts;
+/// ```
+///
+/// ```toml
+/// # 20160107082941_create_posts/metadata.toml
+///
+/// # specifices if a migration is executed inside a
+/// # transaction or not. This configuration is optional
+/// # by default all migrations are run in transactions.
+/// #
+/// # For certain types of migrations, like creating an
+/// # index onto a existing column, it is required
+/// # to set this to false
+/// run_in_transaction = true
+/// ```
+#[derive(Clone)]
+pub struct FileBasedMigrations {
+    base_path: PathBuf,
+}
+
+impl FileBasedMigrations {
+    /// Create a new file based migration source based on a specific path
+    ///
+    /// This methods fails if the path passed as argument is no valid migration directory
+    pub fn from_path(path: impl AsRef<Path>) -> Result<Self, MigrationError> {
+        for dir in migrations_directories(path.as_ref()) {
+            let path = dir?.path();
+            if !migrations_internals::valid_sql_migration_directory(&path) {
+                return Err(MigrationError::UnknownMigrationFormat(path));
+            }
+        }
+        Ok(Self {
+            base_path: path.as_ref().to_path_buf(),
+        })
+    }
+
+    /// Create a new file based migration source by searching the migration diretcory
+    ///
+    /// This method looks in the current and all parent directories for a folder named
+    /// `migrations`
+    ///
+    /// This method fails if no valid migration directory is found
+    pub fn find_migrations_directory() -> Result<Self, MigrationError> {
+        Self::find_migrations_directory_in_path(&std::env::current_dir()?)
+    }
+
+    /// Create a new file based migration source by searching a give path for the migration
+    /// directory
+    ///
+    /// This method looks in the passed directory and all parent directories for a folder
+    /// named `migrations`
+    ///
+    /// This method fails if no valid migration directory is found
+    pub fn find_migrations_directory_in_path(
+        path: impl AsRef<Path>,
+    ) -> Result<Self, MigrationError> {
+        let migrations_directory = search_for_migrations_directory(path.as_ref())?;
+        Self::from_path(&migrations_directory)
+    }
+
+    #[doc(hidden)]
+    pub fn path(&self) -> &Path {
+        &self.base_path
+    }
+}
+
+fn search_for_migrations_directory(path: &Path) -> Result<PathBuf, MigrationError> {
+    migrations_internals::search_for_migrations_directory(path)
+        .ok_or_else(|| MigrationError::MigrationDirectoryNotFound(path.to_path_buf()))
+}
+
+fn migrations_directories<'a>(
+    path: &'a Path,
+) -> impl Iterator<Item = Result<DirEntry, MigrationError>> + 'a {
+    migrations_internals::migrations_directories(path).map(move |e| e.map_err(Into::into))
+}
+
+fn migrations_in_directory<'a, DB: Backend>(
+    path: &'a Path,
+) -> impl Iterator<Item = Result<SqlFileMigration, MigrationError>> + 'a {
+    migrations_directories(path).map(|entry| SqlFileMigration::from_path(&entry?.path()))
+}
+
+impl<DB: Backend> MigrationSource<DB> for FileBasedMigrations {
+    fn migrations(&self) -> migration::Result<Vec<Box<dyn Migration<DB>>>> {
+        migrations_in_directory::<DB>(&self.base_path)
+            .map(|r| Ok(Box::new(r?) as Box<dyn Migration<DB>>))
+            .collect()
+    }
+}
+
+struct SqlFileMigration {
+    base_path: PathBuf,
+    metadata: TomlMetadataWrapper,
+    name: DieselMigrationName,
+}
+
+impl SqlFileMigration {
+    fn from_path(path: &Path) -> Result<Self, MigrationError> {
+        if migrations_internals::valid_sql_migration_directory(path) {
+            let metadata = TomlMetadataWrapper(
+                TomlMetadata::read_from_file(&path.join("metadata.toml")).unwrap_or_default(),
+            );
+            Ok(Self {
+                base_path: path.to_path_buf(),
+                metadata,
+                name: DieselMigrationName::from_path(path)?,
+            })
+        } else {
+            Err(MigrationError::UnknownMigrationFormat(path.to_path_buf()))
+        }
+    }
+}
+
+impl<DB: Backend> Migration<DB> for SqlFileMigration {
+    fn run(&self, conn: &dyn BoxableConnection<DB>) -> migration::Result<()> {
+        Ok(run_sql_from_file(
+            conn,
+            &self.base_path.join("up.sql"),
+            &self.name,
+        )?)
+    }
+
+    fn revert(&self, conn: &dyn BoxableConnection<DB>) -> migration::Result<()> {
+        Ok(run_sql_from_file(
+            conn,
+            &self.base_path.join("down.sql"),
+            &self.name,
+        )?)
+    }
+
+    fn metadata(&self) -> &dyn MigrationMetadata {
+        &self.metadata
+    }
+
+    fn name(&self) -> &dyn MigrationName {
+        &self.name
+    }
+}
+
+#[derive(Debug, PartialEq)]
+pub struct DieselMigrationName {
+    name: String,
+    version: MigrationVersion<'static>,
+}
+
+impl Clone for DieselMigrationName {
+    fn clone(&self) -> Self {
+        Self {
+            name: self.name.clone(),
+            version: self.version.as_owned(),
+        }
+    }
+}
+
+impl DieselMigrationName {
+    fn from_path(path: &Path) -> Result<Self, MigrationError> {
+        let name = path
+            .file_name()
+            .ok_or_else(|| MigrationError::UnknownMigrationFormat(path.to_path_buf()))?
+            .to_string_lossy();
+        Self::from_name(&name)
+    }
+
+    pub(crate) fn from_name(name: &str) -> Result<Self, MigrationError> {
+        let version = migrations_internals::version_from_string(&name)
+            .ok_or_else(|| MigrationError::UnknownMigrationFormat(PathBuf::from(name)))?;
+        Ok(Self {
+            name: name.to_owned(),
+            version: MigrationVersion::from(version),
+        })
+    }
+}
+
+impl MigrationName for DieselMigrationName {
+    fn version(&self) -> MigrationVersion {
+        self.version.as_owned()
+    }
+}
+
+impl Display for DieselMigrationName {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{}", self.name)
+    }
+}
+
+#[derive(Default)]
+#[doc(hidden)]
+pub struct TomlMetadataWrapper(TomlMetadata);
+
+impl TomlMetadataWrapper {
+    #[doc(hidden)]
+    pub const fn new(run_in_transaction: bool) -> Self {
+        Self(TomlMetadata::new(run_in_transaction))
+    }
+}
+
+impl MigrationMetadata for TomlMetadataWrapper {
+    fn run_in_transaction(&self) -> bool {
+        self.0.run_in_transaction
+    }
+}
+
+fn run_sql_from_file<DB: Backend>(
+    conn: &dyn BoxableConnection<DB>,
+    path: &Path,
+    name: &DieselMigrationName,
+) -> Result<(), RunMigrationsError> {
+    let map_io_err = |e| RunMigrationsError::MigrationError(name.clone(), MigrationError::from(e));
+
+    let mut sql = String::new();
+    let mut file = File::open(path).map_err(map_io_err)?;
+    file.read_to_string(&mut sql).map_err(map_io_err)?;
+
+    if sql.is_empty() {
+        return Err(RunMigrationsError::EmptyMigration(name.clone()));
+    }
+
+    conn.batch_execute(&sql)
+        .map_err(|e| RunMigrationsError::QueryError(name.clone(), e))?;
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    extern crate tempfile;
+
+    use super::*;
+
+    use self::tempfile::Builder;
+    use std::fs;
+
+    #[test]
+    fn migration_directory_not_found_if_no_migration_dir_exists() {
+        let dir = Builder::new().prefix("diesel").tempdir().unwrap();
+
+        assert_eq!(
+            Err(MigrationError::MigrationDirectoryNotFound(
+                dir.path().into()
+            )),
+            search_for_migrations_directory(dir.path())
+        );
+    }
+
+    #[test]
+    fn migration_directory_defaults_to_pwd_slash_migrations() {
+        let dir = Builder::new().prefix("diesel").tempdir().unwrap();
+        let temp_path = dir.path().canonicalize().unwrap();
+        let migrations_path = temp_path.join("migrations");
+
+        fs::create_dir(&migrations_path).unwrap();
+
+        assert_eq!(
+            Ok(migrations_path),
+            search_for_migrations_directory(&temp_path)
+        );
+    }
+
+    #[test]
+    fn migration_directory_checks_parents() {
+        let dir = Builder::new().prefix("diesel").tempdir().unwrap();
+        let temp_path = dir.path().canonicalize().unwrap();
+        let migrations_path = temp_path.join("migrations");
+        let child_path = temp_path.join("child");
+
+        fs::create_dir(&child_path).unwrap();
+        fs::create_dir(&migrations_path).unwrap();
+
+        assert_eq!(
+            Ok(migrations_path),
+            search_for_migrations_directory(&child_path)
+        );
+    }
+
+    #[cfg(feature = "sqlite")]
+    type Backend = diesel::sqlite::Sqlite;
+
+    #[cfg(feature = "postgres")]
+    type Backend = diesel::pg::Pg;
+
+    #[cfg(feature = "mysql")]
+    type Backend = diesel::mysql::Mysql;
+
+    #[test]
+    fn migration_paths_in_directory_ignores_files() {
+        let dir = Builder::new().prefix("diesel").tempdir().unwrap();
+        let temp_path = dir.path().canonicalize().unwrap();
+        let migrations_path = temp_path.join("migrations");
+        let file_path = migrations_path.join("README.md");
+
+        fs::create_dir(&migrations_path).unwrap();
+        fs::File::create(&file_path).unwrap();
+
+        let migrations = migrations_in_directory::<Backend>(&migrations_path)
+            .collect::<Result<Vec<_>, _>>()
+            .unwrap();
+
+        assert_eq!(0, migrations.len());
+    }
+
+    #[test]
+    fn migration_paths_in_directory_ignores_dot_directories() {
+        let dir = Builder::new().prefix("diesel").tempdir().unwrap();
+        let temp_path = dir.path().canonicalize().unwrap();
+        let migrations_path = temp_path.join("migrations");
+        let dot_path = migrations_path.join(".hidden_dir");
+
+        fs::create_dir(&migrations_path).unwrap();
+        fs::create_dir(&dot_path).unwrap();
+
+        let migrations = migrations_in_directory::<Backend>(&migrations_path)
+            .collect::<Result<Vec<_>, _>>()
+            .unwrap();
+
+        assert_eq!(0, migrations.len());
+    }
+}
diff --git a/diesel_migrations/src/lib.rs b/diesel_migrations/src/lib.rs
index 93c85f64ce91..158f6a43948a 100644
--- a/diesel_migrations/src/lib.rs
+++ b/diesel_migrations/src/lib.rs
@@ -7,6 +7,7 @@
     clippy::option_map_unwrap_or
 )]
 #![warn(
+    missing_docs,
     clippy::wrong_pub_self_convention,
     clippy::mut_mut,
     clippy::non_ascii_literal,
@@ -28,7 +29,7 @@
 //! migrations directory in the current directory and its parents, stopping when it finds the
 //! directory containing `Cargo.toml`.
 //!
-//! Individual migrations should be a folder containing exactly two files, `up.sql` and `down.sql`.
+//! Individual migrations should be a folder containing two files, `up.sql` and `down.sql`.
 //! `up.sql` will be used to run the migration, while `down.sql` will be used for reverting it. The
 //! folder itself should have the structure `{version}_{migration_name}`. It is recommended that
 //! you use the timestamp of creation for the version.
@@ -37,98 +38,20 @@
 //! and executed with code, for example right after establishing a database connection.
 //! For more information, consult the [`embed_migrations!`](macro.embed_migrations.html) macro.
 //!
-//! ## Example
-//!
-//! ```text
-//! # Directory Structure
-//! - 20151219180527_create_users
-//!     - up.sql
-//!     - down.sql
-//! - 20160107082941_create_posts
-//!     - up.sql
-//!     - down.sql
-//! ```
-//!
-//! ```sql
-//! -- 20151219180527_create_users/up.sql
-//! CREATE TABLE users (
-//!   id SERIAL PRIMARY KEY,
-//!   name VARCHAR NOT NULL,
-//!   hair_color VARCHAR
-//! );
-//! ```
-//!
-//! ```sql
-//! -- 20151219180527_create_users/down.sql
-//! DROP TABLE users;
-//! ```
-//!
-//! ```sql
-//! -- 20160107082941_create_posts/up.sql
-//! CREATE TABLE posts (
-//!   id SERIAL PRIMARY KEY,
-//!   user_id INTEGER NOT NULL,
-//!   title VARCHAR NOT NULL,
-//!   body TEXT
-//! );
-//! ```
-//!
-//! ```sql
-//! -- 20160107082941_create_posts/down.sql
-//! DROP TABLE posts;
-//! ```
 
-extern crate migrations_internals;
-extern crate migrations_macros;
-#[doc(inline)]
-pub use migrations_internals::any_pending_migrations;
-#[doc(inline)]
-pub use migrations_internals::find_migrations_directory;
-#[doc(inline)]
-pub use migrations_internals::mark_migrations_in_directory;
-#[doc(inline)]
-pub use migrations_internals::migration_from;
-#[doc(inline)]
-pub use migrations_internals::migration_paths_in_directory;
-#[doc(inline)]
-pub use migrations_internals::name;
-#[doc(inline)]
-pub use migrations_internals::revert_latest_migration;
-#[doc(inline)]
-pub use migrations_internals::revert_latest_migration_in_directory;
-#[doc(inline)]
-pub use migrations_internals::revert_migration_with_version;
-#[doc(inline)]
-pub use migrations_internals::run_migration_with_version;
-#[doc(inline)]
-pub use migrations_internals::run_migrations;
-#[doc(inline)]
-pub use migrations_internals::run_pending_migrations;
-#[doc(inline)]
-pub use migrations_internals::run_pending_migrations_in_directory;
-#[doc(inline)]
-pub use migrations_internals::search_for_migrations_directory;
-#[doc(inline)]
-pub use migrations_internals::setup_database;
-#[doc(inline)]
-pub use migrations_internals::version_from_path;
-#[doc(inline)]
-pub use migrations_internals::Migration;
-#[doc(inline)]
-pub use migrations_internals::MigrationConnection;
-#[doc(inline)]
-pub use migrations_internals::MigrationError;
-#[doc(inline)]
-pub use migrations_internals::MigrationName;
-#[doc(inline)]
-pub use migrations_internals::RunMigrationsError;
-#[doc(hidden)]
-pub use migrations_macros::*;
+mod embedded_migrations;
+mod errors;
+mod file_based_migrations;
+mod migration_harness;
 
-pub mod connection {
-    #[doc(inline)]
-    pub use migrations_internals::connection::MigrationConnection;
-}
-
-#[doc(inline)]
+pub use crate::embedded_migrations::EmbeddedMigrations;
+pub use crate::file_based_migrations::FileBasedMigrations;
+pub use crate::migration_harness::{HarnessWithOutput, MigrationHarness};
 pub use migrations_macros::embed_migrations;
+
+#[doc(hidden)]
+pub use crate::embedded_migrations::{EmbeddedMigration, EmbeddedName};
+#[doc(hidden)]
+pub use crate::errors::MigrationError;
+#[doc(hidden)]
+pub use crate::file_based_migrations::TomlMetadataWrapper;
diff --git a/diesel_migrations/src/migration_harness.rs b/diesel_migrations/src/migration_harness.rs
new file mode 100644
index 000000000000..d7de53443399
--- /dev/null
+++ b/diesel_migrations/src/migration_harness.rs
@@ -0,0 +1,271 @@
+use diesel::backend::Backend;
+use diesel::dsl;
+use diesel::expression::bound::Bound;
+use diesel::insertable::ColumnInsertValue;
+use diesel::migration::{
+    Migration, MigrationConnection, MigrationSource, MigrationVersion, Result,
+};
+use diesel::prelude::*;
+use diesel::query_builder::{InsertStatement, ValuesClause};
+use diesel::query_dsl::methods::ExecuteDsl;
+use diesel::query_dsl::LoadQuery;
+use diesel::sql_types::Text;
+use std::cell::RefCell;
+use std::collections::HashMap;
+use std::io::Write;
+
+use crate::errors::MigrationError;
+
+diesel::table! {
+    __diesel_schema_migrations (version) {
+        version -> VarChar,
+        run_on -> Timestamp,
+    }
+}
+
+/// A migration harness is an entity which applies migration to an existing database
+pub trait MigrationHarness<DB: Backend> {
+    /// Checks if the database represented by the current harness has unapplied migrations
+    fn has_pending_migration<S: MigrationSource<DB>>(&self, source: S) -> Result<bool> {
+        self.pending_migrations(source).map(|p| !p.is_empty())
+    }
+
+    /// Execute all unapplied migrations for a given migration source
+    fn run_pending_migrations<S: MigrationSource<DB>>(
+        &self,
+        source: S,
+    ) -> Result<Vec<MigrationVersion>> {
+        self.run_migrations(&self.pending_migrations(source)?)
+    }
+
+    /// Execute all migrations in the given list
+    ///
+    /// This method does not check if a certain migration was already applied or not
+    #[doc(hidden)]
+    fn run_migrations(
+        &self,
+        migrations: &[Box<dyn Migration<DB>>],
+    ) -> Result<Vec<MigrationVersion>> {
+        migrations.iter().map(|m| self.run_migration(m)).collect()
+    }
+
+    /// Execute the next migration from the given migration source
+    fn run_next_migration<S: MigrationSource<DB>>(&self, source: S) -> Result<MigrationVersion> {
+        let pending_migrations = self.pending_migrations(source)?;
+        let next_migration = pending_migrations
+            .first()
+            .ok_or_else(|| MigrationError::NoMigrationRun)?;
+        self.run_migration(next_migration)
+    }
+
+    /// Revert all applied migrations from a given migration source
+    fn revert_all_migrations<S: MigrationSource<DB>>(
+        &self,
+        source: S,
+    ) -> Result<Vec<MigrationVersion>> {
+        let applied_versions = self.applied_migrations()?;
+        let mut migrations = source
+            .migrations()?
+            .into_iter()
+            .map(|m| (m.name().version().as_owned(), m))
+            .collect::<HashMap<_, _>>();
+
+        applied_versions
+            .into_iter()
+            .map(|version| {
+                let migration_to_revert = migrations
+                    .remove(&version)
+                    .ok_or_else(|| MigrationError::UnknownMigrationVersion(version))?;
+                self.revert_migration(&migration_to_revert)
+            })
+            .collect()
+    }
+
+    /// Revert the last migration from a given migration source
+    ///
+    /// This method returns a error if the given migration source does not
+    /// contain the last applied migration
+    fn revert_last_migration<S: MigrationSource<DB>>(&self, source: S) -> Result<MigrationVersion> {
+        let applied_versions = self.applied_migrations()?;
+        let migrations = source.migrations()?;
+        let last_migration_version = applied_versions
+            .first()
+            .ok_or_else(|| MigrationError::NoMigrationRun)?;
+        let migration_to_revert = migrations
+            .iter()
+            .find(|m| m.name().version() == *last_migration_version)
+            .ok_or_else(|| {
+                MigrationError::UnknownMigrationVersion(last_migration_version.as_owned())
+            })?;
+        self.revert_migration(migration_to_revert)
+    }
+
+    /// Get a list of non applied migrations for a specific migration source
+    ///
+    /// The returned migration list is sorted in ascending order by the individual version
+    /// of each migration
+    fn pending_migrations<S: MigrationSource<DB>>(
+        &self,
+        source: S,
+    ) -> Result<Vec<Box<dyn Migration<DB>>>> {
+        let applied_versions = self.applied_migrations()?;
+        let mut migrations = source
+            .migrations()?
+            .into_iter()
+            .map(|m| (m.name().version().as_owned(), m))
+            .collect::<HashMap<_, _>>();
+
+        for applied_version in applied_versions {
+            migrations.remove(&applied_version);
+        }
+
+        let mut migrations = migrations.into_iter().map(|(_, m)| m).collect::<Vec<_>>();
+
+        migrations.sort_unstable_by(|a, b| a.name().version().cmp(&b.name().version()));
+
+        Ok(migrations)
+    }
+
+    /// Apply a single migration
+    ///
+    /// Types implementing this trait should call [`Migration::run`] internally and record
+    /// that a specific migration version was executed afterwards.
+    fn run_migration(&self, migration: &dyn Migration<DB>) -> Result<MigrationVersion<'static>>;
+
+    /// Revert a single migration
+    ///
+    /// Types implementing this trait should call [`Migration::revert`] internally
+    /// and record that a specific migration version was reverted afterwards.
+    fn revert_migration(&self, migration: &dyn Migration<DB>) -> Result<MigrationVersion<'static>>;
+
+    /// Get a list of already applied migration versions
+    fn applied_migrations(&self) -> Result<Vec<MigrationVersion<'static>>>;
+}
+
+impl<C, DB> MigrationHarness<DB> for C
+where
+    DB: Backend,
+    C: Connection<Backend = DB> + MigrationConnection + 'static,
+    dsl::Order<
+        dsl::Select<__diesel_schema_migrations::table, __diesel_schema_migrations::version>,
+        dsl::Desc<__diesel_schema_migrations::version>,
+    >: LoadQuery<C, MigrationVersion<'static>>,
+    for<'a> InsertStatement<
+        __diesel_schema_migrations::table,
+        ValuesClause<
+            ColumnInsertValue<
+                __diesel_schema_migrations::version,
+                Bound<Text, MigrationVersion<'a>>,
+            >,
+            __diesel_schema_migrations::table,
+        >,
+    >: ExecuteDsl<C>,
+{
+    fn run_migration(&self, migration: &dyn Migration<DB>) -> Result<MigrationVersion<'static>> {
+        let apply_migration = || -> Result<()> {
+            migration.run(self)?;
+            diesel::insert_into(__diesel_schema_migrations::table)
+                .values(__diesel_schema_migrations::version.eq(migration.name().version()))
+                .execute(self)?;
+            Ok(())
+        };
+
+        if migration.metadata().run_in_transaction() {
+            self.transaction(apply_migration)?;
+        } else {
+            apply_migration()?;
+        }
+        Ok(migration.name().version().as_owned())
+    }
+
+    fn revert_migration(&self, migration: &dyn Migration<DB>) -> Result<MigrationVersion<'static>> {
+        let revert_migration = || -> Result<()> {
+            migration.revert(self)?;
+            diesel::delete(__diesel_schema_migrations::table.find(migration.name().version()))
+                .execute(self)?;
+            Ok(())
+        };
+
+        if migration.metadata().run_in_transaction() {
+            self.transaction(revert_migration)?;
+        } else {
+            revert_migration()?;
+        }
+        Ok(migration.name().version().as_owned())
+    }
+
+    fn applied_migrations(&self) -> Result<Vec<MigrationVersion<'static>>> {
+        setup_database(self)?;
+        Ok(__diesel_schema_migrations::table
+            .select(__diesel_schema_migrations::version)
+            .order(__diesel_schema_migrations::version.desc())
+            .load(self)?)
+    }
+}
+
+/// A migration harness that writes messages
+/// into some output for each applied/reverted migration
+pub struct HarnessWithOutput<'a, C, W> {
+    connection: &'a C,
+    output: RefCell<W>,
+}
+
+impl<'a, C, W> HarnessWithOutput<'a, C, W> {
+    /// Create a new `HarnessWithOutput` that writes to a specific writer
+    pub fn new<DB>(harness: &'a C, output: W) -> Self
+    where
+        C: MigrationHarness<DB>,
+        DB: Backend,
+        W: Write,
+    {
+        Self {
+            connection: harness,
+            output: RefCell::new(output),
+        }
+    }
+}
+
+impl<'a, C> HarnessWithOutput<'a, C, std::io::Stdout> {
+    /// Create a new `HarnessWithOutput` that writes to stdout
+    pub fn write_to_stdout<DB>(harness: &'a C) -> Self
+    where
+        C: MigrationHarness<DB>,
+        DB: Backend,
+    {
+        Self {
+            connection: harness,
+            output: RefCell::new(std::io::stdout()),
+        }
+    }
+}
+
+impl<'a, C, W, DB> MigrationHarness<DB> for HarnessWithOutput<'a, C, W>
+where
+    W: Write,
+    C: MigrationHarness<DB>,
+    DB: Backend,
+{
+    fn run_migration(&self, migration: &dyn Migration<DB>) -> Result<MigrationVersion<'static>> {
+        if migration.name().version() != MigrationVersion::from("00000000000000") {
+            let mut output = self.output.try_borrow_mut()?;
+            writeln!(output, "Running migration {}", migration.name())?;
+        }
+        self.connection.run_migration(migration)
+    }
+
+    fn revert_migration(&self, migration: &dyn Migration<DB>) -> Result<MigrationVersion<'static>> {
+        if migration.name().version() != MigrationVersion::from("00000000000000") {
+            let mut output = self.output.try_borrow_mut()?;
+            writeln!(output, "Rolling back migration {}", migration.name())?;
+        }
+        self.connection.revert_migration(migration)
+    }
+
+    fn applied_migrations(&self) -> Result<Vec<MigrationVersion<'static>>> {
+        self.connection.applied_migrations()
+    }
+}
+
+fn setup_database<Conn: MigrationConnection>(conn: &Conn) -> QueryResult<usize> {
+    conn.setup()
+}
diff --git a/examples/postgres/advanced-blog-cli/Cargo.toml b/examples/postgres/advanced-blog-cli/Cargo.toml
index 5ca7b74564ba..131cfe34514e 100644
--- a/examples/postgres/advanced-blog-cli/Cargo.toml
+++ b/examples/postgres/advanced-blog-cli/Cargo.toml
@@ -14,7 +14,7 @@ tempfile = "3.1.0"
 
 [dev-dependencies]
 assert_matches = "1.1"
-diesel_migrations = { version = "1.4.0", features = ["postgres"], path = "../../../diesel_migrations" }
+diesel_migrations = { version = "2.0.0", features = ["postgres"], path = "../../../diesel_migrations" }
 lazy_static = "1.0"
 
 [[bin]]
diff --git a/migrations/mysql/20170211150830_index_columns_used_in_benchmarks/down.sql b/migrations/mysql/20170211150830_index_columns_used_in_benchmarks/down.sql
index c19dc97fecc6..f64472a9af3d 100644
--- a/migrations/mysql/20170211150830_index_columns_used_in_benchmarks/down.sql
+++ b/migrations/mysql/20170211150830_index_columns_used_in_benchmarks/down.sql
@@ -1,2 +1,2 @@
-DROP INDEX users_hair_color;
-DROP INDEX posts_user_id;
+DROP INDEX users_hair_color ON users;
+DROP INDEX posts_user_id ON posts;
diff --git a/rust-toolchain b/rust-toolchain
index 32b7211cb618..9db5ea12f522 100644
--- a/rust-toolchain
+++ b/rust-toolchain
@@ -1,1 +1,1 @@
-1.40.0
+1.48.0


Test patch:
diff --git a/diesel_cli/tests/migration_generate.rs b/diesel_cli/tests/migration_generate.rs
index 0b59f4f3ee4a..a40fe25a60ae 100644
--- a/diesel_cli/tests/migration_generate.rs
+++ b/diesel_cli/tests/migration_generate.rs
@@ -2,7 +2,7 @@ use chrono::prelude::*;
 use regex::Regex;
 
 use crate::support::project;
-use migrations_internals::TIMESTAMP_FORMAT;
+pub static TIMESTAMP_FORMAT: &str = "%Y-%m-%d-%H%M%S";
 
 #[test]
 fn migration_generate_creates_a_migration_with_the_proper_name() {
diff --git a/diesel_cli/tests/migration_list.rs b/diesel_cli/tests/migration_list.rs
index c13e5edf0246..bc4c94e23f65 100644
--- a/diesel_cli/tests/migration_list.rs
+++ b/diesel_cli/tests/migration_list.rs
@@ -3,7 +3,7 @@ use std::thread::sleep;
 use std::time::Duration;
 
 use crate::support::{database, project};
-use migrations_internals::TIMESTAMP_FORMAT;
+pub static TIMESTAMP_FORMAT: &str = "%Y-%m-%d-%H%M%S";
 
 #[test]
 fn migration_list_lists_pending_applied_migrations() {
diff --git a/diesel_cli/tests/migration_redo.rs b/diesel_cli/tests/migration_redo.rs
index 182faf7e5715..02b54d812635 100644
--- a/diesel_cli/tests/migration_redo.rs
+++ b/diesel_cli/tests/migration_redo.rs
@@ -146,30 +146,6 @@ Running migration 12345_create_users_table
     );
 }
 
-#[test]
-fn output_contains_path_to_migration_script() {
-    let p = project("output_contains_path_to_migration_script")
-        .folder("migrations")
-        .build();
-    p.create_migration(
-        "output_contains_path_to_migration_script",
-        "CREATE TABLE users (id INTEGER PRIMARY KEY);",
-        "DROP TABLE users};",
-    );
-
-    // Make sure the project is setup
-    p.command("setup").run();
-
-    let result = p.command("migration").arg("redo").run();
-
-    assert!(!result.is_success(), "Result was successful {:?}", result);
-    assert!(
-        result.stdout().contains("down.sql"),
-        "Unexpected stdout {}",
-        result.stdout()
-    );
-}
-
 #[test]
 fn error_migrations_fails() {
     let p = project("redo_error_migrations_fails")
@@ -187,7 +163,9 @@ fn error_migrations_fails() {
     let result = p.command("migration").arg("redo").run();
 
     assert!(!result.is_success());
-    assert!(result.stderr().contains("Failed with: "));
+    assert!(result
+        .stderr()
+        .contains("Failed to run redo_error_migrations_fails with: "));
 }
 
 #[test]
diff --git a/diesel_cli/tests/migration_run.rs b/diesel_cli/tests/migration_run.rs
index f58b6f2f289e..60f6db2e39b2 100644
--- a/diesel_cli/tests/migration_run.rs
+++ b/diesel_cli/tests/migration_run.rs
@@ -109,9 +109,9 @@ fn empty_migrations_are_not_valid() {
     let result = p.command("migration").arg("run").run();
 
     assert!(!result.is_success());
-    assert!(result
-        .stderr()
-        .contains("Failed with: Attempted to run an empty migration."));
+    assert!(result.stderr().contains(
+        "Failed to run 12345_empty_migration with: Attempted to run an empty migration."
+    ));
 }
 
 #[test]
@@ -131,23 +131,9 @@ fn error_migrations_fails() {
     let result = p.command("migration").arg("run").run();
 
     assert!(!result.is_success());
-    assert!(result.stderr().contains("Failed with: "));
-}
-
-#[test]
-fn output_contains_path_to_migration_script() {
-    let p = project("output_contains_path_to_migration_script")
-        .folder("migrations")
-        .build();
-
-    p.command("setup").run();
-
-    p.create_migration("output_contains_path_to_migration_script", "", "");
-
-    let result = p.command("migration").arg("run").run();
-
-    assert!(!result.is_success());
-    assert!(result.stdout().contains("up.sql"));
+    assert!(result
+        .stderr()
+        .contains("Failed to run run_error_migrations_fails with: "));
 }
 
 #[test]
diff --git a/diesel_cli/tests/tests.rs b/diesel_cli/tests/tests.rs
index 818fb4f8e294..fc4581c5bd3d 100644
--- a/diesel_cli/tests/tests.rs
+++ b/diesel_cli/tests/tests.rs
@@ -4,7 +4,6 @@ extern crate chrono;
 extern crate diesel;
 #[macro_use]
 extern crate difference;
-extern crate migrations_internals;
 extern crate regex;
 extern crate tempfile;
 
diff --git a/diesel_tests/build.rs b/diesel_tests/build.rs
index 43131df3f718..c977cd22733c 100644
--- a/diesel_tests/build.rs
+++ b/diesel_tests/build.rs
@@ -3,7 +3,7 @@ extern crate diesel_migrations as migrations;
 extern crate dotenv;
 use self::diesel::*;
 use self::dotenv::dotenv;
-use std::{env, io};
+use std::env;
 
 #[cfg(not(any(feature = "mysql", feature = "sqlite", feature = "postgres")))]
 compile_error!(
@@ -52,14 +52,17 @@ fn database_url_from_env(backend_specific_env_var: &str) -> String {
 }
 
 fn main() {
-    let migrations_dir = migrations::find_migrations_directory()
-        .unwrap()
+    use migrations::MigrationHarness;
+    use std::path::PathBuf;
+
+    let manifest_dir = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
+    let migrations_dir = manifest_dir
+        .join("..")
+        .join("migrations")
         .join(MIGRATION_SUBDIR);
+    let migrations = migrations::FileBasedMigrations::from_path(&migrations_dir).unwrap();
+
     println!("cargo:rerun-if-changed={}", migrations_dir.display());
-    migrations::run_pending_migrations_in_directory(
-        &connection(),
-        &migrations_dir,
-        &mut io::sink(),
-    )
-    .unwrap();
+
+    connection().run_pending_migrations(migrations).unwrap();
 }
diff --git a/diesel_tests/tests/annotations.rs b/diesel_tests/tests/annotations.rs
index 2baffc23c69e..c5530fb0767a 100644
--- a/diesel_tests/tests/annotations.rs
+++ b/diesel_tests/tests/annotations.rs
@@ -122,8 +122,8 @@ mod associations_can_have_nullable_foreign_keys {
         id: i32,
     }
 
-    #[belongs_to(Foo)]
     #[derive(Identifiable, Associations)]
+    #[belongs_to(Foo)]
     pub struct Bar {
         id: i32,
         foo_id: Option<i32>,
diff --git a/diesel_tests/tests/schema/mod.rs b/diesel_tests/tests/schema/mod.rs
index 481bccbc3098..d33cff07dfa4 100644
--- a/diesel_tests/tests/schema/mod.rs
+++ b/diesel_tests/tests/schema/mod.rs
@@ -208,12 +208,14 @@ pub fn connection_without_transaction() -> TestConnection {
 }
 
 #[cfg(feature = "sqlite")]
-diesel_migrations::embed_migrations!("../migrations/sqlite");
+const MIGRATIONS: diesel_migrations::EmbeddedMigrations =
+    diesel_migrations::embed_migrations!("../migrations/sqlite");
 
 #[cfg(feature = "sqlite")]
 pub fn connection_without_transaction() -> TestConnection {
+    use diesel_migrations::MigrationHarness;
     let connection = SqliteConnection::establish(":memory:").unwrap();
-    embedded_migrations::run(&connection).unwrap();
+    connection.run_pending_migrations(MIGRATIONS).unwrap();
     connection
 }
 
diff --git a/examples/postgres/advanced-blog-cli/src/test_helpers.rs b/examples/postgres/advanced-blog-cli/src/test_helpers.rs
index bf6840ffe01d..afaea5fce49b 100644
--- a/examples/postgres/advanced-blog-cli/src/test_helpers.rs
+++ b/examples/postgres/advanced-blog-cli/src/test_helpers.rs
@@ -1,12 +1,13 @@
 use diesel::prelude::*;
-use diesel_migrations::run_pending_migrations;
+use diesel_migrations::{FileBasedMigrations, MigrationHarness};
 use lazy_static::lazy_static;
 use std::sync::{Mutex, MutexGuard};
 
 pub fn connection() -> PgConnection {
     let url = dotenv::var("DATABASE_URL").expect("DATABASE_URL must be set");
     let conn = PgConnection::establish(&url).unwrap();
-    run_pending_migrations(&conn).unwrap();
+    let migrations = FileBasedMigrations::find_migrations_directory().unwrap();
+    conn.run_pending_migrations(migrations).unwrap();
     conn.begin_test_transaction().unwrap();
     conn
 }