Support of rec_lsn added to pagecache_write call.

storage/maria/ma_pagecache.c:
  Support of rec_lsn added to pagecache_write call.
  A function for setting rec_lsn made.

storage/maria/ma_bitmap.c

@@ -149,7 +149,8 @@ static inline my_bool write_changed_bitmap(MARIA_SHARE *share,
                           (uchar*) bitmap->map, PAGECACHE_PLAIN_PAGE,
                           PAGECACHE_LOCK_LEFT_UNLOCKED,
                           PAGECACHE_PIN_LEFT_UNPINNED,
-                          PAGECACHE_WRITE_DELAY, 0));
+                          PAGECACHE_WRITE_DELAY, 0,
+                          LSN_IMPOSSIBLE));
 }
 
 /*

storage/maria/ma_blockrec.c

@@ -1451,7 +1451,8 @@ static my_bool write_tail(MARIA_HA *info,
                              PAGECACHE_LOCK_READ,
                              block_is_read ? PAGECACHE_PIN_LEFT_PINNED :
                              PAGECACHE_PIN,
-                             PAGECACHE_WRITE_DELAY, &page_link.link)))
+                             PAGECACHE_WRITE_DELAY, &page_link.link,
+                             LSN_IMPOSSIBLE)))
   {
     page_link.unlock= PAGECACHE_LOCK_READ_UNLOCK;
     if (block_is_read)
@@ -1544,7 +1545,7 @@ static my_bool write_full_pages(MARIA_HA *info,
                         PAGECACHE_LOCK_LEFT_UNLOCKED,
                         PAGECACHE_PIN_LEFT_UNPINNED,
                         PAGECACHE_WRITE_DELAY,
-                        0))
+                        0, LSN_IMPOSSIBLE))
       DBUG_RETURN(1);
     page++;
     block->used= BLOCKUSED_USED;
@@ -2262,7 +2263,8 @@ static my_bool write_block_record(MARIA_HA *info,
                       PAGECACHE_LOCK_READ,
                       head_block_is_read ? PAGECACHE_PIN_LEFT_PINNED :
                       PAGECACHE_PIN,
-                      PAGECACHE_WRITE_DELAY, &page_link.link))
+                      PAGECACHE_WRITE_DELAY, &page_link.link,
+                      LSN_IMPOSSIBLE))
     goto disk_err;
   page_link.unlock= PAGECACHE_LOCK_READ_UNLOCK;
   if (head_block_is_read)
@@ -3041,7 +3043,8 @@ static my_bool delete_head_or_tail(MARIA_HA *info,
                         buff, share->page_type,
                         lock_at_write,
                         PAGECACHE_PIN_LEFT_PINNED,
-                        PAGECACHE_WRITE_DELAY, &page_link.link))
+                        PAGECACHE_WRITE_DELAY, &page_link.link,
+                        LSN_IMPOSSIBLE))
       DBUG_RETURN(1);
   }
   else /* page is now empty */
@@ -3069,7 +3072,8 @@ static my_bool delete_head_or_tail(MARIA_HA *info,
                         buff, share->page_type,
                         lock_at_write,
                         PAGECACHE_PIN_LEFT_PINNED,
-                        PAGECACHE_WRITE_DELAY, &page_link.link))
+                        PAGECACHE_WRITE_DELAY, &page_link.link,
+                        LSN_IMPOSSIBLE))
       DBUG_RETURN(1);
 
     DBUG_ASSERT(empty_space >= info->s->bitmap.sizes[0]);
@@ -5037,7 +5041,8 @@ uint _ma_apply_redo_insert_row_head_or_tail(MARIA_HA *info, LSN lsn,
                       &info->dfile, page, 0,
                       buff, PAGECACHE_PLAIN_PAGE,
                       unlock_method, unpin_method,
-                      PAGECACHE_WRITE_DELAY, 0))
+                      PAGECACHE_WRITE_DELAY, 0,
+                      LSN_IMPOSSIBLE))
     DBUG_RETURN(my_errno);
 
   /* Fix bitmap */
@@ -5144,7 +5149,8 @@ uint _ma_apply_redo_purge_row_head_or_tail(MARIA_HA *info, LSN lsn,
                       &info->dfile, page, 0,
                       buff, PAGECACHE_PLAIN_PAGE,
                       PAGECACHE_LOCK_WRITE_UNLOCK, PAGECACHE_UNPIN,
-                      PAGECACHE_WRITE_DELAY, 0))
+                      PAGECACHE_WRITE_DELAY, 0,
+                      LSN_IMPOSSIBLE))
     result= my_errno;
 
   /* This will work even if the page was marked as UNALLOCATED_PAGE */
