Backport of revno: 2617.71.1

Bug#42546 Backup: RESTORE fails, thinking it finds an existing table

The problem occured when a MDL locking conflict happened for a non-existent 
table between a CREATE and a INSERT statement. The code for CREATE 
interpreted this lock conflict to mean that the table existed, 
which meant that the statement failed when it should not have.
The problem could occur for CREATE TABLE, CREATE TABLE LIKE and
ALTER TABLE RENAME.

This patch fixes the problem for CREATE TABLE and CREATE TABLE LIKE.
It is based on code backported from the mysql-6.1-fk tree written
by Dmitry Lenev. CREATE now uses normal open_and_lock_tables() code 
to acquire exclusive locks. This means that for the test case in the bug 
description, CREATE will wait until INSERT completes so that it can 
get the exclusive lock. This resolves the reported bug.

The patch also prohibits CREATE TABLE and CREATE TABLE LIKE under 
LOCK TABLES. Note that this is an incompatible change and must 
be reflected in the documentation. Affected test cases have been
updated.

mdl_sync.test contains tests for CREATE TABLE and CREATE TABLE LIKE.

Fixing the issue for ALTER TABLE RENAME is beyond the scope of this
patch. ALTER TABLE cannot be prohibited from working under LOCK TABLES
as this could seriously impact customers and a proper fix would require
a significant rewrite.

21 files changed, 466 insertions, 416 deletions

diff --git a/mysql-test/r/lock_multi.result b/mysql-test/r/lock_multi.result
index ef9292ad8c0..5d12e0efd64 100644
--- a/mysql-test/r/lock_multi.result
+++ b/mysql-test/r/lock_multi.result
@@ -72,9 +72,10 @@ CREATE TABLE t1 (c1 int);
 LOCK TABLE t1 WRITE;
 FLUSH TABLES WITH READ LOCK;
 CREATE TABLE t2 (c1 int);
+ERROR HY000: Table 't2' was not locked with LOCK TABLES
 UNLOCK TABLES;
 UNLOCK TABLES;
-DROP TABLE t1, t2;
+DROP TABLE t1;
 CREATE TABLE t1 (c1 int);
 LOCK TABLE t1 WRITE;
 FLUSH TABLES WITH READ LOCK;
diff --git a/mysql-test/r/mdl_sync.result b/mysql-test/r/mdl_sync.result
index d409157a70b..e5447c32b7d 100644
--- a/mysql-test/r/mdl_sync.result
+++ b/mysql-test/r/mdl_sync.result
@@ -63,6 +63,66 @@ unlock tables;
 # Clean-up.
 drop tables t1, t2, t3, t5;
 #
+# Bug#42546 - Backup: RESTORE fails, thinking it finds an existing table
+#
+DROP TABLE IF EXISTS t1;
+set @save_log_output=@@global.log_output;
+set global log_output=file;
+#
+# Test 1: CREATE TABLE
+#
+# Connection 2 
+# Start insert on the not-yet existing table
+# Wait after taking the MDL lock
+SET DEBUG_SYNC= 'after_open_table_mdl_shared SIGNAL locked WAIT_FOR finish';
+INSERT INTO t1 VALUES(1,"def");
+# Connection 1
+SET DEBUG_SYNC= 'now WAIT_FOR locked';
+# Now INSERT has a MDL on the non-existent table t1.
+#
+# Continue the INSERT once CREATE waits for exclusive lock
+SET DEBUG_SYNC= 'mdl_acquire_exclusive_locks_wait SIGNAL finish';
+# Try to create that table.
+CREATE TABLE t1 (c1 INT, c2 VARCHAR(100), KEY(c1));
+# Connection 2
+# Insert fails
+ERROR 42S02: Table 'test.t1' doesn't exist
+# Connection 1
+SET DEBUG_SYNC= 'RESET';
+SHOW TABLES;
+Tables_in_test
+t1
+DROP TABLE IF EXISTS t1;
+#
+# Test 2: CREATE TABLE LIKE
+#
+CREATE TABLE t2 (c1 INT, c2 VARCHAR(100), KEY(c1));
+# Connection 2 
+# Start insert on the not-yet existing table
+# Wait after taking the MDL
+SET DEBUG_SYNC= 'after_open_table_mdl_shared SIGNAL locked WAIT_FOR finish';
+INSERT INTO t1 VALUES(1,"def");
+# Connection 1
+SET DEBUG_SYNC= 'now WAIT_FOR locked';
+# Now INSERT has a MDL on the non-existent table t1.
+#
+# Continue the INSERT once CREATE waits for exclusive lock
+SET DEBUG_SYNC= 'mdl_acquire_exclusive_locks_wait SIGNAL finish';
+# Try to create that table.
+CREATE TABLE t1 LIKE t2;
+# Connection 2
+# Insert fails
+ERROR 42S02: Table 'test.t1' doesn't exist
+# Connection 1
+SET DEBUG_SYNC= 'RESET';
+SHOW TABLES;
+Tables_in_test
+t1
+t2
+DROP TABLE t2;
+DROP TABLE IF EXISTS t1;
+set global log_output=@save_log_output;
+#
 # Bug #46044 "MDL deadlock on LOCK TABLE + CREATE TABLE HIGH_PRIORITY
 #             FOR UPDATE"
 #
diff --git a/mysql-test/r/merge.result b/mysql-test/r/merge.result
index 3a6cd4d3f5a..0417b91490e 100644
--- a/mysql-test/r/merge.result
+++ b/mysql-test/r/merge.result
@@ -1149,7 +1149,8 @@ SHOW CREATE TABLE t3;
 ERROR 42S02: Table 'test.t3' doesn't exist
 DROP TABLE t1, t2;
 #
-# CREATE ... LIKE
+# Bug#37371 "CREATE TABLE LIKE merge loses UNION parameter"
+# Demonstrate that this is no longer the case.
 #
 # 1. Create like.
 CREATE TABLE t1 (c1 INT);
@@ -1164,26 +1165,26 @@ SHOW CREATE TABLE t4;
 Table	Create Table
 t4	CREATE TABLE `t4` (
   `c1` int(11) DEFAULT NULL
-) ENGINE=MRG_MyISAM DEFAULT CHARSET=latin1
+) ENGINE=MRG_MyISAM DEFAULT CHARSET=latin1 INSERT_METHOD=LAST UNION=(`t1`,`t2`)
 INSERT INTO t4 VALUES (4);
-ERROR HY000: Table 't4' is read only
 DROP TABLE t4;
 #
 # 1. Create like with locked tables.
 LOCK TABLES t3 WRITE, t2 WRITE, t1 WRITE;
 CREATE TABLE t4 LIKE t3;
+ERROR HY000: Table 't4' was not locked with LOCK TABLES
 SHOW CREATE TABLE t4;
 ERROR HY000: Table 't4' was not locked with LOCK TABLES
 INSERT INTO t4 VALUES (4);
 ERROR HY000: Table 't4' was not locked with LOCK TABLES
-UNLOCK TABLES;
+CREATE TEMPORARY TABLE t4 LIKE t3;
 SHOW CREATE TABLE t4;
-Table	Create Table
-t4	CREATE TABLE `t4` (
-  `c1` int(11) DEFAULT NULL
-) ENGINE=MRG_MyISAM DEFAULT CHARSET=latin1
+ERROR HY000: Unable to open underlying table which is differently defined or of non-MyISAM type or doesn't exist
+INSERT INTO t4 VALUES (4);
+ERROR HY000: Unable to open underlying table which is differently defined or of non-MyISAM type or doesn't exist
+UNLOCK TABLES;
 INSERT INTO t4 VALUES (4);
