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

[David Shrewsbury]

This changes the transaction_reader program, which is not distributed and is used for our test suite, to output correct SQL by grouping GPB Transaction messages by transaction ID when the input stream has them intermingled. Doing this allows a single threaded reader to execute the SQL statements as they are read.

=== modified file 'drizzled/message/include.am'
--- drizzled/message/include.am	2010-08-24 16:45:09 +0000
+++ drizzled/message/include.am	2010-09-27 14:31:05 +0000
@@ -60,7 +60,8 @@
 
 noinst_HEADERS+= \
 		drizzled/message/ioutil.h \
-		drizzled/message/all.h
+		drizzled/message/all.h \
+		drizzled/message/transaction_manager.h
 
 
 drizzled_message_schema_reader_SOURCES =  drizzled/message/schema_reader.cc
@@ -87,7 +88,8 @@
 drizzled_message_transaction_writer_LDADD = ${MESSAGE_LDADD} ${top_builddir}/drizzled/algorithm/libhash.la
 drizzled_message_transaction_writer_CXXFLAGS = ${MESSAGE_AM_CXXFLAGS} ${NO_WERROR}
 
-drizzled_message_transaction_reader_SOURCES =  drizzled/message/transaction_reader.cc
+drizzled_message_transaction_reader_SOURCES =  drizzled/message/transaction_reader.cc \
+                                               drizzled/message/transaction_manager.cc
 drizzled_message_transaction_reader_LDADD = ${MESSAGE_LDADD} ${top_builddir}/drizzled/algorithm/libhash.la ${top_builddir}/drizzled/util/libutil.la
 drizzled_message_transaction_reader_CXXFLAGS = ${MESSAGE_AM_CXXFLAGS} ${NO_WERROR}
 
 
=== added file 'drizzled/message/transaction_manager.cc'
--- drizzled/message/transaction_manager.cc	1970-01-01 00:00:00 +0000
+++ drizzled/message/transaction_manager.cc	2010-09-27 14:31:05 +0000
@@ -0,0 +1,76 @@
+/* -*- mode: c++; c-basic-offset: 2; indent-tabs-mode: nil; -*-
+ *  vim:expandtab:shiftwidth=2:tabstop=2:smarttab:
+ *
+ *  Copyright (C) 2010 David Shrewsbury <[e-mail]>
+ *
+ *  This program is free software; you can redistribute it and/or modify
+ *  it under the terms of the GNU General Public License as published by
+ *  the Free Software Foundation; version 2 of the License.
+ *
+ *  This program is distributed in the hope that it will be useful,
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *  GNU General Public License for more details.
+ *
+ *  You should have received a copy of the GNU General Public License
+ *  along with this program; if not, write to the Free Software
+ *  Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
+ */
+
+/**
+ * @file
+ * Implementation of the TransactionManager class.
+ */
+
+#include "config.h"
+#include "drizzled/message/transaction_manager.h"
+
+using namespace std;
+
+namespace drizzled
+{
+namespace message
+{
+
+bool TransactionManager::store(const message::Transaction &transaction)
+{
+  string msg;
+  const message::TransactionContext trx_ctxt= transaction.transaction_context();
+  uint64_t trx_id= trx_ctxt.transaction_id();
+  transaction.SerializeToString(&msg);
+
+  cache[trx_id].push_back(msg);
+  return true;
+}
+
+bool TransactionManager::remove(uint64_t trx_id)
+{
+  cache.erase(trx_id);
+  return true;
+}
+
+bool TransactionManager::contains(uint64_t trx_id)
+{
+  boost::unordered_map< uint64_t, vector<string> >::const_iterator it= cache.find(trx_id);
+
+  if (it != cache.end())
+    return true;
+
+  return false;
+}
+
+uint32_t TransactionManager::getTransactionBufferSize(uint64_t trx_id)
+{
+  return static_cast<uint32_t>(cache[trx_id].size());
+}
+
+bool TransactionManager::getTransactionMessage(message::Transaction &trx,
+                                               uint64_t trx_id,
+                                               uint32_t position)
+{
+  trx.ParseFromString(cache[trx_id].at(position));
+  return true;
+}
+
+} /* namespace message */
+} /* namespace drizzled */
 
