diff --git a/benchmarks/roseTable.cpp b/benchmarks/roseTable.cpp
index 434352d..b9fe283 100644
--- a/benchmarks/roseTable.cpp
+++ b/benchmarks/roseTable.cpp
@@ -1,397 +1,7 @@
-#include <stdio.h>
-#include <stdlib.h>
-#include <errno.h>
-#include "stasis/operations/lsmTable.h"
-
-#include "stasis/transactional.h"
-
-#include "stasis/page/compression/multicolumn-impl.h"
-#include "stasis/page/compression/staticMulticolumn.h"
-#include "stasis/page/compression/for-impl.h"
-#include "stasis/page/compression/rle-impl.h"
-#include "stasis/page/compression/staticTuple.h"
-#include "stasis/page/compression/pageLayout.h"
-
 //#define LEAK_TEST
 
-namespace rose {
-  template<class PAGELAYOUT> 
-  void getTuple(long int i, typename PAGELAYOUT::FMT::TUP & t) {
-    typename PAGELAYOUT::FMT::TUP::TYP0 m = i;
-    typename PAGELAYOUT::FMT::TUP::TYP1 j = i / 65536;
-    typename PAGELAYOUT::FMT::TUP::TYP2 k = i / 12514500;
-    typename PAGELAYOUT::FMT::TUP::TYP3 l = i / 10000000;
-    typename PAGELAYOUT::FMT::TUP::TYP4 n = i / 65536;
-    typename PAGELAYOUT::FMT::TUP::TYP5 o = i / 12514500;
-    typename PAGELAYOUT::FMT::TUP::TYP6 p = i / 10000000;
-    typename PAGELAYOUT::FMT::TUP::TYP7 q = i / 65536;
-    typename PAGELAYOUT::FMT::TUP::TYP8 r = i / 12514500;
-    typename PAGELAYOUT::FMT::TUP::TYP9 s = i / 10000000;
-
-    t.set0(&m);
-    t.set1(&j);
-    t.set2(&k);
-    t.set3(&l);
-    t.set4(&n);
-    t.set5(&o);
-    t.set6(&p);
-    t.set7(&q);
-    t.set8(&r);
-    t.set9(&s);
-  }
-
-  template<class PAGELAYOUT>
-  int main(int argc, char **argv) {
-    unlink("storefile.txt");
-    unlink("logfile.txt");
-
-    sync();
-
-    PAGELAYOUT::initPageLayout();
-
-    bufferManagerNonBlockingSlowHandleType = IO_HANDLE_PFILE;
-
-    Tinit();
-
-    int xid = Tbegin();
-
-    recordid lsmTable = TlsmTableAlloc<PAGELAYOUT>(xid);
-
-    Tcommit(xid);
-
-    lsmTableHandle<PAGELAYOUT>* h = TlsmTableStart<PAGELAYOUT>(lsmTable);
-
-    typename PAGELAYOUT::FMT::TUP t;
-    typename PAGELAYOUT::FMT::TUP s;
-
-    long INSERTS; 
-    int file_mode = 0;
-    char * file = 0;
-    if(argc == 2) {
-      INSERTS = atoll(argv[1]);
-    } else if (argc == 3) {
-      file_mode = 1;
-      assert(!strcmp("-f", argv[1]));
-      file = argv[2];
-    } else {
-      INSERTS = 10 * 1000 * 1000;
-    }
-
-    //    int column[] = { 0 , 1, 2, 3, 4, 5, 6, 7, 8, 9 };
-    //               0   1  2  3   4  5  6  7  8   9 
-    int column[] = { 3 , 4, 1, 11, 0, 5, 6, 9, 10, 14 };
-    static long COUNT = INSERTS / 100;
-    long int count = COUNT;
-
-    struct timeval start_tv, now_tv;
-    double start, now, last_start;
-
-    gettimeofday(&start_tv,0);
-    start = rose::tv_to_double(start_tv);
-    last_start = start;
-
-
-    printf("tuple 'size'%d ; %ld\n", PAGELAYOUT::FMT::TUP::sizeofBytes(), sizeof(typename PAGELAYOUT::FMT::TUP));
-
-    if(file_mode) {
-      typename PAGELAYOUT::FMT::TUP scratch;
-
-      int max_col_number = 0;
-      for(int col =0; col < PAGELAYOUT::FMT::TUP::NN ; col++) {
-	max_col_number = max_col_number < column[col]
-	  ? column[col] : max_col_number;
-      }
-      char ** toks = (char**)malloc(sizeof(char*)*(max_col_number+1));
-      printf("Reading from file %s\n", file);
-      int inserts = 0;
-      size_t line_len = 100;
-      // getline wants malloced memmory (it probably calls realloc...)
-      char * line = (char*) malloc(sizeof(char) * line_len);
-
-      FILE * input = fopen(file, "r");
-      if(!input) {
-	perror("Couldn't open input");
-	return 1;
-      }
-      size_t read_len;
-      COUNT = 100000;
-      count = 100000;
-
-      struct timeval cannonical_start_tv;
-      gettimeofday(&cannonical_start_tv,0);
-      double cannonical_start = tv_to_double(cannonical_start_tv);
-
-      while(-1 != (read_len = getline(&line, &line_len, input))) {
-	int line_tok_count;
-	{
-	  char * saveptr;
-	  int i;
-	  toks[0] = strtok_r(line, ",\n", &saveptr);
-	  for(i = 1; i < (max_col_number+1); i++) {
-	    toks[i] = strtok_r(0, ",\n", &saveptr);
-	    if(!toks[i]) {
-	      break;
-	    }
-	  }
-	  line_tok_count = i;
-	}
-	if(line_tok_count < (max_col_number+1)) {
-	  //	  printf("!");
-	  if(-1 == getline(&line,&line_len,input)) {
-	    // hit eof.
-	  } else {
-	    printf("Not enough tokesn on line %d (found: %d expected: %d\n",
-		   inserts+1, line_tok_count, max_col_number+1);
-	    return 1;
-	  }
-	} else {
-	  //	  printf(".");
-	  inserts ++;
-
-	  if(0 < PAGELAYOUT::FMT::TUP::NN) {
-	    char * endptr;
-	    errno = 0;
-	    typename PAGELAYOUT::FMT::TUP::TYP0 t = strtoll(toks[column[0]], &endptr, 0);
-	    if(strlen(toks[column[0]]) - (size_t)(endptr-toks[column[0]]) > 1) {
-	      printf("couldnt parse token #%d: ->%s<-\n", column[0], toks[column[0]]);
-	      return 1;
-	    }
-	    if(errno) {
-	      printf("Couldn't parse token #%d: ->%s<-", column[0],toks[column[0]]);
-	      perror("strtoll error is");
-	      return 1;
-	    }
-	    scratch.set0(&t);
-	  }
-	  if(1 < PAGELAYOUT::FMT::TUP::NN) {
-	    char * endptr;
-	    errno = 0;
-	    typename PAGELAYOUT::FMT::TUP::TYP1 t = strtoll(toks[column[1]], &endptr, 0);
-	    if(strlen(toks[column[1]]) - (size_t)(endptr-toks[column[1]]) > 1) {
-	      printf("couldnt parse token #%d: ->%s<-\n", column[1], toks[column[1]]);
-	      return 1;
-	    }
-	    if(errno) {
-	      printf("Couldn't parse token #%d: ->%s<-", column[1],toks[column[1]]);
-	      perror("strtoll error is");
-	      return 1;
-	    }
-	    scratch.set1(&t);
-	  }
-	  if(2 < PAGELAYOUT::FMT::TUP::NN) {
-	    char * endptr;
-	    errno = 0;
-	    typename PAGELAYOUT::FMT::TUP::TYP2 t = strtoll(toks[column[2]], &endptr, 0);
-	    if(strlen(toks[column[2]]) - (size_t)(endptr-toks[column[2]]) > 1) {
-	      printf("couldnt parse token #%d: ->%s<-\n", column[2], toks[column[2]]);
-	      return 1;
-	    }
-	    if(errno) {
-	      printf("Couldn't parse token #%d: ->%s<-", column[2],toks[column[2]]);
-	      perror("strtoll error is");
-	      return 1;
-	    }
-	    scratch.set2(&t);
-	  }
-	  if(3 < PAGELAYOUT::FMT::TUP::NN) {
-	    char * endptr;
-	    errno = 0;
-	    typename PAGELAYOUT::FMT::TUP::TYP3 t = strtoll(toks[column[3]], &endptr, 0);
-	    if(strlen(toks[column[3]]) - (size_t)(endptr-toks[column[3]]) > 1) {
-	      printf("couldnt parse token #%d: ->%s<-\n", column[3], toks[column[3]]);
-	      return 1;
-	    }
-	    if(errno) {
-	      printf("Couldn't parse token #%d: ->%s<-", column[3],toks[column[3]]);
-	      perror("strtoll error is");
-	      return 1;
-	    }
-	    scratch.set3(&t);
-	  }
-	  if(4 < PAGELAYOUT::FMT::TUP::NN) {
-	    char * endptr;
-	    errno = 0;
-	    typename PAGELAYOUT::FMT::TUP::TYP4 t = strtoll(toks[column[4]], &endptr, 0);
-	    if(strlen(toks[column[4]]) - (size_t)(endptr-toks[column[4]]) > 1) {
-	      printf("couldnt parse token #%d: ->%s<-\n", column[4], toks[column[4]]);
-	      return 1;
-	    }
-	    if(errno) {
-	      printf("Couldn't parse token #%d: ->%s<-", column[4],toks[column[4]]);
-	      perror("strtoll error is");
-	      return 1;
-	    }
-	    scratch.set4(&t);
-	  }
-	  if(5 < PAGELAYOUT::FMT::TUP::NN) {
-	    char * endptr;
-	    errno = 0;
-	    typename PAGELAYOUT::FMT::TUP::TYP5 t = strtoll(toks[column[5]], &endptr, 0);
-	    if(strlen(toks[column[5]]) - (size_t)(endptr-toks[column[5]]) > 1) {
-	      printf("couldnt parse token #%d: ->%s<-\n", column[5], toks[column[5]]);
-	      return 1;
-	    }
-	    if(errno) {
-	      printf("Couldn't parse token #%d: ->%s<-", column[5],toks[column[5]]);
-	      perror("strtoll error is");
-	      return 1;
-	    }
-	    scratch.set5(&t);
-	  }
-	  if(6 < PAGELAYOUT::FMT::TUP::NN) {
-	    char * endptr;
-	    errno = 0;
-	    typename PAGELAYOUT::FMT::TUP::TYP6 t = strtoll(toks[column[6]], &endptr, 0);
-	    if(strlen(toks[column[6]]) - (size_t)(endptr-toks[column[6]]) > 1) {
-	      printf("couldnt parse token #%d: ->%s<-\n", column[6], toks[column[6]]);
-	      return 1;
-	    }
-	    if(errno) {
-	      printf("Couldn't parse token #%d: ->%s<-", column[6],toks[column[6]]);
-	      perror("strtoll error is");
-	      return 1;
-	    }
-	    scratch.set6(&t);
-	  }
-	  if(7 < PAGELAYOUT::FMT::TUP::NN) {
-	    char * endptr;
-	    errno = 0;
-	    typename PAGELAYOUT::FMT::TUP::TYP7 t = strtoll(toks[column[7]], &endptr, 0);
-	    if(strlen(toks[column[7]]) - (size_t)(endptr-toks[column[7]]) > 1) {
-	      printf("couldnt parse token #%d: ->%s<-\n", column[7], toks[column[7]]);
-	      return 1;
-	    }
-	    if(errno) {
-	      printf("Couldn't parse token #%d: ->%s<-", column[7],toks[column[7]]);
-	      perror("strtoll error is");
-	      return 1;
-	    }
-	    scratch.set7(&t);
-	  }
-	  if(8 < PAGELAYOUT::FMT::TUP::NN) {
-	    char * endptr;
-	    errno = 0;
-	    typename PAGELAYOUT::FMT::TUP::TYP8 t = strtoll(toks[column[8]], &endptr, 0);
-	    if(strlen(toks[column[8]]) - (size_t)(endptr-toks[column[8]]) > 1) {
-	      printf("couldnt parse token #%d: ->%s<-\n", column[8], toks[column[8]]);
-	      return 1;
-	    }
-	    if(errno) {
-	      printf("Couldn't parse token #%d: ->%s<-", column[8],toks[column[8]]);
-	      perror("strtoll error is");
-	      return 1;
-	    }
-	    scratch.set8(&t);
-	  }
-	  if(9 < PAGELAYOUT::FMT::TUP::NN) {
-	    char * endptr;
-	    errno = 0;
-	    typename PAGELAYOUT::FMT::TUP::TYP9 t = strtoll(toks[column[9]], &endptr, 0);
-	    if(strlen(toks[column[9]]) - (size_t)(endptr-toks[column[9]]) > 1) {
-	      printf("couldnt parse token #%d: ->%s<-\n", column[9], toks[column[9]]);
-	      return 1;
-	    }
-	    if(errno) {
-	      printf("Couldn't parse token #%d: ->%s<-", column[9],toks[column[9]]);
-	      perror("strtoll error is");
-	      return 1;
-	    }
-	    scratch.set9(&t);
-	  }
-
-	  //	  abort();
-	  TlsmTableInsert(h,scratch);
-	  count --;
-	  if(!count) {
-	    count = COUNT;
-	    gettimeofday(&now_tv,0);
-	    now = tv_to_double(now_tv);
-	    printf("After %6.1f seconds, wrote %d tuples "
-		   "%9.3f Mtup/sec (avg) %9.3f Mtup/sec (cur) "
-		   "%9.3f Mbyte/sec (avg) %9.3f Mbyte/sec (cur)\n",
-		   inserts, //((inserts+1) * 100) / INSERTS,
-		   now - cannonical_start,
-		   ((double)inserts/1000000.0)/(now-start),
-		   ((double)count/1000000.0)/(now-last_start),
-		   (((double)PAGELAYOUT::FMT::TUP::sizeofBytes())*(double)inserts/1000000.0)/(now-start),
-		   (((double)PAGELAYOUT::FMT::TUP::sizeofBytes())*(double)count/1000000.0)/(now-last_start)
-		   );
-	    last_start = now;
-	  }
-	}
-      }
-      printf("insertions done.\n");
-    } else {
-      for(long int i = 0; i < INSERTS; i++) {
-	getTuple<PAGELAYOUT>(i,t);
-	TlsmTableInsert(h,t);
-	//      getTuple<PAGELAYOUT>(i,t);
-	//      assert(TlsmTableFind(xid,h,t,s));
-	count --;
-	if(!count) {
-	  count = COUNT;
-	  gettimeofday(&now_tv,0);
-	  now = tv_to_double(now_tv);
-	  printf("%3ld%% write "
-		 "%9.3f Mtup/sec (avg) %9.3f Mtup/sec (cur) "
-		 "%9.3f Mbyte/sec (avg) %9.3f Mbyte/sec (cur)\n",
-		 ((i+1) * 100) / INSERTS,
-		 ((double)i/1000000.0)/(now-start),
-		 ((double)count/1000000.0)/(now-last_start),
-		 (((double)PAGELAYOUT::FMT::TUP::sizeofBytes())*(double)i/1000000.0)/(now-start),
-		 (((double)PAGELAYOUT::FMT::TUP::sizeofBytes())*(double)count/1000000.0)/(now-last_start)
-		 );
-	  last_start = now;
-	}
-#ifdef LEAK_TEST
-	if(i == INSERTS-1) {
-	  printf("Running leak test; restarting from zero.\n");
-	  i = 0;
-	}
-#endif
-      }
-      printf("insertions done.\n"); fflush(stdout);
-
-      count = COUNT;
-
-      gettimeofday(&start_tv,0);
-      start = rose::tv_to_double(start_tv);
-      last_start = start;
-
-      for(long int i = 0; i < INSERTS; i++) {
-
-	getTuple<PAGELAYOUT>(i,t);
-
-	typename PAGELAYOUT::FMT::TUP const * const sp = TlsmTableFind(xid,h,t,s);
-	assert(sp);
-	assert(*sp == s);
-	count--;
-	if(!count) {
-	  count = COUNT;
-	  gettimeofday(&now_tv,0);
-	  now = tv_to_double(now_tv);
-	  printf("%3ld%% read "
-		 "%9.3f Mtup/sec (avg) %9.3f Mtup/sec (cur) "
-		 "%9.3f Mbyte/sec (avg) %9.3f Mbyte/sec (cur)\n",
-		 ((i+1) * 100) / INSERTS,
-		 ((double)i/1000000.0)/(now-start),
-		 ((double)count/1000000.0)/(now-last_start),
-		 (((double)PAGELAYOUT::FMT::TUP::sizeofBytes())*(double)i/1000000.0)/(now-start),
-		 (((double)PAGELAYOUT::FMT::TUP::sizeofBytes())*(double)count/1000000.0)/(now-last_start)
-		 );
-	  last_start = now;
-	}
-      }
-    }
-
-    TlsmTableStop<PAGELAYOUT>(h);
-
-    Tdeinit();
-
-    printf("test\n");
-    return 0;
-  }
-}
+#include "roseTable.h"
+#include "stasis/page/compression/compression.h"
 
 int main(int argc, char **argv) {
 
@@ -407,17 +17,29 @@ int main(int argc, char **argv) {
   typedef int32_t typ9;
 
   #define COLS 10
-
   typedef rose::StaticTuple<COLS,typ0,typ1,typ2,typ3,typ4,typ5,typ6,typ7,typ8,typ9> tup;
   using rose::For;
   using rose::Rle;
-
+  int ret;
   // multicolumn is deprecated; want static dispatch!
 
-  /*  return rose::main
-    <rose::SingleColumnTypePageLayout
-      <rose::Multicolumn<tup>,rose::For<int64_t> > >
-      (argc,argv); */
+  rose::plugin_id_t * plugins = (rose::plugin_id_t*)malloc(10 * sizeof(rose::plugin_id_t));
+
+  plugins[0] = rose::plugin_id<rose::Multicolumn<tup>, Rle<typ0>, typ0>();
+  plugins[1] = rose::plugin_id<rose::Multicolumn<tup>, Rle<typ1>, typ1>();
+  plugins[2] = rose::plugin_id<rose::Multicolumn<tup>, For<typ2>, typ2>();
+  plugins[3] = rose::plugin_id<rose::Multicolumn<tup>, Rle<typ3>, typ3>();
+  plugins[4] = rose::plugin_id<rose::Multicolumn<tup>, Rle<typ4>, typ4>();
+  plugins[5] = rose::plugin_id<rose::Multicolumn<tup>, Rle<typ5>, typ5>();
+  plugins[6] = rose::plugin_id<rose::Multicolumn<tup>, For<typ6>, typ6>();
+  plugins[7] = rose::plugin_id<rose::Multicolumn<tup>, For<typ7>, typ7>();
+  plugins[8] = rose::plugin_id<rose::Multicolumn<tup>, For<typ8>, typ8>();
+  plugins[9] = rose::plugin_id<rose::Multicolumn<tup>, Rle<typ9>, typ9>();
+
+  rose::DynamicMultiColumnTypePageLayout<rose::Multicolumn<tup> >::initPageLayout(plugins);
+
+  ret = rose::main
+  <rose::DynamicMultiColumnTypePageLayout<rose::Multicolumn<tup> > >(argc,argv); 
 
   /*  return rose::main
     <rose::MultiColumnTypePageLayout
@@ -432,8 +54,19 @@ int main(int argc, char **argv) {
     >
     (argc,argv);
   */
-  return rose::main
-    <rose::MultiColumnTypePageLayout
+
+  /*  rose::StaticMultiColumnTypePageLayout
+    <COLS,
+    rose::StaticMulticolumn<COLS,tup,
+    Rle<typ0>,Rle<typ1>,
+    For<typ2>,Rle<typ3>,
+    Rle<typ4>,Rle<typ5>,
+    For<typ6>,For<typ7>,
+    For<typ8>,Rle<typ9> >
+    >::initPageLayout();
+
+  ret = rose::main
+    <rose::StaticMultiColumnTypePageLayout
     <COLS,
     rose::StaticMulticolumn<COLS,tup,
     Rle<typ0>,Rle<typ1>,
@@ -444,6 +77,6 @@ int main(int argc, char **argv) {
     >
     >
     (argc,argv);
-
-  return 0;
+  */
+  return ret;
 }
diff --git a/benchmarks/roseTable.h b/benchmarks/roseTable.h
new file mode 100644
index 0000000..473690c
--- /dev/null
+++ b/benchmarks/roseTable.h
@@ -0,0 +1,393 @@
+#include <stdio.h>
+#include <stdlib.h>
+#include <errno.h>
+#include "stasis/operations/lsmTable.h"
+
+#include "stasis/transactional.h"
+
+#include "stasis/page/compression/multicolumn-impl.h"
+#include "stasis/page/compression/staticMulticolumn.h"
+#include "stasis/page/compression/for-impl.h"
+#include "stasis/page/compression/rle-impl.h"
+#include "stasis/page/compression/staticTuple.h"
+#include "stasis/page/compression/pageLayout.h"
+
+namespace rose {
+  template<class PAGELAYOUT> 
+  void getTuple(long int i, typename PAGELAYOUT::FMT::TUP & t) {
+    typename PAGELAYOUT::FMT::TUP::TYP0 m = i;
+    typename PAGELAYOUT::FMT::TUP::TYP1 j = i / 65536;
+    typename PAGELAYOUT::FMT::TUP::TYP2 k = i / 12514500;
+    typename PAGELAYOUT::FMT::TUP::TYP3 l = i / 10000000;
+    typename PAGELAYOUT::FMT::TUP::TYP4 n = i / 65536;
+    typename PAGELAYOUT::FMT::TUP::TYP5 o = i / 12514500;
+    typename PAGELAYOUT::FMT::TUP::TYP6 p = i / 10000000;
+    typename PAGELAYOUT::FMT::TUP::TYP7 q = i / 65536;
+    typename PAGELAYOUT::FMT::TUP::TYP8 r = i / 12514500;
+    typename PAGELAYOUT::FMT::TUP::TYP9 s = i / 10000000;
+
+    t.set0(&m);
+    t.set1(&j);
+    t.set2(&k);
+    t.set3(&l);
+    t.set4(&n);
+    t.set5(&o);
+    t.set6(&p);
+    t.set7(&q);
+    t.set8(&r);
+    t.set9(&s);
+  }
+
+  template<class PAGELAYOUT>
+  int main(int argc, char **argv) {
+    unlink("storefile.txt");
+    unlink("logfile.txt");
+
+    sync();
+
+    //    PAGELAYOUT::initPageLayout();
+
+    bufferManagerNonBlockingSlowHandleType = IO_HANDLE_PFILE;
+
+    Tinit();
+
+    int xid = Tbegin();
+
+    recordid lsmTable = TlsmTableAlloc<PAGELAYOUT>(xid);
+
+    Tcommit(xid);
+
+    lsmTableHandle<PAGELAYOUT>* h = TlsmTableStart<PAGELAYOUT>(lsmTable);
+
+    typename PAGELAYOUT::FMT::TUP t;
+    typename PAGELAYOUT::FMT::TUP s;
+
+    long INSERTS; 
+    int file_mode = 0;
+    char * file = 0;
+    if(argc == 2) {
+      INSERTS = atoll(argv[1]);
+    } else if (argc == 3) {
+      file_mode = 1;
+      assert(!strcmp("-f", argv[1]));
+      file = argv[2];
+    } else {
+      INSERTS = 10 * 1000 * 1000;
+    }
+
+    //    int column[] = { 0 , 1, 2, 3, 4, 5, 6, 7, 8, 9 };
+    //               0   1  2  3   4  5  6  7  8   9 
+    int column[] = { 3 , 4, 1, 11, 0, 5, 6, 9, 10, 14 };
+    static long COUNT = INSERTS / 100;
+    long int count = COUNT;
+
+    struct timeval start_tv, now_tv;
+    double start, now, last_start;
+
+    gettimeofday(&start_tv,0);
+    start = rose::tv_to_double(start_tv);
+    last_start = start;
+
+
+    printf("tuple 'size'%d ; %ld\n", PAGELAYOUT::FMT::TUP::sizeofBytes(), sizeof(typename PAGELAYOUT::FMT::TUP));
+
+    if(file_mode) {
+      typename PAGELAYOUT::FMT::TUP scratch;
+
+      int max_col_number = 0;
+      for(int col =0; col < PAGELAYOUT::FMT::TUP::NN ; col++) {
+	max_col_number = max_col_number < column[col]
+	  ? column[col] : max_col_number;
+      }
+      char ** toks = (char**)malloc(sizeof(char*)*(max_col_number+1));
+      printf("Reading from file %s\n", file);
+      int inserts = 0;
+      size_t line_len = 100;
+      // getline wants malloced memmory (it probably calls realloc...)
+      char * line = (char*) malloc(sizeof(char) * line_len);
+
+      FILE * input = fopen(file, "r");
+      if(!input) {
+	perror("Couldn't open input");
+	return 1;
+      }
+      size_t read_len;
+      COUNT = 100000;
+      count = 100000;
+
+      struct timeval cannonical_start_tv;
+      gettimeofday(&cannonical_start_tv,0);
+      double cannonical_start = tv_to_double(cannonical_start_tv);
+
+      while(-1 != (read_len = getline(&line, &line_len, input))) {
+	int line_tok_count;
+	{
+	  char * saveptr;
+	  int i;
+	  toks[0] = strtok_r(line, ",\n", &saveptr);
+	  for(i = 1; i < (max_col_number+1); i++) {
+	    toks[i] = strtok_r(0, ",\n", &saveptr);
+	    if(!toks[i]) {
+	      break;
+	    }
+	  }
+	  line_tok_count = i;
+	}
+	if(line_tok_count < (max_col_number+1)) {
+	  //	  printf("!");
+	  if(-1 == getline(&line,&line_len,input)) {
+	    // hit eof.
+	  } else {
+	    printf("Not enough tokesn on line %d (found: %d expected: %d\n",
+		   inserts+1, line_tok_count, max_col_number+1);
+	    return 1;
+	  }
+	} else {
+	  //	  printf(".");
+	  inserts ++;
+
+	  if(0 < PAGELAYOUT::FMT::TUP::NN) {
+	    char * endptr;
+	    errno = 0;
+	    typename PAGELAYOUT::FMT::TUP::TYP0 t = strtoll(toks[column[0]], &endptr, 0);
+	    if(strlen(toks[column[0]]) - (size_t)(endptr-toks[column[0]]) > 1) {
+	      printf("couldnt parse token #%d: ->%s<-\n", column[0], toks[column[0]]);
+	      return 1;
+	    }
+	    if(errno) {
+	      printf("Couldn't parse token #%d: ->%s<-", column[0],toks[column[0]]);
+	      perror("strtoll error is");
+	      return 1;
+	    }
+	    scratch.set0(&t);
+	  }
+	  if(1 < PAGELAYOUT::FMT::TUP::NN) {
+	    char * endptr;
+	    errno = 0;
+	    typename PAGELAYOUT::FMT::TUP::TYP1 t = strtoll(toks[column[1]], &endptr, 0);
+	    if(strlen(toks[column[1]]) - (size_t)(endptr-toks[column[1]]) > 1) {
+	      printf("couldnt parse token #%d: ->%s<-\n", column[1], toks[column[1]]);
+	      return 1;
+	    }
+	    if(errno) {
+	      printf("Couldn't parse token #%d: ->%s<-", column[1],toks[column[1]]);
+	      perror("strtoll error is");
+	      return 1;
+	    }
+	    scratch.set1(&t);
+	  }
+	  if(2 < PAGELAYOUT::FMT::TUP::NN) {
+	    char * endptr;
+	    errno = 0;
+	    typename PAGELAYOUT::FMT::TUP::TYP2 t = strtoll(toks[column[2]], &endptr, 0);
+	    if(strlen(toks[column[2]]) - (size_t)(endptr-toks[column[2]]) > 1) {
+	      printf("couldnt parse token #%d: ->%s<-\n", column[2], toks[column[2]]);
+	      return 1;
+	    }
+	    if(errno) {
+	      printf("Couldn't parse token #%d: ->%s<-", column[2],toks[column[2]]);
+	      perror("strtoll error is");
+	      return 1;
+	    }
+	    scratch.set2(&t);
+	  }
+	  if(3 < PAGELAYOUT::FMT::TUP::NN) {
+	    char * endptr;
+	    errno = 0;
+	    typename PAGELAYOUT::FMT::TUP::TYP3 t = strtoll(toks[column[3]], &endptr, 0);
+	    if(strlen(toks[column[3]]) - (size_t)(endptr-toks[column[3]]) > 1) {
+	      printf("couldnt parse token #%d: ->%s<-\n", column[3], toks[column[3]]);
+	      return 1;
+	    }
+	    if(errno) {
+	      printf("Couldn't parse token #%d: ->%s<-", column[3],toks[column[3]]);
+	      perror("strtoll error is");
+	      return 1;
+	    }
+	    scratch.set3(&t);
+	  }
+	  if(4 < PAGELAYOUT::FMT::TUP::NN) {
+	    char * endptr;
+	    errno = 0;
+	    typename PAGELAYOUT::FMT::TUP::TYP4 t = strtoll(toks[column[4]], &endptr, 0);
+	    if(strlen(toks[column[4]]) - (size_t)(endptr-toks[column[4]]) > 1) {
+	      printf("couldnt parse token #%d: ->%s<-\n", column[4], toks[column[4]]);
+	      return 1;
+	    }
+	    if(errno) {
+	      printf("Couldn't parse token #%d: ->%s<-", column[4],toks[column[4]]);
+	      perror("strtoll error is");
+	      return 1;
+	    }
+	    scratch.set4(&t);
+	  }
+	  if(5 < PAGELAYOUT::FMT::TUP::NN) {
+	    char * endptr;
+	    errno = 0;
+	    typename PAGELAYOUT::FMT::TUP::TYP5 t = strtoll(toks[column[5]], &endptr, 0);
+	    if(strlen(toks[column[5]]) - (size_t)(endptr-toks[column[5]]) > 1) {
+	      printf("couldnt parse token #%d: ->%s<-\n", column[5], toks[column[5]]);
+	      return 1;
+	    }
+	    if(errno) {
+	      printf("Couldn't parse token #%d: ->%s<-", column[5],toks[column[5]]);
+	      perror("strtoll error is");
+	      return 1;
+	    }
+	    scratch.set5(&t);
+	  }
+	  if(6 < PAGELAYOUT::FMT::TUP::NN) {
+	    char * endptr;
+	    errno = 0;
+	    typename PAGELAYOUT::FMT::TUP::TYP6 t = strtoll(toks[column[6]], &endptr, 0);
+	    if(strlen(toks[column[6]]) - (size_t)(endptr-toks[column[6]]) > 1) {
+	      printf("couldnt parse token #%d: ->%s<-\n", column[6], toks[column[6]]);
+	      return 1;
+	    }
+	    if(errno) {
+	      printf("Couldn't parse token #%d: ->%s<-", column[6],toks[column[6]]);
+	      perror("strtoll error is");
+	      return 1;
+	    }
+	    scratch.set6(&t);
+	  }
+	  if(7 < PAGELAYOUT::FMT::TUP::NN) {
+	    char * endptr;
+	    errno = 0;
+	    typename PAGELAYOUT::FMT::TUP::TYP7 t = strtoll(toks[column[7]], &endptr, 0);
+	    if(strlen(toks[column[7]]) - (size_t)(endptr-toks[column[7]]) > 1) {
+	      printf("couldnt parse token #%d: ->%s<-\n", column[7], toks[column[7]]);
+	      return 1;
+	    }
+	    if(errno) {
+	      printf("Couldn't parse token #%d: ->%s<-", column[7],toks[column[7]]);
+	      perror("strtoll error is");
+	      return 1;
+	    }
+	    scratch.set7(&t);
+	  }
+	  if(8 < PAGELAYOUT::FMT::TUP::NN) {
+	    char * endptr;
+	    errno = 0;
+	    typename PAGELAYOUT::FMT::TUP::TYP8 t = strtoll(toks[column[8]], &endptr, 0);
+	    if(strlen(toks[column[8]]) - (size_t)(endptr-toks[column[8]]) > 1) {
+	      printf("couldnt parse token #%d: ->%s<-\n", column[8], toks[column[8]]);
+	      return 1;
+	    }
+	    if(errno) {
+	      printf("Couldn't parse token #%d: ->%s<-", column[8],toks[column[8]]);
+	      perror("strtoll error is");
+	      return 1;
+	    }
+	    scratch.set8(&t);
+	  }
+	  if(9 < PAGELAYOUT::FMT::TUP::NN) {
+	    char * endptr;
+	    errno = 0;
+	    typename PAGELAYOUT::FMT::TUP::TYP9 t = strtoll(toks[column[9]], &endptr, 0);
+	    if(strlen(toks[column[9]]) - (size_t)(endptr-toks[column[9]]) > 1) {
+	      printf("couldnt parse token #%d: ->%s<-\n", column[9], toks[column[9]]);
+	      return 1;
+	    }
+	    if(errno) {
+	      printf("Couldn't parse token #%d: ->%s<-", column[9],toks[column[9]]);
+	      perror("strtoll error is");
+	      return 1;
+	    }
+	    scratch.set9(&t);
+	  }
+
+	  //	  abort();
+	  TlsmTableInsert(h,scratch);
+	  count --;
+	  if(!count) {
+	    count = COUNT;
+	    gettimeofday(&now_tv,0);
+	    now = tv_to_double(now_tv);
+	    printf("After %6.1f seconds, wrote %d tuples "
+		   "%9.3f Mtup/sec (avg) %9.3f Mtup/sec (cur) "
+		   "%9.3f Mbyte/sec (avg) %9.3f Mbyte/sec (cur)\n",
+		   now - cannonical_start,
+		   inserts, //((inserts+1) * 100) / INSERTS,
+		   ((double)inserts/1000000.0)/(now-start),
+		   ((double)count/1000000.0)/(now-last_start),
+		   (((double)PAGELAYOUT::FMT::TUP::sizeofBytes())*(double)inserts/1000000.0)/(now-start),
+		   (((double)PAGELAYOUT::FMT::TUP::sizeofBytes())*(double)count/1000000.0)/(now-last_start)
+		   );
+	    last_start = now;
+	  }
+	}
+      }
+      printf("insertions done.\n");
+    } else {
+      for(long int i = 0; i < INSERTS; i++) {
+	getTuple<PAGELAYOUT>(i,t);
+	TlsmTableInsert(h,t);
+	//      getTuple<PAGELAYOUT>(i,t);
+	//      assert(TlsmTableFind(xid,h,t,s));
+	count --;
+	if(!count) {
+	  count = COUNT;
+	  gettimeofday(&now_tv,0);
+	  now = tv_to_double(now_tv);
+	  printf("%3ld%% write "
+		 "%9.3f Mtup/sec (avg) %9.3f Mtup/sec (cur) "
+		 "%9.3f Mbyte/sec (avg) %9.3f Mbyte/sec (cur)\n",
+		 ((i+1) * 100) / INSERTS,
+		 ((double)i/1000000.0)/(now-start),
+		 ((double)count/1000000.0)/(now-last_start),
+		 (((double)PAGELAYOUT::FMT::TUP::sizeofBytes())*(double)i/1000000.0)/(now-start),
+		 (((double)PAGELAYOUT::FMT::TUP::sizeofBytes())*(double)count/1000000.0)/(now-last_start)
+		 );
+	  last_start = now;
+	}
+#ifdef LEAK_TEST
+	if(i == INSERTS-1) {
+	  printf("Running leak test; restarting from zero.\n");
+	  i = 0;
+	}
+#endif
+      }
+      printf("insertions done.\n"); fflush(stdout);
+
+      count = COUNT;
+
+      gettimeofday(&start_tv,0);
+      start = rose::tv_to_double(start_tv);
+      last_start = start;
+
+      for(long int i = 0; i < INSERTS; i++) {
+
+	getTuple<PAGELAYOUT>(i,t);
+
+	typename PAGELAYOUT::FMT::TUP const * const sp = TlsmTableFind(xid,h,t,s);
+	assert(sp);
+	assert(*sp == s);
+	count--;
+	if(!count) {
+	  count = COUNT;
+	  gettimeofday(&now_tv,0);
+	  now = tv_to_double(now_tv);
+	  printf("%3ld%% read "
+		 "%9.3f Mtup/sec (avg) %9.3f Mtup/sec (cur) "
+		 "%9.3f Mbyte/sec (avg) %9.3f Mbyte/sec (cur)\n",
+		 ((i+1) * 100) / INSERTS,
+		 ((double)i/1000000.0)/(now-start),
+		 ((double)count/1000000.0)/(now-last_start),
+		 (((double)PAGELAYOUT::FMT::TUP::sizeofBytes())*(double)i/1000000.0)/(now-start),
+		 (((double)PAGELAYOUT::FMT::TUP::sizeofBytes())*(double)count/1000000.0)/(now-last_start)
+		 );
+	  last_start = now;
+	}
+      }
+    }
+
+    TlsmTableStop<PAGELAYOUT>(h);
+
+    Tdeinit();
+
+    printf("test\n");
+    return 0;
+  }
+}
+