MDEV-6156: Parallel replication incorrectly caches charset between worker threads

Replication caches the character sets used in a query, to be able to quickly
reuse them for the next query in the common case of them not having changed.

In parallel replication, this caching needs to be per-worker-thread. The
code was not modified to handle this correctly, so the caching in one worker
could cause another worker to run a query using the wrong character set,
causing replication corruption.

5 files changed, 43 insertions, 47 deletions

diff --git a/sql/log_event.cc b/sql/log_event.cc
index 6d2854a9f11..0a24d45113b 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -4153,7 +4153,7 @@ int Query_log_event::do_apply_event(rpl_group_info *rgi,
                    (sql_mode & ~(ulong) MODE_NO_DIR_IN_CREATE));
       if (charset_inited)
       {
-        if (rli->cached_charset_compare(charset))
+        if (rgi->cached_charset_compare(charset))
         {
           /* Verify that we support the charsets found in the event. */
           if (!(thd->variables.character_set_client=
@@ -4169,7 +4169,7 @@ int Query_log_event::do_apply_event(rpl_group_info *rgi,
               stop with EE_UNKNOWN_CHARSET in compare_errors (unless set to
               ignore this error).
             */
-            set_slave_thread_default_charset(thd, rli);
+            set_slave_thread_default_charset(thd, rgi);
             goto compare_errors;
           }
           thd->update_charset(); // for the charset change to take effect
@@ -6211,7 +6211,7 @@ int Rotate_log_event::do_update_pos(rpl_group_info *rgi)
       master is 4.0 then the events are in the slave's format (conversion).
     */
     set_slave_thread_options(thd);
-    set_slave_thread_default_charset(thd, rli);
+    set_slave_thread_default_charset(thd, rgi);
     thd->variables.sql_mode= global_system_variables.sql_mode;
     thd->variables.auto_increment_increment=
       thd->variables.auto_increment_offset= 1;
diff --git a/sql/rpl_rli.cc b/sql/rpl_rli.cc
index 0d0c8c9df70..8104d0ff553 100644
--- a/sql/rpl_rli.cc
+++ b/sql/rpl_rli.cc
@@ -82,7 +82,6 @@ Relay_log_info::Relay_log_info(bool is_slave_recovery)
   max_relay_log_size= global_system_variables.max_relay_log_size;
   bzero((char*) &info_file, sizeof(info_file));
   bzero((char*) &cache_buf, sizeof(cache_buf));
-  cached_charset_invalidate();
   mysql_mutex_init(key_relay_log_info_run_lock, &run_lock, MY_MUTEX_INIT_FAST);
   mysql_mutex_init(key_relay_log_info_data_lock,
                    &data_lock, MY_MUTEX_INIT_FAST);
@@ -1200,29 +1199,6 @@ bool Relay_log_info::is_until_satisfied(THD *thd, Log_event *ev)
 }
 
 
-void Relay_log_info::cached_charset_invalidate()
-{
-  DBUG_ENTER("Relay_log_info::cached_charset_invalidate");
-
-  /* Full of zeroes means uninitialized. */
-  bzero(cached_charset, sizeof(cached_charset));
-  DBUG_VOID_RETURN;
-}
-
-
-bool Relay_log_info::cached_charset_compare(char *charset) const
-{
-  DBUG_ENTER("Relay_log_info::cached_charset_compare");
-
-  if (memcmp(cached_charset, charset, sizeof(cached_charset)))
-  {
-    memcpy(const_cast<char*>(cached_charset), charset, sizeof(cached_charset));
-    DBUG_RETURN(1);
-  }
-  DBUG_RETURN(0);
-}
-
-
 void Relay_log_info::stmt_done(my_off_t event_master_log_pos,
                                time_t event_creation_time, THD *thd,
                                rpl_group_info *rgi)
@@ -1503,6 +1479,7 @@ rpl_group_info::rpl_group_info(Relay_log_info *rli)
     deferred_events(NULL), m_annotate_event(0), is_parallel_exec(false)
 {
   reinit(rli);
+  cached_charset_invalidate();
   bzero(&current_gtid, sizeof(current_gtid));
   mysql_mutex_init(key_rpl_group_info_sleep_lock, &sleep_lock,
                    MY_MUTEX_INIT_FAST);
@@ -1585,6 +1562,29 @@ delete_or_keep_event_post_apply(rpl_group_info *rgi,
 }
 
 
+void rpl_group_info::cached_charset_invalidate()
+{
+  DBUG_ENTER("rpl_group_info::cached_charset_invalidate");
+
+  /* Full of zeroes means uninitialized. */
+  bzero(cached_charset, sizeof(cached_charset));
+  DBUG_VOID_RETURN;
+}
+
+
+bool rpl_group_info::cached_charset_compare(char *charset) const
+{
+  DBUG_ENTER("rpl_group_info::cached_charset_compare");
+
+  if (memcmp(cached_charset, charset, sizeof(cached_charset)))
+  {
+    memcpy(const_cast<char*>(cached_charset), charset, sizeof(cached_charset));
+    DBUG_RETURN(1);
+  }
+  DBUG_RETURN(0);
+}
+
+
 void rpl_group_info::cleanup_context(THD *thd, bool error)
 {
   DBUG_ENTER("Relay_log_info::cleanup_context");
diff --git a/sql/rpl_rli.h b/sql/rpl_rli.h
index 48193afce4d..bcb237b6ced 100644
--- a/sql/rpl_rli.h
+++ b/sql/rpl_rli.h
@@ -295,7 +295,6 @@ public:
   /* Condition for UNTIL master_gtid_pos. */
   slave_connection_state until_gtid_pos;
 
-  char cached_charset[6];
   /*
     retried_trans is a cumulative counter: how many times the slave
     has retried a transaction (any) since slave started.
@@ -371,15 +370,6 @@ public:
 	    group_relay_log_pos);
   }
 
-  /*
-    Last charset (6 bytes) seen by slave SQL thread is cached here; it helps
-    the thread save 3 get_charset() per Query_log_event if the charset is not
-    changing from event to event (common situation).
-    When the 6 bytes are equal to 0 is used to mean "cache is invalidated".
-  */
-  void cached_charset_invalidate();
-  bool cached_charset_compare(char *charset) const;
-
   /**
     Helper function to do after statement completion.
 
@@ -546,6 +536,8 @@ struct rpl_group_info
   mysql_mutex_t sleep_lock;
   mysql_cond_t sleep_cond;
 
+  char cached_charset[6];
+
   /*
     trans_retries varies between 0 to slave_transaction_retries and counts how
     many times the slave has retried the present transaction; gets reset to 0
@@ -679,6 +671,15 @@ struct rpl_group_info
     return false;
   }
 
+  /*
+    Last charset (6 bytes) seen by slave SQL thread is cached here; it helps
+    the thread save 3 get_charset() per Query_log_event if the charset is not
+    changing from event to event (common situation).
+    When the 6 bytes are equal to 0 is used to mean "cache is invalidated".
+  */
+  void cached_charset_invalidate();
+  bool cached_charset_compare(char *charset) const;
+
   void clear_tables_to_lock();
   void cleanup_context(THD *, bool);
   void slave_close_thread_tables(THD *);
diff --git a/sql/slave.cc b/sql/slave.cc
index 43c97d0af9c..b5d8758a405 100644
--- a/sql/slave.cc
+++ b/sql/slave.cc
@@ -2879,7 +2879,7 @@ void set_slave_thread_options(THD* thd)
   DBUG_VOID_RETURN;
 }
 
-void set_slave_thread_default_charset(THD* thd, Relay_log_info const *rli)
+void set_slave_thread_default_charset(THD* thd, rpl_group_info *rgi)
 {
   DBUG_ENTER("set_slave_thread_default_charset");
 
@@ -2891,13 +2891,7 @@ void set_slave_thread_default_charset(THD* thd, Relay_log_info const *rli)
     global_system_variables.collation_server;
   thd->update_charset();
 
-  /*
-    We use a const cast here since the conceptual (and externally
-    visible) behavior of the function is to set the default charset of
-    the thread.  That the cache has to be invalidated is a secondary
-    effect.
-   */
-  const_cast<Relay_log_info*>(rli)->cached_charset_invalidate();
+  rgi->cached_charset_invalidate();
   DBUG_VOID_RETURN;
 }
 
@@ -4682,7 +4676,7 @@ err_during_init:
   mysql_cond_broadcast(&rli->data_cond);
   rli->ignore_log_space_limit= 0; /* don't need any lock */
   /* we die so won't remember charset - re-update them on next thread start */
-  rli->cached_charset_invalidate();
+  serial_rgi->cached_charset_invalidate();
 
   /*
     TODO: see if we can do this conditionally in next_event() instead
diff --git a/sql/slave.h b/sql/slave.h
index 3981a9d4f2c..aa3976f6e6c 100644
--- a/sql/slave.h
+++ b/sql/slave.h
@@ -51,6 +51,7 @@
 class Relay_log_info;
 class Master_info;
 class Master_info_index;
+struct rpl_group_info;
 struct rpl_parallel_thread;
 
 int init_intvar_from_file(int* var, IO_CACHE* f, int default_val);
@@ -226,7 +227,7 @@ int init_relay_log_pos(Relay_log_info* rli,const char* log,ulonglong pos,
 int purge_relay_logs(Relay_log_info* rli, THD *thd, bool just_reset,
 		     const char** errmsg);
 void set_slave_thread_options(THD* thd);
-void set_slave_thread_default_charset(THD *thd, Relay_log_info const *rli);
+void set_slave_thread_default_charset(THD *thd, rpl_group_info *rgi);
 int rotate_relay_log(Master_info* mi);
 int apply_event_and_update_pos(Log_event* ev, THD* thd,
                                struct rpl_group_info *rgi,