@@ -5224,7 +5230,8 @@ uint _ma_apply_redo_purge_blocks(MARIA_HA *info,
                           &info->dfile, page+i, 0,
                           buff, PAGECACHE_PLAIN_PAGE,
                           PAGECACHE_LOCK_WRITE_UNLOCK, PAGECACHE_UNPIN,
-                          PAGECACHE_WRITE_DELAY, 0))
+                          PAGECACHE_WRITE_DELAY, 0,
+                          LSN_IMPOSSIBLE))
         DBUG_RETURN(my_errno);
     }
     /** @todo leave bitmap lock to the bitmap code... */

storage/maria/ma_loghandler.c

@@ -2059,6 +2059,7 @@ static my_bool translog_buffer_flush(struct st_translog_buffer *buffer)
                         PAGECACHE_PLAIN_PAGE,
                         PAGECACHE_LOCK_LEFT_UNLOCKED,
                         PAGECACHE_PIN_LEFT_UNPINNED, 0,
+                        LSN_IMPOSSIBLE,
                         &translog_page_validator, (uchar*) &data))
     {
       UNRECOVERABLE_ERROR(("Can't write page (%lu,0x%lx) to pagecache",

storage/maria/ma_page.c

@@ -106,7 +106,7 @@ int _ma_write_keypage(register MARIA_HA *info, register MARIA_KEYDEF *keyinfo,
                               level, buff, PAGECACHE_PLAIN_PAGE,
                               PAGECACHE_LOCK_LEFT_UNLOCKED,
                               PAGECACHE_PIN_LEFT_UNPINNED,
-                              PAGECACHE_WRITE_DELAY, 0));
+                              PAGECACHE_WRITE_DELAY, 0, LSN_IMPOSSIBLE));
 } /* maria_write_keypage */
 
 
@@ -140,7 +140,7 @@ int _ma_dispose(register MARIA_HA *info, MARIA_KEYDEF *keyinfo, my_off_t pos,
                                    PAGECACHE_PLAIN_PAGE,
                                    PAGECACHE_LOCK_LEFT_UNLOCKED,
                                    PAGECACHE_PIN_LEFT_UNPINNED,
-                                   PAGECACHE_WRITE_DELAY, 0,
+                                   PAGECACHE_WRITE_DELAY, 0, LSN_IMPOSSIBLE,
                                    offset, sizeof(buff), 0, 0));
 } /* _ma_dispose */
 

storage/maria/ma_pagecache.c

@@ -632,6 +632,24 @@ static uint pagecache_fwrite(PAGECACHE *pagecache,
            (pageno)<<(pagecache->shift), flags)
 
 
+/**
+  @brief set rec_lsn of pagecache block (if it is needed)
+
+  @param block                   block where to set rec_lsn
+  @param first_REDO_LSN_for_page the LSN to set
+*/
+
+static inline void pagecache_set_block_rec_lsn(PAGECACHE_BLOCK_LINK *block,
+                                               LSN first_REDO_LSN_for_page)
+{
+  if (block->rec_lsn == LSN_MAX)
+    block->rec_lsn= first_REDO_LSN_for_page;
+  else
+    DBUG_ASSERT(cmp_translog_addr(block->rec_lsn,
+                                  first_REDO_LSN_for_page) <= 0);
+}
+
+
 /*
   next_power(value) is 2 at the power of (1+floor(log2(value)));
   e.g. next_power(2)=4, next_power(3)=4.
@@ -2568,12 +2586,7 @@ void pagecache_unlock(PAGECACHE *pagecache,
   {
     DBUG_ASSERT(lock == PAGECACHE_LOCK_WRITE_UNLOCK);
     DBUG_ASSERT(pin == PAGECACHE_UNPIN);
-    if (block->rec_lsn == LSN_MAX)
-      block->rec_lsn= first_REDO_LSN_for_page;
-    else
-      DBUG_ASSERT(cmp_translog_addr(block->rec_lsn,
-                                    first_REDO_LSN_for_page) <= 0);
-
+    pagecache_set_block_rec_lsn(block, first_REDO_LSN_for_page);
   }
   if (lsn != LSN_IMPOSSIBLE)
     check_and_set_lsn(pagecache, lsn, block);
@@ -2726,14 +2739,10 @@ void pagecache_unlock_by_link(PAGECACHE *pagecache,
       with WRITE lock that was temporarly converted to READ lock before
       it's unpinned
     */
