Merge mkindahl@bk-internal.mysql.com:/home/bk/mysql-5.1-new-rpl

into  capulet.net:/home/bk/mysql-5.1-rpl

client/Makefile.am

@@ -96,7 +96,9 @@ DEFS =			-DUNDEF_THREADS_HACK \
 			-DDATADIR="\"$(localstatedir)\""
 
 sql_src=log_event.h mysql_priv.h rpl_constants.h \
-	log_event.cc my_decimal.h my_decimal.cc
+	log_event.cc my_decimal.h my_decimal.cc \
+	log_event_old.h log_event_old.cc \
+	rpl_record_old.h rpl_record_old.cc
 strings_src=decimal.c
 
 link_sources:

client/mysqlbinlog.cc

@@ -1600,10 +1600,12 @@ int main(int argc, char** argv)
 #include "decimal.c"
 #include "my_decimal.cpp"
 #include "log_event.cpp"
+#include "log_event_old.cpp"
 #else
 #include "my_decimal.h"
 #include "decimal.c"
 #include "my_decimal.cc"
 #include "log_event.cc"
+#include "log_event_old.cc"
 #endif
 

client/mysqltest.c

@@ -2573,7 +2573,7 @@ int do_save_master_pos()
 
     if (have_ndbcluster)
     {
-      ulonglong start_epoch= 0, applied_epoch= 0,
+      ulonglong start_epoch= 0, handled_epoch= 0,
 	latest_epoch=0, latest_trans_epoch=0,
 	latest_handled_binlog_epoch= 0, latest_received_binlog_epoch= 0,
 	latest_applied_binlog_epoch= 0;
@@ -2676,9 +2676,9 @@ int do_save_master_pos()
 	if (!row)
 	  die("result does not contain '%s' in '%s'",
 	      binlog, query);
-	if (latest_applied_binlog_epoch > applied_epoch)
+	if (latest_handled_binlog_epoch > handled_epoch)
 	  count= 0;
-	applied_epoch= latest_applied_binlog_epoch;
+	handled_epoch= latest_handled_binlog_epoch;
 	count++;
 	if (latest_handled_binlog_epoch >= start_epoch)
           do_continue= 0;

libmysqld/Makefile.am

@@ -53,7 +53,9 @@ sqlsources = derror.cc field.cc field_conv.cc strfunc.cc filesort.cc \
 	item_func.cc item_strfunc.cc item_sum.cc item_timefunc.cc \
 	item_geofunc.cc item_subselect.cc item_row.cc\
 	item_xmlfunc.cc \
-	key.cc lock.cc log.cc log_event.cc sql_state.c \
+	key.cc lock.cc log.cc sql_state.c \
+	log_event.cc \
+	log_event_old.cc rpl_record_old.cc \
 	protocol.cc net_serv.cc opt_range.cc \
 	opt_sum.cc procedure.cc records.cc sql_acl.cc \
 	sql_load.cc discover.cc sql_locale.cc \

mysql-test/r/ctype_utf8.result

@@ -1480,6 +1480,49 @@ aa
 xxx
 yyy
 DROP TABLE t1;
+create table t1 (
+a varchar(26) not null
+) default character set utf8;
+insert into t1 (a) values ('abcdefghijklmnopqrstuvwxyz');
+select * from t1;
+a
+abcdefghijklmnopqrstuvwxyz
+alter table t1 change a a varchar(20) character set utf8 not null;
+Warnings:
+Warning	1265	Data truncated for column 'a' at row 1
+select * from t1;
+a
+abcdefghijklmnopqrst
+alter table t1 change a a char(15) character set utf8 not null;
+Warnings:
+Warning	1265	Data truncated for column 'a' at row 1
+select * from t1;
+a
+abcdefghijklmno
+alter table t1 change a a char(10) character set utf8 not null;
+Warnings:
+Warning	1265	Data truncated for column 'a' at row 1
+select * from t1;
+a
+abcdefghij
+alter table t1 change a a varchar(5) character set utf8 not null;
+Warnings:
+Warning	1265	Data truncated for column 'a' at row 1
+select * from t1;
+a
+abcde
+drop table t1;
+create table t1 (
+a varchar(4000) not null
+) default character set utf8;
+insert into t1 values (repeat('a',4000));
+alter table t1 change a a varchar(3000) character set utf8 not null;
+Warnings:
+Warning	1265	Data truncated for column 'a' at row 1
+select length(a) from t1;
+length(a)
+3000
+drop table t1;
 set names utf8;
 select hex(char(1 using utf8));
 hex(char(1 using utf8))

mysql-test/r/rpl_ndb_circular.result

+stop slave;
+drop table if exists t1,t2,t3,t4,t5,t6,t7,t8,t9;
+reset master;
+reset slave;
+drop table if exists t1,t2,t3,t4,t5,t6,t7,t8,t9;
+start slave;
+RESET MASTER;
+CHANGE MASTER TO master_host="127.0.0.1",master_port=SLAVE_PORT,master_user="root";
+START SLAVE;
+CREATE TABLE t1 (a int key, b int) ENGINE=ndb;
+SHOW TABLES;
+Tables_in_test
+t1
+INSERT INTO t1 VALUES (1,2);
+INSERT INTO t1 VALUES (2,3);
+SELECT * FROM t1 ORDER BY a;
+a	b
+1	2
+2	3
+show slave status;;
+Slave_IO_State	#
+Master_Host	127.0.0.1
+Master_User	root
+Master_Port	#
+Connect_Retry	1
+Master_Log_File	master-bin.000001
+Read_Master_Log_Pos	#
+Relay_Log_File	#
+Relay_Log_Pos	#
+Relay_Master_Log_File	master-bin.000001
+Slave_IO_Running	Yes
+Slave_SQL_Running	Yes
+Replicate_Do_DB	
+Replicate_Ignore_DB	
+Replicate_Do_Table	
+Replicate_Ignore_Table	
+Replicate_Wild_Do_Table	
+Replicate_Wild_Ignore_Table	
+Last_Errno	0
+Last_Error	
+Skip_Counter	0
+Exec_Master_Log_Pos	#
+Relay_Log_Space	#
+Until_Condition	None
+Until_Log_File	
+Until_Log_Pos	0
+Master_SSL_Allowed	No
+Master_SSL_CA_File	
+Master_SSL_CA_Path	
+Master_SSL_Cert	
+Master_SSL_Cipher	
+Master_SSL_Key	
+Seconds_Behind_Master	#
+Master_SSL_Verify_Server_Cert	No
+SELECT * FROM t1 ORDER BY a;
+a	b
+1	2
+2	3
+show slave status;;
+Slave_IO_State	#
+Master_Host	127.0.0.1
+Master_User	root
+Master_Port	#
+Connect_Retry	60
+Master_Log_File	slave-bin.000001
+Read_Master_Log_Pos	#
+Relay_Log_File	#
+Relay_Log_Pos	#
+Relay_Master_Log_File	slave-bin.000001
+Slave_IO_Running	Yes
+Slave_SQL_Running	Yes
+Replicate_Do_DB	
+Replicate_Ignore_DB	
+Replicate_Do_Table	
+Replicate_Ignore_Table	
+Replicate_Wild_Do_Table	
+Replicate_Wild_Ignore_Table	
+Last_Errno	0
+Last_Error	
+Skip_Counter	0
+Exec_Master_Log_Pos	#
+Relay_Log_Space	#
+Until_Condition	None
+Until_Log_File	
+Until_Log_Pos	0
+Master_SSL_Allowed	No
+Master_SSL_CA_File	
+Master_SSL_CA_Path	
+Master_SSL_Cert	
+Master_SSL_Cipher	
+Master_SSL_Key	
+Seconds_Behind_Master	#
+Master_SSL_Verify_Server_Cert	No
+STOP SLAVE;
+DROP TABLE t1;

mysql-test/r/rpl_ndb_circular_simplex.result

@@ -19,10 +19,10 @@ SHOW SLAVE STATUS;
 Slave_IO_State	#
 Master_Host	127.0.0.1
 Master_User	root
-Master_Port	9308
+Master_Port	SLAVE_PORT
 Connect_Retry	60
 Master_Log_File	slave-bin.000001
-Read_Master_Log_Pos	468
+Read_Master_Log_Pos	#
 Relay_Log_File	#
 Relay_Log_Pos	#
 Relay_Master_Log_File	slave-bin.000001
@@ -49,6 +49,7 @@ Master_SSL_Cert
 Master_SSL_Cipher	
 Master_SSL_Key	
 Seconds_Behind_Master	#
+Master_SSL_Verify_Server_Cert	No
 SELECT * FROM t1 ORDER BY a;
 a	b
 1	2

mysql-test/t/ctype_utf8.test

@@ -1193,6 +1193,41 @@ SELECT DISTINCT id FROM t1 ORDER BY id;
 
 DROP TABLE t1;
 
+#
+# Bug#20095 Changing length of VARCHAR field with UTF8
+# collation does not truncate values
+#
+create table t1 (
+  a varchar(26) not null
+) default character set utf8;
+insert into t1 (a) values ('abcdefghijklmnopqrstuvwxyz');
+select * from t1;
+# varchar to varchar
+alter table t1 change a a varchar(20) character set utf8 not null;
+select * from t1;
+# varchar to char
+alter table t1 change a a char(15) character set utf8 not null;
+select * from t1;
+# char to char
+alter table t1 change a a char(10) character set utf8 not null;
+select * from t1;
+# char to varchar
+alter table t1 change a a varchar(5) character set utf8 not null;
+select * from t1;
+drop table t1;
+
+#
+# Check that do_varstring2_mb produces a warning
+#
+create table t1 (
+  a varchar(4000) not null
+) default character set utf8;
+insert into t1 values (repeat('a',4000));
+alter table t1 change a a varchar(3000) character set utf8 not null;
+select length(a) from t1;
+drop table t1;
+
+
 #
 #  Bug#10504: Character set does not support traditional mode
 #  Bug#14146: CHAR(...USING ...) and CONVERT(CHAR(...) USING...)

mysql-test/t/rpl_ndb_circular.test

+--source include/have_ndb.inc
+--source include/have_binlog_format_row.inc
+--source include/master-slave.inc
+
+# set up circular replication
+
+--connection slave
+RESET MASTER;
+--connection master
+--replace_result $SLAVE_MYPORT SLAVE_PORT
+--eval CHANGE MASTER TO master_host="127.0.0.1",master_port=$SLAVE_MYPORT,master_user="root"
+START SLAVE;
+
+# create the table on the "slave"
+--connection slave
+CREATE TABLE t1 (a int key, b int) ENGINE=ndb;
+#CREATE TABLE t2 (a int key, b int) ENGINE=ndb;
+--save_master_pos
+--connection master
+--sync_with_master
+# now we should have a table on the master as well
+SHOW TABLES;
+
+# insert some values on the slave and master
+--connection master
+INSERT INTO t1 VALUES (1,2);
+--connection slave
+INSERT INTO t1 VALUES (2,3);
+
+# ensure data has propagated both ways
+--connection slave
+--save_master_pos
+--connection master
+--sync_with_master
+--sync_slave_with_master
+
+# connect to slave and ensure data it there.
+--connection slave
+SELECT * FROM t1 ORDER BY a;
+#SELECT * FROM t2 ORDER BY a;
+--replace_column 1 # 4 # 7 # 8 # 9 # 22 # 23 # 33 #
+--query_vertical show slave status;
+# connect to master and ensure data it there.
+--connection master
+SELECT * FROM t1 ORDER BY a;
+#SELECT * FROM t2 ORDER BY a;
+--replace_column 1 # 4 # 7 # 8 # 9 # 22 # 23 # 33 #
+--query_vertical show slave status;
+
+# stop replication on "master" as not to replicate
+# shutdown circularly, eg drop table
+--connection master
+STOP SLAVE;
+
+# cleanup
+--connection master
+DROP TABLE t1;

mysql-test/t/rpl_ndb_circular_simplex.test

@@ -50,8 +50,8 @@ sync_with_master;
 #SHOW BINLOG EVENTS;
 
 # Check that there is no error in replication
---replace_result $MASTER_MYPORT MASTER_PORT
---replace_column 1 # 8 # 9 # 23 # 33 #
+--replace_result $SLAVE_MYPORT SLAVE_PORT
+--replace_column 1 # 7 # 8 # 9 # 23 # 33 #
 query_vertical SHOW SLAVE STATUS;
 
 # Check that we have the data on the master

sql/CMakeLists.txt

@@ -48,6 +48,7 @@ ADD_EXECUTABLE(mysqld ../sql-common/client.c derror.cc des_key_file.cc
                item_create.cc item_func.cc item_geofunc.cc item_row.cc 
                item_strfunc.cc item_subselect.cc item_sum.cc item_timefunc.cc 
                key.cc log.cc lock.cc log_event.cc message.rc 
+               log_event_old.cc rpl_record_old.cc
                message.h mf_iocache.cc my_decimal.cc ../sql-common/my_time.c
                mysqld.cc net_serv.cc 
                nt_servc.cc nt_servc.h opt_range.cc opt_range.h opt_sum.cc 

sql/Makefile.am

@@ -59,6 +59,7 @@ noinst_HEADERS =	item.h item_func.h item_sum.h item_cmpfunc.h \
 			lex.h lex_symbol.h sql_acl.h sql_crypt.h  \
 			log_event.h sql_repl.h slave.h rpl_filter.h \
 			rpl_injector.h \
+			log_event_old.h rpl_record_old.h \
 			stacktrace.h sql_sort.h sql_cache.h set_var.h \
 			spatial.h gstream.h client_settings.h tzfile.h \
 			tztime.h my_decimal.h\
@@ -88,6 +89,7 @@ mysqld_SOURCES =	sql_lex.cc sql_handler.cc sql_partition.cc \
 			procedure.cc sql_test.cc \
 			log.cc log_event.cc init.cc derror.cc sql_acl.cc \
 			unireg.cc des_key_file.cc \
+			log_event_old.cc rpl_record_old.cc \
 			discover.cc time.cc opt_range.cc opt_sum.cc \
 		   	records.cc filesort.cc handler.cc \
 			ha_partition.cc \

sql/field_conv.cc

@@ -433,6 +433,26 @@ static void do_varstring1(Copy_field *copy)
 }
 
 
+static void do_varstring1_mb(Copy_field *copy)
+{
+  int well_formed_error;
+  CHARSET_INFO *cs= copy->from_field->charset();
+  uint from_length= (uint) *(uchar*) copy->from_ptr;
+  const char *from_ptr= copy->from_ptr + 1;
+  uint to_char_length= (copy->to_length - 1) / cs->mbmaxlen;
+  uint length= cs->cset->well_formed_len(cs, from_ptr, from_ptr + from_length,
+                                         to_char_length, &well_formed_error);
+  if (length < from_length)
+  {
+    if (current_thd->count_cuted_fields)
+      copy->to_field->set_warning(MYSQL_ERROR::WARN_LEVEL_WARN,
+                                  WARN_DATA_TRUNCATED, 1);
+  }
+  *(uchar*) copy->to_ptr= (uchar) length;
+  memcpy(copy->to_ptr + 1, from_ptr, length);
+}
+
+
 static void do_varstring2(Copy_field *copy)
 {
   uint length=uint2korr(copy->from_ptr);
@@ -458,6 +478,12 @@ static void do_varstring2_mb(Copy_field *copy)
   const char *from_beg= copy->from_ptr + HA_KEY_BLOB_LENGTH;
   uint length= cs->cset->well_formed_len(cs, from_beg, from_beg + from_length,
                                          char_length, &well_formed_error);
+  if (length < from_length)
+  {
+    if (current_thd->count_cuted_fields)
+      copy->to_field->set_warning(MYSQL_ERROR::WARN_LEVEL_WARN,
+                                  WARN_DATA_TRUNCATED, 1);
+  }  
   int2store(copy->to_ptr, length);
   memcpy(copy->to_ptr+HA_KEY_BLOB_LENGTH, from_beg, length);
 }
@@ -633,8 +659,10 @@ void (*Copy_field::get_copy_func(Field *to,Field *from))(Copy_field*)
           return do_field_string;
         if (to_length != from_length)
           return (((Field_varstring*) to)->length_bytes == 1 ?
-                  do_varstring1 : (from->charset()->mbmaxlen == 1 ?
-                                   do_varstring2 : do_varstring2_mb));
+                  (from->charset()->mbmaxlen == 1 ? do_varstring1 :
+                                                    do_varstring1_mb) :
+                  (from->charset()->mbmaxlen == 1 ? do_varstring2 :
+                                                    do_varstring2_mb));
       }
       else if (to_length < from_length)
 	return (from->charset()->mbmaxlen == 1 ?

sql/log_event.cc

@@ -972,6 +972,15 @@ Log_event* Log_event::read_log_event(const char* buf, uint event_len,
     ev = new Format_description_log_event(buf, event_len, description_event); 
     break;
 #if defined(HAVE_REPLICATION) 
+  case PRE_GA_WRITE_ROWS_EVENT:
+    ev = new Write_rows_log_event_old(buf, event_len, description_event);
+    break;
+  case PRE_GA_UPDATE_ROWS_EVENT:
+    ev = new Update_rows_log_event_old(buf, event_len, description_event);
+    break;
+  case PRE_GA_DELETE_ROWS_EVENT:
+    ev = new Delete_rows_log_event_old(buf, event_len, description_event);
+    break;
   case WRITE_ROWS_EVENT:
     ev = new Write_rows_log_event(buf, event_len, description_event);
     break;

sql/log_event.h

@@ -2429,7 +2429,7 @@ public:
     return Rows_log_event::is_valid() && m_cols_ai.bitmap;
   }
 
-private:
+protected:
   virtual Log_event_type get_type_code() { return (Log_event_type)TYPE_CODE; }
 
 #ifdef MYSQL_CLIENT
@@ -2500,7 +2500,7 @@ public:
   }
 #endif
   
