MDEV-25292 Atomic CREATE OR REPLACE TABLEbb-10.6-midenok-MDEV-25292

High level operation

  Two DDL chains are used for CREATE OR REPLACE execution:
  ddl_log_state_create (C) and ddl_log_state_rm (D).

  1. (C) Write DDL_LOG_RENAME_TABLE_ACTION from normal to backup;
  2. Rename table from normal to backup;
  3. D is replayed only if C is closed (D is linked as a slave to C);
  4. (D) Write DDL_LOG_DROP_ACTION of backup;
  5. (C) Write DDL_LOG_CREATE_TABLE_ACTION of new table;
  6. Create new table;
  7. (C) update XID;

  (In case of success)
  8. (C) close chain;
  9. (D) replay chain.

  (In case of error)
  8. (C) replay chain;
  9. (D) close chain.

Backup table for CREATE OR REPLACE

  - Instead of dropping the table, CREATE OR REPLACE renames table to
    a backup name. The rename is written into ddl_log_state_create.

  - CREATE TABLE without REPLACE is not affected by this commit.

  - Engines having HTON_EXPENSIVE_RENAME flag set are not affected by
    this commit.

  - The backup is done inside mysql_rm_table_no_locks() to be able to
    support ATOMIC DROP TABLE of multiple tables.

  - The ddl logging of CREATE and DROP are linked together. CREATE
    entries are executed before DROP entries.

  - In case of failed CREATE OR REPLACE ddl_log_state_create is
    replayed which restores original table.

  - In case of succeeded CREATE OR REPLACE ddl_log_state_rm is
    replayed which drops the backup (original table).

  - CREATE TABLE .. SELECT XID usage is fixed and now there is no need
    to log DROP TABLE via DDL_CREATE_TABLE_PHASE_LOG (see comments in
    do_postlock()). XID is now correctly updated so it disables
    DDL_LOG_DROP_TABLE_ACTION. Note that binary log is flushed at the
    final stage when the table is ready. So if we have XID in the
    binary log we don't need to drop the table.

  - Three variations of CREATE OR REPLACE handled:

    1. CREATE OR REPLACE TABLE t1 (..);
    2. CREATE OR REPLACE TABLE t1 LIKE t2;
    3. CREATE OR REPLACE TABLE t1 SELECT ..;

  - Test case uses 5 combinations for engines (aria, aria_notrans,
    myisam, ib, expensive_rename) and 2 combinations for binlog types
    (row, stmt). Combinations help to check differences between the
    results. Error failures are tested for the above three variations.

  - Triggers mechanism is unaffected by this change. This is tested in
    create_replace.test.

  - Triggers mechanism is unaffected which is tested.

  - LOCK TABLES is affected because RENAME TABLE cannot work on locked
    table. mysql_do_rename() (original do_rename()) can now work under
    locked_tables_mode.

On XID usage

  Like with all other atomic DDL operations XID is used to avoid
  inconsistency between master and slave in the case of a crash after
  binary log is written and before ddl_log_state_create is closed. On
  recovery XIDs are taken from binary log and corresponding DDL log
  events get disabled.  That is done by
  ddl_log_close_binlogged_events().

On linking two chains together

  Chains are executed in the ascending order of entry_pos of execute
  entries. But entry_pos assignment order is undefined: it may assign
  bigger number for the first chain and then smaller number for the
  second chain. So the execution order in that case will be reverse:
  second chain will be executed first.

40 files changed, 3524 insertions, 424 deletions

diff --git a/mysql-test/include/log_bin.combinations b/mysql-test/include/log_bin.combinations
new file mode 100644
index 00000000000..07042c2cbec
--- /dev/null
+++ b/mysql-test/include/log_bin.combinations
@@ -0,0 +1,8 @@
+[row]
+binlog-format=row
+
+[stmt]
+binlog-format=statement
+
+[mix]
+binlog-format=mixed
diff --git a/mysql-test/include/log_bin.inc b/mysql-test/include/log_bin.inc
new file mode 100644
index 00000000000..aadec436c18
--- /dev/null
+++ b/mysql-test/include/log_bin.inc
@@ -0,0 +1 @@
+--source include/have_log_bin.inc
diff --git a/mysql-test/main/backup_log.result b/mysql-test/main/backup_log.result
index 94590a220d4..a6a364c9370 100644
--- a/mysql-test/main/backup_log.result
+++ b/mysql-test/main/backup_log.result
@@ -69,7 +69,7 @@ ERROR 23000: Duplicate entry '1' for key 'PRIMARY'
 create table t32 (a int) ;
 drop table if exists t30,t31,t32,tmp_t30;
 Warnings:
-Note	1051	Unknown table 'test.t31,test.tmp_t30'
+Note	1051	Unknown table 'test.tmp_t30'
 #
 # Testing create LIKE
 #
@@ -180,17 +180,18 @@ CREATE,MyISAM,0,test,t21,id: 7,,0,,,
 DROP,MyISAM,0,test,t21,id: 7,,0,,,
 CREATE,MyISAM,0,test,t30,id: 8,,0,,,
 CREATE,MyISAM,0,test,t31,id: 9,,0,,,
-DROP,MyISAM,0,test,t31,id: 9,,0,,,
+RENAME,MyISAM,0,test,t31,id: 9,MyISAM,0,test,#sql1,id: 9
 CREATE,MyISAM,0,test,t31,id: 10,,0,,,
-DROP,MyISAM,0,test,t31,id: 10,,0,,,
+RENAME,MyISAM,0,test,t31,id: 10,MyISAM,0,test,#sql1,id: 10
 DROP_AFTER_CREATE,MyISAM,0,test,t31,id: 11,,0,,,
 CREATE,MyISAM,0,test,t32,id: 12,,0,,,
 DROP,MyISAM,0,test,t30,id: 8,,0,,,
+DROP,MyISAM,0,test,t31,id: 10,,0,,,
 DROP,MyISAM,0,test,t32,id: 12,,0,,,
 CREATE,MyISAM,0,test,t40,id: 13,,0,,,
 CREATE,InnoDB,0,test,t41,id: 14,,0,,,
 CREATE,MyISAM,0,test,t42,id: 15,,0,,,
-DROP,MyISAM,0,test,t42,id: 15,,0,,,
+RENAME,MyISAM,0,test,t42,id: 15,MyISAM,0,test,#sql1,id: 15
 CREATE,InnoDB,0,test,t42,id: 16,,0,,,
 DROP,MyISAM,0,test,t40,id: 13,,0,,,
 DROP,InnoDB,0,test,t41,id: 14,,0,,,
diff --git a/mysql-test/main/create_or_replace.result b/mysql-test/main/create_or_replace.result
index 1eaf87d190d..61106baf9aa 100644
--- a/mysql-test/main/create_or_replace.result
+++ b/mysql-test/main/create_or_replace.result
@@ -271,6 +271,7 @@ create or replace table test.t1;
 ERROR 42000: A table must have at least 1 column
 show tables;
 Tables_in_test
+t1
 t2
 select * from information_schema.metadata_lock_info;
 THREAD_ID	LOCK_MODE	LOCK_DURATION	LOCK_TYPE	TABLE_SCHEMA	TABLE_NAME
@@ -283,8 +284,10 @@ create or replace table mysqltest2.t2;
 ERROR 42000: A table must have at least 1 column
 select * from information_schema.metadata_lock_info;
 THREAD_ID	LOCK_MODE	LOCK_DURATION	LOCK_TYPE	TABLE_SCHEMA	TABLE_NAME
+drop table t1;
 create table t1 (i int);
 drop table t1;
+drop table mysqltest2.t2;
 create table test.t1 (i int);
 create table mysqltest2.t2 like test.t1;
 lock table test.t1 write, mysqltest2.t2 write;
@@ -300,6 +303,7 @@ create or replace table test.t1 (a int) select 1 as 'a', 2 as 'a';
 ERROR 42S21: Duplicate column name 'a'
 show tables;
 Tables_in_test
+t1
 t2
 select * from information_schema.metadata_lock_info;
 THREAD_ID	LOCK_MODE	LOCK_DURATION	LOCK_TYPE	TABLE_SCHEMA	TABLE_NAME
@@ -312,8 +316,10 @@ create or replace table mysqltest2.t2 (a int) select 1 as 'a', 2 as 'a';
 ERROR 42S21: Duplicate column name 'a'
 select * from information_schema.metadata_lock_info;
 THREAD_ID	LOCK_MODE	LOCK_DURATION	LOCK_TYPE	TABLE_SCHEMA	TABLE_NAME
+drop table t1;
 create table t1 (i int);
 drop table t1;
+drop table mysqltest2.t2;
 create table test.t1 (i int) engine=innodb;
 create table mysqltest2.t2 like test.t1;
 lock table test.t1 write, mysqltest2.t2 write;
@@ -545,3 +551,161 @@ SELECT * FROM t3;
 ERROR HY000: Table 't3' was not locked with LOCK TABLES
 UNLOCK TABLES;
 DROP TABLE t3;
