Merge 10.4 into 10.5

46 files changed, 486 insertions, 220 deletions

diff --git a/sql/backup.cc b/sql/backup.cc
index 73cd13ffe2a..5976506cf92 100644
--- a/sql/backup.cc
+++ b/sql/backup.cc
@@ -290,6 +290,14 @@ static bool backup_block_commit(THD *thd)
 
   /* We can ignore errors from flush_tables () */
   (void) flush_tables(thd, FLUSH_SYS_TABLES);
+
+  if (mysql_bin_log.is_open())
+  {
+    mysql_mutex_lock(mysql_bin_log.get_log_lock());
+    mysql_file_sync(mysql_bin_log.get_log_file()->file,
+                    MYF(MY_WME|MY_SYNC_FILESIZE));
+    mysql_mutex_unlock(mysql_bin_log.get_log_lock());
+  }
   thd->clear_error();
 
   DBUG_RETURN(0);
diff --git a/sql/ha_partition.cc b/sql/ha_partition.cc
index 45ea9eb1552..10d97328c70 100644
--- a/sql/ha_partition.cc
+++ b/sql/ha_partition.cc
@@ -3461,8 +3461,7 @@ bool ha_partition::init_partition_bitmaps()
 
 /*
   Open handler object
-
-  SYNOPSIS
+SYNOPSIS
     open()
     name                  Full path of table name
     mode                  Open mode flags
@@ -3588,6 +3587,7 @@ int ha_partition::open(const char *name, int mode, uint test_if_locked)
   }
   else
   {
+    check_insert_autoincrement();
     if (unlikely((error= open_read_partitions(name_buff, sizeof(name_buff)))))
       goto err_handler;
     m_num_locks= m_file_sample->lock_count();
@@ -4473,11 +4473,8 @@ exit:
                     table->found_next_number_field->field_index))
   {
     update_next_auto_inc_val();
-    /*
-      The following call is safe as part_share->auto_inc_initialized
-      (tested in the call) is guaranteed to be set for update statements.
-    */
-    set_auto_increment_if_higher(table->found_next_number_field);
+    if (part_share->auto_inc_initialized)
+      set_auto_increment_if_higher(table->found_next_number_field);
   }
   DBUG_RETURN(error);
 }