-ERROR HY000: Table 't4' is read only
+ERROR HY000: Unable to open underlying table which is differently defined or of non-MyISAM type or doesn't exist
 DROP TABLE t4;
 #
 # Rename child.
@@ -1210,6 +1211,7 @@ c1
 1
 2
 3
+4
 RENAME TABLE t2 TO t5;
 SELECT * FROM t3 ORDER BY c1;
 ERROR HY000: Unable to open underlying table which is differently defined or of non-MyISAM type or doesn't exist
@@ -1219,6 +1221,7 @@ c1
 1
 2
 3
+4
 #
 # 3. Normal rename with locked tables.
 LOCK TABLES t1 WRITE, t2 WRITE, t3 WRITE;
@@ -1227,6 +1230,7 @@ c1
 1
 2
 3
+4
 RENAME TABLE t2 TO t5;
 ERROR HY000: Can't execute the given command because you have active locked tables or an active transaction
 SELECT * FROM t3 ORDER BY c1;
@@ -1234,6 +1238,7 @@ c1
 1
 2
 3
+4
 RENAME TABLE t5 TO t2;
 ERROR HY000: Can't execute the given command because you have active locked tables or an active transaction
 SELECT * FROM t3 ORDER BY c1;
@@ -1241,6 +1246,7 @@ c1
 1
 2
 3
+4
 UNLOCK TABLES;
 #
 # 4. Alter table rename.
@@ -1253,6 +1259,7 @@ c1
 1
 2
 3
+4
 #
 # 5. Alter table rename with locked tables.
 LOCK TABLES t1 WRITE, t2 WRITE, t3 WRITE;
@@ -1268,6 +1275,7 @@ c1
 1
 2
 3
+4
 #
 # Rename parent.
 #
@@ -1278,6 +1286,7 @@ c1
 1
 2
 3
+4
 RENAME TABLE t3 TO t5;
 ERROR HY000: Can't execute the given command because you have active locked tables or an active transaction
 SELECT * FROM t3 ORDER BY c1;
@@ -1285,6 +1294,7 @@ c1
 1
 2
 3
+4
 RENAME TABLE t5 TO t3;
 ERROR HY000: Can't execute the given command because you have active locked tables or an active transaction
 SELECT * FROM t3 ORDER BY c1;
@@ -1292,6 +1302,7 @@ c1
 1
 2
 3
+4
 #
 # 5. Alter table rename with locked tables.
 ALTER TABLE t3 RENAME TO t5;
@@ -1306,6 +1317,7 @@ c1
 1
 2
 3
+4
 DROP TABLE t1, t2, t3;
 #
 # Drop locked tables.
diff --git a/mysql-test/r/ps_ddl.result b/mysql-test/r/ps_ddl.result
index f411328ed7c..3d57c8f7332 100644
--- a/mysql-test/r/ps_ddl.result
+++ b/mysql-test/r/ps_ddl.result
@@ -1755,21 +1755,21 @@ SUCCESS
 
 drop table t1;
 deallocate prepare stmt;
-# XXX: no validation of the first table in case of
-# CREATE TEMPORARY TABLE. This is a shortcoming of the current code,
-# but since validation is not strictly necessary, nothing is done
-# about it.
-# Will be fixed as part of work on Bug#21431 "Incomplete support of
-# temporary tables"
 create table t1 (a int);
 insert into t1 (a) values (1);
 prepare stmt from "create temporary table if not exists t2 as select * from t1";
 execute stmt;
 drop table t2;
 execute stmt;
+call p_verify_reprepare_count(0);
+SUCCESS
+
 execute stmt;
 Warnings:
 Note	1050	Table 't2' already exists
+call p_verify_reprepare_count(1);
+SUCCESS
+
 select * from t2;
 a
 1
@@ -1777,6 +1777,9 @@ a
 execute stmt;
 Warnings:
 Note	1050	Table 't2' already exists
+call p_verify_reprepare_count(0);
+SUCCESS
+
 select * from t2;
 a
 1
@@ -1790,7 +1793,7 @@ Note	1050	Table 't2' already exists
 select * from t2;
 a
 1
-call p_verify_reprepare_count(0);
+call p_verify_reprepare_count(1);
 SUCCESS
 
 drop table t1;
diff --git a/mysql-test/t/lock_multi.test b/mysql-test/t/lock_multi.test
index cbb99c04967..31a10f89796 100644
--- a/mysql-test/t/lock_multi.test
+++ b/mysql-test/t/lock_multi.test
@@ -199,6 +199,7 @@ let $wait_condition=
   where state = "Waiting for table" and info = "FLUSH TABLES WITH READ LOCK";
 --source include/wait_condition.inc
 # This must not block.
+--error ER_TABLE_NOT_LOCKED
 CREATE TABLE t2 (c1 int);
 UNLOCK TABLES;
 #
@@ -208,7 +209,7 @@ reap;
 UNLOCK TABLES;
 #
 connection default;
-DROP TABLE t1, t2;
+DROP TABLE t1;
 #
 # Test if CREATE TABLE SELECT with LOCK TABLE deadlocks.
 #
diff --git a/mysql-test/t/mdl_sync.test b/mysql-test/t/mdl_sync.test
index d50c056fda3..fd66f6d539d 100644
--- a/mysql-test/t/mdl_sync.test
+++ b/mysql-test/t/mdl_sync.test
@@ -142,6 +142,101 @@ drop tables t1, t2, t3, t5;
 
 
 --echo #
+--echo # Bug#42546 - Backup: RESTORE fails, thinking it finds an existing table
+--echo #
+
+--disable_warnings
+DROP TABLE IF EXISTS t1;
+--enable_warnings
+set @save_log_output=@@global.log_output;
+set global log_output=file;
+
+connect(con2, localhost, root,,);
+
+--echo #
+--echo # Test 1: CREATE TABLE
+--echo #
+
+--echo # Connection 2 
+connection con2;
+--echo # Start insert on the not-yet existing table
+--echo # Wait after taking the MDL lock
+SET DEBUG_SYNC= 'after_open_table_mdl_shared SIGNAL locked WAIT_FOR finish';
+--send INSERT INTO t1 VALUES(1,"def")
+
+--echo # Connection 1
+connection default;
+SET DEBUG_SYNC= 'now WAIT_FOR locked';
+--echo # Now INSERT has a MDL on the non-existent table t1.
+
+--echo #
+--echo # Continue the INSERT once CREATE waits for exclusive lock
+SET DEBUG_SYNC= 'mdl_acquire_exclusive_locks_wait SIGNAL finish';
+--echo # Try to create that table.
+--send CREATE TABLE t1 (c1 INT, c2 VARCHAR(100), KEY(c1))
+
+--echo # Connection 2
+--echo # Insert fails
+connection con2;
+--error ER_NO_SUCH_TABLE
+--reap
+
+--echo # Connection 1
+connection default;
+--reap;
+SET DEBUG_SYNC= 'RESET';
+SHOW TABLES;
+
+--disable_warnings
+DROP TABLE IF EXISTS t1;
+--enable_warnings
+
+--echo #
+--echo # Test 2: CREATE TABLE LIKE
+--echo #
+
+CREATE TABLE t2 (c1 INT, c2 VARCHAR(100), KEY(c1));
+
+--echo # Connection 2 
+connection con2;
+--echo # Start insert on the not-yet existing table
+--echo # Wait after taking the MDL
+SET DEBUG_SYNC= 'after_open_table_mdl_shared SIGNAL locked WAIT_FOR finish';
+--send INSERT INTO t1 VALUES(1,"def")
+
+--echo # Connection 1
+connection default;
+SET DEBUG_SYNC= 'now WAIT_FOR locked';
+--echo # Now INSERT has a MDL on the non-existent table t1.
+
+--echo #
+--echo # Continue the INSERT once CREATE waits for exclusive lock
+SET DEBUG_SYNC= 'mdl_acquire_exclusive_locks_wait SIGNAL finish';
+--echo # Try to create that table.
+--send CREATE TABLE t1 LIKE t2
+
+--echo # Connection 2
+--echo # Insert fails
+connection con2;
+--error ER_NO_SUCH_TABLE
+--reap
+
+--echo # Connection 1
+connection default;
+--reap
+SET DEBUG_SYNC= 'RESET';
+SHOW TABLES;
+
+DROP TABLE t2;
+disconnect con2;
+--disable_warnings
+DROP TABLE IF EXISTS t1;
+--enable_warnings
+
+set global log_output=@save_log_output;
+
+
+--echo #
 --echo # Bug #46044 "MDL deadlock on LOCK TABLE + CREATE TABLE HIGH_PRIORITY
 --echo #             FOR UPDATE"
 --echo #
