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

18 files changed, 295 insertions, 95 deletions

diff --git a/sql/handler.cc b/sql/handler.cc
index 6af1682e080..dee76d21303 100644
--- a/sql/handler.cc
+++ b/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;
@@ -1266,7 +1267,7 @@ int ha_commit_trans(THD *thd, bool all)
                  ER(ER_WARNING_NOT_COMPLETE_ROLLBACK)););
 
   DBUG_PRINT("info",
-             ("all: %d thd->in_sub_stmt: %d ha_info: %p is_real_trans: %d",
+             ("all: %d  thd->in_sub_stmt: %d  ha_info: %p  is_real_trans: %d",
               all, thd->in_sub_stmt, ha_info, is_real_trans));
   /*
     We must not commit the normal transaction if a statement
@@ -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()))
diff --git a/sql/log.cc b/sql/log.cc
index 9ea1514ef77..f2f83dbe546 100644
--- a/sql/log.cc
+++ b/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);
 }
 
 
diff --git a/sql/log_event.cc b/sql/log_event.cc
index 7f0933792be..4cebcb97461 100644
--- a/sql/log_event.cc
+++ b/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)) ||
-	!unexpected_error_code(expected_error))
+    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 &&
-                        strcmp("BEGIN", query) != 0)
+                    if (!current_stmt_is_commit && is_begin() == 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,8 +6417,7 @@ 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;
+    return 0;
   }
 
   *need_dummy_event= true;
@@ -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));
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index a52251462b1..484ff18eb32 100644
--- a/sql/mysqld.cc
+++ b/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;
diff --git a/sql/mysqld.h b/sql/mysqld.h
index 12b5d857e6e..119e30e6b77 100644
--- a/sql/mysqld.h
+++ b/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;
diff --git a/sql/rpl_gtid.cc b/sql/rpl_gtid.cc
index 3f79a0cb528..a36a15b3c27 100644
--- a/sql/rpl_gtid.cc
+++ b/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;
diff --git a/sql/rpl_rli.cc b/sql/rpl_rli.cc
index 9036f810020..48494401fda 100644
--- a/sql/rpl_rli.cc
+++ b/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();
diff --git a/sql/slave.cc b/sql/slave.cc
index 6886a6345ab..0c321f29c77 100644
--- a/sql/slave.cc
+++ b/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));
 
   /*
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index d1c3bc25210..9e3ce76af98 100644
--- a/sql/sql_class.cc
+++ b/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)
   {
     /*
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 14f58b30f3c..f00b005b470 100644
--- a/sql/sql_class.h
+++ b/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;
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index 719e0bae4ff..cdfa9f516f9 100644
--- a/sql/sql_insert.cc
+++ b/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,7 +4273,8 @@ bool select_create::send_eof()
   if (!table->s->tmp_table)
   {
     trans_commit_stmt(thd);
-    trans_commit_implicit(thd);
+    if (!(thd->variables.option_bits & OPTION_GTID_BEGIN))
+      trans_commit_implicit(thd);
   }
 
   table->file->extra(HA_EXTRA_NO_IGNORE_DUP_KEY);
@@ -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);
 
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 0d0f1a2634f..8ce2225e731 100644
--- a/sql/sql_parse.cc
+++ b/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,11 +2451,14 @@ 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());
-    /* 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();
+    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
@@ -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,12 +5195,15 @@ finish:
   {
     /* No transaction control allowed in sub-statements. */
     DBUG_ASSERT(! thd->in_sub_stmt);
-    /* 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. */
-    trans_commit_implicit(thd);
-    thd->get_stmt_da()->set_overwrite_status(false);
-    thd->mdl_context.release_transactional_locks();
+    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. */
+      trans_commit_implicit(thd);
+      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.
diff --git a/sql/sql_priv.h b/sql/sql_priv.h
index 383888bac30..5e3b80ab7a9 100644
--- a/sql/sql_priv.h
+++ b/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
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index bcbe544b99c..53c98cf0d38 100644
--- a/sql/sql_show.cc
+++ b/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 "));
-  else
-    packet->append(STRING_WITH_LEN("CREATE TABLE "));
+    packet->append(STRING_WITH_LEN("TEMPORARY "));
+  packet->append(STRING_WITH_LEN("TABLE "));
   if (create_info_arg &&
       (create_info_arg->options & HA_LEX_CREATE_IF_NOT_EXISTS))
     packet->append(STRING_WITH_LEN("IF NOT EXISTS "));
diff --git a/sql/sql_show.h b/sql/sql_show.h
index 10276e8b65e..0416f2fdaba 100644
--- a/sql/sql_show.h
+++ b/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);
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 817823063eb..67d6795ecce 100644
--- a/sql/sql_table.cc
+++ b/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);
-
-  bool result;
-  result= write_bin_log(thd, TRUE, thd->query(), thd->query_length(), is_trans);
+    DBUG_RETURN(result);
+  /* Write log if no error or if we already deleted a table */
+  if (!result || thd->log_current_statement)
+    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,
-                                        &local_create_info, &local_alter_info,
-                                        &is_trans, C_ORDINARY_CREATE) > 0)))
+  res= (mysql_create_table_no_lock(thd, table->db, table->table_name,
+                                   &local_create_info, &local_alter_info,
+                                   &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()))
-          goto err;
+        do_logging= TRUE;
     }
     /*
       Case 3 and 4 does nothing under RBR
     */
   }
-  else if (write_bin_log(thd, TRUE, thd->query(), thd->query_length(), is_trans))
-    goto err;
+  else
+    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);
 }
 
diff --git a/sql/sys_vars.cc b/sql/sys_vars.cc
index 864808cff32..5800fab8d4a 100644
--- a/sql/sys_vars.cc
+++ b/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 &&
-      thd->variables.option_bits & OPTION_NOT_AUTOCOMMIT)
-  { // activating autocommit
-
+  if (test_all_bits(thd->variables.option_bits,
+                    (OPTION_AUTOCOMMIT | OPTION_NOT_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) &&
-      !(thd->variables.option_bits & OPTION_NOT_AUTOCOMMIT))
-  { // disabling autocommit
-
+  if ((thd->variables.option_bits &
+       (OPTION_AUTOCOMMIT |OPTION_NOT_AUTOCOMMIT)) == 0)
+  {
+    // 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),
diff --git a/sql/transaction.cc b/sql/transaction.cc
index 256351ee373..213b6c1a4d8 100644
--- a/sql/transaction.cc
+++ b/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;