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);
-      push_dynamic(&info->pinned_pages, (void*) &page_link);
-    }
+    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,28 +1429,32 @@ 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;
 
-  fileid_store(log_data, info->dfile.file);
-  pagerange_store(log_data + FILEID_STORE_SIZE, 1);
-  int5store(log_data + FILEID_STORE_SIZE + PAGERANGE_STORE_SIZE,
-            page);
-  int2store(log_data + FILEID_STORE_SIZE + PAGERANGE_STORE_SIZE + 5,
-            count);
-  log_array[TRANSLOG_INTERNAL_PARTS + 0].str=    (char*) log_data;
-  log_array[TRANSLOG_INTERNAL_PARTS + 0].length= sizeof(log_data);
+  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,
+              page);
+    int2store(log_data + FILEID_STORE_SIZE + PAGERANGE_STORE_SIZE + 5,
+              count);
+    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,
-                            info->trn->short_id, NULL, info->s,
-                            sizeof(log_data),
-                            TRANSLOG_INTERNAL_PARTS + 1, log_array))
-    res= 1;
+    if (translog_write_record(&lsn, LOGREC_REDO_PURGE_BLOCKS,
+                              info->trn->short_id, NULL, info->s,
+                              sizeof(log_data),
+                              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();
@@ -3867,8 +3866,8 @@ static my_bool translog_write_variable_record(LSN *lsn,
                       header_length1, page_rest));
 
   /*
-     header and part which we should read have to fit in one chunk
-     TODO: allow to divide readable header
+    header and part which we should read have to fit in one chunk
+    TODO: allow to divide readable header
   */
   if (page_rest <
       (header_length1 + log_record_type_descriptor[type].read_header_len))

storage/maria/ma_open.c

@@ -72,8 +72,163 @@ MARIA_HA *_ma_test_if_reopen(char *filename)
 }
 
 
