Merge lp:~skinny.moey/drizzle/multi_level_slave into lp:~drizzle-trunk/drizzle/development

[Joe Daly]

This change adds the column ORIGINATING_SERVER_ID and ORIGINATING_COMMIT_ID to the DATA_DICTIONARY.SYS_REPLICATION_LOG and the SYS_REPLICATION_LOG.APPLIER_STATE table. These fields are added to allow a configuration where a 2nd level slave exists, to allow for recovery should a master fail, such as A->B->C (A replicates to B, B replicates to C). With these new columns C can be reconfigured to point to A with no guesswork as to the correct position to start at.

[Joe Daly (skinny.moey)]

on hold at the moment

[Patrick Crews (patrick-crews)]

Seeing a significant number of random failures on the randgen slave plugin tests with this code. Not tied to Joe's work, but it appears to be tied to the multi-master code itself.

Still testing current trunk vs. pre-merge trunk to make sure of this

[Patrick Crews (patrick-crews)]

It would appear that we need to back out the multi-master code from trunk - it is breaking the randgen slave_plugin tests (for single master/slave) with some regularity. Many runs pass with flying colors, but about 1/6 - 1/10 will have a failing test case with differences between master and slave (often the crash test though it varies).

Will recommend working with Shrews in his beta-multi-master branch until this is ironed out.
To repeat:
./dbqp --mode=randgen --randgen-path=/path/to/randgen --suite=slave_plugin

Unmerged revisions

2263. By Joe Daly on 2011-04-01

add originating_server_id and originating_commit_id for applier_state table, this is needed for a slave that does not have the replication log turned on

2262. By Joe Daly on 2011-04-01

merge trunk

2261. By Joe Daly on 2011-04-01

add ORIGINATING_SERVER_ID and +INNODB_REPLICATION_LOG DATA_DICTIONARY to INNODB_REPLICATION_LOG table

2260. By Joe Daly on 2011-03-31

merge trunk

2259. By Joe Daly on 2011-03-30

fix tests, and fix up if block setting originating_server_id and commit_id

2258. By Joe Daly on 2011-03-30

add ORIGINATING_SERVER_ID and ORIGINATING_COMMIT_ID field

=== modified file 'drizzled/execute.cc'
--- drizzled/execute.cc	2011-02-18 16:08:59 +0000
+++ drizzled/execute.cc	2011-04-01 19:01:31 +0000
@@ -67,6 +67,8 @@
     // session. Eventually we will allow someone to change the effective
     // user.
     new_session->user()= _session.user();
