Move all ddl log code to ddl_log.cc and ddl_log.h

Part of prepration for: MDEV-17567 Atomic DDL

No code changes except moving code around

10 files changed, 1276 insertions, 1219 deletions

diff --git a/libmysqld/CMakeLists.txt b/libmysqld/CMakeLists.txt
index d2f7d40c1b4..b3dd89caf24 100644
--- a/libmysqld/CMakeLists.txt
+++ b/libmysqld/CMakeLists.txt
@@ -86,6 +86,7 @@ SET(SQL_EMBEDDED_SOURCES emb_qcache.cc libmysqld.c lib_sql.cc
            ../sql/sql_show.cc ../sql/sql_state.c 
            ../sql/sql_statistics.cc ../sql/sql_string.cc
            ../sql/sql_tablespace.cc ../sql/sql_table.cc ../sql/sql_test.cc
+           ../sql/ddl_log.cc
            ../sql/sql_trigger.cc ../sql/sql_udf.cc ../sql/sql_union.cc
            ../sql/sql_update.cc ../sql/sql_view.cc ../sql/sql_profile.cc
            ../sql/gcalc_tools.cc ../sql/gcalc_slicescan.cc
diff --git a/sql/CMakeLists.txt b/sql/CMakeLists.txt
index 39307ab8183..9721082440d 100644
--- a/sql/CMakeLists.txt
+++ b/sql/CMakeLists.txt
@@ -122,10 +122,12 @@ SET (SQL_SOURCE
                group_by_handler.cc derived_handler.cc select_handler.cc
                sql_statistics.cc sql_string.cc lex_string.h
                sql_table.cc sql_test.cc sql_trigger.cc sql_udf.cc sql_union.cc
+               ddl_log.cc ddl_log.h
                sql_update.cc sql_view.cc strfunc.cc table.cc thr_malloc.cc 
                sql_time.cc tztime.cc unireg.cc item_xmlfunc.cc 
                uniques.cc
-               rpl_tblmap.cc sql_binlog.cc event_scheduler.cc event_data_objects.cc
+               rpl_tblmap.cc sql_binlog.cc event_scheduler.cc
+               event_data_objects.cc
                event_queue.cc event_db_repository.cc 
                sql_tablespace.cc events.cc ../sql-common/my_user.c 
                partition_info.cc rpl_utility.cc rpl_utility_server.cc
diff --git a/sql/ddl_log.cc b/sql/ddl_log.cc
new file mode 100644
index 00000000000..401d4bd0b93
--- /dev/null
+++ b/sql/ddl_log.cc
@@ -0,0 +1,1156 @@
+/*
+   Copyright (c) 2000, 2019, Oracle and/or its affiliates.
+   Copyright (c) 2010, 2020, MariaDB
+
+   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
+   the Free Software Foundation; version 2 of the License.
+
+   This program is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+   GNU General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program; if not, write to the Free Software
+   Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1335  USA
+*/
+
+#include "mariadb.h"
+#include "mysqld.h"
+#include "sql_class.h"                          // init_sql_alloc()
+#include "log.h"                                // sql_print_error()
+#include "ddl_log.h"
+#include "ha_partition.h"                       // PAR_EXT
+
+
+/*--------------------------------------------------------------------------
+
+   MODULE: DDL log
+   -----------------
+
+   This module is used to ensure that we can recover from crashes that occur
+   in the middle of a meta-data operation in MySQL. E.g. DROP TABLE t1, t2;
+   We need to ensure that both t1 and t2 are dropped and not only t1 and
+   also that each table drop is entirely done and not "half-baked".
+
+   To support this we create log entries for each meta-data statement in the
+   ddl log while we are executing. These entries are dropped when the
+   operation is completed.
+
+   At recovery those entries that were not completed will be executed.
+
+   There is only one ddl log in the system and it is protected by a mutex
+   and there is a global struct that contains information about its current
+   state.
+
+   History:
+   First version written in 2006 by Mikael Ronstrom
+   Second version written in 2020 by Monty
+--------------------------------------------------------------------------*/
+
+struct st_global_ddl_log
+{
+  /*
+    We need to adjust buffer size to be able to handle downgrades/upgrades
+    where IO_SIZE has changed. We'll set the buffer size such that we can
+    handle that the buffer size was upto 4 times bigger in the version
+    that wrote the DDL log.
+  */
+  char file_entry_buf[4*IO_SIZE];
+  char file_name_str[FN_REFLEN];
+  char *file_name;
+  DDL_LOG_MEMORY_ENTRY *first_free;
+  DDL_LOG_MEMORY_ENTRY *first_used;
+  uint num_entries;
+  File file_id;
+  uint name_len;
+  uint io_size;
+  bool inited;
+  bool do_release;
+  bool recovery_phase;
+  st_global_ddl_log() : inited(false), do_release(false) {}
+};
+
+st_global_ddl_log global_ddl_log;
+
+mysql_mutex_t LOCK_gdl;
+
+#define DDL_LOG_ENTRY_TYPE_POS 0
+#define DDL_LOG_ACTION_TYPE_POS 1
+#define DDL_LOG_PHASE_POS 2
+#define DDL_LOG_NEXT_ENTRY_POS 4
+#define DDL_LOG_NAME_POS 8
+
+#define DDL_LOG_NUM_ENTRY_POS 0
+#define DDL_LOG_NAME_LEN_POS 4
+#define DDL_LOG_IO_SIZE_POS 8
+
+/**
+  Read one entry from ddl log file.
+
+  @param entry_no                     Entry number to read
+
+  @return Operation status
+    @retval true   Error
+    @retval false  Success
+*/
+
+static bool read_ddl_log_file_entry(uint entry_no)
+{
+  bool error= FALSE;
+  File file_id= global_ddl_log.file_id;
+  uchar *file_entry_buf= (uchar*)global_ddl_log.file_entry_buf;
+  size_t io_size= global_ddl_log.io_size;
+  DBUG_ENTER("read_ddl_log_file_entry");
+
+  mysql_mutex_assert_owner(&LOCK_gdl);
+  if (mysql_file_pread(file_id, file_entry_buf, io_size, io_size * entry_no,
+                       MYF(MY_WME)) != io_size)
+    error= TRUE;
+  DBUG_RETURN(error);
+}
+
+
+/**
+  Write one entry to ddl log file.
+
+  @param entry_no                     Entry number to write
+
+  @return Operation status
+    @retval true   Error
+    @retval false  Success
+*/
+
+static bool write_ddl_log_file_entry(uint entry_no)
+{
+  bool error= FALSE;
+  File file_id= global_ddl_log.file_id;
+  uchar *file_entry_buf= (uchar*)global_ddl_log.file_entry_buf;
+  DBUG_ENTER("write_ddl_log_file_entry");
+
+  mysql_mutex_assert_owner(&LOCK_gdl);
+  if (mysql_file_pwrite(file_id, file_entry_buf,
+                        IO_SIZE, IO_SIZE * entry_no, MYF(MY_WME)) != IO_SIZE)
+    error= TRUE;
+  DBUG_RETURN(error);
+}
+
+
+/**
+  Sync the ddl log file.
+
+  @return Operation status
+    @retval FALSE  Success
+    @retval TRUE   Error
+*/
+
+
+static bool sync_ddl_log_file()
+{
+  DBUG_ENTER("sync_ddl_log_file");
+  DBUG_RETURN(mysql_file_sync(global_ddl_log.file_id, MYF(MY_WME)));
+}
+
+
+/**
+  Write ddl log header.
+
+  @return Operation status
+    @retval TRUE                      Error
+    @retval FALSE                     Success
+*/
+
+static bool write_ddl_log_header()
+{
+  uint16 const_var;
+  DBUG_ENTER("write_ddl_log_header");
+
+  int4store(&global_ddl_log.file_entry_buf[DDL_LOG_NUM_ENTRY_POS],
+            global_ddl_log.num_entries);
+  const_var= FN_REFLEN;
+  int4store(&global_ddl_log.file_entry_buf[DDL_LOG_NAME_LEN_POS],
+            (ulong) const_var);
+  const_var= IO_SIZE;
+  int4store(&global_ddl_log.file_entry_buf[DDL_LOG_IO_SIZE_POS],
+            (ulong) const_var);
+  if (write_ddl_log_file_entry(0UL))
+  {
+    sql_print_error("Error writing ddl log header");
+    DBUG_RETURN(TRUE);
+  }
+  DBUG_RETURN(sync_ddl_log_file());
+}
+
+
+/**
+  Create ddl log file name.
+  @param file_name                   Filename setup
+*/
+
+static inline void create_ddl_log_file_name(char *file_name)
+{
+  strxmov(file_name, mysql_data_home, "/", "ddl_log.log", NullS);
+}
+
+
+/**
+  Read header of ddl log file.
+
+  When we read the ddl log header we get information about maximum sizes
+  of names in the ddl log and we also get information about the number
+  of entries in the ddl log.
+
+  @return Last entry in ddl log (0 if no entries)
+*/
+
+static uint read_ddl_log_header()
+{
+  uchar *file_entry_buf= (uchar*)global_ddl_log.file_entry_buf;
+  char file_name[FN_REFLEN];
+  uint entry_no;
+  bool successful_open= FALSE;
+  DBUG_ENTER("read_ddl_log_header");
+
+  mysql_mutex_init(key_LOCK_gdl, &LOCK_gdl, MY_MUTEX_INIT_SLOW);
+  mysql_mutex_lock(&LOCK_gdl);
+  create_ddl_log_file_name(file_name);
+  if ((global_ddl_log.file_id= mysql_file_open(key_file_global_ddl_log,
+                                               file_name,
+                                               O_RDWR | O_BINARY, MYF(0))) >= 0)
+  {
+    if (read_ddl_log_file_entry(0UL))
+    {
+      /* Write message into error log */
+      sql_print_error("Failed to read ddl log file in recovery");
+    }
+    else
+      successful_open= TRUE;
+  }
+  if (successful_open)
+  {
+    entry_no= uint4korr(&file_entry_buf[DDL_LOG_NUM_ENTRY_POS]);
+    global_ddl_log.name_len= uint4korr(&file_entry_buf[DDL_LOG_NAME_LEN_POS]);
+    global_ddl_log.io_size= uint4korr(&file_entry_buf[DDL_LOG_IO_SIZE_POS]);
+    DBUG_ASSERT(global_ddl_log.io_size <=
+                sizeof(global_ddl_log.file_entry_buf));
+  }
+  else
+  {
+    entry_no= 0;
+  }
+  global_ddl_log.first_free= NULL;
+  global_ddl_log.first_used= NULL;
+  global_ddl_log.num_entries= 0;
+  global_ddl_log.do_release= true;
+  mysql_mutex_unlock(&LOCK_gdl);
+  DBUG_RETURN(entry_no);
+}
+
+
+/**
+  Convert from ddl_log_entry struct to file_entry_buf binary blob.
+
+  @param ddl_log_entry   filled in ddl_log_entry struct.
+*/
+
+static void set_global_from_ddl_log_entry(const DDL_LOG_ENTRY *ddl_log_entry)
+{
+  mysql_mutex_assert_owner(&LOCK_gdl);
+  global_ddl_log.file_entry_buf[DDL_LOG_ENTRY_TYPE_POS]=
+                                    (char)DDL_LOG_ENTRY_CODE;
+  global_ddl_log.file_entry_buf[DDL_LOG_ACTION_TYPE_POS]=
+                                    (char)ddl_log_entry->action_type;
+  global_ddl_log.file_entry_buf[DDL_LOG_PHASE_POS]= 0;
+  int4store(&global_ddl_log.file_entry_buf[DDL_LOG_NEXT_ENTRY_POS],
+            ddl_log_entry->next_entry);
+  DBUG_ASSERT(strlen(ddl_log_entry->name) < FN_REFLEN);
+  strmake(&global_ddl_log.file_entry_buf[DDL_LOG_NAME_POS],
+          ddl_log_entry->name, FN_REFLEN - 1);
+  if (ddl_log_entry->action_type == DDL_LOG_RENAME_ACTION ||
+      ddl_log_entry->action_type == DDL_LOG_REPLACE_ACTION ||
+      ddl_log_entry->action_type == DDL_LOG_EXCHANGE_ACTION)
+  {
+    DBUG_ASSERT(strlen(ddl_log_entry->from_name) < FN_REFLEN);
+    strmake(&global_ddl_log.file_entry_buf[DDL_LOG_NAME_POS + FN_REFLEN],
+          ddl_log_entry->from_name, FN_REFLEN - 1);
+  }
+  else
+    global_ddl_log.file_entry_buf[DDL_LOG_NAME_POS + FN_REFLEN]= 0;
+  DBUG_ASSERT(strlen(ddl_log_entry->handler_name) < FN_REFLEN);
+  strmake(&global_ddl_log.file_entry_buf[DDL_LOG_NAME_POS + (2*FN_REFLEN)],
+          ddl_log_entry->handler_name, FN_REFLEN - 1);
+  if (ddl_log_entry->action_type == DDL_LOG_EXCHANGE_ACTION)
+  {
+    DBUG_ASSERT(strlen(ddl_log_entry->tmp_name) < FN_REFLEN);
+    strmake(&global_ddl_log.file_entry_buf[DDL_LOG_NAME_POS + (3*FN_REFLEN)],
+          ddl_log_entry->tmp_name, FN_REFLEN - 1);
+  }
+  else
+    global_ddl_log.file_entry_buf[DDL_LOG_NAME_POS + (3*FN_REFLEN)]= 0;
+}
+
+
+/**
+  Convert from file_entry_buf binary blob to ddl_log_entry struct.
+
+  @param[out] ddl_log_entry   struct to fill in.
+
+  @note Strings (names) are pointing to the global_ddl_log structure,
+  so LOCK_gdl needs to be hold until they are read or copied.
+*/
+
+static void set_ddl_log_entry_from_global(DDL_LOG_ENTRY *ddl_log_entry,
+                                          const uint read_entry)
+{
+  char *file_entry_buf= (char*) global_ddl_log.file_entry_buf;
+  uint inx;
+  uchar single_char;
+
+  mysql_mutex_assert_owner(&LOCK_gdl);
+  ddl_log_entry->entry_pos= read_entry;
+  single_char= file_entry_buf[DDL_LOG_ENTRY_TYPE_POS];
+  ddl_log_entry->entry_type= (enum ddl_log_entry_code)single_char;
+  single_char= file_entry_buf[DDL_LOG_ACTION_TYPE_POS];
+  ddl_log_entry->action_type= (enum ddl_log_action_code)single_char;
+  ddl_log_entry->phase= file_entry_buf[DDL_LOG_PHASE_POS];
+  ddl_log_entry->next_entry= uint4korr(&file_entry_buf[DDL_LOG_NEXT_ENTRY_POS]);
+  ddl_log_entry->name= &file_entry_buf[DDL_LOG_NAME_POS];
+  inx= DDL_LOG_NAME_POS + global_ddl_log.name_len;
+  ddl_log_entry->from_name= &file_entry_buf[inx];
+  inx+= global_ddl_log.name_len;
+  ddl_log_entry->handler_name= &file_entry_buf[inx];
+  if (ddl_log_entry->action_type == DDL_LOG_EXCHANGE_ACTION)
+  {
+    inx+= global_ddl_log.name_len;
+    ddl_log_entry->tmp_name= &file_entry_buf[inx];
+  }
+  else
+    ddl_log_entry->tmp_name= NULL;
+}
+
+
+/**
+  Read a ddl log entry.
+
+  Read a specified entry in the ddl log.
+
+  @param read_entry               Number of entry to read
+  @param[out] entry_info          Information from entry
+
+  @return Operation status
+    @retval TRUE                     Error
+    @retval FALSE                    Success
+*/
+
+static bool read_ddl_log_entry(uint read_entry, DDL_LOG_ENTRY *ddl_log_entry)
+{
+  DBUG_ENTER("read_ddl_log_entry");
+
+  if (read_ddl_log_file_entry(read_entry))
+  {
+    DBUG_RETURN(TRUE);
+  }
+  set_ddl_log_entry_from_global(ddl_log_entry, read_entry);
+  DBUG_RETURN(FALSE);
+}
+
+
+/**
+  Initialise ddl log.
+
+  Write the header of the ddl log file and length of names. Also set
+  number of entries to zero.
+
+  @return Operation status
+    @retval TRUE                     Error
+    @retval FALSE                    Success
+*/
+
+static bool init_ddl_log()
+{
+  char file_name[FN_REFLEN];
+  DBUG_ENTER("init_ddl_log");
+
+  if (global_ddl_log.inited)
+    goto end;
+
+  global_ddl_log.io_size= IO_SIZE;
+  global_ddl_log.name_len= FN_REFLEN;
+  create_ddl_log_file_name(file_name);
+  if ((global_ddl_log.file_id= mysql_file_create(key_file_global_ddl_log,
+                                                 file_name, CREATE_MODE,
+                                                 O_RDWR | O_TRUNC | O_BINARY,
+                                                 MYF(MY_WME))) < 0)
+  {
+    /* Couldn't create ddl log file, this is serious error */
+    sql_print_error("Failed to open ddl log file");
+    DBUG_RETURN(TRUE);
+  }
+  global_ddl_log.inited= TRUE;
+  if (write_ddl_log_header())
+  {
+    (void) mysql_file_close(global_ddl_log.file_id, MYF(MY_WME));
+    global_ddl_log.inited= FALSE;
+    DBUG_RETURN(TRUE);
+  }
+
+end:
+  DBUG_RETURN(FALSE);
+}
+
+
+/**
+  Sync ddl log file.
+
+  @return Operation status
+    @retval TRUE        Error
+    @retval FALSE       Success
+*/
+
+static bool sync_ddl_log_no_lock()
+{
+  DBUG_ENTER("sync_ddl_log_no_lock");
+
+  mysql_mutex_assert_owner(&LOCK_gdl);
+  if ((!global_ddl_log.recovery_phase) &&
+      init_ddl_log())
+  {
+    DBUG_RETURN(TRUE);
+  }
+  DBUG_RETURN(sync_ddl_log_file());
+}
+
+
+/**
+  @brief Deactivate an individual entry.
+
+  @details For complex rename operations we need to deactivate individual
+  entries.
+
+  During replace operations where we start with an existing table called
+  t1 and a replacement table called t1#temp or something else and where
+  we want to delete t1 and rename t1#temp to t1 this is not possible to
+  do in a safe manner unless the ddl log is informed of the phases in
+  the change.
+
+  Delete actions are 1-phase actions that can be ignored immediately after
+  being executed.
+  Rename actions from x to y is also a 1-phase action since there is no
+  interaction with any other handlers named x and y.
+  Replace action where drop y and x -> y happens needs to be a two-phase
+  action. Thus the first phase will drop y and the second phase will
+  rename x -> y.
+
+  @param entry_no     Entry position of record to change
+
+  @return Operation status
+    @retval TRUE      Error
+    @retval FALSE     Success
+*/
+
+static bool deactivate_ddl_log_entry_no_lock(uint entry_no)
+{
+  uchar *file_entry_buf= (uchar*)global_ddl_log.file_entry_buf;
+  DBUG_ENTER("deactivate_ddl_log_entry_no_lock");
+
+  mysql_mutex_assert_owner(&LOCK_gdl);
+  if (!read_ddl_log_file_entry(entry_no))
+  {
+    if (file_entry_buf[DDL_LOG_ENTRY_TYPE_POS] == DDL_LOG_ENTRY_CODE)
+    {
+      /*
+        Log entry, if complete mark it done (IGNORE).
+        Otherwise increase the phase by one.
+      */
+      if (file_entry_buf[DDL_LOG_ACTION_TYPE_POS] == DDL_LOG_DELETE_ACTION ||
+          file_entry_buf[DDL_LOG_ACTION_TYPE_POS] == DDL_LOG_RENAME_ACTION ||
+          (file_entry_buf[DDL_LOG_ACTION_TYPE_POS] == DDL_LOG_REPLACE_ACTION &&
+           file_entry_buf[DDL_LOG_PHASE_POS] == 1) ||
+          (file_entry_buf[DDL_LOG_ACTION_TYPE_POS] == DDL_LOG_EXCHANGE_ACTION &&
+           file_entry_buf[DDL_LOG_PHASE_POS] >= EXCH_PHASE_TEMP_TO_FROM))
+        file_entry_buf[DDL_LOG_ENTRY_TYPE_POS]= DDL_IGNORE_LOG_ENTRY_CODE;
+      else if (file_entry_buf[DDL_LOG_ACTION_TYPE_POS] == DDL_LOG_REPLACE_ACTION)
+      {
+        DBUG_ASSERT(file_entry_buf[DDL_LOG_PHASE_POS] == 0);
+        file_entry_buf[DDL_LOG_PHASE_POS]= 1;
+      }
+      else if (file_entry_buf[DDL_LOG_ACTION_TYPE_POS] == DDL_LOG_EXCHANGE_ACTION)
+      {
+        DBUG_ASSERT(file_entry_buf[DDL_LOG_PHASE_POS] <=
+                                                 EXCH_PHASE_FROM_TO_NAME);
+        file_entry_buf[DDL_LOG_PHASE_POS]++;
+      }
+      else
+      {
+        DBUG_ASSERT(0);
+      }
+      if (write_ddl_log_file_entry(entry_no))
+      {
+        sql_print_error("Error in deactivating log entry. Position = %u",
+                        entry_no);
+        DBUG_RETURN(TRUE);
+      }
+    }
+  }
+  else
+  {
+    sql_print_error("Failed in reading entry before deactivating it");
+    DBUG_RETURN(TRUE);
+  }
+  DBUG_RETURN(FALSE);
+}
+
+
+/**
+  Execute one action in a ddl log entry
+
+  @param ddl_log_entry              Information in action entry to execute
+
+  @return Operation status
+    @retval TRUE                       Error
+    @retval FALSE                      Success
+*/
+
+static int execute_ddl_log_action(THD *thd, DDL_LOG_ENTRY *ddl_log_entry)
+{
+  bool frm_action= FALSE;
+  LEX_CSTRING handler_name;
+  handler *file= NULL;
+  MEM_ROOT mem_root;
+  int error= 1;
+  char to_path[FN_REFLEN];
+  char from_path[FN_REFLEN];
+  handlerton *hton;
+  DBUG_ENTER("execute_ddl_log_action");
+
+  mysql_mutex_assert_owner(&LOCK_gdl);
+  if (ddl_log_entry->entry_type == DDL_IGNORE_LOG_ENTRY_CODE)
+  {
+    DBUG_RETURN(FALSE);
+  }
+  DBUG_PRINT("ddl_log",
+             ("execute type %c next %u name '%s' from_name '%s' handler '%s'"
+              " tmp_name '%s'",
+             ddl_log_entry->action_type,
+             ddl_log_entry->next_entry,
+             ddl_log_entry->name,
+             ddl_log_entry->from_name,
+             ddl_log_entry->handler_name,
+             ddl_log_entry->tmp_name));
+  handler_name.str= (char*)ddl_log_entry->handler_name;
+  handler_name.length= strlen(ddl_log_entry->handler_name);
+  init_sql_alloc(key_memory_gdl, &mem_root, TABLE_ALLOC_BLOCK_SIZE, 0,
+                 MYF(MY_THREAD_SPECIFIC));
+  if (!strcmp(ddl_log_entry->handler_name, reg_ext))
+    frm_action= TRUE;
+  else
+  {
+    plugin_ref plugin= ha_resolve_by_name(thd, &handler_name, false);
+    if (!plugin)
+    {
+      my_error(ER_UNKNOWN_STORAGE_ENGINE, MYF(0), ddl_log_entry->handler_name);
+      goto error;
+    }
+    hton= plugin_data(plugin, handlerton*);
+    file= get_new_handler((TABLE_SHARE*)0, &mem_root, hton);
+    if (unlikely(!file))
+      goto error;
+  }
+  switch (ddl_log_entry->action_type)
+  {
+    case DDL_LOG_REPLACE_ACTION:
+    case DDL_LOG_DELETE_ACTION:
+    {
+      if (ddl_log_entry->phase == 0)
+      {
+        if (frm_action)
+        {
+          strxmov(to_path, ddl_log_entry->name, reg_ext, NullS);
+          if (unlikely((error= mysql_file_delete(key_file_frm, to_path,
+                                                 MYF(MY_WME |
+                                                     MY_IGNORE_ENOENT)))))
+            break;
+#ifdef WITH_PARTITION_STORAGE_ENGINE
+          strxmov(to_path, ddl_log_entry->name, PAR_EXT, NullS);
+          (void) mysql_file_delete(key_file_partition_ddl_log, to_path,
+                                   MYF(0));
+#endif
+        }
+        else
+        {
+          if (unlikely((error= hton->drop_table(hton, ddl_log_entry->name))))
+          {
+            if (!non_existing_table_error(error))
+              break;
+          }
+        }
+        if ((deactivate_ddl_log_entry_no_lock(ddl_log_entry->entry_pos)))
+          break;
+        (void) sync_ddl_log_no_lock();
+        error= 0;
+        if (ddl_log_entry->action_type == DDL_LOG_DELETE_ACTION)
+          break;
+      }
+      DBUG_ASSERT(ddl_log_entry->action_type == DDL_LOG_REPLACE_ACTION);
+      /*
+        Fall through and perform the rename action of the replace
+        action. We have already indicated the success of the delete
+        action in the log entry by stepping up the phase.
+      */
+    }
+    /* fall through */
+    case DDL_LOG_RENAME_ACTION:
+    {
+      error= TRUE;
+      if (frm_action)
+      {
+        strxmov(to_path, ddl_log_entry->name, reg_ext, NullS);
+        strxmov(from_path, ddl_log_entry->from_name, reg_ext, NullS);
+        if (mysql_file_rename(key_file_frm, from_path, to_path, MYF(MY_WME)))
+          break;
+#ifdef WITH_PARTITION_STORAGE_ENGINE
+        strxmov(to_path, ddl_log_entry->name, PAR_EXT, NullS);
+        strxmov(from_path, ddl_log_entry->from_name, PAR_EXT, NullS);
+        (void) mysql_file_rename(key_file_partition_ddl_log, from_path, to_path, MYF(MY_WME));
+#endif
+      }
+      else
+      {
+        if (file->ha_rename_table(ddl_log_entry->from_name,
+                                  ddl_log_entry->name))
+          break;
+      }
+      if ((deactivate_ddl_log_entry_no_lock(ddl_log_entry->entry_pos)))
+        break;
+      (void) sync_ddl_log_no_lock();
+      error= FALSE;
+      break;
+    }
+    case DDL_LOG_EXCHANGE_ACTION:
+    {
+      /* We hold LOCK_gdl, so we can alter global_ddl_log.file_entry_buf */
+      char *file_entry_buf= (char*)&global_ddl_log.file_entry_buf;
+      /* not yet implemented for frm */
+      DBUG_ASSERT(!frm_action);
+      /*
+        Using a case-switch here to revert all currently done phases,
+        since it will fall through until the first phase is undone.
+      */
+      switch (ddl_log_entry->phase) {
+        case EXCH_PHASE_TEMP_TO_FROM:
+          /* tmp_name -> from_name possibly done */
+          (void) file->ha_rename_table(ddl_log_entry->from_name,
+                                       ddl_log_entry->tmp_name);
+          /* decrease the phase and sync */
+          file_entry_buf[DDL_LOG_PHASE_POS]--;
+          if (write_ddl_log_file_entry(ddl_log_entry->entry_pos))
+            break;
+          if (sync_ddl_log_no_lock())
+            break;
+          /* fall through */
+        case EXCH_PHASE_FROM_TO_NAME:
+          /* from_name -> name possibly done */
+          (void) file->ha_rename_table(ddl_log_entry->name,
+                                       ddl_log_entry->from_name);
+          /* decrease the phase and sync */
+          file_entry_buf[DDL_LOG_PHASE_POS]--;
+          if (write_ddl_log_file_entry(ddl_log_entry->entry_pos))
+            break;
+          if (sync_ddl_log_no_lock())
+            break;
+          /* fall through */
+        case EXCH_PHASE_NAME_TO_TEMP:
+          /* name -> tmp_name possibly done */
+          (void) file->ha_rename_table(ddl_log_entry->tmp_name,
+                                       ddl_log_entry->name);
+          /* disable the entry and sync */
+          file_entry_buf[DDL_LOG_ENTRY_TYPE_POS]= DDL_IGNORE_LOG_ENTRY_CODE;
+          if (write_ddl_log_file_entry(ddl_log_entry->entry_pos))
+            break;
+          if (sync_ddl_log_no_lock())
+            break;
+          error= FALSE;
+          break;
+        default:
+          DBUG_ASSERT(0);
+          break;
+      }
+
+      break;
+    }
+    default:
+      DBUG_ASSERT(0);
+      break;
+  }
+  delete file;
+error:
+  free_root(&mem_root, MYF(0));
+  DBUG_RETURN(error);
+}
+
+
+/**
+  Get a free entry in the ddl log
+
+  @param[out] active_entry     A ddl log memory entry returned
+
+  @return Operation status
+    @retval TRUE               Error
+    @retval FALSE              Success
+*/
+
+static bool get_free_ddl_log_entry(DDL_LOG_MEMORY_ENTRY **active_entry,
+                                   bool *write_header)
+{
+  DDL_LOG_MEMORY_ENTRY *used_entry;
+  DDL_LOG_MEMORY_ENTRY *first_used= global_ddl_log.first_used;
+  DBUG_ENTER("get_free_ddl_log_entry");
+
+  if (global_ddl_log.first_free == NULL)
+  {
+    if (!(used_entry= (DDL_LOG_MEMORY_ENTRY*)my_malloc(key_memory_DDL_LOG_MEMORY_ENTRY,
+                              sizeof(DDL_LOG_MEMORY_ENTRY), MYF(MY_WME))))
+    {
+      sql_print_error("Failed to allocate memory for ddl log free list");
+      DBUG_RETURN(TRUE);
+    }
+    global_ddl_log.num_entries++;
+    used_entry->entry_pos= global_ddl_log.num_entries;
+    *write_header= TRUE;
+  }
+  else
+  {
+    used_entry= global_ddl_log.first_free;
+    global_ddl_log.first_free= used_entry->next_log_entry;
+    *write_header= FALSE;
+  }
+  /*
+    Move from free list to used list
+  */
+  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;
+
+  *active_entry= used_entry;
+  DBUG_RETURN(FALSE);
+}
+
+
+/**
+  Execute one entry in the ddl log.
+
+  Executing an entry means executing a linked list of actions.
+
+  @param first_entry           Reference to first action in entry
+
+  @return Operation status
+    @retval TRUE               Error
+    @retval FALSE              Success
+*/
+
+static bool execute_ddl_log_entry_no_lock(THD *thd, uint first_entry)
+{
+  DDL_LOG_ENTRY ddl_log_entry;
+  uint read_entry= first_entry;
+  DBUG_ENTER("execute_ddl_log_entry_no_lock");
+
+  mysql_mutex_assert_owner(&LOCK_gdl);
+  do
+  {
+    if (read_ddl_log_entry(read_entry, &ddl_log_entry))
+    {
+      /* Write to error log and continue with next log entry */
+      sql_print_error("Failed to read entry = %u from ddl log",
+                      read_entry);
+      break;
+    }
+    DBUG_ASSERT(ddl_log_entry.entry_type == DDL_LOG_ENTRY_CODE ||
+                ddl_log_entry.entry_type == DDL_IGNORE_LOG_ENTRY_CODE);
+
+    if (execute_ddl_log_action(thd, &ddl_log_entry))
+    {
+      /* Write to error log and continue with next log entry */
+      sql_print_error("Failed to execute action for entry = %u from ddl log",
+                      read_entry);
+      break;
+    }
+    read_entry= ddl_log_entry.next_entry;
+  } while (read_entry);
+  DBUG_RETURN(FALSE);
+}
+
+
+/*
+  External interface methods for the DDL log Module
+  ---------------------------------------------------
+*/
+
+/**
+  Write a ddl log entry.
+
+  A careful write of the ddl log is performed to ensure that we can
+  handle crashes occurring during CREATE and ALTER TABLE processing.
+
+  @param ddl_log_entry         Information about log entry
+  @param[out] entry_written    Entry information written into
+
+  @return Operation status
+    @retval TRUE               Error
+    @retval FALSE              Success
+*/
+
+bool write_ddl_log_entry(DDL_LOG_ENTRY *ddl_log_entry,
+                         DDL_LOG_MEMORY_ENTRY **active_entry)
+{
+  bool error, write_header;
+  DBUG_ENTER("write_ddl_log_entry");
+
+  mysql_mutex_assert_owner(&LOCK_gdl);
+  if (init_ddl_log())
+  {
+    DBUG_RETURN(TRUE);
+  }
+  set_global_from_ddl_log_entry(ddl_log_entry);
+  if (get_free_ddl_log_entry(active_entry, &write_header))
+  {
+    DBUG_RETURN(TRUE);
+  }
+  error= FALSE;
+  DBUG_PRINT("ddl_log",
+             ("write type %c next %u name '%s' from_name '%s' handler '%s'"
+              " tmp_name '%s'",
+             (char) global_ddl_log.file_entry_buf[DDL_LOG_ACTION_TYPE_POS],
+             ddl_log_entry->next_entry,
+             (char*) &global_ddl_log.file_entry_buf[DDL_LOG_NAME_POS],
+             (char*) &global_ddl_log.file_entry_buf[DDL_LOG_NAME_POS
+                                                    + FN_REFLEN],
+             (char*) &global_ddl_log.file_entry_buf[DDL_LOG_NAME_POS
+                                                    + (2*FN_REFLEN)],
+             (char*) &global_ddl_log.file_entry_buf[DDL_LOG_NAME_POS
+                                                    + (3*FN_REFLEN)]));
+  if (unlikely(write_ddl_log_file_entry((*active_entry)->entry_pos)))
+  {
+    error= TRUE;
+    sql_print_error("Failed to write entry_no = %u",
+                    (*active_entry)->entry_pos);
+  }
+  if (write_header && likely(!error))
+  {
+    (void) sync_ddl_log_no_lock();
+    if (write_ddl_log_header())
+      error= TRUE;
+  }
+  if (unlikely(error))
+    release_ddl_log_memory_entry(*active_entry);
+  DBUG_RETURN(error);
+}
+
+
+/**
+  @brief Write final entry in the ddl log.
+
+  @details This is the last write in the ddl log. The previous log entries
+  have already been written but not yet synched to disk.
+  We write a couple of log entries that describes action to perform.
+  This entries are set-up in a linked list, however only when a first
+  execute entry is put as the first entry these will be executed.
+  This routine writes this first.
+
+  @param first_entry               First entry in linked list of entries
+                                   to execute, if 0 = NULL it means that
+                                   the entry is removed and the entries
+                                   are put into the free list.
+  @param complete                  Flag indicating we are simply writing
+                                   info about that entry has been completed
+  @param[in,out] active_entry      Entry to execute, 0 = NULL if the entry
+                                   is written first time and needs to be
+                                   returned. In this case the entry written
+                                   is returned in this parameter
+
+  @return Operation status
+    @retval TRUE                   Error
+    @retval FALSE                  Success
+*/
+
+bool write_execute_ddl_log_entry(uint first_entry,
+                                 bool complete,
+                                 DDL_LOG_MEMORY_ENTRY **active_entry)
+{
+  bool write_header= FALSE;
+  char *file_entry_buf= (char*)global_ddl_log.file_entry_buf;
+  DBUG_ENTER("write_execute_ddl_log_entry");
+
+  mysql_mutex_assert_owner(&LOCK_gdl);
+  if (init_ddl_log())
+  {
+    DBUG_RETURN(TRUE);
+  }
+  if (!complete)
+  {
+    /*
+      We haven't synched the log entries yet, we synch them now before
+      writing the execute entry. If complete is true we haven't written
+      any log entries before, we are only here to write the execute
+      entry to indicate it is done.
+    */
+    (void) sync_ddl_log_no_lock();
+    file_entry_buf[DDL_LOG_ENTRY_TYPE_POS]= (char)DDL_LOG_EXECUTE_CODE;
+  }
+  else
+    file_entry_buf[DDL_LOG_ENTRY_TYPE_POS]= (char)DDL_IGNORE_LOG_ENTRY_CODE;
+  file_entry_buf[DDL_LOG_ACTION_TYPE_POS]= 0; /* Ignored for execute entries */
+  file_entry_buf[DDL_LOG_PHASE_POS]= 0;
+  int4store(&file_entry_buf[DDL_LOG_NEXT_ENTRY_POS], first_entry);
+  file_entry_buf[DDL_LOG_NAME_POS]= 0;
+  file_entry_buf[DDL_LOG_NAME_POS + FN_REFLEN]= 0;
+  file_entry_buf[DDL_LOG_NAME_POS + 2*FN_REFLEN]= 0;
+  if (!(*active_entry))
+  {
+    if (get_free_ddl_log_entry(active_entry, &write_header))
+    {
+      DBUG_RETURN(TRUE);
+    }
+    write_header= TRUE;
+  }
+  if (write_ddl_log_file_entry((*active_entry)->entry_pos))
+  {
+    sql_print_error("Error writing execute entry in ddl log");
+    release_ddl_log_memory_entry(*active_entry);
+    DBUG_RETURN(TRUE);
+  }
+  (void) sync_ddl_log_no_lock();
+  if (write_header)
+  {
+    if (write_ddl_log_header())
+    {
+      release_ddl_log_memory_entry(*active_entry);
+      DBUG_RETURN(TRUE);
+    }
+  }
+  DBUG_RETURN(FALSE);
+}
+
+
+/**
+  Deactivate an individual entry.
+
+  @details see deactivate_ddl_log_entry_no_lock.
+
+  @param entry_no     Entry position of record to change
+
+  @return Operation status
+    @retval TRUE      Error
+    @retval FALSE     Success
+*/
+
+bool deactivate_ddl_log_entry(uint entry_no)
+{
+  bool error;
+  DBUG_ENTER("deactivate_ddl_log_entry");
+
+  mysql_mutex_lock(&LOCK_gdl);
+  error= deactivate_ddl_log_entry_no_lock(entry_no);
+  mysql_mutex_unlock(&LOCK_gdl);
+  DBUG_RETURN(error);
+}
+
+
+/**
+  Sync ddl log file.
+
+  @return Operation status
+    @retval TRUE        Error
+    @retval FALSE       Success
+*/
+
+bool sync_ddl_log()
+{
+  bool error;
+  DBUG_ENTER("sync_ddl_log");
+
+  mysql_mutex_lock(&LOCK_gdl);
+  error= sync_ddl_log_no_lock();
+  mysql_mutex_unlock(&LOCK_gdl);
+
+  DBUG_RETURN(error);
+}
+
+
+/**
+  Release a log memory entry.
+  @param log_memory_entry                Log memory entry to release
+*/
+
+void release_ddl_log_memory_entry(DDL_LOG_MEMORY_ENTRY *log_entry)
+{
+  DDL_LOG_MEMORY_ENTRY *first_free= global_ddl_log.first_free;
+  DDL_LOG_MEMORY_ENTRY *next_log_entry= log_entry->next_log_entry;
+  DDL_LOG_MEMORY_ENTRY *prev_log_entry= log_entry->prev_log_entry;
+  DBUG_ENTER("release_ddl_log_memory_entry");
+
+  mysql_mutex_assert_owner(&LOCK_gdl);
+  global_ddl_log.first_free= log_entry;
+  log_entry->next_log_entry= first_free;
+
+  if (prev_log_entry)
+    prev_log_entry->next_log_entry= next_log_entry;
+  else
+    global_ddl_log.first_used= next_log_entry;
+  if (next_log_entry)
+    next_log_entry->prev_log_entry= prev_log_entry;
+  DBUG_VOID_RETURN;
+}
+
+
+/**
+  Execute one entry in the ddl log.
+
+  Executing an entry means executing a linked list of actions.
+
+  @param first_entry           Reference to first action in entry
+
+  @return Operation status
+    @retval TRUE               Error
+    @retval FALSE              Success
+*/
+
+bool execute_ddl_log_entry(THD *thd, uint first_entry)
+{
+  bool error;
+  DBUG_ENTER("execute_ddl_log_entry");
+
+  mysql_mutex_lock(&LOCK_gdl);
+  error= execute_ddl_log_entry_no_lock(thd, first_entry);
+  mysql_mutex_unlock(&LOCK_gdl);
+  DBUG_RETURN(error);
+}
+
+
+/**
+  Close the ddl log.
+*/
+
+static void close_ddl_log()
+{
+  DBUG_ENTER("close_ddl_log");
+  if (global_ddl_log.file_id >= 0)
+  {
+    (void) mysql_file_close(global_ddl_log.file_id, MYF(MY_WME));
+    global_ddl_log.file_id= (File) -1;
+  }
+  DBUG_VOID_RETURN;
+}
+
+
+/**
+  Execute the ddl log at recovery of MySQL Server.
+*/
+
+void execute_ddl_log_recovery()
+{
+  uint num_entries, i;
+  THD *thd;
+  DDL_LOG_ENTRY ddl_log_entry;
+  char file_name[FN_REFLEN];
+  static char recover_query_string[]= "INTERNAL DDL LOG RECOVER IN PROGRESS";
+  DBUG_ENTER("execute_ddl_log_recovery");
+
+  /*
+    Initialise global_ddl_log struct
+  */
+  bzero(global_ddl_log.file_entry_buf, sizeof(global_ddl_log.file_entry_buf));
+  global_ddl_log.inited= FALSE;
+  global_ddl_log.recovery_phase= TRUE;
+  global_ddl_log.io_size= IO_SIZE;
+  global_ddl_log.file_id= (File) -1;
+
+  /*
+    To be able to run this from boot, we allocate a temporary THD
+  */
+  if (!(thd=new THD(0)))
+    DBUG_VOID_RETURN;
+  thd->thread_stack= (char*) &thd;
+  thd->store_globals();
+
+  thd->set_query(recover_query_string, strlen(recover_query_string));
+
+  /* this also initialize LOCK_gdl */
+  num_entries= read_ddl_log_header();
+  mysql_mutex_lock(&LOCK_gdl);
+  for (i= 1; i < num_entries + 1; i++)
+  {
+    if (read_ddl_log_entry(i, &ddl_log_entry))
+    {
+      sql_print_error("Failed to read entry no = %u from ddl log",
+                       i);
+      continue;
+    }
+    if (ddl_log_entry.entry_type == DDL_LOG_EXECUTE_CODE)
+    {
+      if (execute_ddl_log_entry_no_lock(thd, ddl_log_entry.next_entry))
+      {
+        /* Real unpleasant scenario but we continue anyways.  */
+        continue;
+      }
+    }
+  }
+  close_ddl_log();
+  create_ddl_log_file_name(file_name);
+  (void) mysql_file_delete(key_file_global_ddl_log, file_name, MYF(0));
+  global_ddl_log.recovery_phase= FALSE;
+  mysql_mutex_unlock(&LOCK_gdl);
+  thd->reset_query();
+  delete thd;
+  DBUG_VOID_RETURN;
+}
+
+
+/**
+  Release all memory allocated to the ddl log.
+*/
+
+void release_ddl_log()
+{
+  DDL_LOG_MEMORY_ENTRY *free_list;
+  DDL_LOG_MEMORY_ENTRY *used_list;
+  DBUG_ENTER("release_ddl_log");
+
+  if (!global_ddl_log.do_release)
+    DBUG_VOID_RETURN;
+
+  mysql_mutex_lock(&LOCK_gdl);
+  free_list= global_ddl_log.first_free;
+  used_list= global_ddl_log.first_used;
+  while (used_list)
+  {
+    DDL_LOG_MEMORY_ENTRY *tmp= used_list->next_log_entry;
+    my_free(used_list);
+    used_list= tmp;
+  }
+  while (free_list)
+  {
+    DDL_LOG_MEMORY_ENTRY *tmp= free_list->next_log_entry;
+    my_free(free_list);
+    free_list= tmp;
+  }
+  close_ddl_log();
+  global_ddl_log.inited= 0;
+  mysql_mutex_unlock(&LOCK_gdl);
+  mysql_mutex_destroy(&LOCK_gdl);
+  global_ddl_log.do_release= false;
+  DBUG_VOID_RETURN;
+}
+
+
+/*
+---------------------------------------------------------------------------
+
+  END MODULE DDL log
+  --------------------
+
+---------------------------------------------------------------------------
+*/
diff --git a/sql/ddl_log.h b/sql/ddl_log.h
new file mode 100644
index 00000000000..e2ed648b86c
--- /dev/null
+++ b/sql/ddl_log.h
@@ -0,0 +1,111 @@
+/*
+   Copyright (c) 2000, 2019, Oracle and/or its affiliates.
+   Copyright (c) 2010, 2020, MariaDB
+
+   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
+   the Free Software Foundation; version 2 of the License.
+
+   This program is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+   GNU General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program; if not, write to the Free Software
+   Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1335  USA
+*/
+
+/* External interfaces to ddl log functions */
+
+#ifndef DDL_LOG_INCLUDED
+#define DDL_LOG_INCLUDED
+
+enum ddl_log_entry_code
+{
+  /*
+    DDL_LOG_EXECUTE_CODE:
+      This is a code that indicates that this is a log entry to
+      be executed, from this entry a linked list of log entries
+      can be found and executed.
+    DDL_LOG_ENTRY_CODE:
+      An entry to be executed in a linked list from an execute log
+      entry.
+    DDL_IGNORE_LOG_ENTRY_CODE:
+      An entry that is to be ignored
+  */
+  DDL_LOG_EXECUTE_CODE = 'e',
+  DDL_LOG_ENTRY_CODE = 'l',
+  DDL_IGNORE_LOG_ENTRY_CODE = 'i'
+};
+
+enum ddl_log_action_code
+{
+  /*
+    The type of action that a DDL_LOG_ENTRY_CODE entry is to
+    perform.
+    DDL_LOG_DELETE_ACTION:
+      Delete an entity
+    DDL_LOG_RENAME_ACTION:
+      Rename an entity
+    DDL_LOG_REPLACE_ACTION:
+      Rename an entity after removing the previous entry with the
+      new name, that is replace this entry.
+    DDL_LOG_EXCHANGE_ACTION:
+      Exchange two entities by renaming them a -> tmp, b -> a, tmp -> b.
+  */
+  DDL_LOG_DELETE_ACTION = 'd',
+  DDL_LOG_RENAME_ACTION = 'r',
+  DDL_LOG_REPLACE_ACTION = 's',
+  DDL_LOG_EXCHANGE_ACTION = 'e'
+};
+
+enum enum_ddl_log_exchange_phase {
+  EXCH_PHASE_NAME_TO_TEMP= 0,
+  EXCH_PHASE_FROM_TO_NAME= 1,
+  EXCH_PHASE_TEMP_TO_FROM= 2
+};
+
+
+typedef struct st_ddl_log_entry
+{
+  const char *name;
+  const char *from_name;
+  const char *handler_name;
+  const char *tmp_name;
+  uint next_entry;
+  uint entry_pos;
+  enum ddl_log_entry_code entry_type;
+  enum ddl_log_action_code action_type;
+  /*
+    Most actions have only one phase. REPLACE does however have two
+    phases. The first phase removes the file with the new name if
+    there was one there before and the second phase renames the
+    old name to the new name.
+  */
+  char phase;
+} DDL_LOG_ENTRY;
+
+typedef struct st_ddl_log_memory_entry
+{
+  uint entry_pos;
+  struct st_ddl_log_memory_entry *next_log_entry;
+  struct st_ddl_log_memory_entry *prev_log_entry;
+  struct st_ddl_log_memory_entry *next_active_log_entry;
+} DDL_LOG_MEMORY_ENTRY;
+
+
+bool write_ddl_log_entry(DDL_LOG_ENTRY *ddl_log_entry,
+                           DDL_LOG_MEMORY_ENTRY **active_entry);
+bool write_execute_ddl_log_entry(uint first_entry,
+                                   bool complete,
+                                   DDL_LOG_MEMORY_ENTRY **active_entry);
+bool deactivate_ddl_log_entry(uint entry_no);
+void release_ddl_log_memory_entry(DDL_LOG_MEMORY_ENTRY *log_entry);
+bool sync_ddl_log();
+void release_ddl_log();
+void execute_ddl_log_recovery();
+bool execute_ddl_log_entry(THD *thd, uint first_entry);
+
+extern mysql_mutex_t LOCK_gdl;
+#endif /* DDL_LOG_INCLUDED */
diff --git a/sql/ha_partition.cc b/sql/ha_partition.cc
index 2d7d26b44a3..b44af22ec83 100644
--- a/sql/ha_partition.cc
+++ b/sql/ha_partition.cc
@@ -59,6 +59,7 @@
 #include "sql_show.h"                        // append_identifier
 #include "sql_admin.h"                       // SQL_ADMIN_MSG_TEXT_SIZE
 #include "sql_select.h"
+#include "ddl_log.h"
 
 #include "debug_sync.h"
 
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index ce480733e6e..aee9452f528 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -51,6 +51,7 @@
 #include "sql_manager.h"  // stop_handle_manager, start_handle_manager
 #include "sql_expression_cache.h" // subquery_cache_miss, subquery_cache_hit
 #include "sys_vars_shared.h"
+#include "ddl_log.h"
 
 #include <m_ctype.h>
 #include <my_dir.h>
diff --git a/sql/sql_partition.cc b/sql/sql_partition.cc
index d8cc4fdd716..789a19eab26 100644
--- a/sql/sql_partition.cc
+++ b/sql/sql_partition.cc
@@ -68,6 +68,7 @@
 #include "sql_alter.h"                  // Alter_table_ctx
 #include "sql_select.h"
 #include "sql_tablespace.h"             // check_tablespace_name
+#include "ddl_log.h"
 #include "tztime.h"                     // my_tz_OFFSET0
 
 #include <algorithm>
diff --git a/sql/sql_partition_admin.cc b/sql/sql_partition_admin.cc
index 195d2e52c56..48ddac257a4 100644
--- a/sql/sql_partition_admin.cc
+++ b/sql/sql_partition_admin.cc
@@ -32,6 +32,7 @@
 #include "ha_partition.h"                   // ha_partition
 #endif
 #include "sql_base.h"                       // open_and_lock_tables
+#include "ddl_log.h"
 
 #ifndef WITH_PARTITION_STORAGE_ENGINE
 
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index c1153ed9a7a..143e98b5c1e 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -56,6 +56,7 @@
 #include "sql_sequence.h"
 #include "tztime.h"
 #include "sql_insert.h"                        // binlog_drop_table
+#include "ddl_log.h"
 #include <algorithm>
 
 #ifdef __WIN__
@@ -615,1138 +616,6 @@ uint build_tmptable_filename(THD* thd, char *buff, size_t bufflen)
   DBUG_RETURN((uint)length);
 }
 
-/*
---------------------------------------------------------------------------
-
-   MODULE: DDL log
-   -----------------
-
-   This module is used to ensure that we can recover from crashes that occur
-   in the middle of a meta-data operation in MySQL. E.g. DROP TABLE t1, t2;
-   We need to ensure that both t1 and t2 are dropped and not only t1 and
-   also that each table drop is entirely done and not "half-baked".
-
-   To support this we create log entries for each meta-data statement in the
-   ddl log while we are executing. These entries are dropped when the
-   operation is completed.
-
-   At recovery those entries that were not completed will be executed.
-
-   There is only one ddl log in the system and it is protected by a mutex
-   and there is a global struct that contains information about its current
-   state.
-
-   History:
-   First version written in 2006 by Mikael Ronstrom
---------------------------------------------------------------------------
-*/
-
-struct st_global_ddl_log
-{
-  /*
-    We need to adjust buffer size to be able to handle downgrades/upgrades
-    where IO_SIZE has changed. We'll set the buffer size such that we can
-    handle that the buffer size was upto 4 times bigger in the version
-    that wrote the DDL log.
-  */
-  char file_entry_buf[4*IO_SIZE];
-  char file_name_str[FN_REFLEN];
-  char *file_name;
-  DDL_LOG_MEMORY_ENTRY *first_free;
-  DDL_LOG_MEMORY_ENTRY *first_used;
-  uint num_entries;
-  File file_id;
-  uint name_len;
-  uint io_size;
-  bool inited;
-  bool do_release;
-  bool recovery_phase;
-  st_global_ddl_log() : inited(false), do_release(false) {}
-};
-
-st_global_ddl_log global_ddl_log;
-
-mysql_mutex_t LOCK_gdl;
-
-#define DDL_LOG_ENTRY_TYPE_POS 0
-#define DDL_LOG_ACTION_TYPE_POS 1
-#define DDL_LOG_PHASE_POS 2
-#define DDL_LOG_NEXT_ENTRY_POS 4
-#define DDL_LOG_NAME_POS 8
-
-#define DDL_LOG_NUM_ENTRY_POS 0
-#define DDL_LOG_NAME_LEN_POS 4
-#define DDL_LOG_IO_SIZE_POS 8
-
-/**
-  Read one entry from ddl log file.
-
-  @param entry_no                     Entry number to read
-
-  @return Operation status
-    @retval true   Error
-    @retval false  Success
-*/
-
-static bool read_ddl_log_file_entry(uint entry_no)
-{
-  bool error= FALSE;
-  File file_id= global_ddl_log.file_id;
-  uchar *file_entry_buf= (uchar*)global_ddl_log.file_entry_buf;
-  size_t io_size= global_ddl_log.io_size;
-  DBUG_ENTER("read_ddl_log_file_entry");
-
-  mysql_mutex_assert_owner(&LOCK_gdl);
-  if (mysql_file_pread(file_id, file_entry_buf, io_size, io_size * entry_no,
-                       MYF(MY_WME)) != io_size)
-    error= TRUE;
-  DBUG_RETURN(error);
-}
-
-
-/**
-  Write one entry to ddl log file.
-
-  @param entry_no                     Entry number to write
-
-  @return Operation status
-    @retval true   Error
-    @retval false  Success
-*/
-
-static bool write_ddl_log_file_entry(uint entry_no)
-{
-  bool error= FALSE;
-  File file_id= global_ddl_log.file_id;
-  uchar *file_entry_buf= (uchar*)global_ddl_log.file_entry_buf;
-  DBUG_ENTER("write_ddl_log_file_entry");
-
-  mysql_mutex_assert_owner(&LOCK_gdl);
-  if (mysql_file_pwrite(file_id, file_entry_buf,
-                        IO_SIZE, IO_SIZE * entry_no, MYF(MY_WME)) != IO_SIZE)
-    error= TRUE;
-  DBUG_RETURN(error);
-}
-
-
-/**
-  Sync the ddl log file.
-
-  @return Operation status
-    @retval FALSE  Success
-    @retval TRUE   Error
-*/
-
-
-static bool sync_ddl_log_file()
-{
-  DBUG_ENTER("sync_ddl_log_file");
-  DBUG_RETURN(mysql_file_sync(global_ddl_log.file_id, MYF(MY_WME)));
-}
-
-
-/**
-  Write ddl log header.
-
-  @return Operation status
-    @retval TRUE                      Error
-    @retval FALSE                     Success
-*/
-
-static bool write_ddl_log_header()
-{
-  uint16 const_var;
-  DBUG_ENTER("write_ddl_log_header");
-
-  int4store(&global_ddl_log.file_entry_buf[DDL_LOG_NUM_ENTRY_POS],
-            global_ddl_log.num_entries);
-  const_var= FN_REFLEN;
-  int4store(&global_ddl_log.file_entry_buf[DDL_LOG_NAME_LEN_POS],
-            (ulong) const_var);
-  const_var= IO_SIZE;
-  int4store(&global_ddl_log.file_entry_buf[DDL_LOG_IO_SIZE_POS],
-            (ulong) const_var);
-  if (write_ddl_log_file_entry(0UL))
-  {
-    sql_print_error("Error writing ddl log header");
-    DBUG_RETURN(TRUE);
-  }
-  DBUG_RETURN(sync_ddl_log_file());
-}
-
-
-/**
-  Create ddl log file name.
-  @param file_name                   Filename setup
-*/
-
-static inline void create_ddl_log_file_name(char *file_name)
-{
-  strxmov(file_name, mysql_data_home, "/", "ddl_log.log", NullS);
-}
-
-
-/**
-  Read header of ddl log file.
-
-  When we read the ddl log header we get information about maximum sizes
-  of names in the ddl log and we also get information about the number
-  of entries in the ddl log.
-
-  @return Last entry in ddl log (0 if no entries)
-*/
-
-static uint read_ddl_log_header()
-{
-  uchar *file_entry_buf= (uchar*)global_ddl_log.file_entry_buf;
-  char file_name[FN_REFLEN];
-  uint entry_no;
-  bool successful_open= FALSE;
-  DBUG_ENTER("read_ddl_log_header");
-
-  mysql_mutex_init(key_LOCK_gdl, &LOCK_gdl, MY_MUTEX_INIT_SLOW);
-  mysql_mutex_lock(&LOCK_gdl);
-  create_ddl_log_file_name(file_name);
-  if ((global_ddl_log.file_id= mysql_file_open(key_file_global_ddl_log,
-                                               file_name,
-                                               O_RDWR | O_BINARY, MYF(0))) >= 0)
-  {
-    if (read_ddl_log_file_entry(0UL))
-    {
-      /* Write message into error log */
-      sql_print_error("Failed to read ddl log file in recovery");
-    }
-    else
-      successful_open= TRUE;
-  }
-  if (successful_open)
-  {
-    entry_no= uint4korr(&file_entry_buf[DDL_LOG_NUM_ENTRY_POS]);
-    global_ddl_log.name_len= uint4korr(&file_entry_buf[DDL_LOG_NAME_LEN_POS]);
-    global_ddl_log.io_size= uint4korr(&file_entry_buf[DDL_LOG_IO_SIZE_POS]);
-    DBUG_ASSERT(global_ddl_log.io_size <=
-                sizeof(global_ddl_log.file_entry_buf));
-  }
-  else
-  {
-    entry_no= 0;
-  }
-  global_ddl_log.first_free= NULL;
-  global_ddl_log.first_used= NULL;
-  global_ddl_log.num_entries= 0;
-  global_ddl_log.do_release= true;
-  mysql_mutex_unlock(&LOCK_gdl);
-  DBUG_RETURN(entry_no);
-}
-
-
-/**
-  Convert from ddl_log_entry struct to file_entry_buf binary blob.
-
-  @param ddl_log_entry   filled in ddl_log_entry struct.
-*/
-
-static void set_global_from_ddl_log_entry(const DDL_LOG_ENTRY *ddl_log_entry)
-{
-  mysql_mutex_assert_owner(&LOCK_gdl);
-  global_ddl_log.file_entry_buf[DDL_LOG_ENTRY_TYPE_POS]=
-                                    (char)DDL_LOG_ENTRY_CODE;
-  global_ddl_log.file_entry_buf[DDL_LOG_ACTION_TYPE_POS]=
-                                    (char)ddl_log_entry->action_type;
-  global_ddl_log.file_entry_buf[DDL_LOG_PHASE_POS]= 0;
-  int4store(&global_ddl_log.file_entry_buf[DDL_LOG_NEXT_ENTRY_POS],
-            ddl_log_entry->next_entry);
-  DBUG_ASSERT(strlen(ddl_log_entry->name) < FN_REFLEN);
-  strmake(&global_ddl_log.file_entry_buf[DDL_LOG_NAME_POS],
-          ddl_log_entry->name, FN_REFLEN - 1);
-  if (ddl_log_entry->action_type == DDL_LOG_RENAME_ACTION ||
-      ddl_log_entry->action_type == DDL_LOG_REPLACE_ACTION ||
-      ddl_log_entry->action_type == DDL_LOG_EXCHANGE_ACTION)
-  {
-    DBUG_ASSERT(strlen(ddl_log_entry->from_name) < FN_REFLEN);
-    strmake(&global_ddl_log.file_entry_buf[DDL_LOG_NAME_POS + FN_REFLEN],
-          ddl_log_entry->from_name, FN_REFLEN - 1);
-  }
-  else
-    global_ddl_log.file_entry_buf[DDL_LOG_NAME_POS + FN_REFLEN]= 0;
-  DBUG_ASSERT(strlen(ddl_log_entry->handler_name) < FN_REFLEN);
-  strmake(&global_ddl_log.file_entry_buf[DDL_LOG_NAME_POS + (2*FN_REFLEN)],
-          ddl_log_entry->handler_name, FN_REFLEN - 1);
-  if (ddl_log_entry->action_type == DDL_LOG_EXCHANGE_ACTION)
-  {
-    DBUG_ASSERT(strlen(ddl_log_entry->tmp_name) < FN_REFLEN);
-    strmake(&global_ddl_log.file_entry_buf[DDL_LOG_NAME_POS + (3*FN_REFLEN)],
-          ddl_log_entry->tmp_name, FN_REFLEN - 1);
-  }
-  else
-    global_ddl_log.file_entry_buf[DDL_LOG_NAME_POS + (3*FN_REFLEN)]= 0;
-}
-
-
-/**
-  Convert from file_entry_buf binary blob to ddl_log_entry struct.
-
-  @param[out] ddl_log_entry   struct to fill in.
-
-  @note Strings (names) are pointing to the global_ddl_log structure,
-  so LOCK_gdl needs to be hold until they are read or copied.
-*/
-
-static void set_ddl_log_entry_from_global(DDL_LOG_ENTRY *ddl_log_entry,
-                                          const uint read_entry)
-{
-  char *file_entry_buf= (char*) global_ddl_log.file_entry_buf;
-  uint inx;
-  uchar single_char;
-
-  mysql_mutex_assert_owner(&LOCK_gdl);
-  ddl_log_entry->entry_pos= read_entry;
-  single_char= file_entry_buf[DDL_LOG_ENTRY_TYPE_POS];
-  ddl_log_entry->entry_type= (enum ddl_log_entry_code)single_char;
-  single_char= file_entry_buf[DDL_LOG_ACTION_TYPE_POS];
-  ddl_log_entry->action_type= (enum ddl_log_action_code)single_char;
-  ddl_log_entry->phase= file_entry_buf[DDL_LOG_PHASE_POS];
-  ddl_log_entry->next_entry= uint4korr(&file_entry_buf[DDL_LOG_NEXT_ENTRY_POS]);
-  ddl_log_entry->name= &file_entry_buf[DDL_LOG_NAME_POS];
-  inx= DDL_LOG_NAME_POS + global_ddl_log.name_len;
-  ddl_log_entry->from_name= &file_entry_buf[inx];
-  inx+= global_ddl_log.name_len;
-  ddl_log_entry->handler_name= &file_entry_buf[inx];
-  if (ddl_log_entry->action_type == DDL_LOG_EXCHANGE_ACTION)
-  {
-    inx+= global_ddl_log.name_len;
-    ddl_log_entry->tmp_name= &file_entry_buf[inx];
-  }
-  else
-    ddl_log_entry->tmp_name= NULL;
-}
-
-
-/**
-  Read a ddl log entry.
-
-  Read a specified entry in the ddl log.
-
-  @param read_entry               Number of entry to read
-  @param[out] entry_info          Information from entry
-
-  @return Operation status
-    @retval TRUE                     Error
-    @retval FALSE                    Success
-*/
-
-static bool read_ddl_log_entry(uint read_entry, DDL_LOG_ENTRY *ddl_log_entry)
-{
-  DBUG_ENTER("read_ddl_log_entry");
-
-  if (read_ddl_log_file_entry(read_entry))
-  {
-    DBUG_RETURN(TRUE);
-  }
-  set_ddl_log_entry_from_global(ddl_log_entry, read_entry);
-  DBUG_RETURN(FALSE);
-}
-
-
-/**
-  Initialise ddl log.
-
-  Write the header of the ddl log file and length of names. Also set
-  number of entries to zero.
-
-  @return Operation status
-    @retval TRUE                     Error
-    @retval FALSE                    Success
-*/
-
-static bool init_ddl_log()
-{
-  char file_name[FN_REFLEN];
-  DBUG_ENTER("init_ddl_log");
-
-  if (global_ddl_log.inited)
-    goto end;
-
-  global_ddl_log.io_size= IO_SIZE;
-  global_ddl_log.name_len= FN_REFLEN;
-  create_ddl_log_file_name(file_name);
-  if ((global_ddl_log.file_id= mysql_file_create(key_file_global_ddl_log,
-                                                 file_name, CREATE_MODE,
-                                                 O_RDWR | O_TRUNC | O_BINARY,
-                                                 MYF(MY_WME))) < 0)
-  {
-    /* Couldn't create ddl log file, this is serious error */
-    sql_print_error("Failed to open ddl log file");
-    DBUG_RETURN(TRUE);
-  }
-  global_ddl_log.inited= TRUE;
-  if (write_ddl_log_header())
-  {
-    (void) mysql_file_close(global_ddl_log.file_id, MYF(MY_WME));
-    global_ddl_log.inited= FALSE;
-    DBUG_RETURN(TRUE);
-  }
-
-end:
-  DBUG_RETURN(FALSE);
-}
-
-
-/**
-  Sync ddl log file.
-
-  @return Operation status
-    @retval TRUE        Error
-    @retval FALSE       Success
-*/
-
-static bool sync_ddl_log_no_lock()
-{
-  DBUG_ENTER("sync_ddl_log_no_lock");
-
-  mysql_mutex_assert_owner(&LOCK_gdl);
-  if ((!global_ddl_log.recovery_phase) &&
-      init_ddl_log())
-  {
-    DBUG_RETURN(TRUE);
-  }
-  DBUG_RETURN(sync_ddl_log_file());
-}
-
-
-/**
-  @brief Deactivate an individual entry.
-
-  @details For complex rename operations we need to deactivate individual
-  entries.
-
-  During replace operations where we start with an existing table called
-  t1 and a replacement table called t1#temp or something else and where
-  we want to delete t1 and rename t1#temp to t1 this is not possible to
-  do in a safe manner unless the ddl log is informed of the phases in
-  the change.
-
-  Delete actions are 1-phase actions that can be ignored immediately after
-  being executed.
-  Rename actions from x to y is also a 1-phase action since there is no
-  interaction with any other handlers named x and y.
-  Replace action where drop y and x -> y happens needs to be a two-phase
-  action. Thus the first phase will drop y and the second phase will
-  rename x -> y.
-
-  @param entry_no     Entry position of record to change
-
-  @return Operation status
-    @retval TRUE      Error
-    @retval FALSE     Success
-*/
-
-static bool deactivate_ddl_log_entry_no_lock(uint entry_no)
-{
-  uchar *file_entry_buf= (uchar*)global_ddl_log.file_entry_buf;
-  DBUG_ENTER("deactivate_ddl_log_entry_no_lock");
-
-  mysql_mutex_assert_owner(&LOCK_gdl);
-  if (!read_ddl_log_file_entry(entry_no))
-  {
-    if (file_entry_buf[DDL_LOG_ENTRY_TYPE_POS] == DDL_LOG_ENTRY_CODE)
-    {
-      /*
-        Log entry, if complete mark it done (IGNORE).
-        Otherwise increase the phase by one.
-      */
-      if (file_entry_buf[DDL_LOG_ACTION_TYPE_POS] == DDL_LOG_DELETE_ACTION ||
-          file_entry_buf[DDL_LOG_ACTION_TYPE_POS] == DDL_LOG_RENAME_ACTION ||
-          (file_entry_buf[DDL_LOG_ACTION_TYPE_POS] == DDL_LOG_REPLACE_ACTION &&
-           file_entry_buf[DDL_LOG_PHASE_POS] == 1) ||
-          (file_entry_buf[DDL_LOG_ACTION_TYPE_POS] == DDL_LOG_EXCHANGE_ACTION &&
-           file_entry_buf[DDL_LOG_PHASE_POS] >= EXCH_PHASE_TEMP_TO_FROM))
-        file_entry_buf[DDL_LOG_ENTRY_TYPE_POS]= DDL_IGNORE_LOG_ENTRY_CODE;
-      else if (file_entry_buf[DDL_LOG_ACTION_TYPE_POS] == DDL_LOG_REPLACE_ACTION)
-      {
-        DBUG_ASSERT(file_entry_buf[DDL_LOG_PHASE_POS] == 0);
-        file_entry_buf[DDL_LOG_PHASE_POS]= 1;
-      }
-      else if (file_entry_buf[DDL_LOG_ACTION_TYPE_POS] == DDL_LOG_EXCHANGE_ACTION)
-      {
-        DBUG_ASSERT(file_entry_buf[DDL_LOG_PHASE_POS] <=
-                                                 EXCH_PHASE_FROM_TO_NAME);
-        file_entry_buf[DDL_LOG_PHASE_POS]++;
-      }
-      else
-      {
-        DBUG_ASSERT(0);
-      }
-      if (write_ddl_log_file_entry(entry_no))
-      {
-        sql_print_error("Error in deactivating log entry. Position = %u",
-                        entry_no);
-        DBUG_RETURN(TRUE);
-      }
-    }
-  }
-  else
-  {
-    sql_print_error("Failed in reading entry before deactivating it");
-    DBUG_RETURN(TRUE);
-  }
-  DBUG_RETURN(FALSE);
-}
-
-
-/**
-  Execute one action in a ddl log entry
-
-  @param ddl_log_entry              Information in action entry to execute
-
-  @return Operation status
-    @retval TRUE                       Error
-    @retval FALSE                      Success
-*/
-
-static int execute_ddl_log_action(THD *thd, DDL_LOG_ENTRY *ddl_log_entry)
-{
-  bool frm_action= FALSE;
-  LEX_CSTRING handler_name;
-  handler *file= NULL;
-  MEM_ROOT mem_root;
-  int error= 1;
-  char to_path[FN_REFLEN];
-  char from_path[FN_REFLEN];
-  handlerton *hton;
-  DBUG_ENTER("execute_ddl_log_action");
-
-  mysql_mutex_assert_owner(&LOCK_gdl);
-  if (ddl_log_entry->entry_type == DDL_IGNORE_LOG_ENTRY_CODE)
-  {
-    DBUG_RETURN(FALSE);
-  }
-  DBUG_PRINT("ddl_log",
-             ("execute type %c next %u name '%s' from_name '%s' handler '%s'"
-              " tmp_name '%s'",
-             ddl_log_entry->action_type,
-             ddl_log_entry->next_entry,
-             ddl_log_entry->name,
-             ddl_log_entry->from_name,
-             ddl_log_entry->handler_name,
-             ddl_log_entry->tmp_name));
-  handler_name.str= (char*)ddl_log_entry->handler_name;
-  handler_name.length= strlen(ddl_log_entry->handler_name);
-  init_sql_alloc(key_memory_gdl, &mem_root, TABLE_ALLOC_BLOCK_SIZE, 0,
-                 MYF(MY_THREAD_SPECIFIC));
-  if (!strcmp(ddl_log_entry->handler_name, reg_ext))
-    frm_action= TRUE;
-  else
-  {
-    plugin_ref plugin= ha_resolve_by_name(thd, &handler_name, false);
-    if (!plugin)
-    {
-      my_error(ER_UNKNOWN_STORAGE_ENGINE, MYF(0), ddl_log_entry->handler_name);
-      goto error;
-    }
-    hton= plugin_data(plugin, handlerton*);
-    file= get_new_handler((TABLE_SHARE*)0, &mem_root, hton);
-    if (unlikely(!file))
-      goto error;
-  }
-  switch (ddl_log_entry->action_type)
-  {
-    case DDL_LOG_REPLACE_ACTION:
-    case DDL_LOG_DELETE_ACTION:
-    {
-      if (ddl_log_entry->phase == 0)
-      {
-        if (frm_action)
-        {
-          strxmov(to_path, ddl_log_entry->name, reg_ext, NullS);
-          if (unlikely((error= mysql_file_delete(key_file_frm, to_path,
-                                                 MYF(MY_WME |
-                                                     MY_IGNORE_ENOENT)))))
-            break;
-#ifdef WITH_PARTITION_STORAGE_ENGINE
-          strxmov(to_path, ddl_log_entry->name, PAR_EXT, NullS);
-          (void) mysql_file_delete(key_file_partition_ddl_log, to_path,
-                                   MYF(0));
-#endif
-        }
-        else
-        {
-          if (unlikely((error= hton->drop_table(hton, ddl_log_entry->name))))
-          {
-            if (!non_existing_table_error(error))
-              break;
-          }
-        }
-        if ((deactivate_ddl_log_entry_no_lock(ddl_log_entry->entry_pos)))
-          break;
-        (void) sync_ddl_log_no_lock();
-        error= 0;
-        if (ddl_log_entry->action_type == DDL_LOG_DELETE_ACTION)
-          break;
-      }
-      DBUG_ASSERT(ddl_log_entry->action_type == DDL_LOG_REPLACE_ACTION);
-      /*
-        Fall through and perform the rename action of the replace
-        action. We have already indicated the success of the delete
-        action in the log entry by stepping up the phase.
-      */
-    }
-    /* fall through */
-    case DDL_LOG_RENAME_ACTION:
-    {
-      error= TRUE;
-      if (frm_action)
-      {
-        strxmov(to_path, ddl_log_entry->name, reg_ext, NullS);
-        strxmov(from_path, ddl_log_entry->from_name, reg_ext, NullS);
-        if (mysql_file_rename(key_file_frm, from_path, to_path, MYF(MY_WME)))
-          break;
-#ifdef WITH_PARTITION_STORAGE_ENGINE
-        strxmov(to_path, ddl_log_entry->name, PAR_EXT, NullS);
-        strxmov(from_path, ddl_log_entry->from_name, PAR_EXT, NullS);
-        (void) mysql_file_rename(key_file_partition_ddl_log, from_path, to_path, MYF(MY_WME));
-#endif
-      }
-      else
-      {
-        if (file->ha_rename_table(ddl_log_entry->from_name,
-                                  ddl_log_entry->name))
-          break;
-      }
-      if ((deactivate_ddl_log_entry_no_lock(ddl_log_entry->entry_pos)))
-        break;
-      (void) sync_ddl_log_no_lock();
-      error= FALSE;
-      break;
-    }
-    case DDL_LOG_EXCHANGE_ACTION:
-    {
-      /* We hold LOCK_gdl, so we can alter global_ddl_log.file_entry_buf */
-      char *file_entry_buf= (char*)&global_ddl_log.file_entry_buf;
-      /* not yet implemented for frm */
-      DBUG_ASSERT(!frm_action);
-      /*
-        Using a case-switch here to revert all currently done phases,
-        since it will fall through until the first phase is undone.
-      */
-      switch (ddl_log_entry->phase) {
-        case EXCH_PHASE_TEMP_TO_FROM:
-          /* tmp_name -> from_name possibly done */
-          (void) file->ha_rename_table(ddl_log_entry->from_name,
-                                       ddl_log_entry->tmp_name);
-          /* decrease the phase and sync */
-          file_entry_buf[DDL_LOG_PHASE_POS]--;
-          if (write_ddl_log_file_entry(ddl_log_entry->entry_pos))
-            break;
-          if (sync_ddl_log_no_lock())
-            break;
-          /* fall through */
-        case EXCH_PHASE_FROM_TO_NAME:
-          /* from_name -> name possibly done */
-          (void) file->ha_rename_table(ddl_log_entry->name,
-                                       ddl_log_entry->from_name);
-          /* decrease the phase and sync */
-          file_entry_buf[DDL_LOG_PHASE_POS]--;
-          if (write_ddl_log_file_entry(ddl_log_entry->entry_pos))
-            break;
-          if (sync_ddl_log_no_lock())
-            break;
-          /* fall through */
-        case EXCH_PHASE_NAME_TO_TEMP:
-          /* name -> tmp_name possibly done */
-          (void) file->ha_rename_table(ddl_log_entry->tmp_name,
-                                       ddl_log_entry->name);
-          /* disable the entry and sync */
-          file_entry_buf[DDL_LOG_ENTRY_TYPE_POS]= DDL_IGNORE_LOG_ENTRY_CODE;
-          if (write_ddl_log_file_entry(ddl_log_entry->entry_pos))
-            break;
-          if (sync_ddl_log_no_lock())
-            break;
-          error= FALSE;
-          break;
-        default:
-          DBUG_ASSERT(0);
-          break;
-      }
-
-      break;
-    }
-    default:
-      DBUG_ASSERT(0);
-      break;
-  }
-  delete file;
-error:
-  free_root(&mem_root, MYF(0)); 
-  DBUG_RETURN(error);
-}
-
-
-/**
-  Get a free entry in the ddl log
-
-  @param[out] active_entry     A ddl log memory entry returned
-
-  @return Operation status
-    @retval TRUE               Error
-    @retval FALSE              Success
-*/
-
-static bool get_free_ddl_log_entry(DDL_LOG_MEMORY_ENTRY **active_entry,
-                                   bool *write_header)
-{
-  DDL_LOG_MEMORY_ENTRY *used_entry;
-  DDL_LOG_MEMORY_ENTRY *first_used= global_ddl_log.first_used;
-  DBUG_ENTER("get_free_ddl_log_entry");
-
-  if (global_ddl_log.first_free == NULL)
-  {
-    if (!(used_entry= (DDL_LOG_MEMORY_ENTRY*)my_malloc(key_memory_DDL_LOG_MEMORY_ENTRY,
-                              sizeof(DDL_LOG_MEMORY_ENTRY), MYF(MY_WME))))
-    {
-      sql_print_error("Failed to allocate memory for ddl log free list");
-      DBUG_RETURN(TRUE);
-    }
-    global_ddl_log.num_entries++;
-    used_entry->entry_pos= global_ddl_log.num_entries;
-    *write_header= TRUE;
-  }
-  else
-  {
-    used_entry= global_ddl_log.first_free;
-    global_ddl_log.first_free= used_entry->next_log_entry;
-    *write_header= FALSE;
-  }
-  /*
-    Move from free list to used list
-  */
-  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;
-
-  *active_entry= used_entry;
-  DBUG_RETURN(FALSE);
-}
-
-
-/**
-  Execute one entry in the ddl log.
-  
-  Executing an entry means executing a linked list of actions.
-
-  @param first_entry           Reference to first action in entry
-
-  @return Operation status
-    @retval TRUE               Error
-    @retval FALSE              Success
-*/
-
-static bool execute_ddl_log_entry_no_lock(THD *thd, uint first_entry)
-{
-  DDL_LOG_ENTRY ddl_log_entry;
-  uint read_entry= first_entry;
-  DBUG_ENTER("execute_ddl_log_entry_no_lock");
-
-  mysql_mutex_assert_owner(&LOCK_gdl);
-  do
-  {
-    if (read_ddl_log_entry(read_entry, &ddl_log_entry))
-    {
-      /* Write to error log and continue with next log entry */
-      sql_print_error("Failed to read entry = %u from ddl log",
-                      read_entry);
-      break;
-    }
-    DBUG_ASSERT(ddl_log_entry.entry_type == DDL_LOG_ENTRY_CODE ||
-                ddl_log_entry.entry_type == DDL_IGNORE_LOG_ENTRY_CODE);
-
-    if (execute_ddl_log_action(thd, &ddl_log_entry))
-    {
-      /* Write to error log and continue with next log entry */
-      sql_print_error("Failed to execute action for entry = %u from ddl log",
-                      read_entry);
-      break;
-    }
-    read_entry= ddl_log_entry.next_entry;
-  } while (read_entry);
-  DBUG_RETURN(FALSE);
-}
-
-
-/*
-  External interface methods for the DDL log Module
-  ---------------------------------------------------
-*/
-
-/**
-  Write a ddl log entry.
-
-  A careful write of the ddl log is performed to ensure that we can
-  handle crashes occurring during CREATE and ALTER TABLE processing.
-
-  @param ddl_log_entry         Information about log entry
-  @param[out] entry_written    Entry information written into   
-
-  @return Operation status
-    @retval TRUE               Error
-    @retval FALSE              Success
-*/
-
-bool write_ddl_log_entry(DDL_LOG_ENTRY *ddl_log_entry,
-                         DDL_LOG_MEMORY_ENTRY **active_entry)
-{
-  bool error, write_header;
-  DBUG_ENTER("write_ddl_log_entry");
-
-  mysql_mutex_assert_owner(&LOCK_gdl);
-  if (init_ddl_log())
-  {
-    DBUG_RETURN(TRUE);
-  }
-  set_global_from_ddl_log_entry(ddl_log_entry);
-  if (get_free_ddl_log_entry(active_entry, &write_header))
-  {
-    DBUG_RETURN(TRUE);
-  }
-  error= FALSE;
-  DBUG_PRINT("ddl_log",
-             ("write type %c next %u name '%s' from_name '%s' handler '%s'"
-              " tmp_name '%s'",
-             (char) global_ddl_log.file_entry_buf[DDL_LOG_ACTION_TYPE_POS],
-             ddl_log_entry->next_entry,
-             (char*) &global_ddl_log.file_entry_buf[DDL_LOG_NAME_POS],
-             (char*) &global_ddl_log.file_entry_buf[DDL_LOG_NAME_POS
-                                                    + FN_REFLEN],
-             (char*) &global_ddl_log.file_entry_buf[DDL_LOG_NAME_POS
-                                                    + (2*FN_REFLEN)],
-             (char*) &global_ddl_log.file_entry_buf[DDL_LOG_NAME_POS
-                                                    + (3*FN_REFLEN)]));
-  if (unlikely(write_ddl_log_file_entry((*active_entry)->entry_pos)))
-  {
-    error= TRUE;
-    sql_print_error("Failed to write entry_no = %u",
-                    (*active_entry)->entry_pos);
-  }
-  if (write_header && likely(!error))
-  {
-    (void) sync_ddl_log_no_lock();
-    if (write_ddl_log_header())
-      error= TRUE;
-  }
-  if (unlikely(error))
-    release_ddl_log_memory_entry(*active_entry);
-  DBUG_RETURN(error);
-}
-
-
-/**
-  @brief Write final entry in the ddl log.
-
-  @details This is the last write in the ddl log. The previous log entries
-  have already been written but not yet synched to disk.
-  We write a couple of log entries that describes action to perform.
-  This entries are set-up in a linked list, however only when a first
-  execute entry is put as the first entry these will be executed.
-  This routine writes this first.
-
-  @param first_entry               First entry in linked list of entries
-                                   to execute, if 0 = NULL it means that
-                                   the entry is removed and the entries
-                                   are put into the free list.
-  @param complete                  Flag indicating we are simply writing
-                                   info about that entry has been completed
-  @param[in,out] active_entry      Entry to execute, 0 = NULL if the entry
-                                   is written first time and needs to be
-                                   returned. In this case the entry written
-                                   is returned in this parameter
-
-  @return Operation status
-    @retval TRUE                   Error
-    @retval FALSE                  Success
-*/ 
-
-bool write_execute_ddl_log_entry(uint first_entry,
-                                 bool complete,
-                                 DDL_LOG_MEMORY_ENTRY **active_entry)
-{
-  bool write_header= FALSE;
-  char *file_entry_buf= (char*)global_ddl_log.file_entry_buf;
-  DBUG_ENTER("write_execute_ddl_log_entry");
-
-  mysql_mutex_assert_owner(&LOCK_gdl);
-  if (init_ddl_log())
-  {
-    DBUG_RETURN(TRUE);
-  }
-  if (!complete)
-  {
-    /*
-      We haven't synched the log entries yet, we synch them now before
-      writing the execute entry. If complete is true we haven't written
-      any log entries before, we are only here to write the execute
-      entry to indicate it is done.
-    */
-    (void) sync_ddl_log_no_lock();
-    file_entry_buf[DDL_LOG_ENTRY_TYPE_POS]= (char)DDL_LOG_EXECUTE_CODE;
-  }
-  else
-    file_entry_buf[DDL_LOG_ENTRY_TYPE_POS]= (char)DDL_IGNORE_LOG_ENTRY_CODE;
-  file_entry_buf[DDL_LOG_ACTION_TYPE_POS]= 0; /* Ignored for execute entries */
-  file_entry_buf[DDL_LOG_PHASE_POS]= 0;
-  int4store(&file_entry_buf[DDL_LOG_NEXT_ENTRY_POS], first_entry);
-  file_entry_buf[DDL_LOG_NAME_POS]= 0;
-  file_entry_buf[DDL_LOG_NAME_POS + FN_REFLEN]= 0;
-  file_entry_buf[DDL_LOG_NAME_POS + 2*FN_REFLEN]= 0;
-  if (!(*active_entry))
-  {
-    if (get_free_ddl_log_entry(active_entry, &write_header))
-    {
-      DBUG_RETURN(TRUE);
-    }
-    write_header= TRUE;
-  }
-  if (write_ddl_log_file_entry((*active_entry)->entry_pos))
-  {
-    sql_print_error("Error writing execute entry in ddl log");
-    release_ddl_log_memory_entry(*active_entry);
-    DBUG_RETURN(TRUE);
-  }
-  (void) sync_ddl_log_no_lock();
-  if (write_header)
-  {
-    if (write_ddl_log_header())
-    {
-      release_ddl_log_memory_entry(*active_entry);
-      DBUG_RETURN(TRUE);
-    }
-  }
-  DBUG_RETURN(FALSE);
-}
-
-
-/**
-  Deactivate an individual entry.
-
-  @details see deactivate_ddl_log_entry_no_lock.
-
-  @param entry_no     Entry position of record to change
-
-  @return Operation status
-    @retval TRUE      Error
-    @retval FALSE     Success
-*/
-
-bool deactivate_ddl_log_entry(uint entry_no)
-{
-  bool error;
-  DBUG_ENTER("deactivate_ddl_log_entry");
-
-  mysql_mutex_lock(&LOCK_gdl);
-  error= deactivate_ddl_log_entry_no_lock(entry_no);
-  mysql_mutex_unlock(&LOCK_gdl);
-  DBUG_RETURN(error);
-}
-
-
-/**
-  Sync ddl log file.
-
-  @return Operation status
-    @retval TRUE        Error
-    @retval FALSE       Success
-*/
-
-bool sync_ddl_log()
-{
-  bool error;
-  DBUG_ENTER("sync_ddl_log");
-
-  mysql_mutex_lock(&LOCK_gdl);
-  error= sync_ddl_log_no_lock();
-  mysql_mutex_unlock(&LOCK_gdl);
-
-  DBUG_RETURN(error);
-}
-
-
-/**
-  Release a log memory entry.
-  @param log_memory_entry                Log memory entry to release
-*/
-
-void release_ddl_log_memory_entry(DDL_LOG_MEMORY_ENTRY *log_entry)
-{
-  DDL_LOG_MEMORY_ENTRY *first_free= global_ddl_log.first_free;
-  DDL_LOG_MEMORY_ENTRY *next_log_entry= log_entry->next_log_entry;
-  DDL_LOG_MEMORY_ENTRY *prev_log_entry= log_entry->prev_log_entry;
-  DBUG_ENTER("release_ddl_log_memory_entry");
-
-  mysql_mutex_assert_owner(&LOCK_gdl);
-  global_ddl_log.first_free= log_entry;
-  log_entry->next_log_entry= first_free;
-
-  if (prev_log_entry)
-    prev_log_entry->next_log_entry= next_log_entry;
-  else
-    global_ddl_log.first_used= next_log_entry;
-  if (next_log_entry)
-    next_log_entry->prev_log_entry= prev_log_entry;
-  DBUG_VOID_RETURN;
-}
-
-
-/**
-  Execute one entry in the ddl log.
-  
-  Executing an entry means executing a linked list of actions.
-
-  @param first_entry           Reference to first action in entry
-
-  @return Operation status
-    @retval TRUE               Error
-    @retval FALSE              Success
-*/
-
-bool execute_ddl_log_entry(THD *thd, uint first_entry)
-{
-  bool error;
-  DBUG_ENTER("execute_ddl_log_entry");
-
-  mysql_mutex_lock(&LOCK_gdl);
-  error= execute_ddl_log_entry_no_lock(thd, first_entry);
-  mysql_mutex_unlock(&LOCK_gdl);
-  DBUG_RETURN(error);
-}
-
-
-/**
-  Close the ddl log.
-*/
-
-static void close_ddl_log()
-{
-  DBUG_ENTER("close_ddl_log");
-  if (global_ddl_log.file_id >= 0)
-  {
-    (void) mysql_file_close(global_ddl_log.file_id, MYF(MY_WME));
-    global_ddl_log.file_id= (File) -1;
-  }
-  DBUG_VOID_RETURN;
-}
-
-
-/**
-  Execute the ddl log at recovery of MySQL Server.
-*/
-
-void execute_ddl_log_recovery()
-{
-  uint num_entries, i;
-  THD *thd;
-  DDL_LOG_ENTRY ddl_log_entry;
-  char file_name[FN_REFLEN];
-  static char recover_query_string[]= "INTERNAL DDL LOG RECOVER IN PROGRESS";
-  DBUG_ENTER("execute_ddl_log_recovery");
-
-  /*
-    Initialise global_ddl_log struct
-  */
-  bzero(global_ddl_log.file_entry_buf, sizeof(global_ddl_log.file_entry_buf));
-  global_ddl_log.inited= FALSE;
-  global_ddl_log.recovery_phase= TRUE;
-  global_ddl_log.io_size= IO_SIZE;
-  global_ddl_log.file_id= (File) -1;
-
-  /*
-    To be able to run this from boot, we allocate a temporary THD
-  */
-  if (!(thd=new THD(0)))
-    DBUG_VOID_RETURN;
-  thd->thread_stack= (char*) &thd;
-  thd->store_globals();
-
-  thd->set_query(recover_query_string, strlen(recover_query_string));
-
-  /* this also initialize LOCK_gdl */
-  num_entries= read_ddl_log_header();
-  mysql_mutex_lock(&LOCK_gdl);
-  for (i= 1; i < num_entries + 1; i++)
-  {
-    if (read_ddl_log_entry(i, &ddl_log_entry))
-    {
-      sql_print_error("Failed to read entry no = %u from ddl log",
-                       i);
-      continue;
-    }
-    if (ddl_log_entry.entry_type == DDL_LOG_EXECUTE_CODE)
-    {
-      if (execute_ddl_log_entry_no_lock(thd, ddl_log_entry.next_entry))
-      {
-        /* Real unpleasant scenario but we continue anyways.  */
-        continue;
-      }
-    }
-  }
-  close_ddl_log();
-  create_ddl_log_file_name(file_name);
-  (void) mysql_file_delete(key_file_global_ddl_log, file_name, MYF(0));
-  global_ddl_log.recovery_phase= FALSE;
-  mysql_mutex_unlock(&LOCK_gdl);
-  thd->reset_query();
-  delete thd;
-  DBUG_VOID_RETURN;
-}
-
-
-/**
-  Release all memory allocated to the ddl log.
-*/
-
-void release_ddl_log()
-{
-  DDL_LOG_MEMORY_ENTRY *free_list;
-  DDL_LOG_MEMORY_ENTRY *used_list;
-  DBUG_ENTER("release_ddl_log");
-
-  if (!global_ddl_log.do_release)
-    DBUG_VOID_RETURN;
-
-  mysql_mutex_lock(&LOCK_gdl);
-  free_list= global_ddl_log.first_free;
-  used_list= global_ddl_log.first_used;
-  while (used_list)
-  {
-    DDL_LOG_MEMORY_ENTRY *tmp= used_list->next_log_entry;
-    my_free(used_list);
-    used_list= tmp;
-  }
-  while (free_list)
-  {
-    DDL_LOG_MEMORY_ENTRY *tmp= free_list->next_log_entry;
-    my_free(free_list);
-    free_list= tmp;
-  }
-  close_ddl_log();
-  global_ddl_log.inited= 0;
-  mysql_mutex_unlock(&LOCK_gdl);
-  mysql_mutex_destroy(&LOCK_gdl);
-  global_ddl_log.do_release= false;
-  DBUG_VOID_RETURN;
-}
-
-
-/*
----------------------------------------------------------------------------
-
-  END MODULE DDL log
-  --------------------
-
----------------------------------------------------------------------------
-*/
-
 
 /**
    @brief construct a temporary shadow file name.
diff --git a/sql/sql_table.h b/sql/sql_table.h
index 3e469fd797f..4bae4f8610c 100644
--- a/sql/sql_table.h
+++ b/sql/sql_table.h
@@ -38,80 +38,6 @@ typedef struct st_key_cache KEY_CACHE;
 typedef struct st_lock_param_type ALTER_PARTITION_PARAM_TYPE;
 typedef struct st_order ORDER;
 
-enum ddl_log_entry_code
-{
-  /*
-    DDL_LOG_EXECUTE_CODE:
-      This is a code that indicates that this is a log entry to
-      be executed, from this entry a linked list of log entries
-      can be found and executed.
-    DDL_LOG_ENTRY_CODE:
-      An entry to be executed in a linked list from an execute log
-      entry.
-    DDL_IGNORE_LOG_ENTRY_CODE:
-      An entry that is to be ignored
-  */
-  DDL_LOG_EXECUTE_CODE = 'e',
-  DDL_LOG_ENTRY_CODE = 'l',
-  DDL_IGNORE_LOG_ENTRY_CODE = 'i'
-};
-
-enum ddl_log_action_code
-{
-  /*
-    The type of action that a DDL_LOG_ENTRY_CODE entry is to
-    perform.
-    DDL_LOG_DELETE_ACTION:
-      Delete an entity
-    DDL_LOG_RENAME_ACTION:
-      Rename an entity
-    DDL_LOG_REPLACE_ACTION:
-      Rename an entity after removing the previous entry with the
-      new name, that is replace this entry.
-    DDL_LOG_EXCHANGE_ACTION:
-      Exchange two entities by renaming them a -> tmp, b -> a, tmp -> b.
-  */
-  DDL_LOG_DELETE_ACTION = 'd',
-  DDL_LOG_RENAME_ACTION = 'r',
-  DDL_LOG_REPLACE_ACTION = 's',
-  DDL_LOG_EXCHANGE_ACTION = 'e'
-};
-
-enum enum_ddl_log_exchange_phase {
-  EXCH_PHASE_NAME_TO_TEMP= 0,
-  EXCH_PHASE_FROM_TO_NAME= 1,
-  EXCH_PHASE_TEMP_TO_FROM= 2
-};
-
-
-typedef struct st_ddl_log_entry
-{
-  const char *name;
-  const char *from_name;
-  const char *handler_name;
-  const char *tmp_name;
-  uint next_entry;
-  uint entry_pos;
-  enum ddl_log_entry_code entry_type;
-  enum ddl_log_action_code action_type;
-  /*
-    Most actions have only one phase. REPLACE does however have two
-    phases. The first phase removes the file with the new name if
-    there was one there before and the second phase renames the
-    old name to the new name.
-  */
-  char phase;
-} DDL_LOG_ENTRY;
-
-typedef struct st_ddl_log_memory_entry
-{
-  uint entry_pos;
-  struct st_ddl_log_memory_entry *next_log_entry;
-  struct st_ddl_log_memory_entry *prev_log_entry;
-  struct st_ddl_log_memory_entry *next_active_log_entry;
-} DDL_LOG_MEMORY_ENTRY;
-
-
 enum enum_explain_filename_mode
 {
   EXPLAIN_ALL_VERBOSE= 0,
@@ -266,17 +192,6 @@ int write_bin_log(THD *thd, bool clear_error,
                   bool is_trans= FALSE);
 int write_bin_log_with_if_exists(THD *thd, bool clear_error,
                                  bool is_trans, bool add_if_exists);
-bool write_ddl_log_entry(DDL_LOG_ENTRY *ddl_log_entry,
-                           DDL_LOG_MEMORY_ENTRY **active_entry);
-bool write_execute_ddl_log_entry(uint first_entry,
-                                   bool complete,
-                                   DDL_LOG_MEMORY_ENTRY **active_entry);
-bool deactivate_ddl_log_entry(uint entry_no);
-void release_ddl_log_memory_entry(DDL_LOG_MEMORY_ENTRY *log_entry);
-bool sync_ddl_log();
-void release_ddl_log();
-void execute_ddl_log_recovery();
-bool execute_ddl_log_entry(THD *thd, uint first_entry);
 
 template<typename T> class List;
 void promote_first_timestamp_column(List<Create_field> *column_definitions);
@@ -289,7 +204,6 @@ uint explain_filename(THD* thd, const char *from, char *to, uint to_length,
 
 
 extern MYSQL_PLUGIN_IMPORT const LEX_CSTRING primary_key_name;
-extern mysql_mutex_t LOCK_gdl;
 
 bool check_engine(THD *, const char *, const char *, HA_CREATE_INFO *);