BUG#23171 (Illegal slave restart group position):

Second patch to fix skipping code. Moving relay and binary log 
position changing code from do_apply_event [old exec_event()] into
do_update_pos() and doing other changes necessary to support that.

Fixing a bug that can cause deadlock if rotating binary log when committing
a changes to a transactional table that is not inside a transaction and
cause a rotate log.


sql/log.cc:
  Changing condition in binlog_commit() to skip calling
  binlog_end_trans() twice to match condition in binlog_end_trans().
sql/log_event.cc:
  Name change: apply_event_impl() -> do_apply_event()
  Name change: advance_coord_impl() -> do_update_pos()
  
  do_apply_event() now uses pointer to constant RELAY_LOG_INFO to prevent
  inadvertandly changing the position in the code. Doing this would make
  the skipping code loose track of where it is. All position changing code
  shall now be in do_update_pos().
  
  Factoring out relay and binary log position updating code from
  do_apply_event() [previously exec_event()] into do_update_pos().
  Using a safe approach to make it work: will refine the refactoring
  when the skipping code is implemented.
  
  Adding const casts where needed.
  
  Changing signature to use pointers to constant objects where needed.
sql/rpl_rli.cc:
  Making cached_charset_compare() const to work with constant instances of RELAY_LOG_INFO.
  
  Debriding code.
sql/rpl_rli.h:
  Making cached_charset_compare() const to work with constant instances of RELAY_LOG_INFO.
  
  Debriding code.
sql/rpl_utility.cc:
  Using pointer to const RELAY_LOG_INFO to make it work with other code.
sql/rpl_utility.h:
  Using pointer to const RELAY_LOG_INFO to make it work with other code.
sql/slave.cc:
  Using pointer to const RELAY_LOG_INFO to make it work with other code.
  
  Adding const cast where necessary.
sql/slave.h:
  Using pointer to const RELAY_LOG_INFO to make it work with other code.

sql/log.cc

@@ -1453,7 +1453,7 @@ static int binlog_prepare(handlerton *hton, THD *thd, bool all)
     do nothing.
     just pretend we can do 2pc, so that MySQL won't
     switch to 1pc.
-    real work will be done in MYSQL_BIN_LOG::log()
+    real work will be done in TC_LOG_BINLOG::log()
   */
   return 0;
 }
@@ -1467,9 +1467,15 @@ static int binlog_commit(handlerton *hton, THD *thd, bool all)
   IO_CACHE *trans_log= &trx_data->trans_log;
   DBUG_ASSERT(mysql_bin_log.is_open());
 
-  if (all && trx_data->empty())
+  /*
+    The condition here has to be identical to the one inside
+    binlog_end_trans(), guarding the write of the transaction cache to
+    the binary log.
+   */
+  if ((all || !(thd->options & (OPTION_BEGIN | OPTION_NOT_AUTOCOMMIT))) &&
+      trx_data->empty())
   {
-    // we're here because trans_log was flushed in MYSQL_BIN_LOG::log()
+    // we're here because trans_log was flushed in TC_LOG_BINLOG::log()
     trx_data->reset();
     DBUG_RETURN(0);
   }
@@ -3117,8 +3123,10 @@ void MYSQL_BIN_LOG::new_file_impl(bool need_lock)
   {
     tc_log_page_waits++;
     pthread_mutex_lock(&LOCK_prep_xids);
-    while (prepared_xids)
+    while (prepared_xids) {
+      DBUG_PRINT("info", ("prepared_xids=%lu", prepared_xids));
       pthread_cond_wait(&COND_prep_xids, &LOCK_prep_xids);
+    }
     pthread_mutex_unlock(&LOCK_prep_xids);
   }
 
@@ -4905,8 +4913,10 @@ void TC_LOG_BINLOG::unlog(ulong cookie, my_xid xid)
 {
   pthread_mutex_lock(&LOCK_prep_xids);
   DBUG_ASSERT(prepared_xids > 0);
-  if (--prepared_xids == 0)
+  if (--prepared_xids == 0) {
+    DBUG_PRINT("info", ("prepared_xids=%lu", prepared_xids));
     pthread_cond_signal(&COND_prep_xids);
+  }
   pthread_mutex_unlock(&LOCK_prep_xids);
   rotate_and_purge(0);     // as ::write() did not rotate
 }

sql/rpl_rli.cc

@@ -1056,30 +1056,19 @@ void st_relay_log_info::cached_charset_invalidate()
 }
 
 
-bool st_relay_log_info::cached_charset_compare(char *charset)
+bool st_relay_log_info::cached_charset_compare(char *charset) const
 {
   DBUG_ENTER("st_relay_log_info::cached_charset_compare");
 
   if (bcmp(cached_charset, charset, sizeof(cached_charset)))
   {
-    memcpy(cached_charset, charset, sizeof(cached_charset));
+    memcpy(const_cast<char*>(cached_charset), charset, sizeof(cached_charset));
     DBUG_RETURN(1);
   }
   DBUG_RETURN(0);
 }
 
 
