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

7 files changed, 435 insertions, 17 deletions

diff --git a/mysql-test/main/create_or_replace.result b/mysql-test/main/create_or_replace.result
index 2ddd4ad9ff1..38094d1d128 100644
--- a/mysql-test/main/create_or_replace.result
+++ b/mysql-test/main/create_or_replace.result
@@ -913,12 +913,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;
 #
diff --git a/mysql-test/main/create_or_replace.test b/mysql-test/main/create_or_replace.test
index 9fb1ef16aef..43448e23ddc 100644
--- a/mysql-test/main/create_or_replace.test
+++ b/mysql-test/main/create_or_replace.test
@@ -699,9 +699,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;
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..1a25d39a6cb
--- /dev/null
+++ b/mysql-test/suite/atomic/create_replace_broken.result
@@ -0,0 +1,167 @@
+# 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.DATA2
+t1.TRG
+t1.frm
+tr.TRN
+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
+a	INSERT	t1	set @s= 1	BEFORE						
+tr	DELETE	t1	begin end	BEFORE						
+# CRASH POINT: ddl_log_replace_broken_3
+t1.DATA1
+t1.DATA2
+t1.TRG
+t1.frm
+tr.TRN
+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
+a	INSERT	t1	set @s= 1	BEFORE						
+tr	DELETE	t1	begin end	BEFORE						
+# CRASH POINT: ddl_log_replace_broken_4
+t1.DATA1
+t1.DATA2
+t1.TRG
+t1.frm
+tr.TRN
+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
+a	INSERT	t1	set @s= 1	BEFORE						
+tr	DELETE	t1	begin end	BEFORE						
+# CRASH POINT: ddl_log_drop_before_delete_table
+master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated by ddl recovery */
+show create table t1;
+Level	Code	Message
+Error	1146	Table 'test.t1' doesn't exist
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+# CRASH POINT: ddl_log_drop_after_delete_table
+master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated by ddl recovery */
+show create table t1;
+Level	Code	Message
+Error	1146	Table 'test.t1' doesn't exist
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+# CRASH POINT: ddl_log_drop_before_drop_trigger
+master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated by ddl recovery */
+show create table t1;
+Level	Code	Message
+Error	1146	Table 'test.t1' doesn't exist
+Trigger	Event	Table	Statement	Timing	Created	sql_mode	Definer	character_set_client	collation_connection	Database Collation
+# CRASH POINT: ddl_log_drop_before_drop_trigger2
+t1.DATA1
+t1.DATA2
+t1.frm
+master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated by ddl recovery */
+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_drop_after_drop_trigger
+t1.DATA1
+t1.DATA2
+t1.frm
+master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated by ddl recovery */
+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_drop_before_binlog
+Warnings:
+Error	1017	Can't find file: './test/t1.MYI' (errno: 2 "No such file or directory")
+Error	6	Error on delete of './test/t1.MYI' (Errcode: 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_drop_after_binlog
+Warnings:
+Error	1017	Can't find file: './test/t1.MYI' (errno: 2 "No such file or directory")
+Error	6	Error on delete of './test/t1.MYI' (Errcode: 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_save_backup
+t1.DATA1
+t1.DATA2
+t1.frm
+master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated by ddl recovery */
+master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated by ddl recovery */
+master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated by ddl recovery */
+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_install_new
+t1.DATA1
+t1.DATA2
+t1.frm
+master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated by ddl recovery */
+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
+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..ee763668452
--- /dev/null
+++ b/mysql-test/suite/atomic/create_replace_broken.test
@@ -0,0 +1,165 @@
+--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=13;
+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_drop_before_delete_table',
+                  'ddl_log_drop_after_delete_table',
+                  'ddl_log_drop_before_drop_trigger',
+                  'ddl_log_drop_before_drop_trigger2',
+                  'ddl_log_drop_after_drop_trigger',
+                  'ddl_log_drop_before_binlog',
+                  'ddl_log_drop_after_binlog',
+                  'ddl_log_create_after_save_backup',
+                  'ddl_log_create_after_install_new';
+
+#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..899c07a5807 100644
--- a/sql/ddl_log.cc
+++ b/sql/ddl_log.cc
@@ -2832,7 +2832,8 @@ int ddl_log_execute_recovery()
 
       uint cond_entry= (uint)(ddl_log_entry.unique_id >> DDL_LOG_RETRY_BITS);
 
-      if (cond_entry && is_execute_entry_active(cond_entry))
+      if ((cond_entry && is_execute_entry_active(cond_entry)) ||
+          !ddl_log_entry.next_entry)
       {
         if (disable_execute_entry(i))
           error= -1;
@@ -3672,8 +3673,20 @@ bool ddl_log_delete_frm(DDL_LOG_STATE *ddl_state, const char *to_path)
    CREATE OR REPLACE ... is used.
 */
 
-void ddl_log_link_chains(DDL_LOG_STATE *state, DDL_LOG_STATE *master_state)
+bool ddl_log_link_chains(DDL_LOG_STATE *state, DDL_LOG_STATE *master_state)
 {
+  if (!master_state->execute_entry)
+  {
+    mysql_mutex_lock(&LOCK_gdl);
+    if (ddl_log_write_execute_entry(0, master_state->master_chain_pos,
+                                    &master_state->execute_entry))
+    {
+      mysql_mutex_unlock(&LOCK_gdl);
+      return true;
+    }
+    mysql_mutex_unlock(&LOCK_gdl);
+  }
   DBUG_ASSERT(master_state->execute_entry);
   state->master_chain_pos= master_state->execute_entry->entry_pos;
+  return false;
 }
diff --git a/sql/ddl_log.h b/sql/ddl_log.h
index 809b51a0e10..cc09293f5fb 100644
--- a/sql/ddl_log.h
+++ b/sql/ddl_log.h
@@ -354,6 +354,6 @@ bool ddl_log_alter_table(DDL_LOG_STATE *ddl_state,
 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);
+bool ddl_log_link_chains(DDL_LOG_STATE *state, DDL_LOG_STATE *master_state);
 extern mysql_mutex_t LOCK_gdl;
 #endif /* DDL_LOG_INCLUDED */
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 1f075d7d088..7301240598b 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -4423,14 +4423,21 @@ bool HA_CREATE_INFO::finalize_atomic_replace(THD *thd, TABLE_LIST *orig_table)
   char path[FN_REFLEN + 1];
   cpath.str= path;
   bool locked_tables_decremented= false;
+  bool backed_old= true;
+  DDL_LOG_MEMORY_ENTRY *restore_backup;
 
   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 */
-    ddl_log_link_chains(ddl_log_state_rm, ddl_log_state_create);
+    /*
+      Cleanup chain (ddl_log_state_rm) will not be executed unless
+      rollback chain (ddl_log_state_create) is active.
+    */
+    if (ddl_log_link_chains(ddl_log_state_rm, ddl_log_state_create))
+      return true;
 
     cpath.length= build_table_filename(path, sizeof(path) - 1,
                                        backup_name.db.str,
@@ -4451,12 +4458,14 @@ bool HA_CREATE_INFO::finalize_atomic_replace(THD *thd, TABLE_LIST *orig_table)
                               DDL_RENAME_PHASE_TRIGGER,
                               DDL_LOG_FLAG_FROM_IS_TMP))
       return true;
+    restore_backup= ddl_log_state_create->main_entry;
 
     debug_crash_here("ddl_log_create_after_log_rename_backup");
 
     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));