diff --git a/mysql-test/t/merge.test b/mysql-test/t/merge.test
index 2738f79247f..b9e6813a4df 100644
--- a/mysql-test/t/merge.test
+++ b/mysql-test/t/merge.test
@@ -846,7 +846,8 @@ SHOW CREATE TABLE t3;
 DROP TABLE t1, t2;
 #
 --echo #
---echo # CREATE ... LIKE
+--echo # Bug#37371 "CREATE TABLE LIKE merge loses UNION parameter"
+--echo # Demonstrate that this is no longer the case.
 --echo #
 --echo # 1. Create like.
 CREATE TABLE t1 (c1 INT);
@@ -858,20 +859,24 @@ INSERT INTO t2 VALUES (2);
 INSERT INTO t3 VALUES (3);
 CREATE TABLE t4 LIKE t3;
 SHOW CREATE TABLE t4;
---error ER_OPEN_AS_READONLY
 INSERT INTO t4 VALUES (4);
 DROP TABLE t4;
 --echo #
 --echo # 1. Create like with locked tables.
 LOCK TABLES t3 WRITE, t2 WRITE, t1 WRITE;
+--error ER_TABLE_NOT_LOCKED
 CREATE TABLE t4 LIKE t3;
 --error ER_TABLE_NOT_LOCKED
 SHOW CREATE TABLE t4;
 --error ER_TABLE_NOT_LOCKED
 INSERT INTO t4 VALUES (4);
-UNLOCK TABLES;
+CREATE TEMPORARY TABLE t4 LIKE t3;
+--error ER_WRONG_MRG_TABLE
 SHOW CREATE TABLE t4;
---error ER_OPEN_AS_READONLY
+--error ER_WRONG_MRG_TABLE
+INSERT INTO t4 VALUES (4);
+UNLOCK TABLES;
+--error ER_WRONG_MRG_TABLE
 INSERT INTO t4 VALUES (4);
 DROP TABLE t4;
 #
diff --git a/mysql-test/t/ps_ddl.test b/mysql-test/t/ps_ddl.test
index 1ba193983b2..fe17bca1eba 100644
--- a/mysql-test/t/ps_ddl.test
+++ b/mysql-test/t/ps_ddl.test
@@ -1493,27 +1493,24 @@ execute stmt;
 call p_verify_reprepare_count(0);
 drop table t1;
 deallocate prepare stmt;
---echo # XXX: no validation of the first table in case of
---echo # CREATE TEMPORARY TABLE. This is a shortcoming of the current code,
---echo # but since validation is not strictly necessary, nothing is done
---echo # about it.
---echo # Will be fixed as part of work on Bug#21431 "Incomplete support of
---echo # temporary tables"
 create table t1 (a int);
 insert into t1 (a) values (1);
 prepare stmt from "create temporary table if not exists t2 as select * from t1";
 execute stmt;
 drop table t2;
 execute stmt;
+call p_verify_reprepare_count(0);
 execute stmt;
+call p_verify_reprepare_count(1);
 select * from t2;
 execute stmt;
+call p_verify_reprepare_count(0);
 select * from t2;
 drop table t2;
 create temporary table t2 (a varchar(10));
 execute stmt;
 select * from t2;
-call p_verify_reprepare_count(0);
+call p_verify_reprepare_count(1);
 drop table t1;
 create table t1 (x int);
 execute stmt;
diff --git a/sql/mdl.cc b/sql/mdl.cc
index b624b0658ed..a883b21423e 100644
--- a/sql/mdl.cc
+++ b/sql/mdl.cc
@@ -1051,6 +1051,9 @@ MDL_ticket::upgrade_shared_lock_to_exclusive()
         signalled|= notify_shared_lock(thd, conflicting_ticket);
     }
 
+    /* There is a shared or exclusive lock on the object. */
+    DEBUG_SYNC(thd, "mdl_upgrade_shared_lock_to_exclusive_wait");
+
     if (signalled)
       pthread_cond_wait(&COND_mdl, &LOCK_mdl);
     else
diff --git a/sql/mysql_priv.h b/sql/mysql_priv.h
index 03a4462f199..b6ffcb238b2 100644
--- a/sql/mysql_priv.h
+++ b/sql/mysql_priv.h
@@ -1033,6 +1033,7 @@ bool mysql_new_select(LEX *lex, bool move_down);
 void create_select_for_variable(const char *var_name);
 void mysql_init_multi_delete(LEX *lex);
 bool multi_delete_set_locks_and_link_aux_tables(LEX *lex);
+void create_table_set_open_action_and_adjust_tables(LEX *lex);
 void init_max_user_conn(void);
 void init_update_queries(void);
 void free_max_user_conn(void);
@@ -1170,10 +1171,9 @@ int prepare_create_field(Create_field *sql_field,
 			 longlong table_flags);
 CHARSET_INFO* get_sql_field_charset(Create_field *sql_field,
                                     HA_CREATE_INFO *create_info);
