MDEV-16465 Invalid (old?) table or database name or hang in ha_innobase::delete_table and log semaphore wait upon concurrent DDL with foreign keys

ALTER TABLE locks the table with TL_READ_NO_INSERT, to prevent the
source table modifications while it's being copied. But there's an
indirect way of modifying a table, via cascade FK actions.

After previous commits, an attempt to modify an FK parent table
will cause FK children to be prelocked, so the table-being-altered
cannot be modified by a cascade FK action, because ALTER holds a
lock and prelocking will wait.

But if a new FK is being added by this very ALTER, then the target
table is not locked yet (it's a temporary table). So, we have to
lock FK parents explicitly.

mysql-test/suite/innodb/r/foreign-keys.result

@@ -51,3 +51,39 @@ c	d
 6	30
 drop table t2, t1;
 drop user foo;
+create table t1 (f1 int primary key) engine=innodb;
+create table t2 (f2 int primary key) engine=innodb;
+create table t3 (f3 int primary key, foreign key (f3) references t2(f2)) engine=innodb;
+insert into t1 values (1),(2),(3),(4),(5);
+insert into t2 values (1),(2),(3),(4),(5);
+insert into t3 values (1),(2),(3),(4),(5);
+connect con1,localhost,root;
+set debug_sync='alter_table_before_rename_result_table signal g1 wait_for g2';
+alter table t2 add constraint foreign key (f2) references t1(f1) on delete cascade on update cascade;
+connection default;
+set debug_sync='before_execute_sql_command wait_for g1';
+update t1 set f1 = f1 + 100000 limit 2;
+connect con2,localhost,root;
+kill query UPDATE;
+disconnect con2;
+connection default;
+ERROR 70100: Query execution was interrupted
+set debug_sync='now signal g2';
+connection con1;
+show create table t2;
+Table	Create Table
+t2	CREATE TABLE `t2` (
+  `f2` int(11) NOT NULL,
+  PRIMARY KEY (`f2`),
+  CONSTRAINT `t2_ibfk_1` FOREIGN KEY (`f2`) REFERENCES `t1` (`f1`) ON DELETE CASCADE ON UPDATE CASCADE
+) ENGINE=InnoDB DEFAULT CHARSET=latin1
+disconnect con1;
+connection default;
+select * from t2 where f2 not in (select f1 from t1);
+f2
+select * from t3 where f3 not in (select f2 from t2);
+f3
+drop table t3;
+drop table t2;
+drop table t1;
+set debug_sync='reset';

mysql-test/suite/innodb/t/foreign-keys.test

@@ -1,5 +1,6 @@
 --source include/have_innodb.inc
 --source include/have_debug.inc
+--source include/have_debug_sync.inc
 
 --enable_connect_log
 
@@ -72,3 +73,41 @@ connection default;
 select * from t2;
 drop table t2, t1;
 drop user foo;
+
+#
+# MDEV-16465 Invalid (old?) table or database name or hang in ha_innobase::delete_table and log semaphore wait upon concurrent DDL with foreign keys
+#
+create table t1 (f1 int primary key) engine=innodb;
+create table t2 (f2 int primary key) engine=innodb;
+create table t3 (f3 int primary key, foreign key (f3) references t2(f2)) engine=innodb;
+insert into t1 values (1),(2),(3),(4),(5);
+insert into t2 values (1),(2),(3),(4),(5);
+insert into t3 values (1),(2),(3),(4),(5);
+connect con1,localhost,root;
+set debug_sync='alter_table_before_rename_result_table signal g1 wait_for g2';
+send alter table t2 add constraint foreign key (f2) references t1(f1) on delete cascade on update cascade;
+connection default;
+let $conn=`select connection_id()`;
+set debug_sync='before_execute_sql_command wait_for g1';
+send update t1 set f1 = f1 + 100000 limit 2;
+connect con2,localhost,root;
+let $wait_condition= select 1 from information_schema.processlist where state='Waiting for table metadata lock' and info like 'update t1 %';
+source include/wait_condition.inc;
+--replace_result $conn UPDATE
+eval kill query $conn;
+disconnect con2;
+connection default;
+error ER_QUERY_INTERRUPTED;
+reap;
+set debug_sync='now signal g2';
+connection con1;
+reap;
+show create table t2;
+disconnect con1;
+connection default;
+select * from t2 where f2 not in (select f1 from t1);
+select * from t3 where f3 not in (select f2 from t2);
+drop table t3;
+drop table t2;
+drop table t1;
+set debug_sync='reset';

sql/sql_base.cc

@@ -4806,8 +4806,8 @@ handle_routine(THD *thd, Query_tables_list *prelocking_ctx,
   @note this can be changed to use a hash, instead of scanning the linked
   list, if the performance of this function will ever become an issue
 */
-static bool table_already_fk_prelocked(TABLE_LIST *tl, LEX_STRING *db,
+bool table_already_fk_prelocked(TABLE_LIST *tl, LEX_STRING *db,
-                                    LEX_STRING *table, thr_lock_type lock_type)
+                                LEX_STRING *table, thr_lock_type lock_type)
 {
   for (; tl; tl= tl->next_global )
   {

sql/sql_base.h

@@ -148,6 +148,8 @@ my_bool mysql_rm_tmp_tables(void);
 bool rm_temporary_table(handlerton *base, const char *path);
 void close_tables_for_reopen(THD *thd, TABLE_LIST **tables,
                              const MDL_savepoint &start_of_statement_svp);
+bool table_already_fk_prelocked(TABLE_LIST *tl, LEX_STRING *db,
+                                LEX_STRING *table, thr_lock_type lock_type);
 TABLE_LIST *find_table_in_list(TABLE_LIST *table,
                                TABLE_LIST *TABLE_LIST::*link,
                                const char *db_name,

sql/sql_table.cc

@@ -9057,8 +9057,10 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
                        alter_ctx.tmp_name, strlen(alter_ctx.tmp_name),
                        alter_ctx.tmp_name, TL_READ_NO_INSERT);
     /* Table is in thd->temporary_tables */
-    (void) open_temporary_table(thd, &tbl);
+    if (open_temporary_table(thd, &tbl))
+      goto err_new_table_cleanup;
     new_table= tbl.table;
+    DBUG_ASSERT(new_table);
   }
   else
   {
@@ -9067,9 +9069,48 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
     new_table= open_table_uncached(thd, new_db_type, alter_ctx.get_tmp_path(),
                                    alter_ctx.new_db, alter_ctx.tmp_name,
                                    true, true);
+    if (!new_table)
+      goto err_new_table_cleanup;
+
+    /*
+      Normally, an attempt to modify an FK parent table will cause
+      FK children to be prelocked, so the table-being-altered cannot
+      be modified by a cascade FK action, because ALTER holds a lock
+      and prelocking will wait.
+
+      But if a new FK is being added by this very ALTER, then the target
+      table is not locked yet (it's a temporary table). So, we have to
+      lock FK parents explicitly.
+    */
+    if (alter_info->flags & Alter_info::ADD_FOREIGN_KEY)
+    {
+      List <FOREIGN_KEY_INFO> fk_list;
+      List_iterator<FOREIGN_KEY_INFO> fk_list_it(fk_list);
+      FOREIGN_KEY_INFO *fk;
+
+      /* tables_opened can be > 1 only for MERGE tables */
+      DBUG_ASSERT(tables_opened == 1);
+      DBUG_ASSERT(&table_list->next_global == thd->lex->query_tables_last);
+
+      new_table->file->get_foreign_key_list(thd, &fk_list);
+      while ((fk= fk_list_it++))
+      {
+        if (table_already_fk_prelocked(table_list, fk->referenced_db,
+                                       fk->referenced_table, TL_READ_NO_INSERT))
+          continue;
+
+        TABLE_LIST *tl= (TABLE_LIST *) thd->alloc(sizeof(TABLE_LIST));
+        tl->init_one_table_for_prelocking(fk->referenced_db->str, fk->referenced_db->length,
+                           fk->referenced_table->str, fk->referenced_table->length,
+                           NULL, TL_READ_NO_INSERT, false, NULL, 0,
+                           &thd->lex->query_tables_last);
+      }
+
+      if (open_tables(thd, &table_list->next_global, &tables_opened, 0,
+                      &alter_prelocking_strategy))
+        goto err_new_table_cleanup;
+    }
   }
-  if (!new_table)
-    goto err_new_table_cleanup;
   /*
     Note: In case of MERGE table, we do not attach children. We do not
     copy data for MERGE tables. Only the children have data.

sql/table.h

@@ -1695,6 +1695,14 @@ struct TABLE_LIST
                              const char *alias_arg,
                              enum thr_lock_type lock_type_arg)
   {
+    enum enum_mdl_type mdl_type;
+    if (lock_type_arg >= TL_WRITE_ALLOW_WRITE)
+      mdl_type= MDL_SHARED_WRITE;
+    else if (lock_type_arg == TL_READ_NO_INSERT)
+      mdl_type= MDL_SHARED_NO_WRITE;
+    else
+      mdl_type= MDL_SHARED_READ;
+
     bzero((char*) this, sizeof(*this));
     db= (char*) db_name_arg;
     db_length= db_length_arg;
@@ -1702,10 +1710,7 @@ struct TABLE_LIST
     table_name_length= table_name_length_arg;
     alias= (char*) (alias_arg ? alias_arg : table_name_arg);
     lock_type= lock_type_arg;
-    mdl_request.init(MDL_key::TABLE, db, table_name,
+    mdl_request.init(MDL_key::TABLE, db, table_name, mdl_type, MDL_TRANSACTION);
-                     (lock_type >= TL_WRITE_ALLOW_WRITE) ?
-                     MDL_SHARED_WRITE : MDL_SHARED_READ,
-                     MDL_TRANSACTION);
   }
 
   inline void init_one_table_for_prelocking(const char *db_name_arg,