A fix for Bug#41158 "DROP TABLE holds LOCK_open during unlink()".

Remove acquisition of LOCK_open around file system operations,
since such operations are now protected by metadata locks.
Rework table discovery algorithm to not require LOCK_open.

No new tests added since all MDL locking operations are covered
in lock.test and mdl_sync.test, and as long as these tests
pass despite the increased concurrency, consistency must be
unaffected.

mysql-test/t/disabled.def:
  Disable NDB tests due to Bug#55799.
sql/datadict.cc:
  No longer necessary to protect ha_create_table() with
  LOCK_open. Serial execution is now ensured by metadata
  locks.
sql/ha_ndbcluster.cc:
  Do not manipulate with LOCK_open in cluster code.
sql/ha_ndbcluster_binlog.cc:
  Do not manipulate with LOCK_open in cluster code.
sql/ha_ndbcluster_binlog.h:
  Update function signature.
sql/handler.cc:
  Implement ha_check_if_table_exists().
  @todo: some engines provide ha_table_exists_in_engine()
  handlerton call, for those we perhaps shouldn't
  call ha_discover(), to be more efficient.
  Since currently it's only NDB, postpone till
  integration with NDB.
sql/handler.h:
  Declare ha_check_if_table_exists() function.
sql/mdl.cc:
  Remove an obsolete comment.
sql/sql_base.cc:
  Update to a new signature of close_cached_tables():
  from now on we always call it without LOCK_open.
  Update comments.
  Remove get_table_share_with_create(), we should
  not attempt to create a table under LOCK_open.
  Introduce get_table_share_with_discover() instead,
  which would request a back off action if the table
  exists in engine.
  Remove acquisition of LOCK_open for 
  data dictionary operations, such as check_if_table_exists().
  Do not use get_table_share_with_create/discover for views,
  where it's not needed.
  Make tdc_remove_table() optionally acquire LOCK_open
  to simplify usage of this function.
  Use the right mutex in the partitioning code when
  manipulating with thd->open_tables.
sql/sql_base.h:
  Update signatures of changes functions.
sql/sql_insert.cc:
  Do not wrap quick_rm_table() with LOCK_open acquisition, 
  this is unnecessary.
sql/sql_parse.cc:
  Update to the new calling convention of tdc_remove_table().
  Update to the new signature of close_cached_tables().
  Update comments.
sql/sql_rename.cc:
  Update to the new calling convention of tdc_remove_table().
  Remove acquisition of LOCK_open around filesystem
  operations.
sql/sql_show.cc:
  Remove get_trigger_table_impl().
  Do not acquire LOCK_open for a dirty read of the trigger
  file.
sql/sql_table.cc:
  Do not acquire LOCK_open for filesystem operations.
sql/sql_trigger.cc:
  Do not require LOCK_open for trigger file I/O.
sql/sql_truncate.cc:
  Update to the new signature of tdc_remove_table().
sql/sql_view.cc:
  Do not require LOCK_open for view I/O.
  Use tdc_remove_table() to expel view share.
  Update comments.
sql/sys_vars.cc:
  Update to the new signature of close_cached_tables().

18 files changed, 217 insertions, 319 deletions

diff --git a/sql/datadict.cc b/sql/datadict.cc
index 33c3b6bc700..7eea977fd5d 100644
--- a/sql/datadict.cc
+++ b/sql/datadict.cc
@@ -152,9 +152,7 @@ bool dd_recreate_table(THD *thd, const char *db, const char *table_name)
   build_table_filename(path, sizeof(path) - 1, db, table_name, "", 0);
 
   /* Attempt to reconstruct the table. */
-  mysql_mutex_lock(&LOCK_open);
   error= ha_create_table(thd, path, db, table_name, &create_info, TRUE);
-  mysql_mutex_unlock(&LOCK_open);
 
   DBUG_RETURN(error);
 }
diff --git a/sql/ha_ndbcluster.cc b/sql/ha_ndbcluster.cc
index d4a98265c49..0ec2e21056e 100644
--- a/sql/ha_ndbcluster.cc
+++ b/sql/ha_ndbcluster.cc
@@ -680,7 +680,7 @@ int ha_ndbcluster::ndb_err(NdbTransaction *trans)
     bzero((char*) &table_list,sizeof(table_list));
     table_list.db= m_dbname;
     table_list.alias= table_list.table_name= m_tabname;
-    close_cached_tables(thd, &table_list, FALSE, FALSE);
+    close_cached_tables(thd, &table_list, FALSE);
     break;
   }
   default:
@@ -5702,7 +5702,7 @@ int ha_ndbcluster::create(const char *name,
                                m_table->getObjectVersion(),
                                (is_truncate) ?
 			       SOT_TRUNCATE_TABLE : SOT_CREATE_TABLE, 
-			       0, 0, 1);
+			       0, 0);
       break;
     }
   }
@@ -6143,7 +6143,7 @@ int ha_ndbcluster::rename_table(const char *from, const char *to)
                                old_dbname, m_tabname,
                                ndb_table_id, ndb_table_version,
                                SOT_RENAME_TABLE,
-                               m_dbname, new_tabname, 1);
+                               m_dbname, new_tabname);
   }
 
   // If we are moving tables between databases, we need to recreate
@@ -6337,7 +6337,7 @@ retry_temporary_error1:
                              thd->query(), thd->query_length(),
                              share->db, share->table_name,
                              ndb_table_id, ndb_table_version,
-                             SOT_DROP_TABLE, 0, 0, 1);
+                             SOT_DROP_TABLE, 0, 0);
   }
   else if (table_dropped && share && share->op) /* ndbcluster_log_schema_op
                                                    will do a force GCP */
@@ -7019,7 +7019,6 @@ int ndbcluster_drop_database_impl(const char *path)
   while ((tabname=it++))
   {
     tablename_to_filename(tabname, tmp, FN_REFLEN - (tmp - full_path)-1);
-    mysql_mutex_lock(&LOCK_open);
     if (ha_ndbcluster::delete_table(0, ndb, full_path, dbname, tabname))
     {
       const NdbError err= dict->getNdbError();
@@ -7029,7 +7028,6 @@ int ndbcluster_drop_database_impl(const char *path)
         ret= ndb_to_mysql_error(&err);
       }
     }
-    mysql_mutex_unlock(&LOCK_open);
   }
   DBUG_RETURN(ret);      
 }
@@ -7056,7 +7054,7 @@ static void ndbcluster_drop_database(handlerton *hton, char *path)
   ha_ndbcluster::set_dbname(path, db);
   ndbcluster_log_schema_op(thd, 0,
                            thd->query(), thd->query_length(),
-                           db, "", 0, 0, SOT_DROP_DB, 0, 0, 0);
+                           db, "", 0, 0, SOT_DROP_DB, 0, 0);
 #endif
   DBUG_VOID_RETURN;
 }
@@ -7181,7 +7179,6 @@ int ndbcluster_find_all_files(THD *thd)
       my_free(data);
       my_free(pack_data);
 
-      mysql_mutex_lock(&LOCK_open);
       if (discover)
       {
         /* ToDo 4.1 database needs to be created if missing */
@@ -7199,7 +7196,6 @@ int ndbcluster_find_all_files(THD *thd)
                                        TRUE);
       }
 #endif
-      mysql_mutex_unlock(&LOCK_open);
     }
   }
   while (unhandled && retries);
@@ -7292,19 +7288,16 @@ int ndbcluster_find_files(handlerton *hton, THD *thd,
                            file_name->str, reg_ext, 0);
       if (my_access(name, F_OK))
       {
-        mysql_mutex_lock(&LOCK_open);
         DBUG_PRINT("info", ("Table %s listed and need discovery",
                             file_name->str));
         if (ndb_create_table_from_engine(thd, db, file_name->str))
         {
-          mysql_mutex_unlock(&LOCK_open);
           push_warning_printf(current_thd, MYSQL_ERROR::WARN_LEVEL_WARN,
                               ER_TABLE_EXISTS_ERROR,
                               "Discover of table %s.%s failed",
                               db, file_name->str);
           continue;
         }
-        mysql_mutex_unlock(&LOCK_open);
       }
       DBUG_PRINT("info", ("%s existed in NDB _and_ on disk ", file_name->str));
       file_on_disk= TRUE;
@@ -7361,10 +7354,8 @@ int ndbcluster_find_files(handlerton *hton, THD *thd,
       file_name_str= (char*)my_hash_element(&ok_tables, i);
       end= end1 +
         tablename_to_filename(file_name_str, end1, sizeof(name) - (end1 - name));
-      mysql_mutex_lock(&LOCK_open);
       ndbcluster_create_binlog_setup(ndb, name, end-name,
                                      db, file_name_str, TRUE);
-      mysql_mutex_unlock(&LOCK_open);
     }
   }
 #endif
