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.
6 years ago · c52e62a76f
6 changed files with 212 additions and 85 deletions
--- a/storage/maria/ma_blockrec.c
+++ b/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,
--- a/storage/maria/ma_check.c
+++ b/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);
--- a/storage/maria/ma_recovery.c
+++ b/storage/maria/ma_recovery.c
@ -54,6 +54,7 @@ static my_bool skip_DDLs; /**< if REDO phase should skip DDL records */
 static my_bool checkpoint_useful;
 static my_bool in_redo_phase;
 static my_bool trns_created;
+static int aria_undo_aborted= 0;
 static ulong skipped_undo_phase;
 static ulonglong now; /**< for tracking execution time of phases */
 static void (*save_error_handler_hook)(uint, const char *,myf);
@ -115,7 +116,7 @@ prototype_undo_exec_hook(UNDO_BULK_INSERT);
 static int run_redo_phase(LSN lsn, LSN end_lsn,
                          enum maria_apply_log_way apply);
 static uint end_of_redo_phase(my_bool prepare_for_undo_phase);
-static int run_undo_phase(uint uncommitted);
+static int run_undo_phase(LSN end_undo_lsn, uint uncommitted);
 static void display_record_position(const LOG_DESC *log_desc,
                                    const TRANSLOG_HEADER_BUFFER *rec,
                                    uint number);
@ -236,8 +237,8 @@ int maria_recovery_from_log(void)
 #endif
  tprint(trace_file, "TRACE of the last Aria recovery from mysqld\n");
  DBUG_ASSERT(maria_pagecache->inited);
