Replication changes for CREATE OR REPLACE TABLE

- CREATE TABLE is by default executed on the slave as CREATE OR REPLACE
- DROP TABLE is by default executed on the slave as DROP TABLE IF NOT EXISTS

This means that a slave will by default continue even if we try to create
a table that existed on the slave (the table will be deleted and re-created) or
if we try to drop a table that didn't exist on the slave.
This should be safe as instead of having the slave stop because of an inconsistency between
master and slave, it will fix the inconsistency.
Those that would prefer to get a stopped slave instead for the above cases can set slave_ddl_exec_mode to STRICT. 

- Ensure that a CREATE OR REPLACE TABLE which dropped a table is replicated
- DROP TABLE that generated an error on master is handled as an identical DROP TABLE on the slave (IF NOT EXISTS is not added in this case)
- Added slave_ddl_exec_mode variable to decide how DDL's are replicated

New logic for handling BEGIN GTID ... COMMIT from the binary log:

- When we find a BEGIN GTID, we start a transaction and set OPTION_GTID_BEGIN
- When we find COMMIT, we reset OPTION_GTID_BEGIN and execute the normal COMMIT code.
- While OPTION_GTID_BEGIN is set:
  - We don't generate implict commits before or after statements
  - All tables are regarded as transactional tables in the binary log (to ensure things are executed exactly as on the master)
- We reset OPTION_GTID_BEGIN also on rollback

This will help ensuring that we don't get any sporadic commits (and thus new GTID's) on the slave and will help keep the GTID's between master and slave in sync.


mysql-test/extra/rpl_tests/rpl_log.test:
  Added testing of mode slave_ddl_exec_mode=STRICT
mysql-test/r/mysqld--help.result:
  New help messages
mysql-test/suite/rpl/r/create_or_replace_mix.result:
  Testing of CREATE OR REPLACE TABLE with replication
mysql-test/suite/rpl/r/create_or_replace_row.result:
  Testing of CREATE OR REPLACE TABLE with replication
mysql-test/suite/rpl/r/create_or_replace_statement.result:
  Testing replication of create or replace
mysql-test/suite/rpl/r/rpl_gtid_startpos.result:
  Test must be run in slave_ddl_exec_mode=STRICT as part of the test depends on that DROP TABLE should fail on slave.
mysql-test/suite/rpl/r/rpl_row_log.result:
  Updated result
mysql-test/suite/rpl/r/rpl_row_log_innodb.result:
  Updated result
mysql-test/suite/rpl/r/rpl_row_show_relaylog_events.result:
  Updated result
mysql-test/suite/rpl/r/rpl_stm_log.result:
  Updated result
mysql-test/suite/rpl/r/rpl_stm_mix_show_relaylog_events.result:
  Updated result
mysql-test/suite/rpl/r/rpl_temp_table_mix_row.result:
  Updated result
mysql-test/suite/rpl/t/create_or_replace.inc:
  Testing of CREATE OR REPLACE TABLE with replication
mysql-test/suite/rpl/t/create_or_replace_mix.cnf:
  Testing of CREATE OR REPLACE TABLE with replication
mysql-test/suite/rpl/t/create_or_replace_mix.test:
  Testing of CREATE OR REPLACE TABLE with replication
mysql-test/suite/rpl/t/create_or_replace_row.cnf:
  Testing of CREATE OR REPLACE TABLE with replication
mysql-test/suite/rpl/t/create_or_replace_row.test:
  Testing of CREATE OR REPLACE TABLE with replication
mysql-test/suite/rpl/t/create_or_replace_statement.cnf:
  Testing of CREATE OR REPLACE TABLE with replication
mysql-test/suite/rpl/t/create_or_replace_statement.test:
  Testing of CREATE OR REPLACE TABLE with replication
mysql-test/suite/rpl/t/rpl_gtid_startpos.test:
  Test must be run in slave_ddl_exec_mode=STRICT as part of the test depends on that DROP TABLE should fail on slave.
mysql-test/suite/rpl/t/rpl_stm_log.test:
  Removed some lines
mysql-test/suite/sys_vars/r/slave_ddl_exec_mode_basic.result:
  Testing of slave_ddl_exec_mode
mysql-test/suite/sys_vars/t/slave_ddl_exec_mode_basic.test:
  Testing of slave_ddl_exec_mode
sql/handler.cc:
  Regard all tables as transactional in commit if OPTION_GTID_BEGIN is set.
  This is to ensure that statments are not commited too early if non transactional tables are used.
sql/log.cc:
  Regard all tables as transactional in commit if OPTION_GTID_BEGIN is set.
  Also treat 'direct' log events as transactional (to get them logged as they where on the master)
sql/log_event.cc:
  Ensure that the new error from DROP TABLE when trying to drop a view is treated same as the old one.
  Store error code that slave expects in THD.
  Set OPTION_GTID_BEGIN if we find a BEGIN.
  Reset OPTION_GTID_BEGIN if we find a COMMIT.
sql/mysqld.cc:
  Added slave_ddl_exec_mode_options
sql/mysqld.h:
  Added slave_ddl_exec_mode_options
sql/rpl_gtid.cc:
  Reset OPTION_GTID_BEGIN if we record a gtid (safety)
sql/sql_class.cc:
  Regard all tables as transactional in commit if OPTION_GTID_BEGIN is set.
sql/sql_class.h:
  Added to THD: log_current_statement and slave_expected_error
sql/sql_insert.cc:
  Ensure that CREATE OR REPLACE is logged if table was deleted.
  Don't do implicit commit for CREATE if we are under OPTION_GTID_BEGIN
sql/sql_parse.cc:
  Change CREATE TABLE -> CREATE OR REPLACE TABLE for slaves
  Change DROP TABLE -> DROP TABLE IF EXISTS for slaves
  CREATE TABLE doesn't force implicit commit in case of OPTION_GTID_BEGIN
  Don't do commits before or after any statement if OPTION_GTID_BEGIN was set.
sql/sql_priv.h:
  Added OPTION_GTID_BEGIN
sql/sql_show.cc:
  Enhanced store_create_info() to also be able to handle CREATE OR REPLACE
sql/sql_show.h:
  Updated prototype
sql/sql_table.cc:
  Ensure that CREATE OR REPLACE is logged if table was deleted.
sql/sys_vars.cc:
  Added slave_ddl_exec_mode
sql/transaction.cc:
  Added warning if we got a GTID under OPTION_GTID_BEGIN

mysql-test/extra/rpl_tests/rpl_log.test

@@ -18,8 +18,12 @@ start slave;
 --source include/wait_for_slave_to_start.inc
 
 let $VERSION=`select version()`;
-
+# Lets run this test in STRICT MODE (DROP TABLE is not DROP TABLE IF EXISTS)
+connection slave;
+set @save_slave_ddl_exec_mode=@@global.slave_ddl_exec_mode;
+set @@global.slave_ddl_exec_mode=STRICT;
 connection master;
+
 eval create table t1(n int not null auto_increment primary key)ENGINE=$engine_type;
 insert into t1 values (NULL);
 drop table t1;
@@ -141,3 +145,5 @@ drop table t1;
 # End of 4.1 tests
 
 sync_slave_with_master;
+set @@global.slave_ddl_exec_mode=@save_slave_ddl_exec_mode;
+connection master;

mysql-test/r/mysqld--help.result

@@ -856,13 +856,22 @@ The following options may be given as the first argument:
  --skip-slave-start  If set, slave is not autostarted.
  --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. 
  --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. In STRICT mode, replication will
+ that are idempotent. For example, in row based
- stop on any unexpected difference between the master and
+ replication attempts to delete rows that doesn't exist
- the slave
+ 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
@@ -1260,6 +1269,7 @@ skip-networking FALSE
 skip-show-database FALSE
 skip-slave-start FALSE
 slave-compressed-protocol FALSE
+slave-ddl-exec-mode IDEMPOTENT
 slave-exec-mode STRICT
 slave-max-allowed-packet 1073741824
 slave-net-timeout 3600

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