@@ -7426,7 +7417,6 @@ int ndbcluster_find_files(handlerton *hton, THD *thd,
   }
 
   /* Lock mutex before creating .FRM files. */
-  mysql_mutex_lock(&LOCK_open);
   /* Create new files. */
   List_iterator_fast<char> it2(create_list);
   while ((file_name_str=it2++))
@@ -7441,8 +7431,6 @@ int ndbcluster_find_files(handlerton *hton, THD *thd,
     }
   }
 
-  mysql_mutex_unlock(&LOCK_open);
-
   my_hash_free(&ok_tables);
   my_hash_free(&ndb_tables);
 
@@ -8452,8 +8440,7 @@ int handle_trailing_share(NDB_SHARE *share)
   bzero((char*) &table_list,sizeof(table_list));
   table_list.db= share->db;
   table_list.alias= table_list.table_name= share->table_name;
-  mysql_mutex_assert_owner(&LOCK_open);
-  close_cached_tables(thd, &table_list, TRUE, FALSE);
+  close_cached_tables(thd, &table_list, FALSE);
 
   mysql_mutex_lock(&ndbcluster_mutex);
   /* ndb_share reference temporary free */
@@ -10612,13 +10599,13 @@ int ndbcluster_alter_tablespace(handlerton *hton,
                              thd->query(), thd->query_length(),
                              "", alter_info->tablespace_name,
                              0, 0,
-                             SOT_TABLESPACE, 0, 0, 0);
+                             SOT_TABLESPACE, 0, 0);
   else
     ndbcluster_log_schema_op(thd, 0,
                              thd->query(), thd->query_length(),
                              "", alter_info->logfile_group_name,
                              0, 0,
-                             SOT_LOGFILE_GROUP, 0, 0, 0);
+                             SOT_LOGFILE_GROUP, 0, 0);
 #endif
   DBUG_RETURN(FALSE);
 
