Merge 10.3 into 10.4

36 files changed, 366 insertions, 197 deletions

diff --git a/sql/ha_partition.cc b/sql/ha_partition.cc
index 5927b047f77..baf8f88c00c 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();
@@ -4474,11 +4474,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);
 }
@@ -8139,6 +8136,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;
@@ -8169,6 +8167,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,
@@ -8177,7 +8184,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);
@@ -8486,6 +8493,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 6407a607fe7..598c63837c7 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/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 a01b4f9660c..b90804ed9f0 100644
--- a/sql/log.cc
+++ b/sql/log.cc
@@ -5983,7 +5983,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);
 
@@ -6021,7 +6020,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);
 }
 
 
@@ -7733,7 +7732,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);
 }
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index bed82300099..e1a7a39df89 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -675,7 +675,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 */
 
@@ -1995,7 +1997,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();
@@ -4472,7 +4474,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 c51f5a1a6e2..f0d6fc74242 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 69c57b2959c..d645fea6968 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 43944366665..35bc1b83029 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 8568bc16c00..751e4644da0 100644
--- a/sql/sp.cc
+++ b/sql/sp.cc
@@ -1467,7 +1467,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 87664e16004..9581ded8a12 100644
--- a/sql/sp_head.cc
+++ b/sql/sp_head.cc
@@ -4726,6 +4726,7 @@ typedef struct st_sp_table
   uint lock_count;
   uint query_lock_count;
   uint8 trg_event_map;
+  my_bool for_insert_data;
 } SP_TABLE;
 
 
@@ -4821,6 +4822,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
       {
@@ -4844,6 +4846,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;
       }
@@ -4927,7 +4930,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 847d2bd777b..224549f2add 100644
--- a/sql/sql_acl.cc
+++ b/sql/sql_acl.cc
@@ -3786,7 +3786,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)
@@ -3938,7 +3938,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 4c67b14e8f5..799c85a5675 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -2669,6 +2669,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;
         }
@@ -4605,9 +4606,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
     */
@@ -4617,20 +4620,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);
 }
 
 
@@ -4661,6 +4666,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 ||
@@ -4691,7 +4697,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;
@@ -4719,7 +4725,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);
@@ -4727,8 +4734,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;
@@ -4741,10 +4751,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);
 }
 
 
@@ -4761,7 +4771,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 1ffb7fe258e..5fbe5704649 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>
@@ -5865,17 +5865,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.
@@ -5893,44 +5909,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
@@ -5941,17 +5957,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
@@ -5960,10 +5975,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)
@@ -6092,10 +6107,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), "");
 	  }
@@ -6164,10 +6186,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;
       /*
@@ -6187,8 +6207,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
         */