-  res= maria_apply_log(LSN_IMPOSSIBLE, LSN_IMPOSSIBLE, MARIA_LOG_APPLY,
-                       trace_file, TRUE, TRUE, TRUE, &warnings_count);
+  res= maria_apply_log(LSN_IMPOSSIBLE, LSN_IMPOSSIBLE, 0, MARIA_LOG_APPLY,
+                       trace_file, TRUE, TRUE, &warnings_count);
  if (!res)
  {
    if (warnings_count == 0 && recovery_found_crashed_tables == 0)
@ -258,7 +259,9 @@ int maria_recovery_from_log(void)

   @param  from_lsn        LSN from which log reading/applying should start;
                           LSN_IMPOSSIBLE means "use last checkpoint"
-   @param  end_lsn         Apply until this. LSN_IMPOSSIBLE means until end.
+   @param  end_redo_lsn    Apply until this. LSN_IMPOSSIBLE means until end.
+   @param  end_und_lsn     Apply all undo >= end_undo_lsn. Set to LSN_MAX if
+                           no undo's should be applied.
   @param  apply           how log records should be applied or not
   @param  trace_file      trace file where progress/debug messages will go
   @param  skip_DDLs_arg   Should DDL records (CREATE/RENAME/DROP/REPAIR)
@ -275,10 +278,10 @@ int maria_recovery_from_log(void)
     @retval !=0    Error
 */

-int maria_apply_log(LSN from_lsn, LSN end_lsn,
+int maria_apply_log(LSN from_lsn, LSN end_redo_lsn, LSN end_undo_lsn,
                    enum maria_apply_log_way apply,
                    FILE *trace_file,
-                    my_bool should_run_undo_phase, my_bool skip_DDLs_arg,
+                    my_bool skip_DDLs_arg,
                    my_bool take_checkpoints, uint *warnings_count)
 {
  int error= 0;
@ -287,14 +290,13 @@ int maria_apply_log(LSN from_lsn, LSN end_lsn,
  my_bool abort_message_printed= 0;
  DBUG_ENTER("maria_apply_log");

-  DBUG_ASSERT(apply == MARIA_LOG_APPLY || !should_run_undo_phase);
+  DBUG_ASSERT(apply == MARIA_LOG_APPLY || end_undo_lsn == LSN_MAX);
  DBUG_ASSERT(!maria_multi_threaded);
  recovery_warnings= recovery_found_crashed_tables= 0;
  skipped_lsn_err_count= 0;
  maria_recovery_changed_data= 0;
  /* checkpoints can happen only if TRNs have been built */
-  DBUG_ASSERT(should_run_undo_phase || !take_checkpoints);
-  DBUG_ASSERT(end_lsn == LSN_IMPOSSIBLE || should_run_undo_phase == 0);
+  DBUG_ASSERT(end_undo_lsn != LSN_MAX || !take_checkpoints);
  all_active_trans= (struct st_trn_for_recovery *)
    my_malloc(PSI_INSTRUMENT_ME, (SHORT_TRID_MAX + 1) * sizeof(struct st_trn_for_recovery),
              MYF(MY_ZEROFILL));
@ -313,6 +315,7 @@ int maria_apply_log(LSN from_lsn, LSN end_lsn,

  recovery_message_printed= REC_MSG_NONE;
  checkpoint_useful= trns_created= FALSE;
+  aria_undo_aborted= 0;
  tracef= trace_file;
 #ifdef INSTANT_FLUSH_OF_MESSAGES
  /* enable this for instant flush of messages to trace file */
@ -347,7 +350,7 @@ int maria_apply_log(LSN from_lsn, LSN end_lsn,

  now= microsecond_interval_timer();
  in_redo_phase= TRUE;
-  if (run_redo_phase(from_lsn, end_lsn, apply))
+  if (run_redo_phase(from_lsn, end_redo_lsn, apply))
  {
    ma_message_no_user(0, "Redo phase failed");
    trnman_destroy();
@ -355,7 +358,8 @@ int maria_apply_log(LSN from_lsn, LSN end_lsn,
  }
  trnman_destroy();

-  if (end_lsn != LSN_IMPOSSIBLE)
+  if (end_redo_lsn != LSN_IMPOSSIBLE &&
+      (end_undo_lsn == LSN_MAX || end_undo_lsn == LSN_IMPOSSIBLE))
  {
    abort_message_printed= 1;
    if (!trace_file)
@ -367,7 +371,7 @@ int maria_apply_log(LSN from_lsn, LSN end_lsn,
  }

  if ((uncommitted_trans=
-       end_of_redo_phase(should_run_undo_phase)) == (uint)-1)
+       end_of_redo_phase(end_undo_lsn != LSN_MAX)) == (uint)-1)
  {
    ma_message_no_user(0, "End of redo phase failed");
    goto err;
@ -417,13 +421,19 @@ int maria_apply_log(LSN from_lsn, LSN end_lsn,
  }
 #endif

-  if (should_run_undo_phase)
+  if (end_undo_lsn != LSN_MAX)
  {
-    if (run_undo_phase(uncommitted_trans))
+    if (run_undo_phase(end_undo_lsn, uncommitted_trans))
    {
      ma_message_no_user(0, "Undo phase failed");
      goto err;
    }
+    if (aria_undo_aborted)
+      ma_message_no_user(0, "Undo phase aborted in the middle on user request");
+    else if (end_redo_lsn != LSN_IMPOSSIBLE)
+      my_message(HA_ERR_INITIALIZATION,
+                 "Maria recovery aborted as end_lsn followed by end_undo was "
+                 "reached", MYF(0));
  }
  else if (uncommitted_trans > 0)
  {
@ -493,7 +503,8 @@ err:
 err2:
  if (trns_created)
    delete_all_transactions();
-  error= 1;
+  if (!abort_message_printed)
+    error= 1;
  if (close_all_tables())
  {
    ma_message_no_user(0, "closing of tables failed");
@ -521,7 +532,7 @@ end:
      fprintf(stderr, "\n");
      fflush(stderr);
    }
-    if (!error)
+    if (!error && !abort_message_printed)
    {
      ma_message_no_user(ME_NOTE, "recovery done");
      maria_recovery_changed_data= 1;
@ -540,7 +551,7 @@ end:
  {
    my_message(HA_ERR_INITIALIZATION,
               "Aria recovery failed. Please run aria_chk -r on all Aria "
-               "tables and delete all aria_log.######## files", MYF(0));
+               "tables (*.MAI) and delete all aria_log.######## files", MYF(0));
  }
  procent_printed= 0;
  /*
@ -842,7 +853,7 @@ prototype_redo_exec_hook(REDO_CREATE_TABLE)
    if (cmp_translog_addr(share->state.create_rename_lsn, rec->lsn) >= 0)
    {
      tprint(tracef, "Table '%s' has create_rename_lsn " LSN_FMT " more "
-             "recent than record, ignoring creation",
+             "recent than record, ignoring creation\n",
             name, LSN_IN_PARTS(share->state.create_rename_lsn));
      error= 0;
      goto end;
@ -974,7 +985,7 @@ prototype_redo_exec_hook(REDO_RENAME_TABLE)
  }
  old_name= (char *)log_record_buffer.str;
  new_name= old_name + strlen(old_name) + 1;
-  tprint(tracef, "Table '%s' to rename to '%s'; old-name table ", old_name,
+  tprint(tracef, "Table '%s' to be renamed to '%s'; old-name table ", old_name,
         new_name);
  /*
    Here is why we skip CREATE/DROP/RENAME when doing a recovery from
@ -1010,14 +1021,14 @@ prototype_redo_exec_hook(REDO_RENAME_TABLE)
    MARIA_SHARE *share= info->s;
    if (!share->base.born_transactional)
    {
-      tprint(tracef, ", is not transactional, ignoring renaming\n");
+      tprint(tracef, "is not transactional, ignoring renaming");
      ALERT_USER();
      error= 0;
      goto end;
    }
    if (cmp_translog_addr(share->state.create_rename_lsn, rec->lsn) >= 0)
    {
-      tprint(tracef, ", has create_rename_lsn " LSN_FMT " more recent than"
+      tprint(tracef, "has create_rename_lsn " LSN_FMT " more recent than"
             " record, ignoring renaming",
             LSN_IN_PARTS(share->state.create_rename_lsn));
      error= 0;
@ -1060,19 +1071,19 @@ prototype_redo_exec_hook(REDO_RENAME_TABLE)
    /* We should not have open instances on this table. */
    if (share->reopen != 1)
    {
-      tprint(tracef, ", is already open (reopen=%u)\n", share->reopen);
+      tprint(tracef, "is already open (reopen=%u)", share->reopen);
      ALERT_USER();
      goto end;
    }
    if (!share->base.born_transactional)
    {
-      tprint(tracef, ", is not transactional, ignoring renaming\n");
+      tprint(tracef, "is not transactional, ignoring renaming");
      ALERT_USER();
      goto drop;
    }
    if (cmp_translog_addr(share->state.create_rename_lsn, rec->lsn) >= 0)
    {
-      tprint(tracef, ", has create_rename_lsn " LSN_FMT " more recent than"
+      tprint(tracef, "has create_rename_lsn " LSN_FMT " more recent than"
             " record, ignoring renaming",
             LSN_IN_PARTS(share->state.create_rename_lsn));
      /*
@ -1090,7 +1101,7 @@ prototype_redo_exec_hook(REDO_RENAME_TABLE)
    }
    if (maria_is_crashed(info))
    {
-      tprint(tracef, ", is crashed, can't rename it");
+      tprint(tracef, "is crashed, can't rename it");
      ALERT_USER();
      goto end;
    }
@ -1168,7 +1179,7 @@ prototype_redo_exec_hook(REDO_REPAIR_TABLE)
  if (!info)
  {
    /* no such table, don't need to warn */
-    return 0;
+    DBUG_RETURN(0);
  }

  if (maria_is_crashed(info))
@ -1907,7 +1918,7 @@ prototype_redo_exec_hook(UNDO_ROW_INSERT)
  if (cmp_translog_addr(rec->lsn, share->state.is_of_horizon) >= 0)
  {
    tprint(tracef, "   state has LSN " LSN_FMT " older than record, updating"
-           " rows' count\n", LSN_IN_PARTS(share->state.is_of_horizon));
+           " row count\n", LSN_IN_PARTS(share->state.is_of_horizon));
    share->state.state.records++;
    if (share->calc_checksum)
    {
@ -1925,7 +1936,7 @@ prototype_redo_exec_hook(UNDO_ROW_INSERT)
    info->s->state.changed|= (STATE_CHANGED | STATE_NOT_ANALYZED |
                              STATE_NOT_ZEROFILLED | STATE_NOT_MOVABLE);
  }
-  tprint(tracef, "   rows' count %lu\n", (ulong)info->s->state.state.records);
+  tprint(tracef, "   row count: %lu\n", (ulong)info->s->state.state.records);
  /* Unpin all pages, stamp them with UNDO's LSN */
  _ma_unpin_all_pages(info, rec->lsn);
  return 0;
@ -1963,7 +1974,7 @@ prototype_redo_exec_hook(UNDO_ROW_DELETE)
                            STATE_NOT_OPTIMIZED_ROWS | STATE_NOT_ZEROFILLED |
                            STATE_NOT_MOVABLE);
  }
-  tprint(tracef, "   rows' count %lu\n", (ulong)share->state.state.records);
+  tprint(tracef, "   row count: %lu\n", (ulong)share->state.state.records);
  _ma_unpin_all_pages(info, rec->lsn);
  return 0;
 }
@ -2169,7 +2180,7 @@ prototype_redo_exec_hook(CLR_END)
  if (info == NULL)
    DBUG_RETURN(0);
  share= info->s;
-  tprint(tracef, "   CLR_END was about %s, undo_lsn now LSN " LSN_FMT "\n",
+  tprint(tracef, "   CLR_END was about %s, undo_lsn " LSN_FMT "\n",
         log_desc->name, LSN_IN_PARTS(previous_undo_lsn));

  enlarge_buffer(rec);
@ -2226,7 +2237,7 @@ prototype_redo_exec_hook(CLR_END)
                            STATE_NOT_ZEROFILLED | STATE_NOT_MOVABLE);
  }
  if (row_entry)
-    tprint(tracef, "   rows' count %lu\n", (ulong)share->state.state.records);
+    tprint(tracef, "   row count: %lu\n", (ulong)share->state.state.records);
  _ma_unpin_all_pages(info, rec->lsn);
  DBUG_RETURN(0);
 }
@ -2254,7 +2265,7 @@ prototype_redo_exec_hook(DEBUG_INFO)
  data= (char*) log_record_buffer.str + 1;
  switch (debug_info) {
  case LOGREC_DEBUG_INFO_QUERY:
-    tprint(tracef, "Query: %.*b\n", (int) rec->record_length - 1, data);
+    tprint(tracef, "Query: %.*s\n", (int) rec->record_length - 1, data);
    break;
  default:
    DBUG_ASSERT(0);
@ -2327,7 +2338,7 @@ prototype_undo_exec_hook(UNDO_ROW_INSERT)
                                   FILEID_STORE_SIZE);
  info->trn= 0;
  /* trn->undo_lsn is updated in an inwrite_hook when writing the CLR_END */
-  tprint(tracef, "   rows' count %lu\n", (ulong)info->s->state.state.records);
+  tprint(tracef, "   row count: %lu\n", (ulong)info->s->state.state.records);
  tprint(tracef, "   undo_lsn now LSN " LSN_FMT "\n",
         LSN_IN_PARTS(trn->undo_lsn));
  return error;
@ -2367,7 +2378,7 @@ prototype_undo_exec_hook(UNDO_ROW_DELETE)
                                   rec->record_length -
                                   (LSN_STORE_SIZE + FILEID_STORE_SIZE));
  info->trn= 0;
-  tprint(tracef, "   rows' count %lu\n   undo_lsn now LSN " LSN_FMT "\n",
+  tprint(tracef, "   row count: %lu\n   undo_lsn now LSN " LSN_FMT "\n",
         (ulong)share->state.state.records, LSN_IN_PARTS(trn->undo_lsn));
  return error;
 }
@ -2695,8 +2706,8 @@ static int run_redo_phase(LSN lsn, LSN lsn_end, enum maria_apply_log_way apply)
          if (lsn_end != LSN_IMPOSSIBLE && rec2.lsn >= lsn_end)
          {
            tprint(tracef,
-                   "lsn_end reached at " LSN_FMT ". "
-                   "Skipping rest of redo entries",
+                   "lsn_redo_end reached at " LSN_FMT ". "
+                   "Skipping rest of redo entries\n",
                   LSN_IN_PARTS(rec2.lsn));
            translog_destroy_scanner(&scanner);
            translog_free_record_header(&rec);
@ -2781,7 +2792,7 @@ static int run_redo_phase(LSN lsn, LSN lsn_end, enum maria_apply_log_way apply)
      switch (len)
      {
      case RECHEADER_READ_EOF:
-        tprint(tracef, "EOF on the log\n");
+        tprint(tracef, "*** End of log ***\n");
        break;
      case RECHEADER_READ_ERROR:
        tprint(tracef, "Error reading log\n");
@ -2930,7 +2941,7 @@ static uint end_of_redo_phase(my_bool prepare_for_undo_phase)
 }


-static int run_undo_phase(uint uncommitted)
+static int run_undo_phase(LSN end_undo_lsn, uint uncommitted)
 {
  LSN last_undo __attribute__((unused));
  DBUG_ENTER("run_undo_phase");
@ -2956,7 +2967,20 @@ static int run_undo_phase(uint uncommitted)
        fflush(stderr);
      }
      if ((uncommitted--) == 0)
+      {
+        if (aria_undo_aborted <= 0)
+        {
+          aria_undo_aborted= 0;
+          break;
+        }
+      }
+      if (aria_undo_aborted)
+      {
+        tprint(tracef,
+               "lsn_undo_end found. Skipping rest of undo entries\n");
        break;
+      }
+
      trn= trnman_get_any_trn();
      DBUG_ASSERT(trn != NULL);
      llstr(trn->trid, llbuf);
@ -2984,6 +3008,12 @@ static int run_undo_phase(uint uncommitted)
          DBUG_RETURN(1);
        }
        translog_free_record_header(&rec);
+
+        if (last_undo == end_undo_lsn)
+        {
+          aria_undo_aborted= trn->undo_lsn ? 1 : -1;
+          break;
+        }
      }

      /* Force a crash to test recovery of recovery */
@ -2992,6 +3022,7 @@ static int run_undo_phase(uint uncommitted)
        DBUG_ASSERT(--maria_recovery_force_crash_counter > 0);
      }

+      trn->undo_lsn= 0;            /* Avoid abort in trnman_rollbac_trn */
      if (trnman_rollback_trn(trn))
        DBUG_RETURN(1);
      /* We could want to span a few threads (4?) instead of 1 */
@ -3222,7 +3253,10 @@ static MARIA_HA *get_MARIA_HA_from_UNDO_record(const
  }
  DBUG_ASSERT(share->last_version != 0);
  _ma_writeinfo(info, WRITEINFO_UPDATE_KEYFILE); /* to flush state on close */
-  tprint(tracef, ", applying record\n");
+  if (in_redo_phase)
+    tprint(tracef, ", remembering undo\n");
+  else
+    tprint(tracef, ", applying record\n");
  return info;
 }

--- a/storage/maria/ma_recovery.h
+++ b/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;
--- a/storage/maria/maria_def.h
+++ b/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
--- a/storage/maria/maria_read_log.c
+++ b/storage/maria/maria_read_log.c
@ -1,5 +1,6 @@
 /* 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;
 }
-