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

[David Shrewsbury]

This enables logging of the INSERTs produced by LOAD DATA.

As a consequence, I had to introduce a transaction message size threshold, because otherwise, we would have a large Transaction object for LOAD DATA that would contain all of the INSERTs for a single LOAD DATA statement. That could use a lot of memory and create a large GPB message, which has a maximum of 64M. (Right now, the transaction message size threshold is hard coded to 1M. I'll make this configurable in the near future once our sys var changes are merged into trunk.)

This means that we may have multiple Transactions in the replication stream for a single transaction, but they will all have the same transaction id, and all but the last Transaction will have the Statement end_segment attribute set to FALSE.

So a reader of the stream (or transaction log) will know to COMMIT when either:

a) The transaction id changes
b) Or, the Statement(s) within the Transaction have end_segment = TRUE

[lp:~dshrews/drizzle/bug600795]

1722. By David Shrewsbury on 2010-08-28

Merge from trunk, resolve conflict with transaction_services.h

=== modified file 'drizzled/cursor.cc'
--- drizzled/cursor.cc	2010-08-26 17:42:07 +0000
+++ drizzled/cursor.cc	2010-08-28 01:13:43 +0000
@@ -1363,6 +1363,7 @@
     break;
   case SQLCOM_INSERT:
   case SQLCOM_INSERT_SELECT:
+  case SQLCOM_LOAD:
     /*
      * The else block below represents an 
      * INSERT ... ON DUPLICATE KEY UPDATE that

=== modified file 'drizzled/message/transaction_reader.cc'
--- drizzled/message/transaction_reader.cc	2010-07-17 06:28:29 +0000
+++ drizzled/message/transaction_reader.cc	2010-08-28 01:13:43 +0000
@@ -95,20 +95,74 @@
   }
 }
 
+static bool isEndStatement(const message::Statement &statement)
+{
+  switch (statement.type())
+  {
+    case (message::Statement::INSERT):
+    {
+      const message::InsertData &data= statement.insert_data();
+      if (not data.end_segment())
+        return false;
+      break;
+    }
+    case (message::Statement::UPDATE):
+    {
+      const message::UpdateData &data= statement.update_data();
+      if (not data.end_segment())
+        return false;
+      break;
+    }
+    case (message::Statement::DELETE):
+    {
+      const message::DeleteData &data= statement.delete_data();
+      if (not data.end_segment())
+        return false;
+      break;
+    }
+    default:
+      return true;
+  }
+  return true;
+}
+
 static void printTransaction(const message::Transaction &transaction)
 {
+  static uint64_t last_trx_id= 0;
+  bool should_commit= true;
   const message::TransactionContext trx= transaction.transaction_context();
 
   size_t num_statements= transaction.statement_size();
   size_t x;
 
-  cout << "START TRANSACTION;" << endl;
+  /*
+   * One way to determine when a new transaction begins is when the
+   * transaction id changes. We check that here.
+   */
+  if (trx.transaction_id() != last_trx_id)
+    cout << "START TRANSACTION;" << endl;
+
+  last_trx_id= trx.transaction_id();
+
   for (x= 0; x < num_statements; ++x)
   {
     const message::Statement &statement= transaction.statement(x);
+
+    if (should_commit)
+      should_commit= isEndStatement(statement);
+
     printStatement(statement);
   }
-  cout << "COMMIT;" << endl;
+
+  /*
+   * If ALL Statements are end segments, we can commit this Transaction.
+   * We can also check to see if the transaction_id changed, but this
+   * wouldn't work for the last Transaction in the transaction log since
+   * we don't have another Transaction to compare to. Checking for all
+   * end segments (like we do above) covers this case.
+   */
+  if (should_commit)
+    cout << "COMMIT;" << endl;
 }
 
 int main(int argc, char* argv[])

=== modified file 'drizzled/transaction_services.cc'
--- drizzled/transaction_services.cc	2010-08-26 15:33:12 +0000
+++ drizzled/transaction_services.cc	2010-08-28 01:13:43 +0000
@@ -80,6 +80,9 @@
 namespace drizzled
 {
 
+/** @TODO: Make this a system variable */
+static const size_t trx_msg_threshold= 1024 * 1024;
+
 /**
  * @defgroup Transactions
  *
@@ -895,7 +898,7 @@
   return replication_services.isActive();
 }
 
-message::Transaction *TransactionServices::getActiveTransactionMessage(Session *in_session)
+message::Transaction *TransactionServices::getActiveTransactionMessage(Session *in_session, bool should_inc_trx_id)
 {
   message::Transaction *transaction= in_session->getTransactionMessage();
 
@@ -907,7 +910,7 @@
      * deleting transaction message when done with it.
      */
     transaction= new (nothrow) message::Transaction();
-    initTransactionMessage(*transaction, in_session);
+    initTransactionMessage(*transaction, in_session, should_inc_trx_id);
     in_session->setTransactionMessage(transaction);
     return transaction;
   }
