Aria issues:

- Fix for LP#700623 "Aria recovery: ma_blockrec.c:3930: _ma_update_at_original_place: Assertion `block->org_bitmap_value == _ma_bitmap_get_page_bits(info, &info->s->bitmap, page)' failed"
  - Issue was that when deleting a tail page where all index entries where full, the page was marked wrongly in the bitmap.
- If debug_assert_if_crashed_table is set, we now crash when we find Aria corrupted.
- Write more information if we find something wrong with the bitmap.
- Fixed that REPAIR also can fix wrong create_rename_lsn issues (a very unlikely event)
- Define STATE_CRASHED_FLAGS as set of all CRASHED flags (to simplify code)


storage/maria/ha_maria.cc:
  Mark the normal page cache (not the page cache for the logs) so that we can request extra debugging for it.
  Copy the value of debug_assert_if_crashed_table to maria_assert_if_crashed_table so that we can request a crash at exactly the point where we find Aria corrupted.
  Use STATE_CRASHED_FLAGS
storage/maria/ma_bitmap.c:
  Made bits_to_txt extern so that we can use this in maria_chk
  Added extra information to the log files to be able to easier find bitmap failures in recovery. (When compiling with -DEXTRA_DEBUG_BITMAP)
  Added _ma_get_bitmap_description() to request a clear text description of the bitmap.
  Simplify _ma_check_bitmap_data(), as we know the bitmap pattern in the caller.
storage/maria/ma_blockrec.c:
  In delete_head_or_tail(), fixed a bug where we sent wrong information to _ma_bitmap_set() if the directory was full for a page that should be freed.
  This fixed LP#700623 (failure in bitmap found during recovery)
storage/maria/ma_blockrec.h:
  Added definitions for _ma_get_bitmap_description() and bits_to_txt
storage/maria/ma_check.c:
  Simplify call to _ma_check_bitmap_data().
  Write more information if we find something wrong with the bitmap.
  Moved getting clear text information about the bitmap to ma_bitmap.c::_ma_get_bitmap_description()
storage/maria/ma_checkpoint.c:
  More asserts
storage/maria/ma_create.c:
  Fix wrong create_rename_lsn during repair.
  (Create_rename_lsn can be too big if someone restores an old maria_log_file after an Aria file was created)
