Write error message if aria_log.??????? files are missing

This can happen if one uses a backup where not all aria_log.* files
are copied or if the last one is too short. In this case the data
files will contain data that is not in the logs and recovery will fail.

Other things:
- Fixed tprint() to not print extra new line to debug trace

storage/maria/ma_blockrec.c

@@ -6394,6 +6394,7 @@ uint _ma_apply_redo_insert_row_head_or_tail(MARIA_HA *info, LSN lsn,
     }
     else if (lsn_korr(buff) >= lsn)           /* Test if already applied */
     {
+      check_skipped_lsn(info, lsn_korr(buff), 1, page);
       /* Fix bitmap, just in case */
       empty_space= uint2korr(buff + EMPTY_SPACE_OFFSET);
       if (!enough_free_entries_on_page(share, buff))
@@ -6561,6 +6562,7 @@ uint _ma_apply_redo_purge_row_head_or_tail(MARIA_HA *info, LSN lsn,
       Note that in case the page is not anymore a head or tail page
       a future redo will fix the bitmap.
     */
+    check_skipped_lsn(info, lsn_korr(buff), 1, page);
     if ((uint) (buff[PAGE_TYPE_OFFSET] & PAGE_TYPE_MASK) == page_type)
     {
       empty_space= uint2korr(buff+EMPTY_SPACE_OFFSET);
@@ -6731,6 +6733,7 @@ uint _ma_apply_redo_free_head_or_tail(MARIA_HA *info, LSN lsn,
   if (lsn_korr(buff) >= lsn)
   {
     /* Already applied */
+    check_skipped_lsn(info, lsn_korr(buff), 1, page);
     pagecache_unlock_by_link(share->pagecache, page_link.link,
                              PAGECACHE_LOCK_WRITE_UNLOCK,
                              PAGECACHE_UNPIN, LSN_IMPOSSIBLE,
@@ -6908,8 +6911,7 @@ uint _ma_apply_redo_insert_row_blobs(MARIA_HA *info,
             if (lsn_korr(buff) >= lsn)
             {
               /* Already applied */
-              DBUG_PRINT("info", ("already applied %llu >= %llu",
-                                  lsn_korr(buff), lsn));
+              check_skipped_lsn(info, lsn_korr(buff), 1, page);
               pagecache_unlock_by_link(share->pagecache, page_link.link,
                                        PAGECACHE_LOCK_WRITE_UNLOCK,
                                        PAGECACHE_UNPIN, LSN_IMPOSSIBLE,

storage/maria/ma_key_recover.c

@@ -777,7 +777,7 @@ uint _ma_apply_redo_index_new_page(MARIA_HA *info, LSN lsn,
     else if (lsn_korr(buff) >= lsn)
     {
       /* Already applied */
-      DBUG_PRINT("info", ("Page is up to date, skipping redo"));
+      check_skipped_lsn(info, lsn_korr(buff), 0, root_page);
       result= 0;
       goto err;
     }
@@ -865,6 +865,7 @@ uint _ma_apply_redo_index_free_page(MARIA_HA *info,
   if (lsn_korr(buff) >= lsn)
   {
     /* Already applied */
+    check_skipped_lsn(info, lsn_korr(buff), 0, page);
     result= 0;
     goto err;
   }
@@ -966,7 +967,7 @@ uint _ma_apply_redo_index(MARIA_HA *info,
   if (lsn_korr(buff) >= lsn)
   {
     /* Already applied */
-    DBUG_PRINT("info", ("Page is up to date, skipping redo"));
+    check_skipped_lsn(info, lsn_korr(buff), 0, page_pos);
     result= 0;
     goto err;
   }

storage/maria/ma_loghandler.c

@@ -19,6 +19,9 @@
 #include "ma_key_recover.h" /* For some in-write hooks */
 #include "ma_checkpoint.h"
 #include "ma_servicethread.h"
+#include "ma_recovery.h"
+#include "ma_loghandler_lsn.h"
+#include "ma_recovery_util.h"
 
 /*
   On Windows, neither my_open() nor mysql_file_sync() work for directories.
@@ -7903,6 +7906,34 @@ static my_bool translog_sync_files(uint32 min, uint32 max,
 }
 
 
+/**
+   check_skipped_lsn
+
+   Check if lsn skipped in redo is ok
+*/
+
+void check_skipped_lsn(MARIA_HA *info, LSN lsn, my_bool index_file,
+                       pgcache_page_no_t page)
+{
+  if (lsn <= log_descriptor.horizon)
+  {
+    DBUG_PRINT("info", ("Page is up to date, skipping redo"));
+  }
+  else
+  {
+    /* Give error, but don't flood the log */
+    if (skipped_lsn_err_count++ < 10 && ! info->s->redo_error_given++)
+    {
+      eprint(tracef, "Table %s has wrong LSN: " LSN_FMT " on page: %llu",
+             (index_file ? info->s->data_file_name.str :
+              info->s->index_file_name.str),
+             LSN_IN_PARTS(lsn), (ulonglong) page);
+      recovery_found_crashed_tables++;
+    }
+  }
+}
+
+
 /*
   @brief Flushes buffers with LSNs in them less or equal address <lsn>
 

storage/maria/ma_loghandler.h

@@ -384,6 +384,8 @@ int translog_soft_sync_start(void);
 void  translog_soft_sync_end(void);
 void translog_sync();
 void translog_set_group_commit_interval(uint32 interval);
+extern void check_skipped_lsn(MARIA_HA *info, LSN lsn, my_bool index_file,
+                              pgcache_page_no_t page);
 
 /*
   all the rest added because of recovery; should we make

storage/maria/ma_recovery.c

@@ -57,8 +57,7 @@ static my_bool trns_created;
 static ulong skipped_undo_phase;
 static ulonglong now; /**< for tracking execution time of phases */
 static void (*save_error_handler_hook)(uint, const char *,myf);
-static uint recovery_warnings; /**< count of warnings */
-static uint recovery_found_crashed_tables;
+static ulong recovery_warnings; /**< count of warnings */
 HASH tables_to_redo;                          /* For maria_read_log */
 ulong maria_recovery_force_crash_counter;
 TrID max_long_trid= 0; /**< max long trid seen by REDO phase */
@@ -291,6 +290,7 @@ int maria_apply_log(LSN from_lsn, LSN end_lsn,
   DBUG_ASSERT(apply == MARIA_LOG_APPLY || !should_run_undo_phase);
   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);

storage/maria/ma_recovery.h

@@ -33,4 +33,6 @@ int maria_apply_log(LSN lsn, LSN lsn_end, enum maria_apply_log_way apply,
 /* Table of tables to recover */
 extern HASH tables_to_redo;
 extern ulong maria_recovery_force_crash_counter;
+extern ulong recovery_found_crashed_tables;
+extern uint skipped_lsn_err_count;
 C_MODE_END

storage/maria/ma_recovery_util.c

@@ -51,6 +51,8 @@ LSN checkpoint_start= LSN_IMPOSSIBLE;
 my_bool procent_printed;
 FILE *tracef; /**< trace file for debugging */
 
+ulong recovery_found_crashed_tables;
+uint skipped_lsn_err_count;
 
 /** @brief Prints to a trace file if it is not NULL */
 void tprint(FILE *trace_file __attribute__ ((unused)),
@@ -59,11 +61,12 @@ void tprint(FILE *trace_file __attribute__ ((unused)),
   va_list args;
 #ifndef DBUG_OFF
   {
-    char buff[1024], *end;
+    char buff[1024];
+    size_t length;
     va_start(args, format);
-    vsnprintf(buff, sizeof(buff)-1, format, args);
-    if (*(end= strend(buff)) == '\n')
-      *end= 0;                                  /* Don't print end \n */
+    length= my_vsnprintf(buff, sizeof(buff)-1, format, args);
+    if (length && buff[length-1] == '\n')
+      buff[length-1]= 0;                      /* Don't print end \n */
     DBUG_PRINT("info", ("%s", buff));
     va_end(args);
   }

storage/maria/maria_def.h

@@ -500,6 +500,7 @@ typedef struct st_maria_share
   my_bool have_versioning;
   my_bool key_del_used;                         /* != 0 if key_del is locked */
   my_bool deleting;                     /* we are going to delete this table */
+  my_bool redo_error_given;             /* Used during recovery */
   THR_LOCK lock;
   void (*lock_restore_status)(void *);
   /**