Auto-merge from mysql-5.5-merge.

1 files changed, 294 insertions, 1134 deletions

diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 02a874ce62f..3f0a0326c84 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -51,7 +51,6 @@
 #include "sql_parse.h"
 #include "sql_show.h"
 #include "transaction.h"
-#include "keycaches.h"
 #include "datadict.h"  // dd_frm_type()
 
 #ifdef __WIN__
@@ -78,10 +77,6 @@ mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,
                            uint *db_options,
                            handler *file, KEY **key_info_buffer,
                            uint *key_count, int select_field_count);
-static bool
-mysql_prepare_alter_table(THD *thd, TABLE *table,
-                          HA_CREATE_INFO *create_info,
-                          Alter_info *alter_info);
 
 
 /**
@@ -1086,6 +1081,7 @@ static bool get_free_ddl_log_entry(DDL_LOG_MEMORY_ENTRY **active_entry,
   */
   used_entry->next_log_entry= first_used;
   used_entry->prev_log_entry= NULL;
+  used_entry->next_active_log_entry= NULL;
   global_ddl_log.first_used= used_entry;
   if (first_used)
     first_used->prev_log_entry= used_entry;
@@ -1723,7 +1719,6 @@ bool mysql_write_frm(ALTER_PARTITION_PARAM_TYPE *lpt, uint flags)
       completing this we write a new phase to the log entry that will
       deactivate it.
     */
-    mysql_mutex_lock(&LOCK_open);
     if (mysql_file_delete(key_file_frm, frm_name, MYF(MY_WME)) ||
 #ifdef WITH_PARTITION_STORAGE_ENGINE
         lpt->table->file->ha_create_handler_files(path, shadow_path,
@@ -1779,12 +1774,12 @@ bool mysql_write_frm(ALTER_PARTITION_PARAM_TYPE *lpt, uint flags)
 #endif
 
 err:
-    mysql_mutex_unlock(&LOCK_open);
 #ifdef WITH_PARTITION_STORAGE_ENGINE
     deactivate_ddl_log_entry(part_info->frm_log_entry->entry_pos);
     part_info->frm_log_entry= NULL;
     (void) sync_ddl_log();
 #endif
+    ;
   }
 
 end:
@@ -1799,6 +1794,8 @@ end:
     clear_error                   is clear_error to be called
     query                         Query to log
     query_length                  Length of query
+    is_trans                      if the event changes either
+                                  a trans or non-trans engine.
 
   RETURN VALUES
     NONE
@@ -1921,18 +1918,64 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
   String wrong_tables;
   int error= 0;
   int non_temp_tables_count= 0;
-  bool some_tables_deleted=0, tmp_table_deleted=0, foreign_key_error=0;
+  bool foreign_key_error=0;
+  bool non_tmp_error= 0;
+  bool trans_tmp_table_deleted= 0, non_trans_tmp_table_deleted= 0;
+  bool non_tmp_table_deleted= 0;
   String built_query;
-  String built_tmp_query;
+  String built_trans_tmp_query, built_non_trans_tmp_query;
   DBUG_ENTER("mysql_rm_table_part2");
 
-  if (thd->is_current_stmt_binlog_format_row() && !dont_log_query)
+  /*
+    Prepares the drop statements that will be written into the binary
+    log as follows:
+
+    1 - If we are not processing a "DROP TEMPORARY" it prepares a
+    "DROP".
+
+    2 - A "DROP" may result in a "DROP TEMPORARY" but the opposite is
+    not true.
+
+    3 - If the current format is row, the IF EXISTS token needs to be
+    appended because one does not know if CREATE TEMPORARY was previously
+    written to the binary log.
+
+    4 - Add the IF_EXISTS token if necessary, i.e. if_exists is TRUE.
+
+    5 - For temporary tables, there is a need to differentiate tables
+    in transactional and non-transactional storage engines. For that,
+    reason, two types of drop statements are prepared.
+
+    The need to different the type of tables when dropping a temporary
+    table stems from the fact that such drop does not commit an ongoing
+    transaction and changes to non-transactional tables must be written
+    ahead of the transaction in some circumstances.
+  */
+  if (!dont_log_query)
   {
-    built_query.set_charset(system_charset_info);
-    if (if_exists)
-      built_query.append("DROP TABLE IF EXISTS ");
+    if (!drop_temporary)
+    {
+      built_query.set_charset(system_charset_info);
+      if (if_exists)
+        built_query.append("DROP TABLE IF EXISTS ");
+      else
+        built_query.append("DROP TABLE ");
+    }
+
+    if (thd->is_current_stmt_binlog_format_row() || if_exists)
+    {
+      built_trans_tmp_query.set_charset(system_charset_info);
+      built_trans_tmp_query.append("DROP TEMPORARY TABLE IF EXISTS ");
+      built_non_trans_tmp_query.set_charset(system_charset_info);
+      built_non_trans_tmp_query.append("DROP TEMPORARY TABLE IF EXISTS ");
+    }
     else
-      built_query.append("DROP TABLE ");
+    {
+      built_trans_tmp_query.set_charset(system_charset_info);
+      built_trans_tmp_query.append("DROP TEMPORARY TABLE ");
+      built_non_trans_tmp_query.set_charset(system_charset_info);
+      built_non_trans_tmp_query.append("DROP TEMPORARY TABLE ");
+    }
   }
 
   mysql_ha_rm_tables(thd, tables);
@@ -1955,10 +1998,11 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
       if (lock_table_names(thd, tables, NULL, thd->variables.lock_wait_timeout,
                            MYSQL_OPEN_SKIP_TEMPORARY))
         DBUG_RETURN(1);
-      mysql_mutex_lock(&LOCK_open);
       for (table= tables; table; table= table->next_local)
-        tdc_remove_table(thd, TDC_RT_REMOVE_ALL, table->db, table->table_name);
-      mysql_mutex_unlock(&LOCK_open);
+      {
+        tdc_remove_table(thd, TDC_RT_REMOVE_ALL, table->db, table->table_name,
+                         FALSE);
+      }
     }
     else
     {
@@ -2001,6 +2045,7 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
 
   for (table= tables; table; table= table->next_local)
   {
+    bool is_trans;
     char *db=table->db;
     handlerton *table_type;
     enum legacy_db_type frm_db_type= DB_TYPE_UNKNOWN;
@@ -2009,78 +2054,70 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
                          table->db, table->table_name, (long) table->table,
                          table->table ? (long) table->table->s : (long) -1));
 
+    /*
+      drop_temporary_table may return one of the following error codes:
+      .  0 - a temporary table was successfully dropped.
+      .  1 - a temporary table was not found.
+      . -1 - a temporary table is used by an outer statement.
+    */
     if (table->open_type == OT_BASE_ONLY)
       error= 1;