@@ -916,11 +919,17 @@
 }
 
 void TransactionServices::initTransactionMessage(message::Transaction &in_transaction,
-                                          Session *in_session)
+                                                 Session *in_session,
+                                                 bool should_inc_trx_id)
 {
   message::TransactionContext *trx= in_transaction.mutable_transaction_context();
   trx->set_server_id(in_session->getServerId());
-  trx->set_transaction_id(getNextTransactionId());
+
+  if (should_inc_trx_id)
+    trx->set_transaction_id(getNextTransactionId());
+  else
+    trx->set_transaction_id(getCurrentTransactionId());
+
   trx->set_start_timestamp(in_session->getCurrentTimestamp());
 }
 
@@ -1014,7 +1023,7 @@
      * attach it to the transaction, and push it to replicators.
      */
     transaction->Clear();
-    initTransactionMessage(*transaction, in_session);
+    initTransactionMessage(*transaction, in_session, true);
 
     message::Statement *statement= transaction->add_statement();
 
@@ -1029,9 +1038,11 @@
 }
 
 message::Statement &TransactionServices::getInsertStatement(Session *in_session,
-                                                                 Table *in_table)
+                                                            Table *in_table,
+                                                            uint32_t *next_segment_id)
 {
   message::Statement *statement= in_session->getStatementMessage();
+  message::Transaction *transaction= NULL;
 
   /* 
    * Check the type for the current Statement message, if it is anything
@@ -1047,21 +1058,59 @@
   } 
   else if (statement != NULL)
   {
-    const message::InsertHeader &insert_header= statement->insert_header();
-    string old_table_name= insert_header.table_metadata().table_name();
+    /*
+     * 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)
+    {
+      message::InsertData *current_data= statement->mutable_insert_data();
+
+      /* Caller should use this value when adding a new record */
+      *next_segment_id= current_data->segment_id() + 1;
+
+      current_data->set_end_segment(false);
+
+      /* 
+       * Send the trx message to replicators after finalizing the 
+       * statement and transaction. This will also set the Transaction
+       * and Statement objects in Session to NULL.
+       */
+      commitTransactionMessage(in_session);
+
+      /*
+       * Statement and Transaction should now be NULL, so new ones will get
+       * created. We reuse the transaction id since we are segmenting
+       * one transaction.
+       */
+      statement= in_session->getStatementMessage();
+      transaction= getActiveTransactionMessage(in_session, false);
+    }
+    else
+    {
+      const message::InsertHeader &insert_header= statement->insert_header();
+      string old_table_name= insert_header.table_metadata().table_name();
      
-    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();
+      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();
+      }
     }
   } 
 
   if (statement == NULL)
   {
-    message::Transaction *transaction= getActiveTransactionMessage(in_session);
+    /*
+     * Transaction will be non-NULL only if we had to segment it due to
+     * transaction size above.
+     */
+    if (transaction == NULL)
+      transaction= getActiveTransactionMessage(in_session);
+
     /* 
      * Transaction message initialized and set, but no statement created
      * yet.  We construct one and initialize it, here, then return the
@@ -1132,10 +1181,11 @@
     return true;
   }
 
-  message::Statement &statement= getInsertStatement(in_session, in_table);
+  uint32_t next_segment_id= 1;
+  message::Statement &statement= getInsertStatement(in_session, in_table, &next_segment_id);
 
   message::InsertData *data= statement.mutable_insert_data();
-  data->set_segment_id(1);
+  data->set_segment_id(next_segment_id);
   data->set_end_segment(true);
   message::InsertRecord *record= data->add_record();
 
@@ -1169,9 +1219,11 @@
 message::Statement &TransactionServices::getUpdateStatement(Session *in_session,
                                                             Table *in_table,
                                                             const unsigned char *old_record, 
-                                                            const unsigned char *new_record)
+                                                            const unsigned char *new_record,
+                                                            uint32_t *next_segment_id)
 {
   message::Statement *statement= in_session->getStatementMessage();
+  message::Transaction *transaction= NULL;
 
   /*
    * Check the type for the current Statement message, if it is anything
@@ -1187,21 +1239,59 @@
   }
   else if (statement != NULL)
   {
-    const message::UpdateHeader &update_header= statement->update_header();
-    string old_table_name= update_header.table_metadata().table_name();
-
-    string current_table_name;
-    (void) in_table->getShare()->getTableName(current_table_name);
-    if (current_table_name.compare(old_table_name))
+    /*
+     * 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)
     {
-      finalizeStatementMessage(*statement, in_session);
+      message::UpdateData *current_data= statement->mutable_update_data();
+
+      /* Caller should use this value when adding a new record */
+      *next_segment_id= current_data->segment_id() + 1;
+
+      current_data->set_end_segment(false);
+
+      /*
+       * Send the trx message to replicators after finalizing the 
+       * statement and transaction. This will also set the Transaction
+       * and Statement objects in Session to NULL.
+       */
+      commitTransactionMessage(in_session);
+
+      /*
+       * Statement and Transaction should now be NULL, so new ones will get
+       * created. We reuse the transaction id since we are segmenting
+       * one transaction.
+       */
       statement= in_session->getStatementMessage();
+      transaction= getActiveTransactionMessage(in_session, false);
+    }
+    else
+    {
+      const message::UpdateHeader &update_header= statement->update_header();
+      string old_table_name= update_header.table_metadata().table_name();
+
+      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();
+      }
     }
   }
 
   if (statement == NULL)
   {
-    message::Transaction *transaction= getActiveTransactionMessage(in_session);
+    /*
+     * Transaction will be non-NULL only if we had to segment it due to
+     * transaction size above.
+     */
+    if (transaction == NULL)
+      transaction= getActiveTransactionMessage(in_session);
+
     /* 
      * Transaction message initialized and set, but no statement created
      * yet.  We construct one and initialize it, here, then return the
@@ -1288,10 +1378,11 @@
   if (! replication_services.isActive())
     return;
 
-  message::Statement &statement= getUpdateStatement(in_session, in_table, old_record, new_record);
+  uint32_t next_segment_id= 1;
+  message::Statement &statement= getUpdateStatement(in_session, in_table, old_record, new_record, &next_segment_id);
 
   message::UpdateData *data= statement.mutable_update_data();
-  data->set_segment_id(1);
+  data->set_segment_id(next_segment_id);
   data->set_end_segment(true);
   message::UpdateRecord *record= data->add_record();
 
@@ -1375,9 +1466,11 @@
 }
 
 message::Statement &TransactionServices::getDeleteStatement(Session *in_session,
-                                                            Table *in_table)
+                                                            Table *in_table,
+                                                            uint32_t *next_segment_id)
 {
   message::Statement *statement= in_session->getStatementMessage();
+  message::Transaction *transaction= NULL;
 
   /*
    * Check the type for the current Statement message, if it is anything
@@ -1393,21 +1486,59 @@
   }
   else if (statement != NULL)
   {
-    const message::DeleteHeader &delete_header= statement->delete_header();
-    string old_table_name= delete_header.table_metadata().table_name();
-
-    string current_table_name;
-    (void) in_table->getShare()->getTableName(current_table_name);
-    if (current_table_name.compare(old_table_name))
+    /*
+     * 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)
     {
-      finalizeStatementMessage(*statement, in_session);
+      message::DeleteData *current_data= statement->mutable_delete_data();
+
+      /* Caller should use this value when adding a new record */
+      *next_segment_id= current_data->segment_id() + 1;
+
+      current_data->set_end_segment(false);
+
+      /* 
+       * Send the trx message to replicators after finalizing the 
+       * statement and transaction. This will also set the Transaction
+       * and Statement objects in Session to NULL.
+       */
+      commitTransactionMessage(in_session);
+
+      /*
+       * Statement and Transaction should now be NULL, so new ones will get
+       * created. We reuse the transaction id since we are segmenting
+       * one transaction.
+       */
       statement= in_session->getStatementMessage();
+      transaction= getActiveTransactionMessage(in_session, false);
+    }
+    else
+    {
+      const message::DeleteHeader &delete_header= statement->delete_header();
+      string old_table_name= delete_header.table_metadata().table_name();
+
+      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();
+      }
     }
   }
 
   if (statement == NULL)
   {
-    message::Transaction *transaction= getActiveTransactionMessage(in_session);
+    /*
+     * Transaction will be non-NULL only if we had to segment it due to
+     * transaction size above.
+     */
+    if (transaction == NULL)
+      transaction= getActiveTransactionMessage(in_session);
+
     /* 
      * Transaction message initialized and set, but no statement created
      * yet.  We construct one and initialize it, here, then return the
@@ -1469,10 +1600,11 @@
   if (! replication_services.isActive())
     return;
 
-  message::Statement &statement= getDeleteStatement(in_session, in_table);
+  uint32_t next_segment_id;
+  message::Statement &statement= getDeleteStatement(in_session, in_table, &next_segment_id);
 
   message::DeleteData *data= statement.mutable_delete_data();
-  data->set_segment_id(1);
+  data->set_segment_id(next_segment_id);
   data->set_end_segment(true);
   message::DeleteRecord *record= data->add_record();
 

=== modified file 'drizzled/transaction_services.h'
--- drizzled/transaction_services.h	2010-08-26 15:33:12 +0000
+++ drizzled/transaction_services.h	2010-08-28 01:13:43 +0000
@@ -80,27 +80,35 @@
   /**
    * Method which returns the active Transaction message
    * for the supplied Session.  If one is not found, a new Transaction
-   * message is allocated, initialized, and returned.
+   * message is allocated, initialized, and returned. It is possible that
+   * we may want to NOT increment the transaction id for a new Transaction
+   * object (e.g., splitting up Transactions into smaller chunks). The
+   * should_inc_trx_id flag controls if we do this.
    *
-   * @param The session processing the transaction
+   * @param in_session The session processing the transaction
+   * @param should_inc_trx_id If true, increments the transaction id for a new trx
    */
-  message::Transaction *getActiveTransactionMessage(Session *in_session);
+  message::Transaction *getActiveTransactionMessage(Session *in_session,
+                                                    bool should_inc_trx_id= true);
   /** 
    * Method which attaches a transaction context
    * the supplied transaction based on the supplied Session's
    * transaction information.  This method also ensure the
    * transaction message is attached properly to the Session object
    *
-   * @param The transaction message to initialize
-   * @param The Session processing this transaction
+   * @param in_transaction The transaction message to initialize
+   * @param in_session The Session processing this transaction
+   * @param should_inc_trx_id If true, increments the transaction id for a new trx
    */
-  void initTransactionMessage(message::Transaction &in_transaction, Session *in_session);
+  void initTransactionMessage(message::Transaction &in_transaction,
+                              Session *in_session,
+                              bool should_inc_trx_id);
   /** 
    * Helper method which finalizes data members for the 
    * supplied transaction's context.
    *
-   * @param The transaction message to finalize 
-   * @param The Session processing this transaction
+   * @param in_transaction The transaction message to finalize 
+   * @param in_session The Session processing this transaction
    */
   void finalizeTransactionMessage(message::Transaction &in_transaction, Session *in_session);
   /**
@@ -112,9 +120,9 @@
   /**
    * Helper method which initializes a Statement message
    *
-   * @param The statement to initialize
-   * @param The type of the statement
-   * @param The session processing this statement
+   * @param statement The statement to initialize
+   * @param in_type The type of the statement
+   * @param in_session The session processing this statement
    */
   void initStatementMessage(message::Statement &statement,
                             message::Statement::Type in_type,
@@ -123,27 +131,29 @@
    * Finalizes a Statement message and sets the Session's statement
    * message to NULL.
    *
-   * @param The statement to initialize
-   * @param The session processing this statement
+   * @param statement The statement to initialize
+   * @param in_session The session processing this statement
    */
   void finalizeStatementMessage(message::Statement &statement,
                                 Session *in_session);
   /** Helper method which returns an initialized Statement message for methods
    * doing insertion of data.
    *
-   * @param[in] Pointer to the Session doing the processing
-   * @param[in] Pointer to the Table object being inserted into
+   * @param[in] in_session Pointer to the Session doing the processing
+   * @param[in] in_table Pointer to the Table object being inserted into
+   * @param[out] next_segment_id The next Statement segment id to be used
    */
   message::Statement &getInsertStatement(Session *in_session,
-                                         Table *in_table);
+                                         Table *in_table,
+                                         uint32_t *next_segment_id);
 
   /**
    * Helper method which initializes the header message for
    * insert operations.
    *
-   * @param[inout] Statement message container to modify
-   * @param[in] Pointer to the Session doing the processing
-   * @param[in] Pointer to the Table being inserted into
+   * @param[in,out] statement Statement message container to modify
+   * @param[in] in_session Pointer to the Session doing the processing
+   * @param[in] in_table Pointer to the Table being inserted into
    */
   void setInsertHeader(message::Statement &statement,
                        Session *in_session,
@@ -152,24 +162,26 @@
    * Helper method which returns an initialized Statement
    * message for methods doing updates of data.
    *
-   * @param[in] Pointer to the Session doing the processing
-   * @param[in] Pointer to the Table object being updated
-   * @param[in] Pointer to the old data in the record
-   * @param[in] Pointer to the new data in the record
+   * @param[in] in_session Pointer to the Session doing the processing
+   * @param[in] in_table Pointer to the Table object being updated
+   * @param[in] old_record Pointer to the old data in the record
+   * @param[in] new_record Pointer to the new data in the record
+   * @param[out] next_segment_id The next Statement segment id to be used
    */
   message::Statement &getUpdateStatement(Session *in_session,
                                          Table *in_table,
                                          const unsigned char *old_record, 
-                                         const unsigned char *new_record);
+                                         const unsigned char *new_record,
+                                         uint32_t *next_segment_id);
   /**
    * Helper method which initializes the header message for
    * update operations.
    *
-   * @param[inout] Statement message container to modify
-   * @param[in] Pointer to the Session doing the processing
-   * @param[in] Pointer to the Table being updated
-   * @param[in] Pointer to the old data in the record
-   * @param[in] Pointer to the new data in the record
+   * @param[in,out] statement Statement message container to modify
+   * @param[in] in_session Pointer to the Session doing the processing
+   * @param[in] in_table Pointer to the Table being updated
+   * @param[in] old_record Pointer to the old data in the record
+   * @param[in] new_record Pointer to the new data in the record
    */
   void setUpdateHeader(message::Statement &statement,
                        Session *in_session,
@@ -180,19 +192,21 @@
    * Helper method which returns an initialized Statement
    * message for methods doing deletion of data.
    *
-   * @param[in] Pointer to the Session doing the processing
-   * @param[in] Pointer to the Table object being deleted from
+   * @param[in] in_session Pointer to the Session doing the processing
+   * @param[in] in_table Pointer to the Table object being deleted from
+   * @param[out] next_segment_id The next Statement segment id to be used
    */
   message::Statement &getDeleteStatement(Session *in_session,
-                                         Table *in_table);
+                                         Table *in_table,
+                                         uint32_t *next_segment_id);
 
   /**
    * Helper method which initializes the header message for
    * insert operations.
    *
-   * @param[inout] Statement message container to modify
-   * @param[in] Pointer to the Session doing the processing
-   * @param[in] Pointer to the Table being deleted from
+   * @param[in,out] statement Statement message container to modify
+   * @param[in] in_session Pointer to the Session doing the processing
+   * @param[in] in_table Pointer to the Table being deleted from
    */
   void setDeleteHeader(message::Statement &statement,
                        Session *in_session,
@@ -201,22 +215,22 @@
    * Commits a normal transaction (see above) and pushes the transaction
    * message out to the replicators.
    *
-   * @param Pointer to the Session committing the transaction
+   * @param in_session Pointer to the Session committing the transaction
    */
   int commitTransactionMessage(Session *in_session);
   /** 
    * Marks the current active transaction message as being rolled back and
    * pushes the transaction message out to replicators.
    *
-   * @param Pointer to the Session committing the transaction
+   * @param in_session Pointer to the Session committing the transaction
    */
   void rollbackTransactionMessage(Session *in_session);
   /**
    * Creates a new InsertRecord GPB message and pushes it to
    * replicators.
    *
-   * @param Pointer to the Session which has inserted a record
-   * @param Pointer to the Table containing insert information
+   * @param in_session Pointer to the Session which has inserted a record
+   * @param in_table Pointer to the Table containing insert information
    *
    * Grr, returning "true" here on error because of the cursor
    * reversed bool return crap...fix that.
@@ -226,10 +240,10 @@
    * Creates a new UpdateRecord GPB message and pushes it to
    * replicators.
    *
-   * @param Pointer to the Session which has updated a record
-   * @param Pointer to the Table containing update information
-   * @param Pointer to the raw bytes representing the old record/row
-   * @param Pointer to the raw bytes representing the new record/row 
+   * @param in_session Pointer to the Session which has updated a record
+   * @param in_table Pointer to the Table containing update information
+   * @param old_record Pointer to the raw bytes representing the old record/row
+   * @param new_record Pointer to the raw bytes representing the new record/row 
    */
   void updateRecord(Session *in_session, 
                     Table *in_table, 
@@ -249,8 +263,8 @@
    * to the Session's active Transaction GPB message for pushing
    * out to the replicator streams.
    *
-   * @param[in] Pointer to the Session which issued the statement
-   * @param[in] message::Schema message describing new schema
+   * @param[in] in_session Pointer to the Session which issued the statement
+   * @param[in] schema message::Schema message describing new schema
    */
   void createSchema(Session *in_session, const message::Schema &schema);
   /**
@@ -258,8 +272,8 @@
    * to the Session's active Transaction GPB message for pushing
    * out to the replicator streams.
    *
-   * @param[in] Pointer to the Session which issued the statement
-   * @param[in] message::Schema message describing new schema
+   * @param[in] in_session Pointer to the Session which issued the statement
+   * @param[in] schema_name message::Schema message describing new schema
    */
   void dropSchema(Session *in_session, const std::string &schema_name);
   /**
@@ -267,8 +281,8 @@
    * to the Session's active Transaction GPB message for pushing
    * out to the replicator streams.
    *
-   * @param[in] Pointer to the Session which issued the statement
-   * @param[in] message::Table message describing new schema
+   * @param[in] in_session Pointer to the Session which issued the statement
+   * @param[in] table message::Table message describing new schema
    */
   void createTable(Session *in_session, const message::Table &table);
   /**
@@ -276,10 +290,10 @@
    * to the Session's active Transaction GPB message for pushing
    * out to the replicator streams.
    *
-   * @param[in] Pointer to the Session which issued the statement
-   * @param[in] The schema of the table being dropped
-   * @param[in] The table name of the table being dropped
-   * @param[in] Did the user specify an IF EXISTS clause?
+   * @param[in] in_session Pointer to the Session which issued the statement
+   * @param[in] schema_name The schema of the table being dropped
+   * @param[in] table_name The table name of the table being dropped
+   * @param[in] if_exists Did the user specify an IF EXISTS clause?
    */
   void dropTable(Session *in_session,
                      const std::string &schema_name,
@@ -290,8 +304,8 @@
    * to the Session's active Transaction GPB message for pushing
    * out to the replicator streams.
    *
-   * @param[in] Pointer to the Session which issued the statement
-   * @param[in] The Table being truncated
+   * @param[in] in_session Pointer to the Session which issued the statement
+   * @param[in] in_table The Table being truncated
    */
   void truncateTable(Session *in_session, Table *in_table);
   /**
@@ -303,8 +317,8 @@
    * on the I_S, etc.  Not sure what to do with administrative
    * commands like CHECK TABLE, though..
    *
-   * @param Pointer to the Session which issued the statement
-   * @param Query string
+   * @param in_session Pointer to the Session which issued the statement
+   * @param query Query string
    */
   void rawStatement(Session *in_session, const std::string &query);
   /* transactions: interface to plugin::StorageEngine functions */
@@ -334,9 +348,9 @@
    * per statement, and therefore should not need to be idempotent.
    * Put in assert()s to test this.
    *
-   * @param[in] Session pointer
-   * @param[in] Descriptor for the resource which will be participating
-   * @param[in] Pointer to the TransactionalStorageEngine resource
+   * @param[in] session Session pointer
+   * @param[in] monitored Descriptor for the resource which will be participating
+   * @param[in] engine Pointer to the TransactionalStorageEngine resource
    */
   void registerResourceForStatement(Session *session,
                                     plugin::MonitoredInTransaction *monitored,
@@ -356,10 +370,10 @@
    * per statement, and therefore should not need to be idempotent.
    * Put in assert()s to test this.
    *
-   * @param[in] Session pointer
-   * @param[in] Descriptor for the resource which will be participating
-   * @param[in] Pointer to the TransactionalStorageEngine resource
-   * @param[in] Pointer to the XaResourceManager resource manager
+   * @param[in] session Session pointer
+   * @param[in] monitored Descriptor for the resource which will be participating
+   * @param[in] engine Pointer to the TransactionalStorageEngine resource
+   * @param[in] resource_manager Pointer to the XaResourceManager resource manager
    */
   void registerResourceForStatement(Session *session,
                                     plugin::MonitoredInTransaction *monitored,