Merge lp:~dshrews/drizzle/bugs_626566_637590 into lp:~drizzle-trunk/drizzle/development

[David Shrewsbury]

The segment_id of Statement messages wasn't remaining set to its proper value when a Transaction message had to be split up due to exceed message size. This broke the rollback code since it could not identify bulk load messages.

Also fixed calculating the current message size by using the Transaction message rather than its embedded Statement messages.

Also also, during a ROLLBACK, the transaction id was improperly incremented.

[Brian Aker (brianaker)]

Anyway to add a test case for the rollback?

review: Needs Information

[David Shrewsbury (dshrews)]

Good question Brian. Since this is testing the GPB messages themselves, we don't yet have a good way to test that part. But I'm working with Patrick to implement a plan to do just that. So this is something that is coming in the near future.

=== modified file 'drizzled/transaction_services.cc'
--- drizzled/transaction_services.cc	2010-09-11 20:23:04 +0000
+++ drizzled/transaction_services.cc	2010-09-14 00:21:01 +0000
@@ -1072,7 +1072,7 @@
      * attach it to the transaction, and push it to replicators.
      */
     transaction->Clear();
-    initTransactionMessage(*transaction, in_session, true);
+    initTransactionMessage(*transaction, in_session, false);
 
     message::Statement *statement= transaction->add_statement();
 
@@ -1107,12 +1107,14 @@
   } 
   else if (statement != NULL)
   {
+    transaction= getActiveTransactionMessage(in_session);
+
     /*
      * If we've passed our threshold for the statement size (possible for
      * a bulk insert), we'll finalize the Statement and Transaction (doing
      * the Transaction will keep it from getting huge).
      */
-    if (static_cast<size_t>(statement->ByteSize()) >= trx_msg_threshold)
+    if (static_cast<size_t>(transaction->ByteSize()) >= trx_msg_threshold)
     {
       message::InsertData *current_data= statement->mutable_insert_data();
 
@@ -1143,11 +1145,18 @@
      
       string current_table_name;
       (void) in_table->getShare()->getTableName(current_table_name);
+
       if (current_table_name.compare(old_table_name))
       {
         finalizeStatementMessage(*statement, in_session);
         statement= in_session->getStatementMessage();
       }
+      else
+      {
+        /* carry forward the existing segment id */
+        const message::InsertData &current_data= statement->insert_data();
+        *next_segment_id= current_data.segment_id();
+      }
     }
   } 
 
@@ -1288,12 +1297,14 @@
   }
   else if (statement != NULL)
   {
+    transaction= getActiveTransactionMessage(in_session);
+
     /*
      * If we've passed our threshold for the statement size (possible for
      * a bulk insert), we'll finalize the Statement and Transaction (doing
      * the Transaction will keep it from getting huge).
      */
-    if (static_cast<size_t>(statement->ByteSize()) >= trx_msg_threshold)
+    if (static_cast<size_t>(transaction->ByteSize()) >= trx_msg_threshold)
     {
       message::UpdateData *current_data= statement->mutable_update_data();
 
@@ -1329,6 +1340,12 @@
         finalizeStatementMessage(*statement, in_session);
         statement= in_session->getStatementMessage();
       }
+      else
+      {
+        /* carry forward the existing segment id */
+        const message::UpdateData &current_data= statement->update_data();
+        *next_segment_id= current_data.segment_id();
+      }
     }
   }
 
@@ -1535,12 +1552,14 @@
   }
   else if (statement != NULL)
   {
+    transaction= getActiveTransactionMessage(in_session);
+
     /*
      * If we've passed our threshold for the statement size (possible for
      * a bulk insert), we'll finalize the Statement and Transaction (doing
      * the Transaction will keep it from getting huge).
      */
-    if (static_cast<size_t>(statement->ByteSize()) >= trx_msg_threshold)
+    if (static_cast<size_t>(transaction->ByteSize()) >= trx_msg_threshold)
     {
       message::DeleteData *current_data= statement->mutable_delete_data();
 
@@ -1576,6 +1595,12 @@
         finalizeStatementMessage(*statement, in_session);
         statement= in_session->getStatementMessage();
       }
+      else
+      {
+        /* carry forward the existing segment id */
+        const message::DeleteData &current_data= statement->delete_data();
+        *next_segment_id= current_data.segment_id();
+      }
     }
   }