Review and crash-safety fix

9 files changed, 373 insertions, 96 deletions

diff --git a/mysql-test/suite/atomic/alter_partition,innodb.rdiff b/mysql-test/suite/atomic/alter_partition,innodb.rdiff
index f11c1ef833c..3b4d51a3833 100644
--- a/mysql-test/suite/atomic/alter_partition,innodb.rdiff
+++ b/mysql-test/suite/atomic/alter_partition,innodb.rdiff
@@ -195,3 +195,41 @@
  master-bin.000001	#	Query	#	#	use `test`; ALTER TABLE t1 CONVERT PARTITION p1 TO TABLE tp1
  Table	Create Table
  t1	CREATE TABLE `t1` (
+@@ -270,15 +234,12 @@
+ x
+ 12
+ # CRASH: crash_convert_partition_10
+-t1#P#p0.MYD
+-t1#P#p0.MYI
+-t1#P#pn.MYD
+-t1#P#pn.MYI
++t1#P#p0.ibd
++t1#P#pn.ibd
+ t1.frm
+ t1.par
+-tp1.MYD
+-tp1.MYI
+ tp1.frm
++tp1.ibd
+ master-bin.000001	#	Query	#	#	use `test`; ALTER TABLE t1 CONVERT PARTITION p1 TO TABLE tp1
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+@@ -297,15 +258,12 @@
+ x
+ 12
+ # CRASH: crash_convert_partition_11
+-t1#P#p0.MYD
+-t1#P#p0.MYI
+-t1#P#pn.MYD
+-t1#P#pn.MYI
++t1#P#p0.ibd
++t1#P#pn.ibd
+ t1.frm
+ t1.par
+-tp1.MYD
+-tp1.MYI
+ tp1.frm
++tp1.ibd
+ master-bin.000001	#	Query	#	#	use `test`; ALTER TABLE t1 CONVERT PARTITION p1 TO TABLE tp1
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
diff --git a/mysql-test/suite/atomic/alter_partition.result b/mysql-test/suite/atomic/alter_partition.result
index ab4bc80af51..d9055f14aa8 100644
--- a/mysql-test/suite/atomic/alter_partition.result
+++ b/mysql-test/suite/atomic/alter_partition.result
@@ -269,5 +269,59 @@ tp1	CREATE TABLE `tp1` (
 ) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
 x
 12
+# CRASH: crash_convert_partition_10
+t1#P#p0.MYD
+t1#P#p0.MYI
+t1#P#pn.MYD
+t1#P#pn.MYI
+t1.frm
+t1.par
+tp1.MYD
+tp1.MYI
+tp1.frm
+master-bin.000001	#	Query	#	#	use `test`; ALTER TABLE t1 CONVERT PARTITION p1 TO TABLE tp1
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY RANGE (`x`)
+(PARTITION `p0` VALUES LESS THAN (10) ENGINE = DEFAULT_ENGINE,
+ PARTITION `pn` VALUES LESS THAN MAXVALUE ENGINE = DEFAULT_ENGINE)
+x
+2
+22
+Table	Create Table
+tp1	CREATE TABLE `tp1` (
+  `x` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+x
+12
+# CRASH: crash_convert_partition_11
+t1#P#p0.MYD
+t1#P#p0.MYI
+t1#P#pn.MYD
+t1#P#pn.MYI
+t1.frm
+t1.par
+tp1.MYD
+tp1.MYI
+tp1.frm
+master-bin.000001	#	Query	#	#	use `test`; ALTER TABLE t1 CONVERT PARTITION p1 TO TABLE tp1
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY RANGE (`x`)
+(PARTITION `p0` VALUES LESS THAN (10) ENGINE = DEFAULT_ENGINE,
+ PARTITION `pn` VALUES LESS THAN MAXVALUE ENGINE = DEFAULT_ENGINE)
+x
+2
+22
+Table	Create Table
+tp1	CREATE TABLE `tp1` (
+  `x` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+x
+12
 Warnings:
 Note	1051	Unknown table 'test.t1'
diff --git a/mysql-test/suite/atomic/alter_partition.test b/mysql-test/suite/atomic/alter_partition.test
index f4f1c75ed81..2606c328146 100644
--- a/mysql-test/suite/atomic/alter_partition.test
+++ b/mysql-test/suite/atomic/alter_partition.test
@@ -36,7 +36,7 @@ if ($MTR_COMBINATION_ROW)
 
 let $MYSQLD_DATADIR= `SELECT @@datadir`;
 
-let $crash_count= 12;
+let $crash_count= 14;
 let $crash_points='crash_create_before_create_frm',
                   'crash_alter_partition_after_create_frm',
                   'crash_alter_partition_after_write_frm',
@@ -48,10 +48,12 @@ let $crash_points='crash_create_before_create_frm',
                   'crash_convert_partition_6',
                   'crash_convert_partition_7',
                   'crash_convert_partition_8',
-                  'crash_convert_partition_9';
+                  'crash_convert_partition_9',
+                  'crash_convert_partition_10',
+                  'crash_convert_partition_11';
 
 #let $crash_count= 1;
-#let $crash_points= 'crash_convert_partition_8';
+#let $crash_points= 'crash_convert_partition_10';
 
 let $statement_count= 1;
 let $statements= 'ALTER TABLE t1 CONVERT PARTITION p1 TO TABLE tp1';
diff --git a/mysql-test/suite/parts/r/partition_debug.result b/mysql-test/suite/parts/r/partition_debug.result
index 6a69fa0337a..dc46dbdd0b8 100644
--- a/mysql-test/suite/parts/r/partition_debug.result
+++ b/mysql-test/suite/parts/r/partition_debug.result
@@ -3899,7 +3899,6 @@ UNLOCK TABLES;
 DROP TABLE t1;
 set session debug_dbug=@save_dbug;
 set session debug_dbug="+d,fail_convert_partition_9";
-# Already finished DDL logging, so tp2 now exists:
 create or replace table t1 (x int primary key)
 partition by range(x) (
 partition p1 values less than (10),
@@ -3952,6 +3951,183 @@ ERROR HY000: Unknown error
 db.opt
 t1#P#p1.MYD
 t1#P#p1.MYI
+t1#P#p2.MYD
+t1#P#p2.MYI
+t1#P#p3.MYD
+t1#P#p3.MYI
+t1#P#p4.MYD
+t1#P#p4.MYI
+t1#P#p5.MYD
+t1#P#p5.MYI
+t1#P#pn.MYD
+t1#P#pn.MYI
+t1.frm
+t1.par
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) NOT NULL,
+  PRIMARY KEY (`x`)
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+ PARTITION BY RANGE (`x`)
+(PARTITION `p1` VALUES LESS THAN (10) ENGINE = MyISAM,
+ PARTITION `p2` VALUES LESS THAN (20) ENGINE = MyISAM,
+ PARTITION `p3` VALUES LESS THAN (30) ENGINE = MyISAM,
+ PARTITION `p4` VALUES LESS THAN (40) ENGINE = MyISAM,
+ PARTITION `p5` VALUES LESS THAN (50) ENGINE = MyISAM,
+ PARTITION `pn` VALUES LESS THAN MAXVALUE ENGINE = MyISAM)
+SELECT * FROM t1;
+x
+12
+2
+22
+32
+42
+52
+DROP TABLE t1;
+create or replace table t1 (x int primary key)
+partition by range(x) (
+partition p1 values less than (10),
+partition p2 values less than (20),
+partition p3 values less than (30),
+partition p4 values less than (40),
+partition p5 values less than (50),
+partition pn values less than maxvalue);
+insert into t1 values (2), (12), (22), (32), (42), (52);
+# d,fail_convert_partition_9: BEFORE failure (under LOCK TABLE)
+db.opt
+t1#P#p1.MYD
+t1#P#p1.MYI
+t1#P#p2.MYD
+t1#P#p2.MYI
+t1#P#p3.MYD
+t1#P#p3.MYI
+t1#P#p4.MYD
+t1#P#p4.MYI
+t1#P#p5.MYD
+t1#P#p5.MYI
+t1#P#pn.MYD
+t1#P#pn.MYI
+t1.frm
+t1.par
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) NOT NULL,
+  PRIMARY KEY (`x`)
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+ PARTITION BY RANGE (`x`)
+(PARTITION `p1` VALUES LESS THAN (10) ENGINE = MyISAM,
+ PARTITION `p2` VALUES LESS THAN (20) ENGINE = MyISAM,
+ PARTITION `p3` VALUES LESS THAN (30) ENGINE = MyISAM,
+ PARTITION `p4` VALUES LESS THAN (40) ENGINE = MyISAM,
+ PARTITION `p5` VALUES LESS THAN (50) ENGINE = MyISAM,
+ PARTITION `pn` VALUES LESS THAN MAXVALUE ENGINE = MyISAM)
+SELECT * FROM t1;
+x
+12
+2
+22
+32
+42
+52
+LOCK TABLE t1 WRITE;
+alter table t1 convert partition p2 to table tp2;
+ERROR HY000: Unknown error
+# d,fail_convert_partition_9: AFTER failure (under LOCK TABLE)
+db.opt
+t1#P#p1.MYD
+t1#P#p1.MYI
+t1#P#p2.MYD
+t1#P#p2.MYI
+t1#P#p3.MYD
+t1#P#p3.MYI
+t1#P#p4.MYD
+t1#P#p4.MYI
+t1#P#p5.MYD
+t1#P#p5.MYI
+t1#P#pn.MYD
+t1#P#pn.MYI
+t1.frm
+t1.par
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) NOT NULL,
+  PRIMARY KEY (`x`)
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+ PARTITION BY RANGE (`x`)
+(PARTITION `p1` VALUES LESS THAN (10) ENGINE = MyISAM,
+ PARTITION `p2` VALUES LESS THAN (20) ENGINE = MyISAM,
+ PARTITION `p3` VALUES LESS THAN (30) ENGINE = MyISAM,
+ PARTITION `p4` VALUES LESS THAN (40) ENGINE = MyISAM,
+ PARTITION `p5` VALUES LESS THAN (50) ENGINE = MyISAM,
+ PARTITION `pn` VALUES LESS THAN MAXVALUE ENGINE = MyISAM)
+SELECT * FROM t1;
+x
+12
+2
+22
+32
+42
+52
+UNLOCK TABLES;
+DROP TABLE t1;
+set session debug_dbug=@save_dbug;
+set session debug_dbug="+d,fail_convert_partition_11";
+# Already finished DDL logging, so tp2 now exists:
+create or replace table t1 (x int primary key)
+partition by range(x) (
+partition p1 values less than (10),
+partition p2 values less than (20),
+partition p3 values less than (30),
+partition p4 values less than (40),
+partition p5 values less than (50),
+partition pn values less than maxvalue);
+insert into t1 values (2), (12), (22), (32), (42), (52);
+# d,fail_convert_partition_11: BEFORE failure
+db.opt
+t1#P#p1.MYD
+t1#P#p1.MYI
+t1#P#p2.MYD
+t1#P#p2.MYI
+t1#P#p3.MYD
+t1#P#p3.MYI
+t1#P#p4.MYD
+t1#P#p4.MYI
+t1#P#p5.MYD
+t1#P#p5.MYI
+t1#P#pn.MYD
+t1#P#pn.MYI
+t1.frm
+t1.par
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) NOT NULL,
+  PRIMARY KEY (`x`)
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+ PARTITION BY RANGE (`x`)
+(PARTITION `p1` VALUES LESS THAN (10) ENGINE = MyISAM,
+ PARTITION `p2` VALUES LESS THAN (20) ENGINE = MyISAM,
+ PARTITION `p3` VALUES LESS THAN (30) ENGINE = MyISAM,
+ PARTITION `p4` VALUES LESS THAN (40) ENGINE = MyISAM,
+ PARTITION `p5` VALUES LESS THAN (50) ENGINE = MyISAM,
+ PARTITION `pn` VALUES LESS THAN MAXVALUE ENGINE = MyISAM)
+SELECT * FROM t1;
+x
+12
+2
+22
+32
+42
+52
+alter table t1 convert partition p2 to table tp2;
+ERROR HY000: Unknown error
+# d,fail_convert_partition_11: AFTER failure
+db.opt
+t1#P#p1.MYD
+t1#P#p1.MYI
 t1#P#p3.MYD
 t1#P#p3.MYI
 t1#P#p4.MYD
