Merge romeo.(none):/home/bkroot/mysql-5.1-new-rpl

into  romeo.(none):/home/bk/b23171-mysql-5.1-new-rpl


mysql-test/r/rpl_row_tabledefs_2myisam.result:
  Auto merged
mysql-test/r/rpl_row_tabledefs_3innodb.result:
  Auto merged
sql/rpl_rli.cc:
  Auto merged
sql/rpl_rli.h:
  Auto merged
sql/rpl_utility.cc:
  Auto merged
sql/rpl_utility.h:
  Auto merged
sql/slave.h:
  Auto merged
sql/log.cc:
  Manual merge
sql/log_event.cc:
  Manual merge
sql/log_event.h:
  Manual merge
sql/slave.cc:
  Manual merge
sql/sql_binlog.cc:
  Manual merge

mysql-test/r/rpl_row_tabledefs_2myisam.result

@@ -122,7 +122,7 @@ Replicate_Do_Table
 Replicate_Ignore_Table	
 Replicate_Wild_Do_Table	
 Replicate_Wild_Ignore_Table	
-Last_Errno	1364
+Last_Errno	1105
 Last_Error	Error in Write_rows event: error during transaction execution on table test.t1_nodef
 Skip_Counter	0
 Exec_Master_Log_Pos	#

mysql-test/r/rpl_row_tabledefs_3innodb.result

@@ -122,7 +122,7 @@ Replicate_Do_Table
 Replicate_Ignore_Table	
 Replicate_Wild_Do_Table	
 Replicate_Wild_Ignore_Table	
-Last_Errno	1364
+Last_Errno	1105
 Last_Error	Error in Write_rows event: error during transaction execution on table test.t1_nodef
 Skip_Counter	0
 Exec_Master_Log_Pos	#

sql/log.cc

@@ -1548,7 +1548,13 @@ static int binlog_commit(handlerton *hton, THD *thd, bool all)
     (binlog_trx_data*) thd->ha_data[binlog_hton->slot];
   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_xid()
     trx_data->reset();
@@ -2499,7 +2505,7 @@ bool MYSQL_BIN_LOG::open(const char *log_name,
       /*
         Set 'created' to 0, so that in next relay logs this event does not
         trigger cleaning actions on the slave in
-        Format_description_log_event::exec_event().
+        Format_description_log_event::apply_event_impl().
       */
       description_event_for_queue->created= 0;
       /* Don't set log_pos in event header */
@@ -3206,8 +3212,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);
   }
 
@@ -5061,8 +5069,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/log_event.h

@@ -503,6 +503,7 @@ class THD;
 class Format_description_log_event;
 
 struct st_relay_log_info;
+typedef st_relay_log_info RELAY_LOG_INFO;
 
 #ifdef MYSQL_CLIENT
 /*
@@ -591,6 +592,33 @@ typedef struct st_print_event_info
 class Log_event
 {
 public:
+  /**
+     Enumeration of what kinds of skipping (and non-skipping) that can
+     occur when the slave executes an event.
+
+     @see shall_skip
+     @see do_shall_skip
+   */
+  enum enum_skip_reason {
+    /**
+       Don't skip event.
+    */
+    EVENT_SKIP_NOT,
+
+    /**
+       Skip event by ignoring it.
+
+       This means that the slave skip counter will not be changed.
+    */
+    EVENT_SKIP_IGNORE,
+
+    /**
+       Skip event and decrease skip counter.
+    */
+    EVENT_SKIP_COUNT
+  };
+
+
   /*
     The following type definition is to be used whenever data is placed 
     and manipulated in a common buffer. Use this typedef for buffers
@@ -672,16 +700,14 @@ public:
   static void init_show_field_list(List<Item>* field_list);
 #ifdef HAVE_REPLICATION
   int net_send(Protocol *protocol, const char* log_name, my_off_t pos);
+
   /*
     pack_info() is used by SHOW BINLOG EVENTS; as print() it prepares and sends
     a string to display to the user, so it resembles print().
   */
+
   virtual void pack_info(Protocol *protocol);
-  /*
+
-    The SQL slave thread calls exec_event() to execute the event; this is where
-    the slave's data is modified.
-  */
-  virtual int exec_event(struct st_relay_log_info* rli);
 #endif /* HAVE_REPLICATION */
   virtual const char* get_db()
   {
@@ -754,6 +780,127 @@ public:
                                    *description_event);
   /* returns the human readable name of the event's type */
   const char* get_type_str();
