Patch that changes metadata locking subsystem to use mutex per lock and

condition variable per context instead of one mutex and one conditional
variable for the whole subsystem.

This should increase concurrency in this subsystem.

It also opens the way for further changes which are necessary to solve
such bugs as bug #46272 "MySQL 5.4.4, new MDL: unnecessary deadlock"
and bug #37346 "innodb does not detect deadlock between update and alter
table".

Two other notable changes done by this patch:

- MDL subsystem no longer implicitly acquires global intention exclusive
  metadata lock when per-object metadata lock is acquired. Now this has
  to be done by explicit calls outside of MDL subsystem.
- Instead of using separate MDL_context for opening system tables/tables
  for purposes of I_S we now create MDL savepoint in the main context
  before opening tables and rollback to this savepoint after closing
  them. This means that it is now possible to get ER_LOCK_DEADLOCK error
  even not inside a transaction. This might happen in unlikely case when
  one runs DDL on one of system tables while also running DDL on some
  other tables. Cases when this ER_LOCK_DEADLOCK error is not justified
  will be addressed by advanced deadlock detector for MDL subsystem which
  we plan to implement.

mysql-test/include/handler.inc:
  Adjusted handler_myisam.test and handler_innodb.test to the fact that
  exclusive metadata locks on tables are now acquired according to
  alphabetical order of fully qualified table names instead of order
  in which tables are mentioned in statement.
mysql-test/r/handler_innodb.result:
  Adjusted handler_myisam.test and handler_innodb.test to the fact that
  exclusive metadata locks on tables are now acquired according to
  alphabetical order of fully qualified table names instead of order
  in which tables are mentioned in statement.
mysql-test/r/handler_myisam.result:
  Adjusted handler_myisam.test and handler_innodb.test to the fact that
  exclusive metadata locks on tables are now acquired according to
  alphabetical order of fully qualified table names instead of order
  in which tables are mentioned in statement.
mysql-test/r/mdl_sync.result:
  Adjusted mdl_sync.test to the fact that exclusive metadata locks on
  tables are now acquired according to alphabetical order of fully
  qualified table names instead of order in which tables are mentioned
  in statement.
mysql-test/t/mdl_sync.test:
  Adjusted mdl_sync.test to the fact that exclusive metadata locks on
  tables are now acquired according to alphabetical order of fully
  qualified table names instead of order in which tables are mentioned
  in statement.
sql/events.cc:
  Instead of using separate MDL_context for opening system tables we now
  create MDL savepoint in the main context before opening such tables
  and rollback to this savepoint after closing them. To support this
  change methods of THD responsible for saving/restoring open table
  state were changed to use Open_tables_backup class which in addition
  to Open_table_state has a member for this savepoint. As result code
  opening/closing system tables was changed to use Open_tables_backup
  instead of Open_table_state class as well.
sql/ha_ndbcluster.cc:
  Since manipulations with open table state no longer install proxy
  MDL_context it does not make sense to perform them in order to
  satisfy assert in mysql_rm_tables_part2(). Removed them per agreement
  with Cluster team. This has not broken test suite since scenario in
  which deadlock can occur and assertion fails is not covered by tests.
sql/lock.cc:
  MDL subsystem no longer implicitly acquires global intention exclusive
  metadata lock when per-object exclusive metadata lock is acquired.
  Now this has to be done by explicit calls outside of MDL subsystem.
sql/log.cc:
  Instead of using separate MDL_context for opening system tables we now
  create MDL savepoint in the main context before opening such tables
  and rollback to this savepoint after closing them. To support this
  change methods of THD responsible for saving/restoring open table
  state were changed to use Open_tables_backup class which in addition
  to Open_table_state has a member for this savepoint. As result code
  opening/closing system tables was changed to use Open_tables_backup
  instead of Open_table_state class as well.
sql/mdl.cc:
  Changed metadata locking subsystem to use mutex per lock and condition
  variable per context instead of one mutex and one conditional variable
  for the whole subsystem.
  Changed approach to handling of global metadata locks. Instead of
  implicitly acquiring intention exclusive locks when user requests
  per-object upgradeable or exclusive locks now we require them to be
  acquired explicitly in the same way as ordinary metadata locks.
  In fact global lock are now ordinary metadata locks in new GLOBAL
  namespace.
  
  To implement these changes:
  - Removed LOCK_mdl mutex and COND_mdl condition variable.
  - Introduced MDL_lock::m_mutex mutexes which protect individual lock
    objects.
  - Replaced mdl_locks hash with MDL_map class, which has hash for
    MDL_lock objects as a member and separate mutex which protects this
    hash. Methods of this class allow to find(), find_or_create() or
    remove() MDL_lock objects in concurrency-friendly fashion (i.e.
    for most common operation, find_or_create(), we don't acquire
    MDL_lock::m_mutex while holding MDL_map::m_mutex. Thanks to MikaelR
    for this idea and benchmarks!). Added three auxiliary members to
    MDL_lock class (m_is_destroyed, m_ref_usage, m_ref_release) to
    support this concurrency-friendly behavior.
  - Introduced MDL_context::m_ctx_wakeup_cond condition variable to be
    used for waiting until this context's pending request can be
    satisfied or its thread has to perform actions to resolve potential
    deadlock. Context which want to wait add ticket corresponding to the
    request to an appropriate queue of waiters in MDL_lock object so
    they can be noticed when other contexts change state of lock and be
    awaken by them by signalling on MDL_context::m_ctx_wakeup_cond.
    As consequence MDL_ticket objects has to be used for any waiting
    in metadata locking subsystem including one which happens in
    MDL_context::wait_for_locks() method.
    Another consequence is that MDL_context is no longer copyable and
    can't be saved/restored when working with system tables.
  - Made MDL_lock an abstract class, which delegates specifying exact
    compatibility matrix to its descendants. Added MDL_global_lock child
    class for global lock (The old is_lock_type_compatible() method
    became can_grant_lock() method of this class). Added MDL_object_lock
    class to represent per-object lock (The old MDL_lock::can_grant_lock()
    became its method). Choice between two classes happens based on MDL
    namespace in MDL_lock::create() method.
  - Got rid of MDL_lock::type member as its meaning became ambigous for
    global locks.
  - To simplify waking up of contexts waiting for lock split waiting queue
    in MDL_lock class in two queues. One for pending requests for exclusive
    (including intention exclusive) locks and another for requests for
    shared locks.
  - Added virtual wake_up_waiters() method to MDL_lock, MDL_global_lock and
    MDL_object_lock classes which allows to wake up waiting contexts after
    state of lock changes. Replaced old duplicated code with calls to this
    method.
  - Adjusted MDL_context::try_acquire_shared_lock()/exclusive_lock()/
    global_shared_lock(), MDL_ticket::upgrade_shared_lock_to_exclusive_lock()
    and MDL_context::release_ticket() methods to use MDL_map and
    MDL_lock::m_mutex instead of single LOCK_mdl mutex and wake up
    waiters according to the approach described above. The latter method
    also was renamed to MDL_context::release_lock().
  - Changed MDL_context::try_acquire_shared_lock()/exclusive_lock() and
    release_lock() not to handle global locks. They are now supposed to
    be taken explicitly like ordinary metadata locks.
  - Added helper MDL_context::try_acquire_global_intention_exclusive_lock()
    and acquire_global_intention_exclusive_lock() methods.
  - Moved common code from MDL_context::acquire_global_shared_lock() and
    acquire_global_intention_exclusive_lock() to new method -
    MDL_context::acquire_lock_impl().
  - Moved common code from MDL_context::try_acquire_shared_lock(),
    try_acquire_global_intention_exclusive_lock()/exclusive_lock()
    to MDL_context::try_acquire_lock_impl().
  - Since acquiring of several exclusive locks can no longer happen under
    single LOCK_mdl mutex the approach to it had to be changed. Now we do
    it in one by one fashion. This is done in alphabetical order to avoid
    deadlocks. Changed MDL_context::acquire_exclusive_locks() accordingly
    (as part of this change moved code responsible for acquiring single
    exclusive lock to new MDL_context::acquire_exclusive_lock_impl()
    method).
  - Since we no longer have single LOCK_mdl mutex which protects all
    MDL_context::m_is_waiting_in_mdl members using these members to
    determine if we have really awaken context holding conflicting
    shared lock became inconvinient. Got rid of this member and changed
    notify_shared_lock() helper function and process of acquiring
    of/upgrading to exclusive lock not to rely on such information.
    Now in MDL_context::acquire_exclusive_lock_impl() and
    MDL_ticket::upgrade_shared_lock_to_exclusive_lock() we simply
    re-try to wake up threads holding conflicting shared locks after
    small time out.
  - Adjusted MDL_context::can_wait_lead_to_deadlock() and
    MDL_ticket::has_pending_conflicting_lock() to use per-lock
    mutexes instead of LOCK_mdl. To do this introduced
    MDL_lock::has_pending_exclusive_lock() method.
sql/mdl.h:
  Changed metadata locking subsystem to use mutex per lock and condition
  variable per context instead of one mutex and one conditional variable
  for the whole subsystem. In order to implement this change:
  
  - Added MDL_key::cmp() method to be able to sort MDL_key objects
    alphabetically. Changed length fields in MDL_key class to uint16
    as 16-bit is enough for length of any key.
  - Changed MDL_ticket::get_ctx() to return pointer to non-const
    object in order to be able to use MDL_context::awake() method
    for such contexts.
  - Got rid of unlocked versions of can_wait_lead_to_deadlock()/
    has_pending_conflicting_lock() methods in MDL_context and
    MDL_ticket. We no longer has single mutex which protects all
    locks. Thus one always has to use versions of these methods
    which acquire per-lock mutexes.
  - MDL_request_list type of list now counts its elements.
  - Added MDL_context::m_ctx_wakeup_cond condition variable to be used
    for waiting until this context's pending request can be satisfied
    or its thread has to perform actions to resolve potential deadlock.
    Added awake() method to wake up context from such wait.
    Addition of condition variable made MDL_context uncopyable.
    As result we no longer can save/restore MDL_context when working
    with system tables. Instead we create MDL savepoint before opening
    those tables and rollback to it once they are closed.
  - MDL_context::release_ticket() became release_lock() method.
  - Added auxiliary MDL_context::acquire_exclusive_lock_impl() method
    which does all necessary work to acquire exclusive lock on one object
    but should not be used directly as it does not enforce any asserts
    ensuring that no deadlocks are possible.
  - Since we no longer need to know if thread trying to acquire exclusive
    lock managed to wake up any threads having conflicting shared locks
    (as, anyway, we will try to wake up such threads again shortly)
  - MDL_context::m_is_waiting_in_mdl member became unnecessary and
    notify_shared_lock() no longer needs to be friend of MDL_context.
  
  Changed approach to handling of global metadata locks. Instead of
  implicitly acquiring intention exclusive locks when user requests
  per-object upgradeable or exclusive locks now we require them to be
  acquired explicitly in the same way as ordinary metadata locks.
  
  - Added new GLOBAL namespace for such locks.
  - Added new type of lock to be requested MDL_INTENTION_EXCLISIVE.
  - Added MDL_context::try_acquire_global_intention_exclusive_lock()
    and acquire_global_intention_exclusive_lock() methods.
  - Moved common code from MDL_context::acquire_global_shared_lock()
    and acquire_global_intention_exclusive_lock() to new method -
    MDL_context::acquire_lock_impl().
  - Moved common code from MDL_context::try_acquire_shared_lock(),
    try_acquire_global_intention_exclusive_lock()/exclusive_lock()
    to MDL_context::try_acquire_lock_impl().
  - Added helper MDL_context::is_global_lock_owner() method to be
    able easily to find what kind of global lock this context holds.
  - MDL_context::m_has_global_shared_lock became unnecessary as
    global read lock is now represented by ordinary ticket.
  - Removed assert in MDL_context::set_lt_or_ha_sentinel() which became
    false for cases when we execute LOCK TABLES under global read lock
    mode.
sql/mysql_priv.h:
  Instead of using separate MDL_context for opening system tables we now
  create MDL savepoint in the main context before opening such tables
  and rollback to this savepoint after closing them. To support this
  change methods of THD responsible for saving/restoring open table
  state were changed to use Open_tables_backup class which in addition
  to Open_table_state has a member for this savepoint. As result calls
  opening/closing system tables were changed to use Open_tables_backup
  instead of Open_table_state class as well.
sql/sp.cc:
  Instead of using separate MDL_context for opening system tables we now
  create MDL savepoint in the main context before opening such tables
  and rollback to this savepoint after closing them. To support this
  change methods of THD responsible for saving/restoring open table
  state were changed to use Open_tables_backup class which in addition
  to Open_table_state has a member for this savepoint. As result code
  opening/closing system tables was changed to use Open_tables_backup
  instead of Open_table_state class as well.
sql/sp.h:
  Instead of using separate MDL_context for opening system tables we now
  create MDL savepoint in the main context before opening such tables
  and rollback to this savepoint after closing them. To support this
  change methods of THD responsible for saving/restoring open table
  state were changed to use Open_tables_backup class which in addition
  to Open_table_state has a member for this savepoint. As result code
  opening/closing system tables was changed to use Open_tables_backup
  instead of Open_table_state class as well.
sql/sql_base.cc:
  close_thread_tables():
    Since we no longer use separate MDL_context for opening system
    tables we need to avoid releasing all transaction locks when
    closing system table. Releasing metadata lock on system table
    is now responsibility of THD::restore_backup_open_tables_state().
  open_table_get_mdl_lock(),
  Open_table_context::recover_from_failed_open():
    MDL subsystem no longer implicitly acquires global intention exclusive
    metadata lock when per-object upgradable or exclusive metadata lock is
    acquired. So this have to be done explicitly from these calls.
    Changed Open_table_context class to store MDL_request object for
    global intention exclusive lock acquired when opening tables.
  open_table():
    Do not release metadata lock if we have failed to open table as
    this lock might have been acquired by one of previous statements
    in transaction, and therefore should not be released.
  open_system_tables_for_read()/close_system_tables()/
  open_performance_schema_table():
    Instead of using separate MDL_context for opening system tables we now
    create MDL savepoint in the main context before opening such tables
    and rollback to this savepoint after closing them. To support this
    change methods of THD responsible for saving/restoring open table
    state were changed to use Open_tables_backup class which in addition
    to Open_table_state has a member for this savepoint. As result code
    opening/closing system tables was changed to use Open_tables_backup
    instead of Open_table_state class as well.
  close_performance_schema_table():
    Got rid of duplicated code.
sql/sql_class.cc:
  Instead of using separate MDL_context for opening system tables we now
  create MDL savepoint in the main context before opening such tables
  and rollback to this savepoint after closing them. To support this
  change methods of THD responsible for saving/restoring open table
  state were changed to use Open_tables_backup class which in addition
  to Open_table_state has a member for this savepoint. Also releasing
  metadata lock on system table is now responsibility of
  THD::restore_backup_open_tables_state().
  Adjusted assert in THD::cleanup() to take into account fact that now
  we also use MDL sentinel for global read lock.
sql/sql_class.h:
  Instead of using separate MDL_context for opening system tables we now
  create MDL savepoint in the main context before opening such tables
  and rollback to this savepoint after closing them. As result:
  - 'mdl_context' member was moved out of Open_tables_state to THD class.
    enter_locked_tables_mode()/leave_locked_tables_mode() had to follow.
  - Methods of THD responsible for saving/restoring open table state were
    changed to use Open_tables_backup class which in addition to
    Open_table_state has a member for this savepoint.
  Changed Open_table_context class to store MDL_request object for
  global intention exclusive lock acquired when opening tables.
sql/sql_delete.cc:
  MDL subsystem no longer implicitly acquires global intention exclusive
  metadata lock when per-object exclusive metadata lock is acquired.
  Now this has to be done by explicit calls outside of MDL subsystem.
sql/sql_help.cc:
  Instead of using separate MDL_context for opening system tables we now
  create MDL savepoint in the main context before opening such tables
  and rollback to this savepoint after closing them. To support this
  change methods of THD responsible for saving/restoring open table
  state were changed to use Open_tables_backup class which in addition
  to Open_table_state has a member for this savepoint. As result code
  opening/closing system tables was changed to use Open_tables_backup
  instead of Open_table_state class as well.
sql/sql_parse.cc:
  Adjusted assert reload_acl_and_cache() to the fact that global read
  lock now takes full-blown metadata lock.
sql/sql_plist.h:
  Added support for element counting to I_P_List list template.
  One can use policy classes to specify if such counting is needed
  or not needed for particular list.
sql/sql_show.cc:
  Instead of using separate MDL_context for opening tables for I_S
  purposes we now create MDL savepoint in the main context before
  opening tables and rollback to this savepoint after closing them.
  To support this and similar change for system tables methods of
  THD responsible for saving/restoring open table state were changed
  to use Open_tables_backup class which in addition to Open_table_state
  has a member for this savepoint. As result code opening/closing tables
  for I_S purposes was changed to use Open_tables_backup instead of
  Open_table_state class as well.
sql/sql_table.cc:
  mysql_rm_tables_part2():
    Since now global intention exclusive metadata lock is ordinary
    metadata lock we no longer can rely that by releasing MDL locks
    on all tables we will release all locks acquired by this routine.
    So in non-LOCK-TABLES mode we have to release all locks acquired
    explicitly.
  prepare_for_repair(), mysql_alter_table():
    MDL subsystem no longer implicitly acquires global intention
    exclusive metadata lock when per-object exclusive metadata lock
    is acquired. Now this has to be done by explicit calls outside of
    MDL subsystem.
sql/tztime.cc:
  Instead of using separate MDL_context for opening system tables we now
  create MDL savepoint in the main context before opening such tables
  and rollback to this savepoint after closing them. To support this
  change methods of THD responsible for saving/restoring open table
  state were changed to use Open_tables_backup class which in addition
  to Open_table_state has a member for this savepoint. As result code
  opening/closing system tables was changed to use Open_tables_backup
  instead of Open_table_state class as well.
  Also changed code not to use special mechanism for open system tables
  when it is not really necessary.

24 files changed, 1574 insertions, 827 deletions

diff --git a/mysql-test/include/handler.inc b/mysql-test/include/handler.inc
index 5562f7b2558..8342a072ef9 100644
--- a/mysql-test/include/handler.inc
+++ b/mysql-test/include/handler.inc
@@ -543,7 +543,7 @@ disconnect flush;
 #
 
 --disable_warnings
-drop table if exists t1,t2;
+drop table if exists t1, t0;
 --enable_warnings
 create table t1 (c1 int);
 --echo connection: default
@@ -552,31 +552,31 @@ handler t1 read first;
 connect (flush,localhost,root,,);
 connection flush;
 --echo connection: flush
---send rename table t1 to t2;
+--send rename table t1 to t0;
 connection waiter;
 --echo connection: waiter 
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where state = "Waiting for table" and info = "rename table t1 to t2";
+  where state = "Waiting for table" and info = "rename table t1 to t0";
 --source include/wait_condition.inc
 connection default;
 --echo connection: default
 --echo #
 --echo # RENAME placed two pending locks and waits.
---echo # When HANDLER t2 OPEN does open_tables(), it calls
+--echo # When HANDLER t0 OPEN does open_tables(), it calls
 --echo # mysql_ha_flush(), which in turn closes the open HANDLER for t1.
 --echo # RENAME TABLE gets unblocked. If it gets scheduled quickly
 --echo # and manages to complete before open_tables()
---echo # of HANDLER t2 OPEN, open_tables() and therefore the whole
---echo # HANDLER t2 OPEN succeeds. Otherwise open_tables() 
+--echo # of HANDLER t0 OPEN, open_tables() and therefore the whole
+--echo # HANDLER t0 OPEN succeeds. Otherwise open_tables() 
 --echo # notices a pending or active exclusive metadata lock on t2
---echo # and the whole HANDLER t2 OPEN fails with ER_LOCK_DEADLOCK
+--echo # and the whole HANDLER t0 OPEN fails with ER_LOCK_DEADLOCK
 --echo # error.
 --echo #
 --error 0, ER_LOCK_DEADLOCK
-handler t2 open;
+handler t0 open;
 --error 0, ER_UNKNOWN_TABLE
-handler t2 close;
+handler t0 close;
 --echo connection: flush
 connection flush;
 reap;
@@ -585,7 +585,7 @@ handler t1 read next;
 --error ER_UNKNOWN_TABLE
 handler t1 close;
 connection default;
-drop table t2;
+drop table t0;
 connection flush;
 disconnect flush;
 --source include/wait_until_disconnected.inc
@@ -972,35 +972,29 @@ connection default;
 --echo #
 create table t1 (a int, key a (a));
 insert into t1 (a) values (1), (2), (3), (4), (5);
-create table t2 (a int, key a (a));
-insert into t2 (a) values (1), (2), (3), (4), (5);
+create table t0 (a int, key a (a));
+insert into t0 (a) values (1), (2), (3), (4), (5);
 begin;
 select * from t1;
---echo # --> connection con1
-connection con1;
-lock table t2 read;
 --echo # --> connection con2 
 connection con2;
 --echo # Sending:
-send rename table t2 to t3, t1 to t2, t3 to t1;
+send rename table t0 to t3, t1 to t0, t3 to t1;
 --echo # --> connection con1 
 connection con1;
 --echo # Waiting for 'rename table ...' to get blocked...
 let $wait_condition=select count(*)=1 from information_schema.processlist