+    new_session->setOriginatingServerID(_session.getOriginatingServerID());
+    new_session->setOriginatingCommitID(_session.getOriginatingCommitID());
     
     if (Session::schedule(new_session))
     {
 
=== modified file 'drizzled/session.cc'
--- drizzled/session.cc	2011-03-28 02:46:21 +0000
+++ drizzled/session.cc	2011-04-01 19:01:31 +0000
@@ -223,6 +223,7 @@
   tablespace_op(false),
   use_usage(false),
   security_ctx(identifier::User::make_shared()),
+  originating_server_id_set(false),
   client(client_arg)
 {
   client->setSession(this);
@@ -238,6 +239,8 @@
   lex().current_select= 0;
   memset(&variables, 0, sizeof(variables));
   scoreboard_index= -1;
+  originating_server_id= 0;
+  originating_commit_id= 0;
   cleanup_done= abort_on_warning= no_warnings_for_error= false;
 
   /* query_cache init */
 
=== modified file 'drizzled/session.h'
--- drizzled/session.h	2011-03-28 14:49:14 +0000
+++ drizzled/session.h	2011-04-01 19:01:31 +0000
@@ -293,6 +293,32 @@
     scoreboard_index= in_scoreboard_index;
   }
 
+  bool isOriginatingServerIDSet()
+  {
+    return originating_server_id_set;
+  }
+
+  void setOriginatingServerID(uint32_t in_originating_server_id)
+  {
+    originating_server_id= in_originating_server_id;
+    originating_server_id_set= true;
+  }
+
+  uint32_t getOriginatingServerID()
+  {
+    return originating_server_id;
+  }
+
+  void setOriginatingCommitID(uint64_t in_originating_commit_id)
+  {
+    originating_commit_id= in_originating_commit_id;
+  }
+
+  uint64_t getOriginatingCommitID()
+  {
+    return originating_commit_id;
+  }
+
   /**
    * Is this session viewable by the current user?
    */
@@ -1370,7 +1396,10 @@
   bool tablespace_op; /**< This is true in DISCARD/IMPORT TABLESPACE */
   bool use_usage;
   rusage usage;
-  identifier::user::mptr security_ctx;
+  identifier::user::mptr security_ctx; 
+  bool originating_server_id_set;
+  uint32_t originating_server_id;
+  uint64_t originating_commit_id;
   int32_t scoreboard_index;
   plugin::Client *client;
   util::string::shared_ptr _schema;
 
=== modified file 'plugin/innobase/dict/create_replication.cc'
--- plugin/innobase/dict/create_replication.cc	2011-03-22 23:52:29 +0000
+++ plugin/innobase/dict/create_replication.cc	2011-04-01 19:01:31 +0000
@@ -76,7 +76,7 @@
   error = que_eval_sql(info,
                        "PROCEDURE CREATE_SYS_REPLICATION_LOG_PROC () IS\n"
                        "BEGIN\n"
-                       "CREATE TABLE SYS_REPLICATION_LOG(ID INT(8), SEGID INT, COMMIT_ID INT(8), END_TIMESTAMP INT(8), MESSAGE_LEN INT, MESSAGE BLOB);\n" 
+                       "CREATE TABLE SYS_REPLICATION_LOG(ID INT(8), SEGID INT, COMMIT_ID INT(8), END_TIMESTAMP INT(8), ORIGINATING_SERVER_ID INT, ORIGINATING_COMMIT_ID INT(8), MESSAGE_LEN INT, MESSAGE BLOB);\n" 
                        "CREATE UNIQUE CLUSTERED INDEX PRIMARY ON SYS_REPLICATION_LOG (ID, SEGID);\n"
                        "CREATE INDEX COMMIT_IDX ON SYS_REPLICATION_LOG (COMMIT_ID, ID);\n"
                        "END;\n"
@@ -147,6 +147,14 @@
   field->set_type(drizzled::message::Table::Field::BIGINT);
 
   field= table_message->add_field();
+  field->set_name("ORIGINATING_SERVER_ID");
+  field->set_type(drizzled::message::Table::Field::INTEGER);
+
+  field= table_message->add_field();
+  field->set_name("ORIGINATING_COMMIT_ID");
+  field->set_type(drizzled::message::Table::Field::BIGINT);
+
+  field= table_message->add_field();
   field->set_name("MESSAGE_LEN");
   field->set_type(drizzled::message::Table::Field::INTEGER);
 
@@ -191,7 +199,10 @@
 ulint insert_replication_message(const char *message, size_t size, 
                                  trx_t *trx, uint64_t trx_id, 
                                  uint64_t end_timestamp, bool is_end_segment, 
-                                 uint32_t seg_id) 
+                                 uint32_t seg_id, uint32_t server_id,
+                                 bool use_originating_server_id,
+                                 uint32_t originating_server_id,
+                                 uint64_t originating_commit_id) 
 {
   ulint error;
   row_prebuilt_t*	prebuilt;	/* For reading rows */
@@ -251,12 +262,30 @@
   row_mysql_store_col_in_innobase_format(dfield, data, TRUE, (byte*)&end_timestamp, 8, dict_table_is_comp(prebuilt->table));
   dfield_set_data(dfield, data, 8);
 
+  if (not use_originating_server_id)
+  {
+    /* This transaction originated from this server, rather then being
+       replicated to this server reset the values to reflect that */
+    originating_server_id= server_id;
+    originating_commit_id= commit_id;    
+  }
+
   dfield = dtuple_get_nth_field(dtuple, 4);
   data= static_cast<byte*>(mem_heap_alloc(prebuilt->heap, 4));
+  row_mysql_store_col_in_innobase_format(dfield, data, TRUE, (byte*)&originating_server_id, 4, dict_table_is_comp(prebuilt->table));
+  dfield_set_data(dfield, data, 4);
+
+  dfield = dtuple_get_nth_field(dtuple, 5);
+  data= static_cast<byte*>(mem_heap_alloc(prebuilt->heap, 8));
+  row_mysql_store_col_in_innobase_format(dfield, data, TRUE, (byte*)&originating_commit_id, 8, dict_table_is_comp(prebuilt->table));
+  dfield_set_data(dfield, data, 8);
+
+  dfield = dtuple_get_nth_field(dtuple, 6);
+  data= static_cast<byte*>(mem_heap_alloc(prebuilt->heap, 4));
   row_mysql_store_col_in_innobase_format(dfield, data, TRUE, (byte*)&size, 4, dict_table_is_comp(prebuilt->table));
   dfield_set_data(dfield, data, 4);
 
-  dfield = dtuple_get_nth_field(dtuple, 5);
+  dfield = dtuple_get_nth_field(dtuple, 7);
   dfield_set_data(dfield, message, size);
 
   ins_node_t*	node		= prebuilt->ins_node;
@@ -355,8 +384,18 @@
     convert_to_mysql_format(timestampbyte, field, 8);
     ret.end_timestamp= *(uint64_t *)timestampbyte;
 
+    field = rec_get_nth_field_old(rec, 6, &len);
+    byte serverbyte[4];
+    convert_to_mysql_format(serverbyte, field, 4);
+    ret.originating_server_id= *(uint32_t *)serverbyte;
+
+    field = rec_get_nth_field_old(rec, 7, &len);
+    byte originatingcommitbyte[8];
+    convert_to_mysql_format(originatingcommitbyte, field, 8);
+    ret.originating_commit_id= *(uint64_t *)originatingcommitbyte;
+
     // Handler message
-    field = rec_get_nth_field_old(rec, 7, &len);
+    field = rec_get_nth_field_old(rec, 9, &len);
     ret.message= (char *)field;
     ret.message_length= len;
 
 
=== modified file 'plugin/innobase/handler/ha_innodb.cc'
--- plugin/innobase/handler/ha_innodb.cc	2011-03-29 21:04:17 +0000
+++ plugin/innobase/handler/ha_innodb.cc	2011-04-01 19:01:31 +0000
@@ -966,8 +966,15 @@
   uint64_t end_timestamp= message.transaction_context().end_timestamp();
   bool is_end_segment= message.end_segment();
   trx->log_commit_id= TRUE;
+  uint32_t server_id= message.transaction_context().server_id();
+  uint32_t originating_server_id= session.getOriginatingServerID();
+  uint64_t originating_commit_id= session.getOriginatingCommitID();
+  bool use_originating_server_id= session.isOriginatingServerIDSet();
+
   ulint error= insert_replication_message(data, message.ByteSize(), trx, trx_id,
-               end_timestamp, is_end_segment, seg_id);
+               end_timestamp, is_end_segment, seg_id, server_id,
+               use_originating_server_id, originating_server_id, 
+               originating_commit_id);
   (void)error;
 
   delete[] data;
 
=== modified file 'plugin/innobase/handler/replication_dictionary.cc'
--- plugin/innobase/handler/replication_dictionary.cc	2011-02-17 00:14:13 +0000
+++ plugin/innobase/handler/replication_dictionary.cc	2011-04-01 19:01:31 +0000
@@ -79,6 +79,8 @@
   add_field("TRANSACTION_SEGMENT_ID", plugin::TableFunction::NUMBER, 0, false);
   add_field("COMMIT_ID", plugin::TableFunction::NUMBER, 0, false);
   add_field("END_TIMESTAMP", plugin::TableFunction::NUMBER, 0, false);
+  add_field("ORIGINATING_SERVER_ID", plugin::TableFunction::NUMBER, 0, false);
+  add_field("ORIGINATING_COMMIT_ID", plugin::TableFunction::NUMBER, 0, false);
   add_field("TRANSACTION_MESSAGE_STRING", plugin::TableFunction::STRING, transaction_message_threshold, false);
   add_field("TRANSACTION_LENGTH", plugin::TableFunction::NUMBER, 0, false);
 }
