A follow up after the patch for Bug#21074 - even though

we now have exclusive name lock on the table name in mysql_rm_table_part2,
we still should keep LOCK_open - some storage engines are not
ready for locking scope change and assume that LOCK_open is kept.
Still, the binary logging and query cache invalidation calls
moved out of LOCK_open scope.
Fixes some of the broken 5.1-runtime tests (tests break on asserts).


sql/ha_ndbcluster.cc:
  Do not lock LOCK_open for mysql_rm_table_part2 - it does that
  for us now.
sql/mysql_priv.h:
  Remove an unused flag.
sql/sql_class.h:
  Fix an unrelated compiler warning.
sql/sql_db.cc:
  Adjust to the changed signature.
sql/sql_table.cc:
  mysql_rm_table_part2: we need to keep LOCK_open while calling
  storage engine functions, even though now 
  we have an exclusive lock on the table name. Some of them assume that it's 
  kept and attempt to unlock it.

sql/ha_ndbcluster.cc

@@ -7019,8 +7019,6 @@ int ndbcluster_find_files(handlerton *hton, THD *thd,
     }
   }
 
-  // Lock mutex before deleting and creating frm files
-  pthread_mutex_lock(&LOCK_open);
   if (!global_read_lock)
   {
     // Delete old files
@@ -7037,14 +7035,14 @@ int ndbcluster_find_files(handlerton *hton, THD *thd,
                                  FALSE,   /* if_exists */
                                  FALSE,   /* drop_temporary */ 
                                  FALSE,   /* drop_view */
-                                 TRUE,    /* dont_log_query*/ 
-                                 FALSE);  /* need lock open */
+                                 TRUE     /* dont_log_query*/);
 
       /* Clear error message that is returned when table is deleted */
       thd->clear_error();
     }
   }
 
+  pthread_mutex_lock(&LOCK_open);
   // Create new files
   List_iterator_fast<char> it2(create_list);
   while ((file_name=it2++))

sql/mysql_priv.h

@@ -901,8 +901,7 @@ void mysql_client_binlog_statement(THD *thd);
 bool mysql_rm_table(THD *thd,TABLE_LIST *tables, my_bool if_exists,
                     my_bool drop_temporary);
 int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
-                         bool drop_temporary, bool drop_view, bool log_query,
-                         bool need_lock_open);
+                         bool drop_temporary, bool drop_view, bool log_query);
 bool quick_rm_table(handlerton *base,const char *db,
                     const char *table_name, uint flags);
 void close_cached_table(THD *thd, TABLE *table);

sql/sql_class.h

@@ -1999,8 +1999,8 @@ class select_insert :public select_result_interceptor {
 class select_create: public select_insert {
   ORDER *group;
   TABLE_LIST *create_table;
-  TABLE_LIST *select_tables;
   HA_CREATE_INFO *create_info;
+  TABLE_LIST *select_tables;
   Alter_info *alter_info;
   Field **field;
 public:

sql/sql_db.cc

@@ -1113,7 +1113,7 @@ static long mysql_rm_known_files(THD *thd, MY_DIR *dirp, const char *db,
     }
   }
   if (thd->killed ||
-      (tot_list && mysql_rm_table_part2(thd, tot_list, 1, 0, 1, 1, 1)))
+      (tot_list && mysql_rm_table_part2(thd, tot_list, 1, 0, 1, 1)))
     goto err;
 
   /* Remove RAID directories */

sql/sql_table.cc