-    DBUG_ASSERT(lock == PAGECACHE_LOCK_WRITE_UNLOCK || 
+    DBUG_ASSERT(lock == PAGECACHE_LOCK_WRITE_UNLOCK ||
                 lock == PAGECACHE_LOCK_READ_UNLOCK);
     DBUG_ASSERT(pin == PAGECACHE_UNPIN);
-    if (block->rec_lsn == LSN_MAX)
-      block->rec_lsn= first_REDO_LSN_for_page;
-    else
-      DBUG_ASSERT(cmp_translog_addr(block->rec_lsn,
-                                    first_REDO_LSN_for_page) <= 0);
+    pagecache_set_block_rec_lsn(block, first_REDO_LSN_for_page);
   }
   if (lsn != LSN_IMPOSSIBLE)
     check_and_set_lsn(pagecache, lsn, block);
@@ -3153,25 +3162,27 @@ my_bool pagecache_delete_pages(PAGECACHE *pagecache,
 }
 
 
-/*
-  Write a buffer into a cached file.
-
-  SYNOPSIS
-
-    pagecache_write_part()
-      pagecache           pointer to a page cache data structure
-      file                handler for the file to write data to
-      pageno              number of the block of data in the file
-      level               determines the weight of the data
-      buff                buffer with the data
-      type                type of the page
-      lock                lock change
-      pin                 pin page
-      write_mode          how to write page
-      link                link to the page if we pin it
-
-  RETURN VALUE
-    0 if a success, 1 - otherwise.
+/**
+  @brief Writes a buffer into a cached file.
+
+  @param pagecache       pointer to a page cache data structure
+  @param file            handler for the file to write data to
+  @param pageno          number of the block of data in the file
+  @param level           determines the weight of the data
+  @param buff            buffer with the data
+  @param type            type of the page
+  @param lock            lock change
+  @param pin             pin page
+  @param write_mode      how to write page
+  @param link            link to the page if we pin it
+  @param first_REDO_LSN_for_page the lsn to set rec_lsn
+  @param offset          offset in the page
+  @param size            size of data
+  @param validator       read page validator
+  @param validator_data  the validator data
+
+  @retval 0 if a success.
+  @retval 1 Error.
 */
 
 /* description of how to change lock before and after write */
@@ -3235,6 +3246,7 @@ my_bool pagecache_write_part(PAGECACHE *pagecache,
                              enum pagecache_page_pin pin,
                              enum pagecache_write_mode write_mode,
                              PAGECACHE_BLOCK_LINK **link,
+                             LSN first_REDO_LSN_for_page,
                              uint offset, uint size,
                              pagecache_disk_read_validator validator,
                              uchar* validator_data)
@@ -3364,6 +3376,15 @@ restart:
         block->status|= PCBLOCK_READ;
       }
     }
+    if (first_REDO_LSN_for_page)
+    {
+      /* single write action of the last write action */
+      DBUG_ASSERT(lock == PAGECACHE_LOCK_WRITE_UNLOCK ||
+                  lock == PAGECACHE_LOCK_LEFT_UNLOCKED);
+      DBUG_ASSERT(pin == PAGECACHE_UNPIN ||
+                  pin == PAGECACHE_PIN_LEFT_UNPINNED);
+      pagecache_set_block_rec_lsn(block, first_REDO_LSN_for_page);
+    }
 
     if (need_lock_change)
     {

storage/maria/ma_pagecache.h

@@ -201,12 +201,12 @@ extern uchar *pagecache_valid_read(PAGECACHE *pagecache,
                                   pagecache_disk_read_validator validator,
                                   uchar* validator_data);
 
-#define  pagecache_write(P,F,N,L,B,T,O,I,M,K) \
-   pagecache_write_part(P,F,N,L,B,T,O,I,M,K,0,(P)->block_size,0,0)
+#define  pagecache_write(P,F,N,L,B,T,O,I,M,K,R) \
+   pagecache_write_part(P,F,N,L,B,T,O,I,M,K,R,0,(P)->block_size,0,0)
 
-#define  pagecache_inject(P,F,N,L,B,T,O,I,K,V,D) \
+#define  pagecache_inject(P,F,N,L,B,T,O,I,K,R,V,D) \
    pagecache_write_part(P,F,N,L,B,T,O,I,PAGECACHE_WRITE_DONE, \
-                        K,0,(P)->block_size,V,D)
+                        K,R,0,(P)->block_size,V,D)
 
 extern my_bool pagecache_write_part(PAGECACHE *pagecache,
                                     PAGECACHE_FILE *file,
@@ -218,6 +218,7 @@ extern my_bool pagecache_write_part(PAGECACHE *pagecache,
                                     enum pagecache_page_pin pin,
                                     enum pagecache_write_mode write_mode,
                                     PAGECACHE_BLOCK_LINK **link,
+                                    LSN first_REDO_LSN_for_page,
                                     uint offset,
                                     uint size,
                                     pagecache_disk_read_validator validator,

storage/maria/ma_preload.c

@@ -100,7 +100,7 @@ int maria_preload(MARIA_HA *info, ulonglong key_map, my_bool ignore_leaves)
                               PAGECACHE_PLAIN_PAGE,
                               PAGECACHE_LOCK_LEFT_UNLOCKED,
                               PAGECACHE_PIN_LEFT_UNPINNED,
-                              PAGECACHE_WRITE_DONE, 0))
+                              PAGECACHE_WRITE_DONE, 0, LSN_IMPOSSIBLE))
 	    goto err;
 	}
         pos+= block_length;
