Improve logging of Aria redo's and undo's

- Remove extra ',' and quotes
- Remove extra newline and remove double newlines
- Added options --lsn-redo-end and --lsn-undo-end to aria_read_log
- Allow one to give the aria_read_log lsn aruments as number,0xhexnumber,
  the same way as lsn's are written by aria_read_log
- Don't write full pages to redo log with EXTRA_DEBUG as this takes up
  a lot of disk and there has not been a need for this extra loggging for
  a long time. Instead one should use EXTRA_ARIA_DEBUG instead.

storage/maria/ma_blockrec.c

@@ -3682,6 +3682,15 @@ my_bool _ma_write_abort_block_record(MARIA_HA *info)
   _ma_bitmap_unlock(share);
   if (share->now_transactional)
   {
+    /*
+      Write clr to mark end of aborted row insert.
+      The above delete_head_or_tail() calls will only log redo, not undo.
+      The undo just before the row insert is stored in row->orig_undo_lsn.
+
+      When applying undo's, we can skip all undo records between current
+      lsn and row->orig_undo_lsn as logically things are as before the
+      attempted insert.
+    */
     if (_ma_write_clr(info, info->cur_row.orig_undo_lsn,
                       LOGREC_UNDO_ROW_INSERT,
                       share->calc_checksum != 0,

storage/maria/ma_check.c

@@ -590,8 +590,8 @@ int maria_chk_key(HA_CHECK *param, register MARIA_HA *info)
       {
 	_ma_check_print_error(param,"Found %s keys of %s",llstr(keys,buff),
 		    llstr(share->state.state.records,buff2));
-	if (!(param->testflag & T_INFO))
-	DBUG_RETURN(-1);
+	if (!(param->testflag & (T_INFO | T_EXTEND)))
+          DBUG_RETURN(-1);
 	result= -1;
 	continue;
       }
@@ -1119,8 +1119,8 @@ static uint isam_key_length(MARIA_HA *info, register MARIA_KEYDEF *keyinfo)
 
 
 
-static void record_pos_to_txt(MARIA_HA *info, my_off_t recpos,
-                              char *buff)
+static char * record_pos_to_txt(MARIA_HA *info, my_off_t recpos,
+                                char *buff)
 {
   if (info->s->data_file_type != BLOCK_RECORD)
     llstr(recpos, buff);
@@ -1132,6 +1132,7 @@ static void record_pos_to_txt(MARIA_HA *info, my_off_t recpos,
     *(end++)= ':';
     longlong10_to_str(row, end, 10);
   }
+  return buff;
 }
 
 
@@ -1195,11 +1196,14 @@ static int check_keys_in_record(HA_CHECK *param, MARIA_HA *info, int extend,
             _ma_search(info, &key, SEARCH_SAME, share->state.key_root[keynr]);
           if (search_result)
           {
-            record_pos_to_txt(info, start_recpos, llbuff);
             _ma_check_print_error(param,
                                   "Record at: %14s  "
                                   "Can't find key for index: %2d",
-                                  llbuff, keynr+1);
+                                  record_pos_to_txt(info, start_recpos,
+                                                    llbuff),
+                                  keynr+1);
+            if (param->testflag & T_VERBOSE)
+              _ma_print_key(stdout, &key);
             if (param->err_count++ > MAXERR || !(param->testflag & T_VERBOSE))
               return -1;
           }
@@ -2731,8 +2735,11 @@ int maria_repair(HA_CHECK *param, register MARIA_HA *info,
                               "Duplicate key %2d for record at %10s against "
                               "new record at %10s",
                               info->errkey+1,
-                              llstr(sort_param.current_filepos, llbuff),
-                              llstr(info->dup_key_pos,llbuff2));
+                              record_pos_to_txt(info,
+                                                sort_param.current_filepos,
+                                                llbuff),
+                              record_pos_to_txt(info,
+                                                info->dup_key_pos, llbuff2));
       if (param->testflag & T_VERBOSE)
       {
         MARIA_KEY tmp_key;
@@ -4895,10 +4902,12 @@ static int sort_get_next_record(MARIA_SORT_PARAM *sort_param)
           {
             if (param->testflag & T_VERBOSE)
             {
-              record_pos_to_txt(info, info->cur_row.lastpos, llbuff);
               _ma_check_print_info(param,
                                    "Found record with wrong checksum at %s",
-                                   llbuff);
+                                   record_pos_to_txt(info,
+                                                     info->cur_row.lastpos,
+                                                     llbuff));
+
             }
             continue;
           }
@@ -5517,6 +5526,7 @@ static int sort_key_write(MARIA_SORT_PARAM *sort_param, const uchar *a)
   char llbuff[22],llbuff2[22];
   MARIA_SORT_INFO *sort_info=sort_param->sort_info;
   HA_CHECK *param= sort_info->param;
+  MARIA_HA *info= sort_info->info;
   int cmp;
 
   if (sort_info->key_block->inited)
@@ -5559,11 +5569,14 @@ static int sort_key_write(MARIA_SORT_PARAM *sort_param, const uchar *a)
 			   "Duplicate key %2u for record at %10s against "
                             "record at %10s",
                             sort_param->key + 1,
-                            llstr(sort_info->info->cur_row.lastpos, llbuff),
-                            llstr(get_record_for_key(sort_param->keyinfo,
-                                                     sort_info->key_block->
-                                                     lastkey),
-                                  llbuff2));
+                            record_pos_to_txt(info,
+                                              sort_info->info->cur_row.lastpos,
+                                              llbuff),
+                            record_pos_to_txt(info,
+                                              get_record_for_key(sort_param->
+                                                                 keyinfo,
+                                                                 sort_info->key_block->lastkey),
+                                              llbuff2));
     param->testflag|=T_RETRY_WITHOUT_QUICK;
     if (sort_info->param->testflag & T_VERBOSE)
       _ma_print_keydata(stdout,sort_param->seg, a, USE_WHOLE_KEY);

