rec_lsn (first REDO LSN( is now given to the page cache on unpinning

Added maria_clone(), needed by future REPAIR code


storage/maria/unittest/ma_pagecache_consist.c:
  Change mode to -rw-rw-r--
storage/maria/unittest/lockman-t.c:
  Change mode to -rw-rw-r--
storage/maria/unittest/lockman1-t.c:
  Change mode to -rw-rw-r--
storage/maria/unittest/lockman2-t.c:
  Change mode to -rw-rw-r--
storage/maria/unittest/trnman-t.c:
  Change mode to -rw-rw-r--
include/maria.h:
  Added prototype for maria_clone (for future)
storage/maria/ha_maria.cc:
  Move filename to share structure
storage/maria/ma_blockrec.c:
  rec_lsn (first REDO LSN( is now given to the page cache on unpinning
  Removed impossible lock handling in get_head_or_tail_page()
  Changed calls ot translog_write_record() to remember rec_lsn
  Removed some logging in csse of not transactions
storage/maria/ma_delete.c:
  info->filename -> info->s->open_file_name
storage/maria/ma_loghandler.c:
  Indentation fixes
storage/maria/ma_open.c:
  Added maria_clone(), needed by future REPAIR code
storage/maria/ma_packrec.c:
  Fixed typo in comment
storage/maria/ma_pagecache.c:
  Added comment.
  Allow setting REC_LSN in case of read lock
storage/maria/ma_update.c:
  info->filename -> info->s->open_file_name
storage/maria/ma_write.c:
  info->filename -> info->s->open_file_name
storage/maria/maria_def.h:
  info->filename -> info->s->open_file_name
  Added have_rtree to simplify test in ma_clone()
storage/maria/maria_ftdump.c:
  info->filename -> info->s->open_file_name
storage/maria/maria_pack.c:
  info->filename -> info->s->open_file_name
storage/maria/trnman.h:
  Added rec_lsn

include/maria.h

@@ -263,6 +263,7 @@ extern int maria_close(struct st_maria_info *file);
 extern int maria_delete(struct st_maria_info *file, const byte *buff);
 extern struct st_maria_info *maria_open(const char *name, int mode,
 					uint wait_if_locked);
+extern struct st_maria_info *maria_clone(struct st_maria_share *share, int mode);
 extern int maria_panic(enum ha_panic_function function);
 extern int maria_rfirst(struct st_maria_info *file, byte *buf, int inx);
 extern int maria_rkey(struct st_maria_info *file, byte *buf, int inx,

storage/maria/ha_maria.cc

@@ -1058,7 +1058,7 @@ int ha_maria::repair(THD *thd, HA_CHECK &param, bool do_optimize)
   param.thd= thd;
   param.tmpdir= &mysql_tmpdir_list;
   param.out_flag= 0;
-  strmov(fixed_name, file->filename);
+  strmov(fixed_name, file->s->open_file_name);
 
 #ifndef TO_BE_FIXED
   /* QQ: Until we have repair for block format, lie that it succeded */
@@ -1793,11 +1793,11 @@ int ha_maria::info(uint flag)
        if table is symlinked (Ie;  Real name is not same as generated name)
     */
     data_file_name= index_file_name= 0;
-    fn_format(name_buff, file->filename, "", MARIA_NAME_DEXT,
+    fn_format(name_buff, file->s->open_file_name, "", MARIA_NAME_DEXT,
               MY_APPEND_EXT | MY_UNPACK_FILENAME);
     if (strcmp(name_buff, maria_info.data_file_name))
       data_file_name=maria_info.data_file_name;
-    fn_format(name_buff, file->filename, "", MARIA_NAME_IEXT,
+    fn_format(name_buff, file->s->open_file_name, "", MARIA_NAME_IEXT,
               MY_APPEND_EXT | MY_UNPACK_FILENAME);
     if (strcmp(name_buff, maria_info.index_file_name))
       index_file_name=maria_info.index_file_name;

storage/maria/ma_blockrec.c

@@ -553,6 +553,8 @@ static my_bool check_if_zero(byte *pos, uint length)
     We unpin pages in the reverse order as they where pinned; This may not
     be strictly necessary but may simplify things in the future.
 
+    info->s->rec_lsn contains the lsn for the first REDO
+
   RETURN
     0   ok
     1   error (fatal disk error)
@@ -576,8 +578,9 @@ void _ma_unpin_all_pages(MARIA_HA *info, LSN undo_lsn)
   while (pinned_page-- != page_link)
     pagecache_unlock_by_link(info->s->pagecache, pinned_page->link,
                              pinned_page->unlock, PAGECACHE_UNPIN,
-                             0, undo_lsn);
+                             info->trn->rec_lsn, undo_lsn);
 
+  info->trn->rec_lsn= 0;
   info->pinned_pages.elements= 0;
   DBUG_VOID_RETURN;
 }
@@ -1037,7 +1040,6 @@ static my_bool get_head_or_tail_page(MARIA_HA *info,
   else
   {
     byte *dir;
-    /* TODO: lock the page */
     /* Read old page */
     DBUG_ASSERT(share->pagecache->block_size == block_size);
     if (!(res->buff= pagecache_read(share->pagecache,
@@ -1046,13 +1048,8 @@ static my_bool get_head_or_tail_page(MARIA_HA *info,
                                     buff, share->page_type,
                                     lock, &page_link.link)))
       DBUG_RETURN(1);
-    if (lock != PAGECACHE_LOCK_LEFT_UNLOCKED)
-    {
-      page_link.unlock= (lock == PAGECACHE_LOCK_READ ?
-                         PAGECACHE_LOCK_READ_UNLOCK :
-                         PAGECACHE_LOCK_WRITE_UNLOCK);
+    page_link.unlock= PAGECACHE_LOCK_WRITE_UNLOCK;
     push_dynamic(&info->pinned_pages, (void*) &page_link);
-    }
 
     DBUG_ASSERT((res->buff[PAGE_TYPE_OFFSET] & PAGE_TYPE_MASK) == page_type);
     if (!(dir= find_free_position(res->buff, block_size, &res->rownr,
@@ -1144,7 +1141,8 @@ static my_bool write_tail(MARIA_HA *info,
     log_array[TRANSLOG_INTERNAL_PARTS + 0].length= sizeof(log_data);
     log_array[TRANSLOG_INTERNAL_PARTS + 1].str=    (char*) row_pos.data;
     log_array[TRANSLOG_INTERNAL_PARTS + 1].length= length;
-    if (translog_write_record(&lsn, LOGREC_REDO_INSERT_ROW_TAIL,
+    if (translog_write_record(!info->trn->rec_lsn ? &info->trn->rec_lsn : &lsn,
+                              LOGREC_REDO_INSERT_ROW_TAIL,
                               info->trn->short_id, NULL, share,
                               sizeof(log_data) + length,
                               TRANSLOG_INTERNAL_PARTS + 2,
@@ -1400,7 +1398,8 @@ static my_bool free_full_pages(MARIA_HA *info, MARIA_ROW *row)
   log_array[TRANSLOG_INTERNAL_PARTS + 0].length= sizeof(log_data);
   log_array[TRANSLOG_INTERNAL_PARTS + 1].str=    row->extents;
   log_array[TRANSLOG_INTERNAL_PARTS + 1].length= extents_length;
-  if (translog_write_record(&lsn, LOGREC_REDO_PURGE_BLOCKS,
+  if (translog_write_record(!info->trn->rec_lsn ? &info->trn->rec_lsn : &lsn,
+                            LOGREC_REDO_PURGE_BLOCKS,
                             info->trn->short_id, NULL, info->s,
                             sizeof(log_data) + extents_length,
                             TRANSLOG_INTERNAL_PARTS + 2, log_array))
@@ -1417,6 +1416,9 @@ static my_bool free_full_pages(MARIA_HA *info, MARIA_ROW *row)
   NOTES
     This is very similar to free_full_pages()
 
+    We don't have to update trn->rec_lsn here as before calling this function
+    we have already generated REDO's for deleting the HEAD block.
+
   RETURN
     0   ok
     1   error
@@ -1427,13 +1429,16 @@ static my_bool free_full_page_range(MARIA_HA *info, ulonglong page, uint count)
   uchar log_data[FILEID_STORE_SIZE + PAGERANGE_STORE_SIZE +
                  ROW_EXTENT_SIZE];
   LEX_STRING log_array[TRANSLOG_INTERNAL_PARTS + 1];
-  LSN lsn;
   my_bool res= 0;
 
   if (pagecache_delete_pages(info->s->pagecache, &info->dfile,
                              page, count, PAGECACHE_LOCK_WRITE, 0))
     res= 1;
 
+  if (info->s->base.transactional)
+  {
+    LSN lsn;
+    DBUG_ASSERT(info->trn->rec_lsn);
     fileid_store(log_data, info->dfile.file);
     pagerange_store(log_data + FILEID_STORE_SIZE, 1);
     int5store(log_data + FILEID_STORE_SIZE + PAGERANGE_STORE_SIZE,
@@ -1449,6 +1454,7 @@ static my_bool free_full_page_range(MARIA_HA *info, ulonglong page, uint count)
                               TRANSLOG_INTERNAL_PARTS + 1, log_array))
       res= 1;
 
+  }
   pthread_mutex_lock(&info->s->bitmap.bitmap_lock);
   if (_ma_reset_full_page_bits(info, &info->s->bitmap, page,
                                count))
@@ -1951,7 +1957,8 @@ static my_bool write_block_record(MARIA_HA *info,
     log_array[TRANSLOG_INTERNAL_PARTS + 0].length= sizeof(log_data);
     log_array[TRANSLOG_INTERNAL_PARTS + 1].str=    (char*) row_pos->data;
     log_array[TRANSLOG_INTERNAL_PARTS + 1].length= data_length;
-    if (translog_write_record(&lsn, LOGREC_REDO_INSERT_ROW_HEAD,
+    if (translog_write_record(!info->trn->rec_lsn ? &info->trn->rec_lsn : &lsn,
+                              LOGREC_REDO_INSERT_ROW_HEAD,
                               info->trn->short_id, NULL, share,
                               sizeof(log_data) + data_length,
                               TRANSLOG_INTERNAL_PARTS + 2, log_array))
@@ -2066,6 +2073,7 @@ static my_bool write_block_record(MARIA_HA *info,
                                                              log_data);
     log_entry_length+= (log_pos - log_data);
 
+    /* trn->rec_lsn is already set earlier in this function */
     error= translog_write_record(&lsn, LOGREC_REDO_INSERT_ROW_BLOBS,
                                  info->trn->short_id, NULL, share,
                                  log_entry_length, (uint) (log_array_pos -
@@ -2524,7 +2532,7 @@ static my_bool delete_head_or_tail(MARIA_HA *info,
 
     log_array[TRANSLOG_INTERNAL_PARTS + 0].str=    (char*) log_data;
     log_array[TRANSLOG_INTERNAL_PARTS + 0].length= sizeof(log_data);
-    if (translog_write_record(&lsn,
+    if (translog_write_record(!info->trn->rec_lsn ? &info->trn->rec_lsn : &lsn,
                               (head ? LOGREC_REDO_PURGE_ROW_HEAD :
                                LOGREC_REDO_PURGE_ROW_TAIL),
                               info->trn->short_id, NULL, share,
@@ -2557,7 +2565,8 @@ static my_bool delete_head_or_tail(MARIA_HA *info,
                     PAGERANGE_STORE_SIZE, 1);
     log_array[TRANSLOG_INTERNAL_PARTS + 0].str=    (char*) log_data;
     log_array[TRANSLOG_INTERNAL_PARTS + 0].length= sizeof(log_data);
-    if (translog_write_record(&lsn, LOGREC_REDO_PURGE_BLOCKS,
+    if (translog_write_record(!info->trn->rec_lsn ? &info->trn->rec_lsn : &lsn,
+                              LOGREC_REDO_PURGE_BLOCKS,
                               info->trn->short_id, NULL, share,
                               sizeof(log_data), TRANSLOG_INTERNAL_PARTS + 1,
                               log_array))

storage/maria/ma_delete.c

@@ -111,8 +111,8 @@ int maria_delete(MARIA_HA *info,const byte *record)
   allow_break();			/* Allow SIGHUP & SIGINT */
   if (info->invalidator != 0)
   {
-    DBUG_PRINT("info", ("invalidator... '%s' (delete)", info->filename));
-    (*info->invalidator)(info->filename);
+    DBUG_PRINT("info", ("invalidator... '%s' (delete)", info->s->open_file_name));
+    (*info->invalidator)(info->s->open_file_name);
     info->invalidator=0;
   }
   DBUG_RETURN(0);

storage/maria/ma_loghandler.c

@@ -3855,7 +3855,6 @@ static my_bool translog_write_variable_record(LSN *lsn,
   uint page_rest;
   /* Max number of such LSNs per record is 2 */
   byte compressed_LSNs[2 * LSN_STORE_SIZE];
-
   DBUG_ENTER("translog_write_variable_record");
 
   translog_lock();

storage/maria/ma_packrec.c

@@ -1411,7 +1411,7 @@ uint _ma_pack_get_block_info(MARIA_HA *maria, MARIA_BIT_BUFF *bit_buff,
   {
     ref_length=maria->s->pack.ref_length;
     /*
-      We can't use my_pread() here because maria_read_rnd_pack_record assumes
+      We can't use my_pread() here because _ma_read_rnd_pack_record assumes
       position is ok
     */
     VOID(my_seek(file,filepos,MY_SEEK_SET,MYF(0)));

storage/maria/ma_pagecache.c

@@ -2677,7 +2677,13 @@ void pagecache_unlock_by_link(PAGECACHE *pagecache,
   inc_counter_for_resize_op(pagecache);
   if (first_REDO_LSN_for_page)
   {
-    DBUG_ASSERT(lock == PAGECACHE_LOCK_WRITE_UNLOCK);
+    /*
+      LOCK_READ_UNLOCK is ok here as the page may have first locked
+      with WRITE lock that was temporarly converted to READ lock before
+      it's unpinned
+    */
+    DBUG_ASSERT(lock == PAGECACHE_LOCK_WRITE_UNLOCK || 
+                lock == PAGECACHE_LOCK_READ_UNLOCK);
     DBUG_ASSERT(pin == PAGECACHE_UNPIN);
     set_if_bigger(block->rec_lsn, first_REDO_LSN_for_page);
   }

storage/maria/ma_update.c

@@ -196,8 +196,8 @@ int maria_update(register MARIA_HA *info, const byte *oldrec, byte *newrec)
   allow_break();				/* Allow SIGHUP & SIGINT */
   if (info->invalidator != 0)
   {
-    DBUG_PRINT("info", ("invalidator... '%s' (update)", info->filename));
-    (*info->invalidator)(info->filename);
+    DBUG_PRINT("info", ("invalidator... '%s' (update)", info->s->open_file_name));
+    (*info->invalidator)(info->s->open_file_name);
     info->invalidator=0;
   }
   DBUG_RETURN(0);

storage/maria/ma_write.c

@@ -180,8 +180,8 @@ int maria_write(MARIA_HA *info, byte *record)
   VOID(_ma_writeinfo(info, WRITEINFO_UPDATE_KEYFILE));
   if (info->invalidator != 0)
   {
-    DBUG_PRINT("info", ("invalidator... '%s' (update)", info->filename));
-    (*info->invalidator)(info->filename);
+    DBUG_PRINT("info", ("invalidator... '%s' (update)", info->s->open_file_name));
+    (*info->invalidator)(info->s->open_file_name);
     info->invalidator=0;
   }
 

storage/maria/maria_def.h

@@ -222,9 +222,9 @@ typedef struct st_maria_share
   MARIA_PACK pack;			/* Data about packed records */
   MARIA_BLOB *blobs;			/* Pointer to blobs */
   char *unique_file_name;		/* realpath() of index file */
-  char *data_file_name,			/* Resolved path names from
-						   symlinks */
-   *index_file_name;
+  char *data_file_name;			/* Resolved path names from symlinks */
+  char *index_file_name;
+  char *open_file_name;			/* parameter to open filename */
   byte *file_map;			/* mem-map of file if possible */
   PAGECACHE *pagecache;			/* ref to the current key cache */
   MARIA_DECODE_TREE *decode_trees;
@@ -299,6 +299,7 @@ typedef struct st_maria_share
     global_changed,			/* If changed since open */
     not_flushed, concurrent_insert;
   my_bool delay_key_write;
+  my_bool have_rtree;
 #ifdef THREAD
   THR_LOCK lock;
   pthread_mutex_t intern_lock;		/* Locking for use with _locking */
@@ -388,7 +389,6 @@ struct st_maria_info
   DYNAMIC_ARRAY *ft1_to_ft2;		/* used only in ft1->ft2 conversion */
   MEM_ROOT      ft_memroot;             /* used by the parser               */
   MYSQL_FTPARSER_PARAM *ftparser_param;	/* share info between init/deinit */
-  char *filename;			/* parameter to open filename */
   byte *buff;				/* page buffer */
   byte *keyread_buff;                   /* Buffer for last key read */
   byte *lastkey, *lastkey2;		/* Last used search key */

storage/maria/maria_ftdump.c

@@ -100,7 +100,7 @@ int main(int argc,char *argv[])
   if ((inx >= info->s->base.keys) ||
       !(info->s->keyinfo[inx].flag & HA_FULLTEXT))
   {
-    printf("Key %d in table %s is not a FULLTEXT key\n", inx, info->filename);
+    printf("Key %d in table %s is not a FULLTEXT key\n", inx, info->s->open_file_name);
     goto err;
   }
 

storage/maria/maria_pack.c

@@ -509,9 +509,11 @@ static int compress(PACK_MRG_INFO *mrg,char *result_table)
 
   /* Create temporary or join file */
   if (backup)
-    VOID(fn_format(org_name,isam_file->filename,"",MARIA_NAME_DEXT,2));
+    VOID(fn_format(org_name,isam_file->s->open_file_name,"",MARIA_NAME_DEXT,
+                   2));
   else
-    VOID(fn_format(org_name,isam_file->filename,"",MARIA_NAME_DEXT,2+4+16));
+    VOID(fn_format(org_name,isam_file->s->open_file_name,"",MARIA_NAME_DEXT,
+                   2+4+16));
 
   if (init_pagecache(maria_pagecache, MARIA_MIN_PAGE_CACHE_SIZE, 0, 0,
                      maria_block_size) == 0)
@@ -705,7 +707,8 @@ static int compress(PACK_MRG_INFO *mrg,char *result_table)
     {
       if (backup)
       {
-	if (my_rename(org_name,make_old_name(temp_name,isam_file->filename),
+	if (my_rename(org_name,make_old_name(temp_name,
+                                             isam_file->s->open_file_name),
 		      MYF(MY_WME)))
 	  error=1;
 	else

storage/maria/trnman.h

@@ -45,7 +45,7 @@ struct st_transaction
   LF_PINS             *pins;
   TrID                 trid, min_read_from, commit_trid;
   TRN                 *next, *prev;
-  LSN		       undo_lsn;
+  LSN		       rec_lsn, undo_lsn;
   uint                 locked_tables;
   /* Note! if locks.loid is 0, trn is NOT initialized */
 };