@@ -3994,7 +4170,7 @@ partition p4 values less than (40),
 partition p5 values less than (50),
 partition pn values less than maxvalue);
 insert into t1 values (2), (12), (22), (32), (42), (52);
-# d,fail_convert_partition_9: BEFORE failure (under LOCK TABLE)
+# d,fail_convert_partition_11: BEFORE failure (under LOCK TABLE)
 db.opt
 t1#P#p1.MYD
 t1#P#p1.MYI
@@ -4037,7 +4213,7 @@ x
 LOCK TABLE t1 WRITE;
 alter table t1 convert partition p2 to table tp2;
 ERROR 42S01: Table 'tp2' already exists
-# d,fail_convert_partition_9: AFTER failure (under LOCK TABLE)
+# d,fail_convert_partition_11: AFTER failure (under LOCK TABLE)
 db.opt
 t1#P#p1.MYD
 t1#P#p1.MYI
diff --git a/mysql-test/suite/parts/t/partition_debug.test b/mysql-test/suite/parts/t/partition_debug.test
index 48b3392aa4f..91e098f423b 100644
--- a/mysql-test/suite/parts/t/partition_debug.test
+++ b/mysql-test/suite/parts/t/partition_debug.test
@@ -104,6 +104,9 @@ set session debug_dbug="+d,fail_convert_partition_8";
 --source suite/parts/inc/partition_fail.inc
 set session debug_dbug=@save_dbug;
 set session debug_dbug="+d,fail_convert_partition_9";
