Backport of revno ## 2617.31.1, 2617.31.3, 2617.31.4, 2617.31.5,

2617.31.12, 2617.31.15, 2617.31.15, 2617.31.16, 2617.43.1
- initial changeset that introduced the fix for 
Bug#989 and follow up fixes for all test suite failures
introduced in the initial changeset. 
------------------------------------------------------------
revno: 2617.31.1
committer: Davi Arnaut <Davi.Arnaut@Sun.COM>
branch nick: 4284-6.0
timestamp: Fri 2009-03-06 19:17:00 -0300
message:
Bug#989: If DROP TABLE while there's an active transaction, wrong binlog order
WL#4284: Transactional DDL locking

Currently the MySQL server does not keep metadata locks on
schema objects for the duration of a transaction, thus failing
to guarantee the integrity of the schema objects being used
during the transaction and to protect then from concurrent
DDL operations. This also poses a problem for replication as
a DDL operation might be replicated even thought there are
active transactions using the object being modified.

The solution is to defer the release of metadata locks until
a active transaction is either committed or rolled back. This
prevents other statements from modifying the table for the
entire duration of the transaction. This provides commitment
ordering for guaranteeing serializability across multiple
transactions.

- Incompatible change:

If MySQL's metadata locking system encounters a lock conflict,
the usual schema is to use the try and back-off technique to
avoid deadlocks -- this schema consists in releasing all locks
and trying to acquire them all in one go.

But in a transactional context this algorithm can't be utilized
as its not possible to release locks acquired during the course
of the transaction without breaking the transaction commitments.
To avoid deadlocks in this case, the ER_LOCK_DEADLOCK will be
returned if a lock conflict is encountered during a transaction.

Let's consider an example:

A transaction has two statements that modify table t1, then table
t2, and then commits. The first statement of the transaction will
acquire a shared metadata lock on table t1, and it will be kept
utill COMMIT to ensure serializability.

At the moment when the second statement attempts to acquire a
shared metadata lock on t2, a concurrent ALTER or DROP statement
might have locked t2 exclusively. The prescription of the current
locking protocol is that the acquirer of the shared lock backs off
-- gives up all his current locks and retries. This implies that
the entire multi-statement transaction has to be rolled back.

- Incompatible change:

FLUSH commands such as FLUSH PRIVILEGES and FLUSH TABLES WITH READ
LOCK won't cause locked tables to be implicitly unlocked anymore.


mysql-test/extra/binlog_tests/drop_table.test:
  Add test case for Bug#989.
mysql-test/extra/binlog_tests/mix_innodb_myisam_binlog.test:
  Fix test case to reflect the fact that transactions now hold
  metadata locks for the duration of a transaction.
mysql-test/include/mix1.inc:
  Fix test case to reflect the fact that transactions now hold
  metadata locks for the duration of a transaction.
mysql-test/include/mix2.inc:
  Fix test case to reflect the fact that transactions now hold
  metadata locks for the duration of a transaction.