@@ -1430,11 +1430,7 @@ bool mysql_rm_table(THD *thd,TABLE_LIST *tables, my_bool if_exists,
     LOCK_open during wait_if_global_read_lock(), other threads could not
     close their tables. This would make a pretty deadlock.
   */
-  error= mysql_rm_table_part2(thd, tables, if_exists, drop_temporary, 0, 0, 1);
-  pthread_mutex_lock(&thd->mysys_var->mutex);
-  thd->mysys_var->current_mutex= 0;
-  thd->mysys_var->current_cond= 0;
-  pthread_mutex_unlock(&thd->mysys_var->mutex);
+  error= mysql_rm_table_part2(thd, tables, if_exists, drop_temporary, 0, 0);
 
   if (need_start_waiters)
     start_waiting_global_read_lock(thd);
@@ -1477,7 +1473,7 @@ bool mysql_rm_table(THD *thd,TABLE_LIST *tables, my_bool if_exists,
 
 int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
 			 bool drop_temporary, bool drop_view,
-			 bool dont_log_query, bool need_lock_open)
+			 bool dont_log_query)
 {
   TABLE_LIST *table;
   char path[FN_REFLEN], *alias;
@@ -1489,9 +1485,6 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
   String built_query;
   DBUG_ENTER("mysql_rm_table_part2");
 
-  if (need_lock_open)
-    pthread_mutex_lock(&LOCK_open);
-
   LINT_INIT(alias);
   LINT_INIT(path_length);
 
@@ -1503,6 +1496,9 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
     else
       built_query.append("DROP TABLE ");
   }
+
+  pthread_mutex_lock(&LOCK_open);
+
   /*
     If we have the table in the definition cache, we don't have to check the
     .frm file to find if the table is a normal table (not view) and what
@@ -1522,20 +1518,17 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
                            table->table_name_length, table->table_name, 1))
     {
       my_error(ER_BAD_LOG_STATEMENT, MYF(0), "DROP");
+      pthread_mutex_unlock(&LOCK_open);
       DBUG_RETURN(1);
     }
   }
 
   if (!drop_temporary && lock_table_names_exclusively(thd, tables))
   {
-    if (need_lock_open)
-      pthread_mutex_unlock(&LOCK_open);
+    pthread_mutex_unlock(&LOCK_open);
     DBUG_RETURN(1);
   }
 
-  if (need_lock_open)
-    pthread_mutex_unlock(&LOCK_open);
-
   /* Don't give warnings for not found errors, as we already generate notes */
   thd->no_warnings_for_error= 1;
 
@@ -1545,7 +1538,7 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
     handlerton *table_type;
     enum legacy_db_type frm_db_type;
 
-    mysql_ha_flush(thd, table, MYSQL_HA_CLOSE_FINAL, !need_lock_open);
+    mysql_ha_flush(thd, table, MYSQL_HA_CLOSE_FINAL, 1);
     if (!close_temporary_table(thd, table))
     {
       tmp_table_deleted=1;
@@ -1582,8 +1575,6 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
     {
       TABLE *locked_table;
       abort_locked_tables(thd, db, table->table_name);
-      if (need_lock_open)
-        pthread_mutex_lock(&LOCK_open);
       remove_table_from_cache(thd, db, table->table_name,
 	                      RTFC_WAIT_OTHER_THREAD_FLAG |
 			      RTFC_CHECK_KILLED_FLAG);
@@ -1594,13 +1585,10 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
       if ((locked_table= drop_locked_tables(thd, db, table->table_name)))
         table->table= locked_table;
 
-      if (need_lock_open)
-        pthread_mutex_unlock(&LOCK_open);
-
       if (thd->killed)
       {
-        thd->no_warnings_for_error= 0;
-	DBUG_RETURN(-1);
+        error= -1;
+        goto err_with_placeholders;
       }
       alias= (lower_case_table_names == 2) ? table->alias : table->table_name;
       /* remove .frm file and engine files */
@@ -1663,6 +1651,11 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
       wrong_tables.append(String(table->table_name,system_charset_info));
     }
   }
+  /*
+    It's safe to unlock LOCK_open: we have an exclusive lock
+    on the table name.
+  */
+  pthread_mutex_unlock(&LOCK_open);
   thd->tmp_table_used= tmp_table_deleted;
   error= 0;
   if (wrong_tables.length())
@@ -1722,11 +1715,10 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
       */
     }
   }
-  if (need_lock_open)
-    pthread_mutex_lock(&LOCK_open);
+  pthread_mutex_lock(&LOCK_open);
+err_with_placeholders:
   unlock_table_names(thd, tables, (TABLE_LIST*) 0);
-  if (need_lock_open)
-    pthread_mutex_unlock(&LOCK_open);
+  pthread_mutex_unlock(&LOCK_open);
   thd->no_warnings_for_error= 0;
   DBUG_RETURN(error);
 }