+
+#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
+public:
+
+  /**
+     Apply the event to the database.
+
+     This function represents the public interface for applying an
+     event.
+
+     @see do_apply_event
+   */
+  int apply_event(RELAY_LOG_INFO const *rli) {
+    return do_apply_event(rli);
+  }
+
+
+  /**
+     Update the relay log position.
+
+     This function represents the public interface for "stepping over"
+     the event and will update the relay log information.
+
+     @see do_update_pos
+   */
+  int update_pos(RELAY_LOG_INFO *rli)
+  {
+    return do_update_pos(rli);
+  }
+
+  /**
+     Decide if the event shall be skipped, and the reason for skipping
+     it.
+
+     @see do_shall_skip
+   */
+  enum_skip_reason shall_skip(RELAY_LOG_INFO *rli)
+  {
+    return do_shall_skip(rli);
+  }
+
+protected:
+  /**
+    Primitive to apply an event to the database.
+
+    This is where the change to the database is made.
+
+    @note The primitive is protected instead of private, since there
+    is a hierarchy of actions to be performed in some cases.
+
+    @see Format_description_log_event::do_apply_event()
+
+    @param rli Pointer to relay log info structure
+
+    @retval 0     Event applied successfully
+    @retval errno Error code if event application failed
+  */
+  virtual int do_apply_event(RELAY_LOG_INFO const *rli)
+  {
+    return 0;                /* Default implementation does nothing */
+  }
+
+
+  /**
+     Advance relay log coordinates.
+
+     This function is called to advance the relay log coordinates to
+     just after the event.  It is essential that both the relay log
+     coordinate and the group log position is updated correctly, since
+     this function is used also for skipping events.
+
+     Normally, each implementation of do_update_pos() shall:
+
+     - Update the event position to refer to the position just after
+       the event.
+
+     - Update the group log position to refer to the position just
+       after the event <em>if the event is last in a group</em>
+
+     @param rli Pointer to relay log info structure
+
+     @retval 0     Coordinates changed successfully
+     @retval errno Error code if advancing failed (usually just
+                   1). Observe that handler errors are returned by the
+                   do_apply_event() function, and not by this one.
+   */
+  virtual int do_update_pos(RELAY_LOG_INFO *rli);
+
+
+  /**
+     Decide if this event shall be skipped or not and the reason for
+     skipping it.
+
+     The default implementation decide that the event shall be skipped
+     if either:
+
+     - the server id of the event is the same as the server id of the
+       server and <code>rli->replicate_same_server_id</code> is true,
+       or
+
+     - if <code>rli->slave_skip_counter</code> is greater than zero.
+
+     @see do_apply_event
+     @see do_update_pos
+
+     @retval Log_event::EVENT_SKIP_NOT
+     The event shall not be skipped and should be applied.
+
+     @retval Log_event::EVENT_SKIP_IGNORE
+     The event shall be skipped by just ignoring it, i.e., the slave
+     skip counter shall not be changed. This happends if, for example,
+     the originating server id of the event is the same as the server
+     id of the slave.
+
+     @retval Log_event::EVENT_SKIP_COUNT
+     The event shall be skipped because the slave skip counter was
+     non-zero. The caller shall decrease the counter by one.
+   */
+  virtual enum_skip_reason do_shall_skip(RELAY_LOG_INFO *rli);
+
+#endif
 };
 
 /*
@@ -794,10 +941,10 @@ public:
   uint16 error_code;
   ulong thread_id;
   /*
-    For events created by Query_log_event::exec_event (and
+    For events created by Query_log_event::do_apply_event (and
-    Load_log_event::exec_event()) we need the *original* thread id, to be able
+    Load_log_event::do_apply_event()) we need the *original* thread
-    to log the event with the original (=master's) thread id (fix for
+    id, to be able to log the event with the original (=master's)
-    BUG#1686).
+    thread id (fix for BUG#1686).
   */
   ulong slave_proxy_id;
 
@@ -860,9 +1007,6 @@ public:
   const char* get_db() { return db; }
 #ifdef HAVE_REPLICATION
   void pack_info(Protocol* protocol);
-  int exec_event(struct st_relay_log_info* rli);
-  int exec_event(struct st_relay_log_info* rli, const char *query_arg,
-                 uint32 q_len_arg);
 #endif /* HAVE_REPLICATION */
 #else
   void print_query_header(IO_CACHE* file, PRINT_EVENT_INFO* print_event_info);
@@ -891,6 +1035,16 @@ public:
   */
   virtual ulong get_post_header_size_for_derived() { return 0; }
   /* Writes derived event-specific part of post header. */
+
+public:        /* !!! Public in this patch to allow old usage */
+#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
+  virtual int do_apply_event(RELAY_LOG_INFO const *rli);
+  virtual int do_update_pos(RELAY_LOG_INFO *rli);
+
+  int do_apply_event(RELAY_LOG_INFO const *rli,
+                       const char *query_arg,
+                       uint32 q_len_arg);
+#endif /* HAVE_REPLICATION */
 };
 
 
@@ -939,9 +1093,8 @@ public:
   uint16 master_port;
 
 #ifndef MYSQL_CLIENT
-  Slave_log_event(THD* thd_arg, struct st_relay_log_info* rli);
+  Slave_log_event(THD* thd_arg, RELAY_LOG_INFO* rli);
   void pack_info(Protocol* protocol);
-  int exec_event(struct st_relay_log_info* rli);
 #else
   void print(FILE* file, PRINT_EVENT_INFO* print_event_info);
 #endif
@@ -954,6 +1107,11 @@ public:
 #ifndef MYSQL_CLIENT
   bool write(IO_CACHE* file);
 #endif
+
+private:
+#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
+  virtual int do_apply_event(RELAY_LOG_INFO const* rli);
+#endif
 };
 
 #endif /* HAVE_REPLICATION */