@@ -110,7 +112,11 @@
   push(static_cast<uint64_t>(ret.commit_id));
 
   push(static_cast<uint64_t>(ret.end_timestamp));
-  
+ 
+  push(static_cast<uint64_t>(ret.originating_server_id));
+
+  push(static_cast<uint64_t>(ret.originating_commit_id));
+ 
   /* Message in viewable format */
   bool result= message.ParseFromArray(ret.message, ret.message_length);
 
 
=== modified file 'plugin/innobase/include/create_replication.h'
--- plugin/innobase/include/create_replication.h	2011-03-14 05:40:28 +0000
+++ plugin/innobase/include/create_replication.h	2011-04-01 19:01:31 +0000
@@ -45,10 +45,13 @@
 
 UNIV_INTERN ulint dict_create_sys_replication_log(void);
 
-UNIV_INTERN ulint insert_replication_message(const char *message, size_t size, 
-                                             trx_t *trx, uint64_t trx_id, 
+UNIV_INTERN ulint insert_replication_message(const char *message, size_t size,
+                                             trx_t *trx, uint64_t trx_id,
                                              uint64_t end_timestamp, bool is_end_segment,
-                                             uint32_t seg_id);
+                                             uint32_t seg_id, uint32_t server_id,
+                                             bool use_originating_server_id,
+                                             uint32_t originating_server_id,
+                                             uint64_t originating_commit_id);
 
 UNIV_INTERN struct read_replication_state_st *replication_read_init(void);
 UNIV_INTERN void replication_read_deinit(struct read_replication_state_st *);
 
