MDEV-25180 Atomic ALTER TABLE

MDEV-25604 Atomic DDL: Binlog event written upon recovery does not
           have default database

The purpose of this task is to ensure that ALTER TABLE is atomic even if
the MariaDB server would be killed at any point of the alter table.
This means that either the ALTER TABLE succeeds (including that triggers,
the status tables and the binary log are updated) or things should be
reverted to their original state.

If the server crashes before the new version is fully up to date and
commited, it will revert to the original table and remove all
temporary files and tables.
If the new version is commited, crash recovery will use the new version,
and update triggers, the status tables and the binary log.
The one execption is ALTER TABLE .. RENAME .. where no changes are done
to table definition. This one will work as RENAME and roll back unless
the whole statement completed, including updating the binary log (if
enabled).

Other changes:
- Added handlerton->check_version() function to allow the ddl recovery
  code to check, in case of inplace alter table, if the table in the
  storage engine is of the new or old version.
- Added handler->table_version() so that an engine can report the current
  version of the table. This should be changed each time the table
  definition changes.
- Added  ha_signal_ddl_recovery_done() and
  handlerton::signal_ddl_recovery_done() to inform all handlers when
  ddl recovery has been done. (Needed by InnoDB).
- Added handlerton call inplace_alter_table_committed, to signal engine
  that ddl_log has been closed for the alter table query.
- Added new handerton flag
  HTON_REQUIRES_NOTIFY_TABLEDEF_CHANGED_AFTER_COMMIT to signal when we
  should call hton->notify_tabledef_changed() during
  mysql_inplace_alter_table. This was required as MyRocks and InnoDB
  needed the call at different times.
- Added function server_uuid_value() to be able to generate a temporary
  xid when ddl recovery writes the query to the binary log. This is
  needed to be able to handle crashes during ddl log recovery.
- Moved freeing of the frm definition to end of mysql_alter_table() to
  remove duplicate code and have a common exit strategy.

-------
InnoDB part of atomic ALTER TABLE
(Implemented by Marko Mäkelä)
innodb_check_version(): Compare the saved dict_table_t::def_trx_id
to determine whether an ALTER TABLE operation was committed.

We must correctly recover dict_table_t::def_trx_id for this to work.
Before purge removes any trace of DB_TRX_ID from system tables, it
will make an effort to load the user table into the cache, so that
the dict_table_t::def_trx_id can be recovered.

ha_innobase::table_version(): return garbage, or the trx_id that would
be used for committing an ALTER TABLE operation.

In InnoDB, table names starting with #sql-ib will remain special:
they will be dropped on startup. This may be revisited later in
MDEV-18518 when we implement proper undo logging and rollback
for creating or dropping multiple tables in a transaction.

Table names starting with #sql will retain some special meaning:
dict_table_t::parse_name() will not consider such names for
MDL acquisition, and dict_table_rename_in_cache() will treat such
names specially when handling FOREIGN KEY constraints.

Simplify InnoDB DROP INDEX.
Prevent purge wakeup

To ensure that dict_table_t::def_trx_id will be recovered correctly
in case the server is killed before ddl_log_complete(), we will block
the purge of any history in SYS_TABLES, SYS_INDEXES, SYS_COLUMNS
between ha_innobase::commit_inplace_alter_table(commit=true)
(purge_sys.stop_SYS()) and purge_sys.resume_SYS().
The completion callback purge_sys.resume_SYS() must be between
ddl_log_complete() and MDL release.

--------

MyRocks support for atomic ALTER TABLE
(Implemented by Sergui Petrunia)

Implement these SE API functions:
- ha_rocksdb::table_version()
- hton->check_version = rocksdb_check_versionMyRocks data dictionary
  now stores table version for each table.
  (Absence of table version record is interpreted as table_version=0,
  that is, which means no upgrade changes are needed)
- For inplace alter table of a partitioned table, call the underlying
  handlerton when checking if the table is ok. This assumes that the
  partition engine commits all changes at once.

12 files changed, 1311 insertions, 247 deletions

diff --git a/sql/ddl_log.cc b/sql/ddl_log.cc
index 5f9a025c042..93ebc92b421 100644
--- a/sql/ddl_log.cc
+++ b/sql/ddl_log.cc
@@ -90,7 +90,7 @@ const char *ddl_log_action_name[DDL_LOG_LAST_ACTION]=
   "rename table", "rename view",
   "initialize drop table", "drop table",
   "drop view", "drop trigger", "drop db", "create table", "create view",
-  "delete tmp file", "create trigger",
+  "delete tmp file", "create trigger", "alter table", "store query"
 };
 
 /* Number of phases per entry */
@@ -101,6 +101,7 @@ const uchar ddl_log_entry_phases[DDL_LOG_LAST_ACTION]=
   (uchar) DDL_DROP_PHASE_END, 1, 1,
   (uchar) DDL_DROP_DB_PHASE_END, (uchar) DDL_CREATE_TABLE_PHASE_END,
   (uchar) DDL_CREATE_VIEW_PHASE_END, 0, (uchar) DDL_CREATE_TRIGGER_PHASE_END,
+  DDL_ALTER_TABLE_PHASE_END, 1
 };
 
 
@@ -117,13 +118,21 @@ struct st_global_ddl_log
   bool open;
 };
 
-/* The following structure is only used during startup recovery */
+/*
+  The following structure is only used during startup recovery
+  for writing queries to the binary log.
+ */
+
 class st_ddl_recovery {
 public:
   String drop_table;
   String drop_view;
+  String query;
+  String db;
   size_t drop_table_init_length, drop_view_init_length;
   char   current_db[NAME_LEN];
+  uint   execute_entry_pos;
+  ulonglong xid;
 };
 
 static st_global_ddl_log global_ddl_log;