@@ -1023,12 +1181,6 @@ public:
   const char* get_db() { return db; }
 #ifdef HAVE_REPLICATION
   void pack_info(Protocol* protocol);
-  int exec_event(struct st_relay_log_info* rli)
-  {
-    return exec_event(thd->slave_net,rli,0);
-  }
-  int exec_event(NET* net, struct st_relay_log_info* rli,
-		 bool use_rli_only_for_errors);
 #endif /* HAVE_REPLICATION */
 #else
   void print(FILE* file, PRINT_EVENT_INFO* print_event_info);
@@ -1060,6 +1212,17 @@ public:
 	    + LOAD_HEADER_LEN
 	    + sql_ex.data_size() + field_block_len + num_fields);
   }
+
+public:        /* !!! Public in this patch to allow old usage */
+#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
+  virtual int do_apply_event(RELAY_LOG_INFO const* rli)
+  {
+    return do_apply_event(thd->slave_net,rli,0);
+  }
+
+  int do_apply_event(NET *net, RELAY_LOG_INFO const *rli,
+                     bool use_rli_only_for_errors);
+#endif
 };
 
 extern char server_version[SERVER_VERSION_LENGTH];
@@ -1117,7 +1280,6 @@ public:
   Start_log_event_v3();
 #ifdef HAVE_REPLICATION
   void pack_info(Protocol* protocol);
-  int exec_event(struct st_relay_log_info* rli);
 #endif /* HAVE_REPLICATION */
 #else
   Start_log_event_v3() {}
@@ -1137,6 +1299,22 @@ public:
     return START_V3_HEADER_LEN; //no variable-sized part
   }
   virtual bool is_artificial_event() { return artificial_event; }
+
+protected:
+#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
+  virtual int do_apply_event(RELAY_LOG_INFO const *rli);
+  virtual enum_skip_reason do_shall_skip(RELAY_LOG_INFO*)
+  {
+    /*
+      Events from ourself should be skipped, but they should not
+      decrease the slave skip counter.
+     */
+    if (this->server_id == ::server_id)
+      return Log_event::EVENT_SKIP_IGNORE;
+    else
+      return Log_event::EVENT_SKIP_NOT;
+  }
+#endif
 };
 
 
@@ -1162,13 +1340,6 @@ public:
   uchar server_version_split[3];
 
   Format_description_log_event(uint8 binlog_ver, const char* server_ver=0);
-
-#ifndef MYSQL_CLIENT
-#ifdef HAVE_REPLICATION
-  int exec_event(struct st_relay_log_info* rli);
-#endif /* HAVE_REPLICATION */
-#endif
-
   Format_description_log_event(const char* buf, uint event_len,
                                const Format_description_log_event* description_event);
   ~Format_description_log_event() { my_free((gptr)post_header_len, MYF(0)); }
@@ -1191,7 +1362,15 @@ public:
     */
     return FORMAT_DESCRIPTION_HEADER_LEN;
   }
+
   void calc_server_version_split();
+
+protected:
+#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
+  virtual int do_apply_event(RELAY_LOG_INFO const *rli);
+  virtual int do_update_pos(RELAY_LOG_INFO *rli);
+  virtual enum_skip_reason do_shall_skip(RELAY_LOG_INFO *rli);
+#endif
 };
 
 
@@ -1215,7 +1394,6 @@ public:
   {}
 #ifdef HAVE_REPLICATION
   void pack_info(Protocol* protocol);
-  int exec_event(struct st_relay_log_info* rli);
 #endif /* HAVE_REPLICATION */
 #else
   void print(FILE* file, PRINT_EVENT_INFO* print_event_info);
@@ -1230,6 +1408,13 @@ public:
   bool write(IO_CACHE* file);
 #endif
   bool is_valid() const { return 1; }
+
+private:
+#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
+  virtual int do_apply_event(RELAY_LOG_INFO const *rli);
+  virtual int do_update_pos(RELAY_LOG_INFO *rli);
+  virtual enum_skip_reason do_shall_skip(RELAY_LOG_INFO *rli);
+#endif
 };
 
 
@@ -1256,7 +1441,6 @@ class Rand_log_event: public Log_event
   {}
 #ifdef HAVE_REPLICATION
   void pack_info(Protocol* protocol);
-  int exec_event(struct st_relay_log_info* rli);
 #endif /* HAVE_REPLICATION */
 #else
   void print(FILE* file, PRINT_EVENT_INFO* print_event_info);
@@ -1270,6 +1454,13 @@ class Rand_log_event: public Log_event
   bool write(IO_CACHE* file);
 #endif
   bool is_valid() const { return 1; }
+
+private:
+#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
+  virtual int do_apply_event(RELAY_LOG_INFO const *rli);
+  virtual int do_update_pos(RELAY_LOG_INFO *rli);
+  virtual enum_skip_reason do_shall_skip(RELAY_LOG_INFO *rli);
+#endif
 };
 
 /*****************************************************************************
@@ -1293,7 +1484,6 @@ class Xid_log_event: public Log_event
   Xid_log_event(THD* thd_arg, my_xid x): Log_event(thd_arg,0,0), xid(x) {}
 #ifdef HAVE_REPLICATION
   void pack_info(Protocol* protocol);
-  int exec_event(struct st_relay_log_info* rli);
 #endif /* HAVE_REPLICATION */
 #else
   void print(FILE* file, PRINT_EVENT_INFO* print_event_info);