=== modified file 'plugin/innobase/include/read_replication.h'
--- plugin/innobase/include/read_replication.h	2011-03-14 05:40:28 +0000
+++ plugin/innobase/include/read_replication.h	2011-04-01 19:01:31 +0000
@@ -27,6 +27,8 @@
   unsigned long long commit_id;
   unsigned long long end_timestamp;
   unsigned long seg_id;
+  unsigned long originating_server_id;
+  unsigned long long originating_commit_id;
   unsigned long long message_length;
   const char *message;
 };
 
=== modified file 'plugin/innobase/tests/r/innodb-system-table-view.result'
--- plugin/innobase/tests/r/innodb-system-table-view.result	2011-02-10 18:19:37 +0000
+++ plugin/innobase/tests/r/innodb-system-table-view.result	2011-04-01 19:01:31 +0000
@@ -2,7 +2,7 @@
 TABLE_ID	NAME	FLAG	N_COLS	SPACE
 11	SYS_FOREIGN	0	7	0
 12	SYS_FOREIGN_COLS	0	7	0
-13	SYS_REPLICATION_LOG	0	9	0
+13	SYS_REPLICATION_LOG	0	11	0
 SELECT * FROM DATA_DICTIONARY.INNODB_SYS_INDEXES;
 INDEX_ID	NAME	TABLE_ID	TYPE	N_FIELDS	PAGE_NO	SPACE
 11	ID_IND	11	3	1	302	0
@@ -25,8 +25,10 @@
 13	SEGID	1	6	0	4
 13	COMMIT_ID	2	6	0	8
 13	END_TIMESTAMP	3	6	0	8
-13	MESSAGE_LEN	4	6	0	4
-13	MESSAGE	5	5	4129792	0
+13	ORIGINATING_SERVER_ID	4	6	0	4
+13	ORIGINATING_COMMIT_ID	5	6	0	8
+13	MESSAGE_LEN	6	6	0	4
+13	MESSAGE	7	5	4129792	0
 SELECT * FROM DATA_DICTIONARY.INNODB_SYS_FIELDS;
 INDEX_ID	NAME	POS
 11	ID	0
@@ -70,7 +72,7 @@
 NAME	FLAG	N_COLS	SPACE
 SYS_FOREIGN	0	7	0
 SYS_FOREIGN_COLS	0	7	0
-SYS_REPLICATION_LOG	0	9	0
+SYS_REPLICATION_LOG	0	11	0
 test/child	1	5	0
 test/parent	1	4	0
 SELECT name, n_fields
 
=== modified file 'plugin/innobase/tests/r/innodb_replication_log.result'
--- plugin/innobase/tests/r/innodb_replication_log.result	2011-02-21 17:37:00 +0000
+++ plugin/innobase/tests/r/innodb_replication_log.result	2011-04-01 19:01:31 +0000
@@ -10,6 +10,8 @@
   `TRANSACTION_SEGMENT_ID` BIGINT NOT NULL,
   `COMMIT_ID` BIGINT NOT NULL,
   `END_TIMESTAMP` BIGINT NOT NULL,
+  `ORIGINATING_SERVER_ID` BIGINT NOT NULL,
+  `ORIGINATING_COMMIT_ID` BIGINT NOT NULL,
   `TRANSACTION_MESSAGE_STRING` TEXT COLLATE utf8_general_ci NOT NULL,
   `TRANSACTION_LENGTH` BIGINT NOT NULL
 ) ENGINE=FunctionEngine COLLATE = utf8_general_ci REPLICATE = FALSE
 