-    else
-      error= drop_temporary_table(thd, table);
- 
-    switch (error) {
-    case  0:
-      // removed temporary table
-      tmp_table_deleted= 1;
+    else if ((error= drop_temporary_table(thd, table, &is_trans)) == -1)
+    {
+      DBUG_ASSERT(thd->in_sub_stmt);
+      goto err;
+    }
+
+    if ((drop_temporary && if_exists) || !error)
+    {
       /*
-        One needs to always log any temporary table drop if the current
-        statement logging format is set to row. This happens because one
-        might have created a temporary table while the statement logging
-        format was statement and then switched to mixed or row format.
+        This handles the case of temporary tables. We have the following cases:
+
+          . "DROP TEMPORARY" was executed and a temporary table was affected
+          (i.e. drop_temporary && !error) or the if_exists was specified (i.e.
+          drop_temporary && if_exists).
+
+          . "DROP" was executed but a temporary table was affected (.i.e
+          !error).
       */
-      if (thd->is_current_stmt_binlog_format_row())
+      if (!dont_log_query)
       {
-        if (built_tmp_query.is_empty()) 
-        {
-          built_tmp_query.set_charset(system_charset_info);
-          built_tmp_query.append("DROP TEMPORARY TABLE IF EXISTS ");
-        }
+        /*
+          If there is an error, we don't know the type of the engine
+          at this point. So, we keep it in the trx-cache.
+        */
+        is_trans= error ? TRUE : is_trans;
+        if (is_trans)
+          trans_tmp_table_deleted= TRUE;
+        else
+          non_trans_tmp_table_deleted= TRUE;
 
-        built_tmp_query.append("`");
+        String *built_ptr_query=
+          (is_trans ? &built_trans_tmp_query : &built_non_trans_tmp_query);
+        /*
+          Don't write the database name if it is the current one (or if
+          thd->db is NULL).
+        */
+        built_ptr_query->append("`");
         if (thd->db == NULL || strcmp(db,thd->db) != 0)
         {
-          built_tmp_query.append(db);
-          built_tmp_query.append("`.`");
+          built_ptr_query->append(db);
+          built_ptr_query->append("`.`");
         }
-        built_tmp_query.append(table->table_name);
-        built_tmp_query.append("`,");
+        built_ptr_query->append(table->table_name);
+        built_ptr_query->append("`,");
       }
-
-      continue;
-    case -1:
-      DBUG_ASSERT(thd->in_sub_stmt);
-      error= 1;
-      goto err;
-    default:
-      // temporary table not found
-      error= 0;
-    }
-
-    /* Probably a non-temporary table. */
-    if (!drop_temporary)
-      non_temp_tables_count++;
-
-    /*
-      If row-based replication is used and the table is not a
-      temporary table, we add the table name to the drop statement
-      being built.  The string always end in a comma and the comma
-      will be chopped off before being written to the binary log.
-      */
-    if (!drop_temporary && thd->is_current_stmt_binlog_format_row() && !dont_log_query)
-    {
       /*
-        Don't write the database name if it is the current one (or if
-        thd->db is NULL).
+        This means that a temporary table was droped and as such there
+        is no need to proceed with the code that tries to drop a regular
+        table.
       */
-      built_query.append("`");
-      if (thd->db == NULL || strcmp(db,thd->db) != 0)
-      {
-        built_query.append(db);
-        built_query.append("`.`");
-      }
-
-      built_query.append(table->table_name);
-      built_query.append("`,");
+      if (!error) continue;
     }
-
-    if (!drop_temporary)
+    else if (!drop_temporary)
     {
+      non_temp_tables_count++;
+
       if (thd->locked_tables_mode)
       {
         if (wait_while_table_is_used(thd, table->table, HA_EXTRA_FORCE_REOPEN))
@@ -2103,36 +2140,70 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
                                         reg_ext,
                                         table->internal_tmp_table ?
                                         FN_IS_TMP : 0);
+
+      /*
+        This handles the case where a "DROP" was executed and a regular
+        table "may be" dropped as drop_temporary is FALSE and error is
+        TRUE. If the error was FALSE a temporary table was dropped and
+        regardless of the status of drop_tempoary a "DROP TEMPORARY"
+        must be used.
+      */
+      if (!dont_log_query)
+      {
+        /*
+          Note that unless if_exists is TRUE or a temporary table was deleted, 
+          there is no means to know if the statement should be written to the
+          binary log. See further information on this variable in what follows.
+        */
+        non_tmp_table_deleted= (if_exists ? TRUE : non_tmp_table_deleted);
+        /*
+          Don't write the database name if it is the current one (or if
+          thd->db is NULL).
+        */
+        built_query.append("`");
+        if (thd->db == NULL || strcmp(db,thd->db) != 0)
+        {
+          built_query.append(db);
+          built_query.append("`.`");
+        }
+
+        built_query.append(table->table_name);
+        built_query.append("`,");
+      }
     }
-    /*
-      TODO: Investigate what should be done to remove this lock completely.
-            Is exclusive meta-data lock enough ?
-    */
     DEBUG_SYNC(thd, "rm_table_part2_before_delete_table");
     DBUG_EXECUTE_IF("sleep_before_part2_delete_table",
                     my_sleep(100000););
-    mysql_mutex_lock(&LOCK_open);
+    error= 0;
     if (drop_temporary ||
         ((access(path, F_OK) &&
           ha_create_table_from_engine(thd, db, alias)) ||
          (!drop_view &&
           dd_frm_type(thd, path, &frm_db_type) != FRMTYPE_TABLE)))
     {
-      // Table was not found on disk and table can't be created from engine
+      /*
+        One of the following cases happened:
+          . "DROP TEMPORARY" but a temporary table was not found.
+          . "DROP" but table was not found on disk and table can't be
+            created from engine.
+          . ./sql/datadict.cc +32 /Alfranio - TODO: We need to test this.
+      */
       if (if_exists)
 	push_warning_printf(thd, MYSQL_ERROR::WARN_LEVEL_NOTE,
 			    ER_BAD_TABLE_ERROR, ER(ER_BAD_TABLE_ERROR),
 			    table->table_name);
       else
+      {
+        non_tmp_error = (drop_temporary ? non_tmp_error : TRUE);
         error= 1;
+      }
     }
     else
     {
       char *end;
       /*
         Cannot use the db_type from the table, since that might have changed
-        while waiting for the exclusive name lock. We are under LOCK_open,
-        so reading from the frm-file is safe.
+        while waiting for the exclusive name lock.
       */
       if (frm_db_type == DB_TYPE_UNKNOWN)
       {
@@ -2157,7 +2228,7 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
       if (error == HA_ERR_ROW_IS_REFERENCED)
       {
 	/* the table is referenced by a foreign key constraint */
-	foreign_key_error=1;
+	foreign_key_error= 1;
       }
       if (!error || error == ENOENT || error == HA_ERR_NO_SUCH_TABLE)
       {
@@ -2166,14 +2237,14 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
 	strmov(end,reg_ext);
         if (!(new_error= mysql_file_delete(key_file_frm, path, MYF(MY_WME))))
         {
-	  some_tables_deleted=1;
+          non_tmp_table_deleted= TRUE;
           new_error= Table_triggers_list::drop_all_triggers(thd, db,
                                                             table->table_name);
         }
         error|= new_error;
       }
+       non_tmp_error= error ? TRUE : non_tmp_error;
     }
-    mysql_mutex_unlock(&LOCK_open);
     if (error)
     {
       if (wrong_tables.length())
@@ -2187,11 +2258,12 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
                     my_printf_error(ER_BAD_TABLE_ERROR,
                                     ER(ER_BAD_TABLE_ERROR), MYF(0),
                                     table->table_name););
-
   }
   DEBUG_SYNC(thd, "rm_table_part2_before_binlog");
-  thd->thread_specific_used|= tmp_table_deleted;
+  thd->thread_specific_used|= (trans_tmp_table_deleted ||
+                               non_trans_tmp_table_deleted);
   error= 0;
+err:
   if (wrong_tables.length())
   {
     if (!foreign_key_error)
@@ -2202,85 +2274,48 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
     error= 1;
   }
 
-  if (some_tables_deleted || tmp_table_deleted || !error)
+  if (non_trans_tmp_table_deleted ||
+      trans_tmp_table_deleted || non_tmp_table_deleted)
   {
     query_cache_invalidate3(thd, tables, 0);
     if (!dont_log_query && mysql_bin_log.is_open())
     {
-      if (!thd->is_current_stmt_binlog_format_row() ||
-          (non_temp_tables_count > 0 && !tmp_table_deleted))
+      if (non_trans_tmp_table_deleted)
       {
-        /*
-          In this case, we are either using statement-based
-          replication or using row-based replication but have only
-          deleted one or more non-temporary tables (and no temporary
-          tables).  In this case, we can write the original query into
-          the binary log.
-        */
-        error |= write_bin_log(thd, !error, thd->query(), thd->query_length());
+          /* Chop of the last comma */
+          built_non_trans_tmp_query.chop();
+          built_non_trans_tmp_query.append(" /* generated by server */");
+          error |= thd->binlog_query(THD::STMT_QUERY_TYPE,
+                                     built_non_trans_tmp_query.ptr(),
+                                     built_non_trans_tmp_query.length(),
+                                     FALSE, FALSE, FALSE, 0);
       }
-      else if (thd->is_current_stmt_binlog_format_row() &&
-               tmp_table_deleted)
+      if (trans_tmp_table_deleted)
       {
-        if (non_temp_tables_count > 0)
-        {
-          /*
-            In this case we have deleted both temporary and
-            non-temporary tables, so:
-            - since we have deleted a non-temporary table we have to
-              binlog the statement, but
-            - since we have deleted a temporary table we cannot binlog
-              the statement (since the table may have not been created on the
-              slave - check "if" branch below, this might cause the slave to 
-              stop).
-
-            Instead, we write a built statement, only containing the
-            non-temporary tables, to the binary log
-          */
-          built_query.chop();                  // Chop of the last comma
+          /* Chop of the last comma */
+          built_trans_tmp_query.chop();
+          built_trans_tmp_query.append(" /* generated by server */");
+          error |= thd->binlog_query(THD::STMT_QUERY_TYPE,
+                                     built_trans_tmp_query.ptr(),
+                                     built_trans_tmp_query.length(),
+                                     TRUE, FALSE, FALSE, 0);
+      }
+      if (non_tmp_table_deleted)
+      {
+          /* Chop of the last comma */
+          built_query.chop();
           built_query.append(" /* generated by server */");
-          error|= write_bin_log(thd, !error, built_query.ptr(), built_query.length());
-        }
-
-        /*
-          One needs to always log any temporary table drop if the current
-          statement logging format is set to row. This happens because one
-          might have created a temporary table while the statement logging
-          format was statement and then switched to mixed or row format.
-        */
-        if (thd->is_current_stmt_binlog_format_row())
-        {
-          /*
-            In this case we have deleted some temporary tables but we are using
-            row based logging for the statement. However, thread uses mixed mode
-            format, thence we need to log the dropping as we cannot tell for
-            sure whether the create was logged as statement previously or not, ie,
-            before switching to row mode.
-          */
-          built_tmp_query.chop();                  // Chop of the last comma
-          built_tmp_query.append(" /* generated by server */");
-          /*
-            We cannot call the write_bin_log as we do not care about any errors
-            in the master as the statement is always DROP TEMPORARY TABLE IF EXISTS
-            and as such there will be no errors in the slave.
-          */
-          error|= thd->binlog_query(THD::STMT_QUERY_TYPE, built_tmp_query.ptr(),
-                                    built_tmp_query.length(), FALSE, FALSE, FALSE,
-                                    0);
-        }
+          int error_code = (non_tmp_error ?
+            (foreign_key_error ? ER_ROW_IS_REFERENCED : ER_BAD_TABLE_ERROR) : 0);
+          error |= thd->binlog_query(THD::STMT_QUERY_TYPE,
+                                     built_query.ptr(),
+                                     built_query.length(),
+                                     TRUE, FALSE, FALSE,
+                                     error_code);
       }
-
-      /*
-        The remaining cases are:
-        - no tables were deleted and
-        - only temporary tables were deleted and row-based
-          replication is used.
-        In both these cases, nothing should be written to the binary
-        log.
-      */
     }
   }
-err:
+
   if (!drop_temporary)
   {
     /*
@@ -3795,6 +3830,8 @@ void sp_prepare_create_field(THD *thd, Create_field *sql_field)
     internal_tmp_table  Set to 1 if this is an internal temporary table
 			(From ALTER TABLE)
     select_field_count
+    is_trans            identifies the type of engine where the table
+                        was created: either trans or non-trans.
 
   DESCRIPTION
     If one creates a temporary table, this is automatically opened
@@ -3819,7 +3856,8 @@ bool mysql_create_table_no_lock(THD *thd,
                                 HA_CREATE_INFO *create_info,
                                 Alter_info *alter_info,
                                 bool internal_tmp_table,
-                                uint select_field_count)
+                                uint select_field_count,
+                                bool *is_trans)
 {
   char		path[FN_REFLEN + 1];
   uint          path_length;
@@ -4042,7 +4080,6 @@ bool mysql_create_table_no_lock(THD *thd,
   {
     if (create_info->options & HA_LEX_CREATE_IF_NOT_EXISTS)
     {
-      create_info->table_existed= 1;		// Mark that table existed
       push_warning_printf(thd, MYSQL_ERROR::WARN_LEVEL_NOTE,
                           ER_TABLE_EXISTS_ERROR, ER(ER_TABLE_EXISTS_ERROR),
                           alias);
@@ -4053,7 +4090,6 @@ bool mysql_create_table_no_lock(THD *thd,
     goto err;
   }
 
-  mysql_mutex_lock(&LOCK_open);
   if (!internal_tmp_table && !(create_info->options & HA_LEX_CREATE_TMP_TABLE))
   {
     if (!access(path,F_OK))
@@ -4061,7 +4097,7 @@ bool mysql_create_table_no_lock(THD *thd,
       if (create_info->options & HA_LEX_CREATE_IF_NOT_EXISTS)
         goto warn;
       my_error(ER_TABLE_EXISTS_ERROR,MYF(0),table_name);
-      goto unlock_and_end;
+      goto err;
     }
     /*
       We don't assert here, but check the result, because the table could be
@@ -4071,11 +4107,14 @@ bool mysql_create_table_no_lock(THD *thd,
       Then she could create the table. This case is pretty obscure and
       therefore we don't introduce a new error message only for it.
     */
+    mysql_mutex_lock(&LOCK_open);
     if (get_cached_table_share(db, table_name))
     {
+      mysql_mutex_unlock(&LOCK_open);
       my_error(ER_TABLE_EXISTS_ERROR, MYF(0), table_name);
-      goto unlock_and_end;
+      goto err;
     }
+    mysql_mutex_unlock(&LOCK_open);
   }
 
   /*
@@ -4083,7 +4122,7 @@ bool mysql_create_table_no_lock(THD *thd,
     exist in any storage engine. In such a case it should
     be discovered and the error ER_TABLE_EXISTS_ERROR be returned
     unless user specified CREATE TABLE IF EXISTS
-    The LOCK_open mutex has been locked to make sure no
+    An exclusive metadata lock ensures that no
     one else is attempting to discover the table. Since
     it's not on disk as a frm file, no one could be using it!
   */
@@ -4104,17 +4143,16 @@ bool mysql_create_table_no_lock(THD *thd,
         if (create_if_not_exists)
           goto warn;
         my_error(ER_TABLE_EXISTS_ERROR,MYF(0),table_name);
-        goto unlock_and_end;
+        goto err;
         break;
       default:
         DBUG_PRINT("info", ("error: %u from storage engine", retcode));
         my_error(retcode, MYF(0),table_name);
-        goto unlock_and_end;
+        goto err;
     }
   }
 
   thd_proc_info(thd, "creating table");
