diff --git a/include/maria.h b/include/maria.h
index b216affa3fc..43bafddb2ce 100644
--- a/include/maria.h
+++ b/include/maria.h
@@ -391,6 +391,7 @@ typedef struct st_maria_sort_param
 
 /* functions in maria_check */
 void maria_chk_init(HA_CHECK *param);
+void maria_chk_init_for_check(HA_CHECK *param, MARIA_HA *info);
 int maria_chk_status(HA_CHECK *param, MARIA_HA *info);
 int maria_chk_del(HA_CHECK *param, MARIA_HA *info, ulonglong test_flag);
 int maria_chk_size(HA_CHECK *param, MARIA_HA *info);
@@ -432,6 +433,8 @@ int maria_assign_to_pagecache(MARIA_HA *info, ulonglong key_map,
 void maria_change_pagecache(PAGECACHE *old_key_cache,
 			    PAGECACHE *new_key_cache);
 int maria_preload(MARIA_HA *info, ulonglong key_map, my_bool ignore_leaves);
+void maria_versioning(MARIA_HA *info, my_bool versioning);
+void maria_ignore_trids(MARIA_HA *info);
 
 /* fulltext functions */
 FT_INFO *maria_ft_init_search(uint,void *, uint, uchar *, uint,
diff --git a/include/my_base.h b/include/my_base.h
index ddefb5be7b4..719b7c043b4 100644
--- a/include/my_base.h
+++ b/include/my_base.h
@@ -440,9 +440,12 @@ enum ha_base_keytype {
 #define HA_ERR_INITIALIZATION     173    /* Error during initialization */
 #define HA_ERR_FILE_TOO_SHORT	  174	 /* File too short */
 #define HA_ERR_WRONG_CRC	  175	 /* Wrong CRC on page */
-#define HA_ERR_ROWS_EVENT_APPLY   176    /* The event could not be processed */
-                                         /* no other hanlder error happened  */
-#define HA_ERR_LAST               176    /* Copy of last error nr */
+
+/* The event could not be processed; no other handler error happened  */
+#define HA_ERR_ROWS_EVENT_APPLY   176
+#define HA_ERR_ROW_NOT_VISIBLE    177
+
+#define HA_ERR_LAST               177    /* Copy of last error nr */
 
 /* Number of different errors */
 #define HA_ERR_ERRORS            (HA_ERR_LAST - HA_ERR_FIRST + 1)
diff --git a/include/myisamchk.h b/include/myisamchk.h
index 29f523558aa..f4651b81f8b 100644
--- a/include/myisamchk.h
+++ b/include/myisamchk.h
@@ -134,6 +134,11 @@ typedef struct st_handler_check_param
   ha_checksum tmp_record_checksum;
   ulonglong   org_key_map;
   ulonglong   testflag;
+
+  /* Following is used to check if rows are visible */
+  ulonglong max_trid, max_found_trid;
+  ulonglong not_visible_rows_found;
+
   size_t use_buffers, read_buffer_length, write_buffer_length;
   size_t sort_buffer_length, sort_key_blocks;
   ulong rec_per_key_part[HA_MAX_KEY_SEG * HA_MAX_POSSIBLE_KEY];
diff --git a/include/thr_lock.h b/include/thr_lock.h
index a4ca6e6ddf2..821c1bf2949 100644
--- a/include/thr_lock.h
+++ b/include/thr_lock.h
@@ -123,11 +123,12 @@ typedef struct st_thr_lock {
   /* write_lock_count is incremented for write locks and reset on read locks */
   ulong write_lock_count;
   uint read_no_write_count;
-  void (*get_status)(void*, int);	/* When one gets a lock */
+  void (*get_status)(void*, my_bool);	/* When one gets a lock */
   void (*copy_status)(void*,void*);
   void (*update_status)(void*);		/* Before release of write */
   void (*restore_status)(void*);         /* Before release of read */
   my_bool (*check_status)(void *);
+  my_bool allow_multiple_concurrent_insert;
 } THR_LOCK;
 
 
diff --git a/mysql-test/r/maria-mvcc.result b/mysql-test/r/maria-mvcc.result
new file mode 100644
index 00000000000..20648f41622
--- /dev/null
+++ b/mysql-test/r/maria-mvcc.result
@@ -0,0 +1,79 @@
+drop table if exists t1;
+create table t1 (i int) engine=maria;
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `i` int(11) DEFAULT NULL
+) ENGINE=MARIA DEFAULT CHARSET=latin1 PAGE_CHECKSUM=1
+lock tables t1 write concurrent;
+insert into t1 values (1);
+insert into t1 values (2);
+/* should see 1 and 2 */
+select i from t1;
+i
+1
+2
+/* should see nothing */
+select i from t1;
+i
+lock tables t1 write concurrent;
+insert into t1 values (3);
+insert into t1 values (4);
+/* should see 3 and 4 */
+select i from t1;
+i
+3
+4
+unlock tables;
+lock tables t1 write concurrent;
+insert into t1 values (5);
+/* should see 3, 4 and 5 */
+select i from t1;
+i
+3
+4
+5
+insert into t1 values (6);
+/* Should see 1, 2, 6 */
+select i from t1;
+i
+1
+2
+6
+unlock tables;
+lock tables t1 write concurrent;
+/* Should see 1, 2, 3, 4 and 6 */
+select i from t1;
+i
+1
+2
+3
+4
+6
+/* should see 3, 4, 5 */
+select i from t1;
+i
+3
+4
+5
+unlock tables;
+/* should see 1, 2, 3, 4, 5, 6 */
+select i from t1;
+i
+1
+2
+3
+4
+5
+6
+unlock tables;
+/* should see 1, 2, 3, 4, 5, 6 */
+select i from t1;
+i
+1
+2
+3
+4
+5
+6
+drop table t1;
diff --git a/mysql-test/r/maria-page-checksum.result b/mysql-test/r/maria-page-checksum.result
index 1a7ec2e3fdd..d7ab1fc4ba3 100644
--- a/mysql-test/r/maria-page-checksum.result
+++ b/mysql-test/r/maria-page-checksum.result
@@ -1,3 +1,4 @@
+drop table if exists t1;
 select @@global.maria_page_checksum;
 @@global.maria_page_checksum
 1
diff --git a/mysql-test/t/maria-mvcc.test b/mysql-test/t/maria-mvcc.test
new file mode 100644
index 00000000000..3358781e76a
--- /dev/null
+++ b/mysql-test/t/maria-mvcc.test
@@ -0,0 +1,59 @@
+#
+# Testing insert and select on a table with two threads
+# using locking
+#
+
+-- source include/have_maria.inc
+
+--disable_warnings
+drop table if exists t1;
+--enable_warnings
+
+connect (con1,localhost,root,,);
+connection con1;
+
+create table t1 (i int) engine=maria;
+show create table t1;
+lock tables t1 write concurrent;
+insert into t1 values (1);
+insert into t1 values (2);
+/* should see 1 and 2 */
+select i from t1;
+
+connect (con2,localhost,root,,);
+connection con2;
+/* should see nothing */
+select i from t1;
+lock tables t1 write concurrent;
+insert into t1 values (3);
+insert into t1 values (4);
+/* should see 3 and 4 */
+select i from t1;
+unlock tables;
+lock tables t1 write concurrent;
+insert into t1 values (5);
+/* should see 3, 4 and 5 */
+select i from t1;
+
+connection con1;
+insert into t1 values (6);
+/* Should see 1, 2, 6 */
+select i from t1;
+unlock tables;
+lock tables t1 write concurrent;
+/* Should see 1, 2, 3, 4 and 6 */
+select i from t1;
+
+connection con2;
+/* should see 3, 4, 5 */
+select i from t1;
+unlock tables;
+/* should see 1, 2, 3, 4, 5, 6 */
+select i from t1;
+
+connection con1;
+unlock tables;
+/* should see 1, 2, 3, 4, 5, 6 */
+select i from t1;
+
+drop table t1;
diff --git a/mysql-test/t/maria-page-checksum.test b/mysql-test/t/maria-page-checksum.test
index 7bed7393d2d..43fd9b70867 100644
--- a/mysql-test/t/maria-page-checksum.test
+++ b/mysql-test/t/maria-page-checksum.test
@@ -1,3 +1,9 @@
+-- source include/have_maria.inc
+
+--disable_warnings
+drop table if exists t1;
+--enable_warnings
+
 select @@global.maria_page_checksum;
 
 --echo # iteration 1
diff --git a/mysys/my_handler.c b/mysys/my_handler.c
index 3bac59f409c..c5575809bb0 100644
--- a/mysys/my_handler.c
+++ b/mysys/my_handler.c
@@ -629,6 +629,8 @@ static const char *handler_error_messages[]=
   "Got a fatal error during initialzaction of handler",
   "File to short; Expected more data in file",
   "Read page with wrong checksum"
+  "Could not apply row event",
+  "Row is not visible by the current transaction",
 };
 
 
diff --git a/mysys/thr_lock.c b/mysys/thr_lock.c
index 6decd6a6a27..484fd1dd559 100644
--- a/mysys/thr_lock.c
+++ b/mysys/thr_lock.c
@@ -69,9 +69,11 @@ get_status:
 	for concurrent reads.
 
 The lock algorithm allows one to have one TL_WRITE_ALLOW_READ,
-TL_WRITE_CONCURRENT_INSERT or one TL_WRITE_DELAYED lock at the same time as
-multiple read locks.
+TL_WRITE_CONCURRENT_INSERT or one TL_WRITE_DELAYED lock at the same
+time as multiple read locks.
 
+In addition, if lock->allow_multiple_concurrent_insert is set then there can
+be any number of TL_WRITE_CONCURRENT_INSERT locks aktive at the same time.
 */
 
 #if !defined(MAIN) && !defined(DBUG_OFF) && !defined(EXTRA_DEBUG)
@@ -152,7 +154,8 @@ static int check_lock(struct st_lock_list *list, const char* lock_type,
       }
       if (same_owner &&
           !thr_lock_owner_equal(data->owner, first_owner) &&
-	  last_lock_type != TL_WRITE_ALLOW_WRITE)
+	  last_lock_type != TL_WRITE_ALLOW_WRITE &&
+          last_lock_type != TL_WRITE_CONCURRENT_INSERT)
       {
 	fprintf(stderr,
 		"Warning: Found locks from different threads in %s: %s\n",
@@ -205,7 +208,7 @@ static void check_locks(THR_LOCK *lock, const char *where,
       THR_LOCK_DATA *data;
       for (data=lock->read.data ; data ; data=data->next)
       {
-	if ((int) data->type == (int) TL_READ_NO_INSERT)
+	if (data->type == TL_READ_NO_INSERT)
 	  count++;
         /* Protect against infinite loop. */
         DBUG_ASSERT(count <= lock->read_no_write_count);
@@ -254,7 +257,22 @@ static void check_locks(THR_LOCK *lock, const char *where,
 	}	      
       }
       else
-      {						/* Have write lock */
+      {
+        /* We have at least one write lock */
+        if (lock->write.data->type == TL_WRITE_CONCURRENT_INSERT)
+        {
+          THR_LOCK_DATA *data;
+          for (data=lock->write.data->next ; data ; data=data->next)
+          {
+            if (data->type != TL_WRITE_CONCURRENT_INSERT)
+            {
+              fprintf(stderr,
+                      "Warning at '%s': Found TL_WRITE_CONCURRENT_INSERT lock mixed with other write locks\n",
+                      where);
+              break;
+            }
+          }
+        }
 	if (lock->write_wait.data)
 	{
 	  if (!allow_no_locks && 
@@ -514,7 +532,8 @@ thr_lock(THR_LOCK_DATA *data, THR_LOCK_OWNER *owner,
     /* Request for READ lock */
     if (lock->write.data)
     {
-      /* We can allow a read lock even if there is already a write lock
+      /*
+        We can allow a read lock even if there is already a write lock
 	 on the table in one the following cases:
 	 - This thread alread have a write lock on the table
 	 - The write lock is TL_WRITE_ALLOW_READ or TL_WRITE_DELAYED
@@ -558,11 +577,11 @@ thr_lock(THR_LOCK_DATA *data, THR_LOCK_OWNER *owner,
       (*lock->read.last)=data;			/* Add to running FIFO */
       data->prev=lock->read.last;
       lock->read.last= &data->next;
-      if (lock->get_status)
-	(*lock->get_status)(data->status_param, 0);
       if (lock_type == TL_READ_NO_INSERT)
 	lock->read_no_write_count++;
       check_locks(lock,"read lock with no write locks",0);
+      if (lock->get_status)
+	(*lock->get_status)(data->status_param, 0);
       statistic_increment(locks_immediate,&THR_LOCK_lock);
       goto end;
     }
@@ -626,16 +645,18 @@ thr_lock(THR_LOCK_DATA *data, THR_LOCK_OWNER *owner,
 	The following test will not work if the old lock was a
 	TL_WRITE_ALLOW_WRITE, TL_WRITE_ALLOW_READ or TL_WRITE_DELAYED in
 	the same thread, but this will never happen within MySQL.
+
+        The idea is to allow us to get a lock at once if we already have
+        a write lock or if there is no pending write locks and if all
+        write locks are of the same type and are either
+        TL_WRITE_ALLOW_WRITE or TL_WRITE_CONCURRENT_INSERT
       */
       if (thr_lock_owner_equal(data->owner, lock->write.data->owner) ||
-	  (lock_type == TL_WRITE_ALLOW_WRITE &&
-	   !lock->write_wait.data &&
-	   lock->write.data->type == TL_WRITE_ALLOW_WRITE))
+          (!lock->write_wait.data && lock_type == lock->write.data->type &&
+           (lock_type == TL_WRITE_ALLOW_WRITE ||
+            (lock_type == TL_WRITE_CONCURRENT_INSERT &&
+             lock->allow_multiple_concurrent_insert))))
       {
-	/*
-          We have already got a write lock or all locks are
-          TL_WRITE_ALLOW_WRITE
-        */
         DBUG_PRINT("info", ("write_wait.data: 0x%lx  old_type: %d",
                             (ulong) lock->write_wait.data,
                             lock->write.data->type));
@@ -644,8 +665,9 @@ thr_lock(THR_LOCK_DATA *data, THR_LOCK_OWNER *owner,
 	data->prev=lock->write.last;
 	lock->write.last= &data->next;
 	check_locks(lock,"second write lock",0);
-	if (data->lock->get_status)
-	  (*data->lock->get_status)(data->status_param, 0);
+	if (lock->get_status)
+	  (*lock->get_status)(data->status_param,
+                              lock_type == TL_WRITE_CONCURRENT_INSERT);
 	statistic_increment(locks_immediate,&THR_LOCK_lock);
 	goto end;
       }
@@ -678,8 +700,8 @@ thr_lock(THR_LOCK_DATA *data, THR_LOCK_OWNER *owner,
 	  (*lock->write.last)=data;		/* Add as current write lock */
 	  data->prev=lock->write.last;
 	  lock->write.last= &data->next;
-	  if (data->lock->get_status)
-	    (*data->lock->get_status)(data->status_param, concurrent_insert);
+	  if (lock->get_status)
+	    (*lock->get_status)(data->status_param, concurrent_insert);
 	  check_locks(lock,"only write lock",0);
 	  statistic_increment(locks_immediate,&THR_LOCK_lock);
 	  goto end;
@@ -809,7 +831,6 @@ static void wake_up_waiters(THR_LOCK *lock)
 {
   THR_LOCK_DATA *data;
   enum thr_lock_type lock_type;
-
   DBUG_ENTER("wake_up_waiters");
 
   if (!lock->write.data)			/* If no active write locks */
@@ -1372,8 +1393,8 @@ my_bool thr_upgrade_write_delay_lock(THR_LOCK_DATA *data)
   {
     if (!lock->read.data)			/* No read locks */
     {						/* We have the lock */
-      if (data->lock->get_status)
-	(*data->lock->get_status)(data->status_param, 0);
+      if (lock->get_status)
+	(*lock->get_status)(data->status_param, 0);
       pthread_mutex_unlock(&lock->mutex);
       DBUG_RETURN(0);
     }
@@ -1511,7 +1532,7 @@ struct st_test {
   enum thr_lock_type lock_type;
 };
 
-THR_LOCK locks[5];			/* 4 locks */
+THR_LOCK locks[6];			/* Number of locks +1 */
 
 struct st_test test_0[] = {{0,TL_READ}};	/* One lock */
 struct st_test test_1[] = {{0,TL_READ},{0,TL_WRITE}}; /* Read and write lock of lock 0 */
@@ -1531,9 +1552,20 @@ struct st_test test_14[] = {{0,TL_WRITE_CONCURRENT_INSERT},{1,TL_READ}};
 struct st_test test_15[] = {{0,TL_WRITE_ALLOW_WRITE},{1,TL_READ}};
 struct st_test test_16[] = {{0,TL_WRITE_ALLOW_WRITE},{1,TL_WRITE_ALLOW_WRITE}};
 
-struct st_test *tests[] = {test_0,test_1,test_2,test_3,test_4,test_5,test_6,
-			   test_7,test_8,test_9,test_10,test_11,test_12,
-			   test_13,test_14,test_15,test_16};
+struct st_test test_17[] = {{5,TL_WRITE_CONCURRENT_INSERT}};
+struct st_test test_18[] = {{5,TL_WRITE_CONCURRENT_INSERT}};
+struct st_test test_19[] = {{5,TL_READ}};
+struct st_test test_20[] = {{5,TL_READ_NO_INSERT}};
+struct st_test test_21[] = {{5,TL_WRITE}};
+
+
+struct st_test *tests[]=
+{
+  test_0, test_1, test_2, test_3, test_4, test_5, test_6, test_7, test_8,
+  test_9, test_10, test_11, test_12, test_13, test_14, test_15, test_16,
+  test_17, test_18, test_19, test_20, test_21
+};
+
 int lock_counts[]= {sizeof(test_0)/sizeof(struct st_test),
 		    sizeof(test_1)/sizeof(struct st_test),
 		    sizeof(test_2)/sizeof(struct st_test),
@@ -1550,7 +1582,12 @@ int lock_counts[]= {sizeof(test_0)/sizeof(struct st_test),
 		    sizeof(test_13)/sizeof(struct st_test),
 		    sizeof(test_14)/sizeof(struct st_test),
 		    sizeof(test_15)/sizeof(struct st_test),
-		    sizeof(test_16)/sizeof(struct st_test)
+		    sizeof(test_16)/sizeof(struct st_test),
+		    sizeof(test_17)/sizeof(struct st_test),
+		    sizeof(test_18)/sizeof(struct st_test),
+		    sizeof(test_19)/sizeof(struct st_test),
+		    sizeof(test_20)/sizeof(struct st_test),
+		    sizeof(test_21)/sizeof(struct st_test)
 };
 
 
@@ -1594,7 +1631,6 @@ static void *test_thread(void *arg)
 
   printf("Thread %s (%d) started\n",my_thread_name(),param); fflush(stdout);
 
-
   thr_lock_info_init(&lock_info);
   thr_lock_owner_init(&owner, &lock_info);
   for (i=0; i < lock_counts[param] ; i++)
@@ -1640,7 +1676,8 @@ int main(int argc __attribute__((unused)),char **argv __attribute__((unused)))
 {
   pthread_t tid;
   pthread_attr_t thr_attr;
-  int i,*param,error;
+  int *param,error;
+  uint i;
   MY_INIT(argv[0]);
   if (argc > 1 && argv[1][0] == '-' && argv[1][1] == '#')
     DBUG_PUSH(argv[1]+2);
@@ -1660,13 +1697,14 @@ int main(int argc __attribute__((unused)),char **argv __attribute__((unused)))
     exit(1);
   }
 
-  for (i=0 ; i < (int) array_elements(locks) ; i++)
+  for (i=0 ; i < array_elements(locks) ; i++)
   {
     thr_lock_init(locks+i);
     locks[i].check_status= test_check_status;
     locks[i].update_status=test_update_status;
     locks[i].copy_status=  test_copy_status;
     locks[i].get_status=   test_get_status;
+    locks[i].allow_multiple_concurrent_insert= 1;
   }
   if ((error=pthread_attr_init(&thr_attr)))
   {
@@ -1692,7 +1730,7 @@ int main(int argc __attribute__((unused)),char **argv __attribute__((unused)))
 #ifdef HAVE_THR_SETCONCURRENCY
   VOID(thr_setconcurrency(2));
 #endif
-  for (i=0 ; i < (int) array_elements(lock_counts) ; i++)
+  for (i=0 ; i < array_elements(lock_counts) ; i++)
   {
     param=(int*) malloc(sizeof(int));
     *param=i;
@@ -1724,7 +1762,7 @@ int main(int argc __attribute__((unused)),char **argv __attribute__((unused)))
   }
   if ((error=pthread_mutex_unlock(&LOCK_thread_count)))
     fprintf(stderr,"Got error: %d from pthread_mutex_unlock\n",error);
-  for (i=0 ; i < (int) array_elements(locks) ; i++)
+  for (i=0 ; i < array_elements(locks) ; i++)
     thr_lock_delete(locks+i);
 #ifdef EXTRA_DEBUG
   if (found_errors)
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index 0d5908ccc1e..f50a2d2ea1e 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -11179,6 +11179,16 @@ table_lock:
 lock_option:
           READ_SYM               { $$= TL_READ_NO_INSERT; }
         | WRITE_SYM              { $$= TL_WRITE_DEFAULT; }
+        | WRITE_SYM CONCURRENT
+          {
+#ifdef HAVE_QUERY_CACHE
+            if (Lex->sphead != 0)
+             $$= TL_WRITE_DEFAULT;
+           else
+#endif
+             $$= TL_WRITE_CONCURRENT_INSERT;
+          }
+
         | LOW_PRIORITY WRITE_SYM { $$= TL_WRITE_LOW_PRIORITY; }
         | READ_SYM LOCAL_SYM     { $$= TL_READ; }
         ;
diff --git a/storage/csv/ha_tina.cc b/storage/csv/ha_tina.cc
index 56d77c2b8b1..a4bd35776d5 100644
--- a/storage/csv/ha_tina.cc
+++ b/storage/csv/ha_tina.cc
@@ -715,7 +715,7 @@ const char **ha_tina::bas_ext() const
   for CSV engine. For more details see mysys/thr_lock.c
 */
 
-void tina_get_status(void* param, int concurrent_insert)
+void tina_get_status(void* param, my_bool concurrent_insert)
 {
   ha_tina *tina= (ha_tina*) param;
   tina->get_status();
diff --git a/storage/maria/ha_maria.cc b/storage/maria/ha_maria.cc
index 53837a5e027..248fc9a3702 100644
--- a/storage/maria/ha_maria.cc
+++ b/storage/maria/ha_maria.cc
@@ -951,7 +951,8 @@ int ha_maria::check(THD * thd, HA_CHECK_OPT * check_opt)
                                               0)))))
     return HA_ADMIN_ALREADY_DONE;
 
-  error= maria_chk_status(&param, file);                // Not fatal
+  maria_chk_init_for_check(&param, file);
+  (void) maria_chk_status(&param, file);                // Not fatal
   error= maria_chk_size(&param, file);
   if (!error)
     error|= maria_chk_del(&param, file, param.testflag);
@@ -2768,7 +2769,7 @@ static int ha_maria_init(void *p)
   maria_hton->flags= HTON_CAN_RECREATE | HTON_SUPPORT_LOG_TABLES;
   bzero(maria_log_pagecache, sizeof(*maria_log_pagecache));
   maria_tmpdir= &mysql_tmpdir_list;             /* For REDO */
-  res= maria_init() || ma_control_file_open(TRUE) ||
+  res= maria_init() || ma_control_file_open(TRUE, TRUE) ||
     !init_pagecache(maria_pagecache,
                     (size_t) pagecache_buffer_size, pagecache_division_limit,
                     pagecache_age_threshold, maria_block_size, 0) ||
diff --git a/storage/maria/ma_blockrec.c b/storage/maria/ma_blockrec.c
index 7ac1df772d5..1a522b51b43 100644
--- a/storage/maria/ma_blockrec.c
+++ b/storage/maria/ma_blockrec.c
@@ -345,11 +345,11 @@ static uchar *store_page_range(uchar *to, MARIA_BITMAP_BLOCK *block,
                                uint block_size, ulong length,
                                uint *tot_ranges);
 static size_t fill_insert_undo_parts(MARIA_HA *info, const uchar *record,
-                                     LEX_STRING *log_parts,
+                                     LEX_CUSTRING *log_parts,
                                      uint *log_parts_count);
 static size_t fill_update_undo_parts(MARIA_HA *info, const uchar *oldrec,
                                      const uchar *newrec,
-                                     LEX_STRING *log_parts,
+                                     LEX_CUSTRING *log_parts,
                                      uint *log_parts_count);
 
 /****************************************************************************
@@ -518,7 +518,8 @@ my_bool _ma_init_block_record(MARIA_HA *info)
                                          MYF(MY_WME))))
     goto err;
 
-  row->base_length= new_row->base_length= share->base_length;
+  info->row_base_length= share->base_length;
+  info->row_flag= share->base.default_row_flag;
 
   /*
     We need to reserve 'EXTRA_LENGTH_FIELDS' number of parts in
@@ -723,6 +724,7 @@ my_bool enough_free_entries(uchar *buff, uint block_size, uint wanted_entries)
    @brief Extend a record area to fit a given size block
 
    @fn extend_area_on_page()
+   @param info                  Handler if head page and 0 if tail page
    @param buff			Page buffer
    @param dir			Pointer to dir entry in buffer
    @param rownr			Row number we working on
@@ -753,7 +755,8 @@ my_bool enough_free_entries(uchar *buff, uint block_size, uint wanted_entries)
   @retval 1   error (wrong info in block)
 */
 
-static my_bool extend_area_on_page(uchar *buff, uchar *dir,
+static my_bool extend_area_on_page(MARIA_HA *info,
+                                   uchar *buff, uchar *dir,
                                    uint rownr, uint block_size,
                                    uint request_length,
                                    uint *empty_space, uint *ret_offset,
@@ -831,7 +834,9 @@ static my_bool extend_area_on_page(uchar *buff, uchar *dir,
         int2store(dir, rec_offset);
         /* Reset length, as this may be a deleted block */
         int2store(dir+2, 0);
-        _ma_compact_block_page(buff, block_size, rownr, 1);
+        _ma_compact_block_page(buff, block_size, rownr, 1,
+                               info ? info->trn->min_read_from: 0,
+                               info ? info->s->base.min_block_length : 0);
         rec_offset= uint2korr(dir);
         length=     uint2korr(dir+2);
         if (length < request_length)
@@ -934,7 +939,8 @@ static uint empty_space_on_page(uchar *buff, uint block_size)
 */
 
 static inline my_bool
-make_space_for_directory(uchar *buff, uint block_size, uint max_entry,
+make_space_for_directory(MARIA_HA *info,
+                         uchar *buff, uint block_size, uint max_entry,
                          uint count, uchar *first_dir, uint *empty_space,
                          uint *first_pos)
 {
@@ -952,7 +958,9 @@ make_space_for_directory(uchar *buff, uint block_size, uint max_entry,
     if ((uint) (first_dir - buff) < *first_pos + length_needed)
     {
       /* Create place for directory */
-      _ma_compact_block_page(buff, block_size, max_entry - 1, 0);
+      _ma_compact_block_page(buff, block_size, max_entry - 1, 0,
+                             info ? info->trn->min_read_from : 0,
+                             info ? info->s->base.min_block_length : 0);
       *first_pos= (uint2korr(first_dir) + uint2korr(first_dir + 2));
       *empty_space= uint2korr(buff + EMPTY_SPACE_OFFSET);
       if (*empty_space < length_needed)
@@ -982,6 +990,7 @@ make_space_for_directory(uchar *buff, uint block_size, uint max_entry,
 
   SYNOPSIS
   find_free_position()
+    info                Handler if head page and 0 otherwise
     buff                Page
     block_size          Size of page
     res_rownr           Store index to free position here
@@ -1013,7 +1022,8 @@ make_space_for_directory(uchar *buff, uint block_size, uint max_entry,
     #      Pointer to directory entry on page
 */
 
-static uchar *find_free_position(uchar *buff, uint block_size, uint *res_rownr,
+static uchar *find_free_position(MARIA_HA *info,
+                                 uchar *buff, uint block_size, uint *res_rownr,
                                  uint *res_length, uint *empty_space)
 {
   uint max_entry, free_entry;
@@ -1062,7 +1072,7 @@ static uchar *find_free_position(uchar *buff, uint block_size, uint *res_rownr,
   if (max_entry == MAX_ROWS_PER_PAGE)
     DBUG_RETURN(0);
 
-  if (make_space_for_directory(buff, block_size, max_entry, 1,
+  if (make_space_for_directory(info, buff, block_size, max_entry, 1,
                                first_dir, empty_space, &first_pos))
     DBUG_RETURN(0);
 
@@ -1082,6 +1092,7 @@ static uchar *find_free_position(uchar *buff, uint block_size, uint *res_rownr,
    @brief Enlarge page directory to hold more entries
 
    @fn extend_directory()
+   @param info          Handler if head page and 0 otherwise
    @param buff		Page buffer
    @param block_size	Block size
    @param max_entry	Number of directory entries on page
@@ -1100,8 +1111,9 @@ static uchar *find_free_position(uchar *buff, uint block_size, uint *res_rownr,
    @retval 1  error (No data on page, fatal error)
 */
 
-static my_bool extend_directory(uchar *buff, uint block_size, uint max_entry,
-                                uint new_entry, uint *empty_space)
+static my_bool extend_directory(MARIA_HA *info, uchar *buff, uint block_size,
+                                uint max_entry, uint new_entry,
+                                uint *empty_space)
 {
   uint length, first_pos;
   uchar *dir, *first_dir;
@@ -1114,7 +1126,7 @@ static my_bool extend_directory(uchar *buff, uint block_size, uint max_entry,
   */
   first_dir= dir_entry_pos(buff, block_size, max_entry) + DIR_ENTRY_SIZE;
 
-  if (make_space_for_directory(buff, block_size, max_entry,
+  if (make_space_for_directory(info, buff, block_size, max_entry,
                                new_entry - max_entry + 1,
                                first_dir, empty_space, &first_pos))
     DBUG_RETURN(1);
@@ -1303,7 +1315,7 @@ static void calc_record_size(MARIA_HA *info, const uchar *record,
   }
   row->field_lengths_length= (uint) (field_length_data - row->field_lengths);
   /*
-    - row->base_length is base information we must have on a page in first
+    - info->row_base_length is base information we must have on a page in first
       extent:
       - flag byte (1) + is_nulls_extended (0 | 1) + null_bytes + pack_bytes +
         table_checksum (0 | 1)
@@ -1315,7 +1327,7 @@ static void calc_record_size(MARIA_HA *info, const uchar *record,
     - head_length is the amount of data for the head page
      (ie, all fields except blobs)
   */
-  row->min_length=   (row->base_length +
+  row->min_length=   (info->row_base_length +
                       (share->base.max_field_lengths ?
                        size_to_store_key_length(row->field_lengths_length) :
                        0));
@@ -1340,9 +1352,6 @@ static void calc_record_size(MARIA_HA *info, const uchar *record,
     Move up all rows to start of page.
     Move blocks that are directly after each other with one memmove.
 
-  TODO LATER
-    Remove TRANSID from rows that are visible to all transactions
-
   SYNOPSIS
     _ma_compact_block_page()
     buff                Page to compact
@@ -1350,14 +1359,17 @@ static void calc_record_size(MARIA_HA *info, const uchar *record,
     rownr               Put empty data after this row
     extend_block	If 1, extend the block at 'rownr' to cover the
                         whole block.
+    min_read_from       If <> 0, remove all trid's that are less than this
 */
 
 
 void _ma_compact_block_page(uchar *buff, uint block_size, uint rownr,
-                            my_bool extend_block)
+                            my_bool extend_block, TrID min_read_from,
+                            uint min_row_length)
 {
   uint max_entry= (uint) buff[DIR_COUNT_OFFSET];
   uint page_pos, next_free_pos, start_of_found_block, diff, end_of_found_block;
+  uint freed_size= 0;
   uchar *dir, *end;
   DBUG_ENTER("_ma_compact_block_page");
   DBUG_PRINT("enter", ("rownr: %u", rownr));
@@ -1379,13 +1391,29 @@ void _ma_compact_block_page(uchar *buff, uint block_size, uint rownr,
       uint row_length= uint2korr(end + 2);
       DBUG_ASSERT(offset >= page_pos);
       DBUG_ASSERT(buff + offset + row_length <= dir);
+      DBUG_ASSERT(row_length >= min_row_length || row_length == 0);
+
+      /* Row length can be zero if row is to be deleted */
+      if (min_read_from && row_length && (buff[offset] & ROW_FLAG_TRANSID))
+      {
+        TrID transid= transid_korr(buff+offset+1);
+        if (transid < min_read_from)
+        {
+          /* Remove transid from row by moving the start point of the row up */
+          buff[offset + TRANSID_SIZE]= buff[offset] & ~ROW_FLAG_TRANSID;
+          offset+= TRANSID_SIZE;
+          freed_size+= TRANSID_SIZE;
+          row_length-= TRANSID_SIZE;
+          int2store(end+2, row_length);
+        }
+      }
 
       if (offset != next_free_pos)
       {
         uint length= (next_free_pos - start_of_found_block);
         /*
-           There was empty space before this and prev block
-           Check if we have to move previous block up to page start
+          There was empty space before this and prev block
+          Check if we have to move previous block up to page start
         */
         if (page_pos != start_of_found_block)
         {
@@ -1399,6 +1427,27 @@ void _ma_compact_block_page(uchar *buff, uint block_size, uint rownr,
       }
       int2store(end, offset - diff);            /* correct current pos */
       next_free_pos= offset + row_length;
+
+      if (unlikely(row_length < min_row_length) && row_length)
+      {
+        /*
+          This can only happen in the case we compacted transid and
+          the row become 'too short'
+
+          Move the current row down to it's right place and extend it
+          with 0.
+        */
+        DBUG_ASSERT(page_pos != start_of_found_block);
+        uint row_diff= min_row_length - row_length;
+        uint length= (next_free_pos - start_of_found_block);
+        bmove(buff + page_pos, buff + start_of_found_block, length);
+        bzero(buff+ page_pos + length, row_diff);
+        page_pos+= min_row_length;
+        int2store(end+2, min_row_length);
+        freed_size-= row_diff;
+        next_free_pos= start_of_found_block= page_pos;
+        diff= 0;
+      }
     }
   }
   if (page_pos != start_of_found_block)
@@ -1419,11 +1468,53 @@ void _ma_compact_block_page(uchar *buff, uint block_size, uint rownr,
     for (dir= buff + end_of_found_block ; dir <= end ; dir+= DIR_ENTRY_SIZE)
     {
       uint offset= uint2korr(dir);
-      uint row_length= uint2korr(dir + 2);
-      uint row_end= offset + row_length;
+      uint row_length;
+      uint row_end;
       if (!offset)
         continue;
-      DBUG_ASSERT(offset >= start_of_found_block && row_end <= next_free_pos);
+      row_length= uint2korr(dir + 2);
+      row_end= offset + row_length;
+      DBUG_ASSERT(offset >= start_of_found_block &&
+                  row_end <= next_free_pos && row_length >= min_row_length);
+
+      if (min_read_from && (buff[offset] & ROW_FLAG_TRANSID))
+      {
+        TrID transid= transid_korr(buff + offset+1);
+        if (transid < min_read_from)
+        {
+          /* Remove transid from row */
+          buff[offset + TRANSID_SIZE]= buff[offset] & ~ROW_FLAG_TRANSID;
+          offset+= TRANSID_SIZE;
+          row_length-= TRANSID_SIZE;
+          int2store(dir+2, row_length);
+        }
+        if (unlikely(row_length < min_row_length))
+        {
+          /*
+            This can only happen in the case we compacted transid and
+            the row become 'too short'
+          */
+          uint row_diff= min_row_length - row_length;
+          if (next_free_pos < row_end + row_diff)
+          {
+            /*
+              Not enough space for extending next block with enough
+              end 0's. Move current data down to get place for them
+            */
+            uint move_down= row_diff - (next_free_pos - row_end);
+            bmove(buff + offset - move_down, buff + offset, row_length);
+            offset-= move_down;
+          }
+          /*
+            Extend the next block with 0, which will be part of current
+            row when the blocks are joined together later
+          */
+          bzero(buff + next_free_pos - row_diff, row_diff);
+          next_free_pos-= row_diff;
+          int2store(dir+2, min_row_length);
+        }
+        row_end= offset + row_length;
+      }
 
       if (row_end != next_free_pos)
       {
@@ -1441,6 +1532,7 @@ void _ma_compact_block_page(uchar *buff, uint block_size, uint rownr,
       int2store(dir, offset + diff);            /* correct current pos */
       next_free_pos= offset;
     }
+
     if (page_pos != end_of_found_block)
     {
       uint length= (end_of_found_block - next_free_pos);
@@ -1455,16 +1547,15 @@ void _ma_compact_block_page(uchar *buff, uint block_size, uint rownr,
   {
     if (extend_block)
     {
-      /* Extend last block cover whole page */
+      /* Extend last block to cover whole page */
       uint length= ((uint) (dir - buff) - start_of_found_block);
       int2store(dir+2, length);
     }
     else
     {
-      /*
-        TODO:
-        Update (buff + EMPTY_SPACE_OFFSET) if we remove transid from rows
-      */
+      /* Add length gained from freed transaction id's to this page */
+      uint length= uint2korr(buff+ EMPTY_SPACE_OFFSET) + freed_size;
+      int2store(buff + EMPTY_SPACE_OFFSET, length);
     }
     buff[PAGE_TYPE_OFFSET]&= ~(uchar) PAGE_CAN_BE_COMPACTED;
   }
@@ -1590,7 +1681,8 @@ static my_bool get_head_or_tail_page(MARIA_HA *info,
       goto crashed;
 
     DBUG_ASSERT((res->buff[PAGE_TYPE_OFFSET] & PAGE_TYPE_MASK) == page_type);
-    if (!(dir= find_free_position(res->buff, block_size, &res->rownr,
+    if (!(dir= find_free_position(page_type == HEAD_PAGE ? info : 0,
+                                  res->buff, block_size, &res->rownr,
                                   &res->length, &res->empty_space)))
       goto crashed;
 
@@ -1598,7 +1690,11 @@ static my_bool get_head_or_tail_page(MARIA_HA *info,
     {
       if (res->empty_space + res->length >= length)
       {
-        _ma_compact_block_page(res->buff, block_size, res->rownr, 1);
+        _ma_compact_block_page(res->buff, block_size, res->rownr, 1,
+                               page_type == HEAD_PAGE ?
+                               info->trn->min_read_from : 0,
+                               page_type == HEAD_PAGE ?
+                               share->base.min_block_length : 0);
         /* All empty space are now after current position */
         dir= dir_entry_pos(res->buff, block_size, res->rownr);
         res->length= res->empty_space= uint2korr(dir+2);
@@ -1685,8 +1781,8 @@ static my_bool get_rowpos_in_head_or_tail_page(MARIA_HA *info,
   max_entry= (uint) buff[DIR_COUNT_OFFSET];
   if (max_entry <= rownr)
   {
-    if (extend_directory(buff, block_size, max_entry, rownr,
-                         &res->empty_space))
+    if (extend_directory(page_type == HEAD_PAGE ? info : 0, buff, block_size,
+                         max_entry, rownr, &res->empty_space))
       goto err;
   }
 
@@ -1701,7 +1797,8 @@ static my_bool get_rowpos_in_head_or_tail_page(MARIA_HA *info,
   }
 #endif
 
-  if (extend_area_on_page(buff, dir, rownr, block_size, length,
+  if (extend_area_on_page(page_type == HEAD_PAGE ? info : 0, buff, dir,
+                          rownr, block_size, length,
                           &res->empty_space, &rec_offset, &max_length))
     goto err;
 
@@ -2407,7 +2504,7 @@ static my_bool write_block_record(MARIA_HA *info,
   end_of_data= data + row_pos->length;
 
   /* Write header */
-  flag= share->base.default_row_flag;
+  flag= info->row_flag;
   row_extents_in_use= 0;
   if (unlikely(row->total_length > row_pos->length))
   {
@@ -2420,6 +2517,12 @@ static my_bool write_block_record(MARIA_HA *info,
   }
   /* For now we have only a minimum header */
   *data++= (uchar) flag;
+  if (flag & ROW_FLAG_TRANSID)
+  {
+    transid_store(data, info->trn->trid);
+    data+= TRANSID_SIZE;
+  }
+
   if (unlikely(flag & ROW_FLAG_NULLS_EXTENDED))
     *data++= (uchar) (share->base.null_bytes -
                       share->base.original_null_bytes);
@@ -3048,7 +3151,7 @@ static my_bool write_block_record(MARIA_HA *info,
   lsn= LSN_IMPOSSIBLE;
   if (share->now_transactional)
   {
-    LEX_STRING *log_array= info->log_row_parts;
+    LEX_CUSTRING *log_array= info->log_row_parts;
 
     if (undo_lsn != LSN_ERROR)
     {
@@ -3103,7 +3206,7 @@ static my_bool write_block_record(MARIA_HA *info,
                                   log_array[TRANSLOG_INTERNAL_PARTS +
                                             0].length,
                                   TRANSLOG_INTERNAL_PARTS + 1,
-                                  (LEX_CUSTRING *)log_array,
+                                  log_array,
                                   log_data + LSN_STORE_SIZE, &checksum_delta))
           goto disk_err;
       }
@@ -3141,7 +3244,7 @@ static my_bool write_block_record(MARIA_HA *info,
                                    row_length),
                                   TRANSLOG_INTERNAL_PARTS + 2 +
                                   row_parts_count,
-                                  (LEX_CUSTRING *)log_array,
+                                  log_array,
                                   log_data + LSN_STORE_SIZE,
                                   &checksum_delta))
           goto disk_err;
@@ -3477,7 +3580,7 @@ static my_bool _ma_update_block_record2(MARIA_HA *info,
     */
     block.org_bitmap_value= _ma_free_size_to_head_pattern(&share->bitmap,
                                                           org_empty_size);
-    if (extend_area_on_page(buff, dir, rownr, block_size,
+    if (extend_area_on_page(info, buff, dir, rownr, block_size,
                             new_row->total_length, &org_empty_size,
                             &rec_offset, &length))
       goto err;
@@ -3528,7 +3631,9 @@ static my_bool _ma_update_block_record2(MARIA_HA *info,
        (new_row->total_length <= head_length &&
         org_empty_size + head_length >= new_row->total_length)))
   {
-    _ma_compact_block_page(buff, block_size, rownr, 1);
+    _ma_compact_block_page(buff, block_size, rownr, 1,
+                           info->trn->min_read_from,
+                           share->base.min_block_length);
     org_empty_size= 0;
     head_length= uint2korr(dir + 2);
   }
@@ -3626,7 +3731,7 @@ static my_bool _ma_update_at_original_place(MARIA_HA *info,
     of the row
   */
   empty_size= org_empty_size;
-  if (extend_area_on_page(buff, dir, rownr, block_size,
+  if (extend_area_on_page(info, buff, dir, rownr, block_size,
                           length_on_head_page, &empty_size,
                           &rec_offset, &length))
     goto err;
@@ -3657,8 +3762,13 @@ static my_bool _ma_update_at_original_place(MARIA_HA *info,
               _ma_bitmap_get_page_bits(info, &info->s->bitmap, page));
   block->used|= BLOCKUSED_USE_ORG_BITMAP;
 
+  /*
+    We have to use <= below as the new_row may be smaller than the original
+    row as the new row doesn't have transaction id
+  */
+
   DBUG_ASSERT(blocks->count > 1 ||
-              max(new_row->total_length, share->base.min_block_length) ==
+              max(new_row->total_length, share->base.min_block_length) <=
               length_on_head_page);
 
   if ((res= write_block_record(info, oldrec, record, new_row, blocks,
@@ -4032,7 +4142,7 @@ my_bool _ma_delete_block_record(MARIA_HA *info, const uchar *record)
                                                    0].length + row_length +
                                extents_length),
                               TRANSLOG_INTERNAL_PARTS + 2 + row_parts_count,
-                              (LEX_CUSTRING *)info->log_row_parts,
+                              info->log_row_parts,
                               log_data + LSN_STORE_SIZE,
                               &checksum_delta))
       goto err;
@@ -4364,6 +4474,15 @@ int _ma_read_block_record2(MARIA_HA *info, uchar *record,
   cur_row->full_page_count= cur_row->tail_count= 0;
   cur_row->blob_length= 0;
 
+  if (flag & ROW_FLAG_TRANSID)
+  {
+    cur_row->trid= transid_korr(data+1);
+    if (!info->trn)
+      DBUG_RETURN(my_errno= HA_ERR_WRONG_IN_RECORD);     /* File crashed */
+    if (!trnman_can_read_from(info->trn, cur_row->trid))
+      DBUG_RETURN(my_errno= HA_ERR_ROW_NOT_VISIBLE);
+  }
+
   /* Skip trans header (for now, until we have MVCC csupport) */
   data+= total_header_size[(flag & PRECALC_HEADER_BITMASK)];
   if (flag & ROW_FLAG_NULLS_EXTENDED)
@@ -4987,10 +5106,11 @@ int _ma_scan_block_record(MARIA_HA *info, uchar *record,
 
 restart_record_read:
   /* Find next row in current page */
-  if (likely(record_pos < info->scan.number_of_rows))
+  while (likely(record_pos < info->scan.number_of_rows))
   {
     uint length, offset;
     uchar *data, *end_of_data;
+    int error;
 
     while (!(offset= uint2korr(info->scan.dir)))
     {
@@ -5020,7 +5140,10 @@ restart_record_read:
     }
 #endif
     DBUG_PRINT("info", ("rowid: %lu", (ulong) info->cur_row.lastpos));
-    DBUG_RETURN(_ma_read_block_record2(info, record, data, end_of_data));
+    error= _ma_read_block_record2(info, record, data, end_of_data);
+    if (error != HA_ERR_ROW_NOT_VISIBLE)
+      DBUG_RETURN(error);
+    record_pos++;
   }
 
   /* Find next head page in current bitmap */
@@ -5259,7 +5382,7 @@ static ulong ma_get_length(uchar **packet)
 */
 
 static size_t fill_insert_undo_parts(MARIA_HA *info, const uchar *record,
-                                     LEX_STRING *log_parts,
+                                     LEX_CUSTRING *log_parts,
                                      uint *log_parts_count)
 {
   MARIA_SHARE *share= info->s;
@@ -5267,14 +5390,13 @@ static size_t fill_insert_undo_parts(MARIA_HA *info, const uchar *record,
   uchar *field_lengths= info->cur_row.field_lengths;
   size_t row_length;
   MARIA_ROW *cur_row= &info->cur_row;
-  LEX_STRING *start_log_parts;
+  LEX_CUSTRING *start_log_parts;
   DBUG_ENTER("fill_insert_undo_parts");
 
   start_log_parts= log_parts;
 
   /* Store null bits */
-  /* We cast "const uchar*" to char* but won't change its pointed content */
-  log_parts->str=      (char*)record;
+  log_parts->str=      record;
   log_parts->length=   share->base.null_bytes;
   row_length=          log_parts->length;
   log_parts++;
@@ -5439,7 +5561,7 @@ static size_t fill_insert_undo_parts(MARIA_HA *info, const uchar *record,
 
 static size_t fill_update_undo_parts(MARIA_HA *info, const uchar *oldrec,
                                      const uchar *newrec,
-                                     LEX_STRING *log_parts,
+                                     LEX_CUSTRING *log_parts,
                                      uint *log_parts_count)
 {
   MARIA_SHARE *share= info->s;
@@ -5450,7 +5572,7 @@ static size_t fill_update_undo_parts(MARIA_HA *info, const uchar *oldrec,
   uchar *new_field_lengths= new_row->field_lengths;
   size_t row_length= 0;
   uint field_lengths;
-  LEX_STRING *start_log_parts;
+  LEX_CUSTRING *start_log_parts;
   my_bool new_column_is_empty;
   DBUG_ENTER("fill_update_undo_parts");
 
@@ -5467,8 +5589,7 @@ static size_t fill_update_undo_parts(MARIA_HA *info, const uchar *oldrec,
   {
     /* Store changed null bits */
     *field_data++=       (uchar) 255;           /* Special case */
-    /* We cast "const uchar*" to char* but won't change its pointed content */
-    log_parts->str=      (char*) oldrec;
+    log_parts->str=      oldrec;
     log_parts->length=   share->base.null_bytes;
     row_length=          log_parts->length;
     log_parts++;
@@ -5607,8 +5728,7 @@ static size_t fill_update_undo_parts(MARIA_HA *info, const uchar *oldrec,
                           (start_field_data -
                            ma_calc_length_for_store_length(field_lengths)));
   ma_store_length((uchar *) start_log_parts->str, field_lengths);
-  start_log_parts->length= (size_t) ((char*) field_data -
-                                     start_log_parts->str);
+  start_log_parts->length= (size_t) (field_data - start_log_parts->str);
   row_length+= start_log_parts->length;
   DBUG_RETURN(row_length);
 }
@@ -5924,8 +6044,8 @@ uint _ma_apply_redo_insert_row_head_or_tail(MARIA_HA *info, LSN lsn,
         goto crashed_file;
       make_empty_page(info, buff, page_type, 0);
       empty_space= block_size - PAGE_HEADER_SIZE - PAGE_SUFFIX_SIZE;
-      (void) extend_directory(buff, block_size, 0, rownr,
-                              &empty_space);
+      (void) extend_directory(page_type == HEAD_PAGE ? info: 0, buff,
+                              block_size, 0, rownr, &empty_space);
       rec_offset= PAGE_HEADER_SIZE;
       dir= dir_entry_pos(buff, block_size, rownr);
       empty_space+= uint2korr(dir+2);
@@ -5942,11 +6062,12 @@ uint _ma_apply_redo_insert_row_head_or_tail(MARIA_HA *info, LSN lsn,
       if (max_entry <= rownr)
       {
         /* Add directory entry first in directory and data last on page */
-        if (extend_directory(buff, block_size, max_entry, rownr,
-                             &empty_space))
+        if (extend_directory(page_type == HEAD_PAGE ? info : 0, buff,
+                             block_size, max_entry, rownr, &empty_space))
           goto crashed_file;
       }
-      if (extend_area_on_page(buff, dir, rownr, block_size,
+      if (extend_area_on_page(page_type == HEAD_PAGE ? info : 0, buff,
+                              dir, rownr, block_size,
                               (uint) data_length, &empty_space,
                               &rec_offset, &length))
         goto crashed_file;
@@ -6491,7 +6612,7 @@ err:
 /** Execute undo of a row delete (insert the row back where it was) */
 
 my_bool _ma_apply_undo_row_delete(MARIA_HA *info, LSN undo_lsn,
-                                  uchar *header, size_t header_length
+                                  const uchar *header, size_t header_length
                                   __attribute__((unused)))
 {
   MARIA_SHARE *share= info->s;
@@ -6549,16 +6670,17 @@ my_bool _ma_apply_undo_row_delete(MARIA_HA *info, LSN undo_lsn,
 
   null_bits= header;
   header+= share->base.null_bytes;
-  row.empty_bits= header;
+  /* This will not be changed */
+  row.empty_bits= (uchar*) header;
   header+= share->base.pack_bytes;
   if (share->base.max_field_lengths)
   {
     row.field_lengths_length= uint2korr(header);
-    row.field_lengths= header + 2 ;
+    row.field_lengths= (uchar*) header + 2 ;
     header+= 2 + row.field_lengths_length;
   }
   if (share->base.blobs)
-    row.blob_length= ma_get_length(&header);
+    row.blob_length= ma_get_length((uchar**) &header);
 
   /* We need to build up a record (without blobs) in rec_buff */
   if (!(record= my_malloc(share->base.reclength, MYF(MY_WME))))
@@ -6669,7 +6791,7 @@ my_bool _ma_apply_undo_row_delete(MARIA_HA *info, LSN undo_lsn,
       DBUG_ASSERT(0);
     }
   }
-  row.head_length= (row.base_length +
+  row.head_length= (info->row_base_length +
                     share->base.fixed_not_null_fields_length +
                     row.field_lengths_length +
                     size_to_store_key_length(row.field_lengths_length) +
@@ -6734,7 +6856,8 @@ err:
 */
 
 my_bool _ma_apply_undo_row_update(MARIA_HA *info, LSN undo_lsn,
-                                  uchar *header, size_t header_length
+                                  const uchar *header,
+                                  size_t header_length
                                   __attribute__((unused)))
 {
   MARIA_SHARE *share= info->s;
@@ -6773,8 +6896,8 @@ my_bool _ma_apply_undo_row_update(MARIA_HA *info, LSN undo_lsn,
     Set header to point to old field values, generated by
     fill_update_undo_parts()
   */
-  field_length_header= ma_get_length(&header);
-  field_length_data= header;
+  field_length_header= ma_get_length((uchar**) &header);
+  field_length_data= (uchar*) header;
   header+= field_length_header;
   field_length_data_end= header;
 
@@ -6940,3 +7063,88 @@ maria_page_get_lsn(uchar *page,
 #endif
   return lsn_korr(page);
 }
+
+
+/*****************************************************************************
+ Lock handling for concurrent insert
+*****************************************************************************/
+
+/*
+  Create a copy of the current status for the table
+
+  SYNOPSIS
+    _ma_get_status()
+    param		Pointer to Myisam handler
+    concurrent_insert	Set to 1 if we are going to do concurrent inserts
+			(THR_WRITE_CONCURRENT_INSERT was used)
+*/
+
+void _ma_block_get_status(void* param, my_bool concurrent_insert)
+{
+  MARIA_HA *info=(MARIA_HA*) param;
+  DBUG_ENTER("_ma_block_get_status");
+  DBUG_PRINT("info", ("concurrent_insert %d", concurrent_insert));
+  info->row_base_length= info->s->base_length;
+  info->row_flag= info->s->base.default_row_flag;
+  if (concurrent_insert)
+  {
+    info->row_flag|= ROW_FLAG_TRANSID;
+    info->row_base_length+= TRANSID_SIZE;
+  }
+  DBUG_VOID_RETURN;
+}
+
+
+void _ma_block_update_status(void *param __attribute__((unused)))
+{
+}
+
+void _ma_block_restore_status(void *param __attribute__((unused)))
+{
+}
+
+
+/**
+  Check if should allow concurrent inserts
+
+  @return
+  @retval 0  ok to use concurrent inserts
+  @retval 1  not ok
+*/
+
+my_bool _ma_block_check_status(void *param __attribute__((unused)))
+{
+  return (my_bool) 0;
+}
+
+
+/**
+  Enable/disable versioning
+*/
+
+void maria_versioning(MARIA_HA *info, my_bool versioning)
+{
+  /* For now, this is a hack */
+  _ma_block_get_status((void*) info, versioning);
+}
+
+
+/**
+  Enable reading of all rows, ignoring versioning
+
+  @note
+    This is mainly useful in single user applications, like maria_pack,
+    where we want to be able to read all rows without having to read the
+    transaction id from the control file
+*/
+
+void maria_ignore_trids(MARIA_HA *info)
+{
+  if (info->s->base.born_transactional)
+  {
+    if (!info->trn)
+      info->trn= &dummy_transaction_object;
+    /* Ignore transaction id when row is read */
+    info->trn->min_read_from= ~(TrID) 0;
+  }
+}
diff --git a/storage/maria/ma_blockrec.h b/storage/maria/ma_blockrec.h
index 05c5b99e8d0..36d61852086 100644
--- a/storage/maria/ma_blockrec.h
+++ b/storage/maria/ma_blockrec.h
@@ -63,14 +63,13 @@
 
 #define PAGE_TYPE_MASK 7
 enum en_page_type { UNALLOCATED_PAGE, HEAD_PAGE, TAIL_PAGE, BLOB_PAGE, MAX_PAGE_TYPE };
+#define PAGE_CAN_BE_COMPACTED   128             /* Bit in PAGE_TYPE */
 
 #define PAGE_TYPE_OFFSET        LSN_SIZE
 #define DIR_COUNT_OFFSET        (LSN_SIZE+PAGE_TYPE_SIZE)
 #define DIR_FREE_OFFSET         (DIR_COUNT_OFFSET+DIR_COUNT_SIZE)
 #define EMPTY_SPACE_OFFSET      (DIR_FREE_OFFSET+DIR_FREE_SIZE)
 
-#define PAGE_CAN_BE_COMPACTED   128             /* Bit in PAGE_TYPE */
-
 /* Bits used for flag uchar (one byte, first in record) */
 #define ROW_FLAG_TRANSID                1
 #define ROW_FLAG_VER_PTR                2
@@ -174,7 +173,8 @@ my_bool _ma_write_abort_block_record(MARIA_HA *info);
 my_bool _ma_compare_block_record(register MARIA_HA *info,
                                  register const uchar *record);
 void    _ma_compact_block_page(uchar *buff, uint block_size, uint rownr,
-                               my_bool extend_block);
+                               my_bool extend_block, TrID min_read_from,
+                               uint min_row_length);
 TRANSLOG_ADDRESS
 maria_page_get_lsn(uchar *page, pgcache_page_no_t page_no, uchar* data_ptr);
 
@@ -243,9 +243,9 @@ my_bool _ma_apply_redo_bitmap_new_page(MARIA_HA *info, LSN lsn,
 my_bool _ma_apply_undo_row_insert(MARIA_HA *info, LSN undo_lsn,
                                   const uchar *header);
 my_bool _ma_apply_undo_row_delete(MARIA_HA *info, LSN undo_lsn,
-                                  uchar *header, size_t length);
+                                  const uchar *header, size_t length);
 my_bool _ma_apply_undo_row_update(MARIA_HA *info, LSN undo_lsn,
-                                  uchar *header, size_t length);
+                                  const uchar *header, size_t length);
 my_bool _ma_apply_undo_bulk_insert(MARIA_HA *info, LSN undo_lsn);
 
 my_bool write_hook_for_redo(enum translog_record_type type,
@@ -272,3 +272,7 @@ my_bool write_hook_for_undo_bulk_insert(enum translog_record_type type,
 my_bool write_hook_for_file_id(enum translog_record_type type,
                                TRN *trn, MARIA_HA *tbl_info, LSN *lsn,
                                void *hook_arg);
+void _ma_block_get_status(void* param, my_bool concurrent_insert);
+void _ma_block_update_status(void *param);
+void _ma_block_restore_status(void *param);
+my_bool _ma_block_check_status(void *param);
diff --git a/storage/maria/ma_check.c b/storage/maria/ma_check.c
index d1d66f5b538..7507a5c9c31 100644
--- a/storage/maria/ma_check.c
+++ b/storage/maria/ma_check.c
@@ -99,8 +99,11 @@ static my_bool create_new_data_handle(MARIA_SORT_PARAM *param, File new_file);
 static my_bool _ma_flush_table_files_before_swap(HA_CHECK *param,
                                                  MARIA_HA *info);
 static TrID max_trid_in_system(void);
+static void _ma_check_print_not_visible_error(HA_CHECK *param, TrID used_trid);
 
 
+/* Initialize check param with default values */
+
 void maria_chk_init(HA_CHECK *param)
 {
   bzero((uchar*) param,sizeof(*param));
@@ -121,9 +124,30 @@ void maria_chk_init(HA_CHECK *param)
   param->stats_method= MI_STATS_METHOD_NULLS_NOT_EQUAL;
 }
 
+
+/* Initialize check param and maria handler for check of table */
+
+void maria_chk_init_for_check(HA_CHECK *param, MARIA_HA *info)
+{
+  param->not_visible_rows_found= 0;
+  param->max_found_trid= 0;
+
+  /*
+    Set up transaction handler so that we can see all rows. When rows is read
+    we will check the found id against param->max_tried
+  */
+  if (!ma_control_file_inited())
+    param->max_trid= 0;                 /* Give warning for first trid found */
+  else
+    param->max_trid= max_trid_in_system();
+
+  maria_ignore_trids(info);
+}
+
+
 	/* Check the status flags for the table */
 
-int maria_chk_status(HA_CHECK *param, register MARIA_HA *info)
+int maria_chk_status(HA_CHECK *param, MARIA_HA *info)
 {
   MARIA_SHARE *share= info->s;
 
@@ -1600,19 +1624,22 @@ static my_bool check_head_page(HA_CHECK *param, MARIA_HA *info, uchar *record,
     if (length < share->base.min_block_length)
     {
       _ma_check_print_error(param,
-                            "Page %9s:  Row %3u is too short (%d bytes)",
-                            llstr(page, llbuff), row, length);
+                            "Page %9s:  Row %3u is too short "
+                            "(%d of min %d bytes)",
+                            llstr(page, llbuff), row, length,
+                            (uint) share->base.min_block_length);
       DBUG_RETURN(1);
     }
     flag= (uint) (uchar) page_buff[pos];
     if (flag & ~(ROW_FLAG_ALL))
       _ma_check_print_error(param,
-                            "Page %9s: Row %3u has wrong flag: %d",
+                            "Page %9s: Row %3u has wrong flag: %u",
                             llstr(page, llbuff), row, flag);
 
     DBUG_PRINT("info", ("rowid: %s  page: %lu  row: %u",
                         llstr(ma_recordpos(page, row), llbuff),
                         (ulong) page, row));
+    info->cur_row.trid= 0;
     if (_ma_read_block_record2(info, record, page_buff+pos,
                                page_buff+pos+length))
     {
@@ -1623,6 +1650,10 @@ static my_bool check_head_page(HA_CHECK *param, MARIA_HA *info, uchar *record,
         DBUG_RETURN(1);
       continue;
     }
+    set_if_bigger(param->max_found_trid, info->cur_row.trid);
+    if (info->cur_row.trid > param->max_trid)
+      _ma_check_print_not_visible_error(param, info->cur_row.trid);
+
     if (share->calc_checksum)
     {
       ha_checksum checksum= (*share->calc_checksum)(info, record);
@@ -2070,6 +2101,11 @@ int maria_chk_data_link(HA_CHECK *param, MARIA_HA *info, my_bool extend)
            llstr(param->empty, llbuff),llstr(param->link_used, llbuff2));
     if (param->lost)
       printf("Lost space:   %12s", llstr(param->lost, llbuff));
+    if (param->max_found_trid)
+    {
+      printf("Max trans. id: %11s\n",
+             llstr(param->max_found_trid, llbuff));
+    }
   }
   my_free((uchar*) record,MYF(0));
   DBUG_RETURN (error);
@@ -2196,6 +2232,16 @@ static int initialize_variables_for_repair(HA_CHECK *param,
                     share->base.min_block_length);
     sort_info->max_records= (ha_rows) (sort_info->filelength / rec_length);
   }
+
+  /* Set up transaction handler so that we can see all rows */
+  if (!ma_control_file_inited())
+    param->max_trid= 0;                 /* Give warning for first trid found */
+  else
+    param->max_trid= max_trid_in_system();
+
+  maria_ignore_trids(info);
+  /* Don't write transid's during repair */
+  maria_versioning(info, 0);
   return 0;
 }
 
@@ -3143,7 +3189,7 @@ static my_bool maria_zerofill_data(HA_CHECK *param, MARIA_HA *info,
        pos+= block_size, page++)
   {
     uchar *buff;
-    uint page_type;
+    enum en_page_type page_type;
 
     /* Ignore bitmap pages */
     if ((page % share->bitmap.pages_covered) == 0)
@@ -3159,8 +3205,8 @@ static my_bool maria_zerofill_data(HA_CHECK *param, MARIA_HA *info,
                             llstr(pos, llbuff), my_errno);
       goto err;
     }
-    page_type= buff[PAGE_TYPE_OFFSET] & PAGE_TYPE_MASK;
-    switch ((enum en_page_type) page_type) {
+    page_type= (enum en_page_type) (buff[PAGE_TYPE_OFFSET] & PAGE_TYPE_MASK);
+    switch (page_type) {
     case UNALLOCATED_PAGE:
       if (zero_lsn)
         bzero(buff, block_size);
@@ -3192,7 +3238,10 @@ static my_bool maria_zerofill_data(HA_CHECK *param, MARIA_HA *info,
       if (max_entry != 0)
       {
         dir= dir_entry_pos(buff, block_size, max_entry - 1);
-        _ma_compact_block_page(buff, block_size, max_entry -1, 0);
+        _ma_compact_block_page(buff, block_size, max_entry -1, 0,
+                               page_type == HEAD_PAGE ? ~(TrID) 0 : 0,
+                               page_type == HEAD_PAGE ?
+                               share->base.min_block_length : 0);
 
         /* Zerofille the not used part */
         offset= uint2korr(dir) + uint2korr(dir+2);
@@ -4456,8 +4505,15 @@ static int sort_get_next_record(MARIA_SORT_PARAM *sort_param)
           It requires a reliable data_file_length so we set it.
         */
         info->state->data_file_length= sort_info->filelength;
+        info->cur_row.trid= 0;
         flag= _ma_scan_block_record(info, sort_param->record,
                                     info->cur_row.nextpos, 1);
+        set_if_bigger(param->max_found_trid, info->cur_row.trid);
+        if (info->cur_row.trid > param->max_trid)
+        {
+          _ma_check_print_not_visible_error(param, info->cur_row.trid);
+          flag= HA_ERR_ROW_NOT_VISIBLE;
+        }
       }
       if (!flag)
       {
@@ -6463,3 +6519,25 @@ static TrID max_trid_in_system(void)
   /* 'id' may be far bigger, if last shutdown is old */
   return max(id, max_trid_in_control_file);
 }
+
+
+static void _ma_check_print_not_visible_error(HA_CHECK *param, TrID used_trid)
+{
+  char buff[22], buff2[22];
+  if (!param->not_visible_rows_found++)
+  {
+    if (!ma_control_file_inited())
+    {
+      _ma_check_print_warning(param,
+                              "Found row with transaction id %s but no maria_control_file was specified.  The table may be corrupted",
+                              llstr(used_trid, buff));
+    }
+    else
+    {
+      _ma_check_print_error(param,
+                            "Found row with transaction id %s when max transaction id according to maria_control_file is %s",
+                            llstr(used_trid, buff),
+                            llstr(param->max_trid, buff2));
+    }
+  }
+}
diff --git a/storage/maria/ma_control_file.c b/storage/maria/ma_control_file.c
index 419283a7f6f..e703b126b06 100644
--- a/storage/maria/ma_control_file.c
+++ b/storage/maria/ma_control_file.c
@@ -250,7 +250,8 @@ static int lock_control_file(const char *name)
     @retval 1      Error (in which case the file is left closed)
 */
 
-CONTROL_FILE_ERROR ma_control_file_open(my_bool create_if_missing)
+CONTROL_FILE_ERROR ma_control_file_open(my_bool create_if_missing,
+                                        my_bool print_error)
 {
   uchar buffer[CF_MAX_SIZE];
   char name[FN_REFLEN], errmsg_buff[256];
@@ -423,9 +424,10 @@ ok:
   DBUG_RETURN(0);
 
 err:
-  my_printf_error(HA_ERR_INITIALIZATION,
-                  "Error when trying to use maria control file '%s': %s", 0,
-                  name, errmsg);
+  if (print_error)
+    my_printf_error(HA_ERR_INITIALIZATION,
+                    "Got error '%s' when trying to use maria control file "
+                    "'%s'", 0, errmsg, name);
   ma_control_file_end(); /* will unlock file if needed */
   DBUG_RETURN(error);
 }
diff --git a/storage/maria/ma_control_file.h b/storage/maria/ma_control_file.h
index 54c15b3c9f6..52001cd4a4c 100644
--- a/storage/maria/ma_control_file.h
+++ b/storage/maria/ma_control_file.h
@@ -61,7 +61,8 @@ typedef enum enum_control_file_error {
 } CONTROL_FILE_ERROR;
 
 C_MODE_START
-CONTROL_FILE_ERROR ma_control_file_open(my_bool create_if_missing);
+CONTROL_FILE_ERROR ma_control_file_open(my_bool create_if_missing,
+                                        my_bool print_error);
 int ma_control_file_write_and_force(LSN checkpoint_lsn, uint32 logno, TrID trid);
 int ma_control_file_end(void);
 my_bool ma_control_file_inited(void);
diff --git a/storage/maria/ma_delete.c b/storage/maria/ma_delete.c
index d0757b8f274..4a71dddffc8 100644
--- a/storage/maria/ma_delete.c
+++ b/storage/maria/ma_delete.c
@@ -184,7 +184,6 @@ int _ma_ck_delete(register MARIA_HA *info, uint keynr, uchar *key,
     struct st_msg_to_write_hook_for_undo_key msg;
     enum translog_record_type log_type= LOGREC_UNDO_KEY_DELETE;
 
-    info->key_delete_undo_lsn[keynr]= info->trn->undo_lsn;
     lsn_store(log_data, info->trn->undo_lsn);
     key_nr_store(log_data + LSN_STORE_SIZE + FILEID_STORE_SIZE, keynr);
     log_pos= log_data + LSN_STORE_SIZE + FILEID_STORE_SIZE + KEY_NR_STORE_SIZE;
diff --git a/storage/maria/ma_locking.c b/storage/maria/ma_locking.c
index 6d33a4f67e4..692db81e7d0 100644
--- a/storage/maria/ma_locking.c
+++ b/storage/maria/ma_locking.c
@@ -255,7 +255,7 @@ int maria_lock_database(MARIA_HA *info, int lock_type)
 			(THR_WRITE_CONCURRENT_INSERT was used)
 */
 
-void _ma_get_status(void* param, int concurrent_insert)
+void _ma_get_status(void* param, my_bool concurrent_insert)
 {
   MARIA_HA *info=(MARIA_HA*) param;
   DBUG_ENTER("_ma_get_status");
diff --git a/storage/maria/ma_open.c b/storage/maria/ma_open.c
index bdcc10508d8..b95cfce3c3c 100644
--- a/storage/maria/ma_open.c
+++ b/storage/maria/ma_open.c
@@ -118,10 +118,6 @@ static MARIA_HA *maria_clone_internal(MARIA_SHARE *share, int mode,
 		       &info.first_mbr_key, share->base.max_key_length,
 		       &info.maria_rtree_recursion_state,
                        share->have_rtree ? 1024 : 0,
-                       &info.key_write_undo_lsn,
-                       (uint) (sizeof(LSN) * share->base.keys),
-                       &info.key_delete_undo_lsn,
-                       (uint) (sizeof(LSN) * share->base.keys),
                        &changed_fields_bitmap,
                        bitmap_buffer_size(share->base.fields),
 		       NullS))
@@ -780,26 +776,29 @@ MARIA_HA *maria_open(const char *name, int mode, uint open_flags)
     {
       share->concurrent_insert=
 	((share->options & (HA_OPTION_READ_ONLY_DATA | HA_OPTION_TMP_TABLE |
-			   HA_OPTION_COMPRESS_RECORD |
-			   HA_OPTION_TEMP_COMPRESS_RECORD)) ||
+                            HA_OPTION_COMPRESS_RECORD |
+                            HA_OPTION_TEMP_COMPRESS_RECORD)) ||
 	 (open_flags & HA_OPEN_TMP_TABLE) ||
-         share->data_file_type == BLOCK_RECORD ||
+         (share->data_file_type == BLOCK_RECORD &&
+          !share->now_transactional) ||
 	 share->have_rtree) ? 0 : 1;
       if (share->concurrent_insert)
       {
-	share->lock.get_status=_ma_get_status;
-	share->lock.copy_status=_ma_copy_status;
-        /**
-           @todo RECOVERY
-           INSERT DELAYED and concurrent inserts are currently disabled for
-           transactional tables; when enabled again, we should re-evaluate
-           what problems the call to _ma_update_status() by
-           thr_reschedule_write_lock() can do (it may hurt Checkpoint as it
-           would be without intern_lock, and it modifies the state).
-        */
-	share->lock.update_status=_ma_update_status;
-	share->lock.restore_status=_ma_restore_status;
-	share->lock.check_status=_ma_check_status;
+        if (share->data_file_type == BLOCK_RECORD)
+        {
+          share->lock.get_status=    _ma_block_get_status;
+          share->lock.update_status= _ma_block_update_status;
+          share->lock.check_status=  _ma_block_check_status;
+          share->lock.allow_multiple_concurrent_insert= 1;
+        }
+        else
+        {
+          share->lock.get_status=    _ma_get_status;
+          share->lock.copy_status=   _ma_copy_status;
+          share->lock.update_status= _ma_update_status;
+          share->lock.restore_status=_ma_restore_status;
+          share->lock.check_status=  _ma_check_status;
+        }
       }
     }
 #endif
diff --git a/storage/maria/ma_test1.c b/storage/maria/ma_test1.c
index abbad116c9c..4f66fd98be1 100644
--- a/storage/maria/ma_test1.c
+++ b/storage/maria/ma_test1.c
@@ -43,6 +43,7 @@ static uint unique_key=HA_NOSAME;
 static uint die_in_middle_of_transaction;
 static my_bool pagecacheing, null_fields, silent, skip_update, opt_unique;
 static my_bool verbose, skip_delete, transactional;
+static my_bool opt_versioning= 0;
 static MARIA_COLUMNDEF recinfo[4];
 static MARIA_KEYDEF keyinfo[10];
 static HA_KEYSEG keyseg[10];
@@ -77,7 +78,7 @@ int main(int argc,char *argv[])
   if (maria_init() ||
       (init_pagecache(maria_pagecache, maria_block_size * 16, 0, 0,
                       maria_block_size, MY_WME) == 0) ||
-      ma_control_file_open(TRUE) ||
+      ma_control_file_open(TRUE, TRUE) ||
       (init_pagecache(maria_log_pagecache,
                       TRANSLOG_PAGECACHE_SIZE, 0, 0,
                       TRANSLOG_PAGE_SIZE, MY_WME) == 0) ||
@@ -210,6 +211,8 @@ static int run_test(const char *filename)
 
   if (maria_begin(file))
     goto err;
+  if (opt_versioning)
+    maria_versioning(file, 1);
   my_errno=0;
   row_count=deleted=0;
   for (i=49 ; i>=1 ; i-=2 )
@@ -339,6 +342,8 @@ static int run_test(const char *filename)
     goto err;
   if (maria_begin(file))
     goto err;
+  if (opt_versioning)
+    maria_versioning(file, 1);
   if (!skip_delete)
   {
     if (!silent)
@@ -767,7 +772,7 @@ static struct my_option my_long_options[] =
    "Test in transactional mode. (Only works with block format)",
    (uchar**) &transactional, (uchar**) &transactional, 0, GET_BOOL, NO_ARG,
    0, 0, 0, 0, 0, 0},
-  {"unique", 'C', "Undocumented", (uchar**) &opt_unique,
+  {"unique", 'E', "Check unique handling", (uchar**) &opt_unique,
    (uchar**) &opt_unique, 0, GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0},
   {"update-rows", 'u', "Max number of rows to update", (uchar**) &update_count,
    (uchar**) &update_count, 0, GET_UINT, REQUIRED_ARG, 1000, 0, 0, 0, 0, 0},
@@ -775,6 +780,9 @@ static struct my_option my_long_options[] =
    (uchar**) &verbose, 0, GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0},
   {"version", 'V', "Print version number and exit",
    0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0},
+  {"versioning", 'C', "Use row versioning (only works with block format)",
+   (uchar**) &opt_versioning,  (uchar**) &opt_versioning, 0, GET_BOOL,
+   NO_ARG, 0, 0, 0, 0, 0, 0},
   { 0, 0, 0, 0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0}
 };
 
@@ -872,7 +880,8 @@ static void get_options(int argc, char *argv[])
 
   if ((ho_error=handle_options(&argc, &argv, my_long_options, get_one_option)))
     exit(ho_error);
-
+  if (transactional)
+    record_type= BLOCK_RECORD;
   return;
 } /* get options */
 
diff --git a/storage/maria/ma_test2.c b/storage/maria/ma_test2.c
index 7d7eaf997b6..d3b5cca70c5 100644
--- a/storage/maria/ma_test2.c
+++ b/storage/maria/ma_test2.c
@@ -44,6 +44,7 @@ static int silent= 0, opt_quick_mode= 0, transactional= 0, skip_update= 0;
 static int die_in_middle_of_transaction= 0, pack_fields= 1;
 static int pack_seg= HA_SPACE_PACK, pack_type= HA_PACK_KEY, remove_count= -1;
 static int create_flag= 0, srand_arg= 0, checkpoint= 0;
+static my_bool opt_versioning= 0;
 static uint use_blob= 0, update_count= 0;
 static ulong pagecache_size=8192*32;
 static enum data_file_type record_type= DYNAMIC_RECORD;
@@ -83,7 +84,7 @@ int main(int argc, char *argv[])
   if (maria_init() ||
       (init_pagecache(maria_pagecache, pagecache_size, 0, 0,
 		      maria_block_size, MY_WME) == 0) ||
-      ma_control_file_open(TRUE) ||
+      ma_control_file_open(TRUE, TRUE) ||
       (init_pagecache(maria_log_pagecache,
 		      TRANSLOG_PAGECACHE_SIZE, 0, 0,
 		      TRANSLOG_PAGE_SIZE, MY_WME) == 0) ||
@@ -230,6 +231,8 @@ int main(int argc, char *argv[])
   if (!(file=maria_open(filename,2,HA_OPEN_ABORT_IF_LOCKED)))
     goto err;
   maria_begin(file);
+  if (opt_versioning)
+    maria_versioning(file, 1);
   if (testflag == 1)
     goto end;
   if (checkpoint == 1 && ma_checkpoint_execute(CHECKPOINT_MEDIUM, FALSE))
@@ -1136,12 +1139,15 @@ static void get_options(int argc, char **argv)
     case 'g':
       skip_update= TRUE;
       break;
+    case 'C':
+      opt_versioning= 1;
+      break;
     case '?':
     case 'I':
     case 'V':
-      printf("%s  Ver 1.1 for %s at %s\n",progname,SYSTEM_TYPE,MACHINE_TYPE);
+      printf("%s  Ver 1.2 for %s at %s\n",progname,SYSTEM_TYPE,MACHINE_TYPE);
       puts("By Monty, for testing Maria\n");
-      printf("Usage: %s [-?AbBcDIKLPRqSsTVWltv] [-k#] [-f#] [-m#] [-e#] [-E#] [-t#]\n",
+      printf("Usage: %s [-?AbBcCDIKLPRqSsTVWltv] [-k#] [-f#] [-m#] [-e#] [-E#] [-t#]\n",
 	     progname);
       exit(0);
     case '#':
diff --git a/storage/maria/ma_test_recovery b/storage/maria/ma_test_recovery
index 69fd9778a24..0b20264c434 100755
--- a/storage/maria/ma_test_recovery
+++ b/storage/maria/ma_test_recovery
@@ -5,4 +5,4 @@
 
 # This file is deprecated and has been replaced with ma_test_recovery.pl
 
-unittest/ma_test_recovery.pl
+unittest/ma_test_recovery.pl $@
diff --git a/storage/maria/ma_write.c b/storage/maria/ma_write.c
index fe0182e9a19..b87fb9b6eee 100644
--- a/storage/maria/ma_write.c
+++ b/storage/maria/ma_write.c
@@ -403,7 +403,6 @@ static int _ma_ck_write_btree_with_log(MARIA_HA *info, MARIA_KEYDEF *keyinfo,
     struct st_msg_to_write_hook_for_undo_key msg;
 
     /* Save if we need to write a clr record */
-    info->key_write_undo_lsn[keyinfo->key_nr]= info->trn->undo_lsn;
     lsn_store(log_data, info->trn->undo_lsn);
     key_nr_store(log_data + LSN_STORE_SIZE + FILEID_STORE_SIZE,
                   keyinfo->key_nr);
diff --git a/storage/maria/maria_chk.c b/storage/maria/maria_chk.c
index 55db8e6308d..b64a7379c50 100644
--- a/storage/maria/maria_chk.c
+++ b/storage/maria/maria_chk.c
@@ -106,14 +106,21 @@ int main(int argc, char **argv)
   error=0;
   maria_init();
 
+  if (ma_control_file_open(FALSE, opt_require_control_file) &&
+      (opt_require_control_file ||
+       (opt_transaction_logging && (check_param.testflag & T_REP_ANY))))
+  {
+    error= 1;
+    goto end;
+  }
+
   /*
     If we are doing a repair, user may want to store this repair into the log
     so that the log has a complete history and can be used to replay.
   */
   if (opt_transaction_logging && (check_param.testflag & T_REP_ANY))
   {
-    if (ma_control_file_open(FALSE) ||
-        init_pagecache(maria_log_pagecache,
+    if (init_pagecache(maria_log_pagecache,
                        TRANSLOG_PAGECACHE_SIZE, 0, 0,
                        TRANSLOG_PAGE_SIZE, MY_WME) == 0 ||
         translog_init(opt_log_dir, TRANSLOG_FILE_SIZE,
@@ -127,14 +134,6 @@ int main(int argc, char **argv)
       goto end;
     }
   }
-  else
-  {
-    if (ma_control_file_open(FALSE) && opt_require_control_file)
-    {
-      error= 1;
-      goto end;
-    }
-  }
 
   while (--argc >= 0)
   {
@@ -1237,9 +1236,10 @@ static int maria_chk(HA_CHECK *param, char *filename)
       printf("Data records: %7s   Deleted blocks: %7s\n",
              llstr(info->state->records,llbuff),
              llstr(info->state->del,llbuff2));
-    error =maria_chk_status(param,info);
+    maria_chk_init_for_check(param, info);
+    error= maria_chk_status(param,info);
     maria_intersect_keys_active(share->state.key_map, param->keys_in_use);
-    error =maria_chk_size(param,info);
+    error|= maria_chk_size(param,info);
     if (!error || !(param->testflag & (T_FAST | T_FORCE_CREATE)))
       error|=maria_chk_del(param, info,param->testflag);
     if ((!error || (!(param->testflag & (T_FAST | T_FORCE_CREATE)) &&
@@ -1327,7 +1327,15 @@ end2:
                           T_ZEROFILL)))
     error= write_log_record(param);
 
+  if (param->not_visible_rows_found && (param->testflag & T_VERBOSE))
+  {
+    char buff[22];
+    printf("Max transaction id found: %s\n",
+           llstr(param->max_found_trid, buff));
+  }
+
   VOID(fflush(stdout)); VOID(fflush(stderr));
+
   if (param->error_printed)
   {
     if (param->testflag & (T_REP_ANY | T_SORT_RECORDS | T_SORT_INDEX))
diff --git a/storage/maria/maria_def.h b/storage/maria/maria_def.h
index 5b5503fb02c..e220ce36d94 100644
--- a/storage/maria/maria_def.h
+++ b/storage/maria/maria_def.h
@@ -432,10 +432,11 @@ typedef struct st_maria_row
   MARIA_RECORD_POS *tail_positions;
   ha_checksum checksum;
   LSN orig_undo_lsn;			/* Lsn at start of row insert */
+  TrID trid;                            /* Transaction id for current row */
   uchar *empty_bits, *field_lengths;
   uint *null_field_lengths;             /* All null field lengths */
   ulong *blob_lengths;                  /* Length for each blob */
-  ulong base_length, min_length, normal_length, char_length, varchar_length;
+  ulong min_length, normal_length, char_length, varchar_length;
   ulong blob_length, head_length, total_length;
   size_t extents_buffer_length;         /* Size of 'extents' buffer */
   uint field_lengths_length;            /* Length of data in field_lengths */
@@ -470,12 +471,10 @@ struct st_maria_handler
   DYNAMIC_ARRAY pinned_pages;
   /* accumulate indexfile changes between write's */
   TREE *bulk_insert;
-  LEX_STRING *log_row_parts;		/* For logging */
+  LEX_CUSTRING *log_row_parts;		/* For logging */
   DYNAMIC_ARRAY *ft1_to_ft2;		/* used only in ft1->ft2 conversion */
   MEM_ROOT      ft_memroot;             /* used by the parser               */
   MYSQL_FTPARSER_PARAM *ftparser_param;	/* share info between init/deinit */
-  LSN   *key_write_undo_lsn;            /* Pointer to undo for each key */
-  LSN   *key_delete_undo_lsn;           /* Pointer to undo for each key */
   uchar *buff;				/* page buffer */
   uchar *keyread_buff;                   /* Buffer for last key read */
   uchar *lastkey, *lastkey2;		/* Last used search key */
@@ -510,6 +509,8 @@ struct st_maria_handler
   IO_CACHE rec_cache;			/* When cacheing records */
   LIST open_list;
   MY_BITMAP changed_fields;
+  ulong row_base_length;                /* Length of row header */
+  uint row_flag;                        /* Flag to store in row header */
   uint opt_flag;			/* Optim. for space/speed */
   uint update;				/* If file changed since open */
   int lastinx;				/* Last used index */
@@ -608,9 +609,9 @@ struct st_maria_handler
 #define _ma_store_keynr(share, x, nr) x[(share)->keypage_header - KEYPAGE_KEYID_SIZE - KEYPAGE_FLAG_SIZE - KEYPAGE_USED_SIZE]= (nr)
 #define _ma_get_keynr(share, x) ((uchar) x[(share)->keypage_header - KEYPAGE_KEYID_SIZE - KEYPAGE_FLAG_SIZE - KEYPAGE_USED_SIZE])
 #define _ma_store_transid(buff, transid) \
-  int6store((buff) + LSN_STORE_SIZE, (transid))
+  transid_store((buff) + LSN_STORE_SIZE, (transid))
 #define _ma_korr_transid(buff) \
-  uint6korr((buff) + LSN_STORE_SIZE)
+  transid_korr((buff) + LSN_STORE_SIZE)
 #define _ma_get_keypage_flag(share,x) x[(share)->keypage_header - KEYPAGE_USED_SIZE - KEYPAGE_FLAG_SIZE]
 #define _ma_store_keypage_flag(share,x,flag) x[(share)->keypage_header - KEYPAGE_USED_SIZE - KEYPAGE_FLAG_SIZE]= (flag)
 
@@ -1042,7 +1043,7 @@ my_bool _ma_cmp_dynamic_unique(MARIA_HA *info, MARIA_UNIQUEDEF *def,
                                const uchar *record, MARIA_RECORD_POS pos);
 my_bool _ma_unique_comp(MARIA_UNIQUEDEF *def, const uchar *a, const uchar *b,
                         my_bool null_are_equal);
-void _ma_get_status(void *param, int concurrent_insert);
+void _ma_get_status(void *param, my_bool concurrent_insert);
 void _ma_update_status(void *param);
 void _ma_restore_status(void *param);
 void _ma_copy_status(void *to, void *from);
@@ -1148,4 +1149,5 @@ extern my_bool maria_flush_log_for_page(uchar *page,
 extern my_bool maria_flush_log_for_page_none(uchar *page,
                                              pgcache_page_no_t page_no,
                                              uchar *data_ptr);
+void maria_concurrent_inserts(MARIA_HA *info, my_bool concurrent_insert);
 extern PAGECACHE *maria_log_pagecache;
diff --git a/storage/maria/maria_pack.c b/storage/maria/maria_pack.c
index 6902f131f5b..68ef56ae865 100644
--- a/storage/maria/maria_pack.c
+++ b/storage/maria/maria_pack.c
@@ -435,6 +435,7 @@ static MARIA_HA *open_maria_file(char *name,int mode)
     DBUG_RETURN(0);
   }
   VOID(maria_lock_database(isam_file,F_WRLCK));
+  maria_ignore_trids(isam_file);
   DBUG_RETURN(isam_file);
 }
 
@@ -1083,7 +1084,7 @@ static int get_statistic(PACK_MRG_INFO *mrg,HUFF_COUNTS *huff_counts)
     }
     else if (error != HA_ERR_RECORD_DELETED)
     {
-      VOID(fprintf(stderr, "Got error %d while reading rows", error));
+      VOID(fprintf(stderr, "Got error %d while reading rows\n", error));
       break;
     }
 
diff --git a/storage/maria/maria_read_log.c b/storage/maria/maria_read_log.c
index c2ed0c736dd..79b201feefe 100644
--- a/storage/maria/maria_read_log.c
+++ b/storage/maria/maria_read_log.c
@@ -56,7 +56,7 @@ int main(int argc, char **argv)
     goto err;
   }
   /* we don't want to create a control file, it MUST exist */
-  if (ma_control_file_open(FALSE))
+  if (ma_control_file_open(FALSE, TRUE))
   {
     fprintf(stderr, "Can't open control file (%d)\n", errno);
     goto err;
diff --git a/storage/maria/trnman.c b/storage/maria/trnman.c
index 24497faac50..56f59ce6dae 100644
--- a/storage/maria/trnman.c
+++ b/storage/maria/trnman.c
@@ -315,7 +315,13 @@ TRN *trnman_new_trn(pthread_mutex_t *mutex, pthread_cond_t *cond,
   pthread_mutex_unlock(&LOCK_trn_list);
 
   if (unlikely(!trn->min_read_from))
-    trn->min_read_from= trn->trid;
+  {
+    /*
+      We are the only transaction. Set min_read_from so that we can read
+      our own rows
+    */
+    trn->min_read_from= trn->trid + 1;
+  }
 
   trn->commit_trid= 0;
   trn->rec_lsn= trn->undo_lsn= trn->first_undo_lsn= 0;
@@ -534,9 +540,19 @@ int trnman_can_read_from(TRN *trn, TrID trid)
   LF_REQUIRE_PINS(3);
 
   if (trid < trn->min_read_from)
-    return 1; /* can read */
-  if (trid > trn->trid)
-    return 0; /* cannot read */
+    return 1; /* Row is visible by all transactions in the system */
+
+  if (trid >= trn->trid)
+  {
+    /*
+      We have now two cases
+      trid > trn->trid, in which case the row is from a new transaction
+      and not visible, in which case we should return 0.
+      trid == trn->trid in which case the row is from the current transaction
+      and we should return 1
+    */
+    return trid == trn->trid;
+  }
 
   found= lf_hash_search(&trid_to_committed_trn, trn->pins, &trid, sizeof(trid));
   if (found == NULL)
diff --git a/storage/maria/unittest/ma_control_file-t.c b/storage/maria/unittest/ma_control_file-t.c
index f968abd9c1b..f076615fef7 100644
--- a/storage/maria/unittest/ma_control_file-t.c
+++ b/storage/maria/unittest/ma_control_file-t.c
@@ -110,7 +110,7 @@ static CONTROL_FILE_ERROR local_ma_control_file_open(void)
 {
   CONTROL_FILE_ERROR error;
   error_handler_hook= my_ignore_message;
-  error= ma_control_file_open(TRUE);
+  error= ma_control_file_open(TRUE, TRUE);
   error_handler_hook= default_error_handler_hook;
   return error;
 }
diff --git a/storage/maria/unittest/ma_test_all-t b/storage/maria/unittest/ma_test_all-t
index 11e8ef5d743..70f1a8ded14 100755
--- a/storage/maria/unittest/ma_test_all-t
+++ b/storage/maria/unittest/ma_test_all-t
@@ -87,9 +87,9 @@ sub run_tests
   # make the unit test fail during 'make test'. $nr_tests must be right.
   #
 
-  $nr_tests+= run_check_tests(0, 0, 0, 0, 1) * 4;  #
-  $nr_tests+= run_repair_tests(0, 0, 0, 0, 1) * 4; # called 4 times
-  $nr_tests+= run_pack_tests(0, 0, 0, 0, 1) * 4;   #
+  $nr_tests+= run_check_tests(0, 0, 0, 0, 1) * 5;  #
+  $nr_tests+= run_repair_tests(0, 0, 0, 0, 1) * 5; # called 4 times
+  $nr_tests+= run_pack_tests(0, 0, 0, 0, 1) * 5;   #
   $nr_tests+= run_tests_on_warnings_and_errors(0, 0, 0, 1);
   $nr_tests+= run_ma_test_recovery(0, 1);
   $nr_tests+= run_tests_on_clrs(0, 0, 1);
@@ -185,6 +185,19 @@ sub run_tests
   run_repair_tests($suffix, $opt_silent, "-M -T", $opt_verbose, 0);
   run_pack_tests($suffix, $opt_silent, "-M -T", $opt_verbose, 0);
 
+  if ($opt_verbose)
+  {
+    print "\nRunning tests with block row format, transactions and versioning\n";
+  }
+  run_check_tests($suffix, $opt_silent, "-M -T -C", $opt_verbose, 0);
+  run_repair_tests($suffix, $opt_silent, "-M -T -C", $opt_verbose, 0);
+  run_pack_tests($suffix, $opt_silent, "-M -T -C", $opt_verbose, 0);
+
+
+  if ($opt_verbose)
+  {
+    print "\nRunning tests with warnings and recovery\n";
+  }
   run_tests_on_warnings_and_errors($suffix, $opt_silent, $opt_verbose, 0);
   run_ma_test_recovery($opt_verbose, 0);
   run_tests_on_clrs($suffix, $opt_verbose, 0);
@@ -523,7 +536,6 @@ sub ok
   if ($verbose)
   {
     print " " x (62 - $len);
-    print " ";
   }
   $err= $?;
   if ((!$err && !$expected_error) ||
diff --git a/storage/maria/unittest/ma_test_loghandler-t.c b/storage/maria/unittest/ma_test_loghandler-t.c
index 34b2ce682d5..1e8b4938e4f 100644
--- a/storage/maria/unittest/ma_test_loghandler-t.c
+++ b/storage/maria/unittest/ma_test_loghandler-t.c
@@ -196,7 +196,7 @@ int main(int argc __attribute__((unused)), char *argv[])
   }
 #endif
 
-  if (ma_control_file_open(TRUE))
+  if (ma_control_file_open(TRUE, TRUE))
   {
     fprintf(stderr, "Can't init control file (%d)\n", errno);
     exit(1);
diff --git a/storage/maria/unittest/ma_test_loghandler_first_lsn-t.c b/storage/maria/unittest/ma_test_loghandler_first_lsn-t.c
index d8ae57634e4..a56f7d64118 100644
--- a/storage/maria/unittest/ma_test_loghandler_first_lsn-t.c
+++ b/storage/maria/unittest/ma_test_loghandler_first_lsn-t.c
@@ -67,7 +67,7 @@ int main(int argc __attribute__((unused)), char *argv[])
   }
 #endif
 
-  if (ma_control_file_open(TRUE))
+  if (ma_control_file_open(TRUE, TRUE))
   {
     fprintf(stderr, "Can't init control file (%d)\n", errno);
     exit(1);
diff --git a/storage/maria/unittest/ma_test_loghandler_max_lsn-t.c b/storage/maria/unittest/ma_test_loghandler_max_lsn-t.c
index 9cbf05825ce..238a9ec9cbb 100644
--- a/storage/maria/unittest/ma_test_loghandler_max_lsn-t.c
+++ b/storage/maria/unittest/ma_test_loghandler_max_lsn-t.c
@@ -64,7 +64,7 @@ int main(int argc __attribute__((unused)), char *argv[])
   }
 #endif
 
-  if (ma_control_file_open(TRUE))
+  if (ma_control_file_open(TRUE, TRUE))
   {
     fprintf(stderr, "Can't init control file (%d)\n", errno);
     exit(1);
diff --git a/storage/maria/unittest/ma_test_loghandler_multigroup-t.c b/storage/maria/unittest/ma_test_loghandler_multigroup-t.c
index 3dbfc9e579b..ee5a316d501 100644
--- a/storage/maria/unittest/ma_test_loghandler_multigroup-t.c
+++ b/storage/maria/unittest/ma_test_loghandler_multigroup-t.c
@@ -184,7 +184,7 @@ int main(int argc __attribute__((unused)), char *argv[])
   }
 #endif
 
-  if (ma_control_file_open(TRUE))
+  if (ma_control_file_open(TRUE, TRUE))
   {
     fprintf(stderr, "Can't init control file (%d)\n", errno);
     exit(1);
@@ -348,7 +348,7 @@ int main(int argc __attribute__((unused)), char *argv[])
   end_pagecache(&pagecache, 1);
   ma_control_file_end();
 
-  if (ma_control_file_open(TRUE))
+  if (ma_control_file_open(TRUE,TRUE))
   {
     fprintf(stderr, "pass2: Can't init control file (%d)\n", errno);
     exit(1);
diff --git a/storage/maria/unittest/ma_test_loghandler_multithread-t.c b/storage/maria/unittest/ma_test_loghandler_multithread-t.c
index 3a278c8d03b..ba612fab6a6 100644
--- a/storage/maria/unittest/ma_test_loghandler_multithread-t.c
+++ b/storage/maria/unittest/ma_test_loghandler_multithread-t.c
@@ -283,7 +283,7 @@ int main(int argc __attribute__((unused)),
 
   my_thread_global_init();
 
-  if (ma_control_file_open(TRUE))
+  if (ma_control_file_open(TRUE, TRUE))
   {
     fprintf(stderr, "Can't init control file (%d)\n", errno);
     exit(1);
diff --git a/storage/maria/unittest/ma_test_loghandler_noflush-t.c b/storage/maria/unittest/ma_test_loghandler_noflush-t.c
index e07b96ee5d1..54294936cc3 100644
--- a/storage/maria/unittest/ma_test_loghandler_noflush-t.c
+++ b/storage/maria/unittest/ma_test_loghandler_noflush-t.c
@@ -70,7 +70,7 @@ int main(int argc __attribute__((unused)), char *argv[])
   }
 #endif
 
-  if (ma_control_file_open(TRUE))
+  if (ma_control_file_open(TRUE, TRUE))
   {
     fprintf(stderr, "Can't init control file (%d)\n", errno);
     exit(1);
diff --git a/storage/maria/unittest/ma_test_loghandler_nologs-t.c b/storage/maria/unittest/ma_test_loghandler_nologs-t.c
index 9241279703d..e126ff31f3a 100644
--- a/storage/maria/unittest/ma_test_loghandler_nologs-t.c
+++ b/storage/maria/unittest/ma_test_loghandler_nologs-t.c
@@ -67,7 +67,7 @@ int main(int argc __attribute__((unused)), char *argv[])
   }
 #endif
 
-  if (ma_control_file_open(TRUE))
+  if (ma_control_file_open(TRUE, TRUE))
   {
     fprintf(stderr, "Can't init control file (%d)\n", errno);
     exit(1);
@@ -140,7 +140,7 @@ int main(int argc __attribute__((unused)), char *argv[])
     }
   }
 
-  if (ma_control_file_open(TRUE))
+  if (ma_control_file_open(TRUE, TRUE))
   {
     fprintf(stderr, "Can't init control file (%d)\n", errno);
     exit(1);
diff --git a/storage/maria/unittest/ma_test_loghandler_pagecache-t.c b/storage/maria/unittest/ma_test_loghandler_pagecache-t.c
index c937d56925b..16f672d66ca 100644
--- a/storage/maria/unittest/ma_test_loghandler_pagecache-t.c
+++ b/storage/maria/unittest/ma_test_loghandler_pagecache-t.c
@@ -95,7 +95,7 @@ int main(int argc __attribute__((unused)), char *argv[])
   }
 #endif
 
-  if (ma_control_file_open(TRUE))
+  if (ma_control_file_open(TRUE, TRUE))
   {
     fprintf(stderr, "Can't init control file (%d)\n", errno);
     exit(1);
diff --git a/storage/maria/unittest/ma_test_loghandler_purge-t.c b/storage/maria/unittest/ma_test_loghandler_purge-t.c
index 4602ec88548..0c3be2f2428 100644
--- a/storage/maria/unittest/ma_test_loghandler_purge-t.c
+++ b/storage/maria/unittest/ma_test_loghandler_purge-t.c
@@ -67,7 +67,7 @@ int main(int argc __attribute__((unused)), char *argv[])
   }
 #endif
 
-  if (ma_control_file_open(TRUE))
+  if (ma_control_file_open(TRUE, TRUE))
   {
     fprintf(stderr, "Can't init control file (%d)\n", errno);
     exit(1);
diff --git a/storage/maria/unittest/ma_test_recovery.expected b/storage/maria/unittest/ma_test_recovery.expected
index 9da7b662395..b95575173ee 100644
--- a/storage/maria/unittest/ma_test_recovery.expected
+++ b/storage/maria/unittest/ma_test_recovery.expected
@@ -15,6 +15,14 @@ TEST WITH ma_test2 -s -M -T -c -b65000 -d800
 applying log
 testing idempotency
 applying log
+TEST WITH ma_test1 -s -M -T -c -C
+applying log
+testing idempotency
+applying log
+TEST WITH ma_test2 -s -L -K -W -P -M -T -c -d500 -C
+applying log
+testing idempotency
+applying log
 Testing the REDO AND UNDO PHASE
 TEST WITH ma_test1 -s -M -T -c -N --testflag=1 (commit at end)
 TEST WITH ma_test1 -s -M -T -c -N --testflag=2 --test-undo=1 (additional aborted work)
@@ -35,6 +43,16 @@ testing idempotency
 applying log
 testing applying of CLRs to recreate table
 applying log
+TEST WITH ma_test1 -s -M -T -c -N --versioning --testflag=3 (commit at end)
+Terminating after updates
+TEST WITH ma_test1 -s -M -T -c -N --versioning --testflag=4 --test-undo=1 (additional aborted work)
+Terminating after deletes
+Dying on request without maria_commit()/maria_close()
+applying log
+testing idempotency
+applying log
+testing applying of CLRs to recreate table
+applying log
 TEST WITH ma_test1 -s -M -T -c -N --testflag=2 (commit at end)
 Terminating after inserts
 TEST WITH ma_test1 -s -M -T -c -N --testflag=3 --test-undo=1 (additional aborted work)
@@ -122,6 +140,16 @@ testing idempotency
 applying log
 testing applying of CLRs to recreate table
 applying log
+TEST WITH ma_test1 -s -M -T -c -N --versioning --testflag=3 (commit at end)
+Terminating after updates
+TEST WITH ma_test1 -s -M -T -c -N --versioning --testflag=4 --test-undo=2 (additional aborted work)
+Terminating after deletes
+Dying on request without maria_commit()/maria_close()
+applying log
+testing idempotency
+applying log
+testing applying of CLRs to recreate table
+applying log
 TEST WITH ma_test1 -s -M -T -c -N --testflag=2 (commit at end)
 Terminating after inserts
 TEST WITH ma_test1 -s -M -T -c -N --testflag=3 --test-undo=2 (additional aborted work)
@@ -209,6 +237,16 @@ testing idempotency
 applying log
 testing applying of CLRs to recreate table
 applying log
+TEST WITH ma_test1 -s -M -T -c -N --versioning --testflag=3 (commit at end)
+Terminating after updates
+TEST WITH ma_test1 -s -M -T -c -N --versioning --testflag=4 --test-undo=3 (additional aborted work)
+Terminating after deletes
+Dying on request without maria_commit()/maria_close()
+applying log
+testing idempotency
+applying log
+testing applying of CLRs to recreate table
+applying log
 TEST WITH ma_test1 -s -M -T -c -N --testflag=2 (commit at end)
 Terminating after inserts
 TEST WITH ma_test1 -s -M -T -c -N --testflag=3 --test-undo=3 (additional aborted work)
@@ -296,6 +334,16 @@ testing idempotency
 applying log
 testing applying of CLRs to recreate table
 applying log
+TEST WITH ma_test1 -s -M -T -c -N --versioning --testflag=3 (commit at end)
+Terminating after updates
+TEST WITH ma_test1 -s -M -T -c -N --versioning --testflag=4 --test-undo=4 (additional aborted work)
+Terminating after deletes
+Dying on request without maria_commit()/maria_close()
+applying log
+testing idempotency
+applying log
+testing applying of CLRs to recreate table
+applying log
 TEST WITH ma_test1 -s -M -T -c -N --testflag=2 (commit at end)
 Terminating after inserts
 TEST WITH ma_test1 -s -M -T -c -N --testflag=3 --test-undo=4 (additional aborted work)
@@ -383,6 +431,16 @@ testing idempotency
 applying log
 testing applying of CLRs to recreate table
 applying log
+TEST WITH ma_test1 -s -M -T -c -N -b32768 --versioning --testflag=3 (commit at end)
+Terminating after updates
+TEST WITH ma_test1 -s -M -T -c -N -b32768 --versioning --testflag=4 --test-undo=1 (additional aborted work)
+Terminating after deletes
+Dying on request without maria_commit()/maria_close()
+applying log
+testing idempotency
+applying log
+testing applying of CLRs to recreate table
+applying log
 TEST WITH ma_test1 -s -M -T -c -N -b32768 --testflag=2 (commit at end)
 Terminating after inserts
 TEST WITH ma_test1 -s -M -T -c -N -b32768 --testflag=3 --test-undo=1 (additional aborted work)
@@ -470,6 +528,16 @@ testing idempotency
 applying log
 testing applying of CLRs to recreate table
 applying log
+TEST WITH ma_test1 -s -M -T -c -N -b32768 --versioning --testflag=3 (commit at end)
+Terminating after updates
+TEST WITH ma_test1 -s -M -T -c -N -b32768 --versioning --testflag=4 --test-undo=2 (additional aborted work)
+Terminating after deletes
+Dying on request without maria_commit()/maria_close()
+applying log
+testing idempotency
+applying log
+testing applying of CLRs to recreate table
+applying log
 TEST WITH ma_test1 -s -M -T -c -N -b32768 --testflag=2 (commit at end)
 Terminating after inserts
 TEST WITH ma_test1 -s -M -T -c -N -b32768 --testflag=3 --test-undo=2 (additional aborted work)
@@ -557,6 +625,16 @@ testing idempotency
 applying log
 testing applying of CLRs to recreate table
 applying log
+TEST WITH ma_test1 -s -M -T -c -N -b32768 --versioning --testflag=3 (commit at end)
+Terminating after updates
+TEST WITH ma_test1 -s -M -T -c -N -b32768 --versioning --testflag=4 --test-undo=3 (additional aborted work)
+Terminating after deletes
+Dying on request without maria_commit()/maria_close()
+applying log
+testing idempotency
+applying log
+testing applying of CLRs to recreate table
+applying log
 TEST WITH ma_test1 -s -M -T -c -N -b32768 --testflag=2 (commit at end)
 Terminating after inserts
 TEST WITH ma_test1 -s -M -T -c -N -b32768 --testflag=3 --test-undo=3 (additional aborted work)
@@ -644,6 +722,16 @@ testing idempotency
 applying log
 testing applying of CLRs to recreate table
 applying log
+TEST WITH ma_test1 -s -M -T -c -N -b32768 --versioning --testflag=3 (commit at end)
+Terminating after updates
+TEST WITH ma_test1 -s -M -T -c -N -b32768 --versioning --testflag=4 --test-undo=4 (additional aborted work)
+Terminating after deletes
+Dying on request without maria_commit()/maria_close()
+applying log
+testing idempotency
+applying log
+testing applying of CLRs to recreate table
+applying log
 TEST WITH ma_test1 -s -M -T -c -N -b32768 --testflag=2 (commit at end)
 Terminating after inserts
 TEST WITH ma_test1 -s -M -T -c -N -b32768 --testflag=3 --test-undo=4 (additional aborted work)
@@ -731,6 +819,16 @@ testing idempotency
 applying log
 testing applying of CLRs to recreate table
 applying log
+TEST WITH ma_test1 -s -M -T -c -N  -H2 --versioning --testflag=3 (commit at end)
+Terminating after updates
+TEST WITH ma_test1 -s -M -T -c -N  -H2 --versioning --testflag=4 --test-undo=1 (additional aborted work)
+Terminating after deletes
+Dying on request without maria_commit()/maria_close()
+applying log
+testing idempotency
+applying log
+testing applying of CLRs to recreate table
+applying log
 TEST WITH ma_test1 -s -M -T -c -N  -H2 --testflag=2 (commit at end)
 Terminating after inserts
 TEST WITH ma_test1 -s -M -T -c -N  -H2 --testflag=3 --test-undo=1 (additional aborted work)
@@ -818,6 +916,16 @@ testing idempotency
 applying log
 testing applying of CLRs to recreate table
 applying log
+TEST WITH ma_test1 -s -M -T -c -N  -H2 --versioning --testflag=3 (commit at end)
+Terminating after updates
+TEST WITH ma_test1 -s -M -T -c -N  -H2 --versioning --testflag=4 --test-undo=2 (additional aborted work)
+Terminating after deletes
+Dying on request without maria_commit()/maria_close()
+applying log
+testing idempotency
+applying log
+testing applying of CLRs to recreate table
+applying log
 TEST WITH ma_test1 -s -M -T -c -N  -H2 --testflag=2 (commit at end)
 Terminating after inserts
 TEST WITH ma_test1 -s -M -T -c -N  -H2 --testflag=3 --test-undo=2 (additional aborted work)
@@ -905,6 +1013,16 @@ testing idempotency
 applying log
 testing applying of CLRs to recreate table
 applying log
+TEST WITH ma_test1 -s -M -T -c -N  -H2 --versioning --testflag=3 (commit at end)
+Terminating after updates
+TEST WITH ma_test1 -s -M -T -c -N  -H2 --versioning --testflag=4 --test-undo=3 (additional aborted work)
+Terminating after deletes
+Dying on request without maria_commit()/maria_close()
+applying log
+testing idempotency
+applying log
+testing applying of CLRs to recreate table
+applying log
 TEST WITH ma_test1 -s -M -T -c -N  -H2 --testflag=2 (commit at end)
 Terminating after inserts
 TEST WITH ma_test1 -s -M -T -c -N  -H2 --testflag=3 --test-undo=3 (additional aborted work)
@@ -992,6 +1110,16 @@ testing idempotency
 applying log
 testing applying of CLRs to recreate table
 applying log
+TEST WITH ma_test1 -s -M -T -c -N  -H2 --versioning --testflag=3 (commit at end)
+Terminating after updates
+TEST WITH ma_test1 -s -M -T -c -N  -H2 --versioning --testflag=4 --test-undo=4 (additional aborted work)
+Terminating after deletes
+Dying on request without maria_commit()/maria_close()
+applying log
+testing idempotency
+applying log
+testing applying of CLRs to recreate table
+applying log
 TEST WITH ma_test1 -s -M -T -c -N  -H2 --testflag=2 (commit at end)
 Terminating after inserts
 TEST WITH ma_test1 -s -M -T -c -N  -H2 --testflag=3 --test-undo=4 (additional aborted work)
@@ -1079,6 +1207,16 @@ testing idempotency
 applying log
 testing applying of CLRs to recreate table
 applying log
+TEST WITH ma_test1 -s -M -T -c -N -b32768 -H2 --versioning --testflag=3 (commit at end)
+Terminating after updates
+TEST WITH ma_test1 -s -M -T -c -N -b32768 -H2 --versioning --testflag=4 --test-undo=1 (additional aborted work)
+Terminating after deletes
+Dying on request without maria_commit()/maria_close()
+applying log
+testing idempotency
+applying log
+testing applying of CLRs to recreate table
+applying log
 TEST WITH ma_test1 -s -M -T -c -N -b32768 -H2 --testflag=2 (commit at end)
 Terminating after inserts
 TEST WITH ma_test1 -s -M -T -c -N -b32768 -H2 --testflag=3 --test-undo=1 (additional aborted work)
@@ -1166,6 +1304,16 @@ testing idempotency
 applying log
 testing applying of CLRs to recreate table
 applying log
+TEST WITH ma_test1 -s -M -T -c -N -b32768 -H2 --versioning --testflag=3 (commit at end)
+Terminating after updates
+TEST WITH ma_test1 -s -M -T -c -N -b32768 -H2 --versioning --testflag=4 --test-undo=2 (additional aborted work)
+Terminating after deletes
+Dying on request without maria_commit()/maria_close()
+applying log
+testing idempotency
+applying log
+testing applying of CLRs to recreate table
+applying log
 TEST WITH ma_test1 -s -M -T -c -N -b32768 -H2 --testflag=2 (commit at end)
 Terminating after inserts
 TEST WITH ma_test1 -s -M -T -c -N -b32768 -H2 --testflag=3 --test-undo=2 (additional aborted work)
@@ -1253,6 +1401,16 @@ testing idempotency
 applying log
 testing applying of CLRs to recreate table
 applying log
+TEST WITH ma_test1 -s -M -T -c -N -b32768 -H2 --versioning --testflag=3 (commit at end)
+Terminating after updates
+TEST WITH ma_test1 -s -M -T -c -N -b32768 -H2 --versioning --testflag=4 --test-undo=3 (additional aborted work)
+Terminating after deletes
+Dying on request without maria_commit()/maria_close()
+applying log
+testing idempotency
+applying log
+testing applying of CLRs to recreate table
+applying log
 TEST WITH ma_test1 -s -M -T -c -N -b32768 -H2 --testflag=2 (commit at end)
 Terminating after inserts
 TEST WITH ma_test1 -s -M -T -c -N -b32768 -H2 --testflag=3 --test-undo=3 (additional aborted work)
@@ -1340,6 +1498,16 @@ testing idempotency
 applying log
 testing applying of CLRs to recreate table
 applying log
+TEST WITH ma_test1 -s -M -T -c -N -b32768 -H2 --versioning --testflag=3 (commit at end)
+Terminating after updates
+TEST WITH ma_test1 -s -M -T -c -N -b32768 -H2 --versioning --testflag=4 --test-undo=4 (additional aborted work)
+Terminating after deletes
+Dying on request without maria_commit()/maria_close()
+applying log
+testing idempotency
+applying log
+testing applying of CLRs to recreate table
+applying log
 TEST WITH ma_test1 -s -M -T -c -N -b32768 -H2 --testflag=2 (commit at end)
 Terminating after inserts
 TEST WITH ma_test1 -s -M -T -c -N -b32768 -H2 --testflag=3 --test-undo=4 (additional aborted work)
diff --git a/storage/maria/unittest/ma_test_recovery.pl b/storage/maria/unittest/ma_test_recovery.pl
index 39c1a14563f..6879d795be8 100755
--- a/storage/maria/unittest/ma_test_recovery.pl
+++ b/storage/maria/unittest/ma_test_recovery.pl
@@ -11,6 +11,8 @@ $VER= "1.2";
 
 $opt_version= 0;
 $opt_help=    0;
+$opt_verbose= 0;
+$opt_abort_on_error=0;
 
 my $silent= "-s";
 my $maria_path;     # path to "storage/maria"
@@ -33,7 +35,7 @@ sub main
 {
   my ($res, $table);
 
-  if (!GetOptions("help","version"))
+  if (!GetOptions("abort-on-error", "help", "version", "verbose"))
   {
     $flag_exit= 1;
   }
@@ -86,7 +88,10 @@ sub main
   my @t= ("ma_test1$suffix $silent -M -T -c",
           "ma_test2$suffix $silent -L -K -W -P -M -T -c -d500",
           "ma_test2$suffix $silent -M -T -c -b65000",
-          "ma_test2$suffix $silent -M -T -c -b65000 -d800");
+          "ma_test2$suffix $silent -M -T -c -b65000 -d800",
+          "ma_test1$suffix $silent -M -T -c -C",
+          "ma_test2$suffix $silent -L -K -W -P -M -T -c -d500 -C"
+         );
 
   foreach my $prog (@t)
   {
@@ -94,7 +99,7 @@ sub main
     my $prog_no_suffix= $prog;
     $prog_no_suffix=~ s/$suffix// if ($suffix);
     print MY_LOG "TEST WITH $prog_no_suffix\n";
-    $res= `$maria_exe_path/$prog`;
+    $res= my_exec("$maria_exe_path/$prog");
     print MY_LOG $res;
     # derive table's name from program's name
     if ($prog =~ m/ma_(test[0-9]+).*/)
@@ -104,8 +109,8 @@ sub main
     $com=  "$maria_exe_path/maria_chk$suffix -dvv $table ";
     $com.= "| grep -v \"Creation time:\" | grep -v \"file length\" ";
     $com.= "> $tmp/maria_chk_message.good.txt 2>&1";
-    `$com`;
-    my $checksum=`$maria_exe_path/maria_chk$suffix -dss $table`;
+    my_exec($com);
+    my $checksum= my_exec("$maria_exe_path/maria_chk$suffix -dss $table");
     move("$table.MAD", "$tmp/$table-good.MAD") ||
       die "Can't move $table.MAD to $tmp/$table-good.MAD\n";
     move("$table.MAI", "$tmp/$table-good.MAI") ||
@@ -137,6 +142,9 @@ sub main
            "ma_test1$suffix $silent -M -T -c -N blob -H2",
            "--testflag=3",
            "--testflag=4 --test-undo=",
+           "ma_test1$suffix $silent -M -T -c -N blob -H2 --versioning",
+           "--testflag=3",
+           "--testflag=4 --test-undo=",
            "ma_test1$suffix $silent -M -T -c -N blob -H2",
            "--testflag=2",
            "--testflag=3 --test-undo=",
@@ -170,7 +178,7 @@ sub main
           my $prog_no_suffix= $prog;
           $prog_no_suffix=~ s/$suffix// if ($suffix);
           print MY_LOG "TEST WITH $prog_no_suffix $commit_run_args (commit at end)\n";
-          $res= `$maria_exe_path/$prog $commit_run_args`;
+          $res= my_exec("$maria_exe_path/$prog $commit_run_args");
           print MY_LOG $res;
           # derive table's name from program's name
           if ($prog =~ m/ma_(test[0-9]+).*/)
@@ -180,16 +188,16 @@ sub main
           $com=  "$maria_exe_path/maria_chk$suffix -dvv $table ";
           $com.= "| grep -v \"Creation time:\" | grep -v \"file length\" ";
           $com.= "> $tmp/maria_chk_message.good.txt 2>&1";
-          $res= `$com`;
+          $res= my_exec($com);
           print MY_LOG $res;
-          $checksum= `$maria_exe_path/maria_chk$suffix -dss $table`;
+          $checksum= my_exec("$maria_exe_path/maria_chk$suffix -dss $table");
           move("$table.MAD", "$tmp/$table-good.MAD") ||
             die "Can't move $table.MAD to $tmp/$table-good.MAD\n";
           move("$table.MAI", "$tmp/$table-good.MAI") ||
             die "Can't move $table.MAI to $tmp/$table-good.MAI\n";
           unlink <maria_log.* maria_log_control>;
           print MY_LOG "TEST WITH $prog_no_suffix $abort_run_args$test_undo[$j] (additional aborted work)\n";
-          $res= `$maria_exe_path/$prog $abort_run_args$test_undo[$j]`;
+          $res= my_exec("$maria_exe_path/$prog $abort_run_args$test_undo[$j]");
           print MY_LOG $res;
           copy("$table.MAD", "$tmp/$table-before_undo.MAD") ||
             die "Can't copy $table.MAD to $tmp/$table-before_undo.MAD\n";
@@ -271,6 +279,11 @@ sub check_table_is_same
   # Data/key file length is random in ma_test2 (as it uses srand() which
   # may differ between machines).
 
+  if ($opt_verbose)
+  {
+    print "checking if table $table has changed\n";
+  }
+
   $com=  "$maria_exe_path/maria_chk$suffix -dvv $table | grep -v \"Creation time:\" ";
   $com.= "| grep -v \"file length\"> $tmp/maria_chk_message.txt 2>&1";
   $res= `$com`;
@@ -326,7 +339,7 @@ sub apply_log
     $log_md5.= md5_conv($_);
   }
   print MY_LOG "applying log\n";
-  `$maria_exe_path/maria_read_log$suffix -a > $tmp/maria_read_log_$table.txt`;
+  my_exec("$maria_exe_path/maria_read_log$suffix -a > $tmp/maria_read_log_$table.txt");
   foreach (<maria_log.*>)
   {
     $log_md5_2.= md5_conv($_);
@@ -414,6 +427,23 @@ sub physical_cmp
 }
 
 
+sub my_exec
+{
+  my($command)= @_;
+  my $res;
+  if ($opt_verbose)
+  {
+    print "$command\n";
+  }
+  $res= `$command`;
+  if ($? != 0 && $opt_abort_on_error)
+  {
+    exit(1);
+  }
+  return $res;
+}
+
+
 ####
 #### usage
 ####
@@ -429,7 +459,11 @@ Run various maria recovery tests and print the results
 
 Options
 --help             Show this help and exit.
+
+--abort-on-error   Abort at once in case of error.
+--verbose          Show commands while there are executing.
 --version          Show version number and exit.
+
 EOF
   exit(0);
 }
diff --git a/storage/myisam/mi_locking.c b/storage/myisam/mi_locking.c
index ad23a42ab83..2edeefad45f 100644
--- a/storage/myisam/mi_locking.c
+++ b/storage/myisam/mi_locking.c
@@ -281,7 +281,7 @@ int mi_lock_database(MI_INFO *info, int lock_type)
 			(THR_WRITE_CONCURRENT_INSERT was used)
 */
 
-void mi_get_status(void* param, int concurrent_insert)
+void mi_get_status(void* param, my_bool concurrent_insert)
 {
   MI_INFO *info=(MI_INFO*) param;
   DBUG_ENTER("mi_get_status");
diff --git a/storage/myisam/myisamdef.h b/storage/myisam/myisamdef.h
index b0b9226749d..db6344e14f8 100644
--- a/storage/myisam/myisamdef.h
+++ b/storage/myisam/myisamdef.h
@@ -700,7 +700,7 @@ int _mi_cmp_dynamic_unique(MI_INFO *info, MI_UNIQUEDEF *def,
                            const uchar *record, my_off_t pos);
 int mi_unique_comp(MI_UNIQUEDEF *def, const uchar *a, const uchar *b,
                    my_bool null_are_equal);
-void mi_get_status(void *param, int concurrent_insert);
+void mi_get_status(void *param, my_bool concurrent_insert);
 void mi_update_status(void *param);
 void mi_restore_status(void *param);
 void mi_copy_status(void *to, void *from);