-private:
+protected:
   virtual Log_event_type get_type_code() { return (Log_event_type)TYPE_CODE; }
 
 #ifdef MYSQL_CLIENT
@@ -2521,6 +2521,8 @@ private:
 };
 
 
+#include "log_event_old.h"
+
 /**
    Class representing an incident, an occurance out of the ordinary,
    that happened on the master.

sql/log_event_old.cc

+
+#include "mysql_priv.h"
+#include "log_event_old.h"
+#include "rpl_record_old.h"
+
+#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
+int
+Write_rows_log_event_old::do_prepare_row(THD *thd, RELAY_LOG_INFO *rli,
+                                         TABLE *table,
+                                         char const *row_start,
+                                         char const **row_end)
+{
+  DBUG_ASSERT(table != NULL);
+  DBUG_ASSERT(row_start && row_end);
+
+  int error;
+  error= unpack_row_old(rli, table, m_width, table->record[0],
+                        row_start, &m_cols, row_end, &m_master_reclength,
+                        table->write_set, PRE_GA_WRITE_ROWS_EVENT);
+  bitmap_copy(table->read_set, table->write_set);
+  return error;
+}
+
+
+int
+Delete_rows_log_event_old::do_prepare_row(THD *thd, RELAY_LOG_INFO *rli,
+                                          TABLE *table,
+                                          char const *row_start,
+                                          char const **row_end)
+{
+  int error;
+  DBUG_ASSERT(row_start && row_end);
+  /*
+    This assertion actually checks that there is at least as many
+    columns on the slave as on the master.
+  */
+  DBUG_ASSERT(table->s->fields >= m_width);
+
+  error= unpack_row_old(rli, table, m_width, table->record[0],
+                        row_start, &m_cols, row_end, &m_master_reclength,
+                        table->read_set, PRE_GA_DELETE_ROWS_EVENT);
+  /*
+    If we will access rows using the random access method, m_key will
+    be set to NULL, so we do not need to make a key copy in that case.
+   */
+  if (m_key)
+  {
+    KEY *const key_info= table->key_info;
+
+    key_copy(m_key, table->record[0], key_info, 0);
+  }
+
+  return error;
+}
+
+
+int Update_rows_log_event_old::do_prepare_row(THD *thd, RELAY_LOG_INFO *rli,
+                                              TABLE *table,
+                                              char const *row_start,
+                                              char const **row_end)
+{
+  int error;
+  DBUG_ASSERT(row_start && row_end);
+  /*
+    This assertion actually checks that there is at least as many
+    columns on the slave as on the master.
+  */
+  DBUG_ASSERT(table->s->fields >= m_width);
+
+  /* record[0] is the before image for the update */
+  error= unpack_row_old(rli, table, m_width, table->record[0],
+                        row_start, &m_cols, row_end, &m_master_reclength,
+                        table->read_set, PRE_GA_UPDATE_ROWS_EVENT);
+  row_start = *row_end;
+  /* m_after_image is the after image for the update */
+  error= unpack_row_old(rli, table, m_width, m_after_image,
+                        row_start, &m_cols, row_end, &m_master_reclength,
+                        table->write_set, PRE_GA_UPDATE_ROWS_EVENT);
+
+  DBUG_DUMP("record[0]", table->record[0], table->s->reclength);
+  DBUG_DUMP("m_after_image", m_after_image, table->s->reclength);
+
+
+  /*
+    If we will access rows using the random access method, m_key will
+    be set to NULL, so we do not need to make a key copy in that case.
+   */
+  if (m_key)
+  {
+    KEY *const key_info= table->key_info;
+
+    key_copy(m_key, table->record[0], key_info, 0);
+  }
+
+  return error;
+}
+
+#endif

