Fixed that the we don't change CREATE to CREATE OR REPLACE, except if the slave removed an existing table as part of CREATE.

This will help the following replicaition scenario:
MariaDB 10.0 master (statement replication) -> MariaDB 10.0 slave (row based replication) -> MySQL or MariaDB 5.x slave


mysql-test/r/mysqld--help.result:
  Updated help text
mysql-test/suite/rpl/r/create_or_replace_mix.result:
  Added more tests
mysql-test/suite/rpl/r/create_or_replace_row.result:
  Added more tests
mysql-test/suite/rpl/r/create_or_replace_statement.result:
  Added more tests
mysql-test/suite/rpl/t/create_or_replace.inc:
  Added more tests
sql/handler.h:
  Added org_options so that we can detect what come from the query and what was possible added later.
sql/sql_insert.cc:
  Only write CREATE OR REPLACE if was originally specified or if we delete a conflicting table as part of create
sql/sql_parse.cc:
  Remember orginal create options
sql/sql_table.cc:
  Only write CREATE OR REPLACE if was originally specified or if we delete a conflicting table as part of create
sql/sys_vars.cc:
  Updated help text

10 files changed, 222 insertions, 23 deletions

diff --git a/mysql-test/r/mysqld--help.result b/mysql-test/r/mysqld--help.result
index 5e37fde47a7..5a468cdf236 100644
--- a/mysql-test/r/mysqld--help.result
+++ b/mysql-test/r/mysqld--help.result
@@ -871,12 +871,12 @@ The following options may be given as the first argument:
  --slave-compressed-protocol 
  Use compression on master/slave protocol
  --slave-ddl-exec-mode=name 
- Modes for how replication events should be executed.
- Legal values are STRICT and IDEMPOTENT (default). In
- IDEMPOTENT mode, replication will not stop for DDL
- operations that are idempotent. This means that CREATE
- TABLE is treated CREATE TABLE OR REPLACE and DROP TABLE
- is threated as DROP TABLE IF EXISTS. 
+ How replication events should be executed. Legal values
+ are STRICT and IDEMPOTENT (default). In IDEMPOTENT mode,
+ replication will not stop for DDL operations that are
+ idempotent. This means that CREATE TABLE is treated as
+ CREATE TABLE OR REPLACE and DROP TABLE is treated as DROP
+ TABLE IF EXISTS.
  --slave-domain-parallel-threads=# 
  Maximum number of parallel threads to use on slave for
  events in a single replication domain. When using
@@ -886,14 +886,13 @@ The following options may be given as the first argument:
  as many threads as it wants, up to the value of
  slave_parallel_threads.
  --slave-exec-mode=name 
- Modes for how replication events should be executed.
- Legal values are STRICT (default) and IDEMPOTENT. In
- IDEMPOTENT mode, replication will not stop for operations
- that are idempotent. For example, in row based
- replication attempts to delete rows that doesn't exist
- will be ignored.In STRICT mode, replication will stop on
- any unexpected difference between the master and the
- slave
+ How replication events should be executed. Legal values
+ are STRICT (default) and IDEMPOTENT. In IDEMPOTENT mode,
+ replication will not stop for operations that are
+ idempotent. For example, in row based replication
+ attempts to delete rows that doesn't exist will be
+ ignored. In STRICT mode, replication will stop on any
+ unexpected difference between the master and the slave
  --slave-load-tmpdir=name 
  The location where the slave should put its temporary
  files when replicating a LOAD DATA INFILE command
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 88837ebbf46..839032a305c 100644
--- a/mysql-test/suite/rpl/r/create_or_replace_mix.result
+++ b/mysql-test/suite/rpl/r/create_or_replace_mix.result
@@ -115,6 +115,59 @@ t1	CREATE TABLE `t1` (
 ) ENGINE=MyISAM DEFAULT CHARSET=latin1
 drop table t1;
 #
