Merge from mysql-5.5-runtime to mysql-5.5-bugteam

No conflicts

51 files changed, 1467 insertions, 1385 deletions

diff --git a/sql/event_data_objects.cc b/sql/event_data_objects.cc
index 890d8be3be4..ef14a061677 100644
--- a/sql/event_data_objects.cc
+++ b/sql/event_data_objects.cc
@@ -290,7 +290,6 @@ Event_basic::load_time_zone(THD *thd, const LEX_STRING tz_name)
 */
 
 Event_queue_element::Event_queue_element():
-  status_changed(FALSE), last_executed_changed(FALSE),
   on_completion(Event_parse_data::ON_COMPLETION_DROP),
   status(Event_parse_data::ENABLED), expression(0), dropped(FALSE),
   execution_count(0)
@@ -539,7 +538,6 @@ Event_queue_element::load_from_row(THD *thd, TABLE *table)
                                                    TIME_NO_ZERO_DATE);
     last_executed= my_tz_OFFSET0->TIME_to_gmt_sec(&time,&not_used);
   }
-  last_executed_changed= FALSE;
 
   if ((ptr= get_field(&mem_root, table->field[ET_FIELD_STATUS])) == NullS)
     DBUG_RETURN(TRUE);
@@ -935,7 +933,6 @@ Event_queue_element::compute_next_execution_time()
       DBUG_PRINT("info",("One-time event will be dropped: %d.", dropped));
 
       status= Event_parse_data::DISABLED;
-      status_changed= TRUE;
     }
     goto ret;
   }
@@ -955,7 +952,6 @@ Event_queue_element::compute_next_execution_time()
       dropped= TRUE;
     DBUG_PRINT("info", ("Dropped: %d", dropped));
     status= Event_parse_data::DISABLED;
-    status_changed= TRUE;
 
     goto ret;
   }
@@ -1018,7 +1014,6 @@ Event_queue_element::compute_next_execution_time()
         if (on_completion == Event_parse_data::ON_COMPLETION_DROP)
           dropped= TRUE;
         status= Event_parse_data::DISABLED;
-        status_changed= TRUE;
       }
       else
       {
@@ -1108,7 +1103,6 @@ Event_queue_element::compute_next_execution_time()
           execute_at= 0;
           execute_at_null= TRUE;
           status= Event_parse_data::DISABLED;
-          status_changed= TRUE;
           if (on_completion == Event_parse_data::ON_COMPLETION_DROP)
             dropped= TRUE;
         }
@@ -1144,50 +1138,11 @@ void
 Event_queue_element::mark_last_executed(THD *thd)
 {
   last_executed= (my_time_t) thd->query_start();
-  last_executed_changed= TRUE;
 
   execution_count++;
 }
 
 
-/*
-  Saves status and last_executed_at to the disk if changed.
-
-  SYNOPSIS
-    Event_queue_element::update_timing_fields()
-      thd - thread context
-
-  RETURN VALUE
-    FALSE   OK
-    TRUE    Error while opening mysql.event for writing or during
-            write on disk
-*/
-
-bool
-Event_queue_element::update_timing_fields(THD *thd)
-{
-  Event_db_repository *db_repository= Events::get_db_repository();
-  int ret;
-
-  DBUG_ENTER("Event_queue_element::update_timing_fields");
-
-  DBUG_PRINT("enter", ("name: %*s", (int) name.length, name.str));
-
-  /* No need to update if nothing has changed */
-  if (!(status_changed || last_executed_changed))
-    DBUG_RETURN(0);
-
-  ret= db_repository->update_timing_fields_for_event(thd,
-                                                     dbname, name,
-                                                     last_executed_changed,
-                                                     last_executed,
-                                                     status_changed,
-                                                     (ulonglong) status);
-  last_executed_changed= status_changed= FALSE;
-  DBUG_RETURN(ret);
-}
-
-
 static
 void
 append_datetime(String *buf, Time_zone *time_zone, my_time_t secs,
diff --git a/sql/event_data_objects.h b/sql/event_data_objects.h
index 9d17213bcb8..46740812d31 100644
--- a/sql/event_data_objects.h
+++ b/sql/event_data_objects.h
@@ -82,10 +82,6 @@ protected:
 
 class Event_queue_element : public Event_basic
 {
-protected:
-  bool status_changed;
-  bool last_executed_changed;
-
 public:
   int on_completion;
   int status;
@@ -117,9 +113,6 @@ public:
 
   void
   mark_last_executed(THD *thd);
-
-  bool
-  update_timing_fields(THD *thd);
 };
 
 
diff --git a/sql/event_db_repository.cc b/sql/event_db_repository.cc
index db508e4ea41..053558aa0c3 100644
--- a/sql/event_db_repository.cc
+++ b/sql/event_db_repository.cc
@@ -622,6 +622,12 @@ Event_db_repository::create_event(THD *thd, Event_parse_data *parse_data,
   TABLE *table= NULL;
   sp_head *sp= thd->lex->sphead;
   ulong saved_mode= thd->variables.sql_mode;
+  /*
+    Take a savepoint to release only the lock on mysql.event
+    table at the end but keep the global read lock and
+    possible other locks taken by the caller.
+  */
+  MDL_savepoint mdl_savepoint= thd->mdl_context.mdl_savepoint();
 
   DBUG_ENTER("Event_db_repository::create_event");
 
@@ -699,8 +705,8 @@ Event_db_repository::create_event(THD *thd, Event_parse_data *parse_data,
   ret= 0;
 
 end:
-  if (table)
-    close_mysql_tables(thd);
+  close_thread_tables(thd);
+  thd->mdl_context.rollback_to_savepoint(mdl_savepoint);
 
   thd->variables.sql_mode= saved_mode;
   DBUG_RETURN(test(ret));
@@ -734,6 +740,12 @@ Event_db_repository::update_event(THD *thd, Event_parse_data *parse_data,
   TABLE *table= NULL;
   sp_head *sp= thd->lex->sphead;
   ulong saved_mode= thd->variables.sql_mode;
+  /*
+    Take a savepoint to release only the lock on mysql.event
+    table at the end but keep the global read lock and
+    possible other locks taken by the caller.
+  */
+  MDL_savepoint mdl_savepoint= thd->mdl_context.mdl_savepoint();
   int ret= 1;
 
   DBUG_ENTER("Event_db_repository::update_event");
@@ -811,8 +823,8 @@ Event_db_repository::update_event(THD *thd, Event_parse_data *parse_data,
   ret= 0;
 
 end:
-  if (table)
-    close_mysql_tables(thd);
+  close_thread_tables(thd);
+  thd->mdl_context.rollback_to_savepoint(mdl_savepoint);
 
   thd->variables.sql_mode= saved_mode;
   DBUG_RETURN(test(ret));
@@ -838,6 +850,12 @@ Event_db_repository::drop_event(THD *thd, LEX_STRING db, LEX_STRING name,
                                 bool drop_if_exists)
 {
   TABLE *table= NULL;
+  /*
+    Take a savepoint to release only the lock on mysql.event
+    table at the end but keep the global read lock and
+    possible other locks taken by the caller.
+  */
+  MDL_savepoint mdl_savepoint= thd->mdl_context.mdl_savepoint();
   int ret= 1;
 
   DBUG_ENTER("Event_db_repository::drop_event");
@@ -866,8 +884,8 @@ Event_db_repository::drop_event(THD *thd, LEX_STRING db, LEX_STRING name,
   ret= 0;
 
 end:
-  if (table)
-    close_mysql_tables(thd);
+  close_thread_tables(thd);
+  thd->mdl_context.rollback_to_savepoint(mdl_savepoint);
 
   DBUG_RETURN(test(ret));
 }
@@ -940,7 +958,7 @@ Event_db_repository::drop_schema_events(THD *thd, LEX_STRING schema)
   TABLE *table= NULL;
   READ_RECORD read_record_info;
   enum enum_events_table_field field= ET_FIELD_DB;
-  MDL_ticket *mdl_savepoint= thd->mdl_context.mdl_savepoint();
+  MDL_savepoint mdl_savepoint= thd->mdl_context.mdl_savepoint();
   DBUG_ENTER("Event_db_repository::drop_schema_events");
   DBUG_PRINT("enter", ("field=%d schema=%s", field, schema.str));
 
@@ -1042,15 +1060,14 @@ Event_db_repository::
 update_timing_fields_for_event(THD *thd,
                                LEX_STRING event_db_name,
                                LEX_STRING event_name,
-                               bool update_last_executed,
                                my_time_t last_executed,
-                               bool update_status,
                                ulonglong status)
 {
   TABLE *table= NULL;
   Field **fields;
   int ret= 1;
   bool save_binlog_row_based;
+  MYSQL_TIME time;
 
   DBUG_ENTER("Event_db_repository::update_timing_fields_for_event");
 
@@ -1075,20 +1092,12 @@ update_timing_fields_for_event(THD *thd,
   /* Don't update create on row update. */
   table->timestamp_field_type= TIMESTAMP_NO_AUTO_SET;
 
-  if (update_last_executed)
-  {
-    MYSQL_TIME time;
-    my_tz_OFFSET0->gmt_sec_to_TIME(&time, last_executed);
+  my_tz_OFFSET0->gmt_sec_to_TIME(&time, last_executed);
+  fields[ET_FIELD_LAST_EXECUTED]->set_notnull();
+  fields[ET_FIELD_LAST_EXECUTED]->store_time(&time, MYSQL_TIMESTAMP_DATETIME);
 
-    fields[ET_FIELD_LAST_EXECUTED]->set_notnull();
-    fields[ET_FIELD_LAST_EXECUTED]->store_time(&time,
-                                               MYSQL_TIMESTAMP_DATETIME);
-  }
-  if (update_status)
-  {
-    fields[ET_FIELD_STATUS]->set_notnull();
-    fields[ET_FIELD_STATUS]->store(status, TRUE);
-  }
+  fields[ET_FIELD_STATUS]->set_notnull();
+  fields[ET_FIELD_STATUS]->store(status, TRUE);
 
   if ((ret= table->file->ha_update_row(table->record[1], table->record[0])))
   {
diff --git a/sql/event_db_repository.h b/sql/event_db_repository.h
index ea7f3bbac0e..58484d17f06 100644
--- a/sql/event_db_repository.h
+++ b/sql/event_db_repository.h
@@ -101,9 +101,7 @@ public:
   update_timing_fields_for_event(THD *thd,
                                  LEX_STRING event_db_name,
                                  LEX_STRING event_name,
-                                 bool update_last_executed,
                                  my_time_t last_executed,
-                                 bool update_status,
                                  ulonglong status);
 public:
   static bool
diff --git a/sql/event_queue.cc b/sql/event_queue.cc
index f0310c676d1..92b8a2c9071 100644
--- a/sql/event_queue.cc
+++ b/sql/event_queue.cc
@@ -17,6 +17,8 @@
 #include "unireg.h"
 #include "event_queue.h"
 #include "event_data_objects.h"
+#include "event_db_repository.h"
+#include "events.h"
 #include "sql_audit.h"
 #include "tztime.h"     // my_tz_find, my_tz_OFFSET0, struct Time_zone
 #include "log.h"        // sql_print_error
@@ -444,7 +446,6 @@ Event_queue::recalculate_activation_times(THD *thd)
   for (i= 0; i < queue.elements; i++)
   {
     ((Event_queue_element*)queue_element(&queue, i))->compute_next_execution_time();
-    ((Event_queue_element*)queue_element(&queue, i))->update_timing_fields(thd);
   }
   queue_fix(&queue);
   /*
@@ -567,6 +568,8 @@ Event_queue::get_top_for_execution_if_time(THD *thd,
 {
   bool ret= FALSE;
   *event_name= NULL;
+  my_time_t last_executed;
+  int status;
   DBUG_ENTER("Event_queue::get_top_for_execution_if_time");
 
   LOCK_QUEUE_DATA();
@@ -632,8 +635,14 @@ Event_queue::get_top_for_execution_if_time(THD *thd,
 
     top->execution_count++;
     (*event_name)->dropped= top->dropped;
+    /*
+      Save new values of last_executed timestamp and event status on stack
+      in order to be able to update event description in system table once
+      QUEUE_DATA lock is released.
+    */
+    last_executed= top->last_executed;
+    status= top->status;
 
-    top->update_timing_fields(thd);
     if (top->status == Event_parse_data::DISABLED)
     {
       DBUG_PRINT("info", ("removing from the queue"));
@@ -656,9 +665,16 @@ end:
                       ret, (long) *event_name));
 
   if (*event_name)
+  {
     DBUG_PRINT("info", ("db: %s  name: %s",
                         (*event_name)->dbname.str, (*event_name)->name.str));
 
+    Event_db_repository *db_repository= Events::get_db_repository();
+    (void) db_repository->update_timing_fields_for_event(thd,
+                            (*event_name)->dbname, (*event_name)->name,
+                            last_executed, (ulonglong) status);
+  }
+
   DBUG_RETURN(ret);
 }
 
@@ -741,11 +757,13 @@ Event_queue::cond_wait(THD *thd, struct timespec *abstime, const char* msg,
 
   thd->enter_cond(&COND_queue_state, &LOCK_event_queue, msg);
 
-  DBUG_PRINT("info", ("mysql_cond_%swait", abstime? "timed":""));
-  if (!abstime)
-    mysql_cond_wait(&COND_queue_state, &LOCK_event_queue);
-  else
-    mysql_cond_timedwait(&COND_queue_state, &LOCK_event_queue, abstime);
+  if (!thd->killed)
+  {
+    if (!abstime)
+      mysql_cond_wait(&COND_queue_state, &LOCK_event_queue);
+    else
+      mysql_cond_timedwait(&COND_queue_state, &LOCK_event_queue, abstime);
+  }
 
   mutex_last_locked_in_func= func;
   mutex_last_locked_at_line= line;
diff --git a/sql/events.cc b/sql/events.cc
index e7e47801586..23a0dc9eb52 100644
--- a/sql/events.cc
+++ b/sql/events.cc
@@ -30,6 +30,7 @@
 #include "event_scheduler.h"
 #include "sp_head.h" // for Stored_program_creation_ctx
 #include "set_var.h"
+#include "lock.h"   // lock_object_name
 
 /**
   @addtogroup Event_Scheduler
@@ -77,7 +78,6 @@ Event_queue *Events::event_queue;
 Event_scheduler *Events::scheduler;
 Event_db_repository *Events::db_repository;
 ulong Events::opt_event_scheduler= Events::EVENTS_OFF;
-mysql_mutex_t Events::LOCK_event_metadata;
 bool Events::check_system_tables_error= FALSE;
 
 
@@ -340,7 +340,9 @@ Events::create_event(THD *thd, Event_parse_data *parse_data,
   if ((save_binlog_row_based= thd->is_current_stmt_binlog_format_row()))
     thd->clear_current_stmt_binlog_format_row();
 
-  mysql_mutex_lock(&LOCK_event_metadata);
+  if (lock_object_name(thd, MDL_key::EVENT,
+                       parse_data->dbname.str, parse_data->name.str))
+    DBUG_RETURN(TRUE);
 
   /* On error conditions my_error() is called so no need to handle here */
   if (!(ret= db_repository->create_event(thd, parse_data, if_not_exists)))
@@ -388,7 +390,6 @@ Events::create_event(THD *thd, Event_parse_data *parse_data,
       }
     }
   }
-  mysql_mutex_unlock(&LOCK_event_metadata);
   /* Restore the state of binlog format */
   DBUG_ASSERT(!thd->is_current_stmt_binlog_format_row());
   if (save_binlog_row_based)
@@ -472,7 +473,9 @@ Events::update_event(THD *thd, Event_parse_data *parse_data,
   if ((save_binlog_row_based= thd->is_current_stmt_binlog_format_row()))
     thd->clear_current_stmt_binlog_format_row();
 
-  mysql_mutex_lock(&LOCK_event_metadata);
+  if (lock_object_name(thd, MDL_key::EVENT,
+                       parse_data->dbname.str, parse_data->name.str))
+    DBUG_RETURN(TRUE);
 
   /* On error conditions my_error() is called so no need to handle here */
   if (!(ret= db_repository->update_event(thd, parse_data,
@@ -502,7 +505,6 @@ Events::update_event(THD *thd, Event_parse_data *parse_data,
       ret= write_bin_log(thd, TRUE, thd->query(), thd->query_length());
     }
   }
-  mysql_mutex_unlock(&LOCK_event_metadata);
   /* Restore the state of binlog format */
   DBUG_ASSERT(!thd->is_current_stmt_binlog_format_row());
   if (save_binlog_row_based)
@@ -556,7 +558,9 @@ Events::drop_event(THD *thd, LEX_STRING dbname, LEX_STRING name, bool if_exists)
   if ((save_binlog_row_based= thd->is_current_stmt_binlog_format_row()))
     thd->clear_current_stmt_binlog_format_row();
 
-  mysql_mutex_lock(&LOCK_event_metadata);
+  if (lock_object_name(thd, MDL_key::EVENT,
+                       dbname.str, name.str))
+    DBUG_RETURN(TRUE);
   /* On error conditions my_error() is called so no need to handle here */
   if (!(ret= db_repository->drop_event(thd, dbname, name, if_exists)))
   {
@@ -566,7 +570,6 @@ Events::drop_event(THD *thd, LEX_STRING dbname, LEX_STRING name, bool if_exists)
     DBUG_ASSERT(thd->query() && thd->query_length());
     ret= write_bin_log(thd, TRUE, thd->query(), thd->query_length());
   }
-  mysql_mutex_unlock(&LOCK_event_metadata);
   /* Restore the state of binlog format */
   DBUG_ASSERT(!thd->is_current_stmt_binlog_format_row());
   if (save_binlog_row_based)
@@ -595,15 +598,12 @@ Events::drop_schema_events(THD *thd, char *db)
   DBUG_PRINT("enter", ("dropping events from %s", db));
 
   /*
-    sic: no check if the scheduler is disabled or system tables
+    Sic: no check if the scheduler is disabled or system tables
     are damaged, as intended.
   */
-
-  mysql_mutex_lock(&LOCK_event_metadata);
   if (event_queue)
     event_queue->drop_schema_events(thd, db_lex);
   db_repository->drop_schema_events(thd, db_lex);
-  mysql_mutex_unlock(&LOCK_event_metadata);
 
   DBUG_VOID_RETURN;
 }
@@ -915,12 +915,11 @@ Events::deinit()
 }
 
 #ifdef HAVE_PSI_INTERFACE
-PSI_mutex_key key_LOCK_event_metadata, key_LOCK_event_queue,
+PSI_mutex_key key_LOCK_event_queue,
               key_event_scheduler_LOCK_scheduler_state;
 
 static PSI_mutex_info all_events_mutexes[]=
 {
-  { &key_LOCK_event_metadata, "LOCK_event_metadata", PSI_FLAG_GLOBAL},
   { &key_LOCK_event_queue, "LOCK_event_queue", PSI_FLAG_GLOBAL},
   { &key_event_scheduler_LOCK_scheduler_state, "Event_scheduler::LOCK_scheduler_state", PSI_FLAG_GLOBAL}
 };
@@ -974,23 +973,6 @@ Events::init_mutexes()
 #ifdef HAVE_PSI_INTERFACE
   init_events_psi_keys();
 #endif
-
-  mysql_mutex_init(key_LOCK_event_metadata,
-                   &LOCK_event_metadata, MY_MUTEX_INIT_FAST);
-}
-
-
-/*
-  Destroys Events mutexes
-
-  SYNOPSIS
-    Events::destroy_mutexes()
-*/
-
-void
-Events::destroy_mutexes()
-{
-  mysql_mutex_destroy(&LOCK_event_metadata);
 }
 
 
diff --git a/sql/events.h b/sql/events.h
index f3ebc6da4ad..a337b29049a 100644
--- a/sql/events.h
+++ b/sql/events.h
@@ -26,8 +26,7 @@
 */
 
 #ifdef HAVE_PSI_INTERFACE
-extern PSI_mutex_key key_LOCK_event_metadata,
-                     key_event_scheduler_LOCK_scheduler_state;
+extern PSI_mutex_key key_event_scheduler_LOCK_scheduler_state;
 extern PSI_cond_key key_event_scheduler_COND_state;
 extern PSI_thread_key key_thread_event_scheduler, key_thread_event_worker;
 #endif /* HAVE_PSI_INTERFACE */
@@ -79,7 +78,6 @@ public:
   enum enum_opt_event_scheduler { EVENTS_OFF, EVENTS_ON, EVENTS_DISABLED };
   /* Protected using LOCK_global_system_variables only. */
   static ulong opt_event_scheduler;
-  static mysql_mutex_t LOCK_event_metadata;
   static bool check_if_system_tables_error();
   static bool start();
   static bool stop();
diff --git a/sql/ha_ndbcluster.cc b/sql/ha_ndbcluster.cc
index 6b07cf76d79..20745db9ce9 100644
--- a/sql/ha_ndbcluster.cc
+++ b/sql/ha_ndbcluster.cc
@@ -7382,38 +7382,35 @@ int ndbcluster_find_files(handlerton *hton, THD *thd,
   }
 
   /*
+    Delete old files.
+
     ndbcluster_find_files() may be called from I_S code and ndbcluster_binlog
     thread in situations when some tables are already open. This means that
     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. This is violation of metadata
-    locking protocol which has to be closed ASAP.
+    while holding shared meta-data lock on other tables. This might lead to a
+    deadlock but such a deadlock should be detected by MDL deadlock detector.
+
     XXX: the scenario described above is not covered with any test.
   */
-  if (!global_read_lock)
-  {
-    // Delete old files
-    List_iterator_fast<char> it3(delete_list);
-    while ((file_name_str= it3++))
-    {
-      DBUG_PRINT("info", ("Remove table %s/%s", db, file_name_str));
-      // Delete the table and all related files
-      TABLE_LIST table_list;
-      table_list.init_one_table(db, strlen(db), file_name_str,
-                                strlen(file_name_str), file_name_str,
-                                TL_WRITE);
-      table_list.mdl_request.set_type(MDL_EXCLUSIVE);
-      (void)mysql_rm_table_part2(thd, &table_list,
-                                 FALSE,   /* if_exists */
-                                 FALSE,   /* drop_temporary */ 
-                                 FALSE,   /* drop_view */
-                                 TRUE     /* dont_log_query*/);
-      trans_commit_implicit(thd); /* Safety, should be unnecessary. */
-      thd->mdl_context.release_transactional_locks();
-      /* Clear error message that is returned when table is deleted */
-      thd->clear_error();
-    }
+  List_iterator_fast<char> it3(delete_list);
+  while ((file_name_str= it3++))
+  {
+    DBUG_PRINT("info", ("Remove table %s/%s", db, file_name_str));
+    /* Delete the table and all related files. */
+    TABLE_LIST table_list;
+    table_list.init_one_table(db, strlen(db), file_name_str,
+                              strlen(file_name_str), file_name_str,
+                              TL_WRITE);
+    table_list.mdl_request.set_type(MDL_EXCLUSIVE);
+    (void)mysql_rm_table_part2(thd, &table_list,
+                               FALSE,   /* if_exists */
+                               FALSE,   /* drop_temporary */
+                               FALSE,   /* drop_view */
+                               TRUE     /* dont_log_query*/);
+    trans_commit_implicit(thd); /* Safety, should be unnecessary. */
+    thd->mdl_context.release_transactional_locks();
+    /* Clear error message that is returned when table is deleted */
+    thd->clear_error();
   }
 
   /* Lock mutex before creating .FRM files. */
diff --git a/sql/handler.cc b/sql/handler.cc
index 3552a53972d..38b57c16ee0 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -29,8 +29,6 @@
 #include "sql_cache.h"                   // query_cache, query_cache_*
 #include "key.h"     // key_copy, key_unpack, key_cmp_if_same, key_cmp
 #include "sql_table.h"                   // build_table_filename
-#include "lock.h"               // wait_if_global_read_lock,
-                                // start_waiting_global_read_lock
 #include "sql_parse.h"                          // check_stack_overrun
 #include "sql_acl.h"            // SUPER_ACL
 #include "sql_base.h"           // free_io_cache
@@ -41,6 +39,7 @@
 #include "transaction.h"
 #include <errno.h>
 #include "probes_mysql.h"
+#include "debug_sync.h"         // DEBUG_SYNC
 
 #ifdef WITH_PARTITION_STORAGE_ENGINE
 #include "ha_partition.h"
@@ -1155,6 +1154,7 @@ int ha_commit_trans(THD *thd, bool all)
   {
     uint rw_ha_count;
     bool rw_trans;
+    MDL_request mdl_request;
 
     DBUG_EXECUTE_IF("crash_commit_before", DBUG_SUICIDE(););
 
@@ -1166,11 +1166,27 @@ int ha_commit_trans(THD *thd, bool all)
     /* rw_trans is TRUE when we in a transaction changing data */
     rw_trans= is_real_trans && (rw_ha_count > 0);
 
-    if (rw_trans &&
-        thd->global_read_lock.wait_if_global_read_lock(thd, FALSE, FALSE))
+    if (rw_trans)
     {
-      ha_rollback_trans(thd, all);
-      DBUG_RETURN(1);
+      /*
+        Acquire a metadata lock which will ensure that COMMIT is blocked
+        by an active FLUSH TABLES WITH READ LOCK (and vice versa:
+        COMMIT in progress blocks FTWRL).
+
+        We allow the owner of FTWRL to COMMIT; we assume that it knows
+        what it does.
+      */
+      mdl_request.init(MDL_key::COMMIT, "", "", MDL_INTENTION_EXCLUSIVE,
+                       MDL_EXPLICIT);
+
+      if (thd->mdl_context.acquire_lock(&mdl_request,
+                                        thd->variables.lock_wait_timeout))
+      {
+        ha_rollback_trans(thd, all);
+        DBUG_RETURN(1);
+      }
+
+      DEBUG_SYNC(thd, "ha_commit_trans_after_acquire_commit_lock");
     }
 
     if (rw_trans &&
@@ -1225,8 +1241,16 @@ int ha_commit_trans(THD *thd, bool all)
     DBUG_EXECUTE_IF("crash_commit_after", DBUG_SUICIDE(););
     RUN_HOOK(transaction, after_commit, (thd, FALSE));
 end:
-    if (rw_trans)
-      thd->global_read_lock.start_waiting_global_read_lock(thd);
+    if (rw_trans && mdl_request.ticket)
+    {
+      /*
+        We do not always immediately release transactional locks
+        after ha_commit_trans() (see uses of ha_enable_transaction()),
+        thus we release the commit blocker lock as soon as it's
+        not needed.
+      */
+      thd->mdl_context.release_lock(mdl_request.ticket);
+    }
   }
   /* Free resources and perform other cleanup even for 'empty' transactions. */
   else if (is_real_trans)
diff --git a/sql/item.h b/sql/item.h
index 8e8199ecac8..e65bacf4cb7 100644
--- a/sql/item.h
+++ b/sql/item.h
@@ -2572,7 +2572,7 @@ public:
     DBUG_ASSERT(fixed);
     return (*ref)->get_time(ltime);
   }
-  virtual bool basic_const_item() const { return (*ref)->basic_const_item(); }
+  virtual bool basic_const_item() const { return ref && (*ref)->basic_const_item(); }
   bool is_outer_field() const
   {
     DBUG_ASSERT(fixed);
diff --git a/sql/lock.cc b/sql/lock.cc
index 0181a544824..2e141e1c9fb 100644
--- a/sql/lock.cc
+++ b/sql/lock.cc
@@ -777,8 +777,11 @@ bool lock_schema_name(THD *thd, const char *db)
     return TRUE;
   }
 
-  global_request.init(MDL_key::GLOBAL, "", "", MDL_INTENTION_EXCLUSIVE);
-  mdl_request.init(MDL_key::SCHEMA, db, "", MDL_EXCLUSIVE);
+  if (thd->global_read_lock.can_acquire_protection())
+    return TRUE;
+  global_request.init(MDL_key::GLOBAL, "", "", MDL_INTENTION_EXCLUSIVE,
+                      MDL_STATEMENT);
+  mdl_request.init(MDL_key::SCHEMA, db, "", MDL_EXCLUSIVE, MDL_TRANSACTION);
 
   mdl_requests.push_front(&mdl_request);
   mdl_requests.push_front(&global_request);
@@ -793,13 +796,13 @@ bool lock_schema_name(THD *thd, const char *db)
 
 
 /**
-  Obtain an exclusive metadata lock on the stored routine name.
+  Obtain an exclusive metadata lock on an object name.
 
   @param thd         Thread handle.
-  @param is_function Stored routine type (only functions or procedures
-                     are name-locked.
-  @param db          The schema the routine belongs to.
-  @param name        Routine name.
+  @param mdl_type    Object type (currently functions, procedures
+                     and events can be name-locked).
+  @param db          The schema the object belongs to.
+  @param name        Object name in the schema.
 
   This function assumes that no metadata locks were acquired
   before calling it. Additionally, it cannot be called while
@@ -815,12 +818,9 @@ bool lock_schema_name(THD *thd, const char *db)
                  or this connection was killed.
 */
 
-bool lock_routine_name(THD *thd, bool is_function,
+bool lock_object_name(THD *thd, MDL_key::enum_mdl_namespace mdl_type,
                        const char *db, const char *name)
 {
-  MDL_key::enum_mdl_namespace mdl_type= (is_function ?
-                                         MDL_key::FUNCTION :
-                                         MDL_key::PROCEDURE);
   MDL_request_list mdl_requests;
   MDL_request global_request;
   MDL_request schema_request;
@@ -836,9 +836,13 @@ 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);
-  schema_request.init(MDL_key::SCHEMA, db, "", MDL_INTENTION_EXCLUSIVE);
-  mdl_request.init(mdl_type, db, name, MDL_EXCLUSIVE);
+  if (thd->global_read_lock.can_acquire_protection())
+    return TRUE;
+  global_request.init(MDL_key::GLOBAL, "", "", MDL_INTENTION_EXCLUSIVE,
+                      MDL_STATEMENT);
+  schema_request.init(MDL_key::SCHEMA, db, "", MDL_INTENTION_EXCLUSIVE,
+                      MDL_TRANSACTION);
+  mdl_request.init(mdl_type, db, name, MDL_EXCLUSIVE, MDL_TRANSACTION);
 
   mdl_requests.push_front(&mdl_request);
   mdl_requests.push_front(&schema_request);
@@ -888,45 +892,24 @@ static void print_lock_error(int error, const char *table)
 /****************************************************************************
   Handling of global read locks
 
+  Global read lock is implemented using metadata lock infrastructure.
+
   Taking the global read lock is TWO steps (2nd step is optional; without
   it, COMMIT of existing transactions will be allowed):
   lock_global_read_lock() THEN make_global_read_lock_block_commit().
 
-  The global locks are handled through the global variables:
-  global_read_lock
-    count of threads which have the global read lock (i.e. have completed at
-    least the first step above)
-  global_read_lock_blocks_commit
-    count of threads which have the global read lock and block
-    commits (i.e. are in or have completed the second step above)
-  waiting_for_read_lock
-    count of threads which want to take a global read lock but cannot
-  protect_against_global_read_lock
-    count of threads which have set protection against global read lock.
-
-  access to them is protected with a mutex LOCK_global_read_lock
-
-  (XXX: one should never take LOCK_open if LOCK_global_read_lock is
-  taken, otherwise a deadlock may occur. Other mutexes could be a
-  problem too - grep the code for global_read_lock if you want to use
-  any other mutex here) Also one must not hold LOCK_open when calling
-  wait_if_global_read_lock(). When the thread with the global read lock
-  tries to close its tables, it needs to take LOCK_open in
-  close_thread_table().
-
   How blocking of threads by global read lock is achieved: that's
-  advisory. Any piece of code which should be blocked by global read lock must
-  be designed like this:
-  - call to wait_if_global_read_lock(). When this returns 0, no global read
-  lock is owned; if argument abort_on_refresh was 0, none can be obtained.
-  - job
-  - if abort_on_refresh was 0, call to start_waiting_global_read_lock() to
-  allow other threads to get the global read lock. I.e. removal of the
-  protection.
-  (Note: it's a bit like an implementation of rwlock).
-
-  [ I am sorry to mention some SQL syntaxes below I know I shouldn't but found
-  no better descriptive way ]
+  semi-automatic. We assume that any statement which should be blocked
+  by global read lock will either open and acquires write-lock on tables
+  or acquires metadata locks on objects it is going to modify. For any
+  such statement global IX metadata lock is automatically acquired for
+  its duration (in case of LOCK TABLES until end of LOCK TABLES mode).
+  And lock_global_read_lock() simply acquires global S metadata lock
+  and thus prohibits execution of statements which modify data (unless
+  they modify only temporary tables). If deadlock happens it is detected
+  by MDL subsystem and resolved in the standard fashion (by backing-off
+  metadata locks acquired so far and restarting open tables process
+  if possible).
 
   Why does FLUSH TABLES WITH READ LOCK need to block COMMIT: because it's used
   to read a non-moving SHOW MASTER STATUS, and a COMMIT writes to the binary
@@ -960,11 +943,6 @@ static void print_lock_error(int error, const char *table)
 
 ****************************************************************************/
 
-volatile uint global_read_lock=0;
-volatile uint global_read_lock_blocks_commit=0;
-static volatile uint protect_against_global_read_lock=0;
-static volatile uint waiting_for_read_lock=0;
-
 /**
   Take global read lock, wait if there is protection against lock.
 
@@ -985,85 +963,17 @@ bool Global_read_lock::lock_global_read_lock(THD *thd)
   if (!m_state)
   {
     MDL_request mdl_request;
-    const char *old_message;
-    const char *new_message= "Waiting to get readlock";
-    (void) mysql_mutex_lock(&LOCK_global_read_lock);
-
-#if defined(ENABLED_DEBUG_SYNC)
-    /*
-      The below sync point fires if we have to wait for
-      protect_against_global_read_lock.
-
-      WARNING: Beware to use WAIT_FOR with this sync point. We hold
-      LOCK_global_read_lock here.
-
-      Call the sync point before calling enter_cond() as it does use
-      enter_cond() and exit_cond() itself if a WAIT_FOR action is
-      executed in spite of the above warning.
-
-      Pre-set proc_info so that it is available immediately after the
-      sync point sends a SIGNAL. This makes tests more reliable.
-    */
-    if (protect_against_global_read_lock)
-    {
-      thd_proc_info(thd, new_message);
-      DEBUG_SYNC(thd, "wait_lock_global_read_lock");
-    }
-#endif /* defined(ENABLED_DEBUG_SYNC) */
-
-    old_message=thd->enter_cond(&COND_global_read_lock, &LOCK_global_read_lock,
-                                new_message);
-    DBUG_PRINT("info",
-	       ("waiting_for: %d  protect_against: %d",
-		waiting_for_read_lock, protect_against_global_read_lock));
-
-    waiting_for_read_lock++;
-    while (protect_against_global_read_lock && !thd->killed)
-      mysql_cond_wait(&COND_global_read_lock, &LOCK_global_read_lock);
-    waiting_for_read_lock--;
-    if (thd->killed)
-    {
-      thd->exit_cond(old_message);
-      DBUG_RETURN(1);
-    }
-    m_state= GRL_ACQUIRED;
-    global_read_lock++;
-    thd->exit_cond(old_message); // this unlocks LOCK_global_read_lock
-    /*
-      When we perform FLUSH TABLES or ALTER TABLE under LOCK TABLES,
-      tables being reopened are protected only by meta-data locks at
-      some point. To avoid sneaking in with our global read lock at
-      this moment we have to take global shared meta data lock.
-
-      TODO: We should change this code to acquire global shared metadata
-            lock before acquiring global read lock. But in order to do
-            this we have to get rid of all those places in which
-            wait_if_global_read_lock() is called before acquiring
-            metadata locks first. Also long-term we should get rid of
-            redundancy between metadata locks, global read lock and DDL
-            blocker (see WL#4399 and WL#4400).
-    */
 
     DBUG_ASSERT(! thd->mdl_context.is_lock_owner(MDL_key::GLOBAL, "", "",
                                                  MDL_SHARED));
-    mdl_request.init(MDL_key::GLOBAL, "", "", MDL_SHARED);
+    mdl_request.init(MDL_key::GLOBAL, "", "", MDL_SHARED, MDL_EXPLICIT);
 
     if (thd->mdl_context.acquire_lock(&mdl_request,
                                       thd->variables.lock_wait_timeout))
-    {
-      /* Our thread was killed -- return back to initial state. */
-      mysql_mutex_lock(&LOCK_global_read_lock);
-      if (!(--global_read_lock))
-      {
-        DBUG_PRINT("signal", ("Broadcasting COND_global_read_lock"));
-        mysql_cond_broadcast(&COND_global_read_lock);
-      }
-      mysql_mutex_unlock(&LOCK_global_read_lock);
-      m_state= GRL_NONE;
       DBUG_RETURN(1);
-    }
-    thd->mdl_context.move_ticket_after_trans_sentinel(mdl_request.ticket);
+
     m_mdl_global_shared_lock= mdl_request.ticket;
+    m_state= GRL_ACQUIRED;
   }
   /*
     We DON'T set global_read_lock_blocks_commit now, it will be set after
@@ -1089,166 +999,22 @@ bool Global_read_lock::lock_global_read_lock(THD *thd)
 
 void Global_read_lock::unlock_global_read_lock(THD *thd)
 {
-  uint tmp;
   DBUG_ENTER("unlock_global_read_lock");
-  DBUG_PRINT("info",
-             ("global_read_lock: %u  global_read_lock_blocks_commit: %u",
-              global_read_lock, global_read_lock_blocks_commit));
 
   DBUG_ASSERT(m_mdl_global_shared_lock && m_state);
 
-  thd->mdl_context.release_lock(m_mdl_global_shared_lock);
-  m_mdl_global_shared_lock= NULL;
-
-  mysql_mutex_lock(&LOCK_global_read_lock);
-  tmp= --global_read_lock;
-  if (m_state == GRL_ACQUIRED_AND_BLOCKS_COMMIT)
-    --global_read_lock_blocks_commit;
-  mysql_mutex_unlock(&LOCK_global_read_lock);
-  /* Send the signal outside the mutex to avoid a context switch */
-  if (!tmp)
+  if (m_mdl_blocks_commits_lock)
   {
-    DBUG_PRINT("signal", ("Broadcasting COND_global_read_lock"));
-    mysql_cond_broadcast(&COND_global_read_lock);
+    thd->mdl_context.release_lock(m_mdl_blocks_commits_lock);
+    m_mdl_blocks_commits_lock= NULL;
   }
+  thd->mdl_context.release_lock(m_mdl_global_shared_lock);
+  m_mdl_global_shared_lock= NULL;
   m_state= GRL_NONE;
 
   DBUG_VOID_RETURN;
 }
 
-/**
-  Wait if the global read lock is set, and optionally seek protection against
-  global read lock.
-
-  See also "Handling of global read locks" above.
-
-  @param thd              Reference to thread.
-  @param abort_on_refresh If True, abort waiting if a refresh occurs,
-                          do NOT seek protection against GRL.
-                          If False, wait until the GRL is released and seek
-                          protection against GRL.
-  @param is_not_commit    If False, called from a commit operation,
-                          wait only if commit blocking is also enabled.
-
-  @retval False  Success, protection against global read lock is set
-                          (if !abort_on_refresh)
-  @retval True   Failure, wait was aborted or thread was killed.
-*/
-
-#define must_wait (global_read_lock &&                             \
-                   (is_not_commit ||                               \
-                    global_read_lock_blocks_commit))
-
-bool Global_read_lock::
-wait_if_global_read_lock(THD *thd, bool abort_on_refresh,
-                         bool is_not_commit)
-{
-  const char *UNINIT_VAR(old_message);
-  bool result= 0, need_exit_cond;
-  DBUG_ENTER("wait_if_global_read_lock");
-
-  /*
-    If we already have protection against global read lock,
-    just increment the counter.
-  */
-  if (unlikely(m_protection_count > 0))
-  {
-    if (!abort_on_refresh)
-      m_protection_count++;
-    DBUG_RETURN(FALSE);
-  }
-  /*
-    Assert that we do not own LOCK_open. If we would own it, other
-    threads could not close their tables. This would make a pretty
-    deadlock.
-  */
-  mysql_mutex_assert_not_owner(&LOCK_open);
-
-  mysql_mutex_lock(&LOCK_global_read_lock);
-  if ((need_exit_cond= must_wait))
-  {
-    if (m_state)		// This thread had the read locks
-    {
-      if (is_not_commit)
-        my_message(ER_CANT_UPDATE_WITH_READLOCK,
-                   ER(ER_CANT_UPDATE_WITH_READLOCK), MYF(0));
-      mysql_mutex_unlock(&LOCK_global_read_lock);
-      /*
-        We allow FLUSHer to COMMIT; we assume FLUSHer knows what it does.
-        This allowance is needed to not break existing versions of innobackup
-        which do a BEGIN; INSERT; FLUSH TABLES WITH READ LOCK; COMMIT.
-      */
-      DBUG_RETURN(is_not_commit);
-    }
-    old_message=thd->enter_cond(&COND_global_read_lock, &LOCK_global_read_lock,
-				"Waiting for release of readlock");
-    while (must_wait && ! thd->killed &&
-	   (!abort_on_refresh || !thd->open_tables ||
-            thd->open_tables->s->version == refresh_version))
-    {
-      DBUG_PRINT("signal", ("Waiting for COND_global_read_lock"));
-      mysql_cond_wait(&COND_global_read_lock, &LOCK_global_read_lock);
-      DBUG_PRINT("signal", ("Got COND_global_read_lock"));
-    }
-    if (thd->killed)
-      result=1;
-  }
-  if (!abort_on_refresh && !result)
-  {
-    m_protection_count++;
-    protect_against_global_read_lock++;
-    DBUG_PRINT("sql_lock", ("protect_against_global_read_lock incr: %u",
-                            protect_against_global_read_lock));
-  }
-  /*
-    The following is only true in case of a global read locks (which is rare)
-    and if old_message is set
-  */
-  if (unlikely(need_exit_cond))
-    thd->exit_cond(old_message); // this unlocks LOCK_global_read_lock
-  else
-    mysql_mutex_unlock(&LOCK_global_read_lock);
-  DBUG_RETURN(result);
-}
-
-
-/**
-  Release protection against global read lock and restart
-  global read lock waiters.
-
-  Should only be called if we have protection against global read lock.
-
-  See also "Handling of global read locks" above.
-
-  @param thd     Reference to thread.
-*/
-
-void Global_read_lock::start_waiting_global_read_lock(THD *thd)
-{
-  bool tmp;
-  DBUG_ENTER("start_waiting_global_read_lock");
-  /*
-    Ignore request if we do not have protection against global read lock.
-    (Note that this is a violation of the interface contract, hence the assert).
-  */
-  DBUG_ASSERT(m_protection_count > 0);
-  if (unlikely(m_protection_count == 0))
-    DBUG_VOID_RETURN;
-  /* Decrement local read lock protection counter, return if we still have it */
-  if (unlikely(--m_protection_count > 0))
-    DBUG_VOID_RETURN;
-  if (unlikely(m_state))
-    DBUG_VOID_RETURN;
-  mysql_mutex_lock(&LOCK_global_read_lock);
-  DBUG_ASSERT(protect_against_global_read_lock);
-  tmp= (!--protect_against_global_read_lock &&
-        (waiting_for_read_lock || global_read_lock_blocks_commit));
-  mysql_mutex_unlock(&LOCK_global_read_lock);
-  if (tmp)
-    mysql_cond_broadcast(&COND_global_read_lock);
-  DBUG_VOID_RETURN;
-}
-
 
 /**
   Make global read lock also block commits.
@@ -1267,8 +1033,7 @@ void Global_read_lock::start_waiting_global_read_lock(THD *thd)
 
 bool Global_read_lock::make_global_read_lock_block_commit(THD *thd)
 {
-  bool error;
-  const char *old_message;
+  MDL_request mdl_request;
   DBUG_ENTER("make_global_read_lock_block_commit");
   /*
     If we didn't succeed lock_global_read_lock(), or if we already suceeded
@@ -1276,42 +1041,32 @@ bool Global_read_lock::make_global_read_lock_block_commit(THD *thd)
   */
   if (m_state != GRL_ACQUIRED)
     DBUG_RETURN(0);
-  mysql_mutex_lock(&LOCK_global_read_lock);
-  /* increment this BEFORE waiting on cond (otherwise race cond) */
-  global_read_lock_blocks_commit++;
-  /* For testing we set up some blocking, to see if we can be killed */
-  DBUG_EXECUTE_IF("make_global_read_lock_block_commit_loop",
-                  protect_against_global_read_lock++;);
-  old_message= thd->enter_cond(&COND_global_read_lock, &LOCK_global_read_lock,
-                               "Waiting for all running commits to finish");
-  while (protect_against_global_read_lock && !thd->killed)
-    mysql_cond_wait(&COND_global_read_lock, &LOCK_global_read_lock);
-  DBUG_EXECUTE_IF("make_global_read_lock_block_commit_loop",
-                  protect_against_global_read_lock--;);
-  if ((error= test(thd->killed)))
-    global_read_lock_blocks_commit--; // undo what we did
-  else
-    m_state= GRL_ACQUIRED_AND_BLOCKS_COMMIT;
-  thd->exit_cond(old_message); // this unlocks LOCK_global_read_lock
-  DBUG_RETURN(error);
+
+  mdl_request.init(MDL_key::COMMIT, "", "", MDL_SHARED, MDL_EXPLICIT);
+
+  if (thd->mdl_context.acquire_lock(&mdl_request,
+                                    thd->variables.lock_wait_timeout))
+    DBUG_RETURN(TRUE);
+
+  m_mdl_blocks_commits_lock= mdl_request.ticket;
+  m_state= GRL_ACQUIRED_AND_BLOCKS_COMMIT;
+
+  DBUG_RETURN(FALSE);
 }
 
 
 /**
-  Broadcast COND_global_read_lock.
-
-  TODO/FIXME: Dmitry thinks that we broadcast on COND_global_read_lock
-              when old instance of table is closed to avoid races
-              between incrementing refresh_version and
-              wait_if_global_read_lock(thd, TRUE, FALSE) call.
-              Once global read lock implementation starts using MDL
-              infrastructure this will became unnecessary and should
-              be removed.
+  Set explicit duration for metadata locks which are used to implement GRL.
+
+  @param thd     Reference to thread.
 */
 
-void broadcast_refresh(void)
+void Global_read_lock::set_explicit_lock_duration(THD *thd)
 {
-  mysql_cond_broadcast(&COND_global_read_lock);
+  if (m_mdl_global_shared_lock)
+    thd->mdl_context.set_lock_duration(m_mdl_global_shared_lock, MDL_EXPLICIT);
+  if (m_mdl_blocks_commits_lock)
+    thd->mdl_context.set_lock_duration(m_mdl_blocks_commits_lock, MDL_EXPLICIT);
 }
 
 /**
diff --git a/sql/lock.h b/sql/lock.h
index c097c8d269e..6f779595af8 100644
--- a/sql/lock.h
+++ b/sql/lock.h
@@ -2,6 +2,7 @@
 #define LOCK_INCLUDED
 
 #include "thr_lock.h"                           /* thr_lock_type */
+#include "mdl.h"
 
 // Forward declarations
 struct TABLE;
@@ -18,11 +19,10 @@ void mysql_lock_remove(THD *thd, MYSQL_LOCK *locked,TABLE *table);
 void mysql_lock_abort(THD *thd, TABLE *table, bool upgrade_lock);
 bool mysql_lock_abort_for_thread(THD *thd, TABLE *table);
 MYSQL_LOCK *mysql_lock_merge(MYSQL_LOCK *a,MYSQL_LOCK *b);
-void broadcast_refresh(void);
 /* Lock based on name */
 bool lock_schema_name(THD *thd, const char *db);
 /* Lock based on stored routine name */
-bool lock_routine_name(THD *thd, bool is_function, const char *db,
-                       const char *name);
+bool lock_object_name(THD *thd, MDL_key::enum_mdl_namespace mdl_type,
+                      const char *db, const char *name);
 
 #endif /* LOCK_INCLUDED */
diff --git a/sql/log_event.cc b/sql/log_event.cc
index d3aa94fcce2..537865197da 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -3366,6 +3366,8 @@ int Query_log_event::do_apply_event(Relay_log_info const *rli,
       if (!parser_state.init(thd, thd->query(), thd->query_length()))
       {
         mysql_parse(thd, thd->query(), thd->query_length(), &parser_state);
+        /* Finalize server status flags after executing a statement. */
+        thd->update_server_status();
         log_slow_statement(thd);
       }
 
@@ -4974,6 +4976,8 @@ error:
   */
   if (! thd->in_multi_stmt_transaction_mode())
     thd->mdl_context.release_transactional_locks();
+  else
+    thd->mdl_context.release_statement_locks();
 
   DBUG_EXECUTE_IF("LOAD_DATA_INFILE_has_fatal_error",
                   thd->is_slave_error= 0; thd->is_fatal_error= 1;);
diff --git a/sql/mdl.cc b/sql/mdl.cc
index d53ddcee0c8..3aed54ce12d 100644
--- a/sql/mdl.cc
+++ b/sql/mdl.cc
@@ -68,8 +68,6 @@ static void init_mdl_psi_keys(void)
 }
 #endif /* HAVE_PSI_INTERFACE */
 
-void notify_shared_lock(THD *thd, MDL_ticket *conflicting_ticket);
-
 
 /**
   Thread state names to be used in case when we have to wait on resource
@@ -78,12 +76,14 @@ void notify_shared_lock(THD *thd, MDL_ticket *conflicting_ticket);
 
 const char *MDL_key::m_namespace_to_wait_state_name[NAMESPACE_END]=
 {
-  "Waiting for global metadata lock",
+  "Waiting for global read lock",
   "Waiting for schema metadata lock",
   "Waiting for table metadata lock",
   "Waiting for stored function metadata lock",
   "Waiting for stored procedure metadata lock",
-  NULL
+  "Waiting for trigger metadata lock",
+  "Waiting for event metadata lock",
+  "Waiting for commit lock"
 };
 
 static bool mdl_initialized= 0;
@@ -100,7 +100,6 @@ 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:
@@ -110,6 +109,10 @@ private:
   HASH m_locks;
   /* Protects access to m_locks hash. */
   mysql_mutex_t m_mutex;
+  /** Pre-allocated MDL_lock object for GLOBAL namespace. */
+  MDL_lock *m_global_lock;
+  /** Pre-allocated MDL_lock object for COMMIT namespace. */
+  MDL_lock *m_commit_lock;
 };
 
 
@@ -354,18 +357,6 @@ public:
 
   inline static MDL_lock *create(const MDL_key *key);
 
-  void notify_shared_locks(MDL_context *ctx)
-  {
-    Ticket_iterator it(m_granted);
-    MDL_ticket *conflicting_ticket;
-
-    while ((conflicting_ticket= it++))
-    {
-      if (conflicting_ticket->get_ctx() != ctx)
-        notify_shared_lock(ctx->get_thd(), conflicting_ticket);
-    }
-  }
-
   void reschedule_waiters();
 
   void remove_ticket(Ticket_list MDL_lock::*queue, MDL_ticket *ticket);
@@ -373,6 +364,9 @@ public:
   bool visit_subgraph(MDL_ticket *waiting_ticket,
                       MDL_wait_for_graph_visitor *gvisitor);
 
+  virtual bool needs_notification(const MDL_ticket *ticket) const = 0;
+  virtual void notify_conflicting_locks(MDL_context *ctx) = 0;
+
   /** List of granted tickets for this lock. */
   Ticket_list m_granted;
   /** Tickets for contexts waiting to acquire a lock. */
@@ -442,6 +436,11 @@ public:
   {
     return m_waiting_incompatible;
   }
+  virtual bool needs_notification(const MDL_ticket *ticket) const
+  {
+    return (ticket->get_type() == MDL_SHARED);
+  }
+  virtual void notify_conflicting_locks(MDL_context *ctx);
 
 private:
   static const bitmap_t m_granted_incompatible[MDL_TYPE_END];
@@ -469,6 +468,11 @@ public:
   {
     return m_waiting_incompatible;
   }
+  virtual bool needs_notification(const MDL_ticket *ticket) const
+  {
+    return ticket->is_upgradable_or_exclusive();
+  }
+  virtual void notify_conflicting_locks(MDL_context *ctx);
 
 private:
   static const bitmap_t m_granted_incompatible[MDL_TYPE_END];
@@ -536,9 +540,14 @@ void mdl_destroy()
 
 void MDL_map::init()
 {
+  MDL_key global_lock_key(MDL_key::GLOBAL, "", "");
+  MDL_key commit_lock_key(MDL_key::COMMIT, "", "");
+
   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);
 }
 
 
@@ -552,6 +561,8 @@ void MDL_map::destroy()
   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);
 }
 
 
