A post-review fix for type-aware metadata locks.

DDL no longer aborts mysql_lock_tables(), and hence
we no longer need to support need_reopen flag of this
call. 
Remove the flag, and all the code in the server
that was responsible for handling the case when
it was set. This allowed to simplify: 
open_and_lock_tables_derived(), the delayed thread,
multi-update.

Rename MYSQL_LOCK_IGNORE_FLUSH to MYSQL_OPEN_IGNORE_FLUSH,
since we now only support this flag in open_table().

Rename MYSQL_LOCK_PERF_SCHEMA to MYSQL_LOCK_LOG_TABLE,
to avoid confusion.

Move the wait for the global read lock for cases
when we do updates in SELECT f1() or DO (UPDATE) to 
open_table() from mysql_lock_tables(). When waiting
for the read lock, we could raise need_reopen flag,
which is no longer present in mysql_lock_tables().
Since the block responsible for waiting for GRL
was moved, MYSQL_LOCK_IGNORE_GLOBAL_READ_LOCK
was renamed to MYSQL_OPEN_IGNORE_GLOBAL_READ_LOCK.


mysql-test/r/mdl_sync.result:
  Update test results (see comments for mdl_sync.test).
mysql-test/t/mdl_sync.test:
  Update tests: an abort mysql_lock_tables() called for an
  INSERT no longer auto-closes SQL HANDLERS, since it
  no longer leads to back-off and retry.
sql/ha_ndbcluster_binlog.cc:
  Remove unused variables.
sql/lock.cc:
  Remove support for need_reopen parameter of mysql_lock_tables().
  Update comments.
sql/log_event_old.cc:
  Remove the loop responsible for handling need_reopen
  out parameter of mysql_lock_tables().
sql/mysql_priv.h:
  Update open and lock tables flag names.
sql/share/errmsg-utf8.txt:
  Add a new error message to report when
  thr_multi_lock() is aborted.
sql/sql_base.cc:
  Update comments. Rename MYSQL_LOCK_IGNORE_FLUSH
  to MYSQL_OPEN_IGNORE_FLUSH.
sql/sql_class.h:
  Remove unused code.
sql/sql_db.cc:
  Remove an unused bit of code.
sql/sql_handler.cc:
  For backward compatibility, we still want to back off and
  retry when a call to mysql_lock_tables() is aborted
  from within an SQL HANDLER. Write an internal error
  handler to support the case.
sql/sql_insert.cc:
  Call mysql_lock_tables() no longer has need_reopen
  out parameter. Simplify the code by removing 
  the crud that took care of it.
  MYSQL_LOCK_IGNORE_FLUSH is now only supported by
  open_tables().
sql/sql_show.cc:
  Rename MYSQL_LOCK_IGNORE_FLUSH to MYSQL_OPEN_IGNORE_FLUSH
sql/sql_table.cc:
  Remove an unused parameter.
sql/sql_update.cc:
  Remove the need_reopen loop from multi-update.
  We no also longer need to cleanup the parse tree in case
  when mysql_lock_tables() is aborted and thus an infinite
  source of multi-update bugs is gone.
sql/tztime.cc:
  Rename MYSQL_LOCK_IGNORE_FLUSH to MYSQL_OPEN_IGNORE_FLUSH,
  since from now on this flag is only supported by open_table().

16 files changed, 254 insertions, 410 deletions

diff --git a/mysql-test/r/mdl_sync.result b/mysql-test/r/mdl_sync.result
index ff6daf6443e..984f0df3d0e 100644
--- a/mysql-test/r/mdl_sync.result
+++ b/mysql-test/r/mdl_sync.result
@@ -1678,14 +1678,21 @@ insert into t2 values (1);;
 # 
 # Switching to connection 'handler_con1'.
 # Wait until INSERT is blocked on table-level lock.
-# The below statement should not cause deadlock.
+# Sending 'alter table t1 drop column j'. It should not cause
+# deadlock.
 alter table t1 drop column j;
-unlock tables;
+# Switching to connection 'handler_con2'.
+# Wait until ALTER is blocked during upgrade.
 # 
 # Switching to connection 'default'.
 # Reap INSERT.
+ERROR HY000: Wait on a lock was aborted due to a pending exclusive lock
 handler t1 close;
 #
+# Switching to connection 'handler_con1'.
+# Reaping 'alter table t1 drop column j'
+unlock tables;
+# Switching to connection 'default'.
 # Then, check the scenario in which upgrade of SNRW lock to X
 # lock is blocked by HANDLER which is open in connection currently
 # waiting to get SW lock on the same table.
@@ -2248,6 +2255,8 @@ SET DEBUG_SYNC= 'RESET';
 # Bug#50786 Assertion `thd->mdl_context.trans_sentinel() == __null' 
 #           failed in open_ltable()
 #
+# Supress warnings written to the log file
+call mtr.add_suppression("Wait on a lock was aborted due to a pending exclusive lock");
 DROP TABLE IF EXISTS t1, t2;
 CREATE TABLE t1 (i INT);
 CREATE TABLE t2 (i INT);
@@ -2271,7 +2280,6 @@ SET DEBUG_SYNC= 'now WAIT_FOR parked';
 # Sending:
 SELECT 1;
 # connection: con3
-# Sending:
 ALTER TABLE t1 ADD COLUMN j INT;
 # connection: default
 SET DEBUG_SYNC= 'now SIGNAL go';
@@ -2284,8 +2292,6 @@ HANDLER t1 CLOSE;
 # Reaping SELECT 1
 1
 1
-# connection: con3
-# Reaping ALTER TABLE t1 ADD COLUMN j INT
 # connection: default
 DROP TABLE t1, t2;
 SET DEBUG_SYNC= 'RESET';
diff --git a/mysql-test/t/mdl_sync.test b/mysql-test/t/mdl_sync.test
index b9a9241ce32..ef434e33cfa 100644
--- a/mysql-test/t/mdl_sync.test
+++ b/mysql-test/t/mdl_sync.test
@@ -2277,17 +2277,32 @@ let $wait_condition=
   select count(*) = 1 from information_schema.processlist
   where state = "Table lock" and info = "insert into t2 values (1)";
 --source include/wait_condition.inc
---echo # The below statement should not cause deadlock.
-alter table t1 drop column j;
-unlock tables;
+--echo # Sending 'alter table t1 drop column j'. It should not cause
+--echo # deadlock.
+send alter table t1 drop column j;
+--echo # Switching to connection 'handler_con2'.
+connection handler_con2;
+--echo # Wait until ALTER is blocked during upgrade.
+let $wait_condition=
+  select count(*) = 1 from information_schema.processlist
+  where state = "Waiting for table" and info = "alter table t1 drop column j";
+--source include/wait_condition.inc
 --echo # 
 --echo # Switching to connection 'default'.
 connection default;
 --echo # Reap INSERT.
+--error ER_LOCK_ABORTED
 --reap
 handler t1 close;
-
 --echo #