diff --git a/sql/ha_ndbcluster_binlog.cc b/sql/ha_ndbcluster_binlog.cc
index 26fdb8e1425..72e3092f9a8 100644
--- a/sql/ha_ndbcluster_binlog.cc
+++ b/sql/ha_ndbcluster_binlog.cc
@@ -360,7 +360,6 @@ ndbcluster_binlog_open_table(THD *thd, NDB_SHARE *share,
   int error;
   DBUG_ENTER("ndbcluster_binlog_open_table");
   
-  mysql_mutex_assert_owner(&LOCK_open);
   init_tmp_table_share(thd, table_share, share->db, 0, share->table_name, 
                        share->key);
   if ((error= open_table_def(thd, table_share, 0)))
@@ -376,7 +375,9 @@ ndbcluster_binlog_open_table(THD *thd, NDB_SHARE *share,
     free_table_share(table_share);
     DBUG_RETURN(error);
   }
+  mysql_mutex_lock(&LOCK_open);
   assign_new_table_id(table_share);
+  mysql_mutex_unlock(&LOCK_open);
 
   if (!reopen)
   {
@@ -625,7 +626,7 @@ ndbcluster_binlog_log_query(handlerton *hton, THD *thd, enum_binlog_command binl
   {
     ndbcluster_log_schema_op(thd, 0, query, query_length,
                              db, table_name, 0, 0, type,
-                             0, 0, 0);
+                             0, 0);
   }
   DBUG_VOID_RETURN;
 }
@@ -908,9 +909,7 @@ int ndbcluster_setup_binlog_table_shares(THD *thd)
   if (!ndb_schema_share &&
       ndbcluster_check_ndb_schema_share() == 0)
   {
-    mysql_mutex_lock(&LOCK_open);
     ndb_create_table_from_engine(thd, NDB_REP_DB, NDB_SCHEMA_TABLE);
-    mysql_mutex_unlock(&LOCK_open);
     if (!ndb_schema_share)
     {
       ndbcluster_create_schema_table(thd);
@@ -922,9 +921,7 @@ int ndbcluster_setup_binlog_table_shares(THD *thd)
   if (!ndb_apply_status_share &&
       ndbcluster_check_ndb_apply_status_share() == 0)
   {
-    mysql_mutex_lock(&LOCK_open);
     ndb_create_table_from_engine(thd, NDB_REP_DB, NDB_APPLY_TABLE);
-    mysql_mutex_unlock(&LOCK_open);
     if (!ndb_apply_status_share)
     {
       ndbcluster_create_ndb_apply_status_table(thd);
@@ -934,12 +931,10 @@ int ndbcluster_setup_binlog_table_shares(THD *thd)
   }
   if (!ndbcluster_find_all_files(thd))
   {
-    mysql_mutex_lock(&LOCK_open);
     ndb_binlog_tables_inited= TRUE;
     if (opt_ndb_extra_logging)
       sql_print_information("NDB Binlog: ndb tables writable");
-    close_cached_tables(NULL, NULL, TRUE, FALSE);
-    mysql_mutex_unlock(&LOCK_open);
+    close_cached_tables(NULL, NULL, FALSE);
     /* Signal injector thread that all is setup */
     mysql_cond_signal(&injector_cond);
   }
@@ -1276,8 +1271,7 @@ int ndbcluster_log_schema_op(THD *thd, NDB_SHARE *share,
                              uint32 ndb_table_id,
                              uint32 ndb_table_version,
                              enum SCHEMA_OP_TYPE type,
-                             const char *new_db, const char *new_table_name,
-                             int have_lock_open)
+                             const char *new_db, const char *new_table_name)
 {
   DBUG_ENTER("ndbcluster_log_schema_op");
   Thd_ndb *thd_ndb= get_thd_ndb(thd);
@@ -1580,11 +1574,6 @@ end:
 
     int max_timeout= DEFAULT_SYNC_TIMEOUT;
     mysql_mutex_lock(&ndb_schema_object->mutex);
-    if (have_lock_open)
-    {
-      mysql_mutex_assert_owner(&LOCK_open);
-      mysql_mutex_unlock(&LOCK_open);
-    }
     while (1)
     {
       struct timespec abstime;
@@ -1640,10 +1629,6 @@ end:
                              "distributing", ndb_schema_object->key);
       }
     }
-    if (have_lock_open)
-    {
-      mysql_mutex_lock(&LOCK_open);
-    }
     mysql_mutex_unlock(&ndb_schema_object->mutex);
   }
 
@@ -1726,7 +1711,6 @@ ndb_handle_schema_change(THD *thd, Ndb *ndb, NdbEventOperation *pOp,
     {
       DBUG_DUMP("frm", (uchar*) altered_table->getFrmData(), 
                 altered_table->getFrmLength());
-      mysql_mutex_lock(&LOCK_open);
       Ndb_table_guard ndbtab_g(dict, tabname);
       const NDBTAB *old= ndbtab_g.get_table();
       if (!old &&
@@ -1752,7 +1736,7 @@ ndb_handle_schema_change(THD *thd, Ndb *ndb, NdbEventOperation *pOp,
       bzero((char*) &table_list,sizeof(table_list));
       table_list.db= (char *)dbname;
       table_list.alias= table_list.table_name= (char *)tabname;
-      close_cached_tables(thd, &table_list, TRUE, FALSE);
+      close_cached_tables(thd, &table_list, FALSE);
 
       if ((error= ndbcluster_binlog_open_table(thd, share,
                                                table_share, table, 1)))
@@ -1763,8 +1747,6 @@ ndb_handle_schema_change(THD *thd, Ndb *ndb, NdbEventOperation *pOp,
       table_share= share->table_share;
       dbname= table_share->db.str;
       tabname= table_share->table_name.str;
-
-      mysql_mutex_unlock(&LOCK_open);
     }
     my_free(data);
     my_free(pack_data);
@@ -1858,7 +1840,7 @@ ndb_handle_schema_change(THD *thd, Ndb *ndb, NdbEventOperation *pOp,
     bzero((char*) &table_list,sizeof(table_list));
     table_list.db= (char *)dbname;
     table_list.alias= table_list.table_name= (char *)tabname;
-    close_cached_tables(thd, &table_list, FALSE, FALSE);
+    close_cached_tables(thd, &table_list, FALSE);
     /* ndb_share reference create free */
     DBUG_PRINT("NDB_SHARE", ("%s create free  use_count: %u",
                              share->key, share->use_count));
@@ -1979,7 +1961,7 @@ ndb_binlog_thread_handle_schema_event(THD *thd, Ndb *ndb,
             bzero((char*) &table_list,sizeof(table_list));
             table_list.db= schema->db;
             table_list.alias= table_list.table_name= schema->name;
-            close_cached_tables(thd, &table_list, FALSE, FALSE);
+            close_cached_tables(thd, &table_list, FALSE);
           }
           /* ndb_share reference temporary free */
           if (share)
@@ -1991,7 +1973,6 @@ ndb_binlog_thread_handle_schema_event(THD *thd, Ndb *ndb,
         }
         // fall through
         case SOT_CREATE_TABLE:
-          mysql_mutex_lock(&LOCK_open);
           if (ndbcluster_check_if_local_table(schema->db, schema->name))
           {
             DBUG_PRINT("info", ("NDB Binlog: Skipping locally defined table '%s.%s'",
@@ -2005,7 +1986,6 @@ ndb_binlog_thread_handle_schema_event(THD *thd, Ndb *ndb,
           {
             print_could_not_discover_error(thd, schema);
           }
-          mysql_mutex_unlock(&LOCK_open);
           log_query= 1;
           break;
         case SOT_DROP_DB:
@@ -2096,7 +2076,7 @@ ndb_binlog_thread_handle_schema_event(THD *thd, Ndb *ndb,
       mysql_mutex_unlock(&ndb_schema_share_mutex);
       /* end protect ndb_schema_share */
 
-      close_cached_tables(NULL, NULL, FALSE, FALSE);
+      close_cached_tables(NULL, NULL, FALSE);
       // fall through
     case NDBEVENT::TE_ALTER:
       ndb_handle_schema_change(thd, ndb, pOp, tmp_share);
@@ -2253,7 +2233,7 @@ ndb_binlog_thread_handle_schema_event_post_epoch(THD *thd,
           bzero((char*) &table_list,sizeof(table_list));
           table_list.db= schema->db;
           table_list.alias= table_list.table_name= schema->name;
-          close_cached_tables(thd, &table_list, FALSE, FALSE);
+          close_cached_tables(thd, &table_list, FALSE);
         }
         if (schema_type != SOT_ALTER_TABLE)
           break;
@@ -2274,7 +2254,6 @@ ndb_binlog_thread_handle_schema_event_post_epoch(THD *thd,
             free_share(&share);
             share= 0;
           }
-          mysql_mutex_lock(&LOCK_open);
           if (ndbcluster_check_if_local_table(schema->db, schema->name))
           {
             DBUG_PRINT("info", ("NDB Binlog: Skipping locally defined table '%s.%s'",
@@ -2288,7 +2267,6 @@ ndb_binlog_thread_handle_schema_event_post_epoch(THD *thd,
           {
             print_could_not_discover_error(thd, schema);
           }
-          mysql_mutex_unlock(&LOCK_open);
         }
         break;
       default:
@@ -3154,8 +3132,6 @@ ndbcluster_handle_drop_table(Ndb *ndb, const char *event_name,
 #ifdef SYNC_DROP_
   thd->proc_info= "Syncing ndb table schema operation and binlog";
   mysql_mutex_lock(&share->mutex);
-  mysql_mutex_assert_owner(&LOCK_open);
-  mysql_mutex_unlock(&LOCK_open);
   int max_timeout= DEFAULT_SYNC_TIMEOUT;
   while (share->op)
   {
@@ -3181,7 +3157,6 @@ ndbcluster_handle_drop_table(Ndb *ndb, const char *event_name,
                            type_str, share->key);
     }
   }
-  mysql_mutex_lock(&LOCK_open);
   mysql_mutex_unlock(&share->mutex);
 #else
   mysql_mutex_lock(&share->mutex);
diff --git a/sql/ha_ndbcluster_binlog.h b/sql/ha_ndbcluster_binlog.h
index 4d2a49588b4..5dbcf0fa43f 100644
--- a/sql/ha_ndbcluster_binlog.h
+++ b/sql/ha_ndbcluster_binlog.h
@@ -158,8 +158,7 @@ int ndbcluster_log_schema_op(THD *thd, NDB_SHARE *share,
                              uint32 ndb_table_version,
                              enum SCHEMA_OP_TYPE type,
                              const char *new_db,
-                             const char *new_table_name,
-                             int have_lock_open);
+                             const char *new_table_name);
 int ndbcluster_handle_drop_table(Ndb *ndb, const char *event_name,
                                  NDB_SHARE *share,
                                  const char *type_str);
diff --git a/sql/handler.cc b/sql/handler.cc
index 9893b3cac16..567dbe6ea49 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -3670,6 +3670,34 @@ int ha_create_table_from_engine(THD* thd, const char *db, const char *name)
   DBUG_RETURN(error != 0);
 }
 
+
+/**
+  Try to find a table in a storage engine. 
+
+  @param db   Normalized table schema name
+  @param name Normalized table name.
+  @param[out] exists Only valid if the function succeeded.
+
+  @retval TRUE   An error is found
+  @retval FALSE  Success, check *exists
+*/
+
+bool
+ha_check_if_table_exists(THD* thd, const char *db, const char *name,
+                         bool *exists)
+{
+  uchar *frmblob= NULL;
+  size_t frmlen;
+  DBUG_ENTER("ha_check_if_table_exists");
+
+  *exists= ! ha_discover(thd, db, name, &frmblob, &frmlen);
+  if (*exists)
+    my_free(frmblob);
+
+  DBUG_RETURN(FALSE);
+}
+
+
 void st_ha_check_opt::init()
 {
   flags= sql_flags= 0;
diff --git a/sql/handler.h b/sql/handler.h
index cad97c1f751..4229769290d 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -2120,6 +2120,8 @@ bool ha_show_status(THD *thd, handlerton *db_type, enum ha_stat_type stat);
 
 /* discovery */
 int ha_create_table_from_engine(THD* thd, const char *db, const char *name);
+bool ha_check_if_table_exists(THD* thd, const char *db, const char *name,
+                             bool *exists);
 int ha_discover(THD* thd, const char* dbname, const char* name,
                 uchar** frmblob, size_t* frmlen);
 int ha_find_files(THD *thd,const char *db,const char *path,
diff --git a/sql/mdl.cc b/sql/mdl.cc
index 5b50b11647c..1178428e983 100644
--- a/sql/mdl.cc
+++ b/sql/mdl.cc
@@ -494,14 +494,6 @@ mdl_locks_key(const uchar *record, size_t *length,
   the associated condition variable: LOCK_mdl and COND_mdl.
   These locking primitives are implementation details of the MDL
   subsystem and are private to it.
-
-  Note, that even though the new implementation adds acquisition
-  of a new global mutex to the execution flow of almost every SQL
-  statement, the design capitalizes on that to later save on
-  look ups in the table definition cache. This leads to reduced
-  contention overall and on LOCK_open in particular.
-  Please see the description of MDL_context::acquire_lock()
-  for details.
 */
 
 void mdl_init()
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index be0ea9c0815..9aea868a197 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -314,8 +314,6 @@ void table_def_start_shutdown(void)
   if (table_def_inited)
   {
     mysql_mutex_lock(&LOCK_open);
-    /* Free all cached but unused TABLEs and TABLE_SHAREs first. */
-    close_cached_tables(NULL, NULL, TRUE, FALSE);
     /*
       Ensure that TABLE and TABLE_SHARE objects which are created for
       tables that are open during process of plugins' shutdown are
@@ -324,6 +322,8 @@ void table_def_start_shutdown(void)
     */
     table_def_shutdown_in_progress= TRUE;
     mysql_mutex_unlock(&LOCK_open);
+    /* Free all cached but unused TABLEs and TABLE_SHAREs. */
+    close_cached_tables(NULL, NULL, FALSE);
   }
 }
 
@@ -516,10 +516,10 @@ TABLE_SHARE *get_table_share(THD *thd, TABLE_LIST *table_list, char *key,
   }
 
   /*
-    We assign a new table id under the protection of the LOCK_open and
-    the share's own mutex.  We do this insted of creating a new mutex
+    We assign a new table id under the protection of the LOCK_open.
+    We do this instead of creating a new mutex
     and using it for the sole purpose of serializing accesses to a
-    static variable, we assign the table id here.  We assign it to the
+    static variable, we assign the table id here. We assign it to the
     share before inserting it into the table_def_cache to be really
     sure that it cannot be read from the cache without having a table
     id assigned.
@@ -547,7 +547,7 @@ TABLE_SHARE *get_table_share(THD *thd, TABLE_LIST *table_list, char *key,
   DBUG_RETURN(share);
 
 found:
-  /* 
+  /*
      We found an existing table definition. Return it if we didn't get
      an error when reading the table definition from file.
   */
@@ -589,21 +589,21 @@ found:
 }
 
 
-/*
+/**
   Get a table share. If it didn't exist, try creating it from engine
 
-  For arguments and return values, see get_table_from_share()
+  For arguments and return values, see get_table_share()
 */
 
-static TABLE_SHARE
-*get_table_share_with_create(THD *thd, TABLE_LIST *table_list,
-                             char *key, uint key_length,
-                             uint db_flags, int *error,
-                             my_hash_value_type hash_value)
+static TABLE_SHARE *
+get_table_share_with_discover(THD *thd, TABLE_LIST *table_list,
+                              char *key, uint key_length,
+                              uint db_flags, int *error,
+                              my_hash_value_type hash_value)
 
 {
   TABLE_SHARE *share;
-  int tmp;
+  bool exists;
   DBUG_ENTER("get_table_share_with_create");
 
   share= get_table_share(thd, table_list, key, key_length, db_flags, error,
@@ -617,10 +617,15 @@ static TABLE_SHARE
     from the pre-locking list. In this case we still need to try
     auto-discover before returning a NULL share.
 
+    Or, we're inside SHOW CREATE VIEW, which
+    also installs a silencer for ER_NO_SUCH_TABLE error.
+
     If share is NULL and the error is ER_NO_SUCH_TABLE, this is
-    the same as above, only that the error was not silenced by 
-    pre-locking. Once again, we need to try to auto-discover
-    the share.
+    the same as above, only that the error was not silenced by
+    pre-locking or SHOW CREATE VIEW.
+
+    In both these cases it won't harm to try to discover the
+    table.
 
     Finally, if share is still NULL, it's a real error and we need
     to abort.
@@ -628,20 +633,25 @@ static TABLE_SHARE
     @todo Rework alternative ways to deal with ER_NO_SUCH TABLE.
   */
   if (share || (thd->is_error() && thd->stmt_da->sql_errno() != ER_NO_SUCH_TABLE))
-
     DBUG_RETURN(share);
 
+  *error= 0;
+
   /* Table didn't exist. Check if some engine can provide it */
-  tmp= ha_create_table_from_engine(thd, table_list->db,
-                                   table_list->table_name);
-  if (tmp < 0)
+  if (ha_check_if_table_exists(thd, table_list->db, table_list->table_name,
+                               &exists))
+  {
+    thd->clear_error();
+    /* Conventionally, the storage engine API does not report errors. */
+    my_error(ER_OUT_OF_RESOURCES, MYF(0));
+  }
+  else if (! exists)
   {
     /*
       No such table in any engine.
       Hide "Table doesn't exist" errors if the table belongs to a view.
       The check for thd->is_error() is necessary to not push an
-      unwanted error in case of pre-locking, which silences
-      "no such table" errors.
+      unwanted error in case the error was already silenced.
       @todo Rework the alternative ways to deal with ER_NO_SUCH TABLE.
     */
     if (thd->is_error())
@@ -659,27 +669,16 @@ static TABLE_SHARE
                  view->view_db.str, view->view_name.str);
       }
     }
-    DBUG_RETURN(0);
   }
-  if (tmp)
+  else
   {
-    /* Give right error message */
     thd->clear_error();
-    DBUG_PRINT("error", ("Discovery of %s/%s failed", table_list->db,
-                         table_list->table_name));
-    my_printf_error(ER_UNKNOWN_ERROR,
-                    "Failed to open '%-.64s', error while "
-                    "unpacking from engine",
-                    MYF(0), table_list->table_name);
-    DBUG_RETURN(0);
+    *error= 7; /* Run auto-discover. */
   }
-  /* Table existed in engine. Let's open it */
-  thd->warning_info->clear_warning_info(thd->query_id);
-  thd->clear_error();                           // Clear error message
-  DBUG_RETURN(get_table_share(thd, table_list, key, key_length,
-                              db_flags, error, hash_value));
+  DBUG_RETURN(NULL);
 }
 
+
 /**
   Mark that we are not using table share anymore.
 
@@ -926,7 +925,6 @@ static void kill_delayed_threads_for_table(TABLE_SHARE *share)
 
   @param thd Thread context
   @param tables List of tables to remove from the cache
-  @param have_lock If LOCK_open is locked
   @param wait_for_refresh Wait for a impending flush
 
   @note THD can be NULL, but then wait_for_refresh must be FALSE
@@ -940,16 +938,14 @@ static void kill_delayed_threads_for_table(TABLE_SHARE *share)
         lock taken by thread trying to obtain global read lock.
 */
 
-bool close_cached_tables(THD *thd, TABLE_LIST *tables, bool have_lock,
-                         bool wait_for_refresh)
+bool close_cached_tables(THD *thd, TABLE_LIST *tables, bool wait_for_refresh)
 {
   bool result= FALSE;
   bool found= TRUE;
   DBUG_ENTER("close_cached_tables");
   DBUG_ASSERT(thd || (!wait_for_refresh && !tables));
 
-  if (!have_lock)
-    mysql_mutex_lock(&LOCK_open);
+  mysql_mutex_lock(&LOCK_open);
   if (!tables)
   {
     refresh_version++;				// Force close of open tables
@@ -978,7 +974,7 @@ bool close_cached_tables(THD *thd, TABLE_LIST *tables, bool have_lock,
         kill_delayed_threads_for_table(share);
         /* tdc_remove_table() also sets TABLE_SHARE::version to 0. */
         tdc_remove_table(thd, TDC_RT_REMOVE_UNUSED, table->db,
-                         table->table_name);
+                         table->table_name, TRUE);
 	found=1;
       }
     }
@@ -986,15 +982,11 @@ bool close_cached_tables(THD *thd, TABLE_LIST *tables, bool have_lock,
       wait_for_refresh=0;			// Nothing to wait for
   }
 
-  if (!have_lock)
-    mysql_mutex_unlock(&LOCK_open);
+  mysql_mutex_unlock(&LOCK_open);
 
   if (!wait_for_refresh)
     DBUG_RETURN(result);
 
-  /* Code below assume that LOCK_open is released. */
-  DBUG_ASSERT(!have_lock);
-
   if (thd->locked_tables_mode)
   {
     /*
@@ -1108,7 +1100,7 @@ err_with_reopen:
 */
 
 bool close_cached_connection_tables(THD *thd, bool if_wait_for_refresh,
-                                    LEX_STRING *connection, bool have_lock)
+                                    LEX_STRING *connection)
 {
   uint idx;
   TABLE_LIST tmp, *tables= NULL;
@@ -1118,8 +1110,7 @@ bool close_cached_connection_tables(THD *thd, bool if_wait_for_refresh,
 
   bzero(&tmp, sizeof(TABLE_LIST));
 
-  if (!have_lock)
-    mysql_mutex_lock(&LOCK_open);
+  mysql_mutex_lock(&LOCK_open);
 
   for (idx= 0; idx < table_def_cache.records; idx++)
   {
@@ -1147,12 +1138,10 @@ bool close_cached_connection_tables(THD *thd, bool if_wait_for_refresh,
     tables= (TABLE_LIST *) memdup_root(thd->mem_root, (char*)&tmp, 
                                        sizeof(TABLE_LIST));
   }
+  mysql_mutex_unlock(&LOCK_open);
 
   if (tables)
-    result= close_cached_tables(thd, tables, TRUE, FALSE);
-
-  if (!have_lock)
-    mysql_mutex_unlock(&LOCK_open);
+    result= close_cached_tables(thd, tables, FALSE);
 
   if (if_wait_for_refresh)
   {
@@ -1355,9 +1344,8 @@ close_all_tables_for_name(THD *thd, TABLE_SHARE *share,
     }
   }
   /* Remove the table share from the cache. */
-  mysql_mutex_lock(&LOCK_open);
-  tdc_remove_table(thd, TDC_RT_REMOVE_ALL, db, table_name);
-  mysql_mutex_unlock(&LOCK_open);
+  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.
@@ -2201,10 +2189,9 @@ bool wait_while_table_is_used(THD *thd, TABLE *table,
              table->mdl_ticket, thd->variables.lock_wait_timeout))
     DBUG_RETURN(TRUE);
 
-  mysql_mutex_lock(&LOCK_open);
   tdc_remove_table(thd, TDC_RT_REMOVE_NOT_OWN,
-                   table->s->db.str, table->s->table_name.str);
-  mysql_mutex_unlock(&LOCK_open);
+                   table->s->db.str, table->s->table_name.str,
+                   FALSE);
   /* extra() call must come only after all instances above are closed */
   (void) table->file->extra(function);
   DBUG_RETURN(FALSE);
@@ -2245,9 +2232,8 @@ void drop_open_table(THD *thd, TABLE *table, const char *db_name,
     table->file->extra(HA_EXTRA_PREPARE_FOR_DROP);
     close_thread_table(thd, &thd->open_tables);
     /* Remove the table share from the table cache. */
-    mysql_mutex_lock(&LOCK_open);
-    tdc_remove_table(thd, TDC_RT_REMOVE_ALL, db_name, table_name);
-    mysql_mutex_unlock(&LOCK_open);
+    tdc_remove_table(thd, TDC_RT_REMOVE_ALL, db_name, table_name,
+                     FALSE);
     /* Remove the table from the storage engine and rm the .frm. */
     quick_rm_table(table_type, db_name, table_name, 0);
   }
@@ -2279,16 +2265,20 @@ void drop_open_table(THD *thd, TABLE *table, const char *db_name,
 bool check_if_table_exists(THD *thd, TABLE_LIST *table, bool *exists)
 {
   char path[FN_REFLEN + 1];
-  int rc= 0;
+  TABLE_SHARE *share;
   DBUG_ENTER("check_if_table_exists");
 
-  mysql_mutex_assert_not_owner(&LOCK_open);
-
   *exists= TRUE;
 
+  DBUG_ASSERT(thd->mdl_context.
+              is_lock_owner(MDL_key::TABLE, table->db,
+                            table->table_name, MDL_SHARED));
+
   mysql_mutex_lock(&LOCK_open);
+  share= get_cached_table_share(table->db, table->table_name);
+  mysql_mutex_unlock(&LOCK_open);
 
-  if (get_cached_table_share(table->db, table->table_name))
+  if (share)
     goto end;
 
   build_table_filename(path, sizeof(path) - 1, table->db, table->table_name,
@@ -2298,24 +2288,14 @@ bool check_if_table_exists(THD *thd, TABLE_LIST *table, bool *exists)
     goto end;
 
   /* .FRM file doesn't exist. Check if some engine can provide it. */
-
-  rc= ha_create_table_from_engine(thd, table->db, table->table_name);
-
-  if (rc < 0)
-  {
-    /* Table does not exists in engines as well. */
-    *exists= FALSE;
-    rc= 0;
-  }
-  else if (rc)
+  if (ha_check_if_table_exists(thd, table->db, table->table_name, exists))
   {
-    my_printf_error(ER_UNKNOWN_ERROR, "Failed to open '%-.64s', error while "
+    my_printf_error(ER_OUT_OF_RESOURCES, "Failed to open '%-.64s', error while "
                     "unpacking from engine", MYF(0), table->table_name);
+    DBUG_RETURN(TRUE);
   }
-
 end:
-  mysql_mutex_unlock(&LOCK_open);
-  DBUG_RETURN(test(rc));
+  DBUG_RETURN(FALSE);
 }
 
 
@@ -2812,11 +2792,25 @@ bool open_table(THD *thd, TABLE_LIST *table_list, MEM_ROOT *mem_root,
 
   mysql_mutex_lock(&LOCK_open);
 
-  if (!(share= get_table_share_with_create(thd, table_list, key,
-                                           key_length, OPEN_VIEW,
-                                           &error,
-                                           hash_value)))
-    goto err_unlock2;
+  if (!(share= get_table_share_with_discover(thd, table_list, key,
+                                             key_length, OPEN_VIEW,
+                                             &error,
+                                             hash_value)))
+  {
+    mysql_mutex_unlock(&LOCK_open);
+    /*
+      If thd->is_error() is not set, we either need discover
+      (error == 7), or the error was silenced by the prelocking
+      handler (error == 0), in which case we should skip this
+      table.
+    */
+    if (error == 7 && !thd->is_error())
+    {
+      (void) ot_ctx->request_backoff_action(Open_table_context::OT_DISCOVER,
+                                            table_list);
+    }
+    DBUG_RETURN(TRUE);
+  }
 
   if (share->is_view)
   {
@@ -3010,7 +3004,6 @@ err_lock:
   mysql_mutex_lock(&LOCK_open);
 err_unlock:
   release_table_share(share);
-err_unlock2:
   mysql_mutex_unlock(&LOCK_open);
 
   DBUG_RETURN(TRUE);
@@ -3633,10 +3626,10 @@ bool tdc_open_view(THD *thd, TABLE_LIST *table_list, const char *alias,
                            cache_key_length);
   mysql_mutex_lock(&LOCK_open);
 
-  if (!(share= get_table_share_with_create(thd, table_list, cache_key,
-                                           cache_key_length, 
-                                           OPEN_VIEW, &error,
-                                           hash_value)))
+  if (!(share= get_table_share(thd, table_list, cache_key,
+                               cache_key_length,
+                               OPEN_VIEW, &error,
+                               hash_value)))
     goto err;
 
   if (share->is_view &&
@@ -3738,10 +3731,10 @@ static bool auto_repair_table(THD *thd, TABLE_LIST *table_list)
                            cache_key_length);
   mysql_mutex_lock(&LOCK_open);
 
-  if (!(share= get_table_share_with_create(thd, table_list, cache_key,
-                                           cache_key_length,
-                                           OPEN_VIEW, &not_used,
-                                           hash_value)))
+  if (!(share= get_table_share(thd, table_list, cache_key,
+                               cache_key_length,
+                               OPEN_VIEW, &not_used,
+                               hash_value)))
     goto end_unlock;
 
   if (share->is_view)
@@ -3786,7 +3779,8 @@ static bool auto_repair_table(THD *thd, TABLE_LIST *table_list)
   release_table_share(share);
   /* Remove the repaired share from the table cache. */
   tdc_remove_table(thd, TDC_RT_REMOVE_ALL,
-                   table_list->db, table_list->table_name);
+                   table_list->db, table_list->table_name,
+                   TRUE);
 end_unlock:
   mysql_mutex_unlock(&LOCK_open);
   return result;
@@ -3908,12 +3902,10 @@ recover_from_failed_open(THD *thd)
                                       MYSQL_OPEN_SKIP_TEMPORARY)))
           break;
 
-        mysql_mutex_lock(&LOCK_open);
         tdc_remove_table(thd, TDC_RT_REMOVE_ALL, m_failed_table->db,
-                         m_failed_table->table_name);
+                         m_failed_table->table_name, FALSE);
         ha_create_table_from_engine(thd, m_failed_table->db,
                                     m_failed_table->table_name);
-        mysql_mutex_unlock(&LOCK_open);
 
         thd->warning_info->clear_warning_info(thd->query_id);
         thd->clear_error();                 // Clear error message
@@ -3927,10 +3919,8 @@ recover_from_failed_open(THD *thd)
                                       MYSQL_OPEN_SKIP_TEMPORARY)))
           break;
 
-        mysql_mutex_lock(&LOCK_open);
         tdc_remove_table(thd, TDC_RT_REMOVE_ALL, m_failed_table->db,
-                         m_failed_table->table_name);
-        mysql_mutex_unlock(&LOCK_open);
+                         m_failed_table->table_name, FALSE);
 
         result= auto_repair_table(thd, m_failed_table);
         thd->mdl_context.release_transactional_locks();
@@ -8541,7 +8531,7 @@ void tdc_flush_unused_tables()
   mysql_mutex_lock(&LOCK_open);
   while (unused_tables)
     free_cache_entry(unused_tables);
-  (void) mysql_mutex_unlock(&LOCK_open);
+  mysql_mutex_unlock(&LOCK_open);
 }
 
 
@@ -8646,20 +8636,25 @@ bool mysql_notify_thread_having_shared_lock(THD *thd, THD *in_use,
                                                 remove TABLE_SHARE).
    @param  db           Name of database
    @param  table_name   Name of table
+   @param  has_lock     If TRUE, LOCK_open is already acquired
 
    @note It assumes that table instances are already not used by any
    (other) thread (this should be achieved by using meta-data locks).
 */
 
 void tdc_remove_table(THD *thd, enum_tdc_remove_table_type remove_type,
-                      const char *db, const char *table_name)
+                      const char *db, const char *table_name,
+                      bool has_lock)
 {
   char key[MAX_DBKEY_LENGTH];
   uint key_length;
   TABLE *table;
   TABLE_SHARE *share;
 
-  mysql_mutex_assert_owner(&LOCK_open);
+  if (! has_lock)
+    mysql_mutex_lock(&LOCK_open);
+  else
+    mysql_mutex_assert_owner(&LOCK_open);
 
   DBUG_ASSERT(remove_type == TDC_RT_REMOVE_UNUSED ||
               thd->mdl_context.is_lock_owner(MDL_key::TABLE, db, table_name,
@@ -8700,6 +8695,9 @@ void tdc_remove_table(THD *thd, enum_tdc_remove_table_type remove_type,
     else
       (void) my_hash_delete(&table_def_cache, (uchar*) share);
   }
+
+  if (! has_lock)
+    mysql_mutex_unlock(&LOCK_open);
 }
 
 
@@ -8905,11 +8903,6 @@ static int alter_close_tables(ALTER_PARTITION_PARAM_TYPE *lpt)
   TABLE *table;
   DBUG_ENTER("alter_close_tables");
   /*
-    We must keep LOCK_open while manipulating with thd->open_tables.
-    Another thread may be working on it.
-  */
-  mysql_mutex_lock(&LOCK_open);
-  /*
     We can safely remove locks for all tables with the same name:
     later they will all be closed anyway in
     alter_partition_lock_handling().
@@ -8919,9 +8912,20 @@ static int alter_close_tables(ALTER_PARTITION_PARAM_TYPE *lpt)
     if (!strcmp(table->s->table_name.str, share->table_name.str) &&
 	!strcmp(table->s->db.str, share->db.str))
     {
+      /*
+        No need to take LOCK_thd_data to protect mysql_lock_remove(),
+        since mysql_lock_abort_for_thread() only aborts waiting
+        locks, and our lock is already granted.
+      */
       mysql_lock_remove(thd, thd->lock, table);
+      /*
+        Protect members of thd->open_tables concurrently used
+        in mysql_notify_thread_having_shared_lock().
+      */
+      mysql_mutex_lock(&thd->LOCK_thd_data);
       table->file->close();
       table->db_stat= 0;                        // Mark file closed
+      mysql_mutex_unlock(&thd->LOCK_thd_data);
       /*
         Ensure that we won't end up with a crippled table instance
         in the table cache if an error occurs before we reach
@@ -8930,10 +8934,10 @@ static int alter_close_tables(ALTER_PARTITION_PARAM_TYPE *lpt)
       */
       tdc_remove_table(thd, TDC_RT_REMOVE_UNUSED,
                        table->s->db.str,
-                       table->s->table_name.str);
+                       table->s->table_name.str,
+                       FALSE);
     }
   }
-  mysql_mutex_unlock(&LOCK_open);
   DBUG_RETURN(0);
 }
 
diff --git a/sql/sql_base.h b/sql/sql_base.h
index 05401a8cc6d..379aa67f203 100644
--- a/sql/sql_base.h
+++ b/sql/sql_base.h
@@ -226,16 +226,15 @@ TABLE *open_performance_schema_table(THD *thd, TABLE_LIST *one_table,
                                      Open_tables_state *backup);
 void close_performance_schema_table(THD *thd, Open_tables_state *backup);
 
-bool close_cached_tables(THD *thd, TABLE_LIST *tables, bool have_lock,
-                         bool wait_for_refresh);
+bool close_cached_tables(THD *thd, TABLE_LIST *tables, bool wait_for_refresh);
 bool close_cached_connection_tables(THD *thd, bool wait_for_refresh,
-                                    LEX_STRING *connect_string,
-                                    bool have_lock = FALSE);
+                                    LEX_STRING *connect_string);
 void close_all_tables_for_name(THD *thd, TABLE_SHARE *share,
                                bool remove_from_locked_tables);
 OPEN_TABLE_LIST *list_open_tables(THD *thd, const char *db, const char *wild);
 void tdc_remove_table(THD *thd, enum_tdc_remove_table_type remove_type,
-                      const char *db, const char *table_name);
+                      const char *db, const char *table_name,
+                      bool has_lock);
 bool tdc_open_view(THD *thd, TABLE_LIST *table_list, const char *alias,
                    char *cache_key, uint cache_key_length,
                    MEM_ROOT *mem_root, uint flags);
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index a0d347f48de..ce4535307c8 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -3600,11 +3600,9 @@ static TABLE *create_table_from_items(THD *thd, HA_CREATE_INFO *create_info,
         */
         if (open_table(thd, create_table, thd->mem_root, &ot_ctx))
         {
-          mysql_mutex_lock(&LOCK_open);
           quick_rm_table(create_info->db_type, create_table->db,
                          table_case_name(create_info, create_table->table_name),
                          0);
-          mysql_mutex_unlock(&LOCK_open);
         }
         else
           table= create_table->table;
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 2ef8e9761b1..ba2c9d07845 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -1789,11 +1789,9 @@ static bool flush_tables_with_read_lock(THD *thd, TABLE_LIST *all_tables)
        table_list= table_list->next_global)
   {
     /* Remove the table from cache. */
-    mysql_mutex_lock(&LOCK_open);
     tdc_remove_table(thd, TDC_RT_REMOVE_ALL,
                      table_list->db,
-                     table_list->table_name);
-    mysql_mutex_unlock(&LOCK_open);
+                     table_list->table_name, FALSE);
 
     /* Skip views and temporary tables. */
     table_list->required_type= FRMTYPE_TABLE; /* Don't try to flush views. */
@@ -3414,7 +3412,7 @@ end_with_restore_list:
       /* So that DROP TEMPORARY TABLE gets to binlog at commit/rollback */
       thd->variables.option_bits|= OPTION_KEEP_LOG;
     }
-    /* DDL and binlog write order protected by LOCK_open */
+    /* DDL and binlog write order are protected by metadata locks. */
     res= mysql_rm_table(thd, first_table, lex->drop_if_exists,
 			lex->drop_temporary);
   }