+--source suite/parts/inc/partition_fail.inc
+set session debug_dbug=@save_dbug;
+set session debug_dbug="+d,fail_convert_partition_11";
 --echo # Already finished DDL logging, so tp2 now exists:
 --source suite/parts/inc/partition_fail.inc
 set session debug_dbug=@save_dbug;
diff --git a/sql/ddl_log.cc b/sql/ddl_log.cc
index ea33f075938..47b0077588f 100644
--- a/sql/ddl_log.cc
+++ b/sql/ddl_log.cc
@@ -77,6 +77,8 @@
 #define DDL_LOG_MAGIC_LENGTH 4
 /* How many times to try to execute a ddl log entry that causes crashes */
 #define DDL_LOG_MAX_RETRY 3
+#define DDL_LOG_RETRY_MASK 0xFF
+#define DDL_LOG_RETRY_BITS 8
 
 uchar ddl_log_file_magic[]=
 { (uchar) 254, (uchar) 254, (uchar) 11, (uchar) 2 };
@@ -155,7 +157,7 @@ mysql_mutex_t LOCK_gdl;
 #define DDL_LOG_XID_POS 10
 /* Used to store unique uuid from the .frm file */
 #define DDL_LOG_UUID_POS 18
-/* ID_POS can be used to store something unique, like file size (4 bytes) */
+/* ID_POS can be used to store something unique, like file size (8 bytes) */
 #define DDL_LOG_ID_POS DDL_LOG_UUID_POS + MY_UUID_SIZE
 #define DDL_LOG_END_POS DDL_LOG_ID_POS + 8
 