=== added file 'drizzled/message/transaction_manager.h'
--- drizzled/message/transaction_manager.h	1970-01-01 00:00:00 +0000
+++ drizzled/message/transaction_manager.h	2010-09-27 14:31:05 +0000
@@ -0,0 +1,129 @@
+/* -*- mode: c++; c-basic-offset: 2; indent-tabs-mode: nil; -*-
+ *  vim:expandtab:shiftwidth=2:tabstop=2:smarttab:
+ *
+ *  Copyright (c) 2010 David Shrewsbury <[e-mail]>
+ *
+ *  This program is free software; you can redistribute it and/or modify
+ *  it under the terms of the GNU General Public License as published by
+ *  the Free Software Foundation; version 2 of the License.
+ *
+ *  This program is distributed in the hope that it will be useful,
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *  GNU General Public License for more details.
+ *
+ *  You should have received a copy of the GNU General Public License
+ *  along with this program; if not, write to the Free Software
+ *  Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
+ */
+
+/**
+ * @file
+ * Declaration of a class used to manage Transaction messages so that they
+ * can be retrieved as a collection.
+ */
+
+#ifndef DRIZZLED_MESSAGE_TRANSACTION_MANAGER_H
+#define DRIZZLED_MESSAGE_TRANSACTION_MANAGER_H
+
+#include "drizzled/message/transaction.pb.h"
+#include <string>
+#include <vector>
+#include <boost/unordered_map.hpp>
+
+typedef std::vector<std::string> MsgBufferType;
+
+namespace drizzled
+{
+namespace message
+{
+
+/**
+ * Simple (example) Transaction message buffer and content manager.
+ *
+ * @details
+ * This class groups Transaction messages together by transaction ID and
+ * buffers them together in memory. Obviously, this could eat up a lot of
+ * memory if you have very large transactions. A more robust implementation
+ * would buffer larger transactions to disk rather than memory.
+ *
+ * @note
+ * Once you have a complete transaction and have processed it, you should
+ * call remove() to remove the cache contents for that transaction from memory.
+ */
+class TransactionManager
+{
+public:
+  /**
+   * Store the given Transaction message in a buffer.
+   *
+   * @param[in] transaction Pointer to the Transaction message to store.
+   *
+   * @retval true Success
+   * @retval false Failure
+   */
+  bool store(const drizzled::message::Transaction &transaction);
+
+  /**
+   * Clear the buffer contents for a given transaction ID.
+   *
+   * @param[in] trx_id The transaction ID for the transaction to remove
+   *
+   * @retval true Success
+   * @retval false Failure
+   */
+  bool remove(uint64_t trx_id);
+
+  /**
+   * Check to see if any Transaction messages exist for a given transaction.
+   *
+   * @param[in] trx_id The transaction ID to check for.
+   *
+   * @retval true Transaction messages exist
+   * @retval false No Transaction messages found
+   */
+  bool contains(uint64_t trx_id);
+
+  /**
+   * Return number of cached elements for the given transaction ID.
+   *
+   * @param[in] trx_id Transaction ID
+   *
+   * @returns The number of cached elements associated with trx_id.
+   */
+  uint32_t getTransactionBufferSize(uint64_t trx_id);
+
+  /**
+   * Retrieve a Transaction message from the managed cache.
+   *
+   * Caller must supply a Transaction message to populate. The Transaction
+   * message to retrieve is indexed by a combination of transaction ID and
+   * position.
+   *
+   * @param[out] transaction Transaction message to populate
+   * @param[in] trx_id Transaction ID
+   * @param[in] position Index into the buffer associated with trx_id
+   *
+   * @retval true Success
+   * @retval false Failure
+   */
+  bool getTransactionMessage(drizzled::message::Transaction &transaction,
+                             uint64_t trx_id,
+                             uint32_t position);
+
+private:
+  /**
+   * Our message buffer cache, mapped by the transaction ID.
+   *
+   * We organize Transactions messages by grouping them by transaction ID,
+   * then storing the messages in std::vectors in std::string format. The
+   * string format is convenient because it can be easily copied around
+   * (GPB messages do not provide deep copying).
+   */
+  boost::unordered_map<uint64_t,MsgBufferType> cache;
+};
+
+} /* namespace message */
+} /* namespace drizzled */
+
+#endif /* DRIZZLED_MESSAGE_TRANSACTION_MANAGER_H */
 
