WL#3072 - Maria recovery

When applying a REDO_INDEX_[FREE|NEW]_PAGE, update state members only
if record is newer than current state.
Print a warning when an old Maria is zero-ing out portions of the
control file which it does not know.


storage/maria/ma_control_file.c:
  Print a warning when an old Maria is zero-ing out portions of the
  control file which it does not know; it may help us when helping
  troubleshoot customers' and users' problems.
storage/maria/ma_key_recover.c:
  When applying a REDO_INDEX_[FREE|NEW]_PAGE, we should update the state
  members (key_root, key_del) only if this REDO is newer than the current
  state. We cannot just update it unconditionally, because that may
  change it to an old state, which may not be later corrected if all later
  REDOs are entirely skipped due to their page not being in the dirty pages
  list of the checkpoint record.

storage/maria/ma_control_file.c

@@ -20,6 +20,7 @@
 
 #ifndef EXTRACT_DEFINITIONS
 #include "maria_def.h"
+#include "ma_checkpoint.h"
 #endif
 
 /*
@@ -487,8 +488,14 @@ int ma_control_file_write_and_force(LSN checkpoint_lsn, uint32 logno,
       we cannot maintain, so that any future version notices we didn't
       maintain its extra data.
     */
-    bzero(buffer + CF_CHANGEABLE_TOTAL_SIZE,
-          cf_changeable_size - CF_CHANGEABLE_TOTAL_SIZE);
+    uint zeroed= cf_changeable_size - CF_CHANGEABLE_TOTAL_SIZE;
+    char msg[150];
+    bzero(buffer + CF_CHANGEABLE_TOTAL_SIZE, zeroed);
+    my_snprintf(msg, sizeof(msg),
+                "Control file must be from a newer version; zero-ing out %u"
+                " unknown bytes in control file at offset %u", zeroed,
+                cf_changeable_size + cf_create_time_size);
+    ma_message_no_user(ME_JUST_WARNING, msg);
   }
   else
   {
@@ -512,6 +519,7 @@ int ma_control_file_write_and_force(LSN checkpoint_lsn, uint32 logno,
   last_logno= logno;
   max_trid_in_control_file= trid;
 
+  cf_changeable_size= CF_CHANGEABLE_TOTAL_SIZE; /* no more warning */
   DBUG_RETURN(0);
 }
 

storage/maria/ma_key_recover.c

@@ -611,20 +611,20 @@ uint _ma_apply_redo_index_new_page(MARIA_HA *info, LSN lsn,
   header+= PAGE_STORE_SIZE * 2 + KEY_NR_STORE_SIZE + 1;
   length-= PAGE_STORE_SIZE * 2 + KEY_NR_STORE_SIZE + 1;
 
-  /* free_page is 0 if we shouldn't set key_del */
-  if (free_page)
+  file_size= (my_off_t) (root_page + 1) * share->block_size;
+  if (cmp_translog_addr(lsn, share->state.is_of_horizon) >= 0)
   {
-    if (free_page != IMPOSSIBLE_PAGE_NO)
-      share->state.key_del= (my_off_t) free_page * share->block_size;
-    else
-      share->state.key_del= HA_OFFSET_ERROR;
+    /* free_page is 0 if we shouldn't set key_del */
+    if (free_page)
+    {
+      if (free_page != IMPOSSIBLE_PAGE_NO)
+        share->state.key_del= (my_off_t) free_page * share->block_size;
+      else
+        share->state.key_del= HA_OFFSET_ERROR;
+    }
+    if (page_type_flag)     /* root page */
+      share->state.key_root[key_nr]= file_size - share->block_size;
   }
-  file_size= (my_off_t) (root_page + 1) * share->block_size;
-
-  /* If root page */
-  if (page_type_flag &&
-      cmp_translog_addr(lsn, share->state.is_of_horizon) >= 0)
-    share->state.key_root[key_nr]= file_size - share->block_size;
 
   if (file_size > info->state->key_file_length)
   {
@@ -723,7 +723,9 @@ uint _ma_apply_redo_index_free_page(MARIA_HA *info,
                           STATE_NOT_SORTED_PAGES | STATE_NOT_ZEROFILLED |
                           STATE_NOT_MOVABLE);
 
-  share->state.key_del= (my_off_t) page * share->block_size;
+  if (cmp_translog_addr(lsn, share->state.is_of_horizon) >= 0)
+    share->state.key_del= (my_off_t) page * share->block_size;
+
   old_link=  ((free_page != IMPOSSIBLE_PAGE_NO) ?
               (my_off_t) free_page * share->block_size :
               HA_OFFSET_ERROR);