Fixed several bugs found by running *.test with maria engine

Renamed HA_EXTRA_PREAPRE_FOR_DELETE to HA_EXTRA_PERPARE_FOR_DROP
Added HA_EXTRA_PREPARE_FOR_RENAME (as we in the code before used HA_EXTRA_PREPARE_FOR_DELETE also for renames which confused things)
Allow multiple write locks for same page by same file handle
Don't write table state if table is not changed


include/my_base.h:
  Renamed HA_EXTRA_PREAPRE_FOR_DELETE to HA_EXTRA_PERPARE_FOR_DROP
  Added HA_EXTRA_PREPARE_FOR_RENAME (as we in the code before used HA_EXTRA_PREPARE_FOR_DELETE also for renames which confused things)
mysql-test/r/maria.result:
  More tests of things that failed in other tests
mysql-test/t/maria.test:
  More tests of things that failed in other tests
sql/ha_partition.cc:
  HA_EXTRA_PREPARE_FOR_DELETE -> HA_EXTRA_PREPARE_FOR_DROP
  Use HA_EXTRA_PREPARE_FOR_RENAME for renames
sql/ha_partition.h:
  HA_EXTRA_PREPARE_FOR_DELETE -> HA_EXTRA_PREPARE_FOR_DROP
  Use HA_EXTRA_PREPARE_FOR_RENAME for renames
sql/lock.cc:
  Fixed comment
sql/sql_table.cc:
  Fixed wrong usage of HA_EXTRA_PREAPRE_FOR_DELETE
storage/maria/ha_maria.cc:
  Added missing _ma_renable_logging_for_table()  (When using with ALTER TABLE + repair index)
  Enabled fast generation of index
storage/maria/ma_bitmap.c:
  Fixed bug when resetting full pages when page was a tail page
storage/maria/ma_blockrec.c:
  Fixed several bugs found by running *.test with maria engine:
  During update we keep old changed pages locked with a write lock to be able to reuse them.
  - Fixed bug with allocated but not used tail part
  - Fixed bug with blob that only had tail part
  - Fixed bug when update reused a page (needed multiple write locks for same page)
  - Fixed bug when first extent was a tail block
storage/maria/ma_check.c:
  Better error message when bitmap is destroyed
storage/maria/ma_close.c:
  Only write status if file was changed.
  Fixed bug when maria_chk -e file_name changed the file.
storage/maria/ma_dynrec.c:
  Removed not used argument to _ma_state_info_read_dsk
storage/maria/ma_extra.c:
  HA_EXTRA_PREPARE_FOR_DELETE -> HA_EXTRA_PREPARE_FOR_DROP
  Use HA_EXTRA_PREPARE_FOR_RENAME for renames
  Only ignore flushing of pages for DROP (not rename)
storage/maria/ma_locking.c:
  Removed not used argument to _ma_state_info_read_dsk
storage/maria/ma_open.c:
  Removed not used argument to _ma_state_info_read_dsk
storage/maria/ma_pagecache.c:
  Allow multiple write locks for same page by same file handle
  (Not yet complete, Sanja will fix)
storage/maria/ma_recovery.c:
  HA_EXTRA_PREPARE_FOR_DELETE -> HA_EXTRA_PREPARE_FOR_DROP
storage/maria/maria_def.h:
  Removed not used argument to _ma_state_info_read_dsk
storage/myisam/mi_extra.c:
  HA_EXTRA_PREPARE_FOR_DELETE -> HA_EXTRA_PREPARE_FOR_DROP
  Use HA_EXTRA_PREPARE_FOR_RENAME for renames
  Only ignore flushing of pages for DROP (not rename)
storage/myisammrg/ha_myisammrg.cc:
  HA_EXTRA_PREPARE_FOR_DELETE -> HA_EXTRA_PREPARE_FOR_DROP
  Use HA_EXTRA_PREPARE_FOR_RENAME for renames

include/my_base.h

@@ -135,7 +135,7 @@ enum ha_extra_function {
   HA_EXTRA_RESET_STATE,			/* Reset positions */
   HA_EXTRA_IGNORE_DUP_KEY,		/* Dup keys don't rollback everything*/
   HA_EXTRA_NO_IGNORE_DUP_KEY,
-  HA_EXTRA_PREPARE_FOR_DELETE,
+  HA_EXTRA_PREPARE_FOR_DROP,
   HA_EXTRA_PREPARE_FOR_UPDATE,		/* Remove read cache if problems */
   HA_EXTRA_PRELOAD_BUFFER_SIZE,         /* Set buffer size for preloading */
   /*
@@ -185,7 +185,9 @@ enum ha_extra_function {
     Inform handler that an "INSERT...ON DUPLICATE KEY UPDATE" will be
     executed. This condition is unset by HA_EXTRA_NO_IGNORE_DUP_KEY.
   */
-  HA_EXTRA_INSERT_WITH_UPDATE
+  HA_EXTRA_INSERT_WITH_UPDATE,
+  /* Inform handler that we will do a rename */
+  HA_EXTRA_PREPARE_FOR_RENAME
 };
 
 	/* The following is parameter to ha_panic() */

mysql-test/r/maria.result

@@ -1598,6 +1598,9 @@ insert into t1 values (1),(2),(3),(4),(NULL),(NULL),(NULL),(NULL);
 analyze table t1;
 Table	Op	Msg_type	Msg_text
 test.t1	analyze	status	OK