+--echo # Switching to connection 'handler_con1'.
+connection handler_con1;
+--echo # Reaping 'alter table t1 drop column j'
+--reap
+unlock tables;
+--echo # Switching to connection 'default'.
+connection default;
+
 --echo # Then, check the scenario in which upgrade of SNRW lock to X
 --echo # lock is blocked by HANDLER which is open in connection currently
 --echo # waiting to get SW lock on the same table.
@@ -3220,6 +3235,8 @@ disconnect con2;
 --echo #           failed in open_ltable()
 --echo #
 
+--echo # Supress warnings written to the log file
+call mtr.add_suppression("Wait on a lock was aborted due to a pending exclusive lock");
 --disable_warnings
 DROP TABLE IF EXISTS t1, t2;
 --enable_warnings
@@ -3279,16 +3296,10 @@ let $wait_condition=
 # since the latter waits on a table-level lock while having a HANDLER
 # open. This will cause mysql_lock_tables() in con1 fail which before
 # triggered the assert.
---echo # Sending:
---send ALTER TABLE t1 ADD COLUMN j INT
+ALTER TABLE t1 ADD COLUMN j INT;
 
 --echo # connection: default
 connection default;
-let $wait_condition=
-  SELECT COUNT(*) = 1 FROM information_schema.processlist
-  WHERE state = "Waiting for table" 
-  AND info = "ALTER TABLE t1 ADD COLUMN j INT";
---source include/wait_condition.inc
 SET DEBUG_SYNC= 'now SIGNAL go';
 
 --echo # connection: con1
@@ -3302,11 +3313,6 @@ connection con2;
 --echo # Reaping SELECT 1
 --reap
 
---echo # connection: con3
-connection con3;
---echo # Reaping ALTER TABLE t1 ADD COLUMN j INT
---reap
-
 --echo # connection: default
 connection default;
 DROP TABLE t1, t2;
diff --git a/sql/ha_ndbcluster_binlog.cc b/sql/ha_ndbcluster_binlog.cc
index 279d5b4c242..7097c0a1a46 100644
--- a/sql/ha_ndbcluster_binlog.cc
+++ b/sql/ha_ndbcluster_binlog.cc
@@ -2346,7 +2346,6 @@ static int open_ndb_binlog_index(THD *thd, TABLE **ndb_binlog_index)
   thd->proc_info= "Opening " NDB_REP_DB "." NDB_REP_TABLE;
 
   tables->required_type= FRMTYPE_TABLE;