@@ -0,0 +1,162 @@
+include/rpl_init.inc [topology=1->2]
+create table t2 (a int) engine=myisam;
+insert into t2 values (0),(1),(2),(2);
+create temporary table t3 (a_in_temporary int) engine=myisam;
+#
+# Check how create table and create or replace table are logged
+#
+create table t1 (to_be_deleted int);
+CREATE TABLE t1 AS SELECT 1 AS f1;
+CREATE OR REPLACE TABLE t1 AS SELECT 2 AS f1;
+CREATE OR REPLACE table t1 like t2;
+CREATE OR REPLACE table t1 like t3;
+drop table t1;
+binlog from server 1
+include/show_binlog_events.inc
+Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; create table t2 (a int) engine=myisam
+master-bin.000001	#	Gtid	#	#	BEGIN GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; insert into t2 values (0),(1),(2),(2)
+master-bin.000001	#	Query	#	#	COMMIT
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; create temporary table t3 (a_in_temporary int) engine=myisam
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; CREATE TABLE t1 AS SELECT 1 AS f1
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 AS SELECT 2 AS f1
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE table t1 like t2
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE table t1 like t3
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; DROP TABLE `t1` /* generated by server */
+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 t2 (a int) engine=myisam
+slave-bin.000001	#	Gtid	#	#	BEGIN GTID #-#-#
+slave-bin.000001	#	Query	#	#	use `test`; insert into t2 values (0),(1),(2),(2)
+slave-bin.000001	#	Query	#	#	COMMIT
+slave-bin.000001	#	Gtid	#	#	GTID #-#-#
+slave-bin.000001	#	Query	#	#	use `test`; create temporary table t3 (a_in_temporary int) engine=myisam
+slave-bin.000001	#	Gtid	#	#	GTID #-#-#
+slave-bin.000001	#	Query	#	#	use `test`; create table t1 (to_be_deleted int)
+slave-bin.000001	#	Gtid	#	#	GTID #-#-#
+slave-bin.000001	#	Query	#	#	use `test`; CREATE TABLE t1 AS SELECT 1 AS f1
+slave-bin.000001	#	Gtid	#	#	GTID #-#-#
+slave-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 AS SELECT 2 AS f1
+slave-bin.000001	#	Gtid	#	#	GTID #-#-#
+slave-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE table t1 like t2
+slave-bin.000001	#	Gtid	#	#	GTID #-#-#
+slave-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE table t1 like t3
+slave-bin.000001	#	Gtid	#	#	GTID #-#-#
+slave-bin.000001	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated by server */
+#
+# Ensure that also failed create_or_replace are logged
+#
+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'
+binlog from server 1
+include/show_binlog_events.inc
+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)
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; create or replace table t1
+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`; CREATE OR REPLACE TABLE `t1` (
+  `a` int(11) NOT NULL,
+  PRIMARY KEY (`a`)
+)
+master-bin.000001	#	Table_map	#	#	table_id: # (test.t1)
+master-bin.000001	#	Write_rows_v1	#	#	table_id: # flags: STMT_END_F
+master-bin.000001	#	Query	#	#	ROLLBACK
+show tables;
+Tables_in_test
+t1
+t2
+drop table if exists t1,t2;
+Warnings:
+Note	1051	Unknown table 'test.t1'
+#
+# Ensure that CREATE are run as CREATE OR REPLACE on slave
+#
+create table t1 (server_2_to_be_delete int);
+create table t1 (new_table int);
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `new_table` int(11) DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+drop table t1;
+#
+# Ensure that DROP TABLE is run as DROP IF NOT EXISTS
+#
+create table t1 (server_1_ver_1 int);
+create table t4 (server_1_ver_2 int);
+drop table t1;
+drop table t1,t4;
+create table t1 (server_2_ver_2 int);
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `server_2_ver_2` int(11) DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+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`; DROP TABLE `t1` /* generated by server */
+slave-bin.000001	#	Gtid	#	#	GTID #-#-#
+slave-bin.000001	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1`,`t4` /* generated by server */
+slave-bin.000001	#	Gtid	#	#	GTID #-#-#
+slave-bin.000001	#	Query	#	#	use `test`; create table t1 (server_2_ver_2 int)
+drop table t1;
+#
+# Ensure that CREATE ... SELECT is recorded as one GTID on the slave
+#
+create table t1 (a int);
+insert into t1 values (0),(1),(2);
+create table t2 engine=myisam select * from t1;
+create or replace table t2 engine=innodb select * from t1;
+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 (a int)
+slave-bin.000001	#	Gtid	#	#	BEGIN GTID #-#-#
+slave-bin.000001	#	Table_map	#	#	table_id: # (test.t1)
+slave-bin.000001	#	Write_rows_v1	#	#	table_id: # flags: STMT_END_F
+slave-bin.000001	#	Query	#	#	COMMIT
+slave-bin.000001	#	Gtid	#	#	BEGIN GTID #-#-#
+slave-bin.000001	#	Query	#	#	use `test`; CREATE TABLE `t2` (
+  `a` int(11) DEFAULT NULL
+) ENGINE=MyISAM
+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	#	#	BEGIN GTID #-#-#
+slave-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE `t2` (
+  `a` int(11) DEFAULT NULL
+) ENGINE=InnoDB
+slave-bin.000001	#	Table_map	#	#	table_id: # (test.t2)
+slave-bin.000001	#	Write_rows_v1	#	#	table_id: # flags: STMT_END_F
+slave-bin.000001	#	Xid	#	#	COMMIT /* XID */
+drop table t1;
+drop table t2;
+include/rpl_end.inc

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

@@ -0,0 +1,184 @@
+include/rpl_init.inc [topology=1->2]
+create table t2 (a int) engine=myisam;
+insert into t2 values (0),(1),(2),(2);
+create temporary table t3 (a_in_temporary int) engine=myisam;
+#
+# Check how create table and create or replace table are logged
+#
+create table t1 (to_be_deleted int);
+CREATE TABLE t1 AS SELECT 1 AS f1;
+CREATE OR REPLACE TABLE t1 AS SELECT 2 AS f1;
+CREATE OR REPLACE table t1 like t2;
+CREATE OR REPLACE table t1 like t3;
+drop table t1;
+binlog from server 1
+include/show_binlog_events.inc
+Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; create table t2 (a int) engine=myisam
+master-bin.000001	#	Gtid	#	#	BEGIN GTID #-#-#
+master-bin.000001	#	Table_map	#	#	table_id: # (test.t2)
+master-bin.000001	#	Write_rows_v1	#	#	table_id: # flags: STMT_END_F
+master-bin.000001	#	Query	#	#	COMMIT
+master-bin.000001	#	Gtid	#	#	BEGIN GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; CREATE TABLE `t1` (
+  `f1` int(1) NOT NULL DEFAULT '0'
+)
+master-bin.000001	#	Table_map	#	#	table_id: # (test.t1)
+master-bin.000001	#	Write_rows_v1	#	#	table_id: # flags: STMT_END_F
+master-bin.000001	#	Query	#	#	COMMIT
+master-bin.000001	#	Gtid	#	#	BEGIN GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE `t1` (
+  `f1` int(1) NOT NULL DEFAULT '0'
+)
+master-bin.000001	#	Table_map	#	#	table_id: # (test.t1)
+master-bin.000001	#	Write_rows_v1	#	#	table_id: # flags: STMT_END_F
+master-bin.000001	#	Query	#	#	COMMIT
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE table t1 like t2
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE `t1` (
+  `a_in_temporary` int(11) DEFAULT NULL
+)
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; DROP TABLE `t1` /* generated by server */
+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 t2 (a int) engine=myisam
+slave-bin.000001	#	Gtid	#	#	BEGIN GTID #-#-#
+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 t1 (to_be_deleted int)
+slave-bin.000001	#	Gtid	#	#	BEGIN GTID #-#-#
+slave-bin.000001	#	Query	#	#	use `test`; CREATE TABLE `t1` (
+  `f1` int(1) NOT NULL DEFAULT '0'
+)
+slave-bin.000001	#	Table_map	#	#	table_id: # (test.t1)
+slave-bin.000001	#	Write_rows_v1	#	#	table_id: # flags: STMT_END_F
+slave-bin.000001	#	Query	#	#	COMMIT
+slave-bin.000001	#	Gtid	#	#	BEGIN GTID #-#-#
+slave-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE `t1` (
+  `f1` int(1) NOT NULL DEFAULT '0'
+)
+slave-bin.000001	#	Table_map	#	#	table_id: # (test.t1)
+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 OR REPLACE table t1 like t2
+slave-bin.000001	#	Gtid	#	#	GTID #-#-#
+slave-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE `t1` (
+  `a_in_temporary` int(11) DEFAULT NULL
+)
+slave-bin.000001	#	Gtid	#	#	GTID #-#-#
+slave-bin.000001	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated by server */
+#
+# Ensure that also failed create_or_replace are logged
+#
+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'
+binlog from server 1
+include/show_binlog_events.inc
+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)
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; create or replace table t1
+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`; CREATE OR REPLACE TABLE `t1` (
+  `a` int(11) NOT NULL,
+  PRIMARY KEY (`a`)
+)
+master-bin.000001	#	Table_map	#	#	table_id: # (test.t1)
+master-bin.000001	#	Write_rows_v1	#	#	table_id: # flags: STMT_END_F
+master-bin.000001	#	Query	#	#	ROLLBACK
+show tables;
+Tables_in_test
+t1
+t2
+drop table if exists t1,t2;
+Warnings:
+Note	1051	Unknown table 'test.t1'
+#
+# Ensure that CREATE are run as CREATE OR REPLACE on slave
+#
+create table t1 (server_2_to_be_delete int);
+create table t1 (new_table int);
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `new_table` int(11) DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+drop table t1;
+#
+# Ensure that DROP TABLE is run as DROP IF NOT EXISTS
+#
+create table t1 (server_1_ver_1 int);
+create table t4 (server_1_ver_2 int);
+drop table t1;
+drop table t1,t4;
+create table t1 (server_2_ver_2 int);
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `server_2_ver_2` int(11) DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+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`; DROP TABLE `t1` /* generated by server */
+slave-bin.000001	#	Gtid	#	#	GTID #-#-#
+slave-bin.000001	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1`,`t4` /* generated by server */
+slave-bin.000001	#	Gtid	#	#	GTID #-#-#
+slave-bin.000001	#	Query	#	#	use `test`; create table t1 (server_2_ver_2 int)
+drop table t1;
+#
+# Ensure that CREATE ... SELECT is recorded as one GTID on the slave
+#
+create table t1 (a int);
+insert into t1 values (0),(1),(2);
+create table t2 engine=myisam select * from t1;
+create or replace table t2 engine=innodb select * from t1;
+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 (a int)
+slave-bin.000001	#	Gtid	#	#	BEGIN GTID #-#-#
+slave-bin.000001	#	Table_map	#	#	table_id: # (test.t1)
+slave-bin.000001	#	Write_rows_v1	#	#	table_id: # flags: STMT_END_F
+slave-bin.000001	#	Query	#	#	COMMIT
+slave-bin.000001	#	Gtid	#	#	BEGIN GTID #-#-#
+slave-bin.000001	#	Query	#	#	use `test`; CREATE TABLE `t2` (
+  `a` int(11) DEFAULT NULL
+) ENGINE=MyISAM
+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	#	#	BEGIN GTID #-#-#
+slave-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE `t2` (
+  `a` int(11) DEFAULT NULL
+) ENGINE=InnoDB
+slave-bin.000001	#	Table_map	#	#	table_id: # (test.t2)
+slave-bin.000001	#	Write_rows_v1	#	#	table_id: # flags: STMT_END_F
+slave-bin.000001	#	Xid	#	#	COMMIT /* XID */
+drop table t1;
+drop table t2;
+include/rpl_end.inc

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

@@ -0,0 +1,144 @@
+include/rpl_init.inc [topology=1->2]
+create table t2 (a int) engine=myisam;
+insert into t2 values (0),(1),(2),(2);
+create temporary table t3 (a_in_temporary int) engine=myisam;
+#
+# Check how create table and create or replace table are logged
+#
+create table t1 (to_be_deleted int);
+CREATE TABLE t1 AS SELECT 1 AS f1;
+CREATE OR REPLACE TABLE t1 AS SELECT 2 AS f1;
+CREATE OR REPLACE table t1 like t2;
+CREATE OR REPLACE table t1 like t3;
+drop table t1;
+binlog from server 1
+include/show_binlog_events.inc
+Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; create table t2 (a int) engine=myisam
+master-bin.000001	#	Gtid	#	#	BEGIN GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; insert into t2 values (0),(1),(2),(2)
+master-bin.000001	#	Query	#	#	COMMIT
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; create temporary table t3 (a_in_temporary int) engine=myisam
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; CREATE TABLE t1 AS SELECT 1 AS f1
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 AS SELECT 2 AS f1
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE table t1 like t2
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE table t1 like t3
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; DROP TABLE `t1` /* generated by server */
+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 t2 (a int) engine=myisam
+slave-bin.000001	#	Gtid	#	#	BEGIN GTID #-#-#
+slave-bin.000001	#	Query	#	#	use `test`; insert into t2 values (0),(1),(2),(2)
+slave-bin.000001	#	Query	#	#	COMMIT
+slave-bin.000001	#	Gtid	#	#	GTID #-#-#
+slave-bin.000001	#	Query	#	#	use `test`; create temporary table t3 (a_in_temporary int) engine=myisam
+slave-bin.000001	#	Gtid	#	#	GTID #-#-#
+slave-bin.000001	#	Query	#	#	use `test`; create table t1 (to_be_deleted int)
+slave-bin.000001	#	Gtid	#	#	GTID #-#-#
+slave-bin.000001	#	Query	#	#	use `test`; CREATE TABLE t1 AS SELECT 1 AS f1
+slave-bin.000001	#	Gtid	#	#	GTID #-#-#
+slave-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 AS SELECT 2 AS f1
+slave-bin.000001	#	Gtid	#	#	GTID #-#-#
+slave-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE table t1 like t2
+slave-bin.000001	#	Gtid	#	#	GTID #-#-#
+slave-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE table t1 like t3
+slave-bin.000001	#	Gtid	#	#	GTID #-#-#
+slave-bin.000001	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated by server */
+#
+# Ensure that also failed create_or_replace are logged
+#
+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'
+binlog from server 1
+include/show_binlog_events.inc
+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)
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; create or replace table t1
+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	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; create or replace table t1 (a int primary key) select a from t2
+show tables;
+Tables_in_test
+t2
+drop table if exists t1,t2;
+Warnings:
+Note	1051	Unknown table 'test.t1'
+#
+# Ensure that CREATE are run as CREATE OR REPLACE on slave
+#
+create table t1 (server_2_to_be_delete int);
+create table t1 (new_table int);
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `new_table` int(11) DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+drop table t1;
+#
+# Ensure that DROP TABLE is run as DROP IF NOT EXISTS
+#
+create table t1 (server_1_ver_1 int);
+create table t4 (server_1_ver_2 int);
+drop table t1;
+drop table t1,t4;
+create table t1 (server_2_ver_2 int);
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `server_2_ver_2` int(11) DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+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`; DROP TABLE `t1` /* generated by server */
+slave-bin.000001	#	Gtid	#	#	GTID #-#-#
+slave-bin.000001	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1`,`t4` /* generated by server */
+slave-bin.000001	#	Gtid	#	#	GTID #-#-#
+slave-bin.000001	#	Query	#	#	use `test`; create table t1 (server_2_ver_2 int)
+drop table t1;
+#
+# Ensure that CREATE ... SELECT is recorded as one GTID on the slave
+#
+create table t1 (a int);
+insert into t1 values (0),(1),(2);
+create table t2 engine=myisam select * from t1;
+create or replace table t2 engine=innodb select * from t1;
+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 (a int)
+slave-bin.000001	#	Gtid	#	#	BEGIN GTID #-#-#
+slave-bin.000001	#	Query	#	#	use `test`; insert into t1 values (0),(1),(2)
+slave-bin.000001	#	Query	#	#	COMMIT
+slave-bin.000001	#	Gtid	#	#	GTID #-#-#
+slave-bin.000001	#	Query	#	#	use `test`; create table t2 engine=myisam select * from t1
+slave-bin.000001	#	Gtid	#	#	GTID #-#-#
+slave-bin.000001	#	Query	#	#	use `test`; create or replace table t2 engine=innodb select * from t1
+drop table t1;
+drop table t2;
+include/rpl_end.inc

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

