More merge fixes:

- mdl.cc and mdl.h merged completely
- mysql_system_tables*.sql merged completely
- Fixed wrong merge of lock_tables
- Added some missing functions:
 - bool THD::notify_shared_lock()
 - Dynamic_array::pop, Dynamic_array::del
- Added MDL_context_owner to THD
- Added metadata_locks_hash_instances

8 files changed, 381 insertions, 151 deletions

diff --git a/sql/mdl.cc b/sql/mdl.cc
index 65595c351a2..22d9d4e53e5 100644
--- a/sql/mdl.cc
+++ b/sql/mdl.cc
@@ -1,4 +1,4 @@
-/* Copyright (c) 2007, 2011, Oracle and/or its affiliates.
+/* Copyright (c) 2007, 2012, Oracle and/or its affiliates.
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -14,9 +14,9 @@
    51 Franklin Street, Suite 500, Boston, MA 02110-1335 USA */
 
 
-#include "mdl.h"
 #include "sql_class.h"
 #include "debug_sync.h"
+#include "sql_array.h"
 #include <hash.h>
 #include <mysqld_error.h>
 #include <mysql/plugin.h>
@@ -29,7 +29,7 @@ static PSI_mutex_key key_MDL_wait_LOCK_wait_status;
 
 static PSI_mutex_info all_mdl_mutexes[]=
 {
-  { &key_MDL_map_mutex, "MDL_map::mutex", PSI_FLAG_GLOBAL},
+  { &key_MDL_map_mutex, "MDL_map::mutex", 0},
   { &key_MDL_wait_LOCK_wait_status, "MDL_wait::LOCK_wait_status", 0}
 };
 
@@ -113,22 +113,26 @@ class MDL_object_lock_cache_adapter;
 
 
 /**
-  A collection of all MDL locks. A singleton,
-  there is only one instance of the map in the server.
+  A partition in a collection of all MDL locks.
+  MDL_map is partitioned for scalability reasons.
   Maps MDL_key to MDL_lock instances.
 */
 
-class MDL_map
+class MDL_map_partition
 {
 public:
-  void init();
-  void destroy();
-  MDL_lock *find_or_insert(const MDL_key *key);
-  void remove(MDL_lock *lock);
+  MDL_map_partition();
+  ~MDL_map_partition();
+  inline MDL_lock *find_or_insert(const MDL_key *mdl_key,
+                                  my_hash_value_type hash_value);
+  inline void remove(MDL_lock *lock);
+  my_hash_value_type get_key_hash(const MDL_key *mdl_key) const
+  {
+    return my_calc_hash(&m_locks, mdl_key->ptr(), mdl_key->length());
+  }
 private:
   bool move_from_hash_to_lock_mutex(MDL_lock *lock);
-private:
-  /** All acquired locks in the server. */
+  /** A partition of all acquired locks in the server. */
   HASH m_locks;
   /* Protects access to m_locks hash. */
   mysql_mutex_t m_mutex;
@@ -151,6 +155,30 @@ private:
                    I_P_List_counter>
           Lock_cache;
   Lock_cache m_unused_locks_cache;
+};
+
+
+/**
+  Start-up parameter for the number of partitions of the MDL_lock hash.
+*/
+ulong mdl_locks_hash_partitions;
+
+/**
+  A collection of all MDL locks. A singleton,
+  there is only one instance of the map in the server.
+  Contains instances of MDL_map_partition
+*/
+
+class MDL_map
+{
+public:
+  void init();
+  void destroy();
+  MDL_lock *find_or_insert(const MDL_key *key);
+  void remove(MDL_lock *lock);
+private:
+  /** Array of partitions where the locks are actually stored. */
+  Dynamic_array<MDL_map_partition *> m_partitions;
   /** Pre-allocated MDL_lock object for GLOBAL namespace. */
   MDL_lock *m_global_lock;
   /** Pre-allocated MDL_lock object for COMMIT namespace. */
@@ -398,7 +426,8 @@ public:
   bool can_grant_lock(enum_mdl_type type, MDL_context *requstor_ctx,
                       bool ignore_lock_priority) const;
 
-  inline static MDL_lock *create(const MDL_key *key);
+  inline static MDL_lock *create(const MDL_key *key,
+                                 MDL_map_partition *map_part);
 
   void reschedule_waiters();
 
@@ -425,13 +454,14 @@ public:
 
 public:
 
-  MDL_lock(const MDL_key *key_arg)
+  MDL_lock(const MDL_key *key_arg, MDL_map_partition *map_part)
   : key(key_arg),
     m_hog_lock_count(0),
     m_ref_usage(0),
     m_ref_release(0),
     m_is_destroyed(FALSE),
-    m_version(0)
+    m_version(0),
+    m_map_part(map_part)
   {
     mysql_prlock_init(key_MDL_lock_rwlock, &m_rwlock);
   }
@@ -444,18 +474,18 @@ public:
 public:
   /**
     These three members are used to make it possible to separate
-    the mdl_locks.m_mutex mutex and MDL_lock::m_rwlock in
+    the MDL_map_partition::m_mutex mutex and MDL_lock::m_rwlock 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_rwlock, thus
+    have both the MDL_map_partition::m_mutex and MDL_lock::m_rwlock, 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
+    MDL_map_partition::m_mutex, but when 'm_is_destroyed' is set to TRUE, this
     member is moved to be protected by the MDL_lock::m_rwlock.
     This means that the MDL_map::find_or_insert() which only
     holds the MDL_lock::m_rwlock 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.
+    without acquiring MDL_map_partition::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_rwlock.
     Note since we are only interested in equality of these two
@@ -469,19 +499,23 @@ public:
   /**
     We use the same idea and an additional version counter to support
     caching of unused MDL_lock object for further re-use.
-    This counter is incremented while holding both MDL_map::m_mutex and
-    MDL_lock::m_rwlock locks each time when a MDL_lock is moved from
-    the hash to the unused objects list (or destroyed).
+    This counter is incremented while holding both MDL_map_partition::m_mutex
+    and MDL_lock::m_rwlock locks each time when a MDL_lock is moved from
+    the partitioned hash to the paritioned unused objects list (or destroyed).
     A thread, which has found a MDL_lock object for the key in the hash
-    and then released the MDL_map::m_mutex before acquiring the
+    and then released the MDL_map_partition::m_mutex before acquiring the
     MDL_lock::m_rwlock, can determine that this object was moved to the
     unused objects list (or destroyed) while it held no locks by comparing
-    the version value which it read while holding the MDL_map::m_mutex
+    the version value which it read while holding the MDL_map_partition::m_mutex
     with the value read after acquiring the MDL_lock::m_rwlock.
     Note that since it takes several years to overflow this counter such
     theoretically possible overflows should not have any practical effects.
   */
   ulonglong m_version;
+  /**
+    Partition of MDL_map where the lock is stored.
+  */
+  MDL_map_partition *m_map_part;
 };
 
 