+analyze table t1;
+Table	Op	Msg_type	Msg_text
+test.t1	analyze	status	Table is already up to date
 show keys from t1;
 Table	Non_unique	Key_name	Seq_in_index	Column_name	Collation	Cardinality	Sub_part	Packed	Null	Index_type	Comment
 t1	1	a	1	a	A	8	NULL	NULL	YES	BTREE	
@@ -1887,10 +1890,79 @@ t1_name varchar(255) default null,
 t1_id int(10) unsigned not null auto_increment,
 key (t1_name),
 primary key (t1_id)
-) auto_increment = 1000 default charset=latin1;
+) engine=maria auto_increment = 1000 default charset=latin1;
 lock tables t1 write;
 INSERT INTO `t1` VALUES ('bla',1000),('bla',1001),('bla',1002);
 check table t1;
 Table	Op	Msg_type	Msg_text
 test.t1	check	status	OK
 unlock tables;
+create table t2 like t1;
+insert into t2 select * from t1;
+analyze table t2;
+Table	Op	Msg_type	Msg_text
+test.t2	analyze	status	OK
+delete from t2;
+insert into t2 select * from t1;
+analyze table t2;
+Table	Op	Msg_type	Msg_text
+test.t2	analyze	status	OK
+drop table t1,t2;
+create table t1 (a bigint auto_increment, primary key(a), b char(255), c varchar(20000));
+update t1 set b=repeat('a',100) where a between 1 and 100;
+check table t1;
+Table	Op	Msg_type	Msg_text
+test.t1	check	status	OK
+update t1 set c=repeat('a',8192*2) where a between 200 and 202;
+check table t1;
+Table	Op	Msg_type	Msg_text
+test.t1	check	status	OK
+drop table t1;
+CREATE TABLE t1 (
+auto int(5) unsigned NOT NULL auto_increment,
+string char(10) default "hello",
+tiny tinyint(4) DEFAULT '0' NOT NULL ,
+short smallint(6) DEFAULT '1' NOT NULL ,
+medium mediumint(8) DEFAULT '0' NOT NULL,
+long_int int(11) DEFAULT '0' NOT NULL,
+longlong bigint(13) DEFAULT '0' NOT NULL,
+real_float float(13,1) DEFAULT 0.0 NOT NULL,
+real_double double(16,4),
+utiny tinyint(3) unsigned DEFAULT '0' NOT NULL,
+ushort smallint(5) unsigned zerofill DEFAULT '00000' NOT NULL,
+umedium mediumint(8) unsigned DEFAULT '0' NOT NULL,
+ulong int(11) unsigned DEFAULT '0' NOT NULL,
+ulonglong bigint(13) unsigned DEFAULT '0' NOT NULL,
+time_stamp timestamp,
+date_field date,	
+time_field time,	
+date_time datetime,
+blob_col blob,
+tinyblob_col tinyblob,
+mediumblob_col mediumblob  not null default '',
+longblob_col longblob  not null default '',
+options enum('one','two','tree') not null ,
+flags set('one','two','tree') not null default '',
+PRIMARY KEY (auto),
+KEY (utiny),
+KEY (tiny),
+KEY (short),
+KEY any_name (medium),
+KEY (longlong),
+KEY (real_float),
+KEY (ushort),
+KEY (umedium),
+KEY (ulong),
+KEY (ulonglong,ulong),
+KEY (options,flags)
+) engine=maria;
+insert into t1 values (10,1,1,1,1,1,1,1,1,1,1,1,1,1,NULL,0,0,0,1,1,1,1,'one','one');
+create table t2 (primary key (auto)) engine=maria row_format=page select auto+1 as auto,1 as t1, 'a' as t2, repeat('a',256) as t3, binary repeat('b',256) as t4, repeat('a',4096) as t5, binary repeat('b',4096) as t6, '' as t7, binary '' as t8 from t1;
+check table t1,t2;
+Table	Op	Msg_type	Msg_text
+test.t1	check	status	OK
+test.t2	check	status	OK
+select t1,t2,length(t3),length(t4),length(t5),length(t6),t7,t8 from t2;
+t1	t2	length(t3)	length(t4)	length(t5)	length(t6)	t7	t8
+1	a	256	256	4096	4096		
+drop table t1,t2;

mysql-test/t/maria.test

@@ -711,6 +711,7 @@ analyze table t1;
 show index from t1;
 
 set maria_stats_method=DEFAULT;
+
 drop table t1;
 
 #
@@ -952,6 +953,7 @@ create table t1 (a int, key(a));
 
 insert into t1 values (1),(2),(3),(4),(NULL),(NULL),(NULL),(NULL);
 analyze table t1;
+analyze table t1;
 show keys from t1;
 
 alter table t1 disable keys;
@@ -1155,17 +1157,103 @@ drop table t1;
 
 # CHECK TABLE was reporting
 # "Size of datafile is: 0         Should be: 16384"
+#
+
 create table `t1` (
     t1_name varchar(255) default null,
     t1_id int(10) unsigned not null auto_increment,
     key (t1_name),
     primary key (t1_id)
-) auto_increment = 1000 default charset=latin1;
+) engine=maria auto_increment = 1000 default charset=latin1;
 lock tables t1 write;
 INSERT INTO `t1` VALUES ('bla',1000),('bla',1001),('bla',1002);
 check table t1;
 unlock tables;
 
