A 5.5 version of the fix for Bug #54360 "Deadlock DROP/ALTER/CREATE

DATABASE with open HANDLER"

Remove LOCK_create_db, database name locks, and use metadata locks instead.
This exposes CREATE/DROP/ALTER DATABASE statements to the graph-based
deadlock detector in MDL, and paves the way for a safe, deadlock-free
implementation of RENAME DATABASE.

Database DDL statements will now take exclusive metadata locks on
the database name, while table/view/routine DDL statements take
intention exclusive locks on the database name. This prevents race
conditions between database DDL and table/view/routine DDL.
(e.g. DROP DATABASE with concurrent CREATE/ALTER/DROP TABLE)

By adding database name locks, this patch implements
WL#4450 "DDL locking: CREATE/DROP DATABASE must use database locks" and
WL#4985 "DDL locking: namespace/hierarchical locks".

The patch also changes code to use init_one_table() where appropriate.
The new lock_table_names() function requires TABLE_LIST::db_length to
be set correctly, and this is taken care of by init_one_table().

This patch also adds a simple template to help work with 
the mysys HASH data structure.

Most of the patch was written by Konstantin Osipov.

20 files changed, 443 insertions, 631 deletions

diff --git a/sql/lock.cc b/sql/lock.cc
index de0f39018f7..dcee018276f 100644
--- a/sql/lock.cc
+++ b/sql/lock.cc
@@ -747,62 +747,48 @@ static MYSQL_LOCK *get_lock_data(THD *thd, TABLE **table_ptr, uint count,
 }
 
 
-/*****************************************************************************
-  Lock table based on the name.
-  This is used when we need total access to a closed, not open table
-*****************************************************************************/
-
 /**
-   Obtain exclusive metadata locks on the list of tables.
+  Obtain an exclusive metadata lock on a schema name.
 
-   @param thd         Thread handle
-   @param table_list  List of tables to lock
+  @param thd         Thread handle.
+  @param db          The database name.
 
-   @note This function assumes that no metadata locks were acquired
-         before calling it. Also it cannot be called while holding
-         LOCK_open mutex. Both these invariants are enforced by asserts
-         in MDL_context::acquire_locks().
-   @note Initialization of MDL_request members of TABLE_LIST elements
-         is a responsibility of the caller.
+  This function cannot be called while holding LOCK_open mutex.
+  To avoid deadlocks, we do not try to obtain exclusive metadata
+  locks in LOCK TABLES mode, since in this mode there may be
+  other metadata locks already taken by the current connection,
+  and we must not wait for MDL locks while holding locks.
 
-   @retval FALSE  Success.
-   @retval TRUE   Failure (OOM or thread was killed).
+  @retval FALSE  Success.
+  @retval TRUE   Failure: we're in LOCK TABLES mode, or out of memory,
+                 or this connection was killed.
 */
 
-bool lock_table_names(THD *thd, TABLE_LIST *table_list)
+bool lock_schema_name(THD *thd, const char *db)
 {
   MDL_request_list mdl_requests;
   MDL_request global_request;
-  TABLE_LIST *lock_table;
+  MDL_request mdl_request;
 
-  global_request.init(MDL_key::GLOBAL, "", "", MDL_INTENTION_EXCLUSIVE);
+  if (thd->locked_tables_mode)
+  {
+    my_message(ER_LOCK_OR_ACTIVE_TRANSACTION,
+               ER(ER_LOCK_OR_ACTIVE_TRANSACTION), MYF(0));
+    return TRUE;
+  }
 
-  for (lock_table= table_list; lock_table; lock_table= lock_table->next_local)
-    mdl_requests.push_front(&lock_table->mdl_request);
+  global_request.init(MDL_key::GLOBAL, "", "", MDL_INTENTION_EXCLUSIVE);
+  mdl_request.init(MDL_key::SCHEMA, db, "", MDL_EXCLUSIVE);
 
+  mdl_requests.push_front(&mdl_request);
   mdl_requests.push_front(&global_request);
 
   if (thd->mdl_context.acquire_locks(&mdl_requests,
                                      thd->variables.lock_wait_timeout))
-    return 1;
-
-  return 0;
-}
-
-
-/**
-   Release all metadata locks previously obtained by lock_table_names().
-
-   @param thd  Thread handle.
-
-   @note Cannot be called while holding LOCK_open mutex.
-*/
+    return TRUE;
 
-void unlock_table_names(THD *thd)
-{
-  DBUG_ENTER("unlock_table_names");
-  thd->mdl_context.release_transactional_locks();
-  DBUG_VOID_RETURN;
+  DEBUG_SYNC(thd, "after_wait_locked_schema_name");
+  return FALSE;
 }
 
 
@@ -837,6 +823,7 @@ bool lock_routine_name(THD *thd, bool is_function,
                                          MDL_key::PROCEDURE);
   MDL_request_list mdl_requests;
   MDL_request global_request;
+  MDL_request schema_request;
   MDL_request mdl_request;
 
   if (thd->locked_tables_mode)