@@ -1307,6 +1497,11 @@ class Xid_log_event: public Log_event
   bool write(IO_CACHE* file);
 #endif
   bool is_valid() const { return 1; }
+
+private:
+#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
+  virtual int do_apply_event(RELAY_LOG_INFO const *rli);
+#endif
 };
 
 /*****************************************************************************
@@ -1336,7 +1531,6 @@ public:
     val_len(val_len_arg), type(type_arg), charset_number(charset_number_arg)
     { is_null= !val; }
   void pack_info(Protocol* protocol);
-  int exec_event(struct st_relay_log_info* rli);
 #else
   void print(FILE* file, PRINT_EVENT_INFO* print_event_info);
 #endif
@@ -1348,6 +1542,13 @@ public:
   bool write(IO_CACHE* file);
 #endif
   bool is_valid() const { return 1; }
+
+private:
+#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
+  virtual int do_apply_event(RELAY_LOG_INFO const *rli);
+  virtual int do_update_pos(RELAY_LOG_INFO *rli);
+  virtual enum_skip_reason do_shall_skip(RELAY_LOG_INFO *rli);
+#endif
 };
 
 
@@ -1362,7 +1563,6 @@ public:
 #ifndef MYSQL_CLIENT
   Stop_log_event() :Log_event()
   {}
-  int exec_event(struct st_relay_log_info* rli);
 #else
   void print(FILE* file, PRINT_EVENT_INFO* print_event_info);
 #endif
@@ -1373,6 +1573,22 @@ public:
   ~Stop_log_event() {}
   Log_event_type get_type_code() { return STOP_EVENT;}
   bool is_valid() const { return 1; }
+
+private:
+#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
+  virtual int do_update_pos(RELAY_LOG_INFO *rli);
+  virtual enum_skip_reason do_shall_skip(RELAY_LOG_INFO *rli)
+  {
+    /*
+      Events from ourself should be skipped, but they should not
+      decrease the slave skip counter.
+     */
+    if (this->server_id == ::server_id)
+      return Log_event::EVENT_SKIP_IGNORE;
+    else
+      return Log_event::EVENT_SKIP_NOT;
+  }
+#endif
 };
 
 /*****************************************************************************
@@ -1399,7 +1615,6 @@ public:
 		   ulonglong pos_arg, uint flags);
 #ifdef HAVE_REPLICATION
   void pack_info(Protocol* protocol);
-  int exec_event(struct st_relay_log_info* rli);
 #endif /* HAVE_REPLICATION */
 #else
   void print(FILE* file, PRINT_EVENT_INFO* print_event_info);
@@ -1418,6 +1633,12 @@ public:
 #ifndef MYSQL_CLIENT
   bool write(IO_CACHE* file);
 #endif
+
+private:
+#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
+  virtual int do_update_pos(RELAY_LOG_INFO *rli);
+  virtual enum_skip_reason do_shall_skip(RELAY_LOG_INFO *rli);
+#endif
 };
 
 
@@ -1452,7 +1673,6 @@ public:
 			bool using_trans);
 #ifdef HAVE_REPLICATION
   void pack_info(Protocol* protocol);
-  int exec_event(struct st_relay_log_info* rli);
 #endif /* HAVE_REPLICATION */
 #else
   void print(FILE* file, PRINT_EVENT_INFO* print_event_info);
@@ -1486,6 +1706,11 @@ public:
   */
   bool write_base(IO_CACHE* file);
 #endif
+
+private:
+#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
+  virtual int do_apply_event(RELAY_LOG_INFO const *rli);
+#endif
 };
 
 
@@ -1518,7 +1743,6 @@ public:
   Append_block_log_event(THD* thd, const char* db_arg, char* block_arg,
 			 uint block_len_arg, bool using_trans);
 #ifdef HAVE_REPLICATION
-  int exec_event(struct st_relay_log_info* rli);
   void pack_info(Protocol* protocol);
   virtual int get_create_or_append() const;
 #endif /* HAVE_REPLICATION */
@@ -1536,6 +1760,11 @@ public:
   bool write(IO_CACHE* file);
   const char* get_db() { return db; }
 #endif
+
+private:
+#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
+  virtual int do_apply_event(RELAY_LOG_INFO const *rli);
+#endif
 };
 
 
@@ -1555,7 +1784,6 @@ public:
   Delete_file_log_event(THD* thd, const char* db_arg, bool using_trans);
 #ifdef HAVE_REPLICATION
   void pack_info(Protocol* protocol);
-  int exec_event(struct st_relay_log_info* rli);
 #endif /* HAVE_REPLICATION */
 #else
   void print(FILE* file, PRINT_EVENT_INFO* print_event_info);
@@ -1572,6 +1800,11 @@ public:
   bool write(IO_CACHE* file);
   const char* get_db() { return db; }
 #endif