sql/log_event_old.h

+#ifndef LOG_EVENT_OLD_H
+#define LOG_EVENT_OLD_H
+
+/*
+  Need to include this file at the proper position of log_event.h
+ */
+
+
+class Write_rows_log_event_old : public Write_rows_log_event
+{
+public:
+  enum
+  {
+    /* Support interface to THD::binlog_prepare_pending_rows_event */
+    TYPE_CODE = PRE_GA_WRITE_ROWS_EVENT
+  };
+
+#if defined(HAVE_REPLICATION)
+  Write_rows_log_event_old(const char *buf, uint event_len,
+                           const Format_description_log_event *descr)
+    : Write_rows_log_event(buf, event_len, descr)
+  {
+  }
+#endif
+
+private:
+  virtual Log_event_type get_type_code() { return (Log_event_type)TYPE_CODE; }
+
+#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
+  virtual int do_prepare_row(THD*, RELAY_LOG_INFO*, TABLE*,
+                             char const *row_start, char const **row_end);
+#endif
+};
+
+
+class Update_rows_log_event_old : public Update_rows_log_event
+{
+public:
+  enum 
+  {
+    /* Support interface to THD::binlog_prepare_pending_rows_event */
+    TYPE_CODE = PRE_GA_UPDATE_ROWS_EVENT
+  };
+
+#if defined(HAVE_REPLICATION)
+  Update_rows_log_event_old(const char *buf, uint event_len, 
+                            const Format_description_log_event *descr)
+    : Update_rows_log_event(buf, event_len, descr)
+  {
+  }
+#endif
+
+private:
+  virtual Log_event_type get_type_code() { return (Log_event_type)TYPE_CODE; }
+
+#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
+  virtual int do_prepare_row(THD*, RELAY_LOG_INFO*, TABLE*,
+                             char const *row_start, char const **row_end);
+#endif /* !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION) */
+};
+
+
+class Delete_rows_log_event_old : public Delete_rows_log_event
+{
+public:
+  enum 
+  {
+    /* Support interface to THD::binlog_prepare_pending_rows_event */
+    TYPE_CODE = PRE_GA_DELETE_ROWS_EVENT
+  };
+
+#if defined(HAVE_REPLICATION)
+  Delete_rows_log_event_old(const char *buf, uint event_len,
+                            const Format_description_log_event *descr)
+    : Delete_rows_log_event(buf, event_len, descr)
+  {
+  }
+#endif
+
+private:
+  virtual Log_event_type get_type_code() { return (Log_event_type)TYPE_CODE; }
+
+#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
+  virtual int do_prepare_row(THD*, RELAY_LOG_INFO*, TABLE*,
+                             char const *row_start, char const **row_end);
+#endif
+};
+
+
+#endif
+