storage/maria/ma_recovery.h

@@ -26,10 +26,11 @@ C_MODE_START
 enum maria_apply_log_way
 { MARIA_LOG_APPLY, MARIA_LOG_DISPLAY_HEADER, MARIA_LOG_CHECK };
 int maria_recovery_from_log(void);
-int maria_apply_log(LSN lsn, LSN lsn_end, enum maria_apply_log_way apply,
+int maria_apply_log(LSN lsn, LSN lsn_end, LSN lsn_undo_end,
+                    enum maria_apply_log_way apply,
                     FILE *trace_file,
-                    my_bool execute_undo_phase, my_bool skip_DDLs,
-                    my_bool take_checkpoints, uint *warnings_count);
+                    my_bool skip_DDLs, my_bool take_checkpoints,
+                    uint *warnings_count);
 /* Table of tables to recover */
 extern HASH tables_to_redo;
 extern ulong maria_recovery_force_crash_counter;

storage/maria/maria_def.h

@@ -36,16 +36,23 @@ C_MODE_START
 #include <waiting_threads.h>
 #include <mysql/psi/mysql_file.h>
 
-/* For testing recovery */
-#ifdef TO_BE_REMOVED
-#define IDENTICAL_PAGES_AFTER_RECOVERY 1
-#endif
 /* Do extra sanity checking */
 #define SANITY_CHECKS 1
 #ifdef EXTRA_DEBUG
 #define EXTRA_DEBUG_KEY_CHANGES
+#endif
+/*
+  The following defines can be used when one has problems with redo logging
+  Setting this will log the full key page which can be compared with the
+  redo-changed key page. This will however make the aria log files MUCH bigger.
+*/
+#if defined(EXTRA_ARIA_DEBUG)
 #define EXTRA_STORE_FULL_PAGE_IN_KEY_CHANGES
 #endif
+/* For testing recovery */
+#ifdef TO_BE_REMOVED
+#define IDENTICAL_PAGES_AFTER_RECOVERY 1
+#endif
 
 #define MAX_NONMAPPED_INSERTS 1000
 #define MARIA_MAX_TREE_LEVELS 32

storage/maria/maria_read_log.c

 /* Copyright (C) 2007 MySQL AB
    Copyright (C) 2010 Monty Program Ab
+   Copyright (C) 2020 MariaDB Corporation Ab
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -29,15 +30,51 @@ const char *default_dbug_option= "d:t:O,\\aria_read_log.trace";
 const char *default_dbug_option= "d:t:o,/tmp/aria_read_log.trace";
 #endif
 #endif /* DBUG_OFF */
