Review and crash-safety fix

2025-07-30 16:24:05 +03:00 · 2021-09-27 15:53:52 +03:00
parent 428024524c
commit 7da721be31
9 changed files with 373 additions and 96 deletions
--- 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` (
--- 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'
--- 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';
--- 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
--- 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;
--- 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);
+}
--- 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 */
--- a/sql/sql_partition.cc
+++ b/sql/sql_partition.cc
@ -6225,48 +6225,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
@ -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) &&
--- 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;