MDEV-25292 Disable atomic replace for slave-generated or-replace

Also fixes wrong rollback of ddl_log_state_rm in case of non-atomic replace. There is no backup to remove in that case.

Author: midenok

mysql-test/suite/rpl/r/create_or_replace_mix.result

@@ -150,7 +150,7 @@ slave-bin.000001	#	Query	#	#	use `test`; create table t4 (server_2_to_be_delete
 slave-bin.000001	#	Gtid	#	#	GTID #-#-#
 slave-bin.000001	#	Query	#	#	use `test`; create table t1 (new_table int)
 slave-bin.000001	#	Gtid	#	#	BEGIN GTID #-#-#
-slave-bin.000001	#	Query	#	#	use `test`; CREATE TABLE `t2` (
+slave-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE `t2` (
  `a` int(11) DEFAULT NULL
 )
 slave-bin.000001	#	Annotate_rows	#	#	create table t2 select * from t9

mysql-test/suite/rpl/r/create_or_replace_statement.result

@@ -150,7 +150,7 @@ slave-bin.000001	#	Query	#	#	use `test`; create table t4 (server_2_to_be_delete
 slave-bin.000001	#	Gtid	#	#	GTID #-#-#
 slave-bin.000001	#	Query	#	#	use `test`; create table t1 (new_table int)
 slave-bin.000001	#	Gtid	#	#	BEGIN GTID #-#-#
-slave-bin.000001	#	Query	#	#	use `test`; CREATE TABLE `t2` (
+slave-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE `t2` (
  `a` int(11) DEFAULT NULL
 )
 slave-bin.000001	#	Annotate_rows	#	#	create table t2 select * from t9

sql/handler.h

@@ -2466,7 +2466,8 @@ struct Table_specification_st: public HA_CREATE_INFO,
  }
  bool is_atomic_replace() const
  {
- return or_replace() && is_atomic_replace_usable();
+ return or_replace() && !or_replace_slave_generated() &&
+ is_atomic_replace_usable();
  }
 };
 

sql/sql_table.cc

@@ -4398,6 +4398,8 @@ bool HA_CREATE_INFO::finalize_atomic_replace(THD *thd, TABLE_LIST *orig_table)
  char path[FN_REFLEN + 1];
  cpath.str= path;
 
+ DBUG_ASSERT(is_atomic_replace());
+
  debug_crash_here("ddl_log_create_before_install_new");
  if (old_hton)
  {
@@ -4498,7 +4500,10 @@ void HA_CREATE_INFO::finalize_ddl(THD *thd, bool roll_back)
  debug_crash_here("ddl_log_create_log_complete");
  ddl_log_complete(ddl_log_state_create);
  debug_crash_here("ddl_log_create_log_complete2");
- (void) ddl_log_revert(thd, ddl_log_state_rm);
+ if (is_atomic_replace())
+ (void) ddl_log_revert(thd, ddl_log_state_rm);
+ else
+ ddl_log_complete(ddl_log_state_rm);
  debug_crash_here("ddl_log_create_log_complete3");
  }
 }