+# Check how CREATE is logged on slave in case of conflicts
+#
+create table t1 (server_2_to_be_delete int);
+create table t2 (server_2_to_be_delete int);
+create table t4 (server_2_to_be_delete int);
+set @org_binlog_format=@@binlog_format;
+set @@global.binlog_format="ROW";
+stop slave;
+include/wait_for_slave_to_stop.inc
+start slave;
+include/wait_for_slave_to_start.inc
+create temporary table t9 (a int);
+insert into t9 values(1);
+create table t1 (new_table int);
+create table t2 select * from t9;
+create table t4 like t9;
+create table t5 select * from t9;
+binlog from server 2
+include/show_binlog_events.inc
+Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+slave-bin.000001	#	Gtid	#	#	GTID #-#-#
+slave-bin.000001	#	Query	#	#	use `test`; create table t1 (server_2_to_be_delete int)
+slave-bin.000001	#	Gtid	#	#	GTID #-#-#
+slave-bin.000001	#	Query	#	#	use `test`; create table t2 (server_2_to_be_delete int)
+slave-bin.000001	#	Gtid	#	#	GTID #-#-#
+slave-bin.000001	#	Query	#	#	use `test`; create table t4 (server_2_to_be_delete int)
+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` (
+  `a` int(11) DEFAULT NULL
+)
+slave-bin.000001	#	Table_map	#	#	table_id: # (test.t2)
+slave-bin.000001	#	Write_rows_v1	#	#	table_id: # flags: STMT_END_F
+slave-bin.000001	#	Query	#	#	COMMIT
+slave-bin.000001	#	Gtid	#	#	GTID #-#-#
+slave-bin.000001	#	Query	#	#	use `test`; CREATE TABLE `t4` (
+  `a` int(11) DEFAULT NULL
+)
+slave-bin.000001	#	Gtid	#	#	BEGIN GTID #-#-#
+slave-bin.000001	#	Query	#	#	use `test`; CREATE TABLE `t5` (
+  `a` int(11) DEFAULT NULL
+)
+slave-bin.000001	#	Table_map	#	#	table_id: # (test.t5)
+slave-bin.000001	#	Write_rows_v1	#	#	table_id: # flags: STMT_END_F
+slave-bin.000001	#	Query	#	#	COMMIT
+set @@global.binlog_format=@org_binlog_format;
+stop slave;
+include/wait_for_slave_to_stop.inc
+start slave;
+include/wait_for_slave_to_start.inc
+drop table t1,t2,t4,t5,t9;
+#
 # Ensure that DROP TABLE is run as DROP IF NOT EXISTS
 #
 create table t1 (server_1_ver_1 int);
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 9921ece7588..6e29d02e3bc 100644
--- a/mysql-test/suite/rpl/r/create_or_replace_row.result
+++ b/mysql-test/suite/rpl/r/create_or_replace_row.result
@@ -137,6 +137,59 @@ t1	CREATE TABLE `t1` (
 ) ENGINE=MyISAM DEFAULT CHARSET=latin1
 drop table t1;
 #
+# Check how CREATE is logged on slave in case of conflicts
+#
+create table t1 (server_2_to_be_delete int);
+create table t2 (server_2_to_be_delete int);
+create table t4 (server_2_to_be_delete int);
+set @org_binlog_format=@@binlog_format;
+set @@global.binlog_format="ROW";
+stop slave;
+include/wait_for_slave_to_stop.inc
+start slave;
+include/wait_for_slave_to_start.inc
+create temporary table t9 (a int);
+insert into t9 values(1);
+create table t1 (new_table int);
+create table t2 select * from t9;
+create table t4 like t9;
+create table t5 select * from t9;
+binlog from server 2
+include/show_binlog_events.inc
+Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+slave-bin.000001	#	Gtid	#	#	GTID #-#-#
+slave-bin.000001	#	Query	#	#	use `test`; create table t1 (server_2_to_be_delete int)
+slave-bin.000001	#	Gtid	#	#	GTID #-#-#
+slave-bin.000001	#	Query	#	#	use `test`; create table t2 (server_2_to_be_delete int)
+slave-bin.000001	#	Gtid	#	#	GTID #-#-#
+slave-bin.000001	#	Query	#	#	use `test`; create table t4 (server_2_to_be_delete int)
+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` (
+  `a` int(11) DEFAULT NULL
+)
+slave-bin.000001	#	Table_map	#	#	table_id: # (test.t2)
+slave-bin.000001	#	Write_rows_v1	#	#	table_id: # flags: STMT_END_F
+slave-bin.000001	#	Query	#	#	COMMIT
+slave-bin.000001	#	Gtid	#	#	GTID #-#-#
+slave-bin.000001	#	Query	#	#	use `test`; CREATE TABLE `t4` (
+  `a` int(11) DEFAULT NULL
+)
+slave-bin.000001	#	Gtid	#	#	BEGIN GTID #-#-#
+slave-bin.000001	#	Query	#	#	use `test`; CREATE TABLE `t5` (
+  `a` int(11) DEFAULT NULL
+)
+slave-bin.000001	#	Table_map	#	#	table_id: # (test.t5)
+slave-bin.000001	#	Write_rows_v1	#	#	table_id: # flags: STMT_END_F
+slave-bin.000001	#	Query	#	#	COMMIT
+set @@global.binlog_format=@org_binlog_format;
+stop slave;
+include/wait_for_slave_to_stop.inc
+start slave;
+include/wait_for_slave_to_start.inc
+drop table t1,t2,t4,t5,t9;
+#
 # Ensure that DROP TABLE is run as DROP IF NOT EXISTS
 #
 create table t1 (server_1_ver_1 int);
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 bb799848185..8550976e87a 100644
--- a/mysql-test/suite/rpl/r/create_or_replace_statement.result
+++ b/mysql-test/suite/rpl/r/create_or_replace_statement.result
@@ -115,6 +115,59 @@ t1	CREATE TABLE `t1` (
 ) ENGINE=MyISAM DEFAULT CHARSET=latin1
 drop table t1;
 #
