Merge lp:~stewart/drizzle/bug646898-functionengine-mem-usage-position into lp:drizzle/7.0

[Stewart Smith]

greatly reduce the memory usage of function_engine tables when ::position() is used.
by storing packed rows instead of the full row image, we can decrease memory usage by more than a order of magnitude in some situations.

  Replace FunctionCursor row_cache of full rows with one that is the
  packed row (using similar routines as ARHCIVE does before
  compression). This means we use minimal memory for each row. The ref
  for ::position() is now a offset into the row_cache buffer instead of
  an incrementing index into an array.

  This change GREATLY reduces the memory required for various queries on
  table function tables.

  e.g. (measured using valgrind --tool=massif) running the
  information_schema_dictionary test suite
  BEFORE: peaked at 77.5MB heap usage
  AFTER: peaked at 33.7MB heap usage

  for schema_dictionary suite:
  BEFORE: peaked at 782.6MB heap usage
  AFTER: peaked at 31.05MB heap usage

  (i.e. memory requirements can be reduced by moer than an order of
  magnitude)

  A future patch may want to also overflow to disk if needed.

http://hudson.drizzle.org/view/Drizzle-param/job/drizzle-param/753/

[Brian Aker (brianaker)]

Thanks, in general I think we need to find a generic solution to this problem at some point.

=== modified file 'plugin/function_engine/cursor.cc'
--- plugin/function_engine/cursor.cc	2011-02-17 00:14:13 +0000
+++ plugin/function_engine/cursor.cc	2011-02-23 01:11:55 +0000
@@ -23,6 +23,7 @@
 #include <plugin/function_engine/cursor.h>
 #include <drizzled/session.h>
 #include <drizzled/internal/my_sys.h>
+#include <drizzled/field/blob.h>
 
 #include <unistd.h>
 #include <fcntl.h>
@@ -48,7 +49,7 @@
   tool= static_cast<Function *>(getEngine())->getFunction(name); 
 //  assert(tool);
 
-  record_id= 0;
+  row_cache_position= 0;
 
   if (not tool)
     return HA_ERR_NO_SUCH_TABLE;
@@ -100,15 +101,55 @@
   return more_rows ? 0 : HA_ERR_END_OF_FILE;
 }
 
+uint32_t FunctionCursor::max_row_length()
+{
+  uint32_t length= (uint32_t)(getTable()->getRecordLength() + getTable()->sizeFields()*2);
+
+  uint32_t *ptr, *end;
+  for (ptr= getTable()->getBlobField(), end=ptr + getTable()->sizeBlobFields();
+       ptr != end ;
+       ptr++)
+  {
+      length += 2 + ((Field_blob*)getTable()->getField(*ptr))->get_length();
+  }
+
+  return length;
+}
+
+unsigned int FunctionCursor::pack_row(const unsigned char *record)
+{
+  unsigned char *ptr;
+
+  record_buffer.resize(max_row_length());
+
+  /* Copy null bits */
+  memcpy(&record_buffer[0], record, getTable()->getShare()->null_bytes);
+  ptr= &record_buffer[0] + getTable()->getShare()->null_bytes;
+
+  for (Field **field=getTable()->getFields() ; *field ; field++)
+  {
+    if (!((*field)->is_null()))
+      ptr= (*field)->pack(ptr, record + (*field)->offset(record));
+  }
+
+  return((unsigned int) (ptr - &record_buffer[0]));
+}
+
 void FunctionCursor::position(const unsigned char *record)
 {
-  if (row_cache.size() <= record_id * getTable()->getShare()->getRecordLength())
+  uint32_t max_length= max_row_length();
+
+  if (row_cache.size() <= row_cache_position + max_length)
   {
-    row_cache.resize(row_cache.size() + getTable()->getShare()->getRecordLength() * 100); // Hardwired at adding an additional 100 rows of storage
+    row_cache.resize(row_cache.size() +  max_length);
   }
-  memcpy(&row_cache[record_id * getTable()->getShare()->getRecordLength()], record, getTable()->getShare()->getRecordLength());
-  internal::my_store_ptr(ref, ref_length, record_id);
-  record_id++;
+
+  unsigned int r_pack_length;
+  r_pack_length= pack_row(record);
+  internal::my_store_ptr(ref, ref_length, row_cache_position);
+
+  memcpy(&row_cache[row_cache_position], &record_buffer[0], r_pack_length);
+  row_cache_position+= r_pack_length;
 }
 
 
@@ -118,7 +159,7 @@
     estimate_of_rows= rows_returned;
 
   row_cache.clear();
-  record_id= 0;
+  row_cache_position= 0;
 }
 
 int FunctionCursor::extra(enum ha_extra_function operation)
@@ -151,8 +192,20 @@
   ha_statistic_increment(&system_status_var::ha_read_rnd_count);
   size_t position_id= (size_t)internal::my_get_ptr(pos, ref_length);
 
-  assert(position_id * getTable()->getShare()->getRecordLength() < row_cache.size());
-  memcpy(buf, &row_cache[position_id * getTable()->getShare()->getRecordLength()], getTable()->getShare()->getRecordLength());
+  const unsigned char *ptr;
+  ptr= &row_cache[position_id];
+
+  /* Copy null bits */
+  memcpy(buf, ptr, getTable()->getNullBytes());
+  ptr+= getTable()->getNullBytes();
+  // and copy fields
+  for (Field **field= getTable()->getFields() ; *field ; field++)
+  {
+    if (!((*field)->is_null()))
+    {
+      ptr= (*field)->unpack(buf + (*field)->offset(getTable()->getInsertRecord()), ptr);
+    }
+  }
 
   return 0;
 }

=== modified file 'plugin/function_engine/cursor.h'
--- plugin/function_engine/cursor.h	2010-12-18 04:43:40 +0000
+++ plugin/function_engine/cursor.h	2011-02-23 01:11:55 +0000
@@ -30,13 +30,17 @@
 {
   drizzled::plugin::TableFunction *tool;
   drizzled::plugin::TableFunction::Generator *generator;
-  size_t record_id;
-  std::vector<unsigned char *> row_cache;
+  size_t row_cache_position;
+  std::vector<unsigned char> row_cache;
   drizzled::ha_rows estimate_of_rows;
   drizzled::ha_rows rows_returned;
 
   void wipeCache();
 
+  std::vector <unsigned char> record_buffer; // for pack_row
+  uint32_t max_row_length();
+  unsigned int pack_row(const unsigned char *record);
+
 public:
   FunctionCursor(drizzled::plugin::StorageEngine &engine,
                  drizzled::Table &table_arg);