MDEV-29791 ER_NO_SUCH_TABLE_IN_ENGINE or "File not found"

	   upon concurrent DDL with CREATE OR REPLACE

Race condition happens when atomic C-O-R drops backup table by
replaying ddl_log_state_rm in finalize_ddl(). When it already dropped
table (DDL_DROP_PHASE_TABLE) and before DDL_DROP_PHASE_TRIGGER
triggers still exist. At that point another thread tries to drop the
trigger and finds no such table.

The fix protects triggers with MDL lock when renaming original table
to backup. Thus we can drop triggers in another thread either before
that and rename to backup will be already without triggers or after
the C-O-R is finished and at that point there will be no table nor
triggers.

8 files changed, 79 insertions, 1 deletions

diff --git a/mysql-test/main/create_replace_debug.result b/mysql-test/main/create_replace_debug.result
index 3a3ef21bdec..2e54122e204 100644
--- a/mysql-test/main/create_replace_debug.result
+++ b/mysql-test/main/create_replace_debug.result
@@ -101,3 +101,23 @@ create or replace table t2 (y int) engine innodb select * from t1;
 ERROR HY000: Out of memory.
 set @@debug_dbug= @old_dbug;
 drop table t1;
+#
+# MDEV-29791 ER_NO_SUCH_TABLE_IN_ENGINE or "File not found"
+#            upon concurrent DDL with CREATE OR REPLACE
+#
+connect  con1,localhost,root,,test;
+create table t1 (a int) engine innodb;
+create trigger tr before delete on t1 for each row begin end;
+set debug_sync= 'ddl_log_before_drop_triggers signal s1 wait_for s2';
+create or replace table t1 (c int) engine innodb;
+connection default;
+set debug_sync= 'now wait_for s1';
+set debug_sync= 'now signal s2';
+drop trigger if exists tr;
+Warnings:
+Note	1360	Trigger does not exist
+set debug_sync= 'reset';
+connection con1;
+drop table if exists t1;
+disconnect con1;
+connection default;
diff --git a/mysql-test/main/create_replace_debug.test b/mysql-test/main/create_replace_debug.test
index 86e68c47743..e8f4f2c95a0 100644
--- a/mysql-test/main/create_replace_debug.test
+++ b/mysql-test/main/create_replace_debug.test
@@ -97,3 +97,26 @@ set @@debug_dbug= '+d,ha_end_bulk_insert_fail';
 create or replace table t2 (y int) engine innodb select * from t1;
 set @@debug_dbug= @old_dbug;
 drop table t1;