+#
+# Check that an empty table uses fast recreate of index when we fill it
+# with insert ... select.
+
+create table t2 like t1;
+insert into t2 select * from t1;
+
+# This should say that the table is already up to date
+analyze table t2;
+delete from t2;
+insert into t2 select * from t1;
+analyze table t2;
+
+drop table t1,t2;
+
+#
+# Test when expanding a row so that it doesn't fit into the same page
+#
+
+create table t1 (a bigint auto_increment, primary key(a), b char(255), c varchar(20000));
+
+let $1=1000;
+--disable_query_log
+--disable_warnings
+while ($1)
+{
+  insert into t1 () values();
+  dec $1;
+}
+--enable_query_log
+update t1 set b=repeat('a',100) where a between 1 and 100;
+check table t1;
+update t1 set c=repeat('a',8192*2) where a between 200 and 202;
+check table t1;
+drop table t1;
+
+#
+# Test tail pages for blobs
+#
+
+CREATE TABLE t1 (
+  auto int(5) unsigned NOT NULL auto_increment,
+  string char(10) default "hello",
+  tiny tinyint(4) DEFAULT '0' NOT NULL ,
+  short smallint(6) DEFAULT '1' NOT NULL ,
+  medium mediumint(8) DEFAULT '0' NOT NULL,
+  long_int int(11) DEFAULT '0' NOT NULL,
+  longlong bigint(13) DEFAULT '0' NOT NULL,
+  real_float float(13,1) DEFAULT 0.0 NOT NULL,
+  real_double double(16,4),
+  utiny tinyint(3) unsigned DEFAULT '0' NOT NULL,
+  ushort smallint(5) unsigned zerofill DEFAULT '00000' NOT NULL,
+  umedium mediumint(8) unsigned DEFAULT '0' NOT NULL,
+  ulong int(11) unsigned DEFAULT '0' NOT NULL,
+  ulonglong bigint(13) unsigned DEFAULT '0' NOT NULL,
+  time_stamp timestamp,
+  date_field date,	
+  time_field time,	
+  date_time datetime,
+  blob_col blob,
+  tinyblob_col tinyblob,
+  mediumblob_col mediumblob  not null default '',
+  longblob_col longblob  not null default '',
+  options enum('one','two','tree') not null ,
+  flags set('one','two','tree') not null default '',
+  PRIMARY KEY (auto),
+  KEY (utiny),
+  KEY (tiny),
+  KEY (short),
+  KEY any_name (medium),
+  KEY (longlong),
+  KEY (real_float),
+  KEY (ushort),
+  KEY (umedium),
+  KEY (ulong),
+  KEY (ulonglong,ulong),
+  KEY (options,flags)
+) engine=maria;
+insert into t1 values (10,1,1,1,1,1,1,1,1,1,1,1,1,1,NULL,0,0,0,1,1,1,1,'one','one');
+create table t2 (primary key (auto)) engine=maria row_format=page select auto+1 as auto,1 as t1, 'a' as t2, repeat('a',256) as t3, binary repeat('b',256) as t4, repeat('a',4096) as t5, binary repeat('b',4096) as t6, '' as t7, binary '' as t8 from t1;
+check table t1,t2;
+select t1,t2,length(t3),length(t4),length(t5),length(t6),t7,t8 from t2;
+drop table t1,t2;
+
 # End of 5.2 tests
 
 --disable_result_log

sql/ha_partition.cc

@@ -4621,11 +4621,14 @@ void ha_partition::get_dynamic_partition_info(PARTITION_INFO *stat_info,
     about this call). We pass this along to all underlying MyISAM handlers
     and ignore it for the rest.
 
-  HA_EXTRA_PREPARE_FOR_DELETE:
+  HA_EXTRA_PREPARE_FOR_DROP:
     Only used by MyISAM, called in preparation for a DROP TABLE.
     It's used mostly by Windows that cannot handle dropping an open file.
     On other platforms it has the same effect as HA_EXTRA_FORCE_REOPEN.
 
+  HA_EXTRA_PREPARE_FOR_RENAME:
+    Informs the handler we are about to attempt a rename of the table.
+
   HA_EXTRA_READCHECK:
   HA_EXTRA_NO_READCHECK:
     Only one call to HA_EXTRA_NO_READCHECK from ha_open where it says that
@@ -4751,14 +4754,15 @@ int ha_partition::extra(enum ha_extra_function operation)
   }
 
   /* Category 3), used by MyISAM handlers */
-  case HA_EXTRA_PREPARE_FOR_DELETE:
-    DBUG_RETURN(prepare_for_delete());
+  case HA_EXTRA_PREPARE_FOR_RENAME:
+    DBUG_RETURN(prepare_for_rename());
     break;
   case HA_EXTRA_NORMAL:
   case HA_EXTRA_QUICK:
   case HA_EXTRA_NO_READCHECK:
   case HA_EXTRA_PREPARE_FOR_UPDATE:
   case HA_EXTRA_FORCE_REOPEN:
+  case HA_EXTRA_PREPARE_FOR_DROP:
   case HA_EXTRA_FLUSH_CACHE:
   {
     if (m_myisam)
@@ -4910,24 +4914,24 @@ void ha_partition::prepare_extra_cache(uint cachesize)
     0                     Success
 */
 
-int ha_partition::prepare_for_delete()
+int ha_partition::prepare_for_rename()
 {
   int result= 0, tmp;
   handler **file;
-  DBUG_ENTER("ha_partition::prepare_for_delete()");
+  DBUG_ENTER("ha_partition::prepare_for_rename()");
   
   if (m_new_file != NULL)
   {
     for (file= m_new_file; *file; file++)
-      if ((tmp= (*file)->extra(HA_EXTRA_PREPARE_FOR_DELETE)))
+      if ((tmp= (*file)->extra(HA_EXTRA_PREPARE_FOR_RENAME)))
         result= tmp;      
     for (file= m_reorged_file; *file; file++)
-      if ((tmp= (*file)->extra(HA_EXTRA_PREPARE_FOR_DELETE)))
+      if ((tmp= (*file)->extra(HA_EXTRA_PREPARE_FOR_RENAME)))
         result= tmp;   
     DBUG_RETURN(result);   
   }
   
-  DBUG_RETURN(loop_extra(HA_EXTRA_PREPARE_FOR_DELETE));
+  DBUG_RETURN(loop_extra(HA_EXTRA_PREPARE_FOR_RENAME));
 }
 
 /*

sql/ha_partition.h

@@ -211,7 +211,7 @@ class ha_partition :public handler
   }
   virtual void change_table_ptr(TABLE *table_arg, TABLE_SHARE *share);
 private:
-  int prepare_for_delete();
+  int prepare_for_rename();
   int copy_partitions(ulonglong *copied, ulonglong *deleted);
   void cleanup_new_partition(uint part_count);
   int prepare_new_partition(TABLE *table, HA_CREATE_INFO *create_info,

sql/lock.cc

@@ -31,8 +31,8 @@
       This is followed by a call to thr_multi_lock() for all tables.
 
   - When statement is done, we call mysql_unlock_tables().
-    This will call thr_multi_unlock() followed by
-    table_handler->external_lock(thd, F_UNLCK) for each table.
+    table_handler->external_lock(thd, F_UNLCK) followed by
+    thr_multi_unlock() for each table.
 
   - Note that mysql_unlock_tables() may be called several times as
     MySQL in some cases can free some tables earlier than others.

sql/sql_table.cc

@@ -3694,8 +3694,9 @@ mysql_rename_table(handlerton *base, const char *old_db,
     wait_while_table_is_used()
     thd			Thread handler
     table		Table to remove from cache
-    function		HA_EXTRA_PREPARE_FOR_DELETE if table is to be deleted
+    function		HA_EXTRA_PREPARE_FOR_DROP if table is to be deleted
 			HA_EXTRA_FORCE_REOPEN if table is not be used
+                        HA_EXTRA_PREPARE_FOR_REANME if table is to be renamed
   NOTES
    When returning, the table will be unusable for other threads until
    the table is closed.
@@ -3745,7 +3746,7 @@ void close_cached_table(THD *thd, TABLE *table)
 {
   DBUG_ENTER("close_cached_table");
 
-  wait_while_table_is_used(thd, table, HA_EXTRA_PREPARE_FOR_DELETE);
+  wait_while_table_is_used(thd, table, HA_EXTRA_FORCE_REOPEN);
   /* Close lock if this is not got with LOCK TABLES */
   if (thd->lock)
   {
@@ -6532,7 +6533,7 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
   if (lower_case_table_names)
     my_casedn_str(files_charset_info, old_name);
 
-  wait_while_table_is_used(thd, table, HA_EXTRA_PREPARE_FOR_DELETE);
+  wait_while_table_is_used(thd, table, HA_EXTRA_PREPARE_FOR_RENAME);
   close_data_files_and_morph_locks(thd, db, table_name);
 
   error=0;

storage/maria/ha_maria.cc

@@ -1324,7 +1324,10 @@ int ha_maria::repair(THD *thd, HA_CHECK &param, bool do_optimize)
   }
   thd->proc_info= old_proc_info;
   if (!thd->locked_tables)
+  {
+    _ma_reenable_logging_for_table(file->s);
     maria_lock_database(file, F_UNLCK);
+  }
   DBUG_RETURN(error ? HA_ADMIN_FAILED :
               !optimize_done ? HA_ADMIN_ALREADY_DONE : HA_ADMIN_OK);
 }
@@ -1624,10 +1627,8 @@ void ha_maria::start_bulk_insert(ha_rows rows)
   if (!rows || (rows > MARIA_MIN_ROWS_TO_USE_WRITE_CACHE))
     maria_extra(file, HA_EXTRA_WRITE_CACHE, (void*) &size);
 