+# Check how CREATE is logged on slave in case of conflicts
+#
+create table t1 (server_2_to_be_delete int);
+create table t2 (server_2_to_be_delete int);
+create table t4 (server_2_to_be_delete int);
+set @org_binlog_format=@@binlog_format;
+set @@global.binlog_format="ROW";
+stop slave;
+include/wait_for_slave_to_stop.inc
+start slave;
+include/wait_for_slave_to_start.inc
+create temporary table t9 (a int);
+insert into t9 values(1);
+create table t1 (new_table int);
+create table t2 select * from t9;
+create table t4 like t9;
+create table t5 select * from t9;
+binlog from server 2
+include/show_binlog_events.inc
+Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+slave-bin.000001	#	Gtid	#	#	GTID #-#-#
+slave-bin.000001	#	Query	#	#	use `test`; create table t1 (server_2_to_be_delete int)
+slave-bin.000001	#	Gtid	#	#	GTID #-#-#
+slave-bin.000001	#	Query	#	#	use `test`; create table t2 (server_2_to_be_delete int)
+slave-bin.000001	#	Gtid	#	#	GTID #-#-#
+slave-bin.000001	#	Query	#	#	use `test`; create table t4 (server_2_to_be_delete int)
+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 OR REPLACE TABLE `t2` (
+  `a` int(11) DEFAULT NULL
+)
+slave-bin.000001	#	Table_map	#	#	table_id: # (test.t2)
+slave-bin.000001	#	Write_rows_v1	#	#	table_id: # flags: STMT_END_F
+slave-bin.000001	#	Query	#	#	COMMIT
+slave-bin.000001	#	Gtid	#	#	GTID #-#-#
+slave-bin.000001	#	Query	#	#	use `test`; CREATE TABLE `t4` (
+  `a` int(11) DEFAULT NULL
+)
+slave-bin.000001	#	Gtid	#	#	BEGIN GTID #-#-#
+slave-bin.000001	#	Query	#	#	use `test`; CREATE TABLE `t5` (
+  `a` int(11) DEFAULT NULL
+)
+slave-bin.000001	#	Table_map	#	#	table_id: # (test.t5)
+slave-bin.000001	#	Write_rows_v1	#	#	table_id: # flags: STMT_END_F
+slave-bin.000001	#	Query	#	#	COMMIT
+set @@global.binlog_format=@org_binlog_format;
+stop slave;
+include/wait_for_slave_to_stop.inc
+start slave;
+include/wait_for_slave_to_start.inc
+drop table t1,t2,t4,t5,t9;
+#
 # Ensure that DROP TABLE is run as DROP IF NOT EXISTS
 #
 create table t1 (server_1_ver_1 int);
diff --git a/mysql-test/suite/rpl/t/create_or_replace.inc b/mysql-test/suite/rpl/t/create_or_replace.inc
index b7ba4bc2ba6..7d0dc487061 100644
--- a/mysql-test/suite/rpl/t/create_or_replace.inc
+++ b/mysql-test/suite/rpl/t/create_or_replace.inc
@@ -96,6 +96,43 @@ connection server_1;
 drop table t1;
 
 --echo #
+--echo # Check how CREATE is logged on slave in case of conflicts
+--echo #
+
+save_master_pos;
+connection server_2;
+sync_with_master;
+--let $binlog_start=query_get_value(SHOW MASTER STATUS, Position, 1)
+create table t1 (server_2_to_be_delete int);
+create table t2 (server_2_to_be_delete int);
+create table t4 (server_2_to_be_delete int);
+set @org_binlog_format=@@binlog_format;
+set @@global.binlog_format="ROW";
+stop slave;
+--source include/wait_for_slave_to_stop.inc
+start slave;
+--source include/wait_for_slave_to_start.inc
+connection server_1;
+create temporary table t9 (a int);
+insert into t9 values(1);
+create table t1 (new_table int);
+create table t2 select * from t9;
+create table t4 like t9;
+create table t5 select * from t9;
+save_master_pos;
+connection server_2;
+sync_with_master;
+--echo binlog from server 2
+--source include/show_binlog_events.inc
+set @@global.binlog_format=@org_binlog_format;
+stop slave;
+--source include/wait_for_slave_to_stop.inc
+start slave;
+--source include/wait_for_slave_to_start.inc
+connection server_1;
+drop table t1,t2,t4,t5,t9;
+
+--echo #
 --echo # Ensure that DROP TABLE is run as DROP IF NOT EXISTS
 --echo #
 