@@ -569,43 +580,28 @@ MDL_lock* MDL_map::find_or_insert(const MDL_key *mdl_key)
   MDL_lock *lock;
   my_hash_value_type hash_value;
 
-  hash_value= my_calc_hash(&m_locks, mdl_key->ptr(), mdl_key->length());
-
-retry:
-  mysql_mutex_lock(&m_mutex);
-  if (!(lock= (MDL_lock*) my_hash_search_using_hash_value(&m_locks,
-                                                          hash_value,
-                                                          mdl_key->ptr(),
-                                                          mdl_key->length())))
+  if (mdl_key->mdl_namespace() == MDL_key::GLOBAL ||
+      mdl_key->mdl_namespace() == MDL_key::COMMIT)
   {
-    lock= MDL_lock::create(mdl_key);
-    if (!lock || my_hash_insert(&m_locks, (uchar*)lock))
-    {
-      mysql_mutex_unlock(&m_mutex);
-      MDL_lock::destroy(lock);
-      return NULL;
-    }
-  }
-
-  if (move_from_hash_to_lock_mutex(lock))
-    goto retry;
+    /*
+      Avoid locking 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.
 
-  return lock;
-}
+      It works since these namespaces contain only one element so keys
+      for them look like '<namespace-id>\0\0'.
+    */
+    DBUG_ASSERT(mdl_key->length() == 3);
 
+    lock= (mdl_key->mdl_namespace() == MDL_key::GLOBAL) ? m_global_lock :
+                                                          m_commit_lock;
 
-/**
-  Find MDL_lock object corresponding to the key.
+    mysql_prlock_wrlock(&lock->m_rwlock);
 
-  @retval non-NULL - MDL_lock instance for the key with locked
-                     MDL_lock::m_rwlock.
-  @retval NULL     - There was no MDL_lock for the key.
-*/
+    return lock;
+  }
 
-MDL_lock* MDL_map::find(const MDL_key *mdl_key)
-{
-  MDL_lock *lock;
-  my_hash_value_type hash_value;
 
   hash_value= my_calc_hash(&m_locks, mdl_key->ptr(), mdl_key->length());
 
@@ -616,8 +612,13 @@ retry:
                                                           mdl_key->ptr(),
                                                           mdl_key->length())))
   {
-    mysql_mutex_unlock(&m_mutex);
-    return NULL;
+    lock= MDL_lock::create(mdl_key);
+    if (!lock || my_hash_insert(&m_locks, (uchar*)lock))
+    {
+      mysql_mutex_unlock(&m_mutex);
+      MDL_lock::destroy(lock);
+      return NULL;
+    }
   }
 
   if (move_from_hash_to_lock_mutex(lock))
