diff --git a/mysql-test/suite/maria/r/maria-big.result b/mysql-test/suite/maria/r/maria-big.result
index b45816367a5..aca08b4df72 100644
--- a/mysql-test/suite/maria/r/maria-big.result
+++ b/mysql-test/suite/maria/r/maria-big.result
@@ -1,3 +1,4 @@
+set global max_allowed_packet=400000000;
 set storage_engine=maria;
 affected rows: 0
 set global maria_log_file_size=4294967295;
@@ -61,8 +62,6 @@ count(*)
 affected rows: 1
 drop table t1, t2;
 affected rows: 0
-set @@max_allowed_packet=400000000;
-affected rows: 0
 create table t1 (a int, b longtext);
 affected rows: 0
 insert into t1 values (1,"123456789012345678901234567890"),(2,"09876543210987654321");
diff --git a/mysql-test/suite/maria/r/maria-recovery-big.result b/mysql-test/suite/maria/r/maria-recovery-big.result
index ecbcba9ff82..ecc53de437e 100644
--- a/mysql-test/suite/maria/r/maria-recovery-big.result
+++ b/mysql-test/suite/maria/r/maria-recovery-big.result
@@ -4,7 +4,6 @@ create database mysqltest;
 use mysqltest;
 * TEST of recovery with blobs
 * shut down mysqld, removed logs, restarted it
-set @@max_allowed_packet=32000000;
 create table t1 (a int, b longtext) engine=maria table_checksum=1;
 * copied t1 for feeding_recovery
 insert into t1 values (1,"123456789012345678901234567890"),(2,"09876543210987654321");
diff --git a/mysql-test/suite/maria/r/maria-recovery3.result b/mysql-test/suite/maria/r/maria-recovery3.result
index c1acef57a5a..8c0661cba16 100644
--- a/mysql-test/suite/maria/r/maria-recovery3.result
+++ b/mysql-test/suite/maria/r/maria-recovery3.result
@@ -25,5 +25,43 @@ Checksum-check
 ok
 use mysqltest;
 drop table t1;
+* TEST of logging of BLOBs
+CREATE TABLE `t1` (
+`blob` blob,
+`blob_key` blob
+) ENGINE=maria ROW_FORMAT=page
+;
+* copied t1 for feeding_recovery
+* compared t1 to old version
+set global maria_checkpoint_interval=0;
+INSERT INTO `t1` VALUES (NULL,repeat('A',5198));
+INSERT INTO `t1` VALUES (NULL,repeat('B',65535));
+INSERT INTO `t1` VALUES (repeat('K',5198),repeat('L',2325));
+INSERT INTO `t1` VALUES (repeat('C',65535),NULL);
+INSERT INTO `t1` VALUES (NULL,repeat('D',65535));
+INSERT INTO `t1` VALUES (repeat('E',65535),repeat('F',16111));
+INSERT INTO `t1` VALUES (repeat('G',65535),repeat('H',65535));
+INSERT INTO `t1` VALUES (repeat('I',5198),repeat('J',65535));
+check table t1 extended;
+Table	Op	Msg_type	Msg_text
+mysqltest.t1	check	status	OK
+flush table t1;
+* copied t1 for comparison
+* compared t1 to old version
+SET SESSION debug="+d,maria_flush_whole_log,maria_crash";
+* crashing mysqld intentionally
+set global maria_checkpoint_interval=1;
+ERROR HY000: Lost connection to MySQL server during query
+* copied t1 back for feeding_recovery
+* recovery happens
+check table t1 extended;
+Table	Op	Msg_type	Msg_text
+mysqltest.t1	check	status	OK
+* testing that checksum after recovery is as expected
+Checksum-check
+ok
+* compared t1 to old version
+use mysqltest;
+drop table t1;
 drop database mysqltest_for_comparison;
 drop database mysqltest;
diff --git a/mysql-test/suite/maria/t/maria-big.test b/mysql-test/suite/maria/t/maria-big.test
index aac26ec2e60..92177c0a9af 100644
--- a/mysql-test/suite/maria/t/maria-big.test
+++ b/mysql-test/suite/maria/t/maria-big.test
@@ -2,6 +2,11 @@
 --source include/have_maria.inc
 --source include/big_test.inc
 