@@ -110,6 +110,8 @@ DROP TABLE t1;
 SET SQL_LOG_BIN=1;
 RESET SLAVE;
 SET GLOBAL gtid_slave_pos="";
+SET @save_slave_ddl_exec_mode=@@global.slave_ddl_exec_mode;
+SET GLOBAL slave_ddl_exec_mode=STRICT;
 include/start_slave.inc
 SELECT * FROM t1 ORDER BY a;
 a
@@ -225,4 +227,5 @@ a
 1
 2
 DROP TABLE t1;
+set @@global.slave_ddl_exec_mode=@save_slave_ddl_exec_mode;
 include/rpl_end.inc

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

@@ -6,6 +6,8 @@ reset master;
 reset slave;
 start slave;
 include/wait_for_slave_to_start.inc
+set @save_slave_ddl_exec_mode=@@global.slave_ddl_exec_mode;
+set @@global.slave_ddl_exec_mode=STRICT;
 create table t1(n int not null auto_increment primary key)ENGINE=MyISAM;
 insert into t1 values (NULL);
 drop table t1;
@@ -287,4 +289,5 @@ a	b
 5	1
 6	1
 drop table t1;
+set @@global.slave_ddl_exec_mode=@save_slave_ddl_exec_mode;
 include/rpl_end.inc

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

@@ -6,6 +6,8 @@ reset master;
 reset slave;
 start slave;
 include/wait_for_slave_to_start.inc
+set @save_slave_ddl_exec_mode=@@global.slave_ddl_exec_mode;
+set @@global.slave_ddl_exec_mode=STRICT;
 create table t1(n int not null auto_increment primary key)ENGINE=InnoDB;
 insert into t1 values (NULL);
 drop table t1;
@@ -287,4 +289,5 @@ a	b
 5	1
 6	1
 drop table t1;
+set @@global.slave_ddl_exec_mode=@save_slave_ddl_exec_mode;
 include/rpl_end.inc

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

@@ -172,7 +172,7 @@ include/show_events.inc
 Log_name	Pos	Event_type	Server_id	End_log_pos	Info
 slave-bin.000002	#	Binlog_checkpoint	#	#	slave-bin.000002
 slave-bin.000002	#	Gtid	#	#	GTID #-#-#
-slave-bin.000002	#	Query	#	#	use `test`; DROP TABLE `t1` /* generated by server */
+slave-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated by server */
 ******** [slave] SHOW BINLOG EVENTS IN <FILE> LIMIT 2 ********
 include/show_events.inc
 Log_name	Pos	Event_type	Server_id	End_log_pos	Info
@@ -181,7 +181,7 @@ slave-bin.000002	#	Gtid	#	#	GTID #-#-#
 ******** [slave] SHOW BINLOG EVENTS IN <FILE> LIMIT 2,3 ********
 include/show_events.inc
 Log_name	Pos	Event_type	Server_id	End_log_pos	Info
-slave-bin.000002	#	Query	#	#	use `test`; DROP TABLE `t1` /* generated by server */
+slave-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated by server */
 ******** [slave] SHOW BINLOG EVENTS  ********
 include/show_events.inc
 Log_name	Pos	Event_type	Server_id	End_log_pos	Info

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

@@ -6,6 +6,8 @@ reset master;
 reset slave;
 start slave;
 include/wait_for_slave_to_start.inc
+set @save_slave_ddl_exec_mode=@@global.slave_ddl_exec_mode;
+set @@global.slave_ddl_exec_mode=STRICT;
 create table t1(n int not null auto_increment primary key)ENGINE=MyISAM;
 insert into t1 values (NULL);
 drop table t1;
@@ -286,4 +288,5 @@ a	b
 5	1
 6	1
 drop table t1;
+set @@global.slave_ddl_exec_mode=@save_slave_ddl_exec_mode;
 include/rpl_end.inc

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

@@ -154,7 +154,7 @@ include/show_events.inc
 Log_name	Pos	Event_type	Server_id	End_log_pos	Info
 slave-bin.000002	#	Binlog_checkpoint	#	#	slave-bin.000002
 slave-bin.000002	#	Gtid	#	#	GTID #-#-#
-slave-bin.000002	#	Query	#	#	use `test`; DROP TABLE `t1` /* generated by server */
+slave-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated by server */
 ******** [slave] SHOW BINLOG EVENTS IN <FILE> LIMIT 2 ********
 include/show_events.inc
 Log_name	Pos	Event_type	Server_id	End_log_pos	Info
@@ -163,7 +163,7 @@ slave-bin.000002	#	Gtid	#	#	GTID #-#-#
 ******** [slave] SHOW BINLOG EVENTS IN <FILE> LIMIT 2,3 ********
 include/show_events.inc
 Log_name	Pos	Event_type	Server_id	End_log_pos	Info
-slave-bin.000002	#	Query	#	#	use `test`; DROP TABLE `t1` /* generated by server */
+slave-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated by server */
 ******** [slave] SHOW BINLOG EVENTS  ********
 include/show_events.inc
 Log_name	Pos	Event_type	Server_id	End_log_pos	Info

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

@@ -62,7 +62,7 @@ slave-bin.000001	#	Query	#	#	use `test`; ALTER TABLE t1_tmp ADD COLUMN b INT
 slave-bin.000001	#	Gtid	#	#	GTID #-#-#
 slave-bin.000001	#	Query	#	#	DROP TEMPORARY TABLE IF EXISTS `test`.`t1_tmp` /* generated by server */
 slave-bin.000001	#	Gtid	#	#	GTID #-#-#
-slave-bin.000001	#	Query	#	#	use `test`; DROP TABLE `t2` /* generated by server */
+slave-bin.000001	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t2` /* generated by server */
 slave-bin.000001	#	Gtid	#	#	BEGIN GTID #-#-#
 slave-bin.000001	#	Table_map	#	#	table_id: # (test.t1)
 slave-bin.000001	#	Write_rows_v1	#	#	table_id: # flags: STMT_END_F
@@ -73,7 +73,7 @@ slave-bin.000001	#	Gtid	#	#	BEGIN GTID #-#-#
 slave-bin.000001	#	Query	#	#	use `test`; INSERT INTO t1 VALUES (2)
 slave-bin.000001	#	Xid	#	#	COMMIT /* XID */
 slave-bin.000001	#	Gtid	#	#	GTID #-#-#
-slave-bin.000001	#	Query	#	#	use `test`; DROP TABLE `t3`,`t1` /* generated by server */
+slave-bin.000001	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t3`,`t1` /* generated by server */
 
 # Bug#55478 Row events wrongly apply on the temporary table of the same name
 # ==========================================================================

mysql-test/suite/rpl/t/create_or_replace.inc

@@ -0,0 +1,137 @@
+# Test CREATE OR REPLACE TABLE in replication
+--source include/have_innodb.inc
+
+--let $rpl_topology=1->2
+--source include/rpl_init.inc
+
+# Create help tables
+create table t2 (a int) engine=myisam;
+insert into t2 values (0),(1),(2),(2);
+create temporary table t3 (a_in_temporary int) engine=myisam;
+
+--echo #
+--echo # Check how create table and create or replace table are logged
+--echo #
+
+save_master_pos;
+connection server_2;
+sync_with_master;
+create table t1 (to_be_deleted int);
+
+connection server_1;
+CREATE TABLE t1 AS SELECT 1 AS f1;
+CREATE OR REPLACE TABLE t1 AS SELECT 2 AS f1;
+CREATE OR REPLACE table t1 like t2;
+CREATE OR REPLACE table t1 like t3;
+drop table t1;
+
+--echo binlog from server 1
+--source include/show_binlog_events.inc
+save_master_pos;
+connection server_2;
+sync_with_master;
+--echo binlog from server 2
+--source include/show_binlog_events.inc
+
+connection server_1;
+
+--echo #
+--echo # Ensure that also failed create_or_replace are logged
+--echo #
+
+--let $binlog_start=query_get_value(SHOW MASTER STATUS, Position, 1)
+
+create table t1 (a int);
+--error ER_TABLE_MUST_HAVE_COLUMNS
+create or replace table t1;
+drop table if exists t1;
+# The following is not logged as t1 does not exists;
+--error ER_DUP_ENTRY
+create or replace table t1 (a int primary key) select a from t2;
+
+create table t1 (a int);
+# This should be logged as we will delete t1
+--error ER_DUP_ENTRY
+create or replace table t1 (a int primary key) select a from t2;
+
+--echo binlog from server 1
+--source include/show_binlog_events.inc
+save_master_pos;
+connection server_2;
+sync_with_master;
+show tables;
+connection server_1;
+
+drop table if exists t1,t2;
+
+--echo #
+--echo # Ensure that CREATE are run as CREATE OR REPLACE on slave
+--echo #
+
+save_master_pos;
+connection server_2;
+sync_with_master;
+create table t1 (server_2_to_be_delete int);
+connection server_1;
+create table t1 (new_table int);
+
+save_master_pos;
+connection server_2;
+sync_with_master;
+
+show create table t1;
+connection server_1;
+drop table t1;
+
+--echo #
+--echo # Ensure that DROP TABLE is run as DROP IF NOT EXISTS
+--echo #
+
+create table t1 (server_1_ver_1 int);
+create table t4 (server_1_ver_2 int);
+
+save_master_pos;
+connection server_2;
+sync_with_master;
+--let $binlog_start=query_get_value(SHOW MASTER STATUS, Position, 1)
+
+# Drop the table on the slave
+drop table t1;
+connection server_1;
+drop table t1,t4;
+create table t1 (server_2_ver_2 int);
+save_master_pos;
+connection server_2;
+sync_with_master;
+show create table t1;
+--echo binlog from server 2
+--source include/show_binlog_events.inc
+connection server_1;
+drop table t1;
+
+--echo #
+--echo # Ensure that CREATE ... SELECT is recorded as one GTID on the slave
+--echo #
+
+save_master_pos;
+connection server_2;
+sync_with_master;
+--let $binlog_start=query_get_value(SHOW MASTER STATUS, Position, 1)
+connection server_1;
+
+create table t1 (a int);
+insert into t1 values (0),(1),(2);
+create table t2 engine=myisam select * from t1;
+create or replace table t2 engine=innodb select * from t1;
+save_master_pos;
+connection server_2;
+sync_with_master;
+--echo binlog from server 2
+--source include/show_binlog_events.inc
+connection server_1;
+drop table t1;
+
+# Clean up
+drop table t2;
+
+--source include/rpl_end.inc