+#
+# MDEV-25292 Atomic CREATE OR REPLACE TABLE
+#
+create table t1 (a int);
+insert t1 values (1), (1);
+create table t2 (c int);
+create or replace table t2 (a int, b int, key k (a), key k (b));
+ERROR 42000: Duplicate key name 'k'
+show create table t2;
+Table	Create Table
+t2	CREATE TABLE `t2` (
+  `c` int(11) DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+create or replace table t2 (a int, b int, key k (a), key k (b)) as select a, a as b from t1;
+ERROR 42000: Duplicate key name 'k'
+show create table t2;
+Table	Create Table
+t2	CREATE TABLE `t2` (
+  `c` int(11) DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+create or replace table t2 (a int primary key) as select * from t1;
+ERROR 23000: Duplicate entry '1' for key 'PRIMARY'
+show create table t2;
+Table	Create Table
+t2	CREATE TABLE `t2` (
+  `c` int(11) DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+set @old_mode= @@sql_mode;
+set @@sql_mode='ALLOW_INVALID_DATES';
+create table t3 (dt datetime default '2008-02-31 00:00:00');
+set @@sql_mode= @old_mode;
+create or replace table t2 like t3;
+ERROR 42000: Invalid default value for 'dt'
+show create table t2;
+Table	Create Table
+t2	CREATE TABLE `t2` (
+  `c` int(11) DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+# LOCK TABLES
+lock tables t2 write, t1 write;
+flush tables;
+show open tables like 't2';
+Database	Table	In_use	Name_locked
+test	t2	1	0
+create or replace table t2 (y int);
+flush tables;
+show open tables like 't2';
+Database	Table	In_use	Name_locked
+test	t2	1	0
+create or replace table t2 like t1;
+flush tables;
+show open tables like 't2';
+Database	Table	In_use	Name_locked
+test	t2	1	0
+create or replace table t2 (y int) as select * from t1;
+flush tables;
+show open tables like 't2';
+Database	Table	In_use	Name_locked
+test	t2	1	0
+unlock tables;
+# SP
+create or replace procedure sp()
+begin
+show open tables like 't2';
+create or replace table t2 (y int);
+select 't2 is removed from prelocked:' as '';
+show open tables like 't2';
+select 'but it is ok, it can be opened again:' as '';
+insert into t2 values (2);
+show open tables like 't2';
+create or replace table t2 like t1;
+show open tables like 't2';
+create or replace table t2 (y int) as select * from t1;
+show open tables like 't2';
+select * from t2;
+show open tables  like 't2';
+end $
+flush tables;
+call sp;
+Database	Table	In_use	Name_locked
+
+t2 is removed from prelocked:
+Database	Table	In_use	Name_locked
+
+but it is ok, it can be opened again:
+Database	Table	In_use	Name_locked
+test	t2	0	0
+Database	Table	In_use	Name_locked
+Database	Table	In_use	Name_locked
+test	t2	0	0
+y	a
+NULL	1
+NULL	1
+Database	Table	In_use	Name_locked
+test	t2	0	0
+call sp;
+Database	Table	In_use	Name_locked
+test	t2	0	0
+
+t2 is removed from prelocked:
+Database	Table	In_use	Name_locked
+
+but it is ok, it can be opened again:
+Database	Table	In_use	Name_locked
+test	t2	0	0
+Database	Table	In_use	Name_locked
+Database	Table	In_use	Name_locked
+test	t2	0	0
+y	a
+NULL	1
+NULL	1
+Database	Table	In_use	Name_locked
+test	t2	0	0
+# SP under LOCK TABLES
+lock tables t2 write, t1 write;
+call sp;
+Database	Table	In_use	Name_locked
+test	t2	1	0
+
+t2 is removed from prelocked:
+Database	Table	In_use	Name_locked
+test	t2	1	0
+
+but it is ok, it can be opened again:
+Database	Table	In_use	Name_locked
+test	t2	1	0
+Database	Table	In_use	Name_locked
+test	t2	1	0
+Database	Table	In_use	Name_locked
+test	t2	1	0
+y	a
+NULL	1
+NULL	1
+Database	Table	In_use	Name_locked
+test	t2	1	0
+call sp;
+Database	Table	In_use	Name_locked
+test	t2	1	0
+
+t2 is removed from prelocked:
+Database	Table	In_use	Name_locked
+test	t2	1	0
+
+but it is ok, it can be opened again:
+Database	Table	In_use	Name_locked
+test	t2	1	0
+Database	Table	In_use	Name_locked
+test	t2	1	0
+Database	Table	In_use	Name_locked
+test	t2	1	0
+y	a
+NULL	1
+NULL	1
+Database	Table	In_use	Name_locked
+test	t2	1	0
+unlock tables;
+drop procedure sp;
+drop tables t1, t2, t3;
diff --git a/mysql-test/main/create_or_replace.test b/mysql-test/main/create_or_replace.test
index 1b4994e811f..b459ceaf113 100644
--- a/mysql-test/main/create_or_replace.test
+++ b/mysql-test/main/create_or_replace.test
@@ -227,8 +227,10 @@ create or replace table mysqltest2.t2;
 --replace_column 1 #
 --sorted_result
 select * from information_schema.metadata_lock_info;
+drop table t1;
 create table t1 (i int);
 drop table t1;
+drop table mysqltest2.t2;
 
 create table test.t1 (i int);
 create table mysqltest2.t2 like test.t1;
@@ -247,8 +249,10 @@ create or replace table mysqltest2.t2 (a int) select 1 as 'a', 2 as 'a';
 --replace_column 1 #
 --sorted_result
 select * from information_schema.metadata_lock_info;
+drop table t1;
 create table t1 (i int);
 drop table t1;
+drop table mysqltest2.t2;
 
 create table test.t1 (i int) engine=innodb;
 create table mysqltest2.t2 like test.t1;
@@ -486,3 +490,68 @@ CREATE OR REPLACE TEMPORARY TABLE t2(c INT DEFAULT '');
 SELECT * FROM t3;
 UNLOCK TABLES;
 DROP TABLE t3;
+
+
+--echo #
+--echo # MDEV-25292 Atomic CREATE OR REPLACE TABLE
+--echo #
+create table t1 (a int);
+insert t1 values (1), (1);
+create table t2 (c int);
+--error ER_DUP_KEYNAME
+create or replace table t2 (a int, b int, key k (a), key k (b));
+show create table t2;
+--error ER_DUP_KEYNAME
+create or replace table t2 (a int, b int, key k (a), key k (b)) as select a, a as b from t1;
+show create table t2;
+--error ER_DUP_ENTRY
+create or replace table t2 (a int primary key) as select * from t1;
+show create table t2;
+set @old_mode= @@sql_mode;
+set @@sql_mode='ALLOW_INVALID_DATES';
+create table t3 (dt datetime default '2008-02-31 00:00:00');
+set @@sql_mode= @old_mode;
+--error ER_INVALID_DEFAULT
+create or replace table t2 like t3;
+show create table t2;
+--echo # LOCK TABLES
+lock tables t2 write, t1 write;
+flush tables;
+show open tables like 't2';
+create or replace table t2 (y int);
+flush tables;
+show open tables like 't2';
+create or replace table t2 like t1;
+flush tables;
+show open tables like 't2';
+create or replace table t2 (y int) as select * from t1;
+flush tables;
+show open tables like 't2';
+unlock tables;
+--echo # SP
+--delimiter $
+create or replace procedure sp()
+begin
+  show open tables like 't2';
+  create or replace table t2 (y int);
+  select 't2 is removed from prelocked:' as '';
+  show open tables like 't2';
+  select 'but it is ok, it can be opened again:' as '';
+  insert into t2 values (2);
+  show open tables like 't2';
+  create or replace table t2 like t1;
+  show open tables like 't2';
+  create or replace table t2 (y int) as select * from t1;
+  show open tables like 't2';
+  select * from t2;
+  show open tables  like 't2';
+end $
+--delimiter ;
+flush tables;
+call sp; call sp;
+--echo # SP under LOCK TABLES
+lock tables t2 write, t1 write;
+call sp; call sp;
+unlock tables;
+drop procedure sp;
+drop tables t1, t2, t3;
diff --git a/mysql-test/main/default.result b/mysql-test/main/default.result
index 492cda6b9f5..94693c70eea 100644
--- a/mysql-test/main/default.result
+++ b/mysql-test/main/default.result
@@ -430,14 +430,6 @@ a	b	c	e
 2	-1	1	1
 drop table t1;
 #
-# Create or replace can delete a table on error
-#
-create table t1 (a int);
-create or replace table t1 (a int default b, b int default a);
-ERROR 01000: Expression for field `a` is referring to uninitialized field `b`
-show create table t1;
-ERROR 42S02: Table 'test.t1' doesn't exist
-#
 # Refering to other columns
 #
 create or replace table t1 (a int default 1, b int default a);
@@ -455,11 +447,11 @@ ERROR 01000: Expression for field `a` is referring to uninitialized field `a`
 create or replace table t1 (a int default b, b int default (1+1));
 create or replace table t1 (a int default 1, b int as (c), c int as (a+1));
 ERROR 01000: Expression for field `b` is referring to uninitialized field `c`
-CREATE TABLE t1 (a INT DEFAULT (DEFAULT(a)));
+CREATE OR REPLACE TABLE t1 (a INT DEFAULT (DEFAULT(a)));
 ERROR 01000: Expression for field `a` is referring to uninitialized field `a`
-CREATE TABLE t1 (a INT DEFAULT(DEFAULT(b)), b INT DEFAULT(DEFAULT(a)));
+CREATE OR REPLACE TABLE t1 (a INT DEFAULT(DEFAULT(b)), b INT DEFAULT(DEFAULT(a)));
 ERROR 01000: Expression for field `a` is referring to uninitialized field `b`
-CREATE TABLE t1 (a INT DEFAULT(DEFAULT(b)) NOT NULL, b INT DEFAULT(DEFAULT(a)) NOT NULL);
+CREATE OR REPLACE TABLE t1 (a INT DEFAULT(DEFAULT(b)) NOT NULL, b INT DEFAULT(DEFAULT(a)) NOT NULL);
 ERROR 01000: Expression for field `a` is referring to uninitialized field `b`
 #
 # Allow defaults to refer to not default fields
diff --git a/mysql-test/main/default.test b/mysql-test/main/default.test
index c0561deac67..bc318cd0853 100644
--- a/mysql-test/main/default.test
+++ b/mysql-test/main/default.test
@@ -314,15 +314,6 @@ select * from t1;
 drop table t1;
 
 --echo #
---echo # Create or replace can delete a table on error
---echo #
-create table t1 (a int);
---error ER_EXPRESSION_REFERS_TO_UNINIT_FIELD
-create or replace table t1 (a int default b, b int default a);
---error ER_NO_SUCH_TABLE
-show create table t1;
-
---echo #
 --echo # Refering to other columns
 --echo #
 
@@ -344,11 +335,11 @@ create or replace table t1 (a int default b, b int default (1+1));
 --error ER_EXPRESSION_REFERS_TO_UNINIT_FIELD
 create or replace table t1 (a int default 1, b int as (c), c int as (a+1));
 --error ER_EXPRESSION_REFERS_TO_UNINIT_FIELD
-CREATE TABLE t1 (a INT DEFAULT (DEFAULT(a)));
+CREATE OR REPLACE TABLE t1 (a INT DEFAULT (DEFAULT(a)));
 --error ER_EXPRESSION_REFERS_TO_UNINIT_FIELD
-CREATE TABLE t1 (a INT DEFAULT(DEFAULT(b)), b INT DEFAULT(DEFAULT(a)));
+CREATE OR REPLACE TABLE t1 (a INT DEFAULT(DEFAULT(b)), b INT DEFAULT(DEFAULT(a)));
 --error ER_EXPRESSION_REFERS_TO_UNINIT_FIELD
-CREATE TABLE t1 (a INT DEFAULT(DEFAULT(b)) NOT NULL, b INT DEFAULT(DEFAULT(a)) NOT NULL);
+CREATE OR REPLACE TABLE t1 (a INT DEFAULT(DEFAULT(b)) NOT NULL, b INT DEFAULT(DEFAULT(a)) NOT NULL);
 
 --echo #
 --echo # Allow defaults to refer to not default fields
diff --git a/mysql-test/main/long_unique_bugs.result b/mysql-test/main/long_unique_bugs.result
index 043361f77f8..e514e17011e 100644
--- a/mysql-test/main/long_unique_bugs.result
+++ b/mysql-test/main/long_unique_bugs.result
@@ -343,7 +343,8 @@ c	d
 create or replace table t2 (a int, b blob, unique(b)) as select * from t1;
 ERROR 23000: Duplicate entry 'bar' for key 'b'
 select * from t2;
-ERROR 42S02: Table 'test.t2' doesn't exist
+c	d
+3	bar
 create or replace table t2 (a int, b blob, unique(b)) ignore as select * from t1;
 Warnings:
 Warning	1062	Duplicate entry 'bar' for key 'b'
diff --git a/mysql-test/main/long_unique_bugs.test b/mysql-test/main/long_unique_bugs.test
index 3e0e35e6162..e6b006fc2cd 100644
--- a/mysql-test/main/long_unique_bugs.test
+++ b/mysql-test/main/long_unique_bugs.test
@@ -422,7 +422,6 @@ delete from t2 using t1, t2 where t1.a=t2.c and t1.b='foo';
 # CREATE...SELECT
 --error ER_DUP_ENTRY
 create or replace table t2 (a int, b blob, unique(b)) as select * from t1;
---error ER_NO_SUCH_TABLE
   select * from t2;
 create or replace table t2 (a int, b blob, unique(b)) ignore as select * from t1;
   select * from t2;
diff --git a/mysql-test/suite/atomic/create_replace,expensive_rename,row.rdiff b/mysql-test/suite/atomic/create_replace,expensive_rename,row.rdiff
new file mode 100644
index 00000000000..95aa7d5eee3
--- /dev/null
+++ b/mysql-test/suite/atomic/create_replace,expensive_rename,row.rdiff
@@ -0,0 +1,354 @@
+--- create_replace.result
++++ create_replace,expensive_rename,row.reject
+@@ -1,3 +1,4 @@
++set @@debug_dbug="+d,ddl_log_expensive_rename";
+ set @@default_storage_engine=Aria;
+ create table t1 (c int);
+ create or replace table t1 (a int, b int, key k (a), key k (b));
+@@ -11,60 +12,28 @@
+ flush tables;
+ # QUERY: CREATE OR REPLACE TABLE t1 (new int)
+ # CRASH POINT: ddl_log_replace_after_rename_table
++# No crash!
+ t1.MAD
+ t1.MAI
+-t1.TRG
+ t1.frm
++master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 (new int)
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+-  `old` int(11) DEFAULT NULL
++  `new` int(11) DEFAULT NULL
+ ) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+-a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_before_create_frm
+-t1.MAD
+-t1.MAI
+-t1.TRG
+-t1.frm
+-Table	Create Table
+-t1	CREATE TABLE `t1` (
+-  `old` int(11) DEFAULT NULL
+-) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
++master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated by ddl recovery */
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+-a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_before_create_table
+-t1.MAD
+-t1.MAI
+-t1.TRG
+-t1.frm
+-Table	Create Table
+-t1	CREATE TABLE `t1` (
+-  `old` int(11) DEFAULT NULL
+-) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
++master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated by ddl recovery */
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+-a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_create_table
+-t1.MAD
+-t1.MAI
+-t1.TRG
+-t1.frm
+-Table	Create Table
+-t1	CREATE TABLE `t1` (
+-  `old` int(11) DEFAULT NULL
+-) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
++master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated by ddl recovery */
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+-a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_drop
+-t1.MAD
+-t1.MAI
+-t1.TRG
+-t1.frm
+-Table	Create Table
+-t1	CREATE TABLE `t1` (
+-  `old` int(11) DEFAULT NULL
+-) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
++master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated by ddl recovery */
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+-a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_send_data
+ # No crash!
+ t1.MAD
+@@ -77,16 +46,8 @@
+ ) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # CRASH POINT: ddl_log_create_before_binlog
+-t1.MAD
+-t1.MAI
+-t1.TRG
+-t1.frm
+-Table	Create Table
+-t1	CREATE TABLE `t1` (
+-  `old` int(11) DEFAULT NULL
+-) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
++master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated by ddl recovery */
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+-a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_prepare_eof
+ # No crash!
+ t1.MAD
+@@ -141,60 +102,29 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # QUERY: CREATE OR REPLACE TABLE t1 LIKE const_table
+ # CRASH POINT: ddl_log_replace_after_rename_table
+-t1.MAD
+-t1.MAI
+-t1.TRG
++# No crash!
++t1.MYD
++t1.MYI
+ t1.frm
++master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 LIKE const_table
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+-  `old` int(11) DEFAULT NULL
+-) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
++  `new` int(11) DEFAULT NULL,
++  `b` int(11) DEFAULT NULL
++) ENGINE=MyISAM DEFAULT CHARSET=latin1
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+-a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_before_create_frm
+-t1.MAD
+-t1.MAI
+-t1.TRG
+-t1.frm
+-Table	Create Table
+-t1	CREATE TABLE `t1` (
+-  `old` int(11) DEFAULT NULL
+-) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
++master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated by ddl recovery */
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+-a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_before_create_table
+-t1.MAD
+-t1.MAI
+-t1.TRG
+-t1.frm
+-Table	Create Table
+-t1	CREATE TABLE `t1` (
+-  `old` int(11) DEFAULT NULL
+-) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
++master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated by ddl recovery */
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+-a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_create_table
+-t1.MAD
+-t1.MAI
+-t1.TRG
+-t1.frm
+-Table	Create Table
+-t1	CREATE TABLE `t1` (
+-  `old` int(11) DEFAULT NULL
+-) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
++master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated by ddl recovery */
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+-a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_drop
+-t1.MAD
+-t1.MAI
+-t1.TRG
+-t1.frm
+-Table	Create Table
+-t1	CREATE TABLE `t1` (
+-  `old` int(11) DEFAULT NULL
+-) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
++master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated by ddl recovery */
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+-a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_send_data
+ # No crash!
+ t1.MYD
+@@ -208,16 +138,8 @@
+ ) ENGINE=MyISAM DEFAULT CHARSET=latin1
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # CRASH POINT: ddl_log_create_before_binlog
+-t1.MAD
+-t1.MAI
+-t1.TRG
+-t1.frm
+-Table	Create Table
+-t1	CREATE TABLE `t1` (
+-  `old` int(11) DEFAULT NULL
+-) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
++master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated by ddl recovery */
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+-a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_prepare_eof
+ # No crash!
+ t1.MYD
+@@ -277,108 +199,54 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # QUERY: CREATE OR REPLACE TABLE t1 SELECT * from const_table
+ # CRASH POINT: ddl_log_replace_after_rename_table
++# No crash!
+ t1.MAD
+ t1.MAI
+-t1.TRG
+ t1.frm
++master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE `t1` (
++  `new` int(11) DEFAULT NULL,
++  `b` int(11) DEFAULT NULL
++) PAGE_CHECKSUM=1
++master-bin.000001	#	Annotate_rows	#	#	CREATE OR REPLACE TABLE t1 SELECT * from const_table
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+-  `old` int(11) DEFAULT NULL
++  `new` int(11) DEFAULT NULL,
++  `b` int(11) DEFAULT NULL
+ ) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+-old
++new	b
++1	1
++2	2
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+-a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_before_create_frm
+-t1.MAD
+-t1.MAI
+-t1.TRG
+-t1.frm
+-Table	Create Table
+-t1	CREATE TABLE `t1` (
+-  `old` int(11) DEFAULT NULL
+-) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+-old
++master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated by ddl recovery */
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+-a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_before_create_table
+-t1.MAD
+-t1.MAI
+-t1.TRG
+-t1.frm
+-Table	Create Table
+-t1	CREATE TABLE `t1` (
+-  `old` int(11) DEFAULT NULL
+-) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+-old
++master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated by ddl recovery */
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+-a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_create_table
+-t1.MAD
+-t1.MAI
+-t1.TRG
+-t1.frm
+-Table	Create Table
+-t1	CREATE TABLE `t1` (
+-  `old` int(11) DEFAULT NULL
+-) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+-old
++master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated by ddl recovery */
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+-a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_drop
+-t1.MAD
+-t1.MAI
+-t1.TRG
+-t1.frm
+-Table	Create Table
+-t1	CREATE TABLE `t1` (
+-  `old` int(11) DEFAULT NULL
+-) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+-old
++master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated by ddl recovery */
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+-a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_send_data
+-t1.MAD
+-t1.MAI
+-t1.TRG
+-t1.frm
+-Table	Create Table
+-t1	CREATE TABLE `t1` (
+-  `old` int(11) DEFAULT NULL
+-) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+-old
++master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated by ddl recovery */
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+-a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_before_binlog
+-t1.MAD
+-t1.MAI
+-t1.TRG
+-t1.frm
+-Table	Create Table
+-t1	CREATE TABLE `t1` (
+-  `old` int(11) DEFAULT NULL
+-) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+-old
++master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated by ddl recovery */
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+-a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_prepare_eof
+-t1.MAD
+-t1.MAI
+-t1.frm
+-master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 SELECT * from const_table
+-Table	Create Table
+-t1	CREATE TABLE `t1` (
+-  `new` int(11) DEFAULT NULL,
+-  `b` int(11) DEFAULT NULL
+-) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+-new	b
+-1	1
+-2	2
++master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated by ddl recovery */
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # CRASH POINT: ddl_log_create_after_binlog
+ t1.MAD
+ t1.MAI
+ t1.frm
+-master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 SELECT * from const_table
++master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE `t1` (
++  `new` int(11) DEFAULT NULL,
++  `b` int(11) DEFAULT NULL
++) PAGE_CHECKSUM=1
++master-bin.000001	#	Annotate_rows	#	#	CREATE OR REPLACE TABLE t1 SELECT * from const_table
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+   `new` int(11) DEFAULT NULL,
+@@ -392,7 +260,11 @@
+ t1.MAD
+ t1.MAI
+ t1.frm
+-master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 SELECT * from const_table
++master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE `t1` (
++  `new` int(11) DEFAULT NULL,
++  `b` int(11) DEFAULT NULL
++) PAGE_CHECKSUM=1
++master-bin.000001	#	Annotate_rows	#	#	CREATE OR REPLACE TABLE t1 SELECT * from const_table
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+   `new` int(11) DEFAULT NULL,
+@@ -406,7 +278,11 @@
+ t1.MAD
+ t1.MAI
+ t1.frm
+-master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 SELECT * from const_table
++master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE `t1` (
++  `new` int(11) DEFAULT NULL,
++  `b` int(11) DEFAULT NULL
++) PAGE_CHECKSUM=1
++master-bin.000001	#	Annotate_rows	#	#	CREATE OR REPLACE TABLE t1 SELECT * from const_table
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+   `new` int(11) DEFAULT NULL,
+@@ -420,7 +296,11 @@
+ t1.MAD
+ t1.MAI
+ t1.frm
+-master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 SELECT * from const_table
++master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE `t1` (
++  `new` int(11) DEFAULT NULL,
++  `b` int(11) DEFAULT NULL
++) PAGE_CHECKSUM=1
++master-bin.000001	#	Annotate_rows	#	#	CREATE OR REPLACE TABLE t1 SELECT * from const_table
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+   `new` int(11) DEFAULT NULL,
diff --git a/mysql-test/suite/atomic/create_replace,expensive_rename.rdiff b/mysql-test/suite/atomic/create_replace,expensive_rename.rdiff
new file mode 100644
index 00000000000..44b13476be8
--- /dev/null
+++ b/mysql-test/suite/atomic/create_replace,expensive_rename.rdiff
@@ -0,0 +1,286 @@
+--- create_replace.result
++++ create_replace,expensive_rename.reject
+@@ -1,3 +1,4 @@
++set @@debug_dbug="+d,ddl_log_expensive_rename";
+ set @@default_storage_engine=Aria;
+ create table t1 (c int);
+ create or replace table t1 (a int, b int, key k (a), key k (b));
+@@ -11,60 +12,28 @@
+ flush tables;
+ # QUERY: CREATE OR REPLACE TABLE t1 (new int)
+ # CRASH POINT: ddl_log_replace_after_rename_table
++# No crash!
+ t1.MAD
+ t1.MAI
+-t1.TRG
+ t1.frm
++master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 (new int)
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+-  `old` int(11) DEFAULT NULL
++  `new` int(11) DEFAULT NULL
+ ) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+-a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_before_create_frm
+-t1.MAD
+-t1.MAI
+-t1.TRG
+-t1.frm
+-Table	Create Table
+-t1	CREATE TABLE `t1` (
+-  `old` int(11) DEFAULT NULL
+-) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
++master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated by ddl recovery */
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+-a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_before_create_table
+-t1.MAD
+-t1.MAI
+-t1.TRG
+-t1.frm
+-Table	Create Table
+-t1	CREATE TABLE `t1` (
+-  `old` int(11) DEFAULT NULL
+-) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
++master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated by ddl recovery */
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+-a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_create_table
+-t1.MAD
+-t1.MAI
+-t1.TRG
+-t1.frm
+-Table	Create Table
+-t1	CREATE TABLE `t1` (
+-  `old` int(11) DEFAULT NULL
+-) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
++master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated by ddl recovery */
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+-a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_drop
+-t1.MAD
+-t1.MAI
+-t1.TRG
+-t1.frm
+-Table	Create Table
+-t1	CREATE TABLE `t1` (
+-  `old` int(11) DEFAULT NULL
+-) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
++master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated by ddl recovery */
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+-a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_send_data
+ # No crash!
+ t1.MAD
+@@ -77,16 +46,8 @@
+ ) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # CRASH POINT: ddl_log_create_before_binlog
+-t1.MAD
+-t1.MAI
+-t1.TRG
+-t1.frm
+-Table	Create Table
+-t1	CREATE TABLE `t1` (
+-  `old` int(11) DEFAULT NULL
+-) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
++master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated by ddl recovery */
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+-a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_prepare_eof
+ # No crash!
+ t1.MAD
+@@ -141,60 +102,29 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # QUERY: CREATE OR REPLACE TABLE t1 LIKE const_table
+ # CRASH POINT: ddl_log_replace_after_rename_table
+-t1.MAD
+-t1.MAI
+-t1.TRG
++# No crash!
++t1.MYD
++t1.MYI
+ t1.frm
++master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 LIKE const_table
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+-  `old` int(11) DEFAULT NULL
+-) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
++  `new` int(11) DEFAULT NULL,
++  `b` int(11) DEFAULT NULL
++) ENGINE=MyISAM DEFAULT CHARSET=latin1
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+-a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_before_create_frm
+-t1.MAD
+-t1.MAI
+-t1.TRG
+-t1.frm
+-Table	Create Table
+-t1	CREATE TABLE `t1` (
+-  `old` int(11) DEFAULT NULL
+-) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
++master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated by ddl recovery */
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+-a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_before_create_table
+-t1.MAD
+-t1.MAI
+-t1.TRG
+-t1.frm
+-Table	Create Table
+-t1	CREATE TABLE `t1` (
+-  `old` int(11) DEFAULT NULL
+-) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
++master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated by ddl recovery */
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+-a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_create_table
+-t1.MAD
+-t1.MAI
+-t1.TRG
+-t1.frm
+-Table	Create Table
+-t1	CREATE TABLE `t1` (
+-  `old` int(11) DEFAULT NULL
+-) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
++master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated by ddl recovery */
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+-a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_drop
+-t1.MAD
+-t1.MAI
+-t1.TRG
+-t1.frm
+-Table	Create Table
+-t1	CREATE TABLE `t1` (
+-  `old` int(11) DEFAULT NULL
+-) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
++master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated by ddl recovery */
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+-a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_send_data
+ # No crash!
+ t1.MYD
+@@ -208,16 +138,8 @@
+ ) ENGINE=MyISAM DEFAULT CHARSET=latin1
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # CRASH POINT: ddl_log_create_before_binlog
+-t1.MAD
+-t1.MAI
+-t1.TRG
+-t1.frm
+-Table	Create Table
+-t1	CREATE TABLE `t1` (
+-  `old` int(11) DEFAULT NULL
+-) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
++master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated by ddl recovery */
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+-a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_prepare_eof
+ # No crash!
+ t1.MYD
+@@ -277,89 +199,38 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # QUERY: CREATE OR REPLACE TABLE t1 SELECT * from const_table
+ # CRASH POINT: ddl_log_replace_after_rename_table
++# No crash!
+ t1.MAD
+ t1.MAI
+-t1.TRG
+ t1.frm
++master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 SELECT * from const_table
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+-  `old` int(11) DEFAULT NULL
++  `new` int(11) DEFAULT NULL,
++  `b` int(11) DEFAULT NULL
+ ) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+-old
++new	b
++1	1
++2	2
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+-a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_before_create_frm
+-t1.MAD
+-t1.MAI
+-t1.TRG
+-t1.frm
+-Table	Create Table
+-t1	CREATE TABLE `t1` (
+-  `old` int(11) DEFAULT NULL
+-) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+-old
++master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated by ddl recovery */
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+-a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_before_create_table
+-t1.MAD
+-t1.MAI
+-t1.TRG
+-t1.frm
+-Table	Create Table
+-t1	CREATE TABLE `t1` (
+-  `old` int(11) DEFAULT NULL
+-) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+-old
++master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated by ddl recovery */
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+-a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_create_table
+-t1.MAD
+-t1.MAI
+-t1.TRG
+-t1.frm
+-Table	Create Table
+-t1	CREATE TABLE `t1` (
+-  `old` int(11) DEFAULT NULL
+-) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+-old
++master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated by ddl recovery */
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+-a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_drop
+-t1.MAD
+-t1.MAI
+-t1.TRG
+-t1.frm
+-Table	Create Table
+-t1	CREATE TABLE `t1` (
+-  `old` int(11) DEFAULT NULL
+-) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+-old
++master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated by ddl recovery */
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+-a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_send_data
+-t1.MAD
+-t1.MAI
+-t1.TRG
+-t1.frm
+-Table	Create Table
+-t1	CREATE TABLE `t1` (
+-  `old` int(11) DEFAULT NULL
+-) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+-old
++master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated by ddl recovery */
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+-a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_before_binlog
+-t1.MAD
+-t1.MAI
+-t1.TRG
+-t1.frm
+-Table	Create Table
+-t1	CREATE TABLE `t1` (
+-  `old` int(11) DEFAULT NULL
+-) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+-old
++master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated by ddl recovery */
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+-a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_prepare_eof
+ t1.MAD
+ t1.MAI
diff --git a/mysql-test/suite/atomic/create_replace,ib,row.rdiff b/mysql-test/suite/atomic/create_replace,ib,row.rdiff
new file mode 100644
index 00000000000..49eede23bf3
--- /dev/null
+++ b/mysql-test/suite/atomic/create_replace,ib,row.rdiff
@@ -0,0 +1,389 @@
+--- create_replace.result
++++ create_replace,ib,row.reject
+@@ -1,4 +1,4 @@
+-set @@default_storage_engine=Aria;
++set @@default_storage_engine=InnoDB;
+ create table t1 (c int);
+ create or replace table t1 (a int, b int, key k (a), key k (b));
+ ERROR 42000: Duplicate key name 'k'
+@@ -11,10 +11,9 @@
+ flush tables;
+ # QUERY: CREATE OR REPLACE TABLE t1 (new int)
+ # CRASH POINT: ddl_log_replace_after_rename_table
+-t1.MAD
+-t1.MAI
+ t1.TRG
+ t1.frm
++t1.ibd
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+   `old` int(11) DEFAULT NULL
+@@ -22,10 +21,9 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_before_create_frm
+-t1.MAD
+-t1.MAI
+ t1.TRG
+ t1.frm
++t1.ibd
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+   `old` int(11) DEFAULT NULL
+@@ -33,10 +31,9 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_before_create_table
+-t1.MAD
+-t1.MAI
+ t1.TRG
+ t1.frm
++t1.ibd
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+   `old` int(11) DEFAULT NULL
+@@ -44,10 +41,9 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_create_table
+-t1.MAD
+-t1.MAI
+ t1.TRG
+ t1.frm
++t1.ibd
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+   `old` int(11) DEFAULT NULL
+@@ -55,10 +51,9 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_drop
+-t1.MAD
+-t1.MAI
+ t1.TRG
+ t1.frm
++t1.ibd
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+   `old` int(11) DEFAULT NULL
+@@ -67,9 +62,8 @@
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_send_data
+ # No crash!
+-t1.MAD
+-t1.MAI
+ t1.frm
++t1.ibd
+ master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 (new int)
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+@@ -77,10 +71,9 @@
+ ) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # CRASH POINT: ddl_log_create_before_binlog
+-t1.MAD
+-t1.MAI
+ t1.TRG
+ t1.frm
++t1.ibd
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+   `old` int(11) DEFAULT NULL
+@@ -89,9 +82,8 @@
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_prepare_eof
+ # No crash!
+-t1.MAD
+-t1.MAI
+ t1.frm
++t1.ibd
+ master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 (new int)
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+@@ -99,9 +91,8 @@
+ ) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # CRASH POINT: ddl_log_create_after_binlog
+-t1.MAD
+-t1.MAI
+ t1.frm
++t1.ibd
+ master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 (new int)
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+@@ -110,9 +101,8 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # CRASH POINT: ddl_log_create_log_complete
+ # No crash!
+-t1.MAD
+-t1.MAI
+ t1.frm
++t1.ibd
+ master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 (new int)
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+@@ -120,9 +110,8 @@
+ ) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # CRASH POINT: ddl_log_replace_before_remove_backup
+-t1.MAD
+-t1.MAI
+ t1.frm
++t1.ibd
+ master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 (new int)
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+@@ -130,9 +119,8 @@
+ ) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # CRASH POINT: ddl_log_replace_after_remove_backup
+-t1.MAD
+-t1.MAI
+ t1.frm
++t1.ibd
+ master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 (new int)
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+@@ -141,10 +129,9 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # QUERY: CREATE OR REPLACE TABLE t1 LIKE const_table
+ # CRASH POINT: ddl_log_replace_after_rename_table
+-t1.MAD
+-t1.MAI
+ t1.TRG
+ t1.frm
++t1.ibd
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+   `old` int(11) DEFAULT NULL
+@@ -152,10 +139,9 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_before_create_frm
+-t1.MAD
+-t1.MAI
+ t1.TRG
+ t1.frm
++t1.ibd
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+   `old` int(11) DEFAULT NULL
+@@ -163,10 +149,9 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_before_create_table
+-t1.MAD
+-t1.MAI
+ t1.TRG
+ t1.frm
++t1.ibd
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+   `old` int(11) DEFAULT NULL
+@@ -174,10 +159,9 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_create_table
+-t1.MAD
+-t1.MAI
+ t1.TRG
+ t1.frm
++t1.ibd
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+   `old` int(11) DEFAULT NULL
+@@ -185,10 +169,9 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_drop
+-t1.MAD
+-t1.MAI
+ t1.TRG
+ t1.frm
++t1.ibd
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+   `old` int(11) DEFAULT NULL
+@@ -208,10 +191,9 @@
+ ) ENGINE=MyISAM DEFAULT CHARSET=latin1
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # CRASH POINT: ddl_log_create_before_binlog
+-t1.MAD
+-t1.MAI
+ t1.TRG
+ t1.frm
++t1.ibd
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+   `old` int(11) DEFAULT NULL
+@@ -277,10 +259,9 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # QUERY: CREATE OR REPLACE TABLE t1 SELECT * from const_table
+ # CRASH POINT: ddl_log_replace_after_rename_table
+-t1.MAD
+-t1.MAI
+ t1.TRG
+ t1.frm
++t1.ibd
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+   `old` int(11) DEFAULT NULL
+@@ -289,10 +270,9 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_before_create_frm
+-t1.MAD
+-t1.MAI
+ t1.TRG
+ t1.frm
++t1.ibd
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+   `old` int(11) DEFAULT NULL
+@@ -301,10 +281,9 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_before_create_table
+-t1.MAD
+-t1.MAI
+ t1.TRG
+ t1.frm
++t1.ibd
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+   `old` int(11) DEFAULT NULL
+@@ -313,10 +292,9 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_create_table
+-t1.MAD
+-t1.MAI
+ t1.TRG
+ t1.frm
++t1.ibd
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+   `old` int(11) DEFAULT NULL
+@@ -325,10 +303,9 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_drop
+-t1.MAD
+-t1.MAI
+ t1.TRG
+ t1.frm
++t1.ibd
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+   `old` int(11) DEFAULT NULL
+@@ -337,10 +314,9 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_send_data
+-t1.MAD
+-t1.MAI
+ t1.TRG
+ t1.frm
++t1.ibd
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+   `old` int(11) DEFAULT NULL
+@@ -349,10 +325,9 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_before_binlog
+-t1.MAD
+-t1.MAI
+ t1.TRG
+ t1.frm
++t1.ibd
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+   `old` int(11) DEFAULT NULL
+@@ -361,24 +336,24 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_prepare_eof
+-t1.MAD
+-t1.MAI
++t1.TRG
+ t1.frm
+-master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 SELECT * from const_table
++t1.ibd
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+-  `new` int(11) DEFAULT NULL,
+-  `b` int(11) DEFAULT NULL
++  `old` int(11) DEFAULT NULL
+ ) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+-new	b
+-1	1
+-2	2
++old
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
++a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_binlog
+-t1.MAD
+-t1.MAI
+ t1.frm
+-master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 SELECT * from const_table
++t1.ibd
++master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE `t1` (
++  `new` int(11) DEFAULT NULL,
++  `b` int(11) DEFAULT NULL
++)
++master-bin.000001	#	Annotate_rows	#	#	CREATE OR REPLACE TABLE t1 SELECT * from const_table
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+   `new` int(11) DEFAULT NULL,
+@@ -389,10 +364,13 @@
+ 2	2
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # CRASH POINT: ddl_log_create_log_complete
+-t1.MAD
+-t1.MAI
+ t1.frm
+-master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 SELECT * from const_table
++t1.ibd
++master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE `t1` (
++  `new` int(11) DEFAULT NULL,
++  `b` int(11) DEFAULT NULL
++)
++master-bin.000001	#	Annotate_rows	#	#	CREATE OR REPLACE TABLE t1 SELECT * from const_table
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+   `new` int(11) DEFAULT NULL,
+@@ -403,10 +381,13 @@
+ 2	2
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # CRASH POINT: ddl_log_replace_before_remove_backup
+-t1.MAD
+-t1.MAI
+ t1.frm
+-master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 SELECT * from const_table
++t1.ibd
++master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE `t1` (
++  `new` int(11) DEFAULT NULL,
++  `b` int(11) DEFAULT NULL
++)
++master-bin.000001	#	Annotate_rows	#	#	CREATE OR REPLACE TABLE t1 SELECT * from const_table
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+   `new` int(11) DEFAULT NULL,
+@@ -417,10 +398,13 @@
+ 2	2
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # CRASH POINT: ddl_log_replace_after_remove_backup
+-t1.MAD
+-t1.MAI
+ t1.frm
+-master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 SELECT * from const_table
++t1.ibd
++master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE `t1` (
++  `new` int(11) DEFAULT NULL,
++  `b` int(11) DEFAULT NULL
++)
++master-bin.000001	#	Annotate_rows	#	#	CREATE OR REPLACE TABLE t1 SELECT * from const_table
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+   `new` int(11) DEFAULT NULL,
diff --git a/mysql-test/suite/atomic/create_replace,ib.rdiff b/mysql-test/suite/atomic/create_replace,ib.rdiff
new file mode 100644
index 00000000000..d6eae72a541
--- /dev/null
+++ b/mysql-test/suite/atomic/create_replace,ib.rdiff
@@ -0,0 +1,357 @@
+--- create_replace.result
++++ create_replace,ib.reject
+@@ -1,4 +1,4 @@
+-set @@default_storage_engine=Aria;
++set @@default_storage_engine=InnoDB;
+ create table t1 (c int);
+ create or replace table t1 (a int, b int, key k (a), key k (b));
+ ERROR 42000: Duplicate key name 'k'
+@@ -11,10 +11,9 @@
+ flush tables;
+ # QUERY: CREATE OR REPLACE TABLE t1 (new int)
+ # CRASH POINT: ddl_log_replace_after_rename_table
+-t1.MAD
+-t1.MAI
+ t1.TRG
+ t1.frm
++t1.ibd
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+   `old` int(11) DEFAULT NULL
+@@ -22,10 +21,9 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_before_create_frm
+-t1.MAD
+-t1.MAI
+ t1.TRG
+ t1.frm
++t1.ibd
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+   `old` int(11) DEFAULT NULL
+@@ -33,10 +31,9 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_before_create_table
+-t1.MAD
+-t1.MAI
+ t1.TRG
+ t1.frm
++t1.ibd
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+   `old` int(11) DEFAULT NULL
+@@ -44,10 +41,9 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_create_table
+-t1.MAD
+-t1.MAI
+ t1.TRG
+ t1.frm
++t1.ibd
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+   `old` int(11) DEFAULT NULL
+@@ -55,10 +51,9 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_drop
+-t1.MAD
+-t1.MAI
+ t1.TRG
+ t1.frm
++t1.ibd
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+   `old` int(11) DEFAULT NULL
+@@ -67,9 +62,8 @@
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_send_data
+ # No crash!
+-t1.MAD
+-t1.MAI
+ t1.frm
++t1.ibd
+ master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 (new int)
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+@@ -77,10 +71,9 @@
+ ) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # CRASH POINT: ddl_log_create_before_binlog
+-t1.MAD
+-t1.MAI
+ t1.TRG
+ t1.frm
++t1.ibd
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+   `old` int(11) DEFAULT NULL
+@@ -89,9 +82,8 @@
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_prepare_eof
+ # No crash!
+-t1.MAD
+-t1.MAI
+ t1.frm
++t1.ibd
+ master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 (new int)
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+@@ -99,9 +91,8 @@
+ ) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # CRASH POINT: ddl_log_create_after_binlog
+-t1.MAD
+-t1.MAI
+ t1.frm
++t1.ibd
+ master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 (new int)
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+@@ -110,9 +101,8 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # CRASH POINT: ddl_log_create_log_complete
+ # No crash!
+-t1.MAD
+-t1.MAI
+ t1.frm
++t1.ibd
+ master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 (new int)
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+@@ -120,9 +110,8 @@
+ ) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # CRASH POINT: ddl_log_replace_before_remove_backup
+-t1.MAD
+-t1.MAI
+ t1.frm
++t1.ibd
+ master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 (new int)
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+@@ -130,9 +119,8 @@
+ ) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # CRASH POINT: ddl_log_replace_after_remove_backup
+-t1.MAD
+-t1.MAI
+ t1.frm
++t1.ibd
+ master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 (new int)
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+@@ -141,10 +129,9 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # QUERY: CREATE OR REPLACE TABLE t1 LIKE const_table
+ # CRASH POINT: ddl_log_replace_after_rename_table
+-t1.MAD
+-t1.MAI
+ t1.TRG
+ t1.frm
++t1.ibd
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+   `old` int(11) DEFAULT NULL
+@@ -152,10 +139,9 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_before_create_frm
+-t1.MAD
+-t1.MAI
+ t1.TRG
+ t1.frm
++t1.ibd
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+   `old` int(11) DEFAULT NULL
+@@ -163,10 +149,9 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_before_create_table
+-t1.MAD
+-t1.MAI
+ t1.TRG
+ t1.frm
++t1.ibd
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+   `old` int(11) DEFAULT NULL
+@@ -174,10 +159,9 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_create_table
+-t1.MAD
+-t1.MAI
+ t1.TRG
+ t1.frm
++t1.ibd
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+   `old` int(11) DEFAULT NULL
+@@ -185,10 +169,9 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_drop
+-t1.MAD
+-t1.MAI
+ t1.TRG
+ t1.frm
++t1.ibd
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+   `old` int(11) DEFAULT NULL
+@@ -208,10 +191,9 @@
+ ) ENGINE=MyISAM DEFAULT CHARSET=latin1
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # CRASH POINT: ddl_log_create_before_binlog
+-t1.MAD
+-t1.MAI
+ t1.TRG
+ t1.frm
++t1.ibd
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+   `old` int(11) DEFAULT NULL
+@@ -277,10 +259,9 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # QUERY: CREATE OR REPLACE TABLE t1 SELECT * from const_table
+ # CRASH POINT: ddl_log_replace_after_rename_table
+-t1.MAD
+-t1.MAI
+ t1.TRG
+ t1.frm
++t1.ibd
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+   `old` int(11) DEFAULT NULL
+@@ -289,10 +270,9 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_before_create_frm
+-t1.MAD
+-t1.MAI
+ t1.TRG
+ t1.frm
++t1.ibd
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+   `old` int(11) DEFAULT NULL
+@@ -301,10 +281,9 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_before_create_table
+-t1.MAD
+-t1.MAI
+ t1.TRG
+ t1.frm
++t1.ibd
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+   `old` int(11) DEFAULT NULL
+@@ -313,10 +292,9 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_create_table
+-t1.MAD
+-t1.MAI
+ t1.TRG
+ t1.frm
++t1.ibd
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+   `old` int(11) DEFAULT NULL
+@@ -325,10 +303,9 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_drop
+-t1.MAD
+-t1.MAI
+ t1.TRG
+ t1.frm
++t1.ibd
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+   `old` int(11) DEFAULT NULL
+@@ -337,10 +314,9 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_send_data
+-t1.MAD
+-t1.MAI
+ t1.TRG
+ t1.frm
++t1.ibd
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+   `old` int(11) DEFAULT NULL
+@@ -349,10 +325,9 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_before_binlog
+-t1.MAD
+-t1.MAI
+ t1.TRG
+ t1.frm
++t1.ibd
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+   `old` int(11) DEFAULT NULL
+@@ -361,9 +336,8 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_prepare_eof
+-t1.MAD
+-t1.MAI
+ t1.frm
++t1.ibd
+ master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 SELECT * from const_table
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+@@ -375,9 +349,8 @@
+ 2	2
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # CRASH POINT: ddl_log_create_after_binlog
+-t1.MAD
+-t1.MAI
+ t1.frm
++t1.ibd
+ master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 SELECT * from const_table
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+@@ -389,9 +362,8 @@
+ 2	2
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # CRASH POINT: ddl_log_create_log_complete
+-t1.MAD
+-t1.MAI
+ t1.frm
++t1.ibd
+ master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 SELECT * from const_table
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+@@ -403,9 +375,8 @@
+ 2	2
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # CRASH POINT: ddl_log_replace_before_remove_backup
+-t1.MAD
+-t1.MAI
+ t1.frm
++t1.ibd
+ master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 SELECT * from const_table
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+@@ -417,9 +388,8 @@
+ 2	2
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # CRASH POINT: ddl_log_replace_after_remove_backup
+-t1.MAD
+-t1.MAI
+ t1.frm
++t1.ibd
+ master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 SELECT * from const_table
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
diff --git a/mysql-test/suite/atomic/create_replace,myisam,row.rdiff b/mysql-test/suite/atomic/create_replace,myisam,row.rdiff
new file mode 100644
index 00000000000..b7809b8259d
--- /dev/null
+++ b/mysql-test/suite/atomic/create_replace,myisam,row.rdiff
@@ -0,0 +1,437 @@
+--- create_replace.result
++++ create_replace,myisam,row.reject
+@@ -1,4 +1,4 @@
+-set @@default_storage_engine=Aria;
++set @@default_storage_engine=MyISAM;
+ create table t1 (c int);
+ create or replace table t1 (a int, b int, key k (a), key k (b));
+ ERROR 42000: Duplicate key name 'k'
+@@ -11,8 +11,8 @@
+ flush tables;
+ # QUERY: CREATE OR REPLACE TABLE t1 (new int)
+ # CRASH POINT: ddl_log_replace_after_rename_table
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.TRG
+ t1.frm
+ Table	Create Table
+@@ -22,8 +22,8 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_before_create_frm
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.TRG
+ t1.frm
+ Table	Create Table
+@@ -33,8 +33,8 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_before_create_table
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.TRG
+ t1.frm
+ Table	Create Table
+@@ -44,8 +44,8 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_create_table
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.TRG
+ t1.frm
+ Table	Create Table
+@@ -55,8 +55,8 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_drop
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.TRG
+ t1.frm
+ Table	Create Table
+@@ -67,8 +67,8 @@
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_send_data
+ # No crash!
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.frm
+ master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 (new int)
+ Table	Create Table
+@@ -77,8 +77,8 @@
+ ) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # CRASH POINT: ddl_log_create_before_binlog
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.TRG
+ t1.frm
+ Table	Create Table
+@@ -89,8 +89,8 @@
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_prepare_eof
+ # No crash!
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.frm
+ master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 (new int)
+ Table	Create Table
+@@ -99,8 +99,8 @@
+ ) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # CRASH POINT: ddl_log_create_after_binlog
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.frm
+ master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 (new int)
+ Table	Create Table
+@@ -110,8 +110,8 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # CRASH POINT: ddl_log_create_log_complete
+ # No crash!
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.frm
+ master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 (new int)
+ Table	Create Table
+@@ -120,8 +120,8 @@
+ ) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # CRASH POINT: ddl_log_replace_before_remove_backup
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.frm
+ master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 (new int)
+ Table	Create Table
+@@ -130,8 +130,8 @@
+ ) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # CRASH POINT: ddl_log_replace_after_remove_backup
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.frm
+ master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 (new int)
+ Table	Create Table
+@@ -141,8 +141,8 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # QUERY: CREATE OR REPLACE TABLE t1 LIKE const_table
+ # CRASH POINT: ddl_log_replace_after_rename_table
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.TRG
+ t1.frm
+ Table	Create Table
+@@ -152,8 +152,8 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_before_create_frm
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.TRG
+ t1.frm
+ Table	Create Table
+@@ -163,8 +163,8 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_before_create_table
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.TRG
+ t1.frm
+ Table	Create Table
+@@ -174,8 +174,8 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_create_table
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.TRG
+ t1.frm
+ Table	Create Table
+@@ -185,8 +185,8 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_drop
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.TRG
+ t1.frm
+ Table	Create Table
+@@ -205,11 +205,11 @@
+ t1	CREATE TABLE `t1` (
+   `new` int(11) DEFAULT NULL,
+   `b` int(11) DEFAULT NULL
+-) ENGINE=MyISAM DEFAULT CHARSET=latin1
++) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # CRASH POINT: ddl_log_create_before_binlog
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.TRG
+ t1.frm
+ Table	Create Table
+@@ -228,7 +228,7 @@
+ t1	CREATE TABLE `t1` (
+   `new` int(11) DEFAULT NULL,
+   `b` int(11) DEFAULT NULL
+-) ENGINE=MyISAM DEFAULT CHARSET=latin1
++) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # CRASH POINT: ddl_log_create_after_binlog
+ t1.MYD
+@@ -239,7 +239,7 @@
+ t1	CREATE TABLE `t1` (
+   `new` int(11) DEFAULT NULL,
+   `b` int(11) DEFAULT NULL
+-) ENGINE=MyISAM DEFAULT CHARSET=latin1
++) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # CRASH POINT: ddl_log_create_log_complete
+ # No crash!
+@@ -251,7 +251,7 @@
+ t1	CREATE TABLE `t1` (
+   `new` int(11) DEFAULT NULL,
+   `b` int(11) DEFAULT NULL
+-) ENGINE=MyISAM DEFAULT CHARSET=latin1
++) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # CRASH POINT: ddl_log_replace_before_remove_backup
+ t1.MYD
+@@ -262,7 +262,7 @@
+ t1	CREATE TABLE `t1` (
+   `new` int(11) DEFAULT NULL,
+   `b` int(11) DEFAULT NULL
+-) ENGINE=MyISAM DEFAULT CHARSET=latin1
++) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # CRASH POINT: ddl_log_replace_after_remove_backup
+ t1.MYD
+@@ -273,12 +273,12 @@
+ t1	CREATE TABLE `t1` (
+   `new` int(11) DEFAULT NULL,
+   `b` int(11) DEFAULT NULL
+-) ENGINE=MyISAM DEFAULT CHARSET=latin1
++) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # QUERY: CREATE OR REPLACE TABLE t1 SELECT * from const_table
+ # CRASH POINT: ddl_log_replace_after_rename_table
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.TRG
+ t1.frm
+ Table	Create Table
+@@ -289,8 +289,8 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_before_create_frm
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.TRG
+ t1.frm
+ Table	Create Table
+@@ -301,8 +301,8 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_before_create_table
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.TRG
+ t1.frm
+ Table	Create Table
+@@ -313,8 +313,8 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_create_table
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.TRG
+ t1.frm
+ Table	Create Table
+@@ -325,8 +325,8 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_drop
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.TRG
+ t1.frm
+ Table	Create Table
+@@ -337,8 +337,8 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_send_data
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.TRG
+ t1.frm
+ Table	Create Table
+@@ -349,8 +349,8 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_before_binlog
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.TRG
+ t1.frm
+ Table	Create Table
+@@ -361,74 +361,80 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_prepare_eof
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
++t1.TRG
+ t1.frm
+-master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 SELECT * from const_table
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+-  `new` int(11) DEFAULT NULL,
+-  `b` int(11) DEFAULT NULL
++  `old` int(11) DEFAULT NULL
+ ) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+-new	b
+-1	1
+-2	2
++old
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
++a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_binlog
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.frm
+-master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 SELECT * from const_table
++master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE `t1` (
++  `new` int(11) DEFAULT NULL,
++  `b` int(11) DEFAULT NULL
++)
++master-bin.000001	#	Annotate_rows	#	#	CREATE OR REPLACE TABLE t1 SELECT * from const_table
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+   `new` int(11) DEFAULT NULL,
+   `b` int(11) DEFAULT NULL
+ ) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+ new	b
+-1	1
+-2	2
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # CRASH POINT: ddl_log_create_log_complete
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.frm
+-master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 SELECT * from const_table
++master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE `t1` (
++  `new` int(11) DEFAULT NULL,
++  `b` int(11) DEFAULT NULL
++)
++master-bin.000001	#	Annotate_rows	#	#	CREATE OR REPLACE TABLE t1 SELECT * from const_table
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+   `new` int(11) DEFAULT NULL,
+   `b` int(11) DEFAULT NULL
+ ) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+ new	b
+-1	1
+-2	2
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # CRASH POINT: ddl_log_replace_before_remove_backup
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.frm
+-master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 SELECT * from const_table
++master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE `t1` (
++  `new` int(11) DEFAULT NULL,
++  `b` int(11) DEFAULT NULL
++)
++master-bin.000001	#	Annotate_rows	#	#	CREATE OR REPLACE TABLE t1 SELECT * from const_table
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+   `new` int(11) DEFAULT NULL,
+   `b` int(11) DEFAULT NULL
+ ) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+ new	b
+-1	1
+-2	2
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # CRASH POINT: ddl_log_replace_after_remove_backup
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.frm
+-master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 SELECT * from const_table
++master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE `t1` (
++  `new` int(11) DEFAULT NULL,
++  `b` int(11) DEFAULT NULL
++)
++master-bin.000001	#	Annotate_rows	#	#	CREATE OR REPLACE TABLE t1 SELECT * from const_table
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+   `new` int(11) DEFAULT NULL,
+   `b` int(11) DEFAULT NULL
+ ) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+ new	b
+-1	1
+-2	2
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ Warnings:
+ Note	1051	Unknown table 'test.t1'
diff --git a/mysql-test/suite/atomic/create_replace,myisam.rdiff b/mysql-test/suite/atomic/create_replace,myisam.rdiff
new file mode 100644
index 00000000000..68203260d54
--- /dev/null
+++ b/mysql-test/suite/atomic/create_replace,myisam.rdiff
@@ -0,0 +1,408 @@
+--- create_replace.result
++++ create_replace,myisam.reject
+@@ -1,4 +1,4 @@
+-set @@default_storage_engine=Aria;
++set @@default_storage_engine=MyISAM;
+ create table t1 (c int);
+ create or replace table t1 (a int, b int, key k (a), key k (b));
+ ERROR 42000: Duplicate key name 'k'
+@@ -11,8 +11,8 @@
+ flush tables;
+ # QUERY: CREATE OR REPLACE TABLE t1 (new int)
+ # CRASH POINT: ddl_log_replace_after_rename_table
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.TRG
+ t1.frm
+ Table	Create Table
+@@ -22,8 +22,8 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_before_create_frm
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.TRG
+ t1.frm
+ Table	Create Table
+@@ -33,8 +33,8 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_before_create_table
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.TRG
+ t1.frm
+ Table	Create Table
+@@ -44,8 +44,8 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_create_table
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.TRG
+ t1.frm
+ Table	Create Table
+@@ -55,8 +55,8 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_drop
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.TRG
+ t1.frm
+ Table	Create Table
+@@ -67,8 +67,8 @@
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_send_data
+ # No crash!
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.frm
+ master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 (new int)
+ Table	Create Table
+@@ -77,8 +77,8 @@
+ ) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # CRASH POINT: ddl_log_create_before_binlog
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.TRG
+ t1.frm
+ Table	Create Table
+@@ -89,8 +89,8 @@
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_prepare_eof
+ # No crash!
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.frm
+ master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 (new int)
+ Table	Create Table
+@@ -99,8 +99,8 @@
+ ) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # CRASH POINT: ddl_log_create_after_binlog
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.frm
+ master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 (new int)
+ Table	Create Table
+@@ -110,8 +110,8 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # CRASH POINT: ddl_log_create_log_complete
+ # No crash!
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.frm
+ master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 (new int)
+ Table	Create Table
+@@ -120,8 +120,8 @@
+ ) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # CRASH POINT: ddl_log_replace_before_remove_backup
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.frm
+ master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 (new int)
+ Table	Create Table
+@@ -130,8 +130,8 @@
+ ) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # CRASH POINT: ddl_log_replace_after_remove_backup
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.frm
+ master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 (new int)
+ Table	Create Table
+@@ -141,8 +141,8 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # QUERY: CREATE OR REPLACE TABLE t1 LIKE const_table
+ # CRASH POINT: ddl_log_replace_after_rename_table
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.TRG
+ t1.frm
+ Table	Create Table
+@@ -152,8 +152,8 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_before_create_frm
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.TRG
+ t1.frm
+ Table	Create Table
+@@ -163,8 +163,8 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_before_create_table
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.TRG
+ t1.frm
+ Table	Create Table
+@@ -174,8 +174,8 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_create_table
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.TRG
+ t1.frm
+ Table	Create Table
+@@ -185,8 +185,8 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_drop
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.TRG
+ t1.frm
+ Table	Create Table
+@@ -205,11 +205,11 @@
+ t1	CREATE TABLE `t1` (
+   `new` int(11) DEFAULT NULL,
+   `b` int(11) DEFAULT NULL
+-) ENGINE=MyISAM DEFAULT CHARSET=latin1
++) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # CRASH POINT: ddl_log_create_before_binlog
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.TRG
+ t1.frm
+ Table	Create Table
+@@ -228,7 +228,7 @@
+ t1	CREATE TABLE `t1` (
+   `new` int(11) DEFAULT NULL,
+   `b` int(11) DEFAULT NULL
+-) ENGINE=MyISAM DEFAULT CHARSET=latin1
++) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # CRASH POINT: ddl_log_create_after_binlog
+ t1.MYD
+@@ -239,7 +239,7 @@
+ t1	CREATE TABLE `t1` (
+   `new` int(11) DEFAULT NULL,
+   `b` int(11) DEFAULT NULL
+-) ENGINE=MyISAM DEFAULT CHARSET=latin1
++) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # CRASH POINT: ddl_log_create_log_complete
+ # No crash!
+@@ -251,7 +251,7 @@
+ t1	CREATE TABLE `t1` (
+   `new` int(11) DEFAULT NULL,
+   `b` int(11) DEFAULT NULL
+-) ENGINE=MyISAM DEFAULT CHARSET=latin1
++) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # CRASH POINT: ddl_log_replace_before_remove_backup
+ t1.MYD
+@@ -262,7 +262,7 @@
+ t1	CREATE TABLE `t1` (
+   `new` int(11) DEFAULT NULL,
+   `b` int(11) DEFAULT NULL
+-) ENGINE=MyISAM DEFAULT CHARSET=latin1
++) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # CRASH POINT: ddl_log_replace_after_remove_backup
+ t1.MYD
+@@ -273,12 +273,12 @@
+ t1	CREATE TABLE `t1` (
+   `new` int(11) DEFAULT NULL,
+   `b` int(11) DEFAULT NULL
+-) ENGINE=MyISAM DEFAULT CHARSET=latin1
++) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # QUERY: CREATE OR REPLACE TABLE t1 SELECT * from const_table
+ # CRASH POINT: ddl_log_replace_after_rename_table
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.TRG
+ t1.frm
+ Table	Create Table
+@@ -289,8 +289,8 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_before_create_frm
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.TRG
+ t1.frm
+ Table	Create Table
+@@ -301,8 +301,8 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_before_create_table
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.TRG
+ t1.frm
+ Table	Create Table
+@@ -313,8 +313,8 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_create_table
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.TRG
+ t1.frm
+ Table	Create Table
+@@ -325,8 +325,8 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_drop
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.TRG
+ t1.frm
+ Table	Create Table
+@@ -337,8 +337,8 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_send_data
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.TRG
+ t1.frm
+ Table	Create Table
+@@ -349,8 +349,8 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_before_binlog
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.TRG
+ t1.frm
+ Table	Create Table
+@@ -361,8 +361,8 @@
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_prepare_eof
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.frm
+ master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 SELECT * from const_table
+ Table	Create Table
+@@ -371,12 +371,10 @@
+   `b` int(11) DEFAULT NULL
+ ) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+ new	b
+-1	1
+-2	2
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # CRASH POINT: ddl_log_create_after_binlog
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.frm
+ master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 SELECT * from const_table
+ Table	Create Table
+@@ -385,12 +383,10 @@
+   `b` int(11) DEFAULT NULL
+ ) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+ new	b
+-1	1
+-2	2
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # CRASH POINT: ddl_log_create_log_complete
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.frm
+ master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 SELECT * from const_table
+ Table	Create Table
+@@ -399,12 +395,10 @@
+   `b` int(11) DEFAULT NULL
+ ) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+ new	b
+-1	1
+-2	2
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # CRASH POINT: ddl_log_replace_before_remove_backup
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.frm
+ master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 SELECT * from const_table
+ Table	Create Table
+@@ -413,12 +407,10 @@
+   `b` int(11) DEFAULT NULL
+ ) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+ new	b
+-1	1
+-2	2
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ # CRASH POINT: ddl_log_replace_after_remove_backup
+-t1.MAD
+-t1.MAI
++t1.MYD
++t1.MYI
+ t1.frm
+ master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 SELECT * from const_table
+ Table	Create Table
+@@ -427,8 +419,6 @@
+   `b` int(11) DEFAULT NULL
+ ) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+ new	b
+-1	1
+-2	2
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+ Warnings:
+ Note	1051	Unknown table 'test.t1'
diff --git a/mysql-test/suite/atomic/create_replace,row.rdiff b/mysql-test/suite/atomic/create_replace,row.rdiff
new file mode 100644
index 00000000000..db206b30dd4
--- /dev/null
+++ b/mysql-test/suite/atomic/create_replace,row.rdiff
@@ -0,0 +1,73 @@
+--- create_replace.result
++++ create_replace,row.reject
+@@ -363,22 +363,24 @@
+ # CRASH POINT: ddl_log_create_after_prepare_eof
+ t1.MAD
+ t1.MAI
++t1.TRG
+ t1.frm
+-master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 SELECT * from const_table
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+-  `new` int(11) DEFAULT NULL,
+-  `b` int(11) DEFAULT NULL
++  `old` int(11) DEFAULT NULL
+ ) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+-new	b
+-1	1
+-2	2
++old
+ Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
++a	INSERT	t1	set @s= 1	BEFORE						
+ # CRASH POINT: ddl_log_create_after_binlog
+ t1.MAD
+ t1.MAI
+ t1.frm
+-master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 SELECT * from const_table
++master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE `t1` (
++  `new` int(11) DEFAULT NULL,
++  `b` int(11) DEFAULT NULL
++) PAGE_CHECKSUM=1
++master-bin.000001	#	Annotate_rows	#	#	CREATE OR REPLACE TABLE t1 SELECT * from const_table
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+   `new` int(11) DEFAULT NULL,
+@@ -392,7 +394,11 @@
+ t1.MAD
+ t1.MAI
+ t1.frm
+-master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 SELECT * from const_table
++master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE `t1` (
++  `new` int(11) DEFAULT NULL,
++  `b` int(11) DEFAULT NULL
++) PAGE_CHECKSUM=1
++master-bin.000001	#	Annotate_rows	#	#	CREATE OR REPLACE TABLE t1 SELECT * from const_table
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+   `new` int(11) DEFAULT NULL,
+@@ -406,7 +412,11 @@
+ t1.MAD
+ t1.MAI
+ t1.frm
+-master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 SELECT * from const_table
++master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE `t1` (
++  `new` int(11) DEFAULT NULL,
++  `b` int(11) DEFAULT NULL
++) PAGE_CHECKSUM=1
++master-bin.000001	#	Annotate_rows	#	#	CREATE OR REPLACE TABLE t1 SELECT * from const_table
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+   `new` int(11) DEFAULT NULL,
+@@ -420,7 +430,11 @@
+ t1.MAD
+ t1.MAI
+ t1.frm
+-master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 SELECT * from const_table
++master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE `t1` (
++  `new` int(11) DEFAULT NULL,
++  `b` int(11) DEFAULT NULL
++) PAGE_CHECKSUM=1
++master-bin.000001	#	Annotate_rows	#	#	CREATE OR REPLACE TABLE t1 SELECT * from const_table
+ Table	Create Table
+ t1	CREATE TABLE `t1` (
+   `new` int(11) DEFAULT NULL,
diff --git a/mysql-test/suite/atomic/create_replace.combinations b/mysql-test/suite/atomic/create_replace.combinations
new file mode 100644
index 00000000000..72a359c77eb
--- /dev/null
+++ b/mysql-test/suite/atomic/create_replace.combinations
@@ -0,0 +1,5 @@
+[ib]
+[myisam]
+[aria]
+[aria_notrans]
+[expensive_rename]
diff --git a/mysql-test/suite/atomic/create_replace.result b/mysql-test/suite/atomic/create_replace.result
new file mode 100644
index 00000000000..9cc12a60219
--- /dev/null
+++ b/mysql-test/suite/atomic/create_replace.result
@@ -0,0 +1,434 @@
+set @@default_storage_engine=Aria;
+create table t1 (c int);
+create or replace table t1 (a int, b int, key k (a), key k (b));
+ERROR 42000: Duplicate key name 'k'
+create or replace table t1 (a int, b int, key k (a), key k (b));
+ERROR 42000: Duplicate key name 'k'
+drop table t1;
+# Crash recovery
+create table const_table (new int, b int) engine=myisam;
+insert into const_table values (1,1),(2,2);
+flush tables;
+# QUERY: CREATE OR REPLACE TABLE t1 (new int)
+# CRASH POINT: ddl_log_replace_after_rename_table
+t1.MAD
+t1.MAI
+t1.TRG
+t1.frm
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `old` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+a	INSERT	t1	set @s= 1	BEFORE						
+# CRASH POINT: ddl_log_create_before_create_frm
+t1.MAD
+t1.MAI
+t1.TRG
+t1.frm
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `old` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+a	INSERT	t1	set @s= 1	BEFORE						
+# CRASH POINT: ddl_log_create_before_create_table
+t1.MAD
+t1.MAI
+t1.TRG
+t1.frm
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `old` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+a	INSERT	t1	set @s= 1	BEFORE						
+# CRASH POINT: ddl_log_create_after_create_table
+t1.MAD
+t1.MAI
+t1.TRG
+t1.frm
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `old` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+a	INSERT	t1	set @s= 1	BEFORE						
+# CRASH POINT: ddl_log_create_after_drop
+t1.MAD
+t1.MAI
+t1.TRG
+t1.frm
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `old` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+a	INSERT	t1	set @s= 1	BEFORE						
+# CRASH POINT: ddl_log_create_after_send_data
+# No crash!
+t1.MAD
+t1.MAI
+t1.frm
+master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 (new int)
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `new` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+# CRASH POINT: ddl_log_create_before_binlog
+t1.MAD
+t1.MAI
+t1.TRG
+t1.frm
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `old` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+a	INSERT	t1	set @s= 1	BEFORE						
+# CRASH POINT: ddl_log_create_after_prepare_eof
+# No crash!
+t1.MAD
+t1.MAI
+t1.frm
+master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 (new int)
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `new` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+# CRASH POINT: ddl_log_create_after_binlog
+t1.MAD
+t1.MAI
+t1.frm
+master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 (new int)
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `new` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+# CRASH POINT: ddl_log_create_log_complete
+# No crash!
+t1.MAD
+t1.MAI
+t1.frm
+master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 (new int)
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `new` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+# CRASH POINT: ddl_log_replace_before_remove_backup
+t1.MAD
+t1.MAI
+t1.frm
+master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 (new int)
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `new` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+# CRASH POINT: ddl_log_replace_after_remove_backup
+t1.MAD
+t1.MAI
+t1.frm
+master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 (new int)
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `new` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+# QUERY: CREATE OR REPLACE TABLE t1 LIKE const_table
+# CRASH POINT: ddl_log_replace_after_rename_table
+t1.MAD
+t1.MAI
+t1.TRG
+t1.frm
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `old` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+a	INSERT	t1	set @s= 1	BEFORE						
+# CRASH POINT: ddl_log_create_before_create_frm
+t1.MAD
+t1.MAI
+t1.TRG
+t1.frm
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `old` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+a	INSERT	t1	set @s= 1	BEFORE						
+# CRASH POINT: ddl_log_create_before_create_table
+t1.MAD
+t1.MAI
+t1.TRG
+t1.frm
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `old` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+a	INSERT	t1	set @s= 1	BEFORE						
+# CRASH POINT: ddl_log_create_after_create_table
+t1.MAD
+t1.MAI
+t1.TRG
+t1.frm
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `old` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+a	INSERT	t1	set @s= 1	BEFORE						
+# CRASH POINT: ddl_log_create_after_drop
+t1.MAD
+t1.MAI
+t1.TRG
+t1.frm
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `old` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+a	INSERT	t1	set @s= 1	BEFORE						
+# CRASH POINT: ddl_log_create_after_send_data
+# No crash!
+t1.MYD
+t1.MYI
+t1.frm
+master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 LIKE const_table
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `new` int(11) DEFAULT NULL,
+  `b` int(11) DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+# CRASH POINT: ddl_log_create_before_binlog
+t1.MAD
+t1.MAI
+t1.TRG
+t1.frm
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `old` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+a	INSERT	t1	set @s= 1	BEFORE						
+# CRASH POINT: ddl_log_create_after_prepare_eof
+# No crash!
+t1.MYD
+t1.MYI
+t1.frm
+master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 LIKE const_table
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `new` int(11) DEFAULT NULL,
+  `b` int(11) DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+# CRASH POINT: ddl_log_create_after_binlog
+t1.MYD
+t1.MYI
+t1.frm
+master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 LIKE const_table
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `new` int(11) DEFAULT NULL,
+  `b` int(11) DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+# CRASH POINT: ddl_log_create_log_complete
+# No crash!
+t1.MYD
+t1.MYI
+t1.frm
+master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 LIKE const_table
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `new` int(11) DEFAULT NULL,
+  `b` int(11) DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+# CRASH POINT: ddl_log_replace_before_remove_backup
+t1.MYD
+t1.MYI
+t1.frm
+master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 LIKE const_table
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `new` int(11) DEFAULT NULL,
+  `b` int(11) DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+# CRASH POINT: ddl_log_replace_after_remove_backup
+t1.MYD
+t1.MYI
+t1.frm
+master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 LIKE const_table
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `new` int(11) DEFAULT NULL,
+  `b` int(11) DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+# QUERY: CREATE OR REPLACE TABLE t1 SELECT * from const_table
+# CRASH POINT: ddl_log_replace_after_rename_table
+t1.MAD
+t1.MAI
+t1.TRG
+t1.frm
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `old` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+old
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+a	INSERT	t1	set @s= 1	BEFORE						
+# CRASH POINT: ddl_log_create_before_create_frm
+t1.MAD
+t1.MAI
+t1.TRG
+t1.frm
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `old` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+old
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+a	INSERT	t1	set @s= 1	BEFORE						
+# CRASH POINT: ddl_log_create_before_create_table
+t1.MAD
+t1.MAI
+t1.TRG
+t1.frm
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `old` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+old
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+a	INSERT	t1	set @s= 1	BEFORE						
+# CRASH POINT: ddl_log_create_after_create_table
+t1.MAD
+t1.MAI
+t1.TRG
+t1.frm
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `old` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+old
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+a	INSERT	t1	set @s= 1	BEFORE						
+# CRASH POINT: ddl_log_create_after_drop
+t1.MAD
+t1.MAI
+t1.TRG
+t1.frm
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `old` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+old
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+a	INSERT	t1	set @s= 1	BEFORE						
+# CRASH POINT: ddl_log_create_after_send_data
+t1.MAD
+t1.MAI
+t1.TRG
+t1.frm
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `old` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+old
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+a	INSERT	t1	set @s= 1	BEFORE						
+# CRASH POINT: ddl_log_create_before_binlog
+t1.MAD
+t1.MAI
+t1.TRG
+t1.frm
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `old` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+old
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+a	INSERT	t1	set @s= 1	BEFORE						
+# CRASH POINT: ddl_log_create_after_prepare_eof
+t1.MAD
+t1.MAI
+t1.frm
+master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 SELECT * from const_table
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `new` int(11) DEFAULT NULL,
+  `b` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+new	b
+1	1
+2	2
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+# CRASH POINT: ddl_log_create_after_binlog
+t1.MAD
+t1.MAI
+t1.frm
+master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 SELECT * from const_table
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `new` int(11) DEFAULT NULL,
+  `b` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+new	b
+1	1
+2	2
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+# CRASH POINT: ddl_log_create_log_complete
+t1.MAD
+t1.MAI
+t1.frm
+master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 SELECT * from const_table
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `new` int(11) DEFAULT NULL,
+  `b` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+new	b
+1	1
+2	2
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+# CRASH POINT: ddl_log_replace_before_remove_backup
+t1.MAD
+t1.MAI
+t1.frm
+master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 SELECT * from const_table
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `new` int(11) DEFAULT NULL,
+  `b` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+new	b
+1	1
+2	2
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+# CRASH POINT: ddl_log_replace_after_remove_backup
+t1.MAD
+t1.MAI
+t1.frm
+master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 SELECT * from const_table
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `new` int(11) DEFAULT NULL,
+  `b` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1
+new	b
+1	1
+2	2
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+Warnings:
+Note	1051	Unknown table 'test.t1'
diff --git a/mysql-test/suite/atomic/create_replace.test b/mysql-test/suite/atomic/create_replace.test
new file mode 100644
index 00000000000..9ada7abb21d
--- /dev/null
+++ b/mysql-test/suite/atomic/create_replace.test
@@ -0,0 +1,182 @@
+--source include/have_debug.inc
+--source include/have_sequence.inc
+--source include/have_innodb.inc
+--source include/log_bin.inc
+--source include/not_valgrind.inc
+
+let $default_engine=InnoDB;
+let $extra_option=;
+let $save_debug=`select @@debug_dbug`;
+let $show_error=0;
+let $drop_error=0;
+
+if ($MTR_COMBINATION_MYISAM)
+{
+  let $default_engine=MyISAM;
+}
+if ($MTR_COMBINATION_ARIA)
+{
+  let $default_engine=Aria;
+  let $extra_option=transactional=1;
+}
+if ($MTR_COMBINATION_ARIA_NOTRANS)
+{
+  let $default_engine=Aria;
+  let $extra_option=transactional=0;
+}
+if ($MTR_COMBINATION_EXPENSIVE_RENAME)
+{
+  let $default_engine=Aria;
+  let $extra_option=transactional=0;
+  let $show_error=0, ER_NO_SUCH_TABLE;
+  let $drop_error=0, ER_BAD_TABLE_ERROR;
+  set @@debug_dbug="+d,ddl_log_expensive_rename";
+}
+
+if ($MTR_COMBINATION_STMT)
+{
+  let $binlog_format=include/set_binlog_format_statement.sql;
+}
+if ($MTR_COMBINATION_ROW)
+{
+  let $binlog_format=include/set_binlog_format_row.sql;
+}
+if ($MTR_COMBINATION_MIX)
+{
+  --skip same as stmt
+}
+
+--eval set @@default_storage_engine=$default_engine
+
+# Put entry_pos into higher position, so drop chain executes before create chain
+create table t1 (c int);
+--error ER_DUP_KEYNAME
+create or replace table t1 (a int, b int, key k (a), key k (b));
+--error ER_DUP_KEYNAME
+create or replace table t1 (a int, b int, key k (a), key k (b));
+--error $drop_error
+drop table t1;
+
+--echo # Crash recovery
+
+let $MYSQLD_DATADIR= `SELECT @@datadir`;
+
+let $crash_count=12;
+let $crash_points='ddl_log_replace_after_rename_table',
+                  'ddl_log_create_before_create_frm',
+                  'ddl_log_create_before_create_table',
+                  'ddl_log_create_after_create_table',
+                  'ddl_log_create_after_drop',
+                  'ddl_log_create_after_send_data',
+                  'ddl_log_create_before_binlog',
+                  'ddl_log_create_after_prepare_eof',
+                  'ddl_log_create_after_binlog',
+                  'ddl_log_create_log_complete',
+                  'ddl_log_replace_before_remove_backup',
+                  'ddl_log_replace_after_remove_backup';
+
+#let $crash_count=2;
+#let $crash_points='ddl_log_create_after_prepare_eof', 'ddl_log_create_after_binlog';
+
+let $statement_count=3;
+let $statements='CREATE OR REPLACE TABLE t1 (new int)',
+                'CREATE OR REPLACE TABLE t1 LIKE const_table',
+                'CREATE OR REPLACE TABLE t1 SELECT * from const_table';
+
+#let $statement_count=1;
+#let $statements='CREATE OR REPLACE TABLE t1 SELECT * from const_table';
+
+create table const_table (new int, b int) engine=myisam;
+insert into const_table values (1,1),(2,2);
+flush tables;
+
+let $old_debug=`select @@debug_dbug`;
+
+let $keep_include_silent=1;
+let $grep_script=CREATE|DROP;
+--disable_query_log
+
+let $r=0;
+while ($r < $statement_count)
+{
+  inc $r;
+  let $statement=`select ELT($r, $statements)`;
+  --echo # QUERY: $statement
+
+  let $c=0;
+  while ($c < $crash_count)
+  {
+    inc $c;
+    let $crash=`select ELT($c, $crash_points)`;
+
+    --eval set @@default_storage_engine=$default_engine
+    create or replace table t1 (old int);
+    create trigger a before insert on t1 for each row set @s= 1;
+    flush tables;
+    if (!$c)
+    {
+      lock tables t1 write;
+    }
+
+    --source $binlog_format
+
+    RESET MASTER;
+    --echo # CRASH POINT: $crash
+    --exec echo "restart" > $MYSQLTEST_VARDIR/tmp/mysqld.1.expect
+    --disable_reconnect
+    --eval set @@debug_dbug="+d,$crash",@debug_crash_counter=1
+    let $errno=0;
+    --error 0,2013
+    --eval $statement;
+    let $error=$errno;
+    --enable_reconnect
+    --source include/wait_until_connected_again.inc
+    --disable_query_log
+    --eval set @@debug_dbug="$old_debug"
+
+    if ($error == 0)
+    {
+      --echo # No crash!
+    }
+    # Check which tables still exists
+    --list_files $MYSQLD_DATADIR/test t*
+    --list_files $MYSQLD_DATADIR/test *sql*
+
+    --let $binlog_file=master-bin.000001
+    --source include/show_binlog_events.inc
+    if ($error)
+    {
+      --let $binlog_file=master-bin.000002
+      --source include/show_binlog_events.inc
+    }
+
+    if ($c > 5)
+    {
+      if ($MTR_COMBINATION_MYISAM)
+      {
+        # MyISAM warns normally about table marked as crashed, but not for --ps-protocol
+        --disable_warnings
+      }
+    }
+    --replace_result $default_engine DEFAULT_ENGINE ' PAGE_CHECKSUM=1' ''
+    --error $show_error
+    show create table t1;
+    if (`select locate('SELECT', '$statement')`)
+    {
+      --error $show_error
+      select * from t1;
+    }
+    --enable_warnings
+    --replace_column 6 '' 7 '' 8 '' 9 '' 10 '' 11 ''
+    show triggers;
+    # Drop the tables. The warnings will show what was dropped
+    --disable_warnings
+    --error $drop_error
+    drop table t1;
+    --enable_warnings
+  }
+}
+drop table if exists t1,const_table;
+--eval set @@debug_dbug="$save_debug"
+
+--enable_query_log
diff --git a/mysql-test/suite/atomic/create_table.result b/mysql-test/suite/atomic/create_table.result
index acc78c76d25..1b53f9cfeac 100644
--- a/mysql-test/suite/atomic/create_table.result
+++ b/mysql-test/suite/atomic/create_table.result
@@ -63,40 +63,6 @@ t2.MYD
 t2.MYI
 t2.frm
 master-bin.000001	#	Query	#	#	use `test`; CREATE TABLE t1 (a int)
-query: CREATE OR REPLACE TABLE t2 (a int)
-crash point: ddl_log_create_before_create_frm
-master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t2` /* generated by ddl recovery */
-crash point: storage_engine_middle_of_create
-"No crash!"
-t2.MYD
-t2.MYI
-t2.frm
-master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t2 (a int)
-crash point: ddl_log_create_before_create_table
-master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t2` /* generated by ddl recovery */
-crash point: ddl_log_create_after_create_table
-master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t2` /* generated by ddl recovery */
-crash point: ddl_log_create_after_drop
-master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t2` /* generated by ddl recovery */
-crash point: ddl_log_create_before_binlog
-master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t2` /* generated by ddl recovery */
-crash point: ddl_log_create_after_prepare_eof
-"No crash!"
-t2.MYD
-t2.MYI
-t2.frm
-master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t2 (a int)
-crash point: ddl_log_create_after_binlog
-t2.MYD
-t2.MYI
-t2.frm
-master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t2 (a int)
-crash point: ddl_log_create_log_complete
-"No crash!"
-t2.MYD
-t2.MYI
-t2.frm
-master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t2 (a int)
 query: CREATE TABLE t1 LIKE const_table
 crash point: ddl_log_create_before_create_frm
 t2.MYD
@@ -158,40 +124,6 @@ t2.MYD
 t2.MYI
 t2.frm
 master-bin.000001	#	Query	#	#	use `test`; CREATE TABLE t1 LIKE const_table
-query: CREATE OR REPLACE TABLE t2 LIKE const_table
-crash point: ddl_log_create_before_create_frm
-master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t2` /* generated by ddl recovery */
-crash point: storage_engine_middle_of_create
-"No crash!"
-t2.MYD
-t2.MYI
-t2.frm
-master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t2 LIKE const_table
-crash point: ddl_log_create_before_create_table
-master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t2` /* generated by ddl recovery */
-crash point: ddl_log_create_after_create_table
-master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t2` /* generated by ddl recovery */
-crash point: ddl_log_create_after_drop
-master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t2` /* generated by ddl recovery */
-crash point: ddl_log_create_before_binlog
-master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t2` /* generated by ddl recovery */
-crash point: ddl_log_create_after_prepare_eof
-"No crash!"
-t2.MYD
-t2.MYI
-t2.frm
-master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t2 LIKE const_table
-crash point: ddl_log_create_after_binlog
-t2.MYD
-t2.MYI
-t2.frm
-master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t2 LIKE const_table
-crash point: ddl_log_create_log_complete
-"No crash!"
-t2.MYD
-t2.MYI
-t2.frm
-master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t2 LIKE const_table
 query: CREATE TABLE t1 SELECT * from t2
 crash point: ddl_log_create_before_create_frm
 t2.MYD
@@ -259,47 +191,6 @@ master-bin.000001	#	Query	#	#	use `test`; CREATE TABLE `t1` (
   `seq` bigint(20) unsigned NOT NULL
 )
 master-bin.000001	#	Annotate_rows	#	#	CREATE TABLE t1 SELECT * from t2