@@ -117,7 +117,8 @@ int maria_preload(MARIA_HA *info, ulonglong key_map, my_bool ignore_leaves)
                           PAGECACHE_PLAIN_PAGE,
                           PAGECACHE_LOCK_LEFT_UNLOCKED,
                           PAGECACHE_PIN_LEFT_UNPINNED,
-                          PAGECACHE_WRITE_DONE, 0))
+                          PAGECACHE_WRITE_DONE, 0,
+                          LSN_IMPOSSIBLE))
 	goto err;
       pos+= length;
     }

storage/maria/unittest/ma_pagecache_consist.c

@@ -237,7 +237,7 @@ void writer(int num)
                     PAGECACHE_LOCK_WRITE_UNLOCK,
                     PAGECACHE_UNPIN,
                     PAGECACHE_WRITE_DELAY,
-                    0);
+                    0, LSN_IMPOSSIBLE);
 
     if (i % flush_divider == 0)
       flush_pagecache_blocks(&pagecache, &file1, FLUSH_FORCE_WRITE);
@@ -380,7 +380,7 @@ int main(int argc __attribute__((unused)),
                       PAGECACHE_LOCK_LEFT_UNLOCKED,
                       PAGECACHE_PIN_LEFT_UNPINNED,
                       PAGECACHE_WRITE_DELAY,
-                      0);
+                      0, LSN_IMPOSSIBLE);
     }
     flush_pagecache_blocks(&pagecache, &file1, FLUSH_FORCE_WRITE);
     free(buffr);

storage/maria/unittest/ma_pagecache_single.c

@@ -104,7 +104,7 @@ int simple_read_write_test()
                   PAGECACHE_LOCK_LEFT_UNLOCKED,
                   PAGECACHE_PIN_LEFT_UNPINNED,
                   PAGECACHE_WRITE_DELAY,
