Fix for:

  Bug #20662 "Infinite loop in CREATE TABLE IF NOT EXISTS ... SELECT
              with locked tables"
  Bug #20903 "Crash when using CREATE TABLE .. SELECT and triggers"
  Bug #24738 "CREATE TABLE ... SELECT is not isolated properly"
  Bug #24508 "Inconsistent results of CREATE TABLE ... SELECT when
              temporary table exists"
 
Deadlock occured when one tried to execute CREATE TABLE IF NOT
EXISTS ... SELECT statement under LOCK TABLES which held
read lock on target table.
Attempt to execute the same statement for already existing
target table with triggers caused server crashes.
Also concurrent execution of CREATE TABLE ... SELECT statement
and other statements involving target table suffered from
various races (some of which might've led to deadlocks).
Finally, attempt to execute CREATE TABLE ... SELECT in case
when a temporary table with same name was already present
led to the insertion of data into this temporary table and
creation of empty non-temporary table.
 
All above problems stemmed from the old implementation of CREATE
TABLE ... SELECT in which we created, opened and locked target
table without any special protection in a separate step and not
with the rest of tables used by this statement.
This underminded deadlock-avoidance approach used in server
and created window for races. It also excluded target table
from prelocking causing problems with trigger execution.
  
The patch solves these problems by implementing new approach to
handling of CREATE TABLE ... SELECT for base tables.
We try to open and lock table to be created at the same time as
the rest of tables used by this statement. If such table does not
exist at this moment we create and place in the table cache special
placeholder for it which prevents its creation or any other usage
by other threads.

We still use old approach for creation of temporary tables.

Also note that we decided to postpone introduction of some tests
for concurrent behaviour of CREATE TABLE ... SELECT till 5.1.
The main reason for this is absence in 5.0 ability to set @@debug
variable at runtime, which can be circumvented only by using several
test files with individual .opt files. Since the latter is likely
to slowdown test-suite unnecessary we chose not to push this tests
into 5.0, but run them manually for this version and later push
their optimized version into 5.1


mysql-test/r/create.result:
  Extended test coverage for CREATE TABLE ... SELECT. In particular added
  tests for bug #24508 "Inconsistent results of CREATE TABLE ... SELECT
  when temporary table exists" and bug #20662 "Infinite loop in CREATE
  TABLE IF NOT EXISTS ... SELECT with locked tables".
mysql-test/r/trigger.result:
  Added test case for bug #20903 "Crash when using CREATE TABLE .. SELECT
  and triggers"
mysql-test/t/create.test:
  Extended test coverage for CREATE TABLE ... SELECT. In particular added
  tests for bug #24508 "Inconsistent results of CREATE TABLE ... SELECT
  when temporary table exists" and bug #20662 "Infinite loop in CREATE
  TABLE IF NOT EXISTS ... SELECT with locked tables".
mysql-test/t/trigger.test:
  Added test case for bug #20903 "Crash when using CREATE TABLE .. SELECT
  and triggers"
sql/lock.cc:
  Now for creation of name-lock placeholder in lock_table_name() we use
  auxiliary function table_cache_insert_placeholder().
sql/mysql_priv.h:
  Made build_table_path() function available outside of sql_table.cc file.
  reopen_name_locked_table() now has 3rd argument which controls linking
  in of table being opened into THD::open_tables (this is useful in
  cases when placeholder used for name-locking is already linked into
  this list).
  Added declaration of auxiliary function table_cache_insert_placeholder()
  which is used for creation of table placeholders for name-locking.
  Added declaration of table_cache_has_open_placeholder() function which
  can be used for checking if table cache contains an open placeholder for
  the table and if this placeholder was created by another thread.
  (This function is needed only in 5.0 where we use it in various versions
   of CREATE TABLE in order to protect it from concurrent CREATE TABLE
   ... SELECT operations for the table. Starting from 5.1 we use different
   approach so it is going to be removed there).
  Made close_old_data_files() static within sql_base.cc file. 
  Added auxiliary drop_open_table() routine.
  Moved declaration of refresh_version to table.h header to make it
  accessible from inline methods of TABLE class.
  MYSQL_OPEN_IGNORE_LOCKED_TABLES flag is no longer used. Instead
  MYSQL_OPEN_TEMPORARY_ONLY option was added.
sql/sql_base.cc:
  Added support for the new approach to the handling of CREATE TABLE
  ... SELECT for base tables.
  
  Now we try to open and lock table to be created at the same time as
  the rest of tables used by this statement. If such table does not
  exist at this moment we create and place in the table cache special
  placeholder for it which prevents its creation or any other usage
  by other threads.
  
  Note significant distinctions of this placeholder from the placeholder
  used for normal name-lock: 1) It is treated like open table by other
  name-locks so it does not allow name-lock taking operations like DROP
  TABLE or RENAME TABLE to proceed. 2) it is linked into THD::open_tables
  list and automatically removed during close_thread_tables() call.
  
  open_tables():
    Implemented logic described above. To do this added
    auxiliary check_if_table_exists() function.
    Removed support for MYSQL_OPEN_IGNORE_LOCKED_TABLES option
    which is no longer used.
    Added MYSQL_OPEN_TEMPORARY_ONLY which is used to restrict
    search for temporary tables only.
  close_cached_tables()/close_thread_table()/reopen_tables()/
  close_old_data_files()/table_is_used()/remove_table_from_cache():
    Added support for open placeholders (note that we also use them
    when we need to re-open tables during flush).
  Added auxiliary drop_open_table() routine.
  reopen_name_locked_table():
    Now has 3rd argument which controls linking in of table being
    opened into THD::open_tables (this is useful in cases when
    placeholder used for name-locking is already linked into
    this list).
  Added auxiliary table_cache_insert_placeholder() routine which
  simplifies creation of placeholders used for name-locking.
  Added table_cache_has_open_placeholder() function which can be
  used for checking if table cache contains an open placeholder for
  the table and if this placeholder was created by another thread.
  (This function is needed only in 5.0 where we use it in various versions
   of CREATE TABLE in order to protect it from concurrent CREATE TABLE
   ... SELECT operations for the table. Starting from 5.1 we use different
   approach so it is going to be removed there).
sql/sql_handler.cc:
  Adjusted mysql_ha_mark_tables_for_reopen() routine to properly
  handle placeholders which now can be linked into open tables
  list.
sql/sql_insert.cc:
  Introduced new approach to handling of base tables in CREATE TABLE
  ... SELECT statement.
  
  Now we try to open and lock table to be created at the same time as
  the rest of tables used by this statement. If such table does not
  exist at this moment we create and place in the table cache special
  placeholder for it which prevents its creation or any other usage
  by other threads. By doing this we avoid races which existed with
  previous approach in which we created, opened and locked target in
  separate step without any special protection.
  This also allows properly calculate prelocking set in cases when
  target table already exists and has some on insert triggers.
    
  Note that we don't employ the same approach for temporary tables
  (this is okay as such tables are unaffected by other threads).
  
  Changed create_table_from_items() and select_create methods to
  implement this approach.
sql/sql_parse.cc:
  The new approach to handling of CREATE TABLE ... SELECT for
  base tables assumes that all tables (including table to be
  created) are opened and (or) locked at the same time.
  So in cases when we create base table we have to pass to
  open_and_lock_tables() table list which includes target table.
sql/sql_prepare.cc:
  The new approach to handling of CREATE TABLE ... SELECT for
  base tables assumes that all tables (including table to be
  created) are opened and (or) locked at the same time.
  So in cases when we create base table we have to pass to
  open_and_lock_tables() table list which includes target table.
sql/sql_table.cc:
  Now mysql_create_table_internal(), mysql_create_like_table() and
  mysql_alter_table() not only check that destination table doesn't
  exist on disk but also check that there is no create placeholder
  in table cache for it (i.e. there is no CREATE TABLE ... SELECT
  operation in progress for it). Note that starting from 5.1 we
  use different approach in order to to protect CREATE TABLE ... SELECT
  from concurrent CREATE TABLE (ALTER TABLE ... RENAME) operations,
  the latter simply take name-locks on table before its creation
  (on target table name before renaming).
  
  Also made build_table_path() available from other files and
  asjusted calls to reopen_name_locked_table(), which now takes
  extra argument, which controls linking of open table into
  THD::open_tables list.