=== modified file 'drizzled/message/transaction_reader.cc'
--- drizzled/message/transaction_reader.cc	2010-08-27 13:02:11 +0000
+++ drizzled/message/transaction_reader.cc	2010-09-27 14:31:05 +0000
@@ -39,6 +39,7 @@
 #include <unistd.h>
 #include <drizzled/message/transaction.pb.h>
 #include <drizzled/message/statement_transform.h>
+#include <drizzled/message/transaction_manager.h>
 #include <drizzled/util/convert.h>
 
 #include <google/protobuf/io/coded_stream.h>
@@ -126,6 +127,27 @@
   return true;
 }
 
+static bool isEndTransaction(const message::Transaction &transaction)
+{
+  const message::TransactionContext trx= transaction.transaction_context();
+
+  size_t num_statements= transaction.statement_size();
+
+  /*
+   * If any Statement is partial, then we can expect another Transaction
+   * message.
+   */
+  for (size_t x= 0; x < num_statements; ++x)
+  {
+    const message::Statement &statement= transaction.statement(x);
+
+    if (not isEndStatement(statement))
+      return false;
+  }
+
+  return true;
+}
+
 static void printTransaction(const message::Transaction &transaction)
 {
   static uint64_t last_trx_id= 0;
@@ -137,7 +159,8 @@
 
   /*
    * One way to determine when a new transaction begins is when the
-   * transaction id changes. We check that here.
+   * transaction id changes (if all transactions have their GPB messages
+   * grouped together, which this program will). We check that here.
    */
   if (trx.transaction_id() != last_trx_id)
     cout << "START TRANSACTION;" << endl;
@@ -196,6 +219,8 @@
       do_checksum= true;
   }
 
+  message::TransactionManager trx_mgr;
+
   protobuf::io::ZeroCopyInputStream *raw_input= new protobuf::io::FileInputStream(file);
   protobuf::io::CodedInputStream *coded_input= new protobuf::io::CodedInputStream(raw_input);
 
@@ -276,8 +301,42 @@
       break;
     }
 
-    /* Print the transaction */
-    printTransaction(transaction);
+    if (not isEndTransaction(transaction))
+    {
+      trx_mgr.store(transaction);
+    }
+    else
+    {
+      const message::TransactionContext trx= transaction.transaction_context();
+      uint64_t transaction_id= trx.transaction_id();
+
+      /*
+       * If there are any previous Transaction messages for this transaction,
+       * store this one, then output all of them together.
+       */
+      if (trx_mgr.contains(transaction_id))
+      {
+        trx_mgr.store(transaction);
+
+        uint32_t size= trx_mgr.getTransactionBufferSize(transaction_id);
+        uint32_t idx= 0;
+
+        while (idx != size)
+        {
+          message::Transaction new_trx;
+          trx_mgr.getTransactionMessage(new_trx, transaction_id, idx);
+          printTransaction(new_trx);
+          idx++;
+        }
+
+        /* No longer need this transaction */
+        trx_mgr.remove(transaction_id);
+      }
+      else
+      {
+        printTransaction(transaction);
+      }
+    }
 
     /* Skip 4 byte checksum */
     coded_input->ReadLittleEndian32(&checksum);
@@ -291,10 +350,11 @@
     }
 
     previous_length= length;
-  }
+  } /* end while */
+
   if (buffer)
     free(buffer);
-  
+
   delete coded_input;
   delete raw_input;