-                  0);
+                  0, LSN_IMPOSSIBLE);
   pagecache_read(&pagecache, &file1, 0, 3, (char*)buffr,
                  PAGECACHE_PLAIN_PAGE,
                  PAGECACHE_LOCK_LEFT_UNLOCKED,
@@ -140,7 +140,7 @@ int simple_read_change_write_read_test()
                   PAGECACHE_LOCK_LEFT_UNLOCKED,
                   PAGECACHE_PIN_LEFT_UNPINNED,
                   PAGECACHE_WRITE_DELAY,
-                  0);
+                  0, LSN_IMPOSSIBLE);
   flush_pagecache_blocks(&pagecache, &file1, FLUSH_FORCE_WRITE);
   /* test */
   pagecache_read(&pagecache, &file1, 0, 3, (char*)buffw,
@@ -153,7 +153,7 @@ int simple_read_change_write_read_test()
                   PAGECACHE_LOCK_WRITE_UNLOCK,
                   PAGECACHE_UNPIN,
                   PAGECACHE_WRITE_DELAY,
-                  0);
+                  0, LSN_IMPOSSIBLE);
 
   pagecache_read(&pagecache, &file1, 0, 3, (char*)buffr,
                  PAGECACHE_PLAIN_PAGE,
@@ -194,7 +194,7 @@ int simple_pin_test()
                   PAGECACHE_LOCK_LEFT_UNLOCKED,
                   PAGECACHE_PIN_LEFT_UNPINNED,
                   PAGECACHE_WRITE_DELAY,
-                  0);
+                  0, LSN_IMPOSSIBLE);
   /* test */
   if (flush_pagecache_blocks(&pagecache, &file1, FLUSH_FORCE_WRITE))
   {
@@ -210,14 +210,14 @@ int simple_pin_test()
                   PAGECACHE_LOCK_LEFT_UNLOCKED,
                   PAGECACHE_PIN_LEFT_UNPINNED,
                   PAGECACHE_WRITE_DELAY,
-                  0);
+                  0, LSN_IMPOSSIBLE);
   bfill(buffw + PAGE_SIZE/2, PAGE_SIZE/2, ((unsigned char) 129));
   pagecache_write(&pagecache, &file1, 0, 3, (char*)buffw,
                   PAGECACHE_PLAIN_PAGE,
                   PAGECACHE_LOCK_WRITE_TO_READ,
                   PAGECACHE_PIN_LEFT_PINNED,
                   PAGECACHE_WRITE_DELAY,
-                  0);
+                  0, LSN_IMPOSSIBLE);
   /*
     We have to get error because one page of the file is pinned,
     other page should be flushed
@@ -272,7 +272,7 @@ int simple_delete_forget_test()
                   PAGECACHE_LOCK_LEFT_UNLOCKED,
                   PAGECACHE_PIN_LEFT_UNPINNED,
                   PAGECACHE_WRITE_DELAY,
-                  0);
+                  0, LSN_IMPOSSIBLE);
   flush_pagecache_blocks(&pagecache, &file1, FLUSH_FORCE_WRITE);
   /* test */
   bfill(buffw, PAGE_SIZE, '\2');
@@ -281,7 +281,7 @@ int simple_delete_forget_test()
                   PAGECACHE_LOCK_LEFT_UNLOCKED,
                   PAGECACHE_PIN_LEFT_UNPINNED,
                   PAGECACHE_WRITE_DELAY,
-                  0);
+                  0, LSN_IMPOSSIBLE);
   pagecache_delete(&pagecache, &file1, 0,
                    PAGECACHE_LOCK_WRITE, 0);
   flush_pagecache_blocks(&pagecache, &file1, FLUSH_FORCE_WRITE);
@@ -314,7 +314,7 @@ int simple_delete_flush_test()
                   PAGECACHE_LOCK_WRITE,
                   PAGECACHE_PIN,
                   PAGECACHE_WRITE_DELAY,
-                  0);
+                  0, LSN_IMPOSSIBLE);
   flush_pagecache_blocks(&pagecache, &file1, FLUSH_FORCE_WRITE);
   /* test */
   bfill(buffw, PAGE_SIZE, '\2');
@@ -323,7 +323,7 @@ int simple_delete_flush_test()
                   PAGECACHE_LOCK_LEFT_WRITELOCKED,
                   PAGECACHE_PIN_LEFT_PINNED,
                   PAGECACHE_WRITE_DELAY,
-                  0);
+                  0, LSN_IMPOSSIBLE);
   pagecache_delete(&pagecache, &file1, 0,
                    PAGECACHE_LOCK_LEFT_WRITELOCKED, 1);
   flush_pagecache_blocks(&pagecache, &file1, FLUSH_FORCE_WRITE);
@@ -362,7 +362,7 @@ int simple_big_test()
                     PAGECACHE_LOCK_LEFT_UNLOCKED,
                     PAGECACHE_PIN_LEFT_UNPINNED,
                     PAGECACHE_WRITE_DELAY,
-                    0);
+                    0, LSN_IMPOSSIBLE);
   }
   desc[i].length= 0;
   desc[i].content= '\0';

storage/maria/unittest/ma_test_loghandler_pagecache-t.c

@@ -129,7 +129,7 @@ int main(int argc __attribute__((unused)), char *argv[])
                     PAGECACHE_LOCK_LEFT_UNLOCKED,
                     PAGECACHE_PIN_LEFT_UNPINNED,
                     PAGECACHE_WRITE_DELAY,
-                    0);
+                    0, LSN_IMPOSSIBLE);
     flush_pagecache_blocks(&pagecache, &file1, FLUSH_FORCE_WRITE);
   }
   if ((stat= my_stat(first_translog_file, &st,  MYF(0))) == 0)