@@ -8138,6 +8135,7 @@ int ha_partition::info(uint flag)
   if (flag & HA_STATUS_AUTO)
   {
     bool auto_inc_is_first_in_idx= (table_share->next_number_keypart == 0);
+    bool all_parts_opened= true;
     DBUG_PRINT("info", ("HA_STATUS_AUTO"));
     if (!table->found_next_number_field)
       stats.auto_increment_value= 0;
@@ -8168,6 +8166,15 @@ int ha_partition::info(uint flag)
                    ("checking all partitions for auto_increment_value"));
         do
         {
+          if (!bitmap_is_set(&m_opened_partitions, (uint)(file_array - m_file)))
+          {
+            /*
+              Some partitions aren't opened.
+              So we can't calculate the autoincrement.
+            */
+            all_parts_opened= false;
+            break;
+          }
           file= *file_array;
           file->info(HA_STATUS_AUTO | no_lock_flag);
           set_if_bigger(auto_increment_value,
@@ -8176,7 +8183,7 @@ int ha_partition::info(uint flag)
 
         DBUG_ASSERT(auto_increment_value);
         stats.auto_increment_value= auto_increment_value;
-        if (auto_inc_is_first_in_idx)
+        if (all_parts_opened && auto_inc_is_first_in_idx)
         {
           set_if_bigger(part_share->next_auto_inc_val,
                         auto_increment_value);
@@ -8485,6 +8492,7 @@ int ha_partition::change_partitions_to_open(List<String> *partition_names)
     return 0;
   }
 
+  check_insert_autoincrement();
   if (bitmap_cmp(&m_opened_partitions, &m_part_info->read_partitions) != 0)
     return 0;
 
diff --git a/sql/ha_partition.h b/sql/ha_partition.h
index 0fe98dc4608..ff02ea5aa1c 100644
--- a/sql/ha_partition.h
+++ b/sql/ha_partition.h
@@ -1329,6 +1329,19 @@ private:
     unlock_auto_increment();
   }
 
+  void check_insert_autoincrement()
+  {
+    /*
+      If we INSERT into the table having the AUTO_INCREMENT column,
+      we have to read all partitions for the next autoincrement value
+      unless we already did it.
+    */
+    if (!part_share->auto_inc_initialized &&
+        ha_thd()->lex->sql_command == SQLCOM_INSERT &&
+        table->found_next_number_field)
+      bitmap_set_all(&m_part_info->read_partitions);
+  }
+
 public:
 
   /*
diff --git a/sql/item_strfunc.cc b/sql/item_strfunc.cc
index bf571473211..78c29ac8e33 100644
--- a/sql/item_strfunc.cc
+++ b/sql/item_strfunc.cc
@@ -2666,10 +2666,17 @@ const int FORMAT_MAX_DECIMALS= 30;
 
 bool Item_func_format::fix_length_and_dec()
 {
-  uint32 char_length= args[0]->max_char_length();
-  uint32 max_sep_count= (char_length / 3) + (decimals ? 1 : 0) + /*sign*/1;
+  uint32 char_length= args[0]->type_handler()->Item_decimal_notation_int_digits(args[0]);
+  uint dec= FORMAT_MAX_DECIMALS;
+  if (args[1]->const_item() && !args[1]->is_expensive())
+  {
+    Longlong_hybrid tmp= args[1]->to_longlong_hybrid();
+    if (!args[1]->null_value)
+      dec= tmp.to_uint(FORMAT_MAX_DECIMALS);
+  }
+  uint32 max_sep_count= (char_length / 3) + (dec ? 1 : 0) + /*sign*/1;
   collation.set(default_charset());
-  fix_char_length(char_length + max_sep_count + decimals);
+  fix_char_length(char_length + max_sep_count + dec);
   if (arg_count == 3)
     locale= args[2]->basic_const_item() ? args[2]->locale_from_val_str() : NULL;
   else
diff --git a/sql/keycaches.cc b/sql/keycaches.cc
index f38a43f83cf..60049cdd67d 100644
--- a/sql/keycaches.cc
+++ b/sql/keycaches.cc
@@ -68,7 +68,7 @@ uchar* find_named(I_List<NAMED_ILINK> *list, const char *name, size_t length,
 }
 
 
-bool NAMED_ILIST::delete_element(const char *name, size_t length, void (*free_element)(const char *name, uchar*))
+bool NAMED_ILIST::delete_element(const char *name, size_t length, void (*free_element)(const char *name, void*))
 {
   I_List_iterator<NAMED_ILINK> it(*this);
   NAMED_ILINK *element;
@@ -85,7 +85,7 @@ bool NAMED_ILIST::delete_element(const char *name, size_t length, void (*free_el
   DBUG_RETURN(1);
 }
 
-void NAMED_ILIST::delete_elements(void (*free_element)(const char *name, uchar*))
+void NAMED_ILIST::delete_elements(void (*free_element)(const char *name, void*))
 {
   NAMED_ILINK *element;
   DBUG_ENTER("NAMED_ILIST::delete_elements");
@@ -157,9 +157,9 @@ KEY_CACHE *get_or_create_key_cache(const char *name, size_t length)
 }
 
 
-void free_key_cache(const char *name, KEY_CACHE *key_cache)
+void free_key_cache(const char *name, void *key_cache)
 {
-  end_key_cache(key_cache, 1);		// Can never fail
+  end_key_cache(static_cast<KEY_CACHE *>(key_cache), 1); // Can never fail
   my_free(key_cache);
 }
 
@@ -221,13 +221,12 @@ Rpl_filter *get_or_create_rpl_filter(const char *name, size_t length)
   return filter;
 }
 
-void free_rpl_filter(const char *name, Rpl_filter *filter)
+void free_rpl_filter(const char *name, void *filter)
 {
-  delete filter;
-  filter= 0;
+  delete static_cast<Rpl_filter*>(filter);
 }
 
 void free_all_rpl_filters()
 {
-  rpl_filters.delete_elements((void (*)(const char*, uchar*)) free_rpl_filter);
+  rpl_filters.delete_elements(free_rpl_filter);
 }
diff --git a/sql/keycaches.h b/sql/keycaches.h
index 9da93e5f7ba..68c3dd3a2b0 100644
--- a/sql/keycaches.h
+++ b/sql/keycaches.h
@@ -30,8 +30,8 @@ class NAMED_ILINK;
 class NAMED_ILIST: public I_List<NAMED_ILINK>
 {
   public:
-  void delete_elements(void (*free_element)(const char*, uchar*));
-  bool delete_element(const char *name, size_t length, void (*free_element)(const char*, uchar*));
+  void delete_elements(void (*free_element)(const char*, void*));
+  bool delete_element(const char *name, size_t length, void (*free_element)(const char*, void*));
 };
 
 /* For key cache */
@@ -42,7 +42,7 @@ extern NAMED_ILIST key_caches;
 KEY_CACHE *create_key_cache(const char *name, size_t length);
 KEY_CACHE *get_key_cache(const LEX_CSTRING *cache_name);
 KEY_CACHE *get_or_create_key_cache(const char *name, size_t length);
-void free_key_cache(const char *name, KEY_CACHE *key_cache);
+void free_key_cache(const char *name, void *key_cache);
 bool process_key_caches(process_key_cache_t func, void *param);
 
 /* For Rpl_filter */
@@ -52,7 +52,6 @@ extern NAMED_ILIST rpl_filters;
 Rpl_filter *create_rpl_filter(const char *name, size_t length);
 Rpl_filter *get_rpl_filter(LEX_CSTRING *filter_name);
 Rpl_filter *get_or_create_rpl_filter(const char *name, size_t length);
-void free_rpl_filter(const char *name, Rpl_filter *filter);
 void free_all_rpl_filters(void);
 
 #endif /* KEYCACHES_INCLUDED */
diff --git a/sql/log.cc b/sql/log.cc
index 4f51a9a9c17..98fb4a6d5a6 100644
--- a/sql/log.cc
+++ b/sql/log.cc
@@ -5968,7 +5968,6 @@ MYSQL_BIN_LOG::flush_and_set_pending_rows_event(THD *thd,
   DBUG_ASSERT(WSREP_EMULATE_BINLOG(thd) || mysql_bin_log.is_open());
   DBUG_PRINT("enter", ("event: %p", event));
 
-  int error= 0;
   binlog_cache_mngr *const cache_mngr=
     (binlog_cache_mngr*) thd_get_ha_data(thd, binlog_hton);
 
@@ -6006,7 +6005,7 @@ MYSQL_BIN_LOG::flush_and_set_pending_rows_event(THD *thd,
 
   thd->binlog_set_pending_rows_event(event, is_transactional);
 
-  DBUG_RETURN(error);
+  DBUG_RETURN(0);
 }
 
 
@@ -7718,7 +7717,7 @@ MYSQL_BIN_LOG::queue_for_group_commit(group_commit_entry *orig_entry)
   mysql_mutex_unlock(&LOCK_prepare_ordered);
   DEBUG_SYNC(orig_entry->thd, "commit_after_release_LOCK_prepare_ordered");
 
-  DBUG_PRINT("info", ("Queued for group commit as %s\n",
+  DBUG_PRINT("info", ("Queued for group commit as %s",
                       (orig_queue == NULL) ? "leader" : "participant"));
   DBUG_RETURN(orig_queue == NULL);
 }
@@ -10734,13 +10733,13 @@ bool wsrep_stmt_rollback_is_safe(THD* thd)
     binlog_cache_data * trx_cache = &cache_mngr->trx_cache;
     if (thd->wsrep_sr().fragments_certified() > 0 &&
         (trx_cache->get_prev_position() == MY_OFF_T_UNDEF ||
-         trx_cache->get_prev_position() < thd->wsrep_sr().bytes_certified()))
+         trx_cache->get_prev_position() < thd->wsrep_sr().log_position()))
     {
       WSREP_DEBUG("statement rollback is not safe for streaming replication"
                   " pre-stmt_pos: %llu, frag repl pos: %zu\n"
                   "Thread: %llu, SQL: %s",
                   trx_cache->get_prev_position(),
-                  thd->wsrep_sr().bytes_certified(),
+                  thd->wsrep_sr().log_position(),
                   thd->thread_id, thd->query());
        ret = false;
     }
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index 1df08eb739e..18b8148ed57 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -668,7 +668,9 @@ SHOW_COMP_OPTION have_crypt, have_compress;
 SHOW_COMP_OPTION have_profiling;
 SHOW_COMP_OPTION have_openssl;
 
+#ifndef EMBEDDED_LIBRARY
 static std::atomic<char*> shutdown_user;
+#endif //EMBEDDED_LIBRARY
 
 /* Thread specific variables */
 
@@ -1987,7 +1989,7 @@ static void clean_up(bool print_message)
   tdc_deinit();
   mdl_destroy();
   dflt_key_cache= 0;
-  key_caches.delete_elements((void (*)(const char*, uchar*)) free_key_cache);
+  key_caches.delete_elements(free_key_cache);
   wt_end();
   multi_keycache_free();
   sp_cache_end();
@@ -4303,7 +4305,6 @@ static int init_common_variables()
     return 1;
   }
 
-  global_system_variables.in_subquery_conversion_threshold= IN_SUBQUERY_CONVERSION_THRESHOLD;
 
 #ifdef WITH_WSREP
   /*
diff --git a/sql/opt_subselect.cc b/sql/opt_subselect.cc
index eeefe4200da..484836af23e 100644
--- a/sql/opt_subselect.cc
+++ b/sql/opt_subselect.cc
@@ -2681,9 +2681,17 @@ bool find_eq_ref_candidate(TABLE *table, table_map sj_inner_tables)
       {
         do  /* For all equalities on all key parts */
         {
-          /* Check if this is "t.keypart = expr(outer_tables) */
+          /*
+            Check if this is "t.keypart = expr(outer_tables)
+
+            Don't allow variants that can produce duplicates:
+            - Dont allow "ref or null"
+            - the keyuse (that is, the operation) must be null-rejecting,
+              unless the other expression is non-NULLable.
+          */
           if (!(keyuse->used_tables & sj_inner_tables) &&
-              !(keyuse->optimize & KEY_OPTIMIZE_REF_OR_NULL))
+              !(keyuse->optimize & KEY_OPTIMIZE_REF_OR_NULL) &&
+              (keyuse->null_rejecting || !keyuse->val->maybe_null))
           {
             bound_parts |= 1 << keyuse->keypart;
           }
diff --git a/sql/rpl_rli.cc b/sql/rpl_rli.cc
index 48bc2bb0210..6d55b06b497 100644
--- a/sql/rpl_rli.cc
+++ b/sql/rpl_rli.cc
@@ -1561,7 +1561,7 @@ scan_one_gtid_slave_pos_table(THD *thd, HASH *hash, DYNAMIC_ARRAY *array,
     sub_id= (ulonglong)table->field[1]->val_int();
     server_id= (uint32)table->field[2]->val_int();
     seq_no= (ulonglong)table->field[3]->val_int();
-    DBUG_PRINT("info", ("Read slave state row: %u-%u-%lu sub_id=%lu\n",
+    DBUG_PRINT("info", ("Read slave state row: %u-%u-%lu sub_id=%lu",
                         (unsigned)domain_id, (unsigned)server_id,
                         (ulong)seq_no, (ulong)sub_id));
 
diff --git a/sql/semisync_master.cc b/sql/semisync_master.cc
index f9fb4d9147d..b239a9776a7 100644
--- a/sql/semisync_master.cc
+++ b/sql/semisync_master.cc
@@ -806,7 +806,7 @@ int Repl_semi_sync_master::commit_trx(const char* trx_wait_binlog_name,
     if (!get_master_enabled() || !is_on())
       goto l_end;
 
-    DBUG_PRINT("semisync", ("%s: wait pos (%s, %lu), repl(%d)\n",
+    DBUG_PRINT("semisync", ("%s: wait pos (%s, %lu), repl(%d)",
                             "Repl_semi_sync_master::commit_trx",
                             trx_wait_binlog_name, (ulong)trx_wait_binlog_pos,
                             (int)is_on()));
diff --git a/sql/service_wsrep.cc b/sql/service_wsrep.cc
index b365b393e0b..204d671d53b 100644
--- a/sql/service_wsrep.cc
+++ b/sql/service_wsrep.cc
@@ -55,7 +55,26 @@ extern "C" const char* wsrep_thd_transaction_state_str(const THD *thd)
 
 extern "C" const char *wsrep_thd_query(const THD *thd)
 {
-  return thd ? thd->query() : NULL;
+  if (thd)
+  {
+    switch(thd->lex->sql_command)
+    {
+    case SQLCOM_CREATE_USER:
+      return "CREATE USER";
+    case SQLCOM_GRANT:
+      return "GRANT";
+    case SQLCOM_REVOKE:
+      return "REVOKE";
+    case SQLCOM_SET_OPTION:
+      if (thd->lex->definer)
+	return "SET PASSWORD";
+      /* fallthrough */
+    default:
+      if (thd->query())
+        return thd->query();
+    }
+  }
+  return "NULL";
 }
 
 extern "C" query_id_t wsrep_thd_transaction_id(const THD *thd)
diff --git a/sql/sp.cc b/sql/sp.cc
index c9d1e2e3c4a..058a29e65f7 100644
--- a/sql/sp.cc
+++ b/sql/sp.cc
@@ -1466,7 +1466,7 @@ log:
     /* Such a statement can always go directly to binlog, no trans cache */
     if (thd->binlog_query(THD::STMT_QUERY_TYPE,
                           log_query.ptr(), log_query.length(),
-                          FALSE, FALSE, FALSE, 0))
+                          FALSE, FALSE, FALSE, 0) > 0)
     {
       my_error(ER_ERROR_ON_WRITE, MYF(0), "binary log", -1);
       goto done;
diff --git a/sql/sp_head.cc b/sql/sp_head.cc
index 8c07a060f30..945b3ac8854 100644
--- a/sql/sp_head.cc
+++ b/sql/sp_head.cc
@@ -4722,6 +4722,7 @@ typedef struct st_sp_table
   uint lock_count;
   uint query_lock_count;
   uint8 trg_event_map;
+  my_bool for_insert_data;
 } SP_TABLE;
 
 
@@ -4817,6 +4818,7 @@ sp_head::merge_table_list(THD *thd, TABLE_LIST *table, LEX *lex_for_tmp_check)
         if (tab->query_lock_count > tab->lock_count)
           tab->lock_count++;
         tab->trg_event_map|= table->trg_event_map;
+        tab->for_insert_data|= table->for_insert_data;
       }
       else
       {
@@ -4840,6 +4842,7 @@ sp_head::merge_table_list(THD *thd, TABLE_LIST *table, LEX *lex_for_tmp_check)
         tab->lock_type= table->lock_type;
         tab->lock_count= tab->query_lock_count= 1;
         tab->trg_event_map= table->trg_event_map;
+        tab->for_insert_data= table->for_insert_data;
         if (my_hash_insert(&m_sptabs, (uchar *)tab))
           return FALSE;
       }
@@ -4923,7 +4926,8 @@ sp_head::add_used_tables_to_table_list(THD *thd,
                                            TABLE_LIST::PRELOCK_ROUTINE,
                                            belong_to_view,
                                            stab->trg_event_map,
-                                           query_tables_last_ptr);
+                                           query_tables_last_ptr,
+                                           stab->for_insert_data);
       tab_buff+= ALIGN_SIZE(sizeof(TABLE_LIST));
       result= TRUE;
     }
diff --git a/sql/sql_acl.cc b/sql/sql_acl.cc
index ff8a2bebb8d..4c8eaee3145 100644
--- a/sql/sql_acl.cc
+++ b/sql/sql_acl.cc
@@ -3788,7 +3788,7 @@ bool change_password(THD *thd, LEX_USER *user)
     DBUG_ASSERT(query_length);
     thd->clear_error();
     result= thd->binlog_query(THD::STMT_QUERY_TYPE, buff, query_length,
-                              FALSE, FALSE, FALSE, 0);
+                              FALSE, FALSE, FALSE, 0) > 0;
   }
 end:
   if (result)
@@ -3940,7 +3940,7 @@ int acl_set_default_role(THD *thd, const char *host, const char *user,
       DBUG_ASSERT(query_length);
       thd->clear_error();
       result= thd->binlog_query(THD::STMT_QUERY_TYPE, buff, query_length,
-                                FALSE, FALSE, FALSE, 0);
+                                FALSE, FALSE, FALSE, 0) > 0;
     }
   end:
     close_mysql_tables(thd);
diff --git a/sql/sql_admin.cc b/sql/sql_admin.cc
index 5dcb7ffac72..3b15c5a505e 100644
--- a/sql/sql_admin.cc
+++ b/sql/sql_admin.cc
@@ -1317,7 +1317,7 @@ bool Sql_cmd_analyze_table::execute(THD *thd)
                          "analyze", lock_type, 1, 0, 0, 0,
                          &handler::ha_analyze, 0);
   /* ! we write after unlocking the table */