-query: CREATE OR REPLACE TABLE t2 SELECT * from const_table
-crash point: ddl_log_create_before_create_frm
-master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t2` /* generated by ddl recovery */
-crash point: storage_engine_middle_of_create
-"No crash!"
-t2.MYD
-t2.MYI
-t2.frm
-master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE `t2` (
-  `a` int(11) DEFAULT NULL,
-  `b` int(11) DEFAULT NULL
-)
-master-bin.000001	#	Annotate_rows	#	#	CREATE OR REPLACE TABLE t2 SELECT * from const_table
-crash point: ddl_log_create_before_create_table
-master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t2` /* generated by ddl recovery */
-crash point: ddl_log_create_after_create_table
-master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t2` /* generated by ddl recovery */
-crash point: ddl_log_create_after_drop
-master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t2` /* generated by ddl recovery */
-crash point: ddl_log_create_before_binlog
-master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t2` /* generated by ddl recovery */
-crash point: ddl_log_create_after_prepare_eof
-master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t2` /* generated by ddl recovery */
-crash point: ddl_log_create_after_binlog
-t2.MYD
-t2.MYI
-t2.frm
-master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE `t2` (
-  `a` int(11) DEFAULT NULL,
-  `b` int(11) DEFAULT NULL
-)
-master-bin.000001	#	Annotate_rows	#	#	CREATE OR REPLACE TABLE t2 SELECT * from const_table
-crash point: ddl_log_create_log_complete
-t2.MYD
-t2.MYI
-t2.frm
-master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE `t2` (
-  `a` int(11) DEFAULT NULL,
-  `b` int(11) DEFAULT NULL
-)
-master-bin.000001	#	Annotate_rows	#	#	CREATE OR REPLACE TABLE t2 SELECT * from const_table
 engine: innodb
 query: CREATE TABLE t1 (a int)
 crash point: ddl_log_create_before_create_frm