-void st_relay_log_info::transaction_end(THD* thd)
-{
-  DBUG_ENTER("st_relay_log_info::transaction_end");
-
-  /*
-    Nothing to do here right now.
-   */
-
-  DBUG_VOID_RETURN;
-}
-
 #if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
 void st_relay_log_info::cleanup_context(THD *thd, bool error)
 {

sql/rpl_rli.h

@@ -291,9 +291,7 @@ typedef struct st_relay_log_info
     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);
-
-  void transaction_end(THD*);
+  bool cached_charset_compare(char *charset) const;
 
   void cleanup_context(THD *, bool);
   void clear_tables_to_lock() {

sql/rpl_utility.cc

@@ -109,7 +109,7 @@ field_length_from_packed(enum_field_types const field_type,
 
 */
 int
-table_def::compatible_with(RELAY_LOG_INFO *rli, TABLE *table)
+table_def::compatible_with(RELAY_LOG_INFO const *rli_arg, TABLE *table)
   const
 {
   /*
@@ -117,6 +117,7 @@ table_def::compatible_with(RELAY_LOG_INFO *rli, TABLE *table)
   */
   uint const cols_to_check= min(table->s->fields, size());
   int error= 0;
+  RELAY_LOG_INFO const *rli= const_cast<RELAY_LOG_INFO*>(rli_arg);
 
   TABLE_SHARE const *const tsh= table->s;
 

sql/rpl_utility.h

@@ -116,7 +116,7 @@ class table_def
       1  if the table definition is not compatible with 'table'
       0  if the table definition is compatible with 'table'
   */
-  int compatible_with(RELAY_LOG_INFO *rli, TABLE *table) const;
+  int compatible_with(RELAY_LOG_INFO const *rli, TABLE *table) const;
 
 private:
   my_size_t m_size;           // Number of elements in the types array

sql/slave.cc

@@ -555,7 +555,7 @@ static bool sql_slave_killed(THD* thd, RELAY_LOG_INFO* rli)
     void
 */
 
-void slave_print_msg(enum loglevel level, RELAY_LOG_INFO* rli,
+void slave_print_msg(enum loglevel level, RELAY_LOG_INFO const *rli,
                      int err_code, const char* msg, ...)
 {
   void (*report_function)(const char *, ...);
@@ -577,9 +577,9 @@ void slave_print_msg(enum loglevel level, RELAY_LOG_INFO* rli,
       It's an error, it must be reported in Last_error and Last_errno in SHOW
       SLAVE STATUS.
     */
-    pbuff= rli->last_slave_error;
+    pbuff= const_cast<RELAY_LOG_INFO*>(rli)->last_slave_error;
     pbuffsize= sizeof(rli->last_slave_error);
-    rli->last_slave_errno = err_code;
+    const_cast<RELAY_LOG_INFO*>(rli)->last_slave_errno = err_code;
     report_function= sql_print_error;
     break;
   case WARNING_LEVEL:
@@ -1376,7 +1376,7 @@ void set_slave_thread_options(THD* thd)
   DBUG_VOID_RETURN;
 }
 
-void set_slave_thread_default_charset(THD* thd, RELAY_LOG_INFO *rli)
+void set_slave_thread_default_charset(THD* thd, RELAY_LOG_INFO const *rli)
 {
   DBUG_ENTER("set_slave_thread_default_charset");
 
@@ -1387,7 +1387,14 @@ void set_slave_thread_default_charset(THD* thd, RELAY_LOG_INFO *rli)
   thd->variables.collation_server=
     global_system_variables.collation_server;
   thd->update_charset();
-  rli->cached_charset_invalidate();
+
+  /*
+    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();
   DBUG_VOID_RETURN;
 }
 
@@ -1607,7 +1614,8 @@ static ulong read_event(MYSQL* mysql, MASTER_INFO *mi, bool* suppress_warnings)
 }
 
 
-int check_expected_error(THD* thd, RELAY_LOG_INFO* rli, int expected_error)
+int check_expected_error(THD* thd, RELAY_LOG_INFO const *rli,
+                         int expected_error)
 {
   DBUG_ENTER("check_expected_error");
 

sql/slave.h

@@ -164,9 +164,9 @@ bool show_master_info(THD* thd, MASTER_INFO* mi);
 bool show_binlog_info(THD* thd);
 
 const char *print_slave_db_safe(const char *db);
-int check_expected_error(THD* thd, RELAY_LOG_INFO* rli, int error_code);
+int check_expected_error(THD* thd, RELAY_LOG_INFO const *rli, int error_code);
 void skip_load_data_infile(NET* net);
-void slave_print_msg(enum loglevel level, RELAY_LOG_INFO* rli,
+void slave_print_msg(enum loglevel level, RELAY_LOG_INFO const *rli,
                      int err_code, const char* msg, ...)
   ATTRIBUTE_FORMAT(printf, 4, 5);
 
@@ -184,7 +184,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 *rli);
+void set_slave_thread_default_charset(THD *thd, RELAY_LOG_INFO const *rli);
 void rotate_relay_log(MASTER_INFO* mi);
 
 pthread_handler_t handle_slave_io(void *arg);