mysql-test/suite/rpl/t/create_or_replace_mix.cnf

@@ -0,0 +1,9 @@
+!include suite/rpl/my.cnf
+
+[mysqld.1]
+log-slave-updates
+loose-innodb
+
+[mysqld.2]
+log-slave-updates
+loose-innodb

mysql-test/suite/rpl/t/create_or_replace_mix.test

@@ -0,0 +1,4 @@
+# Testing create or replace table in mixed mode.
+
+--source include/have_binlog_format_mixed.inc
+--source create_or_replace.inc

mysql-test/suite/rpl/t/create_or_replace_row.cnf

@@ -0,0 +1,9 @@
+!include suite/rpl/my.cnf
+
+[mysqld.1]
+log-slave-updates
+loose-innodb
+
+[mysqld.2]
+log-slave-updates
+loose-innodb

mysql-test/suite/rpl/t/create_or_replace_row.test

@@ -0,0 +1,4 @@
+# Testing create or replace table in mixed mode.
+
+--source include/have_binlog_format_row.inc
+--source create_or_replace.inc

mysql-test/suite/rpl/t/create_or_replace_statement.cnf

@@ -0,0 +1,9 @@
+!include suite/rpl/my.cnf
+
+[mysqld.1]
+log-slave-updates
+loose-innodb
+
+[mysqld.2]
+log-slave-updates
+loose-innodb

mysql-test/suite/rpl/t/create_or_replace_statement.test

@@ -0,0 +1,4 @@
+# Testing create or replace table in mixed mode.
+
+--source include/have_binlog_format_statement.inc
+--source create_or_replace.inc

mysql-test/suite/rpl/t/rpl_gtid_startpos.test

@@ -165,7 +165,7 @@ SET GLOBAL gtid_slave_pos="";
 SELECT * FROM t1 ORDER BY a;
 
 
-# Same thing, but this time using SQL_LOG_BIN=0 to avoid polliting the
+# Same thing, but this time using SQL_LOG_BIN=0 to avoid polluting the
 # slave binlog.
 
 --connection server_2
@@ -175,6 +175,9 @@ DROP TABLE t1;
 SET SQL_LOG_BIN=1;
 RESET SLAVE;
 SET GLOBAL gtid_slave_pos="";
+# Ensure that the slave fails because of missing table to be dropped
+SET @save_slave_ddl_exec_mode=@@global.slave_ddl_exec_mode;
+SET GLOBAL slave_ddl_exec_mode=STRICT;
 
 --source include/start_slave.inc
 --sync_with_master
@@ -349,6 +352,7 @@ SELECT * FROM t1 ORDER BY a;
 # Clean up.
 --connection server_1
 DROP TABLE t1;
-
+--connection server_2
+set @@global.slave_ddl_exec_mode=@save_slave_ddl_exec_mode;
 
 --source include/rpl_end.inc

mysql-test/suite/rpl/t/rpl_stm_log.test

@@ -4,7 +4,5 @@
 let $engine_type=MyISAM;
 -- source extra/rpl_tests/rpl_log.test
 
-
 # End of 4.1 tests
-# Adding comment for force manual merge 5.0 -> wl1012: Delete me
 --source include/rpl_end.inc

mysql-test/suite/sys_vars/r/slave_ddl_exec_mode_basic.result

@@ -0,0 +1,39 @@
+SET @start_value = @@global.slave_ddl_exec_mode;
+SELECT @@global.slave_ddl_exec_mode;
+@@global.slave_ddl_exec_mode
+IDEMPOTENT
+SELECT @@slave_ddl_exec_mode = @@GLOBAL.slave_ddl_exec_mode;
+@@slave_ddl_exec_mode = @@GLOBAL.slave_ddl_exec_mode
+1
+1 Expected
+SELECT COUNT(@@slave_ddl_exec_mode);
+COUNT(@@slave_ddl_exec_mode)
+1
+1 Expected
+SELECT COUNT(@@local.slave_ddl_exec_mode);
+ERROR HY000: Variable 'slave_ddl_exec_mode' is a GLOBAL variable
+Expected error 'Variable is a GLOBAL variable'
+SELECT COUNT(@@SESSION.slave_ddl_exec_mode);
+ERROR HY000: Variable 'slave_ddl_exec_mode' is a GLOBAL variable
+Expected error 'Variable is a GLOBAL variable'
+SELECT COUNT(@@GLOBAL.slave_ddl_exec_mode);
+COUNT(@@GLOBAL.slave_ddl_exec_mode)
+1
+1 Expected
+SELECT slave_ddl_exec_mode = @@SESSION.version;
+ERROR 42S22: Unknown column 'slave_ddl_exec_mode' in 'field list'
+Expected error 'Readonly variable'
+SET @@GLOBAL.slave_ddl_exec_mode=STRICT;
+SELECT @@GLOBAL.slave_ddl_exec_mode;
+@@GLOBAL.slave_ddl_exec_mode
+STRICT
+SET @@GLOBAL.slave_ddl_exec_mode=IDEMPOTENT;
+SELECT @@GLOBAL.slave_ddl_exec_mode;
+@@GLOBAL.slave_ddl_exec_mode
+IDEMPOTENT
+SET @@GLOBAL.slave_ddl_exec_mode=XXX;
+ERROR 42000: Variable 'slave_ddl_exec_mode' can't be set to the value of 'XXX'
+SELECT @@GLOBAL.slave_ddl_exec_mode;
+@@GLOBAL.slave_ddl_exec_mode
+IDEMPOTENT
+SET @@global.slave_ddl_exec_mode= @start_value;

mysql-test/suite/sys_vars/t/slave_ddl_exec_mode_basic.test

@@ -0,0 +1,67 @@
+############## mysql-test\t\slave_ddl_exec_mode_basic.test ####################
+#                                                                             #
+# Variable Name: slave_ddl_exec_mode                                          #
+# Scope: GLOBAL & SESSION                                                     #
+# Access Type: Dynamic                                                        #
+# Data Type: Numeric                                                          #
+# Default Value: 1                                                            #
+# Range: 1 - 65536                                                            #
+#                                                                             #
+#                                                                             #
+# Description: Test Cases of Dynamic System Variable slave_ddl_exec_mode      #
+#              that checks the behavior of this variable in the following ways#
+#              * Default Value                                                #
+#              * Valid & Invalid values                                       #
+#              * Scope & Access method                                        #
+#              * Data Integrity                                               #
+#                                                                             #
+###############################################################################
+
+--source include/not_embedded.inc
+--source include/load_sysvars.inc
+
+######################################################################## 
+#                    START OF slave_ddl_exec_mode TESTS               #
+######################################################################## 
+
+SET @start_value = @@global.slave_ddl_exec_mode;
+
+SELECT @@global.slave_ddl_exec_mode;
+
+SELECT @@slave_ddl_exec_mode = @@GLOBAL.slave_ddl_exec_mode;
+--echo 1 Expected
+
+SELECT COUNT(@@slave_ddl_exec_mode);
+--echo 1 Expected
+
+--Error ER_INCORRECT_GLOBAL_LOCAL_VAR
+SELECT COUNT(@@local.slave_ddl_exec_mode);
+--echo Expected error 'Variable is a GLOBAL variable'
+
+--Error ER_INCORRECT_GLOBAL_LOCAL_VAR
+SELECT COUNT(@@SESSION.slave_ddl_exec_mode);
+--echo Expected error 'Variable is a GLOBAL variable'
+
+SELECT COUNT(@@GLOBAL.slave_ddl_exec_mode);
+--echo 1 Expected
+
+--Error ER_BAD_FIELD_ERROR
+SELECT slave_ddl_exec_mode = @@SESSION.version;
+--echo Expected error 'Readonly variable'
+
+
+SET @@GLOBAL.slave_ddl_exec_mode=STRICT;
+SELECT @@GLOBAL.slave_ddl_exec_mode;
+
+SET @@GLOBAL.slave_ddl_exec_mode=IDEMPOTENT;
+SELECT @@GLOBAL.slave_ddl_exec_mode;
+
+--error ER_WRONG_VALUE_FOR_VAR
+SET @@GLOBAL.slave_ddl_exec_mode=XXX;
+SELECT @@GLOBAL.slave_ddl_exec_mode;
+
+SET @@global.slave_ddl_exec_mode= @start_value;
+
+######################################################################## 
+#                    END OF slave_ddl_exec_mode TESTS                  #
+######################################################################## 

plugin/metadata_lock_info/metadata_lock_info.cc

@@ -37,6 +37,7 @@ static const LEX_STRING metadata_lock_info_lock_mode[] = {
   { C_STRING_WITH_LEN("MDL_SHARED_HIGH_PRIO") },
   { C_STRING_WITH_LEN("MDL_SHARED_READ") },
   { C_STRING_WITH_LEN("MDL_SHARED_WRITE") },
+  { C_STRING_WITH_LEN("MDL_SHARED_UPGRADABLE") },
   { C_STRING_WITH_LEN("MDL_SHARED_NO_WRITE") },
   { C_STRING_WITH_LEN("MDL_SHARED_NO_READ_WRITE") },
   { C_STRING_WITH_LEN("MDL_EXCLUSIVE") },

sql/handler.cc

@@ -1251,7 +1251,8 @@ int ha_commit_trans(THD *thd, bool all)
     the changes are not durable as they might be rolled back if the
     enclosing 'all' transaction is rolled back.
   */
-  bool is_real_trans= all || thd->transaction.all.ha_list == 0;
+  bool is_real_trans= ((all || thd->transaction.all.ha_list == 0) &&
+                       !(thd->variables.option_bits & OPTION_GTID_BEGIN));
   Ha_trx_info *ha_info= trans->ha_list;
   bool need_prepare_ordered, need_commit_ordered;
   my_xid xid;
@@ -1471,7 +1472,8 @@ int ha_commit_one_phase(THD *thd, bool all)
     ha_commit_one_phase() can be called with an empty
     transaction.all.ha_list, see why in trans_register_ha()).
   */
-  bool is_real_trans=all || thd->transaction.all.ha_list == 0;
+  bool is_real_trans= ((all || thd->transaction.all.ha_list == 0) &&
+                       !(thd->variables.option_bits & OPTION_GTID_BEGIN));
   int res;
   DBUG_ENTER("ha_commit_one_phase");
   if (is_real_trans && (res= thd->wait_for_prior_commit()))

sql/log.cc

@@ -1679,6 +1679,7 @@ static int binlog_close_connection(handlerton *hton, THD *thd)
     contain updates to non-transactional tables. Or it can be a flush of
     a statement cache.
  */