+
+private:
+#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
+  virtual int do_apply_event(RELAY_LOG_INFO const *rli);
+#endif
 };
 
 
@@ -1591,7 +1824,6 @@ public:
   Execute_load_log_event(THD* thd, const char* db_arg, bool using_trans);
 #ifdef HAVE_REPLICATION
   void pack_info(Protocol* protocol);
-  int exec_event(struct st_relay_log_info* rli);
 #endif /* HAVE_REPLICATION */
 #else
   void print(FILE* file, PRINT_EVENT_INFO* print_event_info);
@@ -1607,6 +1839,11 @@ public:
   bool write(IO_CACHE* file);
   const char* get_db() { return db; }
 #endif
+
+private:
+#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
+  virtual int do_apply_event(RELAY_LOG_INFO const *rli);
+#endif
 };
 
 
@@ -1676,7 +1913,6 @@ public:
                        bool using_trans, bool suppress_use);
 #ifdef HAVE_REPLICATION
   void pack_info(Protocol* protocol);
-  int exec_event(struct st_relay_log_info* rli);
 #endif /* HAVE_REPLICATION */
 #else
   void print(FILE* file, PRINT_EVENT_INFO* print_event_info);
@@ -1695,7 +1931,12 @@ public:
 #ifndef MYSQL_CLIENT
   bool write_post_header_for_derived(IO_CACHE* file);
 #endif
- };
+
+private:
+#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
+  virtual int do_apply_event(RELAY_LOG_INFO const *rli);
+#endif
+};
 
 
 #ifdef MYSQL_CLIENT
@@ -1793,7 +2034,6 @@ public:
 #endif
 
 #if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
-  virtual int exec_event(struct st_relay_log_info *rli);
   virtual void pack_info(Protocol *protocol);
 #endif
 
@@ -1803,6 +2043,11 @@ public:
 
 
 private:
+#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
+  virtual int do_apply_event(RELAY_LOG_INFO const *rli);
+  virtual int do_update_pos(RELAY_LOG_INFO *rli);
+#endif
+
 #ifndef MYSQL_CLIENT
   TABLE      *m_table;
 #endif
@@ -1886,7 +2131,6 @@ public:
   flag_set get_flags(flag_set flags) const { return m_flags & flags; }
 
 #if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
-  virtual int exec_event(struct st_relay_log_info *rli);
   virtual void pack_info(Protocol *protocol);
 #endif
 
@@ -1970,6 +2214,8 @@ protected:
 private:
 
 #if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
+  virtual int do_apply_event(RELAY_LOG_INFO const *rli);
+
   /*
     Primitive to prepare for a sequence of row executions.
 
@@ -2017,7 +2263,7 @@ private:
     RETURN VALUE
       Error code, if something went wrong, 0 otherwise.
    */