sql/sql_trigger.cc:
  reopen_name_locked_tables() now has one more argument which controls
  linking of opened table into the THD::open_tables list.
sql/sql_yacc.yy:
  The new approach to handling of CREATE TABLE ... SELECT statement
  for base tables assumes that all tables including table to be
  created are open and (or) locked at the same time. Therefore
  we need to set correct lock for target table.
sql/table.h:
  Moved declaration of refresh_version variable from mysql_priv.h
  to make it accessible from inline methods of TABLE class. 
  Renamed TABLE::locked_by_flush member to open_placeholder since
  now it is also used for taking exclusive name-lock and not only
  by flush. 
  Introduced TABLE::is_name_opened() helper method which can be used
  to distinguish TABLE instances corresponding to open tables or
  placeholders for them from closed instances (e.g. due to their old
  version). Also introduced TABLE::needs_reopen_or_name_lock() helper
  which allows to check if TABLE instance corresponds to outdated
  version of table or to name-lock placeholder.
  Introduced TABLE_LIST::create member which marks elements of
  table list corresponds to the table to be created.
  Adjusted TABLE_LIST::placeholder() method to take into account 
  name-lock placeholders for tables to be created (this, for example,
  allows to properly handle such placeholders in lock_tables()).

11 files changed, 653 insertions, 184 deletions

diff --git a/sql/lock.cc b/sql/lock.cc
index 233d12d9cc4..9298b33b4d2 100644
--- a/sql/lock.cc
+++ b/sql/lock.cc
@@ -853,7 +853,6 @@ int lock_table_name(THD *thd, TABLE_LIST *table_list)
   TABLE *table;
   char  key[MAX_DBKEY_LENGTH];
   char *db= table_list->db;
-  int table_in_key_offset;
   uint  key_length;
   HASH_SEARCH_STATE state;
   DBUG_ENTER("lock_table_name");
@@ -861,10 +860,8 @@ int lock_table_name(THD *thd, TABLE_LIST *table_list)
 
   safe_mutex_assert_owner(&LOCK_open);
 