-  uint counter;
   thd->clear_error();
   if (open_and_lock_tables(thd, tables, FALSE, 0))
   {
@@ -2374,7 +2373,6 @@ int ndb_add_ndb_binlog_index(THD *thd, void *_row)
 {
   ndb_binlog_index_row &row= *(ndb_binlog_index_row *) _row;
   int error= 0;
-  bool need_reopen;
   /*
     Turn of binlogging to prevent the table changes to be written to
     the binary log.
diff --git a/sql/lock.cc b/sql/lock.cc
index 78a16cea18a..7937878073e 100644
--- a/sql/lock.cc
+++ b/sql/lock.cc
@@ -97,7 +97,7 @@ static void print_lock_error(int error, const char *);
 
 /* Map the return value of thr_lock to an error from errmsg.txt */
 static int thr_lock_errno_to_mysql[]=
-{ 0, 1, ER_LOCK_WAIT_TIMEOUT, ER_LOCK_DEADLOCK };
+{ 0, ER_LOCK_ABORTED, ER_LOCK_WAIT_TIMEOUT, ER_LOCK_DEADLOCK };
 
 /**
   Perform semantic checks for mysql_lock_tables.
@@ -108,8 +108,7 @@ static int thr_lock_errno_to_mysql[]=
   @return 0 if all the check passed, non zero if a check failed.
 */
 static int
-lock_tables_check(THD *thd, TABLE **tables, uint count,
-                  bool *write_lock_used, uint flags)
+lock_tables_check(THD *thd, TABLE **tables, uint count, uint flags)
 {
   uint system_count, i;
   bool is_superuser, log_table_write_query;
@@ -117,10 +116,9 @@ lock_tables_check(THD *thd, TABLE **tables, uint count,
   DBUG_ENTER("lock_tables_check");
 
   system_count= 0;
-  *write_lock_used= FALSE;
   is_superuser= thd->security_ctx->master_access & SUPER_ACL;
   log_table_write_query= (is_log_table_write_query(thd->lex->sql_command)
-                         || ((flags & MYSQL_LOCK_PERF_SCHEMA) != 0));
+                         || ((flags & MYSQL_LOCK_LOG_TABLE) != 0));
 
   for (i=0 ; i<count; i++)
   {
@@ -153,8 +151,6 @@ lock_tables_check(THD *thd, TABLE **tables, uint count,
 
     if (t->reginfo.lock_type >= TL_WRITE_ALLOW_WRITE)
     {
-      *write_lock_used= TRUE;
-
       if (t->s->table_category == TABLE_CATEGORY_SYSTEM)
         system_count++;
 
@@ -273,141 +269,69 @@ static void reset_lock_data_and_free(MYSQL_LOCK **mysql_lock)
    @param tables       An array of pointers to the tables to lock.
    @param count        The number of tables to lock.
    @param flags        Options:
-                 MYSQL_LOCK_IGNORE_GLOBAL_READ_LOCK Ignore a global read lock
                  MYSQL_LOCK_IGNORE_GLOBAL_READ_ONLY Ignore SET GLOBAL READ_ONLY
-                 MYSQL_LOCK_IGNORE_FLUSH            Ignore a flush tables.
                  MYSQL_LOCK_IGNORE_TIMEOUT          Use maximum timeout value.
-   @param need_reopen  Out parameter, TRUE if some tables were altered
-                       or deleted and should be reopened by caller.
-
-   @note Caller of this function should always be ready to handle request to
-         reopen table unless there are external invariants which guarantee
-         that such thing won't be needed (for example we are obtaining lock
-         on table on which we already have exclusive metadata lock).
 
    @retval  A lock structure pointer on success.
-   @retval  NULL on error or if some tables should be reopen.
+   @retval  NULL if an error or if wait on a lock was killed.
 */
 
-MYSQL_LOCK *mysql_lock_tables(THD *thd, TABLE **tables, uint count,
-                              uint flags, bool *need_reopen)
+MYSQL_LOCK *mysql_lock_tables(THD *thd, TABLE **tables, uint count, uint flags)
 {
   int rc;
   MYSQL_LOCK *sql_lock;
-  bool write_lock_used;
-
-  DBUG_ENTER("mysql_lock_tables");
-
-  *need_reopen= FALSE;
-
-  if (lock_tables_check(thd, tables, count, &write_lock_used, flags))
-    DBUG_RETURN (NULL);
-
   ulong timeout= (flags & MYSQL_LOCK_IGNORE_TIMEOUT) ?
     LONG_TIMEOUT : thd->variables.lock_wait_timeout;
 
-  for (;;)
-  {
-    if (! (sql_lock= get_lock_data(thd, tables, count, GET_LOCK_STORE_LOCKS)))
-      break;
+  DBUG_ENTER("mysql_lock_tables");
 
-    if (global_read_lock && write_lock_used &&
-        ! (flags & MYSQL_LOCK_IGNORE_GLOBAL_READ_LOCK))
-    {
-      /*
-	Someone has issued LOCK ALL TABLES FOR READ 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))
-      {
-        /* Clear the lock type of all lock data to avoid reusage. */
-        reset_lock_data_and_free(&sql_lock);
-	break;
-      }
-      if (thd->version != refresh_version)
-      {
-        /* Clear the lock type of all lock data to avoid reusage. */
-        reset_lock_data_and_free(&sql_lock);
-	goto retry;
-      }
-    }
+  if (lock_tables_check(thd, tables, count, flags))
+    DBUG_RETURN(NULL);
 
-    thd_proc_info(thd, "System lock");
-    DBUG_PRINT("info", ("thd->proc_info %s", thd->proc_info));
-    if (sql_lock->table_count && lock_external(thd, sql_lock->table,
-                                               sql_lock->table_count))
-    {
-      /* Clear the lock type of all lock data to avoid reusage. */
-      reset_lock_data_and_free(&sql_lock);
-      break;
-    }
-    DBUG_PRINT("info", ("thd->proc_info %s", thd->proc_info));
-    /* Copy the lock data array. thr_multi_lock() reorders its contens. */
-    memcpy(sql_lock->locks + sql_lock->lock_count, sql_lock->locks,
-           sql_lock->lock_count * sizeof(*sql_lock->locks));
-    /* Lock on the copied half of the lock data array. */
-    rc= thr_lock_errno_to_mysql[(int) thr_multi_lock(sql_lock->locks +
-                                                     sql_lock->lock_count,
-                                                     sql_lock->lock_count,
-                                                     thd->lock_id, timeout)];
-    if (rc > 1)                                 /* a timeout or a deadlock */
-    {
-      if (sql_lock->table_count)
-        (void) unlock_external(thd, sql_lock->table, sql_lock->table_count);
-      reset_lock_data_and_free(&sql_lock);
-      my_error(rc, MYF(0));
-      break;
-    }
-    else if (rc == 1)                           /* aborted or killed */
-    {
-      /*
-        reset_lock_data is required here. If thr_multi_lock fails it
-        resets lock type for tables, which were locked before (and
-        including) one that caused error. Lock type for other tables
-        preserved.
-      */
-      reset_lock_data(sql_lock);
-      sql_lock->lock_count= 0;                  // Locks are already freed
-      // Fall through: unlock, reset lock data, free and retry
-    }
-    else
-    {
-      /* Success */
-      break;
-    }
-    thd_proc_info(thd, 0);
+  if (! (sql_lock= get_lock_data(thd, tables, count, GET_LOCK_STORE_LOCKS)))
+    DBUG_RETURN(NULL);
 
-    /* going to retry, unlock all tables */
-    if (sql_lock->lock_count)
-        thr_multi_unlock(sql_lock->locks, sql_lock->lock_count);
+  thd_proc_info(thd, "System lock");
+  DBUG_PRINT("info", ("thd->proc_info %s", thd->proc_info));
+  if (sql_lock->table_count && lock_external(thd, sql_lock->table,
+                                             sql_lock->table_count))
+  {
+    /* Clear the lock type of all lock data to avoid reusage. */
+    reset_lock_data_and_free(&sql_lock);
+    goto end;
+  }
 
+  /* Copy the lock data array. thr_multi_lock() reorders its contents. */
+  memcpy(sql_lock->locks + sql_lock->lock_count, sql_lock->locks,
+         sql_lock->lock_count * sizeof(*sql_lock->locks));
+  /* Lock on the copied half of the lock data array. */
+  rc= thr_lock_errno_to_mysql[(int) thr_multi_lock(sql_lock->locks +
+                                                   sql_lock->lock_count,
+                                                   sql_lock->lock_count,
+                                                   thd->lock_id, timeout)];
+  if (rc)
+  {
     if (sql_lock->table_count)
       (void) unlock_external(thd, sql_lock->table, sql_lock->table_count);
-
-    /*
-      If thr_multi_lock fails it resets lock type for tables, which
-      were locked before (and including) one that caused error. Lock
-      type for other tables preserved.
-    */
     reset_lock_data_and_free(&sql_lock);
-retry:
-    /* Let upper level close all used tables and retry or give error. */
-    *need_reopen= TRUE;
-    break;
+    if (! thd->killed)
+      my_error(rc, MYF(0));
   }
+end:
   thd_proc_info(thd, 0);
+
   if (thd->killed)
   {
     thd->send_kill_message();
     if (sql_lock)
     {
-      mysql_unlock_tables(thd,sql_lock);
-      sql_lock=0;
+      mysql_unlock_tables(thd, sql_lock);
+      sql_lock= 0;
     }
   }
 
   thd->set_time_after_lock();
-  DBUG_RETURN (sql_lock);
+  DBUG_RETURN(sql_lock);
 }
 
 
diff --git a/sql/log_event_old.cc b/sql/log_event_old.cc
index 9b46ad83b14..e87c3688cac 100644
--- a/sql/log_event_old.cc
+++ b/sql/log_event_old.cc
@@ -1455,8 +1455,6 @@ int Old_rows_log_event::do_apply_event(Relay_log_info const *rli)
   */
   if (!thd->lock)
   {
-    bool need_reopen= 1; /* To execute the first lap of the loop below */
-
     /*
       lock_tables() reads the contents of thd->lex, so they must be
       initialized. Contrary to in
@@ -1465,80 +1463,31 @@ int Old_rows_log_event::do_apply_event(Relay_log_info const *rli)
     */
     lex_start(thd);
 
-    while ((error= lock_tables(thd, rli->tables_to_lock,
-                               rli->tables_to_lock_count, 0,
-                               &need_reopen)))
+    if ((error= lock_tables(thd, rli->tables_to_lock,
+                               rli->tables_to_lock_count, 0)))
     {
-      if (!need_reopen)
-      {
-        if (thd->is_slave_error || thd->is_fatal_error)
-        {
-          /*
-            Error reporting borrowed from Query_log_event with many excessive
-            simplifications (we don't honour --slave-skip-errors)
-          */
-          uint actual_error= thd->net.last_errno;
-          rli->report(ERROR_LEVEL, actual_error,
-                      "Error '%s' in %s event: when locking tables",
-                      (actual_error ? thd->net.last_error :
-                       "unexpected success or fatal error"),
-                      get_type_str());
-          thd->is_fatal_error= 1;
-        }
-        else
-        {
-          rli->report(ERROR_LEVEL, error,
-                      "Error in %s event: when locking tables",
-                      get_type_str());
-        }
-        const_cast<Relay_log_info*>(rli)->slave_close_thread_tables(thd);
-        DBUG_RETURN(error);
-      }
-
-      /*
-        So we need to reopen the tables.
-
-        We need to flush the pending RBR event, since it keeps a
-        pointer to an open table.
-
-        ALTERNATIVE SOLUTION (not implemented): Extract a pointer to
-        the pending RBR event and reset the table pointer after the
-        tables has been reopened.
-
-        NOTE: For this new scheme there should be no pending event:
-        need to add code to assert that is the case.
-       */
-      error= thd->binlog_flush_pending_rows_event(FALSE);
-      if (error)
+      if (thd->is_slave_error || thd->is_fatal_error)
       {
-        rli->report(ERROR_LEVEL, ER_SLAVE_FATAL_ERROR,
-                    ER(ER_SLAVE_FATAL_ERROR),
-                    "call to binlog_flush_pending_rows_event() failed");
-        thd->is_slave_error= 1;
-        DBUG_RETURN(error);
+        /*
+          Error reporting borrowed from Query_log_event with many excessive
+          simplifications (we don't honour --slave-skip-errors)
+        */
+        uint actual_error= thd->net.last_errno;
+        rli->report(ERROR_LEVEL, actual_error,
+                    "Error '%s' in %s event: when locking tables",
+                    (actual_error ? thd->net.last_error :
+                     "unexpected success or fatal error"),
+                    get_type_str());
+        thd->is_fatal_error= 1;
       }
-      TABLE_LIST *tables= rli->tables_to_lock;
-      close_tables_for_reopen(thd, &tables, NULL);
-
-      uint tables_count= rli->tables_to_lock_count;
-      if ((error= open_tables(thd, &tables, &tables_count, 0)))
+      else
       {
-        if (thd->is_slave_error || thd->is_fatal_error)
-        {
-          /*
-            Error reporting borrowed from Query_log_event with many excessive
-            simplifications (we don't honour --slave-skip-errors)
-          */
-          uint actual_error= thd->net.last_errno;
-          rli->report(ERROR_LEVEL, actual_error,
-                      "Error '%s' on reopening tables",
-                      (actual_error ? thd->net.last_error :
-                       "unexpected success or fatal error"));
-          thd->is_slave_error= 1;
-        }
-        const_cast<Relay_log_info*>(rli)->slave_close_thread_tables(thd);
-        DBUG_RETURN(error);
+        rli->report(ERROR_LEVEL, error,
+                    "Error in %s event: when locking tables",
+                    get_type_str());
       }
+      const_cast<Relay_log_info*>(rli)->slave_close_thread_tables(thd);
+      DBUG_RETURN(error);
     }
 
     /*
diff --git a/sql/mysql_priv.h b/sql/mysql_priv.h
index 4e672f1d6c7..b398baa064e 100644
--- a/sql/mysql_priv.h
+++ b/sql/mysql_priv.h
@@ -1590,8 +1590,7 @@ inline bool open_and_lock_tables(THD *thd, TABLE_LIST *tables,
 TABLE *open_n_lock_single_table(THD *thd, TABLE_LIST *table_l,
                                 thr_lock_type lock_type, uint flags);
 bool open_normal_and_derived_tables(THD *thd, TABLE_LIST *tables, uint flags);
-bool lock_tables(THD *thd, TABLE_LIST *tables, uint counter, uint flags,
-                bool *need_reopen);
+bool lock_tables(THD *thd, TABLE_LIST *tables, uint counter, uint flags);
 TABLE *open_temporary_table(THD *thd, const char *path, const char *db,
 			    const char *table_name, bool link_in_list);
 bool rm_temporary_table(handlerton *base, char *path);
@@ -2145,14 +2144,13 @@ extern char *opt_ssl_ca, *opt_ssl_capath, *opt_ssl_cert, *opt_ssl_cipher,
 extern struct st_VioSSLFd * ssl_acceptor_fd;
 #endif /* HAVE_OPENSSL */
 
-MYSQL_LOCK *mysql_lock_tables(THD *thd, TABLE **table, uint count,
-                              uint flags, bool *need_reopen);
+MYSQL_LOCK *mysql_lock_tables(THD *thd, TABLE **table, uint count, uint flags);
 /* mysql_lock_tables() and open_table() flags bits */
-#define MYSQL_LOCK_IGNORE_GLOBAL_READ_LOCK      0x0001
-#define MYSQL_LOCK_IGNORE_FLUSH                 0x0002
+#define MYSQL_OPEN_IGNORE_GLOBAL_READ_LOCK      0x0001
+#define MYSQL_OPEN_IGNORE_FLUSH                 0x0002
 #define MYSQL_OPEN_TEMPORARY_ONLY               0x0004
 #define MYSQL_LOCK_IGNORE_GLOBAL_READ_ONLY      0x0008
-#define MYSQL_LOCK_PERF_SCHEMA                  0x0010
+#define MYSQL_LOCK_LOG_TABLE                    0x0010
 #define MYSQL_OPEN_TAKE_UPGRADABLE_MDL          0x0020
 /**
   Do not try to acquire a metadata lock on the table: we
@@ -2182,8 +2180,8 @@ MYSQL_LOCK *mysql_lock_tables(THD *thd, TABLE **table, uint count,
 #define MYSQL_LOCK_IGNORE_TIMEOUT               0x1000
 
 /** Please refer to the internals manual. */
-#define MYSQL_OPEN_REOPEN  (MYSQL_LOCK_IGNORE_FLUSH |\
-                            MYSQL_LOCK_IGNORE_GLOBAL_READ_LOCK |\
+#define MYSQL_OPEN_REOPEN  (MYSQL_OPEN_IGNORE_FLUSH |\
+                            MYSQL_OPEN_IGNORE_GLOBAL_READ_LOCK |\
                             MYSQL_LOCK_IGNORE_GLOBAL_READ_ONLY |\
                             MYSQL_LOCK_IGNORE_TIMEOUT |\
                             MYSQL_OPEN_GET_NEW_TABLE |\
diff --git a/sql/share/errmsg-utf8.txt b/sql/share/errmsg-utf8.txt
index 9861c03137f..4b680c1788d 100644
--- a/sql/share/errmsg-utf8.txt
+++ b/sql/share/errmsg-utf8.txt
@@ -6321,3 +6321,6 @@ ER_SPATIAL_MUST_HAVE_GEOM_COL 42000
 
 ER_TOO_LONG_INDEX_COMMENT
   eng "Comment for index '%-.64s' is too long (max = %lu)"
+
+ER_LOCK_ABORTED
+  eng "Wait on a lock was aborted due to a pending exclusive lock"
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index b4c9aa576d0..a85a863abd9 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -2421,7 +2421,7 @@ open_table_get_mdl_lock(THD *thd, TABLE_LIST *table_list,
                         required to remedy problem appeared during attempt
                         to open table.
     flags               Bitmap of flags to modify how open works:
-                          MYSQL_LOCK_IGNORE_FLUSH - Open table even if
+                          MYSQL_OPEN_IGNORE_FLUSH - Open table even if
                           someone has done a flush or there is a pending
                           exclusive metadata lock requests against it
                           (i.e. request high priority metadata lock).
@@ -2480,6 +2480,31 @@ 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->version != refresh_version)
+    {
+      (void) ot_ctx->request_backoff_action(Open_table_context::OT_WAIT_TDC);
+      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
@@ -2680,7 +2705,7 @@ bool open_table(THD *thd, TABLE_LIST *table_list, MEM_ROOT *mem_root,
   if (!thd->open_tables)
     thd->version=refresh_version;
   else if ((thd->version != refresh_version) &&
-           ! (flags & MYSQL_LOCK_IGNORE_FLUSH))
+           ! (flags & MYSQL_OPEN_IGNORE_FLUSH))
   {
     /* Someone did a refresh while thread was opening tables */
     mysql_mutex_unlock(&LOCK_open);
@@ -2811,7 +2836,7 @@ bool open_table(THD *thd, TABLE_LIST *table_list, MEM_ROOT *mem_root,
 
   if (share->version != refresh_version)
   {
-    if (!(flags & MYSQL_LOCK_IGNORE_FLUSH))
+    if (!(flags & MYSQL_OPEN_IGNORE_FLUSH))
     {
        /*
          We already have an MDL lock. But we have encountered an old
@@ -3293,7 +3318,6 @@ bool
 Locked_tables_list::reopen_tables(THD *thd)
 {
   Open_table_context ot_ctx_unused(thd, LONG_TIMEOUT);
-  bool lt_refresh_unused;
   size_t reopen_count= 0;
   MYSQL_LOCK *lock;
   MYSQL_LOCK *merged_lock;
@@ -3333,7 +3357,7 @@ Locked_tables_list::reopen_tables(THD *thd)
       break something else.
     */
     lock= mysql_lock_tables(thd, m_reopen_array, reopen_count,
-                            MYSQL_OPEN_REOPEN, &lt_refresh_unused);
+                            MYSQL_OPEN_REOPEN);
     thd->in_lock_tables= 0;
     if (lock == NULL || (merged_lock=
                          mysql_lock_merge(thd->lock, lock)) == NULL)
@@ -5061,7 +5085,6 @@ TABLE *open_ltable(THD *thd, TABLE_LIST *table_list, thr_lock_type lock_type,
   TABLE *table;
   Open_table_context ot_ctx(thd, (lock_flags & MYSQL_LOCK_IGNORE_TIMEOUT) ?
                             LONG_TIMEOUT : thd->variables.lock_wait_timeout);
-  bool refresh;
   bool error;
   DBUG_ENTER("open_ltable");
 
@@ -5073,8 +5096,7 @@ TABLE *open_ltable(THD *thd, TABLE_LIST *table_list, thr_lock_type lock_type,
   /* open_ltable can be used only for BASIC TABLEs */
   table_list->required_type= FRMTYPE_TABLE;
 
-retry:
-  while ((error= open_table(thd, table_list, thd->mem_root, &ot_ctx, 0)) &&
+  while ((error= open_table(thd, table_list, thd->mem_root, &ot_ctx, lock_flags)) &&
          ot_ctx.can_recover_from_failed_open())
   {
     /*
@@ -5120,18 +5142,9 @@ retry:
       DBUG_ASSERT(thd->lock == 0);	// You must lock everything at once
       if ((table->reginfo.lock_type= lock_type) != TL_UNLOCK)
 	if (! (thd->lock= mysql_lock_tables(thd, &table_list->table, 1,
-                                            lock_flags, &refresh)))
+                                            lock_flags)))
         {
-          if (refresh)
-          {
-            close_thread_tables(thd);
-            table_list->table= NULL;
-            table_list->mdl_request.ticket= NULL;
-            thd->mdl_context.rollback_to_savepoint(ot_ctx.start_of_statement_svp());
-            goto retry;
-          }
-          else
-            table= 0;
+          table= 0;
         }
     }
   }
@@ -5168,42 +5181,27 @@ bool open_and_lock_tables(THD *thd, TABLE_LIST *tables,
                           Prelocking_strategy *prelocking_strategy)
 {
   uint counter;
-  bool need_reopen;
-  /*
-    Remember the set of metadata locks which this connection
-    managed to acquire before the start of the current statement.
-    It can be either transaction-scope locks, or HANDLER locks,
-    or LOCK TABLES locks. If mysql_lock_tables() fails with
-    need_reopen request, we'll use it to instruct
-    close_tables_for_reopen() to release all locks of this
-    statement.
-  */
-  MDL_ticket *start_of_statement_svp= thd->mdl_context.mdl_savepoint();
   DBUG_ENTER("open_and_lock_tables");
   DBUG_PRINT("enter", ("derived handling: %d", derived));
 
-  for ( ; ; ) 
-  {
-    if (open_tables(thd, &tables, &counter, flags, prelocking_strategy))
-      DBUG_RETURN(TRUE);
-    DBUG_EXECUTE_IF("sleep_open_and_lock_after_open", {
-      const char *old_proc_info= thd->proc_info;
-      thd->proc_info= "DBUG sleep";
-      my_sleep(6000000);
-      thd->proc_info= old_proc_info;});
-
-    if (!lock_tables(thd, tables, counter, flags,
-                     &need_reopen))
-      break;
-    if (!need_reopen)
-      DBUG_RETURN(TRUE);
-    close_tables_for_reopen(thd, &tables, start_of_statement_svp);
-  }
+  if (open_tables(thd, &tables, &counter, flags, prelocking_strategy))
+    DBUG_RETURN(TRUE);
+
+  DBUG_EXECUTE_IF("sleep_open_and_lock_after_open", {
+                  const char *old_proc_info= thd->proc_info;
+                  thd->proc_info= "DBUG sleep";
+                  my_sleep(6000000);
+                  thd->proc_info= old_proc_info;});
+
+  if (lock_tables(thd, tables, counter, flags))
+    DBUG_RETURN(TRUE);
+
   if (derived &&
       (mysql_handle_derived(thd->lex, &mysql_derived_prepare) ||
        (thd->fill_derived_tables() &&
         mysql_handle_derived(thd->lex, &mysql_derived_filling))))
     DBUG_RETURN(TRUE); /* purecov: inspected */
+
   DBUG_RETURN(FALSE);
 }
 
@@ -5216,7 +5214,7 @@ bool open_and_lock_tables(THD *thd, TABLE_LIST *tables,
     thd		- thread handler
     tables	- list of tables for open
     flags       - bitmap of flags to modify how the tables will be open:
-                  MYSQL_LOCK_IGNORE_FLUSH - open table even if someone has
+                  MYSQL_OPEN_IGNORE_FLUSH - open table even if someone has
                   done a flush or namelock on it.
 
   RETURN
@@ -5261,37 +5259,28 @@ static void mark_real_tables_as_free_for_reuse(TABLE_LIST *table)
 }
 
 
-/*
-  Lock all tables in list
+/**
+  Lock all tables in a list.
 
-  SYNOPSIS
-    lock_tables()
-    thd			Thread handler
-    tables		Tables to lock
-    count		Number of opened tables
-    flags               Options (see mysql_lock_tables() for details)
-    need_reopen         Out parameter which if TRUE indicates that some
-                        tables were dropped or altered during this call
-                        and therefore invoker should reopen tables and
-                        try to lock them once again (in this case
-                        lock_tables() will also return error).
+  @param  thd           Thread handler
+  @param  tables        Tables to lock
+  @param  count         Number of opened tables
+  @param  flags         Options (see mysql_lock_tables() for details)
 
-  NOTES
-    You can't call lock_tables twice, as this would break the dead-lock-free
-    handling thr_lock gives us.  You most always get all needed locks at
-    once.
+  You can't call lock_tables() while holding thr_lock locks, as
+  this would break the dead-lock-free handling thr_lock gives us.
+  You must always get all needed locks at once.
 
-    If query for which we are calling this function marked as requiring
-    prelocking, this function will change locked_tables_mode to
-    LTM_PRELOCKED.
+  If the query for which we are calling this function is marked as
+  requiring prelocking, this function will change
+  locked_tables_mode to LTM_PRELOCKED.
 
-  RETURN VALUES
-   0	ok
-   -1	Error
+  @retval FALSE         Success. 
+  @retval TRUE          A lock wait timeout, deadlock or out of memory.
 */
 
 bool lock_tables(THD *thd, TABLE_LIST *tables, uint count,
-                 uint flags, bool *need_reopen)
+                 uint flags)
 {
   TABLE_LIST *table;
 
@@ -5302,7 +5291,6 @@ bool lock_tables(THD *thd, TABLE_LIST *tables, uint count,
   */
   DBUG_ASSERT(thd->locked_tables_mode <= LTM_LOCK_TABLES ||
               !thd->lex->requires_prelocking());
-  *need_reopen= FALSE;
 
   if (!tables && !thd->lex->requires_prelocking())
     DBUG_RETURN(thd->decide_logging_format(tables));
@@ -5347,7 +5335,7 @@ bool lock_tables(THD *thd, TABLE_LIST *tables, uint count,
     DEBUG_SYNC(thd, "before_lock_tables_takes_lock");
 
     if (! (thd->lock= mysql_lock_tables(thd, start, (uint) (ptr - start),
-                                        flags, need_reopen)))
+                                        flags)))
       DBUG_RETURN(TRUE);
 
     DEBUG_SYNC(thd, "after_lock_tables_takes_lock");
@@ -8869,7 +8857,7 @@ open_system_tables_for_read(THD *thd, TABLE_LIST *table_list,
   thd->reset_n_backup_open_tables_state(backup);
 
   if (open_and_lock_tables(thd, table_list, FALSE,
-                           MYSQL_LOCK_IGNORE_FLUSH |
+                           MYSQL_OPEN_IGNORE_FLUSH |
                            MYSQL_LOCK_IGNORE_TIMEOUT))
   {
     lex->restore_backup_query_tables_list(&query_tables_list_backup);
@@ -8957,11 +8945,11 @@ open_system_table_for_update(THD *thd, TABLE_LIST *one_table)
 TABLE *
 open_log_table(THD *thd, TABLE_LIST *one_table, Open_tables_backup *backup)
 {
-  uint flags= ( MYSQL_LOCK_IGNORE_GLOBAL_READ_LOCK |
+  uint flags= ( MYSQL_OPEN_IGNORE_GLOBAL_READ_LOCK |
                 MYSQL_LOCK_IGNORE_GLOBAL_READ_ONLY |
-                MYSQL_LOCK_IGNORE_FLUSH |
+                MYSQL_OPEN_IGNORE_FLUSH |
                 MYSQL_LOCK_IGNORE_TIMEOUT |
-                MYSQL_LOCK_PERF_SCHEMA);
+                MYSQL_LOCK_LOG_TABLE);
   TABLE *table;
   /* Save value that is changed in mysql_lock_tables() */
   ulonglong save_utime_after_lock= thd->utime_after_lock;
@@ -8989,8 +8977,7 @@ open_log_table(THD *thd, TABLE_LIST *one_table, Open_tables_backup *backup)
       open tables cannot be accepted when restoring the open tables
       state.
     */
-    if (thd->killed)
-      close_thread_tables(thd);
+    close_thread_tables(thd);
     thd->restore_backup_open_tables_state(backup);
   }
 
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 8ce3cee3c36..f8e8942e244 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -1161,9 +1161,7 @@ public:
 class Drop_table_error_handler : public Internal_error_handler
 {
 public:
-  Drop_table_error_handler(Internal_error_handler *err_handler)
-    :m_err_handler(err_handler)
-  { }
+  Drop_table_error_handler() {}
 
 public:
   bool handle_condition(THD *thd,
@@ -1174,7 +1172,6 @@ public:
                         MYSQL_ERROR ** cond_hdl);
 
 private:
-  Internal_error_handler *m_err_handler;
 };
 
 
diff --git a/sql/sql_db.cc b/sql/sql_db.cc
index 75855020127..e700dd93a55 100644
--- a/sql/sql_db.cc
+++ b/sql/sql_db.cc
@@ -934,7 +934,7 @@ bool mysql_rm_db(THD *thd,char *db,bool if_exists, bool silent)
   }
   else
   {
-    Drop_table_error_handler err_handler(thd->get_internal_handler());
+    Drop_table_error_handler err_handler;
     thd->push_internal_handler(&err_handler);
 
     error= -1;
diff --git a/sql/sql_handler.cc b/sql/sql_handler.cc
index 3afeb4164bd..dfa06495e9d 100644
--- a/sql/sql_handler.cc
+++ b/sql/sql_handler.cc
@@ -405,6 +405,56 @@ bool mysql_ha_close(THD *thd, TABLE_LIST *tables)
 }
 
 
+/**
+  A helper class to process an error from mysql_lock_tables().
+  HANDLER READ statement's attempt to lock the subject table
+  may get aborted if there is a pending DDL. In that case
+  we close the table, reopen it, and try to read again.
+  This is implicit and obscure, since HANDLER position
+  is lost in the process, but it's the legacy server
+  behaviour we should preserve.
+*/
+
+class Sql_handler_lock_error_handler: public Internal_error_handler
+{
+public:
+  virtual
+  bool handle_condition(THD *thd,
+                        uint sql_errno,
+                        const char *sqlstate,
+                        MYSQL_ERROR::enum_warning_level level,
+                        const char* msg,
+                        MYSQL_ERROR **cond_hdl);
+
+  bool need_reopen() const { return m_need_reopen; };
+  void init() { m_need_reopen= FALSE; };
+private:
+  bool m_need_reopen;
+};
+
+
+/**
+  Handle an error from mysql_lock_tables().
+  Ignore ER_LOCK_ABORTED errors.
+*/
+
+bool
+Sql_handler_lock_error_handler::
+handle_condition(THD *thd,
+                 uint sql_errno,
+                 const char *sqlstate,
+                 MYSQL_ERROR::enum_warning_level level,
+                 const char* msg,
+                 MYSQL_ERROR **cond_hdl)
+{
+  *cond_hdl= NULL;
+  if (sql_errno == ER_LOCK_ABORTED)
+    m_need_reopen= TRUE;
+
+  return m_need_reopen;
+}
+
+
 /*
   Read from a HANDLER table.
 
@@ -442,7 +492,7 @@ bool mysql_ha_read(THD *thd, TABLE_LIST *tables,
   uint          num_rows;
   uchar		*UNINIT_VAR(key);
   uint		UNINIT_VAR(key_len);
-  bool          need_reopen;
+  Sql_handler_lock_error_handler sql_handler_lock_error;
   DBUG_ENTER("mysql_ha_read");
   DBUG_PRINT("enter",("'%s'.'%s' as '%s'",
                       tables->db, tables->table_name, tables->alias));
@@ -506,8 +556,12 @@ retry:
   thd->open_tables= hash_tables->table;
 
 
-  lock= mysql_lock_tables(thd, &thd->open_tables, 1, 0, &need_reopen);
+  sql_handler_lock_error.init();
+  thd->push_internal_handler(&sql_handler_lock_error);
+
+  lock= mysql_lock_tables(thd, &thd->open_tables, 1, 0);
 
+  thd->pop_internal_handler();
   /*
     In 5.1 and earlier, mysql_lock_tables() could replace the TABLE
     object with another one (reopen it). This is no longer the case
@@ -517,8 +571,9 @@ retry:
   /* Restore previous context. */
   thd->open_tables= backup_open_tables;
 
-  if (need_reopen)
+  if (sql_handler_lock_error.need_reopen())
   {
+    DBUG_ASSERT(!lock && !thd->is_error());
     mysql_ha_close_table(thd, hash_tables);
     goto retry;
   }
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index 26fb4e4e4e4..7423dd9d292 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -2396,7 +2396,8 @@ void kill_delayed_threads(void)
 bool Delayed_insert::open_and_lock_table()
 {
   if (!(table= open_n_lock_single_table(&thd, &table_list,
-                                        TL_WRITE_DELAYED, 0)))
+                                        TL_WRITE_DELAYED,
+                                        MYSQL_OPEN_IGNORE_GLOBAL_READ_LOCK)))
   {
     thd.fatal_error();				// Abort waiting inserts
     return TRUE;
@@ -2557,7 +2558,6 @@ pthread_handler_t handle_delayed_insert(void *arg)
 
       if (di->tables_in_use && ! thd->lock && !thd->killed)
       {
-        bool need_reopen;
         /*
           Request for new delayed insert.
           Lock the table, but avoid to be blocked by a global read lock.
@@ -2568,30 +2568,10 @@ pthread_handler_t handle_delayed_insert(void *arg)
           handler will close the table and finish when the outstanding
           inserts are done.
         */
-        if (! (thd->lock= mysql_lock_tables(thd, &di->table, 1,
-                                            MYSQL_LOCK_IGNORE_GLOBAL_READ_LOCK,
-                                            &need_reopen)))
+        if (! (thd->lock= mysql_lock_tables(thd, &di->table, 1, 0)))
         {
-          if (need_reopen && !thd->killed)
-          {
-            /*
-              We were waiting to obtain TL_WRITE_DELAYED (probably due to
-              someone having or requesting TL_WRITE_ALLOW_READ) and got
-              aborted. Try to reopen table and if it fails die.
-            */
-            TABLE_LIST *tl_ptr = &di->table_list;
-            close_tables_for_reopen(thd, &tl_ptr, NULL);
-            di->table= 0;
-            if (di->open_and_lock_table())
-            {
-              thd->killed= THD::KILL_CONNECTION;
-            }
-          }
-          else
-          {
-            /* Fatal error */
-            thd->killed= THD::KILL_CONNECTION;
-          }
+          /* Fatal error */
+          thd->killed= THD::KILL_CONNECTION;
         }
         mysql_cond_broadcast(&di->cond_client);
       }
@@ -3542,7 +3522,6 @@ static TABLE *create_table_from_items(THD *thd, HA_CREATE_INFO *create_info,
   List_iterator_fast<Item> it(*items);
   Item *item;
   Field *tmp_field;
-  bool not_used;
   DBUG_ENTER("create_table_from_items");
 
   tmp_table.alias= 0;
@@ -3666,8 +3645,7 @@ static TABLE *create_table_from_items(THD *thd, HA_CREATE_INFO *create_info,
     since it won't wait for the table lock (we have exclusive metadata lock on
     the table) and thus can't get aborted.
   */
-  if (! ((*lock)= mysql_lock_tables(thd, &table, 1,
-                                    MYSQL_LOCK_IGNORE_FLUSH, &not_used)) ||
+  if (! ((*lock)= mysql_lock_tables(thd, &table, 1, 0)) ||
         hooks->postlock(&table, 1))
   {
     if (*lock)
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index 8b8e223ad02..542a0378bf4 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -2934,7 +2934,7 @@ fill_schema_show_cols_or_idxs(THD *thd, TABLE_LIST *tables,
   */
   lex->sql_command= SQLCOM_SHOW_FIELDS;
   res= open_normal_and_derived_tables(thd, show_table_list,
-                                      (MYSQL_LOCK_IGNORE_FLUSH |
+                                      (MYSQL_OPEN_IGNORE_FLUSH |
                                        MYSQL_OPEN_FORCE_SHARED_HIGH_PRIO_MDL |
                                        (can_deadlock ?
                                         MYSQL_OPEN_FAIL_ON_MDL_CONFLICT : 0)));
@@ -3516,7 +3516,7 @@ int get_all_tables(THD *thd, TABLE_LIST *tables, COND *cond)
             show_table_list->i_s_requested_object=
               schema_table->i_s_requested_object;
             res= open_normal_and_derived_tables(thd, show_table_list,
-                   (MYSQL_LOCK_IGNORE_FLUSH |
+                   (MYSQL_OPEN_IGNORE_FLUSH |
                     MYSQL_OPEN_FORCE_SHARED_HIGH_PRIO_MDL |
                     (can_deadlock ? MYSQL_OPEN_FAIL_ON_MDL_CONFLICT : 0)));
             lex->sql_command= save_sql_command;
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 8ba8c50b01e..2a9e01daf18 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -1811,7 +1811,7 @@ bool mysql_rm_table(THD *thd,TABLE_LIST *tables, my_bool if_exists,
                     my_bool drop_temporary)
 {
   bool error;
-  Drop_table_error_handler err_handler(thd->get_internal_handler());
+  Drop_table_error_handler err_handler;
 
   DBUG_ENTER("mysql_rm_table");
 
@@ -4426,7 +4426,7 @@ static int prepare_for_repair(THD *thd, TABLE_LIST *table_list,
   MY_STAT stat_info;
   Open_table_context ot_ctx_unused(thd, LONG_TIMEOUT);
   DBUG_ENTER("prepare_for_repair");
-  uint reopen_for_repair_flags= (MYSQL_LOCK_IGNORE_FLUSH |
+  uint reopen_for_repair_flags= (MYSQL_OPEN_IGNORE_FLUSH |
                                  MYSQL_OPEN_HAS_MDL_LOCK);
 
   if (!(check_opt->sql_flags & TT_USEFRM))
@@ -7155,7 +7155,7 @@ view_err:
       tbl.table_name= tbl.alias= tmp_name;
       /* Table is in thd->temporary_tables */
       (void) open_table(thd, &tbl, thd->mem_root, &ot_ctx_unused,
-                        MYSQL_LOCK_IGNORE_FLUSH);
+                        MYSQL_OPEN_IGNORE_FLUSH);
       new_table= tbl.table;
     }
     else
diff --git a/sql/sql_update.cc b/sql/sql_update.cc
index 1643bce8ddd..62a35335374 100644
--- a/sql/sql_update.cc
+++ b/sql/sql_update.cc
@@ -203,33 +203,26 @@ int mysql_update(THD *thd,
   SQL_SELECT	*select;
   READ_RECORD	info;
   SELECT_LEX    *select_lex= &thd->lex->select_lex;
-  bool          need_reopen;
   ulonglong     id;
   List<Item> all_fields;
   THD::killed_state killed_status= THD::NOT_KILLED;
   MDL_ticket *start_of_statement_svp= thd->mdl_context.mdl_savepoint();
   DBUG_ENTER("mysql_update");
 
-  for ( ; ; )
-  {
-    if (open_tables(thd, &table_list, &table_count, 0))
-      DBUG_RETURN(1);
+  if (open_tables(thd, &table_list, &table_count, 0))
+    DBUG_RETURN(1);
 
-    if (table_list->multitable_view)
-    {
-      DBUG_ASSERT(table_list->view != 0);
-      DBUG_PRINT("info", ("Switch to multi-update"));
-      /* pass counter value */
-      thd->lex->table_count= table_count;
-      /* convert to multiupdate */
-      DBUG_RETURN(2);
-    }
-    if (!lock_tables(thd, table_list, table_count, 0, &need_reopen))
-      break;
-    if (!need_reopen)
-      DBUG_RETURN(1);
-    close_tables_for_reopen(thd, &table_list, start_of_statement_svp);
+  if (table_list->multitable_view)
+  {
+    DBUG_ASSERT(table_list->view != 0);
+    DBUG_PRINT("info", ("Switch to multi-update"));
+    /* pass counter value */
+    thd->lex->table_count= table_count;
+    /* convert to multiupdate */
+    DBUG_RETURN(2);
   }
+  if (lock_tables(thd, table_list, table_count, 0))
+    DBUG_RETURN(1);
 
   if (mysql_handle_derived(thd->lex, &mysql_derived_prepare) ||
       (thd->fill_derived_tables() &&
@@ -963,17 +956,14 @@ int mysql_multi_update_prepare(THD *thd)
   uint  table_count= lex->table_count;
   const bool using_lock_tables= thd->locked_tables_mode != LTM_NONE;
   bool original_multiupdate= (thd->lex->sql_command == SQLCOM_UPDATE_MULTI);
-  bool need_reopen= FALSE;
   MDL_ticket *start_of_statement_svp= thd->mdl_context.mdl_savepoint();
   DBUG_ENTER("mysql_multi_update_prepare");
 
   /* following need for prepared statements, to run next time multi-update */
   thd->lex->sql_command= SQLCOM_UPDATE_MULTI;
 
-reopen_tables:
-
   /* open tables and create derived ones, but do not lock and fill them */
-  if (((original_multiupdate || need_reopen) &&
+  if ((original_multiupdate &&
        open_tables(thd, &table_list, &table_count, 0)) ||
       mysql_handle_derived(lex, &mysql_derived_prepare))
     DBUG_RETURN(TRUE);
@@ -1089,58 +1079,11 @@ reopen_tables:
 
   /* now lock and fill tables */
   if (!thd->stmt_arena->is_stmt_prepare() &&
-      lock_tables(thd, table_list, table_count, 0, &need_reopen))
+      lock_tables(thd, table_list, table_count, 0))
   {
-    if (!need_reopen)
-      DBUG_RETURN(TRUE);
-
-    DBUG_PRINT("info", ("lock_tables failed, reopening"));
-
-    /*
-      We have to reopen tables since some of them were altered or dropped
-      during lock_tables() or something was done with their triggers.
-      Let us do some cleanups to be able do setup_table() and setup_fields()
-      once again.
-    */
-    List_iterator_fast<Item> it(*fields);
-    Item *item;
-    while ((item= it++))
-      item->cleanup();
-
-    /*
-      To not to hog memory (as a result of the 
-      unit->reinit_exec_mechanism() call below):
-    */
-    lex->unit.cleanup();
-
-    for (SELECT_LEX *sl= lex->all_selects_list;
-        sl;
-        sl= sl->next_select_in_list())
-    {
-      SELECT_LEX_UNIT *unit= sl->master_unit();
-      unit->reinit_exec_mechanism(); // reset unit->prepared flags
-      /*
-        Reset 'clean' flag back to force normal execution of
-        unit->cleanup() in Prepared_statement::cleanup_stmt()
-        (call to lex->unit.cleanup() above sets this flag to TRUE).
-      */
-      unit->unclean();
-    }
-
-    /*
-      Also we need to cleanup Natural_join_column::table_field items.
-      To not to traverse a join tree we will cleanup whole
-      thd->free_list (in PS execution mode that list may not contain
-      items from 'fields' list, so the cleanup above is necessary to.
-    */
-    cleanup_items(thd->free_list);
-    cleanup_items(thd->stmt_arena->free_list);
-    close_tables_for_reopen(thd, &table_list, start_of_statement_svp);
-
-    DEBUG_SYNC(thd, "multi_update_reopen_tables");
-
-    goto reopen_tables;
+    DBUG_RETURN(TRUE);
   }
+  /* @todo: downgrade the metadata locks here. */
 
   /*
     Check that we are not using table that we are updating, but we should
diff --git a/sql/tztime.cc b/sql/tztime.cc
index 632dca1ce44..7d88b7276f2 100644
--- a/sql/tztime.cc
+++ b/sql/tztime.cc
@@ -1668,7 +1668,7 @@ my_tz_init(THD *org_thd, const char *default_tzname, my_bool bootstrap)
     open all time zone tables to see if they exist.
   */
   if (open_and_lock_tables(thd, tz_tables, FALSE,
-                           MYSQL_LOCK_IGNORE_FLUSH | MYSQL_LOCK_IGNORE_TIMEOUT))
+                           MYSQL_OPEN_IGNORE_FLUSH | MYSQL_LOCK_IGNORE_TIMEOUT))
   {
     sql_print_warning("Can't open and lock time zone table: %s "
                       "trying to live without them", thd->stmt_da->message());