=== modified file 'plugin/schema_dictionary/tests/r/data_dictionary.result'
--- plugin/schema_dictionary/tests/r/data_dictionary.result	2011-03-18 16:53:27 +0000
+++ plugin/schema_dictionary/tests/r/data_dictionary.result	2011-04-01 19:01:31 +0000
@@ -1,7 +1,7 @@
 use data_dictionary;
 SELECT count(*) FROM columns;
 count(*)
-580
+584
 SELECT count(*) FROM indexes;
 count(*)
 2
@@ -356,6 +356,10 @@
 ORDINAL_POSITION
 ORDINAL_POSITION
 ORDINAL_POSITION
+ORIGINATING_COMMIT_ID
+ORIGINATING_COMMIT_ID
+ORIGINATING_SERVER_ID
+ORIGINATING_SERVER_ID
 OTHER_INDEX_SIZE
 PAGES_FREE
 PAGES_FREE
@@ -869,6 +873,8 @@
 DATA_DICTIONARY	INNODB_LOCK_WAITS	REQUESTING_TRX_ID
 DATA_DICTIONARY	INNODB_REPLICATION_LOG	COMMIT_ID
 DATA_DICTIONARY	INNODB_REPLICATION_LOG	END_TIMESTAMP
+DATA_DICTIONARY	INNODB_REPLICATION_LOG	ORIGINATING_COMMIT_ID
+DATA_DICTIONARY	INNODB_REPLICATION_LOG	ORIGINATING_SERVER_ID
 DATA_DICTIONARY	INNODB_REPLICATION_LOG	TRANSACTION_ID
 DATA_DICTIONARY	INNODB_REPLICATION_LOG	TRANSACTION_LENGTH
 DATA_DICTIONARY	INNODB_REPLICATION_LOG	TRANSACTION_MESSAGE_STRING
@@ -995,6 +1001,8 @@
 DATA_DICTIONARY	SYS_REPLICATION_LOG	ID
 DATA_DICTIONARY	SYS_REPLICATION_LOG	MESSAGE
 DATA_DICTIONARY	SYS_REPLICATION_LOG	MESSAGE_LEN
+DATA_DICTIONARY	SYS_REPLICATION_LOG	ORIGINATING_COMMIT_ID
+DATA_DICTIONARY	SYS_REPLICATION_LOG	ORIGINATING_SERVER_ID
 DATA_DICTIONARY	SYS_REPLICATION_LOG	SEGID
 DATA_DICTIONARY	TABLES	AUTO_INCREMENT
 DATA_DICTIONARY	TABLES	ENGINE
@@ -1532,6 +1540,8 @@
 INNODB_LOCK_WAITS	DATA_DICTIONARY	REQUESTING_TRX_ID
 INNODB_REPLICATION_LOG	DATA_DICTIONARY	COMMIT_ID
 INNODB_REPLICATION_LOG	DATA_DICTIONARY	END_TIMESTAMP
+INNODB_REPLICATION_LOG	DATA_DICTIONARY	ORIGINATING_COMMIT_ID
+INNODB_REPLICATION_LOG	DATA_DICTIONARY	ORIGINATING_SERVER_ID
 INNODB_REPLICATION_LOG	DATA_DICTIONARY	TRANSACTION_ID
 INNODB_REPLICATION_LOG	DATA_DICTIONARY	TRANSACTION_LENGTH
 INNODB_REPLICATION_LOG	DATA_DICTIONARY	TRANSACTION_MESSAGE_STRING
@@ -1787,6 +1797,8 @@
 SYS_REPLICATION_LOG	DATA_DICTIONARY	ID
 SYS_REPLICATION_LOG	DATA_DICTIONARY	MESSAGE
 SYS_REPLICATION_LOG	DATA_DICTIONARY	MESSAGE_LEN
+SYS_REPLICATION_LOG	DATA_DICTIONARY	ORIGINATING_COMMIT_ID
+SYS_REPLICATION_LOG	DATA_DICTIONARY	ORIGINATING_SERVER_ID
 SYS_REPLICATION_LOG	DATA_DICTIONARY	SEGID
 TABLES	DATA_DICTIONARY	AUTO_INCREMENT
 TABLES	DATA_DICTIONARY	ENGINE