sql/rpl_injector.cc

@@ -75,9 +75,11 @@ int injector::transaction::use_table(server_id_type sid, table tbl)
   if ((error= check_state(TABLE_STATE)))
     DBUG_RETURN(error);
 
+  server_id_type save_id= m_thd->server_id;
   m_thd->set_server_id(sid);
   error= m_thd->binlog_write_table_map(tbl.get_table(),
                                        tbl.is_transactional());
+  m_thd->set_server_id(save_id);
   DBUG_RETURN(error);
 }
 
@@ -91,9 +93,11 @@ int injector::transaction::write_row (server_id_type sid, table tbl,
    if (int error= check_state(ROW_STATE))
      DBUG_RETURN(error);
 
+   server_id_type save_id= m_thd->server_id;
    m_thd->set_server_id(sid);
    m_thd->binlog_write_row(tbl.get_table(), tbl.is_transactional(), 
                            cols, colcnt, record);
+   m_thd->set_server_id(save_id);
    DBUG_RETURN(0);
 }
 
@@ -107,9 +111,11 @@ int injector::transaction::delete_row(server_id_type sid, table tbl,
    if (int error= check_state(ROW_STATE))
      DBUG_RETURN(error);
 
+   server_id_type save_id= m_thd->server_id;
    m_thd->set_server_id(sid);
    m_thd->binlog_delete_row(tbl.get_table(), tbl.is_transactional(), 
                             cols, colcnt, record);
+   m_thd->set_server_id(save_id);
    DBUG_RETURN(0);
 }
 