-  if (!res && !m_lex->no_write_to_binlog)
+  if (!res && !m_lex->no_write_to_binlog && (!opt_readonly || thd->slave_thread))
   {
     /*
       Presumably, ANALYZE and binlog writing doesn't require synchronization
@@ -1377,7 +1377,7 @@ bool Sql_cmd_optimize_table::execute(THD *thd)
                       "optimize", TL_WRITE, 1, 0, 0, 0,
                       &handler::ha_optimize, 0);
   /* ! we write after unlocking the table */
-  if (!res && !m_lex->no_write_to_binlog)
+  if (!res && !m_lex->no_write_to_binlog && (!opt_readonly || thd->slave_thread))
   {
     /*
       Presumably, OPTIMIZE and binlog writing doesn't require synchronization
@@ -1413,7 +1413,7 @@ bool Sql_cmd_repair_table::execute(THD *thd)
                          &handler::ha_repair, &view_repair);
 
   /* ! we write after unlocking the table */
-  if (!res && !m_lex->no_write_to_binlog)
+  if (!res && !m_lex->no_write_to_binlog && (!opt_readonly || thd->slave_thread))
   {
     /*
       Presumably, REPAIR and binlog writing doesn't require synchronization
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index 15c0e976c26..da061adfc1f 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -2668,6 +2668,7 @@ Locked_tables_list::reopen_tables(THD *thd, bool need_reopen)
         {
           thd->locked_tables_list.unlink_from_list(thd, table_list, false);
           mysql_lock_remove(thd, thd->lock, *prev);
+          (*prev)->file->extra(HA_EXTRA_PREPARE_FOR_FORCED_CLOSE);
           close_thread_table(thd, prev);
           break;
         }
@@ -4604,9 +4605,11 @@ add_internal_tables(THD *thd, Query_tables_list *prelocking_ctx,
                     TABLE_LIST *tables)
 {
   TABLE_LIST *global_table_list= prelocking_ctx->query_tables;
+  DBUG_ENTER("add_internal_tables");
 
   do
   {
+    DBUG_PRINT("info", ("table name: %s", tables->table_name.str));
     /*
       Skip table if already in the list. Can happen with prepared statements
     */
@@ -4616,20 +4619,22 @@ add_internal_tables(THD *thd, Query_tables_list *prelocking_ctx,
 
     TABLE_LIST *tl= (TABLE_LIST *) thd->alloc(sizeof(TABLE_LIST));
     if (!tl)
-      return TRUE;
+      DBUG_RETURN(TRUE);
     tl->init_one_table_for_prelocking(&tables->db,
                                       &tables->table_name,
                                       NULL, tables->lock_type,
                                       TABLE_LIST::PRELOCK_NONE,
                                       0, 0,
-                                      &prelocking_ctx->query_tables_last);
+                                      &prelocking_ctx->query_tables_last,
+                                      tables->for_insert_data);
     /*
       Store link to the new table_list that will be used by open so that
       Item_func_nextval() can find it
     */
     tables->next_local= tl;
+    DBUG_PRINT("info", ("table name: %s added", tables->table_name.str));
   } while ((tables= tables->next_global));
-  return FALSE;
+  DBUG_RETURN(FALSE);
 }
 
 
@@ -4660,6 +4665,7 @@ bool DML_prelocking_strategy::
 handle_table(THD *thd, Query_tables_list *prelocking_ctx,
              TABLE_LIST *table_list, bool *need_prelocking)
 {
+  DBUG_ENTER("handle_table");
   TABLE *table= table_list->table;
   /* We rely on a caller to check that table is going to be changed. */
   DBUG_ASSERT(table_list->lock_type >= TL_WRITE_ALLOW_WRITE ||
@@ -4690,7 +4696,7 @@ handle_table(THD *thd, Query_tables_list *prelocking_ctx,
       {
         if (arena)
           thd->restore_active_arena(arena, &backup);
-        return TRUE;
+        DBUG_RETURN(TRUE);
       }
 
       *need_prelocking= TRUE;
@@ -4718,7 +4724,8 @@ handle_table(THD *thd, Query_tables_list *prelocking_ctx,
                                           NULL, lock_type,
                                           TABLE_LIST::PRELOCK_FK,
                                           table_list->belong_to_view, op,
-                                          &prelocking_ctx->query_tables_last);
+                                          &prelocking_ctx->query_tables_last,
+                                          table_list->for_insert_data);
       }
       if (arena)
         thd->restore_active_arena(arena, &backup);
@@ -4726,8 +4733,11 @@ handle_table(THD *thd, Query_tables_list *prelocking_ctx,
   }
 
   /* Open any tables used by DEFAULT (like sequence tables) */
+  DBUG_PRINT("info", ("table: %p  name: %s  db: %s  flags: %u",
+                      table_list, table_list->table_name.str,
+                      table_list->db.str, table_list->for_insert_data));
   if (table->internal_tables &&
-      ((sql_command_flags[thd->lex->sql_command] & CF_INSERTS_DATA) ||
+      (table_list->for_insert_data ||
        thd->lex->default_used))
   {
     Query_arena *arena, backup;
@@ -4740,10 +4750,10 @@ handle_table(THD *thd, Query_tables_list *prelocking_ctx,
     if (unlikely(error))
     {
       *need_prelocking= TRUE;
-      return TRUE;
+      DBUG_RETURN(TRUE);
     }
   }
-  return FALSE;
+  DBUG_RETURN(FALSE);
 }
 
 
@@ -4760,7 +4770,7 @@ bool open_and_lock_internal_tables(TABLE *table, bool lock_table)
   THD *thd= table->in_use;
   TABLE_LIST *tl;
   MYSQL_LOCK *save_lock,*new_lock;
-  DBUG_ENTER("open_internal_tables");
+  DBUG_ENTER("open_and_lock_internal_tables");
 
   /* remove pointer to old select_lex which is already destroyed */
   for (tl= table->internal_tables ; tl ; tl= tl->next_global)
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index a4861620bf0..cc572065e92 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -49,7 +49,7 @@
 #include <m_ctype.h>
 #include <sys/stat.h>
 #include <thr_alarm.h>
-#ifdef	__WIN__
+#ifdef	__WIN__0
 #include <io.h>
 #endif
 #include <mysys_err.h>
@@ -5848,17 +5848,33 @@ int THD::decide_logging_format(TABLE_LIST *tables)
       Get the capabilities vector for all involved storage engines and
       mask out the flags for the binary log.
     */
-    for (TABLE_LIST *table= tables; table; table= table->next_global)
+    for (TABLE_LIST *tbl= tables; tbl; tbl= tbl->next_global)
     {
-      if (table->placeholder())
+      TABLE *table;
+      TABLE_SHARE *share;
+      handler::Table_flags flags;
+      if (tbl->placeholder())
         continue;
 
-      handler::Table_flags const flags= table->table->file->ha_table_flags();
+      table= tbl->table;
+      share= table->s;
+      flags= table->file->ha_table_flags();
+      if (!share->table_creation_was_logged)
+      {
+        /*
+          This is a temporary table which was not logged in the binary log.
+          Disable statement logging to enforce row level logging.
+        */
+        DBUG_ASSERT(share->tmp_table);
+        flags&= ~HA_BINLOG_STMT_CAPABLE;
+        /* We can only use row logging */
+        set_current_stmt_binlog_format_row();
+      }
 
       DBUG_PRINT("info", ("table: %s; ha_table_flags: 0x%llx",
-                          table->table_name.str, flags));
+                          tbl->table_name.str, flags));
 
-      if (table->table->s->no_replicate)
+      if (share->no_replicate)
       {
         /*
           The statement uses a table that is not replicated.
@@ -5876,44 +5892,44 @@ int THD::decide_logging_format(TABLE_LIST *tables)
         */
         lex->set_stmt_unsafe(LEX::BINLOG_STMT_UNSAFE_SYSTEM_TABLE);
 
-        if (table->lock_type >= TL_WRITE_ALLOW_WRITE)
+        if (tbl->lock_type >= TL_WRITE_ALLOW_WRITE)
         {
           non_replicated_tables_count++;
           continue;
         }
       }
-      if (table == lex->first_not_own_table())
+      if (tbl == lex->first_not_own_table())
         found_first_not_own_table= true;
 
       replicated_tables_count++;
 
-      if (table->prelocking_placeholder != TABLE_LIST::PRELOCK_FK)
+      if (tbl->prelocking_placeholder != TABLE_LIST::PRELOCK_FK)
       {
-        if (table->lock_type <= TL_READ_NO_INSERT)
+        if (tbl->lock_type <= TL_READ_NO_INSERT)
           has_read_tables= true;
-        else if (table->table->found_next_number_field &&
-                 (table->lock_type >= TL_WRITE_ALLOW_WRITE))
+        else if (table->found_next_number_field &&
+                 (tbl->lock_type >= TL_WRITE_ALLOW_WRITE))
         {
           has_auto_increment_write_tables= true;
           has_auto_increment_write_tables_not_first= found_first_not_own_table;
-          if (table->table->s->next_number_keypart != 0)
+          if (share->next_number_keypart != 0)
             has_write_table_auto_increment_not_first_in_pk= true;
         }
       }
 
-      if (table->lock_type >= TL_WRITE_ALLOW_WRITE)
+      if (tbl->lock_type >= TL_WRITE_ALLOW_WRITE)
       {
         bool trans;
         if (prev_write_table && prev_write_table->file->ht !=
-            table->table->file->ht)
+            table->file->ht)
           multi_write_engine= TRUE;
-        if (table->table->s->non_determinstic_insert &&
+        if (share->non_determinstic_insert &&
             !(sql_command_flags[lex->sql_command] & CF_SCHEMA_CHANGE))
           has_write_tables_with_unsafe_statements= true;
 
-        trans= table->table->file->has_transactions();
+        trans= table->file->has_transactions();
 
-        if (table->table->s->tmp_table)
+        if (share->tmp_table)
           lex->set_stmt_accessed_table(trans ? LEX::STMT_WRITES_TEMP_TRANS_TABLE :
                                                LEX::STMT_WRITES_TEMP_NON_TRANS_TABLE);
         else
@@ -5924,17 +5940,16 @@ int THD::decide_logging_format(TABLE_LIST *tables)
         flags_write_some_set |= flags;
         is_write= TRUE;
 
-        prev_write_table= table->table;
+        prev_write_table= table;
 
       }
       flags_access_some_set |= flags;
 
-      if (lex->sql_command != SQLCOM_CREATE_TABLE ||
-          (lex->sql_command == SQLCOM_CREATE_TABLE && lex->tmp_table()))
+      if (lex->sql_command != SQLCOM_CREATE_TABLE || lex->tmp_table())
       {
-        my_bool trans= table->table->file->has_transactions();
+        my_bool trans= table->file->has_transactions();
 
-        if (table->table->s->tmp_table)
+        if (share->tmp_table)
           lex->set_stmt_accessed_table(trans ? LEX::STMT_READS_TEMP_TRANS_TABLE :
                                                LEX::STMT_READS_TEMP_NON_TRANS_TABLE);
         else
@@ -5943,10 +5958,10 @@ int THD::decide_logging_format(TABLE_LIST *tables)
       }
 
       if (prev_access_table && prev_access_table->file->ht !=
-          table->table->file->ht)
+          table->file->ht)
         multi_access_engine= TRUE;
 
-      prev_access_table= table->table;
+      prev_access_table= table;
     }
 
     if (wsrep_binlog_format() != BINLOG_FORMAT_ROW)