@@ -850,9 +837,11 @@ bool lock_routine_name(THD *thd, bool is_function,
   DEBUG_SYNC(thd, "before_wait_locked_pname");
 
   global_request.init(MDL_key::GLOBAL, "", "", MDL_INTENTION_EXCLUSIVE);
+  schema_request.init(MDL_key::SCHEMA, db, "", MDL_INTENTION_EXCLUSIVE);
   mdl_request.init(mdl_type, db, name, MDL_EXCLUSIVE);
 
   mdl_requests.push_front(&mdl_request);
+  mdl_requests.push_front(&schema_request);
   mdl_requests.push_front(&global_request);
 
   if (thd->mdl_context.acquire_locks(&mdl_requests,
diff --git a/sql/lock.h b/sql/lock.h
index 4bdf0085d07..0083dd3ba18 100644
--- a/sql/lock.h
+++ b/sql/lock.h
@@ -62,8 +62,7 @@ bool mysql_lock_abort_for_thread(THD *thd, TABLE *table);
 MYSQL_LOCK *mysql_lock_merge(MYSQL_LOCK *a,MYSQL_LOCK *b);
 void broadcast_refresh(void);
 /* Lock based on name */
-bool lock_table_names(THD *thd, TABLE_LIST *table_list);
-void unlock_table_names(THD *thd);
+bool lock_schema_name(THD *thd, const char *db);
 /* Lock based on stored routine name */
 bool lock_routine_name(THD *thd, bool is_function, const char *db,
                        const char *name);
diff --git a/sql/mdl.cc b/sql/mdl.cc
index 22ad15d2360..631d8f52b5f 100644
--- a/sql/mdl.cc
+++ b/sql/mdl.cc
@@ -406,14 +406,15 @@ public:
 
 
 /**
-  An implementation of the global metadata lock. The only locking modes
-  which are supported at the moment are SHARED and INTENTION EXCLUSIVE.
+  An implementation of the scoped metadata lock. The only locking modes
+  which are supported at the moment are SHARED and INTENTION EXCLUSIVE
+  and EXCLUSIVE
 */
 
-class MDL_global_lock : public MDL_lock
+class MDL_scoped_lock : public MDL_lock
 {
 public:
-  MDL_global_lock(const MDL_key *key_arg)
+  MDL_scoped_lock(const MDL_key *key_arg)
     : MDL_lock(key_arg)
   { }
 
@@ -857,7 +858,8 @@ inline MDL_lock *MDL_lock::create(const MDL_key *mdl_key)
   switch (mdl_key->mdl_namespace())
   {
     case MDL_key::GLOBAL:
-      return new MDL_global_lock(mdl_key);
+    case MDL_key::SCHEMA:
+      return new MDL_scoped_lock(mdl_key);
     case MDL_key::TABLE:
       return new MDL_table_lock(mdl_key);
     default:
@@ -1217,61 +1219,66 @@ void MDL_lock::reschedule_waiters()
 
 
 /**
-  Compatibility (or rather "incompatibility") matrices for global metadata
+  Compatibility (or rather "incompatibility") matrices for scoped metadata
   lock. Arrays of bitmaps which elements specify which granted/waiting locks
   are incompatible with type of lock being requested.
 
-  Here is how types of individual locks are translated to type of global lock:
+  Here is how types of individual locks are translated to type of scoped lock:
 
     ----------------+-------------+
     Type of request | Correspond. |
-    for indiv. lock | global lock |
+    for indiv. lock | scoped lock |
     ----------------+-------------+
     S, SH, SR, SW   |   IS        |
     SNW, SNRW, X    |   IX        |
     SNW, SNRW -> X  |   IX (*)    |
 
   The first array specifies if particular type of request can be satisfied
-  if there is granted global lock of certain type.
+  if there is granted scoped lock of certain type.
 
-             | Type of active |
-     Request |   global lock  |
-      type   | IS(**) IX   S  |
-    ---------+----------------+
-    IS       |  +      +   +  |
-    IX       |  +      +   -  |
-    S        |  +      -   +  |
+             | Type of active   |
+     Request |   scoped lock    |
+      type   | IS(**) IX   S  X |
+    ---------+------------------+
+    IS       |  +      +   +  + |
+    IX       |  +      +   -  - |
+    S        |  +      -   +  - |
+    X        |  +      -   -  - |
 
   The second array specifies if particular type of request can be satisfied
-  if there is already waiting request for the global lock of certain type.
+  if there is already waiting request for the scoped lock of certain type.
   I.e. it specifies what is the priority of different lock types.
 
-             |    Pending   |
-     Request |  global lock |
-      type   | IS(**) IX  S |
-    ---------+--------------+
-    IS       |  +      +  + |
-    IX       |  +      +  - |
-    S        |  +      +  + |
+             |    Pending      |
+     Request |  scoped lock    |
+      type   | IS(**) IX  S  X |
+    ---------+-----------------+
+    IS       |  +      +  +  + |
+    IX       |  +      +  -  - |
+    S        |  +      +  +  - |
+    X        |  +      +  +  + |
 
   Here: "+" -- means that request can be satisfied
         "-" -- means that request can't be satisfied and should wait
 
-  (*)   Since for upgradable locks we always take intention exclusive global
+  (*)   Since for upgradable locks we always take intention exclusive scoped
         lock at the same time when obtaining the shared lock, there is no
         need to obtain such lock during the upgrade itself.
-  (**)  Since intention shared global locks are compatible with all other
+  (**)  Since intention shared scoped locks are compatible with all other
         type of locks we don't even have any accounting for them.
 */
 
-const MDL_lock::bitmap_t MDL_global_lock::m_granted_incompatible[MDL_TYPE_END] =
+const MDL_lock::bitmap_t MDL_scoped_lock::m_granted_incompatible[MDL_TYPE_END] =
 {
-  MDL_BIT(MDL_SHARED), MDL_BIT(MDL_INTENTION_EXCLUSIVE), 0, 0, 0, 0, 0, 0
+  MDL_BIT(MDL_EXCLUSIVE) | MDL_BIT(MDL_SHARED),
+  MDL_BIT(MDL_EXCLUSIVE) | MDL_BIT(MDL_INTENTION_EXCLUSIVE), 0, 0, 0, 0, 0,
+  MDL_BIT(MDL_EXCLUSIVE) | MDL_BIT(MDL_SHARED) | MDL_BIT(MDL_INTENTION_EXCLUSIVE)
 };
 
-const MDL_lock::bitmap_t MDL_global_lock::m_waiting_incompatible[MDL_TYPE_END] =
+const MDL_lock::bitmap_t MDL_scoped_lock::m_waiting_incompatible[MDL_TYPE_END] =
 {
-  MDL_BIT(MDL_SHARED), 0, 0, 0, 0, 0, 0, 0
+  MDL_BIT(MDL_EXCLUSIVE) | MDL_BIT(MDL_SHARED),
+  MDL_BIT(MDL_EXCLUSIVE), 0, 0, 0, 0, 0, 0
 };
 
 
@@ -1912,7 +1919,7 @@ extern "C" int mdl_request_ptr_cmp(const void* ptr1, const void* ptr2)
   @note The list of requests should not contain non-exclusive lock requests.
         There should not be any acquired locks in the context.
 
-  @note Assumes that one already owns global intention exclusive lock.
+  @note Assumes that one already owns scoped intention exclusive lock.
 
   @retval FALSE  Success
   @retval TRUE   Failure
@@ -1929,13 +1936,6 @@ bool MDL_context::acquire_locks(MDL_request_list *mdl_requests,
   if (req_count == 0)
     return FALSE;
 
-  /*
-    To reduce deadlocks, the server acquires all exclusive
-    locks at once. For shared locks, try_acquire_lock() is
-    used instead.
-  */
-  DBUG_ASSERT(m_tickets.is_empty() || m_tickets.front() == m_trans_sentinel);
-
   /* Sort requests according to MDL_key. */
   if (! (sort_buf= (MDL_request **)my_malloc(req_count *
                                              sizeof(MDL_request*),
diff --git a/sql/mdl.h b/sql/mdl.h
index ad3945f524c..5c58289aea2 100644
--- a/sql/mdl.h
+++ b/sql/mdl.h
@@ -40,15 +40,16 @@ class Deadlock_detection_visitor;
   Type of metadata lock request.
 
   @sa Comments for MDL_object_lock::can_grant_lock() and
-      MDL_global_lock::can_grant_lock() for details.
+      MDL_scoped_lock::can_grant_lock() for details.
 */
 
 enum enum_mdl_type {
   /*
-    An intention exclusive metadata lock. Used only for global locks.
+    An intention exclusive metadata lock. Used only for scoped locks.
     Owner of this type of lock can acquire upgradable exclusive locks on
     individual objects.
-    Compatible with other IX locks, but is incompatible with global S lock.
+    Compatible with other IX locks, but is incompatible with scoped S and
+    X locks.
   */
   MDL_INTENTION_EXCLUSIVE= 0,
   /*
@@ -179,6 +180,7 @@ public:
     MDL_key is also used outside of the MDL subsystem.
   */
   enum enum_mdl_namespace { GLOBAL=0,
+                            SCHEMA,
                             TABLE,
                             FUNCTION,
                             PROCEDURE,
@@ -646,6 +648,8 @@ private:
       closes all open HANDLERs.
       However, one can open a few HANDLERs after entering the
       read only mode.
+    * LOCK TABLES locks include intention exclusive locks on
+      involved schemas.
   */
   Ticket_list m_tickets;
   /**
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index 28cad51aa41..0ba796a5eb8 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -27,8 +27,8 @@
                           // reset_status_vars
 #include "strfunc.h"      // find_set_from_flags
 #include "parse_file.h"   // File_parser_dummy_hook
-#include "sql_db.h"       // my_database_names_free,
-                          // my_database_names_init
+#include "sql_db.h"       // my_dboptions_cache_free
+                          // my_dboptions_cache_init
 #include "sql_table.h"    // release_ddl_log, execute_ddl_log_recovery
 #include "sql_connect.h"  // free_max_user_conn, init_max_user_conn,
                           // handle_one_connection
@@ -653,7 +653,7 @@ SHOW_COMP_OPTION have_profiling;
 pthread_key(MEM_ROOT**,THR_MALLOC);
 pthread_key(THD*, THR_THD);
 mysql_mutex_t LOCK_thread_count;
-mysql_mutex_t LOCK_mysql_create_db, LOCK_open,
+mysql_mutex_t LOCK_open,
   LOCK_mapped_file, LOCK_status, LOCK_global_read_lock,
   LOCK_error_log, LOCK_uuid_generator,
   LOCK_delayed_insert, LOCK_delayed_status, LOCK_delayed_create,
@@ -1488,7 +1488,7 @@ void clean_up(bool print_message)
     bitmap_free(&slave_error_mask);
 #endif
   my_tz_free();
-  my_database_names_free();
+  my_dboptions_cache_free();
 #ifndef NO_EMBEDDED_ACCESS_CHECKS
   servers_free(1);
   acl_free(1);
@@ -1597,8 +1597,6 @@ static void wait_for_signal_thread_to_end()
 
 static void clean_up_mutexes()
 {
-  mysql_mutex_destroy(&LOCK_mysql_create_db);
-  mysql_mutex_destroy(&LOCK_lock_db);
   mysql_rwlock_destroy(&LOCK_grant);
   mysql_mutex_destroy(&LOCK_open);
   mysql_mutex_destroy(&LOCK_thread_count);
@@ -3730,7 +3728,7 @@ static int init_common_variables()
   use_temp_pool= 0;
 #endif
 
-  if (my_database_names_init())
+  if (my_dboptions_cache_init())
     return 1;
 
   /*
@@ -3787,9 +3785,6 @@ You should consider changing lower_case_table_names to 1 or 2",
 
 static int init_thread_environment()
 {
-  mysql_mutex_init(key_LOCK_mysql_create_db,
-                   &LOCK_mysql_create_db, MY_MUTEX_INIT_SLOW);
-  mysql_mutex_init(key_LOCK_lock_db, &LOCK_lock_db, MY_MUTEX_INIT_SLOW);
   mysql_mutex_init(key_LOCK_open, &LOCK_open, MY_MUTEX_INIT_FAST);
   mysql_mutex_init(key_LOCK_thread_count, &LOCK_thread_count, MY_MUTEX_INIT_FAST);
   mysql_mutex_init(key_LOCK_mapped_file, &LOCK_mapped_file, MY_MUTEX_INIT_SLOW);
@@ -8007,8 +8002,8 @@ PSI_mutex_key key_BINLOG_LOCK_index, key_BINLOG_LOCK_prep_xids,
   key_LOCK_connection_count, key_LOCK_crypt, key_LOCK_delayed_create,
   key_LOCK_delayed_insert, key_LOCK_delayed_status, key_LOCK_error_log,
   key_LOCK_gdl, key_LOCK_global_read_lock, key_LOCK_global_system_variables,
-  key_LOCK_lock_db, key_LOCK_manager, key_LOCK_mapped_file,
-  key_LOCK_mysql_create_db, key_LOCK_open, key_LOCK_prepared_stmt_count,
+  key_LOCK_manager, key_LOCK_mapped_file,
+  key_LOCK_open, key_LOCK_prepared_stmt_count,
   key_LOCK_rpl_status, key_LOCK_server_started, key_LOCK_status,
   key_LOCK_system_variables_hash, key_LOCK_table_share, key_LOCK_thd_data,
   key_LOCK_user_conn, key_LOCK_uuid_generator, key_LOG_LOCK_log,
@@ -8046,10 +8041,8 @@ static PSI_mutex_info all_server_mutexes[]=
   { &key_LOCK_gdl, "LOCK_gdl", PSI_FLAG_GLOBAL},
   { &key_LOCK_global_read_lock, "LOCK_global_read_lock", PSI_FLAG_GLOBAL},
   { &key_LOCK_global_system_variables, "LOCK_global_system_variables", PSI_FLAG_GLOBAL},
-  { &key_LOCK_lock_db, "LOCK_lock_db", PSI_FLAG_GLOBAL},
   { &key_LOCK_manager, "LOCK_manager", PSI_FLAG_GLOBAL},
   { &key_LOCK_mapped_file, "LOCK_mapped_file", PSI_FLAG_GLOBAL},
-  { &key_LOCK_mysql_create_db, "LOCK_mysql_create_db", PSI_FLAG_GLOBAL},
   { &key_LOCK_open, "LOCK_open", PSI_FLAG_GLOBAL},
   { &key_LOCK_prepared_stmt_count, "LOCK_prepared_stmt_count", PSI_FLAG_GLOBAL},
   { &key_LOCK_rpl_status, "LOCK_rpl_status", PSI_FLAG_GLOBAL},
diff --git a/sql/mysqld.h b/sql/mysqld.h
index e14cd15ceb8..b07d148f507 100644
--- a/sql/mysqld.h
+++ b/sql/mysqld.h
@@ -234,8 +234,8 @@ extern PSI_mutex_key key_BINLOG_LOCK_index, key_BINLOG_LOCK_prep_xids,
   key_LOCK_connection_count, key_LOCK_crypt, key_LOCK_delayed_create,
   key_LOCK_delayed_insert, key_LOCK_delayed_status, key_LOCK_error_log,
   key_LOCK_gdl, key_LOCK_global_read_lock, key_LOCK_global_system_variables,
-  key_LOCK_lock_db, key_LOCK_logger, key_LOCK_manager, key_LOCK_mapped_file,
-  key_LOCK_mysql_create_db, key_LOCK_open, key_LOCK_prepared_stmt_count,
+  key_LOCK_logger, key_LOCK_manager, key_LOCK_mapped_file,
+  key_LOCK_open, key_LOCK_prepared_stmt_count,
   key_LOCK_rpl_status, key_LOCK_server_started, key_LOCK_status,
   key_LOCK_table_share, key_LOCK_thd_data,
   key_LOCK_user_conn, key_LOCK_uuid_generator, key_LOG_LOCK_log,
@@ -323,7 +323,7 @@ extern MYSQL_PLUGIN_IMPORT key_map key_map_full;          /* Should be threaded
 /*
   Server mutex locks and condition variables.
  */
-extern mysql_mutex_t LOCK_mysql_create_db, LOCK_open, LOCK_lock_db,
+extern mysql_mutex_t LOCK_open,
        LOCK_mapped_file, LOCK_user_locks, LOCK_status,
        LOCK_error_log, LOCK_delayed_insert, LOCK_uuid_generator,
        LOCK_delayed_status, LOCK_delayed_create, LOCK_crypt, LOCK_timezone,
diff --git a/sql/sql_acl.cc b/sql/sql_acl.cc
index ec25e4cb68b..99226bad03f 100644
--- a/sql/sql_acl.cc
+++ b/sql/sql_acl.cc
@@ -677,16 +677,15 @@ my_bool acl_reload(THD *thd)
     To avoid deadlocks we should obtain table locks before
     obtaining acl_cache->lock mutex.
   */
-  bzero((char*) tables, sizeof(tables));
-  tables[0].alias= tables[0].table_name= (char*) "host";
-  tables[1].alias= tables[1].table_name= (char*) "user";
-  tables[2].alias= tables[2].table_name= (char*) "db";
-  tables[0].db=tables[1].db=tables[2].db=(char*) "mysql";
+  tables[0].init_one_table(C_STRING_WITH_LEN("mysql"),
+                           C_STRING_WITH_LEN("host"), "host", TL_READ);
+  tables[1].init_one_table(C_STRING_WITH_LEN("mysql"),
+                           C_STRING_WITH_LEN("user"), "user", TL_READ);
+  tables[2].init_one_table(C_STRING_WITH_LEN("mysql"),
+                           C_STRING_WITH_LEN("db"), "db", TL_READ);
   tables[0].next_local= tables[0].next_global= tables+1;
   tables[1].next_local= tables[1].next_global= tables+2;
-  tables[0].lock_type=tables[1].lock_type=tables[2].lock_type=TL_READ;
   tables[0].open_type= tables[1].open_type= tables[2].open_type= OT_BASE_ONLY;
-  init_mdl_requests(tables);
 
   if (open_and_lock_tables(thd, tables, FALSE, MYSQL_LOCK_IGNORE_TIMEOUT))
   {
@@ -1925,9 +1924,8 @@ static bool test_if_create_new_users(THD *thd)
   {
     TABLE_LIST tl;
     ulong db_access;
-    bzero((char*) &tl,sizeof(tl));
-    tl.db=	   (char*) "mysql";
-    tl.table_name=  (char*) "user";
+    tl.init_one_table(C_STRING_WITH_LEN("mysql"),
+                      C_STRING_WITH_LEN("user"), "user", TL_WRITE);
     create_new_users= 1;
 
     db_access=acl_get(sctx->host, sctx->ip,
@@ -3107,20 +3105,18 @@ int mysql_table_grant(THD *thd, TABLE_LIST *table_list,
 
   /* open the mysql.tables_priv and mysql.columns_priv tables */
 
-  bzero((char*) &tables,sizeof(tables));
-  tables[0].alias=tables[0].table_name= (char*) "user";
-  tables[1].alias=tables[1].table_name= (char*) "tables_priv";
-  tables[2].alias=tables[2].table_name= (char*) "columns_priv";
+  tables[0].init_one_table(C_STRING_WITH_LEN("mysql"),
+                           C_STRING_WITH_LEN("user"), "user", TL_WRITE);
+  tables[1].init_one_table(C_STRING_WITH_LEN("mysql"),
+                           C_STRING_WITH_LEN("tables_priv"),
+                           "tables_priv", TL_WRITE);
+  tables[2].init_one_table(C_STRING_WITH_LEN("mysql"),
+                           C_STRING_WITH_LEN("columns_priv"),
+                           "columns_priv", TL_WRITE);
   tables[0].next_local= tables[0].next_global= tables+1;
   /* Don't open column table if we don't need it ! */
-  tables[1].next_local=
-    tables[1].next_global= ((column_priv ||
-			     (revoke_grant &&
-			      ((rights & COL_ACLS) || columns.elements)))
-			    ? tables+2 : 0);
-  tables[0].lock_type=tables[1].lock_type=tables[2].lock_type=TL_WRITE;
-  tables[0].db=tables[1].db=tables[2].db=(char*) "mysql";
-  init_mdl_requests(tables);
+  if (column_priv || (revoke_grant && ((rights & COL_ACLS) || columns.elements)))
+    tables[1].next_local= tables[1].next_global= tables+2;
 
   /*
     This statement will be replicated as a statement, even when using
@@ -3354,13 +3350,11 @@ bool mysql_routine_grant(THD *thd, TABLE_LIST *table_list, bool is_proc,
 
   /* open the mysql.user and mysql.procs_priv tables */
 
-  bzero((char*) &tables,sizeof(tables));
-  tables[0].alias=tables[0].table_name= (char*) "user";
-  tables[1].alias=tables[1].table_name= (char*) "procs_priv";
+  tables[0].init_one_table(C_STRING_WITH_LEN("mysql"),
+                           C_STRING_WITH_LEN("user"), "user", TL_WRITE);
+  tables[1].init_one_table(C_STRING_WITH_LEN("mysql"),
+                           C_STRING_WITH_LEN("procs_priv"), "procs_priv", TL_WRITE);
   tables[0].next_local= tables[0].next_global= tables+1;
-  tables[0].lock_type=tables[1].lock_type=TL_WRITE;
-  tables[0].db=tables[1].db=(char*) "mysql";
-  init_mdl_requests(tables);
 
   /*
     This statement will be replicated as a statement, even when using
@@ -3511,13 +3505,11 @@ bool mysql_grant(THD *thd, const char *db, List <LEX_USER> &list,
   }
 
   /* open the mysql.user and mysql.db tables */
-  bzero((char*) &tables,sizeof(tables));
-  tables[0].alias=tables[0].table_name=(char*) "user";
-  tables[1].alias=tables[1].table_name=(char*) "db";
+  tables[0].init_one_table(C_STRING_WITH_LEN("mysql"),
+                           C_STRING_WITH_LEN("user"), "user", TL_WRITE);
+  tables[1].init_one_table(C_STRING_WITH_LEN("mysql"),
+                           C_STRING_WITH_LEN("db"), "db", TL_WRITE);
   tables[0].next_local= tables[0].next_global= tables+1;
-  tables[0].lock_type=tables[1].lock_type=TL_WRITE;
-  tables[0].db=tables[1].db=(char*) "mysql";
-  init_mdl_requests(tables);
 
   /*
     This statement will be replicated as a statement, even when using
@@ -3930,14 +3922,14 @@ my_bool grant_reload(THD *thd)
   if (!initialized)
     DBUG_RETURN(0);
 
-  bzero((char*) tables, sizeof(tables));
-  tables[0].alias= tables[0].table_name= (char*) "tables_priv";
-  tables[1].alias= tables[1].table_name= (char*) "columns_priv";
-  tables[0].db= tables[1].db= (char *) "mysql";
+  tables[0].init_one_table(C_STRING_WITH_LEN("mysql"),
+                           C_STRING_WITH_LEN("tables_priv"),
+                           "tables_priv", TL_READ);
+  tables[1].init_one_table(C_STRING_WITH_LEN("mysql"),
+                           C_STRING_WITH_LEN("columns_priv"),
+                           "columns_priv", TL_READ);
   tables[0].next_local= tables[0].next_global= tables+1;
-  tables[0].lock_type= tables[1].lock_type= TL_READ;
   tables[0].open_type= tables[1].open_type= OT_BASE_ONLY;
-  init_mdl_requests(tables);
 
   /*
     To avoid deadlocks we should obtain table locks before
@@ -5209,22 +5201,23 @@ int open_grant_tables(THD *thd, TABLE_LIST *tables)
     DBUG_RETURN(-1);
   }
 
-  bzero((char*) tables, GRANT_TABLES*sizeof(*tables));
-  tables->alias= tables->table_name= (char*) "user";
-  (tables+1)->alias= (tables+1)->table_name= (char*) "db";
-  (tables+2)->alias= (tables+2)->table_name= (char*) "tables_priv";
-  (tables+3)->alias= (tables+3)->table_name= (char*) "columns_priv";
-  (tables+4)->alias= (tables+4)->table_name= (char*) "procs_priv";
+  tables->init_one_table(C_STRING_WITH_LEN("mysql"),
+                         C_STRING_WITH_LEN("user"), "user", TL_WRITE);
+  (tables+1)->init_one_table(C_STRING_WITH_LEN("mysql"),
+                             C_STRING_WITH_LEN("db"), "db", TL_WRITE);
+  (tables+2)->init_one_table(C_STRING_WITH_LEN("mysql"),
+                             C_STRING_WITH_LEN("tables_priv"),
+                             "tables_priv", TL_WRITE);
+  (tables+3)->init_one_table(C_STRING_WITH_LEN("mysql"),
+                             C_STRING_WITH_LEN("columns_priv"),
+                             "columns_priv", TL_WRITE);
+  (tables+4)->init_one_table(C_STRING_WITH_LEN("mysql"),
+                             C_STRING_WITH_LEN("procs_priv"),
+                             "procs_priv", TL_WRITE);
   tables->next_local= tables->next_global= tables+1;
   (tables+1)->next_local= (tables+1)->next_global= tables+2;
   (tables+2)->next_local= (tables+2)->next_global= tables+3;
   (tables+3)->next_local= (tables+3)->next_global= tables+4;
-  tables->lock_type= (tables+1)->lock_type=
-    (tables+2)->lock_type= (tables+3)->lock_type= 
-    (tables+4)->lock_type= TL_WRITE;
-  tables->db= (tables+1)->db= (tables+2)->db= 
-    (tables+3)->db= (tables+4)->db= (char*) "mysql";
-  init_mdl_requests(tables);
 
 #ifdef HAVE_REPLICATION
   /*
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index d69a5b1aa77..426c9db2717 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -53,6 +53,7 @@
 #include "rpl_filter.h"
 #include "sql_table.h"                          // build_table_filename
 #include "datadict.h"   // dd_frm_type()
+#include "sql_hset.h"   // Hash_set
 #ifdef  __WIN__
 #include <io.h>
 #endif
@@ -4029,7 +4030,6 @@ end_unlock:
 Open_table_context::Open_table_context(THD *thd, uint flags)
   :m_failed_table(NULL),
    m_start_of_statement_svp(thd->mdl_context.mdl_savepoint()),
-   m_global_mdl_request(NULL),
    m_timeout(flags & MYSQL_LOCK_IGNORE_TIMEOUT ?
              LONG_TIMEOUT : thd->variables.lock_wait_timeout),
    m_flags(flags),
@@ -4041,26 +4041,6 @@ Open_table_context::Open_table_context(THD *thd, uint flags)
 
 
 /**
-  Get MDL_request object for global intention exclusive lock which
-  is acquired during opening tables for statements which take
-  upgradable shared metadata locks.
-*/
-
-MDL_request *Open_table_context::get_global_mdl_request(THD *thd)
-{
-  if (! m_global_mdl_request)
-  {
-    if ((m_global_mdl_request= new (thd->mem_root) MDL_request()))
-    {
-      m_global_mdl_request->init(MDL_key::GLOBAL, "", "",
-                                 MDL_INTENTION_EXCLUSIVE);
-    }
-  }
-  return m_global_mdl_request;
-}
-
-
-/**
   Check if we can back-off and set back off action if we can.
   Otherwise report and return error.
 
@@ -4108,13 +4088,23 @@ request_backoff_action(enum_open_table_action action_arg,
     my_error(ER_LOCK_DEADLOCK, MYF(0));
     return TRUE;
   }
-  m_action= action_arg;
   /*
     If auto-repair or discovery are requested, a pointer to table
     list element must be provided.
   */
-  DBUG_ASSERT((m_action != OT_DISCOVER && m_action != OT_REPAIR) || table);
-  m_failed_table= table;
+  if (table)
+  {
+    DBUG_ASSERT(action_arg == OT_DISCOVER || action_arg == OT_REPAIR);
+    m_failed_table= (TABLE_LIST*) current_thd->alloc(sizeof(TABLE_LIST));
+    if (m_failed_table == NULL)
+      return TRUE;
+    m_failed_table->init_one_table(table->db, table->db_length,
+                                   table->table_name,
+                                   table->table_name_length,
+                                   table->alias, TL_WRITE);
+    m_failed_table->mdl_request.set_type(MDL_EXCLUSIVE);
+  }
+  m_action= action_arg;
   return FALSE;
 }
 
@@ -4136,11 +4126,6 @@ Open_table_context::
 recover_from_failed_open(THD *thd)
 {
   bool result= FALSE;
-  /*
-    Remove reference to released ticket from MDL_request.
-  */
-  if (m_global_mdl_request)
-    m_global_mdl_request->ticket= NULL;
   /* Execute the action. */
   switch (m_action)
   {
@@ -4152,19 +4137,9 @@ recover_from_failed_open(THD *thd)
       break;
     case OT_DISCOVER:
       {
-        MDL_request mdl_global_request;
-        MDL_request mdl_xlock_request(&m_failed_table->mdl_request);
-        MDL_request_list mdl_requests;
-
-        mdl_global_request.init(MDL_key::GLOBAL, "", "",
-                                MDL_INTENTION_EXCLUSIVE);
-        mdl_xlock_request.set_type(MDL_EXCLUSIVE);
-
-        mdl_requests.push_front(&mdl_xlock_request);
-        mdl_requests.push_front(&mdl_global_request);
-
-        if ((result=
-             thd->mdl_context.acquire_locks(&mdl_requests, get_timeout())))
+        if ((result= lock_table_names(thd, m_failed_table, NULL,
+                                      get_timeout(),
+                                      MYSQL_OPEN_SKIP_TEMPORARY)))
           break;
 
         mysql_mutex_lock(&LOCK_open);
@@ -4181,19 +4156,9 @@ recover_from_failed_open(THD *thd)
       }
     case OT_REPAIR:
       {
-        MDL_request mdl_global_request;
-        MDL_request mdl_xlock_request(&m_failed_table->mdl_request);
-        MDL_request_list mdl_requests;
-
-        mdl_global_request.init(MDL_key::GLOBAL, "", "",
-                                MDL_INTENTION_EXCLUSIVE);
-        mdl_xlock_request.set_type(MDL_EXCLUSIVE);
-
-        mdl_requests.push_front(&mdl_xlock_request);
-        mdl_requests.push_front(&mdl_global_request);
-
-        if ((result=
-             thd->mdl_context.acquire_locks(&mdl_requests, get_timeout())))
+        if ((result= lock_table_names(thd, m_failed_table, NULL,
+                                      get_timeout(),
+                                      MYSQL_OPEN_SKIP_TEMPORARY)))
           break;
 
         mysql_mutex_lock(&LOCK_open);
@@ -4688,32 +4653,40 @@ end:
   DBUG_RETURN(error);
 }
 
+extern "C" uchar *schema_set_get_key(const uchar *record, size_t *length,
+                                     my_bool not_used __attribute__((unused)))
+{
+  TABLE_LIST *table=(TABLE_LIST*) record;
+  *length= table->db_length;
+  return (uchar*) table->db;
+}
 
 /**
-  Acquire upgradable (SNW, SNRW) metadata locks on tables to be opened
-  for LOCK TABLES or a DDL statement. Under LOCK TABLES, we can't take
+  Acquire upgradable (SNW, SNRW) metadata locks on tables used by
+  LOCK TABLES or by a DDL statement. Under LOCK TABLES, we can't take
   new locks, so use open_tables_check_upgradable_mdl() instead.
 
-  @param thd           Thread context.
-  @param tables_start  Start of list of tables on which upgradable locks
-                       should be acquired.
-  @param tables_end    End of list of tables.
-  @param ot_ctx        Context of open_tables() operation.
-  @param flags         Bitmap of flags to modify how the tables will be
-                       open, see open_table() description for details.
+  @param thd               Thread context.
+  @param tables_start      Start of list of tables on which upgradable locks
+                           should be acquired.
+  @param tables_end        End of list of tables.
+  @param lock_wait_timeout Seconds to wait before timeout.
+  @param flags             Bitmap of flags to modify how the tables will be
+                           open, see open_table() description for details.
 
   @retval FALSE  Success.
   @retval TRUE   Failure (e.g. connection was killed)
 */
 
-static bool
-open_tables_acquire_upgradable_mdl(THD *thd, TABLE_LIST *tables_start,
-                                   TABLE_LIST *tables_end,
-                                   Open_table_context *ot_ctx,
-                                   uint flags)
+bool
+lock_table_names(THD *thd,
+                 TABLE_LIST *tables_start, TABLE_LIST *tables_end,
+                 ulong lock_wait_timeout, uint flags)
 {
   MDL_request_list mdl_requests;
   TABLE_LIST *table;
+  MDL_request global_request;
+  Hash_set<TABLE_LIST, schema_set_get_key> schema_set;
 
   DBUG_ASSERT(!thd->locked_tables_mode);
 
@@ -4726,30 +4699,37 @@ open_tables_acquire_upgradable_mdl(THD *thd, TABLE_LIST *tables_start,
           (table->open_type != OT_BASE_ONLY &&
            ! (flags & MYSQL_OPEN_SKIP_TEMPORARY) &&
            find_temporary_table(thd, table))))
+    {
+      if (schema_set.insert(table))
+        return TRUE;
       mdl_requests.push_front(&table->mdl_request);
+    }
   }
 
   if (! mdl_requests.is_empty())
   {
-    DEBUG_SYNC(thd, "open_tables_acquire_upgradable_mdl");
-
-    MDL_request *global_request= ot_ctx->get_global_mdl_request(thd);
-
-    if (global_request == NULL)
-      return TRUE;
-    mdl_requests.push_front(global_request);
+    /*
+      Scoped locks: Take intention exclusive locks on all involved
+      schemas.
+    */
+    Hash_set<TABLE_LIST, schema_set_get_key>::Iterator it(schema_set);
+    while ((table= it++))
+    {
+      MDL_request *schema_request= new (thd->mem_root) MDL_request;
+      if (schema_request == NULL)
+        return TRUE;
+      schema_request->init(MDL_key::SCHEMA, table->db, "",
+                           MDL_INTENTION_EXCLUSIVE);
+      mdl_requests.push_front(schema_request);
+    }
+    /* Take the global intention exclusive lock. */
+    global_request.init(MDL_key::GLOBAL, "", "", MDL_INTENTION_EXCLUSIVE);
+    mdl_requests.push_front(&global_request);
   }
 
-  if (thd->mdl_context.acquire_locks(&mdl_requests, ot_ctx->get_timeout()))
+  if (thd->mdl_context.acquire_locks(&mdl_requests, lock_wait_timeout))
     return TRUE;
 
-  for (table= tables_start; table && table != tables_end;
-       table= table->next_global)
-  {
-    if (table->mdl_request.type >= MDL_SHARED_NO_WRITE)
-      table->mdl_request.ticket= NULL;
-  }
-
   return FALSE;
 }
 
@@ -4921,12 +4901,21 @@ restart:
         goto err;
       }
     }
-    else if (open_tables_acquire_upgradable_mdl(thd, *start,
-                                                thd->lex->first_not_own_table(),
-                                                &ot_ctx, flags))
+    else
     {
-      error= TRUE;
-      goto err;
+      TABLE_LIST *table;
+      if (lock_table_names(thd, *start, thd->lex->first_not_own_table(),
+                           ot_ctx.get_timeout(), flags))
+      {
+        error= TRUE;
+        goto err;
+      }
+      for (table= *start; table && table != thd->lex->first_not_own_table();
+           table= table->next_global)
+      {
+        if (table->mdl_request.type >= MDL_SHARED_NO_WRITE)
+          table->mdl_request.ticket= NULL;
+      }
     }
   }
 
diff --git a/sql/sql_base.h b/sql/sql_base.h
index 20a068e27d7..67e5601663a 100644
--- a/sql/sql_base.h
+++ b/sql/sql_base.h
@@ -205,6 +205,9 @@ int setup_ftfuncs(SELECT_LEX* select);
 int init_ftfuncs(THD *thd, SELECT_LEX* select, bool no_order);
 void wait_for_condition(THD *thd, mysql_mutex_t *mutex,
                         mysql_cond_t *cond);
+bool lock_table_names(THD *thd, TABLE_LIST *table_list,
+                      TABLE_LIST *table_list_end, ulong lock_wait_timeout,
+                      uint flags);
 bool open_tables(THD *thd, TABLE_LIST **tables, uint *counter, uint flags,
                  Prelocking_strategy *prelocking_strategy);
 /* open_and_lock_tables with optional derived handling */
@@ -480,8 +483,6 @@ public:
     return m_start_of_statement_svp;
   }
 
-  MDL_request *get_global_mdl_request(THD *thd);
-
   inline ulong get_timeout() const
   {
     return m_timeout;
@@ -499,11 +500,6 @@ private:
   TABLE_LIST *m_failed_table;
   MDL_ticket *m_start_of_statement_svp;
   /**
-    Request object for global intention exclusive lock which is acquired during
-    opening tables for statements which take upgradable shared metadata locks.
-  */
-  MDL_request *m_global_mdl_request;
-  /**
     Lock timeout in seconds. Initialized to LONG_TIMEOUT when opening system
     tables or to the "lock_wait_timeout" system variable for regular tables.
   */
diff --git a/sql/sql_db.cc b/sql/sql_db.cc
index 5e992d2391c..c4a7dc53972 100644
--- a/sql/sql_db.cc
+++ b/sql/sql_db.cc
@@ -58,106 +58,6 @@ static void mysql_change_db_impl(THD *thd,
                                  CHARSET_INFO *new_db_charset);
 
 
-/* Database lock hash */
-HASH lock_db_cache;
-mysql_mutex_t LOCK_lock_db;
-int creating_database= 0;  // how many database locks are made
-
-
-/* Structure for database lock */
-typedef struct my_dblock_st
-{
-  char *name;        /* Database name        */
-  uint name_length;  /* Database length name */
-} my_dblock_t;
-
-
-/*
-  lock_db key.
-*/
-
-extern "C" uchar* lock_db_get_key(my_dblock_t *, size_t *, my_bool not_used);
-
-uchar* lock_db_get_key(my_dblock_t *ptr, size_t *length,
-                       my_bool not_used __attribute__((unused)))
-{
-  *length= ptr->name_length;
-  return (uchar*) ptr->name;
-}
-
-
-/*
-  Free lock_db hash element.
-*/
-
-extern "C" void lock_db_free_element(void *ptr);
-
-void lock_db_free_element(void *ptr)
-{
-  my_free(ptr, MYF(0));
-}
-
-
-/*
-  Put a database lock entry into the hash.
-  
-  DESCRIPTION
-    Insert a database lock entry into hash.
-    LOCK_db_lock must be previously locked.
-  
-  RETURN VALUES
-    0 on success.
-    1 on error.
-*/
-
-static my_bool lock_db_insert(const char *dbname, uint length)
-{
-  my_dblock_t *opt;
-  my_bool error= 0;
-  DBUG_ENTER("lock_db_insert");
-  
-  mysql_mutex_assert_owner(&LOCK_lock_db);
-
-  if (!(opt= (my_dblock_t*) my_hash_search(&lock_db_cache,
-                                           (uchar*) dbname, length)))
-  { 
-    /* Db is not in the hash, insert it */
-    char *tmp_name;
-    if (!my_multi_malloc(MYF(MY_WME | MY_ZEROFILL),
-                         &opt, (uint) sizeof(*opt), &tmp_name, (uint) length+1,
-                         NullS))
-    {
-      error= 1;
-      goto end;
-    }
-    
-    opt->name= tmp_name;
-    strmov(opt->name, dbname);
-    opt->name_length= length;
-    
-    if ((error= my_hash_insert(&lock_db_cache, (uchar*) opt)))
-      my_free(opt, MYF(0));
-  }
-
-end:
-  DBUG_RETURN(error);
-}
-
-
-/*
-  Delete a database lock entry from hash.
-*/
-
-void lock_db_delete(const char *name, uint length)
-{
-  my_dblock_t *opt;
-  mysql_mutex_assert_owner(&LOCK_lock_db);
-  if ((opt= (my_dblock_t *)my_hash_search(&lock_db_cache,
-                                          (const uchar*) name, length)))
-    my_hash_delete(&lock_db_cache, (uchar*) opt);
-}
-
-
 /* Database options hash */
 static HASH dboptions;
 static my_bool dboptions_init= 0;
@@ -233,21 +133,16 @@ static void init_database_names_psi_keys(void)
 }
 #endif
 
-/* 
-  Initialize database option hash and locked database hash.
-
-  SYNOPSIS
-    my_database_names()
+/**
+  Initialize database option cache.
 
-  NOTES
-    Must be called before any other database function is called.
+  @note Must be called before any other database function is called.
 
-  RETURN
-    0	ok
-    1	Fatal error
+  @retval  0	ok
+  @retval  1	Fatal error
 */
 
-bool my_database_names_init(void)
+bool my_dboptions_cache_init(void)
 {
 #ifdef HAVE_PSI_INTERFACE
   init_database_names_psi_keys();
@@ -261,36 +156,30 @@ bool my_database_names_init(void)
     error= my_hash_init(&dboptions, lower_case_table_names ?
                         &my_charset_bin : system_charset_info,
                         32, 0, 0, (my_hash_get_key) dboptions_get_key,
-                        free_dbopt,0) ||
-           my_hash_init(&lock_db_cache, lower_case_table_names ?
-                        &my_charset_bin : system_charset_info,
-                        32, 0, 0, (my_hash_get_key) lock_db_get_key,
-                        lock_db_free_element,0);
-
+                        free_dbopt,0);
   }
   return error;
 }
 
 
 
-/* 
+/**
   Free database option hash and locked databases hash.
 */
 
-void my_database_names_free(void)
+void my_dboptions_cache_free(void)
 {
   if (dboptions_init)
   {
     dboptions_init= 0;
     my_hash_free(&dboptions);
     mysql_rwlock_destroy(&LOCK_dboptions);
-    my_hash_free(&lock_db_cache);
   }
 }
 
 
-/*
-  Cleanup cached options
+/**
+  Cleanup cached options.
 */
 
 void my_dbopt_cleanup(void)
@@ -395,7 +284,7 @@ end:
   Deletes database options from the hash.
 */
 
-void del_dbopt(const char *path)
+static void del_dbopt(const char *path)
 {
   my_dbopt_t *opt;
   mysql_rwlock_wrlock(&LOCK_dboptions);
@@ -664,25 +553,8 @@ int mysql_create_db(THD *thd, char *db, HA_CREATE_INFO *create_info,
     DBUG_RETURN(-1);
   }
 
-  /*
-    Do not create database if another thread is holding read lock.
-    Wait for global read lock before acquiring LOCK_mysql_create_db.
-    After wait_if_global_read_lock() we have protection against another
-    global read lock. If we would acquire LOCK_mysql_create_db first,
-    another thread could step in and get the global read lock before we
-    reach wait_if_global_read_lock(). If this thread tries the same as we
-    (admin a db), it would then go and wait on LOCK_mysql_create_db...
-    Furthermore wait_if_global_read_lock() checks if the current thread
-    has the global read lock and refuses the operation with
-    ER_CANT_UPDATE_WITH_READLOCK if applicable.
-  */
-  if (thd->global_read_lock.wait_if_global_read_lock(thd, FALSE, TRUE))
-  {
-    error= -1;
-    goto exit2;
-  }
-
-  mysql_mutex_lock(&LOCK_mysql_create_db);
+  if (lock_schema_name(thd, db))
+    DBUG_RETURN(-1);
 
   /* Check directory */
   path_len= build_table_filename(path, sizeof(path) - 1, db, "", "", 0);
@@ -786,7 +658,10 @@ not_silent:
       qinfo.db     = db;
       qinfo.db_len = strlen(db);
 
-      /* These DDL methods and logging protected with LOCK_mysql_create_db */
+      /*
+        These DDL methods and logging are protected with the exclusive
+        metadata lock on the schema
+      */
       if (mysql_bin_log.write(&qinfo))
       {
         error= -1;
@@ -797,9 +672,6 @@ not_silent:
   }
 
 exit:
-  mysql_mutex_unlock(&LOCK_mysql_create_db);
-  thd->global_read_lock.start_waiting_global_read_lock(thd);
-exit2:
   DBUG_RETURN(error);
 }
 
@@ -813,22 +685,8 @@ bool mysql_alter_db(THD *thd, const char *db, HA_CREATE_INFO *create_info)
   int error= 0;
   DBUG_ENTER("mysql_alter_db");
 
-  /*
-    Do not alter database if another thread is holding read lock.
-    Wait for global read lock before acquiring LOCK_mysql_create_db.
-    After wait_if_global_read_lock() we have protection against another
-    global read lock. If we would acquire LOCK_mysql_create_db first,
-    another thread could step in and get the global read lock before we
-    reach wait_if_global_read_lock(). If this thread tries the same as we
-    (admin a db), it would then go and wait on LOCK_mysql_create_db...
-    Furthermore wait_if_global_read_lock() checks if the current thread
-    has the global read lock and refuses the operation with
-    ER_CANT_UPDATE_WITH_READLOCK if applicable.
-  */
-  if ((error= thd->global_read_lock.wait_if_global_read_lock(thd, FALSE, TRUE)))
-    goto exit2;
-
-  mysql_mutex_lock(&LOCK_mysql_create_db);
+  if (lock_schema_name(thd, db))
+    DBUG_RETURN(TRUE);
 
   /* 
      Recreate db options file: /dbpath/.db.opt
@@ -866,16 +724,16 @@ bool mysql_alter_db(THD *thd, const char *db, HA_CREATE_INFO *create_info)
     qinfo.db     = db;
     qinfo.db_len = strlen(db);
 
-    /* These DDL methods and logging protected with LOCK_mysql_create_db */
+    /*
+      These DDL methods and logging are protected with the exclusive
+      metadata lock on the schema.
+    */
     if ((error= mysql_bin_log.write(&qinfo)))
       goto exit;
   }
   my_ok(thd, result);
 
 exit:
-  mysql_mutex_unlock(&LOCK_mysql_create_db);
-  thd->global_read_lock.start_waiting_global_read_lock(thd);
-exit2:
   DBUG_RETURN(error);
 }
 
@@ -907,25 +765,9 @@ bool mysql_rm_db(THD *thd,char *db,bool if_exists, bool silent)
   TABLE_LIST* dropped_tables= 0;
   DBUG_ENTER("mysql_rm_db");
 
-  /*
-    Do not drop database if another thread is holding read lock.
-    Wait for global read lock before acquiring LOCK_mysql_create_db.
-    After wait_if_global_read_lock() we have protection against another
-    global read lock. If we would acquire LOCK_mysql_create_db first,
-    another thread could step in and get the global read lock before we
-    reach wait_if_global_read_lock(). If this thread tries the same as we
-    (admin a db), it would then go and wait on LOCK_mysql_create_db...
-    Furthermore wait_if_global_read_lock() checks if the current thread
-    has the global read lock and refuses the operation with
-    ER_CANT_UPDATE_WITH_READLOCK if applicable.
-  */
-  if (thd->global_read_lock.wait_if_global_read_lock(thd, FALSE, TRUE))
-  {
-    error= -1;
-    goto exit2;
-  }
 
-  mysql_mutex_lock(&LOCK_mysql_create_db);
+  if (lock_schema_name(thd, db))
+    DBUG_RETURN(TRUE);
 
   length= build_table_filename(path, sizeof(path) - 1, db, "", "", 0);
   strmov(path+length, MY_DB_OPT_FILE);		// Append db option file name
@@ -1013,7 +855,10 @@ bool mysql_rm_db(THD *thd,char *db,bool if_exists, bool silent)
       qinfo.db     = db;
       qinfo.db_len = strlen(db);
 
-      /* These DDL methods and logging protected with LOCK_mysql_create_db */
+      /*
+        These DDL methods and logging are protected with the exclusive
+        metadata lock on the schema.
+      */
       if (mysql_bin_log.write(&qinfo))
       {
         error= -1;
@@ -1045,7 +890,10 @@ bool mysql_rm_db(THD *thd,char *db,bool if_exists, bool silent)
       tbl_name_len= strlen(tbl->table_name) + 3;
       if (query_pos + tbl_name_len + 1 >= query_end)
       {
-        /* These DDL methods and logging protected with LOCK_mysql_create_db */
+        /*
+          These DDL methods and logging are protected with the exclusive
+          metadata lock on the schema.
+        */
         if (write_to_binlog(thd, query, query_pos -1 - query, db, db_len))
         {
           error= -1;
@@ -1062,7 +910,10 @@ bool mysql_rm_db(THD *thd,char *db,bool if_exists, bool silent)
 
     if (query_pos != query_data_start)
     {
-      /* These DDL methods and logging protected with LOCK_mysql_create_db */
+      /*
+        These DDL methods and logging are protected with the exclusive
+        metadata lock on the schema.
+      */
       if (write_to_binlog(thd, query, query_pos -1 - query, db, db_len))
       {
         error= -1;
@@ -1080,9 +931,6 @@ exit:
   */
   if (thd->db && !strcmp(thd->db, db) && error == 0)
     mysql_change_db_impl(thd, NULL, 0, thd->variables.collation_server);
-  mysql_mutex_unlock(&LOCK_mysql_create_db);
-  thd->global_read_lock.start_waiting_global_read_lock(thd);
-exit2:
   DBUG_RETURN(error);
 }
 
@@ -1099,12 +947,12 @@ static long mysql_rm_known_files(THD *thd, MY_DIR *dirp, const char *db,
   long deleted=0;
   ulong found_other_files=0;
   char filePath[FN_REFLEN];
-  TABLE_LIST *tot_list=0, **tot_list_next;
+  TABLE_LIST *tot_list=0, **tot_list_next_local, **tot_list_next_global;
   List<String> raid_dirs;
   DBUG_ENTER("mysql_rm_known_files");
   DBUG_PRINT("enter",("path: %s", org_path));
 
-  tot_list_next= &tot_list;
+  tot_list_next_local= tot_list_next_global= &tot_list;
 
   for (uint idx=0 ;
        idx < (uint) dirp->number_off_files && !thd->killed ;
@@ -1192,23 +1040,28 @@ static long mysql_rm_known_files(THD *thd, MY_DIR *dirp, const char *db,
       if (!table_list)
         goto err;
       table_list->db= (char*) (table_list+1);
-      table_list->table_name= strmov(table_list->db, db) + 1;
-      (void) filename_to_tablename(file->name, table_list->table_name,
-                                 MYSQL50_TABLE_NAME_PREFIX_LENGTH +
-                                 strlen(file->name) + 1);
+      table_list->db_length= strmov(table_list->db, db) - table_list->db;
+      table_list->table_name= table_list->db + table_list->db_length + 1;
+      table_list->table_name_length= filename_to_tablename(file->name,
+                                       table_list->table_name,
+                                       MYSQL50_TABLE_NAME_PREFIX_LENGTH +
+                                       strlen(file->name) + 1);
       table_list->open_type= OT_BASE_ONLY;
 
       /* To be able to correctly look up the table in the table cache. */
       if (lower_case_table_names)
-        my_casedn_str(files_charset_info, table_list->table_name);
+        table_list->table_name_length= my_casedn_str(files_charset_info,
+                                                     table_list->table_name);
 
       table_list->alias= table_list->table_name;	// If lower_case_table_names=2
       table_list->internal_tmp_table= is_prefix(file->name, tmp_file_prefix);
       table_list->mdl_request.init(MDL_key::TABLE, table_list->db,
                                    table_list->table_name, MDL_EXCLUSIVE);
       /* Link into list */
-      (*tot_list_next)= table_list;
-      tot_list_next= &table_list->next_local;
+      (*tot_list_next_local)= table_list;
+      (*tot_list_next_global)= table_list;
+      tot_list_next_local= &table_list->next_local;
+      tot_list_next_global= &table_list->next_global;
       deleted++;
     }
     else
@@ -1771,60 +1624,6 @@ bool mysql_opt_change_db(THD *thd,
 }
 
 
-static int
-lock_databases(THD *thd, const char *db1, uint length1,
-                         const char *db2, uint length2)
-{
-  mysql_mutex_lock(&LOCK_lock_db);
-  while (!thd->killed &&
-         (my_hash_search(&lock_db_cache,(uchar*) db1, length1) ||
-          my_hash_search(&lock_db_cache,(uchar*) db2, length2)))
-  {
-    wait_for_condition(thd, &LOCK_lock_db, &COND_refresh);
-    mysql_mutex_lock(&LOCK_lock_db);
-  }
-
-  if (thd->killed)
-  {
-    mysql_mutex_unlock(&LOCK_lock_db);
-    return 1;
-  }
-
-  lock_db_insert(db1, length1);
-  lock_db_insert(db2, length2);
-  creating_database++;
-
-  /*
-    Wait if a concurent thread is creating a table at the same time.
-    The assumption here is that it will not take too long until
-    there is a point in time when a table is not created.
-  */
-
-  while (!thd->killed && creating_table)
-  {
-    wait_for_condition(thd, &LOCK_lock_db, &COND_refresh);
-    mysql_mutex_lock(&LOCK_lock_db);
-  }
-
-  if (thd->killed)
-  {
-    lock_db_delete(db1, length1);
-    lock_db_delete(db2, length2);
-    creating_database--;
-    mysql_mutex_unlock(&LOCK_lock_db);
-    mysql_cond_signal(&COND_refresh);
-    return(1);
-  }
-
-  /*
-    We can unlock now as the hash will protect against anyone creating a table
-    in the databases we are using
-  */
-  mysql_mutex_unlock(&LOCK_lock_db);
-  return 0;
-}
-
-
 /**
   Upgrade a 5.0 database.
   This function is invoked whenever an ALTER DATABASE UPGRADE query is executed:
@@ -1866,9 +1665,9 @@ bool mysql_upgrade_db(THD *thd, LEX_STRING *old_db)
   new_db.str= old_db->str + MYSQL50_TABLE_NAME_PREFIX_LENGTH;
   new_db.length= old_db->length - MYSQL50_TABLE_NAME_PREFIX_LENGTH;
 
-  if (lock_databases(thd, old_db->str, old_db->length,
-                          new_db.str, new_db.length))
-    DBUG_RETURN(1);
+  /* Lock the old name, the new name will be locked by mysql_create_db().*/
+  if (lock_schema_name(thd, old_db->str))
+    DBUG_RETURN(-1);
 
   /*
     Let's remember if we should do "USE newdb" afterwards.
@@ -2035,15 +1834,6 @@ bool mysql_upgrade_db(THD *thd, LEX_STRING *old_db)
     error|= mysql_change_db(thd, & new_db, FALSE);
 
 exit:
-  mysql_mutex_lock(&LOCK_lock_db);
-  /* Remove the databases from db lock cache */
-  lock_db_delete(old_db->str, old_db->length);
-  lock_db_delete(new_db.str, new_db.length);
-  creating_database--;
-  /* Signal waiting CREATE TABLE's to continue */
-  mysql_cond_signal(&COND_refresh);
-  mysql_mutex_unlock(&LOCK_lock_db);
-
   DBUG_RETURN(error);
 }
 
diff --git a/sql/sql_db.h b/sql/sql_db.h
index 96b3de80d3a..ecb8deaa397 100644
--- a/sql/sql_db.h
+++ b/sql/sql_db.h
@@ -35,8 +35,8 @@ bool mysql_opt_change_db(THD *thd,
                          LEX_STRING *saved_db_name,
                          bool force_switch,
                          bool *cur_db_changed);
-bool my_database_names_init(void);
-void my_database_names_free(void);
+bool my_dboptions_cache_init(void);
+void my_dboptions_cache_free(void);
 bool check_db_dir_existence(const char *db_name);
 bool load_db_opt(THD *thd, const char *path, HA_CREATE_INFO *create);
 bool load_db_opt_by_name(THD *thd, const char *db_name,
@@ -45,9 +45,6 @@ CHARSET_INFO *get_default_db_collation(THD *thd, const char *db_name);
 bool my_dbopt_init(void);
 void my_dbopt_cleanup(void);
 
-extern int creating_database; // How many database locks are made
-extern HASH lock_db_cache;
-
 #define MY_DB_OPT_FILE "db.opt"
 
 #endif /* SQL_DB_INCLUDED */
diff --git a/sql/sql_help.cc b/sql/sql_help.cc
index 7bea236269a..4e3df950134 100644
--- a/sql/sql_help.cc
+++ b/sql/sql_help.cc
@@ -643,23 +643,24 @@ bool mysqld_help(THD *thd, const char *mask)
   MEM_ROOT *mem_root= thd->mem_root;
   DBUG_ENTER("mysqld_help");
 
-  bzero((uchar*)tables,sizeof(tables));
-  tables[0].alias= tables[0].table_name= (char*) "help_topic";
-  tables[0].lock_type= TL_READ;
+  tables[0].init_one_table(C_STRING_WITH_LEN("mysql"),
+                           C_STRING_WITH_LEN("help_topic"),
+                           "help_topic", TL_READ);
+  tables[1].init_one_table(C_STRING_WITH_LEN("mysql"),
+                           C_STRING_WITH_LEN("help_category"),
+                           "help_category", TL_READ);
+  tables[2].init_one_table(C_STRING_WITH_LEN("mysql"),
+                           C_STRING_WITH_LEN("help_relation"),
+                           "help_relation", TL_READ);
+  tables[3].init_one_table(C_STRING_WITH_LEN("mysql"),
+                           C_STRING_WITH_LEN("help_keyword"),
+                           "help_keyword", TL_READ);
   tables[0].next_global= tables[0].next_local= 
     tables[0].next_name_resolution_table= &tables[1];
-  tables[1].alias= tables[1].table_name= (char*) "help_category";
-  tables[1].lock_type= TL_READ;
   tables[1].next_global= tables[1].next_local= 
     tables[1].next_name_resolution_table= &tables[2];
-  tables[2].alias= tables[2].table_name= (char*) "help_relation";
-  tables[2].lock_type= TL_READ;
   tables[2].next_global= tables[2].next_local= 
     tables[2].next_name_resolution_table= &tables[3];
-  tables[3].alias= tables[3].table_name= (char*) "help_keyword";
-  tables[3].lock_type= TL_READ;
-  tables[0].db= tables[1].db= tables[2].db= tables[3].db= (char*) "mysql";
-  init_mdl_requests(tables);
 
   /*
     HELP must be available under LOCK TABLES. 
diff --git a/sql/sql_hset.h b/sql/sql_hset.h
new file mode 100644
index 00000000000..c3b98b881f5
--- /dev/null
+++ b/sql/sql_hset.h
@@ -0,0 +1,117 @@
+#ifndef SQL_HSET_INCLUDED
+#define SQL_HSET_INCLUDED
+/* Copyright (c) 2010, Oracle and/or its affiliates. All rights reserved.
+
+   This program is free software; you can redistribute it and/or modify
+   it under the terms of the GNU General Public License as published by
+   the Free Software Foundation; version 2 of the License.
+
+   This program is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+   GNU General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program; if not, write to the Free Software
+   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */
+
+#include "my_global.h"
+#include "hash.h"
+
+
+/**
+  A type-safe wrapper around mysys HASH.
+*/
+
+template <typename T, my_hash_get_key K>
+class Hash_set
+{
+public:
+  typedef T Value_type;
+  enum { START_SIZE= 8 };
+  /**
+    Constructs an empty hash. Does not allocate memory, it is done upon
+    the first insert. Thus does not cause or return errors.
+  */
+  Hash_set();
+  /**
+    Destroy the hash by freeing the buckets table. Does
+    not call destructors for the elements.
+  */
+  ~Hash_set();
+  /**
+    Insert a single value into a hash. Does not tell whether
+    the value was inserted -- if an identical value existed,
+    it is not replaced.
+
+    @retval TRUE  Out of memory.
+    @retval FALSE OK. The value either was inserted or existed
+                  in the hash.
+  */
+  bool insert(T *value);
+  /** Is this hash set empty? */
+  bool is_empty() const { return m_hash.records == 0; }
+  /** Returns the number of unique elements. */
+  size_t size() const { return static_cast<size_t>(m_hash.records); }
+  /** An iterator over hash elements. Is not insert-stable. */
+  class Iterator
+  {
+  public:
+    Iterator(Hash_set &hash_set);
+    /**
+      Return the current element and reposition the iterator to the next
+      element.
+    */
+    inline T *operator++(int);
+    void rewind() { m_idx= 0; }
+  private:
+    HASH *m_hash;
+    uint m_idx;
+  };
+private:
+  HASH m_hash;
+};
+
+
+template <typename T, my_hash_get_key K>
+Hash_set<T, K>::Hash_set()
+{
+  my_hash_clear(&m_hash);
+}
+
+
+template <typename T, my_hash_get_key K>
+Hash_set<T, K>::~Hash_set()
+{
+  my_hash_free(&m_hash);
+}
+
+
+template <typename T, my_hash_get_key K>
+bool Hash_set<T, K>::insert(T *value)
+{
+  my_hash_init_opt(&m_hash, &my_charset_bin, START_SIZE, 0, 0, K, 0, MYF(0));
+  size_t key_len;
+  const uchar *key= K(reinterpret_cast<uchar*>(value), &key_len, FALSE);
+  if (my_hash_search(&m_hash, key, key_len) == NULL)
+    return my_hash_insert(&m_hash, reinterpret_cast<uchar *>(value));
+  return FALSE;
+}
+
+
+template <typename T, my_hash_get_key K>
+Hash_set<T, K>::Iterator::Iterator(Hash_set<T, K> &set_arg)
+  :m_hash(&set_arg.m_hash),
+  m_idx(0)
+{}
+
+
+template <typename T, my_hash_get_key K>
+inline T *Hash_set<T, K>::Iterator::operator++(int)
+{
+  if (m_idx < m_hash->records)
+    return reinterpret_cast<T*>(my_hash_element(m_hash, m_idx++));
+  return NULL;
+}
+
+#endif // SQL_HSET_INCLUDED
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 273eadf4205..31eaf4111ce 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -270,10 +270,10 @@ void init_update_queries(void)
   sql_command_flags[SQLCOM_LOAD]=           CF_CHANGES_DATA | CF_REEXECUTION_FRAGILE |
                                             CF_PROTECT_AGAINST_GRL |
                                             CF_CAN_GENERATE_ROW_EVENTS;
-  sql_command_flags[SQLCOM_CREATE_DB]=      CF_CHANGES_DATA | CF_AUTO_COMMIT_TRANS;
-  sql_command_flags[SQLCOM_DROP_DB]=        CF_CHANGES_DATA | CF_AUTO_COMMIT_TRANS;
-  sql_command_flags[SQLCOM_ALTER_DB_UPGRADE]= CF_AUTO_COMMIT_TRANS;
-  sql_command_flags[SQLCOM_ALTER_DB]=       CF_CHANGES_DATA | CF_AUTO_COMMIT_TRANS;
+  sql_command_flags[SQLCOM_CREATE_DB]=      CF_CHANGES_DATA | CF_AUTO_COMMIT_TRANS | CF_PROTECT_AGAINST_GRL;
+  sql_command_flags[SQLCOM_DROP_DB]=        CF_CHANGES_DATA | CF_AUTO_COMMIT_TRANS | CF_PROTECT_AGAINST_GRL;
+  sql_command_flags[SQLCOM_ALTER_DB_UPGRADE]= CF_AUTO_COMMIT_TRANS | CF_PROTECT_AGAINST_GRL;
+  sql_command_flags[SQLCOM_ALTER_DB]=       CF_CHANGES_DATA | CF_AUTO_COMMIT_TRANS | CF_PROTECT_AGAINST_GRL;
   sql_command_flags[SQLCOM_RENAME_TABLE]=   CF_CHANGES_DATA | CF_AUTO_COMMIT_TRANS;
   sql_command_flags[SQLCOM_DROP_INDEX]=     CF_CHANGES_DATA | CF_AUTO_COMMIT_TRANS;
   sql_command_flags[SQLCOM_CREATE_VIEW]=    CF_CHANGES_DATA | CF_REEXECUTION_FRAGILE |
@@ -1803,7 +1803,8 @@ static bool flush_tables_with_read_lock(THD *thd, TABLE_LIST *all_tables)
     current internal MDL asserts, fix after discussing with
     Dmitry.
   */
-  if (lock_table_names(thd, all_tables))
+  if (lock_table_names(thd, all_tables, 0, thd->variables.lock_wait_timeout,
+                       MYSQL_OPEN_SKIP_TEMPORARY))
     goto error;
 
   for (table_list= all_tables; table_list;
@@ -3644,12 +3645,6 @@ end_with_restore_list:
 #endif
     if (check_access(thd, CREATE_ACL, lex->name.str, NULL, NULL, 1, 0))
       break;
-    if (thd->locked_tables_mode)
-    {
-      my_message(ER_LOCK_OR_ACTIVE_TRANSACTION,
-                 ER(ER_LOCK_OR_ACTIVE_TRANSACTION), MYF(0));
-      goto error;
-    }
     res= mysql_create_db(thd,(lower_case_table_names == 2 ? alias :
                               lex->name.str), &create_info, 0);
     break;
@@ -3679,12 +3674,6 @@ end_with_restore_list:
 #endif
     if (check_access(thd, DROP_ACL, lex->name.str, NULL, NULL, 1, 0))
       break;
-    if (thd->locked_tables_mode)
-    {
-      my_message(ER_LOCK_OR_ACTIVE_TRANSACTION,
-                 ER(ER_LOCK_OR_ACTIVE_TRANSACTION), MYF(0));
-      goto error;
-    }
     res= mysql_rm_db(thd, lex->name.str, lex->drop_if_exists, 0);
     break;
   }
@@ -3713,14 +3702,6 @@ end_with_restore_list:
       res= 1;
       break;
     }
-    if (thd->locked_tables_mode)
-    {
-      res= 1;
-      my_message(ER_LOCK_OR_ACTIVE_TRANSACTION,
-                 ER(ER_LOCK_OR_ACTIVE_TRANSACTION), MYF(0));
-      goto error;
-    }
-
     res= mysql_upgrade_db(thd, db);
     if (!res)
       my_ok(thd);
@@ -3753,12 +3734,6 @@ end_with_restore_list:
 #endif
     if (check_access(thd, ALTER_ACL, db->str, NULL, NULL, 1, 0))
       break;
-    if (thd->locked_tables_mode)
-    {
-      my_message(ER_LOCK_OR_ACTIVE_TRANSACTION,
-                 ER(ER_LOCK_OR_ACTIVE_TRANSACTION), MYF(0));
-      goto error;
-    }
     res= mysql_alter_db(thd, db->str, &create_info);
     break;
   }
diff --git a/sql/sql_rename.cc b/sql/sql_rename.cc
index 130a99a374f..301b22bd70e 100644
--- a/sql/sql_rename.cc
+++ b/sql/sql_rename.cc
@@ -24,10 +24,9 @@
 #include "sql_table.h"                         // build_table_filename
 #include "sql_view.h"             // mysql_frm_type, mysql_rename_view
 #include "sql_trigger.h"
-#include "lock.h"       // wait_if_global_read_lock, lock_table_names,
-                        // unlock_table_names,
+#include "lock.h"       // wait_if_global_read_lock
                         // start_waiting_global_read_lock
-#include "sql_base.h"   // tdc_remove_table
+#include "sql_base.h"   // tdc_remove_table, lock_table_names,
 #include "sql_handler.h"                        // mysql_ha_rm_tables
 #include "datadict.h"
 
@@ -144,7 +143,8 @@ bool mysql_rename_tables(THD *thd, TABLE_LIST *table_list, bool silent)
     }
   }
 
-  if (lock_table_names(thd, table_list))
+  if (lock_table_names(thd, table_list, 0, thd->variables.lock_wait_timeout,
+                       MYSQL_OPEN_SKIP_TEMPORARY))
     goto err;
 
   mysql_mutex_lock(&LOCK_open);
@@ -197,7 +197,7 @@ bool mysql_rename_tables(THD *thd, TABLE_LIST *table_list, bool silent)
   if (!error)
     query_cache_invalidate3(thd, table_list, 0);
 
-  unlock_table_names(thd);
+  thd->mdl_context.release_transactional_locks();
 
 err:
   thd->global_read_lock.start_waiting_global_read_lock(thd);
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 97c8f59d7df..74acd134910 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -22,17 +22,17 @@
 #include "sql_rename.h" // do_rename
 #include "sql_parse.h"                        // test_if_data_home_dir
 #include "sql_cache.h"                          // query_cache_*
-#include "sql_base.h"                          // open_temporary_table
-#include "lock.h"       // wait_if_global_read_lock, lock_table_names,
+#include "sql_base.h"   // open_temporary_table, lock_table_names
+#include "lock.h"       // wait_if_global_read_lock
                         // start_waiting_global_read_lock,
-                        // unlock_table_names, mysql_unlock_tables
+                        // mysql_unlock_tables
 #include "strfunc.h"    // find_type2, find_set
 #include "sql_view.h" // view_checksum 
 #include "sql_truncate.h"                       // regenerate_locked_table 
 #include "sql_partition.h"                      // mem_alloc_error,
                                                 // generate_partition_syntax,
                                                 // partition_info
-#include "sql_db.h" // load_db_opt_by_name, lock_db_cache, creating_database
+#include "sql_db.h"                             // load_db_opt_by_name
 #include "sql_time.h"                  // make_truncated_value_warning
 #include "records.h"             // init_read_record, end_read_record
 #include "filesort.h"            // filesort_free_buffers
@@ -58,8 +58,6 @@
 #include <io.h>
 #endif
 
-int creating_table= 0;        // How many mysql_create_table are running
-
 const char *primary_key_name="PRIMARY";
 
 static bool check_if_keyname_exists(const char *name,KEY *start, KEY *end);
@@ -1954,7 +1952,8 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
   {
     if (!thd->locked_tables_mode)
     {
-      if (lock_table_names(thd, tables))
+      if (lock_table_names(thd, tables, NULL, thd->variables.lock_wait_timeout,
+                           MYSQL_OPEN_SKIP_TEMPORARY))
         DBUG_RETURN(1);
       mysql_mutex_lock(&LOCK_open);
       for (table= tables; table; table= table->next_local)
@@ -2296,7 +2295,7 @@ err:
       leaving LOCK TABLES mode if we have dropped only temporary tables.
     */
     if (! thd->locked_tables_mode)
-      unlock_table_names(thd);
+      thd->mdl_context.release_transactional_locks();
     else
     {
       if (thd->lock && thd->lock->table_count == 0 && non_temp_tables_count > 0)
@@ -4226,24 +4225,6 @@ bool mysql_create_table(THD *thd, TABLE_LIST *create_table,
   bool result;
   DBUG_ENTER("mysql_create_table");
 
-  /* Wait for any database locks */
-  mysql_mutex_lock(&LOCK_lock_db);
-  while (!thd->killed &&
-         my_hash_search(&lock_db_cache, (uchar*)create_table->db,
-                        create_table->db_length))
-  {
-    wait_for_condition(thd, &LOCK_lock_db, &COND_refresh);
-    mysql_mutex_lock(&LOCK_lock_db);
-  }
-
-  if (thd->killed)
-  {
-    mysql_mutex_unlock(&LOCK_lock_db);
-    DBUG_RETURN(TRUE);
-  }
-  creating_table++;
-  mysql_mutex_unlock(&LOCK_lock_db);
-
   /*
     Open or obtain an exclusive metadata lock on table being created.
   */
@@ -4282,10 +4263,6 @@ bool mysql_create_table(THD *thd, TABLE_LIST *create_table,
   }
 
 unlock:
-  mysql_mutex_lock(&LOCK_lock_db);
-  if (!--creating_table && creating_database)
-    mysql_cond_signal(&COND_refresh);
-  mysql_mutex_unlock(&LOCK_lock_db);
   DBUG_RETURN(result);
 }
 
@@ -4458,8 +4435,6 @@ static int prepare_for_repair(THD *thd, TABLE_LIST *table_list,
   {
     char key[MAX_DBKEY_LENGTH];
     uint key_length;
-    MDL_request mdl_global_request;
-    MDL_request_list mdl_requests;
     /*
       If the table didn't exist, we have a shared metadata lock
       on it that is left from mysql_admin_table()'s attempt to 
@@ -4479,12 +4454,9 @@ static int prepare_for_repair(THD *thd, TABLE_LIST *table_list,
                                  table_list->db, table_list->table_name,
                                  MDL_EXCLUSIVE);
 
-    mdl_global_request.init(MDL_key::GLOBAL, "", "", MDL_INTENTION_EXCLUSIVE);
-    mdl_requests.push_front(&table_list->mdl_request);
-    mdl_requests.push_front(&mdl_global_request);
-
-    if (thd->mdl_context.acquire_locks(&mdl_requests,
-                                       thd->variables.lock_wait_timeout))
+    if (lock_table_names(thd, table_list, table_list->next_global,
+                         thd->variables.lock_wait_timeout,
+                         MYSQL_OPEN_SKIP_TEMPORARY))
       DBUG_RETURN(0);
     has_mdl_lock= TRUE;
 
@@ -6577,6 +6549,7 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
 
   Alter_table_prelocking_strategy alter_prelocking_strategy(alter_info);
 
+  DEBUG_SYNC(thd, "alter_table_before_open_tables");
   error= open_and_lock_tables(thd, table_list, FALSE, 0,
                               &alter_prelocking_strategy);
 
diff --git a/sql/sql_table.h b/sql/sql_table.h
index 40b24605bd6..dca4b706605 100644
--- a/sql/sql_table.h
+++ b/sql/sql_table.h
@@ -210,7 +210,6 @@ uint explain_filename(THD* thd, const char *from, char *to, uint to_length,
 
 
 extern MYSQL_PLUGIN_IMPORT const char *primary_key_name;
-extern int creating_table;    // How many mysql_create_table() are running
 extern mysql_mutex_t LOCK_gdl;
 
 #endif /* SQL_TABLE_INCLUDED */
diff --git a/sql/sql_truncate.cc b/sql/sql_truncate.cc
index 901ab8e987d..6bbf86cbe55 100644
--- a/sql/sql_truncate.cc
+++ b/sql/sql_truncate.cc
@@ -242,9 +242,10 @@ static bool open_and_lock_table_for_truncate(THD *thd, TABLE_LIST *table_ref,
                                              MDL_ticket **ticket_downgrade)
 {
   TABLE *table= NULL;
-  MDL_ticket *mdl_ticket= NULL;
   DBUG_ENTER("open_and_lock_table_for_truncate");
 
+  DBUG_ASSERT(table_ref->lock_type == TL_WRITE);
+  DBUG_ASSERT(table_ref->mdl_request.type == MDL_SHARED_NO_READ_WRITE);
   /*
     Before doing anything else, acquire a metadata lock on the table,
     or ensure we have one.  We don't use open_and_lock_tables()
@@ -266,6 +267,7 @@ static bool open_and_lock_table_for_truncate(THD *thd, TABLE_LIST *table_ref,
 
     *hton_can_recreate= ha_check_storage_engine_flag(table->s->db_type(),
                                                      HTON_CAN_RECREATE);
+    table_ref->mdl_request.ticket= table->mdl_ticket;
   }
   else
   {
@@ -273,21 +275,12 @@ static bool open_and_lock_table_for_truncate(THD *thd, TABLE_LIST *table_ref,
       Even though we could use the previous execution branch here just as
       well, we must not try to open the table:
     */
-    MDL_request mdl_global_request, mdl_request;
-    MDL_request_list mdl_requests;
-
-    mdl_global_request.init(MDL_key::GLOBAL, "", "", MDL_INTENTION_EXCLUSIVE);
-    mdl_request.init(MDL_key::TABLE, table_ref->db, table_ref->table_name,
-                     MDL_SHARED_NO_READ_WRITE);
-    mdl_requests.push_front(&mdl_request);
-    mdl_requests.push_front(&mdl_global_request);
-
-    if (thd->mdl_context.acquire_locks(&mdl_requests,
-                                        thd->variables.lock_wait_timeout))
+    DBUG_ASSERT(table_ref->next_global == NULL);
+    if (lock_table_names(thd, table_ref, NULL,
+                         thd->variables.lock_wait_timeout,
+                         MYSQL_OPEN_SKIP_TEMPORARY))
       DBUG_RETURN(TRUE);
 
-    mdl_ticket= mdl_request.ticket;
-
     if (dd_check_storage_engine_flag(thd, table_ref->db, table_ref->table_name,
                                      HTON_CAN_RECREATE, hton_can_recreate))
       DBUG_RETURN(TRUE);
@@ -313,7 +306,9 @@ static bool open_and_lock_table_for_truncate(THD *thd, TABLE_LIST *table_ref,
     else
     {
       ulong timeout= thd->variables.lock_wait_timeout;
-      if (thd->mdl_context.upgrade_shared_lock_to_exclusive(mdl_ticket, timeout))
+      if (thd->mdl_context.
+          upgrade_shared_lock_to_exclusive(table_ref->mdl_request.ticket,
+                                           timeout))
         DBUG_RETURN(TRUE);
       mysql_mutex_lock(&LOCK_open);
       tdc_remove_table(thd, TDC_RT_REMOVE_ALL, table_ref->db,
@@ -335,15 +330,14 @@ static bool open_and_lock_table_for_truncate(THD *thd, TABLE_LIST *table_ref,
     table_ref->required_type= FRMTYPE_TABLE;
     /* We don't need to load triggers. */
     DBUG_ASSERT(table_ref->trg_event_map == 0);
-    /* Work around partition parser rules using alter table's. */
-    if (thd->lex->alter_info.flags & ALTER_ADMIN_PARTITION)
-    {
-      table_ref->lock_type= TL_WRITE;
-      table_ref->mdl_request.set_type(MDL_SHARED_WRITE);
-    }
-    /* Ensure proper lock types (e.g. from the parser). */
-    DBUG_ASSERT(table_ref->lock_type == TL_WRITE);
-    DBUG_ASSERT(table_ref->mdl_request.type == MDL_SHARED_WRITE);
+    /*
+      Even though we have an MDL lock on the table here, we don't
+      pass MYSQL_OPEN_HAS_MDL_LOCK to open_and_lock_tables
+      since to truncate a MERGE table, we must open and lock
+      merge children, and on those we don't have an MDL lock.
+      Thus clear the ticket to satisfy MDL asserts.
+    */
+    table_ref->mdl_request.ticket= NULL;
 
     /*
       Open the table as it will handle some required preparations.
diff --git a/sql/sql_view.cc b/sql/sql_view.cc
index 69abe70e863..be13349b5a1 100644
--- a/sql/sql_view.cc
+++ b/sql/sql_view.cc
@@ -19,10 +19,10 @@
 #include "sql_priv.h"
 #include "unireg.h"
 #include "sql_view.h"
-#include "sql_base.h"                     // find_table_in_global_list
+#include "sql_base.h"    // find_table_in_global_list, lock_table_names
 #include "sql_parse.h"                          // sql_parse
 #include "sql_cache.h"                          // query_cache_*
-#include "lock.h"        // wait_if_global_read_lock, lock_table_names
+#include "lock.h"        // wait_if_global_read_lock
 #include "sql_show.h"    // append_identifier
 #include "sql_table.h"                         // build_table_filename
 #include "sql_db.h"            // mysql_opt_change_db, mysql_change_db
@@ -1652,7 +1652,8 @@ bool mysql_drop_view(THD *thd, TABLE_LIST *views, enum_drop_mode drop_mode)
     DBUG_RETURN(TRUE);
   }
 
-  if (lock_table_names(thd, views))
+  if (lock_table_names(thd, views, 0, thd->variables.lock_wait_timeout,
+                       MYSQL_OPEN_SKIP_TEMPORARY))
     DBUG_RETURN(TRUE);
 
   mysql_mutex_lock(&LOCK_open);
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index 73935d2a0b1..934ec9c35ab 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -6442,6 +6442,8 @@ alter_commands:
             lex->sql_command= SQLCOM_TRUNCATE;
             lex->alter_info.flags|= ALTER_ADMIN_PARTITION;
             lex->check_opt.init();
+            lex->query_tables->mdl_request.set_type(MDL_SHARED_NO_READ_WRITE);
+            lex->query_tables->lock_type= TL_WRITE;
           }
         | reorg_partition_rule
         ;
@@ -10695,7 +10697,7 @@ truncate:
             lex->select_lex.sql_cache= SELECT_LEX::SQL_CACHE_UNSPECIFIED;
             lex->select_lex.init_order();
             YYPS->m_lock_type= TL_WRITE;
-            YYPS->m_mdl_type= MDL_SHARED_WRITE;
+            YYPS->m_mdl_type= MDL_SHARED_NO_READ_WRITE;
           }
           table_name
           {}