Fixed bug in Aria with aria_log files that are exactly 8K

In the case one has an old Aria log file that ands with a Aria checkpoint
and the server restarts after next recovery, just after created a
new Aria log file (of 8K), the Aria recovery code would abort.
If one would try to delete all Aria log files after this (but not the
aria_control_file), the server would crash during recovery.

The problem was that translog_get_last_page_addr() would regard a log file
of exactly 8K as illegal and the rest of the code could not handle this
case.

Another issue was that if there was a crash directly after the log file
head was written to the next page, the code in translog_get_next_chunk()
would crash.

This patch fixes most of the issues, but not all. For Sanja to look at!

Things fixed:
- Added code to ignore 8K log files.
- Removed ASSERT in translog_get_next_chunk() that checks if page only
  contains the log page header.

storage/maria/aria_read_log.c

@@ -139,6 +139,12 @@ int main(int argc, char **argv)
   if (opt_display_only)
     printf("You are using --display-only, NOTHING will be written to disk\n");
 
+  if (translog_get_horizon() == LSN_IMPOSSIBLE)
+  {
+     fprintf(stdout, "The transaction log is empty\n");
+     goto end;
+  }
+
   lsn= translog_first_lsn_in_log();
   if (lsn == LSN_ERROR)
   {
@@ -148,6 +154,7 @@ int main(int argc, char **argv)
   if (lsn == LSN_IMPOSSIBLE)
   {
     fprintf(stdout, "The transaction log is empty\n");
+    goto end;
   }
   if (opt_start_from_checkpoint && !opt_start_from_lsn &&
       last_checkpoint_lsn != LSN_IMPOSSIBLE)
@@ -300,7 +307,7 @@ static struct my_option my_long_options[] =
 
 static void print_version(void)
 {
-  printf("%s Ver 1.5 for %s on %s\n",
+  printf("%s Ver 1.6 for %s on %s\n",
               my_progname_short, SYSTEM_TYPE, MACHINE_TYPE);
 }
 
@@ -308,7 +315,7 @@ static void print_version(void)
 static void usage(void)
 {
   print_version();
-  puts("Copyright (C) 2007 MySQL AB, 2009-2011 Monty Program Ab, 2020 MariaDB Corporation");
+  puts("Copyright (C) 2007 MySQL AB, 2009-2011 Monty Program Ab, 2022 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");
 

storage/maria/ma_loghandler.c

@@ -478,7 +478,7 @@ static my_bool translog_page_validator(int res, PAGECACHE_IO_HOOK_ARGS *args);
 static my_bool translog_get_next_chunk(TRANSLOG_SCANNER_DATA *scanner);
 static uint32 translog_first_file(TRANSLOG_ADDRESS horizon, int is_protected);
 LSN translog_next_LSN(TRANSLOG_ADDRESS addr, TRANSLOG_ADDRESS horizon);
-
+static void translog_free_link(PAGECACHE_BLOCK_LINK *direct_link);
 
 /*
   Initialize log_record_type_descriptors
@@ -3116,8 +3116,11 @@ restart:
                                          PAGECACHE_PLAIN_PAGE,
                                          PAGECACHE_LOCK_LEFT_UNLOCKED,
                                          NULL)))
+            {
+              translog_unlock();
               DBUG_RETURN(NULL);
             }
+          }
           else
             skipped_data= 0;  /* Read after skipped in buffer data */
           /*
@@ -3217,6 +3220,11 @@ restart:
                           PAGECACHE_LOCK_READ :
                           PAGECACHE_LOCK_LEFT_UNLOCKED),
                          direct_link);
+  if (!buffer && direct_link)
+  {
+    translog_free_link(*direct_link);
+    *direct_link= 0;
+  }
   DBUG_PRINT("info", ("Direct link is assigned to : %p * %p",
                       direct_link,
                       (direct_link ? *direct_link : NULL)));
@@ -3786,17 +3794,27 @@ my_bool translog_init_with_table(const char *directory,
     }
     else if (LSN_OFFSET(last_page) == 0)
     {
-      if (LSN_FILE_NO(last_page) == 1)
+      if (LSN_FILE_NO(last_page) == 1 ||
+          !translog_is_file(LSN_FILE_NO(last_page-1)))
       {
         logs_found= 0;                          /* file #1 has no pages */
         DBUG_PRINT("info", ("log found. But is is empty => no log assumed"));
       }
       else
+      {
+        do
         {
           last_page-= LSN_ONE_FILE;
           if (translog_get_last_page_addr(&last_page, &pageok, 0))
             goto err;
         }
+        while (LSN_OFFSET(last_page) == 0 && LSN_FILE_NO(last_page) >= 1);
+        if (LSN_OFFSET(last_page) == 0)
+        {
+          /* All files have a size less than TRANSLOG_PAGE_SIZE */
+          logs_found= 0;
+        }
+      }
     }
     if (logs_found)
     {
@@ -3893,8 +3911,9 @@ my_bool translog_init_with_table(const char *directory,
         old_log_was_recovered= 1;
         /* This file is not written till the end so it should be last */
         last_page= current_file_last_page;
-        /* TODO: issue warning */
       }
+      if (LSN_OFFSET(current_file_last_page) >= TRANSLOG_PAGE_SIZE)
+      {
         do
         {
           TRANSLOG_VALIDATOR_DATA data;
@@ -3923,6 +3942,7 @@ my_bool translog_init_with_table(const char *directory,
           last_valid_page= current_page;
           current_page+= TRANSLOG_PAGE_SIZE; /* increase offset */
         } while (current_page <= current_file_last_page);
+      }
       current_page+= LSN_ONE_FILE;
       current_page= LSN_REPLACE_OFFSET(current_page, TRANSLOG_PAGE_SIZE);
     } while (LSN_FILE_NO(current_page) <= LSN_FILE_NO(last_page) &&
@@ -4014,7 +4034,7 @@ my_bool translog_init_with_table(const char *directory,
   }
   DBUG_PRINT("info", ("Logs found: %d  was recovered: %d",
                       logs_found, old_log_was_recovered));
-  if (!logs_found)
+  if (!logs_found && !readonly)
   {
     TRANSLOG_FILE *file= (TRANSLOG_FILE*)my_malloc(PSI_INSTRUMENT_ME,
                                            sizeof(TRANSLOG_FILE), MYF(MY_WME));
@@ -4064,6 +4084,10 @@ my_bool translog_init_with_table(const char *directory,
     translog_start_buffer(log_descriptor.buffers, &log_descriptor.bc, 0);
     translog_new_page_header(&log_descriptor.horizon, &log_descriptor.bc);
   }
+  else if (readonly && !logs_found)
+  {
+    log_descriptor.horizon= LSN_IMPOSSIBLE;
+  }
 
   /* all LSNs that are on disk are flushed */
   log_descriptor.log_start= log_descriptor.sent_to_disk=
@@ -4145,6 +4169,8 @@ my_bool translog_init_with_table(const char *directory,
         uint32 file_no= LSN_FILE_NO(page_addr);
         my_bool last_page_ok;
         /* it is beginning of the current file */
+        do
+        {
           if (unlikely(file_no == 1))
           {
             /*
@@ -4160,6 +4186,7 @@ my_bool translog_init_with_table(const char *directory,
           file_no--;
           page_addr= MAKE_LSN(file_no, TRANSLOG_PAGE_SIZE);
           translog_get_last_page_addr(&page_addr, &last_page_ok, 0);
+        } while (LSN_OFFSET(page_addr) == 0);
         /* page should be OK as it is not the last file */
         DBUG_ASSERT(last_page_ok);
       }
@@ -6905,17 +6932,19 @@ translog_get_next_chunk(TRANSLOG_SCANNER_DATA *scanner)
       /* if it is log end it have to be caught before */
       DBUG_ASSERT(LSN_FILE_NO(scanner->horizon) >
                   LSN_FILE_NO(scanner->page_addr));
+      do
+      {
         scanner->page_addr+= LSN_ONE_FILE;
         scanner->page_addr= LSN_REPLACE_OFFSET(scanner->page_addr,
                                                TRANSLOG_PAGE_SIZE);
         if (translog_scanner_set_last_page(scanner))
           DBUG_RETURN(1);
+      } while (!LSN_OFFSET(scanner->last_file_page));
     }
     else
     {
       scanner->page_addr+= TRANSLOG_PAGE_SIZE; /* offset increased */
     }
-
     if (translog_scanner_get_page(scanner))
       DBUG_RETURN(1);
 
@@ -6926,7 +6955,9 @@ translog_get_next_chunk(TRANSLOG_SCANNER_DATA *scanner)
       scanner->page_offset= 0;
       DBUG_RETURN(0);
     }
+#ifdef CHECK_EMPTY_PAGE
     DBUG_ASSERT(scanner->page[scanner->page_offset] != TRANSLOG_FILLER);
+#endif
   }
   DBUG_RETURN(0);
 }

storage/maria/ma_pagecache.c

@@ -5227,7 +5227,7 @@ int flush_pagecache_blocks_with_filter(PAGECACHE *pagecache,
 {
   int res;
   DBUG_ENTER("flush_pagecache_blocks_with_filter");
-  DBUG_PRINT("enter", ("pagecache: %p", pagecache));
+  DBUG_PRINT("enter", ("pagecache: %p  fd: %di", pagecache, file->file));
 
   if (pagecache->disk_blocks <= 0)
     DBUG_RETURN(0);