Fixed some bugs in the Maria storage engine

- Changed default recovery mode from OFF to NORMAL to get automatic repair of not properly closed tables.
- Fixed a rase condition when two threads calls external_lock and thr_lock() in different order. When this happend the transaction that called external lock first
  and thr_lock() last did not see see the rows from the other transaction, even if if it had to wait in thr_lock() for other to complete.
- Fixed that one can run maria_chk on an automatcally recovered tables without warnings about too small transaction id
- Don't give warning that crashed table could not be repaired if repair was disabled (and thus not run)
- Fixed a error result from flush_key_cache() which caused a DBUG_ASSERT() when one was using concurrent reads on non transactional tables that was updated.

client/mysqldump.c:
  Add "" around error message to make it more readable
client/mysqltest.cc:
  Free environment variables
mysql-test/r/mysqldump.result:
  Updated results
mysql-test/r/openssl_1.result:
  Updated results
mysql-test/suite/maria/r/maria-recover.result:
  Updated results
mysql-test/suite/maria/r/maria3.result:
  Updated results
mysql-test/suite/maria/t/maria3.test:
  Added more test of temporary tables
storage/maria/ha_maria.cc:
  Changed default recovery mode from OFF to NORMAL to get automatic repair of not properly closed tables.
  Start transaction in ma_block_get_status() instead of in ha_maria::external_lock().
  - This fixes a rase condition when two threads calls external lock and thr_lock() in different order. When this happend the transaction that called external lock first and thr_lock() last did not see see the rows from the other transaction, even if if it had to wait in thr_lock() for other to complete.
  Store latest transaction id in controll file if recovery was done.
  - This allows one to run maria_chk on an automatcally recovered tables without warnings about too small transaction id
storage/maria/ha_maria.h:
  Don't give warning that crashed table could not be repaired if repair was disabled (and thus not run)
storage/maria/ma_blockrec.h:
  Added new function "_ma_block_get_status_no_versioning()"
storage/maria/ma_init.c:
  Added hook to create trn in ma_block_get_status() if we are using MariaDB
storage/maria/ma_open.c:
  Ensure we call _ma_block_get_status_no_versioning() for transactional tables without versioning (like tables with fulltext)
storage/maria/ma_pagecache.c:
  Allow one to flush blocks that are pinned for read.
  This fixed a error result from flush_key_cache() which caused a DBUG_ASSERT() when one was using concurrent reads on non transactional tables that was updated.
storage/maria/ma_recovery.c:
  Set maria_recovery_changed_data to 1 if recover changed something.
  Set max_trid_in_control_file to max found trn if we found a bigger trn.
  The allows will ensure that the control file is up to date after recovery which allows one to run maria_chk on the tables without warnings about too big trn
storage/maria/ma_state.c:
  Call maria_create_trn_hook() in _ma_setup_live_state() instead of ha_maria::external_lock()
  This ensures that 'state' and trn are in sync and thus fixes the race condition mentioned for ha_maria.cc
storage/maria/ma_static.c:
  Added maria_create_trn_hook() and maria_recovery_changed_data
storage/maria/maria_def.h:
  Added MARIA_HANDLER->external_ptr, which is used to hold MariaDB thd.
  Added some new external variables
  Removed reference to non existing function: maria_concurrent_inserts()

17 files changed, 190 insertions, 52 deletions

