MDEV-29770 Broken table cannot be CREATE OR REPLACE -ed anymore

finalize_atomic_replace():

If rename to backup fails we fall back to the mechanism of dropping
old table.  For that we write into cleanup chain these events
(executed in reverse order):

  1. rename from temporary to original;
  2. drop the original table.

After binlogging is done, in finalize_ddl() we execute that cleanup
chain in case of success. In case of error finalize_ddl() closes
cleanup chain and executes rollback chain which drops backup (at that
point non-existent) and the new table under tmp name.

mysql_rename_table():

ha_rename_table() is non-atomic, it can make partial changes and
fail. In that case we revert back these partitial changes to make it
more atomic-friendly.

ddl_log API changes:

DROP sequence of actions can now be written into non-empty
chain. After the sequence is replayed in straight order the chain
continues from what was before DROP_INIT action. That is done by
remembering last entry in ddl_log_drop_init() and update next_entry to
that in each ddl_log_drop(). Of course, each next ddl_log_drop()
updates previous DROP (or DROP_INIT) to point to itself as it was
before.

ddl_log_start_atomic_block()/ddl_log_commit_atomic_block(): these
functions allow to write multiple chain entries without updating
execute entry. Required when the block of several actions must be
atomic.

7 files changed, 791 insertions, 16 deletions

diff --git a/mysql-test/main/create_or_replace.result b/mysql-test/main/create_or_replace.result
index 6267838fcca..e63df3c5c13 100644
--- a/mysql-test/main/create_or_replace.result
+++ b/mysql-test/main/create_or_replace.result
@@ -915,12 +915,10 @@ Warning	1265	Data truncated for column 'stat_description' at row 2
 Warning	1265	Data truncated for column 'stat_description' at row 3
 lock table t write;
 create or replace table t (y int);
-ERROR HY000: Error on rename of './test/t' to './test/#sql-backup-t' (errno: 168 "Unknown (generic) error from engine")
 unlock tables;
 alter table mysql.innodb_index_stats modify stat_description varchar(1024) not null;
 select * from t;
-x
-77
+y
 drop table t;
 set sql_mode= default;
 #
@@ -949,3 +947,42 @@ CREATE TABLE t2 (a INT) ENGINE=Aria;
 INSERT INTO t2 VALUES (2),(1);
 CREATE OR REPLACE TABLE t3 AS SELECT t2.* FROM t2, t1 WHERE t2.a = t1.f OR t2.a = t1.id;
 DROP TABLE t1, t2, t3;
+#
+# MDEV-29770 Broken table cannot be CREATE OR REPLACE -ed anymore
+#
+create table t1 (a int);
+create trigger tr before delete on t1 for each row begin end;
+create or replace table t1 (x int);
+Warnings:
+Error	1017	Can't find file: './test/t1.MYI' (errno: 2 "No such file or directory")
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci
+show triggers;
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+create trigger tr before delete on t1 for each row begin end;
+create or replace table t1 select 1 as y;
+Warnings:
+Error	1017	Can't find file: './test/t1.MYI' (errno: 2 "No such file or directory")
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `y` int(1) NOT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci
+show triggers;
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+create trigger tr before delete on t1 for each row begin end;
+create table t2 (z int);
+create or replace table t1 like t2;
+Warnings:
+Error	1017	Can't find file: './test/t1.MYI' (errno: 2 "No such file or directory")
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `z` int(11) DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci
+show triggers;
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+drop tables if exists t1, t2;
diff --git a/mysql-test/main/create_or_replace.test b/mysql-test/main/create_or_replace.test
index 882f98aa217..b3aa6d38ccd 100644
--- a/mysql-test/main/create_or_replace.test
+++ b/mysql-test/main/create_or_replace.test
@@ -707,9 +707,6 @@ create table t (x int) engine innodb;
 insert into t values (77);
 alter table mysql.innodb_index_stats modify stat_description char(10);
 lock table t write;
---replace_regex /#sql-backup-.+-.+-/#sql-backup-/
---replace_result $MYSQLD_DATADIR ./
---error ER_ERROR_ON_RENAME
 create or replace table t (y int);
 # cleanup
 unlock tables;
