Fixed recovery problem in Aria where bitmap had wrong information after recovery.

LP#619731: Aria recovery corruption "Page 1: Row: 1 has an extent with wrong information in bitmap

storage/maria/ma_bitmap.c:
  Don't send broadcast if no one is waiting for it
storage/maria/ma_blockrec.c:
  Don't update bitmap if the page is not in the dirty_page list (or LSN is after checkpoint start)
  Fixes the case where we have in the log redo_free_block followed by another redo entry for the same page which is ignored.
  Also fixed that _ma_apply_redo_insert_row_blobs() doesn't update the bitmap in similar circumstances.
storage/maria/ma_blockrec.h:
  Updated prototype
storage/maria/ma_check.c:
  Added printing of bitmap information if used with maria_chk -vvv (for debugging)
storage/maria/ma_recovery.c:
  Updated call parameters to _ma_apply_redo_free_blocks().

1 files changed, 58 insertions, 5 deletions

diff --git a/storage/maria/ma_check.c b/storage/maria/ma_check.c
index 1cb14567804..631fa879504 100644
--- a/storage/maria/ma_check.c
+++ b/storage/maria/ma_check.c
@@ -100,6 +100,9 @@ static my_bool _ma_flush_table_files_before_swap(HA_CHECK *param,
 static TrID max_trid_in_system(void);
 static void _ma_check_print_not_visible_error(HA_CHECK *param, TrID used_trid);
 void retry_if_quick(MARIA_SORT_PARAM *param, int error);
+static void print_bitmap_description(MARIA_SHARE *share,
+                                     pgcache_page_no_t page,
+                                     uchar *buff);
 
 
 /* Initialize check param with default values */
@@ -1842,6 +1845,8 @@ static int check_block_record(HA_CHECK *param, MARIA_HA *info, int extend,
       }
       param->used+= block_size;
       param->link_used+= block_size;
+      if (param->verbose > 2)
+        print_bitmap_description(share, page, bitmap_buff);
       continue;
     }
     /* Skip pages marked as empty in bitmap */
@@ -2177,12 +2182,17 @@ int maria_chk_data_link(HA_CHECK *param, MARIA_HA *info, my_bool extend)
            llstr(param->del_length, llbuff2));
     printf("Empty space:  %12s    Linkdata:     %10s\n",
            llstr(param->empty, llbuff),llstr(param->link_used, llbuff2));
-    if (param->lost)
-      printf("Lost space:   %12s", llstr(param->lost, llbuff));
-    if (param->max_found_trid)
+    if (share->data_file_type == BLOCK_RECORD)
     {
-      printf("Max trans. id: %11s\n",
-             llstr(param->max_found_trid, llbuff));
+      printf("Full pages:   %12s    Tail count: %12s\n",
+             llstr(param->full_page_count, llbuff),
+             llstr(param->tail_count, llbuff2));
+      printf("Lost space:   %12s\n", llstr(param->lost, llbuff));
+      if (param->max_found_trid)
+      {
+        printf("Max trans. id: %11s\n",
+               llstr(param->max_found_trid, llbuff));
+      }
     }
   }
   my_free(record,MYF(0));
@@ -6799,3 +6809,46 @@ void retry_if_quick(MARIA_SORT_PARAM *sort_param, int error)
     param->testflag|=T_RETRY_WITHOUT_QUICK;
   }
 }
+
+/* Print information about bitmap page */
+
+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);
+}