@@ -123,9 +129,11 @@ int injector::transaction::update_row(server_id_type sid, table tbl,
    if (int error= check_state(ROW_STATE))
      DBUG_RETURN(error);
 
+   server_id_type save_id= m_thd->server_id;
    m_thd->set_server_id(sid);
    m_thd->binlog_update_row(tbl.get_table(), tbl.is_transactional(),
 		            cols, colcnt, before, after);
+   m_thd->set_server_id(save_id);
    DBUG_RETURN(0);
 }
 

sql/rpl_record_old.cc

+
+#include "mysql_priv.h"
+#include "rpl_record_old.h"
+
+my_size_t
+pack_row_old(THD *thd, TABLE *table, MY_BITMAP const* cols,
+             byte *row_data, const byte *record)
+{
+  Field **p_field= table->field, *field;
+  int n_null_bytes= table->s->null_bytes;
+  byte *ptr;
+  uint i;
+  my_ptrdiff_t const rec_offset= record - table->record[0];
+  my_ptrdiff_t const def_offset= table->s->default_values - table->record[0];
+  memcpy(row_data, record, n_null_bytes);
+  ptr= row_data+n_null_bytes;
+
+  for (i= 0 ; (field= *p_field) ; i++, p_field++)
+  {
+    if (bitmap_is_set(cols,i))
+    {
+      my_ptrdiff_t const offset=
+        field->is_null(rec_offset) ? def_offset : rec_offset;
+      field->move_field_offset(offset);
+      ptr= (byte*)field->pack((char *) ptr, field->ptr);
+      field->move_field_offset(-offset);
+    }
+  }
+  return (static_cast<my_size_t>(ptr - row_data));
+}
+
+
+/*
+  Unpack a row into a record.
+
+  SYNOPSIS
+    unpack_row()
+    rli     Relay log info
+    table   Table to unpack into
+    colcnt  Number of columns to read from record
+    record  Record where the data should be unpacked
+    row     Packed row data
+    cols    Pointer to columns data to fill in
+    row_end Pointer to variable that will hold the value of the
+            one-after-end position for the row
+    master_reclength
+            Pointer to variable that will be set to the length of the
+            record on the master side
+    rw_set  Pointer to bitmap that holds either the read_set or the
+            write_set of the table
+
+  DESCRIPTION
+
+      The row is assumed to only consist of the fields for which the
+      bitset represented by 'arr' and 'bits'; the other parts of the
+      record are left alone.
+
+      At most 'colcnt' columns are read: if the table is larger than
+      that, the remaining fields are not filled in.
+
+  RETURN VALUE
+
+      Error code, or zero if no error. The following error codes can
+      be returned:
+
+      ER_NO_DEFAULT_FOR_FIELD
+        Returned if one of the fields existing on the slave but not on
+        the master does not have a default value (and isn't nullable)
+ */
+#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
+int
+unpack_row_old(RELAY_LOG_INFO *rli,
+               TABLE *table, uint const colcnt, byte *record,
+               char const *row, MY_BITMAP const *cols,
+               char const **row_end, ulong *master_reclength,
+               MY_BITMAP* const rw_set, Log_event_type const event_type)
+{
+  DBUG_ASSERT(record && row);
+  my_ptrdiff_t const offset= record - (byte*) table->record[0];
+  my_size_t master_null_bytes= table->s->null_bytes;
+
+  if (colcnt != table->s->fields)
+  {
+    Field **fptr= &table->field[colcnt-1];
+    do
+      master_null_bytes= (*fptr)->last_null_byte();
+    while (master_null_bytes == Field::LAST_NULL_BYTE_UNDEF &&
+           fptr-- > table->field);
+
+    /*
+      If master_null_bytes is LAST_NULL_BYTE_UNDEF (0) at this time,
+      there were no nullable fields nor BIT fields at all in the
+      columns that are common to the master and the slave. In that
+      case, there is only one null byte holding the X bit.
+
+      OBSERVE! There might still be nullable columns following the
+      common columns, so table->s->null_bytes might be greater than 1.
+     */
+    if (master_null_bytes == Field::LAST_NULL_BYTE_UNDEF)
+      master_null_bytes= 1;
+  }
+
+  DBUG_ASSERT(master_null_bytes <= table->s->null_bytes);
+  memcpy(record, row, master_null_bytes);            // [1]
+  int error= 0;
+
+  bitmap_set_all(rw_set);
+
+  Field **const begin_ptr = table->field;
+  Field **field_ptr;
+  char const *ptr= row + master_null_bytes;
+  Field **const end_ptr= begin_ptr + colcnt;
+  for (field_ptr= begin_ptr ; field_ptr < end_ptr ; ++field_ptr)
+  {
+    Field *const f= *field_ptr;
+
+    if (bitmap_is_set(cols, field_ptr -  begin_ptr))
+    {
+      f->move_field_offset(offset);
+      ptr= f->unpack(f->ptr, ptr);
+      f->move_field_offset(-offset);
+      /* Field...::unpack() cannot return 0 */
+      DBUG_ASSERT(ptr != NULL);
+    }
+    else
+      bitmap_clear_bit(rw_set, field_ptr - begin_ptr);
+  }
+
+  *row_end = ptr;
+  if (master_reclength)
+  {
+    if (*field_ptr)
+      *master_reclength = (*field_ptr)->ptr - (char*) table->record[0];
+    else
+      *master_reclength = table->s->reclength;
+  }
+
+  /*
+    Set properties for remaining columns, if there are any. We let the
+    corresponding bit in the write_set be set, to write the value if
+    it was not there already. We iterate over all remaining columns,
+    even if there were an error, to get as many error messages as
+    possible.  We are still able to return a pointer to the next row,
+    so redo that.
+
+    This generation of error messages is only relevant when inserting
+    new rows.
+   */
+  for ( ; *field_ptr ; ++field_ptr)
+  {
+    uint32 const mask= NOT_NULL_FLAG | NO_DEFAULT_VALUE_FLAG;
+
+    DBUG_PRINT("debug", ("flags = 0x%x, mask = 0x%x, flags & mask = 0x%x",
+                         (*field_ptr)->flags, mask,
+                         (*field_ptr)->flags & mask));
+
+    if (event_type == WRITE_ROWS_EVENT &&
+        ((*field_ptr)->flags & mask) == mask)
+    {
+      slave_print_msg(ERROR_LEVEL, rli, ER_NO_DEFAULT_FOR_FIELD,
+                      "Field `%s` of table `%s`.`%s` "
+                      "has no default value and cannot be NULL",
+                      (*field_ptr)->field_name, table->s->db.str,
+                      table->s->table_name.str);
+      error = ER_NO_DEFAULT_FOR_FIELD;
+    }
+    else
+      (*field_ptr)->set_default();
+  }
+
+  return error;
+}
+#endif

sql/rpl_record_old.h

+#ifndef RPL_RECORD_OLD_H
+#define RPL_RECORD_OLD_H
+
+#ifndef MYSQL_CLIENT
+my_size_t pack_row_old(THD *thd, TABLE *table, MY_BITMAP const* cols,
+                       byte *row_data, const byte *record);
+
+#ifdef HAVE_REPLICATION
+int unpack_row_old(RELAY_LOG_INFO *rli,
+                   TABLE *table, uint const colcnt, byte *record,
+                   char const *row, MY_BITMAP const *cols,
+                   char const **row_end, ulong *master_reclength,
+                   MY_BITMAP* const rw_set,
+                   Log_event_type const event_type);
+#endif
+#endif
+#endif