@@ -295,6 +304,7 @@ static bool write_ddl_log_file_entry(uint entry_pos)
 static bool update_phase(uint entry_pos, uchar phase)
 {
   DBUG_ENTER("update_phase");
+  DBUG_PRINT("enter", ("phase: %d", (int) phase));
 
   DBUG_RETURN(mysql_file_pwrite(global_ddl_log.file_id, &phase, 1,
                                 global_ddl_log.io_size * entry_pos +
@@ -304,6 +314,25 @@ static bool update_phase(uint entry_pos, uchar phase)
 }
 
 
+/*
+  Update flags in ddl log entry
+
+  This is not synced as it usually followed by a phase change, which will sync.
+*/
+
+static bool update_flags(uint entry_pos, uint16 flags)
+{
+  uchar buff[2];
+  DBUG_ENTER("update_flags");
+
+  int2store(buff, flags);
+  DBUG_RETURN(mysql_file_pwrite(global_ddl_log.file_id, buff, sizeof(buff),
+                                global_ddl_log.io_size * entry_pos +
+                                DDL_LOG_FLAG_POS,
+                                MYF(MY_WME | MY_NABP)));
+}
+
+
 static bool update_next_entry_pos(uint entry_pos, uint next_entry)
 {
   uchar buff[4];
@@ -323,7 +352,7 @@ static bool update_xid(uint entry_pos, ulonglong xid)
   DBUG_ENTER("update_xid");
 
   int8store(buff, xid);
-  DBUG_RETURN(mysql_file_pwrite(global_ddl_log.file_id, buff, 8,
+  DBUG_RETURN(mysql_file_pwrite(global_ddl_log.file_id, buff, sizeof(buff),
                                 global_ddl_log.io_size * entry_pos +
                                 DDL_LOG_XID_POS,
                                 MYF(MY_WME | MY_NABP)) ||
@@ -559,6 +588,7 @@ static void set_global_from_ddl_log_entry(const DDL_LOG_ENTRY *ddl_log_entry)
   pos= store_string(pos, end, &ddl_log_entry->from_db);
   pos= store_string(pos, end, &ddl_log_entry->from_name);
   pos= store_string(pos, end, &ddl_log_entry->tmp_name);
+  pos= store_string(pos, end, &ddl_log_entry->extra_name);
   bzero(pos, global_ddl_log.io_size - (pos - file_entry_buf));
 }
 
@@ -580,6 +610,7 @@ static size_t ddl_log_free_space_in_entry(const DDL_LOG_ENTRY *ddl_log_entry)
   length+= ddl_log_entry->from_db.length;
   length+= ddl_log_entry->from_name.length;
   length+= ddl_log_entry->tmp_name.length;
+  length+= ddl_log_entry->extra_name.length;
   return global_ddl_log.io_size - length - 3;   // 3 is for storing next string
 }
 
@@ -621,6 +652,7 @@ static void set_ddl_log_entry_from_global(DDL_LOG_ENTRY *ddl_log_entry,
   ddl_log_entry->from_db=      get_string(&pos, end);
   ddl_log_entry->from_name=    get_string(&pos, end);
   ddl_log_entry->tmp_name=     get_string(&pos, end);
+  ddl_log_entry->extra_name=   get_string(&pos, end);
 }
 
 
@@ -818,6 +850,19 @@ static bool ddl_log_increment_phase_no_lock(uint entry_pos)
 
 
 /*
+  Increment phase and sync ddl log. This expects LOCK_gdl to be locked
+*/
+
+static bool increment_phase(uint entry_pos)
+{
+  if (ddl_log_increment_phase_no_lock(entry_pos))
+    return 1;
+  ddl_log_sync_no_lock();
+  return 0;
+}
+
+
+/*
   Ignore errors from the file system about:
   - Non existing tables or file (from drop table or delete file)
   - Error about tables files that already exists.
@@ -935,7 +980,7 @@ static void ddl_log_to_binary_log(THD *thd, String *query)
 static bool ddl_log_drop_to_binary_log(THD *thd, DDL_LOG_ENTRY *ddl_log_entry,
                                   String *query)
 {
-  DBUG_ENTER("ddl_log_binary_log");
+  DBUG_ENTER("ddl_log_drop_to_binary_log");
   if (mysql_bin_log.is_open())
   {
     if (!ddl_log_entry->next_entry ||
@@ -960,6 +1005,235 @@ static bool ddl_log_drop_to_binary_log(THD *thd, DDL_LOG_ENTRY *ddl_log_entry,
   DBUG_RETURN(0);
 }
 
+/*
+  Create a new handler based on handlerton name
+*/
+
+static handler *create_handler(THD *thd, MEM_ROOT *mem_root,
+                               LEX_CSTRING *name)
+{
+  handlerton *hton;
+  handler *file;
+  plugin_ref plugin= my_plugin_lock_by_name(thd, name,
+                                            MYSQL_STORAGE_ENGINE_PLUGIN);
+  if (!plugin)
+  {
+    my_error(ER_UNKNOWN_STORAGE_ENGINE, MYF(ME_ERROR_LOG), name->str);
+    return 0;
+  }
+  hton= plugin_hton(plugin);
+  if (!ha_storage_engine_is_enabled(hton))
+  {
+    my_error(ER_STORAGE_ENGINE_DISABLED, MYF(ME_ERROR_LOG), name->str);
+    return 0;
+  }
+  if ((file= hton->create(hton, (TABLE_SHARE*) 0, mem_root)))
+    file->init();
+  return file;
+}
+
+
+/*
+  Rename a table and its .frm file for a ddl_log_entry
+
+  We first rename the table and then the .frm file as some engines,
+  like connect, needs the .frm file to exists to be able to do an rename.
+*/
+
+static void execute_rename_table(DDL_LOG_ENTRY *ddl_log_entry, handler *file,
+                                 const LEX_CSTRING *from_db,
+                                 const LEX_CSTRING *from_table,
+                                 const LEX_CSTRING *to_db,
+                                 const LEX_CSTRING *to_table,
+                                 uint flags,
+                                 char *from_path, char *to_path)
+{
+  uint to_length=0, fr_length=0;
+  DBUG_ENTER("execute_rename_table");
+
+  if (file->needs_lower_case_filenames())
+  {
+    build_lower_case_table_filename(from_path, FN_REFLEN,
+                                    from_db, from_table,
+                                    flags & FN_FROM_IS_TMP);
+    build_lower_case_table_filename(to_path, FN_REFLEN,
+                                    to_db, to_table, flags & FN_TO_IS_TMP);
+  }
+  else
+  {
+    fr_length= build_table_filename(from_path, FN_REFLEN,
+                                    from_db->str, from_table->str, "",
+                                    flags & FN_TO_IS_TMP);
+    to_length= build_table_filename(to_path, FN_REFLEN,
+                                    to_db->str, to_table->str, "",
+                                    flags & FN_TO_IS_TMP);
+  }
+  file->ha_rename_table(from_path, to_path);
+  if (file->needs_lower_case_filenames())
+  {
+    /*
+      We have to rebuild the file names as the .frm file should be used
+      without lower case conversion
+    */
+    fr_length= build_table_filename(from_path, FN_REFLEN,
+                                    from_db->str, from_table->str, reg_ext,
+                                    flags & FN_FROM_IS_TMP);
+    to_length= build_table_filename(to_path, FN_REFLEN,
+                                    to_db->str, to_table->str, reg_ext,
+                                    flags & FN_TO_IS_TMP);
+  }
+  else
+  {
+    strmov(from_path+fr_length, reg_ext);
+    strmov(to_path+to_length,   reg_ext);
+  }
+  if (!access(from_path, F_OK))
+    (void) mysql_file_rename(key_file_frm, from_path, to_path, MYF(MY_WME));
+  DBUG_VOID_RETURN;
+}
+
+
+/*
+  Update triggers
+
+  If swap_tables == 0  (Restoring the original in case of failed rename)
+    Convert triggers for db.name -> from_db.from_name
+  else (Doing the rename in case of ALTER TABLE ... RENAME)
+    Convert triggers for from_db.from_name -> db.extra_name
+*/
+
+static void rename_triggers(THD *thd, DDL_LOG_ENTRY *ddl_log_entry,
+                            bool swap_tables)
+{
+  LEX_CSTRING to_table, from_table, to_db, from_db, from_converted_name;
+  char to_path[FN_REFLEN+1], from_path[FN_REFLEN+1], conv_path[FN_REFLEN+1];
+
+  if (!swap_tables)
+  {
+    from_db=    ddl_log_entry->db;
+    from_table= ddl_log_entry->name;
+    to_db=      ddl_log_entry->from_db;
+    to_table=   ddl_log_entry->from_name;
+  }
+  else
+  {
+    from_db=    ddl_log_entry->from_db;
+    from_table= ddl_log_entry->from_name;
+    to_db=      ddl_log_entry->db;
+    to_table=   ddl_log_entry->extra_name;
+  }
+
+  build_filename_and_delete_tmp_file(from_path, sizeof(from_path),
+                                     &from_db, &from_table,
+                                     TRG_EXT, key_file_trg);
+  build_filename_and_delete_tmp_file(to_path, sizeof(to_path),
+                                     &to_db, &to_table,
+                                     TRG_EXT, key_file_trg);
+  if (lower_case_table_names)
+  {
+    uint errors;
+    from_converted_name.str= conv_path;
+    from_converted_name.length=
+      strconvert(system_charset_info, from_table.str, from_table.length,
+                 files_charset_info, conv_path, FN_REFLEN, &errors);
+  }
+  else
+    from_converted_name= from_table;
+
+  if (!access(to_path, F_OK))
+  {
+    /*
+      The original file was never renamed or we crashed in recovery
+      just after renaming back the file.
+      In this case the current file is correct and we can remove any
+      left over copied files
+    */
+    (void) mysql_file_delete(key_file_trg, from_path, MYF(0));
+  }
+  else if (!access(from_path, F_OK))
+  {
+    /* .TRG file was renamed. Rename it back */
+    /*
+      We have to create a MDL lock as change_table_names() checks that we
+      have a mdl locks for the table
+    */
+    MDL_request mdl_request;
+    TRIGGER_RENAME_PARAM trigger_param;
+    int error __attribute__((unused));
+    MDL_REQUEST_INIT(&mdl_request, MDL_key::TABLE,
+                     from_db.str,
+                     from_converted_name.str,
+                     MDL_EXCLUSIVE, MDL_EXPLICIT);
+    error= thd->mdl_context.acquire_lock(&mdl_request, 1);
+    /* acquire_locks() should never fail during recovery */
+    DBUG_ASSERT(error == 0);
+
+    (void) Table_triggers_list::prepare_for_rename(thd,
+                                                   &trigger_param,
+                                                   &from_db,
+                                                   &from_table,
+                                                   &from_converted_name,
+                                                   &to_db,
+                                                   &to_table);
+    (void) Table_triggers_list::change_table_name(thd,
+                                                  &trigger_param,
+                                                  &from_db,
+                                                  &from_table,
+                                                  &from_converted_name,
+                                                  &to_db,
+                                                  &to_table);
+    thd->mdl_context.release_lock(mdl_request.ticket);
+  }
+}
+
+
+/*
+  Update stat tables
+
+  If swap_tables == 0
+    Convert stats for from_db.from_table -> db.name
+  else
+    Convert stats for db.name -> from_db.from_table
+*/
+
+static void rename_in_stat_tables(THD *thd, DDL_LOG_ENTRY *ddl_log_entry,
+                                  bool swap_tables)
+{
+  LEX_CSTRING from_table, to_table, from_db, to_db, from_converted_name;
+  char conv_path[FN_REFLEN+1];
+
+  if (!swap_tables)
+  {
+    from_db=    ddl_log_entry->db;
+    from_table= ddl_log_entry->name;
+    to_db=      ddl_log_entry->from_db;
+    to_table=   ddl_log_entry->from_name;
+  }
+  else
+  {
+    from_db=    ddl_log_entry->from_db;
+    from_table= ddl_log_entry->from_name;
+    to_db=      ddl_log_entry->db;
+    to_table=   ddl_log_entry->extra_name;
+  }
+  if (lower_case_table_names)
+  {
+    uint errors;
+    from_converted_name.str= conv_path;
+    from_converted_name.length=
+      strconvert(system_charset_info, from_table.str, from_table.length,
+                 files_charset_info, conv_path, FN_REFLEN, &errors);
+  }
+  else
+    from_converted_name= from_table;
+
+  (void) rename_table_in_stat_tables(thd,
+                                     &from_db,
+                                     &from_converted_name,
+                                     &to_db,
+                                     &to_table);
+}
+
 
 /**
   Execute one action in a ddl log entry
@@ -1009,17 +1283,9 @@ static int ddl_log_execute_action(THD *thd, MEM_ROOT *mem_root,
     frm_action= TRUE;
   else if (ddl_log_entry->handler_name.length)
   {
-    plugin_ref plugin= my_plugin_lock_by_name(thd, &handler_name,
-                                              MYSQL_STORAGE_ENGINE_PLUGIN);
-    if (!plugin)
-    {
-      my_error(ER_UNKNOWN_STORAGE_ENGINE, MYF(0), ddl_log_entry->handler_name);
-      goto end;
-    }
-    hton= plugin_hton(plugin);
-    file= get_new_handler((TABLE_SHARE*)0, mem_root, hton);
-    if (unlikely(!file))
+    if (!(file= create_handler(thd, mem_root, &handler_name)))
       goto end;
+    hton= file->ht;
   }
 
   switch (ddl_log_entry->action_type) {
@@ -1049,9 +1315,8 @@ static int ddl_log_execute_action(THD *thd, MEM_ROOT *mem_root,
             break;
         }
       }
-      if (ddl_log_increment_phase_no_lock(entry_pos))
+      if (increment_phase(entry_pos))
         break;
-      (void) ddl_log_sync_no_lock();
       error= 0;
       if (ddl_log_entry->action_type == DDL_LOG_DELETE_ACTION)
         break;
@@ -1082,9 +1347,8 @@ static int ddl_log_execute_action(THD *thd, MEM_ROOT *mem_root,
     else
       (void) file->ha_rename_table(ddl_log_entry->from_name.str,
                                    ddl_log_entry->name.str);
-    if (ddl_log_increment_phase_no_lock(entry_pos))
+    if (increment_phase(entry_pos))
       break;
-    (void) ddl_log_sync_no_lock();
     break;
   }
   case DDL_LOG_EXCHANGE_ACTION:
@@ -1135,130 +1399,35 @@ static int ddl_log_execute_action(THD *thd, MEM_ROOT *mem_root,
     /*
       We should restore things by renaming from
       'entry->name' to 'entry->from_name'
-
-      In the following code 'to_' stands for what the table was renamed to
-      that we have to rename back.
     */
-    size_t fr_length, to_length;
-    LEX_CSTRING from_table, to_table, to_converted_name;
-    from_table= ddl_log_entry->from_name;
-    to_table= ddl_log_entry->name;
-
-    /* Some functions wants to have the lower case table name as an argument */
-    if (lower_case_table_names)
-    {
-      uint errors;
-      to_converted_name.str= to_path;
-      to_converted_name.length=
-        strconvert(system_charset_info, to_table.str, to_table.length,
-                   files_charset_info, from_path, FN_REFLEN, &errors);
-    }
-    else
-      to_converted_name= to_table;
-
     switch (ddl_log_entry->phase) {
     case DDL_RENAME_PHASE_TRIGGER:
-    {
-      MDL_request mdl_request;
-      TRIGGER_RENAME_PARAM trigger_param;
-
-      build_filename_and_delete_tmp_file(to_path, sizeof(to_path),
-                                         &ddl_log_entry->db,
-                                         &ddl_log_entry->name,
-                                         TRG_EXT,
-                                         key_file_trg);
-      build_filename_and_delete_tmp_file(from_path, sizeof(from_path),
-                                         &ddl_log_entry->from_db,
-                                         &ddl_log_entry->from_name,
-                                         TRG_EXT, key_file_trg);
-
-      if (!access(from_path, F_OK))
-      {
-        /*
-          The original file was never renamed or we crashed in recovery
-          just after renaming back the file.
-          In this case the current file is correct and we can remove any
-          left over copied files
-        */
-        (void) mysql_file_delete(key_file_trg, to_path, MYF(0));
-      }
-      else if (!access(to_path, F_OK))
-      {
-        /* .TRG file was renamed. Rename it back */
-        /*
-          We have to create a MDL lock as change_table_names() checks that we
-          have a mdl locks for the table
-        */
-        MDL_REQUEST_INIT(&mdl_request, MDL_key::TABLE,
-                         ddl_log_entry->db.str,
-                         to_converted_name.str,
-                         MDL_EXCLUSIVE, MDL_EXPLICIT);
-        error= thd->mdl_context.acquire_lock(&mdl_request, 1);
-        /* acquire_locks() should never fail during recovery */
-        DBUG_ASSERT(error == 0);
-        (void) Table_triggers_list::prepare_for_rename(thd,
-                                                       &trigger_param,
-                                                       &ddl_log_entry->db,
-                                                       &to_table,
-                                                       &to_converted_name,
-                                                       &ddl_log_entry->from_db,
-                                                       &from_table);
-        (void) Table_triggers_list::change_table_name(thd,
-                                                      &trigger_param,
-                                                      &ddl_log_entry->db,
-                                                      &to_table,
-                                                      &to_converted_name,
-                                                      &ddl_log_entry->from_db,
-                                                      &from_table);
-        thd->mdl_context.release_lock(mdl_request.ticket);
-      }
-      if (ddl_log_increment_phase_no_lock(entry_pos))
+      rename_triggers(thd, ddl_log_entry, 0);
+      if (increment_phase(entry_pos))
         break;
-      (void) ddl_log_sync_no_lock();
-    }
     /* fall through */
     case DDL_RENAME_PHASE_STAT:
-    {
-      (void) rename_table_in_stat_tables(thd,
-                                         &ddl_log_entry->db,
-                                         &to_converted_name,
-                                         &ddl_log_entry->from_db,
-                                         &from_table);
-      if (ddl_log_increment_phase_no_lock(entry_pos))
-        break;
-      (void) ddl_log_sync_no_lock();
-    }
+      /*
+        Stat tables must be updated last so that we can handle a rename of
+        a stat table. For now we just rememeber that we have to update it
+      */
+      update_flags(ddl_log_entry->entry_pos, DDL_LOG_FLAG_UPDATE_STAT);
+      ddl_log_entry->flags|= DDL_LOG_FLAG_UPDATE_STAT;
     /* fall through */
     case DDL_RENAME_PHASE_TABLE:
       /* Restore frm and table to original names */
-      to_length= build_table_filename(to_path, sizeof(to_path) - 1,
-                                      ddl_log_entry->db.str,
-                                      ddl_log_entry->name.str,
-                                      reg_ext, 0);
-      fr_length= build_table_filename(from_path, sizeof(from_path) - 1,
-                                      ddl_log_entry->from_db.str,
-                                      ddl_log_entry->from_name.str,
-                                      reg_ext, 0);
-      (void) mysql_file_rename(key_file_frm, to_path, from_path, MYF(MY_WME));
+      execute_rename_table(ddl_log_entry, file,
+                           &ddl_log_entry->db, &ddl_log_entry->name,
+                           &ddl_log_entry->from_db, &ddl_log_entry->from_name,
+                           0,
+                           from_path, to_path);
 
-      if (file->needs_lower_case_filenames())
-      {
-        build_lower_case_table_filename(to_path, sizeof(to_path) - 1,
-                                        &ddl_log_entry->db,
-                                        &to_table, 0);
-        build_lower_case_table_filename(from_path, sizeof(from_path) - 1,
-                                        &ddl_log_entry->from_db,
-                                        &from_table, 0);
-      }
-      else
+      if (ddl_log_entry->flags & DDL_LOG_FLAG_UPDATE_STAT)
       {
-        /* remove extension from file name */
-        DBUG_ASSERT(to_length != 0 && fr_length != 0);
-        to_path[to_length - reg_ext_length]= 0;
-        from_path[fr_length - reg_ext_length]= 0;
+        /* Update stat tables last */
+        rename_in_stat_tables(thd, ddl_log_entry, 0);
       }
 
-      file->ha_rename_table(to_path, from_path);
       /* disable the entry and sync */
       (void) update_phase(entry_pos, DDL_LOG_FINAL_PHASE);
       break;
@@ -1348,19 +1517,17 @@ static int ddl_log_execute_action(THD *thd, MEM_ROOT *mem_root,
         /* Not found or already deleted. Delete .frm if it exists */
         strxnmov(to_path, sizeof(to_path)-1, path.str, reg_ext, NullS);
         mysql_file_delete(key_file_frm, to_path, MYF(MY_WME|MY_IGNORE_ENOENT));
+        error= 0;
       }
-      if (ddl_log_increment_phase_no_lock(entry_pos))
+      if (increment_phase(entry_pos))
         break;
-      (void) ddl_log_sync_no_lock();
       /* Fall through */
     case DDL_DROP_PHASE_TRIGGER:
       Table_triggers_list::drop_all_triggers(thd, &db, &table,
                                              MYF(MY_WME | MY_IGNORE_ENOENT));
-      if (ddl_log_increment_phase_no_lock(entry_pos))
+      if (increment_phase(entry_pos))
         break;
-      (void) ddl_log_sync_no_lock();
       /* Fall through */
-
     case DDL_DROP_PHASE_BINLOG:
       if (strcmp(recovery_state.current_db, db.str))
       {
@@ -1374,7 +1541,7 @@ static int ddl_log_execute_action(THD *thd, MEM_ROOT *mem_root,
       if (ddl_log_drop_to_binary_log(thd, ddl_log_entry,
                                      &recovery_state.drop_table))
       {
-        if (ddl_log_increment_phase_no_lock(entry_pos))
+        if (increment_phase(entry_pos))
           break;
       }
       break;
@@ -1408,7 +1575,7 @@ static int ddl_log_execute_action(THD *thd, MEM_ROOT *mem_root,
       if (ddl_log_drop_to_binary_log(thd, ddl_log_entry,
                                      &recovery_state.drop_view))
       {
-        if (ddl_log_increment_phase_no_lock(entry_pos))
+        if (increment_phase(entry_pos))
           break;
       }
     }
@@ -1502,9 +1669,9 @@ static int ddl_log_execute_action(THD *thd, MEM_ROOT *mem_root,
       mysql_file_delete_with_symlink(key_file_misc, to_path, "", MYF(0));
 
       (void) rm_dir_w_symlink(path.str, 0);
-      if (ddl_log_increment_phase_no_lock(entry_pos))
+      if (increment_phase(entry_pos))
         break;
-      /* Fall through */
+      /* fall through */
     case DDL_DROP_DB_PHASE_LOG:
     {
       String *query= &recovery_state.drop_table;
@@ -1578,6 +1745,7 @@ static int ddl_log_execute_action(THD *thd, MEM_ROOT *mem_root,
       }
     }
     (void) update_phase(entry_pos, DDL_LOG_FINAL_PHASE);
+    error= 0;
     break;
   }
   case DDL_LOG_CREATE_VIEW_ACTION:
@@ -1733,6 +1901,349 @@ static int ddl_log_execute_action(THD *thd, MEM_ROOT *mem_root,
     }
     break;
   }
+  case DDL_LOG_ALTER_TABLE_ACTION:
+  {
+    handlerton *org_hton, *partition_hton;
+    handler *org_file;
+    bool is_renamed= ddl_log_entry->flags & DDL_LOG_FLAG_ALTER_RENAME;
+    bool new_version_ready= 0, new_version_unusable= 0;
+    LEX_CSTRING db, table;
+    db=     ddl_log_entry->db;
+    table=  ddl_log_entry->name;
+
+    if (!(org_file= create_handler(thd, mem_root,
+                                   &ddl_log_entry->from_handler_name)))
+      goto end;
+    /* Handlerton of the final table and any temporary tables */
+    org_hton= org_file->ht;
+    /*
+      partition_hton is the hton for the new file, or
+      in case of ALTER of a partitioned table, the underlying
+      table
+    */
+    partition_hton= hton;
+
+    if (ddl_log_entry->flags & DDL_LOG_FLAG_ALTER_PARTITION)
+    {
+      /*
+        The from and to tables where both using the partition engine.
+      */
+      hton= org_hton;
+    }
+    switch (ddl_log_entry->phase) {
+    case DDL_ALTER_TABLE_PHASE_RENAME_FAILED:
+      /*
+        We come here when the final rename of temporary table (#sql-alter) to
+        the original name failed. Now we have to delete the temporary table
+        and restore the backup.
+      */
+      quick_rm_table(thd, hton, &db, &table, FN_IS_TMP);
+      if (!is_renamed)
+      {
+        execute_rename_table(ddl_log_entry, file,
+                             &ddl_log_entry->from_db,
+                             &ddl_log_entry->extra_name, // #sql-backup
+                             &ddl_log_entry->from_db,
+                             &ddl_log_entry->from_name,
+                             FN_FROM_IS_TMP,
+                             from_path, to_path);
+      }
+      (void) update_phase(entry_pos, DDL_LOG_FINAL_PHASE);
+      break;
+    case DDL_ALTER_TABLE_PHASE_PREPARE_INPLACE:
+      /* We crashed before ddl_log_update_unique_id() was called */
+      new_version_unusable= 1;
+      /* fall through */
+    case DDL_ALTER_TABLE_PHASE_INPLACE_COPIED:
+      /* The inplace alter table is committed and ready to be used */
+      if (!new_version_unusable)
+        new_version_ready= 1;
+    /* fall through */
+    case DDL_ALTER_TABLE_PHASE_INPLACE:
+    {
+      int fr_length, to_length;
+      /*
+        Inplace alter table was used.
+        On disk there are now a table with the original name, the
+        original .frm file and potentially a #sql-alter...frm file
+        with the new definition.
+      */
+      fr_length= build_table_filename(from_path, sizeof(from_path) - 1,
+                                      ddl_log_entry->db.str,
+                                      ddl_log_entry->name.str,
+                                      reg_ext, 0);
+      to_length= build_table_filename(to_path, sizeof(to_path) - 1,
+                                      ddl_log_entry->from_db.str,
+                                      ddl_log_entry->from_name.str,
+                                      reg_ext, 0);
+      if (!access(from_path, F_OK))             // Does #sql-alter.. exists?
+      {
+        LEX_CUSTRING version= {ddl_log_entry->uuid, MY_UUID_SIZE};
+        /*
+          Temporary .frm file exists.  This means that that the table in
+          the storage engine can be of either old or new version.
+          If old version, delete the new .frm table and keep the old one.
+          If new version, replace the old .frm with the new one.
+        */
+        to_path[to_length - reg_ext_length]= 0;  // Remove .frm
+        if (!new_version_unusable &&
+            ( !partition_hton->check_version || new_version_ready ||
+              !partition_hton->check_version(partition_hton,
+                                             to_path, &version,
+                                             ddl_log_entry->unique_id)))
+        {
+          /* Table is up to date */
+
+          /*
+            Update state so that if we crash and retry the ddl log entry,
+            we know that we can use the new table even if .frm is renamed.
+          */
+          if (ddl_log_entry->phase != DDL_ALTER_TABLE_PHASE_INPLACE_COPIED)
+            (void) update_phase(entry_pos,
+                                DDL_ALTER_TABLE_PHASE_INPLACE_COPIED);
+          /* Replace old .frm file with new one */
+          to_path[to_length - reg_ext_length]= FN_EXTCHAR;
+          (void) mysql_file_rename(key_file_frm, from_path, to_path,
+                                   MYF(MY_WME));
+          new_version_ready= 1;
+        }
+        else
+        {
+          DBUG_ASSERT(!new_version_ready);
+          /*
+            Use original version of the .frm file.
+            Remove temporary #sql-alter.frm file and the #sql-alter table.
+            We have also to remove the temporary table as some storage engines,
+            like InnoDB, may use it as an internal temporary table
+            during inplace alter table.
+          */
+          from_path[fr_length - reg_ext_length]= 0;
+          error= org_hton->drop_table(org_hton, from_path);
+          if (non_existing_table_error(error))
+            error= 0;
+          from_path[fr_length - reg_ext_length]= FN_EXTCHAR;
+          mysql_file_delete(key_file_frm, from_path,
+                            MYF(MY_WME|MY_IGNORE_ENOENT));
+          (void) update_phase(entry_pos, DDL_LOG_FINAL_PHASE);
+          break;
+        }
+      }
+      if (is_renamed && new_version_ready)
+      {
+        /* After the renames above, the original table is now in from_name */
+        ddl_log_entry->name= ddl_log_entry->from_name;
+        /* Rename db.name -> db.extra_name */
+        execute_rename_table(ddl_log_entry, file,
+                             &ddl_log_entry->db, &ddl_log_entry->name,
+                             &ddl_log_entry->db, &ddl_log_entry->extra_name,
+                             0,
+                             from_path, to_path);
+      }
+      (void) update_phase(entry_pos, DDL_ALTER_TABLE_PHASE_UPDATE_TRIGGERS);
+      goto update_triggers;
+    }
+    case DDL_ALTER_TABLE_PHASE_COPIED:
+    {
+      char *from_end;
+      /*
+        New table is created and we have the query for the binary log.
+        We should remove the original table and in the next stage replace
+        it with the new one.
+      */
+      build_table_filename(from_path, sizeof(from_path) - 1,
+                           ddl_log_entry->from_db.str,
+                           ddl_log_entry->from_name.str,
+                           "", 0);
+      build_table_filename(to_path, sizeof(to_path) - 1,
+                           ddl_log_entry->db.str,
+                           ddl_log_entry->name.str,
+                           "", 0);
+      from_end= strend(from_path);
+      if (likely(org_hton))
+      {
+        error= org_hton->drop_table(org_hton, from_path);
+        if (non_existing_table_error(error))
+          error= 0;
+      }
+      strmov(from_end, reg_ext);
+      mysql_file_delete(key_file_frm, from_path,
+                        MYF(MY_WME|MY_IGNORE_ENOENT));
+      *from_end= 0;                             // Remove extension
+
+      (void) update_phase(entry_pos, DDL_ALTER_TABLE_PHASE_OLD_RENAMED);
+    }
+    /* fall through */
+    case DDL_ALTER_TABLE_PHASE_OLD_RENAMED:
+    {
+      /*
+        The new table (from_path) is up to date.
+        Original table is either renamed as backup table (normal case),
+        only frm is renamed (in case of engine change) or deleted above.
+      */
+      if (!is_renamed)
+      {
+        uint length;
+        /* Rename new "temporary" table to the original wanted name */
+        execute_rename_table(ddl_log_entry, file,
+                             &ddl_log_entry->db,
+                             &ddl_log_entry->name,
+                             &ddl_log_entry->from_db,
+                             &ddl_log_entry->from_name,
+                             FN_FROM_IS_TMP,
+                             from_path, to_path);
+
+        /*
+          Remove backup (only happens if alter table used without rename).
+          Backup name is always in lower case, so there is no need for
+          converting table names.
+        */
+        length= build_table_filename(from_path, sizeof(from_path) - 1,
+                                     ddl_log_entry->from_db.str,
+                                     ddl_log_entry->extra_name.str,
+                                     "", FN_IS_TMP);
+        if (likely(org_hton))
+        {
+          if (ddl_log_entry->flags & DDL_LOG_FLAG_ALTER_ENGINE_CHANGED)
+          {
+            /* Only frm is renamed, storage engine files have original name */
+            build_table_filename(to_path, sizeof(from_path) - 1,
+                                 ddl_log_entry->from_db.str,
+                                 ddl_log_entry->from_name.str,
+                                 "", 0);
+            error= org_hton->drop_table(org_hton, to_path);
+          }
+          else
+            error= org_hton->drop_table(org_hton, from_path);
+          if (non_existing_table_error(error))
+            error= 0;
+        }
+        strmov(from_path + length, reg_ext);
+        mysql_file_delete(key_file_frm, from_path,
+                          MYF(MY_WME|MY_IGNORE_ENOENT));
+      }
+      else
+        execute_rename_table(ddl_log_entry, file,
+                             &ddl_log_entry->db, &ddl_log_entry->name,
+                             &ddl_log_entry->db, &ddl_log_entry->extra_name,
+                             FN_FROM_IS_TMP,
+                             from_path, to_path);
+      (void) update_phase(entry_pos, DDL_ALTER_TABLE_PHASE_UPDATE_TRIGGERS);
+    }
+    /* fall through */
+    case DDL_ALTER_TABLE_PHASE_UPDATE_TRIGGERS:
+    update_triggers:
+    {
+      if (is_renamed)
+      {
+        // rename_triggers will rename from: from_db.from_name -> db.extra_name
+        rename_triggers(thd, ddl_log_entry, 1);
+        (void) update_phase(entry_pos, DDL_ALTER_TABLE_PHASE_UPDATE_STATS);
+      }
+    }
+    /* fall through */
+    case DDL_ALTER_TABLE_PHASE_UPDATE_STATS:
+      if (is_renamed)
+      {
+        ddl_log_entry->name= ddl_log_entry->from_name;
+        ddl_log_entry->from_name= ddl_log_entry->extra_name;
+        rename_in_stat_tables(thd, ddl_log_entry, 1);
+        (void) update_phase(entry_pos, DDL_ALTER_TABLE_PHASE_UPDATE_STATS);
+      }
+      /* fall through */
+    case DDL_ALTER_TABLE_PHASE_UPDATE_BINARY_LOG:
+    {
+      /* Write ALTER TABLE query to binary log */
+      if (recovery_state.query.length() && mysql_bin_log.is_open())
+      {
+        LEX_CSTRING save_db;
+        /* Reuse old xid value if possible */
+        if (!recovery_state.xid)
+          recovery_state.xid= server_uuid_value();
+        thd->binlog_xid= recovery_state.xid;
+        update_xid(recovery_state.execute_entry_pos, thd->binlog_xid);
+
+        mysql_mutex_unlock(&LOCK_gdl);
+        save_db= thd->db;
+        lex_string_set3(&thd->db, recovery_state.db.ptr(),
+                        recovery_state.db.length());
+        (void) thd->binlog_query(THD::STMT_QUERY_TYPE,
+                                 recovery_state.query.ptr(),
+                                 recovery_state.query.length(),
+                                 TRUE, FALSE, FALSE, 0);
+        thd->binlog_xid= 0;
+        thd->db= save_db;
+        mysql_mutex_lock(&LOCK_gdl);
+      }
+      recovery_state.query.length(0);
+      (void) update_phase(entry_pos, DDL_LOG_FINAL_PHASE);
+      break;
+    }
+    /*
+      The following cases are when alter table failed and we have to roll
+      back
+    */
+    case DDL_ALTER_TABLE_PHASE_CREATED:
+    {
+      /*
+        Temporary table should have been created. Delete it.
+      */
+      if (likely(hton))
+      {
+        error= hton->drop_table(hton, ddl_log_entry->tmp_name.str);
+        if (non_existing_table_error(error))
+          error= 0;
+      }
+      (void) update_phase(entry_pos, DDL_ALTER_TABLE_PHASE_INIT);
+    }
+    /* fall through */
+    case DDL_ALTER_TABLE_PHASE_INIT:
+    {
+      /*
+        A temporary .frm and possible a .par files should have been created
+      */
+      strxmov(to_path, ddl_log_entry->tmp_name.str, reg_ext, NullS);
+      mysql_file_delete(key_file_frm, to_path, MYF(MY_WME|MY_IGNORE_ENOENT));
+      strxmov(to_path, ddl_log_entry->tmp_name.str, PAR_EXT, NullS);
+      mysql_file_delete(key_file_partition_ddl_log, to_path,
+                        MYF(MY_WME|MY_IGNORE_ENOENT));
+      (void) update_phase(entry_pos, DDL_LOG_FINAL_PHASE);
+      break;
+    }
+    }
+    delete org_file;
+    break;
+  }
+  case DDL_LOG_STORE_QUERY_ACTION:
+  {
+    /*
+      Read query for next ddl command
+    */
+    if (ddl_log_entry->flags)
+    {
+      /*
+        First QUERY event. Allocate query string.
+        Query length is stored in unique_id
+      */
+      if (recovery_state.query.alloc((size_t) (ddl_log_entry->unique_id+1)))
+        goto end;
+      recovery_state.query.length(0);
+      recovery_state.db.copy(ddl_log_entry->db.str, ddl_log_entry->db.length,
+                             system_charset_info);
+    }
+    if (unlikely(recovery_state.query.length() +
+                 ddl_log_entry->extra_name.length >
+                 recovery_state.query.alloced_length()))
+    {
+      /* Impossible length. Ignore query */
+      recovery_state.query.length(0);
+      error= 1;
+      my_error(ER_INTERNAL_ERROR, MYF(0),
+               "DDL log: QUERY event has impossible length");
+      break;
+    }
+    recovery_state.query.qs_append(&ddl_log_entry->extra_name);
+    break;
+  }
   default:
     DBUG_ASSERT(0);
     break;
@@ -1740,7 +2251,7 @@ static int ddl_log_execute_action(THD *thd, MEM_ROOT *mem_root,
 
 end:
   delete file;
-  if ((error= no_such_table_handler.unhandled_errors > 0))
+  if ((error= (no_such_table_handler.unhandled_errors > 0)))
     my_errno= no_such_table_handler.first_error;
   thd->pop_internal_handler();
   DBUG_RETURN(error);
@@ -1819,6 +2330,8 @@ void ddl_log_release_memory_entry(DDL_LOG_MEMORY_ENTRY *log_entry)
     global_ddl_log.first_used= next_log_entry;
   if (next_log_entry)
     next_log_entry->prev_log_entry= prev_log_entry;
+  // Ensure we get a crash if we try to access this link again.
+  log_entry->next_active_log_entry= (DDL_LOG_MEMORY_ENTRY*) 0x1;
   DBUG_VOID_RETURN;
 }
 
@@ -2056,6 +2569,8 @@ bool ddl_log_sync()
 
   Executing an entry means executing a linked list of actions.
 
+  This function is called for recovering partitioning in case of error.
+
   @param first_entry           Reference to first action in entry
 
   @return Operation status
@@ -2169,9 +2684,12 @@ int ddl_log_execute_recovery()
   thd->thread_stack= (char*) &thd;
   thd->store_globals();
   thd->init();                                  // Needed for error messages
+
   thd->log_all_errors= (global_system_variables.log_warnings >= 3);
   recovery_state.drop_table.free();
   recovery_state.drop_view.free();
+  recovery_state.query.free();
+  recovery_state.db.free();
 
   thd->set_query(recover_query_string, strlen(recover_query_string));
 
@@ -2185,6 +2703,13 @@ int ddl_log_execute_recovery()
     }
     if (ddl_log_entry.entry_type == DDL_LOG_EXECUTE_CODE)
     {
+      /*
+        Remeber information about executive ddl log entry,
+        used for binary logging during recovery
+      */
+      recovery_state.execute_entry_pos= i;
+      recovery_state.xid= ddl_log_entry.xid;
+
       /* purecov: begin tested */
       if (ddl_log_entry.unique_id > DDL_LOG_MAX_RETRY)
       {
@@ -2200,6 +2725,7 @@ int ddl_log_execute_recovery()
         continue;
       }
       /* purecov: end tested */
+
       if (ddl_log_execute_entry_no_lock(thd, ddl_log_entry.next_entry))
       {
         /* Real unpleasant scenario but we have to continue anyway  */
@@ -2211,10 +2737,13 @@ int ddl_log_execute_recovery()
   }
   recovery_state.drop_table.free();
   recovery_state.drop_view.free();
+  recovery_state.query.free();
+  recovery_state.db.free();
   close_ddl_log();
   mysql_mutex_unlock(&LOCK_gdl);
   thd->reset_query();
   delete thd;
+  set_current_thd(original_thd);
 
   /*
     Create a new ddl_log to get rid of old stuff and ensure that header matches
@@ -2280,7 +2809,7 @@ static void add_log_entry(DDL_LOG_STATE *state,
                           DDL_LOG_MEMORY_ENTRY *log_entry)
 {
   log_entry->next_active_log_entry= state->list;
-  state->list= log_entry;
+  state->main_entry= state->list= log_entry;
 }
 
 
@@ -2332,6 +2861,8 @@ void ddl_log_complete(DDL_LOG_STATE *state)
 
 /**
   Revert (execute) all entries in the ddl log
+
+  This is called for failed rename table, create trigger or drop trigger.
 */
 
 void ddl_log_revert(THD *thd, DDL_LOG_STATE *state)
@@ -2355,13 +2886,48 @@ void ddl_log_revert(THD *thd, DDL_LOG_STATE *state)
 
 
 /*
-  Update phase of last created ddl log entry
+  Update phase of main ddl log entry (usually the last one created,
+  except in case of query events, the one before the query event).
 */
 
 bool ddl_log_update_phase(DDL_LOG_STATE *state, uchar phase)
 {
   DBUG_ENTER("ddl_log_update_phase");
-  DBUG_RETURN(update_phase(state->list->entry_pos, phase));
+  if (likely(state->list))
+    DBUG_RETURN(update_phase(state->main_entry->entry_pos, phase));
+  DBUG_RETURN(0);
+}
+
+
+/*
+  Update flag bits in main ddl log entry (usually last created, except in case
+  of query events, the one before the query event.
+*/
+
+bool ddl_log_add_flag(DDL_LOG_STATE *state, uint16 flags)
+{
+  DBUG_ENTER("ddl_log_update_phase");
+  if (likely(state->list))
+  {
+    state->flags|= flags;
+    DBUG_RETURN(update_flags(state->main_entry->entry_pos, state->flags));
+  }
+  DBUG_RETURN(0);
+}
+
+
+/**
+   Update unique_id (used for inplace alter table)
+*/
+
+bool ddl_log_update_unique_id(DDL_LOG_STATE *state, ulonglong id)
+{
+  DBUG_ENTER("ddl_log_update_unique_id");
+  DBUG_PRINT("enter", ("id: %llu", id));
+  /* The following may not be true in case of temporary tables */
+  if (likely(state->list))
+    DBUG_RETURN(update_unique_id(state->main_entry->entry_pos, id));
+  DBUG_RETURN(0);
 }
 
 
@@ -2396,6 +2962,8 @@ bool ddl_log_update_xid(DDL_LOG_STATE *state, ulonglong xid)
 
 /*
   Write ddl_log_entry and write or update ddl_execute_entry
+
+  Will update DDL_LOG_STATE->flags
 */
 
 static bool ddl_log_write(DDL_LOG_STATE *ddl_state,
@@ -2417,6 +2985,7 @@ static bool ddl_log_write(DDL_LOG_STATE *ddl_state,
     DBUG_RETURN(1);
   }
   add_log_entry(ddl_state, log_entry);
+  ddl_state->flags|= ddl_log_entry->flags;      // Update cache
   DBUG_RETURN(0);
 }
 
@@ -2678,7 +3247,7 @@ bool ddl_log_create_table(THD *thd, DDL_LOG_STATE *ddl_state,
   ddl_log_entry.db=           *const_cast<LEX_CSTRING*>(db);
   ddl_log_entry.name=         *const_cast<LEX_CSTRING*>(table);
   ddl_log_entry.tmp_name=     *const_cast<LEX_CSTRING*>(path);
-  ddl_log_entry.flags=        only_frm;
+  ddl_log_entry.flags=        only_frm ? DDL_LOG_FLAG_ONLY_FRM : 0;
 
   DBUG_RETURN(ddl_log_write(ddl_state, &ddl_log_entry));
 }
@@ -2750,3 +3319,156 @@ bool ddl_log_create_trigger(THD *thd, DDL_LOG_STATE *ddl_state,
   ddl_log_entry.phase=        (uchar) phase;
   DBUG_RETURN(ddl_log_write(ddl_state, &ddl_log_entry));
 }
+
+
+/**
+   Log ALTER TABLE
+
+   $param backup_name   Name of backup table. In case of ALTER TABLE rename
+                        this is the final table name
+*/
+
+bool ddl_log_alter_table(THD *thd, DDL_LOG_STATE *ddl_state,
+                         handlerton *org_hton,
+                         const LEX_CSTRING *db, const LEX_CSTRING *table,
+                         handlerton *new_hton,
+                         handlerton *partition_underlying_hton,
+                         const LEX_CSTRING *new_db,
+                         const LEX_CSTRING *new_table,
+                         const LEX_CSTRING *frm_path,
+                         const LEX_CSTRING *backup_name,
+                         const LEX_CUSTRING *version,
+                         ulonglong table_version,
+                         bool is_renamed)
+{
+  DDL_LOG_ENTRY ddl_log_entry;
+  DBUG_ENTER("ddl_log_alter_table");
+  DBUG_ASSERT(new_hton);
+  DBUG_ASSERT(org_hton);
+
+  bzero(&ddl_log_entry, sizeof(ddl_log_entry));
+  ddl_log_entry.action_type=  DDL_LOG_ALTER_TABLE_ACTION;
+  if (new_hton)
+    lex_string_set(&ddl_log_entry.handler_name,
+                   ha_resolve_storage_engine_name(new_hton));
+  /* Store temporary table name */
+  ddl_log_entry.db=           *const_cast<LEX_CSTRING*>(new_db);
+  ddl_log_entry.name=         *const_cast<LEX_CSTRING*>(new_table);
+  if (org_hton)
+    lex_string_set(&ddl_log_entry.from_handler_name,
+                   ha_resolve_storage_engine_name(org_hton));
+  ddl_log_entry.from_db=      *const_cast<LEX_CSTRING*>(db);
+  ddl_log_entry.from_name=    *const_cast<LEX_CSTRING*>(table);
+  ddl_log_entry.tmp_name=     *const_cast<LEX_CSTRING*>(frm_path);
+  ddl_log_entry.extra_name=   *const_cast<LEX_CSTRING*>(backup_name);
+  ddl_log_entry.flags=        is_renamed ? DDL_LOG_FLAG_ALTER_RENAME : 0;
+  ddl_log_entry.unique_id=    table_version;
+
+  /*
+    If we are doing an inplace of a partition engine, we need to log the
+    underlaying engine. We store this is in ddl_log_entry.handler_name
+  */
+  if (new_hton == org_hton && partition_underlying_hton != new_hton)
+  {
+    lex_string_set(&ddl_log_entry.handler_name,
+                   ha_resolve_storage_engine_name(partition_underlying_hton));
+    ddl_log_entry.flags|= DDL_LOG_FLAG_ALTER_PARTITION;
+  }
+  DBUG_ASSERT(version->length == MY_UUID_SIZE);
+  memcpy(ddl_log_entry.uuid, version->str, version->length);
+  DBUG_RETURN(ddl_log_write(ddl_state, &ddl_log_entry));
+}
+
+
+/*
+  Store query that later should be logged to binary log
+
+  The links of the query log event is
+
+  execute_log_event -> first log_query_event [-> log_query_event...] ->
+  action_log_event (probably a LOG_ALTER_TABLE_ACTION event)
+
+  This ensures that when we execute the log_query_event it can collect
+  the full query from the log_query_events and then execute the
+  action_log_event with the original query stored in 'recovery_state.query'.
+
+  The query is stored in ddl_log_entry.extra_name as this is the last string
+  stored in the log block (makes it easier to check and debug).
+*/
+
+bool ddl_log_store_query(THD *thd, DDL_LOG_STATE *ddl_state,
+                         const char *query, size_t length)
+{
+  DDL_LOG_ENTRY ddl_log_entry;
+  DDL_LOG_MEMORY_ENTRY *first_entry, *next_entry= 0;
+  DDL_LOG_MEMORY_ENTRY *original_entry= ddl_state->list;
+  size_t max_query_length;
+  uint entry_pos, next_entry_pos= 0, parent_entry_pos;
+  DBUG_ENTER("ddl_log_store_query");
+  DBUG_ASSERT(length <= UINT_MAX32);
+  DBUG_ASSERT(length > 0);
+  DBUG_ASSERT(ddl_state->list);
+
+  bzero(&ddl_log_entry, sizeof(ddl_log_entry));
+  ddl_log_entry.action_type=  DDL_LOG_STORE_QUERY_ACTION;
+  ddl_log_entry.unique_id=    length;
+  ddl_log_entry.flags=        1;                // First entry
+  ddl_log_entry.db= thd->db;                    // Current database
+
+  max_query_length= ddl_log_free_space_in_entry(&ddl_log_entry);
+
+  mysql_mutex_lock(&LOCK_gdl);
+  ddl_log_entry.entry_type= DDL_LOG_ENTRY_CODE;
+
+  if (ddl_log_get_free_entry(&first_entry))
+    goto err;
+  parent_entry_pos= ddl_state->list->entry_pos;
+  entry_pos= first_entry->entry_pos;
+  add_log_entry(ddl_state, first_entry);
+
+  while (length)
+  {
+    size_t write_length= MY_MIN(length, max_query_length);
+    ddl_log_entry.extra_name.str= query;
+    ddl_log_entry.extra_name.length= write_length;
+
+    query+= write_length;
+    length-= write_length;
+
+    if (length > 0)
+    {
+      if (ddl_log_get_free_entry(&next_entry))
+        goto err;
+      ddl_log_entry.next_entry= next_entry_pos= next_entry->entry_pos;
+      add_log_entry(ddl_state, next_entry);
+    }
+    else
+    {
+      /* point next link of last query_action event to the original action */
+      ddl_log_entry.next_entry= parent_entry_pos;
+    }
+    set_global_from_ddl_log_entry(&ddl_log_entry);
+    if (unlikely(write_ddl_log_file_entry(entry_pos)))
+      goto err;
+    entry_pos= next_entry_pos;
+    ddl_log_entry.flags= 0;           // Only first entry has this set
+    ddl_log_entry.db.length= 0;       // Don't need DB anymore
+    ddl_log_entry.extra_name.length= 0;
+    max_query_length= ddl_log_free_space_in_entry(&ddl_log_entry);
+  }
+  if (ddl_log_write_execute_entry(first_entry->entry_pos,
+                                  &ddl_state->execute_entry))
+    goto err;
+
+  /* Set the original entry to be used for future PHASE updates */
+  ddl_state->main_entry= original_entry;
+  mysql_mutex_unlock(&LOCK_gdl);
+  DBUG_RETURN(0);
+err:
+  /*
+    Allocated ddl_log entries will be released by the
+    ddl_log_release_entries() call in dl_log_complete()
+  */
+  mysql_mutex_unlock(&LOCK_gdl);
+  DBUG_RETURN(1);
+}
diff --git a/sql/ddl_log.h b/sql/ddl_log.h
index 3aabde31733..7033539963d 100644
--- a/sql/ddl_log.h
+++ b/sql/ddl_log.h
@@ -86,6 +86,8 @@ enum ddl_log_action_code
   DDL_LOG_CREATE_VIEW_ACTION=13,
   DDL_LOG_DELETE_TMP_FILE_ACTION=14,
   DDL_LOG_CREATE_TRIGGER_ACTION=15,
+  DDL_LOG_ALTER_TABLE_ACTION=16,
+  DDL_LOG_STORE_QUERY_ACTION=17,
   DDL_LOG_LAST_ACTION                          /* End marker */
 };
 
@@ -142,6 +144,35 @@ enum enum_ddl_log_create_trigger_phase {
   DDL_CREATE_TRIGGER_PHASE_END
 };
 
+enum enum_ddl_log_alter_table_phase {
+  DDL_ALTER_TABLE_PHASE_INIT,
+  DDL_ALTER_TABLE_PHASE_RENAME_FAILED,
+  DDL_ALTER_TABLE_PHASE_INPLACE_COPIED,
+  DDL_ALTER_TABLE_PHASE_INPLACE,
+  DDL_ALTER_TABLE_PHASE_PREPARE_INPLACE,
+  DDL_ALTER_TABLE_PHASE_CREATED,
+  DDL_ALTER_TABLE_PHASE_COPIED,
+  DDL_ALTER_TABLE_PHASE_OLD_RENAMED,
+  DDL_ALTER_TABLE_PHASE_UPDATE_TRIGGERS,
+  DDL_ALTER_TABLE_PHASE_UPDATE_STATS,
+  DDL_ALTER_TABLE_PHASE_UPDATE_BINARY_LOG,
+  DDL_ALTER_TABLE_PHASE_END
+};
+
+
+/*
+  Flags stored in DDL_LOG_ENTRY.flags
+  The flag values can be reused for different commands
+*/
+#define DDL_LOG_FLAG_ALTER_RENAME         (1 << 0)
+#define DDL_LOG_FLAG_ALTER_ENGINE_CHANGED (1 << 1)
+#define DDL_LOG_FLAG_ONLY_FRM             (1 << 2)
+#define DDL_LOG_FLAG_UPDATE_STAT          (1 << 3)
+/*
+  Set when using ALTER TABLE on a partitioned table and the table
+  engine is not changed
+*/
+#define DDL_LOG_FLAG_ALTER_PARTITION      (1 << 4)
 
 /*
   Setting ddl_log_entry.phase to this has the same effect as setting
@@ -155,10 +186,11 @@ typedef struct st_ddl_log_entry
   LEX_CSTRING name;
   LEX_CSTRING from_name;
   LEX_CSTRING handler_name;
-  LEX_CSTRING tmp_name;
   LEX_CSTRING db;
   LEX_CSTRING from_db;
   LEX_CSTRING from_handler_name;
+  LEX_CSTRING tmp_name;                /* frm file or temporary file name */
+  LEX_CSTRING extra_name;              /* Backup table name */
   uchar uuid[MY_UUID_SIZE];            // UUID for new frm file
 
   ulonglong xid;                       // Xid stored in the binary log
@@ -210,6 +242,12 @@ typedef struct st_ddl_log_state
   DDL_LOG_MEMORY_ENTRY *list;
   /* One execute entry per list */
   DDL_LOG_MEMORY_ENTRY *execute_entry;
+  /*
+    Entry used for PHASE updates. Normally same as first in 'list', but in
+    case of a query log event, this points to the main event.
+  */
+  DDL_LOG_MEMORY_ENTRY *main_entry;
+  uint16 flags;                                 /* Cache for flags */
   bool is_active() { return list != 0; }
 } DDL_LOG_STATE;
 
@@ -232,6 +270,8 @@ void ddl_log_complete(DDL_LOG_STATE *ddl_log_state);
 void ddl_log_revert(THD *thd, DDL_LOG_STATE *ddl_log_state);
 
 bool ddl_log_update_phase(DDL_LOG_STATE *entry, uchar phase);
+bool ddl_log_add_flag(DDL_LOG_STATE *entry, uint16 flag);
+bool ddl_log_update_unique_id(DDL_LOG_STATE *state, ulonglong id);
 bool ddl_log_update_xid(DDL_LOG_STATE *state, ulonglong xid);
 bool ddl_log_disable_entry(DDL_LOG_STATE *state);
 bool ddl_log_increment_phase(uint entry_pos);
@@ -294,5 +334,19 @@ bool ddl_log_create_trigger(THD *thd, DDL_LOG_STATE *ddl_state,
                             const LEX_CSTRING *db, const LEX_CSTRING *table,
                             const LEX_CSTRING *trigger_name,
                             enum_ddl_log_create_trigger_phase phase);
+bool ddl_log_alter_table(THD *thd, DDL_LOG_STATE *ddl_state,
+                         handlerton *org_hton,
+                         const LEX_CSTRING *db, const LEX_CSTRING *table,
+                         handlerton *new_hton,
+                         handlerton *partition_underlying_hton,
+                         const LEX_CSTRING *new_db,
+                         const LEX_CSTRING *new_table,
+                         const LEX_CSTRING *frm_path,
+                         const LEX_CSTRING *backup_table_name,
+                         const LEX_CUSTRING *version,
+                         ulonglong table_version,
+                         bool is_renamed);
+bool ddl_log_store_query(THD *thd, DDL_LOG_STATE *ddl_log_state,
+                         const char *query, size_t length);
 extern mysql_mutex_t LOCK_gdl;
 #endif /* DDL_LOG_INCLUDED */
diff --git a/sql/ha_partition.cc b/sql/ha_partition.cc
index b6c1a33fc91..f252c0c3137 100644
--- a/sql/ha_partition.cc
+++ b/sql/ha_partition.cc
@@ -304,8 +304,10 @@ ha_partition::ha_partition(handlerton *hton, TABLE_SHARE *share)
 
 void ha_partition::ha_partition_init()
 {
+  DBUG_ENTER("ha_partition::ha_partition_init");
   init_alloc_root(PSI_INSTRUMENT_ME, &m_mem_root, 512, 512, MYF(0));
   init_handler_variables();
+  DBUG_VOID_RETURN;
 }
 
 /*
diff --git a/sql/handler.cc b/sql/handler.cc
index 82c0f11905a..fa99c24771c 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -943,6 +943,24 @@ void ha_kill_query(THD* thd, enum thd_kill_levels level)
 }
 
 
+static my_bool signal_ddl_recovery_done(THD *, plugin_ref plugin, void *)
+{
+  handlerton *hton= plugin_hton(plugin);
+  if (hton->signal_ddl_recovery_done)
+    (hton->signal_ddl_recovery_done)(hton);
+  return 0;
+}
+
+
+void ha_signal_ddl_recovery_done()
+{
+  DBUG_ENTER("ha_signal_ddl_recovery_done");
+  plugin_foreach(NULL, signal_ddl_recovery_done, MYSQL_STORAGE_ENGINE_PLUGIN,
+                 NULL);
+  DBUG_VOID_RETURN;
+}
+
+
 /*****************************************************************************
   Backup functions
 ******************************************************************************/
@@ -4954,19 +4972,9 @@ Alter_inplace_info::Alter_inplace_info(HA_CREATE_INFO *create_info_arg,
     alter_info(alter_info_arg),
     key_info_buffer(key_info_arg),
     key_count(key_count_arg),
-    index_drop_count(0),
-    index_drop_buffer(nullptr),
-    index_add_count(0),
-    index_add_buffer(nullptr),
-    index_altered_ignorability_count(0),
     rename_keys(current_thd->mem_root),
-    handler_ctx(nullptr),
-    group_commit_ctx(nullptr),
-    handler_flags(0),
     modified_part_info(modified_part_info_arg),
     ignore(ignore_arg),
-    online(false),
-    unsupported_reason(nullptr),
     error_if_not_empty(error_non_empty)
   {}
 
diff --git a/sql/handler.h b/sql/handler.h
index 029e981c038..78ca0f92a38 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -1546,6 +1546,37 @@ struct handlerton
 			    THD *victim_thd, my_bool signal);
    int (*set_checkpoint)(handlerton *hton, const XID* xid);
    int (*get_checkpoint)(handlerton *hton, XID* xid);
+  /**
+     Check if the version of the table matches the version in the .frm
+     file.
+
+     This is mainly used to verify in recovery to check if an inplace
+     ALTER TABLE succeded.
+     Storage engines that does not support inplace alter table does not
+     have to implement this function.
+
+     @param hton      handlerton
+     @param path      Path for table
+     @param version   The unique id that is stored in the .frm file for
+                      CREATE and updated for each ALTER TABLE (but not for
+                      simple renames).
+                      This is the ID used for the final table.
+     @param create_id The value returned from handler->table_version() for
+                      the original table (before ALTER TABLE).
+
+     @retval 0     If id matches or table is newer than create_id (depending
+                   on what version check the engine supports. This means that
+                   The (inplace) alter table did succeed.
+     @retval # > 0 Alter table did not succeed.
+
+     Related to handler::discover_check_version().
+   */
+  int (*check_version)(handlerton *hton, const char *path,
+                       const LEX_CUSTRING *version, ulonglong create_id);
+
+  /* Called for all storage handlers after ddl recovery is done */
+  void (*signal_ddl_recovery_done)(handlerton *hton);
+
    /*
      Optional clauses in the CREATE/ALTER TABLE
    */
@@ -1820,6 +1851,12 @@ handlerton *ha_default_tmp_handlerton(THD *thd);
 */
 #define HTON_REQUIRES_CLOSE_AFTER_TRUNCATE (1 << 18)
 
+/*
+  Used by mysql_inplace_alter_table() to decide if we should call
+  hton->notify_tabledef_changed() before commit (MyRocks) or after (InnoDB).
+*/
+#define HTON_REQUIRES_NOTIFY_TABLEDEF_CHANGED_AFTER_COMMIT (1 << 19)
+
 class Ha_trx_info;
 
 struct THD_TRANS
@@ -2485,27 +2522,27 @@ public:
   uint key_count;
 
   /** Size of index_drop_buffer array. */
-  uint index_drop_count;
+  uint index_drop_count= 0;
 
   /**
      Array of pointers to KEYs to be dropped belonging to the TABLE instance
      for the old version of the table.
   */
-  KEY  **index_drop_buffer;
+  KEY  **index_drop_buffer= nullptr;
 
   /** Size of index_add_buffer array. */
-  uint index_add_count;
+  uint index_add_count= 0;
 
   /**
      Array of indexes into key_info_buffer for KEYs to be added,
      sorted in increasing order.
   */
-  uint *index_add_buffer;
+  uint *index_add_buffer= nullptr;
 
-  KEY_PAIR  *index_altered_ignorability_buffer;
+  KEY_PAIR  *index_altered_ignorability_buffer= nullptr;
 
   /** Size of index_altered_ignorability_buffer array. */
-  uint index_altered_ignorability_count;
+  uint index_altered_ignorability_count= 0;
 
   /**
      Old and new index names. Used for index rename.
@@ -2536,7 +2573,7 @@ public:
 
      @see inplace_alter_handler_ctx for information about object lifecycle.
   */
-  inplace_alter_handler_ctx *handler_ctx;
+  inplace_alter_handler_ctx *handler_ctx= nullptr;
 
   /**
     If the table uses several handlers, like ha_partition uses one handler
@@ -2548,13 +2585,13 @@ public:
 
     @see inplace_alter_handler_ctx for information about object lifecycle.
   */
-  inplace_alter_handler_ctx **group_commit_ctx;
+  inplace_alter_handler_ctx **group_commit_ctx= nullptr;
 
   /**
      Flags describing in detail which operations the storage engine is to
      execute. Flags are defined in sql_alter.h
   */
-  alter_table_operations handler_flags;
+  alter_table_operations handler_flags= 0;
 
   /* Alter operations involving parititons are strored here */
   ulong partition_flags;
@@ -2565,13 +2602,24 @@ public:
      with partitions to be dropped or changed marked as such + all partitions
      to be added in the new version of table marked as such.
   */
-  partition_info *modified_part_info;
+  partition_info * const modified_part_info;
 
   /** true for ALTER IGNORE TABLE ... */
   const bool ignore;
 
   /** true for online operation (LOCK=NONE) */
-  bool online;
+  bool online= false;
+
+  /**
+    When ha_commit_inplace_alter_table() is called the the engine can
+    set this to a function to be called after the ddl log
+    is committed.
+  */
+  typedef void (inplace_alter_table_commit_callback)(void *);
+  inplace_alter_table_commit_callback *inplace_alter_table_committed= nullptr;
+
+  /* This will be used as the argument to the above function when called */
+  void *inplace_alter_table_committed_argument= nullptr;
 
   /** which ALGORITHM and LOCK are supported by the storage engine */
   enum_alter_inplace_result inplace_supported;
@@ -2588,10 +2636,10 @@ public:
      Please set to a properly localized string, for example using
      my_get_err_msg(), so that the error message as a whole is localized.
   */
-  const char *unsupported_reason;
+  const char *unsupported_reason= nullptr;
 
   /** true when InnoDB should abort the alter when table is not empty */
-  bool error_if_not_empty;
+  const bool error_if_not_empty;
 
   Alter_inplace_info(HA_CREATE_INFO *create_info_arg,
                      Alter_info *alter_info_arg,
@@ -4044,6 +4092,15 @@ public:
   { return 0; }
   virtual int extra_opt(enum ha_extra_function operation, ulong arg)
   { return extra(operation); }
+  /*
+    Table version id for the the table. This should change for each
+    sucessfull ALTER TABLE.
+    This is used by the handlerton->check_version() to ask the engine
+    if the table definition has been updated.
+    Storage engines that does not support inplace alter table does not
+    have to support this call.
+  */
+  virtual ulonglong table_version() const { return 0; }
 
   /**
     In an UPDATE or DELETE, if the row under the cursor was locked by another
@@ -5179,6 +5236,7 @@ TYPELIB *ha_known_exts(void);
 int ha_panic(enum ha_panic_function flag);
 void ha_close_connection(THD* thd);
 void ha_kill_query(THD* thd, enum thd_kill_levels level);
+void ha_signal_ddl_recovery_done();
 bool ha_flush_logs();
 void ha_drop_database(const char* path);
 void ha_checkpoint_state(bool disable);
diff --git a/sql/item_func.cc b/sql/item_func.cc
index c31a60304ad..965809ff460 100644
--- a/sql/item_func.cc
+++ b/sql/item_func.cc
@@ -6863,14 +6863,18 @@ void uuid_short_init()
                (((ulonglong) server_start_time) << 24));
 }
 
-
-longlong Item_func_uuid_short::val_int()
+ulonglong server_uuid_value()
 {
   ulonglong val;
   mysql_mutex_lock(&LOCK_short_uuid_generator);
   val= uuid_value++;
   mysql_mutex_unlock(&LOCK_short_uuid_generator);
-  return (longlong) val;
+  return val;
+}
+
+longlong Item_func_uuid_short::val_int()
+{
+  return (longlong) server_uuid_value();
 }
 
 
diff --git a/sql/item_func.h b/sql/item_func.h
index 560a9fa898e..3293a5893a3 100644
--- a/sql/item_func.h
+++ b/sql/item_func.h
@@ -4046,6 +4046,7 @@ public:
 
 
 void uuid_short_init();
+ulonglong server_uuid_value();
 
 class Item_func_uuid_short :public Item_longlong_func
 {
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index 9cc67419564..dff18b9585e 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -5313,6 +5313,10 @@ static int init_server_components()
   }
 #endif
 
+  if (ddl_log_execute_recovery() > 0)
+    unireg_abort(1);
+  ha_signal_ddl_recovery_done();
+
   if (opt_myisam_log)
     (void) mi_log(1);
 
@@ -5681,9 +5685,6 @@ int mysqld_main(int argc, char **argv)
 
   initialize_information_schema_acl();
 
-  if (ddl_log_execute_recovery() > 0)
-    unireg_abort(1);
-
   /*
     Change EVENTS_ORIGINAL to EVENTS_OFF (the default value) as there is no
     point in using ORIGINAL during startup
diff --git a/sql/share/errmsg-utf8.txt b/sql/share/errmsg-utf8.txt
index 63dc9cf024c..ed68576b5b5 100644
--- a/sql/share/errmsg-utf8.txt
+++ b/sql/share/errmsg-utf8.txt
@@ -7990,3 +7990,5 @@ ER_WITH_TIES_NEEDS_ORDER
         eng "FETCH ... WITH TIES requires ORDER BY clause to be present"
 ER_REMOVED_ORPHAN_TRIGGER
         eng "Dropped orphan trigger '%-.64s', originally created for table: '%-.192s'"
+ER_STORAGE_ENGINE_DISABLED
+        eng "Storage engine %s is disabled"
diff --git a/sql/sql_alter.h b/sql/sql_alter.h
index b922cbf0637..b8aa2595a58 100644
--- a/sql/sql_alter.h
+++ b/sql/sql_alter.h
@@ -324,6 +324,7 @@ public:
   LEX_CSTRING  new_alias;
   LEX_CSTRING  tmp_name;
   char         tmp_buff[80];
+
   /**
     Indicates that if a row is deleted during copying of data from old version
     of table to the new version ER_FK_CANNOT_DELETE_PARENT error should be
diff --git a/sql/sql_partition.cc b/sql/sql_partition.cc
index 22efecb9210..07926712583 100644
--- a/sql/sql_partition.cc
+++ b/sql/sql_partition.cc
@@ -6168,8 +6168,9 @@ static void release_part_info_log_entries(DDL_LOG_MEMORY_ENTRY *log_entry)
 
   while (log_entry)
   {
+    DDL_LOG_MEMORY_ENTRY *next= log_entry->next_active_log_entry;
     ddl_log_release_memory_entry(log_entry);
-    log_entry= log_entry->next_active_log_entry;
+    log_entry= next;
   }
   DBUG_VOID_RETURN;
 }
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 7b3b0eb1136..95337d641a5 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -4900,7 +4900,7 @@ mysql_rename_table(handlerton *base, const LEX_CSTRING *old_db,
     DBUG_RETURN(TRUE);
   }
 
-  if (file->needs_lower_case_filenames())
+  if (file && file->needs_lower_case_filenames())
   {
     build_lower_case_table_filename(lc_from, sizeof(lc_from) -1,
                                     old_db, old_name, flags & FN_FROM_IS_TMP);
@@ -7023,6 +7023,41 @@ static bool is_inplace_alter_impossible(TABLE *table,
 }
 
 
+/*
+  Notify engine that table definition has changed as part of inplace alter
+  table
+*/
+
+static bool notify_tabledef_changed(TABLE_LIST *table_list)
+{
+  TABLE *table= table_list->table;
+  DBUG_ENTER("notify_tabledef_changed");
+
+  if (table->file->partition_ht()->notify_tabledef_changed)
+  {
+    char db_buff[FN_REFLEN], table_buff[FN_REFLEN];
+    handlerton *hton= table->file->ht;
+    LEX_CSTRING tmp_db, tmp_table;
+
+    tmp_db.str=       db_buff;
+    tmp_table.str=    table_buff;
+    tmp_db.length=    tablename_to_filename(table_list->db.str,
+                                            db_buff, sizeof(db_buff));
+    tmp_table.length= tablename_to_filename(table_list->table_name.str,
+                                            table_buff, sizeof(table_buff));
+    if ((hton->notify_tabledef_changed)(hton, &tmp_db, &tmp_table,
+                                        table->s->frm_image,
+                                        &table->s->tabledef_version,
+                                        table->file))
+    {
+      my_error(HA_ERR_INCOMPATIBLE_DEFINITION, MYF(0));
+      DBUG_RETURN(true);
+    }
+  }
+  DBUG_RETURN(0);
+}
+
+
 /**
   Perform in-place alter table.
 
@@ -7058,6 +7093,7 @@ static bool mysql_inplace_alter_table(THD *thd,
                                       TABLE *altered_table,
                                       Alter_inplace_info *ha_alter_info,
                                       MDL_request *target_mdl_request,
+                                      DDL_LOG_STATE *ddl_log_state,
                                       TRIGGER_RENAME_PARAM *trigger_param,
                                       Alter_table_ctx *alter_ctx)
 {
@@ -7065,7 +7101,7 @@ static bool mysql_inplace_alter_table(THD *thd,
   handlerton *db_type= table->s->db_type();
   Alter_info *alter_info= ha_alter_info->alter_info;
   bool reopen_tables= false;
-  bool res;
+  bool res, commit_succeded_with_error= 0;
   const enum_alter_inplace_result inplace_supported=
     ha_alter_info->inplace_supported;
   DBUG_ENTER("mysql_inplace_alter_table");
@@ -7178,10 +7214,27 @@ static bool mysql_inplace_alter_table(THD *thd,
     break;
   }
 
+  ddl_log_update_phase(ddl_log_state, DDL_ALTER_TABLE_PHASE_PREPARE_INPLACE);
+
   if (table->file->ha_prepare_inplace_alter_table(altered_table,
                                                   ha_alter_info))
     goto rollback;
 
+  debug_crash_here("ddl_log_alter_after_prepare_inplace");
+
+  /*
+    Store the new table_version() as it may have not been available before
+    in some engines, like InnoDB.
+  */
+  ddl_log_update_unique_id(ddl_log_state,
+                           table->file->table_version());
+  /*
+    Mark that we have started inplace alter table. DDL recover will
+    have to decide if it should use the old or new version of the table, based
+    on if the new version did commit or not.
+  */
+  ddl_log_update_phase(ddl_log_state, DDL_ALTER_TABLE_PHASE_INPLACE);
+
   /*
     Downgrade the lock if storage engine has told us that exclusive lock was
     necessary only for prepare phase (unless we are not under LOCK TABLES) and
@@ -7227,12 +7280,17 @@ static bool mysql_inplace_alter_table(THD *thd,
   if (backup_reset_alter_copy_lock(thd))
     goto rollback;
 
+  /* Crashing here should cause the original table to be used */
+  debug_crash_here("ddl_log_alter_after_copy");
   /*
     If we are killed after this point, we should ignore and continue.
     We have mostly completed the operation at this point, there should
     be no long waits left.
   */
 
+  DEBUG_SYNC(thd, "alter_table_inplace_before_commit");
+  THD_STAGE_INFO(thd, stage_alter_inplace_commit);
+
   DBUG_EXECUTE_IF("alter_table_rollback_new_index", {
       table->file->ha_commit_inplace_alter_table(altered_table,
                                                  ha_alter_info,
@@ -7241,8 +7299,15 @@ static bool mysql_inplace_alter_table(THD *thd,
       goto cleanup;
     });
 
-  DEBUG_SYNC(thd, "alter_table_inplace_before_commit");
-  THD_STAGE_INFO(thd, stage_alter_inplace_commit);
+  /*
+    Notify the engine that the table definition has changed so that it can
+    store the new ID as part of the commit
+  */
+
+  if (!(table->file->partition_ht()->flags &
+        HTON_REQUIRES_NOTIFY_TABLEDEF_CHANGED_AFTER_COMMIT) &&
+      notify_tabledef_changed(table_list))
+    goto rollback;
 
   {
     TR_table trt(thd, true);
@@ -7269,28 +7334,26 @@ static bool mysql_inplace_alter_table(THD *thd,
     DEBUG_SYNC(thd, "alter_table_inplace_after_commit");
   }
 
-  /* Notify the engine that the table definition has changed */
+  /*
+    We are new ready to use the new table. Update the state in the
+    ddl log so that we recovery know that the new table is ready and
+    in case of crash it should use the new one and log the query
+    to the binary log.
+  */
+  ddl_log_update_phase(ddl_log_state, DDL_ALTER_TABLE_PHASE_INPLACE_COPIED);
+  debug_crash_here("ddl_log_alter_after_log");
 
-  if (table->file->partition_ht()->notify_tabledef_changed)
+  if ((table->file->partition_ht()->flags &
+       HTON_REQUIRES_NOTIFY_TABLEDEF_CHANGED_AFTER_COMMIT) &&
+      notify_tabledef_changed(table_list))
   {
-    char db_buff[FN_REFLEN], table_buff[FN_REFLEN];
-    handlerton *hton= table->file->ht;
-    LEX_CSTRING tmp_db, tmp_table;
-
-    tmp_db.str=       db_buff;
-    tmp_table.str=    table_buff;
-    tmp_db.length=    tablename_to_filename(table_list->db.str,
-                                         db_buff, sizeof(db_buff));
-    tmp_table.length= tablename_to_filename(table_list->table_name.str,
-                                            table_buff, sizeof(table_buff));
-    if ((hton->notify_tabledef_changed)(hton, &tmp_db, &tmp_table,
-                                        table->s->frm_image,
-                                        &table->s->tabledef_version,
-                                        table->file))
-    {
-      my_error(HA_ERR_INCOMPATIBLE_DEFINITION, MYF(0));
-      DBUG_RETURN(true);
-    }
+    /*
+      The above should never fail. If it failed, the new structure is
+      commited and we have no way to roll back.
+      The best we can do is to continue, but send an error to the
+      user that something when wrong
+    */
+    commit_succeded_with_error= 1;
   }
 
   close_all_tables_for_name(thd, table->s,
@@ -7303,7 +7366,8 @@ static bool mysql_inplace_alter_table(THD *thd,
   /*
     Replace the old .FRM with the new .FRM, but keep the old name for now.
     Rename to the new name (if needed) will be handled separately below.
-
+  */
+  /*
     TODO: remove this check of thd->is_error() (now it intercept
     errors in some val_*() methods and bring some single place to
     such error interception).
@@ -7316,6 +7380,7 @@ static bool mysql_inplace_alter_table(THD *thd,
     // Since changes were done in-place, we can't revert them.
     DBUG_RETURN(true);
   }
+  debug_crash_here("ddl_log_alter_after_rename_frm");
 
   // Rename altered table in case of ALTER TABLE ... RENAME
   if (alter_ctx->is_table_renamed())
@@ -7331,6 +7396,7 @@ static bool mysql_inplace_alter_table(THD *thd,
       */
       DBUG_RETURN(true);
     }
+    debug_crash_here("ddl_log_alter_before_rename_triggers");
     if (Table_triggers_list::change_table_name(thd, trigger_param,
                                                &alter_ctx->db,
                                                &alter_ctx->alias,
@@ -7346,13 +7412,15 @@ static bool mysql_inplace_alter_table(THD *thd,
                                 &alter_ctx->new_db, &alter_ctx->new_alias,
                                 &alter_ctx->db, &alter_ctx->alias,
                                 NO_FK_CHECKS);
+      ddl_log_disable_entry(ddl_log_state);
       DBUG_RETURN(true);
     }
     rename_table_in_stat_tables(thd, &alter_ctx->db, &alter_ctx->alias,
                                 &alter_ctx->new_db, &alter_ctx->new_alias);
+    debug_crash_here("ddl_log_alter_after_rename_triggers");
   }
 
-  DBUG_RETURN(false);
+  DBUG_RETURN(commit_succeded_with_error);
 
  rollback:
   table->file->ha_commit_inplace_alter_table(altered_table,
@@ -8846,6 +8914,13 @@ simple_tmp_rename_or_index_change(THD *thd, TABLE_LIST *table_list,
   @return Operation status
     @retval false           Success
     @retval true            Failure
+
+  @notes
+    Normally with ALTER TABLE we roll forward as soon as data is copied
+    or new table is committed.  For an ALTER TABLE that only does a RENAME,
+    we will roll back unless the RENAME fully completes.
+    If we crash while using enable/disable keys, this may have completed
+    and will not be rolled back.
 */
 
 static bool
@@ -8856,11 +8931,13 @@ simple_rename_or_index_change(THD *thd, TABLE_LIST *table_list,
 {
   TABLE *table= table_list->table;
   MDL_ticket *mdl_ticket= table->mdl_ticket;
+  DDL_LOG_STATE ddl_log_state;
   int error= 0;
   enum ha_extra_function extra_func= thd->locked_tables_mode
                                        ? HA_EXTRA_NOT_USED
                                        : HA_EXTRA_FORCE_REOPEN;
   DBUG_ENTER("simple_rename_or_index_change");
+  bzero(&ddl_log_state, sizeof(ddl_log_state));
 
   if (keys_onoff != Alter_info::LEAVE_AS_IS)
   {
@@ -8895,37 +8972,52 @@ simple_rename_or_index_change(THD *thd, TABLE_LIST *table_list,
     close_all_tables_for_name(thd, table->s, HA_EXTRA_PREPARE_FOR_RENAME,
                               NULL);
 
+    (void) ddl_log_rename_table(thd, &ddl_log_state, old_db_type,
+                                &alter_ctx->db, &alter_ctx->table_name,
+                                &alter_ctx->new_db, &alter_ctx->new_alias);
     if (mysql_rename_table(old_db_type, &alter_ctx->db, &alter_ctx->table_name,
                            &alter_ctx->new_db, &alter_ctx->new_alias, 0))
       error= -1;
-    else if (Table_triggers_list::change_table_name(thd, trigger_param,
-                                                 &alter_ctx->db,
-                                                 &alter_ctx->alias,
-                                                 &alter_ctx->table_name,
-                                                 &alter_ctx->new_db,
-                                                 &alter_ctx->new_alias))
+    if (!error)
+      ddl_log_update_phase(&ddl_log_state, DDL_RENAME_PHASE_TRIGGER);
+    debug_crash_here("ddl_log_alter_before_rename_triggers");
+    if (!error &&
+        Table_triggers_list::change_table_name(thd, trigger_param,
+                                               &alter_ctx->db,
+                                               &alter_ctx->alias,
+                                               &alter_ctx->table_name,
+                                               &alter_ctx->new_db,
+                                               &alter_ctx->new_alias))
     {
       (void) mysql_rename_table(old_db_type,
                                 &alter_ctx->new_db, &alter_ctx->new_alias,
                                 &alter_ctx->db, &alter_ctx->table_name,
                                 NO_FK_CHECKS);
+      ddl_log_disable_entry(&ddl_log_state);
       error= -1;
     }
-    /* Update stat tables last. This is to be able to handle rename of a stat table */
+    /*
+      Update stat tables last. This is to be able to handle rename of
+      a stat table.
+    */
     if (error == 0)
       (void) rename_table_in_stat_tables(thd, &alter_ctx->db,
                                          &alter_ctx->table_name,
                                          &alter_ctx->new_db,
                                          &alter_ctx->new_alias);
+    debug_crash_here("ddl_log_alter_after_rename_triggers");
   }
 
   if (likely(!error))
   {
+    thd->binlog_xid= thd->query_id;
+    ddl_log_update_xid(&ddl_log_state, thd->binlog_xid);
     error= write_bin_log(thd, TRUE, thd->query(), thd->query_length());
-
+    thd->binlog_xid= 0;
     if (likely(!error))
       my_ok(thd);
   }
+  ddl_log_complete(&ddl_log_state);
   table_list->table= NULL;                    // For query cache
   query_cache_invalidate3(thd, table_list, 0);
 
@@ -9054,6 +9146,8 @@ bool mysql_alter_table(THD *thd, const LEX_CSTRING *new_db,
   bool engine_changed, error, frm_is_created= false;
   bool no_ha_table= true;  /* We have not created table in storage engine yet */
   TABLE *table, *new_table;
+  DDL_LOG_STATE ddl_log_state;
+
 #ifdef WITH_PARTITION_STORAGE_ENGINE
   bool partition_changed= false;
   bool fast_alter_partition= false;
@@ -9076,14 +9170,24 @@ bool mysql_alter_table(THD *thd, const LEX_CSTRING *new_db,
   bool varchar= create_info->varchar, table_creation_was_logged= 0;
   bool binlog_as_create_select= 0, log_if_exists= 0;
   uint tables_opened;
-  handlerton *new_db_type, *old_db_type;
+  handlerton *new_db_type, *old_db_type= nullptr;
   ha_rows copied=0, deleted=0;
   LEX_CUSTRING frm= {0,0};
-  char index_file[FN_REFLEN], data_file[FN_REFLEN];
+  LEX_CSTRING backup_name;
+  char index_file[FN_REFLEN], data_file[FN_REFLEN], backup_name_buff[60];
+  uchar uuid_buffer[MY_UUID_SIZE];
   MDL_request target_mdl_request;
   MDL_ticket *mdl_ticket= 0;
   Alter_table_prelocking_strategy alter_prelocking_strategy;
   TRIGGER_RENAME_PARAM trigger_param;
+
+  /*
+    Callback function that an engine can request to be called after executing
+    inplace alter table.
+  */
+  Alter_inplace_info::inplace_alter_table_commit_callback
+    *inplace_alter_table_committed= 0;
+  void *inplace_alter_table_committed_argument= 0;
   DBUG_ENTER("mysql_alter_table");
 
   /*
@@ -9128,6 +9232,13 @@ bool mysql_alter_table(THD *thd, const LEX_CSTRING *new_db,
   }
 
   THD_STAGE_INFO(thd, stage_init_update);
+  bzero(&ddl_log_state, sizeof(ddl_log_state));
+
+  /* Temporary name for backup of original table */
+  backup_name.str= backup_name_buff;
+  backup_name.length= my_snprintf(backup_name_buff, sizeof(backup_name_buff)-1,
+                                  "%s-backup-%lx-%llx", tmp_file_prefix,
+                                  current_pid, thd->thread_id);
 
   /* Check if the new table type is a shared table */
   if (ha_check_if_updates_are_ignored(thd, create_info->db_type, "ALTER"))
@@ -9752,9 +9863,47 @@ do_continue:;
 
   DEBUG_SYNC(thd, "alter_table_before_create_table_no_lock");
 
+  /* Create a new table version id for the new table */
+  my_uuid(uuid_buffer);
+  create_info->tabledef_version.str= uuid_buffer;
+  create_info->tabledef_version.length= MY_UUID_SIZE;
+
+  if (!table->s->tmp_table)
+  {
+    LEX_CSTRING path_to_frm= alter_ctx.get_tmp_cstring_path();
+    LEX_CSTRING tmp_table= backup_name;
+    if (alter_ctx.is_table_renamed())
+      tmp_table= alter_ctx.new_alias;
+
+    if (ddl_log_alter_table(thd, &ddl_log_state,
+                            old_db_type,
+                            &alter_ctx.db, &alter_ctx.table_name,
+                            new_db_type,
+                            table->file->partition_ht(),
+                            &alter_ctx.new_db, &alter_ctx.tmp_name,
+                            &path_to_frm,
+                            &tmp_table,
+                            &create_info->tabledef_version,
+                            table->file->table_version(),
+                            alter_ctx.is_table_renamed()) ||
+        ddl_log_store_query(thd, &ddl_log_state,
+                            thd->query(), thd->query_length()))
+    {
+      error= 1;
+      goto err_cleanup;
+    }
+  }
+
   tmp_disable_binlog(thd);
   create_info->options|=HA_CREATE_TMP_ALTER;
   create_info->alias= alter_ctx.table_name;
+  /*
+    Create the .frm file for the new table. Storage engine table will not be
+    created at this stage.
+
+    No ddl logging needed as ddl_log_alter_query will take care of failed
+    table creations.
+  */
   error= create_table_impl(thd, (DDL_LOG_STATE*) 0, (DDL_LOG_STATE*) 0,
                            alter_ctx.db, alter_ctx.table_name,
                            alter_ctx.new_db, alter_ctx.tmp_name,
@@ -9763,11 +9912,11 @@ do_continue:;
                            C_ALTER_TABLE_FRM_ONLY, NULL,
                            &key_info, &key_count, &frm);
   reenable_binlog(thd);
+
+  debug_crash_here("ddl_log_alter_after_create_frm");
+
   if (unlikely(error))
-  {
-    my_free(const_cast<uchar*>(frm.str));
-    DBUG_RETURN(true);
-  }
+    goto err_cleanup;
 
   if (alter_info->algorithm(thd) != Alter_info::ALTER_TABLE_ALGORITHM_COPY)
   {
@@ -9809,7 +9958,6 @@ do_continue:;
       */
       table->file->ha_create_partitioning_metadata(alter_ctx.get_tmp_path(),
                                                    NULL, CHF_DELETE_FLAG);
-      my_free(const_cast<uchar*>(frm.str));
       goto end_inplace;
     }
 
@@ -9894,19 +10042,19 @@ do_continue:;
       */
       enum_check_fields org_count_cuted_fields= thd->count_cuted_fields;
       thd->count_cuted_fields= CHECK_FIELD_WARN;
-      int res= mysql_inplace_alter_table(thd,
-                                         table_list, table, &altered_table,
+      int res= mysql_inplace_alter_table(thd, table_list, table, &altered_table,
                                          &ha_alter_info,
-                                         &target_mdl_request,
+                                         &target_mdl_request, &ddl_log_state,
                                          &trigger_param,
                                          &alter_ctx);
       thd->count_cuted_fields= org_count_cuted_fields;
-      my_free(const_cast<uchar*>(frm.str));
-
+      inplace_alter_table_committed= ha_alter_info.inplace_alter_table_committed;
+      inplace_alter_table_committed_argument=
+        ha_alter_info.inplace_alter_table_committed_argument;
       if (res)
       {
         cleanup_table_after_inplace_alter(&altered_table);
-        DBUG_RETURN(true);
+        goto err_cleanup;
       }
       cleanup_table_after_inplace_alter_keep_files(&altered_table);
 
@@ -9958,11 +10106,15 @@ do_continue:;
                   MYSQL_LOCK_USE_MALLOC))
     goto err_new_table_cleanup;
 
+  ddl_log_update_phase(&ddl_log_state, DDL_ALTER_TABLE_PHASE_CREATED);
+
   if (ha_create_table(thd, alter_ctx.get_tmp_path(),
                       alter_ctx.new_db.str, alter_ctx.new_name.str,
                       create_info, &frm, frm_is_created))
     goto err_new_table_cleanup;
 
+  debug_crash_here("ddl_log_alter_after_create_table");
+
   /* Mark that we have created table in storage engine. */
   no_ha_table= false;
   DEBUG_SYNC(thd, "alter_table_intermediate_table_created");
@@ -10107,10 +10259,16 @@ do_continue:;
     /* We don't replicate alter table statement on temporary tables */
     if (!thd->is_current_stmt_binlog_format_row() &&
         table_creation_was_logged &&
-        !binlog_as_create_select &&
-        write_bin_log_with_if_exists(thd, true, false, log_if_exists))
-      DBUG_RETURN(true);
-    my_free(const_cast<uchar*>(frm.str));
+        !binlog_as_create_select)
+    {
+      int tmp_error;
+      thd->binlog_xid= thd->query_id;
+      ddl_log_update_xid(&ddl_log_state, thd->binlog_xid);
+      tmp_error= write_bin_log_with_if_exists(thd, true, false, log_if_exists);
+      thd->binlog_xid= 0;
+      if (tmp_error)
+        goto err_cleanup;
+    }
     goto end_temporary;
   }
 
@@ -10157,6 +10315,18 @@ do_continue:;
       (mysql_execute_command()) to release metadata locks.
   */
 
+  debug_crash_here("ddl_log_alter_after_copy");      // Use old table
+  /*
+    We are new ready to use the new table. Update the state in the
+    ddl log so that we recovery know that the new table is ready and
+    in case of crash it should use the new one and log the query
+    to the binary log.
+  */
+  if (engine_changed)
+    ddl_log_add_flag(&ddl_log_state, DDL_LOG_FLAG_ALTER_ENGINE_CHANGED);
+  ddl_log_update_phase(&ddl_log_state, DDL_ALTER_TABLE_PHASE_COPIED);
+  debug_crash_here("ddl_log_alter_after_log");       // Use new table
+
   THD_STAGE_INFO(thd, stage_rename_result_table);
 
   if (wait_while_table_is_used(thd, table, HA_EXTRA_PREPARE_FOR_RENAME))
@@ -10168,29 +10338,20 @@ do_continue:;
                             HA_EXTRA_NOT_USED,
                             NULL);
   table_list->table= table= NULL;                  /* Safety */
-  my_free(const_cast<uchar*>(frm.str));
-
-  /*
-    Rename the old table to temporary name to have a backup in case
-    anything goes wrong while renaming the new table.
-    We only have to do this if name of the table is not changed.
-    If we are changing to use another table handler, we don't
-    have to do the rename as the table names will not interfer.
-  */
-  char backup_name_buff[FN_LEN];
-  LEX_CSTRING backup_name;
-  backup_name.str= backup_name_buff;
 
   DBUG_PRINT("info", ("is_table_renamed: %d  engine_changed: %d",
                       alter_ctx.is_table_renamed(), engine_changed));
 
   if (!alter_ctx.is_table_renamed())
   {
-    backup_name.length= my_snprintf(backup_name_buff, sizeof(backup_name_buff),
-                                    "%s-backup-%lx-%llx", tmp_file_prefix,
-                                    current_pid, thd->thread_id);
-    if (lower_case_table_names)
-      my_casedn_str(files_charset_info, backup_name_buff);
+    /*
+      Rename the old table to temporary name to have a backup in case
+      anything goes wrong while renaming the new table.
+
+      We only have to do this if name of the table is not changed.
+      If we are changing to use another table handler, we don't
+      have to do the rename as the table names will not interfer.
+    */
     if (mysql_rename_table(old_db_type, &alter_ctx.db, &alter_ctx.table_name,
                            &alter_ctx.db, &backup_name,
                            FN_TO_IS_TMP |
@@ -10209,6 +10370,18 @@ do_continue:;
     PSI_CALL_drop_table_share(0, alter_ctx.db.str, (int) alter_ctx.db.length,
                               alter_ctx.table_name.str, (int) alter_ctx.table_name.length);
   }
+  debug_crash_here("ddl_log_alter_after_rename_to_backup");
+
+  if (!alter_ctx.is_table_renamed())
+  {
+    /*
+      We should not set this stage in case of rename as we in this case
+      must execute DDL_ALTER_TABLE_PHASE_COPIED to remove the orignal table
+    */
+    ddl_log_update_phase(&ddl_log_state, DDL_ALTER_TABLE_PHASE_OLD_RENAMED);
+  }
+
+  debug_crash_here("ddl_log_alter_after_rename_to_backup_log");
 
   // Rename the new table to the correct name.
   if (mysql_rename_table(new_db_type, &alter_ctx.new_db, &alter_ctx.tmp_name,
@@ -10216,6 +10389,7 @@ do_continue:;
                          FN_FROM_IS_TMP))
   {
     // Rename failed, delete the temporary table.
+    ddl_log_update_phase(&ddl_log_state, DDL_ALTER_TABLE_PHASE_RENAME_FAILED);
     (void) quick_rm_table(thd, new_db_type, &alter_ctx.new_db,
                           &alter_ctx.tmp_name, FN_IS_TMP);
 
@@ -10230,10 +10404,12 @@ do_continue:;
     }
     goto err_with_mdl;
   }
+  debug_crash_here("ddl_log_alter_after_rename_to_original");
 
   // Check if we renamed the table and if so update trigger files.
   if (alter_ctx.is_table_renamed())
   {
+    debug_crash_here("ddl_log_alter_before_rename_triggers");
     if (Table_triggers_list::change_table_name(thd, &trigger_param,
                                                &alter_ctx.db,
                                                &alter_ctx.alias,
@@ -10254,12 +10430,15 @@ do_continue:;
     }
     rename_table_in_stat_tables(thd, &alter_ctx.db, &alter_ctx.alias,
                                 &alter_ctx.new_db, &alter_ctx.new_alias);
+    debug_crash_here("ddl_log_alter_after_rename_triggers");
   }
 
   // ALTER TABLE succeeded, delete the backup of the old table.
   error= quick_rm_table(thd, old_db_type, &alter_ctx.db, &backup_name,
                         FN_IS_TMP |
                         (engine_changed ? NO_HA_TABLE | NO_PAR_TABLE: 0));
+
+  debug_crash_here("ddl_log_alter_after_delete_backup");
   if (engine_changed)
   {
     /* the .frm file was removed but not the original table */
@@ -10268,6 +10447,7 @@ do_continue:;
                            NO_FRM_RENAME |
                            (engine_changed ? 0 : FN_IS_TMP));
   }
+  debug_crash_here("ddl_log_alter_after_drop_original_table");
   if (binlog_as_create_select)
   {
     /*
@@ -10275,7 +10455,10 @@ do_continue:;
       DROP + CREATE + data statement to the binary log
     */
     thd->variables.option_bits&= ~OPTION_BIN_COMMIT_OFF;
+    thd->binlog_xid= thd->query_id;
+    ddl_log_update_xid(&ddl_log_state, thd->binlog_xid);
     binlog_hton->commit(binlog_hton, thd, 1);
+    thd->binlog_xid= 0;
   }
 
   if (error)
@@ -10295,7 +10478,6 @@ end_inplace:
     goto err_with_mdl_after_alter;
 
   THD_STAGE_INFO(thd, stage_end);
-
   DEBUG_SYNC(thd, "alter_table_before_main_binlog");
 
   DBUG_ASSERT(!(mysql_bin_log.is_open() &&
@@ -10303,9 +10485,27 @@ end_inplace:
                 (create_info->tmp_table())));
   if (!binlog_as_create_select)
   {
-    if (write_bin_log_with_if_exists(thd, true, false, log_if_exists))
-      DBUG_RETURN(true);
+    int tmp_error;
+    thd->binlog_xid= thd->query_id;
+    ddl_log_update_xid(&ddl_log_state, thd->binlog_xid);
+    tmp_error= write_bin_log_with_if_exists(thd, true, false, log_if_exists);
+    thd->binlog_xid= 0;
+    if (tmp_error)
+      goto err_cleanup;
   }
+
+  /*
+    We have to close the ddl log as soon as possible, after binlogging the
+    query, for inplace alter table.
+  */
+  ddl_log_complete(&ddl_log_state);
+  if (inplace_alter_table_committed)
+  {
+    /* Signal to storage engine that ddl log is committed */
+    (*inplace_alter_table_committed)(inplace_alter_table_committed_argument);
+    inplace_alter_table_committed= 0;
+  }
+
   table_list->table= NULL;			// For query cache
   query_cache_invalidate3(thd, table_list, false);
 
@@ -10319,6 +10519,8 @@ end_inplace:
   }
 
 end_temporary:
+  my_free(const_cast<uchar*>(frm.str));
+
   thd->variables.option_bits&= ~OPTION_BIN_COMMIT_OFF;
 
   my_snprintf(alter_ctx.tmp_buff, sizeof(alter_ctx.tmp_buff),
@@ -10333,7 +10535,6 @@ err_new_table_cleanup:
   DBUG_PRINT("error", ("err_new_table_cleanup"));
   thd->variables.option_bits&= ~OPTION_BIN_COMMIT_OFF;
 
-  my_free(const_cast<uchar*>(frm.str));
   /*
     No default value was provided for a DATE/DATETIME field, the
     current sql_mode doesn't allow the '0000-00-00' value and
@@ -10358,6 +10559,14 @@ err_new_table_cleanup:
                           (FN_IS_TMP | (no_ha_table ? NO_HA_TABLE : 0)),
                           alter_ctx.get_tmp_path());
 
+err_cleanup:
+  my_free(const_cast<uchar*>(frm.str));
+  ddl_log_complete(&ddl_log_state);
+  if (inplace_alter_table_committed)
+  {
+    /* Signal to storage engine that ddl log is committed */
+    (*inplace_alter_table_committed)(inplace_alter_table_committed_argument);
+  }
   DBUG_RETURN(true);
 
 err_with_mdl_after_alter:
@@ -10374,6 +10583,7 @@ err_with_mdl_after_alter:
     write_bin_log_with_if_exists(thd, FALSE, FALSE, log_if_exists);
 
 err_with_mdl:
+  ddl_log_complete(&ddl_log_state);
   /*
     An error happened while we were holding exclusive name metadata lock
     on table being altered. To be safe under LOCK TABLES we should
@@ -10383,7 +10593,7 @@ err_with_mdl:
   thd->locked_tables_list.unlink_all_closed_tables(thd, NULL, 0);
   if (!table_list->table)
     thd->mdl_context.release_all_locks_for_name(mdl_ticket);
-  DBUG_RETURN(true);
+  goto err_cleanup;
 }