-  table_in_key_offset= strmov(key, db) - key + 1;
-  key_length= (uint)(strmov(key + table_in_key_offset, table_list->table_name)
-                     - key) + 1;
-
+  key_length= (uint)(strmov(strmov(key, db) + 1, table_list->table_name) -
+                     key) + 1;
 
   /* Only insert the table if we haven't insert it already */
   for (table=(TABLE*) hash_first(&open_cache, (byte*)key, key_length, &state);
@@ -873,29 +870,11 @@ int lock_table_name(THD *thd, TABLE_LIST *table_list)
     if (table->in_use == thd)
       DBUG_RETURN(0);
 
-  /*
-    Create a table entry with the right key and with an old refresh version
-    Note that we must use my_malloc() here as this is freed by the table
-    cache
-  */
-  if (!(table= (TABLE*) my_malloc(sizeof(*table)+key_length,
-				  MYF(MY_WME | MY_ZEROFILL))))
-    DBUG_RETURN(-1);
-  table->s= &table->share_not_to_be_used;
-  memcpy((table->s->table_cache_key= (char*) (table+1)), key, key_length);
-  table->s->db= table->s->table_cache_key;
-  table->s->table_name= table->s->table_cache_key + table_in_key_offset;
-  table->s->key_length=key_length;
-  table->in_use=thd;
-  table->locked_by_name=1;
-  table_list->table=table;
-
-  if (my_hash_insert(&open_cache, (byte*) table))
-  {
-    my_free((gptr) table,MYF(0));
+  if (!(table= table_cache_insert_placeholder(thd, key, key_length)))
     DBUG_RETURN(-1);
-  }
-  
+
+  table_list->table= table;
+
   /* Return 1 if table is in use */
   DBUG_RETURN(test(remove_table_from_cache(thd, db, table_list->table_name,
                                            RTFC_NO_FLAG)));
diff --git a/sql/mysql_priv.h b/sql/mysql_priv.h
index 868f3584448..37a5fc31f1d 100644
--- a/sql/mysql_priv.h
+++ b/sql/mysql_priv.h
@@ -674,6 +674,8 @@ struct Query_cache_query_flags
 #define query_cache_invalidate_by_MyISAM_filename_ref NULL
 #endif /*HAVE_QUERY_CACHE*/
 
+uint build_table_path(char *buff, size_t bufflen, const char *db,
+                      const char *table, const char *ext);
 bool mysql_create_db(THD *thd, char *db, HA_CREATE_INFO *create, bool silent);
 bool mysql_alter_db(THD *thd, const char *db, HA_CREATE_INFO *create);
 bool mysql_rm_db(THD *thd,char *db,bool if_exists, bool silent);
@@ -856,12 +858,14 @@ bool mysql_create_or_drop_trigger(THD *thd, TABLE_LIST *tables, bool create);
 TABLE *open_ltable(THD *thd, TABLE_LIST *table_list, thr_lock_type update);
 TABLE *open_table(THD *thd, TABLE_LIST *table_list, MEM_ROOT* mem,
 		  bool *refresh, uint flags);
-bool reopen_name_locked_table(THD* thd, TABLE_LIST* table);
+bool reopen_name_locked_table(THD* thd, TABLE_LIST* table_list, bool link_in);
+TABLE *table_cache_insert_placeholder(THD *thd, const char *key,
+                                      uint key_length);
+bool table_cache_has_open_placeholder(THD *thd, const char *db,
+                                      const char *table_name);
 TABLE *find_locked_table(THD *thd, const char *db,const char *table_name);
 bool reopen_table(TABLE *table,bool locked);
 bool reopen_tables(THD *thd,bool get_locks,bool in_refresh);
-void close_old_data_files(THD *thd, TABLE *table, bool abort_locks,
-			  bool send_refresh);
 bool close_data_tables(THD *thd,const char *db, const char *table_name);
 bool wait_for_tables(THD *thd);
 bool table_is_used(TABLE *table, bool wait_for_name_lock);
@@ -1017,6 +1021,8 @@ void add_join_natural(TABLE_LIST *a,TABLE_LIST *b,List<String> *using_fields,
                       SELECT_LEX *lex);
 bool add_proc_to_list(THD *thd, Item *item);
 TABLE *unlink_open_table(THD *thd,TABLE *list,TABLE *find);
+void drop_open_table(THD *thd, TABLE *table, const char *db_name,
+                     const char *table_name);
 void update_non_unique_table_error(TABLE_LIST *update,
                                    const char *operation,
                                    TABLE_LIST *duplicate);
@@ -1269,7 +1275,7 @@ extern double log_01[32];
 extern ulonglong log_10_int[20];
 extern ulonglong keybuff_size;
 extern ulonglong thd_startup_options;
-extern ulong refresh_version,flush_version, thread_id;
+extern ulong flush_version, thread_id;
 extern ulong binlog_cache_use, binlog_cache_disk_use;
 extern ulong aborted_threads,aborted_connects;
 extern ulong delayed_insert_timeout;
@@ -1443,7 +1449,7 @@ MYSQL_LOCK *mysql_lock_tables(THD *thd, TABLE **table, uint count,
 #define MYSQL_LOCK_IGNORE_GLOBAL_READ_LOCK      0x0001
 #define MYSQL_LOCK_IGNORE_FLUSH                 0x0002
 #define MYSQL_LOCK_NOTIFY_IF_NEED_REOPEN        0x0004
-#define MYSQL_OPEN_IGNORE_LOCKED_TABLES         0x0008
+#define MYSQL_OPEN_TEMPORARY_ONLY               0x0008
 
 void mysql_unlock_tables(THD *thd, MYSQL_LOCK *sql_lock);
 void mysql_unlock_read_tables(THD *thd, MYSQL_LOCK *sql_lock);
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index 6e6611d54d2..db4491a19ae 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -94,6 +94,8 @@ static bool open_new_frm(THD *thd, const char *path, const char *alias,
                          uint db_stat, uint prgflag,
                          uint ha_open_flags, TABLE *outparam,
                          TABLE_LIST *table_desc, MEM_ROOT *mem_root);
+static void close_old_data_files(THD *thd, TABLE *table, bool morph_locks,
+                                 bool send_refresh);
 
 extern "C" byte *table_cache_key(const byte *record,uint *length,
 				 my_bool not_used __attribute__((unused)))
@@ -374,7 +376,21 @@ bool close_cached_tables(THD *thd, bool if_wait_for_refresh,
       for (uint idx=0 ; idx < open_cache.records ; idx++)
       {
 	TABLE *table=(TABLE*) hash_element(&open_cache,idx);
-	if ((table->s->version) < refresh_version && table->db_stat)
+        /*
+          Note that we wait here only for tables which are actually open, and
+          not for placeholders with TABLE::open_placeholder set. Waiting for
+          latter will cause deadlock in the following scenario, for example:
+
+          conn1: lock table t1 write;
+          conn2: lock table t2 write;
+          conn1: flush tables;
+          conn2: flush tables;
+
+          It also does not make sense to wait for those of placeholders that
+          are employed by CREATE TABLE as in this case table simply does not
+          exist yet.
+        */
+	if (table->needs_reopen_or_name_lock() && table->db_stat)
 	{
 	  found=1;
           DBUG_PRINT("signal", ("Waiting for COND_refresh"));
@@ -616,10 +632,10 @@ bool close_thread_table(THD *thd, TABLE **table_ptr)
   TABLE *table= *table_ptr;
   DBUG_ENTER("close_thread_table");
   DBUG_ASSERT(table->key_read == 0);
-  DBUG_ASSERT(table->file->inited == handler::NONE);
+  DBUG_ASSERT(!table->file || table->file->inited == handler::NONE);
 
   *table_ptr=table->next;
-  if (table->s->version != refresh_version ||
+  if (table->needs_reopen_or_name_lock() ||
       thd->version != refresh_version || !table->db_stat)
   {
     VOID(hash_delete(&open_cache,(byte*) table));
@@ -627,6 +643,12 @@ bool close_thread_table(THD *thd, TABLE **table_ptr)
   }
   else
   {
+    /*
+      Open placeholders have TABLE::db_stat set to 0, so they should be
+      handled by the first alternative.
+    */
+    DBUG_ASSERT(!table->open_placeholder);
+
     if (table->s->flush_version != flush_version)
     {
       table->s->flush_version= flush_version;
@@ -1114,6 +1136,43 @@ TABLE *unlink_open_table(THD *thd, TABLE *list, TABLE *find)
 }
 
 
+/**
+    @brief Auxiliary routine which closes and drops open table.
+
+    @param  thd         Thread handle
+    @param  table       TABLE object for table to be dropped
+    @param  db_name     Name of database for this table
+    @param  table_name  Name of this table
+
+    @note This routine assumes that table to be closed is open only
+          by calling thread so we needn't wait until other threads
+          will close the table. Also unless called under implicit or
+          explicit LOCK TABLES mode it assumes that table to be
+          dropped is already unlocked. In the former case it will
+          also remove lock on the table. But one should not rely on
+          this behaviour as it may change in future.
+*/
+
+void drop_open_table(THD *thd, TABLE *table, const char *db_name,
+                     const char *table_name)
+{
+  if (table->s->tmp_table)
+    close_temporary_table(thd, db_name, table_name);
+  else
+  {
+    enum db_type table_type= table->s->db_type;
+    VOID(pthread_mutex_lock(&LOCK_open));
+    /*
+      unlink_open_table() also tells threads waiting for refresh or close
+      that something has happened.
+    */
+    thd->open_tables= unlink_open_table(thd, thd->open_tables, table);
+    quick_rm_table(table_type, db_name, table_name);
+    VOID(pthread_mutex_unlock(&LOCK_open));
+  }
+}
+
+
 /*
    When we call the following function we must have a lock on
    LOCK_open ; This lock will be unlocked on return.
@@ -1152,6 +1211,11 @@ void wait_for_refresh(THD *thd)
       table_list  TABLE_LIST object for table to be open, TABLE_LIST::table
                   member should point to TABLE object which was used for
                   name-locking.
+      link_in     TRUE  - if TABLE object for table to be opened should be
+                          linked into THD::open_tables list.
+                  FALSE - placeholder used for name-locking is already in
+                          this list so we only need to preserve TABLE::next
+                          pointer.
 
   NOTE
     This function assumes that its caller already acquired LOCK_open mutex.
@@ -1161,7 +1225,7 @@ void wait_for_refresh(THD *thd)
     TRUE  - Error
 */
 
-bool reopen_name_locked_table(THD* thd, TABLE_LIST* table_list)
+bool reopen_name_locked_table(THD* thd, TABLE_LIST* table_list, bool link_in)
 {
   TABLE *table= table_list->table;
   TABLE_SHARE *share;
@@ -1199,12 +1263,33 @@ bool reopen_name_locked_table(THD* thd, TABLE_LIST* table_list)
   share= table->s;
   share->db= share->table_cache_key;
   share->key_length=key_length;
+  /*
+    We want to prevent other connections from opening this table until end
+    of statement as it is likely that modifications of table's metadata are
+    not yet finished (for example CREATE TRIGGER have to change .TRG file,
+    or we might want to drop table if CREATE TABLE ... SELECT fails).
+    This also allows us to assume that no other connection will sneak in
+    before we will get table-level lock on this table.
+  */
   share->version=0;
   share->flush_version=0;
   table->in_use = thd;
   check_unused();
-  table->next = thd->open_tables;
-  thd->open_tables = table;
+
+  if (link_in)
+  {
+    table->next= thd->open_tables;
+    thd->open_tables= table;
+  }
+  else
+  {
+    /*
+      TABLE object should be already in THD::open_tables list so we just
+      need to set TABLE::next correctly.
+    */
+    table->next= orig_table.next;
+  }
+
   table->tablenr=thd->current_tablenr++;
   table->used_fields=0;
   table->const_table=0;
@@ -1216,6 +1301,167 @@ bool reopen_name_locked_table(THD* thd, TABLE_LIST* table_list)
 }
 
 
+/**
+    @brief Create and insert into table cache placeholder for table
+           which will prevent its opening (or creation) (a.k.a lock
+           table name).
+
+    @param  thd         Thread context
+    @param  key         Table cache key for name to be locked
+    @param  key_length  Table cache key length
+
+    @return Pointer to TABLE object used for name locking or 0 in
+            case of failure.
+*/
+
+TABLE *table_cache_insert_placeholder(THD *thd, const char *key,
+                                      uint key_length)
+{
+  TABLE *table;
+  char *key_buff;
+  DBUG_ENTER("table_cache_insert_placeholder");
+
+  safe_mutex_assert_owner(&LOCK_open);
+
+  /*
+    Create a table entry with the right key and with an old refresh version
+    Note that we must use my_multi_malloc() here as this is freed by the
+    table cache
+  */
+  if (!my_multi_malloc(MYF(MY_WME | MY_ZEROFILL),
+                       &table, sizeof(*table),
+                       &key_buff, key_length,
+                       NULL))
+    DBUG_RETURN(NULL);
+
+  table->s= &table->share_not_to_be_used;
+  memcpy(key_buff, key, key_length);
+  table->s->table_cache_key= key_buff;
+  table->s->db= table->s->table_cache_key;
+  table->s->table_name= table->s->table_cache_key + strlen(table->s->db) + 1;
+  table->s->key_length= key_length;
+  table->in_use= thd;
+  table->locked_by_name= 1;
+
+  if (my_hash_insert(&open_cache, (byte*)table))
+  {
+    my_free((gptr) table, MYF(0));
+    DBUG_RETURN(NULL);
+  }
+
+  DBUG_RETURN(table);
+}
+
+
+/**
+    @brief Check if table cache contains an open placeholder for the
+           table and if this placeholder was created by another thread.
+
+    @param  thd         Thread context
+    @param  db          Name of database for table in question
+    @param  table_name  Table name
+
+    @note The presence of open placeholder indicates that either some
+          other thread is trying to create table in question and obtained
+          an exclusive name-lock on it or that this table already exists
+          and is being flushed at the moment.
+
+    @note One should acquire LOCK_open mutex before calling this function.
+
+    @note This function is a hack which was introduced in 5.0 only to
+          minimize code changes. It doesn't present in 5.1.
+
+    @retval  TRUE   Table cache contains open placeholder for the table
+                    which was created by some other thread.
+    @retval  FALSE  Otherwise.
+*/
+
+bool table_cache_has_open_placeholder(THD *thd, const char *db,
+                                      const char *table_name)
+{
+  char key[MAX_DBKEY_LENGTH];
+  uint key_length;
+  HASH_SEARCH_STATE state;
+  TABLE *search;
+  DBUG_ENTER("table_cache_has_open_placeholder");
+
+  safe_mutex_assert_owner(&LOCK_open);
+
+  key_length=(uint) (strmov(strmov(key,db)+1,table_name)-key)+1;
+  for (search= (TABLE*) hash_first(&open_cache, (byte*) key, key_length,
+                                   &state);
+      search ;
+      search= (TABLE*) hash_next(&open_cache, (byte*) key, key_length,
+                                 &state))
+  {
+    if (search->in_use == thd)
+      continue;
+    if (search->open_placeholder)
+        DBUG_RETURN(1);
+  }
+  DBUG_RETURN(0);
+}
+
+
+/**
+    @brief Check that table exists on disk or in some storage engine.
+
+    @param  thd          Thread context
+    @param  table        Table list element
+    @param  exists[out]  Out parameter which is set to TRUE if table
+                         exists and to FALSE otherwise.
+
+    @note This function assumes that caller owns LOCK_open mutex.
+          It also assumes that the fact that there are no name-locks
+          on the table was checked beforehand.
+
+    @note If there is no .FRM file for the table but it exists in one
+          of engines (e.g. it was created on another node of NDB cluster)
+          this function will fetch and create proper .FRM file for it.
+
+    @retval  TRUE   Some error occured
+    @retval  FALSE  No error. 'exists' out parameter set accordingly.
+*/
+
+bool check_if_table_exists(THD *thd, TABLE_LIST *table, bool *exists)
+{
+  char path[FN_REFLEN];
+  int rc;
+  DBUG_ENTER("check_if_table_exists");
+
+  safe_mutex_assert_owner(&LOCK_open);
+
+  *exists= TRUE;
+
+  build_table_path(path, sizeof(path), table->db, table->table_name, reg_ext);
+
+  if (!access(path, F_OK))
+    DBUG_RETURN(FALSE);
+
+  /* .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;
+    DBUG_RETURN(FALSE);
+  }
+  else if (!rc)
+  {
+    /* Table exists in some engine and .FRM for it was created. */
+    DBUG_RETURN(FALSE);
+  }
+  else /* (rc > 0) */
+  {
+    my_printf_error(ER_UNKNOWN_ERROR, "Failed to open '%-.64s', error while "
+                    "unpacking from engine", MYF(0), table->table_name);
+    DBUG_RETURN(TRUE);
+  }
+}
+
+
 /*
   Open a table.
 
@@ -1231,12 +1477,17 @@ bool reopen_name_locked_table(THD* thd, TABLE_LIST* table_list)
                           MYSQL_LOCK_IGNORE_FLUSH - Open table even if
                           someone has done a flush or namelock on it.
                           No version number checking is done.
-                          MYSQL_OPEN_IGNORE_LOCKED_TABLES - Open table
-                          ignoring set of locked tables and prelocked mode.
+                          MYSQL_OPEN_TEMPORARY_ONLY - Open only temporary
+                          table not the base table or view.
 
   IMPLEMENTATION
     Uses a cache of open tables to find a table not in use.
 
+    If table list element for the table to be opened has "create" flag
+    set and table does not exist, this function will automatically insert
+    a placeholder for exclusive name lock into the open tables cache and
+    will return the TABLE instance that corresponds to this placeholder.
+
   RETURN
     NULL  Open failed.  If refresh is set then one should close
           all other tables and retry the open.
@@ -1305,6 +1556,12 @@ TABLE *open_table(THD *thd, TABLE_LIST *table_list, MEM_ROOT *mem_root,
     }
   }
 
+  if (flags & MYSQL_OPEN_TEMPORARY_ONLY)
+  {
+    my_error(ER_NO_SUCH_TABLE, MYF(0), table_list->db, table_list->table_name);
+    DBUG_RETURN(0);
+  }
+
   /*
     The table is not temporary - if we're in pre-locked or LOCK TABLES
     mode, let's try to find the requested table in the list of pre-opened
@@ -1312,8 +1569,7 @@ TABLE *open_table(THD *thd, TABLE_LIST *table_list, MEM_ROOT *mem_root,
     open not pre-opened tables in pre-locked/LOCK TABLES mode.
     TODO: move this block into a separate function.
   */
-  if (!(flags & MYSQL_OPEN_IGNORE_LOCKED_TABLES) &&
-      (thd->locked_tables || thd->prelocked_mode))
+  if (thd->locked_tables || thd->prelocked_mode)
   {						// Using table locks
     TABLE *best_table= 0;
     int best_distance= INT_MIN;
@@ -1495,7 +1751,7 @@ TABLE *open_table(THD *thd, TABLE_LIST *table_list, MEM_ROOT *mem_root,
       c1: name lock t2; -- blocks
       c2: open t1; -- blocks
     */
-    if (table->s->version != refresh_version)
+    if (table->needs_reopen_or_name_lock())
     {
       DBUG_PRINT("note",
                  ("Found table '%s.%s' with different refresh version",
@@ -1507,6 +1763,14 @@ TABLE *open_table(THD *thd, TABLE_LIST *table_list, MEM_ROOT *mem_root,
         continue;
       }
 
+      /* Avoid self-deadlocks by detecting self-dependencies. */
+      if (table->open_placeholder && table->in_use == thd)
+      {
+	VOID(pthread_mutex_unlock(&LOCK_open));
+        my_error(ER_UPDATE_TABLE_USED, MYF(0), table->s->table_name);
+        DBUG_RETURN(0);
+      }
+
       /*
         Back off, part 1: mark the table as "unused" for the
         purpose of name-locking by setting table->db_stat to 0. Do
@@ -1523,6 +1787,14 @@ TABLE *open_table(THD *thd, TABLE_LIST *table_list, MEM_ROOT *mem_root,
         and wait till the operation is complete: when any
         operation that juggles with table->s->version completes,
         it broadcasts COND_refresh condition variable.
+        If 'old' table we met is in use by current thread we return
+        without waiting since in this situation it's this thread
+        which is responsible for broadcasting on COND_refresh
+        (and this was done already in close_old_data_files()).
+        Good example of such situation is when we have statement
+        that needs two instances of table and FLUSH TABLES comes
+        after we open first instance but before we open second
+        instance.
       */
       if (table->in_use != thd)
       {
@@ -1564,6 +1836,40 @@ TABLE *open_table(THD *thd, TABLE_LIST *table_list, MEM_ROOT *mem_root,
     while (open_cache.records > table_cache_size && unused_tables)
       VOID(hash_delete(&open_cache,(byte*) unused_tables)); /* purecov: tested */
 
+    if (table_list->create)
+    {
+      bool exists;
+
+      if (check_if_table_exists(thd, table_list, &exists))
+      {
+        VOID(pthread_mutex_unlock(&LOCK_open));
+        DBUG_RETURN(NULL);
+      }
+
+      if (!exists)
+      {
+        /*
+          Table to be created, so we need to create placeholder in table-cache.
+        */
+        if (!(table= table_cache_insert_placeholder(thd, key, key_length)))
+        {
+          VOID(pthread_mutex_unlock(&LOCK_open));
+          DBUG_RETURN(NULL);
+        }
+        /*
+          Link placeholder to the open tables list so it will be automatically
+          removed once tables are closed. Also mark it so it won't be ignored
+          by other trying to take name-lock.
+        */
+        table->open_placeholder= 1;
+        table->next= thd->open_tables;
+        thd->open_tables= table;
+        VOID(pthread_mutex_unlock(&LOCK_open));
+        DBUG_RETURN(table);
+      }
+      /* Table exists. Let us try to open it. */
+    }
+
     /* make a new table */
     if (!(table=(TABLE*) my_malloc(sizeof(*table),MYF(MY_WME))))
     {
@@ -1794,9 +2100,24 @@ bool close_data_tables(THD *thd,const char *db, const char *table_name)
 }
 
 
-/*
-  Reopen all tables with closed data files
-  One should have lock on LOCK_open when calling this
+/**
+    @brief Reopen all tables with closed data files.
+
+    @param thd         Thread context
+    @param get_locks   Should we get locks after reopening tables ?
+    @param in_refresh  Are we in FLUSH TABLES ? TODO: It seems that
+                       we can remove this parameter.
+
+    @note Since this function can't properly handle prelocking and
+          create placeholders it should be used in very special
+          situations like FLUSH TABLES or ALTER TABLE. In general
+          case one should just repeat open_tables()/lock_tables()
+          combination when one needs tables to be reopened (for
+          example see open_and_lock_tables()).
+
+    @note One should have lock on LOCK_open when calling this.
+
+    @return FALSE in case of success, TRUE - otherwise.
 */
 
 bool reopen_tables(THD *thd,bool get_locks,bool in_refresh)
@@ -1841,7 +2162,7 @@ bool reopen_tables(THD *thd,bool get_locks,bool in_refresh)
       if (in_refresh)
       {
 	table->s->version=0;
-	table->locked_by_flush=0;
+	table->open_placeholder= 0;
       }
     }
   }
@@ -1867,35 +2188,71 @@ bool reopen_tables(THD *thd,bool get_locks,bool in_refresh)
   DBUG_RETURN(error);
 }
 
-/*
-  Close handlers for tables in list, but leave the TABLE structure
-  intact so that we can re-open these quickly
-  abort_locks is set if called from flush_tables.
+
+/**
+    @brief Close handlers for tables in list, but leave the TABLE structure
+           intact so that we can re-open these quickly.
+
+    @param thd           Thread context
+    @param table         Head of the list of TABLE objects
+    @param morph_locks   TRUE  - remove locks which we have on tables being closed
+                                 but ensure that no DML or DDL will sneak in before
+                                 we will re-open the table (i.e. temporarily morph
+                                 our table-level locks into name-locks).
+                         FALSE - otherwise
+    @param send_refresh  Should we awake waiters even if we didn't close any tables?
 */
 
-void close_old_data_files(THD *thd, TABLE *table, bool abort_locks,
+void close_old_data_files(THD *thd, TABLE *table, bool morph_locks,
 			  bool send_refresh)
 {
   DBUG_ENTER("close_old_data_files");
   bool found=send_refresh;
   for (; table ; table=table->next)
   {
-    if (table->s->version != refresh_version)
+    if (table->needs_reopen_or_name_lock())
     {
       found=1;
-      if (!abort_locks)				// If not from flush tables
+      /*
+        Note that it is safe to update version even for open placeholders
+        as later in this function we reset TABLE::open_placeholder and thus
+        effectively remove them from the table cache.
+      */
+      if (!morph_locks)                         // If not from flush tables
 	table->s->version= refresh_version;	// Let other threads use table
       if (table->db_stat)
       {
-	if (abort_locks)
-	{
-	  mysql_lock_abort(thd,table);		// Close waiting threads
-	  mysql_lock_remove(thd, thd->locked_tables,table);
-	  table->locked_by_flush=1;		// Will be reopened with locks
-	}
+        if (morph_locks)
+        {
+          /*
+            Wake up threads waiting for table-level lock on this table
+            so they won't sneak in when we will temporarily remove our
+            lock on it. This will also give them a chance to close their
+            instances of this table.
+          */
+          mysql_lock_abort(thd, table);
+          mysql_lock_remove(thd, thd->locked_tables, table);
+          /*
+            We want to protect the table from concurrent DDL operations
+            (like RENAME TABLE) until we will re-open and re-lock it.
+          */
+          table->open_placeholder= 1;
+        }
 	table->file->close();
 	table->db_stat=0;
       }
+      else if (table->open_placeholder)
+      {
+        /*
+          We come here only in close-for-back-off scenario. So we have to
+          "close" create placeholder here to avoid deadlocks (for example,
+          in case of concurrent execution of CREATE TABLE t1 SELECT * FROM t2
+          and RENAME TABLE t2 TO t1). In close-for-re-open scenario we will
+          probably want to let it stay.
+        */
+        DBUG_ASSERT(!morph_locks);
+        table->open_placeholder= 0;
+      }
     }
   }
   if (found)
@@ -1923,9 +2280,8 @@ bool table_is_used(TABLE *table, bool wait_for_name_lock)
          search= (TABLE*) hash_next(&open_cache, (byte*) key,
                                     key_length, &state))
     {
-      if (search->locked_by_flush ||
-	  search->locked_by_name && wait_for_name_lock ||
-	  search->db_stat && search->s->version < refresh_version)
+      if (search->locked_by_name && wait_for_name_lock ||
+	  search->is_name_opened() && search->needs_reopen_or_name_lock())
 	return 1;				// Table is used
     }
   } while ((table=table->next));
@@ -5661,7 +6017,7 @@ bool remove_table_from_cache(THD *thd, const char *db, const char *table_name,
       else if (in_use != thd)
       {
         in_use->some_tables_deleted=1;
-        if (table->db_stat)
+        if (table->is_name_opened())
         {
           DBUG_PRINT("info", ("Found another active instance of the table"));
   	  result=1;
diff --git a/sql/sql_handler.cc b/sql/sql_handler.cc
index d1a5ab7dfa8..e1318aa2736 100644
--- a/sql/sql_handler.cc
+++ b/sql/sql_handler.cc
@@ -670,7 +670,7 @@ int mysql_ha_flush(THD *thd, TABLE_LIST *tables, uint mode_flags,
     while (*table_ptr)
     {
       if ((mode_flags & MYSQL_HA_FLUSH_ALL) ||
-          ((*table_ptr)->s->version != refresh_version))
+          (*table_ptr)->needs_reopen_or_name_lock())
       {
         /* The first time it is required, lock for close_thread_table(). */
         if (! did_lock && ! is_locked)
@@ -771,15 +771,22 @@ void mysql_ha_mark_tables_for_reopen(THD *thd, TABLE *table)
   safe_mutex_assert_owner(&LOCK_open);
   for (; table; table= table->next)
   {
-    TABLE_LIST *hash_tables;
-    if ((hash_tables= (TABLE_LIST*) hash_search(&thd->handler_tables_hash,
-                                                (byte*) table->alias,
-                                                strlen(table->alias) + 1)))
+    /*
+      Some elements in open table list, for example placeholders used for
+      name-locking, can have alias set to 0.
+    */
+    if (table->alias)
     {
-      /* Mark table as ready for reopen. */
-      hash_tables->table= NULL;
-      /* End open index/table scans. */
-      table->file->ha_index_or_rnd_end();
+      TABLE_LIST *hash_tables;
+      if ((hash_tables= (TABLE_LIST*) hash_search(&thd->handler_tables_hash,
+                                                  (byte*) table->alias,
+                                                  strlen(table->alias) + 1)))
+      {
+        /* Mark table as ready for reopen. */
+        hash_tables->table= NULL;
+        /* End open index/table scans. */
+        table->file->ha_index_or_rnd_end();
+      }
     }
   }
   DBUG_VOID_RETURN;
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index f1d86224adb..4ba820ddefd 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -2179,7 +2179,7 @@ bool delayed_insert::handle_inserts(void)
 
   thd.proc_info="insert";
   max_rows= delayed_insert_limit;
-  if (thd.killed || table->s->version != refresh_version)
+  if (thd.killed || table->needs_reopen_or_name_lock())
   {
     thd.killed= THD::KILL_CONNECTION;
     max_rows= ~(ulong)0;                        // Do as much as possible
@@ -2791,8 +2791,8 @@ bool select_insert::send_eof()
 ***************************************************************************/
 
 /*
-  Create table from lists of fields and items (or open existing table
-  with same name).
+  Create table from lists of fields and items (or just return TABLE
+  object for pre-opened existing table).
 
   SYNOPSIS
     create_table_from_items()
@@ -2807,18 +2807,24 @@ bool select_insert::send_eof()
                           of fields for the table (corresponding fields will
                           be added to the end of alter_info->create_list)
       lock         out    Pointer to the MYSQL_LOCK object for table created
-                          (open) will be returned in this parameter. Since
-                          this table is not included in THD::lock caller is
-                          responsible for explicitly unlocking this table.
+                          (or open temporary table) will be returned in this
+                          parameter. Since this table is not included in
+                          THD::lock caller is responsible for explicitly
+                          unlocking this table.
 
   NOTES
-    If 'create_info->options' bitmask has HA_LEX_CREATE_IF_NOT_EXISTS
-    flag and table with name provided already exists then this function will
-    simply open existing table.
-    Also note that create, open and lock sequence in this function is not
-    atomic and thus contains gap for deadlock and can cause other troubles.
-    Since this function contains some logic specific to CREATE TABLE ... SELECT
-    it should be changed before it can be used in other contexts.
+    This function behaves differently for base and temporary tables:
+    - For base table we assume that either table exists and was pre-opened
+      and locked at open_and_lock_tables() stage (and in this case we just
+      emit error or warning and return pre-opened TABLE object) or special
+      placeholder was put in table cache that guarantees that this table
+      won't be created or opened until the placeholder will be removed
+      (so there is an exclusive lock on this table).
+    - We don't pre-open existing temporary table, instead we either open
+      or create and then open table in this function.
+
+    Since this function contains some logic specific to CREATE TABLE ...
+    SELECT it should be changed before it can be used in other contexts.
 
   RETURN VALUES
     non-zero  Pointer to TABLE object for table created or opened
@@ -2841,6 +2847,25 @@ static TABLE *create_table_from_items(THD *thd, HA_CREATE_INFO *create_info,
   bool not_used;
   DBUG_ENTER("create_table_from_items");
 
+  DBUG_EXECUTE_IF("sleep_create_select_before_check_if_exists", my_sleep(6000000););
+
+  if (!(create_info->options & HA_LEX_CREATE_TMP_TABLE) &&
+      create_table->table->db_stat)
+  {
+    /* Table already exists and was open at open_and_lock_tables() stage. */
+    if (create_info->options & HA_LEX_CREATE_IF_NOT_EXISTS)
+    {
+      create_info->table_existed= 1;		// Mark that table existed
+      push_warning_printf(thd, MYSQL_ERROR::WARN_LEVEL_NOTE,
+                          ER_TABLE_EXISTS_ERROR, ER(ER_TABLE_EXISTS_ERROR),
+                          create_table->table_name);
+      DBUG_RETURN(create_table->table);
+    }
+
+    my_error(ER_TABLE_EXISTS_ERROR, MYF(0), create_table->table_name);
+    DBUG_RETURN(0);
+  }
+
   tmp_table.alias= 0;
   tmp_table.timestamp_field= 0;
   tmp_table.s= &tmp_table.share_not_to_be_used;
@@ -2869,8 +2894,15 @@ static TABLE *create_table_from_items(THD *thd, HA_CREATE_INFO *create_info,
       cr_field->flags &= ~NOT_NULL_FLAG;
     alter_info->create_list.push_back(cr_field);
   }
+
+  DBUG_EXECUTE_IF("sleep_create_select_before_create", my_sleep(6000000););
+
   /*
-    create and lock table
+    Create and lock table.
+
+    Note that we either creating (or opening existing) temporary table or
+    creating base table on which name we have exclusive lock. So code below
+    should not cause deadlocks or races.
 
     We don't log the statement, it will be logged later.
 
@@ -2880,59 +2912,70 @@ static TABLE *create_table_from_items(THD *thd, HA_CREATE_INFO *create_info,
     don't want to delete from it) 2) it would be written before the CREATE
     TABLE, which is a wrong order. So we keep binary logging disabled when we
     open_table().
-    NOTE: By locking table which we just have created (or for which we just have
-    have found that it already exists) separately from other tables used by the
-    statement we create potential window for deadlock.
-    TODO: create and open should be done atomic !
   */
   {
     tmp_disable_binlog(thd);
     if (!mysql_create_table(thd, create_table->db, create_table->table_name,
                             create_info, alter_info, 0, select_field_count))
     {
-      /*
-        If we are here in prelocked mode we either create temporary table
-        or prelocked mode is caused by the SELECT part of this statement.
-      */
-      DBUG_ASSERT(!thd->prelocked_mode ||
-                  create_info->options & HA_LEX_CREATE_TMP_TABLE ||
-                  thd->lex->requires_prelocking());
 
-      /*
-        NOTE: We don't want to ignore set of locked tables here if we are
-              under explicit LOCK TABLES since it will open gap for deadlock
-              too wide (and also is not backward compatible).
-      */
-      if (! (table= open_table(thd, create_table, thd->mem_root, (bool*) 0,
-                               (MYSQL_LOCK_IGNORE_FLUSH |
-                                ((thd->prelocked_mode == PRELOCKED) ?
-                                 MYSQL_OPEN_IGNORE_LOCKED_TABLES:0)))))
-        quick_rm_table(create_info->db_type, create_table->db,
-                       table_case_name(create_info, create_table->table_name));
+      if (create_info->table_existed &&
+          !(create_info->options & HA_LEX_CREATE_TMP_TABLE))
+      {
+        /*
+          This means that someone created table underneath server
+          or it was created via different mysqld front-end to the
+          cluster. We don't have much options but throw an error.
+        */
+        my_error(ER_TABLE_EXISTS_ERROR, MYF(0), create_table->table_name);
+        DBUG_RETURN(0);
+      }
+
+      DBUG_EXECUTE_IF("sleep_create_select_before_open", my_sleep(6000000););
+
+      if (!(create_info->options & HA_LEX_CREATE_TMP_TABLE))
+      {
+        VOID(pthread_mutex_lock(&LOCK_open));
+        if (reopen_name_locked_table(thd, create_table, FALSE))
+        {
+          quick_rm_table(create_info->db_type, create_table->db,
+                         table_case_name(create_info,
+                                         create_table->table_name));
+        }
+        else
+          table= create_table->table;
+        VOID(pthread_mutex_unlock(&LOCK_open));
+      }
+      else
+      {
+        if (!(table= open_table(thd, create_table, thd->mem_root, (bool*) 0,
+                                MYSQL_OPEN_TEMPORARY_ONLY)) &&
+            !create_info->table_existed)
+        {
+          /*
+            This shouldn't happen as creation of temporary table should make
+            it preparable for open. But let us do close_temporary_table() here
+            just in case.
+          */
+          close_temporary_table(thd, create_table->db, create_table->table_name);
+        }
+      }
     }
     reenable_binlog(thd);
     if (!table)                                   // open failed
       DBUG_RETURN(0);
   }
 
-  /*
-    FIXME: What happens if trigger manages to be created while we are
-           obtaining this lock ? May be it is sensible just to disable
-           trigger execution in this case ? Or will MYSQL_LOCK_IGNORE_FLUSH
-           save us from that ?
-  */
+  DBUG_EXECUTE_IF("sleep_create_select_before_lock", my_sleep(6000000););
+
   table->reginfo.lock_type=TL_WRITE;
   if (! ((*lock)= mysql_lock_tables(thd, &table, 1,
                                     MYSQL_LOCK_IGNORE_FLUSH, &not_used)))
   {
-    VOID(pthread_mutex_lock(&LOCK_open));
-    hash_delete(&open_cache,(byte*) table);
-    VOID(pthread_mutex_unlock(&LOCK_open));
-    quick_rm_table(create_info->db_type, create_table->db,
-		   table_case_name(create_info, create_table->table_name));
+    if (!create_info->table_existed)
+      drop_open_table(thd, table, create_table->db, create_table->table_name);
     DBUG_RETURN(0);
   }
-  table->file->extra(HA_EXTRA_WRITE_CACHE);
   DBUG_RETURN(table);
 }
 
@@ -2983,8 +3026,10 @@ select_create::prepare(List<Item> &values, SELECT_LEX_UNIT *u)
                           (thd->variables.sql_mode &
                            (MODE_STRICT_TRANS_TABLES |
                             MODE_STRICT_ALL_TABLES)));
-  DBUG_RETURN(check_that_all_fields_are_given_values(thd, table,
-                                                     table_list));
+  if (check_that_all_fields_are_given_values(thd, table, table_list))
+    DBUG_RETURN(1);
+  table->file->extra(HA_EXTRA_WRITE_CACHE);
+  DBUG_RETURN(0);
 }
 
 
@@ -3016,31 +3061,18 @@ bool select_create::send_eof()
   {
     table->file->extra(HA_EXTRA_NO_IGNORE_DUP_KEY);
     table->file->extra(HA_EXTRA_WRITE_CANNOT_REPLACE);
-    VOID(pthread_mutex_lock(&LOCK_open));
-    mysql_unlock_tables(thd, lock);
-    /*
-      TODO:
-      Check if we can remove the following two rows.
-      We should be able to just keep the table in the table cache.
-    */
-    if (!table->s->tmp_table)
+    if (lock)
     {
-      ulong version= table->s->version;
-      hash_delete(&open_cache,(byte*) table);
-      /* Tell threads waiting for refresh that something has happened */
-      if (version != refresh_version)
-        broadcast_refresh();
+      mysql_unlock_tables(thd, lock);
+      lock= 0;
     }
-    lock=0;
-    table=0;
-    VOID(pthread_mutex_unlock(&LOCK_open));
   }
   return tmp;
 }
 
+
 void select_create::abort()
 {
-  VOID(pthread_mutex_lock(&LOCK_open));
   if (lock)
   {
     mysql_unlock_tables(thd, lock);
@@ -3050,22 +3082,10 @@ void select_create::abort()
   {
     table->file->extra(HA_EXTRA_NO_IGNORE_DUP_KEY);
     table->file->extra(HA_EXTRA_WRITE_CANNOT_REPLACE);
-    enum db_type table_type=table->s->db_type;
-    if (!table->s->tmp_table)
-    {
-      ulong version= table->s->version;
-      hash_delete(&open_cache,(byte*) table);
-      if (!create_info->table_existed)
-        quick_rm_table(table_type, create_table->db, create_table->table_name);
-      /* Tell threads waiting for refresh that something has happened */
-      if (version != refresh_version)
-        broadcast_refresh();
-    }
-    else if (!create_info->table_existed)
-      close_temporary_table(thd, create_table->db, create_table->table_name);
+    if (!create_info->table_existed)
+      drop_open_table(thd, table, create_table->db, create_table->table_name);
     table=0;
   }
-  VOID(pthread_mutex_unlock(&LOCK_open));
 }
 
 
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 30abb7c2c0a..4ce0dc43144 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -3014,7 +3014,13 @@ mysql_execute_command(THD *thd)
       select_lex->options|= SELECT_NO_UNLOCK;
       unit->set_limit(select_lex);
 
-      if (!(res= open_and_lock_tables(thd, select_tables)))
+      if (!(lex->create_info.options & HA_LEX_CREATE_TMP_TABLE))
+      {
+        lex->link_first_table_back(create_table, link_to_local);
+        create_table->create= TRUE;
+      }
+
+      if (!(res= open_and_lock_tables(thd, lex->query_tables)))
       {
         /*
           Is table which we are changing used somewhere in other parts
@@ -3023,6 +3029,7 @@ mysql_execute_command(THD *thd)
         if (!(create_info.options & HA_LEX_CREATE_TMP_TABLE))
         {
           TABLE_LIST *duplicate;
+          create_table= lex->unlink_first_table(&link_to_local);
           if ((duplicate= unique_table(thd, create_table, select_tables, 0)))
           {
             update_non_unique_table_error(create_table, "CREATE", duplicate);
@@ -3066,6 +3073,9 @@ mysql_execute_command(THD *thd)
           delete sel_result;
         }
       }
+      else if (!(lex->create_info.options & HA_LEX_CREATE_TMP_TABLE))
+        create_table= lex->unlink_first_table(&link_to_local);
+
     }
     else
     {
diff --git a/sql/sql_prepare.cc b/sql/sql_prepare.cc
index d3cc3613b4a..fcd70ad92ff 100644
--- a/sql/sql_prepare.cc
+++ b/sql/sql_prepare.cc
@@ -1491,8 +1491,21 @@ static bool mysql_test_create_table(Prepared_statement *stmt)
 
   if (select_lex->item_list.elements)
   {
+    if (!(lex->create_info.options & HA_LEX_CREATE_TMP_TABLE))
+    {
+      lex->link_first_table_back(create_table, link_to_local);
+      create_table->create= TRUE;
+    }
+
+    if (open_and_lock_tables(stmt->thd, lex->query_tables))
+      DBUG_RETURN(TRUE);
+
+    if (!(lex->create_info.options & HA_LEX_CREATE_TMP_TABLE))
+      create_table= lex->unlink_first_table(&link_to_local);
+
     select_lex->context.resolve_in_select_list= TRUE;
-    res= select_like_stmt_test_with_open_n_lock(stmt, tables, 0, 0);
+
+    res= select_like_stmt_test(stmt, 0, 0);
   }
 
   /* put tables back for PS rexecuting */
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 42d59a10712..079cc0d6456 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -62,8 +62,8 @@ static void set_tmp_file_path(char *buf, size_t bufsize, THD *thd);
     #                   Size of path
  */
 
-static uint build_table_path(char *buff, size_t bufflen, const char *db,
-                             const char *table, const char *ext)
+uint build_table_path(char *buff, size_t bufflen, const char *db,
+                      const char *table, const char *ext)
 {
   strxnmov(buff, bufflen-1, mysql_data_home, "/", db, "/", table, ext,
            NullS);
@@ -1722,7 +1722,14 @@ bool mysql_create_table(THD *thd,const char *db, const char *table_name,
   VOID(pthread_mutex_lock(&LOCK_open));
   if (!internal_tmp_table && !(create_info->options & HA_LEX_CREATE_TMP_TABLE))
   {
-    if (!access(path,F_OK))
+    /*
+      Inspecting table cache for placeholders created by concurrent
+      CREATE TABLE ... SELECT statements to avoid interfering with them
+      is 5.0-only solution. Starting from 5.1 we solve this problem by
+      obtaining name-lock on the table to be created first.
+    */
+    if (table_cache_has_open_placeholder(thd, db, table_name) ||
+        !access(path, F_OK))
     {
       if (create_info->options & HA_LEX_CREATE_IF_NOT_EXISTS)
         goto warn;
@@ -2051,7 +2058,7 @@ static int prepare_for_restore(THD* thd, TABLE_LIST* table,
     to finish the restore in the handler later on
   */
   pthread_mutex_lock(&LOCK_open);
-  if (reopen_name_locked_table(thd, table))
+  if (reopen_name_locked_table(thd, table, TRUE))
   {
     unlock_table_name(thd, table);
     pthread_mutex_unlock(&LOCK_open);
@@ -2158,7 +2165,7 @@ static int prepare_for_repair(THD* thd, TABLE_LIST *table_list,
     to finish the repair in the handler later on.
   */
   pthread_mutex_lock(&LOCK_open);
-  if (reopen_name_locked_table(thd, table_list))
+  if (reopen_name_locked_table(thd, table_list, TRUE))
   {
     unlock_table_name(thd, table_list);
     pthread_mutex_unlock(&LOCK_open);
@@ -2803,10 +2810,24 @@ bool mysql_create_like_table(THD* thd, TABLE_LIST* table,
   }
   else
   {
+    bool exists;
     strxmov(dst_path, mysql_data_home, "/", db, "/", table_name,
 	    reg_ext, NullS);
     fn_format(dst_path, dst_path, "", "", MYF(MY_UNPACK_FILENAME));
-    if (!access(dst_path, F_OK))
+
+    /*
+      Note that this critical section should actually cover most
+      of mysql_create_like_table() function. See bugs #18950 and
+      #23667 for more information.
+      Also note that starting from 5.1 we obtain name-lock on
+      target table instead of inspecting table cache for presence
+      of open placeholders (see comment in mysql_create_table()).
+    */
+    pthread_mutex_lock(&LOCK_open);
+    exists= (table_cache_has_open_placeholder(thd, db, table_name) ||
+             !access(dst_path, F_OK));
+    pthread_mutex_unlock(&LOCK_open);
+    if (exists)
       goto table_exists;
   }
 
@@ -3160,9 +3181,14 @@ view_err:
       else
       {
 	char dir_buff[FN_REFLEN];
+        bool exists;
 	strxnmov(dir_buff, FN_REFLEN, mysql_real_data_home, new_db, NullS);
-	if (!access(fn_format(new_name_buff,new_name_buff,dir_buff,reg_ext,0),
-		    F_OK))
+        VOID(pthread_mutex_lock(&LOCK_open));
+        exists= (table_cache_has_open_placeholder(thd, new_db, new_name) ||
+                 !access(fn_format(new_name_buff, new_name_buff, dir_buff,
+                                   reg_ext, 0), F_OK));
+        VOID(pthread_mutex_unlock(&LOCK_open));
+        if (exists)
 	{
 	  /* Table will be closed in do_command() */
 	  my_error(ER_TABLE_EXISTS_ERROR, MYF(0), new_alias);
@@ -3247,8 +3273,22 @@ view_err:
     if (!error && (new_name != table_name || new_db != db))
     {
       thd->proc_info="rename";
-      /* Then do a 'simple' rename of the table */
-      if (!access(new_name_buff,F_OK))
+      /*
+        Then do a 'simple' rename of the table. First we need to close all
+        instances of 'source' table.
+      */
+      close_cached_table(thd, table);
+      /*
+        Then, we want check once again that target table does not exist.
+        Note that we can't fully rely on results of previous check since
+        no lock was taken on target table during it. We also can't do this
+        before calling close_cached_table() as the latter temporarily
+        releases LOCK_open mutex.
+        Also note that starting from 5.1 we use approach with obtaining
+        of name-lock on target table.
+      */
+      if (table_cache_has_open_placeholder(thd, new_db, new_name) ||
+          !access(new_name_buff,F_OK))
       {
 	my_error(ER_TABLE_EXISTS_ERROR, MYF(0), new_name);
 	error= -1;
@@ -3256,7 +3296,6 @@ view_err:
       else
       {
 	*fn_ext(new_name)=0;
-	close_cached_table(thd, table);
 	if (mysql_rename_table(old_db_type,db,table_name,new_db,new_alias))
 	  error= -1;
         else if (Table_triggers_list::change_table_name(thd, db, table_name,
@@ -3806,17 +3845,6 @@ view_err:
 	      current_pid, thd->thread_id);
   if (lower_case_table_names)
     my_casedn_str(files_charset_info, old_name);
-  if (new_name != table_name || new_db != db)
-  {
-    if (!access(new_name_buff,F_OK))
-    {
-      error=1;
-      my_error(ER_TABLE_EXISTS_ERROR, MYF(0), new_name_buff);
-      VOID(quick_rm_table(new_db_type,new_db,tmp_name));
-      VOID(pthread_mutex_unlock(&LOCK_open));
-      goto err;
-    }
-  }
 
 #if (!defined( __WIN__) && !defined( __EMX__) && !defined( OS2))
   if (table->file->has_transactions())
@@ -3835,6 +3863,22 @@ view_err:
     table->file->extra(HA_EXTRA_FORCE_REOPEN);	// Don't use this file anymore
 #endif
 
+  if (new_name != table_name || new_db != db)
+  {
+    /*
+      Check that there is no table with target name. See the
+      comment describing code for 'simple' ALTER TABLE ... RENAME.
+    */
+    if (table_cache_has_open_placeholder(thd, new_db, new_name) ||
+        !access(new_name_buff,F_OK))
+    {
+      error=1;
+      my_error(ER_TABLE_EXISTS_ERROR, MYF(0), new_name_buff);
+      VOID(quick_rm_table(new_db_type,new_db,tmp_name));
+      VOID(pthread_mutex_unlock(&LOCK_open));
+      goto err;
+    }
+  }
 
   error=0;
   if (!need_copy_table)
diff --git a/sql/sql_trigger.cc b/sql/sql_trigger.cc
index 26b557a8247..cf6db22fbcb 100644
--- a/sql/sql_trigger.cc
+++ b/sql/sql_trigger.cc
@@ -271,7 +271,7 @@ bool mysql_create_or_drop_trigger(THD *thd, TABLE_LIST *tables, bool create)
   /* We also don't allow creation of triggers on views. */
   tables->required_type= FRMTYPE_TABLE;
 
-  if (reopen_name_locked_table(thd, tables))
+  if (reopen_name_locked_table(thd, tables, TRUE))
   {
     unlock_table_name(thd, tables);
     goto end;
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index 390a991c7b6..12d35d5edd3 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -1489,9 +1489,7 @@ create:
 	  lex->sql_command= SQLCOM_CREATE_TABLE;
 	  if (!lex->select_lex.add_table_to_list(thd, $5, NULL,
 						 TL_OPTION_UPDATING,
-						 (using_update_log ?
-						  TL_READ_NO_INSERT:
-						  TL_READ)))
+						 TL_WRITE))
 	    MYSQL_YYABORT;
           lex->alter_info.reset();
 	  lex->col_list.empty();
diff --git a/sql/table.h b/sql/table.h
index 61232a39f0e..da2c90ab212 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -188,6 +188,8 @@ typedef struct st_table_share
 } TABLE_SHARE;
 
 
+extern ulong refresh_version;
+
 /* Information for one open table */
 
 struct st_table {
@@ -268,7 +270,24 @@ struct st_table {
   my_bool force_index;
   my_bool distinct,const_table,no_rows;
   my_bool key_read, no_keyread;
-  my_bool locked_by_flush;
+  /*
+    Placeholder for an open table which prevents other connections
+    from taking name-locks on this table. Typically used with
+    TABLE_SHARE::version member to take an exclusive name-lock on
+    this table name -- a name lock that not only prevents other
+    threads from opening the table, but also blocks other name
+    locks. This is achieved by:
+    - setting open_placeholder to 1 - this will block other name
+      locks, as wait_for_locked_table_name will be forced to wait,
+      see table_is_used for details.
+    - setting version to 0 - this will force other threads to close
+      the instance of this table and wait (this is the same approach
+      as used for usual name locks).
+    An exclusively name-locked table currently can have no handler
+    object associated with it (db_stat is always 0), but please do
+    not rely on that.
+  */
+  my_bool open_placeholder;
   my_bool locked_by_name;
   my_bool fulltext_searched;
   my_bool no_cache;
@@ -291,6 +310,13 @@ struct st_table {
 
   bool fill_item_list(List<Item> *item_list) const;
   void reset_item_list(List<Item> *item_list) const;
+  /* Is table open or should be treated as such by name-locking? */
+  inline bool is_name_opened() { return db_stat || open_placeholder; }
+  /*
+    Is this instance of the table should be reopen or represents a name-lock?
+  */
+  inline bool needs_reopen_or_name_lock()
+  { return s->version != refresh_version; }
 };
 
 enum enum_schema_table_state
@@ -648,6 +674,12 @@ typedef struct st_table_list
     used for implicit LOCK TABLES only and won't be used in real statement.
   */
   bool          prelocking_placeholder;
+  /*
+    This TABLE_LIST object corresponds to the table to be created
+    so it is possible that it does not exist (used in CREATE TABLE
+    ... SELECT implementation).
+  */
+  bool          create;
 
   enum enum_schema_table_state schema_table_state;
   void calc_md5(char *buffer);
@@ -655,7 +687,11 @@ typedef struct st_table_list
   int view_check_option(THD *thd, bool ignore_failure);
   bool setup_underlying(THD *thd);
   void cleanup_items();
-  bool placeholder() {return derived || view || schema_table || !table; }
+  bool placeholder()
+  {
+    return derived || view || schema_table || create && !table->db_stat ||
+           !table;
+  }
   void print(THD *thd, String *str);
   bool check_single_table(st_table_list **table, table_map map,
                           st_table_list *view);