WL#3072 Maria recovery

Misc changes:
- fix for benign Valgrind error, compiler warnings
- fix for a segfault in execution of maria_delete_all_rows() and one
when taking multiple checkpoints
- fix for too paranoid assertion
- adding ability to take checkpoints at the end of the REDO phase
and at the end of recovery.
- other minor changes


storage/maria/ha_maria.cc:
  The checkpoint done after Recovery is finished, is moved to
  maria_recover().
storage/maria/ma_bitmap.c:
  fix for Valgrind error: the "shadow debug copy" of the bitmap page
  started unitialized and so ma_print_bitmap() would use it uninitialized
storage/maria/ma_checkpoint.c:
  * reset pointers to NULL after freeing them, or we segfault at
  next checkpoint in my_realloc().
  * fix for compiler warnings.
storage/maria/ma_delete_all.c:
  info->trn is NULL for non-transactional tables
storage/maria/ma_locking.c:
  correct assertion (it fired wrongly in execution of REDO_DROP_TABLE
  due to the maria_extra(HA_PREPARE_FOR_DROP)->_ma_decrement_open_count()
  ->maria_lock_database(F_UNLCK); another solution would have been to
  not call _ma_decrement_open_count() (it's ok to have a wrong open
  count in a table which we are dropping), but the same problem
  would still exist for REDO_RENAME_TABLE.
storage/maria/ma_loghandler.c:
  fail early if UNRECOVERABLE_ERROR
storage/maria/ma_recovery.c:
  * new argument to maria_apply_log(): should it take checkpoints
  (at end of REDO phase and at the very end) or no.
  * moving the call to translog_next_LSN() into
  parse_checkpoint_record() ("hide the details").
  * Refining an error detection for something which could happen
  if there is a checkpoint record in the log.
  * Using close_one_table() instead of maria_extra(HA_EXTRA_PREPARE_FOR_DROP|RENAME),
  as it looks safer, and also changing how close_one_table() works:
  it now limits itself to scanning all_tables[], thus having one loopp
  instead of two, which should be faster (as a result, it does not
  close tables not registered in this array, which is ok as there
  should not be any).
storage/maria/ma_recovery.h:
  new parameter
storage/maria/maria_read_log.c:
  update to new prototype

storage/maria/ha_maria.cc

@@ -2401,9 +2401,7 @@ static int ha_maria_init(void *p)
                   MYSQL_VERSION_ID, server_id, maria_log_pagecache,
                   TRANSLOG_DEFAULT_FLAGS) ||
     maria_recover() ||
-    ma_checkpoint_init(FALSE) ||
-    /* One checkpoint after Recovery */
-    ma_checkpoint_execute(CHECKPOINT_FULL, FALSE);
+    ma_checkpoint_init(FALSE);
   maria_multi_threaded= TRUE;
   return res;
 }

storage/maria/ma_bitmap.c

@@ -217,6 +217,9 @@ my_bool _ma_bitmap_init(MARIA_SHARE *share, File file)
   bitmap->page= ~(ulonglong) 0;
   bitmap->used_size= bitmap->total_size;
   bfill(bitmap->map, share->block_size, 255);
+#ifndef DBUG_OFF
+  memcpy(bitmap->map + bitmap->block_size, bitmap->map, bitmap->block_size);
+#endif
   if (share->state.first_bitmap_with_space == ~(ulonglong) 0)
   {
     /* Start scanning for free space from start of file */

storage/maria/ma_checkpoint.c

@@ -390,6 +390,7 @@ void ma_checkpoint_end(void)
     pthread_mutex_unlock(&LOCK_checkpoint);
     my_free((uchar *)dfiles, MYF(MY_ALLOW_ZERO_PTR));
     my_free((uchar *)kfiles, MYF(MY_ALLOW_ZERO_PTR));
+    dfiles= kfiles= NULL;
     pthread_mutex_destroy(&LOCK_checkpoint);
     pthread_cond_destroy(&COND_checkpoint);
     checkpoint_inited= FALSE;
@@ -563,6 +564,8 @@ pthread_handler_t ma_checkpoint_background(void *arg __attribute__((unused)))
     TRANSLOG_ADDRESS log_horizon_at_last_checkpoint= LSN_IMPOSSIBLE;
     ulonglong pagecache_flushes_at_last_checkpoint= 0;
     struct timespec abstime;
+    LINT_INIT(kfile);
+    LINT_INIT(dfile);
     switch((sleeps++) % time_between_checkpoints)
     {
     case 0:
@@ -731,6 +734,7 @@ static int collect_tables(LEX_STRING *str, LSN checkpoint_start_log_horizon)
     *state_copies_end, /**< cache ends here */
     *state_copy; /**< iterator in cache */
   TRANSLOG_ADDRESS state_copies_horizon; /**< horizon of states' _copies_ */
+  LINT_INIT(state_copies_horizon);
   DBUG_ENTER("collect_tables");
 
   /* let's make a list of distinct shares */

storage/maria/ma_delete_all.c

@@ -95,6 +95,7 @@ int maria_delete_all_rows(MARIA_HA *info)
     The operations above on the index/data file will be forced to disk at
     Checkpoint or maria_close() time. So we can reset:
   */
+  if (log_record)
     info->trn->rec_lsn= LSN_IMPOSSIBLE;
 
   VOID(_ma_writeinfo(info,WRITEINFO_UPDATE_KEYFILE));

storage/maria/ma_locking.c

@@ -140,8 +140,13 @@ int maria_lock_database(MARIA_HA *info, int lock_type)
       }
       info->opt_flag&= ~(READ_CACHE_USED | WRITE_CACHE_USED);
       info->lock_type= F_UNLCK;
-      /* verify that user of the table cleaned up after itself */
-      DBUG_ASSERT(share->now_transactional == share->base.born_transactional);
+      /*
+        Verify that user of the table cleaned up after itself. Not in
+        recovery, as for example maria_extra(HA_EXTRA_PREPARE_FOR_RENAME) may
+        call us here, with transactionality temporarily disabled.
+      */
+      DBUG_ASSERT(maria_in_recovery ||
+                  share->now_transactional == share->base.born_transactional);
       break;
     case F_RDLCK:
       if (info->lock_type == F_WRLCK)

storage/maria/ma_loghandler.c

@@ -55,6 +55,7 @@
     DBUG_PRINT("error", E); \
     printf E; \
     putchar('\n'); \
+    DBUG_ASSERT(0); \
   } while(0);
 
 /* Maximum length of compressed LSNs (the worst case of whole LSN storing) */

storage/maria/ma_recovery.h

@@ -26,5 +26,6 @@
 C_MODE_START
 int maria_recover(void);
 int maria_apply_log(LSN lsn, my_bool apply, FILE *trace_file,
-                    my_bool execute_undo_phase, my_bool skip_DDLs);
+                    my_bool execute_undo_phase, my_bool skip_DDLs,
+                    my_bool take_checkpoints);
 C_MODE_END

storage/maria/maria_read_log.c

@@ -103,7 +103,7 @@ int main(int argc, char **argv)
 
   fprintf(stdout, "TRACE of the last maria_read_log\n");
   if (maria_apply_log(lsn, opt_apply, opt_silent ? NULL : stdout,
-                      opt_apply_undo, FALSE))
+                      opt_apply_undo, FALSE, FALSE))
     goto err;
   fprintf(stdout, "%s: SUCCESS\n", my_progname_short);