-  create_info->table_existed= 0;		// Mark that table is created
 
 #ifdef HAVE_READLINK
   {
@@ -4142,7 +4180,7 @@ bool mysql_create_table_no_lock(THD *thd,
       if (test_if_data_home_dir(dirpath))
       {
         my_error(ER_WRONG_ARGUMENTS, MYF(0), "DATA DIRECTORY");
-        goto unlock_and_end;
+        goto err;
       }
     }
     if (create_info->index_file_name)
@@ -4151,7 +4189,7 @@ bool mysql_create_table_no_lock(THD *thd,
       if (test_if_data_home_dir(dirpath))
       {
         my_error(ER_WRONG_ARGUMENTS, MYF(0), "INDEX DIRECTORY");
-        goto unlock_and_end;
+        goto err;
       }
     }
   }
@@ -4159,7 +4197,7 @@ bool mysql_create_table_no_lock(THD *thd,
 #ifdef WITH_PARTITION_STORAGE_ENGINE
   if (check_partition_dirs(thd->lex->part_info))
   {
-    goto unlock_and_end;
+    goto err;
   }
 #endif /* WITH_PARTITION_STORAGE_ENGINE */
 
@@ -4182,23 +4220,25 @@ bool mysql_create_table_no_lock(THD *thd,
   if (rea_create_table(thd, path, db, table_name,
                        create_info, alter_info->create_list,
                        key_count, key_info_buffer, file))
-    goto unlock_and_end;
+    goto err;
 
   if (create_info->options & HA_LEX_CREATE_TMP_TABLE)
   {
+    TABLE *table= NULL;
     /* Open table and put in temporary table list */
-    if (!(open_temporary_table(thd, path, db, table_name, 1)))
+    if (!(table= open_temporary_table(thd, path, db, table_name, 1)))
     {
       (void) rm_temporary_table(create_info->db_type, path);
-      goto unlock_and_end;
+      goto err;
     }
+
+    if (is_trans != NULL)
+      *is_trans= table->file->has_transactions();
+
     thd->thread_specific_used= TRUE;
   }
 
   error= FALSE;
-unlock_and_end:
-  mysql_mutex_unlock(&LOCK_open);
-
 err:
   thd_proc_info(thd, "After create");
   delete file;
@@ -4209,8 +4249,7 @@ warn:
   push_warning_printf(thd, MYSQL_ERROR::WARN_LEVEL_NOTE,
                       ER_TABLE_EXISTS_ERROR, ER(ER_TABLE_EXISTS_ERROR),
                       alias);
-  create_info->table_existed= 1;		// Mark that table existed
-  goto unlock_and_end;
+  goto err;
 }
 
 
@@ -4229,6 +4268,7 @@ bool mysql_create_table(THD *thd, TABLE_LIST *create_table,
                         Alter_info *alter_info)
 {
   bool result;
+  bool is_trans= FALSE;
   DBUG_ENTER("mysql_create_table");
 
   /*
@@ -4245,7 +4285,7 @@ bool mysql_create_table(THD *thd, TABLE_LIST *create_table,
 
   result= mysql_create_table_no_lock(thd, create_table->db,
                                      create_table->table_name, create_info,
-                                     alter_info, FALSE, 0);
+                                     alter_info, FALSE, 0, &is_trans);
 
   /*
     Don't write statement if:
@@ -4257,7 +4297,7 @@ bool mysql_create_table(THD *thd, TABLE_LIST *create_table,
       (!thd->is_current_stmt_binlog_format_row() ||
        (thd->is_current_stmt_binlog_format_row() &&
         !(create_info->options & HA_LEX_CREATE_TMP_TABLE))))
-    result= write_bin_log(thd, TRUE, thd->query(), thd->query_length());
+    result= write_bin_log(thd, TRUE, thd->query(), thd->query_length(), is_trans);
 
 end:
   DBUG_RETURN(result);
@@ -4393,885 +4433,6 @@ mysql_rename_table(handlerton *base, const char *old_db,
 }
 
 
-static int send_check_errmsg(THD *thd, TABLE_LIST* table,
-			     const char* operator_name, const char* errmsg)
-
-{
-  Protocol *protocol= thd->protocol;
-  protocol->prepare_for_resend();
-  protocol->store(table->alias, system_charset_info);
-  protocol->store((char*) operator_name, system_charset_info);
-  protocol->store(STRING_WITH_LEN("error"), system_charset_info);
-  protocol->store(errmsg, system_charset_info);
-  thd->clear_error();
-  if (protocol->write())
-    return -1;
-  return 1;
-}
-
-
-static int prepare_for_repair(THD *thd, TABLE_LIST *table_list,
-			      HA_CHECK_OPT *check_opt)
-{
-  int error= 0;
-  TABLE tmp_table, *table;
-  TABLE_SHARE *share;
-  bool has_mdl_lock= FALSE;
-  char from[FN_REFLEN],tmp[FN_REFLEN+32];
-  const char **ext;
-  MY_STAT stat_info;
-  Open_table_context ot_ctx(thd, (MYSQL_OPEN_IGNORE_FLUSH |
-                                  MYSQL_OPEN_HAS_MDL_LOCK |
-                                  MYSQL_LOCK_IGNORE_TIMEOUT));
-  DBUG_ENTER("prepare_for_repair");
-
-  if (!(check_opt->sql_flags & TT_USEFRM))
-    DBUG_RETURN(0);
-
-  if (!(table= table_list->table))
-  {
-    char key[MAX_DBKEY_LENGTH];
-    uint key_length;
-    /*
-      If the table didn't exist, we have a shared metadata lock
-      on it that is left from mysql_admin_table()'s attempt to 
-      open it. Release the shared metadata lock before trying to
-      acquire the exclusive lock to satisfy MDL asserts and avoid
-      deadlocks.
-    */
-    thd->mdl_context.release_transactional_locks();
-    /*
-      Attempt to do full-blown table open in mysql_admin_table() has failed.
-      Let us try to open at least a .FRM for this table.
-    */
-    my_hash_value_type hash_value;
-
-    key_length= create_table_def_key(thd, key, table_list, 0);
-    table_list->mdl_request.init(MDL_key::TABLE,
-                                 table_list->db, table_list->table_name,
-                                 MDL_EXCLUSIVE);
-
-    if (lock_table_names(thd, table_list, table_list->next_global,
-                         thd->variables.lock_wait_timeout,
-                         MYSQL_OPEN_SKIP_TEMPORARY))
-      DBUG_RETURN(0);
-    has_mdl_lock= TRUE;
-
-    hash_value= my_calc_hash(&table_def_cache, (uchar*) key, key_length);
-    mysql_mutex_lock(&LOCK_open);
-    if (!(share= (get_table_share(thd, table_list, key, key_length, 0,
-                                  &error, hash_value))))
-    {
-      mysql_mutex_unlock(&LOCK_open);
-      DBUG_RETURN(0);				// Can't open frm file
-    }
-
-    if (open_table_from_share(thd, share, "", 0, 0, 0, &tmp_table, FALSE))
-    {
-      release_table_share(share);
-      mysql_mutex_unlock(&LOCK_open);
-      DBUG_RETURN(0);                           // Out of memory
-    }
-    mysql_mutex_unlock(&LOCK_open);
-    table= &tmp_table;
-  }
-
-  /* A MERGE table must not come here. */
-  DBUG_ASSERT(table->file->ht->db_type != DB_TYPE_MRG_MYISAM);
-
-  /*
-    REPAIR TABLE ... USE_FRM for temporary tables makes little sense.
-  */
-  if (table->s->tmp_table)
-  {
-    error= send_check_errmsg(thd, table_list, "repair",
-			     "Cannot repair temporary table from .frm file");
-    goto end;
-  }
-
-  /*
-    User gave us USE_FRM which means that the header in the index file is
-    trashed.
-    In this case we will try to fix the table the following way:
-    - Rename the data file to a temporary name
-    - Truncate the table
-    - Replace the new data file with the old one
-    - Run a normal repair using the new index file and the old data file
-  */
-
-  if (table->s->frm_version != FRM_VER_TRUE_VARCHAR)
-  {
-    error= send_check_errmsg(thd, table_list, "repair",
-                             "Failed repairing incompatible .frm file");
-    goto end;
-  }
-
-  /*
-    Check if this is a table type that stores index and data separately,
-    like ISAM or MyISAM. We assume fixed order of engine file name
-    extentions array. First element of engine file name extentions array
-    is meta/index file extention. Second element - data file extention. 
-  */
-  ext= table->file->bas_ext();
-  if (!ext[0] || !ext[1])
-    goto end;					// No data file
-
-  // Name of data file
-  strxmov(from, table->s->normalized_path.str, ext[1], NullS);
-  if (!mysql_file_stat(key_file_misc, from, &stat_info, MYF(0)))
-    goto end;				// Can't use USE_FRM flag
-
-  my_snprintf(tmp, sizeof(tmp), "%s-%lx_%lx",
-	      from, current_pid, thd->thread_id);
-
-  if (table_list->table)
-  {
-    /*
-      Table was successfully open in mysql_admin_table(). Now we need
-      to close it, but leave it protected by exclusive metadata lock.
-    */
-    if (wait_while_table_is_used(thd, table, HA_EXTRA_FORCE_REOPEN))
-      goto end;
-    close_all_tables_for_name(thd, table_list->table->s, FALSE);
-    table_list->table= 0;
-  }
-  /*
-    After this point we have an exclusive metadata lock on our table
-    in both cases when table was successfully open in mysql_admin_table()
-    and when it was open in prepare_for_repair().
-  */
-
-  if (my_rename(from, tmp, MYF(MY_WME)))
-  {
-    error= send_check_errmsg(thd, table_list, "repair",
-			     "Failed renaming data file");
-    goto end;
-  }
-  if (dd_recreate_table(thd, table_list->db, table_list->table_name))
-  {
-    error= send_check_errmsg(thd, table_list, "repair",
-			     "Failed generating table from .frm file");
-    goto end;
-  }
-  /*
-    'FALSE' for 'using_transactions' means don't postpone
-    invalidation till the end of a transaction, but do it
-    immediately.
-  */
-  query_cache_invalidate3(thd, table_list, FALSE);
-  if (mysql_file_rename(key_file_misc, tmp, from, MYF(MY_WME)))
-  {
-    error= send_check_errmsg(thd, table_list, "repair",
-			     "Failed restoring .MYD file");
-    goto end;
-  }
-
-  if (thd->locked_tables_list.reopen_tables(thd))
-    goto end;
-
-  /*
-    Now we should be able to open the partially repaired table
-    to finish the repair in the handler later on.
-  */
-  if (open_table(thd, table_list, thd->mem_root, &ot_ctx))
-  {
-    error= send_check_errmsg(thd, table_list, "repair",
-                             "Failed to open partially repaired table");
-    goto end;
-  }
-
-end:
-  thd->locked_tables_list.unlink_all_closed_tables(thd, NULL, 0);
-  if (table == &tmp_table)
-  {
-    mysql_mutex_lock(&LOCK_open);
-    closefrm(table, 1);				// Free allocated memory
-    mysql_mutex_unlock(&LOCK_open);
-  }
-  /* In case of a temporary table there will be no metadata lock. */
-  if (error && has_mdl_lock)
-    thd->mdl_context.release_transactional_locks();
-
-  DBUG_RETURN(error);
-}
-
-
-
-/*
-  RETURN VALUES
-    FALSE Message sent to net (admin operation went ok)
-    TRUE  Message should be sent by caller 
-          (admin operation or network communication failed)
-*/
-static bool mysql_admin_table(THD* thd, TABLE_LIST* tables,
-                              HA_CHECK_OPT* check_opt,
-                              const char *operator_name,
-                              thr_lock_type lock_type,
-                              bool open_for_modify,
-                              bool no_warnings_for_error,
-                              uint extra_open_options,
-                              int (*prepare_func)(THD *, TABLE_LIST *,
-                                                  HA_CHECK_OPT *),
-                              int (handler::*operator_func)(THD *,
-                                                            HA_CHECK_OPT *),
-                              int (view_operator_func)(THD *, TABLE_LIST*))
-{
-  TABLE_LIST *table;
-  SELECT_LEX *select= &thd->lex->select_lex;
-  List<Item> field_list;
-  Item *item;
-  Protocol *protocol= thd->protocol;
-  LEX *lex= thd->lex;
-  int result_code;
-  DBUG_ENTER("mysql_admin_table");
-
-  field_list.push_back(item = new Item_empty_string("Table", NAME_CHAR_LEN*2));
-  item->maybe_null = 1;
-  field_list.push_back(item = new Item_empty_string("Op", 10));
-  item->maybe_null = 1;
-  field_list.push_back(item = new Item_empty_string("Msg_type", 10));
-  item->maybe_null = 1;
-  field_list.push_back(item = new Item_empty_string("Msg_text", 255));
-  item->maybe_null = 1;
-  if (protocol->send_result_set_metadata(&field_list,
-                            Protocol::SEND_NUM_ROWS | Protocol::SEND_EOF))
-    DBUG_RETURN(TRUE);
-
-  mysql_ha_rm_tables(thd, tables);
-
-  for (table= tables; table; table= table->next_local)
-  {
-    char table_name[NAME_LEN*2+2];
-    char* db = table->db;
-    bool fatal_error=0;
-    bool open_error;
-
-    DBUG_PRINT("admin", ("table: '%s'.'%s'", table->db, table->table_name));
-    DBUG_PRINT("admin", ("extra_open_options: %u", extra_open_options));
-    strxmov(table_name, db, ".", table->table_name, NullS);
-    thd->open_options|= extra_open_options;
-    table->lock_type= lock_type;
-    /*
-      To make code safe for re-execution we need to reset type of MDL
-      request as code below may change it.
-      To allow concurrent execution of read-only operations we acquire
-      weak metadata lock for them.
-    */
-    table->mdl_request.set_type((lock_type >= TL_WRITE_ALLOW_WRITE) ?
-                                MDL_SHARED_NO_READ_WRITE : MDL_SHARED_READ);
-    /* open only one table from local list of command */
-    {
-      TABLE_LIST *save_next_global, *save_next_local;
-      save_next_global= table->next_global;
-      table->next_global= 0;
-      save_next_local= table->next_local;
-      table->next_local= 0;
-      select->table_list.first= table;
-      /*
-        Time zone tables and SP tables can be add to lex->query_tables list,
-        so it have to be prepared.
-        TODO: Investigate if we can put extra tables into argument instead of
-        using lex->query_tables
-      */
-      lex->query_tables= table;
-      lex->query_tables_last= &table->next_global;
-      lex->query_tables_own_last= 0;
-      thd->no_warnings_for_error= no_warnings_for_error;
-      if (view_operator_func == NULL)
-        table->required_type=FRMTYPE_TABLE;
-
-      open_error= open_and_lock_tables(thd, table, TRUE, 0);
-      thd->no_warnings_for_error= 0;
-      table->next_global= save_next_global;
-      table->next_local= save_next_local;
-      thd->open_options&= ~extra_open_options;
-      /*
-        Under locked tables, we know that the table can be opened,
-        so any errors opening the table are logical errors.
-        In these cases it does not make sense to try to repair.
-      */
-      if (open_error && thd->locked_tables_mode)
-      {
-        result_code= HA_ADMIN_FAILED;
-        goto send_result;
-      }
-#ifdef WITH_PARTITION_STORAGE_ENGINE
-      if (table->table)
-      {
-        /*
-          Set up which partitions that should be processed
-          if ALTER TABLE t ANALYZE/CHECK/OPTIMIZE/REPAIR PARTITION ..
-          CACHE INDEX/LOAD INDEX for specified partitions
-        */
-        Alter_info *alter_info= &lex->alter_info;
-
-        if (alter_info->flags & ALTER_ADMIN_PARTITION)
-        {
-          if (!table->table->part_info)
-          {
-            my_error(ER_PARTITION_MGMT_ON_NONPARTITIONED, MYF(0));
-            DBUG_RETURN(TRUE);
-          }
-          uint num_parts_found;
-          uint num_parts_opt= alter_info->partition_names.elements;
-          num_parts_found= set_part_state(alter_info, table->table->part_info,
-                                          PART_ADMIN);
-          if (num_parts_found != num_parts_opt &&
-              (!(alter_info->flags & ALTER_ALL_PARTITION)))
-          {
-            char buff[FN_REFLEN + MYSQL_ERRMSG_SIZE];
-            size_t length;
-            DBUG_PRINT("admin", ("sending non existent partition error"));
-            protocol->prepare_for_resend();
-            protocol->store(table_name, system_charset_info);
-            protocol->store(operator_name, system_charset_info);
-            protocol->store(STRING_WITH_LEN("error"), system_charset_info);
-            length= my_snprintf(buff, sizeof(buff),
-                                ER(ER_DROP_PARTITION_NON_EXISTENT),
-                                table_name);
-            protocol->store(buff, length, system_charset_info);
-            if(protocol->write())
-              goto err;
-            my_eof(thd);
-            goto err;
-          }
-        }
-      }
-#endif
-    }
-    DBUG_PRINT("admin", ("table: 0x%lx", (long) table->table));
-
-    if (prepare_func)
-    {
-      DBUG_PRINT("admin", ("calling prepare_func"));
-      switch ((*prepare_func)(thd, table, check_opt)) {
-      case  1:           // error, message written to net
-        trans_rollback_stmt(thd);
-        trans_rollback(thd);
-        close_thread_tables(thd);
-        thd->mdl_context.release_transactional_locks();
-        DBUG_PRINT("admin", ("simple error, admin next table"));
-        continue;
-      case -1:           // error, message could be written to net
-        /* purecov: begin inspected */
-        DBUG_PRINT("admin", ("severe error, stop"));
-        goto err;
-        /* purecov: end */
-      default:           // should be 0 otherwise
-        DBUG_PRINT("admin", ("prepare_func succeeded"));
-        ;
-      }
-    }
-
-    /*
-      CHECK TABLE command is only command where VIEW allowed here and this
-      command use only temporary teble method for VIEWs resolving => there
-      can't be VIEW tree substitition of join view => if opening table
-      succeed then table->table will have real TABLE pointer as value (in
-      case of join view substitution table->table can be 0, but here it is
-      impossible)
-    */
-    if (!table->table)
-    {
-      DBUG_PRINT("admin", ("open table failed"));
-      if (thd->warning_info->is_empty())
-        push_warning(thd, MYSQL_ERROR::WARN_LEVEL_WARN,
-                     ER_CHECK_NO_SUCH_TABLE, ER(ER_CHECK_NO_SUCH_TABLE));
-      /* if it was a view will check md5 sum */
-      if (table->view &&
-          view_checksum(thd, table) == HA_ADMIN_WRONG_CHECKSUM)
-        push_warning(thd, MYSQL_ERROR::WARN_LEVEL_WARN,
-                     ER_VIEW_CHECKSUM, ER(ER_VIEW_CHECKSUM));
-      if (thd->stmt_da->is_error() &&
-          (thd->stmt_da->sql_errno() == ER_NO_SUCH_TABLE ||
-           thd->stmt_da->sql_errno() == ER_FILE_NOT_FOUND))
-        /* A missing table is just issued as a failed command */
-        result_code= HA_ADMIN_FAILED;
-      else
-        /* Default failure code is corrupt table */
-        result_code= HA_ADMIN_CORRUPT;
-      goto send_result;
-    }
-
-    if (table->view)
-    {
-      DBUG_PRINT("admin", ("calling view_operator_func"));
-      result_code= (*view_operator_func)(thd, table);
-      goto send_result;
-    }
-
-    if (table->schema_table)
-    {
-      result_code= HA_ADMIN_NOT_IMPLEMENTED;
-      goto send_result;
-    }
-
-    if ((table->table->db_stat & HA_READ_ONLY) && open_for_modify)
-    {
-      /* purecov: begin inspected */
-      char buff[FN_REFLEN + MYSQL_ERRMSG_SIZE];
-      size_t length;
-      enum_sql_command save_sql_command= lex->sql_command;
-      DBUG_PRINT("admin", ("sending error message"));
-      protocol->prepare_for_resend();
-      protocol->store(table_name, system_charset_info);
-      protocol->store(operator_name, system_charset_info);
-      protocol->store(STRING_WITH_LEN("error"), system_charset_info);
-      length= my_snprintf(buff, sizeof(buff), ER(ER_OPEN_AS_READONLY),
-                          table_name);
-      protocol->store(buff, length, system_charset_info);
-      trans_commit_stmt(thd);
-      trans_commit(thd);
-      close_thread_tables(thd);
-      thd->mdl_context.release_transactional_locks();
-      lex->reset_query_tables_list(FALSE);
-      /*
-        Restore Query_tables_list::sql_command value to make statement
-        safe for re-execution.
-      */
-      lex->sql_command= save_sql_command;
-      table->table=0;				// For query cache
-      if (protocol->write())
-	goto err;
-      thd->stmt_da->reset_diagnostics_area();
-      continue;
-      /* purecov: end */
-    }
-
-    /*
-      Close all instances of the table to allow MyISAM "repair"
-      to rename files.
-      @todo: This code does not close all instances of the table.
-      It only closes instances in other connections, but if this
-      connection has LOCK TABLE t1 a READ, t1 b WRITE,
-      both t1 instances will be kept open.
-      There is no need to execute this branch for InnoDB, which does
-      repair by recreate. There is no need to do it for OPTIMIZE,
-      which doesn't move files around.
-      Hence, this code should be moved to prepare_for_repair(),
-      and executed only for MyISAM engine.
-    */
-    if (lock_type == TL_WRITE && !table->table->s->tmp_table)
-    {
-      if (wait_while_table_is_used(thd, table->table,
-                                   HA_EXTRA_PREPARE_FOR_RENAME))
-        goto err;
-      DEBUG_SYNC(thd, "after_admin_flush");
-      /* Flush entries in the query cache involving this table. */
-      query_cache_invalidate3(thd, table->table, 0);
-      /*
-        XXX: hack: switch off open_for_modify to skip the
-        flush that is made later in the execution flow. 
-      */
-      open_for_modify= 0;
-    }
-
-    if (table->table->s->crashed && operator_func == &handler::ha_check)
-    {
-      /* purecov: begin inspected */
-      DBUG_PRINT("admin", ("sending crashed warning"));
-      protocol->prepare_for_resend();
-      protocol->store(table_name, system_charset_info);
-      protocol->store(operator_name, system_charset_info);
-      protocol->store(STRING_WITH_LEN("warning"), system_charset_info);
-      protocol->store(STRING_WITH_LEN("Table is marked as crashed"),
-                      system_charset_info);
-      if (protocol->write())
-        goto err;
-      /* purecov: end */
-    }
-
-    if (operator_func == &handler::ha_repair &&
-        !(check_opt->sql_flags & TT_USEFRM))
-    {
-      if ((table->table->file->check_old_types() == HA_ADMIN_NEEDS_ALTER) ||
-          (table->table->file->ha_check_for_upgrade(check_opt) ==
-           HA_ADMIN_NEEDS_ALTER))
-      {
-        DBUG_PRINT("admin", ("recreating table"));
-        trans_rollback_stmt(thd);
-        trans_rollback(thd);
-        close_thread_tables(thd);
-        thd->mdl_context.release_transactional_locks();
-        tmp_disable_binlog(thd); // binlogging is done by caller if wanted
-        result_code= mysql_recreate_table(thd, table);
-        reenable_binlog(thd);
-        /*
-          mysql_recreate_table() can push OK or ERROR.
-          Clear 'OK' status. If there is an error, keep it:
-          we will store the error message in a result set row 
-          and then clear.
-        */
-        if (thd->stmt_da->is_ok())
-          thd->stmt_da->reset_diagnostics_area();
-        table->table= NULL;
-        result_code= result_code ? HA_ADMIN_FAILED : HA_ADMIN_OK;
-        goto send_result;
-      }
-    }
-
-    DBUG_PRINT("admin", ("calling operator_func '%s'", operator_name));
-    result_code = (table->table->file->*operator_func)(thd, check_opt);
-    DBUG_PRINT("admin", ("operator_func returned: %d", result_code));
-
-send_result:
-
-    lex->cleanup_after_one_table_open();
-    thd->clear_error();  // these errors shouldn't get client
-    {
-      List_iterator_fast<MYSQL_ERROR> it(thd->warning_info->warn_list());
-      MYSQL_ERROR *err;
-      while ((err= it++))
-      {
-        protocol->prepare_for_resend();
-        protocol->store(table_name, system_charset_info);
-        protocol->store((char*) operator_name, system_charset_info);
-        protocol->store(warning_level_names[err->get_level()].str,
-                        warning_level_names[err->get_level()].length,
-                        system_charset_info);
-        protocol->store(err->get_message_text(), system_charset_info);
-        if (protocol->write())
-          goto err;
-      }
-      thd->warning_info->clear_warning_info(thd->query_id);
-    }
-    protocol->prepare_for_resend();
-    protocol->store(table_name, system_charset_info);
-    protocol->store(operator_name, system_charset_info);
-
-send_result_message:
-
-    DBUG_PRINT("info", ("result_code: %d", result_code));
-    switch (result_code) {
-    case HA_ADMIN_NOT_IMPLEMENTED:
-      {
-       char buf[MYSQL_ERRMSG_SIZE];
-       size_t length=my_snprintf(buf, sizeof(buf),
-				ER(ER_CHECK_NOT_IMPLEMENTED), operator_name);
-	protocol->store(STRING_WITH_LEN("note"), system_charset_info);
-	protocol->store(buf, length, system_charset_info);
-      }
-      break;
-
-    case HA_ADMIN_NOT_BASE_TABLE:
-      {
-        char buf[MYSQL_ERRMSG_SIZE];
-        size_t length= my_snprintf(buf, sizeof(buf),
-                                 ER(ER_BAD_TABLE_ERROR), table_name);
-        protocol->store(STRING_WITH_LEN("note"), system_charset_info);
-        protocol->store(buf, length, system_charset_info);
-      }
-      break;
-
-    case HA_ADMIN_OK:
-      protocol->store(STRING_WITH_LEN("status"), system_charset_info);
-      protocol->store(STRING_WITH_LEN("OK"), system_charset_info);
-      break;
-
-    case HA_ADMIN_FAILED:
-      protocol->store(STRING_WITH_LEN("status"), system_charset_info);
-      protocol->store(STRING_WITH_LEN("Operation failed"),
-                      system_charset_info);
-      break;
-
-    case HA_ADMIN_REJECT:
-      protocol->store(STRING_WITH_LEN("status"), system_charset_info);
-      protocol->store(STRING_WITH_LEN("Operation need committed state"),
-                      system_charset_info);
-      open_for_modify= FALSE;
-      break;
-
-    case HA_ADMIN_ALREADY_DONE:
-      protocol->store(STRING_WITH_LEN("status"), system_charset_info);
-      protocol->store(STRING_WITH_LEN("Table is already up to date"),
-                      system_charset_info);
-      break;
-
-    case HA_ADMIN_CORRUPT:
-      protocol->store(STRING_WITH_LEN("error"), system_charset_info);
-      protocol->store(STRING_WITH_LEN("Corrupt"), system_charset_info);
-      fatal_error=1;
-      break;
-
-    case HA_ADMIN_INVALID:
-      protocol->store(STRING_WITH_LEN("error"), system_charset_info);
-      protocol->store(STRING_WITH_LEN("Invalid argument"),
-                      system_charset_info);
-      break;
-
-    case HA_ADMIN_TRY_ALTER:
-    {
-      /*
-        This is currently used only by InnoDB. ha_innobase::optimize() answers
-        "try with alter", so here we close the table, do an ALTER TABLE,
-        reopen the table and do ha_innobase::analyze() on it.
-        We have to end the row, so analyze could return more rows.
-      */
-      trans_commit_stmt(thd);
-      trans_commit(thd);
-      close_thread_tables(thd);
-      thd->mdl_context.release_transactional_locks();
-      DEBUG_SYNC(thd, "ha_admin_try_alter");
-      protocol->store(STRING_WITH_LEN("note"), system_charset_info);
-      protocol->store(STRING_WITH_LEN(
-          "Table does not support optimize, doing recreate + analyze instead"),
-                      system_charset_info);
-      if (protocol->write())
-        goto err;
-      DBUG_PRINT("info", ("HA_ADMIN_TRY_ALTER, trying analyze..."));
-      TABLE_LIST *save_next_local= table->next_local,
-                 *save_next_global= table->next_global;
-      table->next_local= table->next_global= 0;
-      tmp_disable_binlog(thd); // binlogging is done by caller if wanted
-      result_code= mysql_recreate_table(thd, table);
-      reenable_binlog(thd);
-      /*
-        mysql_recreate_table() can push OK or ERROR.
-        Clear 'OK' status. If there is an error, keep it:
-        we will store the error message in a result set row 
-        and then clear.
-      */
-      if (thd->stmt_da->is_ok())
-        thd->stmt_da->reset_diagnostics_area();
-      trans_commit_stmt(thd);
-      trans_commit(thd);
-      close_thread_tables(thd);
-      thd->mdl_context.release_transactional_locks();
-      table->table= NULL;
-      if (!result_code) // recreation went ok
-      {
-        /* Clear the ticket released above. */
-        table->mdl_request.ticket= NULL;
-        DEBUG_SYNC(thd, "ha_admin_open_ltable");
-        table->mdl_request.set_type(MDL_SHARED_WRITE);
-        if ((table->table= open_ltable(thd, table, lock_type, 0)))
-        {
-          result_code= table->table->file->ha_analyze(thd, check_opt);
-          if (result_code == HA_ADMIN_ALREADY_DONE)
-            result_code= HA_ADMIN_OK;
-          else if (result_code)  // analyze failed
-            table->table->file->print_error(result_code, MYF(0));
-        }
-        else
-          result_code= -1; // open failed
-      }
-      /* Start a new row for the final status row */
-      protocol->prepare_for_resend();
-      protocol->store(table_name, system_charset_info);
-      protocol->store(operator_name, system_charset_info);
-      if (result_code) // either mysql_recreate_table or analyze failed
-      {
-        DBUG_ASSERT(thd->is_error());
-        if (thd->is_error())
-        {
-          const char *err_msg= thd->stmt_da->message();
-          if (!thd->vio_ok())
-          {
-            sql_print_error("%s", err_msg);
-          }
-          else
-          {
-            /* Hijack the row already in-progress. */
-            protocol->store(STRING_WITH_LEN("error"), system_charset_info);
-            protocol->store(err_msg, system_charset_info);
-            if (protocol->write())
-              goto err;
-            /* Start off another row for HA_ADMIN_FAILED */
-            protocol->prepare_for_resend();
-            protocol->store(table_name, system_charset_info);
-            protocol->store(operator_name, system_charset_info);
-          }
-          thd->clear_error();
-        }
-      }
-      result_code= result_code ? HA_ADMIN_FAILED : HA_ADMIN_OK;
-      table->next_local= save_next_local;
-      table->next_global= save_next_global;
-      goto send_result_message;
-    }
-    case HA_ADMIN_WRONG_CHECKSUM:
-    {
-      protocol->store(STRING_WITH_LEN("note"), system_charset_info);
-      protocol->store(ER(ER_VIEW_CHECKSUM), strlen(ER(ER_VIEW_CHECKSUM)),
-                      system_charset_info);
-      break;
-    }
-
-    case HA_ADMIN_NEEDS_UPGRADE:
-    case HA_ADMIN_NEEDS_ALTER:
-    {
-      char buf[MYSQL_ERRMSG_SIZE];
-      size_t length;
-
-      protocol->store(STRING_WITH_LEN("error"), system_charset_info);
-      length=my_snprintf(buf, sizeof(buf), ER(ER_TABLE_NEEDS_UPGRADE),
-                         table->table_name);
-      protocol->store(buf, length, system_charset_info);
-      fatal_error=1;
-      break;
-    }
-
-    default:				// Probably HA_ADMIN_INTERNAL_ERROR
-      {
-        char buf[MYSQL_ERRMSG_SIZE];
-        size_t length=my_snprintf(buf, sizeof(buf),
-                                "Unknown - internal error %d during operation",
-                                result_code);
-        protocol->store(STRING_WITH_LEN("error"), system_charset_info);
-        protocol->store(buf, length, system_charset_info);
-        fatal_error=1;
-        break;
-      }
-    }
-    if (table->table)
-    {
-      if (table->table->s->tmp_table)
-      {
-        /*
-          If the table was not opened successfully, do not try to get
-          status information. (Bug#47633)
-        */
-        if (open_for_modify && !open_error)
-          table->table->file->info(HA_STATUS_CONST);
-      }
-      else if (open_for_modify || fatal_error)
-      {
-        mysql_mutex_lock(&LOCK_open);
-        tdc_remove_table(thd, TDC_RT_REMOVE_UNUSED,
-                         table->db, table->table_name);
-        mysql_mutex_unlock(&LOCK_open);
-        /*
-          May be something modified. Consequently, we have to
-          invalidate the query cache.
-        */
-        table->table= 0;			// For query cache
-        query_cache_invalidate3(thd, table, 0);
-      }
-    }
-    /* Error path, a admin command failed. */
-    trans_commit_stmt(thd);
-    trans_commit_implicit(thd);
-    close_thread_tables(thd);
-    thd->mdl_context.release_transactional_locks();
-
-    /*
-      If it is CHECK TABLE v1, v2, v3, and v1, v2, v3 are views, we will run
-      separate open_tables() for each CHECK TABLE argument.
-      Right now we do not have a separate method to reset the prelocking
-      state in the lex to the state after parsing, so each open will pollute
-      this state: add elements to lex->srotuines_list, TABLE_LISTs to
-      lex->query_tables. Below is a lame attempt to recover from this
-      pollution.
-      @todo: have a method to reset a prelocking context, or use separate
-      contexts for each open.
-    */
-    for (Sroutine_hash_entry *rt=
-           (Sroutine_hash_entry*)thd->lex->sroutines_list.first;
-         rt; rt= rt->next)
-      rt->mdl_request.ticket= NULL;
-
-    if (protocol->write())
-      goto err;
-  }
-
-  my_eof(thd);
-  DBUG_RETURN(FALSE);
-
-err:
-  trans_rollback_stmt(thd);
-  trans_rollback(thd);
-  close_thread_tables(thd);			// Shouldn't be needed
-  thd->mdl_context.release_transactional_locks();
-  if (table)
-    table->table=0;
-  DBUG_RETURN(TRUE);
-}
-
-
-bool mysql_repair_table(THD* thd, TABLE_LIST* tables, HA_CHECK_OPT* check_opt)
-{
-  DBUG_ENTER("mysql_repair_table");
-  DBUG_RETURN(mysql_admin_table(thd, tables, check_opt,
-				"repair", TL_WRITE, 1,
-                                test(check_opt->sql_flags & TT_USEFRM),
-                                HA_OPEN_FOR_REPAIR,
-				&prepare_for_repair,
-				&handler::ha_repair, 0));
-}
-
-
-bool mysql_optimize_table(THD* thd, TABLE_LIST* tables, HA_CHECK_OPT* check_opt)
-{
-  DBUG_ENTER("mysql_optimize_table");
-  DBUG_RETURN(mysql_admin_table(thd, tables, check_opt,
-				"optimize", TL_WRITE, 1,0,0,0,
-				&handler::ha_optimize, 0));
-}
-
-
-/*
-  Assigned specified indexes for a table into key cache
-
-  SYNOPSIS
-    mysql_assign_to_keycache()
-    thd		Thread object
-    tables	Table list (one table only)
-
-  RETURN VALUES
-   FALSE ok
-   TRUE  error
-*/
-
-bool mysql_assign_to_keycache(THD* thd, TABLE_LIST* tables,
-			     LEX_STRING *key_cache_name)
-{
-  HA_CHECK_OPT check_opt;
-  KEY_CACHE *key_cache;
-  DBUG_ENTER("mysql_assign_to_keycache");
-
-  check_opt.init();
-  mysql_mutex_lock(&LOCK_global_system_variables);
-  if (!(key_cache= get_key_cache(key_cache_name)))
-  {
-    mysql_mutex_unlock(&LOCK_global_system_variables);
-    my_error(ER_UNKNOWN_KEY_CACHE, MYF(0), key_cache_name->str);
-    DBUG_RETURN(TRUE);
-  }
-  mysql_mutex_unlock(&LOCK_global_system_variables);
-  check_opt.key_cache= key_cache;
-  DBUG_RETURN(mysql_admin_table(thd, tables, &check_opt,
-				"assign_to_keycache", TL_READ_NO_INSERT, 0, 0,
-				0, 0, &handler::assign_to_keycache, 0));
-}
-
-
-/*
-  Preload specified indexes for a table into key cache
-
-  SYNOPSIS
-    mysql_preload_keys()
-    thd		Thread object
-    tables	Table list (one table only)
-
-  RETURN VALUES
-    FALSE ok
-    TRUE  error
-*/
-
-bool mysql_preload_keys(THD* thd, TABLE_LIST* tables)
-{
-  DBUG_ENTER("mysql_preload_keys");
-  /*
-    We cannot allow concurrent inserts. The storage engine reads
-    directly from the index file, bypassing the cache. It could read
-    outdated information if parallel inserts into cache blocks happen.
-  */
-  DBUG_RETURN(mysql_admin_table(thd, tables, 0,
-				"preload_keys", TL_READ_NO_INSERT, 0, 0, 0, 0,
-				&handler::preload_keys, 0));
-}
-
-
 /*
   Create a table identical to the specified table
 
@@ -5293,6 +4454,7 @@ bool mysql_create_like_table(THD* thd, TABLE_LIST* table, TABLE_LIST* src_table,
   HA_CREATE_INFO local_create_info;
   Alter_info local_alter_info;
   bool res= TRUE;
+  bool is_trans= FALSE;
   uint not_used;
   DBUG_ENTER("mysql_create_like_table");
 
@@ -5344,7 +4506,7 @@ bool mysql_create_like_table(THD* thd, TABLE_LIST* table, TABLE_LIST* src_table,
 
   if ((res= mysql_create_table_no_lock(thd, table->db, table->table_name,
                                        &local_create_info, &local_alter_info,
-                                       FALSE, 0)))
+                                       FALSE, 0, &is_trans)))
     goto err;
 
   /*
@@ -5352,11 +4514,9 @@ bool mysql_create_like_table(THD* thd, TABLE_LIST* table, TABLE_LIST* src_table,
     non-temporary table.
   */
   DBUG_ASSERT((create_info->options & HA_LEX_CREATE_TMP_TABLE) ||
-              local_create_info.table_existed ||
               thd->mdl_context.is_lock_owner(MDL_key::TABLE, table->db,
                                              table->table_name,
                                              MDL_EXCLUSIVE));