diff --git a/sql/handler.h b/sql/handler.h
index 37c2c86dce8..13b783b964b 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -1591,6 +1591,7 @@ struct HA_CREATE_INFO
   uint stats_sample_pages;
   uint null_bits;                       /* NULL bits at start of record */
   uint options;				/* OR of HA_CREATE_ options */
+  uint org_options;                     /* original options from query */
   uint merge_insert_method;
   uint extra_size;                      /* length of extra data segment */
   SQL_I_List<TABLE_LIST> merge_list;
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index 59223a6091f..d61af758ced 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -4164,8 +4164,9 @@ select_create::binlog_show_create_table(TABLE **tables, uint count)
 
   result= store_create_info(thd, &tmp_table_list, &query, create_info,
                             /* show_database */ TRUE,
-                            MY_TEST(create_info->options &
-                                    HA_LEX_CREATE_REPLACE));
+                            MY_TEST(create_info->org_options &
+                                    HA_LEX_CREATE_REPLACE) ||
+                            create_info->table_was_deleted);
   DBUG_ASSERT(result == 0); /* store_create_info() always return 0 */
 
   if (mysql_bin_log.is_open())
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index e0b205c40c6..6a5073be3a7 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -2893,6 +2893,7 @@ case SQLCOM_PREPARE:
       CREATE TABLE OR EXISTS failures by dropping the table and
       retrying the create.
     */
+    create_info.org_options= create_info.options;
     if (thd->slave_thread &&
         slave_ddl_exec_mode_options == SLAVE_EXEC_MODE_IDEMPOTENT &&
         !(lex->create_info.options & HA_LEX_CREATE_IF_NOT_EXISTS))
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index b2107a0ccba..dec010bb24f 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -5460,8 +5460,9 @@ bool mysql_create_like_table(THD* thd, TABLE_LIST* table,
           int result __attribute__((unused))=
             store_create_info(thd, table, &query,
                               create_info, FALSE /* show_database */,
-                              MY_TEST(create_info->options &
-                                      HA_LEX_CREATE_REPLACE));
+                              MY_TEST(create_info->org_options &
+                                      HA_LEX_CREATE_REPLACE) ||
+                              create_info->table_was_deleted);
 
           DBUG_ASSERT(result == 0); // store_create_info() always return 0
           do_logging= FALSE;
diff --git a/sql/sys_vars.cc b/sql/sys_vars.cc
index 520e4c21468..d22e8498642 100644
--- a/sql/sys_vars.cc
+++ b/sql/sys_vars.cc
@@ -2717,11 +2717,11 @@ static Sys_var_mybool Sys_slave_compressed_protocol(
 static const char *slave_exec_mode_names[]= {"STRICT", "IDEMPOTENT", 0};
 static Sys_var_enum Slave_exec_mode(
        "slave_exec_mode",
-       "Modes for how replication events should be executed. Legal values "
+       "How replication events should be executed. Legal values "
        "are STRICT (default) and IDEMPOTENT. In IDEMPOTENT mode, "
        "replication will not stop for operations that are idempotent. "
        "For example, in row based replication attempts to delete rows that "
-       "doesn't exist will be ignored."
+       "doesn't exist will be ignored. "
        "In STRICT mode, replication will stop on any unexpected difference "
        "between the master and the slave",
        GLOBAL_VAR(slave_exec_mode_options), CMD_LINE(REQUIRED_ARG),
@@ -2729,11 +2729,11 @@ static Sys_var_enum Slave_exec_mode(
 
 static Sys_var_enum Slave_ddl_exec_mode(
        "slave_ddl_exec_mode",
-       "Modes for how replication events should be executed. Legal values "
+       "How replication events should be executed. Legal values "
        "are STRICT and IDEMPOTENT (default). In IDEMPOTENT mode, "
        "replication will not stop for DDL operations that are idempotent. "
-       "This means that CREATE TABLE is treated CREATE TABLE OR REPLACE and "
-       "DROP TABLE is threated as DROP TABLE IF EXISTS. ",
+       "This means that CREATE TABLE is treated as CREATE TABLE OR REPLACE and "
+       "DROP TABLE is treated as DROP TABLE IF EXISTS.",
        GLOBAL_VAR(slave_ddl_exec_mode_options), CMD_LINE(REQUIRED_ARG),
        slave_exec_mode_names, DEFAULT(SLAVE_EXEC_MODE_IDEMPOTENT));