@@ -2515,6 +2517,7 @@ bool ddl_log_write_entry(DDL_LOG_ENTRY *ddl_log_entry,
 
   @param first_entry               First entry in linked list of entries
                                    to execute.
+  @param cond_entry                Check and don't execute if cond_entry is active
   @param[in,out] active_entry      Entry to execute, 0 = NULL if the entry
                                    is written first time and needs to be
                                    returned. In this case the entry written
@@ -2525,6 +2528,7 @@ bool ddl_log_write_entry(DDL_LOG_ENTRY *ddl_log_entry,
 */
 
 bool ddl_log_write_execute_entry(uint first_entry,
+                                 uint cond_entry,
                                  DDL_LOG_MEMORY_ENTRY **active_entry)
 {
   uchar *file_entry_buf= global_ddl_log.file_entry_buf;
@@ -2541,6 +2545,7 @@ bool ddl_log_write_execute_entry(uint first_entry,
 
   file_entry_buf[DDL_LOG_ENTRY_TYPE_POS]= (uchar)DDL_LOG_EXECUTE_CODE;
   int4store(file_entry_buf + DDL_LOG_NEXT_ENTRY_POS, first_entry);
+  int8store(file_entry_buf + DDL_LOG_ID_POS, (cond_entry << DDL_LOG_RETRY_BITS));
 
   if (!(*active_entry))
   {
@@ -2763,13 +2768,13 @@ int ddl_log_execute_recovery()
       recovery_state.xid= ddl_log_entry.xid;
 
       /* purecov: begin tested */
-      if (ddl_log_entry.unique_id > DDL_LOG_MAX_RETRY)
+      if ((ddl_log_entry.unique_id & DDL_LOG_RETRY_MASK) > DDL_LOG_MAX_RETRY)
       {
         error= -1;
         continue;
       }
       update_unique_id(i, ++ddl_log_entry.unique_id);
-      if (ddl_log_entry.unique_id > DDL_LOG_MAX_RETRY)
+      if ((ddl_log_entry.unique_id & DDL_LOG_RETRY_MASK) > DDL_LOG_MAX_RETRY)
       {
         sql_print_error("DDL_LOG: Aborting executing entry %u after %llu "
                         "retries", i, ddl_log_entry.unique_id);
@@ -2778,6 +2783,15 @@ int ddl_log_execute_recovery()
       }
       /* purecov: end tested */
 
+      uint cond_entry= (uint)(ddl_log_entry.unique_id >> DDL_LOG_RETRY_BITS);
+
+      if (cond_entry && is_execute_entry_active(cond_entry))
+      {
+        if (disable_execute_entry(i))
+          error= -1;
+        continue;
+      }
+
       if (ddl_log_execute_entry_no_lock(thd, ddl_log_entry.next_entry))
       {
         /* Real unpleasant scenario but we have to continue anyway  */
@@ -3534,3 +3548,32 @@ err:
   mysql_mutex_unlock(&LOCK_gdl);
   DBUG_RETURN(1);
 }
+
+
+/*
+  Log an delete frm file
+*/
+
+/*
+  TODO: Partitioning atomic DDL refactoring: this should be replaced with
+        ddl_log_create_table().
+*/
+bool ddl_log_delete_frm(DDL_LOG_STATE *ddl_state, const char *to_path)
+{
+  DDL_LOG_ENTRY ddl_log_entry;
+  DDL_LOG_MEMORY_ENTRY *log_entry;
+  DBUG_ENTER("ddl_log_delete_frm");
+  bzero(&ddl_log_entry, sizeof(ddl_log_entry));
+  ddl_log_entry.action_type= DDL_LOG_DELETE_ACTION;
+  ddl_log_entry.next_entry= ddl_state->list ? ddl_state->list->entry_pos : 0;
+
+  lex_string_set(&ddl_log_entry.handler_name, reg_ext);
+  lex_string_set(&ddl_log_entry.name, to_path);
+
+  mysql_mutex_assert_owner(&LOCK_gdl);
+  if (ddl_log_write_entry(&ddl_log_entry, &log_entry))
+    DBUG_RETURN(1);
+
+  ddl_log_add_entry(ddl_state, log_entry);
+  DBUG_RETURN(0);
+}
diff --git a/sql/ddl_log.h b/sql/ddl_log.h
index afdd775d7a0..9960855a813 100644
--- a/sql/ddl_log.h
+++ b/sql/ddl_log.h
@@ -262,8 +262,14 @@ int ddl_log_execute_recovery();
 bool ddl_log_write_entry(DDL_LOG_ENTRY *ddl_log_entry,
                            DDL_LOG_MEMORY_ENTRY **active_entry);
 
+bool ddl_log_write_execute_entry(uint first_entry, uint cond_entry,
+                                 DDL_LOG_MEMORY_ENTRY** active_entry);
+inline
 bool ddl_log_write_execute_entry(uint first_entry,
-                                 DDL_LOG_MEMORY_ENTRY **active_entry);
+                                 DDL_LOG_MEMORY_ENTRY **active_entry)
+{
+  return ddl_log_write_execute_entry(first_entry, 0, active_entry);
+}
 bool ddl_log_disable_execute_entry(DDL_LOG_MEMORY_ENTRY **active_entry);
 
 void ddl_log_complete(DDL_LOG_STATE *ddl_log_state);
@@ -349,5 +355,6 @@ bool ddl_log_alter_table(THD *thd, DDL_LOG_STATE *ddl_state,
                          bool is_renamed);
 bool ddl_log_store_query(THD *thd, DDL_LOG_STATE *ddl_log_state,
                          const char *query, size_t length);
+bool ddl_log_delete_frm(DDL_LOG_STATE *ddl_state, const char *to_path);
 extern mysql_mutex_t LOCK_gdl;
 #endif /* DDL_LOG_INCLUDED */
diff --git a/sql/sql_partition.cc b/sql/sql_partition.cc
index df116f9c9ba..bdd11740539 100644
--- a/sql/sql_partition.cc
+++ b/sql/sql_partition.cc
@@ -6226,48 +6226,6 @@ static void release_part_info_log_entries(DDL_LOG_MEMORY_ENTRY *log_entry)
 
 
 /*
-  Log an delete frm file
-  SYNOPSIS
-    write_log_delete_frm()
-    lpt                            Struct for parameters
-    to_path                        Name to delete
-  RETURN VALUES
-    TRUE                           Error
-    FALSE                          Success
-  DESCRIPTION
-    Support routine that writes a delete of an frm file into the
-    ddl log. It also inserts an entry that keeps track of used space into
-    the partition info object
-*/
-
-
-/*
-  TODO: Partitioning atomic DDL refactoring: this should be replaced with
-        ddl_log_create_table().
-*/
-static bool write_log_delete_frm(ALTER_PARTITION_PARAM_TYPE *lpt,
-                                 const char *to_path)
-{
-  DDL_LOG_ENTRY ddl_log_entry;
-  DDL_LOG_MEMORY_ENTRY *log_entry;
-  DBUG_ENTER("write_log_delete_frm");
-  bzero(&ddl_log_entry, sizeof(ddl_log_entry));
-  ddl_log_entry.action_type= DDL_LOG_DELETE_ACTION;
-  ddl_log_entry.next_entry= lpt->part_info->list ? lpt->part_info->list->entry_pos : 0;
-
-  lex_string_set(&ddl_log_entry.handler_name, reg_ext);
-  lex_string_set(&ddl_log_entry.name, to_path);
-
-  if (ddl_log_write_entry(&ddl_log_entry, &log_entry))
-  {
-    DBUG_RETURN(true);
-  }
-  ddl_log_add_entry(lpt->part_info, log_entry);
-  DBUG_RETURN(false);
-}
-
-
-/*
   Log an rename frm file
   SYNOPSIS
     write_log_replace_frm()
@@ -6541,11 +6499,10 @@ static bool write_log_convert_out_partition(ALTER_PARTITION_PARAM_TYPE *lpt,
   DDL_LOG_MEMORY_ENTRY *main_entry= lpt->part_info->main_entry;
   bool res= log_drop_or_convert_action(lpt, next_entry, path, from_name, false);
   /*
-    NOTE: main_entry is "drop shadow frm", we have to keep it like this,
-    because partitioning crash-safety disables it at install shadow FRM phase
-    That is not really needed though, because shadow frm is replaced with
-    backup frm so there is nothing to drop. But we avoid spurious action by
-    disabling it.
+    NOTE: main_entry is "drop shadow frm", we have to keep it like this
+    because partitioning crash-safety disables it at install shadow FRM phase.
+    This is needed to avoid spurious drop action when the shadow frm is replaced
+    by the backup frm and there is nothing to drop.
   */
   lpt->part_info->main_entry= main_entry;
   return res;
@@ -6556,10 +6513,9 @@ static bool write_log_convert_out_partition(ALTER_PARTITION_PARAM_TYPE *lpt,
   Write the log entry to ensure that the shadow frm file is removed at
   crash.
   SYNOPSIS
-    write_log_drop_shadow_frm()
+    write_log_drop_frm()
     lpt                      Struct containing parameters
-    install_frm              Should we log action to install shadow frm or should
-                             the action be to remove the shadow frm file.
+
   RETURN VALUES
     TRUE                     Error
     FALSE                    Success
@@ -6568,38 +6524,43 @@ static bool write_log_convert_out_partition(ALTER_PARTITION_PARAM_TYPE *lpt,
     file and its corresponding handler file.
 */
 
-static bool write_log_drop_shadow_frm(ALTER_PARTITION_PARAM_TYPE *lpt,
-                                      uint flags= 0)
+static bool write_log_drop_frm(ALTER_PARTITION_PARAM_TYPE *lpt,
+                               DDL_LOG_STATE *drop_chain)
 {
-  partition_info *part_info= lpt->part_info;
-  DDL_LOG_MEMORY_ENTRY *log_entry;
   char path[FN_REFLEN + 1];
-  DBUG_ENTER("write_log_drop_shadow_frm");
-  const bool drop_backup= (flags & WFRM_DROP_BACKUP);
-
-  DBUG_ASSERT(!drop_backup || !part_info->is_active());
+  DBUG_ENTER("write_log_drop_frm");
+  const DDL_LOG_STATE *main_chain= lpt->part_info;
+  const bool drop_backup= (drop_chain != main_chain);
 
   build_table_shadow_filename(path, sizeof(path) - 1, lpt, drop_backup);
   mysql_mutex_lock(&LOCK_gdl);
-  if (write_log_delete_frm(lpt, (const char*)path))
+  if (ddl_log_delete_frm(drop_chain, (const char*)path))
     goto error;
 
-  log_entry= part_info->list;
-  if (ddl_log_write_execute_entry(log_entry->entry_pos,
-                                  &part_info->execute_entry))
+  if (ddl_log_write_execute_entry(drop_chain->list->entry_pos,
+                                  drop_backup ?
+                                    main_chain->execute_entry->entry_pos : 0,
+                                  &drop_chain->execute_entry))
     goto error;
   mysql_mutex_unlock(&LOCK_gdl);
   DBUG_RETURN(FALSE);
 
 error:
-  release_part_info_log_entries(part_info->list);
+  release_part_info_log_entries(drop_chain->list);
   mysql_mutex_unlock(&LOCK_gdl);
-  part_info->list= NULL;
+  drop_chain->list= NULL;
   my_error(ER_DDL_LOG_ERROR, MYF(0));
   DBUG_RETURN(TRUE);
 }
 
 
+static inline
+bool write_log_drop_shadow_frm(ALTER_PARTITION_PARAM_TYPE *lpt)
+{
+  return write_log_drop_frm(lpt, lpt->part_info);
+}
+
+
 /*
   Log renaming of shadow frm to real frm name and dropping of old frm
   SYNOPSIS
@@ -6910,6 +6871,10 @@ static void write_log_completed(ALTER_PARTITION_PARAM_TYPE *lpt,
      NONE
 */
 
+/*
+  TODO: Partitioning atomic DDL refactoring: this should be replaced with
+        ddl_log_release_entries().
+*/
 static void release_log_entries(partition_info *part_info)
 {
   mysql_mutex_lock(&LOCK_gdl);
@@ -7414,6 +7379,9 @@ uint fast_alter_partition_table(THD *thd, TABLE *table,
   }
   else if (alter_info->partition_flags & ALTER_PARTITION_CONVERT_OUT)
   {
+    DDL_LOG_STATE chain_drop_backup;
+    bzero(&chain_drop_backup, sizeof(chain_drop_backup));
+
     if (mysql_write_frm(lpt, WFRM_WRITE_CONVERTED_TO) ||
         ERROR_INJECT("convert_partition_1") ||
         write_log_drop_shadow_frm(lpt) ||
@@ -7428,40 +7396,27 @@ uint fast_alter_partition_table(THD *thd, TABLE *table,
         ERROR_INJECT("convert_partition_6") ||
         alter_partition_convert_out(lpt) ||
         ERROR_INJECT("convert_partition_7") ||
-        (frm_install= true, false) ||
+        write_log_drop_frm(lpt, &chain_drop_backup) ||
         mysql_write_frm(lpt, WFRM_INSTALL_SHADOW|WFRM_BACKUP_ORIGINAL) ||
         log_partition_alter_to_ddl_log(lpt) ||
-        (frm_install= false, false) ||
         ERROR_INJECT("convert_partition_8") ||
         ((!thd->lex->no_write_to_binlog) &&
           ((thd->binlog_xid= thd->query_id),
            ddl_log_update_xid(lpt->part_info, thd->binlog_xid),
            write_bin_log(thd, false, thd->query(), thd->query_length()),
            (thd->binlog_xid= 0))) ||
-        (ddl_log_complete(lpt->part_info), false) ||
-        /*
-          TODO:
-
-          1. Add DDL_LOG_EXECUTE_IF_CLOSED to ddl_log_entry_code.
-             Execute entry is executed only if another entry is active.
-             This requires ddl log file extension or store entry_pos into some
-             string field of execute entry: name, tmp_name, etc. These are
-             not used now for execute entry.
-          2. Log WFRM_DROP_BACKUP into separate "cleanup" chain and execute it
-             only if the main chain is closed. That must be logged before
-             WFRM_BACKUP_ORIGINAL is done.
-
-        */
-        write_log_drop_shadow_frm(lpt, WFRM_DROP_BACKUP) ||
         ERROR_INJECT("convert_partition_9"))
     {
+      ddl_log_complete(&chain_drop_backup);
       (void) ddl_log_revert(thd, lpt->part_info);
-      handle_alter_part_error(lpt, true, true, frm_install);
+      handle_alter_part_error(lpt, true, true, false);
       goto err;
     }
-    /* Drop backup frm */
-    (void) ddl_log_revert(thd, lpt->part_info);
-    if (alter_partition_lock_handling(lpt))
+    ddl_log_complete(lpt->part_info);
+    ERROR_INJECT("convert_partition_10");
+    (void) ddl_log_revert(thd, &chain_drop_backup);
+    if (alter_partition_lock_handling(lpt) ||
+        ERROR_INJECT("convert_partition_11"))
       goto err;
   }
   else if ((alter_info->partition_flags & ALTER_PARTITION_ADD) &&
diff --git a/sql/sql_table.h b/sql/sql_table.h
index 4f22bf086ae..eaa03bfaf8c 100644
--- a/sql/sql_table.h
+++ b/sql/sql_table.h
@@ -59,7 +59,6 @@ enum enum_explain_filename_mode
 #define WFRM_KEEP_SHARE 4
 #define WFRM_WRITE_CONVERTED_TO 8
 #define WFRM_BACKUP_ORIGINAL 16
-#define WFRM_DROP_BACKUP 32
 
 /* Flags for conversion functions. */
 static const uint FN_FROM_IS_TMP=  1 << 0;