@@ -7025,11 +7045,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)
@@ -7055,7 +7076,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);
   }
 
   /*
@@ -7071,7 +7092,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);
+    }
   }
 
   /*
@@ -7114,7 +7138,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 */
 
     /*
@@ -7155,7 +7179,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 c98c0e4621c..3f732a76e53 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -1617,12 +1617,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
@@ -6474,6 +6478,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).
@@ -6678,6 +6687,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;
+};
+
 
 /**
   This class resembles the SQL Standard schema qualified object name:
diff --git a/sql/sql_delete.cc b/sql/sql_delete.cc
index a965a7115d2..c4bc32262e8 100644
--- a/sql/sql_delete.cc
+++ b/sql/sql_delete.cc
@@ -945,7 +945,7 @@ cleanup:
                                         transactional_table, FALSE, FALSE,
                                         errcode);
 
-      if (log_result)
+      if (log_result > 0)
       {
 	error=1;
       }
@@ -1640,7 +1640,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 7528239ca7b..3ad3cf9151f 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -1190,13 +1190,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;
       }
     }
@@ -3947,6 +3947,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");
@@ -3997,18 +3998,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))
@@ -4059,8 +4064,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
@@ -4112,16 +4118,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();
   }
 
@@ -4190,6 +4200,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;
@@ -4343,6 +4354,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
@@ -4355,8 +4374,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);
@@ -4366,6 +4388,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);
 }
 
@@ -4569,7 +4592,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())
@@ -4693,8 +4716,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
@@ -4780,7 +4801,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);
@@ -4809,7 +4831,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 6f9f2200a00..16bb53ca66c 100644
--- a/sql/sql_lex.cc
+++ b/sql/sql_lex.cc
@@ -7713,7 +7713,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 8d09407606a..159c05fb98f 100644
--- a/sql/sql_lex.h
+++ b/sql/sql_lex.h
@@ -2175,6 +2175,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
@@ -2226,7 +2234,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)));
  
@@ -4531,6 +4539,8 @@ public:
                                 Item_result return_type,
                                 const LEX_CSTRING &soname);
   Spvar_definition *row_field_name(THD *thd, const Lex_ident_sys_st &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 c0b78ead6d6..516813a0264 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -606,11 +606,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 |
@@ -3307,6 +3308,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)
   {
@@ -6751,11 +6756,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) &&
@@ -6768,20 +6769,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;
 }
 
 /**
@@ -6956,7 +6952,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
@@ -6968,12 +6963,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.
@@ -6984,7 +6976,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;
     }
 
@@ -6994,8 +6986,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
@@ -7009,15 +6999,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;
 }
 
 
@@ -10315,3 +10301,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_partition.cc b/sql/sql_partition.cc
index 65d79ac2d25..0c7be11bd59 100644
--- a/sql/sql_partition.cc
+++ b/sql/sql_partition.cc
@@ -146,7 +146,7 @@ Item* convert_charset_partition_constant(Item *item, CHARSET_INFO *cs)
   item= item->safe_charset_converter(thd, cs);
   context->table_list= NULL;
   thd->where= "convert character set partition constant";
-  if (item->fix_fields_if_needed(thd, (Item**)NULL))
+  if (item && item->fix_fields_if_needed(thd, (Item**)NULL))
     item= NULL;
   thd->where= save_where;
   context->table_list= save_list;
diff --git a/sql/sql_prepare.cc b/sql/sql_prepare.cc
index 8088b6923f2..5bf4e0a9c2f 100644
--- a/sql/sql_prepare.cc
+++ b/sql/sql_prepare.cc
@@ -3294,7 +3294,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)))
   {
@@ -3532,7 +3532,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 2d38af625d1..dcfcba0a01e 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -2979,8 +2979,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);
@@ -8314,8 +8318,7 @@ TABLE *create_schema_table(THD *thd, TABLE_LIST *table_list)
     item->maybe_null= (fields_info->field_flags & MY_I_S_MAYBE_NULL);
     field_count++;
   }
-  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= cs;
   tmp_table_param->field_count= field_count;
@@ -8894,6 +8897,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 bb6caef0ebb..db0992ec908 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);
 }
 
@@ -5271,9 +5271,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();
@@ -5290,11 +5294,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;
     }
@@ -6883,7 +6889,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_update.cc b/sql/sql_update.cc
index 258aa470b7b..f5bb298fdba 100644
--- a/sql/sql_update.cc
+++ b/sql/sql_update.cc
@@ -1263,7 +1263,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
       }
@@ -2992,7 +2992,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 b130fbc099b..cfaa5141a3b 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;
   }
 
@@ -1501,6 +1501,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
@@ -1551,6 +1552,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 249ed584aee..3fa2c85f160 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -7795,6 +7795,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
           {
@@ -13492,6 +13493,7 @@ insert:
             Lex->pop_select(); //main select
             if (Lex->check_main_unit_semantics())
               MYSQL_YYABORT;
+            Lex->mark_first_table_as_inserting();
           }
         ;
 
@@ -13516,6 +13518,7 @@ replace:
             Lex->pop_select(); //main select
             if (Lex->check_main_unit_semantics())
               MYSQL_YYABORT;
+            Lex->mark_first_table_as_inserting();
           }
         ;
 
@@ -15004,6 +15007,7 @@ load:
             Lex->pop_select(); //main select
             if (Lex->check_main_unit_semantics())
               MYSQL_YYABORT;
+            Lex->mark_first_table_as_inserting();
           }
           ;
 
diff --git a/sql/sql_yacc_ora.yy b/sql/sql_yacc_ora.yy
index 9370faf7ef5..aebbb145f96 100644
--- a/sql/sql_yacc_ora.yy
+++ b/sql/sql_yacc_ora.yy
@@ -7896,6 +7896,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
           {
@@ -13618,6 +13619,7 @@ insert:
             Lex->pop_select(); //main select
             if (Lex->check_main_unit_semantics())
               MYSQL_YYABORT;
+            Lex->mark_first_table_as_inserting();
           }
         ;
 
@@ -13642,6 +13644,7 @@ replace:
             Lex->pop_select(); //main select
             if (Lex->check_main_unit_semantics())
               MYSQL_YYABORT;
+            Lex->mark_first_table_as_inserting();
           }
         ;
 
@@ -15136,6 +15139,7 @@ load:
             Lex->pop_select(); //main select
             if (Lex->check_main_unit_semantics())
               MYSQL_YYABORT;
+            Lex->mark_first_table_as_inserting();
           }
           ;
 
diff --git a/sql/table.cc b/sql/table.cc
index 663a65d321a..9b027a8b31b 100644
--- a/sql/table.cc
+++ b/sql/table.cc
@@ -694,7 +694,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;
@@ -703,6 +707,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;
@@ -3179,6 +3186,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 1c721624d5d..8297774c4d2 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -750,10 +750,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;
@@ -2060,7 +2065,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);
@@ -2075,6 +2081,7 @@ struct TABLE_LIST
     **last_ptr= this;
     prev_global= *last_ptr;
     *last_ptr= &next_global;
+    for_insert_data= insert_data;
   }
 
 
@@ -2501,6 +2508,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_dummy.cc b/sql/wsrep_dummy.cc
index 75ee9b04cdf..b7752b0fd5e 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_mysqld.cc b/sql/wsrep_mysqld.cc
index bc0c1ab20e3..1ac5b39ff29 100644
--- a/sql/wsrep_mysqld.cc
+++ b/sql/wsrep_mysqld.cc
@@ -1703,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)
 {
@@ -2659,7 +2639,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_* */