+/*
+  Open a new instance of an already opened Maria table
+
+  SYNOPSIS
+    maria_clone_internal()
+    share	Share of already open table
+    mode	Mode of table (O_RDONLY | O_RDWR)
+    data_file   Filedescriptor of data file to use < 0 if one should open
+	        open it.    
+ 
+  RETURN
+    #   Maria handler
+    0   Error
+*/
+
+
+static MARIA_HA *maria_clone_internal(MARIA_SHARE *share, int mode,
+                                      File data_file)
+{
+  int save_errno;
+  uint errpos;
+  MARIA_HA info,*m_info;
+  DBUG_ENTER("maria_clone_internal");
+
+  errpos= 0;
+  bzero((byte*) &info,sizeof(info));
+
+  if (mode == O_RDWR && share->mode == O_RDONLY)
+  {
+    my_errno=EACCES;				/* Can't open in write mode */
+    goto err;
+  }
+  if (data_file >= 0)
+    info.dfile.file= data_file;
+  else if (_ma_open_datafile(&info, share, -1))
+    goto err;
+  errpos= 5;
+
+  /* alloc and set up private structure parts */
+  if (!my_multi_malloc(MY_WME,
+		       &m_info,sizeof(MARIA_HA),
+		       &info.blobs,sizeof(MARIA_BLOB)*share->base.blobs,
+		       &info.buff,(share->base.max_key_block_length*2+
+				   share->base.max_key_length),
+		       &info.lastkey,share->base.max_key_length*3+1,
+		       &info.first_mbr_key, share->base.max_key_length,
+		       &info.maria_rtree_recursion_state,
+                       share->have_rtree ? 1024 : 0,
+		       NullS))
+    goto err;
+  errpos= 6;
+
+  memcpy(info.blobs,share->blobs,sizeof(MARIA_BLOB)*share->base.blobs);
+  info.lastkey2=info.lastkey+share->base.max_key_length;
+
+  info.s=share;
+  info.cur_row.lastpos= HA_OFFSET_ERROR;
+  info.update= (short) (HA_STATE_NEXT_FOUND+HA_STATE_PREV_FOUND);
+  info.opt_flag=READ_CHECK_USED;
+  info.this_unique= (ulong) info.dfile.file; /* Uniq number in process */
+  if (share->data_file_type == COMPRESSED_RECORD)
+    info.this_unique= share->state.unique;
+  info.this_loop=0;				/* Update counter */
+  info.last_unique= share->state.unique;
+  info.last_loop=   share->state.update_count;
+  info.lock_type=F_UNLCK;
+  info.quick_mode=0;
+  info.bulk_insert=0;
+  info.ft1_to_ft2=0;
+  info.errkey= -1;
+  info.page_changed=1;
+  info.keyread_buff= info.buff + share->base.max_key_block_length;
+  if ((*share->init)(&info))
+    goto err;
+
+  pthread_mutex_lock(&share->intern_lock);
+  info.read_record= share->read_record;
+  share->reopen++;
+  share->write_flag=MYF(MY_NABP | MY_WAIT_IF_FULL);
+  if (share->options & HA_OPTION_READ_ONLY_DATA)
+  {
+    info.lock_type=F_RDLCK;
+    share->r_locks++;
+    share->tot_locks++;
+  }
+  if (share->options & HA_OPTION_TMP_TABLE)
+  {
+    share->temporary= share->delay_key_write= 1;
+
+    share->write_flag=MYF(MY_NABP);
+    share->w_locks++;			/* We don't have to update status */
+    share->tot_locks++;
+    info.lock_type=F_WRLCK;
+  }
+  if ((share->options & HA_OPTION_DELAY_KEY_WRITE) &&
+      maria_delay_key_write)
+    share->delay_key_write=1;
+
+  info.state= &share->state.state;	/* Change global values by default */
+  info.trn=   &dummy_transaction_object;
+  pthread_mutex_unlock(&share->intern_lock);
+
+  /* Allocate buffer for one record */
+  /* prerequisites: info->rec_buffer == 0 && info->rec_buff_size == 0 */
+  if (_ma_alloc_buffer(&info.rec_buff, &info.rec_buff_size,
+                       share->base.default_rec_buff_size))
+    goto err;
+
+  bzero(info.rec_buff, share->base.default_rec_buff_size);
+
+  *m_info=info;
+#ifdef THREAD
+  thr_lock_data_init(&share->lock,&m_info->lock,(void*) m_info);
+#endif
+  m_info->open_list.data=(void*) m_info;
+  maria_open_list=list_add(maria_open_list,&m_info->open_list);
+
+  DBUG_RETURN(m_info);
+
+err:
+  save_errno=my_errno ? my_errno : HA_ERR_END_OF_FILE;
+  if ((save_errno == HA_ERR_CRASHED) ||
+      (save_errno == HA_ERR_CRASHED_ON_USAGE) ||
+      (save_errno == HA_ERR_CRASHED_ON_REPAIR))
+    _ma_report_error(save_errno, share->open_file_name);
+  switch (errpos) {
+  case 6:
+    (*share->end)(&info);
+    my_free((gptr) m_info,MYF(0));
+    /* fall through */
+  case 5:
+    if (data_file < 0)
+      VOID(my_close(info.dfile.file, MYF(0)));
+    break;
+  }
+  my_errno=save_errno;
+  DBUG_RETURN (NULL);
+} /* maria_clone_internal */
+
+
+/* Make a clone of a maria table */
+
+MARIA_HA *maria_clone(MARIA_SHARE *share, int mode)
+{
+  MARIA_HA *new_info;
+  pthread_mutex_lock(&THR_LOCK_maria);
+  new_info= maria_clone_internal(share, mode,
+                                 share->data_file_type == BLOCK_RECORD ?
+                                 share->bitmap.file.file : -1);
+  pthread_mutex_unlock(&THR_LOCK_maria);
+  return new_info;
+}
+
+
 /******************************************************************************
-  open a MARIA database.
+  open a MARIA table
+
   See my_base.h for the handle_locking argument
   if handle_locking and HA_OPEN_ABORT_IF_CRASHED then abort if the table
   is marked crashed or if we are not using locking and the table doesn't
@@ -82,7 +237,7 @@ MARIA_HA *_ma_test_if_reopen(char *filename)
 
 MARIA_HA *maria_open(const char *name, int mode, uint open_flags)
 {
-  int kfile,open_mode,save_errno,have_rtree=0;
+  int kfile,open_mode,save_errno;
   uint i,j,len,errpos,head_length,base_pos,info_length,keys,
     key_parts,unique_key_parts,fulltext_keys,uniques;
   char name_buff[FN_REFLEN], org_name[FN_REFLEN], index_name[FN_REFLEN],
@@ -93,6 +248,7 @@ MARIA_HA *maria_open(const char *name, int mode, uint open_flags)
   ulong rec_per_key_part[HA_MAX_POSSIBLE_KEY*HA_MAX_KEY_SEG];
   my_off_t key_root[HA_MAX_POSSIBLE_KEY];
   ulonglong max_key_file_length, max_data_file_length;
+  File data_file= -1;
   DBUG_ENTER("maria_open");
 
   LINT_INIT(m_info);
@@ -288,6 +444,7 @@ MARIA_HA *maria_open(const char *name, int mode, uint open_flags)
 			 &share->unique_file_name,strlen(name_buff)+1,
 			 &share->index_file_name,strlen(index_name)+1,
 			 &share->data_file_name,strlen(data_name)+1,
+                         &share->open_file_name,strlen(name)+1,
 			 &share->state.key_root,keys*sizeof(my_off_t),
 #ifdef THREAD
 			 &share->key_root_lock,sizeof(rw_lock_t)*keys,
@@ -306,6 +463,7 @@ MARIA_HA *maria_open(const char *name, int mode, uint open_flags)
     share->unique_name_length= strlen(name_buff);
     strmov(share->index_file_name,  index_name);
     strmov(share->data_file_name,   data_name);
+    strmov(share->open_file_name,   name);
 
     share->block_size= share->base.block_size;
     {
@@ -317,7 +475,7 @@ MARIA_HA *maria_open(const char *name, int mode, uint open_flags)
         disk_pos_assert(disk_pos + share->keyinfo[i].keysegs * HA_KEYSEG_SIZE,
  			end_pos);
         if (share->keyinfo[i].key_alg == HA_KEY_ALG_RTREE)
-          have_rtree=1;
+          share->have_rtree= 1;
 	share->keyinfo[i].seg=pos;
 	for (j=0 ; j < share->keyinfo[i].keysegs; j++,pos++)
 	{
@@ -458,35 +616,14 @@ MARIA_HA *maria_open(const char *name, int mode, uint open_flags)
       }
     }
     share->columndef[i].type=(int) FIELD_LAST;	/* End marker */