@@ -6854,7 +6852,7 @@ bool reload_acl_and_cache(THD *thd, ulong options, TABLE_LIST *tables,
       tmp_write_to_binlog= 0;
       if (thd->global_read_lock.lock_global_read_lock(thd))
 	return 1;                               // Killed
-      if (close_cached_tables(thd, tables, FALSE, (options & REFRESH_FAST) ?
+      if (close_cached_tables(thd, tables, (options & REFRESH_FAST) ?
                               FALSE : TRUE))
           result= 1;
       
@@ -6894,7 +6892,7 @@ bool reload_acl_and_cache(THD *thd, ulong options, TABLE_LIST *tables,
         }
       }
 
-      if (close_cached_tables(thd, tables, FALSE, (options & REFRESH_FAST) ?
+      if (close_cached_tables(thd, tables, (options & REFRESH_FAST) ?
                               FALSE : TRUE))
         result= 1;
     }
diff --git a/sql/sql_rename.cc b/sql/sql_rename.cc
index 301b22bd70e..f0b53abcb03 100644
--- a/sql/sql_rename.cc
+++ b/sql/sql_rename.cc
@@ -147,13 +147,15 @@ bool mysql_rename_tables(THD *thd, TABLE_LIST *table_list, bool silent)
                        MYSQL_OPEN_SKIP_TEMPORARY))
     goto err;
 
-  mysql_mutex_lock(&LOCK_open);
-
   for (ren_table= table_list; ren_table; ren_table= ren_table->next_local)
     tdc_remove_table(thd, TDC_RT_REMOVE_ALL, ren_table->db,
-                     ren_table->table_name);
+                     ren_table->table_name, FALSE);
 
   error=0;