@@ -494,8 +528,8 @@ public:
 class MDL_scoped_lock : public MDL_lock
 {
 public:
-  MDL_scoped_lock(const MDL_key *key_arg)
-    : MDL_lock(key_arg)
+  MDL_scoped_lock(const MDL_key *key_arg, MDL_map_partition *map_part)
+    : MDL_lock(key_arg, map_part)
   { }
 
   virtual const bitmap_t *incompatible_granted_types_bitmap() const
@@ -535,8 +569,8 @@ private:
 class MDL_object_lock : public MDL_lock
 {
 public:
-  MDL_object_lock(const MDL_key *key_arg)
-    : MDL_lock(key_arg)
+  MDL_object_lock(const MDL_key *key_arg, MDL_map_partition *map_part)
+    : MDL_lock(key_arg, map_part)
   { }
 
   /**
@@ -569,7 +603,7 @@ public:
   }
   virtual bool needs_notification(const MDL_ticket *ticket) const
   {
-    return ticket->is_upgradable_or_exclusive();
+    return (ticket->get_type() >= MDL_SHARED_NO_WRITE);
   }
   virtual void notify_conflicting_locks(MDL_context *ctx);
 
@@ -666,33 +700,62 @@ void mdl_destroy()
 }
 
 
-/** Initialize the global hash containing all MDL locks. */
+/** Initialize the container for all MDL locks. */
 
 void MDL_map::init()
 {
   MDL_key global_lock_key(MDL_key::GLOBAL, "", "");
   MDL_key commit_lock_key(MDL_key::COMMIT, "", "");
 
+  m_global_lock= MDL_lock::create(&global_lock_key, NULL);
+  m_commit_lock= MDL_lock::create(&commit_lock_key, NULL);
+
+  for (uint i= 0; i < mdl_locks_hash_partitions; i++)
+  {
+    MDL_map_partition *part= new (std::nothrow) MDL_map_partition();
+    m_partitions.append(part);
+  }
+}
+
+
+/** Initialize the partition in the container with all MDL locks. */
+
+MDL_map_partition::MDL_map_partition()
+{
   mysql_mutex_init(key_MDL_map_mutex, &m_mutex, NULL);
   my_hash_init(&m_locks, &my_charset_bin, 16 /* FIXME */, 0, 0,
                mdl_locks_key, 0, 0);
-  m_global_lock= MDL_lock::create(&global_lock_key);
-  m_commit_lock= MDL_lock::create(&commit_lock_key);
-}
+};
 
 
 /**
-  Destroy the global hash containing all MDL locks.
+  Destroy the container for all MDL locks.
   @pre It must be empty.
 */
 
 void MDL_map::destroy()
 {
+  MDL_lock::destroy(m_global_lock);
+  MDL_lock::destroy(m_commit_lock);
+
+  while (m_partitions.elements() > 0)
+  {
+    MDL_map_partition *part= m_partitions.pop();
+    delete part;
+  }
+}
+
+
+/**
+  Destroy the partition in container for all MDL locks.
+  @pre It must be empty.
+*/
+
+MDL_map_partition::~MDL_map_partition()
+{
   DBUG_ASSERT(!m_locks.records);
   mysql_mutex_destroy(&m_mutex);
   my_hash_free(&m_locks);
-  MDL_lock::destroy(m_global_lock);
-  MDL_lock::destroy(m_commit_lock);
 
   MDL_object_lock *lock;
   while ((lock= m_unused_locks_cache.pop_front()))
@@ -712,13 +775,12 @@ void MDL_map::destroy()
 MDL_lock* MDL_map::find_or_insert(const MDL_key *mdl_key)
 {
   MDL_lock *lock;
-  my_hash_value_type hash_value;
 
   if (mdl_key->mdl_namespace() == MDL_key::GLOBAL ||
       mdl_key->mdl_namespace() == MDL_key::COMMIT)
   {
     /*
-      Avoid locking m_mutex when lock for GLOBAL or COMMIT namespace is
+      Avoid locking any m_mutex when lock for GLOBAL or COMMIT namespace is
       requested. Return pointer to pre-allocated MDL_lock instance instead.
       Such an optimization allows to save one mutex lock/unlock for any
       statement changing data.
@@ -736,8 +798,27 @@ MDL_lock* MDL_map::find_or_insert(const MDL_key *mdl_key)
     return lock;
   }
 
+  my_hash_value_type hash_value= m_partitions.at(0)->get_key_hash(mdl_key);
+  uint part_id= hash_value % mdl_locks_hash_partitions;
+  MDL_map_partition *part= m_partitions.at(part_id);
+
+  return part->find_or_insert(mdl_key, hash_value);
+}
+
 
-  hash_value= my_calc_hash(&m_locks, mdl_key->ptr(), mdl_key->length());
+/**
+  Find MDL_lock object corresponding to the key and hash value in
+  MDL_map partition, create it if it does not exist.
+
+  @retval non-NULL - Success. MDL_lock instance for the key with
+                     locked MDL_lock::m_rwlock.
+  @retval NULL     - Failure (OOM).
+*/
+
+MDL_lock* MDL_map_partition::find_or_insert(const MDL_key *mdl_key,
+                                            my_hash_value_type hash_value)
+{
+  MDL_lock *lock;
 
 retry:
   mysql_mutex_lock(&m_mutex);
@@ -770,7 +851,7 @@ retry:
     }
     else
     {
-      lock= MDL_lock::create(mdl_key);
+      lock= MDL_lock::create(mdl_key, this);
     }
 
     if (!lock || my_hash_insert(&m_locks, (uchar*)lock))
@@ -801,7 +882,7 @@ retry:
 
 
 /**
-  Release mdl_locks.m_mutex mutex and lock MDL_lock::m_rwlock for lock
+  Release MDL_map_partition::m_mutex mutex and lock MDL_lock::m_rwlock for lock
   object from the hash. Handle situation when object was released
   while we held no locks.
 
@@ -810,7 +891,7 @@ retry:
                   should re-try looking up MDL_lock object in the hash.
 */
 
-bool MDL_map::move_from_hash_to_lock_mutex(MDL_lock *lock)
+bool MDL_map_partition::move_from_hash_to_lock_mutex(MDL_lock *lock)
 {
   ulonglong version;
 
@@ -819,8 +900,8 @@ bool MDL_map::move_from_hash_to_lock_mutex(MDL_lock *lock)
 
   /*
     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.
+    MDL_map_partition::m_mutex under the condition it is present in the hash
+    and m_is_destroyed is FALSE.
   */
   lock->m_ref_usage++;
   /* Read value of the version counter under protection of m_mutex lock. */
@@ -892,28 +973,41 @@ void MDL_map::remove(MDL_lock *lock)
     return;
   }
 
+  lock->m_map_part->remove(lock);
+}
+
+
+/**
+  Destroy MDL_lock object belonging to specific MDL_map
+  partition or delegate this responsibility to whatever
+  thread that holds the last outstanding reference to it.
+*/
+
+void MDL_map_partition::remove(MDL_lock *lock)
+{
   mysql_mutex_lock(&m_mutex);
   my_hash_delete(&m_locks, (uchar*) lock);
   /*
     To let threads holding references to the MDL_lock object know that it was
     moved to the list of unused objects or destroyed, we increment the version
-    counter under protection of both MDL_map::m_mutex and MDL_lock::m_rwlock
-    locks. This allows us to read the version value while having either one
-    of those locks.
+    counter under protection of both MDL_map_partition::m_mutex and
+    MDL_lock::m_rwlock locks. This allows us to read the version value while
+    having either one of those locks.
   */
   lock->m_version++;
 
   if ((lock->key.mdl_namespace() != MDL_key::SCHEMA) &&
-      (m_unused_locks_cache.elements() < mdl_locks_cache_size))
+      (m_unused_locks_cache.elements() <
+       mdl_locks_cache_size/mdl_locks_hash_partitions))
   {
     /*
       This is an object of MDL_object_lock type and the cache of unused
       objects has not reached its maximum size yet. So instead of destroying
       object we move it to the list of unused objects to allow its later
       re-use with possibly different key. Any threads holding references to
-      this object (owning MDL_map::m_mutex or MDL_lock::m_rwlock) will notice
-      this thanks to the fact that we have changed the MDL_lock::m_version
-      counter.
+      this object (owning MDL_map_partition::m_mutex or MDL_lock::m_rwlock)
+      will notice this thanks to the fact that we have changed the
+      MDL_lock::m_version counter.
     */
     DBUG_ASSERT(lock->key.mdl_namespace() != MDL_key::GLOBAL &&
                 lock->key.mdl_namespace() != MDL_key::COMMIT);
@@ -930,8 +1024,8 @@ void MDL_map::remove(MDL_lock *lock)
       has the responsibility to release it.
 
       Setting of m_is_destroyed to TRUE while holding _both_
-      mdl_locks.m_mutex and MDL_lock::m_rwlock mutexes transfers the
-      protection of m_ref_usage from mdl_locks.m_mutex to
+      MDL_map_partition::m_mutex and MDL_lock::m_rwlock mutexes transfers
+      the protection of m_ref_usage from MDL_map_partition::m_mutex to
       MDL_lock::m_rwlock while removal of the object from the hash
       (and cache of unused objects) makes it read-only. Therefore
       whoever acquires MDL_lock::m_rwlock next will see the most up
@@ -961,7 +1055,8 @@ void MDL_map::remove(MDL_lock *lock)
 */
 
 MDL_context::MDL_context()
-  : m_thd(NULL),
+  :
+  m_owner(NULL),
   m_needs_thr_lock_abort(FALSE),
   m_waiting_for(NULL)
 {
@@ -983,9 +1078,9 @@ MDL_context::MDL_context()
 
 void MDL_context::destroy()
 {
-  DBUG_ASSERT(m_tickets[MDL_STATEMENT].is_empty() &&
-              m_tickets[MDL_TRANSACTION].is_empty() &&
-              m_tickets[MDL_EXPLICIT].is_empty());
+  DBUG_ASSERT(m_tickets[MDL_STATEMENT].is_empty());
+  DBUG_ASSERT(m_tickets[MDL_TRANSACTION].is_empty());
+  DBUG_ASSERT(m_tickets[MDL_EXPLICIT].is_empty());
 
   mysql_prlock_destroy(&m_LOCK_waiting_for);
 }
@@ -1050,16 +1145,17 @@ void MDL_request::init(const MDL_key *key_arg,
   @note Also chooses an MDL_lock descendant appropriate for object namespace.
 */
 
-inline MDL_lock *MDL_lock::create(const MDL_key *mdl_key)
+inline MDL_lock *MDL_lock::create(const MDL_key *mdl_key,
+                                  MDL_map_partition *map_part)
 {
   switch (mdl_key->mdl_namespace())
   {
     case MDL_key::GLOBAL:
     case MDL_key::SCHEMA:
     case MDL_key::COMMIT:
-      return new MDL_scoped_lock(mdl_key);
+      return new (std::nothrow) MDL_scoped_lock(mdl_key, map_part);
     default:
-      return new MDL_object_lock(mdl_key);
+      return new (std::nothrow) MDL_object_lock(mdl_key, map_part);
   }
 }
 
@@ -1084,7 +1180,8 @@ MDL_ticket *MDL_ticket::create(MDL_context *ctx_arg, enum_mdl_type type_arg
 #endif
                                )
 {
-  return new MDL_ticket(ctx_arg, type_arg
+  return new (std::nothrow)
+             MDL_ticket(ctx_arg, type_arg
 #ifndef DBUG_OFF
                         , duration_arg
 #endif
@@ -1177,6 +1274,7 @@ void MDL_wait::reset_status()
 /**
   Wait for the status to be assigned to this wait slot.
 
+  @param owner           MDL context owner.
   @param abs_timeout     Absolute time after which waiting should stop.
   @param set_status_on_timeout TRUE  - If in case of timeout waiting
                                        context should close the wait slot by
@@ -1188,7 +1286,7 @@ void MDL_wait::reset_status()
 */
 
 MDL_wait::enum_wait_status
-MDL_wait::timed_wait(THD *thd, struct timespec *abs_timeout,
+MDL_wait::timed_wait(MDL_context_owner *owner, struct timespec *abs_timeout,
                      bool set_status_on_timeout,
                      const PSI_stage_info *wait_state_name)
 {
@@ -1199,16 +1297,16 @@ MDL_wait::timed_wait(THD *thd, struct timespec *abs_timeout,
 
   mysql_mutex_lock(&m_LOCK_wait_status);
 
-  THD_ENTER_COND(thd, &m_COND_wait_status, &m_LOCK_wait_status,
-                  wait_state_name, & old_stage);
-  thd_wait_begin(thd, THD_WAIT_META_DATA_LOCK);
-  while (!m_wait_status && !thd->killed &&
+  owner->ENTER_COND(&m_COND_wait_status, &m_LOCK_wait_status,
+                    wait_state_name, & old_stage);
+  thd_wait_begin(NULL, THD_WAIT_META_DATA_LOCK);
+  while (!m_wait_status && !owner->is_killed() &&
          wait_result != ETIMEDOUT && wait_result != ETIME)
   {
     wait_result= mysql_cond_timedwait(&m_COND_wait_status, &m_LOCK_wait_status,
                                       abs_timeout);
   }
-  thd_wait_end(thd);
+  thd_wait_end(NULL);
 
   if (m_wait_status == EMPTY)
   {
@@ -1224,14 +1322,14 @@ MDL_wait::timed_wait(THD *thd, struct timespec *abs_timeout,
       false, which means that the caller intends to restart the
       wait.
     */
-    if (thd->killed)
+    if (owner->is_killed())
       m_wait_status= KILLED;
     else if (set_status_on_timeout)
       m_wait_status= TIMEOUT;
   }
   result= m_wait_status;
 
-  thd->EXIT_COND(& old_stage);
+  owner->EXIT_COND(& old_stage);
 
   DBUG_RETURN(result);
 }
@@ -1445,7 +1543,7 @@ void MDL_lock::reschedule_waiters()
 
              | Type of active   |
      Request |   scoped lock    |
-      type   | IS(**) IX   S  X |
+      type   | IS(*)  IX   S  X |
     ---------+------------------+
     IS       |  +      +   +  + |
     IX       |  +      +   -  - |
@@ -1458,7 +1556,7 @@ void MDL_lock::reschedule_waiters()
 
              |    Pending      |
      Request |  scoped lock    |
-      type   | IS(**) IX  S  X |
+      type   | IS(*)  IX  S  X |
     ---------+-----------------+
     IS       |  +      +  +  + |
     IX       |  +      +  -  - |
@@ -2010,9 +2108,9 @@ void MDL_object_lock::notify_conflicting_locks(MDL_context *ctx)
         lock or some other non-MDL resource we might need to wake it up
         by calling code outside of MDL.
       */
-      mysql_notify_thread_having_shared_lock(ctx->get_thd(),
-                                 conflicting_ctx->get_thd(),
-                                 conflicting_ctx->get_needs_thr_lock_abort());
+      ctx->get_owner()->
+        notify_shared_lock(conflicting_ctx->get_owner(),
+                           conflicting_ctx->get_needs_thr_lock_abort());
     }
   }
 }
@@ -2042,9 +2140,9 @@ void MDL_scoped_lock::notify_conflicting_locks(MDL_context *ctx)
         insert delayed. We need to kill such threads in order to get
         global shared lock. We do this my calling code outside of MDL.
       */
-      mysql_notify_thread_having_shared_lock(ctx->get_thd(),
-                                 conflicting_ctx->get_thd(),
-                                 conflicting_ctx->get_needs_thr_lock_abort());
+      ctx->get_owner()->
+        notify_shared_lock(conflicting_ctx->get_owner(),
+                           conflicting_ctx->get_needs_thr_lock_abort());
     }
   }
 }
@@ -2117,7 +2215,7 @@ MDL_context::acquire_lock(MDL_request *mdl_request, ulong lock_wait_timeout)
   will_wait_for(ticket);
 
   /* There is a shared or exclusive lock on the object. */
-  DEBUG_SYNC(m_thd, "mdl_acquire_lock_wait");
+  DEBUG_SYNC(get_thd(), "mdl_acquire_lock_wait");
 
   find_deadlock();
 
@@ -2130,7 +2228,7 @@ MDL_context::acquire_lock(MDL_request *mdl_request, ulong lock_wait_timeout)
     while (cmp_timespec(abs_shortwait, abs_timeout) <= 0)
     {
       /* abs_timeout is far away. Wait a short while and notify locks. */
-      wait_status= m_wait.timed_wait(m_thd, &abs_shortwait, FALSE,
+      wait_status= m_wait.timed_wait(m_owner, &abs_shortwait, FALSE,
                                      mdl_request->key.get_wait_state_name());
 
       if (wait_status != MDL_wait::EMPTY)
@@ -2142,11 +2240,11 @@ MDL_context::acquire_lock(MDL_request *mdl_request, ulong lock_wait_timeout)
       set_timespec(abs_shortwait, 1);
     }
     if (wait_status == MDL_wait::EMPTY)
-      wait_status= m_wait.timed_wait(m_thd, &abs_timeout, TRUE,
+      wait_status= m_wait.timed_wait(m_owner, &abs_timeout, TRUE,
                                      mdl_request->key.get_wait_state_name());
   }
   else
-    wait_status= m_wait.timed_wait(m_thd, &abs_timeout, TRUE,
+    wait_status= m_wait.timed_wait(m_owner, &abs_timeout, TRUE,
                                    mdl_request->key.get_wait_state_name());
 
   done_waiting_for();
@@ -2231,8 +2329,7 @@ bool MDL_context::acquire_locks(MDL_request_list *mdl_requests,
   /* Sort requests according to MDL_key. */
   if (! (sort_buf= (MDL_request **)my_malloc(req_count *
                                              sizeof(MDL_request*),
-                                             MYF(MY_WME |
-                                                 MY_THREAD_SPECIFIC))))
+                                             MYF(MY_WME))))
     DBUG_RETURN(TRUE);
 
   for (p_req= sort_buf; p_req < sort_buf + req_count; p_req++)
diff --git a/sql/mdl.h b/sql/mdl.h
index 82d764bc66c..ddbd55ac467 100644
--- a/sql/mdl.h
+++ b/sql/mdl.h
@@ -1,6 +1,6 @@
 #ifndef MDL_H
 #define MDL_H
-/* Copyright (c) 2009, 2011, Oracle and/or its affiliates. All rights reserved.
+/* Copyright (c) 2009, 2012, Oracle and/or its affiliates. All rights reserved.
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -12,8 +12,8 @@
    GNU General Public License for more details.
 
    You should have received a copy of the GNU General Public License
-   along with this program; if not, write to the Free Software
-   Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301  USA */
+   along with this program; if not, write to the Free Software Foundation,
+   51 Franklin Street, Suite 500, Boston, MA 02110-1335 USA */
 
 #if defined(__IBMC__) || defined(__IBMCPP__)
 /* Further down, "next_in_lock" and "next_in_context" have the same type,
@@ -26,10 +26,11 @@
 
 #include "sql_plist.h"
 #include <my_sys.h>
-#include <my_pthread.h>
 #include <m_string.h>
 #include <mysql_com.h>
 
+#include <algorithm>
+
 class THD;
 
 class MDL_context;
@@ -55,6 +56,67 @@ class MDL_ticket;
 #define EXIT_COND(S) exit_cond(S, __func__, __FILE__, __LINE__)
 
 /**
+   An interface to separate the MDL module from the THD, and the rest of the
+   server code.
+ */
+
+class MDL_context_owner
+{
+public:
+  virtual ~MDL_context_owner() {}
+
+  /**
+    Enter a condition wait.
+    For @c enter_cond() / @c exit_cond() to work the mutex must be held before
+    @c enter_cond(); this mutex is then released by @c exit_cond().
+    Usage must be: lock mutex; enter_cond(); your code; exit_cond().
+    @param cond the condition to wait on
+    @param mutex the associated mutex
+    @param [in] stage the stage to enter, or NULL
+    @param [out] old_stage the previous stage, or NULL
+    @param src_function function name of the caller
+    @param src_file file name of the caller
+    @param src_line line number of the caller
+    @sa ENTER_COND(), THD::enter_cond()
+    @sa EXIT_COND(), THD::exit_cond()
+  */
+  virtual void enter_cond(mysql_cond_t *cond, mysql_mutex_t *mutex,
+                          const PSI_stage_info *stage, PSI_stage_info *old_stage,
+                          const char *src_function, const char *src_file,
+                          int src_line) = 0;
+
+  /**
+    @def EXIT_COND(S)
+    End a wait on a condition
+    @param [in] stage the new stage to enter
+    @param src_function function name of the caller
+    @param src_file file name of the caller
+    @param src_line line number of the caller
+    @sa ENTER_COND(), THD::enter_cond()
+    @sa EXIT_COND(), THD::exit_cond()
+  */
+  virtual void exit_cond(const PSI_stage_info *stage,
+                         const char *src_function, const char *src_file,
+                         int src_line) = 0;
+  /**
+     Has the owner thread been killed?
+   */
+  virtual int  is_killed() = 0;
+
+  /**
+     This one is only used for DEBUG_SYNC.
+     (Do not use it to peek/poke into other parts of THD.)
+   */
+  virtual THD* get_thd() = 0;
+
+  /**
+     @see THD::notify_shared_lock()
+   */
+  virtual bool notify_shared_lock(MDL_context_owner *in_use,
+                                  bool needs_thr_lock_abort) = 0;
+};
+
+/**
   Type of metadata lock request.
 
   @sa Comments for MDL_object_lock::can_grant_lock() and
@@ -273,8 +335,16 @@ public:
                     const char *db, const char *name)
   {
     m_ptr[0]= (char) mdl_namespace;
-    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);
+    /*
+      It is responsibility of caller to ensure that db and object names
+      are not longer than NAME_LEN. Still we play safe and try to avoid
+      buffer overruns.
+    */
+    DBUG_ASSERT(strlen(db) <= NAME_LEN && strlen(name) <= NAME_LEN);
+    m_db_name_length= static_cast<uint16>(strmake(m_ptr + 1, db, NAME_LEN) -
+                                          m_ptr - 1);
+    m_length= static_cast<uint16>(strmake(m_ptr + m_db_name_length + 2, name,
+                                          NAME_LEN) - m_ptr + 1);
   }
   void mdl_key_init(const MDL_key *rhs)
   {
@@ -297,7 +367,8 @@ public:
       character set is utf-8, we can safely assume that no
       character starts with a zero byte.
     */
-    return memcmp(m_ptr, rhs->m_ptr, MY_MIN(m_length, rhs->m_length));
+    using std::min;
+    return memcmp(m_ptr, rhs->m_ptr, min(m_length, rhs->m_length));
   }
 
   MDL_key(const MDL_key *rhs)
@@ -624,7 +695,7 @@ public:
   bool set_status(enum_wait_status result_arg);
   enum_wait_status get_status();
   void reset_status();
-  enum_wait_status timed_wait(THD *thd,
+  enum_wait_status timed_wait(MDL_context_owner *owner,
                               struct timespec *abs_timeout,
                               bool signal_timeout,
                               const PSI_stage_info *wait_state_name);
@@ -706,7 +777,7 @@ public:
   void release_transactional_locks();
   void rollback_to_savepoint(const MDL_savepoint &mdl_savepoint);
 
-  inline THD *get_thd() const { return m_thd; }
+  MDL_context_owner *get_owner() { return m_owner; }
 
   /** @pre Only valid if we started waiting for lock. */
   inline uint get_deadlock_weight() const
@@ -719,7 +790,7 @@ public:
                     already has received some signal or closed
                     signal slot.
   */
-  void init(THD *thd_arg) { m_thd= thd_arg; }
+  void init(MDL_context_owner *arg) { m_owner= arg; }
 
   void set_needs_thr_lock_abort(bool needs_thr_lock_abort)
   {
@@ -799,7 +870,7 @@ private:
       involved schemas and global intention exclusive lock.
   */
   Ticket_list m_tickets[MDL_DURATION_END];
-  THD *m_thd;
+  MDL_context_owner *m_owner;
   /**
     TRUE -  if for this context we will break protocol and try to
             acquire table-level locks while having only S lock on
@@ -828,6 +899,7 @@ private:
    */
   MDL_wait_for_subgraph *m_waiting_for;
 private:
+  THD *get_thd() const { return m_owner->get_thd(); }
   MDL_ticket *find_ticket(MDL_request *mdl_req,
                           enum_mdl_duration *duration);
   void release_locks_stored_before(enum_mdl_duration duration, MDL_ticket *sentinel);
@@ -872,8 +944,6 @@ private:
 void mdl_init();
 void mdl_destroy();
 
-extern bool mysql_notify_thread_having_shared_lock(THD *thd, THD *in_use,
-                                                   bool needs_thr_lock_abort);
 
 #ifndef DBUG_OFF
 extern mysql_mutex_t LOCK_open;
@@ -888,6 +958,14 @@ extern ulong mdl_locks_cache_size;
 static const ulong MDL_LOCKS_CACHE_SIZE_DEFAULT = 1024;
 
 /*
+  Start-up parameter for the number of partitions of the hash
+  containing all the MDL_lock objects and a constant for
+  its default value.
+*/
+extern ulong mdl_locks_hash_partitions;
+static const ulong MDL_LOCKS_HASH_PARTITIONS_DEFAULT = 8;
+
+/*
   Metadata locking subsystem tries not to grant more than
   max_write_lock_count high-prio, strong locks successively,
   to avoid starving out weak, low-prio locks.
diff --git a/sql/sql_array.h b/sql/sql_array.h
index f07126bc0ef..9ac27cca63b 100644
--- a/sql/sql_array.h
+++ b/sql/sql_array.h
@@ -124,6 +124,17 @@ public:
     return (insert_dynamic(&array, (uchar*)&el));
   }
 
+  /// Pops the last element. Does nothing if array is empty.
+  Elem& pop()
+  {
+    return *((Elem*)pop_dynamic(&array));
+  }
+
+  void del(uint idx)
+  {
+    delete_dynamic_element(&array, idx);
+  }
+
   int elements()
   {
     return array.elements;
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index 28f4b7e0ef8..0680c8f1b23 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -4860,55 +4860,20 @@ lock_table_names(THD *thd,
       continue;
     }
 
-    /*
-      Write lock on normal tables is not allowed in a read only transaction.
-    */
+    /* Write lock on normal tables is not allowed in a read only transaction. */
     if (thd->tx_read_only)
     {
       my_error(ER_CANT_EXECUTE_IN_READ_ONLY_TRANSACTION, MYF(0));
       DBUG_RETURN(true);
     }
 
-    if (! (flags & MYSQL_OPEN_SKIP_SCOPED_MDL_LOCK) && schema_set.insert(table))
+    if (! (flags & MYSQL_OPEN_SKIP_SCOPED_MDL_LOCK) &&
+        schema_set.insert(table))
       DBUG_RETURN(TRUE);
 
     mdl_requests.push_front(&table->mdl_request);
   }
 
-  if (! (flags & MYSQL_OPEN_SKIP_SCOPED_MDL_LOCK) &&
-      ! mdl_requests.is_empty())
-  {
-    /*
-      Scoped locks: Take intention exclusive locks on all involved
-      schemas.
-    */
-    Hash_set<TABLE_LIST, schema_set_get_key>::Iterator it(schema_set);
-    while ((table= it++))
-    {
-      MDL_request *schema_request= new (thd->mem_root) MDL_request;
-      if (schema_request == NULL)
-        return TRUE;
-      schema_request->init(MDL_key::SCHEMA, table->db, "",
-                           MDL_INTENTION_EXCLUSIVE,
-                           MDL_TRANSACTION);
-      mdl_requests.push_front(schema_request);
-    }
-
-    /*
-      Protect this statement against concurrent global read lock
-      by acquiring global intention exclusive lock with statement
-      duration.
-    */
-    if (thd->global_read_lock.can_acquire_protection())
-      return TRUE;
-    global_request.init(MDL_key::GLOBAL, "", "", MDL_INTENTION_EXCLUSIVE,
-                        MDL_STATEMENT);
-    mdl_requests.push_front(&global_request);
-  }
-
-  if (thd->mdl_context.acquire_locks(&mdl_requests, lock_wait_timeout))
-    return TRUE;
-
   if (mdl_requests.is_empty())
     DBUG_RETURN(FALSE);
 
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index 253e2388d65..33794e0fd7a 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -62,6 +62,7 @@
 #include "debug_sync.h"
 #include "sql_parse.h"                          // is_update_query
 #include "sql_callback.h"
+#include "lock.h"
 #include "sql_connect.h"
 
 /*
@@ -1788,6 +1789,46 @@ void THD::disconnect()
 }
 
 
+bool THD::notify_shared_lock(MDL_context_owner *ctx_in_use,
+                             bool needs_thr_lock_abort)
+{
+  THD *in_use= ctx_in_use->get_thd();
+  bool signalled= FALSE;
+
+  if ((in_use->system_thread & SYSTEM_THREAD_DELAYED_INSERT) &&
+      !in_use->killed)
+  {
+    in_use->killed= KILL_CONNECTION;
+    mysql_mutex_lock(&in_use->mysys_var->mutex);
+    if (in_use->mysys_var->current_cond)
+      mysql_cond_broadcast(in_use->mysys_var->current_cond);
+    mysql_mutex_unlock(&in_use->mysys_var->mutex);
+    signalled= TRUE;
+  }
+
+  if (needs_thr_lock_abort)
+  {
+    mysql_mutex_lock(&in_use->LOCK_thd_data);
+    for (TABLE *thd_table= in_use->open_tables;
+         thd_table ;
+         thd_table= thd_table->next)
+    {
+      /*
+        Check for TABLE::needs_reopen() is needed since in some places we call
+        handler::close() for table instance (and set TABLE::db_stat to 0)
+        and do not remove such instances from the THD::open_tables
+        for some time, during which other thread can see those instances
+        (e.g. see partitioning code).
+      */
+      if (!thd_table->needs_reopen())
+        signalled|= mysql_lock_abort_for_thread(this, thd_table);
+    }
+    mysql_mutex_unlock(&in_use->LOCK_thd_data);
+  }
+  return signalled;
+}
+
+
 /*
   Get error number for killed state
   Note that the error message can't have any parameters.
diff --git a/sql/sql_class.h b/sql/sql_class.h
index c2b2723f78e..81a4b52f887 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -1571,6 +1571,7 @@ void dbug_serve_apcs(THD *thd, int n_calls);
 */
 
 class THD :public Statement,
+           public MDL_context_owner,
            public Open_tables_state
 {
 private:
@@ -2583,6 +2584,37 @@ public:
     mysql_mutex_unlock(&mysys_var->mutex);
     return;
   }
+  virtual int is_killed() { return killed; }
+  virtual THD* get_thd() { return this; }
+
+  /**
+    A callback to the server internals that is used to address
+    special cases of the locking protocol.
+    Invoked when acquiring an exclusive lock, for each thread that
+    has a conflicting shared metadata lock.
+
+    This function:
+    - aborts waiting of the thread on a data lock, to make it notice
+      the pending exclusive lock and back off.
+    - if the thread is an INSERT DELAYED thread, sends it a KILL
+      signal to terminate it.
+
+    @note This function does not wait for the thread to give away its
+          locks. Waiting is done outside for all threads at once.
+
+    @param ctx_in_use           The MDL context owner (thread) to wake up.
+    @param needs_thr_lock_abort Indicates that to wake up thread
+                                this call needs to abort its waiting
+                                on table-level lock.
+
+    @retval  TRUE  if the thread was woken up
+    @retval  FALSE otherwise.
+   */
+  virtual bool notify_shared_lock(MDL_context_owner *ctx_in_use,
+                                  bool needs_thr_lock_abort);
+
+  // End implementation of MDL_context_owner interface.
+
   inline bool is_strict_mode() const
   {
     return variables.sql_mode & (MODE_STRICT_TRANS_TABLES |
diff --git a/sql/sql_prepare.cc b/sql/sql_prepare.cc
index a8ecdc0718b..2460f15ab62 100644
--- a/sql/sql_prepare.cc
+++ b/sql/sql_prepare.cc
@@ -3520,7 +3520,6 @@ Prepared_statement::execute_loop(String *expanded_query,
   Reprepare_observer reprepare_observer;
   bool error;
   int reprepare_attempt= 0;
-  bool need_set_parameters= true;
 
   /* Check if we got an error when sending long data */
   if (state == Query_arena::STMT_ERROR)
@@ -3529,20 +3528,19 @@ Prepared_statement::execute_loop(String *expanded_query,
     return TRUE;
   }
 
-reexecute:
-  if (need_set_parameters &&
-      set_parameters(expanded_query, packet, packet_end))
+  if (set_parameters(expanded_query, packet, packet_end))
     return TRUE;
 
-  /*
-    if set_parameters() has generated warnings,
-    we need to repeat it when reexecuting, to recreate these
-    warnings.
-  */
-  need_set_parameters= thd->get_stmt_da()->statement_warn_count();
-
-  reprepare_observer.reset_reprepare_observer();
+#ifdef NOT_YET_FROM_MYSQL_5_6
+  if (unlikely(thd->security_ctx->password_expired && 
+               !lex->is_change_password))
+  {
+    my_error(ER_MUST_CHANGE_PASSWORD, MYF(0));
+    return true;
+  }
+#endif
 
+reexecute:
   /*
     If the free_list is not empty, we'll wrongly free some externally
     allocated items when cleaning up after validation of the prepared
@@ -3556,18 +3554,20 @@ reexecute:
     the observer method will be invoked to push an error into
     the error stack.
   */
-  if (sql_command_flags[lex->sql_command] &
-      CF_REEXECUTION_FRAGILE)
+
+  if (sql_command_flags[lex->sql_command] & CF_REEXECUTION_FRAGILE)
   {
+    reprepare_observer.reset_reprepare_observer();
     DBUG_ASSERT(thd->m_reprepare_observer == NULL);
-    thd->m_reprepare_observer = &reprepare_observer;
+    thd->m_reprepare_observer= &reprepare_observer;
   }
 
   error= execute(expanded_query, open_cursor) || thd->is_error();
 
   thd->m_reprepare_observer= NULL;
 
-  if (error && !thd->is_fatal_error && !thd->killed &&
+  if ((sql_command_flags[lex->sql_command] & CF_REEXECUTION_FRAGILE) &&
+      error && !thd->is_fatal_error && !thd->killed &&
       reprepare_observer.is_invalidated() &&
       reprepare_attempt++ < MAX_REPREPARE_ATTEMPTS)
   {
diff --git a/sql/sys_vars.cc b/sql/sys_vars.cc
index 42e44b4706d..4c0aeef9c06 100644
--- a/sql/sys_vars.cc
+++ b/sql/sys_vars.cc
@@ -1331,6 +1331,12 @@ static Sys_var_ulong Sys_metadata_locks_cache_size(
        VALID_RANGE(1, 1024*1024), DEFAULT(MDL_LOCKS_CACHE_SIZE_DEFAULT),
        BLOCK_SIZE(1));
 
+static Sys_var_ulong Sys_metadata_locks_hash_instances(
+       "metadata_locks_hash_instances", "Number of metadata locks hash instances",
+       READ_ONLY GLOBAL_VAR(mdl_locks_hash_partitions), CMD_LINE(REQUIRED_ARG),
+       VALID_RANGE(1, 1024), DEFAULT(MDL_LOCKS_HASH_PARTITIONS_DEFAULT),
+       BLOCK_SIZE(1));
+
 /*
   "pseudo_thread_id" variable used in the test suite to detect 32/64bit
   systems.  If you change it to something else then ulong then fix the tests