-#ifdef ASKMONTY
-    /*
-      This code was added to mi_open.c in this cset:
-      "ChangeSet 1.1616.2941.5 2007/01/22 16:34:58 svoj@mysql.com
-      BUG#24401 - MySQL server crashes if you try to retrieve data from
-      corrupted table
-      Accessing a table with corrupted column definition results in server
-      crash.
-      This is fixed by refusing to open such tables. Affects MyISAM only.
-      No test case, since it requires crashed table.
-      storage/myisam/mi_open.c 1.80.2.10 2007/01/22 16:34:57 svoj@mysql.com
-      Refuse to open MyISAM table with summary columns length bigger than
-      length of the record."
-
-      The problem is that the "offset" variable was removed (by Monty in the
-      rows-in-block patch). Monty will know how to merge that.
-      Guilhem will make sure to notify him.
-    */
-    if (offset > share->base.reclength)
+
+    if ((share->data_file_type == BLOCK_RECORD ||
+         share->data_file_type == COMPRESSED_RECORD))
     {
-      /* purecov: begin inspected */
-      my_errno= HA_ERR_CRASHED;
-      goto err;
-      /* purecov: end */
+      if (_ma_open_datafile(&info, share, -1))
+        goto err;
+      data_file= info.dfile.file;
     }