-bool mysql_create_table(THD *thd,const char *db, const char *table_name,
+bool mysql_create_table(THD *thd, TABLE_LIST *create_table,
                         HA_CREATE_INFO *create_info,
-                        Alter_info *alter_info,
-                        bool tmp_table, uint select_field_count);
+                        Alter_info *alter_info);
 bool mysql_create_table_no_lock(THD *thd, const char *db,
                                 const char *table_name,
                                 HA_CREATE_INFO *create_info,
diff --git a/sql/sql_acl.cc b/sql/sql_acl.cc
index fdcd68cc2ea..34680855337 100644
--- a/sql/sql_acl.cc
+++ b/sql/sql_acl.cc
@@ -689,8 +689,7 @@ my_bool acl_reload(THD *thd)
   tables[0].next_local= tables[0].next_global= tables+1;
   tables[1].next_local= tables[1].next_global= tables+2;
   tables[0].lock_type=tables[1].lock_type=tables[2].lock_type=TL_READ;
-  tables[0].skip_temporary= tables[1].skip_temporary=
-    tables[2].skip_temporary= TRUE;
+  tables[0].open_type= tables[1].open_type= tables[2].open_type= OT_BASE_ONLY;
   init_mdl_requests(tables);
 
   if (simple_open_n_lock_tables(thd, tables))
@@ -3797,7 +3796,7 @@ static my_bool grant_reload_procs_priv(THD *thd)
   table.init_one_table("mysql", 5, "procs_priv",
                        strlen("procs_priv"), "procs_priv",
                        TL_READ);
-  table.skip_temporary= 1;
+  table.open_type= OT_BASE_ONLY;
 
   if (simple_open_n_lock_tables(thd, &table))
   {
@@ -3863,7 +3862,7 @@ my_bool grant_reload(THD *thd)
   tables[0].db= tables[1].db= (char *) "mysql";
   tables[0].next_local= tables[0].next_global= tables+1;
   tables[0].lock_type= tables[1].lock_type= TL_READ;
-  tables[0].skip_temporary= tables[1].skip_temporary= TRUE;
+  tables[0].open_type= tables[1].open_type= OT_BASE_ONLY;
   init_mdl_requests(tables);
 
   /*
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index 08ec0c8a2ed..7a3adc89ea9 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -2437,7 +2437,8 @@ bool open_table(THD *thd, TABLE_LIST *table_list, MEM_ROOT *mem_root,
     same name. This block implements the behaviour.
     TODO: move this block into a separate function.
   */
-  if (!table_list->skip_temporary && ! (flags & MYSQL_OPEN_SKIP_TEMPORARY))
+  if (table_list->open_type != OT_BASE_ONLY &&
+      ! (flags & MYSQL_OPEN_SKIP_TEMPORARY))
   {
     for (table= thd->temporary_tables; table ; table=table->next)
     {
@@ -2469,10 +2470,16 @@ bool open_table(THD *thd, TABLE_LIST *table_list, MEM_ROOT *mem_root,
     }
   }
 
-  if (flags & MYSQL_OPEN_TEMPORARY_ONLY)
+  if (table_list->open_type == OT_TEMPORARY_ONLY ||
+      (flags & MYSQL_OPEN_TEMPORARY_ONLY))
   {
-    my_error(ER_NO_SUCH_TABLE, MYF(0), table_list->db, table_list->table_name);
-    DBUG_RETURN(TRUE);
+    if (table_list->open_strategy == TABLE_LIST::OPEN_NORMAL)
+    {
+      my_error(ER_NO_SUCH_TABLE, MYF(0), table_list->db, table_list->table_name);
+      DBUG_RETURN(TRUE);
+    }
+    else
+      DBUG_RETURN(FALSE);
   }
 
   /*
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index 171c5e2cee0..8fd704c4f71 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -3415,46 +3415,43 @@ void select_insert::abort() {
   CREATE TABLE (SELECT) ...
 ***************************************************************************/
 
-/*
+/**
   Create table from lists of fields and items (or just return TABLE
   object for pre-opened existing table).
 
-  SYNOPSIS
-    create_table_from_items()
-      thd          in     Thread object
-      create_info  in     Create information (like MAX_ROWS, ENGINE or
-                          temporary table flag)
-      create_table in     Pointer to TABLE_LIST object providing database
-                          and name for table to be created or to be open
-      alter_info   in/out Initial list of columns and indexes for the table
-                          to be created
-      items        in     List of items which should be used to produce rest
-                          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
-                          (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.
-      hooks
-
-  NOTES
-    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.
-
+  @param thd           [in]     Thread object
+  @param create_info   [in]     Create information (like MAX_ROWS, ENGINE or
+                                temporary table flag)
+  @param create_table  [in]     Pointer to TABLE_LIST object providing database
+                                and name for table to be created or to be open
+  @param alter_info    [in/out] Initial list of columns and indexes for the
+                                table to be created
+  @param items         [in]     List of items which should be used to produce
+                                rest of fields for the table (corresponding
+                                fields will be added to the end of
+                                alter_info->create_list)
+  @param lock          [out]    Pointer to the MYSQL_LOCK object for table
+                                created will be returned in this parameter.
+                                Since this table is not included in THD::lock
+                                caller is responsible for explicitly unlocking
+                                this table.
+  @param hooks         [in]     Hooks to be invoked before and after obtaining
+                                table lock on the table being created.
+
+  @note
+    This function assumes 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 an exclusive
+    metadata lock was acquired on table so we can safely create, open and
+    lock table in it (we don't acquire metadata lock if this create is
+    for temporary table).
+
+  @note
     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
-    0         Error
+  @retval non-zero  Pointer to TABLE object for table created or opened
+  @retval 0         Error
 */
 
 static TABLE *create_table_from_items(THD *thd, HA_CREATE_INFO *create_info,
@@ -3529,14 +3526,12 @@ static TABLE *create_table_from_items(THD *thd, HA_CREATE_INFO *create_info,
     open_table().
   */
   {
-    tmp_disable_binlog(thd);
     if (!mysql_create_table_no_lock(thd, create_table->db,
                                     create_table->table_name,
                                     create_info, alter_info, 0,
                                     select_field_count))
     {
-      if (create_info->table_existed &&
-          !(create_info->options & HA_LEX_CREATE_TMP_TABLE))
+      if (create_info->table_existed)
       {
         /*
           This means that someone created table underneath server
@@ -3572,8 +3567,7 @@ static TABLE *create_table_from_items(THD *thd, HA_CREATE_INFO *create_info,
       {
         Open_table_context ot_ctx_unused(thd);
         if (open_table(thd, create_table, thd->mem_root, &ot_ctx_unused,
-                       MYSQL_OPEN_TEMPORARY_ONLY) &&
-            !create_info->table_existed)
+                       MYSQL_OPEN_TEMPORARY_ONLY))
         {
           /*
             This shouldn't happen as creation of temporary table should make
@@ -3586,7 +3580,6 @@ static TABLE *create_table_from_items(THD *thd, HA_CREATE_INFO *create_info,
           table= create_table->table;
       }
     }
-    reenable_binlog(thd);
     if (!table)                                   // open failed
       DBUG_RETURN(0);
   }
@@ -3610,9 +3603,7 @@ static TABLE *create_table_from_items(THD *thd, HA_CREATE_INFO *create_info,
       mysql_unlock_tables(thd, *lock);
       *lock= 0;
     }
-
-    if (!create_info->table_existed)
-      drop_open_table(thd, table, create_table->db, create_table->table_name);
+    drop_open_table(thd, table, create_table->db, create_table->table_name);
     DBUG_RETURN(0);
   }
   DBUG_RETURN(table);
@@ -3704,7 +3695,7 @@ select_create::prepare(List<Item> &values, SELECT_LEX_UNIT *u)
 
   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)
+  if (create_table->table)
   {
     /* Table already exists and was open at open_and_lock_tables() stage. */
     if (create_info->options & HA_LEX_CREATE_IF_NOT_EXISTS)
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index cafc70dc2ee..e462556c133 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -2245,9 +2245,9 @@ case SQLCOM_PREPARE:
   {
     DBUG_ASSERT(first_table == all_tables && first_table != 0);
     bool link_to_local;
-    // Skip first table, which is the table we are creating
-    TABLE_LIST *create_table= lex->unlink_first_table(&link_to_local);
-    TABLE_LIST *select_tables= lex->query_tables;
+    TABLE_LIST *create_table= first_table;
+    TABLE_LIST *select_tables= lex->create_last_non_select_table->next_global;
+
     /*
       Code below (especially in mysql_create_table() and select_create
       methods) may modify HA_CREATE_INFO structure in LEX, so we have to
@@ -2327,6 +2327,10 @@ case SQLCOM_PREPARE:
     }
 #endif
 
+    /* Set strategies: reset default or 'prepared' values. */
+    create_table->open_strategy= TABLE_LIST::OPEN_IF_EXISTS;
+    create_table->lock_strategy= TABLE_LIST::EXCLUSIVE_DOWNGRADABLE_MDL;
+
     /*
       Close any open handlers for the table
     */
@@ -2389,15 +2393,8 @@ case SQLCOM_PREPARE:
         goto end_with_restore_list;
       }
 
-      if (!(create_info.options & HA_LEX_CREATE_TMP_TABLE))
-      {
-        lex->link_first_table_back(create_table, link_to_local);
-        /* Set strategies: reset default or 'prepared' values. */
-        create_table->open_strategy= TABLE_LIST::OPEN_IF_EXISTS;
-        create_table->lock_strategy= TABLE_LIST::EXCLUSIVE_DOWNGRADABLE_MDL;
-      }
-
-      if (!(res= open_and_lock_tables(thd, lex->query_tables)))
+      if (!(res= open_and_lock_tables_derived(thd, lex->query_tables, TRUE,
+                                              MYSQL_OPEN_TAKE_UPGRADABLE_MDL)))
       {
         /*
           Is table which we are changing used somewhere in other parts
@@ -2406,7 +2403,6 @@ case SQLCOM_PREPARE:
         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);
@@ -2433,6 +2429,13 @@ case SQLCOM_PREPARE:
         }
 
         /*
+          Remove target table from main select and name resolution
+          context. This can't be done earlier as it will break view merging in
+          statements like "CREATE TABLE IF NOT EXISTS existing_view SELECT".
+        */
+        lex->unlink_first_table(&link_to_local);
+
+        /*
           select_create is currently not re-execution friendly and
           needs to be created for every execution of a PS/SP.
         */
@@ -2451,33 +2454,32 @@ case SQLCOM_PREPARE:
           res= handle_select(thd, lex, result, 0);
           delete result;
         }
+        
+        lex->link_first_table_back(create_table, link_to_local);
       }
-      else if (!(create_info.options & HA_LEX_CREATE_TMP_TABLE))
-        create_table= lex->unlink_first_table(&link_to_local);
-
     }
     else
     {
       /* So that CREATE TEMPORARY TABLE gets to binlog at commit/rollback */
       if (create_info.options & HA_LEX_CREATE_TMP_TABLE)
         thd->options|= OPTION_KEEP_LOG;
-      /* regular create */
       if (create_info.options & HA_LEX_CREATE_TABLE_LIKE)
+      {
+        /* CREATE TABLE ... LIKE ... */
         res= mysql_create_like_table(thd, create_table, select_tables,
                                      &create_info);
+      }
       else
       {
-        res= mysql_create_table(thd, create_table->db,
-                                create_table->table_name, &create_info,
-                                &alter_info, 0, 0);
+        /* Regular CREATE TABLE */
+        res= mysql_create_table(thd, create_table,
+                                &create_info, &alter_info);
       }
       if (!res)
-	my_ok(thd);
+        my_ok(thd);
     }
 
-    /* put tables back for PS rexecuting */
 end_with_restore_list:
-    lex->link_first_table_back(create_table, link_to_local);
     break;
   }
   case SQLCOM_CREATE_INDEX:
@@ -2705,7 +2707,8 @@ end_with_restore_list:
         }
 
         /* Ignore temporary tables if this is "SHOW CREATE VIEW" */
-        first_table->skip_temporary= 1;
+        first_table->open_type= OT_BASE_ONLY;
+
       }
       else
       {
@@ -7195,6 +7198,34 @@ bool insert_precheck(THD *thd, TABLE_LIST *tables)
 
 
 /**
+   Set proper open mode and table type for element representing target table
+   of CREATE TABLE statement, also adjust statement table list if necessary.
+*/
+
+void create_table_set_open_action_and_adjust_tables(LEX *lex)
+{
+  TABLE_LIST *create_table= lex->query_tables;
+
+  if (lex->create_info.options & HA_LEX_CREATE_TMP_TABLE)
+    create_table->open_type= OT_TEMPORARY_ONLY;
+  else if (!lex->select_lex.item_list.elements)
+    create_table->open_type= OT_BASE_ONLY;
+
+  if (!lex->select_lex.item_list.elements)
+  {
+    /*
+      Avoid opening and locking target table for ordinary CREATE TABLE
+      or CREATE TABLE LIKE for write (unlike in CREATE ... SELECT we
+      won't do any insertions in it anyway). Not doing this causes
+      problems when running CREATE TABLE IF NOT EXISTS for already
+      existing log table.
+    */
+    create_table->lock_type= TL_READ;
+  }
+}
+
+
+/**
   CREATE TABLE query pre-check.
 
   @param thd			Thread handler
diff --git a/sql/sql_partition.cc b/sql/sql_partition.cc
index e0461533dde..718471cc1b6 100644
--- a/sql/sql_partition.cc
+++ b/sql/sql_partition.cc
@@ -4175,39 +4175,22 @@ bool mysql_unpack_partition(THD *thd,
              ha_resolve_storage_engine_name(default_db_type)));
   if (is_create_table_ind && old_lex->sql_command == SQLCOM_CREATE_TABLE)
   {
-    if (old_lex->create_info.options & HA_LEX_CREATE_TABLE_LIKE)
-    {
-      /*
-        This code is executed when we create table in CREATE TABLE t1 LIKE t2.
-        old_lex->query_tables contains table list element for t2 and the table
-        we are opening has name t1.
-      */
-      if (partition_default_handling(table, part_info, FALSE,
-                                     old_lex->query_tables->table->s->path.str))
-      {
-        result= TRUE;
-        goto end;
-      }
-    }
-    else
-    {
-      /*
-        When we come here we are doing a create table. In this case we
-        have already done some preparatory work on the old part_info
-        object. We don't really need this new partition_info object.
-        Thus we go back to the old partition info object.
-        We need to free any memory objects allocated on item_free_list
-        by the parser since we are keeping the old info from the first
-        parser call in CREATE TABLE.
-        We'll ensure that this object isn't put into table cache also
-        just to ensure we don't get into strange situations with the
-        item objects.
-      */
-      thd->free_items();
-      part_info= thd->work_part_info;
-      table->s->version= 0UL;
-      *work_part_info_used= true;
-    }
+    /*
+      When we come here we are doing a create table. In this case we
+      have already done some preparatory work on the old part_info
+      object. We don't really need this new partition_info object.
+      Thus we go back to the old partition info object.
+      We need to free any memory objects allocated on item_free_list
+      by the parser since we are keeping the old info from the first
+      parser call in CREATE TABLE.
+      We'll ensure that this object isn't put into table cache also
+      just to ensure we don't get into strange situations with the
+      item objects.
+    */
+    thd->free_items();
+    part_info= thd->work_part_info;
+    table->s->version= 0UL;
+    *work_part_info_used= true;
   }
   table->part_info= part_info;
   table->file->set_part_info(part_info);
diff --git a/sql/sql_prepare.cc b/sql/sql_prepare.cc
index 9c7949eaf1d..27fdd1e2a8d 100644
--- a/sql/sql_prepare.cc
+++ b/sql/sql_prepare.cc
@@ -1660,39 +1660,35 @@ static bool mysql_test_create_table(Prepared_statement *stmt)
   LEX *lex= stmt->lex;
   SELECT_LEX *select_lex= &lex->select_lex;
   bool res= FALSE;
-  /* Skip first table, which is the table we are creating */
   bool link_to_local;
-  TABLE_LIST *create_table= lex->unlink_first_table(&link_to_local);
-  TABLE_LIST *tables= lex->query_tables;
+  TABLE_LIST *create_table= lex->query_tables;
+  TABLE_LIST *tables= lex->create_last_non_select_table->next_global;
 
   if (create_table_precheck(thd, tables, create_table))
     DBUG_RETURN(TRUE);
 
+   /*
+     The open and lock strategies will be set again once the
+     statement is executed. These values are only meaningful
+     for the prepare phase.
+   */
+  create_table->open_strategy= TABLE_LIST::OPEN_IF_EXISTS;
+  create_table->lock_strategy= TABLE_LIST::SHARED_MDL;
+
   if (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);
-      /*
-        The open and lock strategies will be set again once the
-        statement is executed. These values are only meaningful
-        for the prepare phase.
-      */
-      create_table->open_strategy= TABLE_LIST::OPEN_IF_EXISTS;
-      create_table->lock_strategy= TABLE_LIST::SHARED_MDL;
-    }
-
     if (open_normal_and_derived_tables(stmt->thd, lex->query_tables, 0))
       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;
 
+    lex->unlink_first_table(&link_to_local);
+
     res= select_like_stmt_test(stmt, 0, 0);
+
+    lex->link_first_table_back(create_table, &link_to_local);
   }