-where state='Waiting for table' and info='rename table t2 to t3, t1 to t2, t3 to t1';
+where state='Waiting for table' and info='rename table t0 to t3, t1 to t0, t3 to t1';
 --source include/wait_condition.inc
 --echo # --> connection default
 connection default;
 --error ER_LOCK_DEADLOCK
-handler t2 open;
+handler t0 open;
 --error ER_LOCK_DEADLOCK
-select * from t2;
+select * from t0;
 handler t1 open;
 commit;
 handler t1 close;
---echo # --> connection con1
-connection con1;
-unlock tables;
 --echo # --> connection con2
 connection con2;
 --echo # Reaping 'rename table ...'...
@@ -1010,7 +1004,7 @@ connection default;
 handler t1 open;
 handler t1 read a prev;
 handler t1 close;
-drop table t2;
+drop table t0;
 --echo #
 --echo # Originally there was a deadlock error in this test.
 --echo # With implementation of deadlock detector
diff --git a/mysql-test/r/handler_innodb.result b/mysql-test/r/handler_innodb.result
index 807e8becea8..a3e3e325e7d 100644
--- a/mysql-test/r/handler_innodb.result
+++ b/mysql-test/r/handler_innodb.result
@@ -560,36 +560,36 @@ c1
 handler t1 close;
 handler t2 close;
 drop table t1,t2;
-drop table if exists t1,t2;
+drop table if exists t1, t0;
 create table t1 (c1 int);
 connection: default
 handler t1 open;
 handler t1 read first;
 c1
 connection: flush
-rename table t1 to t2;;
+rename table t1 to t0;;
 connection: waiter 
 connection: default
 #
 # RENAME placed two pending locks and waits.
-# When HANDLER t2 OPEN does open_tables(), it calls
+# When HANDLER t0 OPEN does open_tables(), it calls
 # mysql_ha_flush(), which in turn closes the open HANDLER for t1.
 # RENAME TABLE gets unblocked. If it gets scheduled quickly
 # and manages to complete before open_tables()
-# of HANDLER t2 OPEN, open_tables() and therefore the whole
-# HANDLER t2 OPEN succeeds. Otherwise open_tables() 
+# of HANDLER t0 OPEN, open_tables() and therefore the whole
+# HANDLER t0 OPEN succeeds. Otherwise open_tables() 
 # notices a pending or active exclusive metadata lock on t2
-# and the whole HANDLER t2 OPEN fails with ER_LOCK_DEADLOCK
+# and the whole HANDLER t0 OPEN fails with ER_LOCK_DEADLOCK
 # error.
 #
-handler t2 open;
-handler t2 close;
+handler t0 open;
+handler t0 close;
 connection: flush
 handler t1 read next;
 ERROR 42S02: Unknown table 't1' in HANDLER
 handler t1 close;
 ERROR 42S02: Unknown table 't1' in HANDLER
-drop table t2;
+drop table t0;
 drop table if exists t1;
 create temporary table t1 (a int, b char(1), key a(a), key b(a,b));
 insert into t1 values (0,"a"),(1,"b"),(2,"c"),(3,"d"),(4,"e"),
@@ -989,8 +989,8 @@ handler t1 close;
 #
 create table t1 (a int, key a (a));
 insert into t1 (a) values (1), (2), (3), (4), (5);
-create table t2 (a int, key a (a));
-insert into t2 (a) values (1), (2), (3), (4), (5);
+create table t0 (a int, key a (a));
+insert into t0 (a) values (1), (2), (3), (4), (5);
 begin;
 select * from t1;
 a
@@ -999,23 +999,19 @@ a
 3
 4
 5
-# --> connection con1
-lock table t2 read;
 # --> connection con2 
 # Sending:
-rename table t2 to t3, t1 to t2, t3 to t1;
+rename table t0 to t3, t1 to t0, t3 to t1;
 # --> connection con1 
 # Waiting for 'rename table ...' to get blocked...
 # --> connection default
-handler t2 open;
+handler t0 open;
 ERROR 40001: Deadlock found when trying to get lock; try restarting transaction
-select * from t2;
+select * from t0;
 ERROR 40001: Deadlock found when trying to get lock; try restarting transaction
 handler t1 open;
 commit;
 handler t1 close;
-# --> connection con1
-unlock tables;
 # --> connection con2
 # Reaping 'rename table ...'...
 # --> connection default
@@ -1024,7 +1020,7 @@ handler t1 read a prev;
 a
 5
 handler t1 close;
-drop table t2;
+drop table t0;
 #
 # Originally there was a deadlock error in this test.
 # With implementation of deadlock detector
diff --git a/mysql-test/r/handler_myisam.result b/mysql-test/r/handler_myisam.result
index adcbf068b97..f5c5bfebd15 100644
--- a/mysql-test/r/handler_myisam.result
+++ b/mysql-test/r/handler_myisam.result
@@ -559,36 +559,36 @@ c1
 handler t1 close;
 handler t2 close;
 drop table t1,t2;
-drop table if exists t1,t2;
+drop table if exists t1, t0;
 create table t1 (c1 int);
 connection: default
 handler t1 open;
 handler t1 read first;
 c1
 connection: flush
-rename table t1 to t2;;
+rename table t1 to t0;;
 connection: waiter 
 connection: default
 #
 # RENAME placed two pending locks and waits.
-# When HANDLER t2 OPEN does open_tables(), it calls
+# When HANDLER t0 OPEN does open_tables(), it calls
 # mysql_ha_flush(), which in turn closes the open HANDLER for t1.
 # RENAME TABLE gets unblocked. If it gets scheduled quickly
 # and manages to complete before open_tables()
-# of HANDLER t2 OPEN, open_tables() and therefore the whole
-# HANDLER t2 OPEN succeeds. Otherwise open_tables() 
+# of HANDLER t0 OPEN, open_tables() and therefore the whole
+# HANDLER t0 OPEN succeeds. Otherwise open_tables() 
 # notices a pending or active exclusive metadata lock on t2
-# and the whole HANDLER t2 OPEN fails with ER_LOCK_DEADLOCK
+# and the whole HANDLER t0 OPEN fails with ER_LOCK_DEADLOCK
 # error.
 #
-handler t2 open;
-handler t2 close;
+handler t0 open;
+handler t0 close;
 connection: flush
 handler t1 read next;
 ERROR 42S02: Unknown table 't1' in HANDLER
 handler t1 close;
 ERROR 42S02: Unknown table 't1' in HANDLER
-drop table t2;
+drop table t0;
 drop table if exists t1;
 create temporary table t1 (a int, b char(1), key a(a), key b(a,b));
 insert into t1 values (0,"a"),(1,"b"),(2,"c"),(3,"d"),(4,"e"),
@@ -986,8 +986,8 @@ handler t1 close;
 #
 create table t1 (a int, key a (a));
 insert into t1 (a) values (1), (2), (3), (4), (5);
-create table t2 (a int, key a (a));
-insert into t2 (a) values (1), (2), (3), (4), (5);
+create table t0 (a int, key a (a));
+insert into t0 (a) values (1), (2), (3), (4), (5);
 begin;
 select * from t1;
 a
@@ -996,23 +996,19 @@ a
 3
 4
 5
-# --> connection con1
-lock table t2 read;
 # --> connection con2 
 # Sending:
-rename table t2 to t3, t1 to t2, t3 to t1;
+rename table t0 to t3, t1 to t0, t3 to t1;
 # --> connection con1 
 # Waiting for 'rename table ...' to get blocked...
 # --> connection default
-handler t2 open;
+handler t0 open;
 ERROR 40001: Deadlock found when trying to get lock; try restarting transaction
-select * from t2;
+select * from t0;
 ERROR 40001: Deadlock found when trying to get lock; try restarting transaction
 handler t1 open;
 commit;
 handler t1 close;
-# --> connection con1
-unlock tables;
 # --> connection con2
 # Reaping 'rename table ...'...
 # --> connection default
@@ -1021,7 +1017,7 @@ handler t1 read a prev;
 a
 5
 handler t1 close;
-drop table t2;
+drop table t0;
 #
 # Originally there was a deadlock error in this test.
 # With implementation of deadlock detector
diff --git a/mysql-test/r/mdl_sync.result b/mysql-test/r/mdl_sync.result
index 0c9b6432e95..8c4d7272e29 100644
--- a/mysql-test/r/mdl_sync.result
+++ b/mysql-test/r/mdl_sync.result
@@ -23,7 +23,7 @@ SET DEBUG_SYNC= 'RESET';
 # Test coverage for basic deadlock detection in metadata
 # locking subsystem.
 #
-drop tables if exists t1, t2, t3, t4;
+drop tables if exists t0, t1, t2, t3, t4, t5;
 create table t1 (i int);
 create table t2 (j int);
 create table t3 (k int);
@@ -90,7 +90,7 @@ commit;
 #
 # Switching to connection 'deadlock_con1'.
 begin;
-insert into t1 values (2);
+insert into t2 values (2);
 #
 # Switching to connection 'default'.
 # Send:
@@ -98,11 +98,11 @@ rename table t2 to t0, t1 to t2, t0 to t1;;
 #
 # Switching to connection 'deadlock_con1'.
 # Wait until the above RENAME TABLE is blocked because it has to wait
-# for 'deadlock_con1' which holds shared metadata lock on 't1'.
+# for 'deadlock_con1' which holds shared metadata lock on 't2'.
 # 
 # The below statement should not wait as doing so will cause deadlock.
 # Instead it should fail and emit ER_LOCK_DEADLOCK statement.
-select * from t2;
+select * from t1;
 ERROR 40001: Deadlock found when trying to get lock; try restarting transaction
 #
 # Let us check that failure of the above statement has not released
@@ -141,7 +141,7 @@ select * from t2;;
 # for an exclusive metadata lock to go away.
 # Send RENAME TABLE statement that will deadlock with the
 # SELECT statement and thus should abort the latter.
-rename table t1 to t0, t2 to t1, t0 to t2;;
+rename table t1 to t5, t2 to t1, t5 to t2;;
 #
 # Switching to connection 'deadlock_con1'.
 # Since the latest RENAME TABLE entered in deadlock with SELECT
@@ -156,15 +156,17 @@ ERROR 40001: Deadlock found when trying to get lock; try restarting transaction
 # Commit transaction to unblock this RENAME TABLE.
 commit;
 #
-# Switching to connection 'deadlock_con3'.
-# Reap RENAME TABLE t1 TO t0 ... .
-#
 # Switching to connection 'deadlock_con2'.
 # Commit transaction to unblock the first RENAME TABLE.
 commit;
 #
 # Switching to connection 'default'.
 # Reap RENAME TABLE t2 TO t0 ... .
+#
+# Switching to connection 'deadlock_con3'.
+# Reap RENAME TABLE t1 TO t5 ... .
+#
+# Switching to connection 'default'.
 drop tables t1, t2, t3, t4;
 #
 # Now, test case which shows that deadlock detection empiric
diff --git a/mysql-test/t/mdl_sync.test b/mysql-test/t/mdl_sync.test
index c817012fb2f..4cbaa689339 100644
--- a/mysql-test/t/mdl_sync.test
+++ b/mysql-test/t/mdl_sync.test
@@ -78,7 +78,7 @@ SET DEBUG_SYNC= 'RESET';
 --echo # locking subsystem.
 --echo #
 --disable_warnings
-drop tables if exists t1, t2, t3, t4;
+drop tables if exists t0, t1, t2, t3, t4, t5;
 --enable_warnings
 
 connect(deadlock_con1,localhost,root,,);
@@ -189,7 +189,7 @@ connection default;
 --echo # Switching to connection 'deadlock_con1'.
 connection deadlock_con1;
 begin;
-insert into t1 values (2);
+insert into t2 values (2);
 
 --echo #
 --echo # Switching to connection 'default'.
@@ -201,7 +201,7 @@ connection default;
 --echo # Switching to connection 'deadlock_con1'.
 connection deadlock_con1;
 --echo # Wait until the above RENAME TABLE is blocked because it has to wait
---echo # for 'deadlock_con1' which holds shared metadata lock on 't1'.
+--echo # for 'deadlock_con1' which holds shared metadata lock on 't2'.
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
   where state = "Waiting for table" and info = "rename table t2 to t0, t1 to t2, t0 to t1";
@@ -210,7 +210,7 @@ let $wait_condition=
 --echo # The below statement should not wait as doing so will cause deadlock.
 --echo # Instead it should fail and emit ER_LOCK_DEADLOCK statement.
 --error ER_LOCK_DEADLOCK
-select * from t2;
+select * from t1;
 
 --echo #
 --echo # Let us check that failure of the above statement has not released
@@ -276,7 +276,7 @@ let $wait_condition=
 
 --echo # Send RENAME TABLE statement that will deadlock with the
 --echo # SELECT statement and thus should abort the latter.
---send rename table t1 to t0, t2 to t1, t0 to t2;
+--send rename table t1 to t5, t2 to t1, t5 to t2;
 
 --echo #
 --echo # Switching to connection 'deadlock_con1'.
@@ -294,18 +294,12 @@ connection deadlock_con1;
 --echo # is blocked.
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where state = "Waiting for table" and info = "rename table t1 to t0, t2 to t1, t0 to t2";
+  where state = "Waiting for table" and info = "rename table t1 to t5, t2 to t1, t5 to t2";
 --source include/wait_condition.inc
 --echo # Commit transaction to unblock this RENAME TABLE.
 commit;
 
 --echo #
---echo # Switching to connection 'deadlock_con3'.
-connection deadlock_con3;
---echo # Reap RENAME TABLE t1 TO t0 ... .
---reap;
-
---echo #
 --echo # Switching to connection 'deadlock_con2'.
 connection deadlock_con2;
 --echo # Commit transaction to unblock the first RENAME TABLE.
@@ -317,6 +311,16 @@ connection default;
 --echo # Reap RENAME TABLE t2 TO t0 ... .
 --reap
 
+--echo #
+--echo # Switching to connection 'deadlock_con3'.
+connection deadlock_con3;
+--echo # Reap RENAME TABLE t1 TO t5 ... .
+--reap;
+
+--echo #
+--echo # Switching to connection 'default'.
+connection default;
+
 drop tables t1, t2, t3, t4;
 
 --echo #