+set global max_allowed_packet=400000000;
+# need new session to use setting above
+connect (root,localhost,root,,test,$MASTER_MYPORT,$MASTER_MYSOCK);
+connection root;
+
 enable_info;
 set storage_engine=maria;
 set global maria_log_file_size=4294967295;
@@ -38,7 +43,6 @@ drop table t1, t2;
 # Test creating a really big blob (up to 16M)
 #
 
-set @@max_allowed_packet=400000000;
 create table t1 (a int, b longtext);
 insert into t1 values (1,"123456789012345678901234567890"),(2,"09876543210987654321");
 
diff --git a/mysql-test/suite/maria/t/maria-recovery-big-master.opt b/mysql-test/suite/maria/t/maria-recovery-big-master.opt
index 425fda95086..d24a11c924f 100644
--- a/mysql-test/suite/maria/t/maria-recovery-big-master.opt
+++ b/mysql-test/suite/maria/t/maria-recovery-big-master.opt
@@ -1 +1 @@
---skip-stack-trace --skip-core-file
+--skip-stack-trace --skip-core-file --max_allowed_packet=32000000
diff --git a/mysql-test/suite/maria/t/maria-recovery-big.test b/mysql-test/suite/maria/t/maria-recovery-big.test
index 9a5eeec56c5..4de8f934ec1 100644
--- a/mysql-test/suite/maria/t/maria-recovery-big.test
+++ b/mysql-test/suite/maria/t/maria-recovery-big.test
@@ -33,7 +33,6 @@ use mysqltest;
 
 --echo * TEST of recovery with blobs
 -- source include/maria_empty_logs.inc
-set @@max_allowed_packet=32000000;
 create table t1 (a int, b longtext) engine=maria table_checksum=1;
 let $mms_tables=1;
 -- source include/maria_make_snapshot_for_feeding_recovery.inc
diff --git a/mysql-test/suite/maria/t/maria-recovery2-master.opt b/mysql-test/suite/maria/t/maria-recovery2-master.opt
index 7fece2f8c1b..9023fb74e8b 100644
--- a/mysql-test/suite/maria/t/maria-recovery2-master.opt
+++ b/mysql-test/suite/maria/t/maria-recovery2-master.opt
@@ -1 +1 @@
---skip-stack-trace --skip-core-file --maria-log-dir-path=../tmp
+--skip-stack-trace --skip-core-file --loose-maria-log-dir-path=$MYSQLTEST_VARDIR/tmp
diff --git a/mysql-test/suite/maria/t/maria-recovery3.test b/mysql-test/suite/maria/t/maria-recovery3.test
index 68cc67fd328..baee7978f2b 100644
--- a/mysql-test/suite/maria/t/maria-recovery3.test
+++ b/mysql-test/suite/maria/t/maria-recovery3.test
@@ -39,7 +39,6 @@ let $mvr_restore_old_snapshot=0;
 # UNDO phase prevents physical comparison, normally,
 # so we'll only use checksums to compare.
 let $mms_compare_physically=0;
-let $mvr_crash_statement= set global maria_checkpoint_interval=1;
 create table t1(a int primary key) engine=maria;
 insert into t1 values(1);
 -- source include/maria_make_snapshot_for_comparison.inc
@@ -65,6 +64,30 @@ drop table t1;
 # before checkpoint happens, test should still pass (though it won't
 # reproduce the conditions of the bug).
 