+  /*
+    An exclusive lock on table names is satisfactory to ensure
+    no other thread accesses this table.
+  */
   if ((ren_table=rename_tables(thd,table_list,0)))
   {
     /* Rename didn't succeed;  rename back the tables in reverse order */
@@ -175,17 +177,6 @@ bool mysql_rename_tables(THD *thd, TABLE_LIST *table_list, bool silent)
 
     error= 1;
   }
-  /*
-    An exclusive lock on table names is satisfactory to ensure
-    no other thread accesses this table.
-    However, NDB assumes that handler::rename_tables is called under
-    LOCK_open. And it indeed is, from ALTER TABLE.
-    TODO: remove this limitation.
-    We still should unlock LOCK_open as early as possible, to provide
-    higher concurrency - query_cache_invalidate can take minutes to
-    complete.
-  */
-  mysql_mutex_unlock(&LOCK_open);
 
   if (!silent && !error)
   {
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index 4d58db2e36c..139568bb9d5 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -7657,7 +7657,7 @@ static bool show_create_trigger_impl(THD *thd,
 */
 
 static
-TABLE_LIST *get_trigger_table_impl(THD *thd, const sp_name *trg_name)
+TABLE_LIST *get_trigger_table(THD *thd, const sp_name *trg_name)
 {
   char trn_path_buff[FN_REFLEN];
   LEX_STRING trn_path= { trn_path_buff, 0 };
@@ -7694,39 +7694,6 @@ TABLE_LIST *get_trigger_table_impl(THD *thd, const sp_name *trg_name)
   return table;
 }
 
-/**
-  Read TRN and TRG files to obtain base table name for the specified
-  trigger name and construct TABE_LIST object for the base table. Acquire
-  LOCK_open when doing this.
-
-  @param thd      Thread context.
-  @param trg_name Trigger name.
-
-  @return TABLE_LIST object corresponding to the base table.
-*/
-
-static
-TABLE_LIST *get_trigger_table(THD *thd, const sp_name *trg_name)
-{
-  /* Acquire LOCK_open (stop the server). */
-
-  mysql_mutex_lock(&LOCK_open);
-
-  /*
-    Load base table name from the TRN-file and create TABLE_LIST object.
-  */
-
-  TABLE_LIST *lst= get_trigger_table_impl(thd, trg_name);
-
-  /* Release LOCK_open (continue the server). */
-
-  mysql_mutex_unlock(&LOCK_open);
-
-  /* That's it. */
-
-  return lst;
-}
-
 
 /**
   SHOW CREATE TRIGGER high-level implementation.
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 02a874ce62f..3f777411103 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -1723,7 +1723,6 @@ bool mysql_write_frm(ALTER_PARTITION_PARAM_TYPE *lpt, uint flags)
       completing this we write a new phase to the log entry that will
       deactivate it.
     */
-    mysql_mutex_lock(&LOCK_open);
     if (mysql_file_delete(key_file_frm, frm_name, MYF(MY_WME)) ||
 #ifdef WITH_PARTITION_STORAGE_ENGINE
         lpt->table->file->ha_create_handler_files(path, shadow_path,
@@ -1779,7 +1778,6 @@ bool mysql_write_frm(ALTER_PARTITION_PARAM_TYPE *lpt, uint flags)
 #endif
 
 err:
-    mysql_mutex_unlock(&LOCK_open);
 #ifdef WITH_PARTITION_STORAGE_ENGINE
     deactivate_ddl_log_entry(part_info->frm_log_entry->entry_pos);
     part_info->frm_log_entry= NULL;
@@ -1955,10 +1953,11 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
       if (lock_table_names(thd, tables, NULL, thd->variables.lock_wait_timeout,
                            MYSQL_OPEN_SKIP_TEMPORARY))
         DBUG_RETURN(1);
-      mysql_mutex_lock(&LOCK_open);
       for (table= tables; table; table= table->next_local)
-        tdc_remove_table(thd, TDC_RT_REMOVE_ALL, table->db, table->table_name);
-      mysql_mutex_unlock(&LOCK_open);
+      {
+        tdc_remove_table(thd, TDC_RT_REMOVE_ALL, table->db, table->table_name,
+                         FALSE);
+      }
     }
     else
     {
@@ -2104,14 +2103,9 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
                                         table->internal_tmp_table ?
                                         FN_IS_TMP : 0);
     }
-    /*
-      TODO: Investigate what should be done to remove this lock completely.
-            Is exclusive meta-data lock enough ?
-    */
     DEBUG_SYNC(thd, "rm_table_part2_before_delete_table");
     DBUG_EXECUTE_IF("sleep_before_part2_delete_table",
                     my_sleep(100000););
-    mysql_mutex_lock(&LOCK_open);
     if (drop_temporary ||
         ((access(path, F_OK) &&
           ha_create_table_from_engine(thd, db, alias)) ||
@@ -2131,8 +2125,7 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
       char *end;
       /*
         Cannot use the db_type from the table, since that might have changed
-        while waiting for the exclusive name lock. We are under LOCK_open,
-        so reading from the frm-file is safe.
+        while waiting for the exclusive name lock.
       */
       if (frm_db_type == DB_TYPE_UNKNOWN)
       {
@@ -2173,7 +2166,6 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
         error|= new_error;
       }
     }
-    mysql_mutex_unlock(&LOCK_open);
     if (error)
     {
       if (wrong_tables.length())
@@ -4053,7 +4045,6 @@ bool mysql_create_table_no_lock(THD *thd,
     goto err;
   }
 
-  mysql_mutex_lock(&LOCK_open);
   if (!internal_tmp_table && !(create_info->options & HA_LEX_CREATE_TMP_TABLE))
   {
     if (!access(path,F_OK))
@@ -4061,7 +4052,7 @@ bool mysql_create_table_no_lock(THD *thd,
       if (create_info->options & HA_LEX_CREATE_IF_NOT_EXISTS)
         goto warn;
       my_error(ER_TABLE_EXISTS_ERROR,MYF(0),table_name);
-      goto unlock_and_end;
+      goto err;
     }
     /*
       We don't assert here, but check the result, because the table could be
@@ -4071,11 +4062,14 @@ bool mysql_create_table_no_lock(THD *thd,
       Then she could create the table. This case is pretty obscure and
       therefore we don't introduce a new error message only for it.
     */
+    mysql_mutex_lock(&LOCK_open);
     if (get_cached_table_share(db, table_name))
     {
+      mysql_mutex_unlock(&LOCK_open);
       my_error(ER_TABLE_EXISTS_ERROR, MYF(0), table_name);
-      goto unlock_and_end;
+      goto err;
     }
+    mysql_mutex_unlock(&LOCK_open);
   }
 
   /*
@@ -4083,7 +4077,7 @@ bool mysql_create_table_no_lock(THD *thd,
     exist in any storage engine. In such a case it should
     be discovered and the error ER_TABLE_EXISTS_ERROR be returned
     unless user specified CREATE TABLE IF EXISTS
-    The LOCK_open mutex has been locked to make sure no
+    An exclusive metadata lock ensures that no
     one else is attempting to discover the table. Since
     it's not on disk as a frm file, no one could be using it!
   */
@@ -4104,12 +4098,12 @@ bool mysql_create_table_no_lock(THD *thd,
         if (create_if_not_exists)
           goto warn;
         my_error(ER_TABLE_EXISTS_ERROR,MYF(0),table_name);
-        goto unlock_and_end;
+        goto err;
         break;
       default:
         DBUG_PRINT("info", ("error: %u from storage engine", retcode));
         my_error(retcode, MYF(0),table_name);
-        goto unlock_and_end;
+        goto err;
     }
   }
 
@@ -4142,7 +4136,7 @@ bool mysql_create_table_no_lock(THD *thd,
       if (test_if_data_home_dir(dirpath))
       {
         my_error(ER_WRONG_ARGUMENTS, MYF(0), "DATA DIRECTORY");
-        goto unlock_and_end;
+        goto err;
       }
     }
     if (create_info->index_file_name)
@@ -4151,7 +4145,7 @@ bool mysql_create_table_no_lock(THD *thd,
       if (test_if_data_home_dir(dirpath))
       {
         my_error(ER_WRONG_ARGUMENTS, MYF(0), "INDEX DIRECTORY");
-        goto unlock_and_end;
+        goto err;
       }
     }
   }
@@ -4159,7 +4153,7 @@ bool mysql_create_table_no_lock(THD *thd,
 #ifdef WITH_PARTITION_STORAGE_ENGINE
   if (check_partition_dirs(thd->lex->part_info))
   {
-    goto unlock_and_end;
+    goto err;
   }
 #endif /* WITH_PARTITION_STORAGE_ENGINE */
 
@@ -4182,7 +4176,7 @@ bool mysql_create_table_no_lock(THD *thd,
   if (rea_create_table(thd, path, db, table_name,
                        create_info, alter_info->create_list,
                        key_count, key_info_buffer, file))
-    goto unlock_and_end;
+    goto err;
 
   if (create_info->options & HA_LEX_CREATE_TMP_TABLE)
   {
@@ -4190,15 +4184,12 @@ bool mysql_create_table_no_lock(THD *thd,
     if (!(open_temporary_table(thd, path, db, table_name, 1)))
     {
       (void) rm_temporary_table(create_info->db_type, path);
-      goto unlock_and_end;
+      goto err;
     }
     thd->thread_specific_used= TRUE;
   }
 
   error= FALSE;
-unlock_and_end:
-  mysql_mutex_unlock(&LOCK_open);
-
 err:
   thd_proc_info(thd, "After create");
   delete file;
@@ -4210,7 +4201,7 @@ warn:
                       ER_TABLE_EXISTS_ERROR, ER(ER_TABLE_EXISTS_ERROR),
                       alias);
   create_info->table_existed= 1;		// Mark that table existed
-  goto unlock_and_end;
+  goto err;
 }
 
 
@@ -4459,20 +4450,19 @@ static int prepare_for_repair(THD *thd, TABLE_LIST *table_list,
 
     hash_value= my_calc_hash(&table_def_cache, (uchar*) key, key_length);
     mysql_mutex_lock(&LOCK_open);
-    if (!(share= (get_table_share(thd, table_list, key, key_length, 0,
-                                  &error, hash_value))))
-    {
-      mysql_mutex_unlock(&LOCK_open);
+    share= get_table_share(thd, table_list, key, key_length, 0,
+                           &error, hash_value);
+    mysql_mutex_unlock(&LOCK_open);
+    if (share == NULL)
       DBUG_RETURN(0);				// Can't open frm file
-    }
 
     if (open_table_from_share(thd, share, "", 0, 0, 0, &tmp_table, FALSE))
     {
+      mysql_mutex_lock(&LOCK_open);
       release_table_share(share);
       mysql_mutex_unlock(&LOCK_open);
       DBUG_RETURN(0);                           // Out of memory
     }
-    mysql_mutex_unlock(&LOCK_open);
     table= &tmp_table;
   }
 
@@ -5136,10 +5126,8 @@ send_result_message:
       }
       else if (open_for_modify || fatal_error)
       {
-        mysql_mutex_lock(&LOCK_open);
         tdc_remove_table(thd, TDC_RT_REMOVE_UNUSED,
-                         table->db, table->table_name);
-        mysql_mutex_unlock(&LOCK_open);
+                         table->db, table->table_name, FALSE);
         /*
           May be something modified. Consequently, we have to
           invalidate the query cache.
@@ -6775,7 +6763,6 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
       else
       {
         *fn_ext(new_name)=0;
-        mysql_mutex_lock(&LOCK_open);
         if (mysql_rename_table(old_db_type,db,table_name,new_db,new_alias, 0))
           error= -1;
         else if (Table_triggers_list::change_table_name(thd, db, table_name,
@@ -6785,7 +6772,6 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
                                     table_name, 0);
           error= -1;
         }
-        mysql_mutex_unlock(&LOCK_open);
       }
     }
 
@@ -7404,7 +7390,6 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
     /* This type cannot happen in regular ALTER. */
     new_db_type= old_db_type= NULL;
   }
-  mysql_mutex_lock(&LOCK_open);
   if (mysql_rename_table(old_db_type, db, table_name, db, old_name,
                          FN_TO_IS_TMP))
   {
@@ -7431,8 +7416,6 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
   if (! error)
     (void) quick_rm_table(old_db_type, db, old_name, FN_IS_TMP);
 
-  mysql_mutex_unlock(&LOCK_open);
-
   if (error)
   {
     /* This shouldn't happen. But let us play it safe. */
diff --git a/sql/sql_trigger.cc b/sql/sql_trigger.cc
index a5664b00287..b81461e8371 100644
--- a/sql/sql_trigger.cc
+++ b/sql/sql_trigger.cc
@@ -394,9 +394,7 @@ bool mysql_create_or_drop_trigger(THD *thd, TABLE_LIST *tables, bool create)
   /*
     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. (Acquiring
-    LOCK_open is not enough because global read lock is held without holding
-    LOCK_open).
+    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))
@@ -516,11 +514,9 @@ bool mysql_create_or_drop_trigger(THD *thd, TABLE_LIST *tables, bool create)
       goto end;
   }
 
-  mysql_mutex_lock(&LOCK_open);
   result= (create ?
            table->triggers->create_trigger(thd, tables, &stmt_query):
            table->triggers->drop_trigger(thd, tables, &stmt_query));
-  mysql_mutex_unlock(&LOCK_open);
 
   if (result)
     goto end;
@@ -1680,9 +1676,6 @@ bool add_table_for_trigger(THD *thd,
   @param db       schema for table
   @param name     name for table
 
-  @note
-    The calling thread should hold the LOCK_open mutex;
-
   @retval
     False   success
   @retval
@@ -1912,14 +1905,10 @@ bool Table_triggers_list::change_table_name(THD *thd, const char *db,
 
   /*
     This method interfaces the mysql server code protected by
-    either LOCK_open mutex or with an exclusive metadata lock.
-    In the future, only an exclusive metadata lock will be enough.
+    an exclusive metadata lock.
   */
-#ifndef DBUG_OFF
-  if (thd->mdl_context.is_lock_owner(MDL_key::TABLE, db, old_table,
-                                     MDL_EXCLUSIVE))
-    mysql_mutex_assert_owner(&LOCK_open);
-#endif
+  DBUG_ASSERT(thd->mdl_context.is_lock_owner(MDL_key::TABLE, db, old_table,
+                                             MDL_EXCLUSIVE));
 
   DBUG_ASSERT(my_strcasecmp(table_alias_charset, db, new_db) ||
               my_strcasecmp(table_alias_charset, old_table, new_table));
diff --git a/sql/sql_truncate.cc b/sql/sql_truncate.cc
index ee5c707cd69..38e32082388 100644
--- a/sql/sql_truncate.cc
+++ b/sql/sql_truncate.cc
@@ -310,10 +310,8 @@ static bool open_and_lock_table_for_truncate(THD *thd, TABLE_LIST *table_ref,
           upgrade_shared_lock_to_exclusive(table_ref->mdl_request.ticket,
                                            timeout))
         DBUG_RETURN(TRUE);
-      mysql_mutex_lock(&LOCK_open);
       tdc_remove_table(thd, TDC_RT_REMOVE_ALL, table_ref->db,
-                       table_ref->table_name);
-      mysql_mutex_unlock(&LOCK_open);
+                       table_ref->table_name, FALSE);
     }
   }
   else
