merge

27 files changed, 267 insertions, 184 deletions

diff --git a/storage/maria/ha_maria.cc b/storage/maria/ha_maria.cc
index 78edf7d6545..62a813415b7 100644
--- a/storage/maria/ha_maria.cc
+++ b/storage/maria/ha_maria.cc
@@ -44,6 +44,7 @@ C_MODE_END
 #ifdef MARIA_CANNOT_ROLLBACK
 #define trans_register_ha(A, B, C)  do { /* nothing */ } while(0)
 #endif
+#define THD_TRN (*(TRN **)thd_ha_data(thd, maria_hton))
 
 ulong pagecache_division_limit, pagecache_age_threshold;
 ulonglong pagecache_buffer_size;
@@ -1278,7 +1279,7 @@ int ha_maria::zerofill(THD * thd, HA_CHECK_OPT *check_opt)
   param.op_name= "zerofill";
   param.testflag= check_opt->flags | T_SILENT | T_ZEROFILL;
   param.sort_buffer_length= THDVAR(thd, sort_buffer_size);
-  error=maria_zerofill(&param, file, share->open_file_name);
+  error=maria_zerofill(&param, file, share->open_file_name.str);
 
   if (!error)
   {
@@ -1355,7 +1356,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, share->open_file_name);
+  strmov(fixed_name, share->open_file_name.str);
 
   // Don't lock tables if we have used LOCK TABLE
   if (!thd->locked_tables &&
@@ -1958,10 +1959,18 @@ bool ha_maria::is_crashed() const
           (my_disable_locking && file->s->state.open_count));
 }
 
