Disable logging of index pages during repair

Fixed failure in unittest/ma_test_loghandler_pagecache-t
Initialize pagecache callbacks explictily, not with pagecache_init().
This is to make things more readable and for the future to make more choices with callbacks


storage/maria/ha_maria.cc:
  Disable logging of index pages during repair
storage/maria/ma_bitmap.c:
  Initialize callbacks explictily, not with pagecache_init(), to make things more readable and for future to have more choices with callbacks
  Use new interface to flush logs from pagecache
storage/maria/ma_check.c:
  Fixed test for wrong keyblocks
  Use default functions to setup callbacks for pagecache
storage/maria/ma_loghandler.c:
  Use dummy functions for log flush callback (NULL doesn't work anymore)
storage/maria/ma_open.c:
  Initialize callbacks explictily, not with pagecache_init(), to make things more readable and for future to have more choices with callbacks
  Prefix external functions with _ma_
storage/maria/ma_pagecache.c:
  Use new simpler interface to flush logs if needed
storage/maria/ma_pagecache.h:
  Changed interface to a faster, simpler one to flush logs.
  Now we have a function that takes care of flushing logs, instead of a function to get lsn address
storage/maria/ma_pagecrc.c:
  Add functions for flushing logs
storage/maria/ma_recovery.c:
  Rename functions
storage/maria/maria_chk.c:
  Use default functions to setup callbacks for pagecache
storage/maria/maria_def.h:
  Prefixd global functions with _ma_
storage/maria/unittest/ma_pagecache_consist.c:
  Use dummy functions for log flush callback (NULL doesn't work anymore)
storage/maria/unittest/ma_pagecache_single.c:
  Use dummy functions for log flush callback (NULL doesn't work anymore)
storage/maria/unittest/ma_test_loghandler_pagecache-t.c:
  Use maria_flush_log_for_page to flush log pages. Fixes failure in unittest

storage/maria/ha_maria.cc

@@ -1294,13 +1294,13 @@ int ha_maria::repair(THD *thd, HA_CHECK &param, bool do_optimize)
     {
       thd_proc_info(thd, "Repair with keycache");
       param.testflag &= ~(T_REP_BY_SORT | T_REP_PARALLEL);
-      /**
-         @todo In REPAIR TABLE EXTENDED this will log
-         REDO_INDEX_NEW_PAGE and UNDO_KEY_INSERT though unneeded.
-         maria_chk -o does not have this problem as it disables
-         transactionality.
+      /*
+        Disable logging of index changes as the repair redo call will
+        make it for us
       */
+      _ma_tmp_disable_logging_for_table(file, 0);
       error= maria_repair(&param, file, fixed_name, param.testflag & T_QUICK);
+      _ma_reenable_logging_for_table(file);
       /**
          @todo RECOVERY BUG we do things with the index file
          (maria_sort_index() after the above which already has logged the

storage/maria/ma_bitmap.c

@@ -133,7 +133,6 @@
 static my_bool _ma_read_bitmap_page(MARIA_SHARE *share,
                                     MARIA_FILE_BITMAP *bitmap,
                                     ulonglong page);
-static TRANSLOG_ADDRESS _ma_bitmap_get_log_address();
 
 /* Write bitmap page to key cache */
 
@@ -2578,39 +2577,53 @@ int _ma_bitmap_create_first(MARIA_SHARE *share)
   @retval TRANSLOG_ADDRESS to flush up to.
 */
 
-TRANSLOG_ADDRESS
-_ma_bitmap_get_log_address(uchar *page __attribute__((unused)),
-                           pgcache_page_no_t page_no __attribute__((unused)),
-                           uchar* data_ptr)
+static my_bool
+flush_log_for_bitmap(uchar *page __attribute__((unused)),
+                     pgcache_page_no_t page_no __attribute__((unused)),
+                     uchar* data_ptr)
 {
 #ifndef DBUG_OFF
   const MARIA_SHARE *share= (MARIA_SHARE*)data_ptr;
 #endif
-  DBUG_ENTER("_ma_bitmap_get_log_address");
+  DBUG_ENTER("flush_log_for_bitmap");
   DBUG_ASSERT(share->page_type == PAGECACHE_LSN_PAGE &&
               share->now_transactional);
   /*
     WAL imposes that UNDOs reach disk before bitmap is flushed. We don't know
     the LSN of the last UNDO about this bitmap page, so we flush whole log.
   */
-  DBUG_RETURN(translog_get_horizon());
+  DBUG_RETURN(translog_flush(translog_get_horizon()));
 }
 
 
+/**
+   @brief Set callbacks for bitmap pages
+
+   @note
+   We don't use pagecache_file_init here, as we want to keep the
+   code readable
+*/
+
 void _ma_bitmap_set_pagecache_callbacks(PAGECACHE_FILE *file,
                                         MARIA_SHARE *share)
 {
+  file->callback_data= (uchar*) share;
+  file->flush_log_callback= maria_flush_log_for_page_none;
+  file->write_fail= maria_page_write_failure;
+
   if (share->temporary)
-    pagecache_file_init(*file, &maria_page_crc_check_none,
-                        &maria_page_filler_set_none,
-                        &maria_page_write_failure,
-                        NULL, share);
+  {
+    file->read_callback=  &maria_page_crc_check_none;
+    file->write_callback= &maria_page_filler_set_none;
+  }
   else
-    pagecache_file_init(*file, &maria_page_crc_check_bitmap,
-                        ((share->options & HA_OPTION_PAGE_CHECKSUM) ?
-                         &maria_page_crc_set_normal :
-                         &maria_page_filler_set_bitmap),
-                        &maria_page_write_failure,
-                        share->now_transactional ?
-                        &_ma_bitmap_get_log_address : NULL, share);
+  {
+    file->read_callback=  &maria_page_crc_check_bitmap;
+    if (share->options & HA_OPTION_PAGE_CHECKSUM)
+      file->write_callback= &maria_page_crc_set_normal;
+    else
+      file->write_callback= &maria_page_filler_set_bitmap;
+    if (share->now_transactional)
+      file->flush_log_callback= flush_log_for_bitmap;
+  }
 }

storage/maria/ma_check.c

@@ -2914,7 +2914,9 @@ static my_bool maria_zerofill_index(HA_CHECK *param, MARIA_HA *info,
     if (transactional)
       bzero(buff, LSN_SIZE);
     length= _ma_get_page_used(share, buff);
-    if (length < block_size)
+    /* Skip mailformed blocks */
+    DBUG_ASSERT(length + share->keypage_header <= block_size);
+    if (length + share->keypage_header < block_size)
       bzero(buff + share->keypage_header + length, block_size - length -
             share->keypage_header);
     pagecache_unlock_by_link(share->pagecache, page_link.link,
@@ -3005,6 +3007,7 @@ static my_bool maria_zerofill_data(HA_CHECK *param, MARIA_HA *info,
         /* Zerofille the not used part */
         offset= uint2korr(dir) + uint2korr(dir+2);
         dir_start= (uint) (dir - buff);
+        DBUG_ASSERT(dir_start >= offset);
         if (dir_start > offset)
           bzero(buff + offset, dir_start - offset);
       }
@@ -5864,17 +5867,11 @@ static my_bool create_new_data_handle(MARIA_SORT_PARAM *param, File new_file)
                                         HA_OPEN_COPY | HA_OPEN_FOR_REPAIR)))
     DBUG_RETURN(1);
 
+  info->s->now_transactional= 0;
   new_info= sort_info->new_info;
-  pagecache_file_init(new_info->s->bitmap.file, &maria_page_crc_check_bitmap,
-                      (new_info->s->options & HA_OPTION_PAGE_CHECKSUM ?
-                       &maria_page_crc_set_normal :
-                       &maria_page_filler_set_bitmap),
-                      &maria_page_write_failure, NULL, new_info->s);
-  pagecache_file_init(new_info->dfile, &maria_page_crc_check_data,
-                      (new_info->s->options & HA_OPTION_PAGE_CHECKSUM ?
-                       &maria_page_crc_set_normal :
-                       &maria_page_filler_set_normal),
-                      &maria_page_write_failure, NULL, new_info->s);
+  _ma_bitmap_set_pagecache_callbacks(&new_info->s->bitmap.file,
+                                     new_info->s);
+  _ma_set_data_pagecache_callbacks(&new_info->dfile, new_info->s);
   change_data_file_descriptor(new_info, new_file);
   maria_lock_database(new_info, F_EXTRA_LCK);
   if ((sort_info->param->testflag & T_UNPACK) &&

storage/maria/ma_loghandler.c

@@ -1354,7 +1354,8 @@ static void translog_file_init(TRANSLOG_FILE *file, uint32 number,
 {
   pagecache_file_init(file->handler, &translog_page_validator,
                       &translog_dummy_callback,
-                      &translog_dummy_write_failure, NULL, file);
+                      &translog_dummy_write_failure,
+                      maria_flush_log_for_page_none, file);
   file->number= number;
   file->was_recovered= 0;
   file->is_sync= is_sync;

storage/maria/ma_open.c

@@ -152,7 +152,7 @@ static MARIA_HA *maria_clone_internal(MARIA_SHARE *share, int mode,
   if (share->options & HA_OPTION_TMP_TABLE)
     info.lock_type= F_WRLCK;
 
-  set_data_pagecache_callbacks(&info.dfile, share);
+  _ma_set_data_pagecache_callbacks(&info.dfile, share);
   bitmap_init(&info.changed_fields, changed_fields_bitmap,
               share->base.fields, 0);
   if ((*share->init)(&info))
@@ -722,7 +722,7 @@ MARIA_HA *maria_open(const char *name, int mode, uint open_flags)
     }
 
     share->kfile.file= kfile;
-    set_index_pagecache_callbacks(&share->kfile, share);
+    _ma_set_index_pagecache_callbacks(&share->kfile, share);
     share->this_process=(ulong) getpid();
     share->last_process= share->state.process;
     share->base.key_parts=key_parts;
@@ -1531,47 +1531,69 @@ uchar *_ma_column_nr_read(uchar *ptr, uint16 *offsets, uint columns)
   return ptr;
 }
 
+/**
+   @brief Set callbacks for data pages
+
+   @note
+   We don't use pagecache_file_init here, as we want to keep the
+   code readable
+*/
 
-void set_data_pagecache_callbacks(PAGECACHE_FILE *file, MARIA_SHARE *share)
+void _ma_set_data_pagecache_callbacks(PAGECACHE_FILE *file,
+                                      MARIA_SHARE *share)
 {
-  /*
-    Note that non-BLOCK_RECORD formats don't use the pagecache for their data
-    files, so it does not matter that maria_page* calls are passed below for
-    them. On the other hand, index file can always have page CRCs, for all
-    data formats.
-  */
+  file->callback_data= (uchar*) share;
+  file->flush_log_callback= &maria_flush_log_for_page_none; /* Do nothing */
+
   if (share->temporary)
-    pagecache_file_init(*file, &maria_page_crc_check_none,
-                        &maria_page_filler_set_none,
-                        &maria_page_write_failure,
-                        NULL, share);
+  {
+    file->read_callback=  &maria_page_crc_check_none;
+    file->write_callback= &maria_page_filler_set_none;
+  }
   else
-    pagecache_file_init(*file, &maria_page_crc_check_data,
-                        ((share->options & HA_OPTION_PAGE_CHECKSUM) ?
-                         &maria_page_crc_set_normal :
-                         &maria_page_filler_set_normal),
-                        &maria_page_write_failure,
-                        share->now_transactional ?
-                        &maria_page_get_lsn : NULL, share);
+  {
+    file->read_callback=  &maria_page_crc_check_data;
+    if (share->options & HA_OPTION_PAGE_CHECKSUM)
+      file->write_callback= &maria_page_crc_set_normal;
+    else
+      file->write_callback= &maria_page_filler_set_normal;
+    if (share->now_transactional)
+      file->flush_log_callback= maria_flush_log_for_page;
+  }
 }
 
 
-void set_index_pagecache_callbacks(PAGECACHE_FILE *file, MARIA_SHARE *share)
+/**
+   @brief Set callbacks for index pages
+
+   @note
+   We don't use pagecache_file_init here, as we want to keep the
+   code readable
+*/
+
+void _ma_set_index_pagecache_callbacks(PAGECACHE_FILE *file,
+                                       MARIA_SHARE *share)
 {
+  file->callback_data= (uchar*) share;
+  file->flush_log_callback= &maria_flush_log_for_page_none; /* Do nothing */
+  file->write_fail= maria_page_write_failure;
+
   if (share->temporary)
-    pagecache_file_init(*file, &maria_page_crc_check_none,
-                        &maria_page_filler_set_none,
-                        &maria_page_write_failure,
-                        NULL, share);
+  {
+    file->read_callback=  &maria_page_crc_check_none;
+    file->write_callback= &maria_page_filler_set_none;
+  }
   else
-    pagecache_file_init(*file, &maria_page_crc_check_index,
-                        ((share->options & HA_OPTION_PAGE_CHECKSUM) ?
-                         &maria_page_crc_set_index :
-                         &maria_page_filler_set_normal),
-                        &maria_page_write_failure,
-                        share->now_transactional ?
-                        &maria_page_get_lsn : NULL,
-                        share);
+  {
+    file->read_callback=  &maria_page_crc_check_index;
+    if (share->options & HA_OPTION_PAGE_CHECKSUM)
+      file->write_callback= &maria_page_crc_set_index;
+    else
+      file->write_callback= &maria_page_filler_set_normal;
+
+    if (share->now_transactional)
+      file->flush_log_callback= maria_flush_log_for_page;
+  }
 }
 
 

storage/maria/ma_pagecache.c

@@ -597,23 +597,12 @@ static uint pagecache_fwrite(PAGECACHE *pagecache,
                              enum pagecache_page_type type,
                              myf flags)
 {
-  TRANSLOG_ADDRESS (*addr_callback)
-    (uchar *page, pgcache_page_no_t offset, uchar *data)=
-    filedesc->get_log_address_callback;
   DBUG_ENTER("pagecache_fwrite");
   DBUG_ASSERT(type != PAGECACHE_READ_UNKNOWN_PAGE);
-  if (addr_callback != NULL)
-  {
-    TRANSLOG_ADDRESS addr=
-      (*addr_callback)(buffer, pageno, filedesc->callback_data);
-    DBUG_PRINT("info", ("Log handler call"));
-    DBUG_ASSERT(LSN_VALID(addr));
-    if (translog_flush(addr))
-    {
-      (*filedesc->write_fail)(filedesc->callback_data);
-      DBUG_RETURN(1);
-    }
-  }
+
+  /* Todo: Integrate this with write_callback so we have only one callback */
+  if ((*filedesc->flush_log_callback)(buffer, pageno, filedesc->callback_data))
+    DBUG_RETURN(1);
   DBUG_PRINT("info", ("write_callback: 0x%lx  data: 0x%lx",
                       (ulong) filedesc->write_callback,
                       (ulong) filedesc->callback_data));
@@ -622,7 +611,6 @@ static uint pagecache_fwrite(PAGECACHE *pagecache,
     DBUG_PRINT("error", ("write callback problem"));
     DBUG_RETURN(1);
   }
-
   if (my_pwrite(filedesc->file, buffer, pagecache->block_size,
                 ((my_off_t) pageno << pagecache->shift), flags))
   {

storage/maria/ma_pagecache.h

@@ -88,9 +88,9 @@ typedef struct st_pagecache_file
   my_bool (*write_callback)(uchar *page, pgcache_page_no_t offset,
                             uchar *data);
   void (*write_fail)(uchar *data);
-  /** Can be NULL */
-  TRANSLOG_ADDRESS (*get_log_address_callback)
-    (uchar *page, pgcache_page_no_t offset, uchar *data);
+  /** Cannot be NULL */
+  my_bool (*flush_log_callback)(uchar *page, pgcache_page_no_t offset,
+                                uchar *data);
   uchar *callback_data;
 } PAGECACHE_FILE;
 
@@ -267,7 +267,7 @@ extern void pagecache_unpin_by_link(PAGECACHE *pagecache,
   do{ \
     (F).read_callback= (RC); (F).write_callback= (WC); \
     (F).write_fail= (WF); \
-    (F).get_log_address_callback= (GLC); (F).callback_data= (uchar*)(D); \
+    (F).flush_log_callback= (GLC); (F).callback_data= (uchar*)(D); \
   } while(0)
 
 #define flush_pagecache_blocks(A,B,C)                   \

storage/maria/ma_pagecrc.c

@@ -291,12 +291,52 @@ my_bool maria_page_filler_set_none(uchar *page __attribute__((unused)),
   return 0;
 }
 
+
 /**
   @brief Write failure callback (mark table as corrupted)
 
   @param data_ptr        Write callback data pointer (pointer to MARIA_SHARE)
 */
-void maria_page_write_failure (uchar* data_ptr)
+
+void maria_page_write_failure(uchar* data_ptr)
 {
   maria_mark_crashed_share((MARIA_SHARE *)data_ptr);
 }
+
+
+/**
+  @brief Maria flush log log if needed
+
+  @param page            The page data to set
+  @param page_no         The page number (<offset>/<page length>)
+  @param data_ptr        Write callback data pointer (pointer to MARIA_SHARE)
+
+  @retval 0  OK
+  @retval 1  error
+*/
+
+my_bool maria_flush_log_for_page(uchar *page,
+                                 pgcache_page_no_t page_no
+                                 __attribute__((unused)),
+                                 uchar *data_ptr)
+{
+  LSN lsn;
+  const MARIA_SHARE *share= (MARIA_SHARE*) data_ptr;
+  DBUG_ENTER("maria_flush_log_for_page");
+  /* share is 0 here only in unittest */
+  DBUG_ASSERT(!share || (share->page_type == PAGECACHE_LSN_PAGE &&
+                         share->now_transactional));
+  lsn= lsn_korr(page);
+  if (translog_flush(lsn))
+    DBUG_RETURN(1);
+  DBUG_RETURN(0);
+}
+
+
+my_bool maria_flush_log_for_page_none(uchar *page __attribute__((unused)),
+                                      pgcache_page_no_t page_no
+                                      __attribute__((unused)),
+                                      uchar *data_ptr __attribute__((unused)))
+{
+  return 0;
+}

storage/maria/ma_recovery.c

@@ -3010,8 +3010,8 @@ void _ma_tmp_disable_logging_for_table(MARIA_HA *info,
   info->trn= &dummy_transaction_object;
   share->page_type= PAGECACHE_PLAIN_PAGE;
   /* Functions below will pick up now_transactional and change callbacks */
-  set_data_pagecache_callbacks(&info->dfile, share);
-  set_index_pagecache_callbacks(&share->kfile, share);
+  _ma_set_data_pagecache_callbacks(&info->dfile, share);
+  _ma_set_index_pagecache_callbacks(&share->kfile, share);
   _ma_bitmap_set_pagecache_callbacks(&share->bitmap.file, share);
 }
 
@@ -3035,8 +3035,8 @@ void _ma_reenable_logging_for_table(MARIA_HA *info)
     share->page_type= PAGECACHE_LSN_PAGE;
     info->trn= NULL; /* safety */
   }
-  set_data_pagecache_callbacks(&info->dfile, share);
-  set_index_pagecache_callbacks(&share->kfile, share);
+  _ma_set_data_pagecache_callbacks(&info->dfile, share);
+  _ma_set_index_pagecache_callbacks(&share->kfile, share);
   _ma_bitmap_set_pagecache_callbacks(&share->bitmap.file, share);
 }
 

storage/maria/maria_chk.c

@@ -1690,12 +1690,9 @@ static int maria_sort_records(HA_CHECK *param,
 
   VOID(my_close(info->dfile.file, MYF(MY_WME)));
   param->out_flag|=O_NEW_DATA;			/* Data in new file */
-  info->dfile.file= new_file;				/* Use new datafile */
-  pagecache_file_init(info->dfile, &maria_page_crc_check_data,
-                      (share->options & HA_OPTION_PAGE_CHECKSUM ?
-                       &maria_page_crc_set_normal :
-                       &maria_page_filler_set_normal),
-                      &maria_page_write_failure, NULL, share);
+  info->dfile.file= new_file;                   /* Use new datafile */
+  _ma_set_data_pagecache_callbacks(&info->dfile, info->s);
+
   info->state->del=0;
   info->state->empty=0;
   share->state.dellink= HA_OFFSET_ERROR;

storage/maria/maria_def.h

@@ -1068,10 +1068,10 @@ int _ma_update_create_rename_lsn(MARIA_SHARE *share,
                                  LSN lsn, my_bool do_sync);
 int _ma_update_create_rename_lsn_sub(MARIA_SHARE *share,
                                      LSN lsn, my_bool do_sync);
-void set_data_pagecache_callbacks(PAGECACHE_FILE *file,
-                                  MARIA_SHARE *share);
-void set_index_pagecache_callbacks(PAGECACHE_FILE *file,
-                                   MARIA_SHARE *share);
+void _ma_set_data_pagecache_callbacks(PAGECACHE_FILE *file,
+                                      MARIA_SHARE *share);
+void _ma_set_index_pagecache_callbacks(PAGECACHE_FILE *file,
+                                       MARIA_SHARE *share);
 void _ma_tmp_disable_logging_for_table(MARIA_HA *info,
                                        my_bool log_incomplete);
 void _ma_reenable_logging_for_table(MARIA_HA *info);
@@ -1106,5 +1106,10 @@ extern my_bool maria_page_filler_set_none(uchar *page,
                                           pgcache_page_no_t page_no,
                                           uchar *data_ptr);
 extern void maria_page_write_failure(uchar* data_ptr);
-
+extern my_bool maria_flush_log_for_page(uchar *page,
+                                        pgcache_page_no_t page_no,
+                                        uchar *data_ptr);
+extern my_bool maria_flush_log_for_page_none(uchar *page,
+                                             pgcache_page_no_t page_no,
+                                             uchar *data_ptr);
 extern PAGECACHE *maria_log_pagecache;

storage/maria/unittest/ma_pagecache_consist.c

@@ -346,7 +346,7 @@ int main(int argc __attribute__((unused)),
     exit(1);
   }
   pagecache_file_init(file1, &dummy_callback, &dummy_callback,
-                      &dummy_fail_callback, NULL, NULL);
+                      &dummy_fail_callback, &dummy_callback, NULL);
   DBUG_PRINT("info", ("file1: %d", file1.file));
   if (chmod(file1_name, S_IRWXU | S_IRWXG | S_IRWXO) != 0)
   {

storage/maria/unittest/ma_pagecache_single.c

@@ -532,7 +532,7 @@ int main(int argc __attribute__((unused)),
     exit(1);
   }
   pagecache_file_init(file1, &dummy_callback, &dummy_callback,
-                      &dummy_fail_callback, NULL, NULL);
+                      &dummy_fail_callback, &dummy_callback, NULL);
   my_close(tmp_file, MYF(0));
   my_delete(file2_name, MYF(0));
 

storage/maria/unittest/ma_test_loghandler_pagecache-t.c

@@ -138,7 +138,7 @@ int main(int argc __attribute__((unused)), char *argv[])
     exit(1);
   }
   pagecache_file_init(file1, &dummy_callback, &dummy_callback,
-                      &dummy_fail_callback, NULL, NULL);
+                      &dummy_fail_callback, maria_flush_log_for_page, NULL);
   if (chmod(file1_name, S_IRWXU | S_IRWXG | S_IRWXO) != 0)
   {
     fprintf(stderr, "Got error during file1 chmod() (errno: %d)\n",