@@ -6075,10 +6090,17 @@ int THD::decide_logging_format(TABLE_LIST *tables)
         {
           /*
             5. Error: Cannot modify table that uses a storage engine
-               limited to row-logging when binlog_format = STATEMENT
+               limited to row-logging when binlog_format = STATEMENT, except
+               if all tables that are updated are temporary tables
           */
-	  if (IF_WSREP((!WSREP_NNULL(this) ||
-			wsrep_cs().mode() == wsrep::client_state::m_local),1))
+          if (!lex->stmt_writes_to_non_temp_table())
+          {
+            /* As all updated tables are temporary, nothing will be logged */
+            set_current_stmt_binlog_format_row();
+          }
+          else if (IF_WSREP((!WSREP(this) ||
+                             wsrep_cs().mode() ==
+                             wsrep::client_state::m_local),1))
 	  {
             my_error((error= ER_BINLOG_STMT_MODE_AND_ROW_ENGINE), MYF(0), "");
 	  }
@@ -6147,10 +6169,8 @@ int THD::decide_logging_format(TABLE_LIST *tables)
                         "ROW" : "STATEMENT"));
 
     if (variables.binlog_format == BINLOG_FORMAT_ROW &&
-        (lex->sql_command == SQLCOM_UPDATE ||
-         lex->sql_command == SQLCOM_UPDATE_MULTI ||
-         lex->sql_command == SQLCOM_DELETE ||
-         lex->sql_command == SQLCOM_DELETE_MULTI))
+        (sql_command_flags[lex->sql_command] &
+         (CF_UPDATES_DATA | CF_DELETES_DATA)))
     {
       String table_names;
       /*
@@ -6170,8 +6190,8 @@ int THD::decide_logging_format(TABLE_LIST *tables)
       }
       if (!table_names.is_empty())
       {
-        bool is_update= (lex->sql_command == SQLCOM_UPDATE ||
-                         lex->sql_command == SQLCOM_UPDATE_MULTI);
+        bool is_update= MY_TEST(sql_command_flags[lex->sql_command] &
+                                CF_UPDATES_DATA);
         /*
           Replace the last ',' with '.' for table_names
         */
@@ -7008,11 +7028,12 @@ void THD::issue_unsafe_warnings()
 
   @see decide_logging_format
 
+  @retval < 0 No logging of query (ok)
   @retval 0 Success
-
-  @retval nonzero If there is a failure when writing the query (e.g.,
-  write failure), then the error code is returned.
+  @retval > 0  If there is a failure when writing the query (e.g.,
+               write failure), then the error code is returned.
 */
+
 int THD::binlog_query(THD::enum_binlog_query_type qtype, char const *query_arg,
                       ulong query_len, bool is_trans, bool direct, 
                       bool suppress_use, int errcode)
@@ -7038,7 +7059,7 @@ int THD::binlog_query(THD::enum_binlog_query_type qtype, char const *query_arg,
       The current statement is to be ignored, and not written to
       the binlog. Do not call issue_unsafe_warnings().
     */
-    DBUG_RETURN(0);
+    DBUG_RETURN(-1);
   }
 
   /*
@@ -7054,7 +7075,10 @@ int THD::binlog_query(THD::enum_binlog_query_type qtype, char const *query_arg,
   {
     int error;
     if (unlikely(error= binlog_flush_pending_rows_event(TRUE, is_trans)))
+    {
+      DBUG_ASSERT(error > 0);
       DBUG_RETURN(error);
+    }
   }
 
   /*
@@ -7097,7 +7121,7 @@ int THD::binlog_query(THD::enum_binlog_query_type qtype, char const *query_arg,
                ("is_current_stmt_binlog_format_row: %d",
                 is_current_stmt_binlog_format_row()));
     if (is_current_stmt_binlog_format_row())
-      DBUG_RETURN(0);
+      DBUG_RETURN(-1);
     /* Fall through */
 
     /*
@@ -7138,7 +7162,7 @@ int THD::binlog_query(THD::enum_binlog_query_type qtype, char const *query_arg,
       }
 
       binlog_table_maps= 0;
-      DBUG_RETURN(error);
+      DBUG_RETURN(error >= 0 ? error : 1);
     }
 
   case THD::QUERY_TYPE_COUNT:
diff --git a/sql/sql_class.h b/sql/sql_class.h
index a208b627b7e..2fb5797b325 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -1630,12 +1630,16 @@ public:
 /**
   @class Sub_statement_state
   @brief Used to save context when executing a function or trigger
+
+  operations on stat tables aren't technically a sub-statement, but they are
+  similar in a sense that they cannot change the transaction status.
 */
 
 /* Defines used for Sub_statement_state::in_sub_stmt */
 
 #define SUB_STMT_TRIGGER 1
 #define SUB_STMT_FUNCTION 2
+#define SUB_STMT_STAT_TABLES 4
 
 
 class Sub_statement_state
@@ -6654,6 +6658,11 @@ public:
 /* Bits in server_command_flags */
 
 /**
+  Statement that deletes existing rows (DELETE, DELETE_MULTI)
+*/
+#define CF_DELETES_DATA (1U << 24)
+
+/**
   Skip the increase of the global query id counter. Commonly set for
   commands that are stateless (won't cause any change on the server
   internal states).
@@ -6858,6 +6867,22 @@ class Sql_mode_save
   sql_mode_t old_mode; // SQL mode saved at construction time.
 };
 
+class Switch_to_definer_security_ctx
+{
+ public:
+  Switch_to_definer_security_ctx(THD *thd, TABLE_LIST *table) :
+    m_thd(thd), m_sctx(thd->security_ctx)
+  {
+    if (table->security_ctx)
+      thd->security_ctx= table->security_ctx;
+  }
+  ~Switch_to_definer_security_ctx() { m_thd->security_ctx = m_sctx; }
+
+ private:
+  THD *m_thd;
+  Security_context *m_sctx;
+};
+
 
 class Sql_mode_instant_set: public Sql_mode_save
 {
diff --git a/sql/sql_delete.cc b/sql/sql_delete.cc
index 8161a761486..0ea00a7d61c 100644
--- a/sql/sql_delete.cc
+++ b/sql/sql_delete.cc
@@ -930,7 +930,7 @@ cleanup:
                                         transactional_table, FALSE, FALSE,
                                         errcode);
 
-      if (log_result)
+      if (log_result > 0)
       {
 	error=1;
       }
@@ -1616,7 +1616,7 @@ bool multi_delete::send_eof()
       if (unlikely(thd->binlog_query(THD::ROW_QUERY_TYPE,
                                      thd->query(), thd->query_length(),
                                      transactional_tables, FALSE, FALSE,
-                                     errcode)) &&
+                                     errcode) > 0) &&
           !normal_tables)
       {
 	local_error=1;  // Log write failed: roll back the SQL statement
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index 7be005bd79b..329b6a130b0 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -1193,13 +1193,13 @@ values_loop_end:
           else if (thd->binlog_query(THD::ROW_QUERY_TYPE,
                                      log_query.c_ptr(), log_query.length(),
                                      transactional_table, FALSE, FALSE,
-                                     errcode))
+                                     errcode) > 0)
             error= 1;
         }
         else if (thd->binlog_query(THD::ROW_QUERY_TYPE,
 			           thd->query(), thd->query_length(),
 			           transactional_table, FALSE, FALSE,
-                                   errcode))
+                                   errcode) > 0)
 	  error= 1;
       }
     }
@@ -3964,6 +3964,7 @@ bool select_insert::prepare_eof()
   int error;
   bool const trans_table= table->file->has_transactions();
   bool changed;
+  bool binary_logged= 0;
   killed_state killed_status= thd->killed;
 
   DBUG_ENTER("select_insert::prepare_eof");
@@ -4014,18 +4015,22 @@ bool select_insert::prepare_eof()
       (likely(!error) || thd->transaction.stmt.modified_non_trans_table))
   {
     int errcode= 0;
+    int res;
     if (likely(!error))
       thd->clear_error();
     else
       errcode= query_error_code(thd, killed_status == NOT_KILLED);
-    if (thd->binlog_query(THD::ROW_QUERY_TYPE,
-                      thd->query(), thd->query_length(),
-                      trans_table, FALSE, FALSE, errcode))
+    res= thd->binlog_query(THD::ROW_QUERY_TYPE,
+                           thd->query(), thd->query_length(),
+                           trans_table, FALSE, FALSE, errcode);
+    if (res > 0)
     {
       table->file->ha_release_auto_increment();
       DBUG_RETURN(true);
     }
+    binary_logged= res == 0 || !table->s->tmp_table;
   }
+  table->s->table_creation_was_logged|= binary_logged;
   table->file->ha_release_auto_increment();
 
   if (unlikely(error))
@@ -4082,8 +4087,9 @@ bool select_insert::send_eof()
   DBUG_RETURN(res);
 }
 
-void select_insert::abort_result_set() {
-
+void select_insert::abort_result_set()
+{
+  bool binary_logged= 0;
   DBUG_ENTER("select_insert::abort_result_set");
   /*
     If the creation of the table failed (due to a syntax error, for
@@ -4135,16 +4141,20 @@ void select_insert::abort_result_set() {
         if(WSREP_EMULATE_BINLOG(thd) || mysql_bin_log.is_open())
         {
           int errcode= query_error_code(thd, thd->killed == NOT_KILLED);
+          int res;
           /* error of writing binary log is ignored */
-          (void) thd->binlog_query(THD::ROW_QUERY_TYPE, thd->query(),
-                                   thd->query_length(),
-                                   transactional_table, FALSE, FALSE, errcode);
+          res= thd->binlog_query(THD::ROW_QUERY_TYPE, thd->query(),
+                                 thd->query_length(),
+                                 transactional_table, FALSE, FALSE, errcode);
+          binary_logged= res == 0 || !table->s->tmp_table;
         }
 	if (changed)
 	  query_cache_invalidate3(thd, table, 1);
     }
     DBUG_ASSERT(transactional_table || !changed ||
 		thd->transaction.stmt.modified_non_trans_table);
+
+    table->s->table_creation_was_logged|= binary_logged;
     table->file->ha_release_auto_increment();
   }
 
@@ -4213,6 +4223,7 @@ TABLE *select_create::create_table_from_items(THD *thd, List<Item> *items,
   /* Add selected items to field list */
   List_iterator_fast<Item> it(*items);
   Item *item;
+  bool save_table_creation_was_logged;
   DBUG_ENTER("select_create::create_table_from_items");
 
   tmp_table.s= &share;
@@ -4367,6 +4378,14 @@ TABLE *select_create::create_table_from_items(THD *thd, List<Item> *items,
 
   table->reginfo.lock_type=TL_WRITE;
   hooks->prelock(&table, 1);                    // Call prelock hooks
+
+  /*
+    Ensure that decide_logging_format(), called by mysql_lock_tables(), works
+    with temporary tables that will be logged later if needed.
+  */
+  save_table_creation_was_logged= table->s->table_creation_was_logged;
+  table->s->table_creation_was_logged= 1;
+
   /*
     mysql_lock_tables() below should never fail with request to reopen table
     since it won't wait for the table lock (we have exclusive metadata lock on
@@ -4379,8 +4398,11 @@ TABLE *select_create::create_table_from_items(THD *thd, List<Item> *items,
     /*
       This can happen in innodb when you get a deadlock when using same table
       in insert and select or when you run out of memory.
+      It can also happen if there was a conflict in
+      THD::decide_logging_format()
     */
-    my_error(ER_CANT_LOCK, MYF(0), my_errno);
+    if (!thd->is_error())
+      my_error(ER_CANT_LOCK, MYF(0), my_errno);
     if (*lock)
     {
       mysql_unlock_tables(thd, *lock);
@@ -4390,6 +4412,7 @@ TABLE *select_create::create_table_from_items(THD *thd, List<Item> *items,
     DBUG_RETURN(0);
     /* purecov: end */
   }
+  table->s->table_creation_was_logged= save_table_creation_was_logged;
   DBUG_RETURN(table);
 }
 
@@ -4593,7 +4616,7 @@ select_create::binlog_show_create_table(TABLE **tables, uint count)
                               /* is_trans */ TRUE,
                               /* direct */ FALSE,
                               /* suppress_use */ FALSE,
-                              errcode);
+                              errcode) > 0;
   }
 #ifdef WITH_WSREP
   if (thd->wsrep_trx().active())
