Check if we can rename triggers before doing an ALTER TABLE ... RENAME

ALTER TABLE .. RENAME, when used with the inplace algorithm, does:
- Do an inplace or online alter to the new definition
- Rename to new name
- Update triggers.

If update triggers would fail, we would rename the table back.
The problem with this approach is that the table would have the new
definition but the rename would fail.  The binary log would also not be
updated.

The solution to this is to very early check if we can rename triggers
and give an error if this would fail.
Both ALTER TABLE ... RENAME and RENAME TABLE is fixed.

This was implemented by moving the pre-check of rename table in triggers
from Table_triggers_list::change_table_name() to
Table_triggers_list::prepare_for_rename().

5 files changed, 176 insertions, 63 deletions

diff --git a/sql/ddl_log.cc b/sql/ddl_log.cc
index 7eae243771b..5f9a025c042 100644
--- a/sql/ddl_log.cc
+++ b/sql/ddl_log.cc
@@ -1160,6 +1160,7 @@ static int ddl_log_execute_action(THD *thd, MEM_ROOT *mem_root,
     case DDL_RENAME_PHASE_TRIGGER:
     {
       MDL_request mdl_request;
+      TRIGGER_RENAME_PARAM trigger_param;
 
       build_filename_and_delete_tmp_file(to_path, sizeof(to_path),
                                          &ddl_log_entry->db,
@@ -1195,14 +1196,20 @@ static int ddl_log_execute_action(THD *thd, MEM_ROOT *mem_root,
         error= thd->mdl_context.acquire_lock(&mdl_request, 1);
         /* acquire_locks() should never fail during recovery */
         DBUG_ASSERT(error == 0);
-
+        (void) Table_triggers_list::prepare_for_rename(thd,
+                                                       &trigger_param,
+                                                       &ddl_log_entry->db,
+                                                       &to_table,
+                                                       &to_converted_name,
+                                                       &ddl_log_entry->from_db,
+                                                       &from_table);
         (void) Table_triggers_list::change_table_name(thd,
+                                                      &trigger_param,
                                                       &ddl_log_entry->db,
                                                       &to_table,
                                                       &to_converted_name,
                                                       &ddl_log_entry->from_db,
                                                       &from_table);
-
         thd->mdl_context.release_lock(mdl_request.ticket);
       }
       if (ddl_log_increment_phase_no_lock(entry_pos))
@@ -1726,7 +1733,6 @@ static int ddl_log_execute_action(THD *thd, MEM_ROOT *mem_root,
     }
     break;
   }
-  }
   default:
     DBUG_ASSERT(0);
     break;
diff --git a/sql/sql_rename.cc b/sql/sql_rename.cc
index 2b2f32cf126..ec782792f30 100644
--- a/sql/sql_rename.cc
+++ b/sql/sql_rename.cc
@@ -1,6 +1,6 @@
 /*
    Copyright (c) 2000, 2013, Oracle and/or its affiliates.
-   Copyright (c) 2011, 2013, Monty Program Ab.
+   Copyright (c) 2011, 2021, Monty Program Ab.
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -339,6 +339,7 @@ do_rename(THD *thd, rename_param *param, DDL_LOG_STATE *ddl_log_state,
   int rc= 1;
   handlerton *hton;
   LEX_CSTRING *old_alias, *new_alias;
+  TRIGGER_RENAME_PARAM rename_param;
   DBUG_ENTER("do_rename");
   DBUG_PRINT("enter", ("skip_error: %d", (int) skip_error));
 
@@ -361,6 +362,15 @@ do_rename(THD *thd, rename_param *param, DDL_LOG_STATE *ddl_log_state,
     if (hton->flags & HTON_TABLE_MAY_NOT_EXIST_ON_SLAVE)
       *force_if_exists= 1;
 
+    /* Check if we can rename triggers */
+    if (Table_triggers_list::prepare_for_rename(thd, &rename_param,
+                                                &ren_table->db,
+                                                old_alias,
+                                                &ren_table->table_name,
+                                                new_db,
+                                                new_alias))
+      DBUG_RETURN(!skip_error);
+
     thd->replication_flags= 0;
 
     if (ddl_log_rename_table(thd, ddl_log_state, hton,
@@ -379,7 +389,9 @@ do_rename(THD *thd, rename_param *param, DDL_LOG_STATE *ddl_log_state,
 
       debug_crash_here("ddl_log_rename_before_rename_trigger");
 
-      if (!(rc= Table_triggers_list::change_table_name(thd, &ren_table->db,
+      if (!(rc= Table_triggers_list::change_table_name(thd,
+                                                       &rename_param,
+                                                       &ren_table->db,
                                                        old_alias,
                                                        &ren_table->table_name,
                                                        new_db,
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index f00aaaee00a..7b3b0eb1136 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -7058,6 +7058,7 @@ static bool mysql_inplace_alter_table(THD *thd,
                                       TABLE *altered_table,
                                       Alter_inplace_info *ha_alter_info,
                                       MDL_request *target_mdl_request,
+                                      TRIGGER_RENAME_PARAM *trigger_param,
                                       Alter_table_ctx *alter_ctx)
 {
   Open_table_context ot_ctx(thd, MYSQL_OPEN_REOPEN | MYSQL_OPEN_IGNORE_KILLED);
@@ -7330,7 +7331,7 @@ static bool mysql_inplace_alter_table(THD *thd,
       */
       DBUG_RETURN(true);
     }
-    if (Table_triggers_list::change_table_name(thd,
+    if (Table_triggers_list::change_table_name(thd, trigger_param,
                                                &alter_ctx->db,
                                                &alter_ctx->alias,
                                                &alter_ctx->table_name,
@@ -7343,7 +7344,8 @@ static bool mysql_inplace_alter_table(THD *thd,
       */
       (void) mysql_rename_table(db_type,
                                 &alter_ctx->new_db, &alter_ctx->new_alias,
-                                &alter_ctx->db, &alter_ctx->alias, NO_FK_CHECKS);
+                                &alter_ctx->db, &alter_ctx->alias,
+                                NO_FK_CHECKS);
       DBUG_RETURN(true);
     }
     rename_table_in_stat_tables(thd, &alter_ctx->db, &alter_ctx->alias,
@@ -8849,6 +8851,7 @@ simple_tmp_rename_or_index_change(THD *thd, TABLE_LIST *table_list,
 static bool
 simple_rename_or_index_change(THD *thd, TABLE_LIST *table_list,
                               Alter_info::enum_enable_or_disable keys_onoff,
+                              TRIGGER_RENAME_PARAM *trigger_param,
                               Alter_table_ctx *alter_ctx)
 {
   TABLE *table= table_list->table;
@@ -8895,7 +8898,7 @@ simple_rename_or_index_change(THD *thd, TABLE_LIST *table_list,
     if (mysql_rename_table(old_db_type, &alter_ctx->db, &alter_ctx->table_name,
                            &alter_ctx->new_db, &alter_ctx->new_alias, 0))
       error= -1;
-    else if (Table_triggers_list::change_table_name(thd,
+    else if (Table_triggers_list::change_table_name(thd, trigger_param,
                                                  &alter_ctx->db,
                                                  &alter_ctx->alias,
                                                  &alter_ctx->table_name,
@@ -9080,6 +9083,7 @@ bool mysql_alter_table(THD *thd, const LEX_CSTRING *new_db,
   MDL_request target_mdl_request;
   MDL_ticket *mdl_ticket= 0;
   Alter_table_prelocking_strategy alter_prelocking_strategy;
+  TRIGGER_RENAME_PARAM trigger_param;
   DBUG_ENTER("mysql_alter_table");
 
   /*
@@ -9504,6 +9508,16 @@ do_continue:;
                             create_info))
     DBUG_RETURN(true);
 
+  /* Check if rename of triggers are supported */
+  if (alter_ctx.is_table_renamed() &&
+      Table_triggers_list::prepare_for_rename(thd, &trigger_param,
+                                              &alter_ctx.db,
+                                              &alter_ctx.alias,
+                                              &alter_ctx.table_name,
+                                              &alter_ctx.new_db,
+                                              &alter_ctx.new_alias))
+    DBUG_RETURN(true);
+
   /*
     Look if we have to do anything at all.
     ALTER can become NOOP after handling
@@ -9549,6 +9563,7 @@ do_continue:;
       }
       res= simple_rename_or_index_change(thd, table_list,
                                          alter_info->keys_onoff,
+                                         &trigger_param,
                                          &alter_ctx);
     }
     else
@@ -9805,7 +9820,7 @@ do_continue:;
                                        &altered_table))
       goto err_new_table_cleanup;
     /*
-      Avoid creating frm again in ha_create_table() if inline alter will not
+      Avoid creating frm again in ha_create_table() if inplace alter will not
       be used.
     */
     frm_is_created= 1;
@@ -9879,9 +9894,12 @@ do_continue:;
       */
       enum_check_fields org_count_cuted_fields= thd->count_cuted_fields;
       thd->count_cuted_fields= CHECK_FIELD_WARN;
-      int res= mysql_inplace_alter_table(thd, table_list, table, &altered_table,
+      int res= mysql_inplace_alter_table(thd,
+                                         table_list, table, &altered_table,
                                          &ha_alter_info,
-                                         &target_mdl_request, &alter_ctx);
+                                         &target_mdl_request,
+                                         &trigger_param,
+                                         &alter_ctx);
       thd->count_cuted_fields= org_count_cuted_fields;
       my_free(const_cast<uchar*>(frm.str));
 
@@ -10216,7 +10234,7 @@ do_continue:;
   // Check if we renamed the table and if so update trigger files.
   if (alter_ctx.is_table_renamed())
   {
-    if (Table_triggers_list::change_table_name(thd,
+    if (Table_triggers_list::change_table_name(thd, &trigger_param,
                                                &alter_ctx.db,
                                                &alter_ctx.alias,
                                                &alter_ctx.table_name,
diff --git a/sql/sql_trigger.cc b/sql/sql_trigger.cc
index e3149799b6e..181cca39f7e 100644
--- a/sql/sql_trigger.cc
+++ b/sql/sql_trigger.cc
@@ -45,6 +45,18 @@ static bool add_table_for_trigger_internal(THD *thd,
                                            TABLE_LIST **table,
                                            char *trn_path_buff);
 
+/*
+  Functions for TRIGGER_RENAME_PARAM
+*/
+
+void TRIGGER_RENAME_PARAM::reset()
+{
+  delete table.triggers;
+  table.triggers= 0;
+  free_root(&table.mem_root, MYF(0));
+}
+
+
 /**
   Trigger_creation_ctx -- creation context of triggers.
 */
@@ -2206,64 +2218,43 @@ bool Trigger::change_on_table_name(void* param_arg)
 }
 
 
-/**
-  Update .TRG and .TRN files after renaming triggers' subject table.
-
-  @param[in,out] thd Thread context
-  @param[in] db Old database of subject table
-  @param[in] old_alias Old alias of subject table
-  @param[in] old_table Old name of subject table. The difference between
-             old_table and old_alias is that in case of lower_case_table_names
-             old_table == lowercase(old_alias)
-  @param[in] new_db New database for subject table
-  @param[in] new_table New name of subject table
-
-  @note
-    This method tries to leave trigger related files in consistent state,
-    i.e. it either will complete successfully, or will fail leaving files
-    in their initial state.
-    Also this method assumes that subject table is not renamed to itself.
-    This method needs to be called under an exclusive table metadata lock.
+/*
+  Check if we can rename triggers in change_table_name()
+  The idea is to ensure that it is close to impossible that
+  change_table_name() should fail.
 
-  @retval FALSE Success
-  @retval TRUE  Error
+  @return 0 ok
+  @return 1 Error: rename of triggers would fail
 */
 
-bool Table_triggers_list::change_table_name(THD *thd, const LEX_CSTRING *db,
-                                            const LEX_CSTRING *old_alias,
-                                            const LEX_CSTRING *old_table,
-                                            const LEX_CSTRING *new_db,
-                                            const LEX_CSTRING *new_table)
+bool
+Table_triggers_list::prepare_for_rename(THD *thd,
+                                        TRIGGER_RENAME_PARAM *param,
+                                        const LEX_CSTRING *db,
+                                        const LEX_CSTRING *old_alias,
+                                        const LEX_CSTRING *old_table,
+                                        const LEX_CSTRING *new_db,
+                                        const LEX_CSTRING *new_table)
 {
-  TABLE table;
+  TABLE *table= &param->table;
   bool result= 0;
-  bool upgrading50to51= FALSE;
-  Trigger *err_trigger;
-  DBUG_ENTER("Triggers::change_table_name");
+  DBUG_ENTER("Table_triggers_lists::prepare_change_table_name");
 
-  table.reset();
   init_sql_alloc(key_memory_Table_trigger_dispatcher,
-                 &table.mem_root, 8192, 0, MYF(0));
-
-  /*
-    This method interfaces the mysql server code protected by
-    an exclusive metadata lock.
-  */
-  DBUG_ASSERT(thd->mdl_context.is_lock_owner(MDL_key::TABLE, db->str,
-                                             old_table->str,
-                                             MDL_EXCLUSIVE));
+                 &table->mem_root, 8192, 0, MYF(0));
 
   DBUG_ASSERT(my_strcasecmp(table_alias_charset, db->str, new_db->str) ||
-              my_strcasecmp(table_alias_charset, old_alias->str, new_table->str));
+              my_strcasecmp(table_alias_charset, old_alias->str,
+                            new_table->str));
 
-  if (Table_triggers_list::check_n_load(thd, db, old_table, &table, TRUE))
+  if (Table_triggers_list::check_n_load(thd, db, old_table, table, TRUE))
   {
     result= 1;
     goto end;
   }
-  if (table.triggers)
+  if (table->triggers)
   {
-    if (table.triggers->check_for_broken_triggers())
+    if (table->triggers->check_for_broken_triggers())
     {
       result= 1;
       goto end;
@@ -2284,7 +2275,7 @@ bool Table_triggers_list::change_table_name(THD *thd, const LEX_CSTRING *db,
       if (check_n_cut_mysql50_prefix(db->str, dbname, sizeof(dbname)) &&
           !my_strcasecmp(table_alias_charset, dbname, new_db->str))
       {
-        upgrading50to51= TRUE;
+        param->upgrading50to51= TRUE;
       }
       else
       {
@@ -2293,14 +2284,70 @@ bool Table_triggers_list::change_table_name(THD *thd, const LEX_CSTRING *db,
         goto end;
       }
     }
-    if (unlikely(table.triggers->change_table_name_in_triggers(thd, db, new_db,
+  }
+
+end:
+  param->got_error= result;
+  DBUG_RETURN(result);
+}
+
+
+/**
+  Update .TRG and .TRN files after renaming triggers' subject table.
+
+  @param[in,out] thd Thread context
+  @param[in] db Old database of subject table
+  @param[in] old_alias Old alias of subject table
+  @param[in] old_table Old name of subject table. The difference between
+             old_table and old_alias is that in case of lower_case_table_names
+             old_table == lowercase(old_alias)
+  @param[in] new_db New database for subject table
+  @param[in] new_table New name of subject table
+
+  @note
+    This method tries to leave trigger related files in consistent state,
+    i.e. it either will complete successfully, or will fail leaving files
+    in their initial state.
+    Also this method assumes that subject table is not renamed to itself.
+    This method needs to be called under an exclusive table metadata lock.
+
+  @retval FALSE Success
+  @retval TRUE  Error
+*/
+
+bool Table_triggers_list::change_table_name(THD *thd,
+                                            TRIGGER_RENAME_PARAM *param,
+                                            const LEX_CSTRING *db,
+                                            const LEX_CSTRING *old_alias,
+                                            const LEX_CSTRING *old_table,
+                                            const LEX_CSTRING *new_db,
+                                            const LEX_CSTRING *new_table)
+{
+  TABLE *table= &param->table;
+  bool result= 0;
+  bool upgrading50to51= FALSE;
+  Trigger *err_trigger;
+  DBUG_ENTER("Table_triggers_list::change_table_name");
+
+  DBUG_ASSERT(!param->got_error);
+  /*
+    This method interfaces the mysql server code protected by
+    an exclusive metadata lock.
+  */
+  DBUG_ASSERT(thd->mdl_context.is_lock_owner(MDL_key::TABLE, db->str,
+                                             old_table->str,
+                                             MDL_EXCLUSIVE));
+
+  if (table->triggers)
+  {
+    if (unlikely(table->triggers->change_table_name_in_triggers(thd, db, new_db,
                                                                old_alias,
                                                                new_table)))
     {
       result= 1;
       goto end;
     }
-    if ((err_trigger= table.triggers->
+    if ((err_trigger= table->triggers->
          change_table_name_in_trignames( upgrading50to51 ? db : NULL,
                                          new_db, new_table, 0)))
     {
@@ -2310,10 +2357,10 @@ bool Table_triggers_list::change_table_name(THD *thd, const LEX_CSTRING *db,
         We assume that we will be able to undo our changes without errors
         (we can't do much if there will be an error anyway).
       */
-      (void) table.triggers->change_table_name_in_trignames(
+      (void) table->triggers->change_table_name_in_trignames(
                                upgrading50to51 ? new_db : NULL, db,
                                old_alias, err_trigger);
-      (void) table.triggers->change_table_name_in_triggers(
+      (void) table->triggers->change_table_name_in_triggers(
                                thd, db, new_db,
                                new_table, old_alias);
       result= 1;
@@ -2322,8 +2369,6 @@ bool Table_triggers_list::change_table_name(THD *thd, const LEX_CSTRING *db,
   }
 
 end:
-  delete table.triggers;
-  free_root(&table.mem_root, MYF(0));
   DBUG_RETURN(result);
 }
 
diff --git a/sql/sql_trigger.h b/sql/sql_trigger.h
index 98e527a50c0..739669c86a5 100644
--- a/sql/sql_trigger.h
+++ b/sql/sql_trigger.h
@@ -79,6 +79,30 @@ struct st_trg_execution_order
 };
 
 
+/*
+  Parameter to change_table_name_in_triggers()
+*/
+
+class TRIGGER_RENAME_PARAM
+{
+public:
+  TABLE table;
+  bool upgrading50to51;
+  bool got_error;
+
+  TRIGGER_RENAME_PARAM()
+  {
+    upgrading50to51= got_error= 0;
+    table.reset();
+  }
+  ~TRIGGER_RENAME_PARAM()
+  {
+    reset();
+  }
+  void reset();
+};
+
+
 class Table_triggers_list;
 
 /**
@@ -237,7 +261,14 @@ public:
                            TABLE *table, bool names_only);
   static bool drop_all_triggers(THD *thd, const LEX_CSTRING *db,
                                 const LEX_CSTRING *table_name, myf MyFlags);
-  static bool change_table_name(THD *thd, const LEX_CSTRING *db,
+  static bool prepare_for_rename(THD *thd, TRIGGER_RENAME_PARAM *param,
+                                 const LEX_CSTRING *db,
+                                 const LEX_CSTRING *old_alias,
+                                 const LEX_CSTRING *old_table,
+                                 const LEX_CSTRING *new_db,
+                                 const LEX_CSTRING *new_table);
+  static bool change_table_name(THD *thd, TRIGGER_RENAME_PARAM *param,
+                                const LEX_CSTRING *db,
                                 const LEX_CSTRING *old_alias,
                                 const LEX_CSTRING *old_table,
                                 const LEX_CSTRING *new_db,
@@ -315,6 +346,7 @@ private:
   }
 };
 
+
 bool add_table_for_trigger(THD *thd,
                            const sp_name *trg_name,
                            bool continue_if_not_exist,