-  else if (lex->create_info.options & HA_LEX_CREATE_TABLE_LIKE)
+  else
   {
     /*
       Check that the source table exist, and also record
@@ -1704,8 +1700,6 @@ static bool mysql_test_create_table(Prepared_statement *stmt)
       DBUG_RETURN(TRUE);
   }
 
-  /* put tables back for PS rexecuting */
-  lex->link_first_table_back(create_table, link_to_local);
   DBUG_RETURN(res);
 }
 
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 44a0d21580f..e9474d9add6 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -4084,7 +4084,6 @@ bool mysql_create_table_no_lock(THD *thd,
     thd->thread_specific_used= TRUE;
   }
 
-  write_create_table_bin_log(thd, create_info, internal_tmp_table);
   error= FALSE;
 unlock_and_end:
   pthread_mutex_unlock(&LOCK_open);
@@ -4109,21 +4108,18 @@ warn:
   Database and name-locking aware wrapper for mysql_create_table_no_lock(),
 */
 
-bool mysql_create_table(THD *thd, const char *db, const char *table_name,
+bool mysql_create_table(THD *thd, TABLE_LIST *create_table,
                         HA_CREATE_INFO *create_info,
-                        Alter_info *alter_info,
-                        bool internal_tmp_table,
-                        uint select_field_count)
+                        Alter_info *alter_info)
 {
-  MDL_request target_mdl_request;
-  bool has_target_mdl_lock= FALSE;
   bool result;
   DBUG_ENTER("mysql_create_table");
 
   /* Wait for any database locks */
   pthread_mutex_lock(&LOCK_lock_db);
   while (!thd->killed &&
-         my_hash_search(&lock_db_cache,(uchar*) db, strlen(db)))
+         my_hash_search(&lock_db_cache, (uchar*)create_table->db,
+                        create_table->db_length))
   {
     wait_for_condition(thd, &LOCK_lock_db, &COND_refresh);
     pthread_mutex_lock(&LOCK_lock_db);
@@ -4137,47 +4133,47 @@ bool mysql_create_table(THD *thd, const char *db, const char *table_name,
   creating_table++;
   pthread_mutex_unlock(&LOCK_lock_db);
 
-  if (!(create_info->options & HA_LEX_CREATE_TMP_TABLE))
+  /*
+    Open or obtain an exclusive metadata lock on table being created.
+  */
+  if (open_and_lock_tables_derived(thd, thd->lex->query_tables, FALSE,
+                                   MYSQL_OPEN_TAKE_UPGRADABLE_MDL))
   {
-    target_mdl_request.init(MDL_key::TABLE, db, table_name, MDL_EXCLUSIVE);
-    if (thd->mdl_context.try_acquire_exclusive_lock(&target_mdl_request))
-    {
-      result= TRUE;
-      goto unlock;
-    }
-    if (target_mdl_request.ticket == NULL)
-    {
-      /* Table exists and is locked by some other thread. */
-      if (create_info->options & HA_LEX_CREATE_IF_NOT_EXISTS)
-      {
-        push_warning_printf(thd, MYSQL_ERROR::WARN_LEVEL_NOTE,
-                            ER_TABLE_EXISTS_ERROR, ER(ER_TABLE_EXISTS_ERROR),
-                            table_name);
-        create_info->table_existed= 1;
-        result= FALSE;
-        write_create_table_bin_log(thd, create_info, internal_tmp_table);
-      }
-      else
-      {
-        my_error(ER_TABLE_EXISTS_ERROR,MYF(0),table_name);
-        result= TRUE;
-      }
-      goto unlock;
-    }
-    /* Got lock. */
-    DEBUG_SYNC(thd, "locked_table_name");
-    has_target_mdl_lock= TRUE;
+    result= TRUE;
+    goto unlock;
   }
 
-  result= mysql_create_table_no_lock(thd, db, table_name, create_info,
-                                     alter_info,
-                                     internal_tmp_table,
-                                     select_field_count);
+  /* Got lock. */
+  DEBUG_SYNC(thd, "locked_table_name");
 
-unlock:
-  if (has_target_mdl_lock)
-    thd->mdl_context.release_lock(target_mdl_request.ticket);
+  result= mysql_create_table_no_lock(thd, create_table->db,
+                                     create_table->table_name, create_info,
+                                     alter_info, FALSE, 0);
+
+  /*
+    Don't write statement if:
+    - Table creation has failed
+    - Table has already existed
+    - Row-based logging is used and we are creating a temporary table
+    Otherwise, the statement shall be binlogged.
+  */
+  if (!result &&
+      !create_info->table_existed &&
+      (!thd->current_stmt_binlog_row_based ||
+       (thd->current_stmt_binlog_row_based &&
+        !(create_info->options & HA_LEX_CREATE_TMP_TABLE))))
+    write_bin_log(thd, TRUE, thd->query(), thd->query_length());
 
+  if (!(create_info->options & HA_LEX_CREATE_TMP_TABLE))
+  {
+    /*
+      close_thread_tables() takes care about both closing open tables (which
+      might be still around in case of error) and releasing metadata locks.
+    */
+    close_thread_tables(thd);
+  }
+
+unlock:
   pthread_mutex_lock(&LOCK_lock_db);
   if (!--creating_table && creating_database)
     pthread_cond_signal(&COND_refresh);
@@ -5158,55 +5154,6 @@ bool mysql_preload_keys(THD* thd, TABLE_LIST* tables)
 }
 
 
-
-/**
-  @brief          Create frm file based on I_S table
-
-  @param[in]      thd                      thread handler
-  @param[in]      schema_table             I_S table           
-  @param[in]      dst_path                 path where frm should be created
-  @param[in]      create_info              Create info
-
-  @return         Operation status
-    @retval       0                        success
-    @retval       1                        error
-*/
-
-
-bool mysql_create_like_schema_frm(THD* thd, TABLE_LIST* schema_table,
-                                  char *dst_path, HA_CREATE_INFO *create_info)
-{
-  HA_CREATE_INFO local_create_info;
-  Alter_info alter_info;
-  bool tmp_table= (create_info->options & HA_LEX_CREATE_TMP_TABLE);
-  uint keys= schema_table->table->s->keys;
-  uint db_options= 0;
-  DBUG_ENTER("mysql_create_like_schema_frm");
-
-  bzero((char*) &local_create_info, sizeof(local_create_info));
-  local_create_info.db_type= schema_table->table->s->db_type();
-  local_create_info.row_type= schema_table->table->s->row_type;
-  local_create_info.default_table_charset=default_charset_info;
-  alter_info.flags= (ALTER_CHANGE_COLUMN | ALTER_RECREATE);
-  schema_table->table->use_all_columns();
-  if (mysql_prepare_alter_table(thd, schema_table->table,
-                                &local_create_info, &alter_info))
-    DBUG_RETURN(1);
-  if (mysql_prepare_create_table(thd, &local_create_info, &alter_info,
-                                 tmp_table, &db_options,
-                                 schema_table->table->file,
-                                 &schema_table->table->s->key_info, &keys, 0))
-    DBUG_RETURN(1);
-  local_create_info.max_rows= 0;
-  if (mysql_create_frm(thd, dst_path, NullS, NullS,
-                       &local_create_info, alter_info.create_list,
-                       keys, schema_table->table->s->key_info,
-                       schema_table->table->file))
-    DBUG_RETURN(1);
-  DBUG_RETURN(0);
-}
-
-
 /*
   Create a table identical to the specified table
 
@@ -5225,12 +5172,8 @@ bool mysql_create_like_schema_frm(THD* thd, TABLE_LIST* schema_table,
 bool mysql_create_like_table(THD* thd, TABLE_LIST* table, TABLE_LIST* src_table,
                              HA_CREATE_INFO *create_info)
 {
-  char src_path[FN_REFLEN + 1], dst_path[FN_REFLEN + 1];
-  uint dst_path_length;
-  bool has_mdl_lock= FALSE;
-  char *db= table->db;
-  char *table_name= table->table_name;
-  int  err;
+  HA_CREATE_INFO local_create_info;
+  Alter_info local_alter_info;
   bool res= TRUE;
   uint not_used;
 #ifdef WITH_PARTITION_STORAGE_ENGINE
@@ -5242,161 +5185,63 @@ bool mysql_create_like_table(THD* thd, TABLE_LIST* table, TABLE_LIST* src_table,
 
 
   /*
-    By opening source table and thus acquiring shared metadata lock on it
-    we guarantee that it exists and no concurrent DDL operation will mess
-    with it. Later we also take an exclusive metadata lock on target table
-    name, which makes copying of .frm file, call to ha_create_table() and
-    binlogging atomic against concurrent DML and DDL operations on target
-    table. Thus by holding both these "locks" we ensure that our statement
-    is properly isolated from all concurrent operations which matter.
-  */
-  if (open_tables(thd, &src_table, &not_used, 0))
-    DBUG_RETURN(TRUE);
-
-  /*
-    For bug#25875, Newly created table through CREATE TABLE .. LIKE
-                   has no ndb_dd attributes;
-    Add something to get possible tablespace info from src table,
-    it can get valid tablespace name only for disk-base ndb table
+    We the open source table to get its description in HA_CREATE_INFO
+    and Alter_info objects. This also acquires a shared metadata lock
+    on this table which ensures that no concurrent DDL operation will
+    mess with it.
+    Also in case when we create non-temporary table open_tables()
+    call obtains an exclusive metadata lock on target table ensuring
+    that we can safely perform table creation.
+    Thus by holding both these locks we ensure that our statement is
+    properly isolated from all concurrent operations which matter.
   */
-  if ((src_table->table->file->get_tablespace_name(thd, ts_name, FN_LEN)))
-  {
-    create_info->tablespace= ts_name;
-    create_info->storage_media= HA_SM_DISK;
-  }
-
-  strxmov(src_path, src_table->table->s->path.str, reg_ext, NullS);
+  if (open_tables(thd, &thd->lex->query_tables, &not_used, 0))
+    goto err;
+  src_table->table->use_all_columns();
 
-  DBUG_EXECUTE_IF("sleep_create_like_before_check_if_exists", my_sleep(6000000););
+  /* Fill HA_CREATE_INFO and Alter_info with description of source table. */
+  bzero((char*) &local_create_info, sizeof(local_create_info));
+  local_create_info.db_type= src_table->table->s->db_type();
+  local_create_info.row_type= src_table->table->s->row_type;
+  if (mysql_prepare_alter_table(thd, src_table->table, &local_create_info,
+                                &local_alter_info))
+    goto err;
+#ifdef WITH_PARTITION_STORAGE_ENGINE
+  /* Partition info is not handled by mysql_prepare_alter_table() call. */
+  if (src_table->table->part_info)
+    thd->work_part_info= src_table->table->part_info->get_clone();
+#endif
 
   /*
-    Check that destination tables does not exist. Note that its name
-    was already checked when it was added to the table list.
-  */
-  if (create_info->options & HA_LEX_CREATE_TMP_TABLE)
-  {
-    if (find_temporary_table(thd, db, table_name))
-      goto table_exists;
-    dst_path_length= build_tmptable_filename(thd, dst_path, sizeof(dst_path));
-    create_info->table_options|= HA_CREATE_DELAY_KEY_WRITE;
-  }
-  else
-  {
-    table->mdl_request.init(MDL_key::TABLE, db, table_name, MDL_EXCLUSIVE);
-    if (thd->mdl_context.try_acquire_exclusive_lock(&table->mdl_request))
-      DBUG_RETURN(TRUE);
-
-    if (table->mdl_request.ticket == NULL)
-      goto table_exists;
+    Adjust description of source table before using it for creation of
+    target table.
 
-    DEBUG_SYNC(thd, "locked_table_name");
-    has_mdl_lock= TRUE;
-
-    dst_path_length= build_table_filename(dst_path, sizeof(dst_path) - 1,
-                                          db, table_name, reg_ext, 0);
-    if (!access(dst_path, F_OK))
-      goto table_exists;
-  }
-
-  DBUG_EXECUTE_IF("sleep_create_like_before_copy", my_sleep(6000000););
-
-  if (opt_sync_frm && !(create_info->options & HA_LEX_CREATE_TMP_TABLE))
-    flags|= MY_SYNC;
-
-  /*
-    Create a new table by copying from source table
-    and sync the new table if the flag MY_SYNC is set
-
-    TODO: Obtaining LOCK_open mutex here is actually a legacy from the
-          times when some operations (e.g. I_S implementation) ignored
-          exclusive metadata lock on target table. Also some engines
-          (e.g. NDB cluster) require that LOCK_open should be held
-          during the call to ha_create_table() (See bug #28614 for more
-          info). So we should double check and probably fix this code
-          to not acquire this mutex.
+    Similarly to SHOW CREATE TABLE we ignore MAX_ROWS attribute of
+    temporary table which represents I_S table.
   */
-  pthread_mutex_lock(&LOCK_open);
   if (src_table->schema_table)
-  {
-    if (mysql_create_like_schema_frm(thd, src_table, dst_path, create_info))
-    {
-      pthread_mutex_unlock(&LOCK_open);
-      goto err;
-    }
-  }
-  else if (my_copy(src_path, dst_path, flags))
-  {
-    if (my_errno == ENOENT)
-      my_error(ER_BAD_DB_ERROR,MYF(0),db);
-    else
-      my_error(ER_CANT_CREATE_FILE,MYF(0),dst_path,my_errno);
-    pthread_mutex_unlock(&LOCK_open);
+    local_create_info.max_rows= 0;
+  /* Set IF NOT EXISTS option as in the CREATE TABLE LIKE statement. */
+  local_create_info.options|= create_info->options&HA_LEX_CREATE_IF_NOT_EXISTS;
+  /* Replace type of source table with one specified in the statement. */
+  local_create_info.options&= ~HA_LEX_CREATE_TMP_TABLE;
+  local_create_info.options|= create_info->options & HA_LEX_CREATE_TMP_TABLE;
+  /* Reset auto-increment counter for the new table. */
+  local_create_info.auto_increment_value= 0;
+
+  if ((res= mysql_create_table_no_lock(thd, table->db, table->table_name,
+                                       &local_create_info, &local_alter_info,
+                                       FALSE, 0)) ||
+      local_create_info.table_existed)
     goto err;
-  }
 
   /*
-    As mysql_truncate don't work on a new table at this stage of
-    creation, instead create the table directly (for both normal
-    and temporary tables).
-  */
-#ifdef WITH_PARTITION_STORAGE_ENGINE
-  /*
-    For partitioned tables we need to copy the .par file as well since
-    it is used in open_table_def to even be able to create a new handler.
-    There is no way to find out here if the original table is a
-    partitioned table so we copy the file and ignore any errors.
+    Ensure that we have an exclusive lock on target table if we are creating
+    non-temporary table.
   */
-  fn_format(tmp_path, dst_path, reg_ext, ".par", MYF(MY_REPLACE_EXT));
-  strmov(dst_path, tmp_path);
-  fn_format(tmp_path, src_path, reg_ext, ".par", MYF(MY_REPLACE_EXT));
-  strmov(src_path, tmp_path);
-  my_copy(src_path, dst_path, MYF(MY_DONT_OVERWRITE_FILE));
-#endif
-
-  DBUG_EXECUTE_IF("sleep_create_like_before_ha_create", my_sleep(6000000););
-
-  dst_path[dst_path_length - reg_ext_length]= '\0';  // Remove .frm
-  if (thd->variables.keep_files_on_create)
-    create_info->options|= HA_CREATE_KEEP_FILES;
-  err= ha_create_table(thd, dst_path, db, table_name, create_info, 1);
-  pthread_mutex_unlock(&LOCK_open);
-
-  if (create_info->options & HA_LEX_CREATE_TMP_TABLE)
-  {
-    if (err || !open_temporary_table(thd, dst_path, db, table_name, 1))
-    {
-      (void) rm_temporary_table(create_info->db_type,
-				dst_path); /* purecov: inspected */
-      goto err;     /* purecov: inspected */
-    }
-    thd->thread_specific_used= TRUE;
-  }
-  else if (err)
-  {
-    (void) quick_rm_table(create_info->db_type, db,
-			  table_name, 0); /* purecov: inspected */
-    goto err;	    /* purecov: inspected */
-  }
-
-goto binlog;
-
-table_exists:
-  if (create_info->options & HA_LEX_CREATE_IF_NOT_EXISTS)
-  {
-    char warn_buff[MYSQL_ERRMSG_SIZE];
-    my_snprintf(warn_buff, sizeof(warn_buff),
-		ER(ER_TABLE_EXISTS_ERROR), table_name);
-    push_warning(thd, MYSQL_ERROR::WARN_LEVEL_NOTE,
-		 ER_TABLE_EXISTS_ERROR,warn_buff);
-  }
-  else
-  {
-    my_error(ER_TABLE_EXISTS_ERROR, MYF(0), table_name);
-    goto err;
-  }
-
-binlog:
-  DBUG_EXECUTE_IF("sleep_create_like_before_binlogging", my_sleep(6000000););
+  DBUG_ASSERT((create_info->options & HA_LEX_CREATE_TMP_TABLE) ||
+              thd->mdl_context.is_exclusive_lock_owner(MDL_key::TABLE, table->db,
+                                                       table->table_name));
 
   /*
     We have to write the query before we unlock the tables.
@@ -5463,12 +5308,7 @@ binlog:
   else
     write_bin_log(thd, TRUE, thd->query(), thd->query_length());
 
-  res= FALSE;
-
 err:
-  if (has_mdl_lock)
-    thd->mdl_context.release_lock(table->mdl_request.ticket);
-
   DBUG_RETURN(res);
 }
 
diff --git a/sql/sql_trigger.cc b/sql/sql_trigger.cc
index 823dca5d93f..8719938d85d 100644
--- a/sql/sql_trigger.cc
+++ b/sql/sql_trigger.cc
@@ -445,7 +445,7 @@ bool mysql_create_or_drop_trigger(THD *thd, TABLE_LIST *tables, bool create)
     Also prevent DROP TRIGGER from opening temporary table which might
     shadow base table on which trigger to be dropped is defined.
   */
-  tables->skip_temporary= TRUE;
+  tables->open_type= OT_BASE_ONLY;
 
   /* Keep consistent with respect to other DDL statements */
   mysql_ha_rm_tables(thd, tables);
diff --git a/sql/sql_view.cc b/sql/sql_view.cc
index b36ff6b6743..0f920bca101 100644
--- a/sql/sql_view.cc
+++ b/sql/sql_view.cc
@@ -1259,7 +1259,7 @@ bool mysql_make_view(THD *thd, File_parser *parser, TABLE_LIST *table,
          tbl;
          tbl= (view_tables_tail= tbl)->next_global)
     {
-      tbl->skip_temporary= 1;
+      tbl->open_type= OT_BASE_ONLY;
       tbl->belong_to_view= top_view;
       tbl->referencing_view= table;
       tbl->prelocking_placeholder= table->prelocking_placeholder;
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index 826ec63a93e..1e6a7c4d50f 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -1781,6 +1781,7 @@ create:
                                   ha_resolve_storage_engine_name(lex->create_info.db_type),
                                   $5->table.str);
             }