@@ -4717,8 +4740,6 @@ bool select_create::send_eof()
     }
 #endif /* WITH_WSREP */
   }
-  else if (!thd->is_current_stmt_binlog_format_row())
-    table->s->table_creation_was_logged= 1;
 
   /*
     exit_done must only be set after last potential call to
@@ -4804,7 +4825,8 @@ void select_create::abort_result_set()
   if (table)
   {
     bool tmp_table= table->s->tmp_table;
-
+    bool table_creation_was_logged= (!tmp_table ||
+                                     table->s->table_creation_was_logged);
     if (tmp_table)
     {
       DBUG_ASSERT(saved_tmp_table_share);
@@ -4833,7 +4855,9 @@ void select_create::abort_result_set()
       /* Remove logging of drop, create + insert rows */
       binlog_reset_cache(thd);
       /* Original table was deleted. We have to log it */
-      log_drop_table(thd, &create_table->db, &create_table->table_name, tmp_table);
+      if (table_creation_was_logged)
+        log_drop_table(thd, &create_table->db, &create_table->table_name,
+                       tmp_table);
     }
   }
   DBUG_VOID_RETURN;
diff --git a/sql/sql_lex.cc b/sql/sql_lex.cc
index 34dcb0bcbe9..e5e6b1f3afd 100644
--- a/sql/sql_lex.cc
+++ b/sql/sql_lex.cc
@@ -7723,7 +7723,7 @@ uint binlog_unsafe_map[256];
 
 #define UNSAFE(a, b, c) \
   { \
-  DBUG_PRINT("unsafe_mixed_statement", ("SETTING BASE VALUES: %s, %s, %02X\n", \
+  DBUG_PRINT("unsafe_mixed_statement", ("SETTING BASE VALUES: %s, %s, %02X", \
     LEX::stmt_accessed_table_string(a), \
     LEX::stmt_accessed_table_string(b), \
     c)); \
diff --git a/sql/sql_lex.h b/sql/sql_lex.h
index e34cd2ebfd7..7fc905528d6 100644
--- a/sql/sql_lex.h
+++ b/sql/sql_lex.h
@@ -2208,6 +2208,14 @@ public:
                 ((1U << STMT_READS_TEMP_TRANS_TABLE) |
                  (1U << STMT_WRITES_TEMP_TRANS_TABLE))) != 0);
   }
+  inline bool stmt_writes_to_non_temp_table()
+  {
+    DBUG_ENTER("THD::stmt_writes_to_non_temp_table");
+
+    DBUG_RETURN((stmt_accessed_table_flag &
+                ((1U << STMT_WRITES_TRANS_TABLE) |
+                 (1U << STMT_WRITES_NON_TRANS_TABLE))));
+  }
 
   /**
     Checks if a temporary non-transactional table is about to be accessed
@@ -2259,7 +2267,7 @@ public:
       unsafe= (binlog_unsafe_map[stmt_accessed_table_flag] & condition);
 
 #if !defined(DBUG_OFF)
-      DBUG_PRINT("LEX::is_mixed_stmt_unsafe", ("RESULT %02X %02X %02X\n", condition,
+      DBUG_PRINT("LEX::is_mixed_stmt_unsafe", ("RESULT %02X %02X %02X", condition,
               binlog_unsafe_map[stmt_accessed_table_flag],
               (binlog_unsafe_map[stmt_accessed_table_flag] & condition)));
  
@@ -4593,6 +4601,8 @@ public:
                           const Lex_length_and_dec_st &attr);
   bool set_cast_type_udt(Lex_cast_type_st *type,
                          const LEX_CSTRING &name);
+
+  void mark_first_table_as_inserting();
 };
 
 
diff --git a/sql/sql_load.cc b/sql/sql_load.cc
index 3fa6e095f10..89cc3f8da64 100644
--- a/sql/sql_load.cc
+++ b/sql/sql_load.cc
@@ -2053,7 +2053,7 @@ int READ_INFO::read_xml(THD *thd)
       chr= read_value(delim, &value);
       if (attribute.length() > 0 && value.length() > 0)
       {
-        DBUG_PRINT("read_xml", ("lev:%i att:%s val:%s\n",
+        DBUG_PRINT("read_xml", ("lev:%i att:%s val:%s",
                                 level + 1,
                                 attribute.c_ptr_safe(),
                                 value.c_ptr_safe()));
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 0493176b264..beda00592e1 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -610,11 +610,12 @@ void init_update_queries(void)
                                             CF_CAN_GENERATE_ROW_EVENTS |
                                             CF_OPTIMIZER_TRACE |
                                             CF_CAN_BE_EXPLAINED |
-                                            CF_SP_BULK_SAFE;
+                                            CF_SP_BULK_SAFE | CF_DELETES_DATA;
   sql_command_flags[SQLCOM_DELETE_MULTI]=   CF_CHANGES_DATA | CF_REEXECUTION_FRAGILE |
                                             CF_CAN_GENERATE_ROW_EVENTS |
                                             CF_OPTIMIZER_TRACE |
-                                            CF_CAN_BE_EXPLAINED;
+                                            CF_CAN_BE_EXPLAINED |
+                                            CF_DELETES_DATA;
   sql_command_flags[SQLCOM_REPLACE]=        CF_CHANGES_DATA | CF_REEXECUTION_FRAGILE |
                                             CF_CAN_GENERATE_ROW_EVENTS |
                                             CF_OPTIMIZER_TRACE |
@@ -3313,6 +3314,10 @@ mysql_execute_command(THD *thd)
 #endif
   DBUG_ENTER("mysql_execute_command");
 
+  // check that we correctly marked first table for data insertion
+  DBUG_ASSERT(!(sql_command_flags[lex->sql_command] & CF_INSERTS_DATA) ||
+              first_table->for_insert_data);
+
   if (thd->security_ctx->password_expired &&
       lex->sql_command != SQLCOM_SET_OPTION)
   {
@@ -6840,11 +6845,7 @@ check_access(THD *thd, ulong want_access, const char *db, ulong *save_priv,
 bool check_single_table_access(THD *thd, ulong privilege, 
                                TABLE_LIST *all_tables, bool no_errors)
 {
-  Security_context * backup_ctx= thd->security_ctx;
-
-  /* we need to switch to the saved context (if any) */
-  if (all_tables->security_ctx)
-    thd->security_ctx= all_tables->security_ctx;
+  Switch_to_definer_security_ctx backup_sctx(thd, all_tables);
 
   const char *db_name;
   if ((all_tables->view || all_tables->field_translation) &&
@@ -6857,20 +6858,15 @@ bool check_single_table_access(THD *thd, ulong privilege,
                    &all_tables->grant.privilege,
                    &all_tables->grant.m_internal,
                    0, no_errors))
-    goto deny;
+    return 1;
 
   /* Show only 1 table for check_grant */
   if (!(all_tables->belong_to_view &&
         (thd->lex->sql_command == SQLCOM_SHOW_FIELDS)) &&
       check_grant(thd, privilege, all_tables, FALSE, 1, no_errors))
-    goto deny;
+    return 1;
 
-  thd->security_ctx= backup_ctx;
   return 0;
-
-deny:
-  thd->security_ctx= backup_ctx;
-  return 1;
 }
 
 /**
@@ -7045,7 +7041,6 @@ check_table_access(THD *thd, ulong requirements,TABLE_LIST *tables,
 {
   TABLE_LIST *org_tables= tables;
   TABLE_LIST *first_not_own_table= thd->lex->first_not_own_table();
-  Security_context *sctx= thd->security_ctx, *backup_ctx= thd->security_ctx;
   uint i= 0;
   /*
     The check that first_not_own_table is not reached is for the case when
@@ -7057,12 +7052,9 @@ check_table_access(THD *thd, ulong requirements,TABLE_LIST *tables,
   {
     TABLE_LIST *const table_ref= tables->correspondent_table ?
       tables->correspondent_table : tables;
+    Switch_to_definer_security_ctx backup_ctx(thd, table_ref);
 
     ulong want_access= requirements;
-    if (table_ref->security_ctx)
-      sctx= table_ref->security_ctx;
-    else
-      sctx= backup_ctx;
 
     /*
        Register access for view underlying table.
@@ -7073,7 +7065,7 @@ check_table_access(THD *thd, ulong requirements,TABLE_LIST *tables,
     if (table_ref->schema_table_reformed)
     {
       if (check_show_access(thd, table_ref))
-        goto deny;
+        return 1;
       continue;
     }
 
@@ -7083,8 +7075,6 @@ check_table_access(THD *thd, ulong requirements,TABLE_LIST *tables,
     if (table_ref->is_anonymous_derived_table())
       continue;
 
-    thd->security_ctx= sctx;
-
     if (table_ref->sequence)
     {
       /* We want to have either SELECT or INSERT rights to sequences depending
@@ -7098,15 +7088,11 @@ check_table_access(THD *thd, ulong requirements,TABLE_LIST *tables,
                      &table_ref->grant.privilege,
                      &table_ref->grant.m_internal,
                      0, no_errors))
-      goto deny;
+      return 1;
   }
-  thd->security_ctx= backup_ctx;
   return check_grant(thd,requirements,org_tables,
                      any_combination_of_privileges_will_do,
                      number, no_errors);
-deny:
-  thd->security_ctx= backup_ctx;
-  return TRUE;
 }
 
 
@@ -10401,3 +10387,14 @@ CHARSET_INFO *find_bin_collation(CHARSET_INFO *cs)
   }
   return cs;
 }
+
+void LEX::mark_first_table_as_inserting()
+{
+  TABLE_LIST *t= first_select_lex()->table_list.first;
+  DBUG_ENTER("Query_tables_list::mark_tables_with_important_flags");
+  DBUG_ASSERT(sql_command_flags[sql_command] & CF_INSERTS_DATA);
+  t->for_insert_data= TRUE;
+  DBUG_PRINT("info", ("table_list: %p  name: %s  db: %s  command: %u",
+                      t, t->table_name.str,t->db.str, sql_command));
+  DBUG_VOID_RETURN;
+}
diff --git a/sql/sql_prepare.cc b/sql/sql_prepare.cc
index e5723cb3b78..424a039a5be 100644
--- a/sql/sql_prepare.cc
+++ b/sql/sql_prepare.cc
@@ -3303,7 +3303,7 @@ void mysql_sql_stmt_execute(THD *thd)
   /* Query text for binary, general or slow log, if any of them is open */
   String expanded_query;
   DBUG_ENTER("mysql_sql_stmt_execute");