+# Test for BUG#41493 Maria: two recovery failures (wrong logging of BLOB pages)
+--echo * TEST of logging of BLOBs
+let $mvr_restore_old_snapshot=1;
+let $mms_compare_physically=1;
+CREATE TABLE `t1` (
+`blob` blob,
+`blob_key` blob
+) ENGINE=maria ROW_FORMAT=page
+;
+-- source include/maria_make_snapshot_for_feeding_recovery.inc
+set global maria_checkpoint_interval=0; # no checkpoints
+INSERT INTO `t1` VALUES (NULL,repeat('A',5198));
+INSERT INTO `t1` VALUES (NULL,repeat('B',65535));
+INSERT INTO `t1` VALUES (repeat('K',5198),repeat('L',2325));
+INSERT INTO `t1` VALUES (repeat('C',65535),NULL);
+INSERT INTO `t1` VALUES (NULL,repeat('D',65535));
+INSERT INTO `t1` VALUES (repeat('E',65535),repeat('F',16111));
+INSERT INTO `t1` VALUES (repeat('G',65535),repeat('H',65535));
+INSERT INTO `t1` VALUES (repeat('I',5198),repeat('J',65535));
+check table t1 extended;
+-- source include/maria_make_snapshot_for_comparison.inc
+-- source include/maria_verify_recovery.inc
+drop table t1;
+
 # clean up everything
 let $mms_purpose=comparison;
 eval drop database mysqltest_for_$mms_purpose;
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index 175358140e1..b10a7789079 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -3493,6 +3493,12 @@ static TABLE *create_table_from_items(THD *thd, HA_CREATE_INFO *create_info,
                                     MYSQL_LOCK_IGNORE_FLUSH, &not_used)) ||
         hooks->postlock(&table, 1))
   {
+    /* purecov: begin tested */
+    /*
+      This can happen in innodb when you get a deadlock when using same table
+      in insert and select
+    */
+    my_error(ER_CANT_LOCK, MYF(0), my_errno);
     if (*lock)
     {
       mysql_unlock_tables(thd, *lock);
@@ -3502,6 +3508,7 @@ static TABLE *create_table_from_items(THD *thd, HA_CREATE_INFO *create_info,
     if (!create_info->table_existed)
       drop_open_table(thd, table, create_table->db, create_table->table_name);
     DBUG_RETURN(0);
+    /* purecov: end */
   }
   DBUG_RETURN(table);
 }
diff --git a/storage/maria/ha_maria.cc b/storage/maria/ha_maria.cc
index 0725422d1c5..34c7bd1455e 100644
--- a/storage/maria/ha_maria.cc
+++ b/storage/maria/ha_maria.cc
@@ -2264,6 +2264,9 @@ int ha_maria::extra_opt(enum ha_extra_function operation, ulong cache_size)
 
 int ha_maria::delete_all_rows()
 {
+  THD *thd= current_thd;
+  (void) translog_log_debug_info(file->trn, LOGREC_DEBUG_INFO_QUERY,
+                                 (uchar*) thd->query, thd->query_length);
   if (file->s->now_transactional &&
       ((table->in_use->options & (OPTION_NOT_AUTOCOMMIT | OPTION_BEGIN)) ||
        table->in_use->locked_tables))
@@ -2280,6 +2283,9 @@ int ha_maria::delete_all_rows()
 
 int ha_maria::delete_table(const char *name)
 {
+  THD *thd= current_thd;
+  (void) translog_log_debug_info(0, LOGREC_DEBUG_INFO_QUERY,
+                                 (uchar*) thd->query, thd->query_length);
   return maria_delete_table(name);
 }
 
@@ -2353,6 +2359,16 @@ 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
     {
@@ -2377,9 +2393,10 @@ int ha_maria::external_lock(THD *thd, int lock_type)
         external lock of the table
       */
       file->state= &file->s->state.state;
-      if (trn && trnman_has_locked_tables(trn))
+      if (trn)
       {
-        if (!trnman_decrement_locked_tables(trn))
+        if (trnman_has_locked_tables(trn) &&
+            !trnman_decrement_locked_tables(trn))
         {
           /*
             OK should not have been sent to client yet (ACID).
@@ -2402,6 +2419,7 @@ int ha_maria::external_lock(THD *thd, int lock_type)
           }
 #endif
         }
+        trnman_set_flags(trn, trnman_get_flags(trn) & ~ TRN_STATE_INFO_LOGGED);
       }
     }
   } /* if transactional table */
@@ -2436,6 +2454,16 @@ int ha_maria::start_stmt(THD *thd, thr_lock_type lock_type)
       call to start_stmt().
     */
     trnman_new_statement(trn);
+
+#ifdef EXTRA_DEBUG
+    if (!(trnman_get_flags(trn) & TRN_STATE_INFO_LOGGED) &&
+        trnman_get_flags(trn) & TRN_STATE_TABLES_CAN_CHANGE)
+    {
+      trnman_set_flags(trn, trnman_get_flags(trn) | TRN_STATE_INFO_LOGGED);
+      (void) translog_log_debug_info(trn, LOGREC_DEBUG_INFO_QUERY,
+                                     (uchar*) thd->query, thd->query_length);
+    }
+#endif
   }
   return 0;
 }
@@ -2648,6 +2676,7 @@ int ha_maria::create(const char *name, register TABLE *table_arg,
   TABLE_SHARE *share= table_arg->s;
   uint options= share->db_options_in_use;
   enum data_file_type row_type;
+  THD *thd= current_thd;
   DBUG_ENTER("ha_maria::create");
 
   for (i= 0; i < share->keys; i++)
@@ -2712,6 +2741,9 @@ int ha_maria::create(const char *name, register TABLE *table_arg,
        ha_create_info->page_checksum ==  HA_CHOICE_YES)
     create_flags|= HA_CREATE_PAGE_CHECKSUM;
 
+  (void) translog_log_debug_info(0, LOGREC_DEBUG_INFO_QUERY,
+                                 (uchar*) thd->query, thd->query_length);
+
   /* TODO: Check that the following fn_format is really needed */
   error=
     maria_create(fn_format(buff, name, "", "",
@@ -2728,6 +2760,9 @@ int ha_maria::create(const char *name, register TABLE *table_arg,
 
 int ha_maria::rename_table(const char *from, const char *to)
 {
+  THD *thd= current_thd;
+  (void) translog_log_debug_info(0, LOGREC_DEBUG_INFO_QUERY,
+                                 (uchar*) thd->query, thd->query_length);
   return maria_rename(from, to);
 }
 
@@ -2872,6 +2907,8 @@ static int maria_commit(handlerton *hton __attribute__ ((unused)),
   TRN *trn= THD_TRN;
   DBUG_ENTER("maria_commit");
   trnman_reset_locked_tables(trn, 0);
+  trnman_set_flags(trn, trnman_get_flags(trn) & ~TRN_STATE_INFO_LOGGED);
+
   /* statement or transaction ? */
   if ((thd->options & (OPTION_NOT_AUTOCOMMIT | OPTION_BEGIN)) && !all)
     DBUG_RETURN(0); // end of statement
diff --git a/storage/maria/ma_blockrec.c b/storage/maria/ma_blockrec.c
index b8538a52cb0..0b338e633d7 100644
--- a/storage/maria/ma_blockrec.c
+++ b/storage/maria/ma_blockrec.c
@@ -3146,8 +3146,8 @@ static my_bool write_block_record(MARIA_HA *info,
 
           log_pos= store_page_range(log_pos, tmp_block, block_size,
                                     blob_length, &extents);
-          tmp_block+= tmp_block->sub_blocks;
         }
+        tmp_block+= tmp_block->sub_blocks;
       }
     }
 
@@ -6182,6 +6182,7 @@ err:
                              PAGECACHE_UNPIN, LSN_IMPOSSIBLE,
                              LSN_IMPOSSIBLE, 0, FALSE);
   _ma_mark_file_crashed(share);
+  DBUG_ASSERT(0); /* catch recovery errors early */
   DBUG_RETURN((my_errno= error));
 }
 
@@ -6280,6 +6281,7 @@ err:
                            PAGECACHE_UNPIN, LSN_IMPOSSIBLE,
                            LSN_IMPOSSIBLE, 0, FALSE);
   _ma_mark_file_crashed(share);
+  DBUG_ASSERT(0);
   DBUG_RETURN((my_errno= error));
 
 }
@@ -6337,6 +6339,7 @@ uint _ma_apply_redo_free_blocks(MARIA_HA *info,
     if (res)
     {
       _ma_mark_file_crashed(share);
+      DBUG_ASSERT(0);
       DBUG_RETURN(res);
     }
   }
@@ -6420,6 +6423,7 @@ uint _ma_apply_redo_free_head_or_tail(MARIA_HA *info, LSN lsn,
 
 err:
   _ma_mark_file_crashed(share);
+  DBUG_ASSERT(0);
   DBUG_RETURN(1);
 }
 
@@ -6431,6 +6435,10 @@ err:
    @parma  lsn             LSN to put on pages
    @param  header          Header (with FILEID)
    @param  redo_lsn        REDO record's LSN
+   @param[out] number_of_blobs Number of blobs found in log record
+   @param[out] number_of_ranges Number of ranges found
+   @param[out] first_page  First page touched
+   @param[out] last_page   Last page touched
 
    @note Write full pages (full head & blob pages)
 
@@ -6441,13 +6449,18 @@ err:
 
 uint _ma_apply_redo_insert_row_blobs(MARIA_HA *info,
                                      LSN lsn, const uchar *header,
-                                     LSN redo_lsn)
+                                     LSN redo_lsn,
+                                     uint * const number_of_blobs,
+                                     uint * const number_of_ranges,
+                                     pgcache_page_no_t * const first_page,
+                                     pgcache_page_no_t * const last_page)
 {
   MARIA_SHARE *share= info->s;
   const uchar *data;
   uint      data_size= FULL_PAGE_SIZE(share->block_size);
   uint      blob_count, ranges;
   uint16    sid;
+  pgcache_page_no_t first_page2= ULONGLONG_MAX, last_page2= 0;
   DBUG_ENTER("_ma_apply_redo_insert_row_blobs");
 
   share->state.changed|= (STATE_CHANGED | STATE_NOT_ZEROFILLED |
@@ -6455,9 +6468,9 @@ uint _ma_apply_redo_insert_row_blobs(MARIA_HA *info,
 
   sid= fileid_korr(header);
   header+= FILEID_STORE_SIZE;
-  ranges= pagerange_korr(header);
+  *number_of_ranges= ranges= pagerange_korr(header);
   header+= PAGERANGE_STORE_SIZE;
-  blob_count= pagerange_korr(header);
+  *number_of_blobs= blob_count= pagerange_korr(header);
   header+= PAGERANGE_STORE_SIZE;
   DBUG_ASSERT(ranges >= blob_count);
 
@@ -6495,6 +6508,8 @@ uint _ma_apply_redo_insert_row_blobs(MARIA_HA *info,
         enum pagecache_page_pin unpin_method;
         uint length;
 
+        set_if_smaller(first_page2, page);
+        set_if_bigger(last_page2, page);
         if (_ma_redo_not_needed_for_page(sid, redo_lsn, page, FALSE))
           continue;
 
@@ -6545,15 +6560,22 @@ uint _ma_apply_redo_insert_row_blobs(MARIA_HA *info,
           }
           else
           {
+#ifndef DBUG_OFF
+            uchar found_page_type= (buff[PAGE_TYPE_OFFSET] & PAGE_TYPE_MASK);
+#endif
             if (lsn_korr(buff) >= lsn)
             {
               /* Already applied */
+              DBUG_PRINT("info", ("already applied %llu >= %llu",
+                                  lsn_korr(buff), lsn));
               pagecache_unlock_by_link(share->pagecache, page_link.link,
                                        PAGECACHE_LOCK_WRITE_UNLOCK,
                                        PAGECACHE_UNPIN, LSN_IMPOSSIBLE,
                                        LSN_IMPOSSIBLE, 0, FALSE);
               continue;
             }
+            DBUG_ASSERT((found_page_type == (uchar) BLOB_PAGE) ||
+                        (found_page_type == (uchar) UNALLOCATED_PAGE));
           }
           unlock_method= PAGECACHE_LOCK_WRITE_UNLOCK;
           unpin_method=  PAGECACHE_UNPIN;
@@ -6595,10 +6617,13 @@ uint _ma_apply_redo_insert_row_blobs(MARIA_HA *info,
         goto err;
     }
   }
+  *first_page= first_page2;
+  *last_page=  last_page2;
   DBUG_RETURN(0);
 
 err:
   _ma_mark_file_crashed(share);
+  DBUG_ASSERT(0);
   DBUG_RETURN(1);
 }
 
diff --git a/storage/maria/ma_blockrec.h b/storage/maria/ma_blockrec.h
index 9b373d1d0ea..7262205d10e 100644
--- a/storage/maria/ma_blockrec.h
+++ b/storage/maria/ma_blockrec.h
@@ -236,7 +236,11 @@ uint _ma_apply_redo_free_blocks(MARIA_HA *info, LSN lsn,
 uint _ma_apply_redo_free_head_or_tail(MARIA_HA *info, LSN lsn,
                                       const uchar *header);
 uint _ma_apply_redo_insert_row_blobs(MARIA_HA *info, LSN lsn,
-                                     const uchar *header, LSN redo_lsn);
+                                     const uchar *header, LSN redo_lsn,
+                                     uint * const number_of_blobs,
+                                     uint * const number_of_ranges,
+                                     pgcache_page_no_t * const first_page,
+                                     pgcache_page_no_t * const last_page);
 my_bool _ma_apply_redo_bitmap_new_page(MARIA_HA *info, LSN lsn,
                                        const uchar *header);
 my_bool _ma_apply_undo_row_insert(MARIA_HA *info, LSN undo_lsn,
diff --git a/storage/maria/ma_create.c b/storage/maria/ma_create.c
index 5552fae6028..f9ea1dab67e 100644
--- a/storage/maria/ma_create.c
+++ b/storage/maria/ma_create.c
@@ -652,7 +652,8 @@ int maria_create(const char *name, enum data_file_type datafile_type,
   /* There are only 16 bits for the total header length. */
   if (info_length > 65535)
   {
-    my_printf_error(0, "Maria table '%s' has too many columns and/or "
+    my_printf_error(HA_WRONG_CREATE_OPTION,
+                    "Maria table '%s' has too many columns and/or "
                     "indexes and/or unique constraints.",
                     MYF(0), name + dirname_length(name));
     my_errno= HA_WRONG_CREATE_OPTION;
diff --git a/storage/maria/ma_loghandler.c b/storage/maria/ma_loghandler.c
index a0c8d0e63a3..a0440aa46f3 100644
--- a/storage/maria/ma_loghandler.c
+++ b/storage/maria/ma_loghandler.c
@@ -685,6 +685,10 @@ static LOG_DESC INIT_LOGREC_IMPORTED_TABLE=
 {LOGRECTYPE_VARIABLE_LENGTH, 0, 0, NULL, NULL, NULL, 0,
  "imported_table", LOGREC_IS_GROUP_ITSELF, NULL, NULL};
 
+static LOG_DESC INIT_LOGREC_DEBUG_INFO=
+{LOGRECTYPE_VARIABLE_LENGTH, 0, 0, NULL, NULL, NULL, 0,
+ "info", LOGREC_IS_GROUP_ITSELF, NULL, NULL};
+
 const myf log_write_flags= MY_WME | MY_NABP | MY_WAIT_IF_FULL;
 
 void translog_table_init()
@@ -774,6 +778,9 @@ void translog_table_init()
     INIT_LOGREC_REDO_BITMAP_NEW_PAGE;
   log_record_type_descriptor[LOGREC_IMPORTED_TABLE]=
     INIT_LOGREC_IMPORTED_TABLE;
+  log_record_type_descriptor[LOGREC_DEBUG_INFO]=
+    INIT_LOGREC_DEBUG_INFO;
+
   for (i= LOGREC_FIRST_FREE; i < LOGREC_NUMBER_OF_TYPES; i++)
     log_record_type_descriptor[i].rclass= LOGRECTYPE_NOT_ALLOWED;
 #ifndef DBUG_OFF
@@ -8299,6 +8306,46 @@ void translog_set_file_size(uint32 size)
   DBUG_VOID_RETURN;
 }
 
+
+/**
+   Write debug information to log if we EXTRA_DEBUG is enabled
+*/
+
+my_bool translog_log_debug_info(TRN *trn __attribute__((unused)),
+                                enum translog_debug_info_type type
+                                __attribute__((unused)),
+                                uchar *info __attribute__((unused)),
+                                size_t length __attribute__((unused)))
+{
+#ifdef EXTRA_DEBUG
+  LEX_CUSTRING log_array[TRANSLOG_INTERNAL_PARTS + 2];
+  uchar debug_type;
+  LSN lsn;
+
+  if (!trn)
+  {
+    /*
+      We can't log the current transaction because we don't have
+      an active transaction. Use a temporary transaction object instead
+    */
+    trn= &dummy_transaction_object;
+  }
+  debug_type= (uchar) type;
+  log_array[TRANSLOG_INTERNAL_PARTS + 0].str= &debug_type;
+  log_array[TRANSLOG_INTERNAL_PARTS + 0].length= 1;
+  log_array[TRANSLOG_INTERNAL_PARTS + 1].str= info;
+  log_array[TRANSLOG_INTERNAL_PARTS + 1].length= length;
+  return translog_write_record(&lsn, LOGREC_DEBUG_INFO,
+                               trn, NULL,
+                               (translog_size_t) (1+ length),
+                               sizeof(log_array)/sizeof(log_array[0]),
+                               log_array, NULL, NULL);
+#else
+  return 0;
+#endif
+}
+
+
 #ifdef MARIA_DUMP_LOG
 #include <my_getopt.h>
 extern void translog_example_table_init();
diff --git a/storage/maria/ma_loghandler.h b/storage/maria/ma_loghandler.h
index 95d711d3eee..dba6283e303 100644
--- a/storage/maria/ma_loghandler.h
+++ b/storage/maria/ma_loghandler.h
@@ -144,6 +144,7 @@ enum translog_record_type
   LOGREC_UNDO_BULK_INSERT,
   LOGREC_REDO_BITMAP_NEW_PAGE,
   LOGREC_IMPORTED_TABLE,
+  LOGREC_DEBUG_INFO,
   LOGREC_FIRST_FREE,
   LOGREC_RESERVED_FUTURE_EXTENSION= 63
 };
@@ -167,6 +168,12 @@ enum en_key_op
   KEY_OP_COMPACT_PAGE   /* Compact key page */
 };
 
+
+enum translog_debug_info_type
+{
+  LOGREC_DEBUG_INFO_QUERY
+};
+
 /* Size of log file; One log file is restricted to 4G */
 typedef uint32 translog_size_t;
 
@@ -323,6 +330,9 @@ translog_assign_id_to_share_from_recovery(struct st_maria_share *share,
 extern my_bool translog_walk_filenames(const char *directory,
                                        my_bool (*callback)(const char *,
                                                            const char *));
+extern my_bool translog_log_debug_info(TRN *trn,
+                                       enum translog_debug_info_type type,
+                                       uchar *info, size_t length);
 
 enum enum_translog_status
 {
diff --git a/storage/maria/ma_recovery.c b/storage/maria/ma_recovery.c
index 77dfc6b568a..a63df9c1b79 100644
--- a/storage/maria/ma_recovery.c
+++ b/storage/maria/ma_recovery.c
@@ -98,6 +98,7 @@ prototype_redo_exec_hook(UNDO_KEY_DELETE);
 prototype_redo_exec_hook(UNDO_KEY_DELETE_WITH_ROOT);
 prototype_redo_exec_hook(COMMIT);
 prototype_redo_exec_hook(CLR_END);
+prototype_redo_exec_hook(DEBUG_INFO);
 prototype_undo_exec_hook(UNDO_ROW_INSERT);
 prototype_undo_exec_hook(UNDO_ROW_DELETE);
 prototype_undo_exec_hook(UNDO_ROW_UPDATE);
@@ -488,6 +489,11 @@ static void display_record_position(const LOG_DESC *log_desc,
          number ? "" : "   ", number, LSN_IN_PARTS(rec->lsn),
          rec->short_trid, log_desc->name, rec->type,
          (ulong)rec->record_length);
+  if (rec->type == LOGREC_DEBUG_INFO)
+  {
+    /* Print some extra information */
+    (*log_desc->record_execute_in_redo_phase)(rec);
+  }
 }
 
 
@@ -1412,6 +1418,9 @@ prototype_redo_exec_hook(REDO_INSERT_ROW_BLOBS)
 {
   int error= 1;
   uchar *buff;
+  uint number_of_blobs, number_of_ranges;
+  pgcache_page_no_t first_page, last_page;
+  char llbuf1[22], llbuf2[22];
   MARIA_HA *info= get_MARIA_HA_from_REDO_record(rec);
   if (info == NULL)
     return 0;
@@ -1426,11 +1435,19 @@ prototype_redo_exec_hook(REDO_INSERT_ROW_BLOBS)
   }
   buff= log_record_buffer.str;
   if (_ma_apply_redo_insert_row_blobs(info, current_group_end_lsn,
-                                      buff, rec->lsn))
+                                      buff, rec->lsn, &number_of_blobs,
+                                      &number_of_ranges,
+                                      &first_page, &last_page))
     goto end;
+  llstr(first_page, llbuf1);
+  llstr(last_page, llbuf2);
+  tprint(tracef, " %u blobs %u ranges, first page %s last %s",
+         number_of_blobs, number_of_ranges, llbuf1, llbuf2);
+
   error= 0;
 
 end:
+  tprint(tracef, " \n");
   return error;
 }
 
@@ -1992,6 +2009,37 @@ prototype_redo_exec_hook(CLR_END)
 }
 
 
+/**
+   Hock to print debug information (like MySQL query)
+*/
+
+prototype_redo_exec_hook(DEBUG_INFO)
+{
+  uchar *data;
+  enum translog_debug_info_type debug_info;
+
+  enlarge_buffer(rec);
+  if (log_record_buffer.str == NULL ||
+      translog_read_record(rec->lsn, 0, rec->record_length,
+                           log_record_buffer.str, NULL) !=
+      rec->record_length)
+  {
+    eprint(tracef, "Failed to read record debug record");
+    return 1;
+  }
+  debug_info= (enum translog_debug_info_type) log_record_buffer.str[0];
+  data= log_record_buffer.str + 1;
+  switch (debug_info) {
+  case LOGREC_DEBUG_INFO_QUERY:
+    tprint(tracef, "Query: %s\n", (char*) data);
+    break;
+  default:
+    DBUG_ASSERT(0);
+  }
+  return 0;
+}
+
+
 /**
   In some cases we have to skip execution of an UNDO record during the UNDO
   phase.
@@ -2350,6 +2398,7 @@ static int run_redo_phase(LSN lsn, enum maria_apply_log_way apply)
   install_redo_exec_hook(UNDO_BULK_INSERT);
   install_undo_exec_hook(UNDO_BULK_INSERT);
   install_redo_exec_hook(IMPORTED_TABLE);
+  install_redo_exec_hook(DEBUG_INFO);
 
   current_group_end_lsn= LSN_IMPOSSIBLE;
 #ifndef DBUG_OFF
@@ -3392,6 +3441,7 @@ static void print_redo_phase_progress(TRANSLOG_ADDRESS addr)
   }
 }
 
+
 #ifdef MARIA_EXTERNAL_LOCKING
 #error Marias Checkpoint and Recovery are really not ready for it
 #endif
diff --git a/storage/maria/trnman.c b/storage/maria/trnman.c
index e353d0d1869..06b3b828d50 100644
--- a/storage/maria/trnman.c
+++ b/storage/maria/trnman.c
@@ -89,6 +89,18 @@ void trnman_reset_locked_tables(TRN *trn, uint locked_tables)
   trn->locked_tables= locked_tables;
 }
 
+#ifdef EXTRA_DEBUG
+uint16 trnman_get_flags(TRN *trn)
+{
+  return trn->flags;
+}
+
+void trnman_set_flags(TRN *trn, uint16 flags)
+{
+  trn->flags= flags;
+}
+#endif
+
 /** Wake up threads waiting for this transaction */
 static void wt_thd_release_self(TRN *trn)
 {
diff --git a/storage/maria/trnman.h b/storage/maria/trnman.h
index 4bb26d4d702..afe01d4ad10 100644
--- a/storage/maria/trnman.h
+++ b/storage/maria/trnman.h
@@ -53,6 +53,7 @@ struct st_ma_transaction
   LSN_WITH_FLAGS       first_undo_lsn;
   uint                 locked_tables;
   uint16               short_id;
+  uint16               flags;         /**< Various flags */
 };
 
 #define TRANSACTION_LOGGED_LONG_ID ULL(0x8000000000000000)
diff --git a/storage/maria/trnman_public.h b/storage/maria/trnman_public.h
index 5ac89ef7b42..9523eb5de8f 100644
--- a/storage/maria/trnman_public.h
+++ b/storage/maria/trnman_public.h
@@ -69,5 +69,17 @@ my_bool trnman_exists_active_transactions(TrID min_id, TrID max_id,
 void trnman_lock();
 void trnman_unlock();
 my_bool trman_is_inited();
+#ifdef EXTRA_DEBUG
+uint16 trnman_get_flags(TRN *);
+void trnman_set_flags(TRN *, uint16 flags);
+#else
+#define trnman_get_flags(A) 0
+#define trnman_set_flags(A, B) do { } while (0)
+#endif
+
+/* Flag bits */
+#define TRN_STATE_INFO_LOGGED       1  /* Query is logged */
+#define TRN_STATE_TABLES_CAN_CHANGE 2  /* Things can change during trans. */
+
 C_MODE_END
 #endif
diff --git a/storage/myisam/mi_create.c b/storage/myisam/mi_create.c
index 560a537e87c..1ad58bee3e3 100644
--- a/storage/myisam/mi_create.c
+++ b/storage/myisam/mi_create.c
@@ -496,7 +496,8 @@ int mi_create(const char *name,uint keys,MI_KEYDEF *keydefs,
   /* There are only 16 bits for the total header length. */
   if (info_length > 65535)
   {
-    my_printf_error(0, "MyISAM table '%s' has too many columns and/or "
+    my_printf_error(HA_WRONG_CREATE_OPTION,
+                    "MyISAM table '%s' has too many columns and/or "
                     "indexes and/or unique constraints.",
                     MYF(0), name + dirname_length(name));
     my_errno= HA_WRONG_CREATE_OPTION;