-  virtual int do_prepare_row(THD*, RELAY_LOG_INFO*, TABLE*,
+  virtual int do_prepare_row(THD*, RELAY_LOG_INFO const*, TABLE*,
                              char const *row_start, char const **row_end) = 0;
 
   /*
@@ -2088,7 +2334,7 @@ private:
 
   virtual int do_before_row_operations(TABLE *table);
   virtual int do_after_row_operations(TABLE *table, int error);
-  virtual int do_prepare_row(THD*, RELAY_LOG_INFO*, TABLE*,
+  virtual int do_prepare_row(THD*, RELAY_LOG_INFO const*, TABLE*,
                              char const *row_start, char const **row_end);
   virtual int do_exec_row(TABLE *table);
 #endif
@@ -2153,7 +2399,7 @@ private:
 
   virtual int do_before_row_operations(TABLE *table);
   virtual int do_after_row_operations(TABLE *table, int error);
-  virtual int do_prepare_row(THD*, RELAY_LOG_INFO*, TABLE*,
+  virtual int do_prepare_row(THD*, RELAY_LOG_INFO const*, TABLE*,
                              char const *row_start, char const **row_end);
   virtual int do_exec_row(TABLE *table);
 #endif /* !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION) */
@@ -2224,7 +2470,7 @@ private:
 
   virtual int do_before_row_operations(TABLE *table);
   virtual int do_after_row_operations(TABLE *table, int error);
-  virtual int do_prepare_row(THD*, RELAY_LOG_INFO*, TABLE*,
+  virtual int do_prepare_row(THD*, RELAY_LOG_INFO const*, TABLE*,
                              char const *row_start, char const **row_end);
   virtual int do_exec_row(TABLE *table);
 #endif

sql/rpl_rli.cc

@@ -29,14 +29,15 @@ int init_strvar_from_file(char *var, int max_size, IO_CACHE *f,
 
 
 st_relay_log_info::st_relay_log_info()
-  :no_storage(FALSE), info_fd(-1), cur_log_fd(-1), save_temporary_tables(0),
+  :no_storage(FALSE), replicate_same_server_id(::replicate_same_server_id),
+   info_fd(-1), cur_log_fd(-1), save_temporary_tables(0),
    cur_log_old_open_count(0), group_master_log_pos(0), log_space_total(0),
    ignore_log_space_limit(0), last_master_timestamp(0), slave_skip_counter(0),
    abort_pos_wait(0), slave_run_id(0), sql_thd(0), last_slave_errno(0),
    inited(0), abort_slave(0), slave_running(0), until_condition(UNTIL_NONE),
    until_log_pos(0), retried_trans(0),
    tables_to_lock(0), tables_to_lock_count(0),
-   unsafe_to_stop_at(0)
+   last_event_start_time(0)
 {
   DBUG_ENTER("st_relay_log_info::st_relay_log_info");
 
@@ -1001,6 +1002,22 @@ bool st_relay_log_info::is_until_satisfied()
     log_pos= group_relay_log_pos;
   }
 
+#ifndef DBUG_OFF
+  {
+    char buf[32];
+    DBUG_PRINT("info", ("group_master_log_name='%s', group_master_log_pos=%s",
+                        group_master_log_name, llstr(group_master_log_pos, buf)));
+    DBUG_PRINT("info", ("group_relay_log_name='%s', group_relay_log_pos=%s",
+                        group_relay_log_name, llstr(group_relay_log_pos, buf)));
+    DBUG_PRINT("info", ("(%s) log_name='%s', log_pos=%s",
+                        until_condition == UNTIL_MASTER_POS ? "master" : "relay",
+                        log_name, llstr(log_pos, buf)));
+    DBUG_PRINT("info", ("(%s) until_log_name='%s', until_log_pos=%s",
+                        until_condition == UNTIL_MASTER_POS ? "master" : "relay",
+                        until_log_name, llstr(until_log_pos, buf)));
+  }
+#endif
+
   if (until_log_names_cmp_result == UNTIL_LOG_NAMES_CMP_UNKNOWN)
   {
     /*
@@ -1056,30 +1073,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)
 {
@@ -1106,7 +1112,7 @@ void st_relay_log_info::cleanup_context(THD *thd, bool error)
   m_table_map.clear_tables();
   close_thread_tables(thd);
   clear_tables_to_lock();
-  unsafe_to_stop_at= 0;
+  last_event_start_time= 0;
   DBUG_VOID_RETURN;
 }
 

sql/rpl_rli.h

@@ -58,6 +58,15 @@ typedef struct st_relay_log_info
   */
   bool no_storage;
 
+  /*
+    If true, events with the same server id should be replicated. This
+    field is set on creation of a relay log info structure by copying
+    the value of ::replicate_same_server_id and can be overridden if
+    necessary. For example of when this is done, check sql_binlog.cc,
+    where the BINLOG statement can be used to execute "raw" events.
+   */
+  bool replicate_same_server_id;
+
   /*** The following variables can only be read when protect by data lock ****/
 
   /*
@@ -292,14 +301,19 @@ 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);
+  bool cached_charset_compare(char *charset) const;
-
-  void transaction_end(THD*);
 
   void cleanup_context(THD *, bool);
   void clear_tables_to_lock();
 
-  time_t unsafe_to_stop_at;
+  /*
+    Used by row-based replication to detect that it should not stop at
+    this event, but give it a chance to send more events. The time
+    where the last event inside a group started is stored here. If the
+    variable is zero, we are not in a group (but may be in a
+    transaction).
+   */
+  time_t last_event_start_time;
 } RELAY_LOG_INFO;
 
 

sql/rpl_utility.cc

@@ -108,7 +108,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
 {
   /*
@@ -116,6 +116,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

@@ -117,7 +117,7 @@ public:
     @retval 1  if the table definition is not compatible with @c table
     @retval 0  if the table definition is compatible with @c 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

@@ -519,11 +519,11 @@ static bool sql_slave_killed(THD* thd, RELAY_LOG_INFO* rli)
       really one minute of idleness, we don't timeout if the slave SQL thread
       is actively working.
     */
-    if (!rli->unsafe_to_stop_at)
+    if (rli->last_event_start_time == 0)
       DBUG_RETURN(1);
     DBUG_PRINT("info", ("Slave SQL thread is in an unsafe situation, giving "
                         "it some grace period"));
-    if (difftime(time(0), rli->unsafe_to_stop_at) > 60)
+    if (difftime(time(0), rli->last_event_start_time) > 60)
     {
       slave_print_msg(ERROR_LEVEL, rli, 0,
                       "SQL thread had to stop in an unsafe situation, in "
@@ -557,7 +557,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 *, ...);
@@ -579,9 +579,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:
@@ -813,7 +813,7 @@ do not trust column Seconds_Behind_Master of SHOW SLAVE STATUS");
   {
     if ((master_row= mysql_fetch_row(master_res)) &&
         (::server_id == strtoul(master_row[1], 0, 10)) &&
-        !replicate_same_server_id)
+        !mi->rli.replicate_same_server_id)
       errmsg= "The slave I/O thread stops because master and slave have equal \
 MySQL server ids; these ids must be different for replication to work (or \
 the --replicate-same-server-id option must be used on slave but this does \
@@ -1390,7 +1390,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");
 
@@ -1401,7 +1401,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;
 }
 
@@ -1609,7 +1616,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");
 
@@ -1715,77 +1723,42 @@ static int exec_relay_log_event(THD* thd, RELAY_LOG_INFO* rli)
   if (ev)
   {
     int type_code = ev->get_type_code();
-    int exec_res;
+    int exec_res= 0;
 
     /*
-      Queries originating from this server must be skipped.
-      Low-level events (Format_desc, Rotate, Stop) from this server
-      must also be skipped. But for those we don't want to modify
-      group_master_log_pos, because these events did not exist on the master.
-      Format_desc is not completely skipped.
-      Skip queries specified by the user in slave_skip_counter.
-      We can't however skip events that has something to do with the
-      log files themselves.
-      Filtering on own server id is extremely important, to ignore execution of
-      events created by the creation/rotation of the relay log (remember that
-      now the relay log starts with its Format_desc, has a Rotate etc).
     */
 
-    DBUG_PRINT("info",("type_code=%d, server_id=%d",type_code,ev->server_id));
+    DBUG_PRINT("info",("type_code=%d (%s), server_id=%d",
+                       type_code, ev->get_type_str(), ev->server_id));
+    DBUG_PRINT("info", ("thd->options={ %s%s}",
+                        FLAGSTR(thd->options, OPTION_NOT_AUTOCOMMIT),
+                        FLAGSTR(thd->options, OPTION_BEGIN)));
 
-    if ((ev->server_id == (uint32) ::server_id &&
-         !replicate_same_server_id &&
-         type_code != FORMAT_DESCRIPTION_EVENT) ||
-        (rli->slave_skip_counter &&
-         type_code != ROTATE_EVENT && type_code != STOP_EVENT &&
-         type_code != START_EVENT_V3 && type_code!= FORMAT_DESCRIPTION_EVENT))
-    {
-      DBUG_PRINT("info", ("event skipped"));
-      /*
-        We only skip the event here and do not increase the group log
-        position.  In the event that we have to restart, this means
-        that we might have to skip the event again, but that is a
-        minor issue.
 
-        If we were to increase the group log position when skipping an
-        event, it might be that we are restarting at the wrong
-        position and have events before that we should have executed,
-        so not increasing the group log position is a sure bet in this
-        case.
 
-        In this way, we just step the group log position when we
+    /*
-        *know* that we are at the end of a group.
+      Execute the event to change the database and update the binary
-       */
+      log coordinates, but first we set some data that is needed for
-      rli->inc_event_relay_log_pos();
+      the thread.
 
-      /*
+      The event will be executed unless it is supposed to be skipped.
-        Protect against common user error of setting the counter to 1
+
-        instead of 2 while recovering from an insert which used auto_increment,
+      Queries originating from this server must be skipped.  Low-level
-        rand or user var.
+      events (Format_description_log_event, Rotate_log_event,
-      */
+      Stop_log_event) from this server must also be skipped. But for
-      if (rli->slave_skip_counter &&
+      those we don't want to modify 'group_master_log_pos', because
-          !((type_code == INTVAR_EVENT ||
+      these events did not exist on the master.
-             type_code == RAND_EVENT ||
+      Format_description_log_event is not completely skipped.
-             type_code == USER_VAR_EVENT) &&
+
-            rli->slave_skip_counter == 1) &&
+      Skip queries specified by the user in 'slave_skip_counter'.  We
-          /*
+      can't however skip events that has something to do with the log
-            The events from ourselves which have something to do with the relay
+      files themselves.
-            log itself must be skipped, true, but they mustn't decrement
+
-            rli->slave_skip_counter, because the user is supposed to not see
+      Filtering on own server id is extremely important, to ignore
-            these events (they are not in the master's binlog) and if we
+      execution of events created by the creation/rotation of the relay
-            decremented, START SLAVE would for example decrement when it sees
+      log (remember that now the relay log starts with its Format_desc,
-            the Rotate, so the event which the user probably wanted to skip
+      has a Rotate etc).
-            would not be skipped.
+    */
-          */
-          !(ev->server_id == (uint32) ::server_id &&
-            (type_code == ROTATE_EVENT || type_code == STOP_EVENT ||
-             type_code == START_EVENT_V3 || type_code == FORMAT_DESCRIPTION_EVENT)))
-        --rli->slave_skip_counter;
-      pthread_mutex_unlock(&rli->data_lock);
-      delete ev;
-      DBUG_RETURN(0);                                 // avoid infinite update loops
-    }
-    pthread_mutex_unlock(&rli->data_lock);
 
     thd->server_id = ev->server_id; // use the original server id for logging
     thd->set_time();                            // time the query
@@ -1793,13 +1766,63 @@ static int exec_relay_log_event(THD* thd, RELAY_LOG_INFO* rli)
     if (!ev->when)
       ev->when = time(NULL);
     ev->thd = thd; // because up to this point, ev->thd == 0
-    DBUG_PRINT("info", ("thd->options={ %s%s}",
-                        FLAGSTR(thd->options, OPTION_NOT_AUTOCOMMIT),
-                        FLAGSTR(thd->options, OPTION_BEGIN)));
 
-    exec_res = ev->exec_event(rli);
+    int reason= ev->shall_skip(rli);
-    DBUG_PRINT("info", ("exec_event result: %d", exec_res));
+    if (reason == Log_event::EVENT_SKIP_COUNT)
-    DBUG_ASSERT(rli->sql_thd==thd);
+      --rli->slave_skip_counter;
+    pthread_mutex_unlock(&rli->data_lock);
+    if (reason == Log_event::EVENT_SKIP_NOT)
+      exec_res= ev->apply_event(rli);
+#ifndef DBUG_OFF
+    else
+    {
+      /*
+        This only prints information to the debug trace.
+
+        TODO: Print an informational message to the error log?
+       */
+      static const char *const explain[] = {
+        "event was not skipped",                  // EVENT_SKIP_NOT,
+        "event originated from this server",      // EVENT_SKIP_IGNORE,
+        "event skip counter was non-zero"         // EVENT_SKIP_COUNT
+      };
+      DBUG_PRINT("info", ("%s was skipped because %s",
+                          ev->get_type_str(), explain[reason]));
+    }
+#endif
+
+    DBUG_PRINT("info", ("apply_event error = %d", exec_res));
+    if (exec_res == 0)
+    {
+      int error= ev->update_pos(rli);
+      char buf[22];
+      DBUG_PRINT("info", ("update_pos error = %d", error));
+      DBUG_PRINT("info", ("group %s %s",
+                          llstr(rli->group_relay_log_pos, buf),
+                          rli->group_relay_log_name));
+      DBUG_PRINT("info", ("event %s %s",
+                          llstr(rli->event_relay_log_pos, buf),
+                          rli->event_relay_log_name));
+      /*
+        The update should not fail, so print an error message and
+        return an error code.
+
+        TODO: Replace this with a decent error message when merged
+        with BUG#24954 (which adds several new error message).
+      */
+      if (error)
+      {
+        slave_print_msg(ERROR_LEVEL, rli, ER_UNKNOWN_ERROR,
+                        "It was not possible to update the positions"
+                        " of the relay log information: the slave may"
+                        " be in an inconsistent state."
+                        " Stopped in %s position %s",
+                        rli->group_relay_log_name,
+                        llstr(rli->group_relay_log_pos, buf));
+        DBUG_RETURN(1);
+      }
+    }
+
     /*
        Format_description_log_event should not be deleted because it will be
        used to read info about the relay log's format; it will be deleted when
@@ -2366,13 +2389,17 @@ Slave SQL thread aborted. Can't execute init_slave query");
     THD_CHECK_SENTRY(thd);
     if (exec_relay_log_event(thd,rli))
     {
+      DBUG_PRINT("info", ("exec_relay_log_event() failed"));
       // do not scare the user if SQL thread was simply killed or stopped
       if (!sql_slave_killed(thd,rli))
       {
         /*
-          retrieve as much info as possible from the thd and, error codes and warnings
+          retrieve as much info as possible from the thd and, error
-          and print this to the error log as to allow the user to locate the error
+          codes and warnings and print this to the error log as to
+          allow the user to locate the error
         */
+        DBUG_PRINT("info", ("thd->net.last_errno=%d; rli->last_slave_errno=%d",
+                            thd->net.last_errno, rli->last_slave_errno));
         if (thd->net.last_errno != 0)
         {
           if (rli->last_slave_errno == 0)
@@ -2699,6 +2726,7 @@ static int queue_binlog_ver_1_event(MASTER_INFO *mi, const char *buf,
     my_free((char*) tmp_buf, MYF(MY_ALLOW_ZERO_PTR));
     DBUG_RETURN(1);
   }
+
   pthread_mutex_lock(&mi->data_lock);
   ev->log_pos= mi->master_log_pos; /* 3.23 events don't contain log_pos */
   switch (ev->get_type_code()) {
@@ -2962,7 +2990,7 @@ int queue_event(MASTER_INFO* mi,const char* buf, ulong event_len)
   pthread_mutex_lock(log_lock);
 
   if ((uint4korr(buf + SERVER_ID_OFFSET) == ::server_id) &&
-      !replicate_same_server_id)
+      !mi->rli.replicate_same_server_id)
   {
     /*
       Do not write it to the relay log.

sql/slave.h

@@ -162,9 +162,9 @@ bool show_binlog_info(THD* thd);
 bool rpl_master_has_bug(RELAY_LOG_INFO *rli, uint bug_id);
 
 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);
 
@@ -182,7 +182,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);

sql/sql_binlog.cc

@@ -163,9 +163,17 @@ void mysql_client_binlog_statement(THD* thd)
                           (ulong) uint4korr(bufptr+EVENT_LEN_OFFSET)));
 #endif
       ev->thd= thd;
-      if (IF_DBUG(int err= ) ev->exec_event(thd->rli_fake))
+      /*
+        We go directly to the application phase, since we don't need
+        to check if the event shall be skipped or not.
+
+        Neither do we have to update the log positions, since that is
+        not used at all: the rli_fake instance is used only for error
+        reporting.
+       */
+      if (IF_DBUG(int err= ) ev->apply_event(thd->rli_fake))
       {
-        DBUG_PRINT("error", ("exec_event() returned: %d", err));
+        DBUG_PRINT("info", ("apply_event() returned: %d", err));
         /*
           TODO: Maybe a better error message since the BINLOG statement
           now contains several events.