-  DBUG_PRINT("info", ("EXECUTE: %.*s\n", (int) name->length, name->str));
+  DBUG_PRINT("info", ("EXECUTE: %.*s", (int) name->length, name->str));
 
   if (!(stmt= (Prepared_statement*) thd->stmt_map.find_by_name(name)))
   {
@@ -3541,7 +3541,7 @@ void mysql_sql_stmt_close(THD *thd)
 {
   Prepared_statement* stmt;
   const LEX_CSTRING *name= &thd->lex->prepared_stmt.name();
-  DBUG_PRINT("info", ("DEALLOCATE PREPARE: %.*s\n", (int) name->length,
+  DBUG_PRINT("info", ("DEALLOCATE PREPARE: %.*s", (int) name->length,
                       name->str));
 
   if (! (stmt= (Prepared_statement*) thd->stmt_map.find_by_name(name)))
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index 4b47f9683eb..43914a0003d 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -2860,8 +2860,12 @@ void mysqld_list_processes(THD *thd,const char *user, bool verbose)
     else
       protocol->store_null();
     protocol->store(thd_info->state_info, system_charset_info);
-    protocol->store(thd_info->query_string.str(),
-                    thd_info->query_string.charset());
+    if (thd_info->query_string.length())
+      protocol->store(thd_info->query_string.str(),
+                      thd_info->query_string.length(),
+                      thd_info->query_string.charset());
+    else
+      protocol->store_null();
     if (!thd->variables.old_mode &&
         !(thd->variables.old_behavior & OLD_MODE_NO_PROGRESS_INFO))
       protocol->store(thd_info->progress, 3, &store_buffer);
@@ -8019,8 +8023,7 @@ TABLE *create_schema_table(THD *thd, TABLE_LIST *table_list)
 
   mark_all_fields_used_in_query(thd, fields_info, &bitmap, all_items);
 
-  TMP_TABLE_PARAM *tmp_table_param =
-    (TMP_TABLE_PARAM*) (thd->alloc(sizeof(TMP_TABLE_PARAM)));
+  TMP_TABLE_PARAM *tmp_table_param = new (thd->mem_root) TMP_TABLE_PARAM;
   tmp_table_param->init();
   tmp_table_param->table_charset= system_charset_info;
   tmp_table_param->field_count= field_count;
@@ -8577,6 +8580,7 @@ bool get_schema_tables_result(JOIN *join,
         cond= tab->cache_select->cond;
       }
 
+      Switch_to_definer_security_ctx backup_ctx(thd, table_list);
       if (table_list->schema_table->fill_table(thd, table_list, cond))
       {
         result= 1;
diff --git a/sql/sql_statistics.cc b/sql/sql_statistics.cc
index abf6a8a3cad..a94fb1196b4 100644
--- a/sql/sql_statistics.cc
+++ b/sql/sql_statistics.cc
@@ -230,10 +230,8 @@ index_stat_def= {INDEX_STAT_N_FIELDS, index_stat_fields, 4, index_stat_pk_col};
   Open all statistical tables and lock them
 */
 
-static
-inline int open_stat_tables(THD *thd, TABLE_LIST *tables,
-                            Open_tables_backup *backup,
-                            bool for_write)
+static int open_stat_tables(THD *thd, TABLE_LIST *tables,
+                            Open_tables_backup *backup, bool for_write)
 {
   int rc;
 
@@ -241,12 +239,14 @@ inline int open_stat_tables(THD *thd, TABLE_LIST *tables,
   thd->push_internal_handler(&deh);
   init_table_list_for_stat_tables(tables, for_write);
   init_mdl_requests(tables);
+  thd->in_sub_stmt|= SUB_STMT_STAT_TABLES;
   rc= open_system_tables_for_read(thd, tables, backup);
+  thd->in_sub_stmt&= ~SUB_STMT_STAT_TABLES;
   thd->pop_internal_handler();
 
 
   /* If the number of tables changes, we should revise the check below. */
-  DBUG_ASSERT(STATISTICS_TABLES == 3);
+  compile_time_assert(STATISTICS_TABLES == 3);
 
   if (!rc &&
       (stat_table_intact.check(tables[TABLE_STAT].table, &table_stat_def) ||
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index f824abbcabc..2917a320603 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -1987,7 +1987,7 @@ int write_bin_log(THD *thd, bool clear_error,
       errcode= query_error_code(thd, TRUE);
     error= thd->binlog_query(THD::STMT_QUERY_TYPE,
                              query, query_length, is_trans, FALSE, FALSE,
-                             errcode);
+                             errcode) > 0;
     thd_proc_info(thd, 0);
   }
   return error;
@@ -2603,24 +2603,24 @@ err:
           /* Chop of the last comma */
           built_non_trans_tmp_query.chop();
           built_non_trans_tmp_query.append(" /* generated by server */");
-          error |= thd->binlog_query(THD::STMT_QUERY_TYPE,
-                                     built_non_trans_tmp_query.ptr(),
-                                     built_non_trans_tmp_query.length(),
-                                     FALSE, FALSE,
-                                     is_drop_tmp_if_exists_added,
-                                     0);
+          error |= (thd->binlog_query(THD::STMT_QUERY_TYPE,
+                                      built_non_trans_tmp_query.ptr(),
+                                      built_non_trans_tmp_query.length(),
+                                      FALSE, FALSE,
+                                      is_drop_tmp_if_exists_added,
+                                      0) > 0);
       }
       if (trans_tmp_table_deleted)
       {
           /* Chop of the last comma */
           built_trans_tmp_query.chop();
           built_trans_tmp_query.append(" /* generated by server */");
-          error |= thd->binlog_query(THD::STMT_QUERY_TYPE,
-                                     built_trans_tmp_query.ptr(),
-                                     built_trans_tmp_query.length(),
-                                     TRUE, FALSE,
-                                     is_drop_tmp_if_exists_added,
-                                     0);
+          error |= (thd->binlog_query(THD::STMT_QUERY_TYPE,
+                                      built_trans_tmp_query.ptr(),
+                                      built_trans_tmp_query.length(),
+                                      TRUE, FALSE,
+                                      is_drop_tmp_if_exists_added,
+                                      0) > 0);
       }
       if (non_tmp_table_deleted)
       {
@@ -2629,11 +2629,11 @@ err:
           built_query.append(" /* generated by server */");
           int error_code = non_tmp_error ?  thd->get_stmt_da()->sql_errno()
                                          : 0;
-          error |= thd->binlog_query(THD::STMT_QUERY_TYPE,
-                                     built_query.ptr(),
-                                     built_query.length(),
-                                     TRUE, FALSE, FALSE,
-                                     error_code);
+          error |= (thd->binlog_query(THD::STMT_QUERY_TYPE,
+                                      built_query.ptr(),
+                                      built_query.length(),
+                                      TRUE, FALSE, FALSE,
+                                      error_code) > 0);
       }
     }
   }
@@ -2716,7 +2716,7 @@ bool log_drop_table(THD *thd, const LEX_CSTRING *db_name,
                                "failed CREATE OR REPLACE */"));
   error= thd->binlog_query(THD::STMT_QUERY_TYPE,
                            query.ptr(), query.length(),
-                           FALSE, FALSE, temporary_table, 0);
+                           FALSE, FALSE, temporary_table, 0) > 0;
   DBUG_RETURN(error);
 }
 
@@ -5281,9 +5281,13 @@ bool mysql_create_table(THD *thd, TABLE_LIST *create_table,
   }
 
 err:
-  /* In RBR we don't need to log CREATE TEMPORARY TABLE */
-  if (!result && thd->is_current_stmt_binlog_format_row() && create_info->tmp_table())
+  /* In RBR or readonly server we don't need to log CREATE TEMPORARY TABLE */
+  if (!result && create_info->tmp_table() &&
+      (thd->is_current_stmt_binlog_format_row() || (opt_readonly && !thd->slave_thread)))
+  {
+    /* Note that table->s->table_creation_was_logged is not set! */
     DBUG_RETURN(result);
+  }
 
   if (create_info->tmp_table())
     thd->transaction.stmt.mark_created_temp_table();
@@ -5300,11 +5304,13 @@ err:
       */
       thd->locked_tables_list.unlock_locked_table(thd, mdl_ticket);
     }
-    else if (likely(!result) && create_info->tmp_table() && create_info->table)
+    else if (likely(!result) && create_info->table)
     {
       /*
-        Remember that tmp table creation was logged so that we know if
+        Remember that table creation was logged so that we know if
         we should log a delete of it.
+        If create_info->table was not set, it's a normal table and
+        table_creation_was_logged will be set when the share is created.
       */
       create_info->table->s->table_creation_was_logged= 1;
     }
@@ -6893,7 +6899,7 @@ static bool fill_alter_inplace_info(THD *thd, TABLE *table, bool varchar,
           }
 
           if (field->vcol_info->is_in_partitioning_expr() ||
-              field->flags & PART_KEY_FLAG)
+              field->flags & PART_KEY_FLAG || field->stored_in_db())
           {
             if (value_changes)
               ha_alter_info->handler_flags|= ALTER_COLUMN_VCOL;
diff --git a/sql/sql_type.cc b/sql/sql_type.cc
index f43599d682b..323f693302f 100644
--- a/sql/sql_type.cc
+++ b/sql/sql_type.cc
@@ -4030,7 +4030,6 @@ uint32 Type_handler_temporal_result::max_display_length(const Item *item) const
   return item->max_length;
 }
 
-
 uint32 Type_handler_string_result::max_display_length(const Item *item) const
 {
   return item->max_length;
@@ -4048,6 +4047,50 @@ uint32 Type_handler_bit::max_display_length(const Item *item) const
   return item->max_length;
 }
 
+/*************************************************************************/
+
+uint32 
+Type_handler_decimal_result::Item_decimal_notation_int_digits(const Item *item)
+                                                              const
+{
+  return item->decimal_int_part();
+}
+
+
+uint32 
+Type_handler_temporal_result::Item_decimal_notation_int_digits(const Item *item)
+                                                               const
+{
+  return item->decimal_int_part();
+}
+
+
+uint32 
+Type_handler_bit::Item_decimal_notation_int_digits(const Item *item)
+                                                              const
+{
+  return Bit_decimal_notation_int_digits(item);
+}
+
+
+uint32
+Type_handler_general_purpose_int::Item_decimal_notation_int_digits(
+                                                   const Item *item) const
+{
+  return type_limits_int()->precision();
+}
+
+/*************************************************************************/
+
+/*
+    Decimal to binary digits ratio converges to log2(10) thus using 3 as
+    a divisor.
+*/
+uint32
+Type_handler_bit::Bit_decimal_notation_int_digits(const Item *item)
+{
+  return item->max_length/3+1;
+}
 
 /*************************************************************************/
 
diff --git a/sql/sql_type.h b/sql/sql_type.h
index dda5768e753..3943a3f761f 100644
--- a/sql/sql_type.h
+++ b/sql/sql_type.h
@@ -3752,6 +3752,7 @@ public:
                           SORT_FIELD_ATTR *attr) const= 0;
 
   virtual uint32 max_display_length(const Item *item) const= 0;