+#define CHECK_UNTIL_WE_FULLY_IMPLEMENTED_VERSIONING(msg) \
+  do { \
+    if (file->lock.type == TL_WRITE_CONCURRENT_INSERT) \
+    { \
+      my_error(ER_CHECK_NOT_IMPLEMENTED, MYF(0), msg); \
+      return 1; \
+    } \
+  } while(0)
 
 int ha_maria::update_row(const uchar * old_data, uchar * new_data)
 {
-  DBUG_ASSERT(file->lock.type != TL_WRITE_CONCURRENT_INSERT);
+  CHECK_UNTIL_WE_FULLY_IMPLEMENTED_VERSIONING("UPDATE in WRITE CONCURRENT");
   ha_statistic_increment(&SSV::ha_update_count);
   if (table->timestamp_field_type & TIMESTAMP_AUTO_SET_ON_UPDATE)
     table->timestamp_field->set_time();
@@ -1971,7 +1980,7 @@ int ha_maria::update_row(const uchar * old_data, uchar * new_data)
 
 int ha_maria::delete_row(const uchar * buf)
 {
-  DBUG_ASSERT(file->lock.type != TL_WRITE_CONCURRENT_INSERT);
+  CHECK_UNTIL_WE_FULLY_IMPLEMENTED_VERSIONING("DELETE in WRITE CONCURRENT");
   ha_statistic_increment(&SSV::ha_delete_count);
   return maria_delete(file, buf);
 }
@@ -2175,11 +2184,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->s->open_file_name, "", MARIA_NAME_DEXT,
+    fn_format(name_buff, file->s->open_file_name.str, "", 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->s->open_file_name, "", MARIA_NAME_IEXT,
+      data_file_name =maria_info.data_file_name;
+    fn_format(name_buff, file->s->open_file_name.str, "", 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;
@@ -2201,6 +2210,21 @@ int ha_maria::extra(enum ha_extra_function operation)
 {
   if ((specialflag & SPECIAL_SAFE_MODE) && operation == HA_EXTRA_KEYREAD)
     return 0;
+
+  /*
+    We have to set file->trn here because in some cases we call
+    extern_lock(F_UNLOCK) (which resets file->trn) followed by maria_close()
+    without calling commit/rollback in between.  If file->trn is not set
+    we can't remove file->share from the transaction list in the extra() call.
+  */
+
+  if (!file->trn &&
+      (operation == HA_EXTRA_PREPARE_FOR_DROP ||
+       operation == HA_EXTRA_PREPARE_FOR_RENAME))
+  {
+    THD *thd= table->in_use;
+    file->trn= THD_TRN;
+  }
   return maria_extra(file, operation, 0);
 }
 
@@ -2240,8 +2264,6 @@ int ha_maria::delete_table(const char *name)
   return maria_delete_table(name);
 }
 
-#define THD_TRN (*(TRN **)thd_ha_data(thd, maria_hton))
-
 int ha_maria::external_lock(THD *thd, int lock_type)
 {
   TRN *trn= THD_TRN;
diff --git a/storage/maria/ma_blockrec.c b/storage/maria/ma_blockrec.c
index 71a81edb680..a1e3fabfcca 100644
--- a/storage/maria/ma_blockrec.c
+++ b/storage/maria/ma_blockrec.c
@@ -3073,7 +3073,7 @@ static my_bool write_block_record(MARIA_HA *info,
     }
     else
     {
-      if (my_multi_malloc(MY_WME, &log_array,
+      if (!my_multi_malloc(MY_WME, &log_array,
                           (uint) ((bitmap_blocks->count +
                                    TRANSLOG_INTERNAL_PARTS + 2) *
                                   sizeof(*log_array)),
diff --git a/storage/maria/ma_check.c b/storage/maria/ma_check.c
index fa987673914..c64c879992c 100644
--- a/storage/maria/ma_check.c
+++ b/storage/maria/ma_check.c
@@ -2476,7 +2476,7 @@ int maria_repair(HA_CHECK *param, register MARIA_HA *info,
   {
     /* Get real path for data file */
     if ((new_file= my_create(fn_format(param->temp_filename,
-                                       share->data_file_name, "",
+                                       share->data_file_name.str, "",
                                        DATA_TMP_EXT, 2+4),
                              0,param->tmpfile_createflag,
                              MYF(0))) < 0)
@@ -2680,7 +2680,7 @@ int maria_repair(HA_CHECK *param, register MARIA_HA *info,
       my_close(new_file, MYF(MY_WME));
     new_file= -1;
     change_data_file_descriptor(info, -1);
-    if (maria_change_to_newfile(share->data_file_name,MARIA_NAME_DEXT,
+    if (maria_change_to_newfile(share->data_file_name.str, MARIA_NAME_DEXT,
                                 DATA_TMP_EXT,
                                 (param->testflag & T_BACKUP_DATA ?
                                  MYF(MY_REDEL_MAKE_BACKUP): MYF(0)) |
@@ -2993,7 +2993,7 @@ int maria_sort_index(HA_CHECK *param, register MARIA_HA *info, char *name)
   share->kfile.file = -1;
   pthread_mutex_unlock(&share->intern_lock);
   VOID(my_close(new_file,MYF(MY_WME)));
-  if (maria_change_to_newfile(share->index_file_name, MARIA_NAME_IEXT,
+  if (maria_change_to_newfile(share->index_file_name.str, MARIA_NAME_IEXT,
                               INDEX_TMP_EXT, sync_dir) ||
       _ma_open_keyfile(share))
     goto err2;
@@ -3524,7 +3524,7 @@ int maria_repair_by_sort(HA_CHECK *param, register MARIA_HA *info,
   {
     /* Get real path for data file */
     if ((new_file=my_create(fn_format(param->temp_filename,
-                                      share->data_file_name, "",
+                                      share->data_file_name.str, "",
                                       DATA_TMP_EXT, 2+4),
                             0,param->tmpfile_createflag,
                             MYF(0))) < 0)
@@ -3789,7 +3789,7 @@ int maria_repair_by_sort(HA_CHECK *param, register MARIA_HA *info,
         new_file= -1;
       }
       change_data_file_descriptor(info, -1);
-      if (maria_change_to_newfile(share->data_file_name,MARIA_NAME_DEXT,
+      if (maria_change_to_newfile(share->data_file_name.str, MARIA_NAME_DEXT,
                                   DATA_TMP_EXT,
                                   (param->testflag & T_BACKUP_DATA ?
                                    MYF(MY_REDEL_MAKE_BACKUP): MYF(0)) |
@@ -4070,7 +4070,7 @@ int maria_repair_parallel(HA_CHECK *param, register MARIA_HA *info,
   {
     /* Get real path for data file */
     if ((new_file= my_create(fn_format(param->temp_filename,
-                                       share->data_file_name, "",
+                                       share->data_file_name.str, "",
                                        DATA_TMP_EXT,
                                        2+4),
                              0,param->tmpfile_createflag,
@@ -4397,7 +4397,7 @@ err:
     {
       my_close(new_file,MYF(0));
       info->dfile.file= new_file= -1;
-      if (maria_change_to_newfile(share->data_file_name,MARIA_NAME_DEXT,
+      if (maria_change_to_newfile(share->data_file_name.str, MARIA_NAME_DEXT,
                                   DATA_TMP_EXT,
                                   MYF((param->testflag & T_BACKUP_DATA ?
                                        MY_REDEL_MAKE_BACKUP : 0) |
@@ -6236,7 +6236,7 @@ static my_bool create_new_data_handle(MARIA_SORT_PARAM *param, File new_file)
   MARIA_HA *new_info;
   DBUG_ENTER("create_new_data_handle");
 
-  if (!(sort_info->new_info= maria_open(info->s->open_file_name, O_RDWR,
+  if (!(sort_info->new_info= maria_open(info->s->open_file_name.str, O_RDWR,
                                         HA_OPEN_COPY | HA_OPEN_FOR_REPAIR)))
     DBUG_RETURN(1);
 
diff --git a/storage/maria/ma_checkpoint.c b/storage/maria/ma_checkpoint.c
index 35ae71ecb1c..91cb026d5ed 100644
--- a/storage/maria/ma_checkpoint.c
+++ b/storage/maria/ma_checkpoint.c
@@ -820,7 +820,7 @@ static int collect_tables(LEX_STRING *str, LSN checkpoint_start_log_horizon)
       */
       share->in_checkpoint= MARIA_CHECKPOINT_LOOKS_AT_ME;
       /** @todo avoid strlen() */
-      total_names_length+= strlen(share->open_file_name);
+      total_names_length+= share->open_file_name.length;
     }
   }
 
@@ -894,7 +894,7 @@ static int collect_tables(LEX_STRING *str, LSN checkpoint_start_log_horizon)
        @todo We should not look at tables which didn't change since last
        checkpoint.
     */
-    DBUG_PRINT("info",("looking at table '%s'", share->open_file_name));
+    DBUG_PRINT("info",("looking at table '%s'", share->open_file_name.str));
     if (state_copy == state_copies_end) /* we have no more cached states */
     {
       /*
@@ -978,8 +978,7 @@ static int collect_tables(LEX_STRING *str, LSN checkpoint_start_log_horizon)
     DBUG_PRINT("info", ("ignore_share: %d", ignore_share));
     if (!ignore_share)
     {
-      /** @todo avoid strlen */
-      uint open_file_name_len= strlen(share->open_file_name) + 1;
+      uint open_file_name_len= share->open_file_name.length + 1;
       /* remember the descriptors for background flush */
       *(dfiles_end++)= dfile;
       *(kfiles_end++)= kfile;
@@ -1000,7 +999,7 @@ static int collect_tables(LEX_STRING *str, LSN checkpoint_start_log_horizon)
         If no crash, maria_close() will write the exact value.
       */
       state_copy->state.first_bitmap_with_space= ~(ulonglong)0;
-      memcpy(ptr, share->open_file_name, open_file_name_len);
+      memcpy(ptr, share->open_file_name.str, open_file_name_len);
       ptr+= open_file_name_len;
       if (cmp_translog_addr(share->state.is_of_horizon,
                             checkpoint_start_log_horizon) >= 0)
diff --git a/storage/maria/ma_close.c b/storage/maria/ma_close.c
index c520fdccaa7..266a095dd52 100644
--- a/storage/maria/ma_close.c
+++ b/storage/maria/ma_close.c
@@ -33,7 +33,7 @@ int maria_close(register MARIA_HA *info)
                       (uint) share->tot_locks));
 
   /* Check that we have unlocked key delete-links properly */
-  DBUG_ASSERT(info->used_key_del == 0);
+  DBUG_ASSERT(info->key_del_used == 0);
 
   pthread_mutex_lock(&THR_LOCK_maria);
   if (info->lock_type == F_EXTRA_LCK)
@@ -69,6 +69,10 @@ int maria_close(register MARIA_HA *info)
   if (flag)
   {
     /* Last close of file; Flush everything */
+
+    /* Check that we don't have any dangling pointers from the transaction */
+    DBUG_ASSERT(share->in_trans == 0);
+
     if (share->kfile.file >= 0)
     {
       if ((*share->once_end)(share))
diff --git a/storage/maria/ma_create.c b/storage/maria/ma_create.c
index 4eb97e6578a..417397f0882 100644
--- a/storage/maria/ma_create.c
+++ b/storage/maria/ma_create.c
@@ -1348,9 +1348,9 @@ int _ma_update_state_lsns_sub(MARIA_SHARE *share, LSN lsn, TrID create_trid,
     int res;
     LEX_CUSTRING log_array[TRANSLOG_INTERNAL_PARTS + 1];
     /* table name is logged only for information */
-    log_array[TRANSLOG_INTERNAL_PARTS + 0].str= (uchar*)share->open_file_name;
+    log_array[TRANSLOG_INTERNAL_PARTS + 0].str=    share->open_file_name.str;
     log_array[TRANSLOG_INTERNAL_PARTS + 0].length=
-      strlen(share->open_file_name) + 1;
+      share->open_file_name.length + 1;
     if ((res= translog_write_record(&lsn, LOGREC_IMPORTED_TABLE,
                                     &dummy_transaction_object, NULL,
                                     (translog_size_t)
diff --git a/storage/maria/ma_dbug.c b/storage/maria/ma_dbug.c
index 149cc74fbe9..4ae45ca9b7a 100644
--- a/storage/maria/ma_dbug.c
+++ b/storage/maria/ma_dbug.c
@@ -184,7 +184,7 @@ my_bool _ma_check_table_is_closed(const char *name, const char *where)
   {
     MARIA_HA *info=(MARIA_HA*) pos->data;
     MARIA_SHARE *share= info->s;
-    if (!strcmp(share->unique_file_name,filename))
+    if (!strcmp(share->unique_file_name.str, filename))
     {
       if (share->last_version)
       {
diff --git a/storage/maria/ma_delete.c b/storage/maria/ma_delete.c
index 1d6bb565e02..26cc202f542 100644
--- a/storage/maria/ma_delete.c
+++ b/storage/maria/ma_delete.c
@@ -122,8 +122,9 @@ int maria_delete(MARIA_HA *info,const uchar *record)
   allow_break();			/* Allow SIGHUP & SIGINT */
   if (info->invalidator != 0)
   {
-    DBUG_PRINT("info", ("invalidator... '%s' (delete)", share->open_file_name));
-    (*info->invalidator)(share->open_file_name);
+    DBUG_PRINT("info", ("invalidator... '%s' (delete)",
+                        share->open_file_name.str));
+    (*info->invalidator)(share->open_file_name.str);
     info->invalidator=0;
   }
   DBUG_RETURN(0);
@@ -780,6 +781,7 @@ static int underflow(register MARIA_HA *info, MARIA_KEYDEF *keyinfo,
       first_key)
   {
     size_t tmp_length;
+    uint next_page_flag;
     /* Use page right of anc-page */
     DBUG_PRINT("test",("use right page"));
 
@@ -805,6 +807,7 @@ static int underflow(register MARIA_HA *info, MARIA_KEYDEF *keyinfo,
                            DFLT_INIT_HITS, buff, 0, &next_page_link))
       goto err;
     next_buff_length= _ma_get_page_used(share, buff);
+    next_page_flag=   _ma_get_keypage_flag(share,buff);
     DBUG_DUMP("next", buff, next_buff_length);
 
     /* find keys to make a big key-page */
@@ -829,7 +832,7 @@ static int underflow(register MARIA_HA *info, MARIA_KEYDEF *keyinfo,
     _ma_store_page_used(share, buff, buff_length);
 
     /* Set page flag from combination of both key pages and parting key */
-    page_flag= (_ma_get_keypage_flag(share, buff) |
+    page_flag= (next_page_flag |
                 _ma_get_keypage_flag(share, leaf_buff));
     if (anc_key.flag & (SEARCH_USER_KEY_HAS_TRANSID |
                         SEARCH_PAGE_KEY_HAS_TRANSID))
diff --git a/storage/maria/ma_extra.c b/storage/maria/ma_extra.c
index 8c4f44df91c..7fab28edf42 100644
--- a/storage/maria/ma_extra.c
+++ b/storage/maria/ma_extra.c
@@ -326,7 +326,7 @@ int maria_extra(MARIA_HA *info, enum ha_extra_function function,
     {
       _ma_remove_table_from_trnman(share, info->trn);
       /* Ensure we don't point to the deleted data in trn */
-      info->state= &share->state.state;
+      info->state= info->state_start= &share->state.state;
     }
 
     type= do_flush ? FLUSH_RELEASE : FLUSH_IGNORE_CHANGED;
@@ -347,7 +347,7 @@ int maria_extra(MARIA_HA *info, enum ha_extra_function function,
       if (do_flush)
       {
         /* Save the state so that others can find it from disk. */
-        if (_ma_state_info_write(share, 1 | 2) ||
+        if ((share->changed && _ma_state_info_write(share, 1 | 2)) ||
             my_sync(share->kfile.file, MYF(0)))
           error= my_errno;
         else
diff --git a/storage/maria/ma_info.c b/storage/maria/ma_info.c
index 1bb351bfbb5..1bbfa3cbf7e 100644
--- a/storage/maria/ma_info.c
+++ b/storage/maria/ma_info.c
@@ -81,8 +81,8 @@ int maria_status(MARIA_HA *info, register MARIA_INFO *x, uint flag)
     x->sortkey= -1;				/* No clustering */
     x->rec_per_key	= share->state.rec_per_key_part;
     x->key_map	 	= share->state.key_map;
-    x->data_file_name   = share->data_file_name;
-    x->index_file_name  = share->index_file_name;
+    x->data_file_name   = share->data_file_name.str;
+    x->index_file_name  = share->index_file_name.str;
     x->data_file_type   = share->data_file_type;
   }
   if ((flag & HA_STATUS_TIME) && !my_fstat(info->dfile.file, &state, MYF(0)))
@@ -118,13 +118,14 @@ int maria_status(MARIA_HA *info, register MARIA_INFO *x, uint flag)
     void
 */
 
-void _ma_report_error(int errcode, const char *file_name)
+void _ma_report_error(int errcode, const LEX_STRING *name)
 {
-  size_t  length;
+  size_t length;
+  const char *file_name= name->str;
   DBUG_ENTER("_ma_report_error");
   DBUG_PRINT("enter",("errcode %d, table '%s'", errcode, file_name));
 
-  if ((length= strlen(file_name)) > 64)
+  if ((length= name->length) > 64)
   {
     /* we first remove the directory */
     size_t dir_length= dirname_length(file_name);
diff --git a/storage/maria/ma_key_recover.c b/storage/maria/ma_key_recover.c
index b7e67f1d8a7..3787e69f4a0 100644
--- a/storage/maria/ma_key_recover.c
+++ b/storage/maria/ma_key_recover.c
@@ -203,16 +203,6 @@ my_bool write_hook_for_undo_key(enum translog_record_type type,
     (struct st_msg_to_write_hook_for_undo_key *) hook_arg;
 
   *msg->root= msg->value;
-  /**
-    @todo BUG
-    so we have log mutex and then intern_lock.
-    While in checkpoint we have intern_lock and then log mutex, like when we
-    flush bitmap (flushing bitmap pages can call hook which takes log mutex);
-    and in _ma_update_state_lsns_sub() this is the same.
-    So we can deadlock.
-    Another one is that in translog_assign_id_to_share() we have intern_lock
-    and then log mutex.
-  */
   _ma_fast_unlock_key_del(tbl_info);
   return write_hook_for_undo(type, trn, tbl_info, lsn, 0);
 }
@@ -1209,14 +1199,14 @@ my_bool _ma_apply_undo_key_delete(MARIA_HA *info, LSN undo_lsn,
   @note
     To allow higher concurrency in the common case where we do inserts
     and we don't have any linked blocks we do the following:
-    - Mark in info->used_key_del that we are not using key_del
+    - Mark in info->key_del_used that we are not using key_del
     - Return at once (without marking key_del as used)
 
-    This is safe as we in this case don't write current_key_del into
+    This is safe as we in this case don't write key_del_current into
     the redo log and during recover we are not updating key_del.
 
   @retval 1  Use page at end of file
-  @retval 0  Use page at share->current_key_del
+  @retval 0  Use page at share->key_del_current
 */
 
 my_bool _ma_lock_key_del(MARIA_HA *info, my_bool insert_at_end)
@@ -1224,68 +1214,72 @@ my_bool _ma_lock_key_del(MARIA_HA *info, my_bool insert_at_end)
   MARIA_SHARE *share= info->s;
 
   /*
-    info->used_key_del is 0 initially.
+    info->key_del_used is 0 initially.
     If the caller needs a block (_ma_new()), we look at the free list:
     - looks empty? then caller will create a new block at end of file and
-    remember (through info->used_key_del==2) that it will not change
+    remember (through info->key_del_used==2) that it will not change
     state.key_del and does not need to wake up waiters as nobody will wait for
     it.
     - non-empty? then we wait for other users of the state.key_del list to
-    have finished, then we lock this list (through share->used_key_del==1)
+    have finished, then we lock this list (through share->key_del_used==1)
     because we need to prevent some other thread to also read state.key_del
-    and use the same page as ours. We remember through info->used_key_del==1
+    and use the same page as ours. We remember through info->key_del_used==1
     that we will have to set state.key_del at unlock time and wake up
     waiters.
     If the caller wants to free a block (_ma_dispose()), "empty" and
     "non-empty" are treated as "non-empty" is treated above.
-    When we are ready to unlock, we copy share->current_key_del into
+    When we are ready to unlock, we copy share->key_del_current into
     state.key_del. Unlocking happens when writing the UNDO log record, that
     can make a long lock time.
     Why we wrote "*looks* empty": because we are looking at state.key_del
-    which may be slightly old (share->current_key_del may be more recent and
+    which may be slightly old (share->key_del_current may be more recent and
     exact): when we want a new page, we tolerate to treat "there was no free
     page 1 millisecond ago"  as "there is no free page". It's ok to non-pop
     (_ma_new(), page will be found later anyway) but it's not ok to non-push
     (_ma_dispose(), page would be lost).
-    When we leave this function, info->used_key_del is always 1 or 2.
+    When we leave this function, info->key_del_used is always 1 or 2.
   */
-  if (info->used_key_del != 1)
+  if (info->key_del_used != 1)
   {
-    pthread_mutex_lock(&share->intern_lock);
+    pthread_mutex_lock(&share->key_del_lock);
     if (share->state.key_del == HA_OFFSET_ERROR && insert_at_end)
     {
-      pthread_mutex_unlock(&share->intern_lock);
-      info->used_key_del= 2;                  /* insert-with-append */
+      pthread_mutex_unlock(&share->key_del_lock);
+      info->key_del_used= 2;                  /* insert-with-append */
       return 1;
     }
 #ifdef THREAD
-    while (share->used_key_del)
-      pthread_cond_wait(&share->intern_cond, &share->intern_lock);
+    while (share->key_del_used)
+      pthread_cond_wait(&share->key_del_cond, &share->key_del_lock);
 #endif
-    info->used_key_del= 1;
-    share->used_key_del= 1;
-    share->current_key_del= share->state.key_del;
-    pthread_mutex_unlock(&share->intern_lock);
+    info->key_del_used= 1;
+    share->key_del_used= 1;
+    share->key_del_current= share->state.key_del;
+    pthread_mutex_unlock(&share->key_del_lock);
   }
-  return share->current_key_del == HA_OFFSET_ERROR;
+  return share->key_del_current == HA_OFFSET_ERROR;
 }
 
 
 /**
   @brief copy changes to key_del and unlock it
+
+  @notes
+  In case of many threads using the maria table, we always have a lock
+  on the translog when comming here.
 */
 
 void _ma_unlock_key_del(MARIA_HA *info)
 {
-  DBUG_ASSERT(info->used_key_del);
-  if (info->used_key_del == 1)                  /* Ignore insert-with-append */
+  DBUG_ASSERT(info->key_del_used);
+  if (info->key_del_used == 1)                  /* Ignore insert-with-append */
   {
     MARIA_SHARE *share= info->s;
-    pthread_mutex_lock(&share->intern_lock);
-    share->used_key_del= 0;
-    share->state.key_del= share->current_key_del;
-    pthread_mutex_unlock(&share->intern_lock);
-    pthread_cond_signal(&share->intern_cond);
+    pthread_mutex_lock(&share->key_del_lock);
+    share->key_del_used= 0;
+    share->state.key_del= share->key_del_current;
+    pthread_mutex_unlock(&share->key_del_lock);
+    pthread_cond_signal(&share->key_del_cond);
   }
-  info->used_key_del= 0;
+  info->key_del_used= 0;
 }
diff --git a/storage/maria/ma_key_recover.h b/storage/maria/ma_key_recover.h
index e8973082777..2366d53f143 100644
--- a/storage/maria/ma_key_recover.h
+++ b/storage/maria/ma_key_recover.h
@@ -114,6 +114,6 @@ extern my_bool _ma_lock_key_del(MARIA_HA *info, my_bool insert_at_end);
 extern void _ma_unlock_key_del(MARIA_HA *info);
 static inline void _ma_fast_unlock_key_del(MARIA_HA *info)
 {
-  if (info->used_key_del)
+  if (info->key_del_used)
     _ma_unlock_key_del(info);
 }
diff --git a/storage/maria/ma_keycache.c b/storage/maria/ma_keycache.c
index 9295904dbcf..39fc7d421ae 100644
--- a/storage/maria/ma_keycache.c
+++ b/storage/maria/ma_keycache.c
@@ -105,8 +105,8 @@ int maria_assign_to_pagecache(MARIA_HA *info,
   share->pagecache= pagecache;
 
   /* store the key cache in the global hash structure for future opens */
-  if (multi_pagecache_set((uchar*) share->unique_file_name,
-			  share->unique_name_length,
+  if (multi_pagecache_set((uchar*) share->unique_file_name.str,
+                          share->unique_file_name.length,
 			  share->pagecache))
     error= my_errno;
   pthread_mutex_unlock(&share->intern_lock);
diff --git a/storage/maria/ma_locking.c b/storage/maria/ma_locking.c
index 4ddfa285d37..1ad41e5b213 100644
--- a/storage/maria/ma_locking.c
+++ b/storage/maria/ma_locking.c
@@ -34,7 +34,7 @@ int maria_lock_database(MARIA_HA *info, int lock_type)
                       lock_type, info->lock_type, share->r_locks,
                       share->w_locks,
                       share->global_changed, share->state.open_count,
-                      share->index_file_name));
+                      share->index_file_name.str));
   if (share->options & HA_OPTION_READ_ONLY_DATA ||
       info->lock_type == lock_type)
     DBUG_RETURN(0);
diff --git a/storage/maria/ma_loghandler.c b/storage/maria/ma_loghandler.c
index 42937f22103..f8b264fe450 100644
--- a/storage/maria/ma_loghandler.c
+++ b/storage/maria/ma_loghandler.c
@@ -7686,7 +7686,7 @@ int translog_assign_id_to_share(MARIA_HA *tbl_info, TRN *trn)
   DBUG_ASSERT(share->data_file_type == BLOCK_RECORD);
   /* re-check under mutex to avoid having 2 ids for the same share */
   pthread_mutex_lock(&share->intern_lock);
-  if (likely(share->id == 0))
+  if (unlikely(share->id == 0))
   {
     LSN lsn;
     LEX_CUSTRING log_array[TRANSLOG_INTERNAL_PARTS + 2];
@@ -7717,13 +7717,13 @@ int translog_assign_id_to_share(MARIA_HA *tbl_info, TRN *trn)
       is not realpath-ed, etc) which is good: the log can be moved to another
       directory and continue working.
     */
-    log_array[TRANSLOG_INTERNAL_PARTS + 1].str= (uchar*)share->open_file_name;
-    /**
-       @todo if we had the name's length in MARIA_SHARE we could avoid this
-       strlen()
-    */
+    log_array[TRANSLOG_INTERNAL_PARTS + 1].str= share->open_file_name.str;
     log_array[TRANSLOG_INTERNAL_PARTS + 1].length=
-      strlen(share->open_file_name) + 1;
+      share->open_file_name.length + 1;
+    /*
+      We can't unlock share->intern_lock before the log entry is written to
+      ensure no one uses the id before it's logged.
+    */
     if (unlikely(translog_write_record(&lsn, LOGREC_FILE_ID, trn, tbl_info,
                                        (translog_size_t)
                                        (sizeof(log_data) +
diff --git a/storage/maria/ma_open.c b/storage/maria/ma_open.c
index e35b5eaac86..d1cbeb1310c 100644
--- a/storage/maria/ma_open.c
+++ b/storage/maria/ma_open.c
@@ -62,7 +62,7 @@ MARIA_HA *_ma_test_if_reopen(const char *filename)
   {
     MARIA_HA *info=(MARIA_HA*) pos->data;
     MARIA_SHARE *share= info->s;
-    if (!strcmp(share->unique_file_name,filename) && share->last_version)
+    if (!strcmp(share->unique_file_name.str,filename) && share->last_version)
       return info;
   }
   return 0;
@@ -212,7 +212,7 @@ err:
   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);
+    _ma_report_error(save_errno, &share->open_file_name);
   switch (errpos) {
   case 6:
     (*share->end)(&info);
@@ -480,6 +480,10 @@ MARIA_HA *maria_open(const char *name, int mode, uint open_flags)
     /* Add space for node pointer */
     share->base.max_key_length+= share->base.key_reflength;
 
+    share->unique_file_name.length= strlen(name_buff);
+    share->index_file_name.length=  strlen(index_name);
+    share->data_file_name.length=   strlen(data_name);
+    share->open_file_name.length=   strlen(name);
     if (!my_multi_malloc(MY_WME,
 			 &share,sizeof(*share),
 			 &share->state.rec_per_key_part,
@@ -495,10 +499,14 @@ MARIA_HA *maria_open(const char *name, int mode, uint open_flags)
 			 (share->base.fields+1)*sizeof(MARIA_COLUMNDEF),
                          &share->column_nr, share->base.fields*sizeof(uint16),
 			 &share->blobs,sizeof(MARIA_BLOB)*share->base.blobs,
-			 &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->unique_file_name.str,
+			 share->unique_file_name.length+1,
+			 &share->index_file_name.str,
+                         share->index_file_name.length+1,
+			 &share->data_file_name.str,
+                         share->data_file_name.length+1,
+                         &share->open_file_name.str,
+                         share->open_file_name.length+1,
 			 &share->state.key_root,keys*sizeof(my_off_t),
 			 &share->mmap_lock,sizeof(rw_lock_t),
 			 NullS))
@@ -512,11 +520,10 @@ MARIA_HA *maria_open(const char *name, int mode, uint open_flags)
 	   (char*) nulls_per_key_part, sizeof(long)*key_parts);
     memcpy((char*) share->state.key_root,
 	   (char*) key_root, sizeof(my_off_t)*keys);
-    strmov(share->unique_file_name, name_buff);
-    share->unique_name_length= (uint) strlen(name_buff);
-    strmov(share->index_file_name,  index_name);
-    strmov(share->data_file_name,   data_name);
-    strmov(share->open_file_name,   name);
+    strmov(share->unique_file_name.str, name_buff);
+    strmov(share->index_file_name.str, index_name);
+    strmov(share->data_file_name.str,  data_name);
+    strmov(share->open_file_name.str,  name);
 
     share->block_size= share->base.block_size;   /* Convenience */
     {
@@ -800,8 +807,9 @@ MARIA_HA *maria_open(const char *name, int mode, uint open_flags)
     }
 #ifdef THREAD
     thr_lock_init(&share->lock);
-    VOID(pthread_mutex_init(&share->intern_lock, MY_MUTEX_INIT_FAST));
-    VOID(pthread_cond_init(&share->intern_cond, 0));
+    pthread_mutex_init(&share->intern_lock, MY_MUTEX_INIT_FAST);
+    pthread_mutex_init(&share->key_del_lock, MY_MUTEX_INIT_FAST);
+    pthread_cond_init(&share->key_del_cond, 0);
     for (i=0; i<keys; i++)
       VOID(my_rwlock_init(&share->keyinfo[i].root_lock, NULL));
     VOID(my_rwlock_init(&share->mmap_lock, NULL));
@@ -886,7 +894,12 @@ err:
   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, name);
+  {
+    LEX_STRING tmp_name;
+    tmp_name.str= (char*) name;
+    tmp_name.length= strlen(name);
+    _ma_report_error(save_errno, &tmp_name);
+  }
   if (save_errno == HA_ERR_OLD_FILE) /* uuid is different ? */
     save_errno= HA_ERR_CRASHED_ON_USAGE; /* the code to trigger auto-repair */
   switch (errpos) {
@@ -1186,6 +1199,7 @@ uint _ma_state_info_write(MARIA_SHARE *share, uint pWrite)
   res= _ma_state_info_write_sub(share->kfile.file, &share->state, pWrite);
   if (pWrite & 4)
     pthread_mutex_unlock(&share->intern_lock);
+  share->changed= 0;
   return res;
 }
 
@@ -1202,6 +1216,13 @@ uint _ma_state_info_write(MARIA_SHARE *share, uint pWrite)
                            (should only be needed after ALTER TABLE
                            ENABLE/DISABLE KEYS, and REPAIR/OPTIMIZE).
 
+   @notes
+     For transactional multiuser tables, this function is called
+     with intern_lock & translog_lock or when the last thread who
+     is using the table is closing it.
+     Because of the translog_lock we don't need to have a lock on
+     key_del_lock.
+
    @return Operation status
      @retval 0      OK
      @retval 1      Error
@@ -1711,7 +1732,7 @@ int _ma_open_datafile(MARIA_HA *info, MARIA_SHARE *share,
                       File file_to_dup __attribute__((unused)))
 {
   info->dfile.file= share->bitmap.file.file=
-    my_open(share->data_file_name, share->mode | O_SHARE,
+    my_open(share->data_file_name.str, share->mode | O_SHARE,
             MYF(MY_WME));
   return info->dfile.file >= 0 ? 0 : 1;
 }
@@ -1724,7 +1745,7 @@ int _ma_open_keyfile(MARIA_SHARE *share)
     against a concurrent checkpoint.
   */
   pthread_mutex_lock(&share->intern_lock);
-  share->kfile.file= my_open(share->unique_file_name,
+  share->kfile.file= my_open(share->unique_file_name.str,
                              share->mode | O_SHARE,
                              MYF(MY_WME));
   pthread_mutex_unlock(&share->intern_lock);
diff --git a/storage/maria/ma_page.c b/storage/maria/ma_page.c
index 2cbaf3ec690..afe40aad399 100644
--- a/storage/maria/ma_page.c
+++ b/storage/maria/ma_page.c
@@ -223,8 +223,8 @@ int _ma_dispose(register MARIA_HA *info, my_off_t pos, my_bool page_not_read)
 
   (void) _ma_lock_key_del(info, 0);
 
-  old_link= share->current_key_del;
-  share->current_key_del= pos;
+  old_link= share->key_del_current;
+  share->key_del_current= pos;
   page_no= pos / block_size;
   bzero(buff, share->keypage_header);
   _ma_store_keynr(share, buff, (uchar) MARIA_DELETE_KEY_NR);
@@ -347,7 +347,7 @@ my_off_t _ma_new(register MARIA_HA *info, int level,
   else
   {
     uchar *buff;
-    pos= share->current_key_del;                /* Protected */
+    pos= share->key_del_current;                /* Protected */
     DBUG_ASSERT(share->pagecache->block_size == block_size);
     if (!(buff= pagecache_read(share->pagecache,
                                &share->kfile,
@@ -362,15 +362,15 @@ my_off_t _ma_new(register MARIA_HA *info, int level,
         (single linked list):
       */
 #ifndef DBUG_OFF
-      my_off_t current_key_del;
+      my_off_t key_del_current;
 #endif
-      share->current_key_del= mi_sizekorr(buff+share->keypage_header);
+      share->key_del_current= mi_sizekorr(buff+share->keypage_header);
 #ifndef DBUG_OFF
-      current_key_del= share->current_key_del;
-      DBUG_ASSERT(current_key_del != share->state.key_del &&
-                  (current_key_del != 0) &&
-                  ((current_key_del == HA_OFFSET_ERROR) ||
-                   (current_key_del <=
+      key_del_current= share->key_del_current;
+      DBUG_ASSERT(key_del_current != share->state.key_del &&
+                  (key_del_current != 0) &&
+                  ((key_del_current == HA_OFFSET_ERROR) ||
+                   (key_del_current <=
                     (share->state.state.key_file_length - block_size))));
 #endif
     }
diff --git a/storage/maria/ma_pagecache.c b/storage/maria/ma_pagecache.c
index 3e28835d4a8..8972ee94fc1 100644
--- a/storage/maria/ma_pagecache.c
+++ b/storage/maria/ma_pagecache.c
@@ -305,12 +305,13 @@ struct st_pagecache_block_link
   ulonglong last_hit_time; /* timestamp of the last hit                      */
   WQUEUE
     wqueue[COND_SIZE];    /* queues on waiting requests for new/old pages    */
-  uint requests;          /* number of requests for the block                */
-  uint status;            /* state of the block                              */
-  uint pins;              /* pin counter                                     */
-  uint wlocks;            /* write locks counter                             */
-  uint rlocks;            /* read locks counter                              */
-  uint rlocks_queue;      /* rd. locks waiting wr. lock of this thread       */
+  uint32 requests;        /* number of requests for the block                */
+  uint32 pins;            /* pin counter                                     */
+  uint32 wlocks;          /* write locks counter                             */
+  uint32 rlocks;          /* read locks counter                              */
+  uint32 rlocks_queue;    /* rd. locks waiting wr. lock of this thread       */
+  uint16 status;          /* state of the block                              */
+  int16  error;           /* error code for block in case of error */
   enum PCBLOCK_TEMPERATURE temperature; /* block temperature: cold, warm, hot*/
   enum pagecache_page_type type; /* type of the block                        */
   uint hits_left;         /* number of hits left until promotion             */
@@ -592,16 +593,16 @@ extern my_bool translog_flush(TRANSLOG_ADDRESS lsn);
 
   RETURN
     0   - OK
-    !=0 - Error
+    1   - Error
 */
 
-static uint pagecache_fwrite(PAGECACHE *pagecache,
-                             PAGECACHE_FILE *filedesc,
-                             uchar *buffer,
-                             pgcache_page_no_t pageno,
-                             enum pagecache_page_type type
-                             __attribute__((unused)),
-                             myf flags)
+static my_bool pagecache_fwrite(PAGECACHE *pagecache,
+                                PAGECACHE_FILE *filedesc,
+                                uchar *buffer,
+                                pgcache_page_no_t pageno,
+                                enum pagecache_page_type type
+                                __attribute__((unused)),
+                                myf flags)
 {
   DBUG_ENTER("pagecache_fwrite");
   DBUG_ASSERT(type != PAGECACHE_READ_UNKNOWN_PAGE);
@@ -2068,6 +2069,7 @@ restart:
                             (my_bool)(block->hash_link ? 1 : 0));
           PCBLOCK_INFO(block);
           block->status= error ? PCBLOCK_ERROR : 0;
+          block->error=  (int16) my_errno;
 #ifndef DBUG_OFF
           block->type= PAGECACHE_EMPTY_PAGE;
           if (error)
@@ -2606,6 +2608,7 @@ static void read_block(PAGECACHE *pagecache,
     if (error)
     {
       block->status|= PCBLOCK_ERROR;
+      block->error=   (int16) my_errno;
       my_debug_put_break_here();
     }
     else
@@ -2618,6 +2621,7 @@ static void read_block(PAGECACHE *pagecache,
       {
         DBUG_PRINT("error", ("read callback problem"));
         block->status|= PCBLOCK_ERROR;
+        block->error=  (int16) my_errno;
         my_debug_put_break_here();
       }
     }
@@ -3107,7 +3111,7 @@ uchar *pagecache_read(PAGECACHE *pagecache,
                       enum pagecache_page_lock lock,
                       PAGECACHE_BLOCK_LINK **page_link)
 {
-  int error= 0;
+  my_bool error= 0;
   enum pagecache_page_pin pin= lock_to_pin[test(buff==0)][lock];
   PAGECACHE_BLOCK_LINK *fake_link;
   my_bool reg_request;
@@ -3229,6 +3233,8 @@ restart:
         pagecache_pthread_mutex_lock(&pagecache->cache_lock);
 #endif
       }
+      else
+        my_errno= block->error;
     }
 
     remove_reader(block);
@@ -3286,7 +3292,7 @@ static my_bool pagecache_delete_internal(PAGECACHE *pagecache,
                                          PAGECACHE_HASH_LINK *page_link,
                                          my_bool flush)
 {
-  int error= 0;
+  my_bool error= 0;
   if (block->status & PCBLOCK_CHANGED)
   {
     if (flush)
@@ -3313,6 +3319,7 @@ static my_bool pagecache_delete_internal(PAGECACHE *pagecache,
       if (error)
       {
         block->status|= PCBLOCK_ERROR;
+        block->error=   (int16) my_errno;
         my_debug_put_break_here();
         goto err;
       }
@@ -3361,7 +3368,7 @@ my_bool pagecache_delete_by_link(PAGECACHE *pagecache,
                                  enum pagecache_page_lock lock,
                                  my_bool flush)
 {
-  int error= 0;
+  my_bool error= 0;
   enum pagecache_page_pin pin= PAGECACHE_PIN_LEFT_PINNED;
   DBUG_ENTER("pagecache_delete_by_link");
   DBUG_PRINT("enter", ("fd: %d block 0x%lx  %s  %s",
@@ -3460,7 +3467,7 @@ my_bool pagecache_delete(PAGECACHE *pagecache,
                          enum pagecache_page_lock lock,
                          my_bool flush)
 {
-  int error= 0;
+  my_bool error= 0;
   enum pagecache_page_pin pin= lock_to_pin[0][lock];
   DBUG_ENTER("pagecache_delete");
   DBUG_PRINT("enter", ("fd: %u  page: %lu  %s  %s",
@@ -3648,7 +3655,7 @@ my_bool pagecache_write_part(PAGECACHE *pagecache,
 {
   PAGECACHE_BLOCK_LINK *block= NULL;
   PAGECACHE_BLOCK_LINK *fake_link;
-  int error= 0;
+  my_bool error= 0;
   int need_lock_change= write_lock_change_table[lock].need_lock_change;
   my_bool reg_request;
 #ifndef DBUG_OFF
@@ -3771,6 +3778,7 @@ restart:
         {
           DBUG_PRINT("error", ("read callback problem"));
           block->status|= PCBLOCK_ERROR;
+          block->error= (int16) my_errno;
           my_debug_put_break_here();
         }
         KEYCACHE_DBUG_PRINT("key_cache_insert",
@@ -3860,10 +3868,10 @@ no_key_cache:
       uchar *page_buffer= (uchar *) alloca(pagecache->block_size);
 
       pagecache->global_cache_read++;
-      if ((error= pagecache_fread(pagecache, file,
-                                  page_buffer,
-                                  pageno,
-                                  pagecache->readwrite_flags)))
+      if ((error= (pagecache_fread(pagecache, file,
+                                   page_buffer,
+                                   pageno,
+                                   pagecache->readwrite_flags) != 0)))
         goto end;
       if ((file->read_callback)(page_buffer, pageno, file->callback_data))
       {
@@ -3987,7 +3995,7 @@ static int flush_cached_blocks(PAGECACHE *pagecache,
                                int *first_errno)
 {
   int rc= PCFLUSH_OK;
-  int error;
+  my_bool error;
   uint count= (uint) (end-cache);
   DBUG_ENTER("flush_cached_blocks");
   *first_errno= 0;
@@ -4067,6 +4075,7 @@ static int flush_cached_blocks(PAGECACHE *pagecache,
     if (error)
     {
       block->status|= PCBLOCK_ERROR;
+      block->error=   (int16) my_errno;
       my_debug_put_break_here();
       if (!*first_errno)
         *first_errno= my_errno ? my_errno : -1;
diff --git a/storage/maria/ma_recovery.c b/storage/maria/ma_recovery.c
index ad75a84aa1f..26d29aec4be 100644
--- a/storage/maria/ma_recovery.c
+++ b/storage/maria/ma_recovery.c
@@ -866,7 +866,7 @@ prototype_redo_exec_hook(REDO_RENAME_TABLE)
       ALERT_USER();
       goto end;
     }
-    if (close_one_table(info->s->open_file_name, rec->lsn) ||
+    if (close_one_table(info->s->open_file_name.str, rec->lsn) ||
         maria_close(info))
       goto end;
     info= NULL;
@@ -1008,7 +1008,7 @@ prototype_redo_exec_hook(REDO_REPAIR_TABLE)
   tprint(tracef, "   repairing...\n");
 
   maria_chk_init(&param);
-  param.isam_file_name= name= info->s->open_file_name;
+  param.isam_file_name= name= info->s->open_file_name.str;
   param.testflag= uint8korr(rec->header + FILEID_STORE_SIZE);
   param.tmpdir= maria_tmpdir;
   DBUG_ASSERT(maria_tmpdir);
@@ -1085,7 +1085,7 @@ prototype_redo_exec_hook(REDO_DROP_TABLE)
       ALERT_USER();
       goto end;
     }
-    if (close_one_table(info->s->open_file_name, rec->lsn) ||
+    if (close_one_table(info->s->open_file_name.str, rec->lsn) ||
         maria_close(info))
       goto end;
     info= NULL;
@@ -1141,7 +1141,7 @@ prototype_redo_exec_hook(FILE_ID)
   info= all_tables[sid].info;
   if (info != NULL)
   {
-    tprint(tracef, "   Closing table '%s'\n", info->s->open_file_name);
+    tprint(tracef, "   Closing table '%s'\n", info->s->open_file_name.str);
     prepare_table_for_close(info, rec->lsn);
     if (maria_close(info))
     {
@@ -1201,7 +1201,7 @@ static int new_table(uint16 sid, const char *name, LSN lsn_of_file_id)
       It could be that we have in the log
       FILE_ID(t1,10) ... (t1 was flushed) ... FILE_ID(t1,12);
     */
-    if (close_one_table(share->open_file_name, lsn_of_file_id))
+    if (close_one_table(share->open_file_name.str, lsn_of_file_id))
       goto end;
   }
   if (!share->base.born_transactional)
@@ -1230,7 +1230,7 @@ static int new_table(uint16 sid, const char *name, LSN lsn_of_file_id)
   if (maria_is_crashed(info))
   {
     eprint(tracef, "Table '%s' is crashed, skipping it. Please repair it with"
-           " maria_chk -r", share->open_file_name);
+           " maria_chk -r", share->open_file_name.str);
     error= -1; /* not fatal, try with other tables */
     goto end;
     /*
@@ -2828,7 +2828,7 @@ static MARIA_HA *get_MARIA_HA_from_REDO_record(const
     return NULL;
   }
   share= info->s;
-  tprint(tracef, ", '%s'", share->open_file_name);
+  tprint(tracef, ", '%s'", share->open_file_name.str);
   DBUG_ASSERT(in_redo_phase);
   if (cmp_translog_addr(rec->lsn, share->lsn_of_file_id) <= 0)
   {
@@ -2899,7 +2899,7 @@ static MARIA_HA *get_MARIA_HA_from_UNDO_record(const
     return NULL;
   }
   share= info->s;
-  tprint(tracef, ", '%s'", share->open_file_name);
+  tprint(tracef, ", '%s'", share->open_file_name.str);
   if (cmp_translog_addr(rec->lsn, share->lsn_of_file_id) <= 0)
   {
     tprint(tracef, ", table's LOGREC_FILE_ID has LSN (%lu,0x%lx) more recent"
@@ -3188,7 +3188,7 @@ static my_bool close_one_table(const char *name, TRANSLOG_ADDRESS addr)
        internal_table++)
   {
     MARIA_HA *info= internal_table->info;
-    if ((info != NULL) && !strcmp(info->s->open_file_name, name))
+    if ((info != NULL) && !strcmp(info->s->open_file_name.str, name))
     {
       prepare_table_for_close(info, addr);
       if (maria_close(info))
diff --git a/storage/maria/ma_search.c b/storage/maria/ma_search.c
index 98bac11e8d4..cf7ec5f9724 100644
--- a/storage/maria/ma_search.c
+++ b/storage/maria/ma_search.c
@@ -116,7 +116,8 @@ int _ma_search(register MARIA_HA *info, MARIA_KEY *key, uint32 nextflag,
 	((keyinfo->flag & (HA_NOSAME | HA_NULL_PART)) != HA_NOSAME ||
 	 (key->flag & SEARCH_PART_KEY) || info->s->base.born_transactional))
     {
-      if ((error= _ma_search(info, key, nextflag,
+      if ((error= _ma_search(info, key, (nextflag | SEARCH_FIND) &
+                             ~(SEARCH_BIGGER | SEARCH_SMALLER | SEARCH_LAST),
                              _ma_kpos(nod_flag,keypos))) >= 0 ||
           my_errno != HA_ERR_KEY_NOT_FOUND)
         DBUG_RETURN(error);
@@ -338,10 +339,8 @@ int _ma_seq_search(const MARIA_KEY *key, uchar *page,
                           comp_flag | tmp_key.flag,
                           not_used)) >= 0)
       break;
-#ifdef EXTRA_DEBUG
-    DBUG_PRINT("loop",("page: 0x%lx  key: '%s'  flag: %d", (long) page, t_buff,
-                       flag));
-#endif
+    DBUG_PRINT("loop_extra",("page: 0x%lx  key: '%s'  flag: %d",
+                             (long) page, t_buff, flag));
     memcpy(buff,t_buff,length);
     *ret_pos=page;
   }
@@ -2195,9 +2194,10 @@ int _ma_calc_bin_pack_key_length(const MARIA_KEY *int_key,
 
     if (next_length > ref_length)
     {
-      /* We put a key with different case between two keys with the same prefix
-         Extend next key to have same prefix as
-         this key */
+      /*
+        We put a key with different case between two keys with the same prefix
+        Extend next key to have same prefix as this key
+      */
       s_temp->n_ref_length= ref_length;
       s_temp->prev_length=  next_length-ref_length;
       s_temp->prev_key+=    ref_length;
@@ -2207,13 +2207,13 @@ int _ma_calc_bin_pack_key_length(const MARIA_KEY *int_key,
     }
     /* Check how many characters are identical to next key */
     key= s_temp->key+next_length;
+    s_temp->prev_length= 0;
     while (*key++ == *next_key++) ;
     if ((ref_length= (uint) (key - s_temp->key)-1) == next_length)
     {
       s_temp->next_key_pos=0;
       return (s_temp->move_length= length);  /* Can't pack next key */
     }
-    s_temp->prev_length=0;
     s_temp->n_ref_length=ref_length;
     return s_temp->move_length= (int) (length-(ref_length - next_length) -
                                        next_length_pack +
diff --git a/storage/maria/ma_state.c b/storage/maria/ma_state.c
index 422e35795bd..cfe0de62929 100644
--- a/storage/maria/ma_state.c
+++ b/storage/maria/ma_state.c
@@ -79,6 +79,9 @@ my_bool _ma_setup_live_state(MARIA_HA *info)
 
   pthread_mutex_lock(&share->intern_lock);
   share->in_trans++;
+  DBUG_PRINT("info", ("share: 0x%lx  in_trans: %d",
+                      (ulong) share, share->in_trans));
+
   history= share->state_history;
 
   /*
@@ -359,15 +362,16 @@ my_bool _ma_trnman_end_trans_hook(TRN *trn, my_bool commit,
 {
   my_bool error= 0;
   MARIA_USED_TABLES *tables, *next;
+  DBUG_ENTER("_ma_trnman_end_trans_hook");
   
   for (tables= (MARIA_USED_TABLES*) trn->used_tables;
        tables;
        tables= next)
   {
+    MARIA_SHARE *share= tables->share;
     next= tables->next;
     if (commit)
     {
-      MARIA_SHARE *share= tables->share;
       MARIA_STATE_HISTORY *history;
 
       pthread_mutex_lock(&share->intern_lock);
@@ -405,13 +409,27 @@ my_bool _ma_trnman_end_trans_hook(TRN *trn, my_bool commit,
         /* Remove not visible states */
         share->state_history= _ma_remove_not_visible_states(history, 0, 1);
       }
+      DBUG_PRINT("info", ("share: 0x%lx  in_trans: %d",
+                          (ulong) share, share->in_trans));
+      share->in_trans--;
+      pthread_mutex_unlock(&share->intern_lock);
+    }
+    else
+    {
+#ifndef DBUG_OFF
+      /*
+        We need to keep share->in_trans correct in the debug library
+        because of the assert in maria_close()
+      */
+      pthread_mutex_lock(&share->intern_lock);
       share->in_trans--;
       pthread_mutex_unlock(&share->intern_lock);
+#endif
     }
     my_free(tables, MYF(0));
   }
   trn->used_tables= 0;
-  return error;
+  DBUG_RETURN(error);
 }
 
 
@@ -421,11 +439,18 @@ my_bool _ma_trnman_end_trans_hook(TRN *trn, my_bool commit,
    @notes
      This is used when we unlock a table from a group of locked tables
      just before doing a rename or drop table.
+
+     share->internal_lock must be locked when function is called
 */
 
 void _ma_remove_table_from_trnman(MARIA_SHARE *share, TRN *trn)
 {
   MARIA_USED_TABLES *tables, **prev;
+  DBUG_ENTER("_ma_remove_table_from_trnman");
+  DBUG_PRINT("enter", ("share: 0x%lx  in_trans: %d",
+                       (ulong) share, share->in_trans));
+
+  safe_mutex_assert_owner(&share->intern_lock);
   
   for (prev= (MARIA_USED_TABLES**) &trn->used_tables, tables= *prev;
        tables;
@@ -434,11 +459,14 @@ void _ma_remove_table_from_trnman(MARIA_SHARE *share, TRN *trn)
     if (tables->share == share)
     {
       *prev= tables->next;
+      share->in_trans--;
+      DBUG_PRINT("info", ("in_trans: %d", share->in_trans));
       my_free(tables, MYF(0));
+      break;
     }
-    else
-      prev= &tables->next;
+    prev= &tables->next;
   }
+  DBUG_VOID_RETURN;
 }
 
 
diff --git a/storage/maria/ma_update.c b/storage/maria/ma_update.c
index 29246d2d07e..a178f5d47a7 100644
--- a/storage/maria/ma_update.c
+++ b/storage/maria/ma_update.c
@@ -188,8 +188,9 @@ int maria_update(register MARIA_HA *info, const uchar *oldrec, uchar *newrec)
   allow_break();				/* Allow SIGHUP & SIGINT */
   if (info->invalidator != 0)
   {
-    DBUG_PRINT("info", ("invalidator... '%s' (update)", share->open_file_name));
-    (*info->invalidator)(share->open_file_name);
+    DBUG_PRINT("info", ("invalidator... '%s' (update)",
+                        share->open_file_name.str));
+    (*info->invalidator)(share->open_file_name.str);
     info->invalidator=0;
   }
   DBUG_RETURN(0);
diff --git a/storage/maria/ma_write.c b/storage/maria/ma_write.c
index 8add20d46ef..fb380b6d3e2 100644
--- a/storage/maria/ma_write.c
+++ b/storage/maria/ma_write.c
@@ -280,8 +280,8 @@ int maria_write(MARIA_HA *info, uchar *record)
   if (info->invalidator != 0)
   {
     DBUG_PRINT("info", ("invalidator... '%s' (update)",
-                        share->open_file_name));
-    (*info->invalidator)(share->open_file_name);
+                        share->open_file_name.str));
+    (*info->invalidator)(share->open_file_name.str);
     info->invalidator=0;
   }
 
@@ -1823,11 +1823,11 @@ my_bool _ma_log_new(MARIA_HA *info, my_off_t page, const uchar *buff,
   page_store(log_data + FILEID_STORE_SIZE, page);
 
   /* Store link to next unused page */
-  if (info->used_key_del == 2)
+  if (info->key_del_used == 2)
     page= 0;                                    /* key_del not changed */
   else
-    page= ((share->current_key_del == HA_OFFSET_ERROR) ? IMPOSSIBLE_PAGE_NO :
-           share->current_key_del / share->block_size);
+    page= ((share->key_del_current == HA_OFFSET_ERROR) ? IMPOSSIBLE_PAGE_NO :
+           share->key_del_current / share->block_size);
 
   page_store(log_data + FILEID_STORE_SIZE + PAGE_STORE_SIZE, page);
   key_nr_store(log_data + FILEID_STORE_SIZE + PAGE_STORE_SIZE*2, key_nr);
diff --git a/storage/maria/maria_def.h b/storage/maria/maria_def.h
index 060432c4696..42a78261d20 100644
--- a/storage/maria/maria_def.h
+++ b/storage/maria/maria_def.h
@@ -275,10 +275,10 @@ typedef struct st_maria_share
   MARIA_PACK pack;			/* Data about packed records */
   MARIA_BLOB *blobs;			/* Pointer to blobs */
   uint16 *column_nr;			/* Original column order */
-  char *unique_file_name;		/* realpath() of index file */
-  char *data_file_name;			/* Resolved path names from symlinks */
-  char *index_file_name;
-  char *open_file_name;			/* parameter to open filename */
+  LEX_STRING unique_file_name;		/* realpath() of index file */
+  LEX_STRING data_file_name;		/* Resolved path names from symlinks */
+  LEX_STRING index_file_name;
+  LEX_STRING open_file_name;		/* parameter to open filename */
   uchar *file_map;			/* mem-map of file if possible */
   PAGECACHE *pagecache;			/* ref to the current key cache */
   MARIA_DECODE_TREE *decode_trees;
@@ -337,7 +337,7 @@ typedef struct st_maria_share
   size_t (*file_read)(MARIA_HA *, uchar *, size_t, my_off_t, myf);
   size_t (*file_write)(MARIA_HA *, const uchar *, size_t, my_off_t, myf);
   invalidator_by_filename invalidator;	/* query cache invalidator */
-  my_off_t current_key_del;		/* delete links for index pages */
+  my_off_t key_del_current;		/* delete links for index pages */
   ulong this_process;			/* processid */
   ulong last_process;			/* For table-change-check */
   ulong last_version;			/* Version on start */
@@ -346,7 +346,6 @@ typedef struct st_maria_share
   ulong max_pack_length;
   ulong state_diff_length;
   uint rec_reflength;			/* rec_reflength in use now */
-  uint unique_name_length;
   uint keypage_header;
   uint32 ftparsers;			/* Number of distinct ftparsers
 						   + 1 */
@@ -382,12 +381,13 @@ typedef struct st_maria_share
   */
   my_bool now_transactional;
   my_bool have_versioning;
-  my_bool used_key_del;                         /* != 0 if key_del is locked */
+  my_bool key_del_used;                         /* != 0 if key_del is locked */
 #ifdef THREAD
   THR_LOCK lock;
   void (*lock_restore_status)(void *);
   pthread_mutex_t intern_lock;		/* Locking for use with _locking */
-  pthread_cond_t intern_cond;
+  pthread_mutex_t key_del_lock;
+  pthread_cond_t  key_del_cond;
 #endif
   my_off_t mmaped_length;
   uint nonmmaped_inserts;		/* counter of writing in
@@ -537,7 +537,7 @@ struct st_maria_handler
   int save_lastinx;
   uint preload_buff_size;		/* When preloading indexes */
   uint16 last_used_keyseg;              /* For MARIAMRG */
-  uint8 used_key_del;                   /* != 0 if key_del is used */
+  uint8 key_del_used;                   /* != 0 if key_del is used */
   my_bool was_locked;			/* Was locked in panic */
   my_bool append_insert_at_end;		/* Set if concurrent insert */
   my_bool quick_mode;
@@ -663,7 +663,7 @@ struct st_maria_handler
 */
 #define maria_print_error(SHARE, ERRNO)                         \
   do{ if (!maria_in_ha_maria)                                   \
-      _ma_report_error((ERRNO), (SHARE)->index_file_name); }    \
+      _ma_report_error((ERRNO), &(SHARE)->index_file_name); }    \
   while(0)
 #else
 #define maria_print_error(SHARE, ERRNO) while (0)
@@ -1030,7 +1030,7 @@ extern uint _ma_pack_get_block_info(MARIA_HA *maria, MARIA_BIT_BUFF *bit_buff,
                                     size_t *rec_buff_size,
                                     File file, my_off_t filepos);
 extern void _ma_store_blob_length(uchar *pos, uint pack_length, uint length);
-extern void _ma_report_error(int errcode, const char *file_name);
+extern void _ma_report_error(int errcode, const LEX_STRING *file_name);
 extern my_bool _ma_memmap_file(MARIA_HA *info);
 extern void _ma_unmap_file(MARIA_HA *info);
 extern uint _ma_save_pack_length(uint version, uchar * block_buff,
diff --git a/storage/maria/maria_ftdump.c b/storage/maria/maria_ftdump.c
index 978082adb78..5e3b47b956e 100644
--- a/storage/maria/maria_ftdump.c
+++ b/storage/maria/maria_ftdump.c
@@ -100,7 +100,8 @@ 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->s->open_file_name);
+    printf("Key %d in table %s is not a FULLTEXT key\n", inx,
+           info->s->open_file_name.str);
     goto err;
   }
 
diff --git a/storage/maria/maria_pack.c b/storage/maria/maria_pack.c
index 858f054a9f5..b9466475548 100644
--- a/storage/maria/maria_pack.c
+++ b/storage/maria/maria_pack.c
@@ -510,11 +510,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->s->open_file_name,"",MARIA_NAME_DEXT,
-                   2));
+    VOID(fn_format(org_name,isam_file->s->open_file_name.str,
+                   "",MARIA_NAME_DEXT, 2));
   else
-    VOID(fn_format(org_name,isam_file->s->open_file_name,"",MARIA_NAME_DEXT,
-                   2+4+16));
+    VOID(fn_format(org_name,isam_file->s->open_file_name.str,
+                   "",MARIA_NAME_DEXT, 2+4+16));
 
   if (init_pagecache(maria_pagecache, MARIA_MIN_PAGE_CACHE_SIZE, 0, 0,
                      maria_block_size, MY_WME) == 0)
@@ -709,7 +709,7 @@ static int compress(PACK_MRG_INFO *mrg,char *result_table)
       if (backup)
       {
 	if (my_rename(org_name,make_old_name(temp_name,
-                                             isam_file->s->open_file_name),
+                                             isam_file->s->open_file_name.str),
 		      MYF(MY_WME)))
 	  error=1;
 	else
diff --git a/storage/myisam/mi_search.c b/storage/myisam/mi_search.c
index f7d8d2c901c..2c346ea18f7 100644
--- a/storage/myisam/mi_search.c
+++ b/storage/myisam/mi_search.c
@@ -1802,13 +1802,13 @@ _mi_calc_bin_pack_key_length(MI_KEYDEF *keyinfo,uint nod_flag,uchar *next_key,
     }
     /* Check how many characters are identical to next key */
     key= s_temp->key+next_length;
+    s_temp->prev_length= 0;
     while (*key++ == *next_key++) ;
     if ((ref_length= (uint) (key - s_temp->key)-1) == next_length)
     {
       s_temp->next_key_pos=0;
       return length;                            /* can't pack next key */
     }
-    s_temp->prev_length=0;
     s_temp->n_ref_length=ref_length;
     return (int) (length-(ref_length - next_length) - next_length_pack +
                   get_pack_length(ref_length));