-static my_bool opt_display_only, opt_apply, opt_apply_undo, opt_silent;
-static my_bool opt_check;
+static my_bool opt_display_only, opt_apply, opt_silent, opt_apply_undo;
+static my_bool opt_check, opt_start_from_checkpoint;
 static my_bool opt_print_aria_log_control;
 static const char *opt_tmpdir;
 static ulong opt_translog_buffer_size;
 static ulonglong opt_page_buffer_size;
-static ulonglong opt_start_from_lsn, opt_end_lsn, opt_start_from_checkpoint;
+static ulonglong opt_start_from_lsn, opt_lsn_redo_end, opt_lsn_undo_end;
+static char *start_from_lsn_buf, *lsn_redo_end_buf, *lsn_undo_end_buf;
 static MY_TMPDIR maria_chk_tmpdir;
 
+/*
+  Get lsn from file number and offset
+  Format supported:
+  ulonglong
+  uint,0xhex
+*/
+
+static ulonglong get_lsn(const char *lsn_str)
+{
+  ulong file;
+  ulonglong pos;
+  if (sscanf(lsn_str, " %lu,0x%Lx", &file, &pos) == 2)
+    return MAKE_LSN(file, pos);
+  if (sscanf(lsn_str, " %Lu", &pos) == 1)
+    return pos;
+  return ~(ulonglong) 0;                        /* Error */
+}
+
+static my_bool get_lsn_arg(const char *lsn_string, ulonglong *lsn,
+                           const char *name)
+{
+  ulonglong value;
+  value= get_lsn(lsn_string);
+  if (value != ~(ulonglong) 0)
+  {
+    *lsn= value;
+    return 0;
+  }
+  fprintf(stderr,
+          "Wrong value '%s' for option %s. Value should be in format: "
+          "number,0xhexnumber\n",
+          lsn_string, name);
+  return 1;
+}
+
 
 int main(int argc, char **argv)
 {
@@ -136,17 +173,12 @@ int main(int argc, char **argv)
             LSN_IN_PARTS(lsn));
   }
 
-  if (opt_end_lsn != LSN_IMPOSSIBLE)
-  {
-    /* We can't apply undo if we use end_lsn */
-    opt_apply_undo= 0;
-  }
-
   fprintf(stdout, "TRACE of the last aria_read_log\n");
-  if (maria_apply_log(lsn, opt_end_lsn, opt_apply ?  MARIA_LOG_APPLY :
+  if (maria_apply_log(lsn, opt_lsn_redo_end, opt_lsn_undo_end,
+                      opt_apply ?  MARIA_LOG_APPLY :
                       (opt_check ? MARIA_LOG_CHECK :
                        MARIA_LOG_DISPLAY_HEADER), opt_silent ? NULL : stdout,
-                      opt_apply_undo, FALSE, FALSE, &warnings_count))
+                      FALSE, FALSE, &warnings_count))
     goto err;
   if (warnings_count == 0)
     fprintf(stdout, "%s: SUCCESS\n", my_progname_short);
@@ -204,9 +236,16 @@ static struct my_option my_long_options[] =
   {"display-only", 'd', "display brief info read from records' header",
    &opt_display_only, &opt_display_only, 0, GET_BOOL,
    NO_ARG,0, 0, 0, 0, 0, 0},
-  { "end-lsn", 'e', "Stop applying at this lsn. If end-lsn is used, UNDO:s "
-    "will not be applied", &opt_end_lsn, &opt_end_lsn,
-    0, GET_ULL, REQUIRED_ARG, 0, 0, ~(longlong) 0, 0, 0, 0 },
+  { "end-lsn", 'e', "Alias for lsn-redo-end",
+    &lsn_redo_end_buf, &lsn_redo_end_buf, 0, GET_STR, REQUIRED_ARG, 0, 0,
+    0, 0, 0, 0 },
+  { "lsn-redo-end", 'e', "Stop applying at this lsn during redo. If "
+    "this option is used UNDO:s will not be applied unless --lsn-undo-end is "
+    "given", &lsn_redo_end_buf,
+    &lsn_redo_end_buf, 0, GET_STR, REQUIRED_ARG, 0, 0, 0, 0, 0, 0 },
+  { "lsn-undo-end", 'E', "Stop applying undo after this lsn has been applied",
+    &lsn_undo_end_buf, &lsn_undo_end_buf, 0, GET_STR, REQUIRED_ARG, 0, 0,
+    0, 0, 0, 0 },
   {"aria-log-dir-path", 'h',
     "Path to the directory where to store transactional log",
     (uchar **) &maria_data_root, (uchar **) &maria_data_root, 0,
@@ -246,7 +285,9 @@ static struct my_option my_long_options[] =
     GET_ULONG, REQUIRED_ARG, (long) TRANSLOG_PAGECACHE_SIZE,
     1024L*1024L, (long) ~(ulong) 0, (long) MALLOC_OVERHEAD,
     (long) IO_SIZE, 0},
