fix: correctly handle foreign keys when dropping mysql indexes (#5073)

* fix: correctly handle foreign keys when dropping mysql indexes

* doc: clarify comments

* doc: clarify comment

* chore: add tests and clean up code

schema-engine/connectors/sql-schema-connector/src/apply_migration.rs

@@ -159,9 +159,7 @@ fn render_raw_sql(
             index_id,
             from_drop_and_recreate: _,
         } => vec![renderer.render_create_index(schemas.next.walk(*index_id))],
-        SqlMigrationStep::DropIndex { index_id } => {
-            vec![renderer.render_drop_index(schemas.previous.walk(*index_id))]
-        }
+        SqlMigrationStep::DropIndex { index_id } => vec![renderer.render_drop_index(schemas.previous.walk(*index_id))],
         SqlMigrationStep::RenameIndex { index } => renderer.render_rename_index(schemas.walk(*index)),
         SqlMigrationStep::DropView(drop_view) => {
             let view = schemas.previous.walk(drop_view.view_id);

schema-engine/connectors/sql-schema-connector/src/sql_schema_differ.rs

@@ -362,13 +362,39 @@ fn push_created_index_steps(steps: &mut Vec<SqlMigrationStep>, db: &DifferDataba
 
 fn push_dropped_index_steps(steps: &mut Vec<SqlMigrationStep>, db: &DifferDatabase<'_>) {
     let mut drop_indexes = HashSet::new();
+    let mut recreate_fkeys = HashSet::new();
 
-    for tables in db.non_redefined_table_pairs() {
-        for index in tables.dropped_indexes() {
-            // On MySQL, foreign keys automatically create indexes. These foreign-key-created
-            // indexes should only be dropped as part of the foreign key.
-            if db.flavour.should_skip_fk_indexes() && index::index_covers_fk(tables.previous(), index) {
-                continue;
+    for table in db.non_redefined_table_pairs() {
+        let dropped_fkeys = table.dropped_foreign_keys().map(|fk| fk.id).collect::<HashSet<_>>();
+
+        for index in table.dropped_indexes() {
+            if db.flavour.should_recreate_fks_covered_by_deleted_indexes() {
+                // MySQL requires an index on the referencing columns of every foreign key. The database
+                // will reuse a user-defined index if it exists, as long as the foreign key columns appear
+                // as the leftmost columns of the index. It's also possible for a single index to be
+                // used for more than one foreign key. We therefore have to find all such foreign keys
+                // and drop them before dropping the index and recreate them afterwards.
+                //
+                // For details see 'https://dev.mysql.com/doc/refman/8.0/en/create-table-foreign-keys.html#foreign-key-restrictions'.
+                let covered_fks = index::get_fks_covered_by_index(table.previous(), index)
+                    // We do not care about foreign keys that are meant to be dropped anyway.
+                    .filter(|fk| !dropped_fkeys.contains(&fk.id))
+                    .collect::<Vec<_>>();
+
+                // An edge case: if it looks like a normal index that has gone missing from the
+                // schema file and it precisely matches the columns of a foreign key, we assume it
+                // to be the index created for the foreign key and we do not drop it.
+                // This prevents us from dropping and re-creating the FK index when the schema file
+                // has not changed.
+                if index.is_normal()
+                    && covered_fks
+                        .iter()
+                        .any(|fk| fk.constrained_columns().len() == index.columns().len())
+                {
+                    continue;
+                }
+
+                recreate_fkeys.extend(covered_fks.iter().map(|fk| fk.id));
             }
 
             drop_indexes.insert(index.id);
@@ -388,6 +414,12 @@ fn push_dropped_index_steps(steps: &mut Vec<SqlMigrationStep>, db: &DifferDataba
     for index_id in drop_indexes.into_iter() {
         steps.push(SqlMigrationStep::DropIndex { index_id })
     }
+
+    for foreign_key_id in recreate_fkeys.into_iter() {
+        steps.push(SqlMigrationStep::DropForeignKey { foreign_key_id });
+        // Due to re-ordering this will execute after the DropIndex step.
+        steps.push(SqlMigrationStep::AddForeignKey { foreign_key_id });
+    }
 }
 
 fn push_redefined_table_steps(steps: &mut Vec<SqlMigrationStep>, db: &DifferDatabase<'_>) {

schema-engine/connectors/sql-schema-connector/src/sql_schema_differ/index.rs

@@ -1,15 +1,27 @@
-use sql_schema_describer::walkers::{IndexWalker, TableWalker};
+use either::Either;
+use sql_schema_describer::{
+    walkers::{IndexWalker, TableWalker},
+    ForeignKeyWalker, IndexType,
+};
+use std::iter;
 
-pub(super) fn index_covers_fk(table: TableWalker<'_>, index: IndexWalker<'_>) -> bool {
-    // Only normal indexes can cover foreign keys.
-    if index.index_type() != sql_schema_describer::IndexType::Normal {
-        return false;
+pub(super) fn get_fks_covered_by_index<'a>(
+    table: TableWalker<'a>,
+    index: IndexWalker<'a>,
+) -> impl Iterator<Item = ForeignKeyWalker<'a>> {
+    // Only normal, unique and primary key indexes can cover foreign keys.
+    if !matches!(
+        index.index_type(),
+        IndexType::Normal | IndexType::Unique | IndexType::PrimaryKey
+    ) {
+        return Either::Left(iter::empty());
     }
 
-    table.foreign_keys().any(|fk| {
+    Either::Right(table.foreign_keys().filter(move |fk| {
         let fk_cols = fk.constrained_columns().map(|col| col.name());
         let index_cols = index.column_names();
 
-        fk_cols.len() == index_cols.len() && fk_cols.zip(index_cols).all(|(a, b)| a == b)
-    })
+        // It's sufficient that leftmost columns of the index match the FK columns.
+        fk_cols.len() <= index_cols.len() && fk_cols.zip(index_cols).all(|(a, b)| a == b)
+    }))
 }

schema-engine/connectors/sql-schema-connector/src/sql_schema_differ/sql_schema_differ_flavour.rs

@@ -118,8 +118,8 @@ pub(crate) trait SqlSchemaDifferFlavour {
         true
     }
 
-    /// Whether indexes matching a foreign key should be skipped.
-    fn should_skip_fk_indexes(&self) -> bool {
+    /// Whether foreign keys should be recreated when they are covered by deleted indexes.
+    fn should_recreate_fks_covered_by_deleted_indexes(&self) -> bool {
         false
     }
 

schema-engine/connectors/sql-schema-connector/src/sql_schema_differ/sql_schema_differ_flavour/mysql.rs

@@ -91,7 +91,7 @@ impl SqlSchemaDifferFlavour for MysqlFlavour {
         true
     }
 
-    fn should_skip_fk_indexes(&self) -> bool {
+    fn should_recreate_fks_covered_by_deleted_indexes(&self) -> bool {
         true
     }
 

schema-engine/sql-migration-tests/tests/migrations/diff.rs

@@ -71,11 +71,18 @@ fn from_unique_index_to_without(mut api: TestApi) {
     })
     .unwrap();
 
-    let expected_printed_messages = if api.is_mysql() {
+    let expected_printed_messages = if api.is_vitess() {
         expect![[r#"
             [
                 "-- DropIndex\nDROP INDEX `Post_authorId_key` ON `Post`;\n",
             ]
+            "#]]
+    } else if api.is_mysql() {
+        // MySQL requires dropping the foreign key before dropping the index.
+        expect![[r#"
+            [
+                "-- DropForeignKey\nALTER TABLE `Post` DROP FOREIGN KEY `Post_authorId_fkey`;\n\n-- DropIndex\nDROP INDEX `Post_authorId_key` ON `Post`;\n\n-- AddForeignKey\nALTER TABLE `Post` ADD CONSTRAINT `Post_authorId_fkey` FOREIGN KEY (`authorId`) REFERENCES `User`(`id`) ON DELETE SET NULL ON UPDATE CASCADE;\n",
+            ]
         "#]]
     } else if api.is_sqlite() || api.is_postgres() || api.is_cockroach() {
         expect![[r#"
@@ -1138,7 +1145,7 @@ fn from_multi_file_schema_datamodel_to_url(mut api: TestApi) {
                   provider = "sqlite"
                   url = "{}"
               }}
-    
+
               model cows {{
                 id Int @id
                 meows Boolean

schema-engine/sql-migration-tests/tests/migrations/mysql.rs

@@ -1,7 +1,8 @@
 #![allow(dead_code)]
 
 use indoc::indoc;
-use schema_core::json_rpc::types::*;
+use psl::SourceFile;
+use schema_core::{json_rpc::types::*, schema_connector};
 use sql_migration_tests::test_api::*;
 use std::fmt::Write as _;
 
@@ -634,3 +635,194 @@ fn bigint_defaults_work(api: TestApi) {
     api.schema_push(schema).send().assert_green();
     api.schema_push(schema).send().assert_green().assert_no_steps();
 }
+
+#[test_connector(tags(Mysql), exclude(Vitess))]
+fn foreign_keys_covered_by_deleted_index_are_recreated(api: TestApi) {
+    let schema_a = r#"
+        model User {
+            id Int @id
+            transactions Transaction[]
+        }
+
+        model Account {
+            userId Int
+            id Int
+            transactions Transaction[]
+            @@id([userId, id])
+        }
+
+        model Transaction {
+            id Int @id
+            userId Int
+            accountId Int
+
+            user User @relation(fields: [userId], references: [id])
+            account Account @relation(fields: [userId, accountId], references: [userId, id])
+            @@unique([userId, accountId])
+        //  ^^^
+        //  We want to delete this index, MySQL will put a regular index back in its place
+        //  when we recreate the foreign key.
+        }
+    "#;
+
+    api.schema_push_w_datasource(schema_a).send();
+
+    api.assert_schema().assert_table("Transaction", |table| {
+        table
+            .assert_foreign_keys_count(2)
+            .assert_fk_on_columns(&["userId"], |fk| fk.assert_references("User", &["id"]))
+            .assert_fk_on_columns(&["userId", "accountId"], |fk| {
+                fk.assert_references("Account", &["userId", "id"])
+            })
+            .assert_indexes_count(1)
+            .assert_index_on_columns(&["userId", "accountId"], |idx| idx.assert_is_unique())
+    });
+
+    let schema_b = r#"
+        model User {
+            id Int @id
+            transactions Transaction[]
+        }
+
+        model Account {
+            userId Int
+            id Int
+            transactions Transaction[]
+            @@id([userId, id])
+        }
+
+        model Transaction {
+            id Int @id
+            userId Int
+            accountId Int
+
+            user User @relation(fields: [userId], references: [id])
+            account Account @relation(fields: [userId, accountId], references: [userId, id])
+        }
+    "#;
+
+    api.schema_push_w_datasource(schema_b)
+        .send()
+        .assert_green()
+        .assert_has_executed_steps();
+
+    api.assert_schema().assert_table("Transaction", |table| {
+        table
+            .assert_foreign_keys_count(2)
+            .assert_fk_on_columns(&["userId"], |fk| fk.assert_references("User", &["id"]))
+            .assert_fk_on_columns(&["userId", "accountId"], |fk| {
+                fk.assert_references("Account", &["userId", "id"])
+            })
+            .assert_indexes_count(1)
+            .assert_index_on_columns(&["userId", "accountId"], |idx| idx.assert_is_not_unique())
+    });
+
+    let diff = api.connector_diff(
+        schema_connector::DiffTarget::Datamodel(vec![("schema.prisma".to_string(), SourceFile::new_static(schema_a))]),
+        schema_connector::DiffTarget::Datamodel(vec![("schema.prisma".to_string(), SourceFile::new_static(schema_b))]),
+        None,
+    );
+    expect![[r#"
+        -- DropForeignKey
+        ALTER TABLE `Transaction` DROP FOREIGN KEY `Transaction_userId_fkey`;
+
+        -- DropForeignKey
+        ALTER TABLE `Transaction` DROP FOREIGN KEY `Transaction_userId_accountId_fkey`;
+
+        -- DropIndex
+        DROP INDEX `Transaction_userId_accountId_key` ON `Transaction`;
+
+        -- AddForeignKey
+        ALTER TABLE `Transaction` ADD CONSTRAINT `Transaction_userId_fkey` FOREIGN KEY (`userId`) REFERENCES `User`(`id`) ON DELETE RESTRICT ON UPDATE CASCADE;
+
+        -- AddForeignKey
+        ALTER TABLE `Transaction` ADD CONSTRAINT `Transaction_userId_accountId_fkey` FOREIGN KEY (`userId`, `accountId`) REFERENCES `Account`(`userId`, `id`) ON DELETE RESTRICT ON UPDATE CASCADE;
+    "#]].assert_eq(&diff);
+}
+
+#[test_connector(tags(Mysql), exclude(Vitess))]
+fn foreign_keys_covered_by_deleted_index_are_also_deleted(api: TestApi) {
+    let schema_a = r#"
+        model User {
+            id Int @id
+            transactions Transaction[]
+        }
+
+        model Account {
+            userId Int
+            id Int
+            transactions Transaction[]
+            @@id([userId, id])
+        }
+
+        model Transaction {
+            id Int @id
+            userId Int
+            accountId Int
+
+            user User @relation(fields: [userId], references: [id])
+            account Account @relation(fields: [userId, accountId], references: [userId, id])
+            @@unique([userId, accountId])
+        //  ^^^
+        //  We will delete both the index and the foreign keys. The migration should only
+        //  contain DROP statements.
+        }
+    "#;
+
+    api.schema_push_w_datasource(schema_a).send();
+
+    api.assert_schema().assert_table("Transaction", |table| {
+        table
+            .assert_foreign_keys_count(2)
+            .assert_fk_on_columns(&["userId"], |fk| fk.assert_references("User", &["id"]))
+            .assert_fk_on_columns(&["userId", "accountId"], |fk| {
+                fk.assert_references("Account", &["userId", "id"])
+            })
+            .assert_indexes_count(1)
+            .assert_index_on_columns(&["userId", "accountId"], |idx| idx.assert_is_unique())
+    });
+
+    let schema_b = r#"
+        model User {
+            id Int @id
+        }
+
+        model Account {
+            userId Int
+            id Int
+            @@id([userId, id])
+        }
+
+        model Transaction {
+            id Int @id
+            userId Int
+            accountId Int
+        }
+    "#;
+
+    api.schema_push_w_datasource(schema_b)
+        .send()
+        .assert_green()
+        .assert_has_executed_steps();
+
+    api.assert_schema().assert_table("Transaction", |table| {
+        table.assert_foreign_keys_count(0).assert_indexes_count(0)
+    });
+
+    let diff = api.connector_diff(
+        schema_connector::DiffTarget::Datamodel(vec![("schema.prisma".to_string(), SourceFile::new_static(schema_a))]),
+        schema_connector::DiffTarget::Datamodel(vec![("schema.prisma".to_string(), SourceFile::new_static(schema_b))]),
+        None,
+    );
+    expect![[r#"
+        -- DropForeignKey
+        ALTER TABLE `Transaction` DROP FOREIGN KEY `Transaction_userId_fkey`;
+
+        -- DropForeignKey
+        ALTER TABLE `Transaction` DROP FOREIGN KEY `Transaction_userId_accountId_fkey`;
+
+        -- DropIndex
+        DROP INDEX `Transaction_userId_accountId_key` ON `Transaction`;
+    "#]]
+    .assert_eq(&diff);
+}

schema-engine/sql-schema-describer/src/walkers/index.rs

@@ -39,6 +39,11 @@ impl<'a> IndexWalker<'a> {
         matches!(self.get().tpe, IndexType::Unique)
     }
 
+    /// Is this index a normal index?
+    pub fn is_normal(self) -> bool {
+        matches!(self.get().tpe, IndexType::Normal)
+    }
+
     /// The name of the index.
     pub fn name(self) -> &'a str {
         &self.get().index_name