+            create_table_set_open_action_and_adjust_tables(lex);
           }
         | CREATE opt_unique_or_fulltext INDEX_SYM ident key_alg ON
           table_ident
@@ -3921,7 +3922,7 @@ size_number:
 create2:
           '(' create2a {}
         | opt_create_table_options
-          opt_partitioning
+          opt_create_partitioning
           create3 {}
         | LIKE table_ident
           {
@@ -3955,9 +3956,9 @@ create2:
 
 create2a:
           create_field_list ')' opt_create_table_options
-          opt_partitioning
+          opt_create_partitioning
           create3 {}
-        |  opt_partitioning
+        |  opt_create_partitioning
            create_select ')'
            { Select->set_braces(1);}
            union_opt {}
@@ -3973,6 +3974,19 @@ create3:
           union_opt {}
         ;
 
+opt_create_partitioning:
+          opt_partitioning
+          {
+            /*
+              Remove all tables used in PARTITION clause from the global table
+              list. Partitioning with subqueries is not allowed anyway.
+            */
+            TABLE_LIST *last_non_sel_table= Lex->create_last_non_select_table;
+            last_non_sel_table->next_global= 0;
+            Lex->query_tables_last= &last_non_sel_table->next_global;
+          }
+        ;
+
 /*
  This part of the parser is about handling of the partition information.
 
diff --git a/sql/table.h b/sql/table.h
index eacf4f6085c..b44c2e9f3be 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -1069,6 +1069,16 @@ public:
 };
 
 
+/**
+   Type of table which can be open for an element of table list.
+*/
+
+enum enum_open_type
+{
+  OT_TEMPORARY_OR_BASE= 0, OT_TEMPORARY_ONLY, OT_BASE_ONLY
+};
+
+
 /*
   Table reference in the FROM clause.
 
@@ -1330,7 +1340,11 @@ struct TABLE_LIST
   bool		cacheable_table;	/* stop PS caching */
   /* used in multi-upd/views privilege check */
   bool		table_in_first_from_clause;
-  bool		skip_temporary;		/* this table shouldn't be temporary */
+  /**
+     Specifies which kind of table should be open for this element
+     of table list.
+  */
+  enum enum_open_type open_type;
   /* TRUE if this merged view contain auto_increment field */
   bool          contain_auto_increment;
   bool          multitable_view;        /* TRUE iff this is multitable view */