diff --git a/client/mysqldump.c b/client/mysqldump.c
index dca8ecb7dfb..212baca6be8 100644
--- a/client/mysqldump.c
+++ b/client/mysqldump.c
@@ -964,7 +964,7 @@ static int get_options(int *argc, char ***argv)
 static void DB_error(MYSQL *mysql_arg, const char *when)
 {
   DBUG_ENTER("DB_error");
-  maybe_die(EX_MYSQLERR, "Got error: %d: %s %s",
+  maybe_die(EX_MYSQLERR, "Got error: %d: \"%s\" %s",
           mysql_errno(mysql_arg), mysql_error(mysql_arg), when);
   DBUG_VOID_RETURN;
 }
diff --git a/client/mysqltest.cc b/client/mysqltest.cc
index fba1b579add..8ac3c3ebd92 100644
--- a/client/mysqltest.cc
+++ b/client/mysqltest.cc
@@ -2006,6 +2006,7 @@ VAR *var_init(VAR *v, const char *name, int name_len, const char *val,
 void var_free(void *v)
 {
   my_free(((VAR*) v)->str_val, MYF(MY_WME));
+  my_free(((VAR*) v)->env_s, MYF(MY_ALLOW_ZERO_PTR));
   if (((VAR*)v)->alloced)
     my_free(v, MYF(MY_WME));
 }
diff --git a/mysql-test/r/mysqldump.result b/mysql-test/r/mysqldump.result
index 328f7e2a675..ad77d04d583 100644
--- a/mysql-test/r/mysqldump.result
+++ b/mysql-test/r/mysqldump.result
@@ -1658,8 +1658,8 @@ mysqldump: Couldn't find table: "T_1"
 mysqldump: Couldn't find table: "T_"
 test_sequence
 ------ Testing with illegal database names ------
-mysqldump: Got error: 1049: Unknown database 'mysqldump_test_d' when selecting the database
-mysqldump: Got error: 1049: Unknown database 'mysqld\ump_test_db' when selecting the database
+mysqldump: Got error: 1049: "Unknown database 'mysqldump_test_d'" when selecting the database
+mysqldump: Got error: 1049: "Unknown database 'mysqld\ump_test_db'" when selecting the database
 drop table t1, t2, t3;
 drop database mysqldump_test_db;
 use test;
@@ -1833,7 +1833,7 @@ drop table t1, t2, t3;
 #
 create table t1 (a int);
 mysqldump: Couldn't execute 'SELECT /*!40001 SQL_NO_CACHE */ * FROM `t1` WHERE xx xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx': You have an error in your SQL syntax; check the manual that corresponds to your MySQL server version for the right syntax to use near 'xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx' at line 1 (1064)
-mysqldump: Got error: 1064: You have an error in your SQL syntax; check the manual that corresponds to your MySQL server version for the right syntax to use near 'xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx' at line 1 when retrieving data from server
+mysqldump: Got error: 1064: "You have an error in your SQL syntax; check the manual that corresponds to your MySQL server version for the right syntax to use near 'xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx' at line 1" when retrieving data from server
 
 /*!40101 SET @OLD_CHARACTER_SET_CLIENT=@@CHARACTER_SET_CLIENT */;
 /*!40101 SET @OLD_CHARACTER_SET_RESULTS=@@CHARACTER_SET_RESULTS */;
diff --git a/mysql-test/r/openssl_1.result b/mysql-test/r/openssl_1.result
index d68b436d089..d87aeefbc34 100644
--- a/mysql-test/r/openssl_1.result
+++ b/mysql-test/r/openssl_1.result
@@ -189,7 +189,7 @@ UNLOCK TABLES;
 /*!40111 SET SQL_NOTES=@OLD_SQL_NOTES */;
 
 SSL error: Unable to get private key from 'MYSQL_TEST_DIR/std_data/client-cert.pem'
-mysqldump: Got error: 2026: SSL connection error when trying to connect
+mysqldump: Got error: 2026: "SSL connection error" when trying to connect
 DROP TABLE t1;
 Variable_name	Value
 Ssl_cipher	DHE-RSA-AES256-SHA
diff --git a/mysql-test/suite/maria/r/maria-recover.result b/mysql-test/suite/maria/r/maria-recover.result
index aee4119304f..e9bfc019c81 100644
--- a/mysql-test/suite/maria/r/maria-recover.result
+++ b/mysql-test/suite/maria/r/maria-recover.result
@@ -8,7 +8,7 @@ OFF
 set global maria_recover=default;
 select @@global.maria_recover;
 @@global.maria_recover
-OFF
+NORMAL
 set global maria_recover=normal;
 select @@global.maria_recover;
 @@global.maria_recover
diff --git a/mysql-test/suite/maria/r/maria3.result b/mysql-test/suite/maria/r/maria3.result
index 58e170cd3c3..e632363346d 100644
--- a/mysql-test/suite/maria/r/maria3.result
+++ b/mysql-test/suite/maria/r/maria3.result
@@ -313,7 +313,7 @@ maria_pagecache_age_threshold	300
 maria_pagecache_buffer_size	8384512
 maria_pagecache_division_limit	100
 maria_page_checksum	OFF
-maria_recover	OFF
+maria_recover	NORMAL
 maria_repair_threads	1
 maria_sort_buffer_size	8388608
 maria_stats_method	nulls_unequal
@@ -549,6 +549,30 @@ select count(*) from t1 where a >= 4;
 count(*)
 1
 drop table t1, t2;
+create temporary table t1 (a int, key(a)) transactional=0 row_format=page;
+create temporary table t2 (a int, key(a)) transactional=0 row_format=page;
+insert into t1 values (0),(1),(2),(3),(4);
+insert into t2 select * from t1;
+insert into t1 select NULL from t2;
+select count(*) from t1;
+count(*)
+10
+select count(*) from t1 where a >= 4;
+count(*)
+1
+drop table t1, t2;
+create temporary table t1 (a int, key(a)) transactional=0 row_format=fixed;
+create temporary table t2 (a int, key(a)) transactional=0 row_format=dynamic;
+insert into t1 values (0),(1),(2),(3),(4);
+insert into t2 select * from t1;
+insert into t1 select NULL from t2;
+select count(*) from t1;
+count(*)
+10
+select count(*) from t1 where a >= 4;
+count(*)
+1
+drop table t1, t2;
 create table t1 (i int auto_increment not null primary key) transactional=0;
 check table t1 extended;
 Table	Op	Msg_type	Msg_text
diff --git a/mysql-test/suite/maria/t/maria3.test b/mysql-test/suite/maria/t/maria3.test
index 45765b3b868..f2ae5d73fb2 100644
--- a/mysql-test/suite/maria/t/maria3.test
+++ b/mysql-test/suite/maria/t/maria3.test
@@ -433,6 +433,24 @@ select count(*) from t1;
 select count(*) from t1 where a >= 4;
 drop table t1, t2;
 
+create temporary table t1 (a int, key(a)) transactional=0 row_format=page;
+create temporary table t2 (a int, key(a)) transactional=0 row_format=page;
+insert into t1 values (0),(1),(2),(3),(4);
+insert into t2 select * from t1;
+insert into t1 select NULL from t2;
+select count(*) from t1;
+select count(*) from t1 where a >= 4;
+drop table t1, t2;
+
+create temporary table t1 (a int, key(a)) transactional=0 row_format=fixed;
+create temporary table t2 (a int, key(a)) transactional=0 row_format=dynamic;
+insert into t1 values (0),(1),(2),(3),(4);
+insert into t2 select * from t1;
+insert into t1 select NULL from t2;
+select count(*) from t1;
+select count(*) from t1 where a >= 4;
+drop table t1, t2;
+
 #
 # Test problems with small rows and row_type=page 
 # Bug 35048 "maria table corruption reported when transactional=0"
diff --git a/storage/maria/ha_maria.cc b/storage/maria/ha_maria.cc
index 4e18155b46e..e919dbbc3cc 100644
--- a/storage/maria/ha_maria.cc
+++ b/storage/maria/ha_maria.cc
@@ -202,7 +202,7 @@ static MYSQL_SYSVAR_ENUM(recover, maria_recover_options, PLUGIN_VAR_OPCMDARG,
        "Specifies how corrupted tables should be automatically repaired."
        " Possible values are \"NORMAL\" (the default), \"BACKUP\", \"FORCE\","
        " \"QUICK\", or \"OFF\" which is like not using the option.",
-       NULL, NULL, HA_RECOVER_NONE, &maria_recover_typelib);
+       NULL, NULL, HA_RECOVER_DEFAULT, &maria_recover_typelib);
 
 static MYSQL_THDVAR_ULONG(repair_threads, PLUGIN_VAR_RQCMDARG,
        "Number of threads to use when repairing maria tables. The value of 1 "
@@ -707,8 +707,54 @@ void _ma_check_print_warning(HA_CHECK *param, const char *fmt, ...)
   DBUG_VOID_RETURN;
 }
 
+/*
+  Create a transaction object
+
+  SYNOPSIS
+    info	Maria handler
+
+  RETURN
+    0 		ok
+    #		Error number (HA_ERR_OUT_OF_MEM)
+*/
+
+static int maria_create_trn_for_mysql(MARIA_HA *info)
+{
+  THD *thd= (THD*) info->external_ptr;
+  TRN *trn= THD_TRN;
+  DBUG_ENTER("maria_create_trn_for_mysql");
+
+  if (!trn)  /* no transaction yet - open it now */
+  {
+    trn= trnman_new_trn(& thd->transaction.wt);
+    if (unlikely(!trn))
+      DBUG_RETURN(HA_ERR_OUT_OF_MEM);
+    THD_TRN= trn;
+    if (thd->options & (OPTION_NOT_AUTOCOMMIT | OPTION_BEGIN))
+      trans_register_ha(thd, TRUE, maria_hton);
+  }
+  _ma_set_trn_for_table(info, trn);
+  if (!trnman_increment_locked_tables(trn))
+  {
+    trans_register_ha(thd, FALSE, maria_hton);
+    trnman_new_statement(trn);
+  }
+#ifdef EXTRA_DEBUG
+  if (info->lock_type == F_WRLCK &&
+      ! (trnman_get_flags(trn) & TRN_STATE_INFO_LOGGED))
+  {
+    trnman_set_flags(trn, trnman_get_flags(trn) | TRN_STATE_INFO_LOGGED |
+                     TRN_STATE_TABLES_CAN_CHANGE);
+    (void) translog_log_debug_info(trn, LOGREC_DEBUG_INFO_QUERY,
+                                   (uchar*) thd->query(),
+                                   thd->query_length());
+  }
+#endif
+  DBUG_RETURN(0);
 }
 
+} /* extern "C" */
+
 /**
   Transactional table doing bulk insert with one single UNDO
   (UNDO_BULK_INSERT) and with repair.
@@ -2313,9 +2359,9 @@ int ha_maria::delete_table(const char *name)
   return maria_delete_table(name);
 }
 
+
 int ha_maria::external_lock(THD *thd, int lock_type)
 {
-  TRN *trn= THD_TRN;
   DBUG_ENTER("ha_maria::external_lock");
   /*
     We don't test now_transactional because it may vary between lock/unlock
@@ -2335,22 +2381,7 @@ int ha_maria::external_lock(THD *thd, int lock_type)
     /* Transactional table */
     if (lock_type != F_UNLCK)
     {
-      /* Start of new statement */
-      if (!trn)  /* no transaction yet - open it now */
-      {
-        trn= trnman_new_trn(& thd->transaction.wt);
-        if (unlikely(!trn))
-          DBUG_RETURN(HA_ERR_OUT_OF_MEM);
-        THD_TRN= trn;
-        if (thd->options & (OPTION_NOT_AUTOCOMMIT | OPTION_BEGIN))
-          trans_register_ha(thd, TRUE, maria_hton);
-      }
-      _ma_set_trn_for_table(file, trn);
-      if (!trnman_increment_locked_tables(trn))
-      {
-        trans_register_ha(thd, FALSE, maria_hton);
-        trnman_new_statement(trn);
-      }
+      file->external_ptr= thd;                  // For maria_register_trn()
 
       if (!file->s->lock_key_trees)             // If we don't use versioning
       {
@@ -2383,20 +2414,10 @@ int ha_maria::external_lock(THD *thd, int lock_type)
         DBUG_PRINT("info", ("Disabling logging for table"));
         _ma_tmp_disable_logging_for_table(file, TRUE);
       }
-#ifdef EXTRA_DEBUG
-      if (lock_type == F_WRLCK &&
-          ! (trnman_get_flags(trn) & TRN_STATE_INFO_LOGGED))
-      {
-        trnman_set_flags(trn, trnman_get_flags(trn) | TRN_STATE_INFO_LOGGED |
-                         TRN_STATE_TABLES_CAN_CHANGE);
-        (void) translog_log_debug_info(trn, LOGREC_DEBUG_INFO_QUERY,
-                                       (uchar*) thd->query(),
-                                       thd->query_length());
-      }
-#endif
     }
     else
     {
+      TRN *trn= THD_TRN;
       /* End of transaction */
 
       /*
@@ -2421,6 +2442,8 @@ int ha_maria::external_lock(THD *thd, int lock_type)
       file->state= &file->s->state.state;
       if (trn)
       {
+        DBUG_PRINT("info",
+                   ("locked_tables: %u", trnman_has_locked_tables(trn)));
         if (trnman_has_locked_tables(trn) &&
             !trnman_decrement_locked_tables(trn))
         {
@@ -3187,9 +3210,11 @@ static int ha_maria_init(void *p)
                   MYSQL_VERSION_ID, server_id, maria_log_pagecache,
                   TRANSLOG_DEFAULT_FLAGS, 0) ||
     maria_recovery_from_log() ||
-    ((force_start_after_recovery_failures != 0) && mark_recovery_success()) ||
+    ((force_start_after_recovery_failures != 0 ||
+      maria_recovery_changed_data) && mark_recovery_success()) ||
     ma_checkpoint_init(checkpoint_interval);
   maria_multi_threaded= maria_in_ha_maria= TRUE;
+  maria_create_trn_hook= maria_create_trn_for_mysql;
 
 #if defined(HAVE_REALPATH) && !defined(HAVE_valgrind) && !defined(HAVE_BROKEN_REALPATH)
   /*  We can only test for sub paths if my_symlink.c is using realpath */
diff --git a/storage/maria/ha_maria.h b/storage/maria/ha_maria.h
index 23fb74f1c27..ab4012ca38c 100644
--- a/storage/maria/ha_maria.h
+++ b/storage/maria/ha_maria.h
@@ -141,7 +141,7 @@ public:
   bool check_and_repair(THD * thd);
   bool is_crashed() const;
   bool is_changed() const;
-  bool auto_repair() const { return 1; }
+  bool auto_repair() const { return maria_recover_options != HA_RECOVER_NONE; }
   int optimize(THD * thd, HA_CHECK_OPT * check_opt);
   int restore(THD * thd, HA_CHECK_OPT * check_opt);
   int backup(THD * thd, HA_CHECK_OPT * check_opt);
diff --git a/storage/maria/ma_blockrec.h b/storage/maria/ma_blockrec.h
index cb682eef701..16dedc93b28 100644
--- a/storage/maria/ma_blockrec.h
+++ b/storage/maria/ma_blockrec.h
@@ -279,7 +279,8 @@ my_bool write_hook_for_file_id(enum translog_record_type type,
 my_bool write_hook_for_commit(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_get_status(void *param, my_bool concurrent_insert);
+void _ma_block_get_status_no_versioning(void *param, my_bool concurrent_ins);
 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_init.c b/storage/maria/ma_init.c
index 1f2eddd7e30..9cec978166f 100644
--- a/storage/maria/ma_init.c
+++ b/storage/maria/ma_init.c
@@ -40,6 +40,11 @@ void history_state_free(MARIA_STATE_HISTORY_CLOSED *closed_history)
 }
 
 
+static int dummy_maria_create_trn_hook(MARIA_HA *info __attribute__((unused)))
+{
+  return 0;
+}
+
 /*
   Initialize maria
 
@@ -64,6 +69,7 @@ int maria_init(void)
     pthread_mutex_init(&THR_LOCK_maria,MY_MUTEX_INIT_SLOW);
     _ma_init_block_record_data();
     trnman_end_trans_hook= _ma_trnman_end_trans_hook;
+    maria_create_trn_hook= dummy_maria_create_trn_hook;
     my_handler_error_register();
   }
   hash_init(&maria_stored_state, &my_charset_bin, 32,
diff --git a/storage/maria/ma_open.c b/storage/maria/ma_open.c
index c86c439a702..98ccb3e23cd 100644
--- a/storage/maria/ma_open.c
+++ b/storage/maria/ma_open.c
@@ -892,6 +892,11 @@ MARIA_HA *maria_open(const char *name, int mode, uint open_flags)
           share->lock_restore_status= _ma_restore_status;
         }
       }
+      else if (share->now_transactional)
+      {
+        DBUG_ASSERT(share->data_file_type == BLOCK_RECORD);
+        share->lock.get_status=     _ma_block_get_status_no_versioning;
+      }
     }
 #endif
     /*
diff --git a/storage/maria/ma_pagecache.c b/storage/maria/ma_pagecache.c
index dd676028f05..b95a6f53c06 100644
--- a/storage/maria/ma_pagecache.c
+++ b/storage/maria/ma_pagecache.c
@@ -4187,7 +4187,13 @@ static int flush_cached_blocks(PAGECACHE *pagecache,
   {
     PAGECACHE_BLOCK_LINK *block= *cache;
 
-    if (block->pins)
+    /*
+      This code is only run for non transactional tables.
+      We may have other threads reading the block during flush,
+      as non transactional tables can have many readers while the
+      one writer is doing the flush.
+    */
+    if (block->wlocks)
     {
       KEYCACHE_DBUG_PRINT("flush_cached_blocks",
                           ("block: %u (0x%lx)  pinned",
@@ -4204,13 +4210,9 @@ static int flush_cached_blocks(PAGECACHE *pagecache,
         *first_errno= HA_ERR_INTERNAL_ERROR;
       continue;
     }
-    /* if the block is not pinned then it is not write locked */
-    DBUG_ASSERT(block->wlocks == 0);
-    DBUG_ASSERT(block->pins == 0);
     if (make_lock_and_pin(pagecache, block,
-                          PAGECACHE_LOCK_WRITE, PAGECACHE_PIN, FALSE))
+                          PAGECACHE_LOCK_READ, PAGECACHE_PIN, FALSE))
       DBUG_ASSERT(0);
-    DBUG_ASSERT(block->pins == 1);
 
     KEYCACHE_DBUG_PRINT("flush_cached_blocks",
                         ("block: %u (0x%lx)  to be flushed",
@@ -4222,7 +4224,6 @@ static int flush_cached_blocks(PAGECACHE *pagecache,
     DBUG_PRINT("info", ("block: %u (0x%lx)  pins: %u",
                         PCBLOCK_NUMBER(pagecache, block), (ulong)block,
                         block->pins));
-    DBUG_ASSERT(block->pins == 1);
     /**
        @todo IO If page is contiguous with next page to flush, group flushes
        in one single my_pwrite().
@@ -4241,7 +4242,7 @@ static int flush_cached_blocks(PAGECACHE *pagecache,
     pagecache_pthread_mutex_lock(&pagecache->cache_lock);
 
     if (make_lock_and_pin(pagecache, block,
-                          PAGECACHE_LOCK_WRITE_UNLOCK,
+                          PAGECACHE_LOCK_READ_UNLOCK,
                           PAGECACHE_UNPIN, FALSE))
       DBUG_ASSERT(0);
 
diff --git a/storage/maria/ma_recovery.c b/storage/maria/ma_recovery.c
index 7b3065b0208..e74f52b2ef6 100644
--- a/storage/maria/ma_recovery.c
+++ b/storage/maria/ma_recovery.c
@@ -266,6 +266,7 @@ int maria_apply_log(LSN from_lsn, enum maria_apply_log_way apply,
   DBUG_ASSERT(apply == MARIA_LOG_APPLY || !should_run_undo_phase);
   DBUG_ASSERT(!maria_multi_threaded);
   recovery_warnings= 0;
+  maria_recovery_changed_data= 0;
   /* checkpoints can happen only if TRNs have been built */
   DBUG_ASSERT(should_run_undo_phase || !take_checkpoints);
   all_active_trans= (struct st_trn_for_recovery *)
@@ -465,8 +466,18 @@ end:
       fflush(stderr);
     }
     if (!error)
+    {
       ma_message_no_user(ME_JUST_INFO, "recovery done");
+      maria_recovery_changed_data= 1;
+    }
   }
+  else if (!error && max_trid_in_control_file != max_long_trid)
+  {
+    /* Set max trid in log file so that one can run maria_chk on the tables */
+    max_trid_in_control_file= trnman_get_max_trid();
+    maria_recovery_changed_data= 1;
+  }
+
   if (error)
     my_message(HA_ERR_INITIALIZATION,
                "Maria recovery failed. Please run maria_chk -r on all maria "
diff --git a/storage/maria/ma_state.c b/storage/maria/ma_state.c
index d7ddc73e2b4..c8fcf1a6818 100644
--- a/storage/maria/ma_state.c
+++ b/storage/maria/ma_state.c
@@ -53,12 +53,16 @@
 
 my_bool _ma_setup_live_state(MARIA_HA *info)
 {
-  TRN *trn= info->trn;
+  TRN *trn;
   MARIA_SHARE *share= info->s;
   MARIA_USED_TABLES *tables;
   MARIA_STATE_HISTORY *history;
   DBUG_ENTER("_ma_setup_live_state");
 
+  if (maria_create_trn_hook(info))
+    DBUG_RETURN(1);
+
+  trn= info->trn;
   for (tables= (MARIA_USED_TABLES*) info->trn->used_tables;
        tables;
        tables= tables->next)
@@ -69,6 +73,7 @@ my_bool _ma_setup_live_state(MARIA_HA *info)
       goto end;
     }
   }
+
   /* Table was not used before, create new table state entry */
   if (!(tables= (MARIA_USED_TABLES*) my_malloc(sizeof(*tables),
                                                MYF(MY_WME | MY_ZEROFILL))))
@@ -566,7 +571,8 @@ 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));
+  DBUG_PRINT("enter", ("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)
@@ -589,6 +595,21 @@ void _ma_block_get_status(void* param, my_bool concurrent_insert)
     */
     (void) _ma_setup_live_state(info);
   }
+  else
+  {
+    /*
+      Info->trn is set if this table is already handled and we are
+      called from maria_versioning()
+    */
+    if (info->s->base.born_transactional && !info->trn)
+    {
+      /*
+        Assume for now that this doesn't fail (It can only fail in
+        out of memory conditions)
+      */
+      (void) maria_create_trn_hook(info);
+    }
+  }
   DBUG_VOID_RETURN;
 }
 
@@ -616,6 +637,28 @@ my_bool _ma_block_check_status(void *param __attribute__((unused)))
 }
 
 
+/* Get status when transactional but not versioned */
+
+void _ma_block_get_status_no_versioning(void* param, my_bool concurrent_insert)
+{
+  MARIA_HA *info=(MARIA_HA*) param;
+  DBUG_ENTER("_ma_block_get_status_no_version");
+  DBUG_PRINT("enter", ("concurrent_insert %d", concurrent_insert));
+  DBUG_ASSERT(info->s->base.born_transactional);
+
+  info->state->changed= 0;              /* from _ma_reset_update_flag() */
+  if (!info->trn)
+  {
+    /*
+      Assume for now that this doesn't fail (It can only fail in
+      out of memory conditions)
+    */
+    (void) maria_create_trn_hook(info);
+  }
+  DBUG_VOID_RETURN;
+}
+
+
 /**
   Enable/disable versioning
 */
diff --git a/storage/maria/ma_static.c b/storage/maria/ma_static.c
index b4589469caf..d870b034eb8 100644
--- a/storage/maria/ma_static.c
+++ b/storage/maria/ma_static.c
@@ -37,6 +37,7 @@ my_bool maria_flush= 0, maria_single_user= 0;
 my_bool maria_delay_key_write= 0, maria_page_checksums= 1;
 my_bool maria_inited= FALSE;
 my_bool maria_in_ha_maria= FALSE; /* If used from ha_maria or not */
+my_bool maria_recovery_changed_data= 0;
 pthread_mutex_t THR_LOCK_maria;
 #if defined(THREAD) && !defined(DONT_USE_RW_LOCKS)
 ulong maria_concurrent_insert= 2;
@@ -55,6 +56,7 @@ PAGECACHE *maria_log_pagecache= &maria_log_pagecache_var;
 MY_TMPDIR *maria_tmpdir;                        /* Tempdir for redo */
 char *maria_data_root;
 HASH maria_stored_state;
+int (*maria_create_trn_hook)(MARIA_HA *);
 
 /**
    @brief when transactionality does not matter we can use this transaction
diff --git a/storage/maria/maria_def.h b/storage/maria/maria_def.h
index db6845cb345..868885c961b 100644
--- a/storage/maria/maria_def.h
+++ b/storage/maria/maria_def.h
@@ -482,7 +482,8 @@ typedef struct st_maria_block_scan
 struct st_maria_handler
 {
   MARIA_SHARE *s;			/* Shared between open:s */
-  struct st_ma_transaction *trn;           /* Pointer to active transaction */
+  struct st_ma_transaction *trn;        /* Pointer to active transaction */
+  void *external_ptr;           	/* Pointer to THD in mysql */
   MARIA_STATUS_INFO *state, state_save;
   MARIA_STATUS_INFO *state_start;       /* State at start of transaction */
   MARIA_ROW cur_row;                    /* The active row that we just read */
@@ -788,8 +789,9 @@ extern uint32 maria_read_vec[], maria_readnext_vec[];
 extern uint maria_quick_table_bits;
 extern char *maria_data_root;
 extern uchar maria_zero_string[];
-extern my_bool maria_inited, maria_in_ha_maria;
+extern my_bool maria_inited, maria_in_ha_maria, maria_recovery_changed_data;
 extern HASH maria_stored_state;
+extern int (*maria_create_trn_hook)(MARIA_HA *);
 
 /* This is used by _ma_calc_xxx_key_length och _ma_store_key */
 typedef struct st_maria_s_param
@@ -1242,5 +1244,4 @@ 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;