@@ -348,36 +239,6 @@ t1.ibd
 t2.frm
 t2.ibd
 master-bin.000001	#	Query	#	#	use `test`; CREATE TABLE t1 (a int)
-query: CREATE OR REPLACE TABLE t2 (a int)
-crash point: ddl_log_create_before_create_frm
-master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t2` /* generated by ddl recovery */
-crash point: storage_engine_middle_of_create
-"No crash!"
-t2.frm
-t2.ibd
-master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t2 (a int)
-crash point: ddl_log_create_before_create_table
-master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t2` /* generated by ddl recovery */
-crash point: ddl_log_create_after_create_table
-master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t2` /* generated by ddl recovery */
-crash point: ddl_log_create_after_drop
-master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t2` /* generated by ddl recovery */
-crash point: ddl_log_create_before_binlog
-master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t2` /* generated by ddl recovery */
-crash point: ddl_log_create_after_prepare_eof
-"No crash!"
-t2.frm
-t2.ibd
-master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t2 (a int)
-crash point: ddl_log_create_after_binlog
-t2.frm
-t2.ibd
-master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t2 (a int)
-crash point: ddl_log_create_log_complete
-"No crash!"
-t2.frm
-t2.ibd
-master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t2 (a int)
 query: CREATE TABLE t1 LIKE const_table
 crash point: ddl_log_create_before_create_frm
 t2.frm
@@ -430,40 +291,6 @@ t1.frm
 t2.frm
 t2.ibd
 master-bin.000001	#	Query	#	#	use `test`; CREATE TABLE t1 LIKE const_table
-query: CREATE OR REPLACE TABLE t2 LIKE const_table
-crash point: ddl_log_create_before_create_frm
-master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t2` /* generated by ddl recovery */
-crash point: storage_engine_middle_of_create
-"No crash!"
-t2.MYD
-t2.MYI
-t2.frm
-master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t2 LIKE const_table
-crash point: ddl_log_create_before_create_table
-master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t2` /* generated by ddl recovery */
-crash point: ddl_log_create_after_create_table
-master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t2` /* generated by ddl recovery */
-crash point: ddl_log_create_after_drop
-master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t2` /* generated by ddl recovery */
-crash point: ddl_log_create_before_binlog
-master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t2` /* generated by ddl recovery */
-crash point: ddl_log_create_after_prepare_eof
-"No crash!"
-t2.MYD
-t2.MYI
-t2.frm
-master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t2 LIKE const_table
-crash point: ddl_log_create_after_binlog
-t2.MYD
-t2.MYI
-t2.frm
-master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t2 LIKE const_table
-crash point: ddl_log_create_log_complete
-"No crash!"
-t2.MYD
-t2.MYI
-t2.frm
-master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t2 LIKE const_table
 query: CREATE TABLE t1 SELECT * from t2
 crash point: ddl_log_create_before_create_frm
 t2.frm