@@ -4478,13 +4487,68 @@ 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;
+      debug_crash_here("ddl_log_replace_broken_1");
+      thd->pop_internal_handler();
+      /*
+        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.
+      */
+      if (ddl_log_rename_table(ddl_log_state_create, db_type,
+                               &db, &table_name,
+                               &tmp_name.db, &tmp_name.table_name,
+                               DDL_RENAME_PHASE_TRIGGER,
+                               DDL_LOG_FLAG_FROM_IS_TMP))
+        return true;
+
+      debug_crash_here("ddl_log_replace_broken_2");
+      /* We don't need restore from backup entry anymore, disabling it */
+      ddl_log_state_create->main_entry= restore_backup;
+      ddl_log_update_phase(ddl_log_state_create, DDL_LOG_FINAL_PHASE);
+      debug_crash_here("ddl_log_replace_broken_3");
+
+      /*
+        mysql_rm_table_no_locks() does its own DDL logging but it cannot add
+        to existing chain (ddl_log_drop_table_init() overwrites execute entry).
+        So we organize the separate chain drop_broken_table for that and link
+        the rollback chain so drop_broken_table will be executed first.
+      */
+      DDL_LOG_STATE drop_broken_table;
+      bzero(&drop_broken_table, sizeof(drop_broken_table));
+      if (ddl_log_link_chains(ddl_log_state_create, &drop_broken_table))
+      {
+        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");
+
+      TABLE_LIST table_list;
+      table_list.init_one_table(&db, &table_name, 0, TL_WRITE);
+
+      enum_locked_tables_mode ltm_save= thd->locked_tables_mode;
+      thd->locked_tables_mode= LTM_NONE;
+      if (mysql_rm_table_no_locks(thd, &table_list, &thd->db, &drop_broken_table,
+                                  0, 0, 0, 0, 1, 1))
+      {
+        thd->locked_tables_mode= ltm_save;
+        if (locked_tables_decremented)
+          thd->locked_tables_list.add_back_last_deleted_lock(pos_in_locked_tables);
+        return true;
+      }
+      thd->locked_tables_mode= ltm_save;
+      backed_old= false;
     }
+    else
+      thd->pop_internal_handler();
     debug_crash_here("ddl_log_create_after_save_backup");
   }
 
@@ -4498,8 +4562,9 @@ bool HA_CREATE_INFO::finalize_atomic_replace(THD *thd, TABLE_LIST *orig_table)
   param.lock_triggers= false;
   ulonglong option_bits_save= thd->variables.option_bits;
   thd->variables.option_bits&= ~OPTION_NO_FOREIGN_KEY_CHECKS;
-  if (ddl_log_create_table(ddl_log_state_create, param.from_table_hton,
-                           &cpath, &db, &table_name, false) ||
+  if ((backed_old &&
+       ddl_log_create_table(ddl_log_state_create, param.from_table_hton,
+                           &cpath, &db, &table_name, false)) ||
       rename_table_and_triggers(thd, &param, NULL, &tmp_name, &db, false,
                                 &dummy))
   {
@@ -4792,7 +4857,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;
@@ -4809,6 +4882,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)
@@ -5595,6 +5669,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;