-  can_enable_indexes= maria_is_all_keys_active(file->s->state.key_map,
-                                               file->s->base.keys);
-  /* TODO: Remove when we have repair() working */
-  can_enable_indexes= 0;
+  can_enable_indexes= (maria_is_all_keys_active(file->s->state.key_map,
+                                                file->s->base.keys));
 
   if (!(specialflag & SPECIAL_SAFE_MODE))
   {
@@ -1640,8 +1641,7 @@ void ha_maria::start_bulk_insert(ha_rows rows)
     if (file->state->records == 0 && can_enable_indexes &&
         (!rows || rows >= MARIA_MIN_ROWS_TO_DISABLE_INDEXES))
       maria_disable_non_unique_index(file, rows);
-    else
-      if (!file->bulk_insert &&
+    else if (!file->bulk_insert &&
              (!rows || rows >= MARIA_MIN_ROWS_TO_USE_BULK_INSERT))
     {
       maria_init_bulk_insert(file, thd->variables.bulk_insert_buff_size, rows);

storage/maria/ma_bitmap.c

@@ -451,6 +451,10 @@ static void _ma_print_bitmap(MARIA_FILE_BITMAP *bitmap)
   fprintf(DBUG_FILE,"\nBitmap page changes at page %lu\n",
           (ulong) bitmap->page);
 
+  DBUG_ASSERT(memcmp(bitmap->map + bitmap->block_size -
+                     sizeof(maria_bitmap_marker),
+                     maria_bitmap_marker, sizeof(maria_bitmap_marker)) == 0);
+
   page= (ulong) bitmap->page+1;
   for (pos= bitmap->map, org_pos= bitmap->map + bitmap->block_size ;
        pos < end ;
@@ -536,14 +540,14 @@ static my_bool _ma_read_bitmap_page(MARIA_SHARE *share,
   }
   bitmap->used_size= bitmap->total_size;
   DBUG_ASSERT(share->pagecache->block_size == bitmap->block_size);
-  res= (pagecache_read(share->pagecache,
+  res= ((pagecache_read(share->pagecache,
                        (PAGECACHE_FILE*)&bitmap->file, page, 0,
                        (uchar*) bitmap->map,
                        PAGECACHE_PLAIN_PAGE,
-                       PAGECACHE_LOCK_LEFT_UNLOCKED, 0) == NULL) |
+                       PAGECACHE_LOCK_LEFT_UNLOCKED, 0) == NULL) ||
         memcmp(bitmap->map + bitmap->block_size -
                sizeof(maria_bitmap_marker),
-           maria_bitmap_marker, sizeof(maria_bitmap_marker));
+               maria_bitmap_marker, sizeof(maria_bitmap_marker)));
 #ifndef DBUG_OFF
   if (!res)
     memcpy(bitmap->map + bitmap->block_size, bitmap->map, bitmap->block_size);
@@ -1838,11 +1842,15 @@ my_bool _ma_bitmap_release_unused(MARIA_HA *info, MARIA_BITMAP_BLOCKS *blocks)
   /* Handle all full pages and tail pages (for head page and blob) */
   for (block++; block < end; block++)
   {
+    uint page_count;
     if (!block->page_count)
       continue;                               /* Skip 'filler blocks' */
 
+    page_count= block->page_count;
     if (block->used & BLOCKUSED_TAIL)
     {
+      /* The bitmap page is only one page */
+      page_count= 1;
       if (block->used & BLOCKUSED_USED)
       {
         DBUG_PRINT("info", ("tail empty_space: %u", block->empty_space));
@@ -1861,7 +1869,7 @@ my_bool _ma_bitmap_release_unused(MARIA_HA *info, MARIA_BITMAP_BLOCKS *blocks)
     }
     if (!(block->used & BLOCKUSED_USED) &&
         _ma_reset_full_page_bits(info, bitmap,
-                                 block->page, block->page_count))
+                                 block->page, page_count))
       goto err;
   }
   pthread_mutex_unlock(&info->s->bitmap.bitmap_lock);

storage/maria/ma_check.c

@@ -1706,8 +1706,13 @@ static int check_block_record(HA_CHECK *param, MARIA_HA *info, int extend,
                               full_dir ? 0 : empty_space,
                               &bitmap_pattern))
     {
+      if (bitmap_pattern == ~(uint) 0)
         _ma_check_print_error(param,
-                            "Page %9s:  Wrong data in bitmap.  Page_type: %d  empty_space: %u  Bitmap: %d",
+                              "Page: %9s: Wrong bitmap for data on page",
+                              llstr(pos, llbuff));
+      else
+      _ma_check_print_error(param,
+                            "Page %9s:  Wrong data in bitmap.  Page_type: %d  empty_space: %u  Bitmap-bits: %d",
                             llstr(pos, llbuff), page_type, empty_space,
                             bitmap_pattern);
       if (param->err_count++ > MAXERR || !(param->testflag & T_VERBOSE))

storage/maria/ma_close.c

@@ -65,6 +65,7 @@ int maria_close(register MARIA_HA *info)
 
   if (flag)
   {
+    /* Last close of file; Flush everything */
     if (share->kfile.file >= 0)
     {
       if ((*share->once_end)(share))
@@ -87,7 +88,7 @@ int maria_close(register MARIA_HA *info)
         may be using the file at this point
         IF using --external-locking, which does not apply to Maria.
       */
-      if (share->mode != O_RDONLY)
+      if (share->changed)
 	_ma_state_info_write(share->kfile.file, &share->state, 1);
       if (my_close(share->kfile.file, MYF(0)))
         error= my_errno;

storage/maria/ma_dynrec.c

@@ -1694,7 +1694,7 @@ int _ma_read_rnd_dynamic_record(MARIA_HA *info,
       {						/* Check if changed */
 	info_read=1;
 	info->rec_cache.seek_not_done=1;
-	if (_ma_state_info_read_dsk(share->kfile.file, &share->state, 1))
+	if (_ma_state_info_read_dsk(share->kfile.file, &share->state))
 	  goto panic;
       }
       if (filepos >= info->state->data_file_length)

storage/maria/ma_extra.c

@@ -264,8 +264,8 @@ int maria_extra(MARIA_HA *info, enum ha_extra_function function,
     share->last_version= 0L;			/* Impossible version */
     pthread_mutex_unlock(&THR_LOCK_maria);
     break;
-  case HA_EXTRA_PREPARE_FOR_DELETE:
-    /*  QQ: suggest to rename it to "PREPARE_FOR_DROP" */
+  case HA_EXTRA_PREPARE_FOR_DROP:
+  case HA_EXTRA_PREPARE_FOR_RENAME:
     pthread_mutex_lock(&THR_LOCK_maria);
     share->last_version= 0L;			/* Impossible version */
 #ifdef __WIN__
@@ -284,7 +284,8 @@ int maria_extra(MARIA_HA *info, enum ha_extra_function function,
       Does ENABLE KEYS rebuild them too?
     */
     if (flush_pagecache_blocks(share->pagecache, &share->kfile,
-                               FLUSH_IGNORE_CHANGED))
+                               (function == HA_EXTRA_PREPARE_FOR_DROP ?
+                                FLUSH_IGNORE_CHANGED : FLUSH_RELEASE)))
     {
       error=my_errno;
       share->changed=1;

storage/maria/ma_locking.c

@@ -153,7 +153,7 @@ int maria_lock_database(MARIA_HA *info, int lock_type)
       }
       if (!share->r_locks && !share->w_locks)
       {
-	if (_ma_state_info_read_dsk(share->kfile.file, &share->state, 1))
+	if (_ma_state_info_read_dsk(share->kfile.file, &share->state))
 	{
 	  error=my_errno;
 	  break;
@@ -181,7 +181,7 @@ int maria_lock_database(MARIA_HA *info, int lock_type)
 	{
 	  if (!share->r_locks)
 	  {
-	    if (_ma_state_info_read_dsk(share->kfile.file, &share->state, 1))
+	    if (_ma_state_info_read_dsk(share->kfile.file, &share->state))
 	    {
 	      error=my_errno;
 	      break;
@@ -364,7 +364,7 @@ int _ma_readinfo(register MARIA_HA *info, int lock_type, int check_keybuffer)
     MARIA_SHARE *share=info->s;
     if (!share->tot_locks)
     {
-      if (_ma_state_info_read_dsk(share->kfile.file, &share->state, 1))
+      if (_ma_state_info_read_dsk(share->kfile.file, &share->state))
       {
 	int error=my_errno ? my_errno : -1;
 	my_errno=error;

storage/maria/ma_open.c

@@ -1110,18 +1110,13 @@ uchar *_ma_state_info_read(uchar *ptr, MARIA_STATE_INFO *state)
    @param  pRead           if true, use my_pread(), otherwise my_read()
 */
 
-uint _ma_state_info_read_dsk(File file, MARIA_STATE_INFO *state, my_bool pRead)
+uint _ma_state_info_read_dsk(File file, MARIA_STATE_INFO *state)
 {
   char	buff[MARIA_STATE_INFO_SIZE + MARIA_STATE_EXTRA_SIZE];
 
   if (!maria_single_user)
   {
-    if (pRead)
-    {
-      if (my_pread(file, buff, state->state_length,0L, MYF(MY_NABP)))
-	return 1;
-    }
-    else if (my_read(file, buff, state->state_length,MYF(MY_NABP)))
+    if (my_pread(file, buff, state->state_length, 0L, MYF(MY_NABP)))
       return 1;
     _ma_state_info_read(buff, state);
   }

storage/maria/ma_pagecache.c

@@ -300,23 +300,24 @@ struct st_pagecache_block_link
     *next_changed, **prev_changed; /* for lists of file dirty/clean blocks   */
   struct st_pagecache_hash_link
     *hash_link;           /* backward ptr to referring hash_link             */
+#ifndef DBUG_OFF
+  PAGECACHE_PIN_INFO *pin_list;
+  PAGECACHE_LOCK_INFO *lock_list;
+#endif
+  KEYCACHE_CONDVAR *condvar; /* condition variable for 'no readers' event    */
+  uchar *buffer;           /* buffer for the block page                      */
+  PAGECACHE_FILE *write_locker;
+  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                */
-  uchar *buffer;           /* buffer for the block page                       */
   uint status;            /* state of the block                              */
   uint pins;              /* pin counter                                     */
-#ifndef DBUG_OFF
-  PAGECACHE_PIN_INFO *pin_list;
-  PAGECACHE_LOCK_INFO *lock_list;
-#endif
   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             */
-  ulonglong last_hit_time; /* timestamp of the last hit                      */
   /** @brief LSN when first became dirty; LSN_MAX means "not yet set"        */
   LSN rec_lsn;
-  KEYCACHE_CONDVAR *condvar; /* condition variable for 'no readers' event    */
 };
 
 #ifndef DBUG_OFF
@@ -2147,6 +2148,9 @@ static void info_change_lock(PAGECACHE_BLOCK_LINK *block, my_bool wl)
     get_wrlock()
     pagecache            pointer to a page cache data structure
     block                the block to work with
+    user_file		 Unique handler per handler file. Used to check if
+			 we request many write locks withing the same
+                         statement
 
   RETURN
     0 - OK
@@ -2154,7 +2158,8 @@ static void info_change_lock(PAGECACHE_BLOCK_LINK *block, my_bool wl)
 */
 
 static my_bool get_wrlock(PAGECACHE *pagecache,
-                          PAGECACHE_BLOCK_LINK *block)
+                          PAGECACHE_BLOCK_LINK *block,
+                          PAGECACHE_FILE *user_file)
 {
   PAGECACHE_FILE file= block->hash_link->file;
   pgcache_page_no_t pageno= block->hash_link->pageno;
@@ -2165,7 +2170,7 @@ static my_bool get_wrlock(PAGECACHE *pagecache,
                           file.file, block->hash_link->file.file,
                           pageno, block->hash_link->pageno));
   PCBLOCK_INFO(block);
-  while (block->status & PCBLOCK_WRLOCK)
+  while ((block->status & PCBLOCK_WRLOCK) && block->write_locker != user_file)
   {
     /* Lock failed we will wait */
 #ifdef THREAD
@@ -2197,9 +2202,9 @@ static my_bool get_wrlock(PAGECACHE *pagecache,
       DBUG_RETURN(1);
     }
   }
-  DBUG_ASSERT(block->pins == 0);
   /* we are doing it by global cache mutex protection, so it is OK */
   block->status|= PCBLOCK_WRLOCK;
+  block->write_locker= user_file;
   DBUG_PRINT("info", ("WR lock set, block 0x%lx", (ulong)block));
   DBUG_RETURN(0);
 }
@@ -2223,6 +2228,8 @@ static void release_wrlock(PAGECACHE_BLOCK_LINK *block)
   PCBLOCK_INFO(block);
   DBUG_ASSERT(block->status & PCBLOCK_WRLOCK);
   DBUG_ASSERT(block->pins > 0);
+  if (block->pins > 1)
+    DBUG_VOID_RETURN;                      /* Multiple write locked */
   block->status&= ~PCBLOCK_WRLOCK;
   DBUG_PRINT("info", ("WR lock reset, block 0x%lx", (ulong)block));
 #ifdef THREAD
@@ -2244,6 +2251,7 @@ static void release_wrlock(PAGECACHE_BLOCK_LINK *block)
     block                the block to work with
     lock                 lock change mode
     pin                  pinchange mode
+    file		 File handler requesting pin
 
   RETURN
     0 - OK
@@ -2253,7 +2261,8 @@ static void release_wrlock(PAGECACHE_BLOCK_LINK *block)
 static my_bool make_lock_and_pin(PAGECACHE *pagecache,
                                  PAGECACHE_BLOCK_LINK *block,
                                  enum pagecache_page_lock lock,
-                                 enum pagecache_page_pin pin)
+                                 enum pagecache_page_pin pin,
+                                 PAGECACHE_FILE *file)
 {
   DBUG_ENTER("make_lock_and_pin");
 
@@ -2274,7 +2283,7 @@ static my_bool make_lock_and_pin(PAGECACHE *pagecache,
   switch (lock) {
   case PAGECACHE_LOCK_WRITE:               /* free  -> write */
     /* Writelock and pin the buffer */
-    if (get_wrlock(pagecache, block))
+    if (get_wrlock(pagecache, block, file))
     {
       /* can't lock => need retry */
       goto retry;
@@ -2291,6 +2300,7 @@ static my_bool make_lock_and_pin(PAGECACHE *pagecache,
       implementation)
     */
     release_wrlock(block);
+    /* fall through */
   case PAGECACHE_LOCK_READ_UNLOCK:         /* read  -> free  */
   case PAGECACHE_LOCK_LEFT_READLOCKED:     /* read  -> read  */
     if (pin == PAGECACHE_UNPIN)
@@ -2549,7 +2559,7 @@ void pagecache_unlock(PAGECACHE *pagecache,
   if (lsn != LSN_IMPOSSIBLE)
     check_and_set_lsn(pagecache, lsn, block);
 
-  if (make_lock_and_pin(pagecache, block, lock, pin))
+  if (make_lock_and_pin(pagecache, block, lock, pin, file))
   {
     DBUG_ASSERT(0); /* should not happend */
   }
@@ -2617,7 +2627,7 @@ void pagecache_unpin(PAGECACHE *pagecache,
   */
   if (make_lock_and_pin(pagecache, block,
                         PAGECACHE_LOCK_LEFT_READLOCKED,
-                        PAGECACHE_UNPIN))
+                        PAGECACHE_UNPIN, file))
     DBUG_ASSERT(0);                           /* should not happend */
 
   remove_reader(block);
@@ -2678,7 +2688,7 @@ void pagecache_unlock_by_link(PAGECACHE *pagecache,
       lock == PAGECACHE_LOCK_READ_UNLOCK)
   {
     /* block do not need here so we do not provide it */
-    if (make_lock_and_pin(pagecache, 0, lock, pin))
+    if (make_lock_and_pin(pagecache, 0, lock, pin, 0))
       DBUG_ASSERT(0);                         /* should not happend */
     DBUG_VOID_RETURN;
   }
@@ -2710,7 +2720,7 @@ void pagecache_unlock_by_link(PAGECACHE *pagecache,
   if (lsn != LSN_IMPOSSIBLE)
     check_and_set_lsn(pagecache, lsn, block);
 
-  if (make_lock_and_pin(pagecache, block, lock, pin))
+  if (make_lock_and_pin(pagecache, block, lock, pin, 0))
     DBUG_ASSERT(0);                           /* should not happend */
 
   /*
@@ -2772,7 +2782,7 @@ void pagecache_unpin_by_link(PAGECACHE *pagecache,
   */
   if (make_lock_and_pin(pagecache, block,
                         PAGECACHE_LOCK_LEFT_READLOCKED,
-                        PAGECACHE_UNPIN))
+                        PAGECACHE_UNPIN, 0))
     DBUG_ASSERT(0); /* should not happend */
 
   /*
@@ -2889,7 +2899,7 @@ uchar *pagecache_valid_read(PAGECACHE *pagecache,
                  validator, validator_data);
       DBUG_PRINT("info", ("read is done"));
     }
-    if (make_lock_and_pin(pagecache, block, lock, pin))
+    if (make_lock_and_pin(pagecache, block, lock, pin, file))
     {
       /*
         We failed to write lock the block, cache is unlocked,
@@ -3009,7 +3019,7 @@ my_bool pagecache_delete(PAGECACHE *pagecache,
     if (pin == PAGECACHE_PIN)
       reg_requests(pagecache, block, 1);
     DBUG_ASSERT(block != 0);
-    if (make_lock_and_pin(pagecache, block, lock, pin))
+    if (make_lock_and_pin(pagecache, block, lock, pin, file))
     {
       /*
         We failed to writelock the block, cache is unlocked, and last write
@@ -3059,7 +3069,7 @@ my_bool pagecache_delete(PAGECACHE *pagecache,
     /* Cache is locked, so we can relese page before freeing it */
     make_lock_and_pin(pagecache, block,
                       PAGECACHE_LOCK_WRITE_UNLOCK,
-                      PAGECACHE_UNPIN);
+                      PAGECACHE_UNPIN, file);
     DBUG_ASSERT(link->requests > 0);
     link->requests--;
     /* See NOTE for pagecache_unlock about registering requests. */
@@ -3254,7 +3264,7 @@ my_bool pagecache_write_part(PAGECACHE *pagecache,
                           write_lock_change_table[lock].new_lock,
                           (need_lock_change ?
                            write_pin_change_table[pin].new_pin :
-                           pin)))
+                           pin), file))
     {
       /*
         We failed to writelock the block, cache is unlocked, and last write
@@ -3307,7 +3317,6 @@ my_bool pagecache_write_part(PAGECACHE *pagecache,
       }
     }
 
-
     if (need_lock_change)
     {
       /*
@@ -3316,7 +3325,7 @@ my_bool pagecache_write_part(PAGECACHE *pagecache,
       */
       if (make_lock_and_pin(pagecache, block,
                             write_lock_change_table[lock].unlock_lock,
-                            write_pin_change_table[pin].unlock_pin))
+                            write_pin_change_table[pin].unlock_pin, file))
         DBUG_ASSERT(0);
     }
 
@@ -3474,7 +3483,7 @@ static int flush_cached_blocks(PAGECACHE *pagecache,
     DBUG_ASSERT((block->status & PCBLOCK_WRLOCK) == 0);
     DBUG_ASSERT(block->pins == 0);
     if (make_lock_and_pin(pagecache, block,
-                          PAGECACHE_LOCK_WRITE, PAGECACHE_PIN))
+                          PAGECACHE_LOCK_WRITE, PAGECACHE_PIN, 0))
       DBUG_ASSERT(0);
 
     KEYCACHE_DBUG_PRINT("flush_cached_blocks",
@@ -3497,7 +3506,7 @@ static int flush_cached_blocks(PAGECACHE *pagecache,
 
     make_lock_and_pin(pagecache, block,
                       PAGECACHE_LOCK_WRITE_UNLOCK,
-                      PAGECACHE_UNPIN);
+                      PAGECACHE_UNPIN, 0);
 
     pagecache->global_cache_write++;
     if (error)

storage/maria/ma_recovery.c

@@ -507,7 +507,7 @@ prototype_exec_hook(REDO_DROP_TABLE)
       this table should not be used anymore, and (only on Windows) to close
       open files so they can be deleted
     */
-    if (maria_extra(info, HA_EXTRA_PREPARE_FOR_DELETE, NULL) ||
+    if (maria_extra(info, HA_EXTRA_PREPARE_FOR_DROP, NULL) ||
         maria_close(info))
       goto end;
     info= NULL;

storage/maria/maria_def.h

@@ -864,8 +864,7 @@ extern uint _ma_nommap_pwrite(MARIA_HA *info, uchar *Buffer,
 
 uint _ma_state_info_write(File file, MARIA_STATE_INFO *state, uint pWrite);
 uchar *_ma_state_info_read(uchar *ptr, MARIA_STATE_INFO *state);
-uint _ma_state_info_read_dsk(File file, MARIA_STATE_INFO *state,
-                             my_bool pRead);
+uint _ma_state_info_read_dsk(File file, MARIA_STATE_INFO *state);
 uint _ma_base_info_write(File file, MARIA_BASE_INFO *base);
 int _ma_keyseg_write(File file, const HA_KEYSEG *keyseg);
 char *_ma_keyseg_read(char *ptr, HA_KEYSEG *keyseg);

storage/myisam/mi_extra.c

@@ -255,15 +255,16 @@ int mi_extra(MI_INFO *info, enum ha_extra_function function, void *extra_arg)
     share->last_version= 0L;			/* Impossible version */
     pthread_mutex_unlock(&THR_LOCK_myisam);
     break;
-  case HA_EXTRA_PREPARE_FOR_DELETE:
+  case HA_EXTRA_PREPARE_FOR_RENAME:
+  case HA_EXTRA_PREPARE_FOR_DROP:
     pthread_mutex_lock(&THR_LOCK_myisam);
     share->last_version= 0L;			/* Impossible version */
 #ifdef __WIN__
     /* Close the isam and data files as Win32 can't drop an open table */
     pthread_mutex_lock(&share->intern_lock);
     if (flush_key_blocks(share->key_cache, share->kfile,
-			 (function == HA_EXTRA_FORCE_REOPEN ?
-			  FLUSH_RELEASE : FLUSH_IGNORE_CHANGED)))
+			 (function == HA_EXTRA_PREPARE_FOR_DROP ?
+                          FLUSH_IGNORE_CHANGED : FLUSH_RELEASE)))
     {
       error=my_errno;
       share->changed=1;

storage/myisammrg/ha_myisammrg.cc

@@ -392,7 +392,8 @@ int ha_myisammrg::extra(enum ha_extra_function operation)
   /* As this is just a mapping, we don't have to force the underlying
      tables to be closed */
   if (operation == HA_EXTRA_FORCE_REOPEN ||
-      operation == HA_EXTRA_PREPARE_FOR_DELETE)
+      operation == HA_EXTRA_PREPARE_FOR_DROP ||
+      operation == HA_EXTRA_PREPARE_FOR_RENAME)
     return 0;
   return myrg_extra(file,operation,0);
 }