-
   /*
     We have to write the query before we unlock the tables.
   */
@@ -5427,7 +4587,7 @@ bool mysql_create_like_table(THD* thd, TABLE_LIST* table, TABLE_LIST* src_table,
       Case 3 and 4 does nothing under RBR
     */
   }
-  else if (write_bin_log(thd, TRUE, thd->query(), thd->query_length()))
+  else if (write_bin_log(thd, TRUE, thd->query(), thd->query_length(), is_trans))
     goto err;
 
 err:
@@ -5435,29 +4595,6 @@ err:
 }
 
 
-bool mysql_analyze_table(THD* thd, TABLE_LIST* tables, HA_CHECK_OPT* check_opt)
-{
-  thr_lock_type lock_type = TL_READ_NO_INSERT;
-
-  DBUG_ENTER("mysql_analyze_table");
-  DBUG_RETURN(mysql_admin_table(thd, tables, check_opt,
-				"analyze", lock_type, 1, 0, 0, 0,
-				&handler::ha_analyze, 0));
-}
-
-
-bool mysql_check_table(THD* thd, TABLE_LIST* tables,HA_CHECK_OPT* check_opt)
-{
-  thr_lock_type lock_type = TL_READ_NO_INSERT;
-
-  DBUG_ENTER("mysql_check_table");
-  DBUG_RETURN(mysql_admin_table(thd, tables, check_opt,
-				"check", lock_type,
-				0, 0, HA_OPEN_FOR_REPAIR, 0,
-				&handler::ha_check, &view_checksum));
-}
-
-
 /* table_list should contain just one table */
 static int
 mysql_discard_or_import_tablespace(THD *thd,
@@ -5568,7 +4705,7 @@ is_index_maintenance_unique (TABLE *table, Alter_info *alter_info)
 
 /*
   SYNOPSIS
-    compare_tables()
+    mysql_compare_tables()
       table                     The original table.
       alter_info                Alter options, fields and keys for the new
                                 table.
@@ -5608,17 +4745,16 @@ is_index_maintenance_unique (TABLE *table, Alter_info *alter_info)
     FALSE  success
 */
 
-static
 bool
-compare_tables(TABLE *table,
-               Alter_info *alter_info,
-               HA_CREATE_INFO *create_info,
-               uint order_num,
-               enum_alter_table_change_level *need_copy_table,
-               KEY **key_info_buffer,
-               uint **index_drop_buffer, uint *index_drop_count,
-               uint **index_add_buffer, uint *index_add_count,
-               uint *candidate_key_count)
+mysql_compare_tables(TABLE *table,
+                     Alter_info *alter_info,
+                     HA_CREATE_INFO *create_info,
+                     uint order_num,
+                     enum_alter_table_change_level *need_copy_table,
+                     KEY **key_info_buffer,
+                     uint **index_drop_buffer, uint *index_drop_count,
+                     uint **index_add_buffer, uint *index_add_count,
+                     uint *candidate_key_count)
 {
   Field **f_ptr, *field;
   uint changes= 0, tmp;
@@ -5634,7 +4770,7 @@ compare_tables(TABLE *table,
   */
   bool varchar= create_info->varchar;
   bool not_nullable= true;
-  DBUG_ENTER("compare_tables");
+  DBUG_ENTER("mysql_compare_tables");
 
   /*
     Create a copy of alter_info.
@@ -5645,7 +4781,7 @@ compare_tables(TABLE *table,
     mysql_prepare_create_table.  Unfortunately,
     mysql_prepare_create_table performs its transformations
     "in-place", that is, modifies the argument.  Since we would
-    like to keep compare_tables() idempotent (not altering any
+    like to keep mysql_compare_tables() idempotent (not altering any
     of the arguments) we create a copy of alter_info here and
     pass it to mysql_prepare_create_table, then use the result
     to evaluate possibility of fast ALTER TABLE, and then
@@ -6025,7 +5161,7 @@ blob_length_by_type(enum_field_types type)
   @retval FALSE  success
 */
 
-static bool
+bool
 mysql_prepare_alter_table(THD *thd, TABLE *table,
                           HA_CREATE_INFO *create_info,
                           Alter_info *alter_info)
@@ -6434,7 +5570,7 @@ err:
     Important is the fact, that this function tries to do as little work as
     possible, by finding out whether a intermediate table is needed to copy
     data into and when finishing the altering to use it as the original table.
-    For this reason the function compare_tables() is called, which decides
+    For this reason the function mysql_compare_tables() is called, which decides
     based on all kind of data how similar are the new and the original
     tables.
 
@@ -6775,7 +5911,6 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
       else
       {
         *fn_ext(new_name)=0;
-        mysql_mutex_lock(&LOCK_open);
         if (mysql_rename_table(old_db_type,db,table_name,new_db,new_alias, 0))
           error= -1;
         else if (Table_triggers_list::change_table_name(thd, db, table_name,
@@ -6785,7 +5920,6 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
                                     table_name, 0);
           error= -1;
         }
-        mysql_mutex_unlock(&LOCK_open);
       }
     }
 
@@ -6872,13 +6006,13 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
   {
     enum_alter_table_change_level need_copy_table_res;
     /* Check how much the tables differ. */
-    if (compare_tables(table, alter_info,
-                       create_info, order_num,
-                       &need_copy_table_res,
-                       &key_info_buffer,
-                       &index_drop_buffer, &index_drop_count,
-                       &index_add_buffer, &index_add_count,
-                       &candidate_key_count))
+    if (mysql_compare_tables(table, alter_info,
+                             create_info, order_num,
+                             &need_copy_table_res,
+                             &key_info_buffer,
+                             &index_drop_buffer, &index_drop_count,
+                             &index_add_buffer, &index_add_count,
+                             &candidate_key_count))
       goto err;
    
     DBUG_EXECUTE_IF("alter_table_only_metadata_change", {
@@ -7140,7 +6274,7 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
   error= mysql_create_table_no_lock(thd, new_db, tmp_name,
                                     create_info,
                                     alter_info,
-                                    1, 0);
+                                    1, 0, NULL);
   reenable_binlog(thd);
   if (error)
     goto err;
@@ -7392,7 +6526,7 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
     it should become the actual table. Later, we will recycle the old table.
     However, in case of ALTER TABLE RENAME there might be no intermediate
     table. This is when the old and new tables are compatible, according to
-    compare_table(). Then, we need one additional call to
+    mysql_compare_table(). Then, we need one additional call to
     mysql_rename_table() with flag NO_FRM_RENAME, which does nothing else but
     actual rename in the SE and the FRM is not touched. Note that, if the
     table is renamed and the SE is also changed, then an intermediate table
@@ -7404,7 +6538,6 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
     /* This type cannot happen in regular ALTER. */
     new_db_type= old_db_type= NULL;
   }
-  mysql_mutex_lock(&LOCK_open);
   if (mysql_rename_table(old_db_type, db, table_name, db, old_name,
                          FN_TO_IS_TMP))
   {
@@ -7431,8 +6564,6 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
   if (! error)
     (void) quick_rm_table(old_db_type, db, old_name, FN_IS_TMP);
 
-  mysql_mutex_unlock(&LOCK_open);
-
   if (error)
   {
     /* This shouldn't happen. But let us play it safe. */
@@ -7609,6 +6740,54 @@ err_with_mdl:
 }
 /* mysql_alter_table */
 
+
+
+/**
+  Prepare the transaction for the alter table's copy phase.
+*/
+
+bool mysql_trans_prepare_alter_copy_data(THD *thd)
+{
+  DBUG_ENTER("mysql_prepare_alter_copy_data");
+  /*
+    Turn off recovery logging since rollback of an alter table is to
+    delete the new table so there is no need to log the changes to it.
+    
+    This needs to be done before external_lock.
+  */
+  if (ha_enable_transaction(thd, FALSE))
+    DBUG_RETURN(TRUE);
+  DBUG_RETURN(FALSE);
+}
+
+
+/**
+  Commit the copy phase of the alter table.
+*/
+
+bool mysql_trans_commit_alter_copy_data(THD *thd)
+{
+  bool error= FALSE;
+  DBUG_ENTER("mysql_commit_alter_copy_data");
+
+  if (ha_enable_transaction(thd, TRUE))
+    DBUG_RETURN(TRUE);
+  
+  /*
+    Ensure that the new table is saved properly to disk before installing
+    the new .frm.
+    And that InnoDB's internal latches are released, to avoid deadlock
+    when waiting on other instances of the table before rename (Bug#54747).
+  */
+  if (trans_commit_stmt(thd))
+    error= TRUE;
+  if (trans_commit_implicit(thd))
+    error= TRUE;
+
+  DBUG_RETURN(error);
+}
+
+
 static int
 copy_data_between_tables(TABLE *from,TABLE *to,
 			 List<Create_field> &create,
@@ -7635,14 +6814,7 @@ copy_data_between_tables(TABLE *from,TABLE *to,
   ulonglong prev_insert_id;
   DBUG_ENTER("copy_data_between_tables");
 
-  /*
-    Turn off recovery logging since rollback of an alter table is to
-    delete the new table so there is no need to log the changes to it.
-    
-    This needs to be done before external_lock
-  */
-  error= ha_enable_transaction(thd, FALSE);
-  if (error)
+  if (mysql_trans_prepare_alter_copy_data(thd))
     DBUG_RETURN(-1);
   
   if (!(copy= new Copy_field[to->s->fields]))
@@ -7801,20 +6973,8 @@ copy_data_between_tables(TABLE *from,TABLE *to,
   }
   to->file->extra(HA_EXTRA_NO_IGNORE_DUP_KEY);
 
-  if (ha_enable_transaction(thd, TRUE))
-  {
+  if (mysql_trans_commit_alter_copy_data(thd))
     error= 1;
-    goto err;
-  }
-  
-  /*
-    Ensure that the new table is saved properly to disk so that we
-    can do a rename
-  */
-  if (trans_commit_stmt(thd))
-    error=1;
-  if (trans_commit_implicit(thd))
-    error=1;
 
  err:
   thd->variables.sql_mode= save_sql_mode;