-#endif /* ASKMONTY */
-
-    if (_ma_open_datafile(&info, share, -1))
-      goto err;
     errpos= 5;
 
     share->kfile.file= kfile;
@@ -522,6 +659,13 @@ MARIA_HA *maria_open(const char *name, int mode, uint open_flags)
       }
     }
     share->is_log_table= FALSE;
+    if (open_flags & HA_OPEN_TMP_TABLE) 
+      share->options|= HA_OPTION_TMP_TABLE;
+    if (open_flags & HA_OPEN_DELAY_KEY_WRITE)
+      share->options|= HA_OPTION_DELAY_KEY_WRITE;
+    if (mode == O_RDONLY)
+      share->options|= HA_OPTION_READ_ONLY_DATA;
+
 #ifdef THREAD
     thr_lock_init(&share->lock);
     VOID(pthread_mutex_init(&share->intern_lock,MY_MUTEX_INIT_FAST));
@@ -541,7 +685,7 @@ MARIA_HA *maria_open(const char *name, int mode, uint open_flags)
 			   HA_OPTION_TEMP_COMPRESS_RECORD)) ||
 	 (open_flags & HA_OPEN_TMP_TABLE) ||
          share->data_file_type == BLOCK_RECORD ||
-	 have_rtree) ? 0 : 1;
+	 share->have_rtree) ? 0 : 1;
       if (share->concurrent_insert)
       {
 	share->lock.get_status=_ma_get_status;
@@ -556,106 +700,12 @@ MARIA_HA *maria_open(const char *name, int mode, uint open_flags)
   else
   {
     share= old_info->s;
-    if (mode == O_RDWR && share->mode == O_RDONLY)
-    {
-      my_errno=EACCES;				/* Can't open in write mode */
-      goto err;
-    }
     if (share->data_file_type == BLOCK_RECORD)
-      info.dfile= share->bitmap.file;
-    else if (_ma_open_datafile(&info, share, old_info->dfile.file))
-      goto err;
-    errpos= 5;
-    have_rtree= old_info->maria_rtree_recursion_state != NULL;
-  }
-
-  /* alloc and set up private structure parts */
-  if (!my_multi_malloc(MY_WME,
-		       &m_info,sizeof(MARIA_HA),
-		       &info.blobs,sizeof(MARIA_BLOB)*share->base.blobs,
-		       &info.buff,(share->base.max_key_block_length*2+
-				   share->base.max_key_length),
-		       &info.lastkey,share->base.max_key_length*3+1,
-		       &info.first_mbr_key, share->base.max_key_length,
-		       &info.filename,strlen(name)+1,
-		       &info.maria_rtree_recursion_state,have_rtree ? 1024 : 0,
-		       NullS))
-    goto err;
-  errpos= 6;
-
-  if (!have_rtree)
-    info.maria_rtree_recursion_state= NULL;
-
-  strmov(info.filename,name);
-  memcpy(info.blobs,share->blobs,sizeof(MARIA_BLOB)*share->base.blobs);
-  info.lastkey2=info.lastkey+share->base.max_key_length;
-
-  info.s=share;
-  info.cur_row.lastpos= HA_OFFSET_ERROR;
-  info.update= (short) (HA_STATE_NEXT_FOUND+HA_STATE_PREV_FOUND);
-  info.opt_flag=READ_CHECK_USED;
-  info.this_unique= (ulong) info.dfile.file; /* Uniq number in process */
-  if (share->data_file_type == COMPRESSED_RECORD)
-    info.this_unique= share->state.unique;
-  info.this_loop=0;				/* Update counter */
-  info.last_unique= share->state.unique;
-  info.last_loop=   share->state.update_count;
-  if (mode == O_RDONLY)
-    share->options|=HA_OPTION_READ_ONLY_DATA;
-  info.lock_type=F_UNLCK;
-  info.quick_mode=0;
-  info.bulk_insert=0;
-  info.ft1_to_ft2=0;
-  info.errkey= -1;
-  info.page_changed=1;
-  info.keyread_buff= info.buff + share->base.max_key_block_length;
-  if ((*share->init)(&info))
-    goto err;
-
-  pthread_mutex_lock(&share->intern_lock);
-  info.read_record= share->read_record;
-  share->reopen++;
-  share->write_flag=MYF(MY_NABP | MY_WAIT_IF_FULL);
-  if (share->options & HA_OPTION_READ_ONLY_DATA)
-  {
-    info.lock_type=F_RDLCK;
-    share->r_locks++;
-    share->tot_locks++;
+      data_file= share->bitmap.file.file;       /* Only opened once */
   }
-  if ((open_flags & HA_OPEN_TMP_TABLE) ||
-      (share->options & HA_OPTION_TMP_TABLE))
-  {
-    share->temporary= share->delay_key_write= 1;
 
-    share->write_flag=MYF(MY_NABP);
-    share->w_locks++;			/* We don't have to update status */
-    share->tot_locks++;
-    info.lock_type=F_WRLCK;
-  }
-  if (((open_flags & HA_OPEN_DELAY_KEY_WRITE) ||
-      (share->options & HA_OPTION_DELAY_KEY_WRITE)) &&
-      maria_delay_key_write)
-    share->delay_key_write=1;
-
-  info.state= &share->state.state;	/* Change global values by default */
-  info.trn=   &dummy_transaction_object;
-  pthread_mutex_unlock(&share->intern_lock);
-
-  /* Allocate buffer for one record */
-  /* prerequisites: info->rec_buffer == 0 && info->rec_buff_size == 0 */
-  if (_ma_alloc_buffer(&info.rec_buff, &info.rec_buff_size,
-                       share->base.default_rec_buff_size))
+  if (!(m_info= maria_clone_internal(share, mode, data_file)))
     goto err;
-
-  bzero(info.rec_buff, share->base.default_rec_buff_size);
-
-  *m_info=info;
-#ifdef THREAD
-  thr_lock_data_init(&share->lock,&m_info->lock,(void*) m_info);
-#endif
-  m_info->open_list.data=(void*) m_info;
-  maria_open_list=list_add(maria_open_list,&m_info->open_list);
-
   pthread_mutex_unlock(&THR_LOCK_maria);
   DBUG_RETURN(m_info);
 
@@ -666,13 +716,9 @@ MARIA_HA *maria_open(const char *name, int mode, uint open_flags)
       (save_errno == HA_ERR_CRASHED_ON_REPAIR))
     _ma_report_error(save_errno, name);
   switch (errpos) {
-  case 6:
-    (*share->end)(&info);
-    my_free((gptr) m_info,MYF(0));
-    /* fall through */
   case 5:
-    if (share->data_file_type != BLOCK_RECORD)
-      VOID(my_close(info.dfile.file, MYF(0)));
+    if (data_file >= 0)
+      VOID(my_close(data_file, MYF(0)));
     if (old_info)
       break;					/* Don't remove open table */
     (*share->once_end)(share);    
@@ -693,7 +739,7 @@ MARIA_HA *maria_open(const char *name, int mode, uint open_flags)
     break;
   }
   pthread_mutex_unlock(&THR_LOCK_maria);
-  my_errno=save_errno;
+  my_errno= save_errno;
   DBUG_RETURN (NULL);
 } /* maria_open */
 

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 */
 };