@@ -2115,6 +2127,8 @@
 INNODB_LOCK_WAITS	DATA_DICTIONARY	REQUESTING_TRX_ID
 INNODB_REPLICATION_LOG	DATA_DICTIONARY	COMMIT_ID
 INNODB_REPLICATION_LOG	DATA_DICTIONARY	END_TIMESTAMP
+INNODB_REPLICATION_LOG	DATA_DICTIONARY	ORIGINATING_COMMIT_ID
+INNODB_REPLICATION_LOG	DATA_DICTIONARY	ORIGINATING_SERVER_ID
 INNODB_REPLICATION_LOG	DATA_DICTIONARY	TRANSACTION_ID
 INNODB_REPLICATION_LOG	DATA_DICTIONARY	TRANSACTION_LENGTH
 INNODB_REPLICATION_LOG	DATA_DICTIONARY	TRANSACTION_MESSAGE_STRING
@@ -2370,6 +2384,8 @@
 SYS_REPLICATION_LOG	DATA_DICTIONARY	ID
 SYS_REPLICATION_LOG	DATA_DICTIONARY	MESSAGE
 SYS_REPLICATION_LOG	DATA_DICTIONARY	MESSAGE_LEN
+SYS_REPLICATION_LOG	DATA_DICTIONARY	ORIGINATING_COMMIT_ID
+SYS_REPLICATION_LOG	DATA_DICTIONARY	ORIGINATING_SERVER_ID
 SYS_REPLICATION_LOG	DATA_DICTIONARY	SEGID
 TABLES	DATA_DICTIONARY	AUTO_INCREMENT
 TABLES	DATA_DICTIONARY	ENGINE
 
=== modified file 'plugin/slave/queue_consumer.cc'
--- plugin/slave/queue_consumer.cc	2011-03-19 20:20:32 +0000
+++ plugin/slave/queue_consumer.cc	2011-04-01 19:01:31 +0000
@@ -75,6 +75,8 @@
   for (size_t x= 0; x < completedTransactionIds.size(); x++)
   {
     string commit_id;
+    uint32_t originating_server_id= 0;
+    uint64_t originating_commit_id= 0;
     uint64_t trx_id= completedTransactionIds[x];
 
     vector<string> aggregate_sql;  /* final SQL to execute */
@@ -83,7 +85,8 @@
     message::Transaction transaction;
     uint32_t segment_id= 1;
 
-    while (getMessage(transaction, commit_id, master_id, trx_id, segment_id++))
+    while (getMessage(transaction, commit_id, master_id, trx_id, originating_server_id, 
+                      originating_commit_id, segment_id++))
     {
       convertToSQL(transaction, aggregate_sql, segmented_sql);
       transaction.Clear();
@@ -132,7 +135,8 @@
       }
     }
 
