Bug#51851: Server with SBR locks mutex twice on

LOAD DATA into partitioned MyISAM table

Problem was that both partitioning and myisam
used the same table_share->mutex for different protections
(auto inc and repair).

Solved by adding a specific mutex for the partitioning
auto_increment.

Also adding destroying the ha_data structure in
free_table_share (which is to be propagated
into 5.5).

This is a 5.1 ONLY patch, already fixed in 5.5+.

mysql-test/r/partition_binlog_stmt.result

+DROP TABLE IF EXISTS t1;
+#
+# Bug#51851: Server with SBR locks mutex twice on LOAD DATA into
+#            partitioned MyISAM table
+CREATE TABLE t1
+(id INT NOT NULL AUTO_INCREMENT PRIMARY KEY,
+name TINYBLOB NOT NULL,
+modified TIMESTAMP DEFAULT '0000-00-00 00:00:00',
+INDEX namelocs (name(255))) ENGINE = MyISAM
+PARTITION BY HASH(id) PARTITIONS 2;
+LOAD DATA LOCAL INFILE 'init_file.txt'
+INTO TABLE t1 (name);
+DROP TABLE t1;

mysql-test/t/partition_binlog_stmt.test

+--source include/have_partition.inc
+--source include/have_binlog_format_statement.inc
+
+--disable_warnings
+DROP TABLE IF EXISTS t1;
+--enable_warnings
+
+--echo #
+--echo # Bug#51851: Server with SBR locks mutex twice on LOAD DATA into
+--echo #            partitioned MyISAM table
+--write_file init_file.txt
+abcd
+EOF
+
+CREATE TABLE t1
+(id INT NOT NULL AUTO_INCREMENT PRIMARY KEY,
+ name TINYBLOB NOT NULL,
+ modified TIMESTAMP DEFAULT '0000-00-00 00:00:00',
+ INDEX namelocs (name(255))) ENGINE = MyISAM
+PARTITION BY HASH(id) PARTITIONS 2;
+
+LOAD DATA LOCAL INFILE 'init_file.txt'
+INTO TABLE t1 (name);
+
+--remove_file init_file.txt
+DROP TABLE t1;

sql/ha_partition.cc

@@ -2449,6 +2449,21 @@ bool ha_partition::get_from_handler_file(const char *name, MEM_ROOT *mem_root)
 /****************************************************************************
                 MODULE open/close object
 ****************************************************************************/
+
+
+/**
+  A destructor for partition-specific TABLE_SHARE data.
+*/
+
+void ha_data_partition_destroy(void *ha_data)
+{
+  if (ha_data)
+  {
+    HA_DATA_PARTITION *ha_part_data= (HA_DATA_PARTITION*) ha_data;
+    pthread_mutex_destroy(&ha_part_data->LOCK_auto_inc);
+  }
+}
+
 /*
   Open handler object
 
@@ -2605,6 +2620,8 @@ int ha_partition::open(const char *name, int mode, uint test_if_locked)
     }
     DBUG_PRINT("info", ("table_share->ha_data 0x%p", ha_data));
     bzero(ha_data, sizeof(HA_DATA_PARTITION));
+    table_share->ha_data_destroy= ha_data_partition_destroy;
+    VOID(pthread_mutex_init(&ha_data->LOCK_auto_inc, MY_MUTEX_INIT_FAST));
   }
   if (is_not_tmp_table)
     pthread_mutex_unlock(&table_share->mutex);

sql/ha_partition.h

@@ -44,6 +44,7 @@ typedef struct st_partition_share
 typedef struct st_ha_data_partition
 {
   ulonglong next_auto_inc_val;                 /**< first non reserved value */
+  pthread_mutex_t LOCK_auto_inc;
   bool auto_inc_initialized;
 } HA_DATA_PARTITION;
 
@@ -944,8 +945,9 @@ class ha_partition :public handler
     DBUG_ASSERT(table_share->ha_data && !auto_increment_lock);
     if(table_share->tmp_table == NO_TMP_TABLE)
     {
+      HA_DATA_PARTITION *ha_data= (HA_DATA_PARTITION*) table_share->ha_data;
       auto_increment_lock= TRUE;
-      pthread_mutex_lock(&table_share->mutex);
+      pthread_mutex_lock(&ha_data->LOCK_auto_inc);
     }
   }
   virtual void unlock_auto_increment()
@@ -958,7 +960,8 @@ class ha_partition :public handler
     */
     if(auto_increment_lock && !auto_increment_safe_stmt_log_lock)
     {
-      pthread_mutex_unlock(&table_share->mutex);
+      HA_DATA_PARTITION *ha_data= (HA_DATA_PARTITION*) table_share->ha_data;
+      pthread_mutex_unlock(&ha_data->LOCK_auto_inc);
       auto_increment_lock= FALSE;
     }
   }

sql/table.cc

@@ -425,6 +425,11 @@ void free_table_share(TABLE_SHARE *share)
       key_info->flags= 0;
     }
   }
+  if (share->ha_data_destroy)
+  {
+    share->ha_data_destroy(share->ha_data);
+    share->ha_data_destroy= NULL;
+  }
   /* We must copy mem_root from share because share is allocated through it */
   memcpy((char*) &mem_root, (char*) &share->mem_root, sizeof(mem_root));
   free_root(&mem_root, MYF(0));                 // Free's share
@@ -1616,6 +1621,11 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
   delete crypted;
   delete handler_file;
   hash_free(&share->name_hash);
+  if (share->ha_data_destroy)
+  {
+    share->ha_data_destroy(share->ha_data);
+    share->ha_data_destroy= NULL;
+  }
 
   open_table_error(share, error, share->open_errno, errarg);
   DBUG_RETURN(error);

sql/table.h

@@ -463,6 +463,7 @@ typedef struct st_table_share
 
   /** place to store storage engine specific data */
   void *ha_data;
+  void (*ha_data_destroy)(void *); /* An optional destructor for ha_data. */
 
 
   /*