merge with 10.0-monty

49 files changed, 1080 insertions, 522 deletions

diff --git a/sql/ha_ndbcluster_binlog.cc b/sql/ha_ndbcluster_binlog.cc
index fc53183ca7a..531211eb175 100644
--- a/sql/ha_ndbcluster_binlog.cc
+++ b/sql/ha_ndbcluster_binlog.cc
@@ -446,7 +446,7 @@ int ndbcluster_binlog_init_share(NDB_SHARE *share, TABLE *_table)
       alloc_root(mem_root, no_nodes * sizeof(MY_BITMAP));
     for (i= 0; i < no_nodes; i++)
     {
-      bitmap_init(&share->subscriber_bitmap[i],
+      my_bitmap_init(&share->subscriber_bitmap[i],
                   (Uint32*)alloc_root(mem_root, max_ndb_nodes/8),
                   max_ndb_nodes, FALSE);
       bitmap_clear_all(&share->subscriber_bitmap[i]);
@@ -1119,7 +1119,7 @@ ndbcluster_update_slock(THD *thd,
 
   MY_BITMAP slock;
   uint32 bitbuf[SCHEMA_SLOCK_SIZE/4];
-  bitmap_init(&slock, bitbuf, sizeof(bitbuf)*8, false);
+  my_bitmap_init(&slock, bitbuf, sizeof(bitbuf)*8, false);
 
   if (ndbtab == 0)
   {
@@ -1370,7 +1370,7 @@ int ndbcluster_log_schema_op(THD *thd, NDB_SHARE *share,
   {
     int i, updated= 0;
     int no_storage_nodes= g_ndb_cluster_connection->no_db_nodes();
-    bitmap_init(&schema_subscribers, bitbuf, sizeof(bitbuf)*8, FALSE);
+    my_bitmap_init(&schema_subscribers, bitbuf, sizeof(bitbuf)*8, FALSE);
     bitmap_set_all(&schema_subscribers);
 
     /* begin protect ndb_schema_share */
@@ -1908,7 +1908,7 @@ ndb_binlog_thread_handle_schema_event(THD *thd, Ndb *ndb,
       Cluster_schema *schema= (Cluster_schema *)
         sql_alloc(sizeof(Cluster_schema));
       MY_BITMAP slock;
-      bitmap_init(&slock, schema->slock, 8*SCHEMA_SLOCK_SIZE, FALSE);
+      my_bitmap_init(&slock, schema->slock, 8*SCHEMA_SLOCK_SIZE, FALSE);
       uint node_id= g_ndb_cluster_connection->node_id();
       {
         ndbcluster_get_schema(tmp_share, schema);
@@ -3353,7 +3353,7 @@ ndb_binlog_thread_handle_data_event(Ndb *ndb, NdbEventOperation *pOp,
   MY_BITMAP b;
   /* Potential buffer for the bitmap */
   uint32 bitbuf[128 / (sizeof(uint32) * 8)];
-  bitmap_init(&b, n_fields <= sizeof(bitbuf) * 8 ? bitbuf : NULL, 
+  my_bitmap_init(&b, n_fields <= sizeof(bitbuf) * 8 ? bitbuf : NULL, 
               n_fields, FALSE);
   bitmap_set_all(&b);
 
@@ -3573,7 +3573,7 @@ static NDB_SCHEMA_OBJECT *ndb_get_schema_object(const char *key,
       break;
     }
     mysql_mutex_init(key_ndb_schema_object_mutex, &ndb_schema_object->mutex, MY_MUTEX_INIT_FAST);
-    bitmap_init(&ndb_schema_object->slock_bitmap, ndb_schema_object->slock,
+    my_bitmap_init(&ndb_schema_object->slock_bitmap, ndb_schema_object->slock,
                 sizeof(ndb_schema_object->slock)*8, FALSE);
     bitmap_clear_all(&ndb_schema_object->slock_bitmap);
     break;
diff --git a/sql/ha_ndbcluster_cond.h b/sql/ha_ndbcluster_cond.h
index 27675588ed7..9ea0438a348 100644
--- a/sql/ha_ndbcluster_cond.h
+++ b/sql/ha_ndbcluster_cond.h
@@ -350,18 +350,18 @@ class Ndb_cond_traverse_context : public Sql_alloc
     skip(0), collation(NULL), rewrite_stack(NULL)
   {
     // Allocate type checking bitmaps   
-    bitmap_init(&expect_mask, 0, 512, FALSE);
-    bitmap_init(&expect_field_type_mask, 0, 512, FALSE);
-    bitmap_init(&expect_field_result_mask, 0, 512, FALSE);
+    my_bitmap_init(&expect_mask, 0, 512, FALSE);
+    my_bitmap_init(&expect_field_type_mask, 0, 512, FALSE);
+    my_bitmap_init(&expect_field_result_mask, 0, 512, FALSE);
 
     if (stack)
       cond_ptr= stack->ndb_cond;
   };
   ~Ndb_cond_traverse_context()
   {
-    bitmap_free(&expect_mask);
-    bitmap_free(&expect_field_type_mask);
-    bitmap_free(&expect_field_result_mask);
+    my_bitmap_free(&expect_mask);
+    my_bitmap_free(&expect_field_type_mask);
+    my_bitmap_free(&expect_field_result_mask);
     if (rewrite_stack) delete rewrite_stack;
   }
   void expect(Item::Type type)
diff --git a/sql/ha_partition.cc b/sql/ha_partition.cc
index 672348a5c45..6fe3c826913 100644
--- a/sql/ha_partition.cc
+++ b/sql/ha_partition.cc
@@ -3318,10 +3318,10 @@ err:
 void ha_partition::free_partition_bitmaps()
 {
   /* Initialize the bitmap we use to minimize ha_start_bulk_insert calls */
-  bitmap_free(&m_bulk_insert_started);
-  bitmap_free(&m_locked_partitions);
-  bitmap_free(&m_partitions_to_reset);
-  bitmap_free(&m_key_not_found_partitions);
+  my_bitmap_free(&m_bulk_insert_started);
+  my_bitmap_free(&m_locked_partitions);
+  my_bitmap_free(&m_partitions_to_reset);
+  my_bitmap_free(&m_key_not_found_partitions);
 }
 
 
@@ -3333,14 +3333,14 @@ bool ha_partition::init_partition_bitmaps()
 {
   DBUG_ENTER("ha_partition::init_partition_bitmaps");
   /* Initialize the bitmap we use to minimize ha_start_bulk_insert calls */
-  if (bitmap_init(&m_bulk_insert_started, NULL, m_tot_parts + 1, FALSE))
+  if (my_bitmap_init(&m_bulk_insert_started, NULL, m_tot_parts + 1, FALSE))
     DBUG_RETURN(true);
   bitmap_clear_all(&m_bulk_insert_started);
 
   /* Initialize the bitmap we use to keep track of locked partitions */
-  if (bitmap_init(&m_locked_partitions, NULL, m_tot_parts, FALSE))
+  if (my_bitmap_init(&m_locked_partitions, NULL, m_tot_parts, FALSE))
   {
-    bitmap_free(&m_bulk_insert_started);
+    my_bitmap_free(&m_bulk_insert_started);
     DBUG_RETURN(true);
   }
   bitmap_clear_all(&m_locked_partitions);
@@ -3349,10 +3349,10 @@ bool ha_partition::init_partition_bitmaps()
     Initialize the bitmap we use to keep track of partitions which may have
     something to reset in ha_reset().
   */
-  if (bitmap_init(&m_partitions_to_reset, NULL, m_tot_parts, FALSE))
+  if (my_bitmap_init(&m_partitions_to_reset, NULL, m_tot_parts, FALSE))
   {
-    bitmap_free(&m_bulk_insert_started);
-    bitmap_free(&m_locked_partitions);
+    my_bitmap_free(&m_bulk_insert_started);
+    my_bitmap_free(&m_locked_partitions);
     DBUG_RETURN(true);
   }
   bitmap_clear_all(&m_partitions_to_reset);
@@ -3361,11 +3361,11 @@ bool ha_partition::init_partition_bitmaps()
     Initialize the bitmap we use to keep track of partitions which returned
     HA_ERR_KEY_NOT_FOUND from index_read_map.
   */
-  if (bitmap_init(&m_key_not_found_partitions, NULL, m_tot_parts, FALSE))
+  if (my_bitmap_init(&m_key_not_found_partitions, NULL, m_tot_parts, FALSE))
   {
-    bitmap_free(&m_bulk_insert_started);
-    bitmap_free(&m_locked_partitions);
-    bitmap_free(&m_partitions_to_reset);
+    my_bitmap_free(&m_bulk_insert_started);
+    my_bitmap_free(&m_locked_partitions);
+    my_bitmap_free(&m_partitions_to_reset);
     DBUG_RETURN(true);
   }
   bitmap_clear_all(&m_key_not_found_partitions);
diff --git a/sql/handler.cc b/sql/handler.cc
index 521723e4049..5b1e0ed58d1 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
@@ -1476,7 +1477,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)
@@ -5731,7 +5733,7 @@ static int binlog_log_row(TABLE* table,
       the first row handled in this statement. In that case, we need
       to write table maps for all locked tables to the binary log.
     */
-    if (likely(!(error= bitmap_init(&cols,
+    if (likely(!(error= my_bitmap_init(&cols,
                                     use_bitbuf ? bitbuf : NULL,
                                     (n_fields + 7) & ~7UL,
                                     FALSE))))
@@ -5753,7 +5755,7 @@ static int binlog_log_row(TABLE* table,
                            before_record, after_record);
       }
       if (!use_bitbuf)
-        bitmap_free(&cols);
+        my_bitmap_free(&cols);
     }
   }
   return error ? HA_ERR_RBR_LOGGING_FAILED : 0;
diff --git a/sql/handler.h b/sql/handler.h
index 0202fedb1eb..ffb12d11648 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -32,6 +32,7 @@
 #include "sql_cache.h"
 #include "structs.h"                            /* SHOW_COMP_OPTION */
 #include "sql_array.h"          /* Dynamic_array<> */
+#include "mdl.h"
 
 #include <my_compare.h>
 #include <ft_global.h>
@@ -387,6 +388,7 @@ enum enum_alter_inplace_result {
 #define HA_LEX_CREATE_IF_NOT_EXISTS 2
 #define HA_LEX_CREATE_TABLE_LIKE 4
 #define HA_CREATE_TMP_ALTER     8
+#define HA_LEX_CREATE_REPLACE   16
 #define HA_MAX_REC_LENGTH	65535
 
 /* Table caching type */
@@ -1582,9 +1584,15 @@ struct HA_CREATE_INFO
   ulong avg_row_length;
   ulong used_fields;
   ulong key_block_size;
-  uint stats_sample_pages;		/* number of pages to sample during
-					stats estimation, if used, otherwise 0. */
-  enum_stats_auto_recalc stats_auto_recalc;
+  /*
+    number of pages to sample during
+    stats estimation, if used, otherwise 0.
+  */
+  uint stats_sample_pages;
+  uint null_bits;                       /* NULL bits at start of record */
+  uint options;				/* OR of HA_CREATE_ options */
+  uint merge_insert_method;
+  uint extra_size;                      /* length of extra data segment */
   SQL_I_List<TABLE_LIST> merge_list;
   handlerton *db_type;
   /**
@@ -1597,21 +1605,23 @@ struct HA_CREATE_INFO
     If nothing speficied inherits the value of the original table (if present).
   */
   enum row_type row_type;
-  uint null_bits;                       /* NULL bits at start of record */
-  uint options;				/* OR of HA_CREATE_ options */
-  uint merge_insert_method;
-  uint extra_size;                      /* length of extra data segment */
   enum ha_choice transactional;
-  bool varchar;                         ///< 1 if table has a VARCHAR
   enum ha_storage_media storage_media;  ///< DEFAULT, DISK or MEMORY
   enum ha_choice page_checksum;         ///< If we have page_checksums
   engine_option_value *option_list;     ///< list of table create options
+  enum_stats_auto_recalc stats_auto_recalc;
+  bool varchar;                         ///< 1 if table has a VARCHAR
 
   /* the following three are only for ALTER TABLE, check_if_incompatible_data() */
   ha_table_option_struct *option_struct;           ///< structure with parsed table options
   ha_field_option_struct **fields_option_struct;   ///< array of field option structures
   ha_index_option_struct **indexes_option_struct;  ///< array of index option structures
 
+  /* The following is used to remember the old state for CREATE OR REPLACE */
+  TABLE *table;
+  TABLE_LIST *pos_in_locked_tables;
+  MDL_ticket *mdl_ticket;
+
   bool tmp_table() { return options & HA_LEX_CREATE_TMP_TABLE; }
 };
 
diff --git a/sql/hostname.cc b/sql/hostname.cc
index 1200dd2c185..21e652a346f 100644
--- a/sql/hostname.cc
+++ b/sql/hostname.cc
@@ -149,7 +149,7 @@ bool hostname_cache_init()
   Host_entry tmp;
   uint key_offset= (uint) ((char*) (&tmp.ip_key) - (char*) &tmp);
 
-  if (!(hostname_cache= new hash_filo(HOST_CACHE_SIZE,
+  if (!(hostname_cache= new hash_filo(host_cache_size,
                                       key_offset, HOST_ENTRY_KEY_SIZE,
                                       NULL, (my_hash_free_key) free,
                                       &my_charset_bin)))
diff --git a/sql/item_subselect.cc b/sql/item_subselect.cc
index 81c4f0f51e5..88cae108039 100644
--- a/sql/item_subselect.cc
+++ b/sql/item_subselect.cc
@@ -4685,13 +4685,13 @@ ulonglong subselect_hash_sj_engine::rowid_merge_buff_size(
 */
 
 static my_bool
-bitmap_init_memroot(MY_BITMAP *map, uint n_bits, MEM_ROOT *mem_root)
+my_bitmap_init_memroot(MY_BITMAP *map, uint n_bits, MEM_ROOT *mem_root)
 {
   my_bitmap_map *bitmap_buf;
 
   if (!(bitmap_buf= (my_bitmap_map*) alloc_root(mem_root,
                                                 bitmap_buffer_size(n_bits))) ||
-      bitmap_init(map, bitmap_buf, n_bits, FALSE))
+      my_bitmap_init(map, bitmap_buf, n_bits, FALSE))
     return TRUE;
   bitmap_clear_all(map);
   return FALSE;
@@ -4729,9 +4729,9 @@ bool subselect_hash_sj_engine::init(List<Item> *tmp_columns, uint subquery_id)
 
   DBUG_ENTER("subselect_hash_sj_engine::init");
 
-  if (bitmap_init_memroot(&non_null_key_parts, tmp_columns->elements,
+  if (my_bitmap_init_memroot(&non_null_key_parts, tmp_columns->elements,
                             thd->mem_root) ||
-      bitmap_init_memroot(&partial_match_key_parts, tmp_columns->elements,
+      my_bitmap_init_memroot(&partial_match_key_parts, tmp_columns->elements,
                             thd->mem_root))
     DBUG_RETURN(TRUE);
 
@@ -5453,7 +5453,7 @@ Ordered_key::Ordered_key(uint keyid_arg, TABLE *tbl_arg, Item *search_key_arg,
 Ordered_key::~Ordered_key()
 {
   my_free(key_buff);
-  bitmap_free(&null_key);
+  my_bitmap_free(&null_key);
 }
 
 
@@ -5563,7 +5563,7 @@ bool Ordered_key::alloc_keys_buffers()
     lookup offset.
   */
   /* Notice that max_null_row is max array index, we need count, so +1. */
-  if (bitmap_init(&null_key, NULL, (uint)(max_null_row + 1), FALSE))
+  if (my_bitmap_init(&null_key, NULL, (uint)(max_null_row + 1), FALSE))
     return TRUE;
 
   cur_key_idx= HA_POS_ERROR;
@@ -6002,8 +6002,8 @@ subselect_rowid_merge_engine::init(MY_BITMAP *non_null_key_parts,
   */
   if (!has_covering_null_columns)
   {
-    if (bitmap_init_memroot(&matching_keys, merge_keys_count, thd->mem_root) ||
-        bitmap_init_memroot(&matching_outer_cols, merge_keys_count, thd->mem_root))
+    if (my_bitmap_init_memroot(&matching_keys, merge_keys_count, thd->mem_root) ||
+        my_bitmap_init_memroot(&matching_outer_cols, merge_keys_count, thd->mem_root))
       return TRUE;
 
     /*
diff --git a/sql/log.cc b/sql/log.cc
index f531d301b63..f2fc7eb1554 100644
--- a/sql/log.cc
+++ b/sql/log.cc
@@ -525,35 +525,57 @@ bool LOGGER::is_log_table_enabled(uint log_table_type)
 }
 
 
-/* Check if a given table is opened log table */
-int check_if_log_table(size_t db_len, const char *db, size_t table_name_len,
-                       const char *table_name, bool check_if_opened)
+/**
+   Check if a given table is opened log table
+
+   @param table             Table to check
+   @param check_if_opened   Only fail if it's a log table in use
+   @param error_msg	    String to put in error message if not ok.
+                            No error message if 0
+   @return 0 ok
+   @return # Type of log file
+ */
+
+int check_if_log_table(const TABLE_LIST *table,
+                       bool check_if_opened,
+                       const char *error_msg)
 {
-  if (db_len == 5 &&
+  int result= 0;
+  if (table->db_length == 5 &&
       !(lower_case_table_names ?
-        my_strcasecmp(system_charset_info, db, "mysql") :
-        strcmp(db, "mysql")))
+        my_strcasecmp(system_charset_info, table->db, "mysql") :
+        strcmp(table->db, "mysql")))
   {
-    if (table_name_len == 11 && !(lower_case_table_names ?
-                                  my_strcasecmp(system_charset_info,
-                                                table_name, "general_log") :
-                                  strcmp(table_name, "general_log")))
+    const char *table_name= table->table_name;
+
+    if (table->table_name_length == 11 &&
+        !(lower_case_table_names ?
+          my_strcasecmp(system_charset_info,
+                        table_name, "general_log") :
+          strcmp(table_name, "general_log")))
     {
-      if (!check_if_opened || logger.is_log_table_enabled(QUERY_LOG_GENERAL))
-        return QUERY_LOG_GENERAL;
-      return 0;
+      result= QUERY_LOG_GENERAL;
+      goto end;
     }
 
-    if (table_name_len == 8 && !(lower_case_table_names ?
+    if (table->table_name_length == 8 && !(lower_case_table_names ?
       my_strcasecmp(system_charset_info, table_name, "slow_log") :
       strcmp(table_name, "slow_log")))
     {
-      if (!check_if_opened || logger.is_log_table_enabled(QUERY_LOG_SLOW))
-        return QUERY_LOG_SLOW;
-      return 0;
+      result= QUERY_LOG_SLOW;
+      goto end;
     }
   }
   return 0;
+
+end:
+  if (!check_if_opened || logger.is_log_table_enabled(result))
+  {
+    if (error_msg)
+      my_error(ER_BAD_LOG_STATEMENT, MYF(0), error_msg);
+    return result;
+  }
+  return 0;
 }
 
 
@@ -1657,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,
@@ -1664,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()))
@@ -1722,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));
 }
 
 /**
@@ -1739,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));
 }
 
 /**
@@ -5248,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);
@@ -5265,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)))
@@ -5428,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)
@@ -5437,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
@@ -5461,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,
@@ -5469,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);
 }
 
 
@@ -5654,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)
   {
     /*
@@ -5709,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);
@@ -7299,16 +7349,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())
@@ -7329,7 +7380,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);
     }
   }
 
@@ -7337,7 +7388,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);
@@ -7348,7 +7399,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);
     }
   }
 
@@ -7356,16 +7407,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.h b/sql/log.h
index 45381152d97..d6ae7bbb1bb 100644
--- a/sql/log.h
+++ b/sql/log.h
@@ -833,8 +833,8 @@ public:
 };
 
 
-int check_if_log_table(size_t db_len, const char *db, size_t table_name_len,
-                       const char *table_name, bool check_if_opened);
+int check_if_log_table(const TABLE_LIST *table, bool check_if_opened,
+                       const char *errmsg);
 
 class Log_to_csv_event_handler: public Log_event_handler
 {
diff --git a/sql/log_event.cc b/sql/log_event.cc
index 0af7a2ed344..d2d82da4ede 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;
@@ -7250,6 +7264,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();
 
@@ -7271,7 +7286,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));
@@ -9112,8 +9127,8 @@ Rows_log_event::Rows_log_event(THD *thd_arg, TABLE *tbl_arg, ulong tid,
       set_flags(NO_FOREIGN_KEY_CHECKS_F);
   if (thd_arg->variables.option_bits & OPTION_RELAXED_UNIQUE_CHECKS)
       set_flags(RELAXED_UNIQUE_CHECKS_F);
-  /* if bitmap_init fails, caught in is_valid() */
-  if (likely(!bitmap_init(&m_cols,
+  /* if my_bitmap_init fails, caught in is_valid() */
+  if (likely(!my_bitmap_init(&m_cols,
                           m_width <= sizeof(m_bitbuf)*8 ? m_bitbuf : NULL,
                           m_width,
                           false)))
@@ -9127,7 +9142,7 @@ Rows_log_event::Rows_log_event(THD *thd_arg, TABLE *tbl_arg, ulong tid,
   }
   else
   {
-    // Needed because bitmap_init() does not set it to null on failure
+    // Needed because my_bitmap_init() does not set it to null on failure
     m_cols.bitmap= 0;
   }
 }
@@ -9228,8 +9243,8 @@ Rows_log_event::Rows_log_event(const char *buf, uint event_len,
   DBUG_PRINT("debug", ("Reading from %p", ptr_after_width));
   m_width = net_field_length(&ptr_after_width);
   DBUG_PRINT("debug", ("m_width=%lu", m_width));
-  /* if bitmap_init fails, catched in is_valid() */
-  if (likely(!bitmap_init(&m_cols,
+  /* if my_bitmap_init fails, catched in is_valid() */
+  if (likely(!my_bitmap_init(&m_cols,
                           m_width <= sizeof(m_bitbuf)*8 ? m_bitbuf : NULL,
                           m_width,
                           false)))
@@ -9242,7 +9257,7 @@ Rows_log_event::Rows_log_event(const char *buf, uint event_len,
   }
   else
   {
-    // Needed because bitmap_init() does not set it to null on failure
+    // Needed because my_bitmap_init() does not set it to null on failure
     m_cols.bitmap= NULL;
     DBUG_VOID_RETURN;
   }
@@ -9254,8 +9269,8 @@ Rows_log_event::Rows_log_event(const char *buf, uint event_len,
   {
     DBUG_PRINT("debug", ("Reading from %p", ptr_after_width));
 
-    /* if bitmap_init fails, caught in is_valid() */
-    if (likely(!bitmap_init(&m_cols_ai,
+    /* if my_bitmap_init fails, caught in is_valid() */
+    if (likely(!my_bitmap_init(&m_cols_ai,
                             m_width <= sizeof(m_bitbuf_ai)*8 ? m_bitbuf_ai : NULL,
                             m_width,
                             false)))
@@ -9269,7 +9284,7 @@ Rows_log_event::Rows_log_event(const char *buf, uint event_len,
     }
     else
     {
-      // Needed because bitmap_init() does not set it to null on failure
+      // Needed because my_bitmap_init() does not set it to null on failure
       m_cols_ai.bitmap= 0;
       DBUG_VOID_RETURN;
     }
@@ -9300,8 +9315,8 @@ Rows_log_event::Rows_log_event(const char *buf, uint event_len,
 Rows_log_event::~Rows_log_event()
 {
   if (m_cols.bitmap == m_bitbuf) // no my_malloc happened
-    m_cols.bitmap= 0; // so no my_free in bitmap_free
-  bitmap_free(&m_cols); // To pair with bitmap_init().
+    m_cols.bitmap= 0; // so no my_free in my_bitmap_free
+  my_bitmap_free(&m_cols); // To pair with my_bitmap_init().
   my_free(m_rows_buf);
   my_free(m_extra_row_data);
 }
@@ -11960,8 +11975,8 @@ Update_rows_log_event::Update_rows_log_event(THD *thd_arg, TABLE *tbl_arg,
 
 void Update_rows_log_event::init(MY_BITMAP const *cols)
 {
-  /* if bitmap_init fails, caught in is_valid() */
-  if (likely(!bitmap_init(&m_cols_ai,
+  /* if my_bitmap_init fails, caught in is_valid() */
+  if (likely(!my_bitmap_init(&m_cols_ai,
                           m_width <= sizeof(m_bitbuf_ai)*8 ? m_bitbuf_ai : NULL,
                           m_width,
                           false)))
@@ -11980,8 +11995,8 @@ void Update_rows_log_event::init(MY_BITMAP const *cols)
 Update_rows_log_event::~Update_rows_log_event()
 {
   if (m_cols_ai.bitmap == m_bitbuf_ai) // no my_malloc happened
-    m_cols_ai.bitmap= 0; // so no my_free in bitmap_free
-  bitmap_free(&m_cols_ai); // To pair with bitmap_init().
+    m_cols_ai.bitmap= 0; // so no my_free in my_bitmap_free
+  my_bitmap_free(&m_cols_ai); // To pair with my_bitmap_init().
 }
 
 
diff --git a/sql/log_event_old.cc b/sql/log_event_old.cc
index 7b89d5bdf08..0cb78686243 100644
--- a/sql/log_event_old.cc
+++ b/sql/log_event_old.cc
@@ -1244,8 +1244,8 @@ Old_rows_log_event::Old_rows_log_event(THD *thd_arg, TABLE *tbl_arg, ulong tid,
       set_flags(NO_FOREIGN_KEY_CHECKS_F);
   if (thd_arg->variables.option_bits & OPTION_RELAXED_UNIQUE_CHECKS)
       set_flags(RELAXED_UNIQUE_CHECKS_F);
-  /* if bitmap_init fails, caught in is_valid() */
-  if (likely(!bitmap_init(&m_cols,
+  /* if my_bitmap_init fails, caught in is_valid() */
+  if (likely(!my_bitmap_init(&m_cols,
                           m_width <= sizeof(m_bitbuf)*8 ? m_bitbuf : NULL,
                           m_width,
                           false)))
@@ -1259,7 +1259,7 @@ Old_rows_log_event::Old_rows_log_event(THD *thd_arg, TABLE *tbl_arg, ulong tid,
   }
   else
   {
-    // Needed because bitmap_init() does not set it to null on failure
+    // Needed because my_bitmap_init() does not set it to null on failure
     m_cols.bitmap= 0;
   }
 }
@@ -1313,8 +1313,8 @@ Old_rows_log_event::Old_rows_log_event(const char *buf, uint event_len,
   DBUG_PRINT("debug", ("Reading from %p", ptr_after_width));
   m_width = net_field_length(&ptr_after_width);
   DBUG_PRINT("debug", ("m_width=%lu", m_width));
-  /* if bitmap_init fails, catched in is_valid() */
-  if (likely(!bitmap_init(&m_cols,
+  /* if my_bitmap_init fails, catched in is_valid() */
+  if (likely(!my_bitmap_init(&m_cols,
                           m_width <= sizeof(m_bitbuf)*8 ? m_bitbuf : NULL,
                           m_width,
                           false)))
@@ -1327,7 +1327,7 @@ Old_rows_log_event::Old_rows_log_event(const char *buf, uint event_len,
   }
   else
   {
-    // Needed because bitmap_init() does not set it to null on failure
+    // Needed because my_bitmap_init() does not set it to null on failure
     m_cols.bitmap= NULL;
     DBUG_VOID_RETURN;
   }
@@ -1358,8 +1358,8 @@ Old_rows_log_event::Old_rows_log_event(const char *buf, uint event_len,
 Old_rows_log_event::~Old_rows_log_event()
 {
   if (m_cols.bitmap == m_bitbuf) // no my_malloc happened
-    m_cols.bitmap= 0; // so no my_free in bitmap_free
-  bitmap_free(&m_cols); // To pair with bitmap_init().
+    m_cols.bitmap= 0; // so no my_free in my_bitmap_free
+  my_bitmap_free(&m_cols); // To pair with my_bitmap_init().
   my_free(m_rows_buf);
 }
 
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index 984d9cbc968..808a0d47ac9 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -366,7 +366,8 @@ static DYNAMIC_ARRAY all_options;
 /* Global variables */
 
 bool opt_bin_log, opt_bin_log_used=0, opt_ignore_builtin_innodb= 0;
-my_bool opt_log, opt_slow_log, debug_assert_if_crashed_table= 0, opt_help= 0, opt_abort;
+my_bool opt_log, opt_slow_log, debug_assert_if_crashed_table= 0, opt_help= 0;
+static my_bool opt_abort;
 ulonglong log_output_options;
 my_bool opt_userstat_running;
 my_bool opt_log_queries_not_using_indexes= 0;
@@ -478,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;
@@ -1964,7 +1966,7 @@ void clean_up(bool print_message)
   // We must call end_slave() as clean_up may have been called during startup
   end_slave();
   if (use_slave_mask)
-    bitmap_free(&slave_error_mask);
+    my_bitmap_free(&slave_error_mask);
 #endif
   stop_handle_manager();
   release_ddl_log();
@@ -2018,7 +2020,7 @@ void clean_up(bool print_message)
   if (defaults_argv)
     free_defaults(defaults_argv);
   free_tmpdir(&mysql_tmpdir_list);
-  bitmap_free(&temp_pool);
+  my_bitmap_free(&temp_pool);
   free_max_user_conn();
   free_global_user_stats();
   free_global_client_stats();
@@ -4252,7 +4254,7 @@ static int init_common_variables()
 #endif /* defined(ENABLED_DEBUG_SYNC) */
 
 #if (ENABLE_TEMP_POOL)
-  if (use_temp_pool && bitmap_init(&temp_pool,0,1024,1))
+  if (use_temp_pool && my_bitmap_init(&temp_pool,0,1024,1))
     return 1;
 #else
   use_temp_pool= 0;
@@ -7966,7 +7968,6 @@ static int option_cmp(my_option *a, my_option *b)
         return 1;
     }
   }
-  DBUG_ASSERT(a->name == b->name);
   return 0;
 }
 
@@ -7981,9 +7982,16 @@ static void print_help()
   sys_var_add_options(&all_options, sys_var::PARSE_EARLY);
   add_plugin_options(&all_options, &mem_root);
   sort_dynamic(&all_options, (qsort_cmp) option_cmp);
+  sort_dynamic(&all_options, (qsort_cmp) option_cmp);
   add_terminator(&all_options);
 
   my_print_help((my_option*) all_options.buffer);
+
+  /* Add variables that can be shown but not changed, like version numbers */
+  pop_dynamic(&all_options);
+  sys_var_add_options(&all_options, sys_var::SHOW_VALUE_IN_HELP);
+  sort_dynamic(&all_options, (qsort_cmp) option_cmp);
+  add_terminator(&all_options);
   my_print_variables((my_option*) all_options.buffer);
 
   free_root(&mem_root, MYF(0));
@@ -8944,6 +8952,13 @@ static int get_options(int *argc_ptr, char ***argv_ptr)
         max_binlog_size_var->option.def_value;
     }
   }
+
+  /* Ensure that some variables are not set higher than needed */
+  if (back_log > max_connections)
+    back_log= max_connections;
+  if (thread_cache_size > max_connections)
+    thread_cache_size= max_connections;
+  
   return 0;
 }
 
diff --git a/sql/mysqld.h b/sql/mysqld.h
index 1b34c485101..dd80a6cf423 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/opt_range.cc b/sql/opt_range.cc
index 16f32eb9252..1ced92952c2 100644
--- a/sql/opt_range.cc
+++ b/sql/opt_range.cc
@@ -1823,7 +1823,7 @@ QUICK_RANGE_SELECT::QUICK_RANGE_SELECT(THD *thd, TABLE *table, uint key_nr,
     *create_error= 1;
   }
   else
-    bitmap_init(&column_bitmap, bitmap, head->s->fields, FALSE);
+    my_bitmap_init(&column_bitmap, bitmap, head->s->fields, FALSE);
   DBUG_VOID_RETURN;
 }
 
@@ -2847,7 +2847,7 @@ static int fill_used_fields_bitmap(PARAM *param)
   param->fields_bitmap_size= table->s->column_bitmap_size;
   if (!(tmp= (my_bitmap_map*) alloc_root(param->mem_root,
                                   param->fields_bitmap_size)) ||
-      bitmap_init(&param->needed_fields, tmp, table->s->fields, FALSE))
+      my_bitmap_init(&param->needed_fields, tmp, table->s->fields, FALSE))
     return 1;
 
   bitmap_copy(&param->needed_fields, table->read_set);
@@ -4104,7 +4104,7 @@ static int find_used_partitions_imerge_list(PART_PRUNE_PARAM *ppar,
     */
     return find_used_partitions_imerge(ppar, merges.head());
   }
-  bitmap_init(&all_merges, bitmap_buf, n_bits, FALSE);
+  my_bitmap_init(&all_merges, bitmap_buf, n_bits, FALSE);
   bitmap_set_prefix(&all_merges, n_bits);
 
   List_iterator<SEL_IMERGE> it(merges);
@@ -4751,7 +4751,7 @@ static bool create_partition_index_description(PART_PRUNE_PARAM *ppar)
     uint32 bufsize= bitmap_buffer_size(ppar->part_info->num_subparts);
     if (!(buf= (my_bitmap_map*) alloc_root(alloc, bufsize)))
       return TRUE;
-    bitmap_init(&ppar->subparts_bitmap, buf, ppar->part_info->num_subparts,
+    my_bitmap_init(&ppar->subparts_bitmap, buf, ppar->part_info->num_subparts,
                 FALSE);
   }
   range_par->key_parts= key_part;
@@ -5511,7 +5511,7 @@ bool create_fields_bitmap(PARAM *param, MY_BITMAP *fields_bitmap)
   if (!(bitmap_buf= (my_bitmap_map *) alloc_root(param->mem_root,
                                                  param->fields_bitmap_size)))
     return TRUE;
-  if (bitmap_init(fields_bitmap, bitmap_buf, param->table->s->fields, FALSE))
+  if (my_bitmap_init(fields_bitmap, bitmap_buf, param->table->s->fields, FALSE))
     return TRUE;
   
   return FALSE;
@@ -6329,7 +6329,7 @@ ROR_SCAN_INFO *make_ror_scan(const PARAM *param, int idx, SEL_ARG *sel_arg)
                                                 param->fields_bitmap_size)))
     DBUG_RETURN(NULL);
 
-  if (bitmap_init(&ror_scan->covered_fields, bitmap_buf,
+  if (my_bitmap_init(&ror_scan->covered_fields, bitmap_buf,
                   param->table->s->fields, FALSE))
     DBUG_RETURN(NULL);
   bitmap_clear_all(&ror_scan->covered_fields);
@@ -6447,7 +6447,7 @@ ROR_INTERSECT_INFO* ror_intersect_init(const PARAM *param)
   if (!(buf= (my_bitmap_map*) alloc_root(param->mem_root,
                                          param->fields_bitmap_size)))
     return NULL;
-  if (bitmap_init(&info->covered_fields, buf, param->table->s->fields,
+  if (my_bitmap_init(&info->covered_fields, buf, param->table->s->fields,
                   FALSE))
     return NULL;
   info->is_covering= FALSE;
@@ -7024,7 +7024,7 @@ TRP_ROR_INTERSECT *get_best_covering_ror_intersect(PARAM *param,
     covered_fields->bitmap= (my_bitmap_map*)alloc_root(param->mem_root,
                                                param->fields_bitmap_size);
   if (!covered_fields->bitmap ||
-      bitmap_init(covered_fields, covered_fields->bitmap,
+      my_bitmap_init(covered_fields, covered_fields->bitmap,
                   param->table->s->fields, FALSE))
     DBUG_RETURN(0);
   bitmap_clear_all(covered_fields);
diff --git a/sql/opt_table_elimination.cc b/sql/opt_table_elimination.cc
index 46bed0e60e7..9b58f5f3126 100644
--- a/sql/opt_table_elimination.cc
+++ b/sql/opt_table_elimination.cc
@@ -1042,7 +1042,7 @@ bool Dep_analysis_context::setup_equality_modules_deps(List<Dep_module>
  
   void *buf;
   if (!(buf= current_thd->alloc(bitmap_buffer_size(offset))) ||
-      bitmap_init(&expr_deps, (my_bitmap_map*)buf, offset, FALSE))
+      my_bitmap_init(&expr_deps, (my_bitmap_map*)buf, offset, FALSE))
   {
     DBUG_RETURN(TRUE); /* purecov: inspected */
   }
diff --git a/sql/partition_info.cc b/sql/partition_info.cc
index ca455c0d7bc..1ae62011b43 100644
--- a/sql/partition_info.cc
+++ b/sql/partition_info.cc
@@ -385,7 +385,7 @@ bool partition_info::can_prune_insert(THD* thd,
     DBUG_RETURN(true);
   }
   /* Also clears all bits. */
-  if (bitmap_init(used_partitions, bitmap_buf, num_partitions, false))
+  if (my_bitmap_init(used_partitions, bitmap_buf, num_partitions, false))
   {
     /* purecov: begin deadcode */
     /* Cannot happen, due to pre-alloc. */
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_injector.h b/sql/rpl_injector.h
index f4790cc963a..98788955e24 100644
--- a/sql/rpl_injector.h
+++ b/sql/rpl_injector.h
@@ -94,13 +94,13 @@ public:
 
             injector::transaction::table tbl(share->table, true);
             MY_BITMAP cols;
-            bitmap_init(&cols, NULL, (i + 7) / 8, false);
+            my_bitmap_init(&cols, NULL, (i + 7) / 8, false);
             inj->write_row(::server_id, tbl, &cols, row_data);
 
           or
 
             MY_BITMAP cols;
-            bitmap_init(&cols, NULL, (i + 7) / 8, false);
+            my_bitmap_init(&cols, NULL, (i + 7) / 8, false);
             inj->write_row(::server_id, 
                            injector::transaction::table(share->table, true), 
                            &cols, row_data);
diff --git a/sql/rpl_rli.cc b/sql/rpl_rli.cc
index ac93f9640e8..c5c59f3aebb 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/set_var.h b/sql/set_var.h
index a6c3b9daccd..88c5013e28e 100644
--- a/sql/set_var.h
+++ b/sql/set_var.h
@@ -60,7 +60,7 @@ public:
   sys_var *next;
   LEX_CSTRING name;
   enum flag_enum { GLOBAL, SESSION, ONLY_SESSION, SCOPE_MASK=1023,
-                   READONLY=1024, ALLOCATED=2048, PARSE_EARLY=4096 };
+                   READONLY=1024, ALLOCATED=2048, PARSE_EARLY=4096, SHOW_VALUE_IN_HELP=8192 };
   /**
     Enumeration type to indicate for a system variable whether
     it will be written to the binlog or not.
@@ -142,8 +142,9 @@ public:
   }
   bool register_option(DYNAMIC_ARRAY *array, int parse_flags)
   {
-    return (option.id != -1) && ((flags & PARSE_EARLY) == parse_flags) &&
-           insert_dynamic(array, (uchar*)&option);
+    return ((((option.id != -1) && ((flags & PARSE_EARLY) == parse_flags)) ||
+             (flags & parse_flags)) &&
+            insert_dynamic(array, (uchar*)&option));
   }
   void do_deprecated_warning(THD *thd);
 
diff --git a/sql/share/errmsg-utf8.txt b/sql/share/errmsg-utf8.txt
index 25017cef42c..160bf2e6c87 100644
--- a/sql/share/errmsg-utf8.txt
+++ b/sql/share/errmsg-utf8.txt
@@ -2118,13 +2118,9 @@ ER_INSERT_INFO
         spa "Registros: %ld  Duplicados: %ld  Peligros: %ld"
         swe "Rader: %ld  Dubletter: %ld  Varningar: %ld"
         ukr "Записів: %ld  Дублікатів: %ld  Застережень: %ld"
-ER_UPDATE_TABLE_USED  
-        eng "You can't specify target table '%-.192s' for update in FROM clause"
-        ger "Die Verwendung der zu aktualisierenden Zieltabelle '%-.192s' ist in der FROM-Klausel nicht zulässig."
-        jpn "FROM句にある表 '%-.192s' はUPDATEの対象にできません。"
-        rus "Не допускается указание таблицы '%-.192s' в списке таблиц FROM для внесения в нее изменений"
-        swe "INSERT-table '%-.192s' får inte finnas i FROM tabell-listan"
-        ukr "Таблиця '%-.192s' що змінюється не дозволена у переліку таблиць FROM"
+ER_UPDATE_TABLE_USED
+        eng "Table '%-.192s' is specified twice, both as a target for '%s' and as a separate source for data"
+        swe "Table '%-.192s' är använd två gånger. Både för '%s' och för att hämta data"
 ER_NO_SUCH_THREAD  
         cze "Neznámá identifikace threadu: %lu"
         dan "Ukendt tråd id: %lu"
@@ -7067,3 +7063,5 @@ ER_CHANGE_SLAVE_PARALLEL_THREADS_ACTIVE
 	eng "Cannot change @@slave_parallel_threads while another change is in progress"
 ER_PRIOR_COMMIT_FAILED
 	eng "Commit failed due to failure of an earlier commit on which this one depends"
+ER_IT_IS_A_VIEW 42S02
+        eng "'%-.192s' is a view"
diff --git a/sql/slave.cc b/sql/slave.cc
index 6854d2bd6de..a2ed89e0c9c 100644
--- a/sql/slave.cc
+++ b/sql/slave.cc
@@ -573,7 +573,7 @@ void init_slave_skip_errors(const char* arg)
   const char *p;
   DBUG_ENTER("init_slave_skip_errors");
 
-  if (bitmap_init(&slave_error_mask,0,MAX_SLAVE_ERROR,0))
+  if (my_bitmap_init(&slave_error_mask,0,MAX_SLAVE_ERROR,0))
   {
     fprintf(stderr, "Badly out of memory, please check your system status\n");
     exit(1);
@@ -1047,9 +1047,10 @@ static bool sql_slave_killed(rpl_group_info *rgi)
         "documentation for details).";
 
       DBUG_PRINT("info", ("modified_non_trans_table: %d  OPTION_BEGIN: %d  "
-                          "is_in_group: %d",
+                          "OPTION_KEEP_LOG: %d  is_in_group: %d",
                           thd->transaction.all.modified_non_trans_table,
                           test(thd->variables.option_bits & OPTION_BEGIN),
+                          test(thd->variables.option_bits & OPTION_KEEP_LOG),
                           rli->is_in_group()));
 
       if (rli->abort_slave)
@@ -3128,9 +3129,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_alter.cc b/sql/sql_alter.cc
index 01bffaf132f..97b9c127c22 100644
--- a/sql/sql_alter.cc
+++ b/sql/sql_alter.cc
@@ -338,19 +338,8 @@ bool Sql_cmd_discard_import_tablespace::execute(THD *thd)
     it is the case.
     TODO: this design is obsolete and will be removed.
   */
-  int table_kind= check_if_log_table(table_list->db_length, table_list->db,
-                                     table_list->table_name_length,
-                                     table_list->table_name, false);
-
-  if (table_kind)
-  {
-    /* Disable alter of enabled log tables */
-    if (logger.is_log_table_enabled(table_kind))
-    {
-      my_error(ER_BAD_LOG_STATEMENT, MYF(0), "ALTER");
-      return true;
-    }
-  }
+  if (check_if_log_table(table_list, TRUE, "ALTER"))
+    return true;
 
   return
     mysql_discard_or_import_tablespace(thd, table_list,
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index 44d68f35ab2..f4ff9f2fc75 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -784,12 +784,18 @@ static void close_open_tables(THD *thd)
                      access the table cache key
 
   @param[in] extra
-                     HA_EXTRA_PREPRE_FOR_DROP if the table is being dropped
-                     HA_EXTRA_PREPARE_FOR_REANME if the table is being renamed
-                     HA_EXTRA_NOT_USED           no drop/rename
-                     In case of drop/reanme the documented behaviour is to
+                     HA_EXTRA_PREPARE_FOR_DROP
+                        - The table is dropped
+                     HA_EXTRA_PREPARE_FOR_RENAME
+                        - The table is renamed
+                     HA_EXTRA_NOT_USED
+                        - The table is marked as closed in the
+                          locked_table_list but kept there so one can call
+                          locked_table_list->reopen_tables() to put it back.
+                          
+                     In case of drop/rename the documented behavior is to
                      implicitly remove the table from LOCK TABLES
-                     list.
+                     list. 
 
   @pre Must be called with an X MDL lock on the table.
 */
@@ -1477,7 +1483,7 @@ void update_non_unique_table_error(TABLE_LIST *update,
       return;
     }
   }
-  my_error(ER_UPDATE_TABLE_USED, MYF(0), update->alias);
+  my_error(ER_UPDATE_TABLE_USED, MYF(0), update->alias, operation);
 }
 
 
@@ -1588,26 +1594,21 @@ TABLE *find_temporary_table(THD *thd,
   thd->temporary_tables list, it's impossible to tell here whether
   we're dealing with an internal or a user temporary table.
 
-  If is_trans is not null, we return the type of the table:
-  either transactional (e.g. innodb) as TRUE or non-transactional
-  (e.g. myisam) as FALSE.
+  @param thd      Thread handler
+  @param table	  Temporary table to be deleted
+  @param is_trans Is set to the type of the table:
+                  transactional (e.g. innodb) as TRUE or non-transactional
+                  (e.g. myisam) as FALSE.
 
   @retval  0  the table was found and dropped successfully.
-  @retval  1  the table was not found in the list of temporary tables
-              of this thread
   @retval -1  the table is in use by a outer query
 */
 
-int drop_temporary_table(THD *thd, TABLE_LIST *table_list, bool *is_trans)
+int drop_temporary_table(THD *thd, TABLE *table, bool *is_trans)
 {
   DBUG_ENTER("drop_temporary_table");
   DBUG_PRINT("tmptable", ("closing table: '%s'.'%s'",
-                          table_list->db, table_list->table_name));
-
-  if (!is_temporary_table(table_list))
-    DBUG_RETURN(1);
-
-  TABLE *table= table_list->table;
+                          table->s->db.str, table->s->table_name.str));
 
   /* Table might be in use by some outer statement. */
   if (table->query_id && table->query_id != thd->query_id)
@@ -1627,10 +1628,10 @@ int drop_temporary_table(THD *thd, TABLE_LIST *table_list, bool *is_trans)
   */
   mysql_lock_remove(thd, thd->lock, table);
   close_temporary_table(thd, table, 1, 1);
-  table_list->table= NULL;
   DBUG_RETURN(0);
 }
 
+
 /*
   unlink from thd->temporary tables and close temporary table
 */
@@ -2612,9 +2613,9 @@ Locked_tables_list::init_locked_tables(THD *thd)
   {
     TABLE_LIST *src_table_list= table->pos_in_table_list;
     char *db, *table_name, *alias;
-    size_t db_len= src_table_list->db_length;
-    size_t table_name_len= src_table_list->table_name_length;
-    size_t alias_len= strlen(src_table_list->alias);
+    size_t db_len=         table->s->db.length;
+    size_t table_name_len= table->s->table_name.length;
+    size_t alias_len=      table->alias.length();
     TABLE_LIST *dst_table_list;
 
     if (! multi_alloc_root(&m_locked_tables_root,
@@ -2624,23 +2625,15 @@ Locked_tables_list::init_locked_tables(THD *thd)
                            &alias, alias_len + 1,
                            NullS))
     {
-      unlock_locked_tables(0);
+      reset();
       return TRUE;
     }
 
-    memcpy(db, src_table_list->db, db_len + 1);
-    memcpy(table_name, src_table_list->table_name, table_name_len + 1);
-    memcpy(alias, src_table_list->alias, alias_len + 1);
-    /**
-      Sic: remember the *actual* table level lock type taken, to
-      acquire the exact same type in reopen_tables().
-      E.g. if the table was locked for write, src_table_list->lock_type is
-      TL_WRITE_DEFAULT, whereas reginfo.lock_type has been updated from
-      thd->update_lock_default.
-    */
+    memcpy(db,         table->s->db.str, db_len + 1);
+    memcpy(table_name, table->s->table_name.str, table_name_len + 1);
+    strmake(alias,     table->alias.ptr(), alias_len);
     dst_table_list->init_one_table(db, db_len, table_name, table_name_len,
-                                   alias,
-                                   src_table_list->table->reginfo.lock_type);
+                                   alias, table->reginfo.lock_type);
     dst_table_list->table= table;
     dst_table_list->mdl_request.ticket= src_table_list->mdl_request.ticket;
 
@@ -2661,7 +2654,7 @@ Locked_tables_list::init_locked_tables(THD *thd)
                                         (m_locked_tables_count+1));
     if (m_reopen_array == NULL)
     {
-      unlock_locked_tables(0);
+      reset();
       return TRUE;
     }
   }
@@ -2682,42 +2675,50 @@ Locked_tables_list::init_locked_tables(THD *thd)
 void
 Locked_tables_list::unlock_locked_tables(THD *thd)
 {
-  if (thd)
+  DBUG_ASSERT(!thd->in_sub_stmt &&
+              !(thd->state_flags & Open_tables_state::BACKUPS_AVAIL));
+  /*
+    Sic: we must be careful to not close open tables if
+    we're not in LOCK TABLES mode: unlock_locked_tables() is
+    sometimes called implicitly, expecting no effect on
+    open tables, e.g. from begin_trans().
+  */
+  if (thd->locked_tables_mode != LTM_LOCK_TABLES)
+    return;
+
+  for (TABLE_LIST *table_list= m_locked_tables;
+       table_list; table_list= table_list->next_global)
   {
-    DBUG_ASSERT(!thd->in_sub_stmt &&
-                !(thd->state_flags & Open_tables_state::BACKUPS_AVAIL));
     /*
-      Sic: we must be careful to not close open tables if
-      we're not in LOCK TABLES mode: unlock_locked_tables() is
-      sometimes called implicitly, expecting no effect on
-      open tables, e.g. from begin_trans().
+      Clear the position in the list, the TABLE object will be
+      returned to the table cache.
     */
-    if (thd->locked_tables_mode != LTM_LOCK_TABLES)
-      return;
+    if (table_list->table)                    // If not closed
+      table_list->table->pos_in_locked_tables= NULL;
+  }
+  thd->leave_locked_tables_mode();
 
-    for (TABLE_LIST *table_list= m_locked_tables;
-         table_list; table_list= table_list->next_global)
-    {
-      /*
-        Clear the position in the list, the TABLE object will be
-        returned to the table cache.
-      */
-      if (table_list->table)                    // If not closed
-        table_list->table->pos_in_locked_tables= NULL;
-    }
-    thd->leave_locked_tables_mode();
+  DBUG_ASSERT(thd->transaction.stmt.is_empty());
+  close_thread_tables(thd);
+
+  /*
+    We rely on the caller to implicitly commit the
+    transaction and release transactional locks.
+  */
 
-    DBUG_ASSERT(thd->transaction.stmt.is_empty());
-    close_thread_tables(thd);
-    /*
-      We rely on the caller to implicitly commit the
-      transaction and release transactional locks.
-    */
-  }
   /*
     After closing tables we can free memory used for storing lock
     request for metadata locks and TABLE_LIST elements.
   */
+  reset();
+}
+
+/*
+  Free memory allocated for storing locks
+*/
+
+void Locked_tables_list::reset()
+{
   free_root(&m_locked_tables_root, MYF(0));
   m_locked_tables= NULL;
   m_locked_tables_last= &m_locked_tables;
@@ -2782,6 +2783,7 @@ void Locked_tables_list::unlink_from_list(THD *thd,
       m_locked_tables_last= table_list->prev_global;
     else
       table_list->next_global->prev_global= table_list->prev_global;
+    m_locked_tables_count--;
   }
 }
 
@@ -2835,8 +2837,13 @@ unlink_all_closed_tables(THD *thd, MYSQL_LOCK *lock, size_t reopen_count)
         m_locked_tables_last= table_list->prev_global;
       else
         table_list->next_global->prev_global= table_list->prev_global;
+      m_locked_tables_count--;
     }
   }
+
+  /* If no tables left, do an automatic UNLOCK TABLES */
+  if (thd->lock && thd->lock->table_count == 0)
+    unlock_locked_tables(thd);
 }
 
 
@@ -2909,6 +2916,62 @@ Locked_tables_list::reopen_tables(THD *thd)
   return FALSE;
 }
 
+/**
+  Add back a locked table to the locked list that we just removed from it.
+  This is needed in CREATE OR REPLACE TABLE where we are dropping, creating
+  and re-opening a locked table.
+
+  @return 0  0k
+  @return 1  error
+*/
+
+bool Locked_tables_list::restore_lock(THD *thd, TABLE_LIST *dst_table_list,
+                                      TABLE *table, MYSQL_LOCK *lock)
+{
+  MYSQL_LOCK *merged_lock;
+  DBUG_ENTER("restore_lock");
+  DBUG_ASSERT(!strcmp(dst_table_list->table_name, table->s->table_name.str));
+
+  /* Ensure we have the memory to add the table back */
+  if (!(merged_lock= mysql_lock_merge(thd->lock, lock)))
+    DBUG_RETURN(1);
+  thd->lock= merged_lock;
+
+  /* Link to the new table */
+  dst_table_list->table= table;
+  /*
+    The lock type may have changed (normally it should not as create
+    table will lock the table in write mode
+  */
+  dst_table_list->lock_type= table->reginfo.lock_type;
+  table->pos_in_locked_tables= dst_table_list;
+
+  add_back_last_deleted_lock(dst_table_list);
+
+  table->mdl_ticket->downgrade_lock(table->reginfo.lock_type >=
+                                    TL_WRITE_ALLOW_WRITE ? 
+                                    MDL_SHARED_NO_READ_WRITE :
+                                    MDL_SHARED_READ);
+
+  DBUG_RETURN(0);
+}
+
+/*
+  Add back the last deleted lock structure.
+  This should be followed by a call to reopen_tables() to
+  open the table.
+*/
+
+void Locked_tables_list::add_back_last_deleted_lock(TABLE_LIST *dst_table_list)
+{
+  /* Link the lock back in the locked tables list */
+  dst_table_list->prev_global= m_locked_tables_last;
+  *m_locked_tables_last= dst_table_list;
+  m_locked_tables_last= &dst_table_list->next_global;
+  dst_table_list->next_global= 0;
+  m_locked_tables_count++;
+}
+
 
 #ifndef DBUG_OFF
 /* Cause a spurious statement reprepare for debug purposes. */
@@ -4046,9 +4109,9 @@ lock_table_names(THD *thd,
   if (mdl_requests.is_empty())
     DBUG_RETURN(FALSE);
 
-  /* Check if CREATE TABLE was used */
-  create_table= (tables_start && tables_start->open_strategy ==
-                 TABLE_LIST::OPEN_IF_EXISTS);
+  /* Check if CREATE TABLE without REPLACE was used */
+  create_table= (thd->lex->sql_command == SQLCOM_CREATE_TABLE &&
+                 !(thd->lex->create_info.options & HA_LEX_CREATE_REPLACE));
 
   if (!(flags & MYSQL_OPEN_SKIP_SCOPED_MDL_LOCK))
   {
@@ -5294,6 +5357,39 @@ bool lock_tables(THD *thd, TABLE_LIST *tables, uint count,
 }
 
 
+/*
+  Restart transaction for tables
+
+  This is used when we had to do an implicit commit after tables are opened
+  and want to restart transactions on tables.
+
+  This is used in case of:
+  LOCK TABLES xx
+  CREATE OR REPLACE TABLE xx;
+*/
+
+bool restart_trans_for_tables(THD *thd, TABLE_LIST *table)
+{
+  DBUG_ENTER("restart_trans_for_tables");
+
+  if (!thd->locked_tables_mode)
+    DBUG_RETURN(FALSE);
+
+  for (; table; table= table->next_global)
+  {
+    if (table->placeholder())
+      continue;
+
+    if (check_lock_and_start_stmt(thd, thd->lex, table))
+    {
+      DBUG_ASSERT(0);                           // Should never happen
+      DBUG_RETURN(TRUE);
+    }
+  }
+  DBUG_RETURN(FALSE);
+}
+
+
 /**
   Prepare statement for reopening of tables and recalculation of set of
   prelocked tables.
diff --git a/sql/sql_base.h b/sql/sql_base.h
index 3e633fad084..61442843a39 100644
--- a/sql/sql_base.h
+++ b/sql/sql_base.h
@@ -248,7 +248,7 @@ void close_thread_table(THD *thd, TABLE **table_ptr);
 bool close_temporary_tables(THD *thd);
 TABLE_LIST *unique_table(THD *thd, TABLE_LIST *table, TABLE_LIST *table_list,
                          bool check_alias);
-int drop_temporary_table(THD *thd, TABLE_LIST *table_list, bool *is_trans);
+int drop_temporary_table(THD *thd, TABLE *table, bool *is_trans);
 void close_temporary_table(THD *thd, TABLE *table, bool free_share,
                            bool delete_table);
 void close_temporary(TABLE *table, bool free_share, bool delete_table);
@@ -486,6 +486,8 @@ inline bool open_and_lock_tables(THD *thd, TABLE_LIST *tables,
 }
 
 
+bool restart_trans_for_tables(THD *thd, TABLE_LIST *table);
+
 /**
   A context of open_tables() function, used to recover
   from a failed open_table() or open_routine() attempt.
diff --git a/sql/sql_bitmap.h b/sql/sql_bitmap.h
index 7e163b0dbcc..2e07695f605 100644
--- a/sql/sql_bitmap.h
+++ b/sql/sql_bitmap.h
@@ -34,7 +34,7 @@ public:
   Bitmap() { init(); }
   Bitmap(const Bitmap& from) { *this=from; }
   explicit Bitmap(uint prefix_to_set) { init(prefix_to_set); }
-  void init() { bitmap_init(&map, buffer, default_width, 0); }
+  void init() { my_bitmap_init(&map, buffer, default_width, 0); }
   void init(uint prefix_to_set) { init(); set_prefix(prefix_to_set); }
   uint length() const { return default_width; }
   Bitmap& operator=(const Bitmap& map2)
@@ -52,7 +52,7 @@ public:
   void intersect(ulonglong map2buff)
   {
     MY_BITMAP map2;
-    bitmap_init(&map2, (uint32 *)&map2buff, sizeof(ulonglong)*8, 0);
+    my_bitmap_init(&map2, (uint32 *)&map2buff, sizeof(ulonglong)*8, 0);
     bitmap_intersect(&map, &map2);
   }
   /* Use highest bit for all bits above sizeof(ulonglong)*8. */
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index bcaea00b081..ee0cc43dca2 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -5391,6 +5391,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.
@@ -5586,6 +5590,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,
@@ -5619,6 +5627,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.
@@ -5661,6 +5673,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,
@@ -5681,6 +5697,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)
@@ -5700,6 +5720,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.
@@ -5947,6 +5971,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 5c0684f5d68..e3668321bec 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -1518,8 +1518,9 @@ public:
   void unlock_locked_tables(THD *thd);
   ~Locked_tables_list()
   {
-    unlock_locked_tables(0);
+    reset();
   }
+  void reset();
   bool init_locked_tables(THD *thd);
   TABLE_LIST *locked_tables() { return m_locked_tables; }
   void unlink_from_list(THD *thd, TABLE_LIST *table_list,
@@ -1528,6 +1529,9 @@ public:
                                 MYSQL_LOCK *lock,
                                 size_t reopen_count);
   bool reopen_tables(THD *thd);
+  bool restore_lock(THD *thd, TABLE_LIST *dst_table_list, TABLE *table,
+                    MYSQL_LOCK *lock);
+  void add_back_last_deleted_lock(TABLE_LIST *dst_table_list);
 };
 
 
@@ -1971,7 +1975,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) */
@@ -2562,12 +2569,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
@@ -2600,13 +2607,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. */
@@ -2623,6 +2623,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;
@@ -4014,6 +4028,8 @@ class select_create: public select_insert {
   MYSQL_LOCK *m_lock;
   /* m_lock or thd->extra_lock */
   MYSQL_LOCK **m_plock;
+  bool       exit_done;
+
 public:
   select_create (TABLE_LIST *table_arg,
 		 HA_CREATE_INFO *create_info_par,
@@ -4025,7 +4041,7 @@ public:
     create_info(create_info_par),
     select_tables(select_tables_arg),
     alter_info(alter_info_arg),
-    m_plock(NULL)
+    m_plock(NULL), exit_done(0)
     {}
   int prepare(List<Item> &list, SELECT_LEX_UNIT *u);
 
diff --git a/sql/sql_db.cc b/sql/sql_db.cc
index 8068901ebec..fcde2b4acbd 100644
--- a/sql/sql_db.cc
+++ b/sql/sql_db.cc
@@ -798,14 +798,8 @@ bool mysql_rm_db(THD *thd,char *db,bool if_exists, bool silent)
   if ((my_strcasecmp(system_charset_info, MYSQL_SCHEMA_NAME.str, db) == 0))
   {
     for (table= tables; table; table= table->next_local)
-    {
-      if (check_if_log_table(table->db_length, table->db,
-                             table->table_name_length, table->table_name, true))
-      {
-        my_error(ER_BAD_LOG_STATEMENT, MYF(0), "DROP");
+      if (check_if_log_table(table, TRUE, "DROP"))
         goto exit;
-      }
-    }
   }
 
   /* Lock all tables and stored routines about to be dropped. */
@@ -835,7 +829,7 @@ bool mysql_rm_db(THD *thd,char *db,bool if_exists, bool silent)
   thd->push_internal_handler(&err_handler);
   if (!thd->killed &&
       !(tables &&
-        mysql_rm_table_no_locks(thd, tables, true, false, true, true)))
+        mysql_rm_table_no_locks(thd, tables, true, false, true, true, false)))
   {
     /*
       We temporarily disable the binary log while dropping the objects
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index e0ecfcac72e..899e0dead6b 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -1229,7 +1229,7 @@ static bool check_view_insertability(THD * thd, TABLE_LIST *view)
 
   DBUG_ASSERT(view->table != 0 && view->field_translation != 0);
 
-  (void) bitmap_init(&used_fields, used_fields_buff, table->s->fields, 0);
+  (void) my_bitmap_init(&used_fields, used_fields_buff, table->s->fields, 0);
   bitmap_clear_all(&used_fields);
 
   view->contain_auto_increment= 0;
@@ -3803,7 +3803,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;
@@ -3925,6 +3926,16 @@ static TABLE *create_table_from_items(THD *thd, HA_CREATE_INFO *create_info,
 
   DEBUG_SYNC(thd,"create_table_select_before_create");
 
+  /* Check if LOCK TABLES + CREATE OR REPLACE of existing normal table*/
+  if (thd->locked_tables_mode && create_table->table &&
+      !create_info->tmp_table())
+  {
+    /* Remember information about the locked table */
+    create_info->pos_in_locked_tables=
+      create_table->table->pos_in_locked_tables;
+    create_info->mdl_ticket= create_table->table->mdl_ticket;
+  }
+
   /*
     Create and lock table.
 
@@ -3941,52 +3952,63 @@ static TABLE *create_table_from_items(THD *thd, HA_CREATE_INFO *create_info,
     TABLE, which is a wrong order. So we keep binary logging disabled when we
     open_table().
   */
+
+  if (!mysql_create_table_no_lock(thd, create_table->db,
+                                  create_table->table_name,
+                                  create_info, alter_info, NULL,
+                                  select_field_count))
   {
-    if (!mysql_create_table_no_lock(thd, create_table->db,
-                                    create_table->table_name,
-                                    create_info, alter_info, NULL,
-                                    select_field_count))
+    DEBUG_SYNC(thd,"create_table_select_before_open");
+
+    /*
+      If we had a temporary table or a table used with LOCK TABLES,
+      it was closed by mysql_create()
+    */
+    create_table->table= 0;
+
+    if (!create_info->tmp_table())
     {
-      DEBUG_SYNC(thd,"create_table_select_before_open");
+      Open_table_context ot_ctx(thd, MYSQL_OPEN_REOPEN);
+      TABLE_LIST::enum_open_strategy save_open_strategy;
 
-      if (!create_info->tmp_table())
-      {
-        Open_table_context ot_ctx(thd, MYSQL_OPEN_REOPEN);
-        /*
-          Here we open the destination table, on which we already have
-          an exclusive metadata lock.
-        */
-        if (open_table(thd, create_table, thd->mem_root, &ot_ctx))
-        {
-          quick_rm_table(thd, create_info->db_type, create_table->db,
-                         table_case_name(create_info, create_table->table_name),
-                         0);
-        }
-        else
-          table= create_table->table;
-      }
-      else
+      /* Force the newly created table to be opened */
+      save_open_strategy= create_table->open_strategy;
+      create_table->open_strategy= TABLE_LIST::OPEN_NORMAL;
+      /*
+        Here we open the destination table, on which we already have
+        an exclusive metadata lock.
+      */
+      if (open_table(thd, create_table, thd->mem_root, &ot_ctx))
       {
-        if (open_temporary_table(thd, create_table))
-        {
-          /*
-            This shouldn't happen as creation of temporary table should make
-            it preparable for open. Anyway we can't drop temporary table if
-            we are unable to find it.
-          */
-          DBUG_ASSERT(0);
-        }
-        else
-          table= create_table->table;
+        quick_rm_table(thd, create_info->db_type, create_table->db,
+                       table_case_name(create_info, create_table->table_name),
+                       0);
       }
+      /* Restore */
+      create_table->open_strategy= save_open_strategy;
     }
-    if (!table)                                   // open failed
+    else
     {
-      if (!thd->is_error())                     // CREATE ... IF NOT EXISTS
-        my_ok(thd);                             //   succeed, but did nothing
-      DBUG_RETURN(0);
+      if (open_temporary_table(thd, create_table))
+      {
+        /*
+          This shouldn't happen as creation of temporary table should make
+          it preparable for open. Anyway we can't drop temporary table if
+          we are unable to find it.
+        */
+        DBUG_ASSERT(0);
+      }
     }
   }
+  else
+    create_table->table= 0;                     // Create failed
+  
+  if (!(table= create_table->table))
+  {
+    if (!thd->is_error())                     // CREATE ... IF NOT EXISTS
+      my_ok(thd);                             //   succeed, but did nothing
+    DBUG_RETURN(0);
+  }
 
   DEBUG_SYNC(thd,"create_table_select_before_lock");
 
@@ -4003,7 +4025,7 @@ static TABLE *create_table_from_items(THD *thd, HA_CREATE_INFO *create_info,
     /* purecov: begin tested */
     /*
       This can happen in innodb when you get a deadlock when using same table
-      in insert and select
+      in insert and select or when you run out of memory.
     */
     my_error(ER_CANT_LOCK, MYF(0), my_errno);
     if (*lock)
@@ -4101,8 +4123,6 @@ select_create::prepare(List<Item> &values, SELECT_LEX_UNIT *u)
     thd->binlog_start_trans_and_stmt();
   }
 
-  DBUG_ASSERT(create_table->table == NULL);
-
   DEBUG_SYNC(thd,"create_table_select_before_check_if_exists");
 
   if (!(table= create_table_from_items(thd, create_info, create_table,
@@ -4190,7 +4210,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())
@@ -4245,39 +4267,67 @@ void select_create::send_error(uint errcode,const char *err)
 
 bool select_create::send_eof()
 {
-  bool tmp=select_insert::send_eof();
-  if (tmp)
+  if (select_insert::send_eof())
+  {
     abort_result_set();
-  else
+    return 1;
+  }
+
+  exit_done= 1;                                 // Avoid double calls
+  /*
+    Do an implicit commit at end of statement for non-temporary
+    tables.  This can fail, but we should unlock the table
+    nevertheless.
+  */
+  if (!table->s->tmp_table)
   {
-    /*
-      Do an implicit commit at end of statement for non-temporary
-      tables.  This can fail, but we should unlock the table
-      nevertheless.
-    */
-    if (!table->s->tmp_table)
-    {
-      trans_commit_stmt(thd);
+    trans_commit_stmt(thd);
+    if (!(thd->variables.option_bits & OPTION_GTID_BEGIN))
       trans_commit_implicit(thd);
-    }
+  }
 
-    table->file->extra(HA_EXTRA_NO_IGNORE_DUP_KEY);
-    table->file->extra(HA_EXTRA_WRITE_CANNOT_REPLACE);
-    if (m_plock)
+  table->file->extra(HA_EXTRA_NO_IGNORE_DUP_KEY);
+  table->file->extra(HA_EXTRA_WRITE_CANNOT_REPLACE);
+
+  if (m_plock)
+  {
+    MYSQL_LOCK *lock= *m_plock;
+    *m_plock= NULL;
+    m_plock= NULL;
+
+    if (create_info->pos_in_locked_tables)
     {
-      mysql_unlock_tables(thd, *m_plock);
-      *m_plock= NULL;
-      m_plock= NULL;
+      /*
+        If we are under lock tables, we have created a table that was
+        originally locked. We should add back the lock to ensure that
+        all tables in the thd->open_list are locked!
+      */
+      table->mdl_ticket= create_info->mdl_ticket;
+
+      /* The following should never fail, except if out of memory */
+      if (!thd->locked_tables_list.restore_lock(thd,
+                                                create_info->
+                                                pos_in_locked_tables,
+                                                table, lock))
+        return 0;                               // ok
+      /* Fail. Continue without locking the table */
     }
+    mysql_unlock_tables(thd, lock);
   }
-  return tmp;
+  return 0;
 }
 
 
 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 */
+  if (exit_done)
+    DBUG_VOID_RETURN;
+  exit_done= 1;
+
   /*
     In select_insert::abort_result_set() we roll back the statement, including
     truncating the transaction cache of the binary log. To do this, we
@@ -4292,11 +4342,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_load.cc b/sql/sql_load.cc
index 281d1de7877..a0959bdd278 100644
--- a/sql/sql_load.cc
+++ b/sql/sql_load.cc
@@ -258,7 +258,8 @@ int mysql_load(THD *thd,sql_exchange *ex,TABLE_LIST *table_list,
   */
   if (unique_table(thd, table_list, table_list->next_global, 0))
   {
-    my_error(ER_UPDATE_TABLE_USED, MYF(0), table_list->table_name);
+    my_error(ER_UPDATE_TABLE_USED, MYF(0), table_list->table_name,
+             "LOAD DATA");
     DBUG_RETURN(TRUE);
   }
 
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 15d7e19188e..0de3609e932 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;
@@ -2440,11 +2452,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
@@ -2829,6 +2844,7 @@ case SQLCOM_PREPARE:
       goto end_with_restore_list;
     }
 
+    /* Check privileges */
     if ((res= create_table_precheck(thd, select_tables, create_table)))
       goto end_with_restore_list;
 
@@ -2863,6 +2879,23 @@ case SQLCOM_PREPARE:
       create_info.table_charset= 0;
     }
 
+    /*
+      For CREATE TABLE we should not open the table even if it exists.
+      If the table exists, we should either not create it or replace it
+    */
+    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;
@@ -2951,28 +2984,25 @@ case SQLCOM_PREPARE:
         /* Got error or warning. Set res to 1 if error */
         if (!(res= thd->is_error()))
           my_ok(thd);                           // CREATE ... IF NOT EXISTS
+        goto end_with_restore_list;
       }
-      else
+
+      /* Ensure we don't try to create something from which we select from */
+      if ((create_info.options & HA_LEX_CREATE_REPLACE) &&
+          !create_info.tmp_table())
       {
-        /* The table already exists */
-        if (create_table->table)
+        TABLE_LIST *duplicate;
+        if ((duplicate= unique_table(thd, lex->query_tables,
+                                     lex->query_tables->next_global,
+                                     0)))
         {
-          if (create_info.options & HA_LEX_CREATE_IF_NOT_EXISTS)
-          {
-            push_warning_printf(thd, Sql_condition::WARN_LEVEL_NOTE,
-                                ER_TABLE_EXISTS_ERROR,
-                                ER(ER_TABLE_EXISTS_ERROR),
-                                create_info.alias);
-            my_ok(thd);
-          }
-          else
-          {
-            my_error(ER_TABLE_EXISTS_ERROR, MYF(0), create_info.alias);
-            res= 1;
-          }
+          update_non_unique_table_error(lex->query_tables, "CREATE",
+                                        duplicate);
+          res= TRUE;
           goto end_with_restore_list;
         }
-
+      }
+      {
         /*
           Remove target table from main select and name resolution
           context. This can't be done earlier as it will break view merging in
@@ -2980,9 +3010,8 @@ case SQLCOM_PREPARE:
         */
         lex->unlink_first_table(&link_to_local);
 
-        /* So that CREATE TEMPORARY TABLE gets to binlog at commit/rollback */
-        if (create_info.tmp_table())
-          thd->variables.option_bits|= OPTION_KEEP_LOG;
+        /* Store reference to table in case of LOCK TABLES */
+        create_info.table= create_table->table;
 
         /*
           select_create is currently not re-execution friendly and
@@ -3000,18 +3029,18 @@ case SQLCOM_PREPARE:
             CREATE from SELECT give its SELECT_LEX for SELECT,
             and item_list belong to SELECT
           */
-          res= handle_select(thd, lex, result, 0);
+          if (!(res= handle_select(thd, lex, result, 0)))
+          {
+            if (create_info.tmp_table())
+              thd->variables.option_bits|= OPTION_KEEP_LOG;
+          }
           delete result;
         }
-
         lex->link_first_table_back(create_table, link_to_local);
       }
     }
     else
     {
-      /* So that CREATE TEMPORARY TABLE gets to binlog at commit/rollback */
-      if (create_info.tmp_table())
-        thd->variables.option_bits|= OPTION_KEEP_LOG;
       /* regular create */
       if (create_info.options & HA_LEX_CREATE_TABLE_LIKE)
       {
@@ -3026,7 +3055,12 @@ case SQLCOM_PREPARE:
                                 &create_info, &alter_info);
       }
       if (!res)
+      {
+        /* So that CREATE TEMPORARY TABLE gets to binlog at commit/rollback */
+        if (create_info.tmp_table())
+          thd->variables.option_bits|= OPTION_KEEP_LOG;
         my_ok(thd);
+      }
     }
 
 end_with_restore_list:
@@ -3652,6 +3686,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);
@@ -4402,6 +4446,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();
@@ -5153,12 +5198,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())
   {
@@ -6101,6 +6149,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.
@@ -7961,6 +8011,11 @@ bool create_table_precheck(THD *thd, TABLE_LIST *tables,
   want_priv= lex->create_info.tmp_table() ?  CREATE_TMP_ACL :
              (CREATE_ACL | (select_lex->item_list.elements ? INSERT_ACL : 0));
 
+  /* CREATE OR REPLACE on not temporary tables require DROP_ACL */
+  if ((lex->create_info.options & HA_LEX_CREATE_REPLACE) &&
+      !lex->create_info.tmp_table())
+    want_priv|= DROP_ACL;
+                          
   if (check_access(thd, want_priv, create_table->db,
                    &create_table->grant.privilege,
                    &create_table->grant.m_internal,
@@ -7997,8 +8052,8 @@ bool create_table_precheck(THD *thd, TABLE_LIST *tables,
 
         - For temporary MERGE tables we do not track if their child tables are
           base or temporary. As result we can't guarantee that privilege check
-          which was done in presence of temporary child will stay relevant later
-          as this temporary table might be removed.
+          which was done in presence of temporary child will stay relevant
+          later as this temporary table might be removed.
 
       If SELECT_ACL | UPDATE_ACL | DELETE_ACL privileges were not checked for
       the underlying *base* tables, it would create a security breach as in
@@ -8029,6 +8084,12 @@ bool create_table_precheck(THD *thd, TABLE_LIST *tables,
   }
   error= FALSE;
 
+  /*
+    For CREATE TABLE we should not open the table even if it exists.
+    If the table exists, we should either not create it or replace it
+  */
+  lex->query_tables->open_strategy= TABLE_LIST::OPEN_STUB;
+
 err:
   DBUG_RETURN(error);
 }
diff --git a/sql/sql_partition.cc b/sql/sql_partition.cc
index 5e256522119..782a49af813 100644
--- a/sql/sql_partition.cc
+++ b/sql/sql_partition.cc
@@ -651,7 +651,7 @@ static bool create_full_part_field_array(THD *thd, TABLE *table,
     result= TRUE;
     goto end;
   }
-  if (bitmap_init(&part_info->full_part_field_set, bitmap_buf,
+  if (my_bitmap_init(&part_info->full_part_field_set, bitmap_buf,
                   table->s->fields, FALSE))
   {
     mem_alloc_error(table->s->fields);
@@ -1230,9 +1230,9 @@ static bool set_up_partition_bitmaps(THD *thd, partition_info *part_info)
     mem_alloc_error(bitmap_bytes * 2);
     DBUG_RETURN(TRUE);
   }
-  bitmap_init(&part_info->read_partitions, bitmap_buf, bitmap_bits, FALSE);
+  my_bitmap_init(&part_info->read_partitions, bitmap_buf, bitmap_bits, FALSE);
   /* Use the second half of the allocated buffer for lock_partitions */
-  bitmap_init(&part_info->lock_partitions, bitmap_buf + (bitmap_bytes / 4),
+  my_bitmap_init(&part_info->lock_partitions, bitmap_buf + (bitmap_bytes / 4),
               bitmap_bits, FALSE);
   part_info->bitmaps_are_initialized= TRUE;
   part_info->set_partition_bitmaps(NULL);
diff --git a/sql/sql_partition_admin.cc b/sql/sql_partition_admin.cc
index 29ca86fa274..b2a8bca72db 100644
--- a/sql/sql_partition_admin.cc
+++ b/sql/sql_partition_admin.cc
@@ -506,13 +506,8 @@ bool Sql_cmd_alter_table_exchange_partition::
 
   /* Don't allow to exchange with log table */
   swap_table_list= table_list->next_local;
-  if (check_if_log_table(swap_table_list->db_length, swap_table_list->db,
-                         swap_table_list->table_name_length,
-                         swap_table_list->table_name, 0))
-  {
-    my_error(ER_WRONG_USAGE, MYF(0), "PARTITION", "log table");
+  if (check_if_log_table(swap_table_list, FALSE, "ALTER PARTITION"))
     DBUG_RETURN(TRUE);
-  }
 
   /*
     Currently no MDL lock that allows both read and write and is upgradeable
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_rename.cc b/sql/sql_rename.cc
index 6babdd7c636..897aa183b60 100644
--- a/sql/sql_rename.cc
+++ b/sql/sql_rename.cc
@@ -84,12 +84,8 @@ bool mysql_rename_tables(THD *thd, TABLE_LIST *table_list, bool silent)
     for (to_table= 0, ren_table= table_list; ren_table;
          to_table= 1 - to_table, ren_table= ren_table->next_local)
     {
-      int log_table_rename= 0;
-
-      if ((log_table_rename=
-           check_if_log_table(ren_table->db_length, ren_table->db,
-                              ren_table->table_name_length,
-                              ren_table->table_name, 1)))
+      int log_table_rename;
+      if ((log_table_rename= check_if_log_table(ren_table, TRUE, NullS)))
       {
         /*
           as we use log_table_rename as an array index, we need it to start
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index dd23400f387..332230e67dc 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -8252,7 +8252,8 @@ get_best_combination(JOIN *join)
            sub-order
       */
       SJ_MATERIALIZATION_INFO *sjm= cur_pos->table->emb_sj_nest->sj_mat_info;
-      j->records= j->records_read= (ha_rows)(sjm->is_sj_scan? sjm->rows : 1);
+      j->records_read= (sjm->is_sj_scan? sjm->rows : 1);
+      j->records= (ha_rows) j->records_read;
       j->cond_selectivity= 1.0;
       JOIN_TAB *jt;
       JOIN_TAB_RANGE *jt_range;
@@ -11152,7 +11153,7 @@ ha_rows JOIN_TAB::get_examined_rows()
   else
     examined_rows= records_read;
 
-  return examined_rows;
+  return (ha_rows) examined_rows;
 }
 
 
@@ -15354,18 +15355,18 @@ Field *create_tmp_field(THD *thd, TABLE *table,Item *item, Item::Type type,
 void setup_tmp_table_column_bitmaps(TABLE *table, uchar *bitmaps)
 {
   uint field_count= table->s->fields;
-  bitmap_init(&table->def_read_set, (my_bitmap_map*) bitmaps, field_count,
+  my_bitmap_init(&table->def_read_set, (my_bitmap_map*) bitmaps, field_count,
               FALSE);
-  bitmap_init(&table->def_vcol_set,
+  my_bitmap_init(&table->def_vcol_set,
               (my_bitmap_map*) (bitmaps+ bitmap_buffer_size(field_count)),
               field_count, FALSE);
-  bitmap_init(&table->tmp_set,
+  my_bitmap_init(&table->tmp_set,
               (my_bitmap_map*) (bitmaps+ 2*bitmap_buffer_size(field_count)),
               field_count, FALSE);
-  bitmap_init(&table->eq_join_set,
+  my_bitmap_init(&table->eq_join_set,
               (my_bitmap_map*) (bitmaps+ 3*bitmap_buffer_size(field_count)),
               field_count, FALSE);
-  bitmap_init(&table->cond_set,
+  my_bitmap_init(&table->cond_set,
               (my_bitmap_map*) (bitmaps+ 4*bitmap_buffer_size(field_count)),
               field_count, FALSE);
   /* write_set and all_set are copies of read_set */
@@ -23306,7 +23307,7 @@ int JOIN::save_explain_data_intern(Explain_query *output, bool need_tmp_table,
         double examined_rows= tab->get_examined_rows();
 
         eta->rows_set= true;
-        eta->rows= examined_rows;
+        eta->rows= (ha_rows) examined_rows;
 
         /* "filtered"  */
         float f= 0.0; 
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index 2ab34ece903..738384d599f 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 "));
@@ -7681,7 +7686,7 @@ TABLE *create_schema_table(THD *thd, TABLE_LIST *table_list)
     DBUG_RETURN(0);
   my_bitmap_map* bitmaps=
     (my_bitmap_map*) thd->alloc(bitmap_buffer_size(field_count));
-  bitmap_init(&table->def_read_set, (my_bitmap_map*) bitmaps, field_count,
+  my_bitmap_init(&table->def_read_set, (my_bitmap_map*) bitmaps, field_count,
               FALSE);
   table->read_set= &table->def_read_set;
   bitmap_clear_all(table->read_set);
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 68c890d41da..c94343cadfd 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -2031,33 +2031,29 @@ bool mysql_rm_table(THD *thd,TABLE_LIST *tables, my_bool if_exists,
   bool error;
   Drop_table_error_handler err_handler;
   TABLE_LIST *table;
-
   DBUG_ENTER("mysql_rm_table");
 
   /* Disable drop of enabled log tables, must be done before name locking */
   for (table= tables; table; table= table->next_local)
   {
-    if (check_if_log_table(table->db_length, table->db,
-                           table->table_name_length, table->table_name, true))
-    {
-      my_error(ER_BAD_LOG_STATEMENT, MYF(0), "DROP");
+    if (check_if_log_table(table, TRUE, "DROP"))
       DBUG_RETURN(true);
-    }
   }
 
-  if (!in_bootstrap)
+  if (!drop_temporary)
   {
-    for (table= tables; table; table= table->next_local)
+    if (!in_bootstrap)
     {
-      LEX_STRING db_name= { table->db, table->db_length };
-      LEX_STRING table_name= { table->table_name, table->table_name_length };
-      if (table->open_type == OT_BASE_ONLY || !find_temporary_table(thd, table))
-        (void) delete_statistics_for_table(thd, &db_name, &table_name);
+      for (table= tables; table; table= table->next_local)
+      {
+        LEX_STRING db_name= { table->db, table->db_length };
+        LEX_STRING table_name= { table->table_name, table->table_name_length };
+        if (table->open_type == OT_BASE_ONLY ||
+            !find_temporary_table(thd, table))
+          (void) delete_statistics_for_table(thd, &db_name, &table_name);
+      }
     }
-  }
 
-  if (!drop_temporary)
-  {
     if (!thd->locked_tables_mode)
     {
       if (lock_table_names(thd, tables, NULL,
@@ -2107,7 +2103,7 @@ bool mysql_rm_table(THD *thd,TABLE_LIST *tables, my_bool if_exists,
   /* mark for close and remove all cached entries */
   thd->push_internal_handler(&err_handler);
   error= mysql_rm_table_no_locks(thd, tables, if_exists, drop_temporary,
-                                 false, false);
+                                 false, false, false);
   thd->pop_internal_handler();
 
   if (error)
@@ -2172,6 +2168,8 @@ static uint32 comment_length(THD *thd, uint32 comment_pos,
   @param  drop_view       Allow to delete VIEW .frm
   @param  dont_log_query  Don't write query to log files. This will also not
                           generate warnings if the handler files doesn't exists
+  @param  dont_free_locks Don't do automatic UNLOCK TABLE if no more locked
+                          tables
 
   @retval  0  ok
   @retval  1  Error
@@ -2194,7 +2192,8 @@ static uint32 comment_length(THD *thd, uint32 comment_pos,
 
 int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
                             bool drop_temporary, bool drop_view,
-                            bool dont_log_query)
+                            bool dont_log_query,
+                            bool dont_free_locks)
 {
   TABLE_LIST *table;
   char path[FN_REFLEN + 1], wrong_tables_buff[160], *alias= NULL;
@@ -2208,6 +2207,8 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
   bool trans_tmp_table_deleted= 0, non_trans_tmp_table_deleted= 0;
   bool non_tmp_table_deleted= 0;
   bool is_drop_tmp_if_exists_added= 0;
+  bool one_table= tables->next_local == 0;
+  bool was_view= 0;
   String built_query;
   String built_trans_tmp_query, built_non_trans_tmp_query;
   DBUG_ENTER("mysql_rm_table_no_locks");
@@ -2286,7 +2287,7 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
 
   for (table= tables; table; table= table->next_local)
   {
-    bool is_trans;
+    bool is_trans= 0;
     char *db=table->db;
     size_t db_length= table->db_length;
     handlerton *table_type= 0;
@@ -2311,12 +2312,16 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
       .  1 - a temporary table was not found.
       . -1 - a temporary table is used by an outer statement.
     */
-    if (table->open_type == OT_BASE_ONLY)
+    if (table->open_type == OT_BASE_ONLY || !is_temporary_table(table))
       error= 1;
-    else if ((error= drop_temporary_table(thd, table, &is_trans)) == -1)
+    else
     {
-      DBUG_ASSERT(thd->in_sub_stmt);
-      goto err;
+      if ((error= drop_temporary_table(thd, table->table, &is_trans)) == -1)
+      {
+        DBUG_ASSERT(thd->in_sub_stmt);
+        goto err;
+      }
+      table->table= 0;
     }
 
     if ((drop_temporary && if_exists) || !error)
@@ -2413,7 +2418,7 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
     DEBUG_SYNC(thd, "rm_table_no_locks_before_delete_table");
     error= 0;
     if ((drop_temporary || !ha_table_exists(thd, db, alias, &table_type) ||
-         (!drop_view && table_type == view_pseudo_hton)))
+         (!drop_view && (was_view= (table_type == view_pseudo_hton)))))
     {
       /*
         One of the following cases happened:
@@ -2544,7 +2549,10 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
 err:
   if (wrong_tables.length())
   {
-    if (!foreign_key_error)
+    if (one_table && was_view)
+      my_printf_error(ER_IT_IS_A_VIEW, ER(ER_IT_IS_A_VIEW), MYF(0),
+                      wrong_tables.c_ptr_safe());
+    else if (!foreign_key_error)
       my_printf_error(ER_BAD_TABLE_ERROR, ER(ER_BAD_TABLE_ERROR), MYF(0),
                       wrong_tables.c_ptr_safe());
     else
@@ -2610,7 +2618,8 @@ err:
     */
     if (thd->locked_tables_mode)
     {
-      if (thd->lock && thd->lock->table_count == 0 && non_temp_tables_count > 0)
+      if (thd->lock && thd->lock->table_count == 0 &&
+          non_temp_tables_count > 0 && !dont_free_locks)
       {
         thd->locked_tables_list.unlock_locked_tables(thd);
         goto end;
@@ -4517,12 +4526,13 @@ err:
   way to ensure that concurrent operations won't intervene.
   mysql_create_table() is a wrapper that can be used for this.
 
-  @retval false OK
-  @retval true  error
+  @retval 0 OK
+  @retval 1 error
+  @retval -1 table existed but IF EXISTS was used
 */
 
 static
-bool create_table_impl(THD *thd,
+int create_table_impl(THD *thd,
                        const char *db, const char *table_name,
                        const char *path,
                        HA_CREATE_INFO *create_info,
@@ -4535,7 +4545,7 @@ bool create_table_impl(THD *thd,
 {
   const char	*alias;
   handler	*file= 0;
-  bool		error= TRUE;
+  int		error= 1;
   bool          frm_only= create_table_mode == C_ALTER_TABLE_FRM_ONLY;
   bool          internal_tmp_table= create_table_mode == C_ALTER_TABLE || frm_only;
   DBUG_ENTER("mysql_create_table_no_lock");
@@ -4565,22 +4575,74 @@ bool create_table_impl(THD *thd,
   /* Check if table exists */
   if (create_info->tmp_table())
   {
-    if (find_temporary_table(thd, db, table_name))
+    TABLE *tmp_table;
+    if ((tmp_table= find_temporary_table(thd, db, table_name)))
     {
-      if (create_info->options & HA_LEX_CREATE_IF_NOT_EXISTS)
+      if (create_info->options & HA_LEX_CREATE_REPLACE)
+      {
+        bool is_trans;
+        /*
+          We are using CREATE OR REPLACE on an existing temporary table
+          Remove the old table so that we can re-create it.
+        */
+        if (drop_temporary_table(thd, tmp_table, &is_trans))
+          goto err;
+      }
+      else if (create_info->options & HA_LEX_CREATE_IF_NOT_EXISTS)
         goto warn;
-      my_error(ER_TABLE_EXISTS_ERROR, MYF(0), alias);
-      goto err;
+      else
+      {
+        my_error(ER_TABLE_EXISTS_ERROR, MYF(0), alias);
+        goto err;
+      }
     }
   }
-  else  
+  else
   {
     if (!internal_tmp_table && ha_table_exists(thd, db, table_name))
     {
-      if (create_info->options & HA_LEX_CREATE_IF_NOT_EXISTS)
+      if (create_info->options & HA_LEX_CREATE_REPLACE)
+      {
+        TABLE_LIST table_list;
+        table_list.init_one_table(db, strlen(db), table_name,
+                                  strlen(table_name), table_name,
+                                  TL_WRITE_ALLOW_WRITE);
+        table_list.table= create_info->table;
+
+        if (check_if_log_table(&table_list, TRUE, "CREATE OR REPLACE"))
+          goto err;
+        
+        /*
+          Rollback the empty transaction started in mysql_create_table()
+          call to open_and_lock_tables() when we are using LOCK TABLES.
+        */
+        (void) trans_rollback_stmt(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
+        */
+        if (thd->lex->query_tables &&
+            restart_trans_for_tables(thd, thd->lex->query_tables->next_global))
+          goto err;
+      }
+      else if (create_info->options & HA_LEX_CREATE_IF_NOT_EXISTS)
         goto warn;
-      my_error(ER_TABLE_EXISTS_ERROR, MYF(0), table_name);
-      goto err;
+      else
+      {
+        my_error(ER_TABLE_EXISTS_ERROR, MYF(0), table_name);
+        goto err;
+      }
     }
   }
 
@@ -4702,14 +4764,14 @@ bool create_table_impl(THD *thd,
   }
 #endif
 
-  error= FALSE;
+  error= 0;
 err:
   THD_STAGE_INFO(thd, stage_after_create);
   delete file;
   DBUG_RETURN(error);
 
 warn:
-  error= FALSE;
+  error= -1;
   push_warning_printf(thd, Sql_condition::WARN_LEVEL_NOTE,
                       ER_TABLE_EXISTS_ERROR, ER(ER_TABLE_EXISTS_ERROR),
                       alias);
@@ -4720,7 +4782,8 @@ warn:
   Simple wrapper around create_table_impl() to be used
   in various version of CREATE TABLE statement.
 */
-bool mysql_create_table_no_lock(THD *thd,
+
+int mysql_create_table_no_lock(THD *thd,
                                 const char *db, const char *table_name,
                                 HA_CREATE_INFO *create_info,
                                 Alter_info *alter_info, bool *is_trans,
@@ -4728,6 +4791,7 @@ bool mysql_create_table_no_lock(THD *thd,
 {
   KEY *not_used_1;
   uint not_used_2;
+  int res;
   char path[FN_REFLEN + 1];
   LEX_CUSTRING frm= {0,0};
 
@@ -4747,9 +4811,9 @@ bool mysql_create_table_no_lock(THD *thd,
     }
   }
 
-  bool res= create_table_impl(thd, db, table_name, path, create_info,
-                              alter_info, create_table_mode, is_trans,
-                              &not_used_1, &not_used_2, &frm);
+  res= create_table_impl(thd, db, table_name, path, create_info,
+                         alter_info, create_table_mode, is_trans,
+                         &not_used_1, &not_used_2, &frm);
   my_free(const_cast<uchar*>(frm.str));
   return res;
 }
@@ -4771,16 +4835,23 @@ 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");
 
+  DBUG_ASSERT(create_table == thd->lex->query_tables);
+
   /* Open or obtain an exclusive metadata lock on table being created  */
   if (open_and_lock_tables(thd, thd->lex->query_tables, FALSE, 0))
   {
     /* is_error() may be 0 if table existed and we generated a warning */
     DBUG_RETURN(thd->is_error());
   }
-
+  /* The following is needed only in case of lock tables */
+  if ((create_info->table= thd->lex->query_tables->table))
+    pos_in_locked_tables= create_info->table->pos_in_locked_tables;
+  
   /* Got lock. */
   DEBUG_SYNC(thd, "locked_table_name");
 
@@ -4791,15 +4862,42 @@ 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))
-    DBUG_RETURN(1);
+                                 &is_trans, create_table_mode) > 0)
+  {
+    result= 1;
+    goto err;
+  }
+
+  /*
+    Check if we are doing CREATE OR REPLACE TABLE under LOCK TABLES
+    on a non temporary table
+  */
+  if (thd->locked_tables_mode && pos_in_locked_tables &&
+      (create_info->options & HA_LEX_CREATE_REPLACE))
+  {
+    /*
+      Add back the deleted table and re-created table as a locked table
+      This should always work as we have a meta lock on the table.
+     */
+    thd->locked_tables_list.add_back_last_deleted_lock(pos_in_locked_tables);
+    if (thd->locked_tables_list.reopen_tables(thd))
+      thd->locked_tables_list.unlink_all_closed_tables(thd, NULL, 0);
+    else
+    {
+      TABLE *table= pos_in_locked_tables->table;
+      table->mdl_ticket->downgrade_lock(MDL_SHARED_NO_READ_WRITE);
+    }
+  }
 
+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);
 }
 
@@ -4986,18 +5084,20 @@ mysql_rename_table(handlerton *base, const char *old_db,
     TRUE  error
 */
 
-bool mysql_create_like_table(THD* thd, TABLE_LIST* table, TABLE_LIST* src_table,
+bool mysql_create_like_table(THD* thd, TABLE_LIST* table,
+                             TABLE_LIST* src_table,
                              HA_CREATE_INFO *create_info)
 {
   HA_CREATE_INFO local_create_info;
+  TABLE_LIST *pos_in_locked_tables= 0;
   Alter_info local_alter_info;
   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");
 
-
   /*
     We the open source table to get its description in HA_CREATE_INFO
     and Alter_info objects. This also acquires a shared metadata lock
@@ -5014,6 +5114,18 @@ bool mysql_create_like_table(THD* thd, TABLE_LIST* table, TABLE_LIST* src_table,
     res= thd->is_error();
     goto err;
   }
+  /* Ensure we don't try to create something from which we select from */
+  if ((create_info->options & HA_LEX_CREATE_REPLACE) &&
+      !create_info->tmp_table())
+  {
+    TABLE_LIST *duplicate;
+    if ((duplicate= unique_table(thd, table, src_table, 0)))
+    {
+      update_non_unique_table_error(src_table, "CREATE", duplicate);
+      goto err;
+    }
+  }
+
   src_table->table->use_all_columns();
 
   DEBUG_SYNC(thd, "create_table_like_after_open");
@@ -5041,7 +5153,9 @@ bool mysql_create_like_table(THD* thd, TABLE_LIST* table, TABLE_LIST* src_table,
   if (src_table->schema_table)
     local_create_info.max_rows= 0;
   /* Set IF NOT EXISTS option as in the CREATE TABLE LIKE statement. */
-  local_create_info.options|= create_info->options&HA_LEX_CREATE_IF_NOT_EXISTS;
+  local_create_info.options|= (create_info->options &
+                               (HA_LEX_CREATE_IF_NOT_EXISTS | 
+                                HA_LEX_CREATE_REPLACE));
   /* Replace type of source table with one specified in the statement. */
   local_create_info.options&= ~HA_LEX_CREATE_TMP_TABLE;
   local_create_info.options|= create_info->tmp_table();
@@ -5053,19 +5167,53 @@ bool mysql_create_like_table(THD* thd, TABLE_LIST* table, TABLE_LIST* src_table,
   */
   local_create_info.data_file_name= local_create_info.index_file_name= NULL;
 
-  if ((res= mysql_create_table_no_lock(thd, table->db, table->table_name,
-                                       &local_create_info, &local_alter_info,
-                                       &is_trans, C_ORDINARY_CREATE)))
+  /* The following is needed only in case of lock tables */
+  if ((local_create_info.table= thd->lex->query_tables->table))
+    pos_in_locked_tables= local_create_info.table->pos_in_locked_tables;    
+
+  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;
 
   /*
-    Ensure that we have an exclusive lock on target table if we are creating
-    non-temporary table.
+    Check if we are doing CREATE OR REPLACE TABLE under LOCK TABLES
+    on a non temporary table
   */
-  DBUG_ASSERT((create_info->tmp_table()) ||
-              thd->mdl_context.is_lock_owner(MDL_key::TABLE, table->db,
-                                             table->table_name,
-                                             MDL_EXCLUSIVE));
+  if (thd->locked_tables_mode && pos_in_locked_tables &&
+      (create_info->options & HA_LEX_CREATE_REPLACE))
+  {
+    /*
+      Add back the deleted table and re-created table as a locked table
+      This should always work as we have a meta lock on the table.
+     */
+    thd->locked_tables_list.add_back_last_deleted_lock(pos_in_locked_tables);
+    if (thd->locked_tables_list.reopen_tables(thd))
+      thd->locked_tables_list.unlink_all_closed_tables(thd, NULL, 0);
+    else
+    {
+      /*
+        Get pointer to the newly opened table. We need this to ensure we
+        don't reopen the table when doing statment logging below.
+      */
+      table->table= pos_in_locked_tables->table;
+      table->table->mdl_ticket->downgrade_lock(MDL_SHARED_NO_READ_WRITE);
+    }
+  }
+  else
+  {
+    /*
+      Ensure that we have an exclusive lock on target table if we are creating
+      non-temporary table.
+    */
+    DBUG_ASSERT((create_info->tmp_table()) ||
+                thd->mdl_context.is_lock_owner(MDL_key::TABLE, table->db,
+                                               table->table_name,
+                                               MDL_EXCLUSIVE));
+  }
 
   DEBUG_SYNC(thd, "create_table_like_before_binlog");
 
@@ -5108,6 +5256,11 @@ bool mysql_create_like_table(THD* thd, TABLE_LIST* table, TABLE_LIST* src_table,
         {
           if (!table->table)
           {
+            TABLE_LIST::enum_open_strategy save_open_strategy;
+            int open_res;
+            /* Force the newly created table to be opened */
+            save_open_strategy= table->open_strategy;
+            table->open_strategy= TABLE_LIST::OPEN_NORMAL;
 
             /*
               In order for store_create_info() to work we need to open
@@ -5117,18 +5270,36 @@ bool mysql_create_like_table(THD* thd, TABLE_LIST* table, TABLE_LIST* src_table,
               lock on this table. The table will be closed by
               close_thread_table() at the end of this branch.
             */
-            if (open_table(thd, table, thd->mem_root, &ot_ctx))
+            open_res= open_table(thd, table, thd->mem_root, &ot_ctx);
+            /* Restore */
+            table->open_strategy= save_open_strategy;
+            if (open_res)
+            {
+              res= 1;
               goto err;
+            }
             new_table= TRUE;
           }
-
+        }
+        /*
+          We have to re-test if the table was a view as the view may not
+          have been opened until just above.
+        */
+        if (!table->view)
+        {
           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)
           {
@@ -5143,17 +5314,20 @@ bool mysql_create_like_table(THD* thd, TABLE_LIST* table, TABLE_LIST* src_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);
 }
 
@@ -7772,9 +7946,7 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
     it is the case.
     TODO: this design is obsolete and will be removed.
   */
-  int table_kind= check_if_log_table(table_list->db_length, table_list->db,
-                                     table_list->table_name_length,
-                                     table_list->table_name, false);
+  int table_kind= check_if_log_table(table_list, FALSE, NullS);
 
   if (table_kind)
   {
diff --git a/sql/sql_table.h b/sql/sql_table.h
index c42f8aaa39e..fc7eb775bef 100644
--- a/sql/sql_table.h
+++ b/sql/sql_table.h
@@ -187,11 +187,11 @@ bool mysql_create_table(THD *thd, TABLE_LIST *create_table,
 #define C_ALTER_TABLE_FRM_ONLY   -2
 #define C_ASSISTED_DISCOVERY     -3
 
-bool mysql_create_table_no_lock(THD *thd, const char *db,
-                                const char *table_name,
-                                HA_CREATE_INFO *create_info,
-                                Alter_info *alter_info, bool *is_trans,
-                                int create_table_mode);
+int mysql_create_table_no_lock(THD *thd, const char *db,
+                               const char *table_name,
+                               HA_CREATE_INFO *create_info,
+                               Alter_info *alter_info, bool *is_trans,
+                               int create_table_mode);
 
 handler *mysql_create_frm_image(THD *thd,
                                 const char *db, const char *table_name,
@@ -238,7 +238,7 @@ bool mysql_rm_table(THD *thd,TABLE_LIST *tables, my_bool if_exists,
                     my_bool drop_temporary);
 int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
                             bool drop_temporary, bool drop_view,
-                            bool log_query);
+                            bool log_query, bool dont_free_locks);
 bool quick_rm_table(THD *thd, handlerton *base, const char *db,
                     const char *table_name, uint flags);
 void close_cached_table(THD *thd, TABLE *table);
diff --git a/sql/sql_update.cc b/sql/sql_update.cc
index 889eaf5fceb..ccdc91e10df 100644
--- a/sql/sql_update.cc
+++ b/sql/sql_update.cc
@@ -191,7 +191,7 @@ static void prepare_record_for_error_message(int error, TABLE *table)
     DBUG_VOID_RETURN;
 
   /* Create unique_map with all fields used by that index. */
-  bitmap_init(&unique_map, unique_map_buf, table->s->fields, FALSE);
+  my_bitmap_init(&unique_map, unique_map_buf, table->s->fields, FALSE);
   table->mark_columns_used_by_index_no_reset(keynr, &unique_map);
 
   /* Subtract read_set and write_set. */
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index 3315c2d4d17..b206874c68d 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -1665,7 +1665,7 @@ bool my_yyoverflow(short **a, YYSTYPE **b, ulong *yystacksize);
 %type <num>
         type type_with_opt_collate int_type real_type order_dir lock_option
         udf_type opt_if_exists opt_local opt_table_options table_options
-        table_option opt_if_not_exists opt_no_write_to_binlog
+        table_option opt_if_not_exists create_or_replace opt_no_write_to_binlog
         opt_temporary all_or_any opt_distinct
         opt_ignore_leaves fulltext_options spatial_type union_option
         field_def
@@ -1844,7 +1844,6 @@ bool my_yyoverflow(short **a, YYSTYPE **b, ulong *yystacksize);
         statement sp_suid
         sp_c_chistics sp_a_chistics sp_chistic sp_c_chistic xa
         opt_field_or_var_spec fields_or_vars opt_load_data_set_spec
-        view_replace_or_algorithm view_replace
         view_algorithm view_or_trigger_or_sp_or_event
         definer_tail no_definer_tail
         view_suid view_tail view_list_opt view_list view_select
@@ -2342,25 +2341,29 @@ connection_name:
 /* create a table */
 
 create:
-          CREATE opt_table_options TABLE_SYM opt_if_not_exists table_ident
+          create_or_replace opt_table_options TABLE_SYM opt_if_not_exists table_ident
           {
             LEX *lex= thd->lex;
             lex->sql_command= SQLCOM_CREATE_TABLE;
+            if ($1 && $4)
+            {
+               my_error(ER_WRONG_USAGE, MYF(0), "OR REPLACE", "IF NOT EXISTS");
+               MYSQL_YYABORT;
+            }
             if (!lex->select_lex.add_table_to_list(thd, $5, NULL,
                                                    TL_OPTION_UPDATING,
                                                    TL_WRITE, MDL_EXCLUSIVE))
               MYSQL_YYABORT;
-            /*
-              For CREATE TABLE, an non-existing table is not an error.
-              Instruct open_tables() to just take an MDL lock if the
-              table does not exist.
-            */
-            lex->query_tables->open_strategy= TABLE_LIST::OPEN_IF_EXISTS;
             lex->alter_info.reset();
             lex->col_list.empty();
             lex->change=NullS;
             bzero((char*) &lex->create_info,sizeof(lex->create_info));
-            lex->create_info.options=$2 | $4;
+            /*
+              For CREATE TABLE we should not open the table even if it exists.
+              If the table exists, we should either not create it or replace it
+            */
+            lex->query_tables->open_strategy= TABLE_LIST::OPEN_STUB;
+            lex->create_info.options= ($1 | $2 | $4);
             lex->create_info.default_table_charset= NULL;
             lex->name.str= 0;
             lex->name.length= 0;
@@ -2429,14 +2432,22 @@ create:
             lex->name= $4;
             lex->create_info.options=$3;
           }
-        | CREATE
+        | create_or_replace
           {
-            Lex->create_view_mode= VIEW_CREATE_NEW;
+            Lex->create_view_mode= ($1 == 0 ? VIEW_CREATE_NEW :
+                                    VIEW_CREATE_OR_REPLACE);
             Lex->create_view_algorithm= DTYPE_ALGORITHM_UNDEFINED;
             Lex->create_view_suid= TRUE;
           }
           view_or_trigger_or_sp_or_event
-          {}
+          {
+            if ($1 && Lex->sql_command != SQLCOM_CREATE_VIEW)
+            {
+               my_error(ER_WRONG_USAGE, MYF(0), "OR REPLACE",
+                       "TRIGGERS / SP / EVENT");
+               MYSQL_YYABORT;
+            }
+          }
         | CREATE USER clear_privileges grant_list
           {
             Lex->sql_command = SQLCOM_CREATE_USER;
@@ -5516,6 +5527,17 @@ opt_if_not_exists:
           }
          ;
 
+create_or_replace:
+          CREATE /* empty */
+          {
+            $$= 0;
+          }
+        | CREATE OR_SYM REPLACE
+          {
+            $$= HA_LEX_CREATE_REPLACE;
+          }
+         ;
+
 opt_create_table_options:
           /* empty */
         | create_table_options
@@ -15821,7 +15843,7 @@ view_or_trigger_or_sp_or_event:
           {}
         | no_definer no_definer_tail
           {}
-        | view_replace_or_algorithm definer_opt view_tail
+        | view_algorithm definer_opt view_tail
           {}
         ;
 
@@ -15880,20 +15902,6 @@ definer:
 
 **************************************************************************/
 
-view_replace_or_algorithm:
-          view_replace
-          {}
-        | view_replace view_algorithm
-          {}
-        | view_algorithm
-          {}
-        ;
-
-view_replace:
-          OR_SYM REPLACE
-          { Lex->create_view_mode= VIEW_CREATE_OR_REPLACE; }
-        ;
-
 view_algorithm:
           ALGORITHM_SYM EQ UNDEFINED_SYM
           { Lex->create_view_algorithm= DTYPE_ALGORITHM_UNDEFINED; }
diff --git a/sql/sys_vars.cc b/sql/sys_vars.cc
index 77f437243c8..d7c30f532e0 100644
--- a/sql/sys_vars.cc
+++ b/sql/sys_vars.cc
@@ -364,7 +364,7 @@ static Sys_var_ulong Sys_back_log(
        "MySQL can have. This comes into play when the main MySQL thread "
        "gets very many connection requests in a very short time",
        READ_ONLY GLOBAL_VAR(back_log), CMD_LINE(REQUIRED_ARG),
-       VALID_RANGE(1, 65535), DEFAULT(50), BLOCK_SIZE(1));
+       VALID_RANGE(1, 65535), DEFAULT(150), BLOCK_SIZE(1));
 
 static Sys_var_charptr Sys_basedir(
        "basedir", "Path to installation directory. All paths are "
@@ -593,7 +593,7 @@ static bool check_charset_db(sys_var *self, THD *thd, set_var *var)
 }
 static Sys_var_struct Sys_character_set_database(
        "character_set_database",
-       " The character set used by the default database",
+       "The character set used by the default database",
        SESSION_VAR(collation_database), NO_CMD_LINE,
        offsetof(CHARSET_INFO, csname), DEFAULT(&default_charset_info),
        NO_MUTEX_GUARD, IN_BINLOG, ON_CHECK(check_charset_db));
@@ -1062,7 +1062,7 @@ static Sys_var_keycache Sys_key_cache_age_threshold(
 static Sys_var_mybool Sys_large_files_support(
        "large_files_support",
        "Whether mysqld was compiled with options for large file support",
-       READ_ONLY GLOBAL_VAR(opt_large_files),
+       READ_ONLY SHOW_VALUE_IN_HELP GLOBAL_VAR(opt_large_files),
        NO_CMD_LINE, DEFAULT(sizeof(my_off_t) > 4));
 
 static Sys_var_uint Sys_large_page_size(
@@ -1184,7 +1184,8 @@ static Sys_var_mybool Sys_lower_case_file_system(
        "lower_case_file_system",
        "Case sensitivity of file names on the file system where the "
        "data directory is located",
-       READ_ONLY GLOBAL_VAR(lower_case_file_system), NO_CMD_LINE,
+       READ_ONLY SHOW_VALUE_IN_HELP GLOBAL_VAR(lower_case_file_system),
+       NO_CMD_LINE,
        DEFAULT(FALSE));
 
 static Sys_var_uint Sys_lower_case_table_names(
@@ -2145,7 +2146,7 @@ static Sys_var_ulong Sys_preload_buff_size(
 static Sys_var_uint Sys_protocol_version(
        "protocol_version",
        "The version of the client/server protocol used by the MySQL server",
-       READ_ONLY GLOBAL_VAR(protocol_version), NO_CMD_LINE,
+       READ_ONLY SHOW_VALUE_IN_HELP GLOBAL_VAR(protocol_version), NO_CMD_LINE,
        VALID_RANGE(0, ~0), DEFAULT(PROTOCOL_VERSION), BLOCK_SIZE(1));
 
 static Sys_var_proxy_user Sys_proxy_user(
@@ -2604,11 +2605,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",
@@ -2881,7 +2894,8 @@ static Sys_var_mybool Sys_sync_frm(
 static char *system_time_zone_ptr;
 static Sys_var_charptr Sys_system_time_zone(
        "system_time_zone", "The server system time zone",
-       READ_ONLY GLOBAL_VAR(system_time_zone_ptr), NO_CMD_LINE,
+       READ_ONLY SHOW_VALUE_IN_HELP GLOBAL_VAR(system_time_zone_ptr),
+       NO_CMD_LINE,
        IN_SYSTEM_CHARSET, DEFAULT(system_time_zone));
 
 static Sys_var_ulong Sys_table_def_size(
@@ -3091,27 +3105,37 @@ static Sys_var_mybool Sys_timed_mutexes(
 static char *server_version_ptr;
 static Sys_var_charptr Sys_version(
        "version", "Server version",
-       READ_ONLY GLOBAL_VAR(server_version_ptr), NO_CMD_LINE,
+       READ_ONLY SHOW_VALUE_IN_HELP GLOBAL_VAR(server_version_ptr),
+       NO_CMD_LINE,
        IN_SYSTEM_CHARSET, DEFAULT(server_version));
 
 static char *server_version_comment_ptr;
 static Sys_var_charptr Sys_version_comment(
        "version_comment", "version_comment",
-       READ_ONLY GLOBAL_VAR(server_version_comment_ptr), NO_CMD_LINE,
+       READ_ONLY SHOW_VALUE_IN_HELP GLOBAL_VAR(server_version_comment_ptr),
+       NO_CMD_LINE,
        IN_SYSTEM_CHARSET, DEFAULT(MYSQL_COMPILATION_COMMENT));
 
 static char *server_version_compile_machine_ptr;
 static Sys_var_charptr Sys_version_compile_machine(
        "version_compile_machine", "version_compile_machine",
-       READ_ONLY GLOBAL_VAR(server_version_compile_machine_ptr), NO_CMD_LINE,
+       READ_ONLY SHOW_VALUE_IN_HELP
+       GLOBAL_VAR(server_version_compile_machine_ptr), NO_CMD_LINE,
        IN_SYSTEM_CHARSET, DEFAULT(MACHINE_TYPE));
 
 static char *server_version_compile_os_ptr;
 static Sys_var_charptr Sys_version_compile_os(
        "version_compile_os", "version_compile_os",
-       READ_ONLY GLOBAL_VAR(server_version_compile_os_ptr), NO_CMD_LINE,
+       READ_ONLY SHOW_VALUE_IN_HELP GLOBAL_VAR(server_version_compile_os_ptr),
+       NO_CMD_LINE,
        IN_SYSTEM_CHARSET, DEFAULT(SYSTEM_TYPE));
 
+static char *malloc_library;
+static Sys_var_charptr Sys_malloc_library(
+       "version_malloc_library", "Version of the used malloc library",
+       READ_ONLY SHOW_VALUE_IN_HELP GLOBAL_VAR(malloc_library), NO_CMD_LINE,
+       IN_SYSTEM_CHARSET, DEFAULT(MALLOC_LIBRARY));
+
 static Sys_var_ulong Sys_net_wait_timeout(
        "wait_timeout",
        "The number of seconds the server waits for activity on a "
@@ -3190,10 +3214,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;
@@ -3210,16 +3234,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;
@@ -3228,6 +3253,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/sys_vars.h b/sql/sys_vars.h
index bef5fbdd126..6c0228fd6a0 100644
--- a/sql/sys_vars.h
+++ b/sql/sys_vars.h
@@ -56,6 +56,7 @@
 // this means that Sys_var_charptr initial value was malloc()ed
 #define PREALLOCATED sys_var::ALLOCATED+
 #define PARSED_EARLY sys_var::PARSE_EARLY+
+#define SHOW_VALUE_IN_HELP sys_var::SHOW_VALUE_IN_HELP+
 
 /*
   Sys_var_bit meaning is reversed, like in
diff --git a/sql/table.cc b/sql/table.cc
index 87cd2adc542..2e1d18b5da1 100644
--- a/sql/table.cc
+++ b/sql/table.cc
@@ -2010,7 +2010,7 @@ int TABLE_SHARE::init_from_binary_frm_image(THD *thd, bool write,
   if (!(bitmaps= (my_bitmap_map*) alloc_root(&share->mem_root,
                                              share->column_bitmap_size)))
     goto err;
-  bitmap_init(&share->all_set, bitmaps, share->fields, FALSE);
+  my_bitmap_init(&share->all_set, bitmaps, share->fields, FALSE);
   bitmap_set_all(&share->all_set);
 
   delete handler_file;
@@ -2818,17 +2818,17 @@ partititon_err:
   bitmap_size= share->column_bitmap_size;
   if (!(bitmaps= (uchar*) alloc_root(&outparam->mem_root, bitmap_size*6)))
     goto err;
-  bitmap_init(&outparam->def_read_set,
+  my_bitmap_init(&outparam->def_read_set,
               (my_bitmap_map*) bitmaps, share->fields, FALSE);
-  bitmap_init(&outparam->def_write_set,
+  my_bitmap_init(&outparam->def_write_set,
               (my_bitmap_map*) (bitmaps+bitmap_size), share->fields, FALSE);
-  bitmap_init(&outparam->def_vcol_set,
+  my_bitmap_init(&outparam->def_vcol_set,
               (my_bitmap_map*) (bitmaps+bitmap_size*2), share->fields, FALSE);
-  bitmap_init(&outparam->tmp_set,
+  my_bitmap_init(&outparam->tmp_set,
               (my_bitmap_map*) (bitmaps+bitmap_size*3), share->fields, FALSE);
-  bitmap_init(&outparam->eq_join_set,
+  my_bitmap_init(&outparam->eq_join_set,
               (my_bitmap_map*) (bitmaps+bitmap_size*4), share->fields, FALSE);
-  bitmap_init(&outparam->cond_set,
+  my_bitmap_init(&outparam->cond_set,
               (my_bitmap_map*) (bitmaps+bitmap_size*5), share->fields, FALSE);
   outparam->default_column_bitmaps();
 
diff --git a/sql/table.h b/sql/table.h
index 6d8be8f948a..260ff7cf6ba 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -1065,7 +1065,6 @@ public:
   ORDER		*group;
   String	alias;            	  /* alias or table name */
   uchar		*null_flags;
-  my_bitmap_map	*bitmap_init_value;
   MY_BITMAP     def_read_set, def_write_set, def_vcol_set, tmp_set; 
   MY_BITMAP     eq_join_set;         /* used to mark equi-joined fields */
   MY_BITMAP     cond_set;   /* used to mark fields from sargable conditions*/
@@ -1950,7 +1949,7 @@ struct TABLE_LIST
      Indicates that if TABLE_LIST object corresponds to the table/view
      which requires special handling.
   */
-  enum
+  enum enum_open_strategy
   {
     /* Normal open. */
     OPEN_NORMAL= 0,
diff --git a/sql/table_cache.cc b/sql/table_cache.cc
index 6f51ac8276c..6b24f4348ee 100644
--- a/sql/table_cache.cc
+++ b/sql/table_cache.cc
@@ -196,6 +196,7 @@ static void check_unused(THD *thd)
   TABLE *entry;
   TABLE_SHARE *share;
   TDC_iterator tdc_it;
+  DBUG_ENTER("check_unused");
 
   tdc_it.init();
   mysql_mutex_lock(&LOCK_open);
@@ -221,6 +222,7 @@ static void check_unused(THD *thd)
   }
   mysql_mutex_unlock(&LOCK_open);
   tdc_it.deinit();
+  DBUG_VOID_RETURN;
 }
 #else
 #define check_unused(A)
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;