@@ -492,46 +319,22 @@ crash point: ddl_log_create_before_binlog
 t2.frm
 t2.ibd
 crash point: ddl_log_create_after_prepare_eof
+t1.frm
+t1.ibd
 t2.frm
 t2.ibd
 master-bin.000001	#	Query	#	#	use `test`; CREATE TABLE t1 SELECT * from t2
-master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1` /* generated by ddl recovery */
 crash point: ddl_log_create_after_binlog
+t1.frm
+t1.ibd
 t2.frm
 t2.ibd
 master-bin.000001	#	Query	#	#	use `test`; CREATE TABLE t1 SELECT * from t2
-master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1` /* generated by ddl recovery */
 crash point: ddl_log_create_log_complete
 t1.frm
 t1.ibd
 t2.frm
 t2.ibd
 master-bin.000001	#	Query	#	#	use `test`; CREATE TABLE t1 SELECT * from t2
-query: CREATE OR REPLACE TABLE t2 SELECT * from const_table
-crash point: ddl_log_create_before_create_frm
-master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t2` /* generated by ddl recovery */
-crash point: storage_engine_middle_of_create
-"No crash!"
-t2.frm
-t2.ibd
-master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t2 SELECT * from const_table
-crash point: ddl_log_create_before_create_table
-master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t2` /* generated by ddl recovery */
-crash point: ddl_log_create_after_create_table
-master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t2` /* generated by ddl recovery */
-crash point: ddl_log_create_after_drop
-master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t2` /* generated by ddl recovery */
-crash point: ddl_log_create_before_binlog
-master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t2` /* generated by ddl recovery */
-crash point: ddl_log_create_after_prepare_eof
-master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t2 SELECT * from const_table
-master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t2` /* generated by ddl recovery */
-crash point: ddl_log_create_after_binlog
-master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t2 SELECT * from const_table
-master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t2` /* generated by ddl recovery */
-crash point: ddl_log_create_log_complete
-t2.frm
-t2.ibd
-master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t2 SELECT * from const_table
 Warnings:
 Note	1051	Unknown table 'test.t1,test.t2'
diff --git a/mysql-test/suite/atomic/create_table.test b/mysql-test/suite/atomic/create_table.test
index ff53a12ebab..2dc7e55051a 100644
--- a/mysql-test/suite/atomic/create_table.test
+++ b/mysql-test/suite/atomic/create_table.test
@@ -30,13 +30,10 @@ if ($engine_count == "")
 let $crash_count=9;
 let $crash_points='ddl_log_create_before_create_frm', 'storage_engine_middle_of_create', 'ddl_log_create_before_create_table', 'ddl_log_create_after_create_table', 'ddl_log_create_after_drop', 'ddl_log_create_before_binlog', 'ddl_log_create_after_prepare_eof', 'ddl_log_create_after_binlog', 'ddl_log_create_log_complete';
 
-let $statement_count=6;
+let $statement_count=3;
 let $statements='CREATE TABLE t1 (a int)',
-                'CREATE OR REPLACE TABLE t2 (a int)',
                 'CREATE TABLE t1 LIKE const_table',
-                'CREATE OR REPLACE TABLE t2 LIKE const_table',
-                'CREATE TABLE t1 SELECT * from t2',
-                'CREATE OR REPLACE TABLE t2 SELECT * from const_table';
+                'CREATE TABLE t1 SELECT * from t2';
 
 create table const_table (a int, b int) engine=myisam;
 insert into const_table values (1,1),(2,2);
diff --git a/mysql-test/suite/parts/r/backup_log.result b/mysql-test/suite/parts/r/backup_log.result
index 3a220491027..fa42dba04ef 100644
--- a/mysql-test/suite/parts/r/backup_log.result
+++ b/mysql-test/suite/parts/r/backup_log.result
@@ -67,7 +67,7 @@ ERROR 23000: Duplicate entry '1' for key 'PRIMARY'
 create table t32 (a int) PARTITION BY HASH(a) PARTITIONS 2;
 drop table if exists t30,t31,t32,tmp_t30;
 Warnings:
-Note	1051	Unknown table 'test.t31,test.tmp_t30'
+Note	1051	Unknown table 'test.tmp_t30'
 #
 # Testing create LIKE
 #
@@ -206,17 +206,18 @@ CREATE,MyISAM,1,test,t21,id: 7,,0,,,
 DROP,MyISAM,1,test,t21,id: 7,,0,,,
 CREATE,MyISAM,1,test,t30,id: 8,,0,,,
 CREATE,MyISAM,1,test,t31,id: 9,,0,,,
-DROP,MyISAM,1,test,t31,id: 9,,0,,,
+RENAME,MyISAM,1,test,t31,id: 9,MyISAM,1,test,#sql1,id: 9
 CREATE,MyISAM,0,test,t31,id: 10,,0,,,
-DROP,MyISAM,0,test,t31,id: 10,,0,,,
+RENAME,MyISAM,0,test,t31,id: 10,MyISAM,0,test,#sql1,id: 10
 DROP_AFTER_CREATE,MyISAM,1,test,t31,id: 11,,0,,,
 CREATE,MyISAM,1,test,t32,id: 12,,0,,,
 DROP,MyISAM,1,test,t30,id: 8,,0,,,
+DROP,MyISAM,0,test,t31,id: 10,,0,,,
 DROP,MyISAM,1,test,t32,id: 12,,0,,,
 CREATE,MyISAM,1,test,t40,id: 13,,0,,,
 CREATE,InnoDB,1,test,t41,id: 14,,0,,,
 CREATE,partition,0,test,t42,id: 15,,0,,,
-DROP,MyISAM,1,test,t42,id: 15,,0,,,
+RENAME,MyISAM,1,test,t42,id: 15,MyISAM,1,test,#sql1,id: 15
 CREATE,partition,0,test,t42,id: 16,,0,,,
 DROP,MyISAM,1,test,t40,id: 13,,0,,,
 DROP,InnoDB,1,test,t41,id: 14,,0,,,
diff --git a/mysql-test/suite/rpl/r/create_or_replace_mix.result b/mysql-test/suite/rpl/r/create_or_replace_mix.result
index 6c83d27eef9..e55ba0707ee 100644
--- a/mysql-test/suite/rpl/r/create_or_replace_mix.result
+++ b/mysql-test/suite/rpl/r/create_or_replace_mix.result
@@ -64,13 +64,12 @@ create table t1 (a int);
 create or replace table t1;
 ERROR 42000: A table must have at least 1 column
 drop table if exists t1;
-Warnings:
-Note	1051	Unknown table 'test.t1'
 create or replace table t1 (a int primary key) select a from t2;
 ERROR 23000: Duplicate entry '2' for key 'PRIMARY'
 create table t1 (a int);
 create or replace table t1 (a int primary key) select a from t2;
 ERROR 23000: Duplicate entry '2' for key 'PRIMARY'
+drop table t1;
 create temporary table t9 (a int);
 create or replace temporary table t9 (a int primary key) select a from t2;
 ERROR 23000: Duplicate entry '2' for key 'PRIMARY'
@@ -86,11 +85,9 @@ master-bin.000001	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated
 master-bin.000001	#	Gtid	#	#	GTID #-#-#
 master-bin.000001	#	Query	#	#	use `test`; create table t1 (a int)
 master-bin.000001	#	Gtid	#	#	GTID #-#-#
-master-bin.000001	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `test`.`t1`/* Generated to handle failed CREATE OR REPLACE */
+master-bin.000001	#	Query	#	#	use `test`; DROP TABLE `t1` /* generated by server */
 master-bin.000001	#	Gtid	#	#	GTID #-#-#
 master-bin.000001	#	Query	#	#	use `test`; create temporary table t9 (a int)
-master-bin.000001	#	Gtid	#	#	GTID #-#-#
-master-bin.000001	#	Query	#	#	DROP TEMPORARY TABLE IF EXISTS `test`.`t9`/* Generated to handle failed CREATE OR REPLACE */
 connection server_2;
 show tables;
 Tables_in_test
@@ -104,8 +101,6 @@ Log_name	Pos	Event_type	Server_id	End_log_pos	Info
 master-bin.000001	#	Gtid	#	#	GTID #-#-#
 master-bin.000001	#	Query	#	#	use `test`; create table t1 (a int)
 drop table if exists t1,t2;
-Warnings:
-Note	1051	Unknown table 'test.t1'
 drop temporary table if exists t9;
 Warnings:
 Note	1051	Unknown table 'test.t9'
diff --git a/mysql-test/suite/rpl/r/create_or_replace_row.result b/mysql-test/suite/rpl/r/create_or_replace_row.result
index c45daefd671..6385a58b656 100644
--- a/mysql-test/suite/rpl/r/create_or_replace_row.result
+++ b/mysql-test/suite/rpl/r/create_or_replace_row.result
@@ -92,13 +92,12 @@ create table t1 (a int);
 create or replace table t1;
 ERROR 42000: A table must have at least 1 column
 drop table if exists t1;
-Warnings:
-Note	1051	Unknown table 'test.t1'
 create or replace table t1 (a int primary key) select a from t2;
 ERROR 23000: Duplicate entry '2' for key 'PRIMARY'
 create table t1 (a int);
 create or replace table t1 (a int primary key) select a from t2;
 ERROR 23000: Duplicate entry '2' for key 'PRIMARY'
+drop table t1;
 create temporary table t9 (a int);
 create or replace temporary table t9 (a int primary key) select a from t2;
 ERROR 23000: Duplicate entry '2' for key 'PRIMARY'
@@ -113,9 +112,8 @@ master-bin.000001	#	Gtid	#	#	GTID #-#-#
 master-bin.000001	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated by server */
 master-bin.000001	#	Gtid	#	#	GTID #-#-#
 master-bin.000001	#	Query	#	#	use `test`; create table t1 (a int)
-master-bin.000001	#	Gtid	#	#	BEGIN GTID #-#-#
-master-bin.000001	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `test`.`t1`/* Generated to handle failed CREATE OR REPLACE */
-master-bin.000001	#	Query	#	#	ROLLBACK
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; DROP TABLE `t1` /* generated by server */
 connection server_2;
 show tables;
 Tables_in_test
@@ -129,8 +127,6 @@ Log_name	Pos	Event_type	Server_id	End_log_pos	Info
 master-bin.000001	#	Gtid	#	#	GTID #-#-#
 master-bin.000001	#	Query	#	#	use `test`; create table t1 (a int)
 drop table if exists t1,t2;
-Warnings:
-Note	1051	Unknown table 'test.t1'
 drop temporary table if exists t9;
 Warnings:
 Note	1051	Unknown table 'test.t9'
diff --git a/mysql-test/suite/rpl/r/create_or_replace_statement.result b/mysql-test/suite/rpl/r/create_or_replace_statement.result
index 6c83d27eef9..e55ba0707ee 100644
--- a/mysql-test/suite/rpl/r/create_or_replace_statement.result
+++ b/mysql-test/suite/rpl/r/create_or_replace_statement.result
@@ -64,13 +64,12 @@ create table t1 (a int);
 create or replace table t1;
 ERROR 42000: A table must have at least 1 column
 drop table if exists t1;
-Warnings:
-Note	1051	Unknown table 'test.t1'
 create or replace table t1 (a int primary key) select a from t2;
 ERROR 23000: Duplicate entry '2' for key 'PRIMARY'
 create table t1 (a int);
 create or replace table t1 (a int primary key) select a from t2;
 ERROR 23000: Duplicate entry '2' for key 'PRIMARY'
+drop table t1;
 create temporary table t9 (a int);
 create or replace temporary table t9 (a int primary key) select a from t2;
 ERROR 23000: Duplicate entry '2' for key 'PRIMARY'
@@ -86,11 +85,9 @@ master-bin.000001	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated
 master-bin.000001	#	Gtid	#	#	GTID #-#-#
 master-bin.000001	#	Query	#	#	use `test`; create table t1 (a int)
 master-bin.000001	#	Gtid	#	#	GTID #-#-#
-master-bin.000001	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `test`.`t1`/* Generated to handle failed CREATE OR REPLACE */
+master-bin.000001	#	Query	#	#	use `test`; DROP TABLE `t1` /* generated by server */
 master-bin.000001	#	Gtid	#	#	GTID #-#-#
 master-bin.000001	#	Query	#	#	use `test`; create temporary table t9 (a int)
-master-bin.000001	#	Gtid	#	#	GTID #-#-#
-master-bin.000001	#	Query	#	#	DROP TEMPORARY TABLE IF EXISTS `test`.`t9`/* Generated to handle failed CREATE OR REPLACE */
 connection server_2;
 show tables;
 Tables_in_test
@@ -104,8 +101,6 @@ Log_name	Pos	Event_type	Server_id	End_log_pos	Info
 master-bin.000001	#	Gtid	#	#	GTID #-#-#
 master-bin.000001	#	Query	#	#	use `test`; create table t1 (a int)
 drop table if exists t1,t2;
-Warnings:
-Note	1051	Unknown table 'test.t1'
 drop temporary table if exists t9;
 Warnings:
 Note	1051	Unknown table 'test.t9'
diff --git a/mysql-test/suite/rpl/t/create_or_replace.inc b/mysql-test/suite/rpl/t/create_or_replace.inc
index df46cc36e97..9a7914ae03b 100644
--- a/mysql-test/suite/rpl/t/create_or_replace.inc
+++ b/mysql-test/suite/rpl/t/create_or_replace.inc
@@ -53,6 +53,7 @@ create table t1 (a int);
 # This should as a delete as we will delete t1
 --error ER_DUP_ENTRY
 create or replace table t1 (a int primary key) select a from t2;
+drop table t1;
 
 # Same with temporary table
 create temporary table t9 (a int);
diff --git a/mysql-test/suite/sql_sequence/create.result b/mysql-test/suite/sql_sequence/create.result
index 6f70f335d88..84122946870 100644
--- a/mysql-test/suite/sql_sequence/create.result
+++ b/mysql-test/suite/sql_sequence/create.result
@@ -60,8 +60,6 @@ next_not_cached_value	minimum_value	maximum_value	start_value	increment	cache_si
 1	1	9223372036854775806	1	1	1000	0	0
 create or replace sequence t1 engine=archive;
 ERROR HY000: Table storage engine 'ARCHIVE' does not support the create option 'SEQUENCE'
-show create table t1;
-ERROR 42S02: Table 'test.t1' doesn't exist
 create or replace sequence t1 start with 10;
 show create sequence t1;
 Table	Create Table
diff --git a/mysql-test/suite/sql_sequence/create.test b/mysql-test/suite/sql_sequence/create.test
index 61430740560..5f53daabbd7 100644
--- a/mysql-test/suite/sql_sequence/create.test
+++ b/mysql-test/suite/sql_sequence/create.test
@@ -24,12 +24,6 @@ show create table t1;
 select * from t1;
 --error ER_ILLEGAL_HA_CREATE_OPTION
 create or replace sequence t1 engine=archive;
-#
-# The following error should be fixed. We shouldn't delete old table on errors
-#
---error ER_NO_SUCH_TABLE
-show create table t1;
-
 
 # Check start values
 create or replace sequence t1 start with 10;
diff --git a/sql/ddl_log.cc b/sql/ddl_log.cc
index 65a02914866..45677fe2f30 100644
--- a/sql/ddl_log.cc
+++ b/sql/ddl_log.cc
@@ -453,7 +453,7 @@ bool ddl_log_disable_execute_entry(DDL_LOG_MEMORY_ENTRY **active_entry)
 static bool is_execute_entry_active(uint entry_pos)
 {
   uchar buff[1];
-  DBUG_ENTER("disable_execute_entry");
+  DBUG_ENTER("is_execute_entry_active");
 
   if (mysql_file_pread(global_ddl_log.file_id, buff, sizeof(buff),
                        global_ddl_log.io_size * entry_pos +
@@ -1297,9 +1297,8 @@ static int ddl_log_execute_action(THD *thd, MEM_ROOT *mem_root,
 
   mysql_mutex_assert_owner(&LOCK_gdl);
   DBUG_PRINT("ddl_log",
-             ("pos: %u=>%u->%u  type: %u  action: %u (%s) phase: %u  "
+             ("pos: %u->%u  type: %u  action: %u (%s) phase: %u  "
               "handler: '%s'  name: '%s'  from_name: '%s'  tmp_name: '%s'",
-              recovery_state.execute_entry_pos,
               ddl_log_entry->entry_pos,
               ddl_log_entry->next_entry,
               (uint) ddl_log_entry->entry_type,
@@ -1510,6 +1509,20 @@ static int ddl_log_execute_action(THD *thd, MEM_ROOT *mem_root,
   case DDL_LOG_DROP_INIT_ACTION:
   {
     LEX_CSTRING *comment= &ddl_log_entry->tmp_name;
+    const uint master_chain_pos= (uint) ddl_log_entry->unique_id;
+    /*
+      This drop table was depending on another chain and should only be executed
+      if the other chain is not active.
+      One such case is CREATE OR REPLACE TABLE ... which renamed the original table
+      and created this DROP TABLE event to be able to DROP the backup table if we
+      have a crash directly after closing of the CREATE event.
+    */
+    if (master_chain_pos && is_execute_entry_active(master_chain_pos))
+    {
+      DBUG_ASSERT(ddl_log_entry->next_entry);
+      error= disable_execute_entry(ddl_log_entry->next_entry);
+        break;
+    }
     recovery_state.drop_table.length(0);
     recovery_state.drop_table.set_charset(system_charset_info);
     recovery_state.drop_table.append(STRING_WITH_LEN("DROP TABLE IF EXISTS "));
@@ -1568,22 +1581,27 @@ static int ddl_log_execute_action(THD *thd, MEM_ROOT *mem_root,
         break;
       /* Fall through */
     case DDL_DROP_PHASE_BINLOG:
-      if (strcmp(recovery_state.current_db, db.str))
+      if (!(ddl_log_entry->flags & DDL_LOG_FLAG_DROP_SKIP_BINLOG))
       {
-        append_identifier(thd, &recovery_state.drop_table, &db);
-        recovery_state.drop_table.append('.');
-      }
-      append_identifier(thd, &recovery_state.drop_table, &table);
-      recovery_state.drop_table.append(',');
-      /* We don't increment phase as we want to retry this in case of crash */
+        if (strcmp(recovery_state.current_db, db.str))
+        {
+          append_identifier(thd, &recovery_state.drop_table, &db);
+          recovery_state.drop_table.append('.');
+        }
+        append_identifier(thd, &recovery_state.drop_table, &table);
+        recovery_state.drop_table.append(',');
+        /* We don't increment phase as we want to retry this in case of crash */
 
-      if (ddl_log_drop_to_binary_log(thd, ddl_log_entry,
-                                     &recovery_state.drop_table))
-      {
-        if (increment_phase(entry_pos))
-          break;
+        if (ddl_log_drop_to_binary_log(thd, ddl_log_entry,
+                                      &recovery_state.drop_table))
+        {
+          if (increment_phase(entry_pos))
+            break;
+        }
+        break;
       }
-      break;
+      (void) increment_phase(entry_pos);
+      /* Fall through */
     case DDL_DROP_PHASE_RESET:
       /* We have already logged all previous drop's. Clear the query */
       recovery_state.drop_table.length(recovery_state.drop_table_init_length);
@@ -1758,31 +1776,6 @@ static int ddl_log_execute_action(THD *thd, MEM_ROOT *mem_root,
     }
     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));
-    if (ddl_log_entry->phase == DDL_CREATE_TABLE_PHASE_LOG)
-    {
-      /*
-        The server logged CREATE TABLE ... SELECT into binary log
-        before crashing. As the commit failed and we have delete the
-        table above, we have now to log the DROP of the created table.
-      */
-
-      String *query= &recovery_state.drop_table;
-      query->length(0);
-      query->append(STRING_WITH_LEN("DROP TABLE IF EXISTS "));
-      append_identifier(thd, query, &db);
-      query->append('.');
-      append_identifier(thd, query, &table);
-      query->append(&end_comment);
-
-      if (mysql_bin_log.is_open())
-      {
-        mysql_mutex_unlock(&LOCK_gdl);
-        (void) thd->binlog_query(THD::STMT_QUERY_TYPE,
-                                 query->ptr(), query->length(),
-                                 TRUE, FALSE, FALSE, 0);
-        mysql_mutex_lock(&LOCK_gdl);
-      }
-    }
     (void) update_phase(entry_pos, DDL_LOG_FINAL_PHASE);
     error= 0;
     break;
@@ -2749,7 +2742,7 @@ int ddl_log_execute_recovery()
     if (ddl_log_entry.entry_type == DDL_LOG_EXECUTE_CODE)
     {
       /*
-        Remeber information about executive ddl log entry,
+        Remember information about executive ddl log entry,
         used for binary logging during recovery
       */
       recovery_state.execute_entry_pos= i;
@@ -3101,18 +3094,18 @@ bool ddl_log_rename_view(THD *thd, DDL_LOG_STATE *ddl_state,
 */
 
 static bool ddl_log_drop_init(THD *thd, DDL_LOG_STATE *ddl_state,
-                              ddl_log_action_code action_code,
                               const LEX_CSTRING *db,
                               const LEX_CSTRING *comment)
 {
   DDL_LOG_ENTRY ddl_log_entry;
-  DBUG_ENTER("ddl_log_drop_file");
+  DBUG_ENTER("ddl_log_drop_init");
 
   bzero(&ddl_log_entry, sizeof(ddl_log_entry));
 
-  ddl_log_entry.action_type=  action_code;
+  ddl_log_entry.action_type=  DDL_LOG_DROP_INIT_ACTION;
   ddl_log_entry.from_db=      *const_cast<LEX_CSTRING*>(db);
   ddl_log_entry.tmp_name=     *const_cast<LEX_CSTRING*>(comment);
+  ddl_log_entry.unique_id=    ddl_state->master_chain_pos;
 
   DBUG_RETURN(ddl_log_write(ddl_state, &ddl_log_entry));
 }
@@ -3120,17 +3113,17 @@ static bool ddl_log_drop_init(THD *thd, DDL_LOG_STATE *ddl_state,
 
 bool ddl_log_drop_table_init(THD *thd, DDL_LOG_STATE *ddl_state,
                              const LEX_CSTRING *db,
-                             const LEX_CSTRING *comment)
+                             const LEX_CSTRING *comment,
+                             bool skip_binlog)
 {
-  return ddl_log_drop_init(thd, ddl_state, DDL_LOG_DROP_INIT_ACTION,
-                           db, comment);
+  ddl_state->skip_binlog= skip_binlog;
+  return ddl_log_drop_init(thd, ddl_state, db, comment);
 }
 
 bool ddl_log_drop_view_init(THD *thd, DDL_LOG_STATE *ddl_state,
                             const LEX_CSTRING *db)
 {
-  return ddl_log_drop_init(thd, ddl_state, DDL_LOG_DROP_INIT_ACTION,
-                           db, &empty_clex_str);
+  return ddl_log_drop_init(thd, ddl_state, db, &empty_clex_str);
 }
 
 
@@ -3166,6 +3159,8 @@ static bool ddl_log_drop(THD *thd, DDL_LOG_STATE *ddl_state,
   ddl_log_entry.name=         *const_cast<LEX_CSTRING*>(table);
   ddl_log_entry.tmp_name=     *const_cast<LEX_CSTRING*>(path);
   ddl_log_entry.phase=        (uchar) phase;
+  if (ddl_state->skip_binlog)
+    ddl_log_entry.flags= DDL_LOG_FLAG_DROP_SKIP_BINLOG;
 
   mysql_mutex_lock(&LOCK_gdl);
   if (ddl_log_write_entry(&ddl_log_entry, &log_entry))
@@ -3304,6 +3299,7 @@ bool ddl_log_create_table(THD *thd, DDL_LOG_STATE *ddl_state,
   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_FLAG_ONLY_FRM : 0;
+  ddl_log_entry.next_entry=   ddl_state->list ? ddl_state->list->entry_pos : 0;
 
   DBUG_RETURN(ddl_log_write(ddl_state, &ddl_log_entry));
 }
@@ -3528,3 +3524,17 @@ err:
   mysql_mutex_unlock(&LOCK_gdl);
   DBUG_RETURN(1);
 }
+
+
+/*
+   Link the ddl_log_state to another (master) chain. If the master
+   chain is active during DDL recovery, this event will not be executed.
+
+   This is used for DROP TABLE of the original table when
+   CREATE OR REPLACE ... is used.
+*/
+void ddl_log_link_events(DDL_LOG_STATE *state, DDL_LOG_STATE *master_state)
+{
+  DBUG_ASSERT(master_state->execute_entry);
+  state->master_chain_pos= master_state->execute_entry->entry_pos;
+}
diff --git a/sql/ddl_log.h b/sql/ddl_log.h
index a2a6af76a77..b08f8dcb307 100644
--- a/sql/ddl_log.h
+++ b/sql/ddl_log.h
@@ -126,7 +126,6 @@ enum enum_ddl_log_drop_db_phase {
 
 enum enum_ddl_log_create_table_phase {
   DDL_CREATE_TABLE_PHASE_INIT=0,
-  DDL_CREATE_TABLE_PHASE_LOG,
   DDL_CREATE_TABLE_PHASE_END
 };
 
@@ -173,6 +172,7 @@ enum enum_ddl_log_alter_table_phase {
   engine is not changed
 */
 #define DDL_LOG_FLAG_ALTER_PARTITION      (1 << 4)
+#define DDL_LOG_FLAG_DROP_SKIP_BINLOG     (1 << 5)
 
 /*
   Setting ddl_log_entry.phase to this has the same effect as setting
@@ -248,6 +248,12 @@ typedef struct st_ddl_log_state
   */
   DDL_LOG_MEMORY_ENTRY *main_entry;
   uint16 flags;                                 /* Cache for flags */
+  /*
+    Don't log DROP to binlog and tell the upper level that it must skip
+    updating XID. HTON_EXPENSIVE_RENAME engines have this flag false.
+  */
+  bool skip_binlog;
+  ulonglong master_chain_pos;
   bool is_active() { return list != 0; }
 } DDL_LOG_STATE;
 
@@ -293,7 +299,8 @@ bool ddl_log_rename_view(THD *thd, DDL_LOG_STATE *ddl_state,
                          const LEX_CSTRING *new_alias);
 bool ddl_log_drop_table_init(THD *thd, DDL_LOG_STATE *ddl_state,
                              const LEX_CSTRING *db,
-                             const LEX_CSTRING *comment);
+                             const LEX_CSTRING *comment,
+                             bool skip_binlog);
 bool ddl_log_drop_view_init(THD *thd, DDL_LOG_STATE *ddl_state,
                             const LEX_CSTRING *db);
 bool ddl_log_drop_table(THD *thd, DDL_LOG_STATE *ddl_state,
@@ -348,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);
+void ddl_log_link_events(DDL_LOG_STATE *state, DDL_LOG_STATE *master_state);
 extern mysql_mutex_t LOCK_gdl;
 #endif /* DDL_LOG_INCLUDED */
diff --git a/sql/handler.h b/sql/handler.h
index 7ad00ee546b..9c380e5b986 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -1894,6 +1894,12 @@ handlerton *ha_default_tmp_handlerton(THD *thd);
 */
 #define HTON_REQUIRES_NOTIFY_TABLEDEF_CHANGED_AFTER_COMMIT (1 << 20)
 
+/*
+  Indicates that rename table is expensive operation.
+  When set atomic CREATE OR REPLACE TABLE is not used.
+*/
+#define HTON_EXPENSIVE_RENAME (1 << 21)
+
 class Ha_trx_info;
 
 struct THD_TRANS
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 713523b7d75..151b3c2d60e 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -6025,6 +6025,7 @@ class select_insert :public select_result_interceptor {
   ulonglong autoinc_value_of_last_inserted_row; // autogenerated or not
   COPY_INFO info;
   bool insert_into_view;
+  bool binary_logged;
   select_insert(THD *thd_arg, TABLE_LIST *table_list_par, TABLE *table_par,
                 List<Item> *fields_par, List<Item> *update_fields,
                 List<Item> *update_values, enum_duplicates duplic,
diff --git a/sql/sql_db.cc b/sql/sql_db.cc
index 37e136927f2..d159b277605 100644
--- a/sql/sql_db.cc
+++ b/sql/sql_db.cc
@@ -1121,7 +1121,7 @@ mysql_rm_db_internal(THD *thd, const LEX_CSTRING *db, bool if_exists,
   thd->push_internal_handler(&err_handler);
   if (!thd->killed &&
       !(tables &&
-        mysql_rm_table_no_locks(thd, tables, &rm_db, &ddl_log_state, true, false,
+        mysql_rm_table_no_locks(thd, tables, &rm_db, &ddl_log_state, NULL, true, false,
                                 true, false, true, false)))
   {
     debug_crash_here("ddl_log_drop_after_drop_tables");
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index 24ffad7368a..bfe8aa28202 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -3791,7 +3791,8 @@ select_insert::select_insert(THD *thd_arg, TABLE_LIST *table_list_par,
   sel_result(result),
   table_list(table_list_par), table(table_par), fields(fields_par),
   autoinc_value_of_last_inserted_row(0),
-  insert_into_view(table_list_par && table_list_par->view != 0)
+  insert_into_view(table_list_par && table_list_par->view != 0),
+  binary_logged(false)
 {
   bzero((char*) &info,sizeof(info));
   info.handle_duplicates= duplic;
@@ -4123,7 +4124,6 @@ bool select_insert::prepare_eof()
   int error;
   bool const trans_table= table->file->has_transactions_and_rollback();
   bool changed;
-  bool binary_logged= 0;
   killed_state killed_status= thd->killed;
 
   DBUG_ENTER("select_insert::prepare_eof");
@@ -4533,8 +4533,11 @@ TABLE *select_create::create_table_from_items(THD *thd, List<Item> *items,
   {
     if (likely(!thd->is_error()))             // CREATE ... IF NOT EXISTS
       my_ok(thd);                             //   succeed, but did nothing
+    if (ddl_log_state_rm.is_active())
+      (void) ddl_log_revert(thd, &ddl_log_state_create);
+    else
+      ddl_log_complete(&ddl_log_state_create);
     ddl_log_complete(&ddl_log_state_rm);
-    ddl_log_complete(&ddl_log_state_create);
     DBUG_RETURN(NULL);
   }
 
@@ -4573,8 +4576,11 @@ TABLE *select_create::create_table_from_items(THD *thd, List<Item> *items,
       *lock= 0;
     }
     drop_open_table(thd, table, &create_table->db, &create_table->table_name);
+    if (ddl_log_state_rm.is_active())
+      (void) ddl_log_revert(thd, &ddl_log_state_create);
+    else
+      ddl_log_complete(&ddl_log_state_create);
     ddl_log_complete(&ddl_log_state_rm);
-    ddl_log_complete(&ddl_log_state_create);
     DBUG_RETURN(NULL);
     /* purecov: end */
   }
@@ -4634,6 +4640,12 @@ select_create::prepare(List<Item> &_values, SELECT_LEX_UNIT *u)
   private:
     virtual int do_postlock(TABLE **tables, uint count)
     {
+      /*
+         TODO: why row binlogging is done here at stage of
+         create_table_from_items() while statement binlogging is done at stage
+         of send_eof()? To avoid business-logic discrepancies both logging
+         types should be done at send_eof(), i.e. when select_insert succeeded.
+      */
       int error;
       THD *thd= const_cast<THD*>(ptr->get_thd());
       TABLE_LIST *save_next_global= create_table->next_global;
@@ -4650,7 +4662,19 @@ select_create::prepare(List<Item> &_values, SELECT_LEX_UNIT *u)
       TABLE const *const table = *tables;
       if (thd->is_current_stmt_binlog_format_row() &&
           !table->s->tmp_table)
+      {
+        thd->binlog_xid= thd->query_id;
+        /*
+           Remember xid's for the case of row based logging. Note that binary
+           log is not flushed until the end of statement, so it is OK to write it
+           now and if crash happens until we closed ddl_log_state_rm we won't see
+           CREATE OR REPLACE event in the binary log.
+         */
+        ddl_log_update_xid(&ptr->ddl_log_state_create, thd->binlog_xid);
+        if (ptr->ddl_log_state_rm.is_active() && !ptr->ddl_log_state_rm.skip_binlog)
+          ddl_log_update_xid(&ptr->ddl_log_state_rm, thd->binlog_xid);
         return binlog_show_create_table(thd, *tables, ptr->create_info);
+      }
       return 0;
     }
     select_create *ptr;
@@ -4919,20 +4943,13 @@ bool select_create::send_eof()
 
   debug_crash_here("ddl_log_create_before_binlog");
 
-  /*
-    In case of crash, we have to add DROP TABLE to the binary log as
-    the CREATE TABLE will already be logged if we are not using row based
-    replication.
-  */
-  if (!thd->is_current_stmt_binlog_format_row())
+  if (!thd->binlog_xid && !table->s->tmp_table)
   {
-    if (ddl_log_state_create.is_active())       // Not temporary table
-      ddl_log_update_phase(&ddl_log_state_create, DDL_CREATE_TABLE_PHASE_LOG);
-    /*
-      We can ignore if we replaced an old table as ddl_log_state_create will
-      now handle the logging of the drop if needed.
-    */
-    ddl_log_complete(&ddl_log_state_rm);
+    thd->binlog_xid= thd->query_id;
+    /* Remember xid's for the case of row based logging */
+    ddl_log_update_xid(&ddl_log_state_create, thd->binlog_xid);
+    if (ddl_log_state_rm.is_active() && !ddl_log_state_rm.skip_binlog)
+      ddl_log_update_xid(&ddl_log_state_rm, thd->binlog_xid);
   }
 
   if (prepare_eof())
@@ -4963,6 +4980,8 @@ bool select_create::send_eof()
     }
   }
 
+  DBUG_ASSERT(!(table->s->tmp_table && ddl_log_state_rm.skip_binlog));
+
   /*
     Do an implicit commit at end of statement for non-temporary
     tables.  This can fail, but we should unlock the table
@@ -5007,10 +5026,6 @@ bool select_create::send_eof()
       thd->get_stmt_da()->set_overwrite_status(true);
     }
 #endif /* WITH_WSREP */
-    thd->binlog_xid= thd->query_id;
-    /* Remember xid's for the case of row based logging */
-    ddl_log_update_xid(&ddl_log_state_create, thd->binlog_xid);
-    ddl_log_update_xid(&ddl_log_state_rm, thd->binlog_xid);
     trans_commit_stmt(thd);
     if (!(thd->variables.option_bits & OPTION_GTID_BEGIN))
       trans_commit_implicit(thd);
@@ -5054,8 +5069,14 @@ bool select_create::send_eof()
     (as the query was logged before commit!)
   */
   debug_crash_here("ddl_log_create_after_binlog");
-  ddl_log_complete(&ddl_log_state_rm);
   ddl_log_complete(&ddl_log_state_create);
+  debug_crash_here("ddl_log_replace_before_remove_backup");
+  if (ddl_log_revert(thd, &ddl_log_state_rm))
+  {
+    abort_result_set();
+    DBUG_RETURN(true);
+  }
+  debug_crash_here("ddl_log_replace_after_remove_backup");
   debug_crash_here("ddl_log_create_log_complete");
 
   /*
@@ -5142,8 +5163,6 @@ void select_create::abort_result_set()
   if (table)
   {
     bool tmp_table= table->s->tmp_table;
-    bool table_creation_was_logged= (!tmp_table ||
-                                     table->s->table_creation_was_logged);
     if (tmp_table)
     {
       DBUG_ASSERT(saved_tmp_table_share);
@@ -5173,21 +5192,6 @@ void select_create::abort_result_set()
       {
         /* Remove logging of drop, create + insert rows */
         binlog_reset_cache(thd);
-        /* Original table was deleted. We have to log it */
-        if (table_creation_was_logged)
-        {
-          thd->binlog_xid= thd->query_id;
-          ddl_log_update_xid(&ddl_log_state_create, thd->binlog_xid);
-          ddl_log_update_xid(&ddl_log_state_rm, thd->binlog_xid);
-          debug_crash_here("ddl_log_create_before_binlog");
-          log_drop_table(thd, &create_table->db, &create_table->table_name,
-                         &create_info->org_storage_engine_name,
-                         create_info->db_type == partition_hton,
-                         &create_info->tabledef_version,
-                         tmp_table);
-          debug_crash_here("ddl_log_create_after_binlog");
-          thd->binlog_xid= 0;
-        }
       }
       else if (!tmp_table)
       {
@@ -5203,7 +5207,14 @@ void select_create::abort_result_set()
       }
     }
   }
-  ddl_log_complete(&ddl_log_state_rm);
-  ddl_log_complete(&ddl_log_state_create);
+  if (!binary_logged)
+  {
+    if (ddl_log_state_rm.is_active())
+      (void) ddl_log_revert(thd, &ddl_log_state_create);
+    else
+      ddl_log_complete(&ddl_log_state_create);
+    ddl_log_complete(&ddl_log_state_rm);
+  }
+  thd->binlog_xid= 0;
   DBUG_VOID_RETURN;
 }
diff --git a/sql/sql_rename.cc b/sql/sql_rename.cc
index 094578874a8..253b78a4d1e 100644
--- a/sql/sql_rename.cc
+++ b/sql/sql_rename.cc
@@ -199,7 +199,7 @@ bool mysql_rename_tables(THD *thd, TABLE_LIST *table_list, bool silent,
   else
   {
     /* Revert the renames of normal tables with the help of the ddl log */
-    ddl_log_revert(thd, &ddl_log_state);
+    error|= ddl_log_revert(thd, &ddl_log_state);
   }
 
 err:
@@ -228,19 +228,7 @@ do_rename_temporary(THD *thd, TABLE_LIST *ren_table, TABLE_LIST *new_table)
 
 
 /**
-   Parameters for do_rename
-*/
-
-struct rename_param
-{
-  LEX_CSTRING old_alias, new_alias;
-  LEX_CUSTRING old_version;
-  handlerton *from_table_hton;
-};
-
-
-/**
-  check_rename()
+  rename_check()
 
   Check pre-conditions for rename
   - From table should exists
@@ -257,8 +245,8 @@ struct rename_param
   @retval <0  Can't do rename, but no error
 */
 
-static int
-check_rename(THD *thd, rename_param *param,
+int
+rename_check(THD *thd, rename_param *param,
              TABLE_LIST *ren_table,
              const LEX_CSTRING *new_db,
              const LEX_CSTRING *new_table_name,
@@ -316,7 +304,7 @@ check_rename(THD *thd, rename_param *param,
   Rename a single table or a view
 
   SYNPOSIS
-    do_rename()
+    rename_do()
       thd               Thread handle
       ren_table         A table/view to be renamed
       new_db            The database to which the table to be moved to
@@ -333,8 +321,8 @@ check_rename(THD *thd, rename_param *param,
     true      rename failed
 */
 
-static bool
-do_rename(THD *thd, rename_param *param, DDL_LOG_STATE *ddl_log_state,
+bool
+rename_do(THD *thd, rename_param *param, DDL_LOG_STATE *ddl_log_state,
           TABLE_LIST *ren_table, const LEX_CSTRING *new_db,
           bool skip_error, bool *force_if_exists)
 {
@@ -349,8 +337,6 @@ do_rename(THD *thd, rename_param *param, DDL_LOG_STATE *ddl_log_state,
   new_alias= &param->new_alias;
   hton=      param->from_table_hton;
 
-  DBUG_ASSERT(!thd->locked_tables_mode);
-
 #ifdef WITH_WSREP
   if (WSREP(thd) && hton && hton != view_pseudo_hton &&
       !wsrep_should_replicate_ddl(thd, hton))
@@ -522,7 +508,7 @@ rename_tables(THD *thd, TABLE_LIST *table_list, DDL_LOG_STATE *ddl_log_state,
     {
       int error;
       rename_param param;
-      error= check_rename(thd, &param, ren_table, &new_table->db,
+      error= rename_check(thd, &param, ren_table, &new_table->db,
                           &new_table->table_name,
                           &new_table->alias, (skip_error || if_exists));
       if (error < 0)
@@ -530,9 +516,9 @@ rename_tables(THD *thd, TABLE_LIST *table_list, DDL_LOG_STATE *ddl_log_state,
       if (error > 0)
         goto revert_rename;
 
-      if (do_rename(thd, &param, ddl_log_state,
-                    ren_table, &new_table->db,
-                    skip_error, force_if_exists))
+      if (rename_do(thd, &param, ddl_log_state,
+                          ren_table, &new_table->db,
+                          skip_error, force_if_exists))
         goto revert_rename;
     }
   }
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index 52f23ce9d0b..e5432bb7817 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -55,6 +55,7 @@
 #include "sql_window.h"
 #include "tztime.h"
 
+#include "debug.h"
 #include "debug_sync.h"          // DEBUG_SYNC
 #include <m_ctype.h>
 #include <my_bit.h>
@@ -22219,6 +22220,7 @@ end_send(JOIN *join, JOIN_TAB *join_tab, bool end_of_records)
       // error < 0 => duplicate row
       join->duplicate_rows++;
     }
+    debug_crash_here("ddl_log_create_after_send_data");
   }
 
   join->send_records++;
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 24ad0bd4676..3345f9a5e1d 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -1041,7 +1041,7 @@ bool mysql_rm_table(THD *thd,TABLE_LIST *tables, bool if_exists,
 
   /* mark for close and remove all cached entries */
   thd->push_internal_handler(&err_handler);
-  error= mysql_rm_table_no_locks(thd, tables, &thd->db, (DDL_LOG_STATE*) 0,
+  error= mysql_rm_table_no_locks(thd, tables, &thd->db, (DDL_LOG_STATE*) 0, NULL,
                                  if_exists,
                                  drop_temporary,
                                  false, drop_sequence, dont_log_query,
@@ -1100,6 +1100,52 @@ static uint32 get_comment(THD *thd, uint32 comment_pos,
   return 0;
 }
 
+
+static
+bool make_backup_name(THD *thd, TABLE_LIST *orig, TABLE_LIST *res)
+{
+  char res_name[NAME_LEN + 1];
+
+  size_t len= my_snprintf(res_name, sizeof(res_name) - 1,
+                          backup_file_prefix "%lx-%llx", current_pid,
+                          thd->thread_id, orig->table_name.str);
+
+  LEX_CSTRING n= { res_name, len };
+  res->init_one_table(&orig->db, &n, &n, TL_WRITE);
+
+  res->table_name.str= strmake_root(thd->mem_root,
+                                    LEX_STRING_WITH_LEN(res->table_name));
+  if (!res->table_name.str)
+  {
+    my_error(ER_OUT_OF_RESOURCES, MYF(0));
+    return true;
+  }
+  res->alias.str= res->table_name.str;
+  return false;
+}
+
+
+static
+bool close_or_remove_table(THD *thd, TABLE_LIST *table)
+{
+  if (thd->locked_tables_mode == LTM_LOCK_TABLES ||
+      thd->locked_tables_mode == LTM_PRELOCKED_UNDER_LOCK_TABLES)
+  {
+    if (table->table)
+    {
+      if (wait_while_table_is_used(thd, table->table, HA_EXTRA_NOT_USED))
+        return true;
+      close_all_tables_for_name(thd, table->table->s,
+                                HA_EXTRA_PREPARE_FOR_DROP, NULL);
+      table->table= 0;
+    }
+  }
+  else
+    tdc_remove_table(thd, table->db.str, table->table_name.str);
+  return false;
+}
+
+
 /**
   Execute the drop of a sequence, view or table (normal or temporary).
 
@@ -1141,6 +1187,7 @@ static uint32 get_comment(THD *thd, uint32 comment_pos,
 int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables,
                             const LEX_CSTRING *current_db,
                             DDL_LOG_STATE *ddl_log_state,
+                            DDL_LOG_STATE *ddl_log_state_create,
                             bool if_exists,
                             bool drop_temporary, bool drop_view,
                             bool drop_sequence,
@@ -1161,6 +1208,7 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables,
   bool trans_tmp_table_deleted= 0, non_trans_tmp_table_deleted= 0;
   bool is_drop_tmp_if_exists_added= 0, non_tmp_table_deleted= 0;
   bool log_if_exists= if_exists;
+  bool force_if_exists;
   const LEX_CSTRING *object_to_drop= ((drop_sequence) ?
                                       &SEQUENCE_clex_str :
                                       &TABLE_clex_str);
@@ -1170,6 +1218,7 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables,
 
   if (!ddl_log_state)
   {
+    DBUG_ASSERT(!ddl_log_state_create);
     ddl_log_state= &local_ddl_log_state;
     bzero(ddl_log_state, sizeof(*ddl_log_state));
   }
@@ -1237,7 +1286,7 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables,
     bool is_temporary= 0;
     bool was_view= 0, was_table= 0;
     const LEX_CSTRING db= table->db;
-    const LEX_CSTRING table_name= table->table_name;
+    LEX_CSTRING table_name= table->table_name;
     LEX_CSTRING cpath= {0,0};
     handlerton *hton= 0;
     Table_type table_type;
@@ -1359,11 +1408,12 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables,
       path_length= build_table_filename(path, sizeof(path) - 1, db.str,
                                         alias.str, reg_ext, 0);
       path_end= path + path_length - reg_ext_length;
-    }
+    } /* if (!drop_temporary) */
 
     DEBUG_SYNC(thd, "rm_table_no_locks_before_delete_table");
     if (drop_temporary)
     {
+      DBUG_ASSERT(!ddl_log_state_create);
       /* "DROP TEMPORARY" but a temporary table was not found */
       unknown_tables.append(&db);
       unknown_tables.append('.');
@@ -1393,19 +1443,68 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables,
 
     thd->replication_flags= 0;
     was_view= table_type == TABLE_TYPE_VIEW;
+    const bool wrong_drop= (table_type == TABLE_TYPE_UNKNOWN) ||
+                              (was_view && !drop_view) ||
+                              (table_type != TABLE_TYPE_SEQUENCE && drop_sequence);
+
+    if (!wrong_drop && ddl_log_state_create)
+    {
+      rename_param param;
+      TABLE_LIST t;
+
+      if (make_backup_name(thd, table, &t) ||
+          rename_check(thd, &param, table, &table->db, &t.table_name,
+                        &t.table_name, if_exists))
+      {
+        error= 1;
+        goto err;
+      }
+
+      if (param.from_table_hton == view_pseudo_hton ||
+          param.from_table_hton->flags & HTON_EXPENSIVE_RENAME ||
+          DBUG_EVALUATE_IF("ddl_log_expensive_rename", true, false))
+      {
+        ddl_log_state_create= NULL;
+      }
+      else
+      {
+        if (close_or_remove_table(thd, table) ||
+            rename_do(thd, &param, ddl_log_state_create, table,
+                      &table->db, false, &force_if_exists))
+        {
+          error= 1;
+          goto err;
+        }
+        debug_crash_here("ddl_log_replace_after_rename_table");
+        table_name= t.table_name;
+        alias= (lower_case_table_names == 2) ? table->alias : table_name;
+        /* remove .frm file and engine files */
+        path_length= build_table_filename(path, sizeof(path) - 1, db.str,
+                                          alias.str, reg_ext, 0);
+        path_end= path + path_length - reg_ext_length;
+        lex_string_set3(&cpath, path, (size_t) (path_end - path));
+      } /* else !(param.from_table_hton->flags & HTON_EXPENSIVE_RENAME) */
+    } /* if (ddl_log_state_rename) */
 
     if (!table_count++)
     {
       LEX_CSTRING comment= {comment_start, (size_t) comment_len};
-      if (ddl_log_drop_table_init(thd, ddl_log_state, current_db, &comment))
+
+      if (!wrong_drop && ddl_log_state_create)
+      {
+        ddl_log_link_events(ddl_log_state, ddl_log_state_create);
+        dont_log_query= true;
+      }
+
+      if (ddl_log_drop_table_init(thd, ddl_log_state, current_db, &comment,
+                                  ddl_log_state_create != 0))
       {
         error= 1;
         goto err;
       }
     }
 
-    if ((table_type == TABLE_TYPE_UNKNOWN) || (was_view && !drop_view) ||
-        (table_type != TABLE_TYPE_SEQUENCE && drop_sequence))
+    if (wrong_drop)
     {
       /*
         One of the following cases happened:
@@ -1428,24 +1527,17 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables,
         goto err;
       }
 
-      if (thd->locked_tables_mode == LTM_LOCK_TABLES ||
-          thd->locked_tables_mode == LTM_PRELOCKED_UNDER_LOCK_TABLES)
+
+      if (close_or_remove_table(thd, table))
       {
-        if (wait_while_table_is_used(thd, table->table, HA_EXTRA_NOT_USED))
-        {
-          error= -1;
-          goto err;
-        }
-        close_all_tables_for_name(thd, table->table->s,
-                                  HA_EXTRA_PREPARE_FOR_DROP, NULL);
-        table->table= 0;
+        error= 1;
+        goto err;
       }
-      else
-        tdc_remove_table(thd, db.str, table_name.str);
 
       /* Check that we have an exclusive lock on the table to be dropped. */
-      DBUG_ASSERT(thd->mdl_context.is_lock_owner(MDL_key::TABLE, db.str,
-                                                table_name.str, MDL_EXCLUSIVE));
+      DBUG_ASSERT(ddl_log_state_create ||
+                  thd->mdl_context.is_lock_owner(MDL_key::TABLE, db.str,
+                                                 table_name.str, MDL_EXCLUSIVE));
 
       // Remove extension for delete
       *path_end= '\0';
@@ -1468,6 +1560,9 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables,
       }
 
       debug_crash_here("ddl_log_drop_before_delete_table");
+      if (ddl_log_state_create)
+        goto report_error;
+
       error= ha_delete_table(thd, hton, path, &db, &table_name,
                              enoent_warning);
       debug_crash_here("ddl_log_drop_after_delete_table");
@@ -1655,7 +1750,7 @@ report_error:
         backup_log_ddl(&ddl_log);
       }
     }
-    if (!was_view)
+    if (!was_view && !ddl_log_state_create)
       ddl_log_update_phase(ddl_log_state, DDL_DROP_PHASE_BINLOG);
 
     if (!dont_log_query &&
@@ -4308,7 +4403,7 @@ int create_table_impl(THD *thd,
         (void) trans_rollback_stmt(thd);
         /* Remove normal table without logging. Keep tables locked */
         if (mysql_rm_table_no_locks(thd, &table_list, &thd->db,
-                                    ddl_log_state_rm,
+                                    ddl_log_state_rm, ddl_log_state_create,
                                     0, 0, 0, 0, 1, 1))
           goto err;
 
@@ -4479,12 +4574,6 @@ int create_table_impl(THD *thd,
 
   error= 0;
 err:
-  if (unlikely(error) && ddl_log_state_create)
-  {
-    /* Table was never created, so we can ignore the ddl log entry */
-    ddl_log_complete(ddl_log_state_create);
-  }
-
   THD_STAGE_INFO(thd, stage_after_create);
   delete file;
   DBUG_PRINT("exit", ("return: %d", error));
@@ -4565,7 +4654,7 @@ int mysql_create_table_no_lock(THD *thd,
       DBUG_ASSERT(thd->is_error());
       /* Drop the table as it wasn't completely done */
       if (!mysql_rm_table_no_locks(thd, table_list, &thd->db,
-                                   (DDL_LOG_STATE*) 0,
+                                   (DDL_LOG_STATE*) 0, NULL,
                                    1,
                                    create_info->tmp_table(),
                                    false, true /* Sequence*/,
@@ -4732,7 +4821,7 @@ err:
     }
     thd->binlog_xid= thd->query_id;
     ddl_log_update_xid(&ddl_log_state_create, thd->binlog_xid);
-    if (ddl_log_state_rm.is_active())
+    if (ddl_log_state_rm.is_active() && !ddl_log_state_rm.skip_binlog)
       ddl_log_update_xid(&ddl_log_state_rm, thd->binlog_xid);
     debug_crash_here("ddl_log_create_before_binlog");
     if (unlikely(write_bin_log(thd, result ? FALSE : TRUE, thd->query(),
@@ -4754,8 +4843,26 @@ err:
       backup_log_ddl(&ddl_log);
     }
   }
-  ddl_log_complete(&ddl_log_state_rm);
-  ddl_log_complete(&ddl_log_state_create);
+  if (result)
+  {
+    if (ddl_log_state_rm.is_active())
+      result|= ddl_log_revert(thd, &ddl_log_state_create);
+    else
+      ddl_log_complete(&ddl_log_state_create);
+    ddl_log_complete(&ddl_log_state_rm);
+  }
+  else
+  {
+    /*
+      NOTE: write renamed table name into drop-chain and replay it after
+      CREATE OR REPLACE finishes. We need to finalize rename-chain first,
+      so replay would not execute it.
+    */
+    ddl_log_complete(&ddl_log_state_create);
+    debug_crash_here("ddl_log_replace_before_remove_backup");
+    result|= ddl_log_revert(thd, &ddl_log_state_rm);
+    debug_crash_here("ddl_log_replace_after_remove_backup");
+  }
   DBUG_RETURN(result);
 }
 
@@ -5377,9 +5484,9 @@ err:
   {
     thd->binlog_xid= thd->query_id;
     ddl_log_update_xid(&ddl_log_state_create, thd->binlog_xid);
-    if (ddl_log_state_rm.is_active())
+    if (ddl_log_state_rm.is_active() && !ddl_log_state_rm.skip_binlog)
       ddl_log_update_xid(&ddl_log_state_rm, thd->binlog_xid);
-    debug_crash_here("ddl_log_create_before_binlog");      
+    debug_crash_here("ddl_log_create_before_binlog");
     if (res && create_info->table_was_deleted)
     {
       /*
@@ -5415,8 +5522,23 @@ err:
     backup_log_ddl(&ddl_log);
   }
 
-  ddl_log_complete(&ddl_log_state_rm);
-  ddl_log_complete(&ddl_log_state_create);
+  if (res)
+  {
+    if (ddl_log_state_rm.is_active() &&
+        ddl_log_revert(thd, &ddl_log_state_create))
+      res= 1;
+    else
+      ddl_log_complete(&ddl_log_state_create);
+    ddl_log_complete(&ddl_log_state_rm);
+  }
+  else
+  {
+    ddl_log_complete(&ddl_log_state_create);
+    debug_crash_here("ddl_log_replace_before_remove_backup");
+    if (ddl_log_revert(thd, &ddl_log_state_rm))
+      res= 1;
+    debug_crash_here("ddl_log_replace_after_remove_backup");
+  }
   DBUG_RETURN(res != 0);
 }
 
diff --git a/sql/sql_table.h b/sql/sql_table.h
index aacb6c99f15..0dc105b7341 100644
--- a/sql/sql_table.h
+++ b/sql/sql_table.h
@@ -169,6 +169,26 @@ bool mysql_recreate_table(THD *thd, TABLE_LIST *table_list, bool table_copy);
 bool mysql_create_like_table(THD *thd, TABLE_LIST *table,
                              TABLE_LIST *src_table,
                              Table_specification_st *create_info);
+/**
+    Parameters for rename_do()
+*/
+struct rename_param
+{
+  LEX_CSTRING old_alias, new_alias;
+  LEX_CUSTRING old_version;
+  handlerton *from_table_hton;
+};
+bool
+rename_do(THD *thd, rename_param *param, DDL_LOG_STATE *ddl_log_state,
+          TABLE_LIST *ren_table, const LEX_CSTRING *new_db,
+          bool skip_error, bool *force_if_exists);
+int
+rename_check(THD *thd, rename_param *param,
+             TABLE_LIST *ren_table,
+             const LEX_CSTRING *new_db,
+             const LEX_CSTRING *new_table_name,
+             const LEX_CSTRING *new_table_alias,
+             bool if_exists);
 bool mysql_rename_table(handlerton *base, const LEX_CSTRING *old_db,
                         const LEX_CSTRING *old_name, const LEX_CSTRING *new_db,
                         const LEX_CSTRING *new_name, LEX_CUSTRING *id,
@@ -186,6 +206,7 @@ bool mysql_rm_table(THD *thd,TABLE_LIST *tables, bool if_exists,
 int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables,
                             const LEX_CSTRING *db,
                             DDL_LOG_STATE *ddl_log_state,
+                            DDL_LOG_STATE *ddl_log_state_create,
                             bool if_exists,
                             bool drop_temporary, bool drop_view,
                             bool drop_sequence,
diff --git a/sql/table.h b/sql/table.h
index 183ade8f1d9..e8158a23e99 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -75,6 +75,7 @@ typedef ulonglong nested_join_map;
 
 #define tmp_file_prefix "#sql"			/**< Prefix for tmp tables */
 #define tmp_file_prefix_length 4
+#define backup_file_prefix tmp_file_prefix "-backup-"
 #define TMP_TABLE_KEY_EXTRA 8
 
 /**
diff --git a/storage/maria/ha_s3.cc b/storage/maria/ha_s3.cc
index e7428385a60..ea870c8879d 100644
--- a/storage/maria/ha_s3.cc
+++ b/storage/maria/ha_s3.cc
@@ -1030,7 +1030,8 @@ static int ha_s3_init(void *p)
   s3_hton->show_status= 0;
   s3_hton->prepare_for_backup= 0;
   s3_hton->end_backup= 0;
-  s3_hton->flags= ((s3_slave_ignore_updates ? HTON_IGNORE_UPDATES : 0) |
+  s3_hton->flags= HTON_EXPENSIVE_RENAME |
+                  ((s3_slave_ignore_updates ? HTON_IGNORE_UPDATES : 0) |
                    (s3_replicate_alter_as_create_select ?
                     HTON_TABLE_MAY_NOT_EXIST_ON_SLAVE : 0));
   /* Copy global arguments to s3_access_key and s3_secret_key */