diff --git a/sql/events.cc b/sql/events.cc
index d4efcdbdc4a..ef5bf801f1e 100644
--- a/sql/events.cc
+++ b/sql/events.cc
@@ -770,7 +770,7 @@ send_show_create_event(THD *thd, Event_timed *et, Protocol *protocol)
 bool
 Events::show_create_event(THD *thd, LEX_STRING dbname, LEX_STRING name)
 {
-  Open_tables_state open_tables_backup;
+  Open_tables_backup open_tables_backup;
   Event_timed et;
   bool ret;
 
@@ -826,7 +826,7 @@ Events::fill_schema_events(THD *thd, TABLE_LIST *tables, COND * /* cond */)
 {
   char *db= NULL;
   int ret;
-  Open_tables_state open_tables_backup;
+  Open_tables_backup open_tables_backup;
   DBUG_ENTER("Events::fill_schema_events");
 
   if (check_if_system_tables_error())
diff --git a/sql/ha_ndbcluster.cc b/sql/ha_ndbcluster.cc
index b35746102c8..220e5c460db 100644
--- a/sql/ha_ndbcluster.cc
+++ b/sql/ha_ndbcluster.cc
@@ -7285,14 +7285,10 @@ int ndbcluster_find_files(handlerton *hton, THD *thd,
     code below will try to obtain exclusive metadata lock on some table
     while holding shared meta-data lock on other tables. This might lead to
     a deadlock, and therefore is disallowed by assertions of the metadata
-    locking subsystem. In order to temporarily make the code work, we must
-    reset and backup the open tables state, thus hide the existing locks
-    from MDL asserts. But in the essence this is violation of metadata
+    locking subsystem. This is violation of metadata
     locking protocol which has to be closed ASAP.
+    XXX: the scenario described above is not covered with any test.
   */
-  Open_tables_state open_tables_state_backup;
-  thd->reset_n_backup_open_tables_state(&open_tables_state_backup);
-
   if (!global_read_lock)
   {
     // Delete old files
@@ -7316,8 +7312,6 @@ int ndbcluster_find_files(handlerton *hton, THD *thd,
     }
   }
 
-  thd->restore_backup_open_tables_state(&open_tables_state_backup);
-
   /* Lock mutex before creating .FRM files. */
   pthread_mutex_lock(&LOCK_open);
   /* Create new files. */
diff --git a/sql/lock.cc b/sql/lock.cc
index 6cdf2e4a202..9d794b07418 100644
--- a/sql/lock.cc
+++ b/sql/lock.cc
@@ -949,8 +949,11 @@ static MYSQL_LOCK *get_lock_data(THD *thd, TABLE **table_ptr, uint count,
 bool lock_table_names(THD *thd, TABLE_LIST *table_list)
 {
   MDL_request_list mdl_requests;
+  MDL_request global_request;
   TABLE_LIST *lock_table;
 
+  global_request.init(MDL_key::GLOBAL, "", "", MDL_INTENTION_EXCLUSIVE);
+
   for (lock_table= table_list; lock_table; lock_table= lock_table->next_local)
   {
     lock_table->mdl_request.init(MDL_key::TABLE,
@@ -958,8 +961,15 @@ bool lock_table_names(THD *thd, TABLE_LIST *table_list)
                                  MDL_EXCLUSIVE);
     mdl_requests.push_front(&lock_table->mdl_request);
   }
+
+  if (thd->mdl_context.acquire_global_intention_exclusive_lock(&global_request))
+    return 1;
+
   if (thd->mdl_context.acquire_exclusive_locks(&mdl_requests))
+  {
+    thd->mdl_context.release_lock(global_request.ticket);
     return 1;
+  }
   return 0;
 }
 
@@ -1009,6 +1019,7 @@ bool lock_routine_name(THD *thd, bool is_function,
   MDL_key::enum_mdl_namespace mdl_type= (is_function ?
                                          MDL_key::FUNCTION :
                                          MDL_key::PROCEDURE);
+  MDL_request global_request;
   MDL_request mdl_request;
 
   if (thd->locked_tables_mode)
@@ -1021,10 +1032,17 @@ bool lock_routine_name(THD *thd, bool is_function,
   DBUG_ASSERT(name);
   DEBUG_SYNC(thd, "before_wait_locked_pname");
 
+  global_request.init(MDL_key::GLOBAL, "", "", MDL_INTENTION_EXCLUSIVE);
   mdl_request.init(mdl_type, db, name, MDL_EXCLUSIVE);
 
+  if (thd->mdl_context.acquire_global_intention_exclusive_lock(&global_request))
+    return TRUE;
+
   if (thd->mdl_context.acquire_exclusive_lock(&mdl_request))
+  {
+    thd->mdl_context.release_lock(global_request.ticket);
     return TRUE;
+  }
 
   DEBUG_SYNC(thd, "after_wait_locked_pname");
   return FALSE;
diff --git a/sql/log.cc b/sql/log.cc
index a74fc94d09d..e1f98ffa1bf 100644
--- a/sql/log.cc
+++ b/sql/log.cc
@@ -410,7 +410,7 @@ bool Log_to_csv_event_handler::
   bool need_rnd_end= FALSE;
   uint field_index;
   Silence_log_table_errors error_handler;
-  Open_tables_state open_tables_backup;
+  Open_tables_backup open_tables_backup;
   ulonglong save_thd_options;
   bool save_time_zone_used;
 
@@ -572,7 +572,7 @@ bool Log_to_csv_event_handler::
   bool need_close= FALSE;
   bool need_rnd_end= FALSE;
   Silence_log_table_errors error_handler;
-  Open_tables_state open_tables_backup;
+  Open_tables_backup open_tables_backup;
   CHARSET_INFO *client_cs= thd->variables.character_set_client;
   bool save_time_zone_used;
   DBUG_ENTER("Log_to_csv_event_handler::log_slow");
@@ -727,7 +727,7 @@ int Log_to_csv_event_handler::
   TABLE *table;
   LEX_STRING *UNINIT_VAR(log_name);
   int result;
-  Open_tables_state open_tables_backup;
+  Open_tables_backup open_tables_backup;
 
   DBUG_ENTER("Log_to_csv_event_handler::activate_log");
 
diff --git a/sql/mdl.cc b/sql/mdl.cc
index af7f310e598..dce917a1a2e 100644
--- a/sql/mdl.cc
+++ b/sql/mdl.cc
@@ -21,11 +21,40 @@
 
 static bool mdl_initialized= 0;
 
+
+/**
+  A collection of all MDL locks. A singleton,
+  there is only one instance of the map in the server.
+  Maps MDL_key to MDL_lock instances.
+*/
+
+class MDL_map
+{
+public:
+  void init();
+  void destroy();
+  MDL_lock *find(const MDL_key *key);
+  MDL_lock *find_or_insert(const MDL_key *key);
+  void remove(MDL_lock *lock);
+private:
+  bool move_from_hash_to_lock_mutex(MDL_lock *lock);
+private:
+  /** All acquired locks in the server. */
+  HASH m_locks;
+  /* Protects access to m_locks hash. */
+  pthread_mutex_t m_mutex;
+};
+
+
 /**
   The lock context. Created internally for an acquired lock.
   For a given name, there exists only one MDL_lock instance,
   and it exists only when the lock has been granted.
   Can be seen as an MDL subsystem's version of TABLE_SHARE.
+
+  This is an abstract class which lacks information about
+  compatibility rules for lock types. They should be specified
+  in its descendants.
 */
 
 class MDL_lock
@@ -39,78 +68,128 @@ public:
 
   typedef Ticket_list::Iterator Ticket_iterator;
 
-  /** The type of lock (shared or exclusive). */
-  enum
-  {
-    MDL_LOCK_SHARED,
-    MDL_LOCK_EXCLUSIVE,
-  } type;
+public:
   /** The key of the object (data) being protected. */
   MDL_key key;
   /** List of granted tickets for this lock. */
   Ticket_list granted;
+  /** Tickets for contexts waiting to acquire a shared lock. */
+  Ticket_list waiting_shared;
   /**
+    Tickets for contexts waiting to acquire an exclusive lock.
     There can be several upgraders and active exclusive
     locks belonging to the same context. E.g.
     in case of RENAME t1 to t2, t2 to t3, we attempt to
     exclusively lock t2 twice.
   */
-  Ticket_list waiting;
+  Ticket_list waiting_exclusive;
   void   *cached_object;
   mdl_cached_object_release_hook cached_object_release_hook;
+  /** Mutex protecting this lock context. */
+  pthread_mutex_t m_mutex;
 
   bool is_empty() const
   {
-    return (granted.is_empty() && waiting.is_empty());
+    return (granted.is_empty() && waiting_shared.is_empty() &&
+            waiting_exclusive.is_empty());
   }
 
-  bool can_grant_lock(const MDL_context *requestor_ctx,
-                      enum_mdl_type type, bool is_upgrade);
+  bool has_pending_exclusive_lock()
+  {
+    bool has_locks;
+    pthread_mutex_lock(&m_mutex);
+    has_locks= ! waiting_exclusive.is_empty();
+    pthread_mutex_unlock(&m_mutex);
+    return has_locks;
+  }
+  virtual bool can_grant_lock(const MDL_context *requestor_ctx,
+                              enum_mdl_type type, bool is_upgrade)= 0;
+  virtual void wake_up_waiters()= 0;
 
   inline static MDL_lock *create(const MDL_key *key);
-  inline static void destroy(MDL_lock *lock);
-private:
+
   MDL_lock(const MDL_key *key_arg)
-  : type(MDL_LOCK_SHARED),
-    key(key_arg),
+  : key(key_arg),
     cached_object(NULL),
-    cached_object_release_hook(NULL)
+    cached_object_release_hook(NULL),
+    m_ref_usage(0),
+    m_ref_release(0),
+    m_is_destroyed(FALSE)
   {
+    pthread_mutex_init(&m_mutex, NULL);
   }
-};
 
+  virtual ~MDL_lock()
+  {
+    pthread_mutex_destroy(&m_mutex);
+  }
+  inline static void destroy(MDL_lock *lock);
+public:
+  /**
+    These three members are used to make it possible to separate
+    the mdl_locks.m_mutex mutex and MDL_lock::m_mutex in
+    MDL_map::find_or_insert() for increased scalability.
+    The 'm_is_destroyed' member is only set by destroyers that
+    have both the mdl_locks.m_mutex and MDL_lock::m_mutex, thus
+    holding any of the mutexes is sufficient to read it.
+    The 'm_ref_usage; is incremented under protection by
+    mdl_locks.m_mutex, but when 'm_is_destroyed' is set to TRUE, this
+    member is moved to be protected by the MDL_lock::m_mutex.
+    This means that the MDL_map::find_or_insert() which only
+    holds the MDL_lock::m_mutex can compare it to 'm_ref_release'
+    without acquiring mdl_locks.m_mutex again and if equal it can also
+    destroy the lock object safely.
+    The 'm_ref_release' is incremented under protection by
+    MDL_lock::m_mutex.
+    Note since we are only interested in equality of these two
+    counters we don't have to worry about overflows as long as
+    their size is big enough to hold maximum number of concurrent
+    threads on the system.
+  */
+  uint m_ref_usage;
+  uint m_ref_release;
+  bool m_is_destroyed;
+};
 
-static pthread_mutex_t LOCK_mdl;
-static pthread_cond_t  COND_mdl;
-static HASH mdl_locks;
 
 /**
-  An implementation of the global metadata lock. The only
-  locking modes which are supported at the moment are SHARED and
-  INTENTION EXCLUSIVE. Note, that SHARED global metadata lock
-  is acquired automatically when one tries to acquire an EXCLUSIVE
-  or UPGRADABLE SHARED metadata lock on an individual object.
+  An implementation of the global metadata lock. The only locking modes
+  which are supported at the moment are SHARED and INTENTION EXCLUSIVE.
 */
 
-class MDL_global_lock
+class MDL_global_lock : public MDL_lock
 {
 public:
-  uint waiting_shared;
-  uint active_shared;
-  uint active_intention_exclusive;
+  MDL_global_lock(const MDL_key *key_arg)
+    : MDL_lock(key_arg)
+  { }
 
-  bool is_empty() const
-  {
-    return (waiting_shared == 0 && active_shared == 0 &&
-            active_intention_exclusive == 0);
-  }
-  bool is_lock_type_compatible(enum_mdl_type type, bool is_upgrade) const;
+  virtual bool can_grant_lock(const MDL_context *requestor_ctx,
+                              enum_mdl_type type, bool is_upgrade);
+  virtual void wake_up_waiters();
 };
 
 
-static MDL_global_lock global_lock;
+/**
+  An implementation of a per-object lock. Supports SHARED, SHARED_UPGRADABLE,
+  SHARED HIGH PRIORITY and EXCLUSIVE locks.
+*/
+
+class MDL_object_lock : public MDL_lock
+{
+public:
+  MDL_object_lock(const MDL_key *key_arg)
+    : MDL_lock(key_arg)
+  { }
+
+  virtual bool can_grant_lock(const MDL_context *requestor_ctx,
+                              enum_mdl_type type, bool is_upgrade);
+  virtual void wake_up_waiters();
+};
 
 
+static MDL_map mdl_locks;
+
 extern "C"
 {
 static uchar *
@@ -147,12 +226,7 @@ void mdl_init()
 {
   DBUG_ASSERT(! mdl_initialized);
   mdl_initialized= TRUE;
-  pthread_mutex_init(&LOCK_mdl, NULL);
-  pthread_cond_init(&COND_mdl, NULL);
-  my_hash_init(&mdl_locks, &my_charset_bin, 16 /* FIXME */, 0, 0,
-               mdl_locks_key, 0, 0);
-  /* The global lock is zero-initialized by the loader. */
-  DBUG_ASSERT(global_lock.is_empty());
+  mdl_locks.init();
 }
 
 
@@ -168,35 +242,199 @@ void mdl_destroy()
   if (mdl_initialized)
   {
     mdl_initialized= FALSE;
-    DBUG_ASSERT(!mdl_locks.records);
-    DBUG_ASSERT(global_lock.is_empty());
-    pthread_mutex_destroy(&LOCK_mdl);
-    pthread_cond_destroy(&COND_mdl);
-    my_hash_free(&mdl_locks);
+    mdl_locks.destroy();
+  }
+}
+
+
+/** Initialize the global hash containing all MDL locks. */
+
+void MDL_map::init()
+{
+  pthread_mutex_init(&m_mutex, NULL);
+  my_hash_init(&m_locks, &my_charset_bin, 16 /* FIXME */, 0, 0,
+               mdl_locks_key, 0, 0);
+}
+
+
+/**
+  Destroy the global hash containing all MDL locks.
+  @pre It must be empty.
+*/
+
+void MDL_map::destroy()
+{
+  DBUG_ASSERT(!m_locks.records);
+  pthread_mutex_destroy(&m_mutex);
+  my_hash_free(&m_locks);
+}
+
+
+/**
+  Find MDL_lock object corresponding to the key, create it
+  if it does not exist.
+
+  @retval non-NULL - Success. MDL_lock instance for the key with
+                     locked MDL_lock::m_mutex.
+  @retval NULL     - Failure (OOM).
+*/
+
+MDL_lock* MDL_map::find_or_insert(const MDL_key *mdl_key)
+{
+  MDL_lock *lock;
+
+retry:
+  pthread_mutex_lock(&m_mutex);
+  if (!(lock= (MDL_lock*) my_hash_search(&m_locks,
+                                         mdl_key->ptr(),
+                                         mdl_key->length())))
+  {
+    lock= MDL_lock::create(mdl_key);
+    if (!lock || my_hash_insert(&m_locks, (uchar*)lock))
+    {
+      pthread_mutex_unlock(&m_mutex);
+      MDL_lock::destroy(lock);
+      return NULL;
+    }
   }
+
+  if (move_from_hash_to_lock_mutex(lock))
+    goto retry;
+
+  return lock;
 }
 
 
 /**
-  Initialize a metadata locking context.
+  Find MDL_lock object corresponding to the key.
 
-  This is to be called when a new server connection is created.
+  @retval non-NULL - MDL_lock instance for the key with locked
+                     MDL_lock::m_mutex.
+  @retval NULL     - There was no MDL_lock for the key.
 */
 
-void MDL_context::init(THD *thd_arg)
+MDL_lock* MDL_map::find(const MDL_key *mdl_key)
 {
-  m_has_global_shared_lock= FALSE;
-  m_thd= thd_arg;
-  m_lt_or_ha_sentinel= NULL;
+  MDL_lock *lock;
+
+retry:
+  pthread_mutex_lock(&m_mutex);
+  if (!(lock= (MDL_lock*) my_hash_search(&m_locks,
+                                         mdl_key->ptr(),
+                                         mdl_key->length())))
+  {
+    pthread_mutex_unlock(&m_mutex);
+    return NULL;
+  }
+
+  if (move_from_hash_to_lock_mutex(lock))
+    goto retry;
+
+  return lock;
+}
+
+
+/**
+  Release mdl_locks.m_mutex mutex and lock MDL_lock::m_mutex for lock
+  object from the hash. Handle situation when object was released
+  while the held no mutex.
+
+  @retval FALSE - Success.
+  @retval TRUE  - Object was released while we held no mutex, caller
+                  should re-try looking up MDL_lock object in the hash.
+*/
+
+bool MDL_map::move_from_hash_to_lock_mutex(MDL_lock *lock)
+{
+  DBUG_ASSERT(! lock->m_is_destroyed);
+  safe_mutex_assert_owner(&m_mutex);
+
+  /*
+    We increment m_ref_usage which is a reference counter protected by
+    mdl_locks.m_mutex under the condition it is present in the hash and
+    m_is_destroyed is FALSE.
+  */
+  lock->m_ref_usage++;
+  pthread_mutex_unlock(&m_mutex);
+
+  pthread_mutex_lock(&lock->m_mutex);
+  lock->m_ref_release++;
+  if (unlikely(lock->m_is_destroyed))
+  {
+    /*
+      Object was released while we held no mutex, we need to
+      release it if no others hold references to it, while our own
+      reference count ensured that the object as such haven't got
+      its memory released yet. We can also safely compare
+      m_ref_usage and m_ref_release since the object is no longer
+      present in the hash so no one will be able to find it and
+      increment m_ref_usage anymore.
+    */
+    uint ref_usage= lock->m_ref_usage;
+    uint ref_release= lock->m_ref_release;
+    pthread_mutex_unlock(&lock->m_mutex);
+    if (ref_usage == ref_release)
+      MDL_lock::destroy(lock);
+    return TRUE;
+  }
+  return FALSE;
+}
+
+
+/**
+  Destroy MDL_lock object or delegate this responsibility to
+  whatever thread that holds the last outstanding reference to
+  it.
+*/
+
+void MDL_map::remove(MDL_lock *lock)
+{
+  uint ref_usage, ref_release;
+
+  safe_mutex_assert_owner(&lock->m_mutex);
+
+  if (lock->cached_object)
+    (*lock->cached_object_release_hook)(lock->cached_object);
+
   /*
-    FIXME: In reset_n_backup_open_tables_state,
-    we abuse "init" as a reset, i.e. call it on an already
-    constructed non-empty object. This is why we can't
-    rely here on the default constructors of I_P_List
-    to empty the list.
+    Destroy the MDL_lock object, but ensure that anyone that is
+    holding a reference to the object is not remaining, if so he
+    has the responsibility to release it.
+
+    Setting of m_is_destroyed to TRUE while holding _both_
+    mdl_locks.m_mutex and MDL_lock::m_mutex mutexes transfers the
+    protection of m_ref_usage from mdl_locks.m_mutex to
+    MDL_lock::m_mutex while removal of object from the hash makes
+    it read-only.  Therefore whoever acquires MDL_lock::m_mutex next
+    will see most up to date version of m_ref_usage.
+
+    This means that when m_is_destroyed is TRUE and we hold the
+    MDL_lock::m_mutex we can safely read the m_ref_usage
+    member.
   */
-  m_tickets.empty();
-  m_is_waiting_in_mdl= FALSE;
+  pthread_mutex_lock(&m_mutex);
+  my_hash_delete(&m_locks, (uchar*) lock);
+  lock->m_is_destroyed= TRUE;
+  ref_usage= lock->m_ref_usage;
+  ref_release= lock->m_ref_release;
+  pthread_mutex_unlock(&lock->m_mutex);
+  pthread_mutex_unlock(&m_mutex);
+  if (ref_usage == ref_release)
+    MDL_lock::destroy(lock);
+}
+
+
+/**
+  Initialize a metadata locking context.
+
+  This is to be called when a new server connection is created.
+*/
+
+MDL_context::MDL_context()
+  :m_lt_or_ha_sentinel(NULL),
+  m_thd(NULL)
+{
+  pthread_cond_init(&m_ctx_wakeup_cond, NULL);
 }
 
 
@@ -215,7 +453,7 @@ void MDL_context::init(THD *thd_arg)
 void MDL_context::destroy()
 {
   DBUG_ASSERT(m_tickets.is_empty());
-  DBUG_ASSERT(! m_has_global_shared_lock);
+  pthread_cond_destroy(&m_ctx_wakeup_cond);
 }
 
 
@@ -305,13 +543,21 @@ MDL_request::create(MDL_key::enum_mdl_namespace mdl_namespace, const char *db,
 /**
   Auxiliary functions needed for creation/destruction of MDL_lock objects.
 
+  @note Also chooses an MDL_lock descendant appropriate for object namespace.
+
   @todo This naive implementation should be replaced with one that saves
         on memory allocation by reusing released objects.
 */
 
 inline MDL_lock *MDL_lock::create(const MDL_key *mdl_key)
 {
-  return new MDL_lock(mdl_key);
+  switch (mdl_key->mdl_namespace())
+  {
+    case MDL_key::GLOBAL:
+      return new MDL_global_lock(mdl_key);
+    default:
+      return new MDL_object_lock(mdl_key);
+  }
 }
 
 
@@ -321,6 +567,8 @@ void MDL_lock::destroy(MDL_lock *lock)
 }
 
 
+
+
 /**
   Auxiliary functions needed for creation/destruction of MDL_ticket
   objects.
@@ -354,18 +602,21 @@ void MDL_ticket::destroy(MDL_ticket *ticket)
         will probably introduce too much overhead.
 */
 
-#define MDL_ENTER_COND(A, B) mdl_enter_cond(A, B, __func__, __FILE__, __LINE__)
+#define MDL_ENTER_COND(A, B, C, D) \
+        mdl_enter_cond(A, B, C, D, __func__, __FILE__, __LINE__)
 
 static inline const char *mdl_enter_cond(THD *thd,
                                          st_my_thread_var *mysys_var,
+                                         pthread_cond_t *cond,
+                                         pthread_mutex_t *mutex,
                                          const char *calling_func,
                                          const char *calling_file,
                                          const unsigned int calling_line)
 {
-  safe_mutex_assert_owner(&LOCK_mdl);
+  safe_mutex_assert_owner(mutex);
 
-  mysys_var->current_mutex= &LOCK_mdl;
-  mysys_var->current_cond= &COND_mdl;
+  mysys_var->current_mutex= mutex;
+  mysys_var->current_cond= cond;
 
   DEBUG_SYNC(thd, "mdl_enter_cond");
 
@@ -373,18 +624,20 @@ static inline const char *mdl_enter_cond(THD *thd,
                            calling_func, calling_file, calling_line);
 }
 
-#define MDL_EXIT_COND(A, B, C) mdl_exit_cond(A, B, C, __func__, __FILE__, __LINE__)
+#define MDL_EXIT_COND(A, B, C, D) \
+        mdl_exit_cond(A, B, C, D, __func__, __FILE__, __LINE__)
 
 static inline void mdl_exit_cond(THD *thd,
                                  st_my_thread_var *mysys_var,
+                                 pthread_mutex_t *mutex,
                                  const char* old_msg,
                                  const char *calling_func,
                                  const char *calling_file,
                                  const unsigned int calling_line)
 {
-  DBUG_ASSERT(&LOCK_mdl == mysys_var->current_mutex);
+  DBUG_ASSERT(mutex == mysys_var->current_mutex);
 
-  pthread_mutex_unlock(&LOCK_mdl);
+  pthread_mutex_unlock(mutex);
   pthread_mutex_lock(&mysys_var->mutex);
   mysys_var->current_mutex= 0;
   mysys_var->current_cond= 0;
@@ -398,15 +651,14 @@ static inline void mdl_exit_cond(THD *thd,
 
 
 /**
-  Check if request for the lock on particular object can be satisfied given
-  current state of the global metadata lock.
+  Check if request for the global metadata lock can be satisfied given
+  its current state,
 
-  @note In other words, we're trying to check that the individual lock
-        request, implying a form of lock on the global metadata, is
-        compatible with the current state of the global metadata lock.
-
-  @param mdl_request  Request for lock on an individual object, implying a
-                      certain kind of global metadata lock.
+  @param  requestor_ctx  The context that identifies the owner of the request.
+  @param  type_arg       The requested type of global lock. Usually derived
+                         from the type of lock on individual object to be
+                         requested. See table below.
+  @param  is_upgrade     TRUE if we are performing lock upgrade (not unused).
 
   @retval TRUE  - Lock request can be satisfied
   @retval FALSE - There is some conflicting lock
@@ -426,7 +678,7 @@ static inline void mdl_exit_cond(THD *thd,
 
   Here: "+" -- means that request can be satisfied
         "-" -- means that request can't be satisfied and should wait
-        "0" -- means impossible situation which will trigger assert
+        "0" -- means impossible situation.
 
   (*)  Since for upgradable shared locks we always take intention exclusive
        global lock at the same time when obtaining the shared lock, there
@@ -436,62 +688,98 @@ static inline void mdl_exit_cond(THD *thd,
 */
 
 bool
-MDL_global_lock::is_lock_type_compatible(enum_mdl_type type,
-                                         bool is_upgrade) const
+MDL_global_lock::can_grant_lock(const MDL_context *requestor_ctx,
+                                enum_mdl_type type_arg,
+                                bool is_upgrade)
 {
-  switch (type)
+  switch (type_arg)
   {
   case MDL_SHARED:
-  case MDL_SHARED_HIGH_PRIO:
-    return TRUE;
-    break;
-  case MDL_SHARED_UPGRADABLE:
-    if (active_shared || waiting_shared)
+    if (! granted.is_empty() && granted.front()->m_type == MDL_INTENTION_EXCLUSIVE)
     {
       /*
-        We are going to obtain intention exclusive global lock and
-        there is active or pending shared global lock. Have to wait.
+        We are going to obtain global shared lock and there is active
+        intention exclusive lock. Have to wait.
       */
       return FALSE;
     }
-    else
-      return TRUE;
+    return TRUE;
     break;
-  case MDL_EXCLUSIVE:
-    if (is_upgrade)
+  case MDL_INTENTION_EXCLUSIVE:
+    if ((! granted.is_empty() && granted.front()->m_type == MDL_SHARED) ||
+        ! waiting_shared.is_empty())
     {
       /*
-        We are upgrading MDL_SHARED to MDL_EXCLUSIVE.
-
-        There should be no conflicting global locks since for each upgradable
-        shared lock we obtain intention exclusive global lock first.
+        We are going to obtain intention exclusive global lock and
+        there is active or pending shared global lock. Have to wait.
       */
-      DBUG_ASSERT(active_shared == 0 && active_intention_exclusive);
-      return TRUE;
+      return FALSE;
     }
     else
-    {
-      if (active_shared || waiting_shared)
-      {
-        /*
-          We are going to obtain intention exclusive global lock and
-          there is active or pending shared global lock.
-        */
-        return FALSE;
-      }
-      else
-        return TRUE;
-    }
+      return TRUE;
     break;
   default:
     DBUG_ASSERT(0);
+    break;
   }
   return FALSE;
 }
 
 
 /**
-  Check if request for the lock can be satisfied given current state of lock.
+  Wake up contexts which are waiting to acquire the global
+  metadata lock and which may succeed now, when we released it, or
+  removed a blocking request for it from the waiters list.
+  The latter can happen when the context trying to acquire the
+  global shared lock is killed.
+*/
+
+void MDL_global_lock::wake_up_waiters()
+{
+  /*
+    If there are no active locks or they are of INTENTION
+    EXCLUSIVE type and there are no pending requests for global
+    SHARED lock, wake up contexts waiting for an INTENTION
+    EXCLUSIVE lock.
+    This happens when we release the global SHARED lock or abort
+    or remove a pending request for it, i.e. abort the
+    context waiting for it.
+  */
+  if ((granted.is_empty() ||
+       granted.front()->m_type == MDL_INTENTION_EXCLUSIVE) &&
+      waiting_shared.is_empty() && ! waiting_exclusive.is_empty())
+  {
+    MDL_lock::Ticket_iterator it(waiting_exclusive);
+    MDL_ticket *awake_ticket;
+    while ((awake_ticket= it++))
+      awake_ticket->get_ctx()->awake();
+  }
+
+  /*
+    If there are no active locks, wake up contexts waiting for
+    the global shared lock (happens when an INTENTION EXCLUSIVE
+    lock is released).
+
+    We don't wake up contexts waiting for the global shared lock
+    if there is an active global shared lock since such situation
+    is transient and in it contexts marked as waiting for global
+    shared lock must be already woken up and simply have not
+    managed to update lock object yet.
+  */
+  if (granted.is_empty() &&
+      ! waiting_shared.is_empty())
+  {
+    MDL_lock::Ticket_iterator it(waiting_shared);
+    MDL_ticket *awake_ticket;
+    while ((awake_ticket= it++))
+      awake_ticket->get_ctx()->awake();
+  }
+}
+
+
+/**
+  Check if request for the per-object lock can be satisfied given current
+  state of the lock.
 
   @param  requestor_ctx  The context that identifies the owner of the request.
   @param  type_arg       The requested lock type.
@@ -523,8 +811,9 @@ MDL_global_lock::is_lock_type_compatible(enum_mdl_type type,
 */
 
 bool
-MDL_lock::can_grant_lock(const MDL_context *requestor_ctx, enum_mdl_type type_arg,
-                         bool is_upgrade)
+MDL_object_lock::can_grant_lock(const MDL_context *requestor_ctx,
+                                enum_mdl_type type_arg,
+                                bool is_upgrade)
 {
   bool can_grant= FALSE;
 
@@ -532,10 +821,10 @@ MDL_lock::can_grant_lock(const MDL_context *requestor_ctx, enum_mdl_type type_ar
   case MDL_SHARED:
   case MDL_SHARED_UPGRADABLE:
   case MDL_SHARED_HIGH_PRIO:
-    if (type == MDL_lock::MDL_LOCK_SHARED)
+    if (granted.is_empty() || granted.front()->is_shared())
     {
       /* Pending exclusive locks have higher priority over shared locks. */
-      if (waiting.is_empty() || type_arg == MDL_SHARED_HIGH_PRIO)
+      if (waiting_exclusive.is_empty() || type_arg == MDL_SHARED_HIGH_PRIO)
         can_grant= TRUE;
     }
     else if (granted.front()->get_ctx() == requestor_ctx)
@@ -559,7 +848,7 @@ MDL_lock::can_grant_lock(const MDL_context *requestor_ctx, enum_mdl_type type_ar
         There should be no active exclusive locks since we own shared lock
         on the object.
       */
-      DBUG_ASSERT(type == MDL_lock::MDL_LOCK_SHARED);
+      DBUG_ASSERT(granted.front()->is_shared());
 
       while ((conflicting_ticket= it++))
       {
@@ -576,9 +865,13 @@ MDL_lock::can_grant_lock(const MDL_context *requestor_ctx, enum_mdl_type type_ar
         can_grant= TRUE;
       break;
     }
-    else if (type == MDL_lock::MDL_LOCK_SHARED)
+    else if (granted.is_empty())
     {
-      can_grant= granted.is_empty();
+      /*
+        We are trying to acquire fresh MDL_EXCLUSIVE and there are no active
+        shared or exclusive locks.
+      */
+      can_grant= TRUE;
     }
     break;
   default:
@@ -589,6 +882,44 @@ MDL_lock::can_grant_lock(const MDL_context *requestor_ctx, enum_mdl_type type_ar
 
 
 /**
+  Wake up contexts which are waiting to acquire lock on individual object
+  and which may succeed now, when we released some lock on it or removed
+  some pending request from its waiters list (the latter can happen, for
+  example, when context trying to acquire exclusive lock is killed).
+*/
+
+void MDL_object_lock::wake_up_waiters()
+{
+  /*
+    There are no active locks or they are of shared type.
+    We have to wake up contexts waiting for shared lock even if there is
+    a pending exclusive lock as some them might be trying to acquire high
+    priority shared lock.
+  */
+  if ((granted.is_empty() || granted.front()->is_shared()) &&
+      ! waiting_shared.is_empty())
+  {
+    MDL_lock::Ticket_iterator it(waiting_shared);
+    MDL_ticket *waiting_ticket;
+    while ((waiting_ticket= it++))
+      waiting_ticket->get_ctx()->awake();
+  }
+
+  /*
+    There are no active locks (shared or exclusive).
+    Wake up contexts waiting to acquire exclusive locks.
+  */
+  if (granted.is_empty() && ! waiting_exclusive.is_empty())
+  {
+    MDL_lock::Ticket_iterator it(waiting_exclusive);
+    MDL_ticket *waiting_ticket;
+    while ((waiting_ticket= it++))
+      waiting_ticket->get_ctx()->awake();
+  }
+}
+
+
+/**
   Check whether the context already holds a compatible lock ticket
   on an object.
   Start searching the transactional locks. If not
@@ -626,47 +957,207 @@ MDL_context::find_ticket(MDL_request *mdl_request,
 
 
 /**
-  Try to acquire one shared lock.
+  Try to acquire global intention exclusive lock.
 
-  Unlike exclusive locks, shared locks are acquired one by
-  one. This is interface is chosen to simplify introduction of
-  the new locking API to the system. MDL_context::try_acquire_shared_lock()
-  is currently used from open_table(), and there we have only one
-  table to work with.
+  @param[in/out]  mdl_request  Lock request object for lock to be acquired
 
-  In future we may consider allocating multiple shared locks at once.
+  @retval  FALSE   Success. The lock may have not been acquired.
+                   One needs to check value of 'MDL_request::ticket'
+                   to find out what has happened.
+  @retval  TRUE    Error.
+*/
+
+bool
+MDL_context::
+try_acquire_global_intention_exclusive_lock(MDL_request *mdl_request)
+{
+  DBUG_ASSERT(mdl_request->key.mdl_namespace() == MDL_key::GLOBAL &&
+              mdl_request->type == MDL_INTENTION_EXCLUSIVE);
+
+  if (is_global_lock_owner(MDL_SHARED))
+  {
+    my_error(ER_CANT_UPDATE_WITH_READLOCK, MYF(0));
+    return TRUE;
+  }
+
+  return try_acquire_lock_impl(mdl_request);
+}
+
+
+/**
+  Acquire one lock with waiting for conflicting locks to go away if needed.
+
+  @note This is an internal method which should not be used outside of MDL
+        subsystem as in most cases simply waiting for conflicting locks to
+        go away will lead to deadlock.
+
+  @param mdl_request [in/out] Lock request object for lock to be acquired
+
+  @retval  FALSE   Success. MDL_request::ticket points to the ticket
+                   for the lock.
+  @retval  TRUE    Failure (Out of resources or waiting is aborted),
+*/
+
+bool
+MDL_context::acquire_lock_impl(MDL_request *mdl_request)
+{
+  bool not_used;
+  MDL_ticket *ticket;
+  MDL_key *key= &mdl_request->key;
+  MDL_lock *lock;
+  const char *old_msg;
+  st_my_thread_var *mysys_var= my_thread_var;
+
+  DBUG_ASSERT(mdl_request->ticket == NULL);
+  safe_mutex_assert_not_owner(&LOCK_open);
+
+  /*
+    Grant lock without waiting if this context already owns this type of lock
+    on this object.
+
+    The fact that we don't wait in such situation allows to avoid deadlocks
+    in cases when pending request for global shared lock pops up after the
+    moment when thread has acquired its first intention exclusive lock but
+    before it has requested the second instance of such lock.
+  */
+  if ((mdl_request->ticket= find_ticket(mdl_request, &not_used)))
+    return FALSE;
+
+  if (! (ticket= MDL_ticket::create(this, mdl_request->type)))
+    return TRUE;
+
+  /* The below call also implicitly locks MDL_lock::m_mutex. */
+  if (! (lock= mdl_locks.find_or_insert(key)))
+  {
+    MDL_ticket::destroy(ticket);
+    return TRUE;
+  }
+
+  old_msg= MDL_ENTER_COND(m_thd, mysys_var, &m_ctx_wakeup_cond,
+                          &lock->m_mutex);
+
+  if (! lock->can_grant_lock(this, mdl_request->type, FALSE))
+  {
+    if (mdl_request->is_shared())
+      lock->waiting_shared.push_front(ticket);
+    else
+      lock->waiting_exclusive.push_front(ticket);
+
+    do
+    {
+      pthread_cond_wait(&m_ctx_wakeup_cond, &lock->m_mutex);
+    }
+    while (! lock->can_grant_lock(this, mdl_request->type, FALSE) &&
+           ! mysys_var->abort);
+
+    if (mysys_var->abort)
+    {
+      /*
+        We have to do MDL_EXIT_COND here and then re-acquire the lock
+        as there is a chance that we will destroy MDL_lock object and
+        won't be able to call MDL_EXIT_COND after it.
+      */
+      MDL_EXIT_COND(m_thd, mysys_var, &lock->m_mutex, old_msg);
+
+      pthread_mutex_lock(&lock->m_mutex);
+      /* Get rid of pending ticket. */
+      if (mdl_request->is_shared())
+        lock->waiting_shared.remove(ticket);
+      else
+        lock->waiting_exclusive.remove(ticket);
+      if (lock->is_empty())
+        mdl_locks.remove(lock);
+      else
+      {
+        lock->wake_up_waiters();
+        pthread_mutex_unlock(&lock->m_mutex);
+      }
+      MDL_ticket::destroy(ticket);
+      return TRUE;
+    }
+
+    if (mdl_request->is_shared())
+      lock->waiting_shared.remove(ticket);
+    else
+      lock->waiting_exclusive.remove(ticket);
+  }
+
+  lock->granted.push_front(ticket);
+  MDL_EXIT_COND(m_thd, mysys_var, &lock->m_mutex, old_msg);
+
+  ticket->m_state= MDL_ACQUIRED;
+  ticket->m_lock= lock;
+
+  m_tickets.push_front(ticket);
+
+  mdl_request->ticket= ticket;
+
+  return FALSE;
+}
+
+
+/**
+  Acquire global intention exclusive lock.
+
+  @param[in]  mdl_request  Lock request object for lock to be acquired
+
+  @retval  FALSE   Success. The lock has been acquired.
+  @retval  TRUE    Error.
+*/
+
+bool
+MDL_context::acquire_global_intention_exclusive_lock(MDL_request *mdl_request)
+{
+  DBUG_ASSERT(mdl_request->key.mdl_namespace() == MDL_key::GLOBAL &&
+              mdl_request->type == MDL_INTENTION_EXCLUSIVE);
+
+  if (is_global_lock_owner(MDL_SHARED))
+  {
+    my_error(ER_CANT_UPDATE_WITH_READLOCK, MYF(0));
+    return TRUE;
+  }
+
+  /*
+    If this is a non-recursive attempt to acquire global intention
+    exclusive lock we might have to wait until active global shared
+    lock or pending requests will go away. Since we won't hold any
+    resources (except associated with open HANDLERs) while doing it
+    deadlocks are not possible,
+  */
+  DBUG_ASSERT(is_global_lock_owner(MDL_INTENTION_EXCLUSIVE) ||
+              ! has_locks() ||
+              (m_lt_or_ha_sentinel &&
+               m_tickets.front() == m_lt_or_ha_sentinel));
+
+  return acquire_lock_impl(mdl_request);
+}
+
+
+/**
+  Try to acquire one lock.
 
   @param mdl_request [in/out] Lock request object for lock to be acquired
 
   @retval  FALSE   Success. The lock may have not been acquired.
                    Check the ticket, if it's NULL, a conflicting lock
-                   exists and another attempt should be made after releasing
-                   all current locks and waiting for conflicting lock go
-                   away (using MDL_context::wait_for_locks()).
+                   exists.
   @retval  TRUE    Out of resources, an error has been reported.
 */
 
 bool
-MDL_context::try_acquire_shared_lock(MDL_request *mdl_request)
+MDL_context::try_acquire_lock_impl(MDL_request *mdl_request)
 {
   MDL_lock *lock;
   MDL_key *key= &mdl_request->key;
   MDL_ticket *ticket;
   bool is_lt_or_ha;
 
-  DBUG_ASSERT(mdl_request->is_shared() && mdl_request->ticket == NULL);
+  DBUG_ASSERT(mdl_request->ticket == NULL);
 
   /* Don't take chances in production. */
   mdl_request->ticket= NULL;
   safe_mutex_assert_not_owner(&LOCK_open);
 
-  if (m_has_global_shared_lock &&
-      mdl_request->type == MDL_SHARED_UPGRADABLE)
-  {
-    my_error(ER_CANT_UPDATE_WITH_READLOCK, MYF(0));
-    return TRUE;
-  }
-
   /*
     Check whether the context already holds a shared lock on the object,
     and if so, grant the request.
@@ -674,8 +1165,7 @@ MDL_context::try_acquire_shared_lock(MDL_request *mdl_request)
   if ((ticket= find_ticket(mdl_request, &is_lt_or_ha)))
   {
     DBUG_ASSERT(ticket->m_state == MDL_ACQUIRED);
-    /* Only shared locks can be recursive. */
-    DBUG_ASSERT(ticket->is_shared());
+    DBUG_ASSERT(ticket->m_type == mdl_request->type);
     /*
       If the request is for a transactional lock, and we found
       a transactional lock, just reuse the found ticket.
@@ -703,57 +1193,73 @@ MDL_context::try_acquire_shared_lock(MDL_request *mdl_request)
     return FALSE;
   }
 
-  pthread_mutex_lock(&LOCK_mdl);
-
-  if (!global_lock.is_lock_type_compatible(mdl_request->type, FALSE))
-  {
-    pthread_mutex_unlock(&LOCK_mdl);
-    return FALSE;
-  }
-
   if (!(ticket= MDL_ticket::create(this, mdl_request->type)))
-  {
-    pthread_mutex_unlock(&LOCK_mdl);
     return TRUE;
-  }
 
-  if (!(lock= (MDL_lock*) my_hash_search(&mdl_locks,
-                                         key->ptr(), key->length())))
+  /* The below call also implicitly locks MDL_lock::m_mutex. */
+  if (!(lock= mdl_locks.find_or_insert(key)))
   {
-    /* Default lock type is MDL_lock::MDL_LOCK_SHARED */
-    lock= MDL_lock::create(key);
-    if (!lock || my_hash_insert(&mdl_locks, (uchar*)lock))
-    {
-      MDL_lock::destroy(lock);
-      MDL_ticket::destroy(ticket);
-      pthread_mutex_unlock(&LOCK_mdl);
-      return TRUE;
-    }
+    MDL_ticket::destroy(ticket);
+    return TRUE;
   }
 
   if (lock->can_grant_lock(this, mdl_request->type, FALSE))
   {
-    mdl_request->ticket= ticket;
     lock->granted.push_front(ticket);
-    m_tickets.push_front(ticket);
+    pthread_mutex_unlock(&lock->m_mutex);
+
     ticket->m_state= MDL_ACQUIRED;
     ticket->m_lock= lock;
-    if (mdl_request->type == MDL_SHARED_UPGRADABLE)
-      global_lock.active_intention_exclusive++;
+
+    m_tickets.push_front(ticket);
+
+    mdl_request->ticket= ticket;
   }
   else
   {
     /* We can't get here if we allocated a new lock. */
     DBUG_ASSERT(! lock->is_empty());
+    pthread_mutex_unlock(&lock->m_mutex);
     MDL_ticket::destroy(ticket);
   }
-  pthread_mutex_unlock(&LOCK_mdl);
 
   return FALSE;
 }
 
 
 /**
+  Try to acquire one shared lock.
+
+  Unlike exclusive locks, shared locks are acquired one by
+  one. This is interface is chosen to simplify introduction of
+  the new locking API to the system. MDL_context::try_acquire_shared_lock()
+  is currently used from open_table(), and there we have only one
+  table to work with.
+
+  In future we may consider allocating multiple shared locks at once.
+
+  @param mdl_request [in/out] Lock request object for lock to be acquired
+
+  @retval  FALSE   Success. The lock may have not been acquired.
+                   Check the ticket, if it's NULL, a conflicting lock
+                   exists and another attempt should be made after releasing
+                   all current locks and waiting for conflicting lock go
+                   away (using MDL_context::wait_for_locks()).
+  @retval  TRUE    Out of resources, an error has been reported.
+*/
+
+bool
+MDL_context::try_acquire_shared_lock(MDL_request *mdl_request)
+{
+  DBUG_ASSERT(mdl_request->is_shared());
+  DBUG_ASSERT(mdl_request->type != MDL_SHARED_UPGRADABLE ||
+              is_global_lock_owner(MDL_INTENTION_EXCLUSIVE));
+
+  return try_acquire_lock_impl(mdl_request);
+}
+
+
+/**
   Create a copy of a granted ticket. 
   This is used to make sure that HANDLER ticket
   is never shared with a ticket that belongs to
@@ -782,11 +1288,9 @@ MDL_context::clone_ticket(MDL_request *mdl_request)
   ticket->m_lock= mdl_request->ticket->m_lock;
   mdl_request->ticket= ticket;
 
-  pthread_mutex_lock(&LOCK_mdl);
+  pthread_mutex_lock(&ticket->m_lock->m_mutex);
   ticket->m_lock->granted.push_front(ticket);
-  if (mdl_request->type == MDL_SHARED_UPGRADABLE)
-    global_lock.active_intention_exclusive++;
-  pthread_mutex_unlock(&LOCK_mdl);
+  pthread_mutex_unlock(&ticket->m_lock->m_mutex);
 
   m_tickets.push_front(ticket);
 
@@ -799,225 +1303,291 @@ MDL_context::clone_ticket(MDL_request *mdl_request)
 
   @param thd               Current thread context
   @param conflicting_ticket  Conflicting metadata lock
-
-  @retval TRUE   A thread was woken up
-  @retval FALSE  Lock is not a shared one or no thread was woken up
 */
 
-bool notify_shared_lock(THD *thd, MDL_ticket *conflicting_ticket)
+void notify_shared_lock(THD *thd, MDL_ticket *conflicting_ticket)
 {
-  bool woke= FALSE;
   if (conflicting_ticket->is_shared())
   {
     THD *conflicting_thd= conflicting_ticket->get_ctx()->get_thd();
     DBUG_ASSERT(thd != conflicting_thd); /* Self-deadlock */
 
     /*
-      If the thread that holds the conflicting lock is waiting
-      on an MDL lock, wake it up by broadcasting on COND_mdl.
-      Otherwise it must be waiting on a table-level lock
-      or some other non-MDL resource, so delegate its waking up
-      to an external call.
+      If the thread that holds the conflicting lock is waiting in MDL
+      subsystem it has to be woken up by calling MDL_context::awake().
     */
-    if (conflicting_ticket->get_ctx()->is_waiting_in_mdl())
-    {
-      pthread_cond_broadcast(&COND_mdl);
-      woke= TRUE;
-    }
-    else
-      woke= mysql_notify_thread_having_shared_lock(thd, conflicting_thd);
+    conflicting_ticket->get_ctx()->awake();
+    /*
+      If it is waiting on table-level lock or some other non-MDL resource
+      we delegate its waking up to code outside of MDL.
+    */
+    mysql_notify_thread_having_shared_lock(thd, conflicting_thd);
   }
-  return woke;
 }
 
 
 /**
-  Acquire a single exclusive lock. A convenience
-  wrapper around the method acquiring a list of locks.
-*/
+  Auxiliary method for acquiring an exclusive lock.
 
-bool MDL_context::acquire_exclusive_lock(MDL_request *mdl_request)
-{
-  MDL_request_list mdl_requests;
-  mdl_requests.push_front(mdl_request);
-  return acquire_exclusive_locks(&mdl_requests);
-}
+  @param mdl_request  Request for the lock to be acqured.
 
-
-/**
-  Acquire exclusive locks. The context must contain the list of
-  locks to be acquired. There must be no granted locks in the
-  context.
-
-  This is a replacement of lock_table_names(). It is used in
-  RENAME, DROP and other DDL SQL statements.
-
-  @note The MDL context may not have non-exclusive lock requests
-        or acquired locks.
+  @note Should not be used outside of MDL subsystem. Instead one should
+        call acquire_exclusive_lock() or acquire_exclusive_locks() methods
+        which ensure that conditions for deadlock-free lock acquisition are
+        fulfilled.
 
   @retval FALSE  Success
   @retval TRUE   Failure
 */
 
-bool MDL_context::acquire_exclusive_locks(MDL_request_list *mdl_requests)
+bool MDL_context::acquire_exclusive_lock_impl(MDL_request *mdl_request)
 {
   MDL_lock *lock;
-  bool signalled= FALSE;
   const char *old_msg;
-  MDL_request *mdl_request;
   MDL_ticket *ticket;
+  bool not_used;
   st_my_thread_var *mysys_var= my_thread_var;
-  MDL_request_list::Iterator it(*mdl_requests);
+  MDL_key *key= &mdl_request->key;
+
+  DBUG_ASSERT(mdl_request->type == MDL_EXCLUSIVE &&
+              mdl_request->ticket == NULL);
 
   safe_mutex_assert_not_owner(&LOCK_open);
-  /* Exclusive locks must always be acquired first, all at once. */
-  DBUG_ASSERT(! has_locks() ||
-              (m_lt_or_ha_sentinel &&
-               m_tickets.front() == m_lt_or_ha_sentinel));
 
-  if (m_has_global_shared_lock)
+  /* Don't take chances in production. */
+  mdl_request->ticket= NULL;
+
+  /*
+    Check whether the context already holds an exclusive lock on the object,
+    and if so, grant the request.
+  */
+  if ((ticket= find_ticket(mdl_request, &not_used)))
   {
-    my_error(ER_CANT_UPDATE_WITH_READLOCK, MYF(0));
-    return TRUE;
+    DBUG_ASSERT(ticket->m_state == MDL_ACQUIRED);
+    DBUG_ASSERT(ticket->m_type == MDL_EXCLUSIVE);
+    mdl_request->ticket= ticket;
+    return FALSE;
   }
 
-  pthread_mutex_lock(&LOCK_mdl);
+  DBUG_ASSERT(is_global_lock_owner(MDL_INTENTION_EXCLUSIVE));
 
-  old_msg= MDL_ENTER_COND(m_thd, mysys_var);
+  /* Early allocation: ticket will be needed in any case. */
+  if (!(ticket= MDL_ticket::create(this, mdl_request->type)))
+    return TRUE;
 
-  while ((mdl_request= it++))
+  /* The below call also implicitly locks MDL_lock::m_mutex. */
+  if (!(lock= mdl_locks.find_or_insert(key)))
   {
-    MDL_key *key= &mdl_request->key;
-    DBUG_ASSERT(mdl_request->type == MDL_EXCLUSIVE &&
-                mdl_request->ticket == NULL);
-
-    /* Don't take chances in production. */
-    mdl_request->ticket= NULL;
-
-    /* Early allocation: ticket is used as a shortcut to the lock. */
-    if (!(ticket= MDL_ticket::create(this, mdl_request->type)))
-      goto err;
-
-    if (!(lock= (MDL_lock*) my_hash_search(&mdl_locks,
-                                           key->ptr(), key->length())))
-    {
-      lock= MDL_lock::create(key);
-      if (!lock || my_hash_insert(&mdl_locks, (uchar*)lock))
-      {
-        MDL_ticket::destroy(ticket);
-        MDL_lock::destroy(lock);
-        goto err;
-      }
-    }
-
-    mdl_request->ticket= ticket;
-    lock->waiting.push_front(ticket);
-    ticket->m_lock= lock;
+    MDL_ticket::destroy(ticket);
+    return TRUE;
   }
 
-  while (1)
-  {
-    it.rewind();
-    while ((mdl_request= it++))
-    {
-      lock= mdl_request->ticket->m_lock;
-
-      if (!global_lock.is_lock_type_compatible(mdl_request->type, FALSE))
-      {
-        /*
-          Someone owns or wants to acquire the global shared lock so
-          we have to wait until he goes away.
-        */
-        signalled= TRUE;
-        break;
-      }
-      else if (!lock->can_grant_lock(this, mdl_request->type, FALSE))
-      {
-        MDL_ticket *conflicting_ticket;
-        MDL_lock::Ticket_iterator it(lock->granted);
+  lock->waiting_exclusive.push_front(ticket);
 
-        signalled= (lock->type == MDL_lock::MDL_LOCK_EXCLUSIVE);
-
-        while ((conflicting_ticket= it++))
-          signalled|= notify_shared_lock(m_thd, conflicting_ticket);
-
-        break;
-      }
-    }
-    if (!mdl_request)
-      break;
+  old_msg= MDL_ENTER_COND(m_thd, mysys_var, &m_ctx_wakeup_cond,
+                          &lock->m_mutex);
 
+  while (!lock->can_grant_lock(this, mdl_request->type, FALSE))
+  {
     if (m_lt_or_ha_sentinel)
     {
       /*
         We're about to start waiting. Don't do it if we have
         HANDLER locks (we can't have any other locks here).
         Waiting with locks may lead to a deadlock.
+
+        We have to do MDL_EXIT_COND here and then re-acquire the
+        lock as there is a chance that we will destroy MDL_lock
+        object and won't be able to call MDL_EXIT_COND after it.
       */
+      MDL_EXIT_COND(m_thd, mysys_var, &lock->m_mutex, old_msg);
+
+      pthread_mutex_lock(&lock->m_mutex);
+      /* Get rid of pending ticket. */
+      lock->waiting_exclusive.remove(ticket);
+      if (lock->is_empty())
+        mdl_locks.remove(lock);
+      else
+      {
+        /*
+          There can be some contexts waiting to acquire shared
+          lock which now might be able to do it. Wake them up!
+        */
+        lock->wake_up_waiters();
+        pthread_mutex_unlock(&lock->m_mutex);
+      }
+      MDL_ticket::destroy(ticket);
       my_error(ER_LOCK_DEADLOCK, MYF(0));
-      goto err;
+      return TRUE;
     }
 
+    MDL_ticket *conflicting_ticket;
+    MDL_lock::Ticket_iterator it(lock->granted);
+
+    while ((conflicting_ticket= it++))
+      notify_shared_lock(m_thd, conflicting_ticket);
+
     /* There is a shared or exclusive lock on the object. */
     DEBUG_SYNC(m_thd, "mdl_acquire_exclusive_locks_wait");
 
-    if (signalled)
-      pthread_cond_wait(&COND_mdl, &LOCK_mdl);
-    else
+    /*
+      Another thread might have obtained a shared MDL lock on some table
+      but has not yet opened it and/or tried to obtain data lock on it.
+      Also invocation of acquire_exclusive_lock() method and consequently
+      first call to notify_shared_lock() might have happened right after
+      thread holding shared metadata lock in wait_for_locks() method
+      checked that there are no pending conflicting locks but before
+      it has started waiting.
+      In both these cases we need to sleep until these threads will start
+      waiting and try to abort them once again.
+
+      QQ: What is the optimal value for this sleep?
+    */
+    struct timespec abstime;
+    set_timespec(abstime, 1);
+    pthread_cond_timedwait(&m_ctx_wakeup_cond, &lock->m_mutex, &abstime);
+
+    if (mysys_var->abort)
     {
       /*
-        Another thread obtained a shared MDL lock on some table but
-        has not yet opened it and/or tried to obtain data lock on
-        it. In this case we need to wait until this happens and try
-        to abort this thread once again.
+        We have to do MDL_EXIT_COND here and then re-acquire the lock
+        as there is a chance that we will destroy MDL_lock object and
+        won't be able to call MDL_EXIT_COND after it.
       */
-      struct timespec abstime;
-      set_timespec(abstime, 1);
-      pthread_cond_timedwait(&COND_mdl, &LOCK_mdl, &abstime);
+      MDL_EXIT_COND(m_thd, mysys_var, &lock->m_mutex, old_msg);
+
+      pthread_mutex_lock(&lock->m_mutex);
+      /* Get rid of pending ticket. */
+      lock->waiting_exclusive.remove(ticket);
+      if (lock->is_empty())
+        mdl_locks.remove(lock);
+      else
+      {
+        /*
+          There can be some contexts waiting to acquire shared
+          lock which now might be able to do it. Wake them up!
+        */
+        lock->wake_up_waiters();
+        pthread_mutex_unlock(&lock->m_mutex);
+      }
+      MDL_ticket::destroy(ticket);
+      return TRUE;
     }
-    if (mysys_var->abort)
-      goto err;
   }
-  it.rewind();
-  while ((mdl_request= it++))
+
+  lock->waiting_exclusive.remove(ticket);
+  lock->granted.push_front(ticket);
+
+  if (lock->cached_object)
+    (*lock->cached_object_release_hook)(lock->cached_object);
+  lock->cached_object= NULL;
+
+  MDL_EXIT_COND(m_thd, mysys_var, &lock->m_mutex, old_msg);
+
+  ticket->m_state= MDL_ACQUIRED;
+  ticket->m_lock= lock;
+
+  m_tickets.push_front(ticket);
+
+  mdl_request->ticket= ticket;
+
+  return FALSE;
+}
+
+
+/**
+  Acquire an exclusive lock.
+
+  @param mdl_request  Request for the lock to be acqured.
+
+  @note Assumes that one already owns global intention exclusive lock.
+
+  @retval FALSE  Success
+  @retval TRUE   Failure
+*/
+
+bool MDL_context::acquire_exclusive_lock(MDL_request *mdl_request)
+{
+  /* Exclusive locks must always be acquired first, all at once. */
+  DBUG_ASSERT(! m_tickets.is_empty() &&
+              m_tickets.front()->m_lock->key.mdl_namespace() == MDL_key::GLOBAL &&
+              ++Ticket_list::Iterator(m_tickets) == m_lt_or_ha_sentinel);
+
+  return acquire_exclusive_lock_impl(mdl_request);
+}
+
+
+extern "C" int mdl_request_ptr_cmp(const void* ptr1, const void* ptr2)
+{
+  MDL_request *req1= *(MDL_request**)ptr1;
+  MDL_request *req2= *(MDL_request**)ptr2;
+  return req1->key.cmp(&req2->key);
+}
+
+
+/**
+  Acquire exclusive locks. There must be no granted locks in the
+  context.
+
+  This is a replacement of lock_table_names(). It is used in
+  RENAME, DROP and other DDL SQL statements.
+
+  @param  mdl_requests  List of requests for locks to be acquired.
+
+  @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.
+
+  @retval FALSE  Success
+  @retval TRUE   Failure
+*/
+
+bool MDL_context::acquire_exclusive_locks(MDL_request_list *mdl_requests)
+{
+  MDL_request_list::Iterator it(*mdl_requests);
+  MDL_request **sort_buf;
+  uint i;
+
+  /*
+    Exclusive locks must always be acquired first, all at once.
+  */
+  DBUG_ASSERT(! m_tickets.is_empty() &&
+              m_tickets.front()->m_lock->key.mdl_namespace() == MDL_key::GLOBAL &&
+              ++Ticket_list::Iterator(m_tickets) == m_lt_or_ha_sentinel);
+
+  if (mdl_requests->is_empty())
+    return FALSE;
+
+  /* Sort requests according to MDL_key. */
+  if (! (sort_buf= (MDL_request **)my_malloc(mdl_requests->elements() *
+                                             sizeof(MDL_request *),
+                                             MYF(MY_WME))))
+    return TRUE;
+
+  for (i= 0; i < mdl_requests->elements(); i++)
+    sort_buf[i]= it++;
+
+  my_qsort(sort_buf, mdl_requests->elements(), sizeof(MDL_request*),
+           mdl_request_ptr_cmp);
+
+  for (i= 0; i < mdl_requests->elements(); i++)
   {
-    global_lock.active_intention_exclusive++;
-    ticket= mdl_request->ticket;
-    lock= ticket->m_lock;
-    lock->type= MDL_lock::MDL_LOCK_EXCLUSIVE;
-    lock->waiting.remove(ticket);
-    lock->granted.push_front(ticket);
-    m_tickets.push_front(ticket);
-    ticket->m_state= MDL_ACQUIRED;
-    if (lock->cached_object)
-      (*lock->cached_object_release_hook)(lock->cached_object);
-    lock->cached_object= NULL;
+    if (acquire_exclusive_lock_impl(sort_buf[i]))
+      goto err;
   }
-  /* As a side-effect MDL_EXIT_COND() unlocks LOCK_mdl. */
-  MDL_EXIT_COND(m_thd, mysys_var, old_msg);
+  my_free(sort_buf, MYF(0));
   return FALSE;
 
 err:
-  /* Remove our pending tickets from the locks. */
-  it.rewind();
-  while ((mdl_request= it++) && mdl_request->ticket)
+  /* Release locks we have managed to acquire so far. */
+  for (i= 0; i < mdl_requests->elements() && sort_buf[i]->ticket; i++)
   {
-    ticket= mdl_request->ticket;
-    DBUG_ASSERT(ticket->m_state == MDL_PENDING);
-    lock= ticket->m_lock;
-    lock->waiting.remove(ticket);
-    MDL_ticket::destroy(ticket);
+    release_lock(sort_buf[i]->ticket);
     /* Reset lock request back to its initial state. */
-    mdl_request->ticket= NULL;
-    if (lock->is_empty())
-    {
-      my_hash_delete(&mdl_locks, (uchar *)lock);
-      MDL_lock::destroy(lock);
-    }
+    sort_buf[i]->ticket= NULL;
   }
-  /* May be some pending requests for shared locks can be satisfied now. */
-  pthread_cond_broadcast(&COND_mdl);
-  MDL_EXIT_COND(m_thd, mysys_var, old_msg);
+  my_free(sort_buf, MYF(0));
   return TRUE;
 }
 
@@ -1062,6 +1632,12 @@ MDL_ticket::upgrade_shared_lock_to_exclusive()
   DBUG_ASSERT(m_type == MDL_SHARED_UPGRADABLE);
 
   /*
+    Since we should have already acquired an intention exclusive
+    global lock this call is only enforcing asserts.
+  */
+  DBUG_ASSERT(m_ctx->is_global_lock_owner(MDL_INTENTION_EXCLUSIVE));
+
+  /*
     Create an auxiliary ticket to represent a pending exclusive
     lock and add it to the 'waiting' queue for the duration
     of upgrade. During upgrade we abort waits of connections
@@ -1072,24 +1648,21 @@ MDL_ticket::upgrade_shared_lock_to_exclusive()
   if (! (pending_ticket= MDL_ticket::create(m_ctx, MDL_EXCLUSIVE)))
     DBUG_RETURN(TRUE);
 
-  pthread_mutex_lock(&LOCK_mdl);
+  pthread_mutex_lock(&m_lock->m_mutex);
 
-  pending_ticket->m_lock= m_lock;
-  m_lock->waiting.push_front(pending_ticket);
+  m_lock->waiting_exclusive.push_front(pending_ticket);
 
-  old_msg= MDL_ENTER_COND(thd, mysys_var);
-
-  /*
-    Since we should have already acquired an intention exclusive
-    global lock this call is only enforcing asserts.
-  */
-  DBUG_ASSERT(global_lock.is_lock_type_compatible(MDL_EXCLUSIVE, TRUE));
+  old_msg= MDL_ENTER_COND(thd, mysys_var, &m_ctx->m_ctx_wakeup_cond,
+                          &m_lock->m_mutex);
 
   while (1)
   {
     if (m_lock->can_grant_lock(m_ctx, MDL_EXCLUSIVE, TRUE))
       break;
 
+    MDL_ticket *conflicting_ticket;
+    MDL_lock::Ticket_iterator it(m_lock->granted);
+
     /*
       If m_ctx->lt_or_ha_sentinel(), and this sentinel is for HANDLER,
       we can deadlock. However, HANDLER is not allowed under
@@ -1113,12 +1686,7 @@ MDL_ticket::upgrade_shared_lock_to_exclusive()
 
       (*) There is no requirement to upgrade lock in
       CREATE/DROP TRIGGER, it's used there just for convenience.
-    */
-    bool signalled= FALSE;
-    MDL_ticket *conflicting_ticket;
-    MDL_lock::Ticket_iterator it(m_lock->granted);
 
-    /*
       A temporary work-around to avoid deadlocks/livelocks in
       a situation when in one connection ALTER TABLE tries to
       upgrade its metadata lock and in another connection
@@ -1145,53 +1713,57 @@ MDL_ticket::upgrade_shared_lock_to_exclusive()
     while ((conflicting_ticket= it++))
     {
       if (conflicting_ticket->m_ctx != m_ctx)
-        signalled|= notify_shared_lock(thd, conflicting_ticket);
+        notify_shared_lock(thd, conflicting_ticket);
     }
 
     /* There is a shared or exclusive lock on the object. */
     DEBUG_SYNC(thd, "mdl_upgrade_shared_lock_to_exclusive_wait");
 
-    if (signalled)
-      pthread_cond_wait(&COND_mdl, &LOCK_mdl);
-    else
+    /*
+      Another thread might have obtained a shared MDL lock on some table
+      but has not yet opened it and/or tried to obtain data lock on it.
+      Also invocation of acquire_exclusive_lock() method and consequently
+      first call to notify_shared_lock() might have happened right after
+      thread holding shared metadata lock in wait_for_locks() method
+      checked that there are no pending conflicting locks but before
+      it has started waiting.
+      In both these cases we need to sleep until these threads will start
+      waiting and try to abort them once again.
+    */
+    struct timespec abstime;
+    set_timespec(abstime, 1);
+    pthread_cond_timedwait(&m_ctx->m_ctx_wakeup_cond, &m_lock->m_mutex,
+                           &abstime);
+
+    if (mysys_var->abort)
     {
+      m_lock->waiting_exclusive.remove(pending_ticket);
       /*
-        Another thread obtained a shared MDL lock on some table but
-        has not yet opened it and/or tried to obtain data lock on
-        it. In this case we need to wait until this happens and try
-        to abort this thread once again.
+        If there are no other pending requests for exclusive locks
+        we need to wake up threads waiting for a chance to acquire
+        shared lock.
       */
-      struct timespec abstime;
-      set_timespec(abstime, 1);
-      DBUG_PRINT("info", ("Failed to wake-up from table-level lock ... sleeping"));
-      pthread_cond_timedwait(&COND_mdl, &LOCK_mdl, &abstime);
-    }
-    if (mysys_var->abort)
-    {
-      /* Remove and destroy the auxiliary pending ticket. */
-      m_lock->waiting.remove(pending_ticket);
+      m_lock->wake_up_waiters();
+      MDL_EXIT_COND(thd, mysys_var, &m_lock->m_mutex, old_msg);
       MDL_ticket::destroy(pending_ticket);
-      /* Pending requests for shared locks can be satisfied now. */
-      pthread_cond_broadcast(&COND_mdl);
-      MDL_EXIT_COND(thd, mysys_var, old_msg);
       DBUG_RETURN(TRUE);
     }
   }
 
-  m_lock->type= MDL_lock::MDL_LOCK_EXCLUSIVE;
   /* Set the new type of lock in the ticket. */
   m_type= MDL_EXCLUSIVE;
 
   /* Remove and destroy the auxiliary pending ticket. */
-  m_lock->waiting.remove(pending_ticket);
-  MDL_ticket::destroy(pending_ticket);
+  m_lock->waiting_exclusive.remove(pending_ticket);
 
   if (m_lock->cached_object)
     (*m_lock->cached_object_release_hook)(m_lock->cached_object);
   m_lock->cached_object= 0;
 
-  /* As a side-effect MDL_EXIT_COND() unlocks LOCK_mdl. */
-  MDL_EXIT_COND(thd, mysys_var, old_msg);
+  MDL_EXIT_COND(thd, mysys_var, &m_lock->m_mutex, old_msg);
+
+  MDL_ticket::destroy(pending_ticket);
+
   DBUG_RETURN(FALSE);
 }
 
@@ -1222,41 +1794,10 @@ MDL_ticket::upgrade_shared_lock_to_exclusive()
 bool
 MDL_context::try_acquire_exclusive_lock(MDL_request *mdl_request)
 {
-  MDL_lock *lock;
-  MDL_ticket *ticket;
-  MDL_key *key= &mdl_request->key;
+  DBUG_ASSERT(mdl_request->type == MDL_EXCLUSIVE);
+  DBUG_ASSERT(is_global_lock_owner(MDL_INTENTION_EXCLUSIVE));
 
-  DBUG_ASSERT(mdl_request->type == MDL_EXCLUSIVE &&
-              mdl_request->ticket == NULL);
-
-  safe_mutex_assert_not_owner(&LOCK_open);
-
-  mdl_request->ticket= NULL;
-
-  pthread_mutex_lock(&LOCK_mdl);
-
-  if (!(lock= (MDL_lock*) my_hash_search(&mdl_locks,
-                                         key->ptr(), key->length())))
-  {
-    ticket= MDL_ticket::create(this, mdl_request->type);
-    lock= MDL_lock::create(key);
-    if (!ticket || !lock || my_hash_insert(&mdl_locks, (uchar*)lock))
-    {
-      MDL_ticket::destroy(ticket);
-      MDL_lock::destroy(lock);
-      pthread_mutex_unlock(&LOCK_mdl);
-      return TRUE;
-    }
-    mdl_request->ticket= ticket;
-    lock->type= MDL_lock::MDL_LOCK_EXCLUSIVE;
-    lock->granted.push_front(ticket);
-    m_tickets.push_front(ticket);
-    ticket->m_state= MDL_ACQUIRED;
-    ticket->m_lock= lock;
-    global_lock.active_intention_exclusive++;
-  }
-  pthread_mutex_unlock(&LOCK_mdl);
-  return FALSE;
+  return try_acquire_lock_impl(mdl_request);
 }
 
 
@@ -1272,46 +1813,32 @@ MDL_context::try_acquire_exclusive_lock(MDL_request *mdl_request)
 
 bool MDL_context::acquire_global_shared_lock()
 {
-  st_my_thread_var *mysys_var= my_thread_var;
-  const char *old_msg;
+  MDL_request mdl_request;
 
-  safe_mutex_assert_not_owner(&LOCK_open);
-  DBUG_ASSERT(!m_has_global_shared_lock);
+  DBUG_ASSERT(! is_global_lock_owner(MDL_SHARED));
 
-  pthread_mutex_lock(&LOCK_mdl);
+  mdl_request.init(MDL_key::GLOBAL, "", "", MDL_SHARED);
 
-  global_lock.waiting_shared++;
-  old_msg= MDL_ENTER_COND(m_thd, mysys_var);
+  if (acquire_lock_impl(&mdl_request))
+    return TRUE;
 
-  while (!mysys_var->abort && global_lock.active_intention_exclusive)
-    pthread_cond_wait(&COND_mdl, &LOCK_mdl);
+  move_ticket_after_lt_or_ha_sentinel(mdl_request.ticket);
 
-  global_lock.waiting_shared--;
-  if (mysys_var->abort)
-  {
-    /* As a side-effect MDL_EXIT_COND() unlocks LOCK_mdl. */
-    MDL_EXIT_COND(m_thd, mysys_var, old_msg);
-    return TRUE;
-  }
-  global_lock.active_shared++;
-  m_has_global_shared_lock= TRUE;
-  /* As a side-effect MDL_EXIT_COND() unlocks LOCK_mdl. */
-  MDL_EXIT_COND(m_thd, mysys_var, old_msg);
   return FALSE;
 }
 
 
 /**
-  Check if there are any pending exclusive locks which conflict
-  with shared locks held by this thread.
-
-  @pre The caller already has acquired LOCK_mdl.
+  Implement a simple deadlock detection heuristic: check if there
+  are any pending exclusive locks which conflict with shared locks
+  held by this thread. In that case waiting can be circular,
+  i.e. lead to a deadlock.
 
   @return TRUE   If there are any pending conflicting locks.
           FALSE  Otherwise.
 */
 
-bool MDL_context::can_wait_lead_to_deadlock_impl() const
+bool MDL_context::can_wait_lead_to_deadlock() const
 {
   Ticket_iterator ticket_it(m_tickets);
   MDL_ticket *ticket;
@@ -1323,12 +1850,12 @@ bool MDL_context::can_wait_lead_to_deadlock_impl() const
       upgradeable shared metadata locks.
       Otherwise we would also have to check for the presence of pending
       requests for conflicting types of global lock.
-      In addition MDL_ticket::has_pending_conflicting_lock_impl()
+      In addition MDL_ticket::has_pending_conflicting_lock()
       won't work properly for exclusive type of lock.
     */
     DBUG_ASSERT(! ticket->is_upgradable_or_exclusive());
 
-    if (ticket->has_pending_conflicting_lock_impl())
+    if (ticket->has_pending_conflicting_lock())
       return TRUE;
   }
   return FALSE;
@@ -1336,25 +1863,6 @@ bool MDL_context::can_wait_lead_to_deadlock_impl() const
 
 
 /**
-  Implement a simple deadlock detection heuristic: check if there
-  are any pending exclusive locks which conflict with shared locks
-  held by this thread. In that case waiting can be circular,
-  i.e. lead to a deadlock.
-
-  @return TRUE if there are any conflicting locks, FALSE otherwise.
-*/
-
-bool MDL_context::can_wait_lead_to_deadlock() const
-{
-  bool result;
-  pthread_mutex_lock(&LOCK_mdl);
-  result= can_wait_lead_to_deadlock_impl();
-  pthread_mutex_unlock(&LOCK_mdl);
-  return result;
-}
-
-
-/**
   Wait until there will be no locks that conflict with lock requests
   in the given list.
 
@@ -1391,8 +1899,6 @@ MDL_context::wait_for_locks(MDL_request_list *mdl_requests)
             COND_mdl because of above scenario.
     */
     mysql_ha_flush(m_thd);
-    pthread_mutex_lock(&LOCK_mdl);
-    old_msg= MDL_ENTER_COND(m_thd, mysys_var);
 
     /*
       In cases when we wait while still holding some metadata
@@ -1406,9 +1912,8 @@ MDL_context::wait_for_locks(MDL_request_list *mdl_requests)
       negatives) in situations when conflicts are rare (in our
       case this is true since DDL statements should be rare).
     */
-    if (can_wait_lead_to_deadlock_impl())
+    if (can_wait_lead_to_deadlock())
     {
-      MDL_EXIT_COND(m_thd, mysys_var, old_msg);
       my_error(ER_LOCK_DEADLOCK, MYF(0));
       return TRUE;
     }
@@ -1418,83 +1923,104 @@ MDL_context::wait_for_locks(MDL_request_list *mdl_requests)
     {
       MDL_key *key= &mdl_request->key;
       DBUG_ASSERT(mdl_request->ticket == NULL);
-      if (!global_lock.is_lock_type_compatible(mdl_request->type, FALSE))
-        break;
+
       /*
         To avoid starvation we don't wait if we have a conflict against
         request for MDL_EXCLUSIVE lock.
       */
-      if (mdl_request->is_shared() &&
-          (lock= (MDL_lock*) my_hash_search(&mdl_locks, key->ptr(),
-                                            key->length())) &&
-          !lock->can_grant_lock(this, mdl_request->type, FALSE))
+      if (mdl_request->is_shared() ||
+          mdl_request->type == MDL_INTENTION_EXCLUSIVE)
+      {
+        /* The below call also implicitly locks MDL_lock::m_mutex. */
+        if (! (lock= mdl_locks.find(key)))
+          continue;
+
+        if (lock->can_grant_lock(this, mdl_request->type, FALSE))
+        {
+          pthread_mutex_unlock(&lock->m_mutex);
+          continue;
+        }
+
+        MDL_ticket *pending_ticket;
+        if (! (pending_ticket= MDL_ticket::create(this, mdl_request->type)))
+        {
+          pthread_mutex_unlock(&lock->m_mutex);
+          return TRUE;
+        }
+        if (mdl_request->is_shared())
+          lock->waiting_shared.push_front(pending_ticket);
+        else
+          lock->waiting_exclusive.push_front(pending_ticket);
+
+        old_msg= MDL_ENTER_COND(m_thd, mysys_var, &m_ctx_wakeup_cond,
+                                &lock->m_mutex);
+
+        pthread_cond_wait(&m_ctx_wakeup_cond, &lock->m_mutex);
+
+        /*
+          We have to do MDL_EXIT_COND here and then re-acquire the lock
+          as there is a chance that we will destroy MDL_lock object and
+          won't be able to call MDL_EXIT_COND after it.
+        */
+        MDL_EXIT_COND(m_thd, mysys_var, &lock->m_mutex, old_msg);
+
+        pthread_mutex_lock(&lock->m_mutex);
+        if (mdl_request->is_shared())
+          lock->waiting_shared.remove(pending_ticket);
+        else
+          lock->waiting_exclusive.remove(pending_ticket);
+        if (lock->is_empty())
+          mdl_locks.remove(lock);
+        else
+          pthread_mutex_unlock(&lock->m_mutex);
+        MDL_ticket::destroy(pending_ticket);
         break;
+      }
     }
     if (!mdl_request)
     {
-      /* As a side-effect MDL_EXIT_COND() unlocks LOCK_mdl. */
-      MDL_EXIT_COND(m_thd, mysys_var, old_msg);
+      /* There are no conflicts for any locks! */
       break;
     }
-    m_is_waiting_in_mdl= TRUE;
-    pthread_cond_wait(&COND_mdl, &LOCK_mdl);
-    m_is_waiting_in_mdl= FALSE;
-    /* As a side-effect MDL_EXIT_COND() unlocks LOCK_mdl. */
-    MDL_EXIT_COND(m_thd, mysys_var, old_msg);
   }
   return mysys_var->abort;
 }
 
 
 /**
-  Auxiliary function which allows to release particular lock
-  ownership of which is represented by a lock ticket object.
+  Release lock.
+
+  @param ticket Ticket for lock to be released.
 */
 
-void MDL_context::release_ticket(MDL_ticket *ticket)
+void MDL_context::release_lock(MDL_ticket *ticket)
 {
   MDL_lock *lock= ticket->m_lock;
-  DBUG_ENTER("release_ticket");
+  DBUG_ENTER("MDL_context::release_lock");
   DBUG_PRINT("enter", ("db=%s name=%s", lock->key.db_name(),
                                         lock->key.name()));
 
-  safe_mutex_assert_owner(&LOCK_mdl);
+  DBUG_ASSERT(this == ticket->m_ctx);
+  safe_mutex_assert_not_owner(&LOCK_open);
 
   if (ticket == m_lt_or_ha_sentinel)
     m_lt_or_ha_sentinel= ++Ticket_list::Iterator(m_tickets, ticket);
 
-  m_tickets.remove(ticket);
+  pthread_mutex_lock(&lock->m_mutex);
 
-  switch (ticket->m_type)
-  {
-    case MDL_SHARED_UPGRADABLE:
-      global_lock.active_intention_exclusive--;
-      /* Fallthrough. */
-    case MDL_SHARED:
-    case MDL_SHARED_HIGH_PRIO:
-      lock->granted.remove(ticket);
-      break;
-    case MDL_EXCLUSIVE:
-      lock->type= MDL_lock::MDL_LOCK_SHARED;
-      lock->granted.remove(ticket);
-      global_lock.active_intention_exclusive--;
-      break;
-    default:
-      DBUG_ASSERT(0);
-  }
-
-  MDL_ticket::destroy(ticket);
+  lock->granted.remove(ticket);
 
   if (lock->is_empty())
+    mdl_locks.remove(lock);
+  else
   {
-    my_hash_delete(&mdl_locks, (uchar *)lock);
-    DBUG_PRINT("info", ("releasing cached_object cached_object=%p",
-                        lock->cached_object));
-    if (lock->cached_object)
-      (*lock->cached_object_release_hook)(lock->cached_object);
-    MDL_lock::destroy(lock);
+    lock->wake_up_waiters();
+    pthread_mutex_unlock(&lock->m_mutex);
   }
 
+  m_tickets.remove(ticket);
+  MDL_ticket::destroy(ticket);
+
   DBUG_VOID_RETURN;
 }
 
@@ -1522,44 +2048,26 @@ void MDL_context::release_locks_stored_before(MDL_ticket *sentinel)
   Ticket_iterator it(m_tickets);
   DBUG_ENTER("MDL_context::release_locks_stored_before");
 
-  safe_mutex_assert_not_owner(&LOCK_open);
-
   if (m_tickets.is_empty())
     DBUG_VOID_RETURN;
 
-  pthread_mutex_lock(&LOCK_mdl);
   while ((ticket= it++) && ticket != sentinel)
   {
     DBUG_PRINT("info", ("found lock to release ticket=%p", ticket));
-    release_ticket(ticket);
+    release_lock(ticket);
   }
-  /* Inefficient but will do for a while */
-  pthread_cond_broadcast(&COND_mdl);
-  pthread_mutex_unlock(&LOCK_mdl);
+  /*
+    If all locks were released, then the sentinel was not present
+    in the list. It must never happen because the sentinel was
+    bogus, i.e. pointed to a ticket that no longer exists.
+  */
+  DBUG_ASSERT(! m_tickets.is_empty() || sentinel == NULL);
 
   DBUG_VOID_RETURN;
 }
 
 
 /**
-  Release a lock.
-
-  @param ticket    Lock to be released
-*/
-
-void MDL_context::release_lock(MDL_ticket *ticket)
-{
-  DBUG_ASSERT(this == ticket->m_ctx);
-  safe_mutex_assert_not_owner(&LOCK_open);
-
-  pthread_mutex_lock(&LOCK_mdl);
-  release_ticket(ticket);
-  pthread_cond_broadcast(&COND_mdl);
-  pthread_mutex_unlock(&LOCK_mdl);
-}
-
-
-/**
   Release all locks in the context which correspond to the same name/
   object as this lock request.
 
@@ -1569,7 +2077,7 @@ void MDL_context::release_lock(MDL_ticket *ticket)
 
 void MDL_context::release_all_locks_for_name(MDL_ticket *name)
 {
-  /* Use MDL_ticket::lock to identify other locks for the same object. */
+  /* Use MDL_ticket::m_lock to identify other locks for the same object. */
   MDL_lock *lock= name->m_lock;
 
   /* Remove matching lock tickets from the context. */
@@ -1600,11 +2108,18 @@ void MDL_ticket::downgrade_exclusive_lock()
   if (is_shared())
     return;
 
-  pthread_mutex_lock(&LOCK_mdl);
-  m_lock->type= MDL_lock::MDL_LOCK_SHARED;
+  pthread_mutex_lock(&m_lock->m_mutex);
   m_type= MDL_SHARED_UPGRADABLE;
-  pthread_cond_broadcast(&COND_mdl);
-  pthread_mutex_unlock(&LOCK_mdl);
+
+  if (! m_lock->waiting_shared.is_empty())
+  {
+    MDL_lock::Ticket_iterator it(m_lock->waiting_shared);
+    MDL_ticket *ticket;
+    while ((ticket= it++))
+      ticket->get_ctx()->awake();
+  }
+
+  pthread_mutex_unlock(&m_lock->m_mutex);
 }
 
 
@@ -1614,14 +2129,22 @@ void MDL_ticket::downgrade_exclusive_lock()
 
 void MDL_context::release_global_shared_lock()
 {
+  MDL_request mdl_request;
+  MDL_ticket *ticket;
+  bool not_used;
+
+  mdl_request.init(MDL_key::GLOBAL, "", "", MDL_SHARED);
+
   safe_mutex_assert_not_owner(&LOCK_open);
-  DBUG_ASSERT(m_has_global_shared_lock);
 
-  pthread_mutex_lock(&LOCK_mdl);
-  global_lock.active_shared--;
-  m_has_global_shared_lock= FALSE;
-  pthread_cond_broadcast(&COND_mdl);
-  pthread_mutex_unlock(&LOCK_mdl);
+  /*
+    TODO/QQ/FIXME: In theory we always should be able to find
+                   ticket here. But in practice this is not
+                   always TRUE.
+  */
+
+  if ((ticket= find_ticket(&mdl_request, &not_used)))
+    release_lock(ticket);
 }
 
 
@@ -1687,40 +2210,16 @@ MDL_context::is_lock_owner(MDL_key::enum_mdl_namespace mdl_namespace,
   existing shared lock.
 
   @pre The ticket must match an acquired lock.
-  @pre The caller already has acquired LOCK_mdl.
 
   @return TRUE if there is a conflicting lock request, FALSE otherwise.
 */
 
-bool MDL_ticket::has_pending_conflicting_lock_impl() const
-{
-  DBUG_ASSERT(is_shared());
-  safe_mutex_assert_owner(&LOCK_mdl);
-
-  return !m_lock->waiting.is_empty();
-}
-
-
-/**
-  Check if we have any pending exclusive locks which conflict with
-  existing shared lock.
-
-  @pre The ticket must match an acquired lock.
-
-  @return TRUE if there is a pending conflicting lock request,
-          FALSE otherwise.
-*/
-
 bool MDL_ticket::has_pending_conflicting_lock() const
 {
-  bool result;
-
   safe_mutex_assert_not_owner(&LOCK_open);
+  DBUG_ASSERT(is_shared());
 
-  pthread_mutex_lock(&LOCK_mdl);
-  result= has_pending_conflicting_lock_impl();
-  pthread_mutex_unlock(&LOCK_mdl);
-  return result;
+  return m_lock->has_pending_exclusive_lock();
 }
 
 
diff --git a/sql/mdl.h b/sql/mdl.h
index 8edbfbc0777..3ae7cdc743d 100644
--- a/sql/mdl.h
+++ b/sql/mdl.h
@@ -36,11 +36,13 @@ class MDL_ticket;
     (because of that their acquisition involves implicit
      acquisition of global intention-exclusive lock).
 
-  @see Comments for can_grant_lock() and can_grant_global_lock() for details.
+  @sa Comments for MDL_object_lock::can_grant_lock() and
+  MDL_global_lock::can_grant_lock() for details.
 */
 
 enum enum_mdl_type {MDL_SHARED=0, MDL_SHARED_HIGH_PRIO,
-                    MDL_SHARED_UPGRADABLE, MDL_EXCLUSIVE};
+                    MDL_SHARED_UPGRADABLE, MDL_INTENTION_EXCLUSIVE,
+                    MDL_EXCLUSIVE};
 
 
 /** States which a metadata lock ticket can have. */
@@ -78,7 +80,8 @@ public:
   enum enum_mdl_namespace { TABLE=0,
                             FUNCTION,
                             PROCEDURE,
-                            TRIGGER };
+                            TRIGGER,
+                            GLOBAL };
 
   const uchar *ptr() const { return (uchar*) m_ptr; }
   uint length() const { return m_length; }
@@ -93,7 +96,8 @@ public:
   { return (enum_mdl_namespace)(m_ptr[0]); }
 
   /**
-    Construct a metadata lock key from a triplet (mdl_namespace, database and name).
+    Construct a metadata lock key from a triplet (mdl_namespace,
+    database and name).
 
     @remark The key for a table is <mdl_namespace>+<database name>+<table name>
 
@@ -102,11 +106,12 @@ public:
     @param  name          Name of of the object
     @param  key           Where to store the the MDL key.
   */
-  void mdl_key_init(enum_mdl_namespace mdl_namespace, const char *db, const char *name)
+  void mdl_key_init(enum_mdl_namespace mdl_namespace,
+                    const char *db, const char *name)
   {
     m_ptr[0]= (char) mdl_namespace;
-    m_db_name_length= (uint) (strmov(m_ptr + 1, db) - m_ptr - 1);
-    m_length= (uint) (strmov(m_ptr + m_db_name_length + 2, name) - m_ptr + 1);
+    m_db_name_length= (uint16) (strmov(m_ptr + 1, db) - m_ptr - 1);
+    m_length= (uint16) (strmov(m_ptr + m_db_name_length + 2, name) - m_ptr + 1);
   }
   void mdl_key_init(const MDL_key *rhs)
   {
@@ -119,20 +124,34 @@ public:
     return (m_length == rhs->m_length &&
             memcmp(m_ptr, rhs->m_ptr, m_length) == 0);
   }
+  /**
+    Compare two MDL keys lexicographically.
+  */
+  int cmp(const MDL_key *rhs) const
+  {
+    /*
+      The key buffer is always '\0'-terminated. Since key
+      character set is utf-8, we can safely assume that no
+      character starts with a zero byte.
+    */
+    return memcmp(m_ptr, rhs->m_ptr, min(m_length, rhs->m_length)+1);
+  }
+
   MDL_key(const MDL_key *rhs)
   {
     mdl_key_init(rhs);
   }
-  MDL_key(enum_mdl_namespace namespace_arg, const char *db_arg, const char *name_arg)
+  MDL_key(enum_mdl_namespace namespace_arg,
+          const char *db_arg, const char *name_arg)
   {
     mdl_key_init(namespace_arg, db_arg, name_arg);
   }
   MDL_key() {} /* To use when part of MDL_request. */
 
 private:
+  uint16 m_length;
+  uint16 m_db_name_length;
   char m_ptr[MAX_MDLKEY_LENGTH];
-  uint m_length;
-  uint m_db_name_length;
 private:
   MDL_key(const MDL_key &);                     /* not implemented */
   MDL_key &operator=(const MDL_key &);          /* not implemented */
@@ -198,7 +217,7 @@ public:
     DBUG_ASSERT(ticket == NULL);
     type= type_arg;
   }
-  bool is_shared() const { return type < MDL_EXCLUSIVE; }
+  bool is_shared() const { return type < MDL_INTENTION_EXCLUSIVE; }
 
   static MDL_request *create(MDL_key::enum_mdl_namespace mdl_namespace,
                              const char *db, const char *name,
@@ -243,6 +262,17 @@ typedef void (*mdl_cached_object_release_hook)(void *);
 
   @note Multiple shared locks on a same object are represented by a
         single ticket. The same does not apply for other lock types.
+
+  @note There are two groups of MDL_ticket members:
+        - "Externally accessible". These members can be accessed from
+          threads/contexts different than ticket owner in cases when
+          ticket participates in some list of granted or waiting tickets
+          for a lock. Therefore one should change these members before
+          including then to waiting/granted lists or while holding lock
+          protecting those lists.
+        - "Context private". Such members are private to thread/context
+          owning this ticket. I.e. they should not be accessed from other
+          threads/contexts.
 */
 
 class MDL_ticket
@@ -250,12 +280,13 @@ class MDL_ticket
 public:
   /**
     Pointers for participating in the list of lock requests for this context.
+    Context private.
   */
   MDL_ticket *next_in_context;
   MDL_ticket **prev_in_context;
   /**
     Pointers for participating in the list of satisfied/pending requests
-    for the lock.
+    for the lock. Externally accessible.
   */
   MDL_ticket *next_in_lock;
   MDL_ticket **prev_in_lock;
@@ -265,8 +296,8 @@ public:
   void *get_cached_object();
   void set_cached_object(void *cached_object,
                          mdl_cached_object_release_hook release_hook);
-  const MDL_context *get_ctx() const { return m_ctx; }
-  bool is_shared() const { return m_type < MDL_EXCLUSIVE; }
+  MDL_context *get_ctx() const { return m_ctx; }
+  bool is_shared() const { return m_type < MDL_INTENTION_EXCLUSIVE; }
   bool is_upgradable_or_exclusive() const
   {
     return m_type == MDL_SHARED_UPGRADABLE || m_type == MDL_EXCLUSIVE;
@@ -275,6 +306,8 @@ public:
   void downgrade_exclusive_lock();
 private:
   friend class MDL_context;
+  friend class MDL_global_lock;
+  friend class MDL_object_lock;
 
   MDL_ticket(MDL_context *ctx_arg, enum_mdl_type type_arg)
    : m_type(type_arg),
@@ -283,31 +316,31 @@ private:
      m_lock(NULL)
   {}
 
-
   static MDL_ticket *create(MDL_context *ctx_arg, enum_mdl_type type_arg);
   static void destroy(MDL_ticket *ticket);
 private:
-  /** Type of metadata lock. */
+  /** Type of metadata lock. Externally accessible. */
   enum enum_mdl_type m_type;
-  /** State of the metadata lock ticket. */
+  /** State of the metadata lock ticket. Context private. */
   enum enum_mdl_state m_state;
 
-  /** Context of the owner of the metadata lock ticket. */
+  /**
+    Context of the owner of the metadata lock ticket. Externally accessible.
+  */
   MDL_context *m_ctx;
 
-  /** Pointer to the lock object for this lock ticket. */
+  /** Pointer to the lock object for this lock ticket. Context private. */
   MDL_lock *m_lock;
 private:
   MDL_ticket(const MDL_ticket &);               /* not implemented */
   MDL_ticket &operator=(const MDL_ticket &);    /* not implemented */
-
-  bool has_pending_conflicting_lock_impl() const;
 };
 
 
 typedef I_P_List<MDL_request, I_P_List_adapter<MDL_request,
                  &MDL_request::next_in_list,
-                 &MDL_request::prev_in_list> >
+                 &MDL_request::prev_in_list>,
+                 I_P_List_counter>
         MDL_request_list;
 
 /**
@@ -326,21 +359,19 @@ public:
 
   typedef Ticket_list::Iterator Ticket_iterator;
 
-  void init(THD *thd);
+  MDL_context();
   void destroy();
 
   bool try_acquire_shared_lock(MDL_request *mdl_request);
   bool acquire_exclusive_lock(MDL_request *mdl_request);
   bool acquire_exclusive_locks(MDL_request_list *requests);
   bool try_acquire_exclusive_lock(MDL_request *mdl_request);
-  bool acquire_global_shared_lock();
   bool clone_ticket(MDL_request *mdl_request);
 
   bool wait_for_locks(MDL_request_list *requests);
 
   void release_all_locks_for_name(MDL_ticket *ticket);
   void release_lock(MDL_ticket *ticket);
-  void release_global_shared_lock();
 
   bool is_exclusive_lock_owner(MDL_key::enum_mdl_namespace mdl_namespace,
                                const char *db,
@@ -368,7 +399,6 @@ public:
 
   void set_lt_or_ha_sentinel()
   {
-    DBUG_ASSERT(m_lt_or_ha_sentinel == NULL);
     m_lt_or_ha_sentinel= mdl_savepoint();
   }
   MDL_ticket *lt_or_ha_sentinel() const { return m_lt_or_ha_sentinel; }
@@ -385,16 +415,35 @@ public:
   bool can_wait_lead_to_deadlock() const;
 
   inline THD *get_thd() const { return m_thd; }
-  
-  bool is_waiting_in_mdl() const { return m_is_waiting_in_mdl; }
-private:
-  Ticket_list m_tickets;
-  bool m_has_global_shared_lock;
+
   /**
-    Indicates that the owner of this context is waiting in
-    wait_for_locks() method.
+    Wake up context which is waiting for a change of MDL_lock state.
   */
-  bool m_is_waiting_in_mdl;
+  void awake()
+  {
+    pthread_cond_signal(&m_ctx_wakeup_cond);
+  }
+
+  bool try_acquire_global_intention_exclusive_lock(MDL_request *mdl_request);
+  bool acquire_global_intention_exclusive_lock(MDL_request *mdl_request);
+
+  bool acquire_global_shared_lock();
+  void release_global_shared_lock();
+
+  /**
+    Check if this context owns global lock of particular type.
+  */
+  bool is_global_lock_owner(enum_mdl_type type_arg)
+  {
+    MDL_request mdl_request;
+    bool not_used;
+    mdl_request.init(MDL_key::GLOBAL, "", "", type_arg);
+    return find_ticket(&mdl_request, &not_used);
+  }
+
+  void init(THD *thd_arg) { m_thd= thd_arg; }
+private:
+  Ticket_list m_tickets;
   /**
     This member has two uses:
     1) When entering LOCK TABLES mode, remember the last taken
@@ -406,12 +455,27 @@ private:
   */
   MDL_ticket *m_lt_or_ha_sentinel;
   THD *m_thd;
+  /**
+    Condvar which is used for waiting until this context's pending
+    request can be satisfied or this thread has to perform actions
+    to resolve potential deadlock (we subscribe for such notification
+    by adding ticket corresponding to the request to an appropriate
+    queue of waiters).
+  */
+  pthread_cond_t m_ctx_wakeup_cond;
 private:
-  void release_ticket(MDL_ticket *ticket);
-  bool can_wait_lead_to_deadlock_impl() const;
   MDL_ticket *find_ticket(MDL_request *mdl_req,
                           bool *is_lt_or_ha);
   void release_locks_stored_before(MDL_ticket *sentinel);
+
+  bool try_acquire_lock_impl(MDL_request *mdl_request);
+  bool acquire_lock_impl(MDL_request *mdl_request);
+  bool acquire_exclusive_lock_impl(MDL_request *mdl_request);
+
+  friend bool MDL_ticket::upgrade_shared_lock_to_exclusive();
+private:
+  MDL_context(const MDL_context &rhs);          /* not implemented */
+  MDL_context &operator=(MDL_context &rhs);     /* not implemented */
 };
 
 
diff --git a/sql/mysql_priv.h b/sql/mysql_priv.h
index bb5bf428ef0..9f153b5aa0e 100644
--- a/sql/mysql_priv.h
+++ b/sql/mysql_priv.h
@@ -1689,13 +1689,13 @@ void close_open_tables_and_downgrade(ALTER_PARTITION_PARAM_TYPE *lpt);
 
 /* Functions to work with system tables. */
 bool open_system_tables_for_read(THD *thd, TABLE_LIST *table_list,
-                                 Open_tables_state *backup);
-void close_system_tables(THD *thd, Open_tables_state *backup);
+                                 Open_tables_backup *backup);
+void close_system_tables(THD *thd, Open_tables_backup *backup);
 TABLE *open_system_table_for_update(THD *thd, TABLE_LIST *one_table);
 
 TABLE *open_performance_schema_table(THD *thd, TABLE_LIST *one_table,
-                                     Open_tables_state *backup);
-void close_performance_schema_table(THD *thd, Open_tables_state *backup);
+                                     Open_tables_backup *backup);
+void close_performance_schema_table(THD *thd, Open_tables_backup *backup);
 
 bool close_cached_tables(THD *thd, TABLE_LIST *tables, bool have_lock,
                          bool wait_for_refresh);
diff --git a/sql/sp.cc b/sql/sp.cc
index 1375d44cb9b..4cbb0f28b59 100644
--- a/sql/sp.cc
+++ b/sql/sp.cc
@@ -260,7 +260,7 @@ Stored_routine_creation_ctx::load_from_db(THD *thd,
     \#	Pointer to TABLE object of mysql.proc
 */
 
-TABLE *open_proc_table_for_read(THD *thd, Open_tables_state *backup)
+TABLE *open_proc_table_for_read(THD *thd, Open_tables_backup *backup)
 {
   TABLE_LIST table;
 
@@ -382,7 +382,7 @@ db_find_routine(THD *thd, int type, sp_name *name, sp_head **sphp)
   String str(buff, sizeof(buff), &my_charset_bin);
   bool saved_time_zone_used= thd->time_zone_used;
   ulong sql_mode, saved_mode= thd->variables.sql_mode;
-  Open_tables_state open_tables_state_backup;
+  Open_tables_backup open_tables_state_backup;
   Stored_program_creation_ctx *creation_ctx;
 
   DBUG_ENTER("db_find_routine");
@@ -1432,7 +1432,7 @@ sp_routine_exists_in_table(THD *thd, int type, sp_name *name)
 {
   TABLE *table;
   int ret;
-  Open_tables_state open_tables_state_backup;
+  Open_tables_backup open_tables_state_backup;
 
   if (!(table= open_proc_table_for_read(thd, &open_tables_state_backup)))
     ret= SP_OPEN_TABLE_FAILED;
diff --git a/sql/sp.h b/sql/sp.h
index cab4d7dbeb5..c051b584031 100644
--- a/sql/sp.h
+++ b/sql/sp.h
@@ -128,6 +128,6 @@ extern "C" uchar* sp_sroutine_key(const uchar *ptr, size_t *plen,
   Routines which allow open/lock and close mysql.proc table even when
   we already have some tables open and locked.
 */
-TABLE *open_proc_table_for_read(THD *thd, Open_tables_state *backup);
+TABLE *open_proc_table_for_read(THD *thd, Open_tables_backup *backup);
 
 #endif /* _SP_H_ */
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index 8d3ef372842..2f891375163 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -1519,25 +1519,23 @@ void close_thread_tables(THD *thd)
   if (thd->open_tables)
     close_open_tables(thd);
 
-  if (thd->state_flags & Open_tables_state::BACKUPS_AVAIL)
-  {
-    /* We can't have an open HANDLER in the backup open tables state. */
-    DBUG_ASSERT(thd->mdl_context.lt_or_ha_sentinel() == NULL);
-    /*
-      Due to the above assert, this is guaranteed to release *all* locks
-      in the context.
-    */
-    thd->mdl_context.release_transactional_locks();
-  }
-  else if (! thd->in_multi_stmt_transaction())
+  /*
+    - If inside a multi-statement transaction,
+    defer the release of metadata locks until the current
+    transaction is either committed or rolled back. This prevents
+    other statements from modifying the table for the entire
+    duration of this transaction.  This provides commit ordering
+    and guarantees serializability across multiple transactions.
+    - If closing a system table, defer the release of metadata locks
+    to the caller. We have no sentinel in MDL subsystem to guard
+    transactional locks from system tables locks, so don't know
+    which locks are which here.
+    - If in autocommit mode, or outside a transactional context,
+    automatically release metadata locks of the current statement.
+  */
+  if (! thd->in_multi_stmt_transaction() &&
+      ! (thd->state_flags & Open_tables_state::BACKUPS_AVAIL))
   {
-    /*
-      Defer the release of metadata locks until the current transaction
-      is either committed or rolled back. This prevents other statements
-      from modifying the table for the entire duration of this transaction.
-      This provides commitment ordering for guaranteeing serializability
-      across multiple transactions.
-    */
     thd->mdl_context.release_transactional_locks();
   }
 
@@ -2336,10 +2334,9 @@ open_table_get_mdl_lock(THD *thd, TABLE_LIST *table_list,
                         Open_table_context *ot_ctx,
                         uint flags)
 {
-  ot_ctx->add_request(mdl_request);
-
   if (table_list->lock_strategy)
   {
+    MDL_request *global_request;
     /*
       In case of CREATE TABLE .. If NOT EXISTS .. SELECT, the table
       may not yet exist. Let's acquire an exclusive lock for that
@@ -2349,10 +2346,24 @@ open_table_get_mdl_lock(THD *thd, TABLE_LIST *table_list,
       shared locks. This invariant is preserved here and is also
       enforced by asserts in metadata locking subsystem.
     */
+
     mdl_request->set_type(MDL_EXCLUSIVE);
     DBUG_ASSERT(! thd->mdl_context.has_locks() ||
-                thd->handler_tables_hash.records);
+                thd->handler_tables_hash.records ||
+                thd->global_read_lock);
+
+    if (!(global_request= ot_ctx->get_global_mdl_request(thd)))
+      return 1;
 
+    if (! global_request->ticket)
+    {
+      ot_ctx->add_request(global_request);
+      if (thd->mdl_context.acquire_global_intention_exclusive_lock(
+                             global_request))
+      return 1;
+    }
+
+    ot_ctx->add_request(mdl_request);
     if (thd->mdl_context.acquire_exclusive_lock(mdl_request))
       return 1;
   }
@@ -2371,8 +2382,29 @@ open_table_get_mdl_lock(THD *thd, TABLE_LIST *table_list,
     if (flags & MYSQL_LOCK_IGNORE_FLUSH)
       mdl_request->set_type(MDL_SHARED_HIGH_PRIO);
 
+    if (mdl_request->type == MDL_SHARED_UPGRADABLE)
+    {
+      MDL_request *global_request;
+
+      if (!(global_request= ot_ctx->get_global_mdl_request(thd)))
+        return 1;
+      if (! global_request->ticket)
+      {
+        ot_ctx->add_request(global_request);
+        if (thd->mdl_context.try_acquire_global_intention_exclusive_lock(
+                               global_request))
+          return 1;
+        if (! global_request->ticket)
+          goto failure;
+      }
+    }
+
+    ot_ctx->add_request(mdl_request);
+
     if (thd->mdl_context.try_acquire_shared_lock(mdl_request))
       return 1;
+
+failure:
     if (mdl_request->ticket == NULL)
     {
       if (flags & MYSQL_OPEN_FAIL_ON_MDL_CONFLICT)
@@ -2919,8 +2951,6 @@ err_unlock:
   release_table_share(share);
 err_unlock2:
   pthread_mutex_unlock(&LOCK_open);
-  if (! (flags & MYSQL_OPEN_HAS_MDL_LOCK))
-    thd->mdl_context.release_lock(mdl_ticket);
 
   DBUG_RETURN(TRUE);
 }
@@ -3713,11 +3743,34 @@ Open_table_context::Open_table_context(THD *thd)
    m_start_of_statement_svp(thd->mdl_context.mdl_savepoint()),
    m_has_locks((thd->in_multi_stmt_transaction() ||
                 thd->mdl_context.lt_or_ha_sentinel()) &&
-               thd->mdl_context.has_locks())
+               thd->mdl_context.has_locks()),
+   m_global_mdl_request(NULL)
 {}
 
 
 /**
+  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)
+  {
+    char *buff;
+    if ((buff= (char*)thd->alloc(sizeof(MDL_request))))
+    {
+      m_global_mdl_request= new (buff) 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.
 
@@ -3777,6 +3830,11 @@ recover_from_failed_open(THD *thd, MDL_request *mdl_request,
                          TABLE_LIST *table)
 {
   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)
   {
@@ -3787,11 +3845,26 @@ recover_from_failed_open(THD *thd, MDL_request *mdl_request,
       break;
     case OT_DISCOVER:
       {
+        MDL_request mdl_global_request;
         MDL_request mdl_xlock_request(mdl_request);
+
+        mdl_global_request.init(MDL_key::GLOBAL, "", "",
+                                MDL_INTENTION_EXCLUSIVE);
         mdl_xlock_request.set_type(MDL_EXCLUSIVE);
+
+
+        if ((result= thd->mdl_context.acquire_global_intention_exclusive_lock(
+                                        &mdl_global_request)))
+          break;
+
         if ((result=
              thd->mdl_context.acquire_exclusive_lock(&mdl_xlock_request)))
+        {
+          /*
+            We rely on close_thread_tables() to release global lock eventually.
+          */
           break;
+        }
 
         DBUG_ASSERT(mdl_request->key.mdl_namespace() == MDL_key::TABLE);
         pthread_mutex_lock(&LOCK_open);
@@ -3805,16 +3878,30 @@ recover_from_failed_open(THD *thd, MDL_request *mdl_request,
 
         thd->warning_info->clear_warning_info(thd->query_id);
         thd->clear_error();                 // Clear error message
-        thd->mdl_context.release_lock(mdl_xlock_request.ticket);
+        thd->mdl_context.release_transactional_locks();
         break;
       }
     case OT_REPAIR:
       {
+        MDL_request mdl_global_request;
         MDL_request mdl_xlock_request(mdl_request);
+
+        mdl_global_request.init(MDL_key::GLOBAL, "", "",
+                                MDL_INTENTION_EXCLUSIVE);
         mdl_xlock_request.set_type(MDL_EXCLUSIVE);
+
+        if ((result= thd->mdl_context.acquire_global_intention_exclusive_lock(
+                                        &mdl_global_request)))
+          break;
+
         if ((result=
              thd->mdl_context.acquire_exclusive_lock(&mdl_xlock_request)))
+        {
+          /*
+            We rely on close_thread_tables() to release global lock eventually.
+          */
           break;
+        }
 
         DBUG_ASSERT(mdl_request->key.mdl_namespace() == MDL_key::TABLE);
         pthread_mutex_lock(&LOCK_open);
@@ -3824,7 +3911,7 @@ recover_from_failed_open(THD *thd, MDL_request *mdl_request,
         pthread_mutex_unlock(&LOCK_open);
 
         result= auto_repair_table(thd, table);
-        thd->mdl_context.release_lock(mdl_xlock_request.ticket);
+        thd->mdl_context.release_transactional_locks();
         break;
       }
     default:
@@ -3921,6 +4008,13 @@ open_and_process_routine(THD *thd, Query_tables_list *prelocking_ctx,
           mdl_type != MDL_key::PROCEDURE)
       {
         ot_ctx->add_request(&rt->mdl_request);
+
+        /*
+          Since we acquire only shared lock on routines we don't
+          need to care about global intention exclusive locks.
+        */
+        DBUG_ASSERT(rt->mdl_request.type == MDL_SHARED);
+
         if (thd->mdl_context.try_acquire_shared_lock(&rt->mdl_request))
           DBUG_RETURN(TRUE);
 
@@ -8784,7 +8878,7 @@ has_write_table_with_auto_increment(TABLE_LIST *tables)
 
 bool
 open_system_tables_for_read(THD *thd, TABLE_LIST *table_list,
-                            Open_tables_state *backup)
+                            Open_tables_backup *backup)
 {
   Query_tables_list query_tables_list_backup;
   LEX *lex= thd->lex;
@@ -8830,13 +8924,13 @@ error:
   SYNOPSIS
     close_system_tables()
       thd     Thread context
-      backup  Pointer to Open_tables_state instance which holds
+      backup  Pointer to Open_tables_backup instance which holds
               information about tables which were open before we
               decided to access system tables.
 */
 
 void
-close_system_tables(THD *thd, Open_tables_state *backup)
+close_system_tables(THD *thd, Open_tables_backup *backup)
 {
   close_thread_tables(thd);
   thd->restore_backup_open_tables_state(backup);
@@ -8887,7 +8981,7 @@ open_system_table_for_update(THD *thd, TABLE_LIST *one_table)
 */
 TABLE *
 open_performance_schema_table(THD *thd, TABLE_LIST *one_table,
-                              Open_tables_state *backup)
+                              Open_tables_backup *backup)
 {
   uint flags= ( MYSQL_LOCK_IGNORE_GLOBAL_READ_LOCK |
                 MYSQL_LOCK_IGNORE_GLOBAL_READ_ONLY |
@@ -8936,51 +9030,9 @@ open_performance_schema_table(THD *thd, TABLE_LIST *one_table,
   @param thd The current thread
   @param backup [in] the context to restore.
 */
-void close_performance_schema_table(THD *thd, Open_tables_state *backup)
+void close_performance_schema_table(THD *thd, Open_tables_backup *backup)
 {
-  bool found_old_table;
-
-  /*
-    If open_performance_schema_table() fails,
-    this function should not be called.
-  */
-  DBUG_ASSERT(thd->lock != NULL);
-
-  /*
-    Note:
-    We do not create explicitly a separate transaction for the
-    performance table I/O, but borrow the current transaction.
-    lock + unlock will autocommit the change done in the
-    performance schema table: this is the expected result.
-    The current transaction should not be affected by this code.
-    TODO: Note that if a transactional engine is used for log tables,
-    this code will need to be revised, as a separate transaction
-    might be needed.
-  */
-  mysql_unlock_tables(thd, thd->lock);
-  thd->lock= 0;
-
-  pthread_mutex_lock(&LOCK_open);
-
-  found_old_table= false;
-  /*
-    Note that we need to hold LOCK_open while changing the
-    open_tables list. Another thread may work on it.
-    (See: notify_thread_having_shared_lock())
-  */
-  while (thd->open_tables)
-    found_old_table|= close_thread_table(thd, &thd->open_tables);
-
-  if (found_old_table)
-    broadcast_refresh();
-
-  pthread_mutex_unlock(&LOCK_open);
-
-  /* We can't have an open HANDLER in the backup context. */
-  DBUG_ASSERT(thd->mdl_context.lt_or_ha_sentinel() == NULL);
-  thd->mdl_context.release_transactional_locks();
-
-  thd->restore_backup_open_tables_state(backup);
+  close_system_tables(thd, backup);
 }
 
 /**
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index 95c985b2c10..62de06d382c 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -471,6 +471,7 @@ THD::THD()
 {
   ulong tmp;
 
+  mdl_context.init(this);
   /*
     Pass nominal parameters to init_alloc_root only to ensure that
     the destructor works OK in case of an error. The main_mem_root
@@ -1007,7 +1008,8 @@ void THD::cleanup(void)
   */
   DBUG_ASSERT(open_tables == NULL);
   /* All HANDLERs must have been closed by now. */
-  DBUG_ASSERT(mdl_context.lt_or_ha_sentinel() == NULL);
+  DBUG_ASSERT(mdl_context.lt_or_ha_sentinel() == NULL ||
+              global_read_lock);
   /*
     Due to the above assert, this is guaranteed to release *all* in
     this session.
@@ -3024,19 +3026,21 @@ bool Security_context::user_matches(Security_context *them)
   access to mysql.proc table to find definitions of stored routines.
 ****************************************************************************/
 
-void THD::reset_n_backup_open_tables_state(Open_tables_state *backup)
+void THD::reset_n_backup_open_tables_state(Open_tables_backup *backup)
 {
   DBUG_ENTER("reset_n_backup_open_tables_state");
   backup->set_open_tables_state(this);
+  backup->mdl_system_tables_svp= mdl_context.mdl_savepoint();
   reset_open_tables_state(this);
   state_flags|= Open_tables_state::BACKUPS_AVAIL;
   DBUG_VOID_RETURN;
 }
 
 
-void THD::restore_backup_open_tables_state(Open_tables_state *backup)
+void THD::restore_backup_open_tables_state(Open_tables_backup *backup)
 {
   DBUG_ENTER("restore_backup_open_tables_state");
+  mdl_context.rollback_to_savepoint(backup->mdl_system_tables_svp);
   /*
     Before we will throw away current open tables state we want
     to be sure that it was properly cleaned up.
@@ -3046,7 +3050,6 @@ void THD::restore_backup_open_tables_state(Open_tables_state *backup)
               lock == 0 &&
               locked_tables_mode == LTM_NONE &&
               m_reprepare_observer == NULL);
-  mdl_context.destroy();
 
   set_open_tables_state(backup);
   DBUG_VOID_RETURN;
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 5654dcb07a6..dce06f7e0c5 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -978,9 +978,6 @@ public:
     Flags with information about the open tables state.
   */
   uint state_flags;
-
-  MDL_context mdl_context;
-
   /**
      This constructor initializes Open_tables_state instance which can only
      be used as backup storage. To prepare Open_tables_state instance for
@@ -1010,21 +1007,29 @@ public:
     locked_tables_mode= LTM_NONE;
     state_flags= 0U;
     m_reprepare_observer= NULL;
-    mdl_context.init(thd);
-  }
-  void enter_locked_tables_mode(enum_locked_tables_mode mode_arg)
-  {
-    DBUG_ASSERT(locked_tables_mode == LTM_NONE);
-    mdl_context.set_lt_or_ha_sentinel();
-    locked_tables_mode= mode_arg;
-  }
-  void leave_locked_tables_mode()
-  {
-    locked_tables_mode= LTM_NONE;
-    mdl_context.clear_lt_or_ha_sentinel();
   }
 };
 
+
+/**
+  Storage for backup of Open_tables_state. Must
+  be used only to open system tables (TABLE_CATEGORY_SYSTEM
+  and TABLE_CATEGORY_LOG).
+*/
+
+class Open_tables_backup: public Open_tables_state
+{
+public:
+  /**
+    When we backup the open tables state to open a system
+    table or tables, points at the last metadata lock
+    acquired before the backup. Is used to release
+    metadata locks on system tables after they are
+    no longer used.
+  */
+  MDL_ticket *mdl_system_tables_svp;
+};
+
 /**
   @class Sub_statement_state
   @brief Used to save context when executing a function or trigger
@@ -1308,6 +1313,9 @@ public:
   {
     return m_start_of_statement_svp;
   }
+
+  MDL_request *get_global_mdl_request(THD *thd);
+
 private:
   /** List of requests for all locks taken so far. Used for waiting on locks. */
   MDL_request_list m_mdl_requests;
@@ -1320,6 +1328,11 @@ private:
     and we can't safely do back-off (and release them).
   */
   bool m_has_locks;
+  /**
+    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;
 };
 
 
@@ -1426,6 +1439,8 @@ class THD :public Statement,
            public Open_tables_state
 {
 public:
+  MDL_context mdl_context;
+
   /* Used to execute base64 coded binlog events in MySQL server */
   Relay_log_info* rli_fake;
 
@@ -2314,8 +2329,8 @@ public:
   void set_status_var_init();
   bool is_context_analysis_only()
     { return stmt_arena->is_stmt_prepare() || lex->view_prepare_mode; }
-  void reset_n_backup_open_tables_state(Open_tables_state *backup);
-  void restore_backup_open_tables_state(Open_tables_state *backup);
+  void reset_n_backup_open_tables_state(Open_tables_backup *backup);
+  void restore_backup_open_tables_state(Open_tables_backup *backup);
   void reset_sub_statement_state(Sub_statement_state *backup, uint new_state);
   void restore_sub_statement_state(Sub_statement_state *backup);
   void set_n_backup_active_arena(Query_arena *set, Query_arena *backup);
@@ -2567,6 +2582,19 @@ public:
     Protected with LOCK_thd_data mutex.
   */
   void set_query(char *query_arg, uint32 query_length_arg);
+  void enter_locked_tables_mode(enum_locked_tables_mode mode_arg)
+  {
+    DBUG_ASSERT(locked_tables_mode == LTM_NONE);
+    DBUG_ASSERT(! mdl_context.lt_or_ha_sentinel() ||
+                mdl_context.is_global_lock_owner(MDL_SHARED));
+    mdl_context.set_lt_or_ha_sentinel();
+    locked_tables_mode= mode_arg;
+  }
+  void leave_locked_tables_mode()
+  {
+    locked_tables_mode= LTM_NONE;
+    mdl_context.clear_lt_or_ha_sentinel();
+  }
 private:
   /** The current internal error handler for this thread, or NULL. */
   Internal_error_handler *m_internal_handler;
diff --git a/sql/sql_delete.cc b/sql/sql_delete.cc
index 26478b31290..228c001f71b 100644
--- a/sql/sql_delete.cc
+++ b/sql/sql_delete.cc
@@ -1100,7 +1100,7 @@ bool mysql_truncate(THD *thd, TABLE_LIST *table_list, bool dont_send_ok)
   TABLE *table;
   bool error= TRUE;
   uint path_length;
-  MDL_request mdl_request;
+  MDL_request mdl_global_request, mdl_request;
   /*
     Is set if we're under LOCK TABLES, and used
     to downgrade the exclusive lock after the
@@ -1207,10 +1207,21 @@ bool mysql_truncate(THD *thd, TABLE_LIST *table_list, bool dont_send_ok)
         the table can be re-created as an empty table with TRUNCATE
         TABLE, even if the data or index files have become corrupted.
       */
+
+      mdl_global_request.init(MDL_key::GLOBAL, "", "", MDL_INTENTION_EXCLUSIVE);
       mdl_request.init(MDL_key::TABLE, table_list->db, table_list->table_name,
                        MDL_EXCLUSIVE);
+      if (thd->mdl_context.acquire_global_intention_exclusive_lock(
+                             &mdl_global_request))
+        DBUG_RETURN(TRUE);
       if (thd->mdl_context.acquire_exclusive_lock(&mdl_request))
+      {
+        /*
+          We rely on that close_thread_tables() to release global lock
+          in this case.
+        */
         DBUG_RETURN(TRUE);
+      }
       has_mdl_lock= TRUE;
       pthread_mutex_lock(&LOCK_open);
       tdc_remove_table(thd, TDC_RT_REMOVE_ALL, table_list->db,
@@ -1250,7 +1261,7 @@ end:
       my_ok(thd);		// This should return record count
     }
     if (has_mdl_lock)
-      thd->mdl_context.release_lock(mdl_request.ticket);
+      thd->mdl_context.release_transactional_locks();
     if (mdl_ticket)
       mdl_ticket->downgrade_exclusive_lock();
   }
diff --git a/sql/sql_help.cc b/sql/sql_help.cc
index af67db45b36..e9b15e07e9d 100644
--- a/sql/sql_help.cc
+++ b/sql/sql_help.cc
@@ -655,7 +655,12 @@ bool mysqld_help(THD *thd, const char *mask)
   tables[0].db= tables[1].db= tables[2].db= tables[3].db= (char*) "mysql";
   init_mdl_requests(tables);
 
-  Open_tables_state open_tables_state_backup;
+  /*
+    HELP must be available under LOCK TABLES. 
+    Reset and backup the current open tables state to
+    make it possible.
+  */
+  Open_tables_backup open_tables_state_backup;
   if (open_system_tables_for_read(thd, tables, &open_tables_state_backup))
     goto error2;
 
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 486cb9af288..397674471c5 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -6502,7 +6502,8 @@ bool reload_acl_and_cache(THD *thd, ulong options, TABLE_LIST *tables,
 
   DBUG_ASSERT(!thd || thd->locked_tables_mode ||
               !thd->mdl_context.has_locks() ||
-              thd->handler_tables_hash.records);
+              thd->handler_tables_hash.records ||
+              thd->global_read_lock);
 
   /*
     Note that if REFRESH_READ_LOCK bit is set then REFRESH_TABLES is set too
diff --git a/sql/sql_plist.h b/sql/sql_plist.h
index 8f2aee6bd5f..eb239a63467 100644
--- a/sql/sql_plist.h
+++ b/sql/sql_plist.h
@@ -18,7 +18,8 @@
 
 #include <my_global.h>
 
-template <typename T, typename B> class I_P_List_iterator;
+template <typename T, typename B, typename C> class I_P_List_iterator;
+class I_P_List_null_counter;
 
 
 /**
@@ -47,10 +48,14 @@ template <typename T, typename B> class I_P_List_iterator;
                  return &el->prev;
                }
              };
+   @param C  Policy class specifying how counting of elements in the list
+             should be done. Instance of this class is also used as a place
+             where information about number of list elements is stored.
+             @sa I_P_List_null_counter, I_P_List_counter
 */
 
-template <typename T, typename B>
-class I_P_List
+template <typename T, typename B, typename C = I_P_List_null_counter>
+class I_P_List : public C
 {
   T *first;
 
@@ -61,7 +66,7 @@ class I_P_List
   */
 public:
   I_P_List() : first(NULL) { };
-  inline void empty()      { first= NULL; }
+  inline void empty()      { first= NULL; C::reset(); }
   inline bool is_empty() const { return (first == NULL); }
   inline void push_front(T* a)
   {
@@ -70,6 +75,7 @@ public:
       *B::prev_ptr(first)= B::next_ptr(a);
     first= a;
     *B::prev_ptr(a)= &first;
+    C::inc();
   }
   inline void push_back(T *a)
   {
@@ -107,21 +113,23 @@ public:
     if (next)
       *B::prev_ptr(next)= *B::prev_ptr(a);
     **B::prev_ptr(a)= next;
+    C::dec();
   }
   inline T* front() { return first; }
   inline const T *front() const { return first; }
-  void swap(I_P_List<T,B> &rhs)
+  void swap(I_P_List<T, B, C> &rhs)
   {
     swap_variables(T *, first, rhs.first);
     if (first)
       *B::prev_ptr(first)= &first;
     if (rhs.first)
       *B::prev_ptr(rhs.first)= &rhs.first;
+    C::swap(rhs);
   }
 #ifndef _lint
-  friend class I_P_List_iterator<T, B>;
+  friend class I_P_List_iterator<T, B, C>;
 #endif
-  typedef I_P_List_iterator<T, B> Iterator;
+  typedef I_P_List_iterator<T, B, C> Iterator;
 };
 
 
@@ -129,15 +137,15 @@ public:
    Iterator for I_P_List.
 */
 
-template <typename T, typename B>
+template <typename T, typename B, typename C = I_P_List_null_counter>
 class I_P_List_iterator
 {
-  const I_P_List<T, B> *list;
+  const I_P_List<T, B, C> *list;
   T *current;
 public:
-  I_P_List_iterator(const I_P_List<T, B> &a) : list(&a), current(a.first) {}
-  I_P_List_iterator(const I_P_List<T, B> &a, T* current_arg) : list(&a), current(current_arg) {}
-  inline void init(I_P_List<T, B> &a)
+  I_P_List_iterator(const I_P_List<T, B, C> &a) : list(&a), current(a.first) {}
+  I_P_List_iterator(const I_P_List<T, B, C> &a, T* current_arg) : list(&a), current(current_arg) {}
+  inline void init(const I_P_List<T, B, C> &a)
   {
     list= &a;
     current= a.first;
@@ -160,4 +168,39 @@ public:
   }
 };
 
+
+/**
+  Element counting policy class for I_P_List to be used in
+  cases when no element counting should be done.
+*/
+
+class I_P_List_null_counter
+{
+protected:
+  void reset() {}
+  void inc() {}
+  void dec() {}
+  void swap(I_P_List_null_counter &rhs) {}
+};
+
+
+/**
+  Element counting policy class for I_P_List which provides
+  basic element counting.
+*/
+
+class I_P_List_counter
+{
+  uint m_counter;
+protected:
+  I_P_List_counter() : m_counter (0) {}
+  void reset() {m_counter= 0;}
+  void inc() {m_counter++;}
+  void dec() {m_counter--;}
+  void swap(I_P_List_counter &rhs)
+  { swap_variables(uint, m_counter, rhs.m_counter); }
+public:
+  uint elements() const { return m_counter; }
+};
+
 #endif
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index e9d1426b3e3..278e0c1445f 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -2871,7 +2871,7 @@ make_table_name_list(THD *thd, List<LEX_STRING> *table_names, LEX *lex,
                                            due to metadata locks, so to avoid
                                            them we should not wait in case if
                                            conflicting lock is present.
-  @param[in]      open_tables_state_backup pointer to Open_tables_state object
+  @param[in]      open_tables_state_backup pointer to Open_tables_backup object
                                            which is used to save|restore original
                                            status of variables related to
                                            open tables state
@@ -2885,7 +2885,7 @@ static int
 fill_schema_show_cols_or_idxs(THD *thd, TABLE_LIST *tables,
                               ST_SCHEMA_TABLE *schema_table,
                               bool can_deadlock,
-                              Open_tables_state *open_tables_state_backup)
+                              Open_tables_backup *open_tables_state_backup)
 {
   LEX *lex= thd->lex;
   bool res;
@@ -2941,7 +2941,8 @@ fill_schema_show_cols_or_idxs(THD *thd, TABLE_LIST *tables,
                                            table, res, db_name,
                                            table_name));
    thd->temporary_tables= 0;
-   close_tables_for_reopen(thd, &show_table_list, NULL);
+   close_tables_for_reopen(thd, &show_table_list,
+                           open_tables_state_backup->mdl_system_tables_svp);
    DBUG_RETURN(error);
 }
 
@@ -3236,8 +3237,12 @@ end_share:
 
 end_unlock:
   pthread_mutex_unlock(&LOCK_open);
+  /*
+    Don't release the MDL lock, it can be part of a transaction.
+    If it is not, it will be released by the call to
+    MDL_context::rollback_to_savepoint() in the caller.
+  */
 
-  thd->mdl_context.release_lock(table_list.mdl_request.ticket);
   thd->clear_error();
   return res;
 }
@@ -3281,7 +3286,7 @@ int get_all_tables(THD *thd, TABLE_LIST *tables, COND *cond)
   COND *partial_cond= 0;
   uint derived_tables= lex->derived_tables; 
   int error= 1;
-  Open_tables_state open_tables_state_backup;
+  Open_tables_backup open_tables_state_backup;
   bool save_view_prepare_mode= lex->view_prepare_mode;
   Query_tables_list query_tables_list_backup;
 #ifndef NO_EMBEDDED_ACCESS_CHECKS
@@ -3500,7 +3505,8 @@ int get_all_tables(THD *thd, TABLE_LIST *tables, COND *cond)
               res= schema_table->process_table(thd, show_table_list, table,
                                                res, &orig_db_name,
                                                &tmp_lex_string);
-              close_tables_for_reopen(thd, &show_table_list, NULL);
+              close_tables_for_reopen(thd, &show_table_list,
+                                      open_tables_state_backup.mdl_system_tables_svp);
             }
             DBUG_ASSERT(!lex->query_tables_own_last);
             if (res)
@@ -4302,7 +4308,7 @@ int fill_schema_proc(THD *thd, TABLE_LIST *tables, COND *cond)
   TABLE *table= tables->table;
   bool full_access;
   char definer[USER_HOST_BUFF_SIZE];
-  Open_tables_state open_tables_state_backup;
+  Open_tables_backup open_tables_state_backup;
   DBUG_ENTER("fill_schema_proc");
 
   strxmov(definer, thd->security_ctx->priv_user, "@",
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 30d6efff7ec..e8c2af4c87b 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -2207,22 +2207,26 @@ err:
       locked. Additional check for 'non_temp_tables_count' is to avoid
       leaving LOCK TABLES mode if we have dropped only temporary tables.
     */
-    if (thd->locked_tables_mode &&
-        thd->lock && thd->lock->table_count == 0 && non_temp_tables_count > 0)
-    {
-      thd->locked_tables_list.unlock_locked_tables(thd);
-      goto end;
-    }
-    for (table= tables; table; table= table->next_local)
+    if (! thd->locked_tables_mode)
+      unlock_table_names(thd);
+    else
     {
-      if (table->mdl_request.ticket)
+      if (thd->lock && thd->lock->table_count == 0 && non_temp_tables_count > 0)
       {
-        /*
-          Under LOCK TABLES we may have several instances of table open
-          and locked and therefore have to remove several metadata lock
-          requests associated with them.
-        */
-        thd->mdl_context.release_all_locks_for_name(table->mdl_request.ticket);
+        thd->locked_tables_list.unlock_locked_tables(thd);
+        goto end;
+      }
+      for (table= tables; table; table= table->next_local)
+      {
+        if (table->mdl_request.ticket)
+        {
+          /*
+            Under LOCK TABLES we may have several instances of table open
+            and locked and therefore have to remove several metadata lock
+            requests associated with them.
+          */
+          thd->mdl_context.release_all_locks_for_name(table->mdl_request.ticket);
+        }
       }
     }
   }
@@ -4350,6 +4354,14 @@ static int prepare_for_repair(THD *thd, TABLE_LIST *table_list,
   if (!(table= table_list->table))
   {
     /*
+      If the table didn't exist, we have a shared metadata lock
+      on it that is left from mysql_admin_table()'s attempt to 
+      open it. Release the shared metadata lock before trying to
+      acquire the exclusive lock to satisfy MDL asserts and avoid
+      deadlocks.
+    */
+    thd->mdl_context.release_transactional_locks();
+    /*
       Attempt to do full-blown table open in mysql_admin_table() has failed.
       Let us try to open at least a .FRM for this table.
     */
@@ -4360,6 +4372,14 @@ static int prepare_for_repair(THD *thd, TABLE_LIST *table_list,
     table_list->mdl_request.init(MDL_key::TABLE,
                                  table_list->db, table_list->table_name,
                                  MDL_EXCLUSIVE);
+
+    MDL_request mdl_global_request;
+    mdl_global_request.init(MDL_key::GLOBAL, "", "", MDL_INTENTION_EXCLUSIVE);
+
+    if (thd->mdl_context.acquire_global_intention_exclusive_lock(
+               &mdl_global_request))
+      DBUG_RETURN(0);
+
     if (thd->mdl_context.acquire_exclusive_lock(&table_list->mdl_request))
       DBUG_RETURN(0);
     has_mdl_lock= TRUE;
@@ -4491,7 +4511,7 @@ end:
   }
   /* In case of a temporary table there will be no metadata lock. */
   if (error && has_mdl_lock)
-    thd->mdl_context.release_lock(table_list->mdl_request.ticket);
+    thd->mdl_context.release_transactional_locks();
 
   DBUG_RETURN(error);
 }
@@ -6544,6 +6564,13 @@ view_err:
       {
         target_mdl_request.init(MDL_key::TABLE, new_db, new_name,
                                 MDL_EXCLUSIVE);
+        /*
+          Global intention exclusive lock must have been already acquired when
+          table to be altered was open, so there is no need to do it here.
+        */
+        DBUG_ASSERT(thd->
+                    mdl_context.is_global_lock_owner(MDL_INTENTION_EXCLUSIVE));
+
         if (thd->mdl_context.try_acquire_exclusive_lock(&target_mdl_request))
           DBUG_RETURN(TRUE);
         if (target_mdl_request.ticket == NULL)
diff --git a/sql/tztime.cc b/sql/tztime.cc
index 2ec641071ee..aa9780754d7 100644
--- a/sql/tztime.cc
+++ b/sql/tztime.cc
@@ -1563,7 +1563,6 @@ my_tz_init(THD *org_thd, const char *default_tzname, my_bool bootstrap)
 {
   THD *thd;
   TABLE_LIST tz_tables[1+MY_TZ_TABLES_COUNT];
-  Open_tables_state open_tables_state_backup;
   TABLE *table;
   Tz_names_entry *tmp_tzname;
   my_bool return_val= 1;
@@ -1642,7 +1641,8 @@ my_tz_init(THD *org_thd, const char *default_tzname, my_bool bootstrap)
     We need to open only mysql.time_zone_leap_second, but we try to
     open all time zone tables to see if they exist.
   */
-  if (open_system_tables_for_read(thd, tz_tables, &open_tables_state_backup))
+  if (open_and_lock_tables_derived(thd, tz_tables, FALSE,
+                                   MYSQL_LOCK_IGNORE_FLUSH))
   {
     sql_print_warning("Can't open and lock time zone table: %s "
                       "trying to live without them", thd->stmt_da->message());
@@ -1651,6 +1651,9 @@ my_tz_init(THD *org_thd, const char *default_tzname, my_bool bootstrap)
     goto end_with_setting_default_tz;
   }
 
+  for (TABLE_LIST *tl= tz_tables; tl; tl= tl->next_global)
+    tl->table->use_all_columns();
+
   /*
     Now we are going to load leap seconds descriptions that are shared
     between all time zones that use them. We are using index for getting
@@ -1739,7 +1742,8 @@ end_with_close:
   if (time_zone_tables_exist)
   {
     thd->version--; /* Force close to free memory */
-    close_system_tables(thd, &open_tables_state_backup);
+    close_thread_tables(thd);
+    thd->mdl_context.release_transactional_locks();
   }
 
 end_with_cleanup:
@@ -2293,7 +2297,7 @@ my_tz_find(THD *thd, const String *name)
     else if (time_zone_tables_exist)
     {
       TABLE_LIST tz_tables[MY_TZ_TABLES_COUNT];
-      Open_tables_state open_tables_state_backup;
+      Open_tables_backup open_tables_state_backup;
 
       tz_init_table_list(tz_tables);
       init_mdl_requests(tz_tables);