@@ -684,6 +685,17 @@ void MDL_map::remove(MDL_lock *lock)
 {
   uint ref_usage, ref_release;
 
+  if (lock->key.mdl_namespace() == MDL_key::GLOBAL ||
+      lock->key.mdl_namespace() == MDL_key::COMMIT)
+  {
+    /*
+      Never destroy pre-allocated MDL_lock objects for GLOBAL and
+      COMMIT namespaces.
+    */
+    mysql_prlock_unlock(&lock->m_rwlock);
+    return;
+  }
+
   /*
     Destroy the MDL_lock object, but ensure that anyone that is
     holding a reference to the object is not remaining, if so he
@@ -719,8 +731,7 @@ void MDL_map::remove(MDL_lock *lock)
 */
 
 MDL_context::MDL_context()
-  :m_trans_sentinel(NULL),
-  m_thd(NULL),
+  : m_thd(NULL),
   m_needs_thr_lock_abort(FALSE),
   m_waiting_for(NULL)
 {
@@ -742,7 +753,9 @@ MDL_context::MDL_context()
 
 void MDL_context::destroy()
 {
-  DBUG_ASSERT(m_tickets.is_empty());
+  DBUG_ASSERT(m_tickets[MDL_STATEMENT].is_empty() &&
+              m_tickets[MDL_TRANSACTION].is_empty() &&
+              m_tickets[MDL_EXPLICIT].is_empty());
 
   mysql_prlock_destroy(&m_LOCK_waiting_for);
 }
@@ -771,10 +784,12 @@ void MDL_context::destroy()
 void MDL_request::init(MDL_key::enum_mdl_namespace mdl_namespace,
                        const char *db_arg,
                        const char *name_arg,
-                       enum enum_mdl_type mdl_type_arg)
+                       enum_mdl_type mdl_type_arg,
+                       enum_mdl_duration mdl_duration_arg)
 {
   key.mdl_key_init(mdl_namespace, db_arg, name_arg);
   type= mdl_type_arg;
+  duration= mdl_duration_arg;
   ticket= NULL;
 }
 
@@ -789,49 +804,17 @@ void MDL_request::init(MDL_key::enum_mdl_namespace mdl_namespace,
 */
 
 void MDL_request::init(const MDL_key *key_arg,
-                       enum enum_mdl_type mdl_type_arg)
+                       enum_mdl_type mdl_type_arg,
+                       enum_mdl_duration mdl_duration_arg)
 {
   key.mdl_key_init(key_arg);
   type= mdl_type_arg;
+  duration= mdl_duration_arg;
   ticket= NULL;
 }
 
 
 /**
-  Allocate and initialize one lock request.
-
-  Same as mdl_init_lock(), but allocates the lock and the key buffer
-  on a memory root. Necessary to lock ad-hoc tables, e.g.
-  mysql.* tables of grant and data dictionary subsystems.
-
-  @param  mdl_namespace  Id of namespace of object to be locked
-  @param  db             Name of database to which object belongs
-  @param  name           Name of of object
-  @param  root           MEM_ROOT on which object should be allocated
-
-  @note The allocated lock request will have MDL_SHARED type.
-
-  @retval 0      Error if out of memory
-  @retval non-0  Pointer to an object representing a lock request
-*/
-
-MDL_request *
-MDL_request::create(MDL_key::enum_mdl_namespace mdl_namespace, const char *db,
-                    const char *name, enum_mdl_type mdl_type,
-                    MEM_ROOT *root)
-{
-  MDL_request *mdl_request;
-
-  if (!(mdl_request= (MDL_request*) alloc_root(root, sizeof(MDL_request))))
-    return NULL;
-
-  mdl_request->init(mdl_namespace, db, name, mdl_type);
-
-  return mdl_request;
-}
-
-
-/**
   Auxiliary functions needed for creation/destruction of MDL_lock objects.
 
   @note Also chooses an MDL_lock descendant appropriate for object namespace.
@@ -846,6 +829,7 @@ inline MDL_lock *MDL_lock::create(const MDL_key *mdl_key)
   {
     case MDL_key::GLOBAL:
     case MDL_key::SCHEMA:
+    case MDL_key::COMMIT:
       return new MDL_scoped_lock(mdl_key);
     default:
       return new MDL_object_lock(mdl_key);
@@ -867,9 +851,17 @@ void MDL_lock::destroy(MDL_lock *lock)
         on memory allocation by reusing released objects.
 */
 
-MDL_ticket *MDL_ticket::create(MDL_context *ctx_arg, enum_mdl_type type_arg)
+MDL_ticket *MDL_ticket::create(MDL_context *ctx_arg, enum_mdl_type type_arg
+#ifndef DBUG_OFF
+                               , enum_mdl_duration duration_arg
+#endif
+                               )
 {
-  return new MDL_ticket(ctx_arg, type_arg);
+  return new MDL_ticket(ctx_arg, type_arg
+#ifndef DBUG_OFF
+                        , duration_arg
+#endif
+                        );
 }
 
 
@@ -1438,13 +1430,11 @@ bool MDL_ticket::is_incompatible_when_waiting(enum_mdl_type type) const
 /**
   Check whether the context already holds a compatible lock ticket
   on an object.
-  Start searching the transactional locks. If not
-  found in the list of transactional locks, look at LOCK TABLES
-  and HANDLER locks.
+  Start searching from list of locks for the same duration as lock
+  being requested. If not look at lists for other durations.
 
   @param mdl_request  Lock request object for lock to be acquired
-  @param[out] is_transactional FALSE if we pass beyond m_trans_sentinel
-                      while searching for ticket, otherwise TRUE.
+  @param[out] result_duration  Duration of lock which was found.
 
   @note Tickets which correspond to lock types "stronger" than one
         being requested are also considered compatible.
@@ -1454,24 +1444,28 @@ bool MDL_ticket::is_incompatible_when_waiting(enum_mdl_type type) const
 
 MDL_ticket *
 MDL_context::find_ticket(MDL_request *mdl_request,
-                         bool *is_transactional)
+                         enum_mdl_duration *result_duration)
 {
   MDL_ticket *ticket;
-  Ticket_iterator it(m_tickets);
-
-  *is_transactional= TRUE;
+  int i;
 
-  while ((ticket= it++))
+  for (i= 0; i < MDL_DURATION_END; i++)
   {
-    if (ticket == m_trans_sentinel)
-      *is_transactional= FALSE;
+    enum_mdl_duration duration= (enum_mdl_duration)((mdl_request->duration+i) %
+                                                    MDL_DURATION_END);
+    Ticket_iterator it(m_tickets[duration]);
 
-    if (mdl_request->key.is_equal(&ticket->m_lock->key) &&
-        ticket->has_stronger_or_equal_type(mdl_request->type))
-      break;
+    while ((ticket= it++))
+    {
+      if (mdl_request->key.is_equal(&ticket->m_lock->key) &&
+          ticket->has_stronger_or_equal_type(mdl_request->type))
+      {
+        *result_duration= duration;
+        return ticket;
+      }
+    }
   }
-
-  return ticket;
+  return NULL;
 }
 
 
@@ -1549,7 +1543,7 @@ MDL_context::try_acquire_lock_impl(MDL_request *mdl_request,
   MDL_lock *lock;
   MDL_key *key= &mdl_request->key;
   MDL_ticket *ticket;
-  bool is_transactional;
+  enum_mdl_duration found_duration;
 
   DBUG_ASSERT(mdl_request->type != MDL_EXCLUSIVE ||
               is_lock_owner(MDL_key::GLOBAL, "", "", MDL_INTENTION_EXCLUSIVE));
@@ -1563,7 +1557,7 @@ MDL_context::try_acquire_lock_impl(MDL_request *mdl_request,
     Check whether the context already holds a shared lock on the object,
     and if so, grant the request.
   */
-  if ((ticket= find_ticket(mdl_request, &is_transactional)))
+  if ((ticket= find_ticket(mdl_request, &found_duration)))
   {
     DBUG_ASSERT(ticket->m_lock);
     DBUG_ASSERT(ticket->has_stronger_or_equal_type(mdl_request->type));
@@ -1585,7 +1579,9 @@ MDL_context::try_acquire_lock_impl(MDL_request *mdl_request,
       a different alias.
     */
     mdl_request->ticket= ticket;
-    if (!is_transactional && clone_ticket(mdl_request))
+    if ((found_duration != mdl_request->duration ||
+         mdl_request->duration == MDL_EXPLICIT) &&
+        clone_ticket(mdl_request))
     {
       /* Clone failed. */
       mdl_request->ticket= NULL;
@@ -1594,7 +1590,11 @@ MDL_context::try_acquire_lock_impl(MDL_request *mdl_request,
     return FALSE;
   }
 
-  if (!(ticket= MDL_ticket::create(this, mdl_request->type)))
+  if (!(ticket= MDL_ticket::create(this, mdl_request->type
+#ifndef DBUG_OFF
+                                   , mdl_request->duration
+#endif
+                                   )))
     return TRUE;
 
   /* The below call implicitly locks MDL_lock::m_rwlock on success. */
@@ -1612,7 +1612,7 @@ MDL_context::try_acquire_lock_impl(MDL_request *mdl_request,
 
     mysql_prlock_unlock(&lock->m_rwlock);
 
-    m_tickets.push_front(ticket);
+    m_tickets[mdl_request->duration].push_front(ticket);
 
     mdl_request->ticket= ticket;
   }
@@ -1647,7 +1647,11 @@ MDL_context::clone_ticket(MDL_request *mdl_request)
     we effectively downgrade the cloned lock to the level of
     the request.
   */
-  if (!(ticket= MDL_ticket::create(this, mdl_request->type)))
+  if (!(ticket= MDL_ticket::create(this, mdl_request->type
+#ifndef DBUG_OFF
+                                   , mdl_request->duration
+#endif
+                                   )))
     return TRUE;
 
   /* clone() is not supposed to be used to get a stronger lock. */
@@ -1660,36 +1664,74 @@ MDL_context::clone_ticket(MDL_request *mdl_request)
   ticket->m_lock->m_granted.add_ticket(ticket);
   mysql_prlock_unlock(&ticket->m_lock->m_rwlock);
 
-  m_tickets.push_front(ticket);
+  m_tickets[mdl_request->duration].push_front(ticket);
 
   return FALSE;
 }
 
 
 /**
-  Notify a thread holding a shared metadata lock which
-  conflicts with a pending exclusive lock.
+  Notify threads holding a shared metadata locks on object which
+  conflict with a pending X, SNW or SNRW lock.
 
-  @param thd               Current thread context
-  @param conflicting_ticket  Conflicting metadata lock
+  @param  ctx  MDL_context for current thread.
 */
 
-void notify_shared_lock(THD *thd, MDL_ticket *conflicting_ticket)
+void MDL_object_lock::notify_conflicting_locks(MDL_context *ctx)
 {
-  /* Only try to abort locks on which we back off. */
-  if (conflicting_ticket->get_type() < MDL_SHARED_NO_WRITE)
+  Ticket_iterator it(m_granted);
+  MDL_ticket *conflicting_ticket;
+
+  while ((conflicting_ticket= it++))
   {
-    MDL_context *conflicting_ctx= conflicting_ticket->get_ctx();
-    THD *conflicting_thd= conflicting_ctx->get_thd();
-    DBUG_ASSERT(thd != conflicting_thd); /* Self-deadlock */
+    /* Only try to abort locks on which we back off. */
+    if (conflicting_ticket->get_ctx() != ctx &&
+        conflicting_ticket->get_type() < MDL_SHARED_NO_WRITE)
 
-    /*
-      If thread which holds conflicting lock is waiting on table-level
-      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(thd, conflicting_thd,
-                               conflicting_ctx->get_needs_thr_lock_abort());
+    {
+      MDL_context *conflicting_ctx= conflicting_ticket->get_ctx();
+
+      /*
+        If thread which holds conflicting lock is waiting on table-level
+        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());
+    }
+  }
+}
+
+
+/**
+  Notify threads holding scoped IX locks which conflict with a pending S lock.
+
+  @param  ctx  MDL_context for current thread.
+*/
+
+void MDL_scoped_lock::notify_conflicting_locks(MDL_context *ctx)
+{
+  Ticket_iterator it(m_granted);
+  MDL_ticket *conflicting_ticket;
+
+  while ((conflicting_ticket= it++))
+  {
+    if (conflicting_ticket->get_ctx() != ctx &&
+        conflicting_ticket->get_type() == MDL_INTENTION_EXCLUSIVE)
+
+    {
+      MDL_context *conflicting_ctx= conflicting_ticket->get_ctx();
+
+      /*
+        Thread which holds global IX lock can be a handler thread for
+        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());
+    }
   }
 }
 
@@ -1747,8 +1789,8 @@ MDL_context::acquire_lock(MDL_request *mdl_request, ulong lock_wait_timeout)
   */
   m_wait.reset_status();
 
-  if (ticket->is_upgradable_or_exclusive())
-    lock->notify_shared_locks(this);
+  if (lock->needs_notification(ticket))
+    lock->notify_conflicting_locks(this);
 
   mysql_prlock_unlock(&lock->m_rwlock);
 
@@ -1759,7 +1801,7 @@ MDL_context::acquire_lock(MDL_request *mdl_request, ulong lock_wait_timeout)
 
   find_deadlock();
 
-  if (ticket->is_upgradable_or_exclusive())
+  if (lock->needs_notification(ticket))
   {
     struct timespec abs_shortwait;
     set_timespec(abs_shortwait, 1);
@@ -1775,7 +1817,7 @@ MDL_context::acquire_lock(MDL_request *mdl_request, ulong lock_wait_timeout)
         break;
 
       mysql_prlock_wrlock(&lock->m_rwlock);
-      lock->notify_shared_locks(this);
+      lock->notify_conflicting_locks(this);
       mysql_prlock_unlock(&lock->m_rwlock);
       set_timespec(abs_shortwait, 1);
     }
@@ -1818,7 +1860,7 @@ MDL_context::acquire_lock(MDL_request *mdl_request, ulong lock_wait_timeout)
   */
   DBUG_ASSERT(wait_status == MDL_wait::GRANTED);
 
-  m_tickets.push_front(ticket);
+  m_tickets[mdl_request->duration].push_front(ticket);
 
   mdl_request->ticket= ticket;
 
@@ -1859,7 +1901,7 @@ bool MDL_context::acquire_locks(MDL_request_list *mdl_requests,
 {
   MDL_request_list::Iterator it(*mdl_requests);
   MDL_request **sort_buf, **p_req;
-  MDL_ticket *mdl_svp= mdl_savepoint();
+  MDL_savepoint mdl_svp= mdl_savepoint();
   ssize_t req_count= static_cast<ssize_t>(mdl_requests->elements());
 
   if (req_count == 0)
@@ -1928,7 +1970,7 @@ MDL_context::upgrade_shared_lock_to_exclusive(MDL_ticket *mdl_ticket,
                                               ulong lock_wait_timeout)
 {
   MDL_request mdl_xlock_request;
-  MDL_ticket *mdl_svp= mdl_savepoint();
+  MDL_savepoint mdl_svp= mdl_savepoint();
   bool is_new_ticket;
 
   DBUG_ENTER("MDL_ticket::upgrade_shared_lock_to_exclusive");
@@ -1945,7 +1987,8 @@ MDL_context::upgrade_shared_lock_to_exclusive(MDL_ticket *mdl_ticket,
   DBUG_ASSERT(mdl_ticket->m_type == MDL_SHARED_NO_WRITE ||
               mdl_ticket->m_type == MDL_SHARED_NO_READ_WRITE);
 
-  mdl_xlock_request.init(&mdl_ticket->m_lock->key, MDL_EXCLUSIVE);
+  mdl_xlock_request.init(&mdl_ticket->m_lock->key, MDL_EXCLUSIVE,
+                         MDL_TRANSACTION);
 
   if (acquire_lock(&mdl_xlock_request, lock_wait_timeout))
     DBUG_RETURN(TRUE);
@@ -1969,7 +2012,7 @@ MDL_context::upgrade_shared_lock_to_exclusive(MDL_ticket *mdl_ticket,
 
   if (is_new_ticket)
   {
-    m_tickets.remove(mdl_xlock_request.ticket);
+    m_tickets[MDL_TRANSACTION].remove(mdl_xlock_request.ticket);
     MDL_ticket::destroy(mdl_xlock_request.ticket);
   }
 
@@ -2230,10 +2273,12 @@ void MDL_context::find_deadlock()
 /**
   Release lock.
 
-  @param ticket Ticket for lock to be released.
+  @param duration Lock duration.
+  @param ticket   Ticket for lock to be released.
+
 */
 
-void MDL_context::release_lock(MDL_ticket *ticket)
+void MDL_context::release_lock(enum_mdl_duration duration, MDL_ticket *ticket)
 {
   MDL_lock *lock= ticket->m_lock;
   DBUG_ENTER("MDL_context::release_lock");
@@ -2243,12 +2288,9 @@ void MDL_context::release_lock(MDL_ticket *ticket)
   DBUG_ASSERT(this == ticket->get_ctx());
   mysql_mutex_assert_not_owner(&LOCK_open);
 
-  if (ticket == m_trans_sentinel)
-    m_trans_sentinel= ++Ticket_list::Iterator(m_tickets, ticket);
-
   lock->remove_ticket(&MDL_lock::m_granted, ticket);
 
-  m_tickets.remove(ticket);
+  m_tickets[duration].remove(ticket);
   MDL_ticket::destroy(ticket);
 
   DBUG_VOID_RETURN;
@@ -2256,50 +2298,56 @@ void MDL_context::release_lock(MDL_ticket *ticket)
 
 
 /**
+  Release lock with explicit duration.
+
+  @param ticket   Ticket for lock to be released.
+
+*/
+
+void MDL_context::release_lock(MDL_ticket *ticket)
+{
+  DBUG_ASSERT(ticket->m_duration == MDL_EXPLICIT);
+
+  release_lock(MDL_EXPLICIT, ticket);
+}
+
+
+/**
   Release all locks associated with the context. If the sentinel
   is not NULL, do not release locks stored in the list after and
   including the sentinel.
 
-  Transactional locks are added to the beginning of the list, i.e.
-  stored in reverse temporal order. This allows to employ this
-  function to:
+  Statement and transactional locks are added to the beginning of
+  the corresponding lists, i.e. stored in reverse temporal order.
+  This allows to employ this function to:
   - back off in case of a lock conflict.
-  - release all locks in the end of a transaction
+  - release all locks in the end of a statment or transaction
   - rollback to a savepoint.
-
-  The sentinel semantics is used to support LOCK TABLES
-  mode and HANDLER statements: locks taken by these statements
-  survive COMMIT, ROLLBACK, ROLLBACK TO SAVEPOINT.
 */
 
-void MDL_context::release_locks_stored_before(MDL_ticket *sentinel)
+void MDL_context::release_locks_stored_before(enum_mdl_duration duration,
+                                              MDL_ticket *sentinel)
 {
   MDL_ticket *ticket;
-  Ticket_iterator it(m_tickets);
+  Ticket_iterator it(m_tickets[duration]);
   DBUG_ENTER("MDL_context::release_locks_stored_before");
 
-  if (m_tickets.is_empty())
+  if (m_tickets[duration].is_empty())
     DBUG_VOID_RETURN;
 
   while ((ticket= it++) && ticket != sentinel)
   {
     DBUG_PRINT("info", ("found lock to release ticket=%p", ticket));
-    release_lock(ticket);
+    release_lock(duration, ticket);
   }
-  /*
-    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 all locks in the context which correspond to the same name/
-  object as this lock request.
+  Release all explicit locks in the context which correspond to the
+  same name/object as this lock request.
 
   @param ticket    One of the locks for the name/object for which all
                    locks should be released.
@@ -2312,17 +2360,13 @@ void MDL_context::release_all_locks_for_name(MDL_ticket *name)
 
   /* Remove matching lock tickets from the context. */
   MDL_ticket *ticket;
-  Ticket_iterator it_ticket(m_tickets);
+  Ticket_iterator it_ticket(m_tickets[MDL_EXPLICIT]);
 
   while ((ticket= it_ticket++))
   {
     DBUG_ASSERT(ticket->m_lock);
-    /*
-      We rarely have more than one ticket in this loop,
-      let's not bother saving on pthread_cond_broadcast().
-    */
     if (ticket->m_lock == lock)
-      release_lock(ticket);
+      release_lock(MDL_EXPLICIT, ticket);
   }
 }
 
@@ -2377,9 +2421,10 @@ MDL_context::is_lock_owner(MDL_key::enum_mdl_namespace mdl_namespace,
                            enum_mdl_type mdl_type)
 {
   MDL_request mdl_request;
-  bool is_transactional_unused;
-  mdl_request.init(mdl_namespace, db, name, mdl_type);
-  MDL_ticket *ticket= find_ticket(&mdl_request, &is_transactional_unused);
+  enum_mdl_duration not_unused;
+  /* We don't care about exact duration of lock here. */
+  mdl_request.init(mdl_namespace, db, name, mdl_type, MDL_TRANSACTION);
+  MDL_ticket *ticket= find_ticket(&mdl_request, &not_unused);
 
   DBUG_ASSERT(ticket == NULL || ticket->m_lock);
 
@@ -2408,18 +2453,15 @@ bool MDL_ticket::has_pending_conflicting_lock() const
   @note It's safe to iterate and unlock any locks after taken after this
         savepoint because other statements that take other special locks
         cause a implicit commit (ie LOCK TABLES).
-
-  @param mdl_savepont  The last acquired MDL lock when the
-                       savepoint was set.
 */
 
-void MDL_context::rollback_to_savepoint(MDL_ticket *mdl_savepoint)
+void MDL_context::rollback_to_savepoint(const MDL_savepoint &mdl_savepoint)
 {
   DBUG_ENTER("MDL_context::rollback_to_savepoint");
 
   /* If savepoint is NULL, it is from the start of the transaction. */
-  release_locks_stored_before(mdl_savepoint ?
-                              mdl_savepoint : m_trans_sentinel);
+  release_locks_stored_before(MDL_STATEMENT, mdl_savepoint.m_stmt_ticket);
+  release_locks_stored_before(MDL_TRANSACTION, mdl_savepoint.m_trans_ticket);
 
   DBUG_VOID_RETURN;
 }
@@ -2437,65 +2479,150 @@ void MDL_context::rollback_to_savepoint(MDL_ticket *mdl_savepoint)
 void MDL_context::release_transactional_locks()
 {
   DBUG_ENTER("MDL_context::release_transactional_locks");
-  release_locks_stored_before(m_trans_sentinel);
+  release_locks_stored_before(MDL_STATEMENT, NULL);
+  release_locks_stored_before(MDL_TRANSACTION, NULL);
+  DBUG_VOID_RETURN;
+}
+
+
+void MDL_context::release_statement_locks()
+{
+  DBUG_ENTER("MDL_context::release_transactional_locks");
+  release_locks_stored_before(MDL_STATEMENT, NULL);
   DBUG_VOID_RETURN;
 }
 
 
 /**
   Does this savepoint have this lock?
-  
-  @retval TRUE  The ticket is older than the savepoint and
-                is not LT, HA or GLR ticket. Thus it belongs
-                to the savepoint.
-  @retval FALSE The ticket is newer than the savepoint
-                or is an LT, HA or GLR ticket.
+
+  @retval TRUE  The ticket is older than the savepoint or
+                is an LT, HA or GLR ticket. Thus it belongs
+                to the savepoint or has explicit duration.
+  @retval FALSE The ticket is newer than the savepoint.
+                and is not an LT, HA or GLR ticket.
 */
 
-bool MDL_context::has_lock(MDL_ticket *mdl_savepoint,
+bool MDL_context::has_lock(const MDL_savepoint &mdl_savepoint,
                            MDL_ticket *mdl_ticket)
 {
   MDL_ticket *ticket;
   /* Start from the beginning, most likely mdl_ticket's been just acquired. */
-  MDL_context::Ticket_iterator it(m_tickets);
-  bool found_savepoint= FALSE;
+  MDL_context::Ticket_iterator s_it(m_tickets[MDL_STATEMENT]);
+  MDL_context::Ticket_iterator t_it(m_tickets[MDL_TRANSACTION]);
 
-  while ((ticket= it++) && ticket != m_trans_sentinel)
+  while ((ticket= s_it++) && ticket != mdl_savepoint.m_stmt_ticket)
   {
-    /*
-      First met the savepoint. The ticket must be
-      somewhere after it.
-    */
-    if (ticket == mdl_savepoint)
-      found_savepoint= TRUE;
-    /*
-      Met the ticket. If we haven't yet met the savepoint,
-      the ticket is newer than the savepoint.
-    */
     if (ticket == mdl_ticket)
-      return found_savepoint;
+      return FALSE;
   }
-  /* Reached m_trans_sentinel. The ticket must be LT, HA or GRL ticket. */
-  return FALSE;
+
+  while ((ticket= t_it++) && ticket != mdl_savepoint.m_trans_ticket)
+  {
+    if (ticket == mdl_ticket)
+      return FALSE;
+  }
+  return TRUE;
+}
+
+
+/**
+  Change lock duration for transactional lock.
+
+  @param ticket   Ticket representing lock.
+  @param duration Lock duration to be set.
+
+  @note This method only supports changing duration of
+        transactional lock to some other duration.
+*/
+
+void MDL_context::set_lock_duration(MDL_ticket *mdl_ticket,
+                                    enum_mdl_duration duration)
+{
+  DBUG_ASSERT(mdl_ticket->m_duration == MDL_TRANSACTION &&
+              duration != MDL_TRANSACTION);
+
+  m_tickets[MDL_TRANSACTION].remove(mdl_ticket);
+  m_tickets[duration].push_front(mdl_ticket);
+#ifndef DBUG_OFF
+  mdl_ticket->m_duration= duration;
+#endif
 }
 
 
 /**
-  Rearrange the ticket to reside in the part of the list that's
-  beyond m_trans_sentinel. This effectively changes the ticket
-  life cycle, from automatic to manual: i.e. the ticket is no
-  longer released by MDL_context::release_transactional_locks() or
-  MDL_context::rollback_to_savepoint(), it must be released manually.
+  Set explicit duration for all locks in the context.
 */
 
-void MDL_context::move_ticket_after_trans_sentinel(MDL_ticket *mdl_ticket)
+void MDL_context::set_explicit_duration_for_all_locks()
 {
-  m_tickets.remove(mdl_ticket);
-  if (m_trans_sentinel == NULL)
+  int i;
+  MDL_ticket *ticket;
+
+  /*
+    In the most common case when this function is called list
+    of transactional locks is bigger than list of locks with
+    explicit duration. So we start by swapping these two lists
+    and then move elements from new list of transactional
+    locks and list of statement locks to list of locks with
+    explicit duration.
+  */
+
+  m_tickets[MDL_EXPLICIT].swap(m_tickets[MDL_TRANSACTION]);
+
+  for (i= 0; i < MDL_EXPLICIT; i++)
   {
-    m_trans_sentinel= mdl_ticket;
-    m_tickets.push_back(mdl_ticket);
+    Ticket_iterator it_ticket(m_tickets[i]);
+
+    while ((ticket= it_ticket++))
+    {
+      m_tickets[i].remove(ticket);
+      m_tickets[MDL_EXPLICIT].push_front(ticket);
+    }
   }
-  else
-    m_tickets.insert_after(m_trans_sentinel, mdl_ticket);
+
+#ifndef DBUG_OFF
+  Ticket_iterator exp_it(m_tickets[MDL_EXPLICIT]);
+
+  while ((ticket= exp_it++))
+    ticket->m_duration= MDL_EXPLICIT;
+#endif
+}
+
+
+/**
+  Set transactional duration for all locks in the context.
+*/
+
+void MDL_context::set_transaction_duration_for_all_locks()
+{
+  MDL_ticket *ticket;
+
+  /*
+    In the most common case when this function is called list
+    of explicit locks is bigger than two other lists (in fact,
+    list of statement locks is always empty). So we start by
+    swapping list of explicit and transactional locks and then
+    move contents of new list of explicit locks to list of
+    locks with transactional duration.
+  */
+
+  DBUG_ASSERT(m_tickets[MDL_STATEMENT].is_empty());
+
+  m_tickets[MDL_TRANSACTION].swap(m_tickets[MDL_EXPLICIT]);
+
+  Ticket_iterator it_ticket(m_tickets[MDL_EXPLICIT]);
+
+  while ((ticket= it_ticket++))
+  {
+    m_tickets[MDL_EXPLICIT].remove(ticket);
+    m_tickets[MDL_TRANSACTION].push_front(ticket);
+  }
+
+#ifndef DBUG_OFF
+  Ticket_iterator trans_it(m_tickets[MDL_TRANSACTION]);
+
+  while ((ticket= trans_it++))
+    ticket->m_duration= MDL_TRANSACTION;
+#endif
 }
diff --git a/sql/mdl.h b/sql/mdl.h
index 7938d833eac..ff30746e739 100644
--- a/sql/mdl.h
+++ b/sql/mdl.h
@@ -150,6 +150,15 @@ enum enum_mdl_type {
   MDL_TYPE_END};
 
 
+/** Duration of metadata lock. */
+
+enum enum_mdl_duration { MDL_STATEMENT= 0,
+                         MDL_TRANSACTION,
+                         MDL_EXPLICIT,
+                         /* This should be the last ! */
+                         MDL_DURATION_END };
+
+
 /** Maximal length of key for metadata locking subsystem. */
 #define MAX_MDLKEY_LENGTH (1 + NAME_LEN + 1 + NAME_LEN + 1)
 
@@ -167,13 +176,16 @@ class MDL_key
 {
 public:
   /**
-    Object namespaces
+    Object namespaces.
+    Sic: when adding a new member to this enum make sure to
+    update m_namespace_to_wait_state_name array in mdl.cc!
 
     Different types of objects exist in different namespaces
      - TABLE is for tables and views.
      - FUNCTION is for stored functions.
      - PROCEDURE is for stored procedures.
      - TRIGGER is for triggers.
+     - EVENT is for event scheduler events
     Note that although there isn't metadata locking on triggers,
     it's necessary to have a separate namespace for them since
     MDL_key is also used outside of the MDL subsystem.
@@ -184,6 +196,8 @@ public:
                             FUNCTION,
                             PROCEDURE,
                             TRIGGER,
+                            EVENT,
+                            COMMIT,
                             /* This should be the last ! */
                             NAMESPACE_END };
 
@@ -305,6 +319,8 @@ class MDL_request
 public:
   /** Type of metadata lock. */
   enum          enum_mdl_type type;
+  /** Duration for requested lock. */
+  enum enum_mdl_duration duration;
 
   /**
     Pointers for participating in the list of lock requests for this context.
@@ -327,17 +343,16 @@ public:
 
   void init(MDL_key::enum_mdl_namespace namespace_arg,
             const char *db_arg, const char *name_arg,
-            enum_mdl_type mdl_type_arg);
-  void init(const MDL_key *key_arg, enum_mdl_type mdl_type_arg);
+            enum_mdl_type mdl_type_arg,
+            enum_mdl_duration mdl_duration_arg);
+  void init(const MDL_key *key_arg, enum_mdl_type mdl_type_arg,
+            enum_mdl_duration mdl_duration_arg);
   /** Set type of lock request. Can be only applied to pending locks. */
   inline void set_type(enum_mdl_type type_arg)
   {
     DBUG_ASSERT(ticket == NULL);
     type= type_arg;
   }
-  static MDL_request *create(MDL_key::enum_mdl_namespace mdl_namespace,
-                             const char *db, const char *name,
-                             enum_mdl_type mdl_type, MEM_ROOT *root);
 
   /*
     This is to work around the ugliness of TABLE_LIST
@@ -363,6 +378,7 @@ public:
 
   MDL_request(const MDL_request *rhs)
     :type(rhs->type),
+    duration(rhs->duration),
     ticket(NULL),
     key(&rhs->key)
   {}
@@ -484,17 +500,35 @@ public:
 private:
   friend class MDL_context;
 
-  MDL_ticket(MDL_context *ctx_arg, enum_mdl_type type_arg)
+  MDL_ticket(MDL_context *ctx_arg, enum_mdl_type type_arg
+#ifndef DBUG_OFF
+             , enum_mdl_duration duration_arg
+#endif
+            )
    : m_type(type_arg),
+#ifndef DBUG_OFF
+     m_duration(duration_arg),
+#endif
      m_ctx(ctx_arg),
      m_lock(NULL)
   {}
 
-  static MDL_ticket *create(MDL_context *ctx_arg, enum_mdl_type type_arg);
+  static MDL_ticket *create(MDL_context *ctx_arg, enum_mdl_type type_arg
+#ifndef DBUG_OFF
+                            , enum_mdl_duration duration_arg
+#endif
+                            );
   static void destroy(MDL_ticket *ticket);
 private:
   /** Type of metadata lock. Externally accessible. */
   enum enum_mdl_type m_type;
+#ifndef DBUG_OFF
+  /**
+    Duration of lock represented by this ticket.
+    Context private. Debug-only.
+  */
+  enum_mdl_duration m_duration;
+#endif
   /**
     Context of the owner of the metadata lock ticket. Externally accessible.
   */
@@ -512,6 +546,39 @@ private:
 
 
 /**
+  Savepoint for MDL context.
+
+  Doesn't include metadata locks with explicit duration as
+  they are not released during rollback to savepoint.
+*/
+
+class MDL_savepoint
+{
+public:
+  MDL_savepoint() {};
+
+private:
+  MDL_savepoint(MDL_ticket *stmt_ticket, MDL_ticket *trans_ticket)
+    : m_stmt_ticket(stmt_ticket), m_trans_ticket(trans_ticket)
+  {}
+
+  friend class MDL_context;
+
+private:
+  /**
+    Pointer to last lock with statement duration which was taken
+    before creation of savepoint.
+  */
+  MDL_ticket *m_stmt_ticket;
+  /**
+    Pointer to last lock with transaction duration which was taken
+    before creation of savepoint.
+  */
+  MDL_ticket *m_trans_ticket;
+};
+
+
+/**
   A reliable way to wait on an MDL lock.
 */
 
@@ -559,9 +626,7 @@ public:
   typedef I_P_List<MDL_ticket,
                    I_P_List_adapter<MDL_ticket,
                                     &MDL_ticket::next_in_context,
-                                    &MDL_ticket::prev_in_context>,
-                   I_P_List_null_counter,
-                   I_P_List_fast_push_back<MDL_ticket> >
+                                    &MDL_ticket::prev_in_context> >
           Ticket_list;
 
   typedef Ticket_list::Iterator Ticket_iterator;
@@ -584,37 +649,28 @@ public:
                      const char *db, const char *name,
                      enum_mdl_type mdl_type);
 
-  bool has_lock(MDL_ticket *mdl_savepoint, MDL_ticket *mdl_ticket);
+  bool has_lock(const MDL_savepoint &mdl_savepoint, MDL_ticket *mdl_ticket);
 
   inline bool has_locks() const
   {
-    return !m_tickets.is_empty();
-  }
-
-  MDL_ticket *mdl_savepoint()
-  {
-    /*
-      NULL savepoint represents the start of the transaction.
-      Checking for m_trans_sentinel also makes sure we never
-      return a pointer to HANDLER ticket as a savepoint.
-    */
-    return m_tickets.front() == m_trans_sentinel ? NULL : m_tickets.front();
+    return !(m_tickets[MDL_STATEMENT].is_empty() &&
+             m_tickets[MDL_TRANSACTION].is_empty() &&
+             m_tickets[MDL_EXPLICIT].is_empty());
   }
 
-  void set_trans_sentinel()
+  MDL_savepoint mdl_savepoint()
   {
-    m_trans_sentinel= m_tickets.front();
+    return MDL_savepoint(m_tickets[MDL_STATEMENT].front(),
+                         m_tickets[MDL_TRANSACTION].front());
   }
-  MDL_ticket *trans_sentinel() const { return m_trans_sentinel; }
 
-  void reset_trans_sentinel(MDL_ticket *sentinel_arg)
-  {
-    m_trans_sentinel= sentinel_arg;
-  }
-  void move_ticket_after_trans_sentinel(MDL_ticket *mdl_ticket);
+  void set_explicit_duration_for_all_locks();
+  void set_transaction_duration_for_all_locks();
+  void set_lock_duration(MDL_ticket *mdl_ticket, enum_mdl_duration duration);
 
+  void release_statement_locks();
   void release_transactional_locks();
-  void rollback_to_savepoint(MDL_ticket *mdl_savepoint);
+  void rollback_to_savepoint(const MDL_savepoint &mdl_savepoint);
 
   inline THD *get_thd() const { return m_thd; }
 
@@ -655,46 +711,43 @@ public:
   MDL_wait m_wait;
 private:
   /**
-    All MDL tickets acquired by this connection.
-
-    The order of tickets in m_tickets list.
-    ---------------------------------------
-    The entire set of locks acquired by a connection
-    can be separated in two subsets: transactional and
-    non-transactional locks.
-
-    Transactional locks are locks with automatic scope. They
-    are accumulated in the course of a transaction, and
-    released only on COMMIT, ROLLBACK or ROLLBACK TO SAVEPOINT.
-    They must not be (and never are) released manually,
+    Lists of all MDL tickets acquired by this connection.
+
+    Lists of MDL tickets:
+    ---------------------
+    The entire set of locks acquired by a connection can be separated
+    in three subsets according to their: locks released at the end of
+    statement, at the end of transaction and locks are released
+    explicitly.
+
+    Statement and transactional locks are locks with automatic scope.
+    They are accumulated in the course of a transaction, and released
+    either at the end of uppermost statement (for statement locks) or
+    on COMMIT, ROLLBACK or ROLLBACK TO SAVEPOINT (for transactional
+    locks). They must not be (and never are) released manually,
     i.e. with release_lock() call.
 
-    Non-transactional locks are taken for locks that span
+    Locks with explicit duration are taken for locks that span
     multiple transactions or savepoints.
     These are: HANDLER SQL locks (HANDLER SQL is
     transaction-agnostic), LOCK TABLES locks (you can COMMIT/etc
     under LOCK TABLES, and the locked tables stay locked), and
-    SET GLOBAL READ_ONLY=1 global shared lock.
+    locks implementing "global read lock".
 
-    Transactional locks are always prepended to the beginning
-    of the list. In other words, they are stored in reverse
-    temporal order. Thus, when we rollback to a savepoint,
-    we start popping and releasing tickets from the front
-    until we reach the last ticket acquired after the
-    savepoint.
+    Statement/transactional locks are always prepended to the
+    beginning of the appropriate list. In other words, they are
+    stored in reverse temporal order. Thus, when we rollback to
+    a savepoint, we start popping and releasing tickets from the
+    front until we reach the last ticket acquired after the savepoint.
 
-    Non-transactional locks are always stored after
-    transactional ones, and among each other can be
-    split into three sets:
+    Locks with explicit duration stored are not stored in any
+    particular order, and among each other can be split into
+    three sets:
 
     [LOCK TABLES locks] [HANDLER locks] [GLOBAL READ LOCK locks]
 
     The following is known about these sets:
 
-    * we can never have both HANDLER and LOCK TABLES locks
-      together -- HANDLER statements are prohibited under LOCK
-      TABLES, entering LOCK TABLES implicitly closes all open
-      HANDLERs.
     * GLOBAL READ LOCK locks are always stored after LOCK TABLES
       locks and after HANDLER locks. This is because one can't say
       SET GLOBAL read_only=1 or FLUSH TABLES WITH READ LOCK
@@ -709,14 +762,9 @@ private:
       However, one can open a few HANDLERs after entering the
       read only mode.
     * LOCK TABLES locks include intention exclusive locks on
-      involved schemas.
-  */
-  Ticket_list m_tickets;
-  /**
-    Separates transactional and non-transactional locks
-    in m_tickets list, @sa m_tickets.
+      involved schemas and global intention exclusive lock.
   */
-  MDL_ticket *m_trans_sentinel;
+  Ticket_list m_tickets[MDL_DURATION_END];
   THD *m_thd;
   /**
     TRUE -  if for this context we will break protocol and try to
@@ -747,8 +795,9 @@ private:
   MDL_wait_for_subgraph *m_waiting_for;
 private:
   MDL_ticket *find_ticket(MDL_request *mdl_req,
-                          bool *is_transactional);
-  void release_locks_stored_before(MDL_ticket *sentinel);
+                          enum_mdl_duration *duration);
+  void release_locks_stored_before(enum_mdl_duration duration, MDL_ticket *sentinel);
+  void release_lock(enum_mdl_duration duration, MDL_ticket *ticket);
   bool try_acquire_lock_impl(MDL_request *mdl_request,
                              MDL_ticket **out_ticket);
 
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index 04e61635f22..9932fa49ab3 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -604,8 +604,7 @@ pthread_key(MEM_ROOT**,THR_MALLOC);
 pthread_key(THD*, THR_THD);
 mysql_mutex_t LOCK_thread_count;
 mysql_mutex_t
-  LOCK_status, LOCK_global_read_lock,
-  LOCK_error_log, LOCK_uuid_generator,
+  LOCK_status, LOCK_error_log, LOCK_uuid_generator,
   LOCK_delayed_insert, LOCK_delayed_status, LOCK_delayed_create,
   LOCK_crypt,
   LOCK_global_system_variables,
@@ -625,7 +624,6 @@ mysql_mutex_t LOCK_des_key_file;
 mysql_rwlock_t LOCK_grant, LOCK_sys_init_connect, LOCK_sys_init_slave;
 mysql_rwlock_t LOCK_system_variables_hash;
 mysql_cond_t COND_thread_count;
-mysql_cond_t COND_global_read_lock;
 pthread_t signal_thread;
 pthread_attr_t connection_attrib;
 mysql_mutex_t LOCK_server_started;
@@ -1542,7 +1540,6 @@ static void clean_up_mutexes()
   mysql_mutex_destroy(&LOCK_crypt);
   mysql_mutex_destroy(&LOCK_user_conn);
   mysql_mutex_destroy(&LOCK_connection_count);
-  Events::destroy_mutexes();
 #ifdef HAVE_OPENSSL
   mysql_mutex_destroy(&LOCK_des_key_file);
 #ifndef HAVE_YASSL
@@ -1560,12 +1557,10 @@ static void clean_up_mutexes()
   mysql_rwlock_destroy(&LOCK_sys_init_slave);
   mysql_mutex_destroy(&LOCK_global_system_variables);
   mysql_rwlock_destroy(&LOCK_system_variables_hash);
-  mysql_mutex_destroy(&LOCK_global_read_lock);
   mysql_mutex_destroy(&LOCK_uuid_generator);
   mysql_mutex_destroy(&LOCK_prepared_stmt_count);
   mysql_mutex_destroy(&LOCK_error_messages);
   mysql_cond_destroy(&COND_thread_count);
-  mysql_cond_destroy(&COND_global_read_lock);
   mysql_cond_destroy(&COND_thread_cache);
   mysql_cond_destroy(&COND_flush_thread_cache);
   mysql_cond_destroy(&COND_manager);
@@ -3524,8 +3519,6 @@ static int init_thread_environment()
                    &LOCK_global_system_variables, MY_MUTEX_INIT_FAST);
   mysql_rwlock_init(key_rwlock_LOCK_system_variables_hash,
                     &LOCK_system_variables_hash);
-  mysql_mutex_init(key_LOCK_global_read_lock,
-                   &LOCK_global_read_lock, MY_MUTEX_INIT_FAST);
   mysql_mutex_init(key_LOCK_prepared_stmt_count,
                    &LOCK_prepared_stmt_count, MY_MUTEX_INIT_FAST);
   mysql_mutex_init(key_LOCK_error_messages,
@@ -3553,7 +3546,6 @@ static int init_thread_environment()
   mysql_rwlock_init(key_rwlock_LOCK_sys_init_slave, &LOCK_sys_init_slave);
   mysql_rwlock_init(key_rwlock_LOCK_grant, &LOCK_grant);
   mysql_cond_init(key_COND_thread_count, &COND_thread_count, NULL);
-  mysql_cond_init(key_COND_global_read_lock, &COND_global_read_lock, NULL);
   mysql_cond_init(key_COND_thread_cache, &COND_thread_cache, NULL);
   mysql_cond_init(key_COND_flush_thread_cache, &COND_flush_thread_cache, NULL);
   mysql_cond_init(key_COND_manager, &COND_manager, NULL);
@@ -7669,7 +7661,7 @@ PSI_mutex_key key_BINLOG_LOCK_index, key_BINLOG_LOCK_prep_xids,
   key_delayed_insert_mutex, key_hash_filo_lock, key_LOCK_active_mi,
   key_LOCK_connection_count, key_LOCK_crypt, key_LOCK_delayed_create,
   key_LOCK_delayed_insert, key_LOCK_delayed_status, key_LOCK_error_log,
-  key_LOCK_gdl, key_LOCK_global_read_lock, key_LOCK_global_system_variables,
+  key_LOCK_gdl, key_LOCK_global_system_variables,
   key_LOCK_manager,
   key_LOCK_prepared_stmt_count,
   key_LOCK_rpl_status, key_LOCK_server_started, key_LOCK_status,
@@ -7707,7 +7699,6 @@ static PSI_mutex_info all_server_mutexes[]=
   { &key_LOCK_delayed_status, "LOCK_delayed_status", PSI_FLAG_GLOBAL},
   { &key_LOCK_error_log, "LOCK_error_log", PSI_FLAG_GLOBAL},
   { &key_LOCK_gdl, "LOCK_gdl", PSI_FLAG_GLOBAL},
-  { &key_LOCK_global_read_lock, "LOCK_global_read_lock", PSI_FLAG_GLOBAL},
   { &key_LOCK_global_system_variables, "LOCK_global_system_variables", PSI_FLAG_GLOBAL},
   { &key_LOCK_manager, "LOCK_manager", PSI_FLAG_GLOBAL},
   { &key_LOCK_prepared_stmt_count, "LOCK_prepared_stmt_count", PSI_FLAG_GLOBAL},
@@ -7756,7 +7747,7 @@ PSI_cond_key key_PAGE_cond, key_COND_active, key_COND_pool;
 #endif /* HAVE_MMAP */
 
 PSI_cond_key key_BINLOG_COND_prep_xids, key_BINLOG_update_cond,
-  key_COND_cache_status_changed, key_COND_global_read_lock, key_COND_manager,
+  key_COND_cache_status_changed, key_COND_manager,
   key_COND_rpl_status, key_COND_server_started,
   key_delayed_insert_cond, key_delayed_insert_cond_client,
   key_item_func_sleep_cond, key_master_info_data_cond,
@@ -7779,7 +7770,6 @@ static PSI_cond_info all_server_conds[]=
   { &key_BINLOG_COND_prep_xids, "MYSQL_BIN_LOG::COND_prep_xids", 0},
   { &key_BINLOG_update_cond, "MYSQL_BIN_LOG::update_cond", 0},
   { &key_COND_cache_status_changed, "Query_cache::COND_cache_status_changed", 0},
-  { &key_COND_global_read_lock, "COND_global_read_lock", PSI_FLAG_GLOBAL},
   { &key_COND_manager, "COND_manager", PSI_FLAG_GLOBAL},
   { &key_COND_rpl_status, "COND_rpl_status", PSI_FLAG_GLOBAL},
   { &key_COND_server_started, "COND_server_started", PSI_FLAG_GLOBAL},
diff --git a/sql/mysqld.h b/sql/mysqld.h
index 113bc3aa983..376d8440619 100644
--- a/sql/mysqld.h
+++ b/sql/mysqld.h
@@ -100,7 +100,7 @@ extern bool opt_ignore_builtin_innodb;
 extern my_bool opt_character_set_client_handshake;
 extern bool volatile abort_loop;
 extern bool in_bootstrap;
-extern uint volatile thread_count, global_read_lock;
+extern uint volatile thread_count;
 extern uint connection_count;
 extern my_bool opt_safe_user_create;
 extern my_bool opt_safe_show_db, opt_local_infile, opt_myisam_use_mmap;
@@ -227,7 +227,7 @@ extern PSI_mutex_key key_BINLOG_LOCK_index, key_BINLOG_LOCK_prep_xids,
   key_delayed_insert_mutex, key_hash_filo_lock, key_LOCK_active_mi,
   key_LOCK_connection_count, key_LOCK_crypt, key_LOCK_delayed_create,
   key_LOCK_delayed_insert, key_LOCK_delayed_status, key_LOCK_error_log,
-  key_LOCK_gdl, key_LOCK_global_read_lock, key_LOCK_global_system_variables,
+  key_LOCK_gdl, key_LOCK_global_system_variables,
   key_LOCK_logger, key_LOCK_manager,
   key_LOCK_prepared_stmt_count,
   key_LOCK_rpl_status, key_LOCK_server_started, key_LOCK_status,
@@ -248,7 +248,7 @@ extern PSI_cond_key key_PAGE_cond, key_COND_active, key_COND_pool;
 #endif /* HAVE_MMAP */
 
 extern PSI_cond_key key_BINLOG_COND_prep_xids, key_BINLOG_update_cond,
-  key_COND_cache_status_changed, key_COND_global_read_lock, key_COND_manager,
+  key_COND_cache_status_changed, key_COND_manager,
   key_COND_rpl_status, key_COND_server_started,
   key_delayed_insert_cond, key_delayed_insert_cond_client,
   key_item_func_sleep_cond, key_master_info_data_cond,
@@ -321,7 +321,7 @@ extern mysql_mutex_t
        LOCK_user_locks, LOCK_status,
        LOCK_error_log, LOCK_delayed_insert, LOCK_uuid_generator,
        LOCK_delayed_status, LOCK_delayed_create, LOCK_crypt, LOCK_timezone,
-       LOCK_slave_list, LOCK_active_mi, LOCK_manager, LOCK_global_read_lock,
+       LOCK_slave_list, LOCK_active_mi, LOCK_manager,
        LOCK_global_system_variables, LOCK_user_conn,
        LOCK_prepared_stmt_count, LOCK_error_messages, LOCK_connection_count;
 extern MYSQL_PLUGIN_IMPORT mysql_mutex_t LOCK_thread_count;
@@ -334,7 +334,6 @@ extern mysql_rwlock_t LOCK_grant, LOCK_sys_init_connect, LOCK_sys_init_slave;
 extern mysql_rwlock_t LOCK_system_variables_hash;
 extern mysql_cond_t COND_thread_count;
 extern mysql_cond_t COND_manager;
-extern mysql_cond_t COND_global_read_lock;
 extern int32 thread_running;
 extern my_atomic_rwlock_t thread_running_lock;
 
diff --git a/sql/protocol.cc b/sql/protocol.cc
index dd3a5d92a87..03b151e4346 100644
--- a/sql/protocol.cc
+++ b/sql/protocol.cc
@@ -505,11 +505,11 @@ void Protocol::end_statement()
                       thd->stmt_da->get_sqlstate());
     break;
   case Diagnostics_area::DA_EOF:
-    error= send_eof(thd->stmt_da->server_status(),
+    error= send_eof(thd->server_status,
                     thd->stmt_da->statement_warn_count());
     break;
   case Diagnostics_area::DA_OK:
-    error= send_ok(thd->stmt_da->server_status(),
+    error= send_ok(thd->server_status,
                    thd->stmt_da->statement_warn_count(),
                    thd->stmt_da->affected_rows(),
                    thd->stmt_da->last_insert_id(),
diff --git a/sql/rpl_rli.cc b/sql/rpl_rli.cc
index af9b452acd8..a35e7bb1612 100644
--- a/sql/rpl_rli.cc
+++ b/sql/rpl_rli.cc
@@ -1274,6 +1274,9 @@ void Relay_log_info::slave_close_thread_tables(THD *thd)
   */
   if (! thd->in_multi_stmt_transaction_mode())
     thd->mdl_context.release_transactional_locks();
+  else
+    thd->mdl_context.release_statement_locks();
+
   clear_tables_to_lock();
 }
 #endif
diff --git a/sql/slave.cc b/sql/slave.cc
index 8874eb7f9bb..c4be7c7a6f7 100644
--- a/sql/slave.cc
+++ b/sql/slave.cc
@@ -721,9 +721,17 @@ int start_slave_thread(
     while (start_id == *slave_run_id)
     {
       DBUG_PRINT("sleep",("Waiting for slave thread to start"));
-      const char* old_msg = thd->enter_cond(start_cond,cond_lock,
-                                            "Waiting for slave thread to start");
-      mysql_cond_wait(start_cond, cond_lock);
+      const char *old_msg= thd->enter_cond(start_cond, cond_lock,
+                                           "Waiting for slave thread to start");
+      /*
+        It is not sufficient to test this at loop bottom. We must test
+        it after registering the mutex in enter_cond(). If the kill
+        happens after testing of thd->killed and before the mutex is
+        registered, we could otherwise go waiting though thd->killed is
+        set.
+      */
+      if (!thd->killed)
+        mysql_cond_wait(start_cond, cond_lock);
       thd->exit_cond(old_msg);
       mysql_mutex_lock(cond_lock); // re-acquire it as exit_cond() released
       if (thd->killed)
diff --git a/sql/sp.cc b/sql/sp.cc
index 7385a6ffcae..5d424564317 100644
--- a/sql/sp.cc
+++ b/sql/sp.cc
@@ -27,7 +27,7 @@
 #include "sql_acl.h"                       // SUPER_ACL
 #include "sp_head.h"
 #include "sp_cache.h"
-#include "lock.h"                               // lock_routine_name
+#include "lock.h"                               // lock_object_name
 
 #include <my_user.h>
 
@@ -440,7 +440,7 @@ static TABLE *open_proc_table_for_update(THD *thd)
 {
   TABLE_LIST table_list;
   TABLE *table;
-  MDL_ticket *mdl_savepoint= thd->mdl_context.mdl_savepoint();
+  MDL_savepoint mdl_savepoint= thd->mdl_context.mdl_savepoint();
   DBUG_ENTER("open_proc_table_for_update");
 
   table_list.init_one_table("mysql", 5, "proc", 4, "proc", TL_WRITE);
@@ -925,6 +925,8 @@ sp_create_routine(THD *thd, int type, sp_head *sp)
   TABLE *table;
   char definer[USER_HOST_BUFF_SIZE];
   ulong saved_mode= thd->variables.sql_mode;
+  MDL_key::enum_mdl_namespace mdl_type= type == TYPE_ENUM_FUNCTION ?
+                                        MDL_key::FUNCTION : MDL_key::PROCEDURE;
 
   CHARSET_INFO *db_cs= get_default_db_collation(thd, sp->m_db.str);
 
@@ -944,8 +946,7 @@ sp_create_routine(THD *thd, int type, sp_head *sp)
               type == TYPE_ENUM_FUNCTION);
 
   /* Grab an exclusive MDL lock. */
-  if (lock_routine_name(thd, type == TYPE_ENUM_FUNCTION,
-                        sp->m_db.str, sp->m_name.str))
+  if (lock_object_name(thd, mdl_type, sp->m_db.str, sp->m_name.str))
     DBUG_RETURN(SP_OPEN_TABLE_FAILED);
 
   /* Reset sql_mode during data dictionary operations. */
@@ -1193,6 +1194,8 @@ sp_drop_routine(THD *thd, int type, sp_name *name)
   TABLE *table;
   int ret;
   bool save_binlog_row_based;
+  MDL_key::enum_mdl_namespace mdl_type= type == TYPE_ENUM_FUNCTION ?
+                                        MDL_key::FUNCTION : MDL_key::PROCEDURE;
   DBUG_ENTER("sp_drop_routine");
   DBUG_PRINT("enter", ("type: %d  name: %.*s",
 		       type, (int) name->m_name.length, name->m_name.str));
@@ -1201,8 +1204,7 @@ sp_drop_routine(THD *thd, int type, sp_name *name)
               type == TYPE_ENUM_FUNCTION);
 
   /* Grab an exclusive MDL lock. */
-  if (lock_routine_name(thd, type == TYPE_ENUM_FUNCTION,
-                        name->m_db.str, name->m_name.str))
+  if (lock_object_name(thd, mdl_type, name->m_db.str, name->m_name.str))
     DBUG_RETURN(SP_DELETE_ROW_FAILED);
 
   if (!(table= open_proc_table_for_update(thd)))
@@ -1273,6 +1275,8 @@ sp_update_routine(THD *thd, int type, sp_name *name, st_sp_chistics *chistics)
   TABLE *table;
   int ret;
   bool save_binlog_row_based;
+  MDL_key::enum_mdl_namespace mdl_type= type == TYPE_ENUM_FUNCTION ?
+                                        MDL_key::FUNCTION : MDL_key::PROCEDURE;
   DBUG_ENTER("sp_update_routine");
   DBUG_PRINT("enter", ("type: %d  name: %.*s",
 		       type, (int) name->m_name.length, name->m_name.str));
@@ -1281,8 +1285,7 @@ sp_update_routine(THD *thd, int type, sp_name *name, st_sp_chistics *chistics)
               type == TYPE_ENUM_FUNCTION);
 
   /* Grab an exclusive MDL lock. */
-  if (lock_routine_name(thd, type == TYPE_ENUM_FUNCTION,
-                        name->m_db.str, name->m_name.str))
+  if (lock_object_name(thd, mdl_type, name->m_db.str, name->m_name.str))
     DBUG_RETURN(SP_OPEN_TABLE_FAILED);
 
   if (!(table= open_proc_table_for_update(thd)))
@@ -1358,6 +1361,106 @@ err:
 
 
 /**
+  This internal handler is used to trap errors from opening mysql.proc.
+*/
+
+class Lock_db_routines_error_handler : public Internal_error_handler
+{
+public:
+  bool handle_condition(THD *thd,
+                        uint sql_errno,
+                        const char* sqlstate,
+                        MYSQL_ERROR::enum_warning_level level,
+                        const char* msg,
+                        MYSQL_ERROR ** cond_hdl)
+  {
+    if (sql_errno == ER_NO_SUCH_TABLE ||
+        sql_errno == ER_COL_COUNT_DOESNT_MATCH_CORRUPTED)
+      return true;
+    return false;
+  }
+};
+
+
+/**
+   Acquires exclusive metadata lock on all stored routines in the
+   given database.
+
+   @note Will also return false (=success) if mysql.proc can't be opened
+         or is outdated. This allows DROP DATABASE to continue in these
+         cases.
+ */
+
+bool lock_db_routines(THD *thd, char *db)
+{
+  TABLE *table;
+  uint key_len;
+  int nxtres= 0;
+  Open_tables_backup open_tables_state_backup;
+  MDL_request_list mdl_requests;
+  Lock_db_routines_error_handler err_handler;
+  DBUG_ENTER("lock_db_routines");
+
+  /*
+    mysql.proc will be re-opened during deletion, so we can ignore
+    errors when opening the table here. The error handler is
+    used to avoid getting the same warning twice.
+  */
+  thd->push_internal_handler(&err_handler);
+  table= open_proc_table_for_read(thd, &open_tables_state_backup);
+  thd->pop_internal_handler();
+  if (!table)
+  {
+    /*
+      DROP DATABASE should not fail even if mysql.proc does not exist
+      or is outdated. We therefore only abort mysql_rm_db() if we
+      have errors not handled by the error handler.
+    */
+    DBUG_RETURN(thd->is_error() || thd->killed);
+  }
+
+  table->field[MYSQL_PROC_FIELD_DB]->store(db, strlen(db), system_charset_info);
+  key_len= table->key_info->key_part[0].store_length;
+  table->file->ha_index_init(0, 1);
+
+  if (! table->file->index_read_map(table->record[0],
+                                    table->field[MYSQL_PROC_FIELD_DB]->ptr,
+                                    (key_part_map)1, HA_READ_KEY_EXACT))
+  {
+    do
+    {
+      char *sp_name= get_field(thd->mem_root,
+                               table->field[MYSQL_PROC_FIELD_NAME]);
+      longlong sp_type= table->field[MYSQL_PROC_MYSQL_TYPE]->val_int();
+      MDL_request *mdl_request= new (thd->mem_root) MDL_request;
+      mdl_request->init(sp_type == TYPE_ENUM_FUNCTION ?
+                        MDL_key::FUNCTION : MDL_key::PROCEDURE,
+                        db, sp_name, MDL_EXCLUSIVE, MDL_TRANSACTION);
+      mdl_requests.push_front(mdl_request);
+    } while (! (nxtres= table->file->index_next_same(table->record[0],
+                                         table->field[MYSQL_PROC_FIELD_DB]->ptr,
+						     key_len)));
+  }
+  table->file->ha_index_end();
+  if (nxtres != 0 && nxtres != HA_ERR_END_OF_FILE)
+  {
+    table->file->print_error(nxtres, MYF(0));
+    close_system_tables(thd, &open_tables_state_backup);
+    DBUG_RETURN(true);
+  }
+  close_system_tables(thd, &open_tables_state_backup);
+
+  /* We should already hold a global IX lock and a schema X lock. */
+  DBUG_ASSERT(thd->mdl_context.is_lock_owner(MDL_key::GLOBAL, "", "",
+                                             MDL_INTENTION_EXCLUSIVE) &&
+              thd->mdl_context.is_lock_owner(MDL_key::SCHEMA, db, "",
+                                             MDL_EXCLUSIVE));
+  DBUG_RETURN(thd->mdl_context.acquire_locks(&mdl_requests,
+                                             thd->variables.lock_wait_timeout));
+}
+
+
+/**
   Drop all routines in database 'db'
 
   @note Close the thread tables, the calling code might want to
@@ -1370,7 +1473,7 @@ sp_drop_db_routines(THD *thd, char *db)
   TABLE *table;
   int ret;
   uint key_len;
-  MDL_ticket *mdl_savepoint= thd->mdl_context.mdl_savepoint();
+  MDL_savepoint mdl_savepoint= thd->mdl_context.mdl_savepoint();
   DBUG_ENTER("sp_drop_db_routines");
   DBUG_PRINT("enter", ("db: %s", db));
 
@@ -1697,7 +1800,7 @@ bool sp_add_used_routine(Query_tables_list *prelocking_ctx, Query_arena *arena,
       (Sroutine_hash_entry *)arena->alloc(sizeof(Sroutine_hash_entry));
     if (!rn)              // OOM. Error will be reported using fatal_error().
       return FALSE;
-    rn->mdl_request.init(key, MDL_SHARED);
+    rn->mdl_request.init(key, MDL_SHARED, MDL_TRANSACTION);
     if (my_hash_insert(&prelocking_ctx->sroutines, (uchar *)rn))
       return FALSE;
     prelocking_ctx->sroutines_list.link_in_list(rn, &rn->next);
diff --git a/sql/sp.h b/sql/sp.h
index 5d0490586a1..68c45768857 100644
--- a/sql/sp.h
+++ b/sql/sp.h
@@ -84,6 +84,18 @@ enum
 int
 sp_drop_db_routines(THD *thd, char *db);
 
+/**
+   Acquires exclusive metadata lock on all stored routines in the
+   given database.
+
+   @param  thd  Thread handler
+   @param  db   Database name
+
+   @retval  false  Success
+   @retval  true   Failure
+ */
+bool lock_db_routines(THD *thd, char *db);
+
 sp_head *
 sp_find_routine(THD *thd, int type, sp_name *name,
                 sp_cache **cp, bool cache_only);
diff --git a/sql/sp_head.cc b/sql/sp_head.cc
index de20b9547c5..2f165310c28 100644
--- a/sql/sp_head.cc
+++ b/sql/sp_head.cc
@@ -2140,6 +2140,8 @@ sp_head::execute_procedure(THD *thd, List<Item> *args)
 
     if (! thd->in_sub_stmt && ! thd->in_multi_stmt_transaction_mode())
       thd->mdl_context.release_transactional_locks();
+    else if (! thd->in_sub_stmt)
+      thd->mdl_context.release_statement_locks();
 
     thd->rollback_item_tree_changes();
 
@@ -2978,6 +2980,8 @@ sp_lex_keeper::reset_lex_and_exec_core(THD *thd, uint *nextp,
 
     if (! thd->in_sub_stmt && ! thd->in_multi_stmt_transaction_mode())
       thd->mdl_context.release_transactional_locks();
+    else if (! thd->in_sub_stmt)
+      thd->mdl_context.release_statement_locks();
   }
 
   if (m_lex->query_tables_own_last)
@@ -3093,7 +3097,12 @@ sp_instr_stmt::execute(THD *thd, uint *nextp)
       res= m_lex_keeper.reset_lex_and_exec_core(thd, nextp, FALSE, this);
 
       if (thd->stmt_da->is_eof())
+      {
+        /* Finalize server status flags after executing a statement. */
+        thd->update_server_status();
+
         thd->protocol->end_statement();
+      }
 
       query_cache_end_of_result(thd);
 
@@ -4173,7 +4182,8 @@ sp_head::add_used_tables_to_table_list(THD *thd,
       */
       table->mdl_request.init(MDL_key::TABLE, table->db, table->table_name,
                               table->lock_type >= TL_WRITE_ALLOW_WRITE ?
-                              MDL_SHARED_WRITE : MDL_SHARED_READ);
+                              MDL_SHARED_WRITE : MDL_SHARED_READ,
+                              MDL_TRANSACTION);
 
       /* Everyting else should be zeroed */
 
@@ -4217,7 +4227,7 @@ sp_add_to_query_tables(THD *thd, LEX *lex,
   table->select_lex= lex->current_select;
   table->cacheable_table= 1;
   table->mdl_request.init(MDL_key::TABLE, table->db, table->table_name,
-                          mdl_type);
+                          mdl_type, MDL_TRANSACTION);
 
   lex->add_to_query_tables(table);
   return table;
diff --git a/sql/sql_admin.cc b/sql/sql_admin.cc
index 21a05a5baca..f648d219fac 100644
--- a/sql/sql_admin.cc
+++ b/sql/sql_admin.cc
@@ -85,7 +85,7 @@ static int prepare_for_repair(THD *thd, TABLE_LIST *table_list,
     key_length= create_table_def_key(thd, key, table_list, 0);
     table_list->mdl_request.init(MDL_key::TABLE,
                                  table_list->db, table_list->table_name,
-                                 MDL_EXCLUSIVE);
+                                 MDL_EXCLUSIVE, MDL_TRANSACTION);
 
     if (lock_table_names(thd, table_list, table_list->next_global,
                          thd->variables.lock_wait_timeout,
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index 8caae3ee406..60c32a1a376 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -21,7 +21,7 @@
 #include "sql_priv.h"
 #include "unireg.h"
 #include "debug_sync.h"
-#include "lock.h"        // broadcast_refresh, mysql_lock_remove,
+#include "lock.h"        // mysql_lock_remove,
                          // mysql_unlock_tables,
                          // mysql_lock_have_duplicate
 #include "sql_show.h"    // append_identifier
@@ -1285,20 +1285,12 @@ static void mark_used_tables_as_free_for_reuse(THD *thd, TABLE *table)
 
 static void close_open_tables(THD *thd)
 {
-  bool found_old_table= 0;
-
   mysql_mutex_assert_not_owner(&LOCK_open);
 
   DBUG_PRINT("info", ("thd->open_tables: 0x%lx", (long) thd->open_tables));
 
   while (thd->open_tables)
-    found_old_table|= close_thread_table(thd, &thd->open_tables);
-
-  if (found_old_table)
-  {
-    /* Tell threads waiting for refresh that something has happened */
-    broadcast_refresh();
-  }
+    (void) close_thread_table(thd, &thd->open_tables);
 }
 
 
@@ -1364,11 +1356,6 @@ close_all_tables_for_name(THD *thd, TABLE_SHARE *share,
   /* Remove the table share from the cache. */
   tdc_remove_table(thd, TDC_RT_REMOVE_ALL, db, table_name,
                    FALSE);
-  /*
-    There could be a FLUSH thread waiting
-    on the table to go away. Wake it up.
-  */
-  broadcast_refresh();
 }
 
 
@@ -2463,7 +2450,8 @@ open_table_get_mdl_lock(THD *thd, Open_table_context *ot_ctx,
 
     mdl_request_shared.init(&mdl_request->key,
                             (flags & MYSQL_OPEN_FORCE_SHARED_MDL) ?
-                            MDL_SHARED : MDL_SHARED_HIGH_PRIO);
+                            MDL_SHARED : MDL_SHARED_HIGH_PRIO,
+                            MDL_TRANSACTION);
     mdl_request= &mdl_request_shared;
   }
 
@@ -2628,32 +2616,6 @@ bool open_table(THD *thd, TABLE_LIST *table_list, MEM_ROOT *mem_root,
                TMP_TABLE_KEY_EXTRA);
 
   /*
-    We need this to work for all tables, including temporary
-    tables, for backwards compatibility. But not under LOCK
-    TABLES, since under LOCK TABLES one can't use a non-prelocked
-    table.  This code only works for updates done inside DO/SELECT
-    f1() statements, normal DML is handled by means of
-    sql_command_flags.
-  */
-  if (global_read_lock && table_list->lock_type >= TL_WRITE_ALLOW_WRITE &&
-      ! (flags & MYSQL_OPEN_IGNORE_GLOBAL_READ_LOCK) &&
-      ! thd->locked_tables_mode)
-  {
-    /*
-      Someone has issued FLUSH TABLES WITH READ LOCK and we want
-      a write lock. Wait until the lock is gone.
-    */
-    if (thd->global_read_lock.wait_if_global_read_lock(thd, 1, 1))
-      DBUG_RETURN(TRUE);
-
-    if (thd->open_tables && thd->open_tables->s->version != refresh_version)
-    {
-      (void)ot_ctx->request_backoff_action(Open_table_context::OT_REOPEN_TABLES,
-                                           NULL);
-      DBUG_RETURN(TRUE);
-    }
-  }
-  /*
     Unless requested otherwise, try to resolve this table in the list
     of temporary tables of this thread. In MySQL temporary tables
     are always thread-local and "shadow" possible base tables with the
@@ -2824,6 +2786,59 @@ bool open_table(THD *thd, TABLE_LIST *table_list, MEM_ROOT *mem_root,
 
   if (! (flags & MYSQL_OPEN_HAS_MDL_LOCK))
   {
+    /*
+      We are not under LOCK TABLES and going to acquire write-lock/
+      modify the base table. We need to acquire protection against
+      global read lock until end of this statement in order to have
+      this statement blocked by active FLUSH TABLES WITH READ LOCK.
+
+      We don't block acquire this protection under LOCK TABLES as
+      such protection already acquired at LOCK TABLES time and
+      not released until UNLOCK TABLES.
+
+      We don't block statements which modify only temporary tables
+      as these tables are not preserved by backup by any form of
+      backup which uses FLUSH TABLES WITH READ LOCK.
+
+      TODO: The fact that we sometimes acquire protection against
+            GRL only when we encounter table to be write-locked
+            slightly increases probability of deadlock.
+            This problem will be solved once Alik pushes his
+            temporary table refactoring patch and we can start
+            pre-acquiring metadata locks at the beggining of
+            open_tables() call.
+    */
+    if (table_list->mdl_request.type >= MDL_SHARED_WRITE &&
+        ! (flags & (MYSQL_OPEN_IGNORE_GLOBAL_READ_LOCK |
+                    MYSQL_OPEN_FORCE_SHARED_MDL |
+                    MYSQL_OPEN_FORCE_SHARED_HIGH_PRIO_MDL |
+                    MYSQL_OPEN_SKIP_SCOPED_MDL_LOCK)) &&
+        ! ot_ctx->has_protection_against_grl())
+    {
+      MDL_request protection_request;
+      MDL_deadlock_handler mdl_deadlock_handler(ot_ctx);
+
+      if (thd->global_read_lock.can_acquire_protection())
+        DBUG_RETURN(TRUE);
+
+      protection_request.init(MDL_key::GLOBAL, "", "", MDL_INTENTION_EXCLUSIVE,
+                              MDL_STATEMENT);
+
+      /*
+        Install error handler which if possible will convert deadlock error
+        into request to back-off and restart process of opening tables.
+      */
+      thd->push_internal_handler(&mdl_deadlock_handler);
+      bool result= thd->mdl_context.acquire_lock(&protection_request,
+                                                 ot_ctx->get_timeout());
+      thd->pop_internal_handler();
+
+      if (result)
+        DBUG_RETURN(TRUE);
+
+      ot_ctx->set_has_protection_against_grl();
+    }
+
     if (open_table_get_mdl_lock(thd, ot_ctx, &table_list->mdl_request,
                                 flags, &mdl_ticket) ||
         mdl_ticket == NULL)
@@ -3379,7 +3394,6 @@ unlink_all_closed_tables(THD *thd, MYSQL_LOCK *lock, size_t reopen_count)
 
       close_thread_table(thd, &thd->open_tables);
     }
-    broadcast_refresh();
   }
   /* Exclude all closed tables from the LOCK TABLES list. */
   for (TABLE_LIST *table_list= m_locked_tables; table_list; table_list=
@@ -3856,7 +3870,8 @@ Open_table_context::Open_table_context(THD *thd, uint flags)
              LONG_TIMEOUT : thd->variables.lock_wait_timeout),
    m_flags(flags),
    m_action(OT_NO_ACTION),
-   m_has_locks(thd->mdl_context.has_locks())
+   m_has_locks(thd->mdl_context.has_locks()),
+   m_has_protection_against_grl(FALSE)
 {}
 
 
@@ -4013,6 +4028,12 @@ recover_from_failed_open(THD *thd)
     for safety.
   */
   m_failed_table= NULL;
+  /*
+    Reset flag indicating that we have already acquired protection
+    against GRL. It is no longer valid as the corresponding lock was
+    released by close_tables_for_reopen().
+  */
+  m_has_protection_against_grl= FALSE;
   /* Prepare for possible another back-off. */
   m_action= OT_NO_ACTION;
   return result;
@@ -4551,11 +4572,20 @@ lock_table_names(THD *thd,
       if (schema_request == NULL)
         return TRUE;
       schema_request->init(MDL_key::SCHEMA, table->db, "",
-                           MDL_INTENTION_EXCLUSIVE);
+                           MDL_INTENTION_EXCLUSIVE,
+                           MDL_TRANSACTION);
       mdl_requests.push_front(schema_request);
     }
-    /* Take the global intention exclusive lock. */
-    global_request.init(MDL_key::GLOBAL, "", "", MDL_INTENTION_EXCLUSIVE);
+
+    /*
+      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);
   }
 
@@ -5362,7 +5392,7 @@ bool open_and_lock_tables(THD *thd, TABLE_LIST *tables,
                           Prelocking_strategy *prelocking_strategy)
 {
   uint counter;
-  MDL_ticket *mdl_savepoint= thd->mdl_context.mdl_savepoint();
+  MDL_savepoint mdl_savepoint= thd->mdl_context.mdl_savepoint();
   DBUG_ENTER("open_and_lock_tables");
   DBUG_PRINT("enter", ("derived handling: %d", derived));
 
@@ -5419,7 +5449,7 @@ bool open_normal_and_derived_tables(THD *thd, TABLE_LIST *tables, uint flags)
 {
   DML_prelocking_strategy prelocking_strategy;
   uint counter;
-  MDL_ticket *mdl_savepoint= thd->mdl_context.mdl_savepoint();
+  MDL_savepoint mdl_savepoint= thd->mdl_context.mdl_savepoint();
   DBUG_ENTER("open_normal_and_derived_tables");
   DBUG_ASSERT(!thd->fill_derived_tables());
   if (open_tables(thd, &tables, &counter, flags, &prelocking_strategy) ||
@@ -5676,7 +5706,7 @@ bool lock_tables(THD *thd, TABLE_LIST *tables, uint count,
 */
 
 void close_tables_for_reopen(THD *thd, TABLE_LIST **tables,
-                             MDL_ticket *start_of_statement_svp)
+                             const MDL_savepoint &start_of_statement_svp)
 {
   TABLE_LIST *first_not_own_table= thd->lex->first_not_own_table();
   TABLE_LIST *tmp;
diff --git a/sql/sql_base.h b/sql/sql_base.h
index 7ae3971942b..35fa04b3674 100644
--- a/sql/sql_base.h
+++ b/sql/sql_base.h
@@ -159,7 +159,7 @@ thr_lock_type read_lock_type_for_table(THD *thd,
 my_bool mysql_rm_tmp_tables(void);
 bool rm_temporary_table(handlerton *base, char *path);
 void close_tables_for_reopen(THD *thd, TABLE_LIST **tables,
-                             MDL_ticket *start_of_statement_svp);
+                             const MDL_savepoint &start_of_statement_svp);
 TABLE_LIST *find_table_in_list(TABLE_LIST *table,
                                TABLE_LIST *TABLE_LIST::*link,
                                const char *db_name,
@@ -243,7 +243,6 @@ bool open_tables(THD *thd, TABLE_LIST **tables, uint *counter, uint flags,
 bool open_and_lock_tables(THD *thd, TABLE_LIST *tables,
                           bool derived, uint flags,
                           Prelocking_strategy *prelocking_strategy);
-int open_and_lock_tables_derived(THD *thd, TABLE_LIST *tables, bool derived);
 /* simple open_and_lock_tables without derived handling for single table */
 TABLE *open_n_lock_single_table(THD *thd, TABLE_LIST *table_l,
                                 thr_lock_type lock_type, uint flags,
@@ -508,7 +507,7 @@ public:
     the statement, so that we can rollback to it before waiting on
     locks.
   */
-  MDL_ticket *start_of_statement_svp() const
+  const MDL_savepoint &start_of_statement_svp() const
   {
     return m_start_of_statement_svp;
   }
@@ -519,6 +518,21 @@ public:
   }
 
   uint get_flags() const { return m_flags; }
+
+  /**
+    Set flag indicating that we have already acquired metadata lock
+    protecting this statement against GRL while opening tables.
+  */
+  void set_has_protection_against_grl()
+  {
+    m_has_protection_against_grl= TRUE;
+  }
+
+  bool has_protection_against_grl() const
+  {
+    return m_has_protection_against_grl;
+  }
+
 private:
   /**
     For OT_DISCOVER and OT_REPAIR actions, the table list element for
@@ -526,7 +540,7 @@ private:
     should be repaired.
   */
   TABLE_LIST *m_failed_table;
-  MDL_ticket *m_start_of_statement_svp;
+  MDL_savepoint m_start_of_statement_svp;
   /**
     Lock timeout in seconds. Initialized to LONG_TIMEOUT when opening system
     tables or to the "lock_wait_timeout" system variable for regular tables.
@@ -542,6 +556,11 @@ private:
     and we can't safely do back-off (and release them).
   */
   bool m_has_locks;
+  /**
+    Indicates that in the process of opening tables we have acquired
+    protection against global read lock.
+  */
+  bool m_has_protection_against_grl;
 };
 
 
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index bace0295a04..2df7a2c8572 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -1179,36 +1179,70 @@ void add_diff_to_status(STATUS_VAR *to_var, STATUS_VAR *from_var,
 }
 
 
+/**
+  Awake a thread.
+
+  @param[in]  state_to_set    value for THD::killed
+
+  This is normally called from another thread's THD object.
+
+  @note Do always call this while holding LOCK_thd_data.
+*/
+
 void THD::awake(THD::killed_state state_to_set)
 {
   DBUG_ENTER("THD::awake");
-  DBUG_PRINT("enter", ("this: 0x%lx", (long) this));
+  DBUG_PRINT("enter", ("this: %p current_thd: %p", this, current_thd));
   THD_CHECK_SENTRY(this);
   mysql_mutex_assert_owner(&LOCK_thd_data);
 
+  /* Set the 'killed' flag of 'this', which is the target THD object. */
   killed= state_to_set;
+
   if (state_to_set != THD::KILL_QUERY)
   {
-    thr_alarm_kill(thread_id);
-    if (!slave_thread)
-      MYSQL_CALLBACK(thread_scheduler, post_kill_notification, (this));
 #ifdef SIGNAL_WITH_VIO_CLOSE
     if (this != current_thd)
     {
       /*
-        In addition to a signal, let's close the socket of the thread that
-        is being killed. This is to make sure it does not block if the
-        signal is lost. This needs to be done only on platforms where
-        signals are not a reliable interruption mechanism.
-
-        If we're killing ourselves, we know that we're not blocked, so this
-        hack is not used.
+        Before sending a signal, let's close the socket of the thread
+        that is being killed ("this", which is not the current thread).
+        This is to make sure it does not block if the signal is lost.
+        This needs to be done only on platforms where signals are not
+        a reliable interruption mechanism.
+
+        Note that the downside of this mechanism is that we could close
+        the connection while "this" target thread is in the middle of
+        sending a result to the application, thus violating the client-
+        server protocol.
+
+        On the other hand, without closing the socket we have a race
+        condition. If "this" target thread passes the check of
+        thd->killed, and then the current thread runs through
+        THD::awake(), sets the 'killed' flag and completes the
+        signaling, and then the target thread runs into read(), it will
+        block on the socket. As a result of the discussions around
+        Bug#37780, it has been decided that we accept the race
+        condition. A second KILL awakes the target from read().
+
+        If we are killing ourselves, we know that we are not blocked.
+        We also know that we will check thd->killed before we go for
+        reading the next statement.
       */
 
       close_active_vio();
     }
-#endif    
+#endif
+
+    /* Mark the target thread's alarm request expired, and signal alarm. */
+    thr_alarm_kill(thread_id);
+
+    /* Send an event to the scheduler that a thread should be killed. */
+    if (!slave_thread)
+      MYSQL_CALLBACK(thread_scheduler, post_kill_notification, (this));
   }
+
+  /* Broadcast a condition to kick the target if it is waiting on it. */
   if (mysys_var)
   {
     mysql_mutex_lock(&mysys_var->mutex);
@@ -1232,6 +1266,11 @@ void THD::awake(THD::killed_state state_to_set)
       we issue a second KILL or the status it's waiting for happens).
       It's true that we have set its thd->killed but it may not
       see it immediately and so may have time to reach the cond_wait().
+
+      However, where possible, we test for killed once again after
+      enter_cond(). This should make the signaling as safe as possible.
+      However, there is still a small chance of failure on platforms with
+      instruction or memory write reordering.
     */
     if (mysys_var->current_cond && mysys_var->current_mutex)
     {
@@ -3456,11 +3495,15 @@ void THD::set_mysys_var(struct st_my_thread_var *new_mysys_var)
 void THD::leave_locked_tables_mode()
 {
   locked_tables_mode= LTM_NONE;
-  /* Make sure we don't release the global read lock when leaving LTM. */
-  mdl_context.reset_trans_sentinel(global_read_lock.global_shared_lock());
+  mdl_context.set_transaction_duration_for_all_locks();
+  /*
+    Make sure we don't release the global read lock and commit blocker
+    when leaving LTM.
+  */
+  global_read_lock.set_explicit_lock_duration(this);
   /* Also ensure that we don't release metadata locks for open HANDLERs. */
   if (handler_tables_hash.records)
-    mysql_ha_move_tickets_after_trans_sentinel(this);
+    mysql_ha_set_explicit_lock_duration(this);
 }
 
 void THD::get_definer(LEX_USER *definer)
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 0399ddbca57..02f28b54e10 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -869,8 +869,8 @@ struct st_savepoint {
   char                *name;
   uint                 length;
   Ha_trx_info         *ha_list;
-  /** Last acquired lock before this savepoint was set. */
-  MDL_ticket     *mdl_savepoint;
+  /** State of metadata locks before this savepoint was set. */
+  MDL_savepoint        mdl_savepoint;
 };
 
 enum xa_states {XA_NOTR=0, XA_ACTIVE, XA_IDLE, XA_PREPARED, XA_ROLLBACK_ONLY};
@@ -1105,12 +1105,12 @@ 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
+    table or tables, we want to save state of metadata
+    locks which were acquired before the backup. It is used
+    to release metadata locks on system tables after they are
     no longer used.
   */
-  MDL_ticket *mdl_system_tables_svp;
+  MDL_savepoint mdl_system_tables_svp;
 };
 
 /**
@@ -1383,26 +1383,43 @@ public:
   };
 
   Global_read_lock()
-    :m_protection_count(0), m_state(GRL_NONE), m_mdl_global_shared_lock(NULL)
+    : m_state(GRL_NONE),
+      m_mdl_global_shared_lock(NULL),
+      m_mdl_blocks_commits_lock(NULL)
   {}
 
   bool lock_global_read_lock(THD *thd);
   void unlock_global_read_lock(THD *thd);
-  bool wait_if_global_read_lock(THD *thd, bool abort_on_refresh,
-                                bool is_not_commit);
-  void start_waiting_global_read_lock(THD *thd);
+  /**
+    Check if this connection can acquire protection against GRL and
+    emit error if otherwise.
+  */
+  bool can_acquire_protection() const
+  {
+    if (m_state)
+    {
+      my_error(ER_CANT_UPDATE_WITH_READLOCK, MYF(0));
+      return TRUE;
+    }
+    return FALSE;
+  }
   bool make_global_read_lock_block_commit(THD *thd);
   bool is_acquired() const { return m_state != GRL_NONE; }
-  bool has_protection() const { return m_protection_count > 0; }
-  MDL_ticket *global_shared_lock() const { return m_mdl_global_shared_lock; }
+  void set_explicit_lock_duration(THD *thd);
 private:
-  uint           m_protection_count;            // GRL protection count
+  enum_grl_state m_state;
   /**
     In order to acquire the global read lock, the connection must
-    acquire a global shared metadata lock, to prohibit all DDL.
+    acquire shared metadata lock in GLOBAL namespace, to prohibit
+    all DDL.
   */
-  enum_grl_state m_state;
   MDL_ticket *m_mdl_global_shared_lock;
+  /**
+    Also in order to acquire the global read lock, the connection
+    must acquire a shared metadata lock in COMMIT namespace, to
+    prohibit commits.
+  */
+  MDL_ticket *m_mdl_blocks_commits_lock;
 };
 
 
@@ -2000,6 +2017,12 @@ public:
   DYNAMIC_ARRAY user_var_events;        /* For user variables replication */
   MEM_ROOT      *user_var_events_alloc; /* Allocate above array elements here */
 
+  /*
+    If checking this in conjunction with a wait condition, please
+    include a check after enter_cond() if you want to avoid a race
+    condition. For details see the implementation of awake(),
+    especially the "broadcast" part.
+  */
   enum killed_state
   {
     NOT_KILLED=0,
@@ -2258,6 +2281,20 @@ public:
   }
   void set_time_after_lock()  { utime_after_lock= my_micro_time(); }
   ulonglong current_utime()  { return my_micro_time(); }
+  /**
+   Update server status after execution of a top level statement.
+
+   Currently only checks if a query was slow, and assigns
+   the status accordingly.
+   Evaluate the current time, and if it exceeds the long-query-time
+   setting, mark the query as slow.
+  */
+  void update_server_status()
+  {
+    ulonglong end_utime_of_query= current_utime();
+    if (end_utime_of_query > utime_after_lock + variables.long_query_time)
+      server_status|= SERVER_QUERY_WAS_SLOW;
+  }
   inline ulonglong found_rows(void)
   {
     return limit_found_rows;
@@ -2749,7 +2786,7 @@ public:
   {
     DBUG_ASSERT(locked_tables_mode == LTM_NONE);
 
-    mdl_context.set_trans_sentinel();
+    mdl_context.set_explicit_duration_for_all_locks();
     locked_tables_mode= mode_arg;
   }
   void leave_locked_tables_mode();
@@ -3555,20 +3592,10 @@ public:
 #define CF_DIAGNOSTIC_STMT        (1U << 8)
 
 /**
-  SQL statements that must be protected against impending global read lock
-  to prevent deadlock. This deadlock could otherwise happen if the statement
-  starts waiting for the GRL to go away inside mysql_lock_tables while at the
-  same time having "old" opened tables. The thread holding the GRL can be
-  waiting for these "old" opened tables to be closed, causing a deadlock
-  (FLUSH TABLES WITH READ LOCK).
- */
-#define CF_PROTECT_AGAINST_GRL  (1U << 10)
-
-/**
   Identifies statements that may generate row events
   and that may end up in the binary log.
 */
-#define CF_CAN_GENERATE_ROW_EVENTS (1U << 11)
+#define CF_CAN_GENERATE_ROW_EVENTS (1U << 9)
 
 /* Bits in server_command_flags */
 
diff --git a/sql/sql_cursor.cc b/sql/sql_cursor.cc
index 7a9834b4cde..acc591f1ea2 100644
--- a/sql/sql_cursor.cc
+++ b/sql/sql_cursor.cc
@@ -277,7 +277,6 @@ int Materialized_cursor::open(JOIN *join __attribute__((unused)))
     rc= result->send_result_set_metadata(item_list, Protocol::SEND_NUM_ROWS);
     thd->server_status|= SERVER_STATUS_CURSOR_EXISTS;
     result->send_eof();
-    thd->server_status&= ~SERVER_STATUS_CURSOR_EXISTS;
   }
   return rc;
 }
@@ -318,12 +317,10 @@ void Materialized_cursor::fetch(ulong num_rows)
   case 0:
     thd->server_status|= SERVER_STATUS_CURSOR_EXISTS;
     result->send_eof();
-    thd->server_status&= ~SERVER_STATUS_CURSOR_EXISTS;
     break;
   case HA_ERR_END_OF_FILE:
     thd->server_status|= SERVER_STATUS_LAST_ROW_SENT;
     result->send_eof();
-    thd->server_status&= ~SERVER_STATUS_LAST_ROW_SENT;
     close();
     break;
   default:
diff --git a/sql/sql_db.cc b/sql/sql_db.cc
index 517cb9139e9..292fbba2352 100644
--- a/sql/sql_db.cc
+++ b/sql/sql_db.cc
@@ -28,6 +28,8 @@
 #include "sql_acl.h"                     // SELECT_ACL, DB_ACLS,
                                          // acl_get, check_grant_db
 #include "log_event.h"                   // Query_log_event
+#include "sql_base.h"                    // lock_table_names, tdc_remove_table
+#include "sql_handler.h"                 // mysql_ha_rm_tables
 #include <mysys_err.h>
 #include "sp.h"
 #include "events.h"
@@ -44,10 +46,12 @@ const char *del_exts[]= {".frm", ".BAK", ".TMD",".opt", NullS};
 static TYPELIB deletable_extentions=
 {array_elements(del_exts)-1,"del_exts", del_exts, NULL};
 
-static long mysql_rm_known_files(THD *thd, MY_DIR *dirp,
-				 const char *db, const char *path, uint level, 
-                                 TABLE_LIST **dropped_tables);
-         
+static bool find_db_tables_and_rm_known_files(THD *thd, MY_DIR *dirp,
+                                              const char *db,
+                                              const char *path,
+                                              TABLE_LIST **tables,
+                                              bool *found_other_files);
+
 long mysql_rm_arc_files(THD *thd, MY_DIR *dirp, const char *org_path);
 static my_bool rm_dir_w_symlink(const char *org_path, my_bool send_error);
 static void mysql_change_db_impl(THD *thd,
@@ -736,36 +740,37 @@ exit:
 }
 
 
-/*
-  Drop all tables in a database and the database itself
-
-  SYNOPSIS
-    mysql_rm_db()
-    thd			Thread handle
-    db			Database name in the case given by user
-		        It's already validated and set to lower case
-                        (if needed) when we come here
-    if_exists		Don't give error if database doesn't exists
-    silent		Don't generate errors
-
-  RETURN
-    FALSE ok (Database dropped)
-    ERROR Error
+/**
+  Drop all tables, routines and events in a database and the database itself.
+
+  @param  thd        Thread handle
+  @param  db         Database name in the case given by user
+                     It's already validated and set to lower case
+                     (if needed) when we come here
+  @param  if_exists  Don't give error if database doesn't exists
+  @param  silent     Don't write the statement to the binary log and don't
+                     send ok packet to the client
+
+  @retval  false  OK (Database dropped)
+  @retval  true   Error
 */
 
 bool mysql_rm_db(THD *thd,char *db,bool if_exists, bool silent)
 {
-  long deleted=0;
-  int error= 0;
+  ulong deleted_tables= 0;
+  bool error= true;
   char	path[FN_REFLEN+16];
   MY_DIR *dirp;
   uint length;
-  TABLE_LIST* dropped_tables= 0;
+  bool found_other_files= false;
+  TABLE_LIST *tables= NULL;
+  TABLE_LIST *table;
+  Drop_table_error_handler err_handler;
   DBUG_ENTER("mysql_rm_db");
 
 
   if (lock_schema_name(thd, db))
-    DBUG_RETURN(TRUE);
+    DBUG_RETURN(true);
 
   length= build_table_filename(path, sizeof(path) - 1, db, "", "", 0);
   strmov(path+length, MY_DB_OPT_FILE);		// Append db option file name
@@ -777,20 +782,72 @@ bool mysql_rm_db(THD *thd,char *db,bool if_exists, bool silent)
   {
     if (!if_exists)
     {
-      error= -1;
       my_error(ER_DB_DROP_EXISTS, MYF(0), db);
-      goto exit;
+      DBUG_RETURN(true);
     }
     else
+    {
       push_warning_printf(thd, MYSQL_ERROR::WARN_LEVEL_NOTE,
 			  ER_DB_DROP_EXISTS, ER(ER_DB_DROP_EXISTS), db);
+      error= false;
+      goto update_binlog;
+    }
   }
-  else
+
+  thd->push_internal_handler(&err_handler);
+
+  if (find_db_tables_and_rm_known_files(thd, dirp, db, path, &tables,
+                                        &found_other_files))
   {
-    Drop_table_error_handler err_handler;
-    thd->push_internal_handler(&err_handler);
+    thd->pop_internal_handler();
+    goto exit;
+  }
 
-    error= -1;
+  /*
+    Disable drop of enabled log tables, must be done before name locking.
+    This check is only needed if we are dropping the "mysql" database.
+  */
+  if ((my_strcasecmp(system_charset_info, MYSQL_SCHEMA_NAME.str, db) == 0))
+  {
+    for (table= tables; table; table= table->next_local)
+    {
+      if (check_if_log_table(table->db_length, table->db,
+                             table->table_name_length, table->table_name, true))
+      {
+        my_error(ER_BAD_LOG_STATEMENT, MYF(0), "DROP");
+        thd->pop_internal_handler();
+        goto exit;
+      }
+    }
+  }
+
+  /* Lock all tables and stored routines about to be dropped. */
+  if (lock_table_names(thd, tables, NULL, thd->variables.lock_wait_timeout,
+                       MYSQL_OPEN_SKIP_TEMPORARY) ||
+      lock_db_routines(thd, db))
+  {
+    thd->pop_internal_handler();
+    goto exit;
+  }
+
+  /* mysql_ha_rm_tables() requires a non-null TABLE_LIST. */
+  if (tables)
+    mysql_ha_rm_tables(thd, tables);
+
+  for (table= tables; table; table= table->next_local)
+  {
+    tdc_remove_table(thd, TDC_RT_REMOVE_ALL, table->db, table->table_name,
+                     false);
+    deleted_tables++;
+  }
+
+  if (thd->killed ||
+      (tables && mysql_rm_table_no_locks(thd, tables, true, false, true, true)))
+  {
+    tables= NULL;
+  }
+  else
+  {
     /*
       We temporarily disable the binary log while dropping the objects
       in the database. Since the DROP DATABASE statement is always
@@ -808,23 +865,30 @@ bool mysql_rm_db(THD *thd,char *db,bool if_exists, bool silent)
       ha_drop_database(), since NDB otherwise detects the binary log
       as disabled and will not log the drop database statement on any
       other connected server.
-     */
-    if ((deleted= mysql_rm_known_files(thd, dirp, db, path, 0,
-                                       &dropped_tables)) >= 0)
-    {
-      ha_drop_database(path);
-      tmp_disable_binlog(thd);
-      query_cache_invalidate1(db);
-      (void) sp_drop_db_routines(thd, db); /* @todo Do not ignore errors */
+    */
+
+    ha_drop_database(path);
+    tmp_disable_binlog(thd);
+    query_cache_invalidate1(db);
+    (void) sp_drop_db_routines(thd, db); /* @todo Do not ignore errors */
 #ifdef HAVE_EVENT_SCHEDULER
-      Events::drop_schema_events(thd, db);
+    Events::drop_schema_events(thd, db);
 #endif
-      error = 0;
-      reenable_binlog(thd);
-    }
-    thd->pop_internal_handler();
+    reenable_binlog(thd);
+
+    /*
+      If the directory is a symbolic link, remove the link first, then
+      remove the directory the symbolic link pointed at
+    */
+    if (found_other_files)
+      my_error(ER_DB_DROP_RMDIR, MYF(0), path, EEXIST);
+    else
+      error= rm_dir_w_symlink(path, true);
   }
-  if (!silent && deleted>=0)
+  thd->pop_internal_handler();
+
+update_binlog:
+  if (!silent && !error)
   {
     const char *query;
     ulong query_length;
@@ -859,14 +923,13 @@ bool mysql_rm_db(THD *thd,char *db,bool if_exists, bool silent)
       */
       if (mysql_bin_log.write(&qinfo))
       {
-        error= -1;
+        error= true;
         goto exit;
       }
     }
     thd->clear_error();
     thd->server_status|= SERVER_STATUS_DB_DROPPED;
-    my_ok(thd, (ulong) deleted);
-    thd->server_status&= ~SERVER_STATUS_DB_DROPPED;
+    my_ok(thd, deleted_tables);
   }
   else if (mysql_bin_log.is_open())
   {
@@ -880,7 +943,7 @@ bool mysql_rm_db(THD *thd,char *db,bool if_exists, bool silent)
     query_end= query + MAX_DROP_TABLE_Q_LEN;
     db_len= strlen(db);
 
-    for (tbl= dropped_tables; tbl; tbl= tbl->next_local)
+    for (tbl= tables; tbl; tbl= tbl->next_local)
     {
       uint tbl_name_len;
 
@@ -894,7 +957,7 @@ bool mysql_rm_db(THD *thd,char *db,bool if_exists, bool silent)
         */
         if (write_to_binlog(thd, query, query_pos -1 - query, db, db_len))
         {
-          error= -1;
+          error= true;
           goto exit;
         }
         query_pos= query_data_start;
@@ -914,7 +977,7 @@ bool mysql_rm_db(THD *thd,char *db,bool if_exists, bool silent)
       */
       if (write_to_binlog(thd, query, query_pos -1 - query, db, db_len))
       {
-        error= -1;
+        error= true;
         goto exit;
       }
     }
@@ -927,28 +990,23 @@ exit:
     SELECT DATABASE() in the future). For this we free() thd->db and set
     it to 0.
   */
-  if (thd->db && !strcmp(thd->db, db) && error == 0)
+  if (thd->db && !strcmp(thd->db, db) && !error)
     mysql_change_db_impl(thd, NULL, 0, thd->variables.collation_server);
+  my_dirend(dirp);
   DBUG_RETURN(error);
 }
 
-/*
-  Removes files with known extensions plus all found subdirectories that
-  are 2 hex digits (raid directories).
-  thd MUST be set when calling this function!
-*/
 
-static long mysql_rm_known_files(THD *thd, MY_DIR *dirp, const char *db,
-				 const char *org_path, uint level,
-                                 TABLE_LIST **dropped_tables)
+static bool find_db_tables_and_rm_known_files(THD *thd, MY_DIR *dirp,
+                                              const char *db,
+                                              const char *path,
+                                              TABLE_LIST **tables,
+                                              bool *found_other_files)
 {
-  long deleted=0;
-  ulong found_other_files=0;
   char filePath[FN_REFLEN];
   TABLE_LIST *tot_list=0, **tot_list_next_local, **tot_list_next_global;
-  List<String> raid_dirs;
-  DBUG_ENTER("mysql_rm_known_files");
-  DBUG_PRINT("enter",("path: %s", org_path));
+  DBUG_ENTER("find_db_tables_and_rm_known_files");
+  DBUG_PRINT("enter",("path: %s", path));
 
   tot_list_next_local= tot_list_next_global= &tot_list;
 
@@ -965,36 +1023,7 @@ static long mysql_rm_known_files(THD *thd, MY_DIR *dirp, const char *db,
        (file->name[1] == '.' &&  !file->name[2])))
       continue;
 
-    /* Check if file is a raid directory */
-    if ((my_isdigit(system_charset_info, file->name[0]) ||
-	 (file->name[0] >= 'a' && file->name[0] <= 'f')) &&
-	(my_isdigit(system_charset_info, file->name[1]) ||
-	 (file->name[1] >= 'a' && file->name[1] <= 'f')) &&
-	!file->name[2] && !level)
-    {
-      char newpath[FN_REFLEN], *copy_of_path;
-      MY_DIR *new_dirp;
-      String *dir;
-      uint length;
-
-      strxmov(newpath,org_path,"/",file->name,NullS);
-      length= unpack_filename(newpath,newpath);
-      if ((new_dirp = my_dir(newpath,MYF(MY_DONT_SORT))))
-      {
-	DBUG_PRINT("my",("New subdir found: %s", newpath));
-	if ((mysql_rm_known_files(thd, new_dirp, NullS, newpath,1,0)) < 0)
-	  goto err;
-	if (!(copy_of_path= (char*) thd->memdup(newpath, length+1)) ||
-	    !(dir= new (thd->mem_root) String(copy_of_path, length,
-					       &my_charset_bin)) ||
-	    raid_dirs.push_back(dir))
-	  goto err;
-	continue;
-      }
-      found_other_files++;
-      continue;
-    }
-    else if (file->name[0] == 'a' && file->name[1] == 'r' &&
+    if (file->name[0] == 'a' && file->name[1] == 'r' &&
              file->name[2] == 'c' && file->name[3] == '\0')
     {
       /* .frm archive:
@@ -1003,16 +1032,16 @@ static long mysql_rm_known_files(THD *thd, MY_DIR *dirp, const char *db,
       */
       char newpath[FN_REFLEN];
       MY_DIR *new_dirp;
-      strxmov(newpath, org_path, "/", "arc", NullS);
+      strxmov(newpath, path, "/", "arc", NullS);
       (void) unpack_filename(newpath, newpath);
       if ((new_dirp = my_dir(newpath, MYF(MY_DONT_SORT))))
       {
 	DBUG_PRINT("my",("Archive subdir found: %s", newpath));
 	if ((mysql_rm_arc_files(thd, new_dirp, newpath)) < 0)
-	  goto err;
+	  DBUG_RETURN(true);
 	continue;
       }
-      found_other_files++;
+      *found_other_files= true;
       continue;
     }
     if (!(extension= strrchr(file->name, '.')))
@@ -1020,7 +1049,7 @@ static long mysql_rm_known_files(THD *thd, MY_DIR *dirp, const char *db,
     if (find_type(extension, &deletable_extentions,1+2) <= 0)
     {
       if (find_type(extension, ha_known_exts(),1+2) <= 0)
-	found_other_files++;
+	*found_other_files= true;
       continue;
     }
     /* just for safety we use files_charset_info */
@@ -1036,7 +1065,7 @@ static long mysql_rm_known_files(THD *thd, MY_DIR *dirp, const char *db,
                                           strlen(file->name) + 1);
 
       if (!table_list)
-        goto err;
+        DBUG_RETURN(true);
       table_list->db= (char*) (table_list+1);
       table_list->db_length= strmov(table_list->db, db) - table_list->db;
       table_list->table_name= table_list->db + table_list->db_length + 1;
@@ -1054,61 +1083,23 @@ static long mysql_rm_known_files(THD *thd, MY_DIR *dirp, const char *db,
       table_list->alias= table_list->table_name;	// If lower_case_table_names=2
       table_list->internal_tmp_table= is_prefix(file->name, tmp_file_prefix);
       table_list->mdl_request.init(MDL_key::TABLE, table_list->db,
-                                   table_list->table_name, MDL_EXCLUSIVE);
+                                   table_list->table_name, MDL_EXCLUSIVE,
+                                   MDL_TRANSACTION);
       /* Link into list */
       (*tot_list_next_local)= table_list;
       (*tot_list_next_global)= table_list;
       tot_list_next_local= &table_list->next_local;
       tot_list_next_global= &table_list->next_global;
-      deleted++;
     }
     else
     {
-      strxmov(filePath, org_path, "/", file->name, NullS);
+      strxmov(filePath, path, "/", file->name, NullS);
       if (mysql_file_delete_with_symlink(key_file_misc, filePath, MYF(MY_WME)))
-      {
-	goto err;
-      }
+	DBUG_RETURN(true);
     }
   }
-  if (thd->killed ||
-      (tot_list && mysql_rm_table_part2(thd, tot_list, 1, 0, 1, 1)))
-    goto err;
-
-  /* Remove RAID directories */
-  {
-    List_iterator<String> it(raid_dirs);
-    String *dir;
-    while ((dir= it++))
-      if (rmdir(dir->c_ptr()) < 0)
-	found_other_files++;
-  }
-  my_dirend(dirp);  
-  
-  if (dropped_tables)
-    *dropped_tables= tot_list;
-  
-  /*
-    If the directory is a symbolic link, remove the link first, then
-    remove the directory the symbolic link pointed at
-  */
-  if (found_other_files)
-  {
-    my_error(ER_DB_DROP_RMDIR, MYF(0), org_path, EEXIST);
-    DBUG_RETURN(-1);
-  }
-  else
-  {
-    /* Don't give errors if we can't delete 'RAID' directory */
-    if (rm_dir_w_symlink(org_path, level == 0))
-      DBUG_RETURN(-1);
-  }
-
-  DBUG_RETURN(deleted);
-
-err:
-  my_dirend(dirp);
-  DBUG_RETURN(-1);
+  *tables= tot_list;
+  DBUG_RETURN(false);
 }
 
 
diff --git a/sql/sql_error.cc b/sql/sql_error.cc
index 8c038e10a1f..d0982b879e7 100644
--- a/sql/sql_error.cc
+++ b/sql/sql_error.cc
@@ -334,7 +334,6 @@ Diagnostics_area::reset_diagnostics_area()
   /** Don't take chances in production */
   m_message[0]= '\0';
   m_sql_errno= 0;
-  m_server_status= 0;
   m_affected_rows= 0;
   m_last_insert_id= 0;
   m_statement_warn_count= 0;
@@ -365,7 +364,6 @@ Diagnostics_area::set_ok_status(THD *thd, ulonglong affected_rows_arg,
   if (is_error() || is_disabled())
     return;
 
-  m_server_status= thd->server_status;
   m_statement_warn_count= thd->warning_info->statement_warn_count();
   m_affected_rows= affected_rows_arg;
   m_last_insert_id= last_insert_id_arg;
@@ -395,7 +393,6 @@ Diagnostics_area::set_eof_status(THD *thd)
   if (is_error() || is_disabled())
     return;
 
-  m_server_status= thd->server_status;
   /*
     If inside a stored procedure, do not return the total
     number of warnings, since they are not available to the client
diff --git a/sql/sql_error.h b/sql/sql_error.h
index 87e98e27673..14dc5e6d12c 100644
--- a/sql/sql_error.h
+++ b/sql/sql_error.h
@@ -79,12 +79,6 @@ public:
   const char* get_sqlstate() const
   { DBUG_ASSERT(m_status == DA_ERROR); return m_sqlstate; }
 
-  uint server_status() const
-  {
-    DBUG_ASSERT(m_status == DA_OK || m_status == DA_EOF);
-    return m_server_status;
-  }
-
   ulonglong affected_rows() const
   { DBUG_ASSERT(m_status == DA_OK); return m_affected_rows; }
 
@@ -111,15 +105,6 @@ private:
   char m_sqlstate[SQLSTATE_LENGTH+1];
 
   /**
-    Copied from thd->server_status when the diagnostics area is assigned.
-    We need this member as some places in the code use the following pattern:
-    thd->server_status|= ...
-    my_eof(thd);
-    thd->server_status&= ~...
-    Assigned by OK, EOF or ERROR.
-  */
-  uint m_server_status;
-  /**
     The number of rows affected by the last statement. This is
     semantically close to thd->row_count_func, but has a different
     life cycle. thd->row_count_func stores the value returned by
diff --git a/sql/sql_handler.cc b/sql/sql_handler.cc
index a5c126a8521..b5cd3ac9e9a 100644
--- a/sql/sql_handler.cc
+++ b/sql/sql_handler.cc
@@ -55,7 +55,7 @@
 #include "sql_handler.h"
 #include "unireg.h"                    // REQUIRED: for other includes
 #include "sql_base.h"                           // close_thread_tables
-#include "lock.h"            // broadcast_refresh, mysql_unlock_tables
+#include "lock.h"                               // mysql_unlock_tables
 #include "key.h"                                // key_copy
 #include "sql_base.h"                           // insert_fields
 #include "sql_select.h"
@@ -131,11 +131,7 @@ static void mysql_ha_close_table(THD *thd, TABLE_LIST *tables)
     /* Non temporary table. */
     tables->table->file->ha_index_or_rnd_end();
     tables->table->open_by_handler= 0;
-    if (close_thread_table(thd, &tables->table))
-    {
-      /* Tell threads waiting for refresh that something has happened */
-      broadcast_refresh();
-    }
+    (void) close_thread_table(thd, &tables->table);
     thd->mdl_context.release_lock(tables->mdl_request.ticket);
   }
   else if (tables->table)
@@ -183,7 +179,7 @@ bool mysql_ha_open(THD *thd, TABLE_LIST *tables, bool reopen)
   uint          dblen, namelen, aliaslen, counter;
   bool          error;
   TABLE         *backup_open_tables;
-  MDL_ticket    *mdl_savepoint;
+  MDL_savepoint mdl_savepoint;
   DBUG_ENTER("mysql_ha_open");
   DBUG_PRINT("enter",("'%s'.'%s' as '%s'  reopen: %d",
                       tables->db, tables->table_name, tables->alias,
@@ -252,7 +248,13 @@ bool mysql_ha_open(THD *thd, TABLE_LIST *tables, bool reopen)
     memcpy(hash_tables->db, tables->db, dblen);
     memcpy(hash_tables->table_name, tables->table_name, namelen);
     memcpy(hash_tables->alias, tables->alias, aliaslen);
-    hash_tables->mdl_request.init(MDL_key::TABLE, db, name, MDL_SHARED);
+    /*
+      We can't request lock with explicit duration for this table
+      right from the start as open_tables() can't handle properly
+      back-off for such locks.
+    */
+    hash_tables->mdl_request.init(MDL_key::TABLE, db, name, MDL_SHARED,
+                                  MDL_TRANSACTION);
     /* for now HANDLER can be used only for real TABLES */
     hash_tables->required_type= FRMTYPE_TABLE;
 
@@ -332,8 +334,8 @@ bool mysql_ha_open(THD *thd, TABLE_LIST *tables, bool reopen)
   thd->set_open_tables(backup_open_tables);
   if (hash_tables->mdl_request.ticket)
   {
-    thd->mdl_context.
-      move_ticket_after_trans_sentinel(hash_tables->mdl_request.ticket);
+    thd->mdl_context.set_lock_duration(hash_tables->mdl_request.ticket,
+                                       MDL_EXPLICIT);
     thd->mdl_context.set_needs_thr_lock_abort(TRUE);
   }
 
@@ -969,24 +971,23 @@ void mysql_ha_cleanup(THD *thd)
 
 
 /**
-  Move tickets for metadata locks corresponding to open HANDLERs
-  after transaction sentinel in order to protect them from being
-  released at the end of transaction.
+  Set explicit duration for metadata locks corresponding to open HANDLERs
+  to protect them from being released at the end of transaction.
 
   @param thd Thread identifier.
 */
 
-void mysql_ha_move_tickets_after_trans_sentinel(THD *thd)
+void mysql_ha_set_explicit_lock_duration(THD *thd)
 {
   TABLE_LIST *hash_tables;
-  DBUG_ENTER("mysql_ha_move_tickets_after_trans_sentinel");
+  DBUG_ENTER("mysql_ha_set_explicit_lock_duration");
 
   for (uint i= 0; i < thd->handler_tables_hash.records; i++)
   {
     hash_tables= (TABLE_LIST*) my_hash_element(&thd->handler_tables_hash, i);
     if (hash_tables->table && hash_tables->table->mdl_ticket)
-      thd->mdl_context.
-             move_ticket_after_trans_sentinel(hash_tables->table->mdl_ticket);
+      thd->mdl_context.set_lock_duration(hash_tables->table->mdl_ticket,
+                                         MDL_EXPLICIT);
   }
   DBUG_VOID_RETURN;
 }
diff --git a/sql/sql_handler.h b/sql/sql_handler.h
index c5da3c4d468..2eea552d7c9 100644
--- a/sql/sql_handler.h
+++ b/sql/sql_handler.h
@@ -31,6 +31,6 @@ void mysql_ha_flush(THD *thd);
 void mysql_ha_flush_tables(THD *thd, TABLE_LIST *all_tables);
 void mysql_ha_rm_tables(THD *thd, TABLE_LIST *tables);
 void mysql_ha_cleanup(THD *thd);
-void mysql_ha_move_tickets_after_trans_sentinel(THD *thd);
+void mysql_ha_set_explicit_lock_duration(THD *thd);
 
 #endif /* SQL_HANDLER_INCLUDED */
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index eaaf7901e25..bd2aa70806a 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -77,7 +77,8 @@
 #include "sql_audit.h"
 
 #ifndef EMBEDDED_LIBRARY
-static bool delayed_get_table(THD *thd, TABLE_LIST *table_list);
+static bool delayed_get_table(THD *thd, MDL_request *grl_protection_request,
+                              TABLE_LIST *table_list);
 static int write_delayed(THD *thd, TABLE *table, enum_duplicates duplic,
                          LEX_STRING query, bool ignore, bool log_on);
 static void end_delayed_insert(THD *thd);
@@ -529,32 +530,28 @@ void upgrade_lock_type(THD *thd, thr_lock_type *lock_type,
 static
 bool open_and_lock_for_insert_delayed(THD *thd, TABLE_LIST *table_list)
 {
+  MDL_request protection_request;
   DBUG_ENTER("open_and_lock_for_insert_delayed");
 
 #ifndef EMBEDDED_LIBRARY
-  if (thd->locked_tables_mode && thd->global_read_lock.is_acquired())
-  {
-    /*
-      If this connection has the global read lock, the handler thread
-      will not be able to lock the table. It will wait for the global
-      read lock to go away, but this will never happen since the
-      connection thread will be stuck waiting for the handler thread
-      to open and lock the table.
-      If we are not in locked tables mode, INSERT will seek protection
-      against the global read lock (and fail), thus we will only get
-      to this point in locked tables mode.
-    */
-    my_error(ER_CANT_UPDATE_WITH_READLOCK, MYF(0));
-    DBUG_RETURN(TRUE);
-  }
-
   /*
     In order for the deadlock detector to be able to find any deadlocks
-    caused by the handler thread locking this table, we take the metadata
-    lock inside the connection thread. If this goes ok, the ticket is cloned
-    and added to the list of granted locks held by the handler thread.
+    caused by the handler thread waiting for GRL or this table, we acquire
+    protection against GRL (global IX metadata lock) and metadata lock on
+    table to being inserted into inside the connection thread.
+    If this goes ok, the tickets are cloned and added to the list of granted
+    locks held by the handler thread.
   */
-  MDL_ticket *mdl_savepoint= thd->mdl_context.mdl_savepoint();
+  if (thd->global_read_lock.can_acquire_protection())
+    DBUG_RETURN(TRUE);
+
+  protection_request.init(MDL_key::GLOBAL, "", "", MDL_INTENTION_EXCLUSIVE,
+                          MDL_STATEMENT);
+
+  if (thd->mdl_context.acquire_lock(&protection_request,
+                                    thd->variables.lock_wait_timeout))
+    DBUG_RETURN(TRUE);
+
   if (thd->mdl_context.acquire_lock(&table_list->mdl_request,
                                     thd->variables.lock_wait_timeout))
     /*
@@ -564,7 +561,7 @@ bool open_and_lock_for_insert_delayed(THD *thd, TABLE_LIST *table_list)
     DBUG_RETURN(TRUE);
 
   bool error= FALSE;
-  if (delayed_get_table(thd, table_list))
+  if (delayed_get_table(thd, &protection_request, table_list))
     error= TRUE;
   else if (table_list->table)
   {
@@ -589,13 +586,13 @@ bool open_and_lock_for_insert_delayed(THD *thd, TABLE_LIST *table_list)
   }
 
   /*
-    If a lock was acquired above, we should release it after
-    handle_delayed_insert() has cloned the ticket. Note that acquire_lock() can
-    succeed because the connection already has the lock. In this case the ticket
-    will be before the mdl_savepoint and we should not release it here.
+    We can't release protection against GRL and metadata lock on the table
+    being inserted into here. These locks might be required, for example,
+    because this INSERT DELAYED calls functions which may try to update
+    this or another tables (updating the same table is of course illegal,
+    but such an attempt can be discovered only later during statement
+    execution).
   */
-  if (!thd->mdl_context.has_lock(mdl_savepoint, table_list->mdl_request.ticket))
-    thd->mdl_context.release_lock(table_list->mdl_request.ticket);
 
   /*
     Reset the ticket in case we end up having to use normal insert and
@@ -1873,10 +1870,11 @@ public:
   mysql_cond_t cond, cond_client;
   volatile uint tables_in_use,stacked_inserts;
   volatile bool status;
-  /*
+  /**
     When the handler thread starts, it clones a metadata lock ticket
-    for the table to be inserted. This is done to allow the deadlock
-    detector to detect deadlocks resulting from this lock.
+    which protects against GRL and ticket for the table to be inserted.
+    This is done to allow the deadlock detector to detect deadlocks
+    resulting from these locks.
     Before this is done, the connection thread cannot safely exit
     without causing problems for clone_ticket().
     Once handler_thread_initialized has been set, it is safe for the
@@ -1888,6 +1886,11 @@ public:
   I_List<delayed_row> rows;
   ulong group_count;
   TABLE_LIST table_list;			// Argument
+  /**
+    Request for IX metadata lock protecting against GRL which is
+    passed from connection thread to the handler thread.
+  */
+  MDL_request grl_protection;
 
   Delayed_insert()
     :locks_in_memory(0), table(0),tables_in_use(0),stacked_inserts(0),
@@ -2066,7 +2069,8 @@ Delayed_insert *find_handler(THD *thd, TABLE_LIST *table_list)
 */
 
 static
-bool delayed_get_table(THD *thd, TABLE_LIST *table_list)
+bool delayed_get_table(THD *thd, MDL_request *grl_protection_request,
+                       TABLE_LIST *table_list)
 {
   int error;
   Delayed_insert *di;
@@ -2111,7 +2115,10 @@ bool delayed_get_table(THD *thd, TABLE_LIST *table_list)
       /* Replace volatile strings with local copies */
       di->table_list.alias= di->table_list.table_name= di->thd.query();
       di->table_list.db= di->thd.db;
-      /* We need the ticket so that it can be cloned in handle_delayed_insert */
+      /* We need the tickets so that they can be cloned in handle_delayed_insert */
+      di->grl_protection.init(MDL_key::GLOBAL, "", "",
+                              MDL_INTENTION_EXCLUSIVE, MDL_STATEMENT);
+      di->grl_protection.ticket= grl_protection_request->ticket;
       init_mdl_requests(&di->table_list);
       di->table_list.mdl_request.ticket= table_list->mdl_request.ticket;
 
@@ -2651,13 +2658,15 @@ pthread_handler_t handle_delayed_insert(void *arg)
     thd->set_current_stmt_binlog_format_row_if_mixed();
 
     /*
-      Clone the ticket representing the lock on the target table for
-      the insert and add it to the list of granted metadata locks held by
-      the handler thread. This is safe since the handler thread is
-      not holding nor waiting on any metadata locks.
+      Clone tickets representing protection against GRL and the lock on
+      the target table for the insert and add them to the list of granted
+      metadata locks held by the handler thread. This is safe since the
+      handler thread is not holding nor waiting on any metadata locks.
     */
-    if (thd->mdl_context.clone_ticket(&di->table_list.mdl_request))
+    if (thd->mdl_context.clone_ticket(&di->grl_protection) ||
+        thd->mdl_context.clone_ticket(&di->table_list.mdl_request))
     {
+      thd->mdl_context.release_transactional_locks();
       di->handler_thread_initialized= TRUE;
       goto err;
     }
diff --git a/sql/sql_lex.cc b/sql/sql_lex.cc
index d91489b4a7a..7a8a86d3ca4 100644
--- a/sql/sql_lex.cc
+++ b/sql/sql_lex.cc
@@ -417,7 +417,6 @@ void lex_start(THD *thd)
   lex->nest_level=0 ;
   lex->allow_sum_func= 0;
   lex->in_sum_func= NULL;
-  lex->protect_against_global_read_lock= FALSE;
   /*
     ok, there must be a better solution for this, long-term
     I tried "bzero" in the sql_yacc.yy code, but that for
diff --git a/sql/sql_lex.h b/sql/sql_lex.h
index c20c2f0bca2..191562bd3e8 100644
--- a/sql/sql_lex.h
+++ b/sql/sql_lex.h
@@ -2394,22 +2394,6 @@ struct LEX: public Query_tables_list
   bool escape_used;
   bool is_lex_started; /* If lex_start() did run. For debugging. */
 
-  /*
-    Special case for SELECT .. FOR UPDATE and LOCK TABLES .. WRITE.
-
-    Protect from a impending GRL as otherwise the thread might deadlock
-    if it starts waiting for the GRL in mysql_lock_tables.
-
-    The protection is needed because there is a race between setting
-    the global read lock and waiting for all open tables to be closed.
-    The problem is a circular wait where a thread holding "old" open
-    tables will wait for the global read lock to be released while the
-    thread holding the global read lock will wait for all "old" open
-    tables to be closed -- the flush part of flush tables with read
-    lock.
-  */
-  bool protect_against_global_read_lock;
-
   LEX();
 
   virtual ~LEX()
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 4d2c0cbb60a..40cecb75b69 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -18,12 +18,9 @@
 #include "sql_priv.h"
 #include "unireg.h"                    // REQUIRED: for other includes
 #include "sql_parse.h"        // sql_kill, *_precheck, *_prepare
-#include "lock.h"             // wait_if_global_read_lock,
-                              // unlock_global_read_lock,
-                              // try_transactional_lock,
+#include "lock.h"             // try_transactional_lock,
                               // check_transactional_lock,
                               // set_handler_table_locks,
-                              // start_waiting_global_read_lock,
                               // lock_global_read_lock,
                               // make_global_read_lock_block_commit
 #include "sql_base.h"         // find_temporary_tablesx
@@ -260,21 +257,20 @@ void init_update_queries(void)
     the code, in particular in the Query_log_event's constructor.
   */
   sql_command_flags[SQLCOM_CREATE_TABLE]=   CF_CHANGES_DATA | CF_REEXECUTION_FRAGILE |
-                                            CF_AUTO_COMMIT_TRANS | CF_PROTECT_AGAINST_GRL |
+                                            CF_AUTO_COMMIT_TRANS |
                                             CF_CAN_GENERATE_ROW_EVENTS;
   sql_command_flags[SQLCOM_CREATE_INDEX]=   CF_CHANGES_DATA | CF_AUTO_COMMIT_TRANS;
   sql_command_flags[SQLCOM_ALTER_TABLE]=    CF_CHANGES_DATA | CF_WRITE_LOGS_COMMAND |
-                                            CF_AUTO_COMMIT_TRANS | CF_PROTECT_AGAINST_GRL;
+                                            CF_AUTO_COMMIT_TRANS;
   sql_command_flags[SQLCOM_TRUNCATE]=       CF_CHANGES_DATA | CF_WRITE_LOGS_COMMAND |
-                                            CF_AUTO_COMMIT_TRANS | CF_PROTECT_AGAINST_GRL;
+                                            CF_AUTO_COMMIT_TRANS;
   sql_command_flags[SQLCOM_DROP_TABLE]=     CF_CHANGES_DATA | CF_AUTO_COMMIT_TRANS;
   sql_command_flags[SQLCOM_LOAD]=           CF_CHANGES_DATA | CF_REEXECUTION_FRAGILE |
-                                            CF_PROTECT_AGAINST_GRL |
                                             CF_CAN_GENERATE_ROW_EVENTS;
-  sql_command_flags[SQLCOM_CREATE_DB]=      CF_CHANGES_DATA | CF_AUTO_COMMIT_TRANS | CF_PROTECT_AGAINST_GRL;
-  sql_command_flags[SQLCOM_DROP_DB]=        CF_CHANGES_DATA | CF_AUTO_COMMIT_TRANS | CF_PROTECT_AGAINST_GRL;
-  sql_command_flags[SQLCOM_ALTER_DB_UPGRADE]= CF_AUTO_COMMIT_TRANS | CF_PROTECT_AGAINST_GRL;
-  sql_command_flags[SQLCOM_ALTER_DB]=       CF_CHANGES_DATA | CF_AUTO_COMMIT_TRANS | CF_PROTECT_AGAINST_GRL;
+  sql_command_flags[SQLCOM_CREATE_DB]=      CF_CHANGES_DATA | CF_AUTO_COMMIT_TRANS;
+  sql_command_flags[SQLCOM_DROP_DB]=        CF_CHANGES_DATA | CF_AUTO_COMMIT_TRANS;
+  sql_command_flags[SQLCOM_ALTER_DB_UPGRADE]= CF_AUTO_COMMIT_TRANS;
+  sql_command_flags[SQLCOM_ALTER_DB]=       CF_CHANGES_DATA | CF_AUTO_COMMIT_TRANS;
   sql_command_flags[SQLCOM_RENAME_TABLE]=   CF_CHANGES_DATA | CF_AUTO_COMMIT_TRANS;
   sql_command_flags[SQLCOM_DROP_INDEX]=     CF_CHANGES_DATA | CF_AUTO_COMMIT_TRANS;
   sql_command_flags[SQLCOM_CREATE_VIEW]=    CF_CHANGES_DATA | CF_REEXECUTION_FRAGILE |
@@ -285,26 +281,18 @@ void init_update_queries(void)
   sql_command_flags[SQLCOM_CREATE_EVENT]=   CF_CHANGES_DATA | CF_AUTO_COMMIT_TRANS;
   sql_command_flags[SQLCOM_ALTER_EVENT]=    CF_CHANGES_DATA | CF_AUTO_COMMIT_TRANS;
   sql_command_flags[SQLCOM_DROP_EVENT]=     CF_CHANGES_DATA | CF_AUTO_COMMIT_TRANS;
-  sql_command_flags[SQLCOM_CREATE_TRIGGER]= CF_AUTO_COMMIT_TRANS;
-  sql_command_flags[SQLCOM_DROP_TRIGGER]=   CF_AUTO_COMMIT_TRANS;
 
   sql_command_flags[SQLCOM_UPDATE]=	    CF_CHANGES_DATA | CF_REEXECUTION_FRAGILE |
-                                            CF_PROTECT_AGAINST_GRL |
                                             CF_CAN_GENERATE_ROW_EVENTS;
   sql_command_flags[SQLCOM_UPDATE_MULTI]=   CF_CHANGES_DATA | CF_REEXECUTION_FRAGILE |
-                                            CF_PROTECT_AGAINST_GRL |
                                             CF_CAN_GENERATE_ROW_EVENTS;
   sql_command_flags[SQLCOM_INSERT]=	    CF_CHANGES_DATA | CF_REEXECUTION_FRAGILE |
-                                            CF_PROTECT_AGAINST_GRL |
                                             CF_CAN_GENERATE_ROW_EVENTS;
   sql_command_flags[SQLCOM_INSERT_SELECT]=  CF_CHANGES_DATA | CF_REEXECUTION_FRAGILE |
-                                            CF_PROTECT_AGAINST_GRL |
                                             CF_CAN_GENERATE_ROW_EVENTS;
   sql_command_flags[SQLCOM_DELETE]=         CF_CHANGES_DATA | CF_REEXECUTION_FRAGILE |
-                                            CF_PROTECT_AGAINST_GRL |
                                             CF_CAN_GENERATE_ROW_EVENTS;
   sql_command_flags[SQLCOM_DELETE_MULTI]=   CF_CHANGES_DATA | CF_REEXECUTION_FRAGILE |
-                                            CF_PROTECT_AGAINST_GRL |
                                             CF_CAN_GENERATE_ROW_EVENTS;
   sql_command_flags[SQLCOM_REPLACE]=        CF_CHANGES_DATA | CF_REEXECUTION_FRAGILE |
                                             CF_CAN_GENERATE_ROW_EVENTS;
@@ -366,20 +354,19 @@ void init_update_queries(void)
                                                 CF_REEXECUTION_FRAGILE);
 
 
-  sql_command_flags[SQLCOM_CREATE_USER]=       CF_CHANGES_DATA | CF_PROTECT_AGAINST_GRL;
-  sql_command_flags[SQLCOM_RENAME_USER]=       CF_CHANGES_DATA | CF_PROTECT_AGAINST_GRL;
-  sql_command_flags[SQLCOM_DROP_USER]=         CF_CHANGES_DATA | CF_PROTECT_AGAINST_GRL;
+  sql_command_flags[SQLCOM_CREATE_USER]=       CF_CHANGES_DATA;
+  sql_command_flags[SQLCOM_RENAME_USER]=       CF_CHANGES_DATA;
+  sql_command_flags[SQLCOM_DROP_USER]=         CF_CHANGES_DATA;
   sql_command_flags[SQLCOM_GRANT]=             CF_CHANGES_DATA;
   sql_command_flags[SQLCOM_REVOKE]=            CF_CHANGES_DATA;
-  sql_command_flags[SQLCOM_REVOKE_ALL]=        CF_PROTECT_AGAINST_GRL;
   sql_command_flags[SQLCOM_OPTIMIZE]=          CF_CHANGES_DATA;
   sql_command_flags[SQLCOM_CREATE_FUNCTION]=   CF_CHANGES_DATA;
-  sql_command_flags[SQLCOM_CREATE_PROCEDURE]=  CF_CHANGES_DATA | CF_PROTECT_AGAINST_GRL | CF_AUTO_COMMIT_TRANS;
-  sql_command_flags[SQLCOM_CREATE_SPFUNCTION]= CF_CHANGES_DATA | CF_PROTECT_AGAINST_GRL | CF_AUTO_COMMIT_TRANS;
-  sql_command_flags[SQLCOM_DROP_PROCEDURE]=    CF_CHANGES_DATA | CF_PROTECT_AGAINST_GRL | CF_AUTO_COMMIT_TRANS;
-  sql_command_flags[SQLCOM_DROP_FUNCTION]=     CF_CHANGES_DATA | CF_PROTECT_AGAINST_GRL | CF_AUTO_COMMIT_TRANS;
-  sql_command_flags[SQLCOM_ALTER_PROCEDURE]=   CF_CHANGES_DATA | CF_PROTECT_AGAINST_GRL | CF_AUTO_COMMIT_TRANS;
-  sql_command_flags[SQLCOM_ALTER_FUNCTION]=    CF_CHANGES_DATA | CF_PROTECT_AGAINST_GRL | CF_AUTO_COMMIT_TRANS;
+  sql_command_flags[SQLCOM_CREATE_PROCEDURE]=  CF_CHANGES_DATA | CF_AUTO_COMMIT_TRANS;
+  sql_command_flags[SQLCOM_CREATE_SPFUNCTION]= CF_CHANGES_DATA | CF_AUTO_COMMIT_TRANS;
+  sql_command_flags[SQLCOM_DROP_PROCEDURE]=    CF_CHANGES_DATA | CF_AUTO_COMMIT_TRANS;
+  sql_command_flags[SQLCOM_DROP_FUNCTION]=     CF_CHANGES_DATA | CF_AUTO_COMMIT_TRANS;
+  sql_command_flags[SQLCOM_ALTER_PROCEDURE]=   CF_CHANGES_DATA | CF_AUTO_COMMIT_TRANS;
+  sql_command_flags[SQLCOM_ALTER_FUNCTION]=    CF_CHANGES_DATA | CF_AUTO_COMMIT_TRANS;
   sql_command_flags[SQLCOM_INSTALL_PLUGIN]=    CF_CHANGES_DATA;
   sql_command_flags[SQLCOM_UNINSTALL_PLUGIN]=  CF_CHANGES_DATA;
 
@@ -712,6 +699,22 @@ bool do_command(THD *thd)
 
   net_new_transaction(net);
 
+  /*
+    Synchronization point for testing of KILL_CONNECTION.
+    This sync point can wait here, to simulate slow code execution
+    between the last test of thd->killed and blocking in read().
+
+    The goal of this test is to verify that a connection does not
+    hang, if it is killed at this point of execution.
+    (Bug#37780 - main.kill fails randomly)
+
+    Note that the sync point wait itself will be terminated by a
+    kill. In this case it consumes a condition broadcast, but does
+    not change anything else. The consumed broadcast should not
+    matter here, because the read/recv() below doesn't use it.
+  */
+  DEBUG_SYNC(thd, "before_do_command_net_read");
+
   if ((packet_length= my_net_read(net)) == packet_error)
   {
     DBUG_PRINT("info",("Got error %d reading command from socket %s",
@@ -1031,11 +1034,13 @@ bool dispatch_command(enum enum_server_command command, THD *thd,
     while (!thd->killed && (parser_state.m_lip.found_semicolon != NULL) &&
            ! thd->is_error())
     {
-      char *beginning_of_next_stmt= (char*)
-        parser_state.m_lip.found_semicolon;
       /*
         Multiple queries exits, execute them individually
       */
+      char *beginning_of_next_stmt= (char*) parser_state.m_lip.found_semicolon;
+
+      /* Finalize server status flags after executing a statement. */
+      thd->update_server_status();
       thd->protocol->end_statement();
       query_cache_end_of_result(thd);
       ulong length= (ulong)(packet_end - beginning_of_next_stmt);
@@ -1096,7 +1101,7 @@ bool dispatch_command(enum enum_server_command command, THD *thd,
       SHOW statements should not add the used tables to the list of tables
       used in a transaction.
     */
-    MDL_ticket *mdl_savepoint= thd->mdl_context.mdl_savepoint();
+    MDL_savepoint mdl_savepoint= thd->mdl_context.mdl_savepoint();
 
     status_var_increment(thd->status_var.com_stat[SQLCOM_SHOW_FIELDS]);
     if (thd->copy_db_to(&db.str, &db.length))
@@ -1382,6 +1387,8 @@ bool dispatch_command(enum enum_server_command command, THD *thd,
               (thd->open_tables == NULL ||
                (thd->locked_tables_mode == LTM_LOCK_TABLES)));
 
+  /* Finalize server status flags after executing a command. */
+  thd->update_server_status();
   thd->protocol->end_statement();
   query_cache_end_of_result(thd);
 
@@ -1436,8 +1443,7 @@ void log_slow_statement(THD *thd)
     ulonglong end_utime_of_query= thd->current_utime();
     thd_proc_info(thd, "logging slow query");
 
-    if (((end_utime_of_query - thd->utime_after_lock) >
-         thd->variables.long_query_time ||
+    if (((thd->server_status & SERVER_QUERY_WAS_SLOW) ||
          ((thd->server_status &
            (SERVER_QUERY_NO_INDEX_USED | SERVER_QUERY_NO_GOOD_INDEX_USED)) &&
           opt_log_queries_not_using_indexes &&
@@ -1739,16 +1745,6 @@ bool sp_process_definer(THD *thd)
 /**
   Execute command saved in thd and lex->sql_command.
 
-    Before every operation that can request a write lock for a table
-    wait if a global read lock exists. However do not wait if this
-    thread has locked tables already. No new locks can be requested
-    until the other locks are released. The thread that requests the
-    global read lock waits for write locked tables to become unlocked.
-
-    Note that wait_if_global_read_lock() sets a protection against a new
-    global read lock when it succeeds. This needs to be released by
-    start_waiting_global_read_lock() after the operation.
-
   @param thd                       Thread handle
 
   @todo
@@ -1782,7 +1778,6 @@ mysql_execute_command(THD *thd)
   /* have table map for update for multi-update statement (BUG#37051) */
   bool have_table_map_for_update= FALSE;
 #endif
-  /* Saved variable value */
   DBUG_ENTER("mysql_execute_command");
 #ifdef WITH_PARTITION_STORAGE_ENGINE
   thd->work_part_info= 0;
@@ -1974,17 +1969,6 @@ mysql_execute_command(THD *thd)
     thd->mdl_context.release_transactional_locks();
   }
 
-  /*
-    Check if this command needs protection against the global read lock
-    to avoid deadlock. See CF_PROTECT_AGAINST_GRL.
-    start_waiting_global_read_lock() is called at the end of
-    mysql_execute_command().
-  */
-  if (((sql_command_flags[lex->sql_command] & CF_PROTECT_AGAINST_GRL) != 0) &&
-      !thd->locked_tables_mode)
-    if (thd->global_read_lock.wait_if_global_read_lock(thd, FALSE, TRUE))
-      goto error;
-
 #ifndef DBUG_OFF
   if (lex->sql_command != SQLCOM_SET_OPTION)
     DEBUG_SYNC(thd,"before_execute_sql_command");
@@ -2059,10 +2043,6 @@ mysql_execute_command(THD *thd)
     if (res)
       break;
 
-    if (!thd->locked_tables_mode && lex->protect_against_global_read_lock &&
-        thd->global_read_lock.wait_if_global_read_lock(thd, FALSE, TRUE))
-      break;
-
     res= execute_sqlcom_select(thd, all_tables);
     break;
   }
@@ -2321,20 +2301,6 @@ case SQLCOM_PREPARE:
       create_info.default_table_charset= create_info.table_charset;
       create_info.table_charset= 0;
     }
-    /*
-      The create-select command will open and read-lock the select table
-      and then create, open and write-lock the new table. If a global
-      read lock steps in, we get a deadlock. The write lock waits for
-      the global read lock, while the global read lock waits for the
-      select table to be closed. So we wait until the global readlock is
-      gone before starting both steps. Note that
-      wait_if_global_read_lock() sets a protection against a new global
-      read lock when it succeeds. This needs to be released by
-      start_waiting_global_read_lock(). We protect the normal CREATE
-      TABLE in the same way. That way we avoid that a new table is
-      created during a global read lock.
-      Protection against grl is covered by the CF_PROTECT_AGAINST_GRL flag.
-    */
 
 #ifdef WITH_PARTITION_STORAGE_ENGINE
     {
@@ -3128,9 +3094,6 @@ end_with_restore_list:
     if (check_table_access(thd, LOCK_TABLES_ACL | SELECT_ACL, all_tables,
                            FALSE, UINT_MAX, FALSE))
       goto error;
-    if (lex->protect_against_global_read_lock &&
-        thd->global_read_lock.wait_if_global_read_lock(thd, FALSE, TRUE))
-      goto error;
 
     thd->variables.option_bits|= OPTION_TABLE_LOCK;
     thd->in_lock_tables=1;
@@ -3786,13 +3749,22 @@ end_with_restore_list:
       Security_context *backup= NULL;
       LEX_USER *definer= thd->lex->definer;
       /*
-        We're going to issue an implicit GRANT statement.
-        It takes metadata locks and updates system tables.
-        Make sure that sp_create_routine() did not leave any
-        locks in the MDL context, so there is no risk to
-        deadlock.
+        We're going to issue an implicit GRANT statement so we close all
+        open tables. We have to keep metadata locks as this ensures that
+        this statement is atomic against concurent FLUSH TABLES WITH READ
+        LOCK. Deadlocks which can arise due to fact that this implicit
+        statement takes metadata locks should be detected by a deadlock
+        detector in MDL subsystem and reported as errors.
+
+        No need to commit/rollback statement transaction, it's not started.
+
+        TODO: Long-term we should either ensure that implicit GRANT statement
+              is written into binary log as a separate statement or make both
+              creation of routine and implicit GRANT parts of one fully atomic
+              statement.
       */
-      close_mysql_tables(thd);
+      DBUG_ASSERT(thd->transaction.stmt.is_empty());
+      close_thread_tables(thd);
       /*
         Check if the definer exists on slave, 
         then use definer privilege to insert routine privileges to mysql.procs_priv.
@@ -4052,13 +4024,22 @@ create_sp_error:
 
 #ifndef NO_EMBEDDED_ACCESS_CHECKS
       /*
-        We're going to issue an implicit REVOKE statement.
-        It takes metadata locks and updates system tables.
-        Make sure that sp_create_routine() did not leave any
-        locks in the MDL context, so there is no risk to
-        deadlock.
+        We're going to issue an implicit REVOKE statement so we close all
+        open tables. We have to keep metadata locks as this ensures that
+        this statement is atomic against concurent FLUSH TABLES WITH READ
+        LOCK. Deadlocks which can arise due to fact that this implicit
+        statement takes metadata locks should be detected by a deadlock
+        detector in MDL subsystem and reported as errors.
+
+        No need to commit/rollback statement transaction, it's not started.
+
+        TODO: Long-term we should either ensure that implicit REVOKE statement
+              is written into binary log as a separate statement or make both
+              dropping of routine and implicit REVOKE parts of one fully atomic
+              statement.
       */
-      close_mysql_tables(thd);
+      DBUG_ASSERT(thd->transaction.stmt.is_empty());
+      close_thread_tables(thd);
 
       if (sp_result != SP_KEY_NOT_FOUND &&
           sp_automatic_privileges && !opt_noacl &&
@@ -4347,14 +4328,6 @@ error:
   res= TRUE;
 
 finish:
-  if (thd->global_read_lock.has_protection())
-  {
-    /*
-      Release the protection against the global read lock and wake
-      everyone, who might want to set a global read lock.
-    */
-    thd->global_read_lock.start_waiting_global_read_lock(thd);
-  }
 
   DBUG_ASSERT(!thd->in_active_multi_stmt_transaction() ||
                thd->in_multi_stmt_transaction_mode());
@@ -4390,6 +4363,11 @@ finish:
   close_thread_tables(thd);
   thd_proc_info(thd, 0);
 
+#ifndef DBUG_OFF
+  if (lex->sql_command != SQLCOM_SET_OPTION && ! thd->in_sub_stmt)
+    DEBUG_SYNC(thd, "execute_command_after_close_tables");
+#endif
+
   if (stmt_causes_implicit_commit(thd, CF_IMPLICIT_COMMIT_END))
   {
     /* No transaction control allowed in sub-statements. */
@@ -4415,6 +4393,10 @@ finish:
     */
     thd->mdl_context.release_transactional_locks();
   }
+  else if (! thd->in_sub_stmt)
+  {
+    thd->mdl_context.release_statement_locks();
+  }
 
   DBUG_RETURN(res || thd->is_error());
 }
@@ -5885,7 +5867,8 @@ TABLE_LIST *st_select_lex::add_table_to_list(THD *thd,
   ptr->next_name_resolution_table= NULL;
   /* Link table in global list (all used tables) */
   lex->add_to_query_tables(ptr);
-  ptr->mdl_request.init(MDL_key::TABLE, ptr->db, ptr->table_name, mdl_type);
+  ptr->mdl_request.init(MDL_key::TABLE, ptr->db, ptr->table_name, mdl_type,
+                        MDL_TRANSACTION);
   DBUG_RETURN(ptr);
 }
 
diff --git a/sql/sql_prepare.cc b/sql/sql_prepare.cc
index 664d08f27a9..4a1d1ae7fe3 100644
--- a/sql/sql_prepare.cc
+++ b/sql/sql_prepare.cc
@@ -3179,7 +3179,6 @@ 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.
@@ -3251,7 +3250,7 @@ bool Prepared_statement::prepare(const char *packet, uint packet_len)
     Marker used to release metadata locks acquired while the prepared
     statement is being checked.
   */
-  mdl_savepoint= thd->mdl_context.mdl_savepoint();
+  MDL_savepoint mdl_savepoint= thd->mdl_context.mdl_savepoint();
 
   /* 
    The only case where we should have items in the thd->free_list is
diff --git a/sql/sql_rename.cc b/sql/sql_rename.cc
index 8f990eae001..6a7b0b0b3ad 100644
--- a/sql/sql_rename.cc
+++ b/sql/sql_rename.cc
@@ -24,8 +24,7 @@
 #include "sql_table.h"                         // build_table_filename
 #include "sql_view.h"             // mysql_frm_type, mysql_rename_view
 #include "sql_trigger.h"
-#include "lock.h"       // wait_if_global_read_lock
-                        // start_waiting_global_read_lock
+#include "lock.h"       // MYSQL_OPEN_SKIP_TEMPORARY
 #include "sql_base.h"   // tdc_remove_table, lock_table_names,
 #include "sql_handler.h"                        // mysql_ha_rm_tables
 #include "datadict.h"
@@ -63,9 +62,6 @@ bool mysql_rename_tables(THD *thd, TABLE_LIST *table_list, bool silent)
 
   mysql_ha_rm_tables(thd, table_list);
 
-  if (thd->global_read_lock.wait_if_global_read_lock(thd, FALSE, TRUE))
-    DBUG_RETURN(1);
-
   if (logger.is_log_table_enabled(QUERY_LOG_GENERAL) ||
       logger.is_log_table_enabled(QUERY_LOG_SLOW))
   {
@@ -189,7 +185,6 @@ bool mysql_rename_tables(THD *thd, TABLE_LIST *table_list, bool silent)
     query_cache_invalidate3(thd, table_list, 0);
 
 err:
-  thd->global_read_lock.start_waiting_global_read_lock(thd);
   DBUG_RETURN(error || binlog_error);
 }
 
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index d521abd51a3..e9373bc455f 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -47,6 +47,7 @@
 #include "records.h"             // init_read_record, end_read_record
 #include "filesort.h"            // filesort_free_buffers
 #include "sql_union.h"           // mysql_union
+#include "debug_sync.h"          // DEBUG_SYNC
 #include <m_ctype.h>
 #include <my_bit.h>
 #include <hash.h>
@@ -852,6 +853,7 @@ JOIN::optimize()
   if (optimized)
     DBUG_RETURN(0);
   optimized= 1;
+  DEBUG_SYNC(thd, "before_join_optimize");
 
   thd_proc_info(thd, "optimizing");
   row_limit= ((select_distinct || order || group_list) ? HA_POS_ERROR :
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index 6283deed331..ec147a99553 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -671,7 +671,7 @@ mysqld_show_create(THD *thd, TABLE_LIST *table_list)
     Metadata locks taken during SHOW CREATE should be released when
     the statmement completes as it is an information statement.
   */
-  MDL_ticket *mdl_savepoint= thd->mdl_context.mdl_savepoint();
+  MDL_savepoint mdl_savepoint= thd->mdl_context.mdl_savepoint();
 
   /* We want to preserve the tree for views. */
   thd->lex->view_prepare_mode= TRUE;
@@ -3189,7 +3189,7 @@ try_acquire_high_prio_shared_mdl_lock(THD *thd, TABLE_LIST *table,
 {
   bool error;
   table->mdl_request.init(MDL_key::TABLE, table->db, table->table_name,
-                          MDL_SHARED_HIGH_PRIO);
+                          MDL_SHARED_HIGH_PRIO, MDL_TRANSACTION);
 
   if (can_deadlock)
   {
@@ -7748,7 +7748,7 @@ bool show_create_trigger(THD *thd, const sp_name *trg_name)
     Metadata locks taken during SHOW CREATE TRIGGER should be released when
     the statement completes as it is an information statement.
   */
-  MDL_ticket *mdl_savepoint= thd->mdl_context.mdl_savepoint();
+  MDL_savepoint mdl_savepoint= thd->mdl_context.mdl_savepoint();
 
   /*
     Open the table by name in order to load Table_triggers_list object.
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 1c6ca4a48d9..6d5180c38d9 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -23,9 +23,7 @@
 #include "sql_parse.h"                        // test_if_data_home_dir
 #include "sql_cache.h"                          // query_cache_*
 #include "sql_base.h"   // open_table_uncached, lock_table_names
-#include "lock.h"       // wait_if_global_read_lock
-                        // start_waiting_global_read_lock,
-                        // mysql_unlock_tables
+#include "lock.h"       // mysql_unlock_tables
 #include "strfunc.h"    // find_type2, find_set
 #include "sql_view.h" // view_checksum 
 #include "sql_truncate.h"                       // regenerate_locked_table 
@@ -1851,64 +1849,117 @@ bool mysql_rm_table(THD *thd,TABLE_LIST *tables, my_bool if_exists,
 {
   bool error;
   Drop_table_error_handler err_handler;
+  TABLE_LIST *table;
 
   DBUG_ENTER("mysql_rm_table");
 
-  /* mark for close and remove all cached entries */
+  /* Disable drop of enabled log tables, must be done before name locking */
+  for (table= tables; table; table= table->next_local)
+  {
+    if (check_if_log_table(table->db_length, table->db,
+                           table->table_name_length, table->table_name, true))
+    {
+      my_error(ER_BAD_LOG_STATEMENT, MYF(0), "DROP");
+      DBUG_RETURN(true);
+    }
+  }
+
+  mysql_ha_rm_tables(thd, tables);
 
   if (!drop_temporary)
   {
-    if (!thd->locked_tables_mode &&
-       thd->global_read_lock.wait_if_global_read_lock(thd, FALSE, TRUE))
-      DBUG_RETURN(TRUE);
+    if (!thd->locked_tables_mode)
+    {
+      if (lock_table_names(thd, tables, NULL, thd->variables.lock_wait_timeout,
+                           MYSQL_OPEN_SKIP_TEMPORARY))
+        DBUG_RETURN(true);
+      for (table= tables; table; table= table->next_local)
+        tdc_remove_table(thd, TDC_RT_REMOVE_ALL, table->db, table->table_name,
+                         false);
+    }
+    else
+    {
+      for (table= tables; table; table= table->next_local)
+        if (table->open_type != OT_BASE_ONLY &&
+	    find_temporary_table(thd, table))
+        {
+          /*
+            A temporary table.
+
+            Don't try to find a corresponding MDL lock or assign it
+            to table->mdl_request.ticket. There can't be metadata
+            locks for temporary tables: they are local to the session.
+
+            Later in this function we release the MDL lock only if
+            table->mdl_requeset.ticket is not NULL. Thus here we
+            ensure that we won't release the metadata lock on the base
+            table locked with LOCK TABLES as a side effect of temporary
+            table drop.
+          */
+          DBUG_ASSERT(table->mdl_request.ticket == NULL);
+        }
+        else
+        {
+          /*
+            Not a temporary table.
+
+            Since 'tables' list can't contain duplicates (this is ensured
+            by parser) it is safe to cache pointer to the TABLE instances
+            in its elements.
+          */
+          table->table= find_table_for_mdl_upgrade(thd->open_tables, table->db,
+                                                   table->table_name, false);
+          if (!table->table)
+            DBUG_RETURN(true);
+          table->mdl_request.ticket= table->table->mdl_ticket;
+        }
+    }
   }
 
+  /* mark for close and remove all cached entries */
   thd->push_internal_handler(&err_handler);
-  error= mysql_rm_table_part2(thd, tables, if_exists, drop_temporary, 0, 0);
+  error= mysql_rm_table_no_locks(thd, tables, if_exists, drop_temporary,
+                                 false, false);
   thd->pop_internal_handler();
 
-  if (thd->global_read_lock.has_protection())
-    thd->global_read_lock.start_waiting_global_read_lock(thd);
-
   if (error)
     DBUG_RETURN(TRUE);
   my_ok(thd);
   DBUG_RETURN(FALSE);
 }
 
-/*
-  Execute the drop of a normal or temporary table
-
-  SYNOPSIS
-    mysql_rm_table_part2()
-    thd			Thread handler
-    tables		Tables to drop
-    if_exists		If set, don't give an error if table doesn't exists.
-			In this case we give an warning of level 'NOTE'
-    drop_temporary	Only drop temporary tables
-    drop_view		Allow to delete VIEW .frm
-    dont_log_query	Don't write query to log files. This will also not
-			generate warnings if the handler files doesn't exists  
-
-  TODO:
-    When logging to the binary log, we should log
-    tmp_tables and transactional tables as separate statements if we
-    are in a transaction;  This is needed to get these tables into the
-    cached binary log that is only written on COMMIT.
-
-   The current code only writes DROP statements that only uses temporary
-   tables to the cache binary log.  This should be ok on most cases, but
-   not all.
 
- RETURN
-   0	ok
-   1	Error
-   -1	Thread was killed
+/**
+  Execute the drop of a normal or temporary table.
+
+  @param  thd             Thread handler
+  @param  tables          Tables to drop
+  @param  if_exists       If set, don't give an error if table doesn't exists.
+                          In this case we give an warning of level 'NOTE'
+  @param  drop_temporary  Only drop temporary tables
+  @param  drop_view       Allow to delete VIEW .frm
+  @param  dont_log_query  Don't write query to log files. This will also not
+                          generate warnings if the handler files doesn't exists
+
+  @retval  0  ok
+  @retval  1  Error
+  @retval -1  Thread was killed
+
+  @note This function assumes that metadata locks have already been taken.
+        It is also assumed that the tables have been removed from TDC.
+
+  @todo When logging to the binary log, we should log
+        tmp_tables and transactional tables as separate statements if we
+        are in a transaction;  This is needed to get these tables into the
+        cached binary log that is only written on COMMIT.
+        The current code only writes DROP statements that only uses temporary
+        tables to the cache binary log.  This should be ok on most cases, but
+        not all.
 */
 
-int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
-			 bool drop_temporary, bool drop_view,
-			 bool dont_log_query)
+int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
+                            bool drop_temporary, bool drop_view,
+                            bool dont_log_query)
 {
   TABLE_LIST *table;
   char path[FN_REFLEN + 1], *alias= NULL;
@@ -1922,7 +1973,7 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
   bool non_tmp_table_deleted= 0;
   String built_query;
   String built_trans_tmp_query, built_non_trans_tmp_query;
-  DBUG_ENTER("mysql_rm_table_part2");
+  DBUG_ENTER("mysql_rm_table_no_locks");
 
   /*
     Prepares the drop statements that will be written into the binary
@@ -1976,71 +2027,6 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
     }
   }
 
-  mysql_ha_rm_tables(thd, tables);
-
-  /* Disable drop of enabled log tables, must be done before name locking */
-  for (table= tables; table; table= table->next_local)
-  {
-    if (check_if_log_table(table->db_length, table->db,
-                           table->table_name_length, table->table_name, 1))
-    {
-      my_error(ER_BAD_LOG_STATEMENT, MYF(0), "DROP");
-      DBUG_RETURN(1);
-    }
-  }
-
-  if (!drop_temporary)
-  {
-    if (!thd->locked_tables_mode)
-    {
-      if (lock_table_names(thd, tables, NULL, thd->variables.lock_wait_timeout,
-                           MYSQL_OPEN_SKIP_TEMPORARY))
-        DBUG_RETURN(1);
-      for (table= tables; table; table= table->next_local)
-      {
-        tdc_remove_table(thd, TDC_RT_REMOVE_ALL, table->db, table->table_name,
-                         FALSE);
-      }
-    }
-    else
-    {
-      for (table= tables; table; table= table->next_local)
-        if (table->open_type != OT_BASE_ONLY &&
-	    find_temporary_table(thd, table))
-        {
-          /*
-            A temporary table.
-
-            Don't try to find a corresponding MDL lock or assign it
-            to table->mdl_request.ticket. There can't be metadata
-            locks for temporary tables: they are local to the session.
-
-            Later in this function we release the MDL lock only if
-            table->mdl_requeset.ticket is not NULL. Thus here we
-            ensure that we won't release the metadata lock on the base
-            table locked with LOCK TABLES as a side effect of temporary
-            table drop.
-          */
-          DBUG_ASSERT(table->mdl_request.ticket == NULL);
-        }
-        else
-        {
-          /*
-            Not a temporary table.
-
-            Since 'tables' list can't contain duplicates (this is ensured
-            by parser) it is safe to cache pointer to the TABLE instances
-            in its elements.
-          */
-          table->table= find_table_for_mdl_upgrade(thd->open_tables, table->db,
-                                                   table->table_name, FALSE);
-          if (!table->table)
-            DBUG_RETURN(1);
-          table->mdl_request.ticket= table->table->mdl_ticket;
-        }
-    }
-  }
-
   for (table= tables; table; table= table->next_local)
   {
     bool is_trans;
@@ -2053,6 +2039,16 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
                          table->table ? (long) table->table->s : (long) -1));
 
     /*
+      If we are in locked tables mode and are dropping a temporary table,
+      the ticket should be NULL to ensure that we don't release a lock
+      on a base table later.
+    */
+    DBUG_ASSERT(!(thd->locked_tables_mode &&
+                  table->open_type != OT_BASE_ONLY &&
+                  find_temporary_table(thd, table) &&
+                  table->mdl_request.ticket != NULL));
+
+    /*
       drop_temporary_table may return one of the following error codes:
       .  0 - a temporary table was successfully dropped.
       .  1 - a temporary table was not found.
@@ -2127,6 +2123,10 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
         table->table= 0;
       }
 
+      /* Check that we have an exclusive lock on the table to be dropped. */
+      DBUG_ASSERT(thd->mdl_context.is_lock_owner(MDL_key::TABLE, table->db,
+                                                 table->table_name,
+                                                 MDL_EXCLUSIVE));
       if (thd->killed)
       {
         error= -1;
@@ -2169,8 +2169,8 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
         built_query.append("`,");
       }
     }
-    DEBUG_SYNC(thd, "rm_table_part2_before_delete_table");
-    DBUG_EXECUTE_IF("sleep_before_part2_delete_table",
+    DEBUG_SYNC(thd, "rm_table_no_locks_before_delete_table");
+    DBUG_EXECUTE_IF("sleep_before_no_locks_delete_table",
                     my_sleep(100000););
     error= 0;
     if (drop_temporary ||
@@ -2257,7 +2257,7 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
                                     ER(ER_BAD_TABLE_ERROR), MYF(0),
                                     table->table_name););
   }
-  DEBUG_SYNC(thd, "rm_table_part2_before_binlog");
+  DEBUG_SYNC(thd, "rm_table_no_locks_before_binlog");
   thd->thread_specific_used|= (trans_tmp_table_deleted ||
                                non_trans_tmp_table_deleted);
   error= 0;
@@ -5592,7 +5592,6 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
   TABLE *table, *new_table= 0;
   MDL_ticket *mdl_ticket;
   MDL_request target_mdl_request;
-  bool has_target_mdl_lock= FALSE;
   int error= 0;
   char tmp_name[80],old_name[32],new_name_buff[FN_REFLEN + 1];
   char new_alias_buff[FN_REFLEN], *table_name, *db, *new_alias, *alias;
@@ -5754,7 +5753,7 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
       else
       {
         target_mdl_request.init(MDL_key::TABLE, new_db, new_name,
-                                MDL_EXCLUSIVE);
+                                MDL_EXCLUSIVE, MDL_TRANSACTION);
         /*
           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.
@@ -5772,7 +5771,6 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
 	  DBUG_RETURN(TRUE);
         }
         DEBUG_SYNC(thd, "locked_table_name");
-        has_target_mdl_lock= TRUE;
         /*
           Table maybe does not exist, but we got an exclusive lock
           on the name, now we can safely try to find out for sure.
@@ -5959,10 +5957,7 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
         along with the implicit commit.
       */
       if (new_name != table_name || new_db != db)
-      {
-        thd->mdl_context.release_lock(target_mdl_request.ticket);
         thd->mdl_context.release_all_locks_for_name(mdl_ticket);
-      }
       else
         mdl_ticket->downgrade_exclusive_lock(MDL_SHARED_NO_READ_WRITE);
     }
@@ -6667,10 +6662,7 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
       thd->locked_tables_mode == LTM_PRELOCKED_UNDER_LOCK_TABLES)
   {
     if ((new_name != table_name || new_db != db))
-    {
-      thd->mdl_context.release_lock(target_mdl_request.ticket);
       thd->mdl_context.release_all_locks_for_name(mdl_ticket);
-    }
     else
       mdl_ticket->downgrade_exclusive_lock(MDL_SHARED_NO_READ_WRITE);
   }
@@ -6731,8 +6723,6 @@ err:
                                  alter_info->datetime_field->field_name);
     thd->abort_on_warning= save_abort_on_warning;
   }
-  if (has_target_mdl_lock)
-    thd->mdl_context.release_lock(target_mdl_request.ticket);
 
   DBUG_RETURN(TRUE);
 
@@ -6744,9 +6734,6 @@ err_with_mdl:
     tables and release the exclusive metadata lock.
   */
   thd->locked_tables_list.unlink_all_closed_tables(thd, NULL, 0);
-  if (has_target_mdl_lock)
-    thd->mdl_context.release_lock(target_mdl_request.ticket);
-
   thd->mdl_context.release_all_locks_for_name(mdl_ticket);
   DBUG_RETURN(TRUE);
 }
diff --git a/sql/sql_table.h b/sql/sql_table.h
index eb0b1aa94dd..aa5738fd4c9 100644
--- a/sql/sql_table.h
+++ b/sql/sql_table.h
@@ -174,8 +174,9 @@ bool mysql_checksum_table(THD* thd, TABLE_LIST* table_list,
                           HA_CHECK_OPT* check_opt);
 bool mysql_rm_table(THD *thd,TABLE_LIST *tables, my_bool if_exists,
                     my_bool drop_temporary);
-int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
-                         bool drop_temporary, bool drop_view, bool log_query);
+int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
+                            bool drop_temporary, bool drop_view,
+                            bool log_query);
 bool quick_rm_table(handlerton *base,const char *db,
                     const char *table_name, uint flags);
 void close_cached_table(THD *thd, TABLE *table);
diff --git a/sql/sql_trigger.cc b/sql/sql_trigger.cc
index a9b52eee9fc..c2e3cd9944a 100644
--- a/sql/sql_trigger.cc
+++ b/sql/sql_trigger.cc
@@ -24,8 +24,6 @@
 #include "parse_file.h"
 #include "sp.h"
 #include "sql_base.h"                          // find_temporary_table
-#include "lock.h"                    // wait_if_global_read_lock,
-                                     // start_waiting_global_read_lock
 #include "sql_show.h"                // append_definer, append_identifier
 #include "sql_table.h"                        // build_table_filename,
                                               // check_n_cut_mysql50_prefix
@@ -391,15 +389,6 @@ bool mysql_create_or_drop_trigger(THD *thd, TABLE_LIST *tables, bool create)
     DBUG_RETURN(TRUE);
   }
 
-  /*
-    We don't want perform our operations while global read lock is held
-    so we have to wait until its end and then prevent it from occurring
-    again until we are done, unless we are under lock tables.
-  */
-  if (!thd->locked_tables_mode &&
-      thd->global_read_lock.wait_if_global_read_lock(thd, FALSE, TRUE))
-    DBUG_RETURN(TRUE);
-
   if (!create)
   {
     bool if_exists= thd->lex->drop_if_exists;
@@ -547,9 +536,6 @@ end:
   if (!create)
     thd->lex->restore_backup_query_tables_list(&backup);
 
-  if (thd->global_read_lock.has_protection())
-    thd->global_read_lock.start_waiting_global_read_lock(thd);
-
   if (!result)
     my_ok(thd);
 
diff --git a/sql/sql_update.cc b/sql/sql_update.cc
index 8cc8d511fd6..b49ed2beafe 100644
--- a/sql/sql_update.cc
+++ b/sql/sql_update.cc
@@ -1026,9 +1026,17 @@ int mysql_multi_update_prepare(THD *thd)
   /* following need for prepared statements, to run next time multi-update */
   thd->lex->sql_command= SQLCOM_UPDATE_MULTI;
 
-  /* open tables and create derived ones, but do not lock and fill them */
+  /*
+    Open tables and create derived ones, but do not lock and fill them yet.
+
+    During prepare phase acquire only S metadata locks instead of SW locks to
+    keep prepare of multi-UPDATE compatible with concurrent LOCK TABLES WRITE
+    and global read lock.
+  */
   if ((original_multiupdate &&
-       open_tables(thd, &table_list, &table_count, 0)) ||
+       open_tables(thd, &table_list, &table_count,
+                   (thd->stmt_arena->is_stmt_prepare() ?
+                    MYSQL_OPEN_FORCE_SHARED_MDL : 0))) ||
       mysql_handle_derived(lex, &mysql_derived_prepare))
     DBUG_RETURN(TRUE);
   /*
diff --git a/sql/sql_view.cc b/sql/sql_view.cc
index 5fdf7b8d850..54b5eb43ab1 100644
--- a/sql/sql_view.cc
+++ b/sql/sql_view.cc
@@ -22,7 +22,7 @@
 #include "sql_base.h"    // find_table_in_global_list, lock_table_names
 #include "sql_parse.h"                          // sql_parse
 #include "sql_cache.h"                          // query_cache_*
-#include "lock.h"        // wait_if_global_read_lock
+#include "lock.h"        // MYSQL_OPEN_SKIP_TEMPORARY 
 #include "sql_show.h"    // append_identifier
 #include "sql_table.h"                         // build_table_filename
 #include "sql_db.h"            // mysql_opt_change_db, mysql_change_db
@@ -649,13 +649,6 @@ bool mysql_create_view(THD *thd, TABLE_LIST *views,
   }
 #endif
 
-
-  if (thd->global_read_lock.wait_if_global_read_lock(thd, FALSE, TRUE))
-  {
-    res= TRUE;
-    goto err;
-  }
-
   res= mysql_register_view(thd, view, mode);
 
   if (mysql_bin_log.is_open())
@@ -704,7 +697,6 @@ bool mysql_create_view(THD *thd, TABLE_LIST *views,
 
   if (mode != VIEW_CREATE_NEW)
     query_cache_invalidate3(thd, view, 0);
-  thd->global_read_lock.start_waiting_global_read_lock(thd);
   if (res)
     goto err;
 
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index 66d74a398fb..9aa938437b1 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -7385,7 +7385,6 @@ select_lock_type:
             LEX *lex=Lex;
             lex->current_select->set_lock_for_tables(TL_WRITE);
             lex->safe_to_cache_query=0;
-            lex->protect_against_global_read_lock= TRUE;
           }
         | LOCK_SYM IN_SYM SHARE_SYM MODE_SYM
           {
@@ -13182,9 +13181,6 @@ table_lock:
                                             MDL_SHARED_NO_READ_WRITE :
                                             MDL_SHARED_READ)))
               MYSQL_YYABORT;
-            /* If table is to be write locked, protect from a impending GRL. */
-            if (lock_for_write)
-              Lex->protect_against_global_read_lock= TRUE;
           }
         ;
 
diff --git a/sql/table.cc b/sql/table.cc
index f55095d6e82..8cd2e9e9bab 100644
--- a/sql/table.cc
+++ b/sql/table.cc
@@ -5219,7 +5219,8 @@ void init_mdl_requests(TABLE_LIST *table_list)
     table_list->mdl_request.init(MDL_key::TABLE,
                                  table_list->db, table_list->table_name,
                                  table_list->lock_type >= TL_WRITE_ALLOW_WRITE ?
-                                 MDL_SHARED_WRITE : MDL_SHARED_READ);
+                                 MDL_SHARED_WRITE : MDL_SHARED_READ,
+                                 MDL_TRANSACTION);
 }
 
 
diff --git a/sql/table.h b/sql/table.h
index c8e1ad8e658..6aba4f21db2 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -1384,7 +1384,8 @@ struct TABLE_LIST
     lock_type= lock_type_arg;
     mdl_request.init(MDL_key::TABLE, db, table_name,
                      (lock_type >= TL_WRITE_ALLOW_WRITE) ?
-                     MDL_SHARED_WRITE : MDL_SHARED_READ);
+                     MDL_SHARED_WRITE : MDL_SHARED_READ,
+                     MDL_TRANSACTION);
   }
 
   /*
diff --git a/sql/transaction.cc b/sql/transaction.cc
index d3e3ba142b9..b331fea89fe 100644
--- a/sql/transaction.cc
+++ b/sql/transaction.cc
@@ -21,6 +21,7 @@
 #include "sql_priv.h"
 #include "transaction.h"
 #include "rpl_handler.h"
+#include "debug_sync.h"         // DEBUG_SYNC
 
 /* Conditions under which the transaction state must not change. */
 static bool trans_check(THD *thd)
@@ -391,15 +392,15 @@ bool trans_savepoint(THD *thd, LEX_STRING name)
   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
+    Remember locks acquired before the savepoint was set.
+    They are used as a marker to only release locks acquired after
     the setting of this savepoint.
     Note: this works just fine if we're under LOCK TABLES,
     since mdl_savepoint() is guaranteed to be beyond
     the last locked table. This allows to release some
     locks acquired during LOCK TABLES.
   */
-  newsv->mdl_savepoint = thd->mdl_context.mdl_savepoint();
+  newsv->mdl_savepoint= thd->mdl_context.mdl_savepoint();
 
   DBUG_RETURN(FALSE);
 }
@@ -645,17 +646,31 @@ bool trans_xa_commit(THD *thd)
   }
   else if (xa_state == XA_PREPARED && thd->lex->xa_opt == XA_NONE)
   {
-    if (thd->global_read_lock.wait_if_global_read_lock(thd, FALSE, FALSE))
+    MDL_request mdl_request;
+
+    /*
+      Acquire metadata lock which will ensure that COMMIT is blocked
+      by active FLUSH TABLES WITH READ LOCK (and vice versa COMMIT in
+      progress blocks FTWRL).
+
+      We allow FLUSHer to COMMIT; we assume FLUSHer knows what it does.
+    */
+    mdl_request.init(MDL_key::COMMIT, "", "", MDL_INTENTION_EXCLUSIVE,
+                     MDL_TRANSACTION);
+
+    if (thd->mdl_context.acquire_lock(&mdl_request,
+                                      thd->variables.lock_wait_timeout))
     {
       ha_rollback_trans(thd, TRUE);
       my_error(ER_XAER_RMERR, MYF(0));
     }
     else
     {
+      DEBUG_SYNC(thd, "trans_xa_commit_after_acquire_commit_lock");
+
       res= test(ha_commit_one_phase(thd, 1));
       if (res)
         my_error(ER_XAER_RMERR, MYF(0));
-      thd->global_read_lock.start_waiting_global_read_lock(thd);
     }
   }
   else