storage/maria/ma_delete.c:
  Call _ma_set_fatal_error() in case of crashed file
  Remove not needed test of save_errno == HA_ERR_KEY_NOT_FOUND. (Handled by other code
storage/maria/ma_extra.c:
  Call _ma_set_fatal_error() in case of crashed file
  Reset share->bitmap.changed_not_flushed to not cause new ASSERTS to trigger.
  Added _ma_file_callback_to_id() for writing share->id to log file in case of DEBUG logging.
storage/maria/ma_init.c:
  Destroy also translog if it's readonly (as when called by maria_read_log -d)
storage/maria/ma_key.c:
  Call _ma_set_fatal_error() in case of crashed file
storage/maria/ma_key_recover.c:
  STATE_CRASHED -> STATE_CRASHED_FLAGS
storage/maria/ma_keycache.c:
  Call _ma_set_fatal_error() in case of crashed file
storage/maria/ma_locking.c:
  Call _ma_set_fatal_error() in case of crashed file.
  Added _ma_set_fatal_error()
storage/maria/ma_open.c:
  Call _ma_set_fatal_error() in case of crashed file
storage/maria/ma_page.c:
  Call _ma_set_fatal_error() in case of crashed file
storage/maria/ma_pagecache.c:
  Added extra information to log file to simply debugging of bitmap errors.
storage/maria/ma_pagecache.h:
  Added extra_debug flag to allow marking of row and index cache for extra logging (for debugging).
storage/maria/ma_panic.c:
  Flush both data and index blocks in case of HA_PANIC_CLOSE
  Fixed wrong position of 'break'. (Not critical for MariaDB as MariaDB never uses this code)
storage/maria/ma_recovery_util.c:
  Avoid writing extra not needed \n to DBUG log.
storage/maria/ma_rkey.c:
  Call _ma_set_fatal_error() in case of crashed file
storage/maria/ma_search.c:
  Call _ma_set_fatal_error() in case of crashed file
storage/maria/ma_static.c:
  Define maria_assert_if_crashed_table
storage/maria/ma_update.c:
  Call _ma_set_fatal_error() in case of crashed file.
  The new code also avoids a problem where we before would print the error twice.
storage/maria/ma_write.c:
  Call _ma_set_fatal_error() in case of crashed file
storage/maria/maria_chk.c:
  STATE_CRASHED -> STATE_CRASHED_FLAGS
storage/maria/maria_def.h:
  Added STATE_CRASHED_PRINTED to avoid giving error message about crash twice.
  Added STATE_CRASHED_FLAGS to be able to easily detect and set all CRASHED related flags.
  Added prototypes for new functions.
storage/myisam/mi_panic.c:
  Fixed wrong position of 'break'. (Not critical for MariaDB as MariaDB never uses this code)

storage/maria/ha_maria.cc

@@ -1065,8 +1065,7 @@ int ha_maria::check(THD * thd, HA_CHECK_OPT * check_opt)
 
   if (!maria_is_crashed(file) &&
       (((param.testflag & T_CHECK_ONLY_CHANGED) &&
-        !(share->state.changed & (STATE_CHANGED | STATE_CRASHED |
-                                  STATE_CRASHED_ON_REPAIR |
+        !(share->state.changed & (STATE_CHANGED | STATE_CRASHED_FLAGS |
                                   STATE_IN_REPAIR)) &&
         share->state.open_count == 0) ||
        ((param.testflag & T_FAST) && (share->state.open_count ==
@@ -1104,15 +1103,15 @@ int ha_maria::check(THD * thd, HA_CHECK_OPT * check_opt)
   if (!error)
   {
     if ((share->state.changed & (STATE_CHANGED |
-                                 STATE_CRASHED_ON_REPAIR | STATE_IN_REPAIR |
-                                 STATE_CRASHED | STATE_NOT_ANALYZED)) ||
+                                 STATE_CRASHED_FLAGS |
+                                 STATE_IN_REPAIR | STATE_NOT_ANALYZED)) ||
         (param.testflag & T_STATISTICS) || maria_is_crashed(file))
     {
       file->update |= HA_STATE_CHANGED | HA_STATE_ROW_CHANGED;
       pthread_mutex_lock(&share->intern_lock);
       DBUG_PRINT("info", ("Reseting crashed state"));
-      share->state.changed&= ~(STATE_CHANGED | STATE_CRASHED |
-                               STATE_CRASHED_ON_REPAIR | STATE_IN_REPAIR);
+      share->state.changed&= ~(STATE_CHANGED | STATE_CRASHED_FLAGS |
+                               STATE_IN_REPAIR);
       if (!(table->db_stat & HA_READ_ONLY))
         error= maria_update_state_info(&param, file,
                                        UPDATE_TIME | UPDATE_OPEN_COUNT |
@@ -1544,8 +1543,8 @@ int ha_maria::repair(THD *thd, HA_CHECK *param, bool do_optimize)
     if ((share->state.changed & STATE_CHANGED) || maria_is_crashed(file))
     {
       DBUG_PRINT("info", ("Reseting crashed state"));
-      share->state.changed&= ~(STATE_CHANGED | STATE_CRASHED |
-                               STATE_CRASHED_ON_REPAIR | STATE_IN_REPAIR);
+      share->state.changed&= ~(STATE_CHANGED | STATE_CRASHED_FLAGS |
+                               STATE_IN_REPAIR);
       file->update |= HA_STATE_CHANGED | HA_STATE_ROW_CHANGED;
     }
     /*
@@ -2025,8 +2024,7 @@ bool ha_maria::check_and_repair(THD *thd)
   check_opt.flags= T_MEDIUM | T_AUTO_REPAIR;
 
   error= 1;
-  if ((file->s->state.changed &
-       (STATE_CRASHED | STATE_CRASHED_ON_REPAIR | STATE_MOVED)) ==
+  if ((file->s->state.changed & (STATE_CRASHED_FLAGS | STATE_MOVED)) ==
       STATE_MOVED)
   {
     sql_print_information("Zerofilling moved table:  '%s'",
@@ -2077,7 +2075,7 @@ bool ha_maria::check_and_repair(THD *thd)
 
 bool ha_maria::is_crashed() const
 {
-  return (file->s->state.changed & (STATE_CRASHED | STATE_MOVED) ||
+  return (file->s->state.changed & (STATE_CRASHED_FLAGS | STATE_MOVED) ||
           (my_disable_locking && file->s->state.open_count));
 }
 
@@ -3287,6 +3285,8 @@ static int ha_maria_init(void *p)
     ma_checkpoint_init(checkpoint_interval);
   maria_multi_threaded= maria_in_ha_maria= TRUE;
   maria_create_trn_hook= maria_create_trn_for_mysql;
+  maria_pagecache->extra_debug= 1;
+  maria_assert_if_crashed_table= debug_assert_if_crashed_table;
 
 #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 */

storage/maria/ma_bitmap.c

@@ -104,10 +104,11 @@
   - On checkpoint
   (Ie: When we do a checkpoint, we have to ensure that all bitmaps are
   put on disk even if they are not in the page cache).
-  - When explicitely requested (for example on backup or after recvoery,
+  - When explicitely requested (for example on backup or after recovery,
   to simplify things)
 
  The flow of writing a row is that:
+ - Mark the bitmap not flushable (_ma_bitmap_flushable(X, 1))
  - Lock the bitmap
  - Decide which data pages we will write to
  - Mark them full in the bitmap page so that other threads do not try to
@@ -119,6 +120,7 @@
    pages (that is, we marked pages full but when we are done we realize
    we didn't fill them)
  - Unlock the bitmap.
+ - Mark the bitmap flushable (_ma_bitmap_flushable(X, -1))
 */
 
 #include "maria_def.h"
@@ -127,6 +129,12 @@
 #define FULL_HEAD_PAGE 4
 #define FULL_TAIL_PAGE 7
 
+const char *bits_to_txt[]=
+{
+  "empty", "00-30% full", "30-60% full", "60-90% full", "full",
+  "tail 00-40 % full", "tail 40-80 % full", "tail/blob full"
+};
+
 /*#define WRONG_BITMAP_FLUSH 1*/ /*define only for provoking bugs*/
 #undef WRONG_BITMAP_FLUSH
 
@@ -273,6 +281,12 @@ my_bool _ma_bitmap_end(MARIA_SHARE *share)
   delete_dynamic(&share->bitmap.pinned_pages);
   my_free(share->bitmap.map, MYF(MY_ALLOW_ZERO_PTR));
   share->bitmap.map= 0;
+  /*
+    This is to not get an assert in checkpoint. The bitmap will be flushed
+    at once by  _ma_once_end_block_record() as part of the normal flush
+    of the kfile.
+  */
+  share->bitmap.changed_not_flushed= 0;
   return res;
 }
 
@@ -353,6 +367,24 @@ my_bool _ma_bitmap_flush_all(MARIA_SHARE *share)
   my_bool res= 0;
   MARIA_FILE_BITMAP *bitmap= &share->bitmap;
   DBUG_ENTER("_ma_bitmap_flush_all");
+
+#ifdef EXTRA_DEBUG_BITMAP
+  {
+    char buff[160];
+    uint len= my_sprintf(buff,
+                         (buff, "bitmap_flush:  fd: %d  id: %u  "
+                          "changed: %d  changed_not_flushed: %d  "
+                          "flush_all_requsted: %d",
+                          share->bitmap.file.file,
+                          share->id,
+                          bitmap->changed,
+                          bitmap->changed_not_flushed,
+                          bitmap->flush_all_requested));
+    (void) translog_log_debug_info(0, LOGREC_DEBUG_INFO_QUERY,
+                                   (uchar*) buff, len);
+  }
+#endif
+
   pthread_mutex_lock(&bitmap->bitmap_lock);
   if (bitmap->changed || bitmap->changed_not_flushed)
   {
@@ -364,6 +396,15 @@ my_bool _ma_bitmap_flush_all(MARIA_SHARE *share)
       pthread_cond_wait(&bitmap->bitmap_cond, &bitmap->bitmap_lock);
     }
 #endif
+#ifdef EXTRA_DEBUG_BITMAP
+    {
+      char tmp[MAX_BITMAP_INFO_LENGTH];      
+      _ma_get_bitmap_description(bitmap, bitmap->map, bitmap->page, tmp);
+      (void) translog_log_debug_info(0, LOGREC_DEBUG_INFO_QUERY,
+                                     (uchar*) tmp, strlen(tmp));
+    }
+#endif
+
     DBUG_ASSERT(bitmap->flush_all_requested == 1);
     /*
       Bitmap is in a flushable state: its contents in memory are reflected by
@@ -680,7 +721,7 @@ static inline uint pattern_to_size(MARIA_FILE_BITMAP *bitmap, uint pattern)
   Print bitmap for debugging
 
   SYNOPSIS
-  _ma_print_bitmap()
+  _ma_print_bitmap_changes()
   bitmap	Bitmap to print
 
   IMPLEMENTATION
@@ -691,12 +732,6 @@ static inline uint pattern_to_size(MARIA_FILE_BITMAP *bitmap, uint pattern)
 
 #ifndef DBUG_OFF
 
-const char *bits_to_txt[]=
-{
-  "empty", "00-30% full", "30-60% full", "60-90% full", "full",
-  "tail 00-40 % full", "tail 40-80 % full", "tail/blob full"
-};
-
 static void _ma_print_bitmap_changes(MARIA_FILE_BITMAP *bitmap)
 {
   uchar *pos, *end, *org_pos;
@@ -747,7 +782,6 @@ void _ma_print_bitmap(MARIA_FILE_BITMAP *bitmap, uchar *data,
   uchar *pos, *end;
   char llbuff[22];
 
-  end= bitmap->map + bitmap->used_size;
   DBUG_LOCK_FILE;
   fprintf(DBUG_FILE,"\nDump of bitmap page at %s\n", llstr(page, llbuff));
 
@@ -781,6 +815,56 @@ void _ma_print_bitmap(MARIA_FILE_BITMAP *bitmap, uchar *data,
 #endif /* DBUG_OFF */
 
 
+/*
+  Return content of bitmap as a printable string
+*/
+
+void _ma_get_bitmap_description(MARIA_FILE_BITMAP *bitmap,
+                                uchar *bitmap_data,
+                                pgcache_page_no_t page,
+                                char *out)
+{
+  uchar *pos, *end;
+  uint count=0, dot_printed= 0, len;
+  char buff[80], last[80];
+
+  page++;
+  last[0]=0;
+  for (pos= bitmap_data, end= pos+ bitmap->used_size ; pos < end ; pos+= 6)
+  {
+    ulonglong bits= uint6korr(pos);    /* 6 bytes = 6*8/3= 16 patterns */
+    uint i;
+
+    for (i= 0; i < 16 ; i++, bits>>= 3)
+    {
+      if (count > 60)
+      {
+        if (memcmp(buff, last, count))
+        {
+          memcpy(last, buff, count);
+          len= my_sprintf(out, (out, "%8lu: ", (ulong) page - count));
+          memcpy(out+len, buff, count);
+          out+= len + count + 1;
+          out[-1]= '\n';
+          dot_printed= 0;
+        }
+        else if (!(dot_printed++))
+        {
+          out= strmov(out, "...\n");
+        }
+        count= 0;
+      }
+      buff[count++]= '0' + (uint) (bits & 7);
+      page++;
+    }
+  }
+  len= my_sprintf(out, (out, "%8lu: ", (ulong) page - count));
+  memcpy(out+len, buff, count);
+  out[len + count]= '\n';
+  out[len + count + 1]= 0;
+}
+
+
 /***************************************************************************
   Reading & writing bitmap pages
 ***************************************************************************/
@@ -2383,16 +2467,14 @@ my_bool _ma_bitmap_release_unused(MARIA_HA *info, MARIA_BITMAP_BLOCKS *blocks)
         The page has all bits set; The following test is an optimization
         to not set the bits to the same value as before.
       */
+      DBUG_ASSERT(current_bitmap_value ==
+                  _ma_bitmap_get_page_bits(info, bitmap, block->page));
+
       if (bits != current_bitmap_value)
       {
         if (set_page_bits(info, bitmap, block->page, bits))
           goto err;
       }
-      else
-      {
-        DBUG_ASSERT(current_bitmap_value ==
-                    _ma_bitmap_get_page_bits(info, bitmap, block->page));
-      }
     }
     else if (!(block->used & BLOCKUSED_USED) &&
              _ma_bitmap_reset_full_page_bits(info, bitmap,
@@ -2521,17 +2603,15 @@ my_bool _ma_bitmap_set(MARIA_HA *info, pgcache_page_no_t page, my_bool head,
     page_type	    What kind of page this is
     page	    Adress to page
     empty_space     Empty space on page
-    bitmap_pattern  Store here the pattern that was in the bitmap for the
-		    page. This is always updated.
+    bitmap_pattern  Bitmap pattern for page (from bitmap)
 
   RETURN
     0  ok
     1  error
 */
 
-my_bool _ma_check_bitmap_data(MARIA_HA *info,
-                              enum en_page_type page_type, pgcache_page_no_t page,
-                              uint empty_space, uint *bitmap_pattern)
+my_bool _ma_check_bitmap_data(MARIA_HA *info, enum en_page_type page_type,
+                              uint empty_space, uint bitmap_pattern)
 {
   uint bits;
   switch (page_type) {
@@ -2552,8 +2632,7 @@ my_bool _ma_check_bitmap_data(MARIA_HA *info,
     bits= 0; /* to satisfy compiler */
     DBUG_ASSERT(0);
   }
-  return ((*bitmap_pattern= _ma_bitmap_get_page_bits(info, &info->s->bitmap,
-                                                     page)) != bits);
+  return (bitmap_pattern != bits);
 }
 
 

storage/maria/ma_blockrec.c

@@ -4106,11 +4106,11 @@ static my_bool delete_head_or_tail(MARIA_HA *info,
 {
   MARIA_SHARE *share= info->s;
   uint empty_space;
-  uint block_size= share->block_size;
+  int res;
+  my_bool page_is_empty;
   uchar *buff;
   LSN lsn;
   MARIA_PINNED_PAGE page_link;
-  int res;
   enum pagecache_page_lock lock_at_write, lock_at_unpin;
   DBUG_ENTER("delete_head_or_tail");
   DBUG_PRINT("enter", ("id: %lu (%lu:%u)",
@@ -4140,13 +4140,14 @@ static my_bool delete_head_or_tail(MARIA_HA *info,
     lock_at_unpin= PAGECACHE_LOCK_READ_UNLOCK;
   }
 
-  res= delete_dir_entry(buff, block_size, record_number, &empty_space);
+  res= delete_dir_entry(buff, share->block_size, record_number, &empty_space);
   if (res < 0)
     DBUG_RETURN(1);
   if (res == 0) /* after our deletion, page is still not empty */
   {
     uchar log_data[FILEID_STORE_SIZE + PAGE_STORE_SIZE + DIRPOS_STORE_SIZE];
     LEX_CUSTRING log_array[TRANSLOG_INTERNAL_PARTS + 1];
+    page_is_empty= 0;
     if (share->now_transactional)
     {
       /* Log REDO data */
@@ -4167,6 +4168,7 @@ static my_bool delete_head_or_tail(MARIA_HA *info,
   }
   else /* page is now empty */
   {
+    page_is_empty= 1;
     if (share->now_transactional)
     {
       uchar log_data[FILEID_STORE_SIZE + PAGE_STORE_SIZE];
@@ -4198,8 +4200,8 @@ static my_bool delete_head_or_tail(MARIA_HA *info,
     If there is not enough space for all possible tails, mark the
     page full
   */
-  if (!head && !enough_free_entries(buff, share->block_size,
-                                    1 + share->base.blobs))
+  if (!head && !page_is_empty && !enough_free_entries(buff, share->block_size,
+                                                      1 + share->base.blobs))
     empty_space= 0;
 
   DBUG_RETURN(_ma_bitmap_set(info, page, head, empty_space));

storage/maria/ma_blockrec.h

@@ -78,6 +78,10 @@ enum en_page_type { UNALLOCATED_PAGE, HEAD_PAGE, TAIL_PAGE, BLOB_PAGE, MAX_PAGE_
 #define ROW_FLAG_EXTENTS                128
 #define ROW_FLAG_ALL			(1+2+4+8+128)
 
+/* Size for buffer to hold information about bitmap */
+#define MAX_BITMAP_INFO_LENGTH ((MARIA_MAX_KEY_BLOCK_LENGTH*8/3)*(61*11/60)+10)
+
+
 /******** Variables that affects how data pages are utilized ********/
 
 /* Minium size of tail segment */
@@ -181,6 +185,8 @@ TRANSLOG_ADDRESS
 maria_page_get_lsn(uchar *page, pgcache_page_no_t page_no, uchar* data_ptr);
 
 /* ma_bitmap.c */
+extern const char *bits_to_txt[];
+
 my_bool _ma_bitmap_init(MARIA_SHARE *share, File file);
 my_bool _ma_bitmap_end(MARIA_SHARE *share);
 my_bool _ma_bitmap_flush(MARIA_SHARE *share);
@@ -206,8 +212,7 @@ my_bool _ma_bitmap_find_new_place(MARIA_HA *info, MARIA_ROW *new_row,
                                   MARIA_BITMAP_BLOCKS *result_blocks);
 my_bool _ma_check_bitmap_data(MARIA_HA *info,
                               enum en_page_type page_type,
-                              pgcache_page_no_t page,
-                              uint empty_space, uint *bitmap_pattern);
+                              uint empty_space, uint bitmap_pattern);
 my_bool _ma_check_if_right_bitmap_type(MARIA_HA *info,
                                        enum en_page_type page_type,
                                        pgcache_page_no_t page,
@@ -225,6 +230,10 @@ void _ma_bitmap_set_pagecache_callbacks(PAGECACHE_FILE *file,
 void _ma_print_bitmap(MARIA_FILE_BITMAP *bitmap, uchar *data,
                       pgcache_page_no_t page);
 #endif
+void _ma_get_bitmap_description(MARIA_FILE_BITMAP *bitmap,
+                                uchar *bitmap_data,
+                                pgcache_page_no_t page,
+                                char *out);
 
 uint _ma_apply_redo_insert_row_head_or_tail(MARIA_HA *info, LSN lsn,
                                             uint page_type,

storage/maria/ma_check.c

@@ -1822,6 +1822,7 @@ static int check_block_record(HA_CHECK *param, MARIA_HA *info, int extend,
        pos+= block_size, page++)
   {
     uint row_count, real_row_count, empty_space, page_type, bitmap_pattern;
+    uint bitmap_for_page;
     LINT_INIT(row_count);
     LINT_INIT(empty_space);
 
@@ -1856,7 +1857,7 @@ static int check_block_record(HA_CHECK *param, MARIA_HA *info, int extend,
     offset= offset_page & 7;
     data= bitmap_buff + offset_page / 8;
     bitmap_pattern= uint2korr(data);
-    if (!((bitmap_pattern >> offset) & 7))
+    if (!(bitmap_for_page= ((bitmap_pattern >> offset) & 7)))
     {
       param->empty+= block_size;
       param->del_blocks++;
@@ -1879,8 +1880,9 @@ static int check_block_record(HA_CHECK *param, MARIA_HA *info, int extend,
     if (page_type == UNALLOCATED_PAGE || page_type >= MAX_PAGE_TYPE)
     {
       _ma_check_print_error(param,
-                            "Page: %9s  Found wrong page type %d",
-                            llstr(page, llbuff), page_type);
+                            "Page: %9s  Found wrong page type %d. Bitmap: %d '%s'",
+                            llstr(page, llbuff), page_type,
+                            bitmap_for_page, bits_to_txt[bitmap_for_page]);
       if (param->err_count++ > MAXERR || !(param->testflag & T_VERBOSE))
         goto err;
       continue;
@@ -1927,20 +1929,17 @@ static int check_block_record(HA_CHECK *param, MARIA_HA *info, int extend,
       param->used+= block_size;
       break;
     }
-    if (_ma_check_bitmap_data(info, page_type, page,
+    if (_ma_check_bitmap_data(info, page_type,
                               full_dir ? 0 : empty_space,
-                              &bitmap_pattern))
+                              bitmap_for_page))
     {
-      if (bitmap_pattern == ~(uint) 0)
-        _ma_check_print_error(param,
-                              "Page %9s: Wrong bitmap for data on page",
-                              llstr(page, llbuff));
-      else
         _ma_check_print_error(param,
                               "Page %9s:  Wrong data in bitmap.  Page_type: "
-                              "%d  full: %d  empty_space: %u  Bitmap-bits: %d",
+                              "%d  full: %d  empty_space: %u  Bitmap-bits: %d "
+                              "'%s'",
                               llstr(page, llbuff), page_type, full_dir,
-                              empty_space, bitmap_pattern);
+                              empty_space, bitmap_for_page,
+                              bits_to_txt[bitmap_for_page]);
       if (param->err_count++ > MAXERR || !(param->testflag & T_VERBOSE))
         goto err;
     }
@@ -6831,39 +6830,7 @@ static void print_bitmap_description(MARIA_SHARE *share,
                                      pgcache_page_no_t page,
                                      uchar *bitmap_data)
 {
-  uchar *pos, *end;
-  MARIA_FILE_BITMAP *bitmap= &share->bitmap;
-  uint count=0, dot_printed= 0;
-  char buff[80], last[80];
-
-  printf("Bitmap page %lu\n", (ulong) page);
-  page++;
-  last[0]=0;
-  for (pos= bitmap_data, end= pos+ bitmap->used_size ; pos < end ; pos+= 6)
-  {
-    ulonglong bits= uint6korr(pos);    /* 6 bytes = 6*8/3= 16 patterns */
-    uint i;
-
-    for (i= 0; i < 16 ; i++, bits>>= 3)
-    {
-      if (count > 60)
-      {
-        buff[count]= 0;
-        if (strcmp(buff, last))
-        {
-          memcpy(last, buff, count+1);
-          printf("%8lu: %s\n", (ulong) page - count, buff);
-          dot_printed= 0;
-        }
-        else if (!(dot_printed++))
-          printf("...\n");
-        count= 0;
-      }
-      buff[count++]= '0' + (uint) (bits & 7);
-      page++;
-    }
-  }
-  buff[count]= 0;
-  printf("%8lu: %s\n", (ulong) page - count, buff);
-  fputs("\n", stdout);
+  char tmp[MAX_BITMAP_INFO_LENGTH];
+  _ma_get_bitmap_description(&share->bitmap, bitmap_data, page, tmp);
+  printf("Bitmap page %lu\n%s", (ulong) page, tmp);
 }

storage/maria/ma_checkpoint.c

@@ -130,6 +130,9 @@ int ma_checkpoint_execute(CHECKPOINT_LEVEL level, my_bool no_wait)
   /* from then on, we are sure to be and stay the only checkpointer */
 
   result= really_execute_checkpoint();
+  DBUG_EXECUTE_IF("maria_crash_after_checkpoint",
+                  { DBUG_PRINT("maria_crash", ("now")); DBUG_ABORT(); });
+
   pthread_cond_broadcast(&COND_checkpoint);
 end:
   DBUG_RETURN(result);
@@ -1065,6 +1068,14 @@ static int collect_tables(LEX_STRING *str, LSN checkpoint_start_log_horizon)
         */
       }
     }
+#ifdef EXTRA_DEBUG_BITMAP
+    else
+    {
+      DBUG_ASSERT(share->bitmap.changed == 0 &&
+                  share->bitmap.changed_not_flushed == 0);
+    }
+#endif
+
     /*
       _ma_bitmap_flush_all() may wait, so don't keep intern_lock as
       otherwise this would deadlock with allocate_and_write_block_record()

storage/maria/ma_create.c

@@ -1387,7 +1387,13 @@ int _ma_update_state_lsns_sub(MARIA_SHARE *share, LSN lsn, TrID create_trid,
   share->state.skip_redo_lsn= share->state.is_of_horizon= lsn;
   share->state.create_trid= create_trid;
   mi_int8store(trid_buff, create_trid);
-  if (update_create_rename_lsn)
+
+  /*
+    Update create_rename_lsn if update was requested or if the old one had an
+    impossible value.
+  */
+  if (update_create_rename_lsn ||
+      (share->state.create_rename_lsn > lsn && lsn != LSN_IMPOSSIBLE))
   {
     share->state.create_rename_lsn= lsn;
     if (share->id != 0)

storage/maria/ma_delete.c

@@ -135,18 +135,13 @@ err:
     save_errno= HA_ERR_INTERNAL_ERROR;          /* Should never happen */
 
   mi_sizestore(lastpos, info->cur_row.lastpos);
-  if (save_errno != HA_ERR_RECORD_CHANGED)
-  {
-    maria_print_error(share, HA_ERR_CRASHED);
-    maria_mark_crashed(info);		/* mark table crashed */
-  }
   VOID(_ma_writeinfo(info,WRITEINFO_UPDATE_KEYFILE));
   info->update|=HA_STATE_WRITTEN;	/* Buffer changed */
   allow_break();			/* Allow SIGHUP & SIGINT */
-  if (save_errno == HA_ERR_KEY_NOT_FOUND)
+  if (save_errno != HA_ERR_RECORD_CHANGED)
   {
-    maria_print_error(share, HA_ERR_CRASHED);
-    my_errno=HA_ERR_CRASHED;
+    _ma_set_fatal_error(share, HA_ERR_CRASHED);
+    save_errno= HA_ERR_CRASHED;
   }
   DBUG_RETURN(my_errno= save_errno);
 } /* maria_delete */
@@ -213,7 +208,7 @@ my_bool _ma_ck_real_delete(register MARIA_HA *info, MARIA_KEY *key,
 
   if ((old_root=*root) == HA_OFFSET_ERROR)
   {
-    my_errno=HA_ERR_CRASHED;
+    _ma_set_fatal_error(info->s, HA_ERR_CRASHED);
     DBUG_RETURN(1);
   }
   if (!(root_buff= (uchar*)  my_alloca((uint) keyinfo->block_length+
@@ -348,7 +343,7 @@ static int d_search(MARIA_HA *info, MARIA_KEY *key, uint32 comp_flag,
       if (!(tmp_key_length=(*keyinfo->get_key)(&tmp_key, page_flag, nod_flag,
                                                &kpos)))
       {
-        my_errno= HA_ERR_CRASHED;
+        _ma_set_fatal_error(share, HA_ERR_CRASHED);
         DBUG_RETURN(-1);
       }
       root= _ma_row_pos_from_key(&tmp_key);
@@ -410,8 +405,9 @@ static int d_search(MARIA_HA *info, MARIA_KEY *key, uint32 comp_flag,
   {
     if (!nod_flag)
     {
+      /* This should newer happend */
       DBUG_PRINT("error",("Didn't find key"));
-      my_errno=HA_ERR_CRASHED;		/* This should newer happend */
+      _ma_set_fatal_error(share, HA_ERR_CRASHED);
       goto err;
     }
     save_flag=0;

storage/maria/ma_extra.c

@@ -175,8 +175,8 @@ int maria_extra(MARIA_HA *info, enum ha_extra_function function,
     {
       if ((error= flush_io_cache(&info->rec_cache)))
       {
-        maria_print_error(info->s, HA_ERR_CRASHED);
-	maria_mark_crashed(info);			/* Fatal error found */
+        /* Fatal error found */
+        _ma_set_fatal_error(share, HA_ERR_CRASHED);
       }
     }
     break;
@@ -299,6 +299,8 @@ int maria_extra(MARIA_HA *info, enum ha_extra_function function,
     }
     pthread_mutex_lock(&THR_LOCK_maria);
     pthread_mutex_lock(&share->intern_lock); /* protect against Checkpoint */
+    /* Safety against assert in checkpoint */
+    share->bitmap.changed_not_flushed= 0;
     /* this makes the share not be re-used next time the table is opened */
     share->last_version= 0L;			/* Impossible version */
     pthread_mutex_unlock(&share->intern_lock);
@@ -382,8 +384,11 @@ int maria_extra(MARIA_HA *info, enum ha_extra_function function,
     if (share->data_file_type == BLOCK_RECORD &&
         share->bitmap.file.file >= 0)
     {
+      DBUG_ASSERT(share->bitmap.non_flushable == 0 &&
+                  share->bitmap.changed == 0);
       if (do_flush && my_sync(share->bitmap.file.file, MYF(0)))
         error= my_errno;
+      share->bitmap.changed_not_flushed= 0;
     }
     /* For protection against Checkpoint, we set under intern_lock: */
     share->last_version= 0L;			/* Impossible version */
@@ -415,9 +420,9 @@ int maria_extra(MARIA_HA *info, enum ha_extra_function function,
 	error= my_errno;
       if (error)
       {
+	/* Fatal error found */
 	share->changed= 1;
-        maria_print_error(info->s, HA_ERR_CRASHED);
-	maria_mark_crashed(info);			/* Fatal error found */
+        _ma_set_fatal_error(share, HA_ERR_CRASHED);
       }
     }
     break;
@@ -560,6 +565,12 @@ int _ma_sync_table_files(const MARIA_HA *info)
           my_sync(info->s->kfile.file, MYF(MY_WME)));
 }
 
+uint _ma_file_callback_to_id(void *callback_data)
+{
+  MARIA_SHARE *share= (MARIA_SHARE*) callback_data;
+  return share ? share->id : 0;
+}
+
 
 /**
    @brief flushes the data and/or index file of a table
@@ -608,6 +619,7 @@ int _ma_flush_table_files(MARIA_HA *info, uint flush_data_or_index,
       {
         pthread_mutex_lock(&share->bitmap.bitmap_lock);
         share->bitmap.changed= 0;
+        share->bitmap.changed_not_flushed= 0;
         pthread_mutex_unlock(&share->bitmap.bitmap_lock);
       }
       if (flush_pagecache_blocks(share->pagecache, &info->dfile,
@@ -622,7 +634,6 @@ int _ma_flush_table_files(MARIA_HA *info, uint flush_data_or_index,
   if (!error)
     return 0;
 
-  maria_print_error(info->s, HA_ERR_CRASHED);
-  maria_mark_crashed(info);
+  _ma_set_fatal_error(info->s, HA_ERR_CRASHED);
   return 1;
 }

storage/maria/ma_init.c

@@ -99,7 +99,7 @@ void maria_end(void)
                                             trid, recovery_failures);
     }
     trnman_destroy();
-    if (translog_status == TRANSLOG_OK)
+    if (translog_status == TRANSLOG_OK || translog_status == TRANSLOG_READONLY)
       translog_destroy();
     end_pagecache(maria_log_pagecache, TRUE);
     end_pagecache(maria_pagecache, TRUE);

storage/maria/ma_key.c

@@ -636,8 +636,7 @@ int _ma_read_key_record(MARIA_HA *info, uchar *buf, MARIA_RECORD_POS filepos)
     {				/* Read only key */
       if (_ma_put_key_in_record(info,(uint) info->lastinx,buf))
       {
-        maria_print_error(info->s, HA_ERR_CRASHED);
-	my_errno=HA_ERR_CRASHED;
+        _ma_set_fatal_error(info->s, HA_ERR_CRASHED);
 	return -1;
       }
       info->update|= HA_STATE_AKTIV; /* We should find a record */

storage/maria/ma_key_recover.c

@@ -66,7 +66,7 @@ void _ma_unpin_all_pages(MARIA_HA *info, LSN undo_lsn)
 #ifdef EXTRA_DEBUG
     DBUG_ASSERT((!pinned_page->changed ||
                  undo_lsn != LSN_IMPOSSIBLE || !info->s->now_transactional) ||
-                (info->s->state.changed & STATE_CRASHED));
+                (info->s->state.changed & STATE_CRASHED_FLAGS));
 #endif
     pagecache_unlock_by_link(info->s->pagecache, pinned_page->link,
                              pinned_page->unlock, PAGECACHE_UNPIN,

storage/maria/ma_keycache.c

@@ -79,8 +79,8 @@ int maria_assign_to_pagecache(MARIA_HA *info,
   if (flush_pagecache_blocks(share->pagecache, &share->kfile, FLUSH_RELEASE))
   {
     error= my_errno;
-    maria_print_error(info->s, HA_ERR_CRASHED);
-    maria_mark_crashed(info);		/* Mark that table must be checked */
+    /* Mark that table must be checked */
+    _ma_set_fatal_error(share, error);
   }
 
   /*

storage/maria/ma_locking.c

@@ -80,9 +80,8 @@ int maria_lock_database(MARIA_HA *info, int lock_type)
       {
 	if (end_io_cache(&info->rec_cache))
 	{
-	  error=my_errno;
-          maria_print_error(info->s, HA_ERR_CRASHED);
-	  maria_mark_crashed(info);
+	  error= my_errno;
+          _ma_set_fatal_error(share, error);
 	}
       }
       if (!count)
@@ -129,10 +128,7 @@ int maria_lock_database(MARIA_HA *info, int lock_type)
 	  else
 	    share->not_flushed=1;
 	  if (error)
-          {
-            maria_print_error(info->s, HA_ERR_CRASHED);
-	    maria_mark_crashed(info);
-          }
+            _ma_set_fatal_error(share, error);
 	}
       }
       info->opt_flag&= ~(READ_CACHE_USED | WRITE_CACHE_USED);
@@ -528,6 +524,28 @@ void _ma_mark_file_crashed(MARIA_SHARE *share)
   DBUG_VOID_RETURN;
 }
 
+/*
+  Handle a fatal error
+
+  - Mark the table as crashed
+  - Print an error message, if we had not issued an error message before
+    that the table had been crashed.
+  - set my_errno to error
+  - If 'maria_assert_if_crashed_table is set, then assert.
+*/
+
+void _ma_set_fatal_error(MARIA_SHARE *share, int error)
+{
+  maria_mark_crashed_share(share);
+  if (!(share->state.changed & STATE_CRASHED_PRINTED))
+  {
+    share->state.changed|= STATE_CRASHED_PRINTED;
+    maria_print_error(share, error);
+  }
+  my_errno= error;
+  DBUG_ASSERT(!maria_assert_if_crashed_table);
+}
+
 
 /**
    @brief Set uuid of for a Maria file

storage/maria/ma_open.c

@@ -41,10 +41,10 @@ static uchar *_ma_state_info_read(uchar *ptr, MARIA_STATE_INFO *state);
 					pos+=size;}
 
 
-#define disk_pos_assert(pos, end_pos) \
+#define disk_pos_assert(share, pos, end_pos)     \
 if (pos > end_pos)             \
 {                              \
-  my_errno=HA_ERR_CRASHED;     \
+  _ma_set_fatal_error(share, HA_ERR_CRASHED);    \
   goto err;                    \
 }
 
@@ -387,7 +387,7 @@ MARIA_HA *maria_open(const char *name, int mode, uint open_flags)
     errpos= 3;
     if (my_pread(kfile, disk_cache, info_length, 0L, MYF(MY_NABP)))
     {
-      my_errno=HA_ERR_CRASHED;
+      _ma_set_fatal_error(share, HA_ERR_CRASHED);
       goto err;
     }
     len=mi_uint2korr(share->state.header.state_info_length);
@@ -413,9 +413,11 @@ MARIA_HA *maria_open(const char *name, int mode, uint open_flags)
     }
     disk_pos= _ma_base_info_read(disk_cache + base_pos, &share->base);
     share->state.state_length=base_pos;
+    /* For newly opened tables we reset the error-has-been-printed flag */
+    share->state.changed&= ~STATE_CRASHED_PRINTED;
 
     if (!(open_flags & HA_OPEN_FOR_REPAIR) &&
-	((share->state.changed & STATE_CRASHED) ||
+	((share->state.changed & STATE_CRASHED_FLAGS) ||
 	 ((open_flags & HA_OPEN_ABORT_IF_CRASHED) &&
 	  (my_disable_locking && share->state.open_count))))
     {
@@ -456,7 +458,7 @@ MARIA_HA *maria_open(const char *name, int mode, uint open_flags)
     /* sanity check */
     if (share->base.keystart > 65535 || share->base.rec_reflength > 8)
     {
-      my_errno=HA_ERR_CRASHED;
+      _ma_set_fatal_error(share, HA_ERR_CRASHED);
       goto err;
     }
 
@@ -567,7 +569,8 @@ MARIA_HA *maria_open(const char *name, int mode, uint open_flags)
         share->keyinfo[i].share= share;
 	disk_pos=_ma_keydef_read(disk_pos, &share->keyinfo[i]);
         share->keyinfo[i].key_nr= i;
-        disk_pos_assert(disk_pos + share->keyinfo[i].keysegs * HA_KEYSEG_SIZE,
+        disk_pos_assert(share,
+                        disk_pos + share->keyinfo[i].keysegs * HA_KEYSEG_SIZE,
  			end_pos);
         if (share->keyinfo[i].key_alg == HA_KEY_ALG_RTREE)
           share->have_rtree= 1;
@@ -615,7 +618,7 @@ MARIA_HA *maria_open(const char *name, int mode, uint open_flags)
               pos[0].language= pos[-1].language;
               if (!(pos[0].charset= pos[-1].charset))
               {
-                my_errno=HA_ERR_CRASHED;
+                _ma_set_fatal_error(share, HA_ERR_CRASHED);
                 goto err;
               }
               pos++;
@@ -647,7 +650,8 @@ MARIA_HA *maria_open(const char *name, int mode, uint open_flags)
       for (i=0 ; i < uniques ; i++)
       {
 	disk_pos=_ma_uniquedef_read(disk_pos, &share->uniqueinfo[i]);
-        disk_pos_assert(disk_pos + share->uniqueinfo[i].keysegs *
+        disk_pos_assert(share,
+                        disk_pos + share->uniqueinfo[i].keysegs *
 			HA_KEYSEG_SIZE, end_pos);
 	share->uniqueinfo[i].seg=pos;
 	for (j=0 ; j < share->uniqueinfo[i].keysegs; j++,pos++)
@@ -751,7 +755,8 @@ MARIA_HA *maria_open(const char *name, int mode, uint open_flags)
                                            share->base.extra_rec_buff_size,
                                            share->base.max_key_length);
 
-    disk_pos_assert(disk_pos + share->base.fields *MARIA_COLUMNDEF_SIZE,
+    disk_pos_assert(share,
+                    disk_pos + share->base.fields *MARIA_COLUMNDEF_SIZE,
                     end_pos);
     for (i= j= 0 ; i < share->base.fields ; i++)
     {
@@ -1875,7 +1880,7 @@ int maria_enable_indexes(MARIA_HA *info)
     DBUG_PRINT("error", ("data_file_length: %lu  key_file_length: %lu",
                          (ulong) share->state.state.data_file_length,
                          (ulong) share->state.state.key_file_length));
-    maria_print_error(info->s, HA_ERR_CRASHED);
+    _ma_set_fatal_error(share, HA_ERR_CRASHED);
     error= HA_ERR_CRASHED;
   }
   else

storage/maria/ma_page.c

@@ -127,8 +127,7 @@ my_bool _ma_fetch_keypage(MARIA_PAGE *page, MARIA_HA *info,
   {
     DBUG_PRINT("error",("Got errno: %d from pagecache_read",my_errno));
     info->last_keypage=HA_OFFSET_ERROR;
-    maria_print_error(share, HA_ERR_CRASHED);
-    my_errno=HA_ERR_CRASHED;
+    _ma_set_fatal_error(share, HA_ERR_CRASHED);
     DBUG_RETURN(1);
   }
   info->last_keypage= pos;
@@ -159,8 +158,7 @@ my_bool _ma_fetch_keypage(MARIA_PAGE *page, MARIA_HA *info,
                           _ma_get_keynr(share, tmp)));
       DBUG_DUMP("page", tmp, page_size);
       info->last_keypage = HA_OFFSET_ERROR;
-      maria_print_error(share, HA_ERR_CRASHED);
-      my_errno= HA_ERR_CRASHED;
+      _ma_set_fatal_error(share, HA_ERR_CRASHED);
       DBUG_RETURN(1);
     }
   }
@@ -552,8 +550,7 @@ my_bool _ma_compact_keypage(MARIA_PAGE *ma_page, TrID min_read_from)
     {
       DBUG_PRINT("error",("Couldn't find last key:  page_pos: 0x%lx",
                           (long) page));
-      maria_print_error(share, HA_ERR_CRASHED);
-      my_errno=HA_ERR_CRASHED;
+      _ma_set_fatal_error(share, HA_ERR_CRASHED);
       DBUG_RETURN(1);
     }
     if (key_has_transid(page-1))

storage/maria/ma_pagecache.c

@@ -614,6 +614,26 @@ static my_bool pagecache_fwrite(PAGECACHE *pagecache,
   DBUG_ENTER("pagecache_fwrite");
   DBUG_ASSERT(type != PAGECACHE_READ_UNKNOWN_PAGE);
 
+#ifdef EXTRA_DEBUG_BITMAP
+  /*
+    This code is very good when debugging changes in bitmaps or dirty lists
+    The above define should be defined for all Aria files if you want to
+    debug either of the above issues.
+  */
+
+  if (pagecache->extra_debug)
+  {
+    char buff[80];
+    uint len= my_sprintf(buff,
+                         (buff, "fwrite: fd: %d  id: %u  page: %lu",
+                          filedesc->file,
+                          _ma_file_callback_to_id(filedesc->callback_data),
+                          (ulong) pageno));
+    (void) translog_log_debug_info(0, LOGREC_DEBUG_INFO_QUERY,
+                                   (uchar*) buff, len);
+  }
+#endif
+
   /* Todo: Integrate this with write_callback so we have only one callback */
   if ((*filedesc->flush_log_callback)(buffer, pageno, filedesc->callback_data))
     DBUG_RETURN(1);

storage/maria/ma_pagecache.h

@@ -173,6 +173,7 @@ typedef struct st_pagecache
   my_bool resize_in_flush;       /* true during flush of resize operation    */
   my_bool can_be_used;           /* usage of cache for read/write is allowed */
   my_bool in_init;		/* Set to 1 in MySQL during init/resize     */
+  my_bool extra_debug;	        /* set to 1 if one wants extra logging */
   HASH    files_in_flush;       /**< files in flush_pagecache_blocks_int() */
 } PAGECACHE;
 

storage/maria/ma_panic.c

@@ -67,8 +67,8 @@ int maria_panic(enum ha_panic_function flag)
       if (info->s->options & HA_OPTION_READ_ONLY_DATA)
 	break;
 #endif
-      if (flush_pagecache_blocks(info->s->pagecache, &info->s->kfile,
-                                 FLUSH_RELEASE))
+      if (_ma_flush_table_files(info, MARIA_FLUSH_DATA | MARIA_FLUSH_INDEX,
+                                FLUSH_RELEASE, FLUSH_RELEASE))
 	error=my_errno;
       if (info->opt_flag & WRITE_CACHE_USED)
 	if (flush_io_cache(&info->rec_cache))
@@ -92,8 +92,8 @@ int maria_panic(enum ha_panic_function flag)
       if (info->dfile.file >= 0 && my_close(info->dfile.file, MYF(0)))
 	error = my_errno;
       info->s->kfile.file= info->dfile.file= -1;/* Files aren't open anymore */
-      break;
 #endif
+      break;
     case HA_PANIC_READ:			/* Restore to before WRITE */
 #ifdef CANT_OPEN_FILES_TWICE
       {					/* Open closed files */

storage/maria/ma_recovery_util.c

@@ -59,9 +59,11 @@ void tprint(FILE *trace_file __attribute__ ((unused)),
   va_list args;
 #ifndef DBUG_OFF
   {
-    char buff[1024];
+    char buff[1024], *end;
     va_start(args, format);
     vsnprintf(buff, sizeof(buff)-1, format, args);
+    if (*(end= strend(buff)) == '\n')
+      *end= 0;                                  /* Don't print end \n */
     DBUG_PRINT("info", ("%s", buff));
     va_end(args);
   }

storage/maria/ma_rkey.c

@@ -91,8 +91,7 @@ int maria_rkey(MARIA_HA *info, uchar *buf, int inx, const uchar *key_data,
   case HA_KEY_ALG_RTREE:
     if (maria_rtree_find_first(info, &key, nextflag) < 0)
     {
-      maria_print_error(info->s, HA_ERR_CRASHED);
-      my_errno= HA_ERR_CRASHED;
+      _ma_set_fatal_error(share, HA_ERR_CRASHED);
       info->cur_row.lastpos= HA_OFFSET_ERROR;
     }
     break;

storage/maria/ma_search.c

@@ -379,8 +379,7 @@ int _ma_seq_search(const MARIA_KEY *key, const MARIA_PAGE *ma_page,
     length=(*keyinfo->get_key)(&tmp_key, page_flag, nod_flag, &page);
     if (length == 0 || page > end)
     {
-      maria_print_error(share, HA_ERR_CRASHED);
-      my_errno=HA_ERR_CRASHED;
+      _ma_set_fatal_error(share, HA_ERR_CRASHED);
       DBUG_PRINT("error",
                  ("Found wrong key:  length: %u  page: 0x%lx  end: 0x%lx",
                   length, (long) page, (long) end));
@@ -562,8 +561,7 @@ int _ma_prefix_search(const MARIA_KEY *key, const MARIA_PAGE *ma_page,
 
     if (page > end)
     {
-      maria_print_error(share, HA_ERR_CRASHED);
-      my_errno=HA_ERR_CRASHED;
+      _ma_set_fatal_error(share, HA_ERR_CRASHED);
       DBUG_PRINT("error",
                  ("Found wrong key:  length: %u  page: 0x%lx  end: %lx",
                   length, (long) page, (long) end));
@@ -1043,8 +1041,7 @@ uint _ma_get_pack_key(MARIA_KEY *int_key, uint page_flag,
       {
 	if (length > (uint) keyseg->length)
 	{
-          maria_print_error(keyinfo->share, HA_ERR_CRASHED);
-	  my_errno=HA_ERR_CRASHED;
+          _ma_set_fatal_error(keyinfo->share, HA_ERR_CRASHED);
 	  return 0;				/* Error */
 	}
 	if (length == 0)			/* Same key */
@@ -1059,8 +1056,7 @@ uint _ma_get_pack_key(MARIA_KEY *int_key, uint page_flag,
                        ("Found too long null packed key: %u of %u at 0x%lx",
                         length, keyseg->length, (long) *page_pos));
 	    DBUG_DUMP("key", *page_pos, 16);
-            maria_print_error(keyinfo->share, HA_ERR_CRASHED);
-	    my_errno=HA_ERR_CRASHED;
+            _ma_set_fatal_error(keyinfo->share, HA_ERR_CRASHED);
 	    return 0;
 	  }
 	  continue;
@@ -1117,8 +1113,7 @@ uint _ma_get_pack_key(MARIA_KEY *int_key, uint page_flag,
         DBUG_PRINT("error",("Found too long packed key: %u of %u at 0x%lx",
                             length, keyseg->length, (long) *page_pos));
         DBUG_DUMP("key", *page_pos, 16);
-        maria_print_error(keyinfo->share, HA_ERR_CRASHED);
-        my_errno=HA_ERR_CRASHED;
+        _ma_set_fatal_error(keyinfo->share, HA_ERR_CRASHED);
         return 0;                               /* Error */
       }
       store_key_length_inc(key,length);
@@ -1277,8 +1272,7 @@ uint _ma_get_binary_pack_key(MARIA_KEY *int_key, uint page_flag, uint nod_flag,
                  ("Found too long binary packed key: %u of %u at 0x%lx",
                   length, keyinfo->maxlength, (long) *page_pos));
       DBUG_DUMP("key", *page_pos, 16);
-      maria_print_error(keyinfo->share, HA_ERR_CRASHED);
-      my_errno=HA_ERR_CRASHED;
+      _ma_set_fatal_error(keyinfo->share, HA_ERR_CRASHED);
       DBUG_RETURN(0);                                 /* Wrong key */
     }
     /* Key is packed against prev key, take prefix from prev key. */
@@ -1369,8 +1363,7 @@ uint _ma_get_binary_pack_key(MARIA_KEY *int_key, uint page_flag, uint nod_flag,
     if (from_end != page_end)
     {
       DBUG_PRINT("error",("Error when unpacking key"));
-      maria_print_error(keyinfo->share, HA_ERR_CRASHED);
-      my_errno=HA_ERR_CRASHED;
+      _ma_set_fatal_error(keyinfo->share, HA_ERR_CRASHED);
       DBUG_RETURN(0);                                 /* Error */
     }
   }
@@ -1456,8 +1449,7 @@ uchar *_ma_get_key(MARIA_KEY *key, MARIA_PAGE *ma_page, uchar *keypos)
     {
       if (!(*keyinfo->get_key)(key, page_flag, nod_flag, &page))
       {
-        maria_print_error(keyinfo->share, HA_ERR_CRASHED);
-        my_errno=HA_ERR_CRASHED;
+        _ma_set_fatal_error(keyinfo->share, HA_ERR_CRASHED);
         DBUG_RETURN(0);
       }
     }
@@ -1507,8 +1499,7 @@ static my_bool _ma_get_prev_key(MARIA_KEY *key, MARIA_PAGE *ma_page,
     {
       if (! (*keyinfo->get_key)(key, page_flag, nod_flag, &page))
       {
-        maria_print_error(keyinfo->share, HA_ERR_CRASHED);
-        my_errno=HA_ERR_CRASHED;
+        _ma_set_fatal_error(keyinfo->share, HA_ERR_CRASHED);
         DBUG_RETURN(1);
       }
     }
@@ -1561,8 +1552,7 @@ uchar *_ma_get_last_key(MARIA_KEY *key, MARIA_PAGE *ma_page, uchar *endpos)
       {
         DBUG_PRINT("error",("Couldn't find last key:  page: 0x%lx",
                             (long) page));
-        maria_print_error(keyinfo->share, HA_ERR_CRASHED);
-        my_errno=HA_ERR_CRASHED;
+        _ma_set_fatal_error(keyinfo->share, HA_ERR_CRASHED);
         DBUG_RETURN(0);
       }
     }

storage/maria/ma_static.c

@@ -38,6 +38,8 @@ 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, maria_recovery_verbose= 0;
+my_bool maria_assert_if_crashed_table= 0;
+
 pthread_mutex_t THR_LOCK_maria;
 #if defined(THREAD) && !defined(DONT_USE_RW_LOCKS)
 ulong maria_concurrent_insert= 2;

storage/maria/ma_update.c

@@ -216,7 +216,10 @@ err:
 	{
 	  if ((flag++ && _ma_ft_del(info,i,new_key_buff,newrec,pos)) ||
 	      _ma_ft_add(info,i,old_key_buff,oldrec,pos))
+          {
+            _ma_set_fatal_error(share, my_errno);
 	    break;
+          }
 	}
 	else
 	{
@@ -228,16 +231,17 @@ err:
                                         oldrec, pos, info->cur_row.trid);
 	  if ((flag++ && _ma_ck_delete(info, &new_key)) ||
 	      _ma_ck_write(info, &old_key))
+          {
+            _ma_set_fatal_error(share, my_errno);
 	    break;
+          }
 	}
       }
     } while (i-- != 0);
   }
   else
-  {
-    maria_print_error(share, HA_ERR_CRASHED);
-    maria_mark_crashed(info);
-  }
+    _ma_set_fatal_error(share, save_errno);
+
   info->update= (HA_STATE_CHANGED | HA_STATE_AKTIV | HA_STATE_ROW_CHANGED |
 		 key_changed);
 
@@ -245,9 +249,6 @@ err:
   VOID(_ma_writeinfo(info,WRITEINFO_UPDATE_KEYFILE));
   allow_break();				/* Allow SIGHUP & SIGINT */
   if (save_errno == HA_ERR_KEY_NOT_FOUND)
-  {
-    maria_print_error(share, HA_ERR_CRASHED);
-    save_errno=HA_ERR_CRASHED;
-  }
+    _ma_set_fatal_error(share, HA_ERR_CRASHED);
   DBUG_RETURN(my_errno=save_errno);
 } /* maria_update */

storage/maria/ma_write.c

@@ -802,7 +802,7 @@ int _ma_insert(register MARIA_HA *info, MARIA_KEY *key,
   {
     if (t_length >= keyinfo->maxlength*2+MARIA_INDEX_OVERHEAD_SIZE)
     {
-      my_errno=HA_ERR_CRASHED;
+      _ma_set_fatal_error(share, HA_ERR_CRASHED);
       DBUG_RETURN(-1);
     }
     bmove_upp(endpos+t_length, endpos, (uint) (endpos-key_pos));
@@ -811,7 +811,7 @@ int _ma_insert(register MARIA_HA *info, MARIA_KEY *key,
   {
     if (-t_length >= keyinfo->maxlength*2+MARIA_INDEX_OVERHEAD_SIZE)
     {
-      my_errno=HA_ERR_CRASHED;
+      _ma_set_fatal_error(share, HA_ERR_CRASHED);
       DBUG_RETURN(-1);
     }
     bmove(key_pos,key_pos-t_length,(uint) (endpos-key_pos)+t_length);
@@ -1176,7 +1176,7 @@ static uchar *_ma_find_last_pos(MARIA_KEY *int_key, MARIA_PAGE *ma_page,
 
   if (!(length=(*keyinfo->get_key)(&tmp_key, page_flag, 0, &page)))
   {
-    my_errno=HA_ERR_CRASHED;
+    _ma_set_fatal_error(share, HA_ERR_CRASHED);
     DBUG_RETURN(0);
   }
 
@@ -1189,7 +1189,7 @@ static uchar *_ma_find_last_pos(MARIA_KEY *int_key, MARIA_PAGE *ma_page,
     memcpy(int_key->data, key_buff, length);		/* previous key */
     if (!(length=(*keyinfo->get_key)(&tmp_key, page_flag, 0, &page)))
     {
-      my_errno=HA_ERR_CRASHED;
+      _ma_set_fatal_error(share, HA_ERR_CRASHED);
       DBUG_RETURN(0);
     }
   } while (page < end);

storage/maria/maria_chk.c

@@ -1012,8 +1012,8 @@ static int maria_chk(HA_CHECK *param, char *filename)
                             share->state.open_count != 0);
 
     if ((param->testflag & (T_REP_ANY | T_SORT_RECORDS)) &&
-	((share->state.changed & (STATE_CHANGED | STATE_CRASHED |
-				  STATE_CRASHED_ON_REPAIR | STATE_IN_REPAIR) ||
+	((share->state.changed & (STATE_CHANGED | STATE_CRASHED_FLAGS |
+				  STATE_IN_REPAIR) ||
 	  !(param->testflag & T_CHECK_ONLY_CHANGED))))
       need_to_check=1;
 
@@ -1030,8 +1030,8 @@ static int maria_chk(HA_CHECK *param, char *filename)
         need_to_check=1;
     }
     if ((param->testflag & T_CHECK_ONLY_CHANGED) &&
-	(share->state.changed & (STATE_CHANGED | STATE_CRASHED |
-				 STATE_CRASHED_ON_REPAIR | STATE_IN_REPAIR)))
+	(share->state.changed & (STATE_CHANGED | STATE_CRASHED_FLAGS |
+				 STATE_IN_REPAIR)))
       need_to_check=1;
     if (!need_to_check)
     {
@@ -1250,8 +1250,8 @@ static int maria_chk(HA_CHECK *param, char *filename)
     if (!error)
     {
       DBUG_PRINT("info", ("Reseting crashed state"));
-      share->state.changed&= ~(STATE_CHANGED | STATE_CRASHED |
-                               STATE_CRASHED_ON_REPAIR | STATE_IN_REPAIR);
+      share->state.changed&= ~(STATE_CHANGED | STATE_CRASHED_FLAGS |
+                               STATE_IN_REPAIR);
     }
     else
       maria_mark_crashed(info);
@@ -1304,14 +1304,13 @@ static int maria_chk(HA_CHECK *param, char *filename)
     if (!error)
     {
       if (((share->state.changed &
-            (STATE_CHANGED | STATE_CRASHED | STATE_CRASHED_ON_REPAIR |
-             STATE_IN_REPAIR)) ||
+            (STATE_CHANGED | STATE_CRASHED_FLAGS | STATE_IN_REPAIR)) ||
            share->state.open_count != 0)
           && (param->testflag & T_UPDATE_STATE))
         info->update|=HA_STATE_CHANGED | HA_STATE_ROW_CHANGED;
       DBUG_PRINT("info", ("Reseting crashed state"));
-      share->state.changed&= ~(STATE_CHANGED | STATE_CRASHED |
-                               STATE_CRASHED_ON_REPAIR | STATE_IN_REPAIR);
+      share->state.changed&= ~(STATE_CHANGED | STATE_CRASHED_FLAGS |
+                               STATE_IN_REPAIR);
     }
     else if (!maria_is_crashed(info) &&
              (param->testflag & T_UPDATE_STATE))

storage/maria/maria_def.h

@@ -619,6 +619,9 @@ struct st_maria_handler
 #define STATE_NOT_MOVABLE        256
 #define STATE_MOVED              512 /* set if base->uuid != maria_uuid */
 #define STATE_IN_REPAIR  	 1024 /* We are running repair on table */
+#define STATE_CRASHED_PRINTED	 2048
+
+#define STATE_CRASHED_FLAGS (STATE_CRASHED | STATE_CRASHED_ON_REPAIR | STATE_CRASHED_PRINTED)
 
 /* options to maria_read_cache */
 
@@ -701,7 +704,6 @@ struct st_maria_handler
 #endif
 #define DBUG_DUMP_KEY(name, key) DBUG_DUMP(name, (key)->data, (key)->data_length + (key)->ref_length)
 
-
 /* Functions to store length of space packed keys, VARCHAR or BLOB keys */
 
 #define store_key_length(key,length) \
@@ -805,6 +807,7 @@ extern char *maria_data_root;
 extern uchar maria_zero_string[];
 extern my_bool maria_inited, maria_in_ha_maria, maria_recovery_changed_data;
 extern my_bool maria_recovery_verbose;
+extern my_bool maria_assert_if_crashed_table;
 extern HASH maria_stored_state;
 extern int (*maria_create_trn_hook)(MARIA_HA *);
 
@@ -918,6 +921,7 @@ extern int _ma_writeinfo(MARIA_HA *info, uint options);
 extern int _ma_test_if_changed(MARIA_HA *info);
 extern int _ma_mark_file_changed(MARIA_HA *info);
 extern void _ma_mark_file_crashed(MARIA_SHARE *share);
+void _ma_set_fatal_error(MARIA_SHARE *share, int error);
 extern my_bool _ma_set_uuid(MARIA_HA *info, my_bool reset_uuid);
 extern my_bool _ma_check_if_zero(uchar *pos, size_t size);
 extern int _ma_decrement_open_count(MARIA_HA *info);
@@ -1261,3 +1265,4 @@ extern my_bool maria_flush_log_for_page_none(uchar *page,
                                              pgcache_page_no_t page_no,
                                              uchar *data_ptr);
 extern PAGECACHE *maria_log_pagecache;
+extern uint _ma_file_callback_to_id(void *callback_data);

storage/myisam/mi_panic.c

@@ -71,8 +71,8 @@ int mi_panic(enum ha_panic_function flag)
       if (info->dfile >= 0 && my_close(info->dfile,MYF(0)))
 	error = my_errno;
       info->s->kfile=info->dfile= -1;	/* Files aren't open anymore */
-      break;
 #endif
+      break;
     case HA_PANIC_READ:			/* Restore to before WRITE */
 #ifdef CANT_OPEN_FILES_TWICE
       {					/* Open closed files */