-  {"undo", 'u', "Apply UNDO records to tables. (disable with --disable-undo)",
+  {"undo", 'u',
+   "Apply UNDO records to tables. (disable with --disable-undo). "
+   "Will be automatically set if lsn-undo-end is used",
    (uchar **) &opt_apply_undo, (uchar **) &opt_apply_undo, 0,
    GET_BOOL, NO_ARG, 1, 0, 0, 0, 0, 0},
   {"verbose", 'v', "Print more information during apply/undo phase",
@@ -257,10 +298,9 @@ static struct my_option my_long_options[] =
   { 0, 0, 0, 0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0}
 };
 
-
 static void print_version(void)
 {
-  printf("%s Ver 1.4 for %s on %s\n",
+  printf("%s Ver 1.5 for %s on %s\n",
               my_progname_short, SYSTEM_TYPE, MACHINE_TYPE);
 }
 
@@ -268,7 +308,7 @@ static void print_version(void)
 static void usage(void)
 {
   print_version();
-  puts("Copyright (C) 2007 MySQL AB, 2009-2011 Monty Program Ab");
+  puts("Copyright (C) 2007 MySQL AB, 2009-2011 Monty Program Ab, 2020 MariaDB Corporation");
   puts("This software comes with ABSOLUTELY NO WARRANTY. This is free software,");
   puts("and you are welcome to modify and redistribute it under the GPL license\n");
 
@@ -312,6 +352,9 @@ get_one_option(const struct my_option *opt,
   case 'V':
     print_version();
     exit(0);
+  case 'E':
+    opt_apply_undo= TRUE;
+    break;
   case 'T':
   {
     char *pos;
@@ -341,13 +384,34 @@ get_one_option(const struct my_option *opt,
 static void get_options(int *argc,char ***argv)
 {
   int ho_error;
-  my_bool need_help= 0;
+  my_bool need_help= 0, need_abort= 0;
 
   if ((ho_error=handle_options(argc, argv, my_long_options, get_one_option)))
     exit(ho_error);
 
+  if (start_from_lsn_buf)
+  {
+    if (get_lsn_arg(start_from_lsn_buf, &opt_start_from_lsn,
+                    "start-from-lsn"))
+      need_abort= 1;
+  }
+  if (lsn_redo_end_buf)
+  {
+    if (get_lsn_arg(lsn_redo_end_buf, &opt_lsn_redo_end,
+                    "lsn-redo-end"))
+      need_abort= 1;
+  }
+  if (lsn_undo_end_buf)
+  {
+    if (get_lsn_arg(lsn_undo_end_buf, &opt_lsn_undo_end,
+                    "lsn-undo-end"))
+      need_abort= 1;
+  }
+
   if (!opt_apply)
     opt_apply_undo= FALSE;
+  if (!opt_apply_undo)
+    opt_lsn_undo_end= LSN_MAX;
 
   if (*argc > 0)
   {
@@ -356,21 +420,20 @@ static void get_options(int *argc,char ***argv)
   }
   if ((opt_display_only + opt_apply + opt_print_aria_log_control) != 1)
   {
-    need_help= 1;
+    need_abort= 1;
     fprintf(stderr,
             "You must use one and only one of the options 'display-only', \n"
             "'print-log-control-file' and 'apply'\n");
   }
 
-  if (need_help)
+  if (need_help || need_abort)
   {
     fflush(stderr);
-    need_help =1;
-    usage();
+    if (need_help)
+      usage();
     exit(1);
   }
   if (init_tmpdir(&maria_chk_tmpdir, opt_tmpdir))
     exit(1);
   maria_tmpdir= &maria_chk_tmpdir;
 }
-