+
 static int
 binlog_flush_cache(THD *thd, binlog_cache_mngr *cache_mngr,
                    Log_event *end_ev, bool all, bool using_stmt,
@@ -1686,6 +1687,7 @@ binlog_flush_cache(THD *thd, binlog_cache_mngr *cache_mngr,
 {
   int error= 0;
   DBUG_ENTER("binlog_flush_cache");
+  DBUG_PRINT("enter", ("end_ev: %p", end_ev));
 
   if ((using_stmt && !cache_mngr->stmt_cache.empty()) ||
       (using_trx && !cache_mngr->trx_cache.empty()))
@@ -1744,9 +1746,10 @@ static inline int
 binlog_commit_flush_stmt_cache(THD *thd, bool all,
                                binlog_cache_mngr *cache_mngr)
 {
+  DBUG_ENTER("binlog_commit_flush_stmt_cache");
   Query_log_event end_evt(thd, STRING_WITH_LEN("COMMIT"),
                           FALSE, TRUE, TRUE, 0);
-  return (binlog_flush_cache(thd, cache_mngr, &end_evt, all, TRUE, FALSE));
+  DBUG_RETURN(binlog_flush_cache(thd, cache_mngr, &end_evt, all, TRUE, FALSE));
 }
 
 /**
@@ -1761,9 +1764,10 @@ binlog_commit_flush_stmt_cache(THD *thd, bool all,
 static inline int
 binlog_commit_flush_trx_cache(THD *thd, bool all, binlog_cache_mngr *cache_mngr)
 {
+  DBUG_ENTER("binlog_commit_flush_trx_cache");
   Query_log_event end_evt(thd, STRING_WITH_LEN("COMMIT"),
                           TRUE, TRUE, TRUE, 0);
-  return (binlog_flush_cache(thd, cache_mngr, &end_evt, all, FALSE, TRUE));
+  DBUG_RETURN(binlog_flush_cache(thd, cache_mngr, &end_evt, all, FALSE, TRUE));
 }
 
 /**
@@ -5270,6 +5274,10 @@ int THD::binlog_write_table_map(TABLE *table, bool is_transactional,
                        (long) table, table->s->table_name.str,
                        table->s->table_map_id));
 
+  /* Ensure that all events in a GTID group are in the same cache */
+  if (variables.option_bits & OPTION_GTID_BEGIN)
+    is_transactional= 1;
+  
   /* Pre-conditions */
   DBUG_ASSERT(is_current_stmt_binlog_format_row() && mysql_bin_log.is_open());
   DBUG_ASSERT(table->s->table_map_id != ULONG_MAX);
@@ -5287,7 +5295,7 @@ int THD::binlog_write_table_map(TABLE *table, bool is_transactional,
     cache_mngr->get_binlog_cache_log(use_trans_cache(this, is_transactional));
   if (with_annotate && *with_annotate)
   {
-    Annotate_rows_log_event anno(current_thd, is_transactional, false);
+    Annotate_rows_log_event anno(table->in_use, is_transactional, false);
     /* Annotate event should be written not more than once */
     *with_annotate= 0;
     if ((error= anno.write(file)))
@@ -5450,6 +5458,7 @@ MYSQL_BIN_LOG::flush_and_set_pending_rows_event(THD *thd,
 
 
 /* Generate a new global transaction ID, and write it to the binlog */
+
 bool
 MYSQL_BIN_LOG::write_gtid_event(THD *thd, bool standalone,
                                 bool is_transactional, uint64 commit_id)
@@ -5459,6 +5468,16 @@ MYSQL_BIN_LOG::write_gtid_event(THD *thd, bool standalone,
   uint32 server_id= thd->variables.server_id;
   uint64 seq_no= thd->variables.gtid_seq_no;
   int err;
+  DBUG_ENTER("write_gtid_event");
+  DBUG_PRINT("enter", ("standalone: %d", standalone));
+  
+  if (thd->variables.option_bits & OPTION_GTID_BEGIN)
+  {
+    DBUG_PRINT("error", ("OPTION_GTID_BEGIN is set. "
+                         "Master and slave will have different GTID values"));
+    /* Reset the flag, as we will write out a GTID anyway */
+    thd->variables.option_bits&= ~OPTION_GTID_BEGIN;
+  }
 
   /*
     Reset the session variable gtid_seq_no, to reduce the risk of accidentally
@@ -5483,7 +5502,7 @@ MYSQL_BIN_LOG::write_gtid_event(THD *thd, bool standalone,
     seq_no= gtid.seq_no;
   }
   if (err)
-    return true;
+    DBUG_RETURN(true);
 
   Gtid_log_event gtid_event(thd, seq_no, domain_id, standalone,
                             LOG_EVENT_SUPPRESS_USE_F, is_transactional,
@@ -5491,10 +5510,10 @@ MYSQL_BIN_LOG::write_gtid_event(THD *thd, bool standalone,
 
   /* Write the event to the binary log. */
   if (gtid_event.write(&mysql_bin_log.log_file))
-    return true;
+    DBUG_RETURN(true);
   status_var_add(thd->status_var.binlog_bytes_written, gtid_event.data_written);
 
-  return false;
+  DBUG_RETURN(false);
 }
 
 
@@ -5676,14 +5695,22 @@ bool MYSQL_BIN_LOG::write(Log_event *event_info, my_bool *with_annotate)
 {
   THD *thd= event_info->thd;
   bool error= 1;
-  DBUG_ENTER("MYSQL_BIN_LOG::write(Log_event *)");
   binlog_cache_data *cache_data= 0;
   bool is_trans_cache= FALSE;
   bool using_trans= event_info->use_trans_cache();
   bool direct= event_info->use_direct_logging();
   ulong prev_binlog_id;
+  DBUG_ENTER("MYSQL_BIN_LOG::write(Log_event *)");
   LINT_INIT(prev_binlog_id);
 
+  if (thd->variables.option_bits & OPTION_GTID_BEGIN)
+  {
+    DBUG_PRINT("info", ("OPTION_GTID_BEGIN was set"));
+    /* Wait for commit from binary log before we commit */
+    direct= 0;
+    using_trans= 1;
+  }
+
   if (thd->binlog_evt_union.do_union)
   {
     /*
@@ -5731,6 +5758,7 @@ bool MYSQL_BIN_LOG::write(Log_event *event_info, my_bool *with_annotate)
 
     if (direct)
     {
+      DBUG_PRINT("info", ("direct is set"));
       file= &log_file;
       my_org_b_tell= my_b_tell(file);
       mysql_mutex_lock(&LOCK_log);
@@ -7249,16 +7277,17 @@ MYSQL_BIN_LOG::write_transaction_or_stmt(group_commit_entry *entry,
                                          uint64 commit_id)
 {
   binlog_cache_mngr *mngr= entry->cache_mngr;
+  DBUG_ENTER("MYSQL_BIN_LOG::write_transaction_or_stmt");
 
   if (write_gtid_event(entry->thd, false, entry->using_trx_cache, commit_id))
-    return ER_ERROR_ON_WRITE;
+    DBUG_RETURN(ER_ERROR_ON_WRITE);
 
   if (entry->using_stmt_cache && !mngr->stmt_cache.empty() &&
       write_cache(entry->thd, mngr->get_binlog_cache_log(FALSE)))
   {
     entry->error_cache= &mngr->stmt_cache.cache_log;
     entry->commit_errno= errno;
-    return ER_ERROR_ON_WRITE;
+    DBUG_RETURN(ER_ERROR_ON_WRITE);
   }
 
   if (entry->using_trx_cache && !mngr->trx_cache.empty())
@@ -7279,7 +7308,7 @@ MYSQL_BIN_LOG::write_transaction_or_stmt(group_commit_entry *entry,
     {
       entry->error_cache= &mngr->trx_cache.cache_log;
       entry->commit_errno= errno;
-      return ER_ERROR_ON_WRITE;
+      DBUG_RETURN(ER_ERROR_ON_WRITE);
     }
   }
 
@@ -7287,7 +7316,7 @@ MYSQL_BIN_LOG::write_transaction_or_stmt(group_commit_entry *entry,
   {
     entry->error_cache= NULL;
     entry->commit_errno= errno;
-    return ER_ERROR_ON_WRITE;
+    DBUG_RETURN(ER_ERROR_ON_WRITE);
   }
   status_var_add(entry->thd->status_var.binlog_bytes_written,
                  entry->end_event->data_written);
@@ -7298,7 +7327,7 @@ MYSQL_BIN_LOG::write_transaction_or_stmt(group_commit_entry *entry,
     {
       entry->error_cache= NULL;
       entry->commit_errno= errno;
-      return ER_ERROR_ON_WRITE;
+      DBUG_RETURN(ER_ERROR_ON_WRITE);
     }
   }
 
@@ -7306,16 +7335,16 @@ MYSQL_BIN_LOG::write_transaction_or_stmt(group_commit_entry *entry,
   {
     entry->error_cache= &mngr->stmt_cache.cache_log;
     entry->commit_errno= errno;
-    return ER_ERROR_ON_READ;
+    DBUG_RETURN(ER_ERROR_ON_WRITE);
   }
   if (mngr->get_binlog_cache_log(TRUE)->error)  // Error on read
   {
     entry->error_cache= &mngr->trx_cache.cache_log;
     entry->commit_errno= errno;
-    return ER_ERROR_ON_READ;
+    DBUG_RETURN(ER_ERROR_ON_WRITE);
   }
 
-  return 0;
+  DBUG_RETURN(0);
 }
 
 

sql/log_event.cc

@@ -3984,6 +3984,8 @@ bool test_if_equal_repl_errors(int expected_error, int actual_error)
   case ER_AUTOINC_READ_FAILED:
     return (actual_error == ER_AUTOINC_READ_FAILED ||
             actual_error == HA_ERR_AUTOINC_ERANGE);
+  case ER_UNKNOWN_TABLE:
+    return actual_error == ER_IT_IS_A_VIEW;
   default:
     break;
   }
@@ -4018,6 +4020,7 @@ int Query_log_event::do_apply_event(rpl_group_info *rgi,
   rpl_gtid gtid;
   Relay_log_info const *rli= rgi->rli;
   Rpl_filter *rpl_filter= rli->mi->rpl_filter;
+  bool current_stmt_is_commit;
   DBUG_ENTER("Query_log_event::do_apply_event");
 
   /*
@@ -4044,7 +4047,9 @@ int Query_log_event::do_apply_event(rpl_group_info *rgi,
   DBUG_PRINT("info", ("log_pos: %lu", (ulong) log_pos));
 
   clear_all_errors(thd, const_cast<Relay_log_info*>(rli));
-  if (strcmp("COMMIT", query) == 0 && rgi->tables_to_lock)
+  current_stmt_is_commit= is_commit();
+
+  if (current_stmt_is_commit && rgi->tables_to_lock)
   {
     /*
       Cleaning-up the last statement context:
@@ -4093,9 +4098,11 @@ int Query_log_event::do_apply_event(rpl_group_info *rgi,
     thd->variables.pseudo_thread_id= thread_id;		// for temp tables
     DBUG_PRINT("query",("%s", thd->query()));
 
-    if (ignored_error_code((expected_error= error_code)) ||
+    if (!(expected_error= error_code) ||
+        ignored_error_code(expected_error) ||
         !unexpected_error_code(expected_error))
     {
+      thd->slave_expected_error= expected_error;
       if (flags2_inited)
         /*
           all bits of thd->variables.option_bits which are 1 in OPTIONS_WRITTEN_TO_BIN_LOG
@@ -4197,12 +4204,13 @@ int Query_log_event::do_apply_event(rpl_group_info *rgi,
         Record any GTID in the same transaction, so slave state is
         transactionally consistent.
       */
-      if (strcmp("COMMIT", query) == 0 && (sub_id= rgi->gtid_sub_id))
+      if (current_stmt_is_commit && (sub_id= rgi->gtid_sub_id))
       {
         /* Clear the GTID from the RLI so we don't accidentally reuse it. */
         rgi->gtid_sub_id= 0;
 
         gtid= rgi->current_gtid;
+        thd->variables.option_bits&= ~OPTION_GTID_BEGIN;
         if (rpl_global_gtid_slave_state.record_gtid(thd, &gtid, sub_id, true, false))
         {
           rli->report(ERROR_LEVEL, ER_CANNOT_UPDATE_GTID_STATE,
@@ -4232,6 +4240,7 @@ int Query_log_event::do_apply_event(rpl_group_info *rgi,
            concurrency_error_code(expected_error)))
       {
         thd->variables.option_bits|= OPTION_MASTER_SQL_ERROR;
+        thd->variables.option_bits&= ~OPTION_GTID_BEGIN;
       }
       /* Execute the query (note that we bypass dispatch_command()) */
       Parser_state parser_state;
@@ -4395,8 +4404,7 @@ Default database: '%s'. Query: '%s'",
       to shutdown trying to finish incomplete events group.
     */
     DBUG_EXECUTE_IF("stop_slave_middle_group",
-                    if (strcmp("COMMIT", query) != 0 &&
+                    if (!current_stmt_is_commit && is_begin() == 0)
-                        strcmp("BEGIN", query) != 0)
                     {
                       if (thd->transaction.all.modified_non_trans_table)
                         const_cast<Relay_log_info*>(rli)->abort_slave= 1;
@@ -4457,7 +4465,7 @@ Query_log_event::do_shall_skip(rpl_group_info *rgi)
 {
   Relay_log_info *rli= rgi->rli;
   DBUG_ENTER("Query_log_event::do_shall_skip");
-  DBUG_PRINT("debug", ("query: %s; q_len: %d", query, q_len));
+  DBUG_PRINT("debug", ("query: '%s'  q_len: %d", query, q_len));
   DBUG_ASSERT(query && q_len > 0);
   DBUG_ASSERT(thd == rgi->thd);
 
@@ -4473,13 +4481,13 @@ Query_log_event::do_shall_skip(rpl_group_info *rgi)
   {
     if (is_begin())
     {
-      thd->variables.option_bits|= OPTION_BEGIN;
+      thd->variables.option_bits|= OPTION_BEGIN | OPTION_GTID_BEGIN;
       DBUG_RETURN(Log_event::continue_group(rgi));
     }
 
     if (is_commit() || is_rollback())
     {
-      thd->variables.option_bits&= ~OPTION_BEGIN;
+      thd->variables.option_bits&= ~(OPTION_BEGIN | OPTION_GTID_BEGIN);
       DBUG_RETURN(Log_event::EVENT_SKIP_COUNT);
     }
   }
@@ -5906,6 +5914,7 @@ error:
   thd->reset_query();
   thd->get_stmt_da()->set_overwrite_status(true);
   thd->is_error() ? trans_rollback_stmt(thd) : trans_commit_stmt(thd);
+  thd->variables.option_bits&= ~(OPTION_BEGIN | OPTION_GTID_BEGIN);
   thd->get_stmt_da()->set_overwrite_status(false);
   close_thread_tables(thd);
   /*
@@ -6408,7 +6417,6 @@ Gtid_log_event::make_compatible_event(String *packet, bool *need_dummy_event,
   {
     if (*need_dummy_event)
       return Query_log_event::dummy_event(packet, ev_offset, checksum_alg);
-    else
     return 0;
   }
 
@@ -6456,10 +6464,16 @@ Gtid_log_event::do_apply_event(rpl_group_info *rgi)
                                                  this->server_id, this->seq_no))
       return 1;
   }
+
+  DBUG_ASSERT((thd->variables.option_bits & OPTION_GTID_BEGIN) == 0);
   if (flags2 & FL_STANDALONE)
     return 0;
 
   /* Execute this like a BEGIN query event. */
+  thd->variables.option_bits|= OPTION_BEGIN | OPTION_GTID_BEGIN;
+  DBUG_PRINT("info", ("Set OPTION_GTID_BEGIN"));
+  trans_begin(thd, 0);
+
   thd->set_query_and_id(gtid_begin_string, sizeof(gtid_begin_string)-1,
                         &my_charset_bin, next_query_id());
   Parser_state parser_state;
@@ -7252,6 +7266,7 @@ int Xid_log_event::do_apply_event(rpl_group_info *rgi)
   /* For a slave Xid_log_event is COMMIT */
   general_log_print(thd, COM_QUERY,
                     "COMMIT /* implicit, from Xid_log_event */");
+  thd->variables.option_bits&= ~OPTION_GTID_BEGIN;
   res= trans_commit(thd); /* Automatically rolls back on error. */
   thd->mdl_context.release_transactional_locks();
 
@@ -7273,7 +7288,7 @@ Xid_log_event::do_shall_skip(rpl_group_info *rgi)
   if (rgi->rli->slave_skip_counter > 0)
   {
     DBUG_ASSERT(!rgi->rli->get_flag(Relay_log_info::IN_TRANSACTION));
-    thd->variables.option_bits&= ~OPTION_BEGIN;
+    thd->variables.option_bits&= ~(OPTION_BEGIN | OPTION_GTID_BEGIN);
     DBUG_RETURN(Log_event::EVENT_SKIP_COUNT);
   }
   DBUG_RETURN(Log_event::do_shall_skip(rgi));

sql/mysqld.cc

@@ -479,6 +479,7 @@ ulong open_files_limit, max_binlog_size;
 ulong slave_trans_retries;
 uint  slave_net_timeout;
 ulong slave_exec_mode_options;
+ulong slave_ddl_exec_mode_options= SLAVE_EXEC_MODE_IDEMPOTENT;
 ulonglong slave_type_conversions_options;
 ulong thread_cache_size=0;
 ulonglong binlog_cache_size=0;

sql/mysqld.h

@@ -96,7 +96,7 @@ extern uint connection_count;
 extern my_bool opt_safe_user_create;
 extern my_bool opt_safe_show_db, opt_local_infile, opt_myisam_use_mmap;
 extern my_bool opt_slave_compressed_protocol, use_temp_pool;
-extern ulong slave_exec_mode_options;
+extern ulong slave_exec_mode_options, slave_ddl_exec_mode_options;
 extern ulong slave_retried_transactions;
 extern ulonglong slave_type_conversions_options;
 extern my_bool read_only, opt_readonly;

sql/rpl_gtid.cc

@@ -352,7 +352,8 @@ rpl_slave_state::record_gtid(THD *thd, const rpl_gtid *gtid, uint64 sub_id,
   {
     DBUG_PRINT("info", ("resetting OPTION_BEGIN"));
     thd->variables.option_bits&=
-      ~(ulonglong)(OPTION_NOT_AUTOCOMMIT|OPTION_BEGIN|OPTION_BIN_LOG);
+      ~(ulonglong)(OPTION_NOT_AUTOCOMMIT |OPTION_BEGIN |OPTION_BIN_LOG |
+                   OPTION_GTID_BEGIN);
   }
   else
     thd->variables.option_bits&= ~(ulonglong)OPTION_BIN_LOG;

sql/rpl_rli.cc

@@ -1583,6 +1583,7 @@ void rpl_group_info::cleanup_context(THD *thd, bool error)
   if (error)
   {
     trans_rollback_stmt(thd); // if a "statement transaction"
+    /* trans_rollback() also resets OPTION_GTID_BEGIN */
     trans_rollback(thd);      // if a "real transaction"
   }
   m_table_map.clear_tables();

sql/slave.cc

@@ -3125,9 +3125,10 @@ int apply_event_and_update_pos(Log_event* ev, THD* thd,
   DBUG_PRINT("exec_event",("%s(type_code: %d; server_id: %d)",
                            ev->get_type_str(), ev->get_type_code(),
                            ev->server_id));
-  DBUG_PRINT("info", ("thd->options: %s%s; rgi->last_event_start_time: %lu",
+  DBUG_PRINT("info", ("thd->options: '%s%s%s'  rgi->last_event_start_time: %lu",
                       FLAGSTR(thd->variables.option_bits, OPTION_NOT_AUTOCOMMIT),
                       FLAGSTR(thd->variables.option_bits, OPTION_BEGIN),
+                      FLAGSTR(thd->variables.option_bits, OPTION_GTID_BEGIN),
                       (ulong) rgi->last_event_start_time));
 
   /*

sql/sql_class.cc

@@ -5389,6 +5389,10 @@ THD::binlog_prepare_pending_rows_event(TABLE* table, uint32 serv_id,
   /* Fetch the type code for the RowsEventT template parameter */
   int const general_type_code= RowsEventT::TYPE_CODE;
 
+  /* Ensure that all events in a GTID group are in the same cache */
+  if (variables.option_bits & OPTION_GTID_BEGIN)
+    is_transactional= 1;
+
   /*
     There is no good place to set up the transactional data, so we
     have to do it here.
@@ -5584,6 +5588,10 @@ int THD::binlog_write_row(TABLE* table, bool is_trans,
 
   size_t const len= pack_row(table, cols, row_data, record);
 
+  /* Ensure that all events in a GTID group are in the same cache */
+  if (variables.option_bits & OPTION_GTID_BEGIN)
+    is_trans= 1;
+
   Rows_log_event* const ev=
     binlog_prepare_pending_rows_event(table, variables.server_id, cols, colcnt,
                                       len, is_trans,
@@ -5617,6 +5625,10 @@ int THD::binlog_update_row(TABLE* table, bool is_trans,
   size_t const after_size= pack_row(table, cols, after_row,
                                        after_record);
 
+  /* Ensure that all events in a GTID group are in the same cache */
+  if (variables.option_bits & OPTION_GTID_BEGIN)
+    is_trans= 1;
+
   /*
     Don't print debug messages when running valgrind since they can
     trigger false warnings.
@@ -5659,6 +5671,10 @@ int THD::binlog_delete_row(TABLE* table, bool is_trans,
 
   size_t const len= pack_row(table, cols, row_data, record);
 
+  /* Ensure that all events in a GTID group are in the same cache */
+  if (variables.option_bits & OPTION_GTID_BEGIN)
+    is_trans= 1;
+
   Rows_log_event* const ev=
     binlog_prepare_pending_rows_event(table, variables.server_id, cols, colcnt,
 				      len, is_trans,
@@ -5679,6 +5695,10 @@ int THD::binlog_remove_pending_rows_event(bool clear_maps,
   if (!mysql_bin_log.is_open())
     DBUG_RETURN(0);
 
+  /* Ensure that all events in a GTID group are in the same cache */
+  if (variables.option_bits & OPTION_GTID_BEGIN)
+    is_transactional= 1;
+
   mysql_bin_log.remove_pending_rows_event(this, is_transactional);
 
   if (clear_maps)
@@ -5698,6 +5718,10 @@ int THD::binlog_flush_pending_rows_event(bool stmt_end, bool is_transactional)
   if (!mysql_bin_log.is_open())
     DBUG_RETURN(0);
 
+  /* Ensure that all events in a GTID group are in the same cache */
+  if (variables.option_bits & OPTION_GTID_BEGIN)
+    is_transactional= 1;
+
   /*
     Mark the event as the last event of a statement if the stmt_end
     flag is set.
@@ -5945,6 +5969,14 @@ int THD::binlog_query(THD::enum_binlog_query_type qtype, char const *query_arg,
                        show_query_type(qtype), (int) query_len, query_arg));
   DBUG_ASSERT(query_arg && mysql_bin_log.is_open());
 
+  /* If this is withing a BEGIN ... COMMIT group, don't log it */
+  if (variables.option_bits & OPTION_GTID_BEGIN)
+  {
+    direct= 0;
+    is_trans= 1;
+  }
+  DBUG_PRINT("info", ("is_trans: %d  direct: %d", is_trans, direct));
+
   if (get_binlog_local_stmt_filter() == BINLOG_FILTER_SET)
   {
     /*

sql/sql_class.h

@@ -1951,7 +1951,10 @@ public:
   uint in_sub_stmt;
   /* True when opt_userstat_running is set at start of query */
   bool userstat_running;
-  /* True if we want to log all errors */
+  /*
+    True if we have to log all errors. Are set by some engines to temporary
+    force errors to the error log.
+  */
   bool log_all_errors;
 
   /* Do not set socket timeouts for wait_timeout (used with threadpool) */
@@ -2542,12 +2545,12 @@ public:
   */
   LEX_STRING connection_name;
   char       default_master_connection_buff[MAX_CONNECTION_NAME+1];
+  uint8      password; /* 0, 1 or 2 */
+  uint8      failed_com_change_user;
   bool       slave_thread, one_shot_set;
   bool       extra_port;                        /* If extra connection */
 
   bool	     no_errors;
-  uint8      password;
-  uint8      failed_com_change_user;
 
   /**
     Set to TRUE if execution of the current compound statement
@@ -2580,13 +2583,6 @@ public:
   /* for IS NULL => = last_insert_id() fix in remove_eq_conds() */
   bool       substitute_null_with_insert_id;
   bool	     in_lock_tables;
-  /**
-    True if a slave error. Causes the slave to stop. Not the same
-    as the statement execution error (is_error()), since
-    a statement may be expected to return an error, e.g. because
-    it returned an error on master, and this is OK on the slave.
-  */
-  bool       is_slave_error;
   bool       bootstrap, cleanup_done;
 
   /**  is set if some thread specific value(s) used in a statement. */
@@ -2603,6 +2599,20 @@ public:
   /* set during loop of derived table processing */
   bool       derived_tables_processing;
   bool       tablespace_op;	/* This is TRUE in DISCARD/IMPORT TABLESPACE */
+  /* True if we have to log the current statement */
+  bool	     log_current_statement;
+  /**
+    True if a slave error. Causes the slave to stop. Not the same
+    as the statement execution error (is_error()), since
+    a statement may be expected to return an error, e.g. because
+    it returned an error on master, and this is OK on the slave.
+  */
+  bool       is_slave_error;
+  /*
+    In case of a slave, set to the error code the master got when executing
+    the query. 0 if no error on the master.
+  */
+  int	     slave_expected_error;
 
   sp_rcontext *spcont;		// SP runtime context
   sp_cache   *sp_proc_cache;

sql/sql_insert.cc

@@ -3794,7 +3794,8 @@ void select_insert::abort_result_set() {
     */
     changed= (info.copied || info.deleted || info.updated);
     transactional_table= table->file->has_transactions();
-    if (thd->transaction.stmt.modified_non_trans_table)
+    if (thd->transaction.stmt.modified_non_trans_table ||
+        thd->log_current_statement)
     {
         if (!can_rollback_data())
           thd->transaction.all.modified_non_trans_table= TRUE;
@@ -4200,7 +4201,9 @@ select_create::binlog_show_create_table(TABLE **tables, uint count)
   query.length(0);      // Have to zero it since constructor doesn't
 
   result= store_create_info(thd, &tmp_table_list, &query, create_info,
-                            /* show_database */ TRUE);
+                            /* show_database */ TRUE,
+                            test(create_info->options &
+                                 HA_LEX_CREATE_REPLACE));
   DBUG_ASSERT(result == 0); /* store_create_info() always return 0 */
 
   if (mysql_bin_log.is_open())
@@ -4270,6 +4273,7 @@ bool select_create::send_eof()
   if (!table->s->tmp_table)
   {
     trans_commit_stmt(thd);
+    if (!(thd->variables.option_bits & OPTION_GTID_BEGIN))
       trans_commit_implicit(thd);
   }
 
@@ -4307,6 +4311,7 @@ bool select_create::send_eof()
 
 void select_create::abort_result_set()
 {
+  ulonglong save_option_bits;
   DBUG_ENTER("select_create::abort_result_set");
 
   /* Avoid double calls, could happen in case of out of memory on cleanup */
@@ -4328,11 +4333,18 @@ void select_create::abort_result_set()
     We also roll back the statement regardless of whether the creation
     of the table succeeded or not, since we need to reset the binary
     log state.
+    
+    However if there was an orignal table that was deleted, as part of
+    create or replace table, then we must log the statement.
   */
-  tmp_disable_binlog(thd);
+
+  save_option_bits= thd->variables.option_bits;
+  if (!(thd->log_current_statement))
+    thd->variables.option_bits&= ~OPTION_BIN_LOG;
   select_insert::abort_result_set();
   thd->transaction.stmt.modified_non_trans_table= FALSE;
-  reenable_binlog(thd);
+  thd->variables.option_bits= save_option_bits;
+
   /* possible error of writing binary log is ignored deliberately */
   (void) thd->binlog_flush_pending_rows_event(TRUE, TRUE);
 

sql/sql_parse.cc

@@ -195,6 +195,8 @@ static bool some_non_temp_table_to_be_updated(THD *thd, TABLE_LIST *tables)
   @param thd    Thread handle.
   @param mask   Bitmask used for the SQL command match.
 
+  @return 0     No implicit commit
+  @return 1     Do a commit
 */
 static bool stmt_causes_implicit_commit(THD *thd, uint mask)
 {
@@ -207,12 +209,22 @@ static bool stmt_causes_implicit_commit(THD *thd, uint mask)
 
   switch (lex->sql_command) {
   case SQLCOM_DROP_TABLE:
-    skip= lex->drop_temporary;
+    skip= (lex->drop_temporary ||
+           (thd->variables.option_bits & OPTION_GTID_BEGIN));
     break;
   case SQLCOM_ALTER_TABLE:
+    /* If ALTER TABLE of non-temporary table, do implicit commit */
+    skip= (lex->create_info.tmp_table());
+    break;
   case SQLCOM_CREATE_TABLE:
-    /* If CREATE TABLE of non-temporary table, do implicit commit */
+    /*
-    skip= lex->create_info.tmp_table();
+      If CREATE TABLE of non-temporary table and the table is not part
+      if a BEGIN GTID ... COMMIT group, do a implicit commit.
+      This ensures that CREATE ... SELECT will in the same GTID group on the
+      master and slave.
+    */
+    skip= (lex->create_info.tmp_table() ||
+           (thd->variables.option_bits & OPTION_GTID_BEGIN));
     break;
   case SQLCOM_SET_OPTION:
     skip= lex->autocommit ? FALSE : TRUE;
@@ -2439,12 +2451,15 @@ mysql_execute_command(THD *thd)
     DBUG_ASSERT(! thd->in_sub_stmt);
     /* Statement transaction still should not be started. */
     DBUG_ASSERT(thd->transaction.stmt.is_empty());
+    if (!(thd->variables.option_bits & OPTION_GTID_BEGIN))
+    {
       /* Commit the normal transaction if one is active. */
       if (trans_commit_implicit(thd))
         goto error;
       /* Release metadata locks acquired in this transaction. */
       thd->mdl_context.release_transactional_locks();
     }
+  }
   
 #ifndef DBUG_OFF
   if (lex->sql_command != SQLCOM_SET_OPTION)
@@ -2869,6 +2884,17 @@ case SQLCOM_PREPARE:
     */
     lex->query_tables->open_strategy= TABLE_LIST::OPEN_STUB;
 
+    /*
+      If we are a slave, we should add OR REPLACE if we don't have
+      IF EXISTS. This will help a slave to recover from
+      CREATE TABLE OR EXISTS failures by dropping the table and
+      retrying the create.
+    */
+    if (thd->slave_thread &&
+        slave_ddl_exec_mode_options == SLAVE_EXEC_MODE_IDEMPOTENT &&
+        !(lex->create_info.options & HA_LEX_CREATE_IF_NOT_EXISTS))
+      create_info.options|= HA_LEX_CREATE_REPLACE;
+
 #ifdef WITH_PARTITION_STORAGE_ENGINE
     {
       partition_info *part_info= thd->lex->part_info;
@@ -3659,6 +3685,16 @@ end_with_restore_list:
       /* So that DROP TEMPORARY TABLE gets to binlog at commit/rollback */
       thd->variables.option_bits|= OPTION_KEEP_LOG;
     }
+    /*
+      If we are a slave, we should add IF EXISTS if the query executed
+      on the master without an error. This will help a slave to
+      recover from multi-table DROP TABLE that was aborted in the
+      middle.
+    */
+    if (thd->slave_thread && !thd->slave_expected_error &&
+        slave_ddl_exec_mode_options == SLAVE_EXEC_MODE_IDEMPOTENT)
+      lex->check_exists= 1;
+
     /* DDL and binlog write order are protected by metadata locks. */
     res= mysql_rm_table(thd, first_table, lex->check_exists,
 			lex->drop_temporary);
@@ -4407,6 +4443,7 @@ end_with_restore_list:
     bool tx_release= (lex->tx_release == TVL_YES ||
                       (thd->variables.completion_type == 2 &&
                        lex->tx_release != TVL_NO));
+
     if (trans_rollback(thd))
       goto error;
     thd->mdl_context.release_transactional_locks();
@@ -5158,6 +5195,8 @@ finish:
   {
     /* No transaction control allowed in sub-statements. */
     DBUG_ASSERT(! thd->in_sub_stmt);
+    if (!(thd->variables.option_bits & OPTION_GTID_BEGIN))
+    {
       /* If commit fails, we should be able to reset the OK status. */
       thd->get_stmt_da()->set_overwrite_status(true);
       /* Commit the normal transaction if one is active. */
@@ -5165,6 +5204,7 @@ finish:
       thd->get_stmt_da()->set_overwrite_status(false);
       thd->mdl_context.release_transactional_locks();
     }
+  }
   else if (! thd->in_sub_stmt && ! thd->in_multi_stmt_transaction_mode())
   {
     /*
@@ -6106,6 +6146,8 @@ void THD::reset_for_next_command()
   thd->query_start_used= 0;
   thd->query_start_sec_part_used= 0;
   thd->is_fatal_error= thd->time_zone_used= 0;
+  thd->log_current_statement= 0;
+
   /*
     Clear the status flag that are expected to be cleared at the
     beginning of each SQL statement.

sql/sql_priv.h

@@ -110,6 +110,8 @@
 
 /* The following is used to detect a conflict with DISTINCT */
 #define SELECT_ALL              (1ULL << 24)    // SELECT, user, parser
+#define OPTION_GTID_BEGIN       (1ULL << 25)    // GTID BEGIN found in log
+
 /** The following can be set when importing tables in a 'wrong order'
    to suppress foreign key checks */
 #define OPTION_NO_FOREIGN_KEY_CHECKS    (1ULL << 26) // THD, user, binlog

sql/sql_show.cc

@@ -1041,7 +1041,7 @@ mysqld_show_create(THD *thd, TABLE_LIST *table_list)
   if ((table_list->view ?
        view_store_create_info(thd, table_list, &buffer) :
        store_create_info(thd, table_list, &buffer, NULL,
-                         FALSE /* show_database */)))
+                         FALSE /* show_database */, FALSE)))
     goto exit;
 
   if (table_list->view)
@@ -1526,6 +1526,8 @@ static void append_create_options(THD *thd, String *packet,
                       to tailor the format of the statement.  Can be
                       NULL, in which case only SQL_MODE is considered
                       when building the statement.
+    show_database     Add database name to table name
+    create_or_replace Use CREATE OR REPLACE syntax
 
   NOTE
     Currently always return 0, but might return error code in the
@@ -1536,7 +1538,8 @@ static void append_create_options(THD *thd, String *packet,
  */
 
 int store_create_info(THD *thd, TABLE_LIST *table_list, String *packet,
-                      HA_CREATE_INFO *create_info_arg, bool show_database)
+                      HA_CREATE_INFO *create_info_arg, bool show_database,
+                      bool create_or_replace)
 {
   List<Item> field_list;
   char tmp[MAX_FIELD_WIDTH], *for_str, buff[128], def_value_buf[MAX_FIELD_WIDTH];
@@ -1569,10 +1572,12 @@ int store_create_info(THD *thd, TABLE_LIST *table_list, String *packet,
 
   restore_record(table, s->default_values); // Get empty record
 
+  packet->append(STRING_WITH_LEN("CREATE "));
+  if (create_or_replace)
+    packet->append(STRING_WITH_LEN("OR REPLACE "));
   if (share->tmp_table)
-    packet->append(STRING_WITH_LEN("CREATE TEMPORARY TABLE "));
+    packet->append(STRING_WITH_LEN("TEMPORARY "));
-  else
+  packet->append(STRING_WITH_LEN("TABLE "));
-    packet->append(STRING_WITH_LEN("CREATE TABLE "));
   if (create_info_arg &&
       (create_info_arg->options & HA_LEX_CREATE_IF_NOT_EXISTS))
     packet->append(STRING_WITH_LEN("IF NOT EXISTS "));

sql/sql_show.h

@@ -75,7 +75,8 @@ typedef struct system_status_var STATUS_VAR;
 #define IS_FILES_EXTRA               37
 
 int store_create_info(THD *thd, TABLE_LIST *table_list, String *packet,
-                      HA_CREATE_INFO  *create_info_arg, bool show_database);
+                      HA_CREATE_INFO  *create_info_arg, bool show_database,
+                      bool create_or_replace);
 int view_store_create_info(THD *thd, TABLE_LIST *table, String *buff);
 
 int copy_event_to_schema_table(THD *thd, TABLE *sch_table, TABLE *event_table);

sql/sql_table.cc

@@ -4620,6 +4620,14 @@ int create_table_impl(THD *thd,
         /* Remove normal table without logging. Keep tables locked */
         if (mysql_rm_table_no_locks(thd, &table_list, 0, 0, 0, 1, 1))
           goto err;
+
+        /*
+          We have to log this query, even if it failed later to ensure the
+          drop is done.
+        */
+        thd->variables.option_bits|= OPTION_KEEP_LOG;
+        thd->log_current_statement= 1;
+
         /*
           The test of query_tables is to ensure we have any tables in the
           select part
@@ -4627,11 +4635,6 @@ int create_table_impl(THD *thd,
         if (thd->lex->query_tables &&
             restart_trans_for_tables(thd, thd->lex->query_tables->next_global))
           goto err;
-        /*
-          We have to log this query, even if it failed later to ensure the
-          drop is done.
-        */
-        thd->variables.option_bits|= OPTION_KEEP_LOG;
       }
       else if (create_info->options & HA_LEX_CREATE_IF_NOT_EXISTS)
         goto warn;
@@ -4832,6 +4835,7 @@ bool mysql_create_table(THD *thd, TABLE_LIST *create_table,
   const char *db= create_table->db;
   const char *table_name= create_table->table_name;
   bool is_trans= FALSE;
+  bool result= 0;
   int create_table_mode;
   TABLE_LIST *pos_in_locked_tables= 0;
   DBUG_ENTER("mysql_create_table");
@@ -4859,7 +4863,10 @@ bool mysql_create_table(THD *thd, TABLE_LIST *create_table,
   promote_first_timestamp_column(&alter_info->create_list);
   if (mysql_create_table_no_lock(thd, db, table_name, create_info, alter_info,
                                  &is_trans, create_table_mode) > 0)
-    DBUG_RETURN(1);
+  {
+    result= 1;
+    goto err;
+  }
 
   /*
     Check if we are doing CREATE OR REPLACE TABLE under LOCK TABLES
@@ -4877,12 +4884,15 @@ bool mysql_create_table(THD *thd, TABLE_LIST *create_table,
       thd->locked_tables_list.unlink_all_closed_tables(thd, NULL, 0);
   }
 
+err:
   /* In RBR we don't need to log CREATE TEMPORARY TABLE */
   if (thd->is_current_stmt_binlog_format_row() && create_info->tmp_table())
-    DBUG_RETURN(0);
+    DBUG_RETURN(result);
-
+  /* Write log if no error or if we already deleted a table */
-  bool result;
+  if (!result || thd->log_current_statement)
-  result= write_bin_log(thd, TRUE, thd->query(), thd->query_length(), is_trans);
+    if (write_bin_log(thd, result ? FALSE : TRUE, thd->query(),
+                      thd->query_length(), is_trans))
+      result= 1;
   DBUG_RETURN(result);
 }
 
@@ -5079,6 +5089,7 @@ bool mysql_create_like_table(THD* thd, TABLE_LIST* table,
   Alter_table_ctx local_alter_ctx; // Not used
   bool res= TRUE;
   bool is_trans= FALSE;
+  bool do_logging= FALSE;
   uint not_used;
   DBUG_ENTER("mysql_create_like_table");
 
@@ -5155,9 +5166,12 @@ bool mysql_create_like_table(THD* thd, TABLE_LIST* table,
   if ((local_create_info.table= thd->lex->query_tables->table))
     pos_in_locked_tables= local_create_info.table->pos_in_locked_tables;    
 
-  if ((res= (mysql_create_table_no_lock(thd, table->db, table->table_name,
+  res= (mysql_create_table_no_lock(thd, table->db, table->table_name,
                                    &local_create_info, &local_alter_info,
-                                        &is_trans, C_ORDINARY_CREATE) > 0)))
+                                   &is_trans, C_ORDINARY_CREATE) > 0);
+  /* Remember to log if we deleted something */
+  do_logging= thd->log_current_statement;
+  if (res)
     goto err;
 
   /*
@@ -5252,7 +5266,10 @@ bool mysql_create_like_table(THD* thd, TABLE_LIST* table,
             /* Restore */
             table->open_strategy= save_open_strategy;
             if (open_res)
+            {
+              res= 1;
               goto err;
+            }
             new_table= TRUE;
           }
         }
@@ -5264,11 +5281,17 @@ 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 */);
+                              create_info, FALSE /* show_database */,
+                              test(create_info->options &
+                                   HA_LEX_CREATE_REPLACE));
 
           DBUG_ASSERT(result == 0); // store_create_info() always return 0
+          do_logging= FALSE;
           if (write_bin_log(thd, TRUE, query.ptr(), query.length()))
+          {
+            res= 1;
             goto err;
+          }
 
           if (new_table)
           {
@@ -5283,17 +5306,20 @@ bool mysql_create_like_table(THD* thd, TABLE_LIST* table,
         }
       }
       else                                      // Case 1
-        if (write_bin_log(thd, TRUE, thd->query(), thd->query_length()))
+        do_logging= TRUE;
-          goto err;
     }
     /*
       Case 3 and 4 does nothing under RBR
     */
   }
-  else if (write_bin_log(thd, TRUE, thd->query(), thd->query_length(), is_trans))
+  else
-    goto err;
+    do_logging= TRUE;
 
 err:
+  if (do_logging &&
+      write_bin_log(thd, res ? FALSE : TRUE, thd->query(),
+                    thd->query_length(), is_trans))
+    res= 1;
   DBUG_RETURN(res);
 }
 

sql/sys_vars.cc

@@ -2601,11 +2601,23 @@ static Sys_var_enum Slave_exec_mode(
        "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",
        GLOBAL_VAR(slave_exec_mode_options), CMD_LINE(REQUIRED_ARG),
        slave_exec_mode_names, DEFAULT(SLAVE_EXEC_MODE_STRICT));
 
+static Sys_var_enum Slave_ddl_exec_mode(
+       "slave_ddl_exec_mode",
+       "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. ",
+       GLOBAL_VAR(slave_ddl_exec_mode_options), CMD_LINE(REQUIRED_ARG),
+       slave_exec_mode_names, DEFAULT(SLAVE_EXEC_MODE_IDEMPOTENT));
+
 static const char *slave_type_conversions_name[]= {"ALL_LOSSY", "ALL_NON_LOSSY", 0};
 static Sys_var_set Slave_type_conversions(
        "slave_type_conversions",
@@ -3174,10 +3186,10 @@ static bool fix_autocommit(sys_var *self, THD *thd, enum_var_type type)
     return false;
   }
 
-  if (thd->variables.option_bits & OPTION_AUTOCOMMIT &&
+  if (test_all_bits(thd->variables.option_bits,
-      thd->variables.option_bits & OPTION_NOT_AUTOCOMMIT)
+                    (OPTION_AUTOCOMMIT | OPTION_NOT_AUTOCOMMIT)))
-  { // activating autocommit
+  {
-
+    // activating autocommit
     if (trans_commit_stmt(thd) || trans_commit(thd))
     {
       thd->variables.option_bits&= ~OPTION_AUTOCOMMIT;
@@ -3194,16 +3206,17 @@ static bool fix_autocommit(sys_var *self, THD *thd, enum_var_type type)
       transaction implicitly at the end (@sa stmt_causes_implicitcommit()).
     */
     thd->variables.option_bits&=
-                 ~(OPTION_BEGIN | OPTION_KEEP_LOG | OPTION_NOT_AUTOCOMMIT);
+                 ~(OPTION_BEGIN | OPTION_KEEP_LOG | OPTION_NOT_AUTOCOMMIT |
+                   OPTION_GTID_BEGIN);
     thd->transaction.all.modified_non_trans_table= false;
     thd->server_status|= SERVER_STATUS_AUTOCOMMIT;
     return false;
   }
 
-  if (!(thd->variables.option_bits & OPTION_AUTOCOMMIT) &&
+  if ((thd->variables.option_bits &
-      !(thd->variables.option_bits & OPTION_NOT_AUTOCOMMIT))
+       (OPTION_AUTOCOMMIT |OPTION_NOT_AUTOCOMMIT)) == 0)
-  { // disabling autocommit
+  {
-
+    // disabling autocommit
     thd->transaction.all.modified_non_trans_table= false;
     thd->server_status&= ~SERVER_STATUS_AUTOCOMMIT;
     thd->variables.option_bits|= OPTION_NOT_AUTOCOMMIT;
@@ -3212,6 +3225,7 @@ static bool fix_autocommit(sys_var *self, THD *thd, enum_var_type type)
 
   return false; // autocommit value wasn't changed
 }
+
 static Sys_var_bit Sys_autocommit(
        "autocommit", "autocommit",
        SESSION_VAR(option_bits), NO_CMD_LINE, OPTION_AUTOCOMMIT, DEFAULT(TRUE),

sql/transaction.cc

@@ -251,6 +251,10 @@ bool trans_commit_implicit(THD *thd)
   if (trans_check(thd))
     DBUG_RETURN(TRUE);
 
+  if (thd->variables.option_bits & OPTION_GTID_BEGIN)
+    DBUG_PRINT("error", ("OPTION_GTID_BEGIN is set. "
+                         "Master and slave will have different GTID values"));
+
   if (thd->in_multi_stmt_transaction_mode() ||
       (thd->variables.option_bits & OPTION_TABLE_LOCK))
   {
@@ -302,6 +306,8 @@ bool trans_rollback(THD *thd)
   res= ha_rollback_trans(thd, TRUE);
   (void) RUN_HOOK(transaction, after_rollback, (thd, FALSE));
   thd->variables.option_bits&= ~(OPTION_BEGIN | OPTION_KEEP_LOG);
+  /* Reset the binlog transaction marker */
+  thd->variables.option_bits&= ~OPTION_GTID_BEGIN;
   thd->transaction.all.modified_non_trans_table= FALSE;
   thd->lex->start_transaction_opt= 0;