mysql-test/r/flush_block_commit.result:
  Update test case result (WL#4284).
mysql-test/r/flush_block_commit_notembedded.result:
  Update test case result (WL#4284).
mysql-test/r/innodb.result:
  Update test case result (WL#4284).
mysql-test/r/innodb_mysql.result:
  Update test case result (WL#4284).
mysql-test/r/lock.result:
  Add test case result for an effect of WL#4284/Bug#989
  (all locks should be released when a connection terminates).
mysql-test/r/mix2_myisam.result:
  Update test case result (effects of WL#4284/Bug#989).
mysql-test/r/not_embedded_server.result:
  Update test case result (effects of WL#4284/Bug#989).
  Add a test case for interaction of WL#4284 and FLUSH PRIVILEGES.
mysql-test/r/partition_innodb_semi_consistent.result:
  Update test case result (effects of WL#4284/Bug#989).
mysql-test/r/partition_sync.result:
  Temporarily disable the test case for Bug#43867,
  which will be fixed by a subsequent backport.
mysql-test/r/ps.result:
  Add a test case for effect of PREPARE on transactional
  locks: we take a savepoint at beginning of PREAPRE
  and release it at the end. Thus PREPARE does not 
  accumulate metadata locks (Bug#989/WL#4284).
mysql-test/r/read_only_innodb.result:
  Update test case result (effects of WL#4284/Bug#989).
mysql-test/suite/binlog/r/binlog_row_drop_tbl.result:
  Add a test case result (WL#4284/Bug#989).
mysql-test/suite/binlog/r/binlog_row_mix_innodb_myisam.result:
  Update test case result (effects of WL#4284/Bug#989).
mysql-test/suite/binlog/r/binlog_stm_drop_tbl.result:
  Add a test case result (WL#4284/Bug#989).
mysql-test/suite/binlog/r/binlog_stm_mix_innodb_myisam.result:
  Update test case result (effects of WL#4284/Bug#989).
mysql-test/suite/binlog/r/binlog_unsafe.result:
  A side effect of Bug#989 -- slightly different table map ids.
mysql-test/suite/binlog/t/binlog_row_drop_tbl.test:
  Add a test case for WL#4284/Bug#989.
mysql-test/suite/binlog/t/binlog_stm_drop_tbl.test:
  Add a test case for WL#4284/Bug#989.
mysql-test/suite/binlog/t/binlog_stm_row.test:
  Update to the new state name. This
  is actually a follow up to another patch for WL#4284, 
  that changes Locked thread state to Table lock.
mysql-test/suite/ndb/r/ndb_index_ordered.result:
  Remove result for disabled part of the test case.
mysql-test/suite/ndb/t/disabled.def:
  Temporarily disable a test case (Bug#45621).
mysql-test/suite/ndb/t/ndb_index_ordered.test:
  Disable a part of a test case (needs update to
  reflect semantics of Bug#989).
mysql-test/suite/rpl/t/disabled.def:
  Disable tests made meaningless by transactional metadata
  locking.
mysql-test/suite/sys_vars/r/autocommit_func.result:
  Add a commit (Bug#989).
mysql-test/suite/sys_vars/t/autocommit_func.test:
  Add a commit (Bug#989).
mysql-test/t/flush_block_commit.test:
  Fix test case to reflect the fact that transactions now hold
  metadata locks for the duration of a transaction.
mysql-test/t/flush_block_commit_notembedded.test:
  Fix test case to reflect the fact that transactions now hold
  metadata locks for the duration of a transaction.
  Add a test case for transaction-scope locks and the global
  read lock (Bug#989/WL#4284).
mysql-test/t/innodb.test:
  Fix test case to reflect the fact that transactions now hold
  metadata locks for the duration of a transaction
  (effects of Bug#989/WL#4284).
mysql-test/t/lock.test:
  Add a test case for Bug#989/WL#4284.
mysql-test/t/not_embedded_server.test:
  Add a test case for Bug#989/WL#4284.
mysql-test/t/partition_innodb_semi_consistent.test:
  Replace TRUNCATE with DELETE, to not issue
  an implicit commit of a transaction, and not depend
  on metadata locks.
mysql-test/t/partition_sync.test:
  Temporarily disable the test case for Bug#43867,
  which needs a fix to be backported from 6.0.
mysql-test/t/ps.test:
  Add a test case for semantics of PREPARE and transaction-scope
  locks: metadata locks on tables used in PREPARE are enclosed into a
  temporary savepoint, taken at the beginning of PREPARE,
  and released at the end. Thus PREPARE does not effect
  what locks a transaction owns.
mysql-test/t/read_only_innodb.test:
  Fix test case to reflect the fact that transactions now hold
  metadata locks for the duration of a transaction 
  (Bug#989/WL#4284).
  
  Wait for the read_only statement to actually flush tables before
  sending other concurrent statements that depend on its state.
mysql-test/t/xa.test:
  Fix test case to reflect the fact that transactions now hold
  metadata locks for the duration of a transaction 
  (Bug#989/WL#4284).
sql/ha_ndbcluster_binlog.cc:
  Backport bits of changes of ha_ndbcluster_binlog.cc
  from 6.0, to fix the failing binlog test suite with
  WL#4284. WL#4284 implementation does not work
  with 5.1 implementation of ndbcluster binlog index.
sql/log_event.cc:
  Release metadata locks after issuing a commit.
sql/mdl.cc:
  Style changes (WL#4284).
sql/mysql_priv.h:
  Rename parameter to match the name used in the definition (WL#4284).
sql/rpl_injector.cc:
  Release metadata locks on commit (WL#4284).
sql/rpl_rli.cc:
  Remove assert made meaningless, metadata locks are released
  at the end of the transaction.
sql/set_var.cc:
  Close tables and release locks if autocommit mode is set.
sql/slave.cc:
  Release metadata locks after a rollback.
sql/sql_acl.cc:
  Don't implicitly unlock locked tables. Issue a implicit commit
  at the end and unlock tables.
sql/sql_base.cc:
  Defer the release of metadata locks when closing tables
  if not required to.
  Issue a deadlock error if the locking protocol requires
  that a transaction re-acquire its locks.
  
  Release metadata locks when closing tables for reopen.
sql/sql_class.cc:
  Release metadata locks if the thread is killed.
sql/sql_parse.cc:
  Release metadata locks after implicitly committing a active
  transaction, or after explicit commits or rollbacks.
sql/sql_plugin.cc:
  
  Allocate MDL request on the stack as the use of the table
  is contained within the function. It will be removed from
  the context once close_thread_tables is called at the end
  of the function.
sql/sql_prepare.cc:
  The problem is that the prepare phase of the CREATE TABLE
  statement takes a exclusive metadata lock lock and this can
  cause a self-deadlock the thread already holds a shared lock
  on the table being that should be created.
  
  The solution is to make the prepare phase take a shared
  metadata lock when preparing a CREATE TABLE statement. The
  execution of the statement will still acquire a exclusive
  lock, but won't cause any problem as it issues a implicit
  commit.
  
  After some discussions with stakeholders it has been decided that
  metadata locks acquired during a PREPARE statement must be released
  once the statement is prepared even if it is prepared within a multi
  statement transaction.
sql/sql_servers.cc:
  Don't implicitly unlock locked tables. Issue a implicit commit
  at the end and unlock tables.
sql/sql_table.cc:
  Close table and release metadata locks after a admin operation.
sql/table.h:
  The problem is that the prepare phase of the CREATE TABLE
  statement takes a exclusive metadata lock lock and this can
  cause a self-deadlock the thread already holds a shared lock
  on the table being that should be created.
  
  The solution is to make the prepare phase take a shared
  metadata lock when preparing a CREATE TABLE statement. The
  execution of the statement will still acquire a exclusive
  lock, but won't cause any problem as it issues a implicit
  commit.
sql/transaction.cc:
  Release metadata locks after the implicitly committed due
  to a new transaction being started. Also, release metadata
  locks acquired after a savepoint if the transaction is rolled
  back to the save point.
  
  The problem is that in some cases transaction-long metadata locks
  could be released before the transaction was committed. This could
  happen when a active transaction was ended by a "START TRANSACTION"
  or "BEGIN" statement, in which case the metadata locks would be
  released before the actual commit of the active transaction.
  
  The solution is to defer the release of metadata locks to after the
  transaction has been implicitly committed. No test case is provided
  as the effort to provide one is too disproportional to the size of
  the fix.

19 files changed, 255 insertions, 110 deletions

diff --git a/sql/ha_ndbcluster_binlog.cc b/sql/ha_ndbcluster_binlog.cc
index 08abb88e768..7ce318394d4 100644
--- a/sql/ha_ndbcluster_binlog.cc
+++ b/sql/ha_ndbcluster_binlog.cc
@@ -2347,7 +2347,7 @@ static int open_ndb_binlog_index(THD *thd, TABLE **ndb_binlog_index)
   tables->required_type= FRMTYPE_TABLE;
   uint counter;
   thd->clear_error();
-  if (open_tables(thd, &tables, &counter, MYSQL_LOCK_IGNORE_FLUSH))
+  if (simple_open_n_lock_tables(thd, tables))
   {
     if (thd->killed)
       sql_print_error("NDB Binlog: Opening ndb_binlog_index: killed");
@@ -2381,28 +2381,11 @@ int ndb_add_ndb_binlog_index(THD *thd, void *_row)
   ulong saved_options= thd->options;
   thd->options&= ~(OPTION_BIN_LOG);
 
-  for ( ; ; ) /* loop for need_reopen */
+  if (!ndb_binlog_index && open_ndb_binlog_index(thd, &ndb_binlog_index))
   {
-    if (!ndb_binlog_index && open_ndb_binlog_index(thd, &ndb_binlog_index))
-    {
-      error= -1;
-      goto add_ndb_binlog_index_err;
-    }
-
-    if (lock_tables(thd, &binlog_tables, 1, 0, &need_reopen))
-    {
-      if (need_reopen)
-      {
-        TABLE_LIST *p_binlog_tables= &binlog_tables;
-        close_tables_for_reopen(thd, &p_binlog_tables, FALSE);
-        ndb_binlog_index= 0;
-        continue;
-      }
-      sql_print_error("NDB Binlog: Unable to lock table ndb_binlog_index");
-      error= -1;
-      goto add_ndb_binlog_index_err;
-    }
-    break;
+    sql_print_error("NDB Binlog: Unable to lock table ndb_binlog_index");
+    error= -1;
+    goto add_ndb_binlog_index_err;
   }
 
   /*
@@ -2428,13 +2411,6 @@ int ndb_add_ndb_binlog_index(THD *thd, void *_row)
   }
 
 
-  if (! thd->locked_tables_mode)                /* Is always TRUE */
-  {
-    mysql_unlock_tables(thd, thd->lock);
-    thd->lock= 0;
-  }
-  thd->options= saved_options;
-  return 0;
 add_ndb_binlog_index_err:
   close_thread_tables(thd);
   ndb_binlog_index= 0;
@@ -3905,9 +3881,6 @@ restart:
   {
     static char db[]= "";
     thd->db= db;
-    if (ndb_binlog_running)
-      open_ndb_binlog_index(thd, &ndb_binlog_index);
-    thd->db= db;
   }
   do_ndbcluster_binlog_close_connection= BCCC_running;
   for ( ; !((ndbcluster_binlog_terminating ||
diff --git a/sql/log_event.cc b/sql/log_event.cc
index e41bb690e8a..aa8dd6e9bff 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -5314,10 +5314,17 @@ void Xid_log_event::print(FILE* file, PRINT_EVENT_INFO* print_event_info)
 #if defined(HAVE_REPLICATION) && !defined(MYSQL_CLIENT)
 int Xid_log_event::do_apply_event(Relay_log_info const *rli)
 {
+  bool res;
   /* For a slave Xid_log_event is COMMIT */
   general_log_print(thd, COM_QUERY,
                     "COMMIT /* implicit, from Xid_log_event */");
-  return trans_commit(thd);
+  if (!(res= trans_commit(thd)))
+  {
+    close_thread_tables(thd);
+    if (!thd->locked_tables_mode)
+      thd->mdl_context.release_all_locks();
+  }
+  return res;
 }
 
 Log_event::enum_skip_reason
diff --git a/sql/mdl.cc b/sql/mdl.cc
index 1d591bb2244..eb8fcdb323e 100644
--- a/sql/mdl.cc
+++ b/sql/mdl.cc
@@ -491,7 +491,7 @@ void MDL_ticket::destroy(MDL_ticket *ticket)
   @sa THD::enter_cond()/exit_cond()/killed.
 
   @note We can't use THD::enter_cond()/exit_cond()/killed directly here
-        since this will make metadata subsystem dependant on THD class
+        since this will make metadata subsystem dependent on THD class
         and thus prevent us from writing unit tests for it. And usage of
         wrapper functions to access THD::killed/enter_cond()/exit_cond()
         will probably introduce too much overhead.
@@ -881,6 +881,7 @@ static bool notify_shared_lock(THD *thd, MDL_ticket *conflicting_ticket)
   if (conflicting_ticket->is_shared())
   {
     THD *conflicting_thd= conflicting_ticket->get_ctx()->get_thd();
+    DBUG_ASSERT(thd != conflicting_thd); /* Self-deadlock */
     woke= mysql_notify_thread_having_shared_lock(thd, conflicting_thd);
   }
   return woke;
@@ -1089,7 +1090,6 @@ MDL_ticket::upgrade_shared_lock_to_exclusive()
 
   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.
@@ -1164,7 +1164,7 @@ MDL_ticket::upgrade_shared_lock_to_exclusive()
   @param conflict   [out] Indicates that conflicting lock exists
 
   @retval TRUE  Failure either conflicting lock exists or some error
-                occured (probably OOM).
+                occurred (probably OOM).
   @retval FALSE Success, lock was acquired.
 
   FIXME: Compared to lock_table_name_if_not_cached()
diff --git a/sql/mysql_priv.h b/sql/mysql_priv.h
index 3f6ed2b1cb0..159613c58a0 100644
--- a/sql/mysql_priv.h
+++ b/sql/mysql_priv.h
@@ -827,7 +827,7 @@ extern my_decimal decimal_zero;
 void free_items(Item *item);
 void cleanup_items(Item *item);
 class THD;
-void close_thread_tables(THD *thd, bool skip_mdl= 0);
+void close_thread_tables(THD *thd, bool is_back_off= 0);
 
 #ifndef NO_EMBEDDED_ACCESS_CHECKS
 bool check_one_table_access(THD *thd, ulong privilege, TABLE_LIST *tables);
diff --git a/sql/rpl_injector.cc b/sql/rpl_injector.cc
index 738341cc034..9d82307d2e7 100644
--- a/sql/rpl_injector.cc
+++ b/sql/rpl_injector.cc
@@ -83,10 +83,16 @@ int injector::transaction::commit()
      explicitly.
    */
    trans_commit_stmt(m_thd);
-   trans_commit(m_thd);
+   if (!trans_commit(m_thd))
+   {
+     close_thread_tables(m_thd);
+     if (!m_thd->locked_tables_mode)
+       m_thd->mdl_context.release_all_locks();
+   }
    DBUG_RETURN(0);
 }
 
+
 int injector::transaction::use_table(server_id_type sid, table tbl)
 {
   DBUG_ENTER("injector::transaction::use_table");
diff --git a/sql/rpl_rli.cc b/sql/rpl_rli.cc
index 695b160fc01..b4554bb4b6c 100644
--- a/sql/rpl_rli.cc
+++ b/sql/rpl_rli.cc
@@ -1189,6 +1189,8 @@ void Relay_log_info::cleanup_context(THD *thd, bool error)
   }
   m_table_map.clear_tables();
   slave_close_thread_tables(thd);
+  if (error && !thd->locked_tables_mode)
+    thd->mdl_context.release_all_locks();
   clear_flag(IN_STMT);
   /*
     Cleanup for the flags that have been set at do_apply_event.
@@ -1200,13 +1202,6 @@ void Relay_log_info::cleanup_context(THD *thd, bool error)
 
 void Relay_log_info::clear_tables_to_lock()
 {
-  /*
-    Deallocating elements of table list below will also free memory where
-    meta-data locks are stored. So we want to be sure that we don't have
-    any references to this memory left.
-  */
-  DBUG_ASSERT(!current_thd->mdl_context.has_locks());
-
   while (tables_to_lock)
   {
     uchar* to_free= reinterpret_cast<uchar*>(tables_to_lock);
@@ -1225,12 +1220,6 @@ void Relay_log_info::clear_tables_to_lock()
 
 void Relay_log_info::slave_close_thread_tables(THD *thd)
 {
-  /*
-    Since we use same memory chunks for allocation of metadata lock
-    objects for tables as we use for allocating corresponding elements
-    of 'tables_to_lock' list, we have to release metadata locks by
-    closing tables before calling clear_tables_to_lock().
-  */
   close_thread_tables(thd);
   clear_tables_to_lock();
 }
diff --git a/sql/set_var.cc b/sql/set_var.cc
index 266cdd9ad6d..dd009541274 100644
--- a/sql/set_var.cc
+++ b/sql/set_var.cc
@@ -3190,9 +3190,15 @@ static bool set_option_autocommit(THD *thd, set_var *var)
     need to commit any outstanding transactions.
    */
   if (var->save_result.ulong_value != 0 &&
-      (thd->options & OPTION_NOT_AUTOCOMMIT) &&
-      trans_commit(thd))
-    return 1;
+      (thd->options & OPTION_NOT_AUTOCOMMIT))
+  {
+    if (trans_commit(thd))
+      return TRUE;
+
+    close_thread_tables(thd);
+    if (!thd->locked_tables_mode)
+      thd->mdl_context.release_all_locks();
+  }
 
   if (var->save_result.ulong_value != 0)
     thd->options&= ~((sys_var_thd_bit*) var->var)->bit_flag;
diff --git a/sql/slave.cc b/sql/slave.cc
index ed722305b29..0316ae2591d 100644
--- a/sql/slave.cc
+++ b/sql/slave.cc
@@ -2432,6 +2432,9 @@ static int exec_relay_log_event(THD* thd, Relay_log_info* rli)
           {
             exec_res= 0;
             trans_rollback(thd);
+            close_thread_tables(thd);
+            if (!thd->locked_tables_mode)
+              thd->mdl_context.release_all_locks();
             /* chance for concurrent connection to get more locks */
             safe_sleep(thd, min(rli->trans_retries, MAX_SLAVE_RETRY_PAUSE),
                        (CHECK_KILLED_FUNC)sql_slave_killed, (void*)rli);
diff --git a/sql/sql_acl.cc b/sql/sql_acl.cc
index aa96483cf09..f4a182b321f 100644
--- a/sql/sql_acl.cc
+++ b/sql/sql_acl.cc
@@ -30,6 +30,7 @@
 #include <stdarg.h>
 #include "sp_head.h"
 #include "sp.h"
+#include "transaction.h"
 
 time_t mysql_db_table_last_check= 0L;
 
@@ -676,9 +677,6 @@ my_bool acl_reload(THD *thd)
   my_bool return_val= 1;
   DBUG_ENTER("acl_reload");
 
-  /* Can't have locked tables here. */
-  thd->locked_tables_list.unlock_locked_tables(thd);
-
   /*
     To avoid deadlocks we should obtain table locks before
     obtaining acl_cache->lock mutex.
@@ -732,7 +730,10 @@ my_bool acl_reload(THD *thd)
   if (old_initialized)
     pthread_mutex_unlock(&acl_cache->lock);
 end:
+  trans_commit_implicit(thd);
   close_thread_tables(thd);
+  if (!thd->locked_tables_mode)
+    thd->mdl_context.release_all_locks();
   DBUG_RETURN(return_val);
 }
 
@@ -3900,7 +3901,10 @@ my_bool grant_reload(THD *thd)
     free_root(&old_mem,MYF(0));
   }
   rw_unlock(&LOCK_grant);
+  trans_commit_implicit(thd);
   close_thread_tables(thd);
+  if (!thd->locked_tables_mode)
+    thd->mdl_context.release_all_locks();
 
   /*
     It is OK failing to load procs_priv table because we may be
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index 0441458510f..42fc3ba0566 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -1478,11 +1478,22 @@ void close_thread_tables(THD *thd,
   if (thd->open_tables)
     close_open_tables(thd);
 
-  thd->mdl_context.release_all_locks();
   if (!is_back_off)
   {
     thd->mdl_context.remove_all_requests();
   }
+
+  /*
+    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.
+  */
+  if (!thd->in_multi_stmt_transaction() ||
+      (thd->state_flags & Open_tables_state::BACKUPS_AVAIL))
+    thd->mdl_context.release_all_locks();
+
   DBUG_VOID_RETURN;
 }
 
@@ -2284,7 +2295,7 @@ open_table_get_mdl_lock(THD *thd, TABLE_LIST *table_list,
 {
   thd->mdl_context.add_request(mdl_request);
 
-  if (table_list->open_type)
+  if (table_list->lock_strategy)
   {
     /*
       In case of CREATE TABLE .. If NOT EXISTS .. SELECT, the table
@@ -2358,10 +2369,16 @@ open_table_get_mdl_lock(THD *thd, TABLE_LIST *table_list,
   IMPLEMENTATION
     Uses a cache of open tables to find a table not in use.
 
-    If table list element for the table to be opened has "open_type" set
-    to OPEN_OR_CREATE and table does not exist, this function will take
-    exclusive metadata lock on the table, also it will do this if
-    "open_type" is TAKE_EXCLUSIVE_MDL.
+    If TABLE_LIST::open_strategy is set to OPEN_IF_EXISTS, the table is opened
+    only if it exists. If the open strategy is OPEN_STUB, the underlying table
+    is never opened. In both cases, metadata locks are always taken according
+    to the lock strategy.
+
+    This function will take a exclusive metadata lock on the table if
+    TABLE_LIST::lock_strategy is EXCLUSIVE_DOWNGRADABLE_MDL or EXCLUSIVE_MDL.
+    If the lock strategy is EXCLUSIVE_DOWNGRADABLE_MDL and opening the table
+    is successful, the exclusive metadata lock is downgraded to a shared
+    lock.
 
   RETURN
     TRUE  Open failed. "action" parameter may contain type of action
@@ -2595,7 +2612,7 @@ bool open_table(THD *thd, TABLE_LIST *table_list, MEM_ROOT *mem_root,
     DBUG_RETURN(TRUE);
   }
 
-  if (table_list->open_type == TABLE_LIST::OPEN_OR_CREATE)
+  if (table_list->open_strategy == TABLE_LIST::OPEN_IF_EXISTS)
   {
     bool exists;
 
@@ -2609,7 +2626,7 @@ bool open_table(THD *thd, TABLE_LIST *table_list, MEM_ROOT *mem_root,
     }
     /* Table exists. Let us try to open it. */
   }
-  else if (table_list->open_type == TABLE_LIST::TAKE_EXCLUSIVE_MDL)
+  else if (table_list->open_strategy == TABLE_LIST::OPEN_STUB)
   {
     pthread_mutex_unlock(&LOCK_open);
     DBUG_RETURN(FALSE);
@@ -2794,7 +2811,7 @@ bool open_table(THD *thd, TABLE_LIST *table_list, MEM_ROOT *mem_root,
     table exists now we should downgrade our exclusive metadata
     lock on this table to shared metadata lock.
   */
-  if (table_list->open_type == TABLE_LIST::OPEN_OR_CREATE)
+  if (table_list->lock_strategy == TABLE_LIST::EXCLUSIVE_DOWNGRADABLE_MDL)
     mdl_ticket->downgrade_exclusive_lock();
 
   table->mdl_ticket= mdl_ticket;
@@ -3623,7 +3640,7 @@ int open_tables(THD *thd, TABLE_LIST **start, uint *counter, uint flags)
   /* Also used for indicating that prelocking is need */
   TABLE_LIST **query_tables_last_own;
   bool safe_to_ignore_table;
-
+  bool has_locks= thd->mdl_context.has_locks();
   DBUG_ENTER("open_tables");
   /*
     temporary mem_root for new .frm parsing.
@@ -3764,6 +3781,18 @@ int open_tables(THD *thd, TABLE_LIST **start, uint *counter, uint flags)
       if (action)
       {
         /*
+          We have met a exclusive metadata lock or a old version of table and
+          we are inside a transaction that already hold locks. We can't follow
+          the locking protocol in this scenario as it might lead to deadlocks.
+        */
+        if (thd->in_multi_stmt_transaction() && has_locks)
+        {
+          my_error(ER_LOCK_DEADLOCK, MYF(0));
+          result= -1;
+          goto err;
+        }
+
+        /*
           We have met exclusive metadata lock or old version of table. Now we
           have to close all tables which are not up to date/release metadata
           locks. We also have to throw away set of prelocked tables (and thus
@@ -3841,7 +3870,7 @@ int open_tables(THD *thd, TABLE_LIST **start, uint *counter, uint flags)
       Special types of open can succeed but still don't set
       TABLE_LIST::table to anything.
     */
-    if (tables->open_type && !tables->table)
+    if (tables->open_strategy && !tables->table)
       continue;
 
     /*
@@ -4122,7 +4151,7 @@ retry:
   if (!error)
   {
     /*
-      We can't have a view or some special "open_type" in this function
+      We can't have a view or some special "open_strategy" in this function
       so there should be a TABLE instance.
     */
     DBUG_ASSERT(table_list->table);
@@ -4662,6 +4691,8 @@ void close_tables_for_reopen(THD *thd, TABLE_LIST **tables, bool is_back_off)
   for (TABLE_LIST *tmp= *tables; tmp; tmp= tmp->next_global)
     tmp->table= 0;
   close_thread_tables(thd, is_back_off);
+  if (!thd->locked_tables_mode)
+    thd->mdl_context.release_all_locks();
 }
 
 
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index 654d7ad718d..7252f078f81 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -995,8 +995,18 @@ void THD::cleanup(void)
     trans_rollback(this);
     xid_cache_delete(&transaction.xid_state);
   }
+
   locked_tables_list.unlock_locked_tables(this);
 
+  /*
+    If the thread was in the middle of an ongoing transaction (rolled
+    back a few lines above) or under LOCK TABLES (unlocked the tables
+    and left the mode a few lines above), there will be outstanding
+    metadata locks. Release them.
+  */
+  DBUG_ASSERT(open_tables == NULL);
+  mdl_context.release_all_locks();
+
 #if defined(ENABLED_DEBUG_SYNC)
   /* End the Debug Sync Facility. See debug_sync.cc. */
   debug_sync_end_thread(this);
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 579fbcea4ea..ddc163072a7 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -122,11 +122,11 @@ static bool some_non_temp_table_to_be_updated(THD *thd, TABLE_LIST *tables)
   @param mask   Bitmask used for the SQL command match.
 
 */
-static bool opt_implicit_commit(THD *thd, uint mask)
+static bool stmt_causes_implicit_commit(THD *thd, uint mask)
 {
   LEX *lex= thd->lex;
-  bool res= FALSE, skip= FALSE;
-  DBUG_ENTER("opt_implicit_commit");
+  bool skip= FALSE;
+  DBUG_ENTER("stmt_causes_implicit_commit");
 
   if (!(sql_command_flags[lex->sql_command] & mask))
     DBUG_RETURN(FALSE);
@@ -147,15 +147,7 @@ static bool opt_implicit_commit(THD *thd, uint mask)
     break;
   }
 
-  if (!skip)
-  {
-    /* Commit or rollback the statement transaction. */
-    thd->is_error() ? trans_rollback_stmt(thd) : trans_commit_stmt(thd);
-    /* Commit the normal transaction if one is active. */
-    res= trans_commit_implicit(thd);
-  }
-
-  DBUG_RETURN(res);
+  DBUG_RETURN(!skip);
 }
 
 
@@ -1168,6 +1160,9 @@ bool dispatch_command(enum enum_server_command command, THD *thd,
     ulong options= (ulong) (uchar) packet[0];
     if (trans_commit_implicit(thd))
       break;
+    close_thread_tables(thd);
+    if (!thd->locked_tables_mode)
+      thd->mdl_context.release_all_locks();
     if (check_global_access(thd,RELOAD_ACL))
       break;
     general_log_print(thd, command, NullS);
@@ -1196,6 +1191,9 @@ bool dispatch_command(enum enum_server_command command, THD *thd,
       break;
     if (trans_commit_implicit(thd))
       break;
+    close_thread_tables(thd);
+    if (!thd->locked_tables_mode)
+      thd->mdl_context.release_all_locks();
     my_ok(thd);
     break;
   }
@@ -1942,8 +1940,18 @@ mysql_execute_command(THD *thd)
     not run in it's own transaction it may simply never appear on
     the slave in case the outside transaction rolls back.
   */
-  if (opt_implicit_commit(thd, CF_IMPLICT_COMMIT_BEGIN))
-    goto error;
+  if (stmt_causes_implicit_commit(thd, CF_IMPLICT_COMMIT_BEGIN))
+  {
+    /* Commit or rollback the statement transaction. */
+    thd->is_error() ? trans_rollback_stmt(thd) : trans_commit_stmt(thd);
+    /* Commit the normal transaction if one is active. */
+    if (trans_commit_implicit(thd))
+      goto error;
+    /* Close tables and release metadata locks. */
+    close_thread_tables(thd);
+    if (!thd->locked_tables_mode)
+      thd->mdl_context.release_all_locks();
+  }
 
   switch (lex->sql_command) {
 
@@ -2363,7 +2371,9 @@ case SQLCOM_PREPARE:
       if (!(create_info.options & HA_LEX_CREATE_TMP_TABLE))
       {
         lex->link_first_table_back(create_table, link_to_local);
-        create_table->open_type= TABLE_LIST::OPEN_OR_CREATE;
+        /* Set strategies: reset default or 'prepared' values. */
+        create_table->open_strategy= TABLE_LIST::OPEN_IF_EXISTS;
+        create_table->lock_strategy= TABLE_LIST::EXCLUSIVE_DOWNGRADABLE_MDL;
       }
 
       if (!(res= open_and_lock_tables(thd, lex->query_tables)))
@@ -3306,6 +3316,7 @@ end_with_restore_list:
     if (thd->options & OPTION_TABLE_LOCK)
     {
       trans_commit_implicit(thd);
+      thd->mdl_context.release_all_locks();
       thd->options&= ~(OPTION_TABLE_LOCK);
     }
     if (thd->global_read_lock)
@@ -3317,6 +3328,8 @@ end_with_restore_list:
     /* we must end the trasaction first, regardless of anything */
     if (trans_commit_implicit(thd))
       goto error;
+    /* release transactional metadata locks. */
+    thd->mdl_context.release_all_locks();
     if (check_table_access(thd, LOCK_TABLES_ACL | SELECT_ACL, all_tables,
                            FALSE, UINT_MAX, FALSE))
       goto error;
@@ -3346,6 +3359,13 @@ end_with_restore_list:
       */
       trans_rollback_stmt(thd);
       trans_commit_implicit(thd);
+      /*
+        Close tables and release metadata locks otherwise a later call to
+        close_thread_tables might not release the locks if autocommit is off.
+      */
+      close_thread_tables(thd);
+      DBUG_ASSERT(!thd->locked_tables_mode);
+      thd->mdl_context.release_all_locks();
       thd->options&= ~(OPTION_TABLE_LOCK);
     }
     else
@@ -3836,6 +3856,8 @@ end_with_restore_list:
                 thd->locked_tables_mode == LTM_LOCK_TABLES);
     if (trans_commit(thd))
       goto error;
+    if (!thd->locked_tables_mode)
+      thd->mdl_context.release_all_locks();
     /* Begin transaction with the same isolation level. */
     if (lex->tx_chain && trans_begin(thd))
       goto error;
@@ -3849,6 +3871,8 @@ end_with_restore_list:
                 thd->locked_tables_mode == LTM_LOCK_TABLES);
     if (trans_rollback(thd))
       goto error;
+    if (!thd->locked_tables_mode)
+      thd->mdl_context.release_all_locks();
     /* Begin transaction with the same isolation level. */
     if (lex->tx_chain && trans_begin(thd))
       goto error;
@@ -4196,6 +4220,12 @@ create_sp_error:
 
         if (trans_commit_implicit(thd))
           goto error;
+
+        close_thread_tables(thd);
+
+        if (!thd->locked_tables_mode)
+          thd->mdl_context.release_all_locks();
+
 #ifndef NO_EMBEDDED_ACCESS_CHECKS
 	if (sp_automatic_privileges && !opt_noacl &&
 	    sp_revoke_privileges(thd, db, name, 
@@ -4375,11 +4405,15 @@ create_sp_error:
   case SQLCOM_XA_COMMIT:
     if (trans_xa_commit(thd))
       goto error;
+    if (!thd->locked_tables_mode)
+      thd->mdl_context.release_all_locks();
     my_ok(thd);
     break;
   case SQLCOM_XA_ROLLBACK:
     if (trans_xa_rollback(thd))
       goto error;
+    if (!thd->locked_tables_mode)
+      thd->mdl_context.release_all_locks();
     my_ok(thd);
     break;
   case SQLCOM_XA_RECOVER:
@@ -4524,10 +4558,20 @@ finish:
     start_waiting_global_read_lock(thd);
   }
 
-  /* If commit fails, we should be able to reset the OK status. */
-  thd->stmt_da->can_overwrite_status= TRUE;
-  opt_implicit_commit(thd, CF_IMPLICIT_COMMIT_END);
-  thd->stmt_da->can_overwrite_status= FALSE;
+  if (stmt_causes_implicit_commit(thd, CF_IMPLICIT_COMMIT_END))
+  {
+    /* If commit fails, we should be able to reset the OK status. */
+    thd->stmt_da->can_overwrite_status= TRUE;
+    /* Commit or rollback the statement transaction. */
+    thd->is_error() ? trans_rollback_stmt(thd) : trans_commit_stmt(thd);
+    /* Commit the normal transaction if one is active. */
+    trans_commit_implicit(thd);
+    /* Close tables and release metadata locks. */
+    close_thread_tables(thd);
+    if (!thd->locked_tables_mode)
+      thd->mdl_context.release_all_locks();
+    thd->stmt_da->can_overwrite_status= FALSE;
+  }
 
   DBUG_RETURN(res || thd->is_error());
 }
@@ -6508,6 +6552,9 @@ bool reload_acl_and_cache(THD *thd, ulong options, TABLE_LIST *tables,
     query_cache.flush();			// RESET QUERY CACHE
   }
 #endif /*HAVE_QUERY_CACHE*/
+
+  DBUG_ASSERT(thd->locked_tables_mode || !thd->mdl_context.has_locks());
+
   /*
     Note that if REFRESH_READ_LOCK bit is set then REFRESH_TABLES is set too
     (see sql_yacc.yy)
diff --git a/sql/sql_plugin.cc b/sql/sql_plugin.cc
index 77b5552d977..b7efe13e26e 100644
--- a/sql/sql_plugin.cc
+++ b/sql/sql_plugin.cc
@@ -1349,6 +1349,7 @@ static void plugin_load(MEM_ROOT *tmp_root, int *argc, char **argv)
 #ifdef EMBEDDED_LIBRARY
   bool table_exists;
 #endif /* EMBEDDED_LIBRARY */
+  MDL_request mdl_request;
   DBUG_ENTER("plugin_load");
 
   if (!(new_thd= new THD))
@@ -1366,7 +1367,8 @@ static void plugin_load(MEM_ROOT *tmp_root, int *argc, char **argv)
   tables.alias= tables.table_name= (char*)"plugin";
   tables.lock_type= TL_READ;
   tables.db= new_thd->db;
-  alloc_mdl_requests(&tables, tmp_root);
+  tables.mdl_request= &mdl_request;
+  mdl_request.init(0, tables.db, tables.table_name);
 
 #ifdef EMBEDDED_LIBRARY
   /*
diff --git a/sql/sql_prepare.cc b/sql/sql_prepare.cc
index 5efa0cea7a9..9c7949eaf1d 100644
--- a/sql/sql_prepare.cc
+++ b/sql/sql_prepare.cc
@@ -1673,7 +1673,13 @@ static bool mysql_test_create_table(Prepared_statement *stmt)
     if (!(lex->create_info.options & HA_LEX_CREATE_TMP_TABLE))
     {
       lex->link_first_table_back(create_table, link_to_local);
-      create_table->open_type= TABLE_LIST::OPEN_OR_CREATE;
+      /*
+        The open and lock strategies will be set again once the
+        statement is executed. These values are only meaningful
+        for the prepare phase.
+      */
+      create_table->open_strategy= TABLE_LIST::OPEN_IF_EXISTS;
+      create_table->lock_strategy= TABLE_LIST::SHARED_MDL;
     }
 
     if (open_normal_and_derived_tables(stmt->thd, lex->query_tables, 0))
@@ -3130,6 +3136,7 @@ bool Prepared_statement::prepare(const char *packet, uint packet_len)
   bool error;
   Statement stmt_backup;
   Query_arena *old_stmt_arena;
+  MDL_ticket *mdl_savepoint= NULL;
   DBUG_ENTER("Prepared_statement::prepare");
   /*
     If this is an SQLCOM_PREPARE, we also increase Com_prepare_sql.
@@ -3188,6 +3195,13 @@ bool Prepared_statement::prepare(const char *packet, uint packet_len)
   */
   DBUG_ASSERT(thd->change_list.is_empty());
 
+  /*
+    Marker used to release metadata locks acquired while the prepared
+    statement is being checked.
+  */
+  if (thd->in_multi_stmt_transaction())
+    mdl_savepoint= thd->mdl_context.mdl_savepoint();
+
   /* 
    The only case where we should have items in the thd->free_list is
    after stmt->set_params_from_vars(), which may in some cases create
@@ -3211,6 +3225,15 @@ bool Prepared_statement::prepare(const char *packet, uint packet_len)
 
   lex_end(lex);
   cleanup_stmt();
+  /*
+    If not inside a multi-statement transaction, the metadata locks have
+    already been released and the rollback_to_savepoint is a nop.
+    Otherwise, release acquired locks -- a NULL mdl_savepoint means that
+    all locks are going to be released or that the transaction didn't
+    own any locks.
+  */
+  if (!thd->locked_tables_mode)
+    thd->mdl_context.rollback_to_savepoint(mdl_savepoint);
   thd->restore_backup_statement(this, &stmt_backup);
   thd->stmt_arena= old_stmt_arena;
 
diff --git a/sql/sql_servers.cc b/sql/sql_servers.cc
index 40cdfeed946..c46a01cd534 100644
--- a/sql/sql_servers.cc
+++ b/sql/sql_servers.cc
@@ -39,6 +39,7 @@
 #include <stdarg.h>
 #include "sp_head.h"
 #include "sp.h"
+#include "transaction.h"
 
 /*
   We only use 1 mutex to guard the data structures - THR_LOCK_servers.
@@ -224,9 +225,6 @@ bool servers_reload(THD *thd)
   bool return_val= TRUE;
   DBUG_ENTER("servers_reload");
 
-  /* Can't have locked tables here */
-  thd->locked_tables_list.unlock_locked_tables(thd);
-
   DBUG_PRINT("info", ("locking servers_cache"));
   rw_wrlock(&THR_LOCK_servers);
 
@@ -252,7 +250,10 @@ bool servers_reload(THD *thd)
   }
 
 end:
+  trans_commit_implicit(thd);
   close_thread_tables(thd);
+  if (!thd->locked_tables_mode)
+    thd->mdl_context.release_all_locks();
   DBUG_PRINT("info", ("unlocking servers_cache"));
   rw_unlock(&THR_LOCK_servers);
   DBUG_RETURN(return_val);
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 42ae0a89458..8fdaa2cd93a 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -1974,7 +1974,8 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
     }
 
     /* Probably a non-temporary table. */
-    non_temp_tables_count++;
+    if (!drop_temporary)
+      non_temp_tables_count++;
 
     /*
       If row-based replication is used and the table is not a
@@ -4738,6 +4739,8 @@ static bool mysql_admin_table(THD* thd, TABLE_LIST* tables,
       trans_commit_stmt(thd);
       trans_commit(thd);
       close_thread_tables(thd);
+      if (!thd->locked_tables_mode)
+        thd->mdl_context.release_all_locks();
       lex->reset_query_tables_list(FALSE);
       table->table=0;				// For query cache
       if (protocol->write())
@@ -4786,7 +4789,10 @@ static bool mysql_admin_table(THD* thd, TABLE_LIST* tables,
       {
         DBUG_PRINT("admin", ("recreating table"));
         trans_rollback_stmt(thd);
+        trans_rollback(thd);
         close_thread_tables(thd);
+        if (!thd->locked_tables_mode)
+          thd->mdl_context.release_all_locks();
         tmp_disable_binlog(thd); // binlogging is done by caller if wanted
         result_code= mysql_recreate_table(thd, table);
         reenable_binlog(thd);
@@ -4899,14 +4905,17 @@ send_result_message:
         reopen the table and do ha_innobase::analyze() on it.
         We have to end the row, so analyze could return more rows.
       */
+      trans_commit_stmt(thd);
+      trans_commit(thd);
+      close_thread_tables(thd);
+      if (!thd->locked_tables_mode)
+        thd->mdl_context.release_all_locks();
       protocol->store(STRING_WITH_LEN("note"), system_charset_info);
       protocol->store(STRING_WITH_LEN(
           "Table does not support optimize, doing recreate + analyze instead"),
                       system_charset_info);
       if (protocol->write())
         goto err;
-      trans_commit_stmt(thd);
-      close_thread_tables(thd);
       DBUG_PRINT("info", ("HA_ADMIN_TRY_ALTER, trying analyze..."));
       TABLE_LIST *save_next_local= table->next_local,
                  *save_next_global= table->next_global;
@@ -4923,7 +4932,10 @@ send_result_message:
       if (thd->stmt_da->is_ok())
         thd->stmt_da->reset_diagnostics_area();
       trans_commit_stmt(thd);
+      trans_commit(thd);
       close_thread_tables(thd);
+      if (!thd->locked_tables_mode)
+        thd->mdl_context.release_all_locks();
       if (!result_code) // recreation went ok
       {
         if ((table->table= open_ltable(thd, table, lock_type, 0)) &&
@@ -5017,6 +5029,7 @@ send_result_message:
         query_cache_invalidate3(thd, table->table, 0);
       }
     }
+    /* Error path, a admin command failed. */
     trans_commit_stmt(thd);
     trans_commit_implicit(thd);
     close_thread_tables(thd);
diff --git a/sql/sql_view.cc b/sql/sql_view.cc
index b370fc79b17..b36ff6b6743 100644
--- a/sql/sql_view.cc
+++ b/sql/sql_view.cc
@@ -395,7 +395,8 @@ bool mysql_create_view(THD *thd, TABLE_LIST *views,
     goto err;
 
   lex->link_first_table_back(view, link_to_local);
-  view->open_type= TABLE_LIST::TAKE_EXCLUSIVE_MDL;
+  view->open_strategy= TABLE_LIST::OPEN_STUB;
+  view->lock_strategy= TABLE_LIST::EXCLUSIVE_MDL;
 
   if (open_and_lock_tables(thd, lex->query_tables))
   {
diff --git a/sql/table.h b/sql/table.h
index b6ea372b41e..1762ae8785d 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -1350,24 +1350,36 @@ struct TABLE_LIST
   bool          prelocking_placeholder;
   /**
      Indicates that if TABLE_LIST object corresponds to the table/view
-     which requires special handling/meta-data locking.
+     which requires special handling.
   */
   enum
   {
-    /* Normal open, shared metadata lock should be taken. */
-    NORMAL_OPEN= 0,
-    /*
-      It's target table of CREATE TABLE ... SELECT so we should
-      either open table if it exists (and take shared metadata lock)
-      or take exclusive metadata lock if it doesn't exist.
-    */
-    OPEN_OR_CREATE,
+    /* Normal open. */
+    OPEN_NORMAL= 0,
+    /* Associate a table share only if the the table exists. */
+    OPEN_IF_EXISTS,
+    /* Don't associate a table share. */
+    OPEN_STUB
+  } open_strategy;
+  /**
+    Indicates the locking strategy for the object being opened:
+    whether the associated metadata lock is shared or exclusive.
+  */
+  enum
+  {
+    /* Take a shared metadata lock before the object is opened. */
+    SHARED_MDL= 0,
     /*
-      It's target view of CREATE/ALTER VIEW. We should take exclusive
-      metadata lock for this table list element.
+       Take a exclusive metadata lock before the object is opened.
+       If opening is successful, downgrade to a shared lock.
     */
-    TAKE_EXCLUSIVE_MDL
-  } open_type;
+    EXCLUSIVE_DOWNGRADABLE_MDL,
+    /* Take a exclusive metadata lock before the object is opened. */
+    EXCLUSIVE_MDL
+  } lock_strategy;
+  /* For transactional locking. */
+  int           lock_timeout;           /* NOWAIT or WAIT [X]               */
+  bool          lock_transactional;     /* If transactional lock requested. */
   bool          internal_tmp_table;
   /** TRUE if an alias for this table was specified in the SQL. */
   bool          is_alias;
diff --git a/sql/transaction.cc b/sql/transaction.cc
index 7bfaf4846cf..d1c7244ba83 100644
--- a/sql/transaction.cc
+++ b/sql/transaction.cc
@@ -99,6 +99,12 @@ bool trans_begin(THD *thd, uint flags)
   if (trans_commit_implicit(thd))
     DBUG_RETURN(TRUE);
 
+  /*
+    Release transactional metadata locks only after the
+    transaction has been committed.
+  */
+  thd->mdl_context.release_all_locks();
+
   thd->options|= OPTION_BEGIN;
   thd->server_status|= SERVER_STATUS_IN_TRANS;
 
@@ -331,6 +337,13 @@ bool trans_savepoint(THD *thd, LEX_STRING name)
   newsv->prev= thd->transaction.savepoints;
   thd->transaction.savepoints= newsv;
 
+  /*
+    Remember the last acquired lock before the savepoint was set.
+    This is used as a marker to only release locks acquired after
+    the setting of this savepoint.
+  */
+  newsv->mdl_savepoint = thd->mdl_context.mdl_savepoint();
+
   DBUG_RETURN(FALSE);
 }
 
@@ -375,6 +388,10 @@ bool trans_rollback_to_savepoint(THD *thd, LEX_STRING name)
 
   thd->transaction.savepoints= sv;
 
+  /* Release metadata locks that were acquired during this savepoint unit. */
+  if (!res && !thd->locked_tables_mode)
+    thd->mdl_context.rollback_to_savepoint(sv->mdl_savepoint);
+
   DBUG_RETURN(test(res));
 }