@@ -749,3 +746,35 @@ INSERT INTO t2 VALUES (2),(1);
 CREATE OR REPLACE TABLE t3 AS SELECT t2.* FROM t2, t1 WHERE t2.a = t1.f OR t2.a = t1.id;
 # Cleanup
 DROP TABLE t1, t2, t3;
+
+
+--echo #
+--echo # MDEV-29770 Broken table cannot be CREATE OR REPLACE -ed anymore
+--echo #
+create table t1 (a int);
+create trigger tr before delete on t1 for each row begin end;
+
+--let $datadir= `select @@datadir`
+--remove_file $datadir/test/t1.MYI
+--replace_result $MYSQLD_DATADIR ./
+create or replace table t1 (x int);
+show create table t1;
+show triggers;
+
+create trigger tr before delete on t1 for each row begin end;
+--remove_file $datadir/test/t1.MYI
+--replace_result $MYSQLD_DATADIR ./
+create or replace table t1 select 1 as y;
+show create table t1;
+show triggers;
+
+create trigger tr before delete on t1 for each row begin end;
+--remove_file $datadir/test/t1.MYI
+create table t2 (z int);
+--replace_result $MYSQLD_DATADIR ./
+create or replace table t1 like t2;
+show create table t1;
+show triggers;
+
+# Cleanup
+drop tables if exists t1, t2;
diff --git a/mysql-test/suite/atomic/create_replace_broken.result b/mysql-test/suite/atomic/create_replace_broken.result
new file mode 100644
index 00000000000..acf4f26f09a
--- /dev/null
+++ b/mysql-test/suite/atomic/create_replace_broken.result
@@ -0,0 +1,424 @@
+# Crash recovery
+Table	Create Table
+const_table	CREATE TABLE `const_table` (
+  `new` int(11) DEFAULT NULL,
+  `b` int(11) DEFAULT NULL
+) ENGINE=ENGINE DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci
+insert into const_table values (1, 1), (2, 2);
+flush tables;
+# QUERY: CREATE OR REPLACE TABLE t1 (new int)
+# CRASH POINT: ddl_log_replace_broken_1
+t1.DATA1
+t1.TRG
+t1.frm
+tr.TRN
+show create table t1;
+Level	Code	Message
+Error	1017	Can't find file: './test/t1.MYI' (errno: 2 "No such file or directory")
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+a	INSERT	t1	set @s= 1	BEFORE						
+tr	DELETE	t1	begin end	BEFORE						
+# CRASH POINT: ddl_log_replace_broken_2
+t1.DATA1
+t1.TRG
+t1.frm
+tr.TRN
+show create table t1;
+Level	Code	Message
+Error	1017	Can't find file: './test/t1.MYI' (errno: 2 "No such file or directory")
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+a	INSERT	t1	set @s= 1	BEFORE						
+tr	DELETE	t1	begin end	BEFORE						
+# CRASH POINT: ddl_log_replace_broken_3
+t1.DATA1
+t1.TRG
+t1.frm
+tr.TRN
+show create table t1;
+Level	Code	Message
+Error	1017	Can't find file: './test/t1.MYI' (errno: 2 "No such file or directory")
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+a	INSERT	t1	set @s= 1	BEFORE						
+tr	DELETE	t1	begin end	BEFORE						
+# CRASH POINT: ddl_log_replace_broken_4
+t1.DATA1
+t1.TRG
+t1.frm
+tr.TRN
+show create table t1;
+Level	Code	Message
+Error	1017	Can't find file: './test/t1.MYI' (errno: 2 "No such file or directory")
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+a	INSERT	t1	set @s= 1	BEFORE						
+tr	DELETE	t1	begin end	BEFORE						
+# CRASH POINT: ddl_log_replace_broken_5
+t1.DATA1
+t1.TRG
+t1.frm
+tr.TRN
+show create table t1;
+Level	Code	Message
+Error	1017	Can't find file: './test/t1.MYI' (errno: 2 "No such file or directory")
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+a	INSERT	t1	set @s= 1	BEFORE						
+tr	DELETE	t1	begin end	BEFORE						
+# CRASH POINT: ddl_log_create_before_binlog
+t1.DATA1
+t1.TRG
+t1.frm
+tr.TRN
+show create table t1;
+Level	Code	Message
+Error	1017	Can't find file: './test/t1.MYI' (errno: 2 "No such file or directory")
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+a	INSERT	t1	set @s= 1	BEFORE						
+tr	DELETE	t1	begin end	BEFORE						
+# CRASH POINT: ddl_log_create_after_prepare_eof
+Warnings:
+Error	1017	Can't find file: './test/t1.MYI' (errno: 2 "No such file or directory")
+# No crash!
+t1.DATA1
+t1.DATA2
+t1.frm
+master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 (new int)
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `new` int(11) DEFAULT NULL
+) ENGINE=ENGINE DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci
+Level	Code	Message
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+# CRASH POINT: ddl_log_create_after_binlog
+t1.DATA1
+t1.DATA2
+t1.frm
+master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 (new int)
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `new` int(11) DEFAULT NULL
+) ENGINE=ENGINE DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci
+Level	Code	Message
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+# CRASH POINT: ddl_log_create_log_complete
+t1.DATA1
+t1.DATA2
+t1.frm
+master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 (new int)
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `new` int(11) DEFAULT NULL
+) ENGINE=ENGINE DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci
+Level	Code	Message
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+# CRASH POINT: ddl_log_create_log_complete2
+t1.DATA1
+t1.DATA2
+t1.frm
+master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 (new int)
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `new` int(11) DEFAULT NULL
+) ENGINE=ENGINE DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci
+Level	Code	Message
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+# CRASH POINT: ddl_log_create_log_complete3
+t1.DATA1
+t1.DATA2
+t1.frm
+master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 (new int)
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `new` int(11) DEFAULT NULL
+) ENGINE=ENGINE DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci
+Level	Code	Message
+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_broken_1
+t1.DATA1
+t1.TRG
+t1.frm
+tr.TRN
+show create table t1;
+Level	Code	Message
+Error	1017	Can't find file: './test/t1.MYI' (errno: 2 "No such file or directory")
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+a	INSERT	t1	set @s= 1	BEFORE						
+tr	DELETE	t1	begin end	BEFORE						
+# CRASH POINT: ddl_log_replace_broken_2
+t1.DATA1
+t1.TRG
+t1.frm
+tr.TRN
+show create table t1;
+Level	Code	Message
+Error	1017	Can't find file: './test/t1.MYI' (errno: 2 "No such file or directory")
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+a	INSERT	t1	set @s= 1	BEFORE						
+tr	DELETE	t1	begin end	BEFORE						
+# CRASH POINT: ddl_log_replace_broken_3
+t1.DATA1
+t1.TRG
+t1.frm
+tr.TRN
+show create table t1;
+Level	Code	Message
+Error	1017	Can't find file: './test/t1.MYI' (errno: 2 "No such file or directory")
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+a	INSERT	t1	set @s= 1	BEFORE						
+tr	DELETE	t1	begin end	BEFORE						
+# CRASH POINT: ddl_log_replace_broken_4
+t1.DATA1
+t1.TRG
+t1.frm
+tr.TRN
+show create table t1;
+Level	Code	Message
+Error	1017	Can't find file: './test/t1.MYI' (errno: 2 "No such file or directory")
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+a	INSERT	t1	set @s= 1	BEFORE						
+tr	DELETE	t1	begin end	BEFORE						
+# CRASH POINT: ddl_log_replace_broken_5
+t1.DATA1
+t1.TRG
+t1.frm
+tr.TRN
+show create table t1;
+Level	Code	Message
+Error	1017	Can't find file: './test/t1.MYI' (errno: 2 "No such file or directory")
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+a	INSERT	t1	set @s= 1	BEFORE						
+tr	DELETE	t1	begin end	BEFORE						
+# CRASH POINT: ddl_log_create_before_binlog
+t1.DATA1
+t1.TRG
+t1.frm
+tr.TRN
+show create table t1;
+Level	Code	Message
+Error	1017	Can't find file: './test/t1.MYI' (errno: 2 "No such file or directory")
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+a	INSERT	t1	set @s= 1	BEFORE						
+tr	DELETE	t1	begin end	BEFORE						
+# CRASH POINT: ddl_log_create_after_prepare_eof
+Warnings:
+Error	1017	Can't find file: './test/t1.MYI' (errno: 2 "No such file or directory")
+# No crash!
+t1.DATA1
+t1.DATA2
+t1.frm
+master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 LIKE const_table
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `new` int(11) DEFAULT NULL,
+  `b` int(11) DEFAULT NULL
+) ENGINE=ENGINE DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci
+Level	Code	Message
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+# CRASH POINT: ddl_log_create_after_binlog
+t1.DATA1
+t1.DATA2
+t1.frm
+master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 LIKE const_table
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `new` int(11) DEFAULT NULL,
+  `b` int(11) DEFAULT NULL
+) ENGINE=ENGINE DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci
+Level	Code	Message
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+# CRASH POINT: ddl_log_create_log_complete
+t1.DATA1
+t1.DATA2
+t1.frm
+master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 LIKE const_table
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `new` int(11) DEFAULT NULL,
+  `b` int(11) DEFAULT NULL
+) ENGINE=ENGINE DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci
+Level	Code	Message
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+# CRASH POINT: ddl_log_create_log_complete2
+t1.DATA1
+t1.DATA2
+t1.frm
+master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 LIKE const_table
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `new` int(11) DEFAULT NULL,
+  `b` int(11) DEFAULT NULL
+) ENGINE=ENGINE DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci
+Level	Code	Message
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+# CRASH POINT: ddl_log_create_log_complete3
+t1.DATA1
+t1.DATA2
+t1.frm
+master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 LIKE const_table
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `new` int(11) DEFAULT NULL,
+  `b` int(11) DEFAULT NULL
+) ENGINE=ENGINE DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci
+Level	Code	Message
+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_broken_1
+t1.DATA1
+t1.TRG
+t1.frm
+tr.TRN
+show create table t1;
+Level	Code	Message
+Error	1017	Can't find file: './test/t1.MYI' (errno: 2 "No such file or directory")
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+a	INSERT	t1	set @s= 1	BEFORE						
+tr	DELETE	t1	begin end	BEFORE						
+# CRASH POINT: ddl_log_replace_broken_2
+t1.DATA1
+t1.TRG
+t1.frm
+tr.TRN
+show create table t1;
+Level	Code	Message
+Error	1017	Can't find file: './test/t1.MYI' (errno: 2 "No such file or directory")
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+a	INSERT	t1	set @s= 1	BEFORE						
+tr	DELETE	t1	begin end	BEFORE						
+# CRASH POINT: ddl_log_replace_broken_3
+t1.DATA1
+t1.TRG
+t1.frm
+tr.TRN
+show create table t1;
+Level	Code	Message
+Error	1017	Can't find file: './test/t1.MYI' (errno: 2 "No such file or directory")
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+a	INSERT	t1	set @s= 1	BEFORE						
+tr	DELETE	t1	begin end	BEFORE						
+# CRASH POINT: ddl_log_replace_broken_4
+t1.DATA1
+t1.TRG
+t1.frm
+tr.TRN
+show create table t1;
+Level	Code	Message
+Error	1017	Can't find file: './test/t1.MYI' (errno: 2 "No such file or directory")
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+a	INSERT	t1	set @s= 1	BEFORE						
+tr	DELETE	t1	begin end	BEFORE						
+# CRASH POINT: ddl_log_replace_broken_5
+t1.DATA1
+t1.TRG
+t1.frm
+tr.TRN
+show create table t1;
+Level	Code	Message
+Error	1017	Can't find file: './test/t1.MYI' (errno: 2 "No such file or directory")
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+a	INSERT	t1	set @s= 1	BEFORE						
+tr	DELETE	t1	begin end	BEFORE						
+# CRASH POINT: ddl_log_create_before_binlog
+t1.DATA1
+t1.TRG
+t1.frm
+tr.TRN
+show create table t1;
+Level	Code	Message
+Error	1017	Can't find file: './test/t1.MYI' (errno: 2 "No such file or directory")
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+a	INSERT	t1	set @s= 1	BEFORE						
+tr	DELETE	t1	begin end	BEFORE						
+# CRASH POINT: ddl_log_create_after_prepare_eof
+t1.DATA1
+t1.TRG
+t1.frm
+tr.TRN
+show create table t1;
+Level	Code	Message
+Error	1017	Can't find file: './test/t1.MYI' (errno: 2 "No such file or directory")
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+a	INSERT	t1	set @s= 1	BEFORE						
+tr	DELETE	t1	begin end	BEFORE						
+# CRASH POINT: ddl_log_create_after_binlog
+t1.DATA1
+t1.TRG
+t1.frm
+tr.TRN
+show create table t1;
+Level	Code	Message
+Error	1017	Can't find file: './test/t1.MYI' (errno: 2 "No such file or directory")
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+a	INSERT	t1	set @s= 1	BEFORE						
+tr	DELETE	t1	begin end	BEFORE						
+# CRASH POINT: ddl_log_create_log_complete
+t1.DATA1
+t1.DATA2
+t1.frm
+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
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `new` int(11) DEFAULT NULL,
+  `b` int(11) DEFAULT NULL
+) ENGINE=ENGINE DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci
+Level	Code	Message
+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_complete2
+t1.DATA1
+t1.DATA2
+t1.frm
+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
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `new` int(11) DEFAULT NULL,
+  `b` int(11) DEFAULT NULL
+) ENGINE=ENGINE DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci
+Level	Code	Message
+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_complete3
+t1.DATA1
+t1.DATA2
+t1.frm
+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
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `new` int(11) DEFAULT NULL,
+  `b` int(11) DEFAULT NULL
+) ENGINE=ENGINE DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci
+Level	Code	Message
+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_broken.test b/mysql-test/suite/atomic/create_replace_broken.test
new file mode 100644
index 00000000000..278ef687db2
--- /dev/null
+++ b/mysql-test/suite/atomic/create_replace_broken.test
@@ -0,0 +1,163 @@
+--source include/have_debug.inc
+--source include/have_sequence.inc
+--source include/have_innodb.inc
+--source include/have_binlog_format_row.inc
+--source include/not_valgrind.inc
+
+--disable_query_log
+let $extra_option= '';
+let $save_debug=`select @@debug_dbug`;
+let $show_error=0,ER_FILE_NOT_FOUND,ER_NO_SUCH_TABLE;
+let $drop_error=0,ER_BAD_TABLE_ERROR;
+let $default_engine=MyISAM;
+--let $datadir= `select @@datadir`
+
+--eval set @@default_storage_engine=$default_engine
+--enable_query_log
+
+--echo # Crash recovery
+
+let $MYSQLD_DATADIR= `SELECT @@datadir`;
+
+let $crash_count= 11;
+let $crash_points='ddl_log_replace_broken_1',
+                  'ddl_log_replace_broken_2',
+                  'ddl_log_replace_broken_3',
+                  'ddl_log_replace_broken_4',
+                  'ddl_log_replace_broken_5',
+                  'ddl_log_create_before_binlog',
+                  'ddl_log_create_after_prepare_eof',
+                  'ddl_log_create_after_binlog',
+                  'ddl_log_create_log_complete',
+                  'ddl_log_create_log_complete2',
+                  'ddl_log_create_log_complete3';
+
+#let $crash_count=1;
+#let $crash_points='ddl_log_replace_broken_2';
+#                  'ddl_log_create_before_binlog3',
+#                  'ddl_log_create_before_binlog4';
+
+let $statement_count=3;
+let $statements='CREATE OR REPLACE TABLE t1 (new int) EXTRA_OPTION',
+                'CREATE OR REPLACE TABLE t1 LIKE const_table',
+                'CREATE OR REPLACE TABLE t1 EXTRA_OPTION SELECT * from const_table';
+
+#let $statement_count=1;
+#let $statements='CREATE OR REPLACE TABLE t1 EXTRA_OPTION SELECT * from const_table';
+
+--disable_query_log
+let create_const=`select REPLACE('create table const_table (new int, b int) EXTRA_OPTION', ' EXTRA_OPTION', $extra_option)`;
+eval $create_const;
+--replace_result $default_engine ENGINE ' PAGE_CHECKSUM=1' '' ' TRANSACTIONAL=0' ''
+show create table const_table;
+--enable_query_log
+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 REPLACE(ELT($r, $statements), ' EXTRA_OPTION', $extra_option)`;
+  --echo # QUERY: $STATEMENT
+
+  let $c=0;
+  while ($c < $crash_count)
+  {
+    inc $c;
+    let $crash= `select ELT($c, $crash_points)`;
+    let $fk_error= `select '$crash' like 'ddl_log_create_fk_fail%'`;
+
+    --eval set @@default_storage_engine=$default_engine
+    create or replace table t1 (old int);
+    if ($fk_error)
+    {
+      create or replace table t1 (old int primary key, y int) engine innodb;
+      create table t2 (x int references t1(old)) engine innodb;
+    }
+    create trigger a before insert on t1 for each row set @s= 1;
+    flush tables;
+    if ($MTR_COMBINATION_LOCK_TABLES)
+    {
+      lock tables t1 write, const_table read;
+    }
+
+    create trigger tr before delete on t1 for each row begin end;
+    --remove_file $datadir/test/t1.MYI
+
+    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!
+      if ($MTR_COMBINATION_LOCK_TABLES)
+      {
+        unlock tables;
+      }
+    }
+    # Check which tables still exists
+    --replace_result .MAD .DATA1 .MYD .DATA1 .MAI .DATA2 .MYI .DATA2
+    --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 ($default_engine == Aria)
+    {
+      # Again we suppress 'marked as crashed', it works differently in --ps
+      --disable_warnings
+    }
+    --replace_result $default_engine ENGINE InnoDB ENGINE ' PAGE_CHECKSUM=1' '' ' TRANSACTIONAL=0' ''
+    --enable_warnings
+    --enable_query_log
+    --error $show_error
+    show create table t1;
+    --disable_query_log
+    show warnings;
+    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
+    if ($fk_error)
+    {
+      drop table t2;
+    }
+    --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/sql/ddl_log.cc b/sql/ddl_log.cc
index 4cb70ddc30a..7f838ae7c76 100644
--- a/sql/ddl_log.cc
+++ b/sql/ddl_log.cc
@@ -3088,10 +3088,17 @@ static bool ddl_log_write(DDL_LOG_STATE *ddl_state,
   DBUG_ENTER("ddl_log_write");
 
   mysql_mutex_lock(&LOCK_gdl);
-  error= ((ddl_log_write_entry(ddl_log_entry, &log_entry)) ||
-          ddl_log_write_execute_entry(log_entry->entry_pos,
-                                      ddl_state->master_chain_pos,
-                                      &ddl_state->execute_entry));
+  error= ddl_log_write_entry(ddl_log_entry, &log_entry);
+
+  if (!error)
+  {
+    if (ddl_state->atomic_block)
+      ddl_state->atomic_block_entry= log_entry;
+    else
+      error= ddl_log_write_execute_entry(log_entry->entry_pos,
+                                        ddl_state->master_chain_pos,
+                                        &ddl_state->execute_entry);
+  }
   mysql_mutex_unlock(&LOCK_gdl);
   if (error)
   {
@@ -3189,7 +3196,10 @@ static bool ddl_log_drop_init(DDL_LOG_STATE *ddl_state,
   ddl_log_entry.from_db=      *const_cast<LEX_CSTRING*>(db);
   ddl_log_entry.tmp_name=     *const_cast<LEX_CSTRING*>(comment);
 
-  DBUG_RETURN(ddl_log_write(ddl_state, &ddl_log_entry));
+  ddl_state->drop_prev_entry= ddl_state->list;
+  bool result= ddl_log_write(ddl_state, &ddl_log_entry);
+
+  DBUG_RETURN(result);
 }
 
 
@@ -3246,6 +3256,8 @@ static bool ddl_log_drop(DDL_LOG_STATE *ddl_state,
   bzero(&ddl_log_entry, sizeof(ddl_log_entry));
 
   ddl_log_entry.action_type=  action_code;
+  ddl_log_entry.next_entry=   ddl_state->drop_prev_entry ? ddl_state->drop_prev_entry->entry_pos : 0;
+
   if (hton)
     lex_string_set(&ddl_log_entry.handler_name,
                    ha_resolve_storage_engine_name(hton));
@@ -3260,6 +3272,11 @@ static bool ddl_log_drop(DDL_LOG_STATE *ddl_state,
     goto error;
 
   (void) ddl_log_sync_no_lock();
+  /*
+    Sequence of multiple DROP actions must be replayed in the same order as
+    it were written into DDL log. The sequence has single DROP_INIT action
+    in the beginning.
+  */
   if (update_next_entry_pos(ddl_state->list->entry_pos,
                             log_entry->entry_pos))
   {
@@ -3677,3 +3694,32 @@ void ddl_log_link_chains(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;
 }
+
+/*
+  Don't update execute_entry until commit_atomic_block()
+*/
+
+void ddl_log_start_atomic_block(DDL_LOG_STATE *state)
+{
+  DBUG_ASSERT(!state->atomic_block);
+  state->atomic_block= true;
+}
+
+/*
+  Finish atomic block, update execute_entry.
+*/
+
+bool ddl_log_commit_atomic_block(DDL_LOG_STATE *state)
+{
+  if (!state->atomic_block)
+    return false;
+  DBUG_ASSERT(state->atomic_block_entry);
+  mysql_mutex_lock(&LOCK_gdl);
+  bool error= ddl_log_write_execute_entry(state->atomic_block_entry->entry_pos,
+                                          state->master_chain_pos,
+                                          &state->execute_entry);
+  mysql_mutex_unlock(&LOCK_gdl);
+  state->atomic_block= false;
+  state->atomic_block_entry= NULL;
+  return error;
+}
diff --git a/sql/ddl_log.h b/sql/ddl_log.h
index 809b51a0e10..720955a14f2 100644
--- a/sql/ddl_log.h
+++ b/sql/ddl_log.h
@@ -244,13 +244,17 @@ typedef struct st_ddl_log_state
   DDL_LOG_MEMORY_ENTRY *list;
   /* One execute entry per list */
   DDL_LOG_MEMORY_ENTRY *execute_entry;
+  DDL_LOG_MEMORY_ENTRY *atomic_block_entry;
   /*
     Entry used for PHASE updates. Normally same as first in 'list', but in
     case of a query log event, this points to the main event.
   */
   DDL_LOG_MEMORY_ENTRY *main_entry;
+  DDL_LOG_MEMORY_ENTRY *drop_prev_entry;
   uint16 flags;                                 /* Cache for flags */
   uint master_chain_pos;
+  /* Don't update execute_entry while accumulating atomic block of entries */
+  bool atomic_block;
   bool is_active() { return list != 0; }
 } DDL_LOG_STATE;
 
@@ -355,5 +359,7 @@ bool ddl_log_store_query(THD *thd, DDL_LOG_STATE *ddl_log_state,
                          const char *query, size_t length);
 bool ddl_log_delete_frm(DDL_LOG_STATE *ddl_state, const char *to_path);
 void ddl_log_link_chains(DDL_LOG_STATE *state, DDL_LOG_STATE *master_state);
+void ddl_log_start_atomic_block(DDL_LOG_STATE *state);
+bool ddl_log_commit_atomic_block(DDL_LOG_STATE *state);
 extern mysql_mutex_t LOCK_gdl;
 #endif /* DDL_LOG_INCLUDED */
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 56b87229c3a..4905bb18801 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -4430,9 +4430,13 @@ bool HA_CREATE_INFO::finalize_atomic_replace(THD *thd, TABLE_LIST *orig_table)
   DBUG_ASSERT(is_atomic_replace());
 
   debug_crash_here("ddl_log_create_before_install_new");
+  /* If old table exists, rename it to backup_name */
   if (old_hton)
   {
-    /* Old table exists, rename it to backup_name */
+    /*
+      Cleanup chain (ddl_log_state_rm) will not be executed unless
+      rollback chain (ddl_log_state_create) is active.
+    */
     ddl_log_link_chains(ddl_log_state_rm, ddl_log_state_create);
 
     cpath.length= build_table_filename(path, sizeof(path) - 1,
@@ -4460,6 +4464,7 @@ bool HA_CREATE_INFO::finalize_atomic_replace(THD *thd, TABLE_LIST *orig_table)
     if (thd->locked_tables_mode == LTM_LOCK_TABLES ||
         thd->locked_tables_mode == LTM_PRELOCKED_UNDER_LOCK_TABLES)
     {
+      /* NOTE: wait_while_table_is_used() was done in in create_table_impl(). */
       DBUG_ASSERT(thd->mdl_context.is_lock_owner(MDL_key::TABLE, db.str,
                                                  table_name.str,
                                                  MDL_EXCLUSIVE));
@@ -4481,13 +4486,65 @@ bool HA_CREATE_INFO::finalize_atomic_replace(THD *thd, TABLE_LIST *orig_table)
                                                    table_name;
     param.new_alias= backup_name.table_name;
     param.lock_triggers= true;
+    Dummy_error_handler suppress_errors;
+    /*
+      Suppress warnings for rename to backup. If something fails we drop the
+      old table and the warnings are shown in mysql_rm_table_no_locks().
+    */
+    thd->push_internal_handler(&suppress_errors);
     if (rename_table_and_triggers(thd, &param, NULL, orig_table,
                                   &backup_name.db, false, &dummy))
     {
-      if (locked_tables_decremented)
-        thd->locked_tables_list.add_back_last_deleted_lock(pos_in_locked_tables);
-      return true;
+      thd->pop_internal_handler();
+
+      debug_crash_here("ddl_log_replace_broken_1");
+      /* We don't need restore from backup entry anymore, disabling it */
+      ddl_log_update_phase(ddl_log_state_create, DDL_LOG_FINAL_PHASE);
+      debug_crash_here("ddl_log_replace_broken_2");
+
+      /*
+        Something is wrong with the old table! But C-O-R is almost done,
+        so we finish it anyway by dropping the old table and applying new table.
+      */
+
+      ddl_log_start_atomic_block(ddl_log_state_rm);
+      if (ddl_log_rename_table(ddl_log_state_rm, db_type,
+                               &db, &table_name,
+                               &tmp_name.db, &tmp_name.table_name,
+                               DDL_RENAME_PHASE_TRIGGER,
+                               DDL_LOG_FLAG_FROM_IS_TMP))
+      {
+        if (locked_tables_decremented)
+          thd->locked_tables_list.add_back_last_deleted_lock(pos_in_locked_tables);
+        return true;
+      }
+
+      debug_crash_here("ddl_log_replace_broken_3");
+      cpath.length= build_table_filename(path, sizeof(path) - 1, db.str,
+                                         table_name.str, "", 0);
+
+      if (ddl_log_drop_table_init(ddl_log_state_rm, &db, &empty_clex_str) ||
+          ddl_log_drop_table(ddl_log_state_rm, old_hton, &cpath,
+                            &db, &table_name, 0))
+      {
+        if (locked_tables_decremented)
+          thd->locked_tables_list.add_back_last_deleted_lock(pos_in_locked_tables);
+        return true;
+      }
+
+      debug_crash_here("ddl_log_replace_broken_4");
+      if (ddl_log_commit_atomic_block(ddl_log_state_rm))
+      {
+        if (locked_tables_decremented)
+          thd->locked_tables_list.add_back_last_deleted_lock(pos_in_locked_tables);
+        return true;
+      }
+
+      debug_crash_here("ddl_log_replace_broken_5");
+      return false;
     }
+    else
+      thd->pop_internal_handler();
     debug_crash_here("ddl_log_create_after_save_backup");
   }
 
@@ -4795,7 +4852,15 @@ int create_table_impl(THD *thd,
             if (!create_info->table)
             {
               if (open_table(thd, &table_list, &ot_ctx))
-                goto err;
+              {
+                thd->clear_error();
+                /*
+                  Remove from cache, otherwise that broken share will be used
+                  by normal CREATE .. SELECT and it will fail on open_table().
+                */
+                tdc_remove_table(thd, orig_db.str, orig_table_name.str);
+                goto skip_foreign_check;
+              }
               table= table_list.table;
             }
             FOREIGN_KEY_INFO *fk;
@@ -4812,6 +4877,7 @@ int create_table_impl(THD *thd,
               goto err;
             }
           }
+skip_foreign_check:
 
           if (thd->locked_tables_mode == LTM_LOCK_TABLES ||
               thd->locked_tables_mode == LTM_PRELOCKED_UNDER_LOCK_TABLES)
@@ -5600,6 +5666,10 @@ mysql_rename_table(handlerton *base, const LEX_CSTRING *old_db,
     else
       log_query= true;
   }
+  else if (file && error)
+  {
+    file->ha_rename_table(to_base, from_base);
+  }
   if (!error && log_query && !(flags & (FN_TO_IS_TMP | FN_FROM_IS_TMP)))
   {
     backup_log_info ddl_log;