+  virtual uint32 Item_decimal_notation_int_digits(const Item *item) const { return 0; }
   virtual uint32 calc_pack_length(uint32 length) const= 0;
   virtual uint calc_key_length(const Column_definition &def) const;
   virtual void Item_update_null_value(Item *item) const= 0;
@@ -4594,6 +4595,7 @@ public:
                   const Type_std_attributes *item,
                   SORT_FIELD_ATTR *attr) const override;
   uint32 max_display_length(const Item *item) const override;
+  uint32 Item_decimal_notation_int_digits(const Item *item) const override;
   Item *create_typecast_item(THD *thd, Item *item,
                              const Type_cast_attributes &attr) const override;
   bool Item_const_eq(const Item_const *a, const Item_const *b,
@@ -4908,6 +4910,7 @@ public:
   {
     return type_limits_int()->char_length();
   }
+  uint32 Item_decimal_notation_int_digits(const Item *item) const override;
   bool partition_field_check(const LEX_CSTRING &field_name,
                              Item *item_expr) const override
   {
@@ -4946,6 +4949,7 @@ public:
                                  const Type_all_attributes *attr,
                                  const st_value *value) const override;
   uint32 max_display_length(const Item *item) const override;
+  uint32 Item_decimal_notation_int_digits(const Item *item) const override;
   bool can_change_cond_ref_to_const(Item_bool_func2 *target,
                                    Item *target_expr, Item *target_value,
                                    Item_bool_func2 *source,
@@ -5043,6 +5047,12 @@ public:
   Column_definition_attributes_frm_pack(const Column_definition_attributes *at,
                                         uchar *buff) const override;
   uint32 max_display_length(const Item *item) const override;
+  /*
+    The next method returns 309 for long stringified doubles in scientific
+    notation, e.g. FORMAT('1e308', 2).
+  */
+  uint32 Item_decimal_notation_int_digits(const Item *item) const override
+  { return 309; }
   bool Item_const_eq(const Item_const *a, const Item_const *b,
                      bool binary_cmp) const override;
   bool Item_eq_value(THD *thd, const Type_cmp_attributes *attr,
@@ -5472,6 +5482,8 @@ public:
     return PROTOCOL_SEND_SHORT;
   }
   uint32 max_display_length(const Item *item) const override;
+  uint32 Item_decimal_notation_int_digits(const Item *item) const override
+  { return 4; };
   uint32 max_display_length_for_field(const Conv_source &src) const override
   { return 4; }
   uint32 calc_pack_length(uint32 length) const override { return 1; }
@@ -5522,6 +5534,8 @@ public:
     return PROTOCOL_SEND_STRING;
   }
   uint32 max_display_length(const Item *item) const override;
+  uint32 Item_decimal_notation_int_digits(const Item *item) const override;
+  static uint32 Bit_decimal_notation_int_digits(const Item *item); 
   uint32 max_display_length_for_field(const Conv_source &src) const override;
   uint32 calc_pack_length(uint32 length) const override { return length / 8; }
   uint calc_key_length(const Column_definition &def) const override;
@@ -5634,6 +5648,8 @@ public:
   }
   bool type_can_have_auto_increment_attribute() const override { return true; }
   uint32 max_display_length(const Item *item) const override { return 53; }
+  uint32 Item_decimal_notation_int_digits(const Item *item) const override
+  { return 309; }
   uint32 max_display_length_for_field(const Conv_source &src) const override
   { return 22; }
   uint32 calc_pack_length(uint32 length) const override
diff --git a/sql/sql_update.cc b/sql/sql_update.cc
index 1b589d941b2..9f298a51b72 100644
--- a/sql/sql_update.cc
+++ b/sql/sql_update.cc
@@ -1262,7 +1262,7 @@ update_end:
 
       if (thd->binlog_query(THD::ROW_QUERY_TYPE,
                             thd->query(), thd->query_length(),
-                            transactional_table, FALSE, FALSE, errcode))
+                            transactional_table, FALSE, FALSE, errcode) > 0)
       {
         error=1;				// Rollback update
       }
@@ -2991,7 +2991,7 @@ bool multi_update::send_eof()
 
       if (thd->binlog_query(THD::ROW_QUERY_TYPE, thd->query(),
                             thd->query_length(), transactional_tables, FALSE,
-                            FALSE, errcode))
+                            FALSE, errcode) > 0)
 	local_error= 1;				// Rollback update
       thd->set_current_stmt_binlog_format(save_binlog_format);
     }
diff --git a/sql/sql_view.cc b/sql/sql_view.cc
index ca66da1fd0d..2011abf79ba 100644
--- a/sql/sql_view.cc
+++ b/sql/sql_view.cc
@@ -698,7 +698,7 @@ bool mysql_create_view(THD *thd, TABLE_LIST *views,
     thd->reset_unsafe_warnings();
     if (thd->binlog_query(THD::STMT_QUERY_TYPE,
                           buff.ptr(), buff.length(), FALSE, FALSE, FALSE,
-                          errcode))
+                          errcode) > 0)
       res= TRUE;
   }
 
@@ -1498,6 +1498,7 @@ bool mysql_make_view(THD *thd, TABLE_SHARE *share, TABLE_LIST *table,
         privileges of top_view
       */
       tbl->grant.want_privilege= SELECT_ACL;
+
       /*
         After unfolding the view we lose the list of tables referenced in it
         (we will have only a list of underlying tables in case of MERGE
@@ -1548,6 +1549,18 @@ bool mysql_make_view(THD *thd, TABLE_SHARE *share, TABLE_LIST *table,
         views with subqueries in select list.
       */
       view_main_select_tables= lex->first_select_lex()->table_list.first;
+      /*
+        Mergeable view can be used for inserting, so we move the flag down
+      */
+      if (table->for_insert_data)
+      {
+        for (TABLE_LIST *t= view_main_select_tables;
+             t;
+             t= t->next_local)
+        {
+          t->for_insert_data= TRUE;
+        }
+      }
 
       /*
         Let us set proper lock type for tables of the view's main
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index e111b268e65..b06d2f5a80c 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -7836,6 +7836,7 @@ alter:
             Lex->first_select_lex()->db=
               (Lex->first_select_lex()->table_list.first)->db;
             Lex->create_last_non_select_table= Lex->last_table();
+            Lex->mark_first_table_as_inserting();
           }
           alter_commands
           {
@@ -13483,7 +13484,10 @@ insert:
             Select->set_lock_for_tables($4, true);
           }
           insert_field_spec opt_insert_update opt_returning
-          stmt_end {}
+          stmt_end
+          {
+            Lex->mark_first_table_as_inserting();
+          }
           ;
 
 replace:
@@ -13497,7 +13501,10 @@ replace:
             Select->set_lock_for_tables($4, true);
           }
           insert_field_spec opt_returning
-          stmt_end {}
+          stmt_end
+          {
+            Lex->mark_first_table_as_inserting();
+          }
           ;
 
 insert_start: {
@@ -15008,7 +15015,10 @@ load:
           opt_xml_rows_identified_by
           opt_field_term opt_line_term opt_ignore_lines opt_field_or_var_spec
           opt_load_data_set_spec
-          stmt_end {}
+          stmt_end
+          {
+            Lex->mark_first_table_as_inserting();
+          }
           ;
 
 data_or_xml:
diff --git a/sql/sql_yacc_ora.yy b/sql/sql_yacc_ora.yy
index c3727c4819d..c18acd3ef96 100644
--- a/sql/sql_yacc_ora.yy
+++ b/sql/sql_yacc_ora.yy
@@ -7799,6 +7799,7 @@ alter:
             Lex->first_select_lex()->db=
               (Lex->first_select_lex()->table_list.first)->db;
             Lex->create_last_non_select_table= Lex->last_table();
+            Lex->mark_first_table_as_inserting();
           }
           alter_commands
           {
@@ -13471,7 +13472,10 @@ insert:
             Select->set_lock_for_tables($4, true);
           }
           insert_field_spec opt_insert_update opt_returning
-          stmt_end {}
+          stmt_end
+          {
+            Lex->mark_first_table_as_inserting();
+          }
           ;
 
 replace:
@@ -13485,7 +13489,10 @@ replace:
             Select->set_lock_for_tables($4, true);
           }
           insert_field_spec opt_returning
-          stmt_end {}
+          stmt_end
+          {
+            Lex->mark_first_table_as_inserting();
+          }
           ;
 
 insert_start: {
@@ -15003,7 +15010,10 @@ load:
           opt_xml_rows_identified_by
           opt_field_term opt_line_term opt_ignore_lines opt_field_or_var_spec
           opt_load_data_set_spec
-          stmt_end {}
+          stmt_end
+          {
+            Lex->mark_first_table_as_inserting();
+          }
           ;
 
 data_or_xml:
diff --git a/sql/table.cc b/sql/table.cc
index 8dc4cc5e9a0..b745ce22946 100644
--- a/sql/table.cc
+++ b/sql/table.cc
@@ -695,7 +695,11 @@ enum open_frm_error open_table_def(THD *thd, TABLE_SHARE *share, uint flags)
   frmlen= read_length + sizeof(head);
 
   share->init_from_binary_frm_image(thd, false, buf, frmlen);
-  error_given= true; // init_from_binary_frm_image has already called my_error()
+  /*
+    Don't give any additional errors. If there would be a problem,
+    init_from_binary_frm_image would call my_error() itself.
+  */
+  error_given= true;
   my_free(buf);
 
   goto err_not_open;
@@ -704,6 +708,9 @@ err:
   mysql_file_close(file, MYF(MY_WME));
 
 err_not_open:
+  /* Mark that table was created earlier and thus should have been logged */
+  share->table_creation_was_logged= 1;
+
   if (unlikely(share->error && !error_given))
   {
     share->open_errno= my_errno;
@@ -3324,6 +3331,8 @@ ret:
              sql_copy);
     DBUG_RETURN(HA_ERR_GENERIC);
   }
+  /* Treat the table as normal table from binary logging point of view */
+  table_creation_was_logged= 1;
   DBUG_RETURN(0);
 }
 
diff --git a/sql/table.h b/sql/table.h
index 898c31c4aff..b9ab61e85fd 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -752,10 +752,15 @@ struct TABLE_SHARE
   bool system;                          /* Set if system table (one record) */
   bool not_usable_by_query_cache;
   bool online_backup;                   /* Set if on-line backup supported */
+  /*
+    This is used by log tables, for tables that have their own internal
+    binary logging or for tables that doesn't support statement or row logging
+   */
   bool no_replicate;
   bool crashed;
   bool is_view;
   bool can_cmp_whole_record;
+  /* This is set for temporary tables where CREATE was binary logged */
   bool table_creation_was_logged;
   bool non_determinstic_insert;
   bool vcols_need_refixing;
@@ -1989,7 +1994,8 @@ struct TABLE_LIST
                                             prelocking_types prelocking_type,
                                             TABLE_LIST *belong_to_view_arg,
                                             uint8 trg_event_map_arg,