-    if (not executeSQLWithCommitId(aggregate_sql, commit_id, master_id))
+    if (not executeSQLWithCommitId(aggregate_sql, commit_id, master_id, 
+                                   originating_server_id, originating_commit_id))
     {
       if (_ignore_errors)
       {
@@ -143,6 +147,10 @@
         string tmp("UPDATE `sys_replication`.`applier_state`"
                    " SET `last_applied_commit_id` = ");
         tmp.append(commit_id);
+        tmp.append(", `originating_server_id` = ");
+        tmp.append(boost::lexical_cast<string>(originating_server_id));
+        tmp.append(", `originating_commit_id` = ");
+        tmp.append(boost::lexical_cast<string>(originating_commit_id));
         tmp.append(" WHERE `master_id` = ");
         tmp.append(master_id);
         sql.push_back(tmp);
@@ -168,9 +176,12 @@
                               string &commit_id,
                               const string &master_id,
                               uint64_t trx_id,
+                              uint32_t &originating_server_id,
+                              uint64_t &originating_commit_id,
                               uint32_t segment_id)
 {
-  string sql("SELECT `msg`, `commit_order` FROM `sys_replication`.`queue`"
+  string sql("SELECT `msg`, `commit_order`, `originating_server_id`, "
+             " `originating_commit_id` FROM `sys_replication`.`queue`"
              " WHERE `trx_id` = ");
   sql.append(boost::lexical_cast<string>(trx_id));
   sql.append(" AND `seg_id` = ", 16);
@@ -178,12 +189,12 @@
   sql.append(" AND `master_id` = ", 19),
   sql.append(master_id);
 
-  sql::ResultSet result_set(2);
+  sql::ResultSet result_set(4);
   Execute execute(*(_session.get()), true);
   
   execute.run(sql, result_set);
   
-  assert(result_set.getMetaData().getColumnCount() == 2);
+  assert(result_set.getMetaData().getColumnCount() == 4);
 
   /* Really should only be 1 returned row */
   uint32_t found_rows= 0;
@@ -191,6 +202,8 @@
   {
     string msg= result_set.getString(0);
     string com_id= result_set.getString(1);
+    string orig_server_id= result_set.getString(2);
+    string orig_commit_id= result_set.getString(3);
 
     if ((msg == "") || (found_rows == 1))
       break;
@@ -198,10 +211,14 @@
     /* Neither column should be NULL */
     assert(result_set.isNull(0) == false);
     assert(result_set.isNull(1) == false);
+    assert(result_set.isNull(2) == false);
+    assert(result_set.isNull(3) == false);
 
     google::protobuf::TextFormat::ParseFromString(msg, &transaction);
 
     commit_id= com_id;
+    originating_server_id= boost::lexical_cast<uint32_t>(orig_server_id);
+    originating_commit_id= boost::lexical_cast<uint64_t>(orig_commit_id);
     found_rows++;
   }
 
@@ -404,15 +421,24 @@
 
 bool QueueConsumer::executeSQLWithCommitId(vector<string> &sql,
                                            const string &commit_id,
-                                           const string &master_id)
+                                           const string &master_id,
+                                           uint32_t originating_server_id,
+                                           uint64_t originating_commit_id)
 {
   string tmp("UPDATE `sys_replication`.`applier_state`"
              " SET `last_applied_commit_id` = ");
   tmp.append(commit_id);
+  tmp.append(", `originating_server_id` = ");
+  tmp.append(boost::lexical_cast<string>(originating_server_id));
+  tmp.append(", `originating_commit_id` = ");
+  tmp.append(boost::lexical_cast<string>(originating_commit_id));
   tmp.append(" WHERE `master_id` = ");
   tmp.append(master_id);
   sql.push_back(tmp);
-  
+ 
+  _session->setOriginatingServerID(originating_server_id);
+  _session->setOriginatingCommitID(originating_commit_id);
+ 
   return executeSQL(sql);
 }
 
 
=== modified file 'plugin/slave/queue_consumer.h'
--- plugin/slave/queue_consumer.h	2011-03-19 20:20:32 +0000
+++ plugin/slave/queue_consumer.h	2011-04-01 19:01:31 +0000
@@ -111,6 +111,8 @@
                   std::string &commit_id,
                   const std::string &master_id,
                   uint64_t trx_id,
+                  uint32_t &originating_server_id,
+                  uint64_t &originating_commit_id,
                   uint32_t segment_id);
 
   /**
@@ -133,14 +135,19 @@
    * @param sql Batch of SQL statements to execute.
    * @param commit_id Commit ID value to store in state table.
    * @param master_id ID of the master this SQL came from.
+   * @param originating_server_id Server ID of the master where 
+   *   this SQL was originally applied.
+   * @param originating_commit_id Commit ID of the master where
+   *   this SQL was originally applied.
    *
    * @retval true Success
    * @retval false Failure
    */
   bool executeSQLWithCommitId(std::vector<std::string> &sql,
                               const std::string &commit_id,
-                              const std::string &master_id);
-  
+                              const std::string &master_id,
+                              uint32_t originating_server_id, 
+                              uint64_t originating_commit_id);
   /**
    * Remove messages for a given transaction from the queue.
    *
 
=== modified file 'plugin/slave/queue_producer.cc'
--- plugin/slave/queue_producer.cc	2011-03-19 20:20:32 +0000
+++ plugin/slave/queue_producer.cc	2011-04-01 19:01:31 +0000
@@ -296,6 +296,8 @@
 bool QueueProducer::queueInsert(const char *trx_id,
                                 const char *seg_id,
                                 const char *commit_id,
+                                const char *originating_server_id,
+                                const char *originating_commit_id,
                                 const char *msg,
                                 const char *msg_length)
 {
@@ -307,7 +309,8 @@
    * The SQL to insert our results into the local queue.
    */
   string sql= "INSERT INTO `sys_replication`.`queue`"