diff --git a/sql/sql_view.cc b/sql/sql_view.cc
index be13349b5a1..b6671d9096b 100644
--- a/sql/sql_view.cc
+++ b/sql/sql_view.cc
@@ -658,7 +658,6 @@ bool mysql_create_view(THD *thd, TABLE_LIST *views,
     goto err;
   }
 
-  mysql_mutex_lock(&LOCK_open);
   res= mysql_register_view(thd, view, mode);
 
   if (mysql_bin_log.is_open())
@@ -705,7 +704,6 @@ bool mysql_create_view(THD *thd, TABLE_LIST *views,
       res= TRUE;
   }
 
-  mysql_mutex_unlock(&LOCK_open);
   if (mode != VIEW_CREATE_NEW)
     query_cache_invalidate3(thd, view, 0);
   thd->global_read_lock.start_waiting_global_read_lock(thd);
@@ -1656,10 +1654,8 @@ bool mysql_drop_view(THD *thd, TABLE_LIST *views, enum_drop_mode drop_mode)
                        MYSQL_OPEN_SKIP_TEMPORARY))
     DBUG_RETURN(TRUE);
 
-  mysql_mutex_lock(&LOCK_open);
   for (view= views; view; view= view->next_local)
   {
-    TABLE_SHARE *share;
     frm_type_enum type= FRMTYPE_ERROR;
     build_table_filename(path, sizeof(path) - 1,
                          view->db, view->table_name, reg_ext, 0);
@@ -1698,16 +1694,12 @@ bool mysql_drop_view(THD *thd, TABLE_LIST *views, enum_drop_mode drop_mode)
     some_views_deleted= TRUE;
 
     /*
-      For a view, there is only one table_share object which should never
-      be used outside of LOCK_open
+      For a view, there is a TABLE_SHARE object, but its
+      ref_count never goes above 1. Remove it from the table
+      definition cache, in case the view was cached.
     */
-    if ((share= get_cached_table_share(view->db, view->table_name)))
-    {
-      DBUG_ASSERT(share->ref_count == 0);
-      share->ref_count++;
-      share->version= 0;
-      release_table_share(share);
-    }
+    tdc_remove_table(thd, TDC_RT_REMOVE_ALL, view->db, view->table_name,
+                     FALSE);
     query_cache_invalidate3(thd, view, 0);
     sp_cache_invalidate();
   }
@@ -1732,8 +1724,6 @@ bool mysql_drop_view(THD *thd, TABLE_LIST *views, enum_drop_mode drop_mode)
       something_wrong= 1;
   }
 
-  mysql_mutex_unlock(&LOCK_open);
-  
   if (something_wrong)
   {
     DBUG_RETURN(TRUE);
diff --git a/sql/sys_vars.cc b/sql/sys_vars.cc
index 320e6d9253e..2285065aa13 100644
--- a/sql/sys_vars.cc
+++ b/sql/sys_vars.cc
@@ -1492,7 +1492,7 @@ static bool fix_read_only(sys_var *self, THD *thd, enum_var_type type)
     can cause to wait on a read lock, it's required for the client application
     to unlock everything, and acceptable for the server to wait on all locks.
   */
-  if ((result= close_cached_tables(thd, NULL, FALSE, TRUE)))
+  if ((result= close_cached_tables(thd, NULL, TRUE)))
     goto end_with_read_lock;
 
   if ((result= thd->global_read_lock.make_global_read_lock_block_commit(thd)))