-                                            TABLE_LIST ***last_ptr)
+                                            TABLE_LIST ***last_ptr,
+                                            my_bool insert_data)
 
   {
     init_one_table(db_arg, table_name_arg, alias_arg, lock_type_arg);
@@ -2004,6 +2010,7 @@ struct TABLE_LIST
     **last_ptr= this;
     prev_global= *last_ptr;
     *last_ptr= &next_global;
+    for_insert_data= insert_data;
   }
 
 
@@ -2430,6 +2437,8 @@ struct TABLE_LIST
     return period_conditions.is_set();
   }
 
+  my_bool for_insert_data;
+
   /**
      @brief
        Find the bottom in the chain of embedded table VIEWs.
diff --git a/sql/wsrep_binlog.cc b/sql/wsrep_binlog.cc
index cfa709b1055..a81503a50da 100644
--- a/sql/wsrep_binlog.cc
+++ b/sql/wsrep_binlog.cc
@@ -123,7 +123,7 @@ static int wsrep_write_cache_inc(THD*      const thd,
   DBUG_ENTER("wsrep_write_cache_inc");
   my_off_t const saved_pos(my_b_tell(cache));
 
-  if (reinit_io_cache(cache, READ_CACHE, thd->wsrep_sr().bytes_certified(), 0, 0))
+  if (reinit_io_cache(cache, READ_CACHE, thd->wsrep_sr().log_position(), 0, 0))
   {
     WSREP_ERROR("failed to initialize io-cache");
     DBUG_RETURN(1);;
@@ -158,7 +158,7 @@ static int wsrep_write_cache_inc(THD*      const thd,
   }
   if (ret == 0)
   {
-    assert(total_length + thd->wsrep_sr().bytes_certified() == saved_pos);
+    assert(total_length + thd->wsrep_sr().log_position() == saved_pos);
   }
 
 cleanup:
diff --git a/sql/wsrep_client_service.cc b/sql/wsrep_client_service.cc
index 8d58f62bd03..696a097db21 100644
--- a/sql/wsrep_client_service.cc
+++ b/sql/wsrep_client_service.cc
@@ -137,7 +137,8 @@ void Wsrep_client_service::cleanup_transaction()
 }
 
 
-int Wsrep_client_service::prepare_fragment_for_replication(wsrep::mutable_buffer& buffer)
+int Wsrep_client_service::prepare_fragment_for_replication(
+  wsrep::mutable_buffer& buffer, size_t& log_position)
 {
   DBUG_ASSERT(m_thd == current_thd);
   THD* thd= m_thd;
@@ -151,7 +152,7 @@ int Wsrep_client_service::prepare_fragment_for_replication(wsrep::mutable_buffer
   }
 
   const my_off_t saved_pos(my_b_tell(cache));
-  if (reinit_io_cache(cache, READ_CACHE, thd->wsrep_sr().bytes_certified(), 0, 0))
+  if (reinit_io_cache(cache, READ_CACHE, thd->wsrep_sr().log_position(), 0, 0))
   {
     DBUG_RETURN(1);
   }
@@ -185,6 +186,7 @@ int Wsrep_client_service::prepare_fragment_for_replication(wsrep::mutable_buffer
     while (cache->file >= 0 && (length= my_b_fill(cache)));
   }
   DBUG_ASSERT(total_length == buffer.size());
+  log_position= saved_pos;
 cleanup:
   if (reinit_io_cache(cache, WRITE_CACHE, saved_pos, 0, 0))
   {
diff --git a/sql/wsrep_client_service.h b/sql/wsrep_client_service.h
index b1695b7aedf..aa58d99c3cf 100644
--- a/sql/wsrep_client_service.h
+++ b/sql/wsrep_client_service.h
@@ -43,7 +43,7 @@ public:
   void cleanup_transaction();
   bool statement_allowed_for_streaming() const;
   size_t bytes_generated() const;
-  int prepare_fragment_for_replication(wsrep::mutable_buffer&);
+  int prepare_fragment_for_replication(wsrep::mutable_buffer&, size_t&);
   int remove_fragments();
   void emergency_shutdown()
   { throw wsrep::not_implemented_error(); }
diff --git a/sql/wsrep_dummy.cc b/sql/wsrep_dummy.cc
index 2ea434c092b..b5adef17442 100644
--- a/sql/wsrep_dummy.cc
+++ b/sql/wsrep_dummy.cc
@@ -66,7 +66,7 @@ const char *wsrep_thd_exec_mode_str(THD *)
 { return NULL; }
 
 const char *wsrep_thd_query(const THD *)
-{ return 0; }
+{ return "NULL"; }
 
 const char *wsrep_thd_query_state_str(THD *)
 { return 0; }
diff --git a/sql/wsrep_high_priority_service.cc b/sql/wsrep_high_priority_service.cc
index 84c89ccb705..987a1765dcf 100644
--- a/sql/wsrep_high_priority_service.cc
+++ b/sql/wsrep_high_priority_service.cc
@@ -222,6 +222,12 @@ const wsrep::transaction& Wsrep_high_priority_service::transaction() const
   DBUG_RETURN(m_thd->wsrep_trx());
 }
 
+int Wsrep_high_priority_service::next_fragment(const wsrep::ws_meta& ws_meta)
+{
+  DBUG_ENTER(" Wsrep_high_priority_service::next_fragment");
+  DBUG_RETURN(m_thd->wsrep_cs().next_fragment(ws_meta));
+}
+
 int Wsrep_high_priority_service::adopt_transaction(
   const wsrep::transaction& transaction)
 {
@@ -242,7 +248,8 @@ int Wsrep_high_priority_service::adopt_transaction(
 int Wsrep_high_priority_service::append_fragment_and_commit(
   const wsrep::ws_handle& ws_handle,
   const wsrep::ws_meta& ws_meta,
-  const wsrep::const_buffer& data)
+  const wsrep::const_buffer& data,
+  const wsrep::xid& xid WSREP_UNUSED)
 {
   DBUG_ENTER("Wsrep_high_priority_service::append_fragment_and_commit");
   int ret= start_transaction(ws_handle, ws_meta);
diff --git a/sql/wsrep_high_priority_service.h b/sql/wsrep_high_priority_service.h
index 5657a2e82fc..a14498037d7 100644
--- a/sql/wsrep_high_priority_service.h
+++ b/sql/wsrep_high_priority_service.h
@@ -34,13 +34,15 @@ public:
   ~Wsrep_high_priority_service();
   int start_transaction(const wsrep::ws_handle&,
                         const wsrep::ws_meta&);
+  int next_fragment(const wsrep::ws_meta&);
   const wsrep::transaction& transaction() const;
   int adopt_transaction(const wsrep::transaction&);
   int apply_write_set(const wsrep::ws_meta&, const wsrep::const_buffer&,
                       wsrep::mutable_buffer&) = 0;
   int append_fragment_and_commit(const wsrep::ws_handle&,
                                  const wsrep::ws_meta&,
-                                 const wsrep::const_buffer&);
+                                 const wsrep::const_buffer&,
+                                 const wsrep::xid&);
   int remove_fragments(const wsrep::ws_meta&);
   int commit(const wsrep::ws_handle&, const wsrep::ws_meta&);
   int rollback(const wsrep::ws_handle&, const wsrep::ws_meta&);
diff --git a/sql/wsrep_mysqld.cc b/sql/wsrep_mysqld.cc
index a6f9d3a6b17..e35d13be4c6 100644
--- a/sql/wsrep_mysqld.cc
+++ b/sql/wsrep_mysqld.cc
@@ -627,6 +627,7 @@ static wsrep::gtid wsrep_server_initial_position()
  */
 static void wsrep_init_provider_status_variables()
 {
+  wsrep_inited= 1;
   const wsrep::provider& provider=
     Wsrep_server_state::instance().provider();
   strncpy(provider_name,
@@ -711,9 +712,7 @@ int wsrep_init()
       WSREP_ERROR("wsrep::init() failed: %d, must shutdown", err);
     }
     else
-    {
       wsrep_init_provider_status_variables();
-    }
     return err;
   }
 
@@ -749,7 +748,6 @@ int wsrep_init()
     Wsrep_server_state::instance().unload_provider();
     return 1;
   }
-  wsrep_inited= 1;
 
   wsrep_init_provider_status_variables();
   wsrep_capabilities_export(Wsrep_server_state::instance().provider().capabilities(),
@@ -1705,26 +1703,6 @@ static bool wsrep_can_run_in_toi(THD *thd, const char *db, const char *table,
   }
 }
 
-#if UNUSED /* 323f269d4099 (Jan Lindström     2018-07-19) */
-static const char* wsrep_get_query_or_msg(const THD* thd)
-{
-  switch(thd->lex->sql_command)
-  {
-    case SQLCOM_CREATE_USER:
-      return "CREATE USER";
-    case SQLCOM_GRANT:
-      return "GRANT";
-    case SQLCOM_REVOKE:
-      return "REVOKE";
-    case SQLCOM_SET_OPTION:
-      if (thd->lex->definer)
-        return "SET PASSWORD";
-      /* fallthrough */
-    default:
-      return thd->query();
-   }
-}
-#endif //UNUSED
 
 static int wsrep_create_sp(THD *thd, uchar** buf, size_t* buf_len)
 {
@@ -2662,7 +2640,7 @@ void* start_wsrep_THD(void *arg)
     need to know the start of the stack so that we could check for
     stack overruns.
   */
-  DBUG_PRINT("wsrep", ("handle_one_connection called by thread %lld\n",
+  DBUG_PRINT("wsrep", ("handle_one_connection called by thread %lld",
                        (long long)thd->thread_id));
   /* now that we've called my_thread_init(), it is safe to call DBUG_* */
 
diff --git a/sql/wsrep_storage_service.cc b/sql/wsrep_storage_service.cc
index 6dfe3eee448..2ad817fe25a 100644
--- a/sql/wsrep_storage_service.cc
+++ b/sql/wsrep_storage_service.cc
@@ -100,7 +100,8 @@ void Wsrep_storage_service::adopt_transaction(const wsrep::transaction& transact
 int Wsrep_storage_service::append_fragment(const wsrep::id& server_id,
                                            wsrep::transaction_id transaction_id,
                                            int flags,
-                                           const wsrep::const_buffer& data)
+                                           const wsrep::const_buffer& data,
+                                           const wsrep::xid& xid WSREP_UNUSED)
 {
   DBUG_ENTER("Wsrep_storage_service::append_fragment");
   DBUG_ASSERT(m_thd == current_thd);
diff --git a/sql/wsrep_storage_service.h b/sql/wsrep_storage_service.h
index 6208300930f..f39543a89bc 100644
--- a/sql/wsrep_storage_service.h
+++ b/sql/wsrep_storage_service.h
@@ -33,7 +33,8 @@ public:
   int append_fragment(const wsrep::id&,
                       wsrep::transaction_id,
                       int flags,
-                      const wsrep::const_buffer&);
+                      const wsrep::const_buffer&,
+                      const wsrep::xid&);
   int update_fragment_meta(const wsrep::ws_meta&);
   int remove_fragments();
   int commit(const wsrep::ws_handle&, const wsrep::ws_meta&);