+
+--echo #
+--echo # MDEV-29791 ER_NO_SUCH_TABLE_IN_ENGINE or "File not found"
+--echo #            upon concurrent DDL with CREATE OR REPLACE
+--echo #
+--connect (con1,localhost,root,,test)
+create table t1 (a int) engine innodb;
+create trigger tr before delete on t1 for each row begin end;
+set debug_sync= 'ddl_log_before_drop_triggers signal s1 wait_for s2';
+send create or replace table t1 (c int) engine innodb;
+
+--connection default
+set debug_sync= 'now wait_for s1';
+set debug_sync= 'now signal s2';
+drop trigger if exists tr;
+set debug_sync= 'reset';
+
+# cleanup
+--connection con1
+--reap
+drop table if exists t1;
+--disconnect con1
+--connection default
diff --git a/sql/ddl_log.cc b/sql/ddl_log.cc
index afcf2e6497e..4cb70ddc30a 100644
--- a/sql/ddl_log.cc
+++ b/sql/ddl_log.cc
@@ -1601,6 +1601,7 @@ static int ddl_log_execute_action(THD *thd, MEM_ROOT *mem_root,
         break;
       /* Fall through */
     case DDL_DROP_PHASE_TRIGGER:
+      DEBUG_SYNC(thd, "ddl_log_before_drop_triggers");
       Table_triggers_list::drop_all_triggers(thd, &db, &table,
                                              fn_flags,
                                              MYF(MY_WME | MY_IGNORE_ENOENT));
diff --git a/sql/sql_rename.cc b/sql/sql_rename.cc
index c9886232d45..4989bcb18ba 100644
--- a/sql/sql_rename.cc
+++ b/sql/sql_rename.cc
@@ -337,6 +337,7 @@ rename_table_and_triggers(THD *thd, rename_param *param,
   LEX_CSTRING *old_alias, *new_alias;
   TRIGGER_RENAME_PARAM rename_param;
   rename_param.rename_flags= param->rename_flags;
+  rename_param.lock_triggers= param->lock_triggers;
   DBUG_ENTER("rename_table_and_triggers");
   DBUG_PRINT("enter", ("skip_error: %d", (int) skip_error));
 
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index e16757798c7..5dddd034eb8 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -4480,6 +4480,7 @@ bool HA_CREATE_INFO::finalize_atomic_replace(THD *thd, TABLE_LIST *orig_table)
     param.old_alias= lower_case_table_names == 2 ? orig_table->alias :
                                                    table_name;
     param.new_alias= backup_name.table_name;
+    param.lock_triggers= true;
     if (rename_table_and_triggers(thd, &param, NULL, orig_table,
                                   &backup_name.db, false, &dummy))
     {
@@ -4497,6 +4498,7 @@ bool HA_CREATE_INFO::finalize_atomic_replace(THD *thd, TABLE_LIST *orig_table)
   param.old_version= tabledef_version;
   param.old_alias= tmp_name.table_name;
   param.new_alias= table_name;
+  param.lock_triggers= false;
   ulonglong option_bits_save= thd->variables.option_bits;
   thd->variables.option_bits&= ~OPTION_NO_FOREIGN_KEY_CHECKS;
   if (ddl_log_create_table(ddl_log_state_create, param.from_table_hton,
diff --git a/sql/sql_table.h b/sql/sql_table.h
index b95e6d6bdb0..5502140babb 100644
--- a/sql/sql_table.h
+++ b/sql/sql_table.h
@@ -147,9 +147,11 @@ struct rename_param
   LEX_CUSTRING old_version;
   handlerton *from_table_hton;
   int rename_flags; /* FN_FROM_IS_TMP, FN_TO_IS_TMP, etc */
+  bool lock_triggers; /* Acquire MDL_EXCLUSIVE for triggers */
   rename_param() :
     from_table_hton(NULL),
-    rename_flags(0) {}
+    rename_flags(0),
+    lock_triggers(false) {}
 };
 bool
 rename_table_and_triggers(THD *thd, rename_param *param,
diff --git a/sql/sql_trigger.cc b/sql/sql_trigger.cc
index 81dcda8b39f..c7c9c25c01e 100644
--- a/sql/sql_trigger.cc
+++ b/sql/sql_trigger.cc
@@ -2275,6 +2275,21 @@ bool Trigger::change_on_table_name(void* param_arg)
 }
 
 
+bool Trigger::lock_trigger(void* param)
+{
+  THD *thd= (THD *) param;
+  MDL_request mdl_request;
+
+  MDL_REQUEST_INIT(&mdl_request, MDL_key::TRIGGER, body->m_db.str,
+                   body->m_name.str, MDL_EXCLUSIVE, MDL_STATEMENT);
+  if (thd->mdl_context.acquire_lock(&mdl_request,
+                                    thd->variables.lock_wait_timeout))
+    return true;
+
+  return false;
+}
+
+
 /*
   Check if we can rename triggers in change_table_name()
   The idea is to ensure that it is close to impossible that
@@ -2342,6 +2357,12 @@ Table_triggers_list::prepare_for_rename(THD *thd,
         goto end;
       }
     }
+    if (param->lock_triggers &&
+        table->triggers->lock_triggers(thd))
+    {
+      result= 1;
+      goto end;
+    }
   }
 
 end:
diff --git a/sql/sql_trigger.h b/sql/sql_trigger.h
index d1aaa970289..0ad7db3f381 100644
--- a/sql/sql_trigger.h
+++ b/sql/sql_trigger.h
@@ -90,11 +90,13 @@ public:
   bool upgrading50to51;
   bool got_error;
   int rename_flags;
+  bool lock_triggers;
 
   TRIGGER_RENAME_PARAM()
   {
     upgrading50to51= got_error= 0;
     rename_flags= 0;
+    lock_triggers= false;
     table.reset();
   }
   ~TRIGGER_RENAME_PARAM()
@@ -154,6 +156,7 @@ public:
   bool change_on_table_name(void* param_arg);
   bool change_table_name(void* param_arg);
   bool add_to_file_list(void* param_arg);
+  bool lock_trigger(void* param);
 };
 
 typedef bool (Trigger::*Triggers_processor)(void *arg);
@@ -327,6 +330,11 @@ public:
 
   Trigger* for_all_triggers(Triggers_processor func, void *arg);
 
+  bool lock_triggers(THD *thd)
+  {
+    return (bool) for_all_triggers(&Trigger::lock_trigger, (void *) thd);
+  }
+
 private:
   bool prepare_record_accessors(TABLE *table);
   Trigger *change_table_name_in_trignames(const LEX_CSTRING *old_db_name,