-              " (`master_id`, `trx_id`, `seg_id`, `commit_order`, `msg`) VALUES (";
+              " (`master_id`, `trx_id`, `seg_id`, `commit_order`,"
+              "  `originating_server_id`, `originating_commit_id`, `msg`) VALUES (";
   sql.append(boost::lexical_cast<string>(masterId()));
   sql.append(", ", 2);
   sql.append(trx_id);
@@ -315,6 +318,10 @@
   sql.append(seg_id);
   sql.append(", ", 2);
   sql.append(commit_id);
+  sql.append(", ", 2);
+  sql.append(originating_server_id);
+  sql.append(", ", 2);
+  sql.append(originating_commit_id);
   sql.append(", '", 3);
 
   /*
@@ -395,7 +402,8 @@
   /*
    * The SQL to pull everything we need from the master.
    */
-  string sql= "SELECT `id`, `segid`, `commit_id`, `message`, `message_len` "
+  string sql= "SELECT `id`, `segid`, `commit_id`, `originating_server_id`,"
+              " `originating_commit_id`, `message`, `message_len` "
               " FROM `data_dictionary`.`sys_replication_log` WHERE `id` IN (";
 
   for (size_t x= 0; x < trx_id_list.size(); x++)
@@ -439,7 +447,7 @@
 
   while ((row= drizzle_row_next(&result)) != NULL)
   {
-    if (not queueInsert(row[0], row[1], row[2], row[3], row[4]))
+    if (not queueInsert(row[0], row[1], row[2], row[3], row[4], row[5], row[6]))
     {
       errmsg_printf(error::ERROR,
                     _("Replication slave: Unable to insert into queue."));
 
=== modified file 'plugin/slave/queue_producer.h'
--- plugin/slave/queue_producer.h	2011-03-19 20:20:32 +0000
+++ plugin/slave/queue_producer.h	2011-04-01 19:01:31 +0000
@@ -184,6 +184,8 @@
   bool queueInsert(const char *trx_id,
                    const char *seg_id,
                    const char *commit_id,
+                   const char *originating_server_id,
+                   const char *originating_commit_id,
                    const char *msg,
                    const char *msg_length);
 
 
=== modified file 'plugin/slave/replication_schema.cc'
--- plugin/slave/replication_schema.cc	2011-03-19 20:20:32 +0000
+++ plugin/slave/replication_schema.cc	2011-04-01 19:01:31 +0000
@@ -73,6 +73,7 @@
    * Table: applier_state
    * Version 1.0: Initial definition
    * Version 1.1: Added master_id and changed PK to master_id
+   * Version 1.2: Added originating_server_id and originating_commit_id
    */
 
   sql.clear();
@@ -80,10 +81,12 @@
   sql.push_back("CREATE TABLE IF NOT EXISTS `sys_replication`.`applier_state` ("
                 " `master_id` BIGINT NOT NULL,"
                 " `last_applied_commit_id` BIGINT NOT NULL,"
+                " `originating_server_id` BIGINT NOT NULL,"
+                " `originating_commit_id` BIGINT NOT NULL,"
                 " `status` VARCHAR(20) NOT NULL,"
                 " `error_msg` VARCHAR(250),"
                 " PRIMARY KEY(`master_id`))"
-                " COMMENT = 'VERSION 1.1'");
+                " COMMENT = 'VERSION 1.2'");
 
   if (not executeSQL(sql))
     return false;
@@ -101,6 +104,8 @@
                 " `trx_id` BIGINT NOT NULL,"
                 " `seg_id` INT NOT NULL,"
                 " `commit_order` BIGINT,"
+                " `originating_server_id` INT NOT NULL,"
+                " `originating_commit_id` BIGINT NOT NULL,"
                 " `msg` BLOB,"
                 " PRIMARY KEY(`master_id`, `trx_id`, `seg_id`))"
                 " COMMENT = 'VERSION 1.1'");
@@ -151,9 +156,11 @@
   {
     sql.clear();
     sql.push_back("INSERT INTO `sys_replication`.`applier_state`"
-                  " (`master_id`, `last_applied_commit_id`, `status`) VALUES ("
+                  " (`master_id`, `last_applied_commit_id`," 
+                  "  `originating_server_id`, `originating_commit_id`,"
+                  "  `status`) VALUES ("
                   + boost::lexical_cast<string>(master_id)
-                  + ",0 , 'STOPPED')");
+                  + ",0 , 0, 0, 'STOPPED')");
     if (not executeSQL(sql))
       return false;
   }