Implementation of MDEV-5491: CREATE OR REPLACE TABLE

Using CREATE OR REPLACE TABLE is be identical to

DROP TABLE IF EXISTS table_name;
CREATE TABLE ...;

Except that:

* CREATE OR REPLACE is be atomic (now one can create the same table between drop and create).
* Temporary tables will not shadow the table name for the DROP as the CREATE TABLE tells us already if we are using a temporary table or not.
* If the table was locked with LOCK TABLES, the new table will be locked with the same lock after it's created.

Implementation details:
- We don't anymore open the to-be-created table during CREATE TABLE, which the original code did.
  - There is no need to open a table we are planning to create. It's enough to check if the table exists or not.
- Removed some of duplicated code for CREATE IF NOT EXISTS.
- Give an error when using CREATE OR REPLACE with IF NOT EXISTS (conflicting options).
- As a side effect of the code changes, we don't anymore have to internally re-prepare prepared statements with CREATE TABLE if the table exists.
- Made one code path for all testing if log table are in use.
- Better error message if one tries to create/drop/alter a log table in use
- Added back disabled rpl_row_create_table test as it now seams to work and includes a lot of interesting tests.
- Added HA_LEX_CREATE_REPLACE to mark if we are using CREATE OR REPLACE
- Aligned CREATE OR REPLACE parsing code in sql_yacc.yy for TABLE and VIEW
- Changed interface for drop_temporary_table() to make it more reusable
- Changed Locked_tables_list::init_locked_tables() to work on the table object instead of the table list object. Before this it used a mix of both, which was not good.
- Locked_tables_list::unlock_locked_tables(THD *thd) now requires a valid thd argument. Old usage of calling this with 0 i changed to instead call Locked_tables_list::reset()
- Added functions Locked_tables_list:restore_lock() and Locked_tables_list::add_back_last_deleted_lock() to be able to easily add back a locked table to the lock list.
- Added restart_trans_for_tables() to be able to restart a transaction.
- DROP_ACL is required if one uses CREATE TABLE OR REPLACE.
- Added drop of normal and temporary tables in create_table_imp() if CREATE OR REPLACE was used.
- Added reacquiring of table locks in mysql_create_table() and mysql_create_like_table()




mysql-test/include/commit.inc:
  With new code we get fewer status increments
mysql-test/r/commit_1innodb.result:
  With new code we get fewer status increments
mysql-test/r/create.result:
  Added testing of create or replace with timeout
mysql-test/r/create_or_replace.result:
  Basic testing of CREATE OR REPLACE TABLE
mysql-test/r/partition_exchange.result:
  New error message
mysql-test/r/ps_ddl.result:
  Fewer reprepares with new code
mysql-test/suite/archive/discover.result:
  Don't rediscover archive tables if the .frm file exists
  (Sergei will look at this if there is a better way...)
mysql-test/suite/archive/discover.test:
  Don't rediscover archive tables if the .frm file exists
  (Sergei will look at this if there is a better way...)
mysql-test/suite/funcs_1/r/innodb_views.result:
  New error message
mysql-test/suite/funcs_1/r/memory_views.result:
  New error message
mysql-test/suite/rpl/disabled.def:
  rpl_row_create_table should now be safe to use
mysql-test/suite/rpl/r/rpl_row_create_table.result:
  Updated results after adding back disabled test
mysql-test/suite/rpl/t/rpl_create_if_not_exists.test:
  Added comment
mysql-test/suite/rpl/t/rpl_row_create_table.test:
  Added CREATE OR REPLACE TABLE test
mysql-test/t/create.test:
  Added CREATE OR REPLACE TABLE test
mysql-test/t/create_or_replace-master.opt:
  Create logs
mysql-test/t/create_or_replace.test:
  Basic testing of CREATE OR REPLACE TABLE
mysql-test/t/partition_exchange.test:
  Error number changed as we are now using same code for all log table change issues
mysql-test/t/ps_ddl.test:
  Fewer reprepares with new code
sql/handler.h:
  Moved things around a bit in a structure to get better alignment.
  Added HA_LEX_CREATE_REPLACE to mark if we are using CREATE OR REPLACE
  Added 3 elements to end of HA_CREATE_INFO to be able to store state to add backs locks in case of LOCK TABLES.
sql/log.cc:
  Reimplemented check_if_log_table():
  - Simpler and faster usage
  - Can give error messages
  
  This gives us one code path for allmost all error messages if log tables are in use
sql/log.h:
  New interface for check_if_log_table()
sql/slave.cc:
  More logging
sql/sql_alter.cc:
  New interface for check_if_log_table()
sql/sql_base.cc:
  More documentation
  Changed interface for drop_temporary_table() to make it more reusable
  Changed Locked_tables_list::init_locked_tables() to work on the table object instead of the table list object. Before this it used a mix of both, which was not good.
  Locked_tables_list::unlock_locked_tables(THD *thd) now requires a valid thd argument.  Old usage of calling this with 0 i changed to instead call Locked_tables_list::reset()
  Added functions Locked_tables_list:restore_lock() and Locked_tables_list::add_back_last_deleted_lock() to be able to easily add back a locked table to the lock list.
  Check for command number instead of open_strategy of CREATE TABLE was used.
  Added restart_trans_for_tables() to be able to restart a transaction.  This was needed in "create or replace ... select" between the drop table and the select.
sql/sql_base.h:
  Added and updated function prototypes
sql/sql_class.h:
  Added new prototypes to Locked_tables_list class
  Added extra argument to select_create to avoid double call to eof() or send_error()
  - I needed this in some edge case where the table was not created against expections.
sql/sql_db.cc:
  New interface for check_if_log_table()
sql/sql_insert.cc:
  Remember position to lock information so that we can reaquire table lock for LOCK TABLES + CREATE OR REPLACE TABLE SELECT. Later add back the lock by calling restore_lock().
  Removed one not needed indentation level in create_table_from_items()
  Ensure we don't call send_eof() or abort_result_set() twice.
sql/sql_lex.h:
  Removed variable that I temporarly added in an earlier changeset
sql/sql_parse.cc:
  Removed old test code (marked with QQ)
  Ensure that we have open_strategy set as TABLE_LIST::OPEN_STUB in CREATE TABLE
  Removed some IF NOT EXISTS code as this is now handled in create_table_table_impl().
  Set OPTION_KEEP_LOGS later. This code had to be moved as the test for IF EXISTS has changed place.
  DROP_ACL is required if one uses CREATE TABLE OR REPLACE.
sql/sql_partition_admin.cc:
  New interface for check_if_log_table()
sql/sql_rename.cc:
  New interface for check_if_log_table()
sql/sql_table.cc:
  New interface for check_if_log_table()
  Moved some code in mysql_rm_table() under a common test.
  - Safe as temporary tables doesn't have statistics.
  - !is_temporary_table(table) test was moved out from drop_temporary_table() and merged with upper level code.
  - Added drop of normal and temporary tables in create_table_imp() if CREATE OR REPLACE was used.
  - Added reacquiring of table locks in mysql_create_table() and mysql_create_like_table()
  - In mysql_create_like_table(), restore table->open_strategy() if it was changed.
  - Re-test if table was a view after opening it.
sql/sql_table.h:
  New prototype for mysql_create_table_no_lock()
sql/sql_yacc.yy:
  Added syntax for CREATE OR REPLACE TABLE
  Reuse new code for CREATE OR REPLACE VIEW
sql/table.h:
  Added name for enum type
sql/table_cache.cc:
  More DBUG

38 files changed, 1195 insertions, 408 deletions

diff --git a/mysql-test/include/commit.inc b/mysql-test/include/commit.inc
index bdb6f48f095..e72ebba8527 100644
--- a/mysql-test/include/commit.inc
+++ b/mysql-test/include/commit.inc
@@ -751,7 +751,7 @@ call p_verify_status_increment(4, 4, 4, 4);
 --echo # Sic: no table is created.
 create table if not exists t2 (a int) select 6 union select 7;
 --echo # Sic: first commits the statement, and then the transaction.
-call p_verify_status_increment(2, 0, 2, 0);
+call p_verify_status_increment(0, 0, 0, 0);
 create table t3 select a from t2;
 call p_verify_status_increment(2, 0, 4, 4);
 alter table t3 add column (b int);
diff --git a/mysql-test/r/commit_1innodb.result b/mysql-test/r/commit_1innodb.result
index 3583e8ed396..1e173221b15 100644
--- a/mysql-test/r/commit_1innodb.result
+++ b/mysql-test/r/commit_1innodb.result
@@ -830,7 +830,7 @@ create table if not exists t2 (a int) select 6 union select 7;
 Warnings:
 Note	1050	Table 't2' already exists
 # Sic: first commits the statement, and then the transaction.
-call p_verify_status_increment(2, 0, 2, 0);
+call p_verify_status_increment(0, 0, 0, 0);
 SUCCESS
 
 create table t3 select a from t2;
diff --git a/mysql-test/r/create.result b/mysql-test/r/create.result
index 7eba25d8ea3..41a2200c13f 100644
--- a/mysql-test/r/create.result
+++ b/mysql-test/r/create.result
@@ -2602,6 +2602,8 @@ create table t1 (a int, b int) select 2,2;
 ERROR 42S01: Table 't1' already exists
 create table t1 like t2;
 ERROR 42S01: Table 't1' already exists
+create or replace table t1 (a int, b int) select 2,2;
+ERROR HY000: Lock wait timeout exceeded; try restarting transaction
 select * from t1;
 a	b
 1	1
diff --git a/mysql-test/r/create_or_replace.result b/mysql-test/r/create_or_replace.result
new file mode 100644
index 00000000000..42a3d1fe17c
--- /dev/null
+++ b/mysql-test/r/create_or_replace.result
@@ -0,0 +1,293 @@
+drop table if exists t1,t2;
+CREATE TABLE t2 (a int);
+INSERT INTO t2 VALUES(1),(2),(3);
+#
+# Check first syntax and wrong usage
+#
+CREATE OR REPLACE TABLE IF NOT EXISTS t1 (a int);
+ERROR HY000: Incorrect usage of OR REPLACE and IF NOT EXISTS
+create or replace trigger trg before insert on t1 for each row set @a:=1;
+ERROR HY000: Incorrect usage of OR REPLACE and TRIGGERS / SP / EVENT
+create or replace table mysql.general_log (a int);
+ERROR HY000: You cannot 'CREATE OR REPLACE' a log table if logging is enabled
+create or replace table mysql.slow_log (a int);
+ERROR HY000: You cannot 'CREATE OR REPLACE' a log table if logging is enabled
+#
+# Usage when table doesn't exist
+#
+CREATE OR REPLACE TABLE t1 (a int);
+CREATE TABLE t1 (a int);
+ERROR 42S01: Table 't1' already exists
+DROP TABLE t1;
+CREATE OR REPLACE TEMPORARY TABLE t1 (a int);
+CREATE TEMPORARY TABLE t1 (a int, b int, c int);
+ERROR 42S01: Table 't1' already exists
+DROP TEMPORARY TABLE t1;
+#
+# Testing with temporary tables
+#
+CREATE OR REPLACE TABLE t1 (a int);
+CREATE OR REPLACE TEMPORARY TABLE t1 (a int);
+CREATE OR REPLACE TEMPORARY TABLE t1 (a int, b int);
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TEMPORARY TABLE `t1` (
+  `a` int(11) DEFAULT NULL,
+  `b` int(11) DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+DROP TEMPORARY TABLE t1;
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `a` int(11) DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+DROP TABLE t1;
+create temporary table t1 (i int) engine=InnoDB;
+create or replace temporary table t1 (a int, b int) engine=InnoDB;
+create or replace temporary table t1 (j int);
+show create table t1;
+Table	Create Table
+t1	CREATE TEMPORARY TABLE `t1` (
+  `j` int(11) DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+drop table t1;
+CREATE OR REPLACE TABLE t1 (a int);
+LOCK TABLES t1 write;
+CREATE OR REPLACE TEMPORARY TABLE t1 (a int);
+CREATE OR REPLACE TEMPORARY TABLE t1 (a int, b int);
+CREATE OR REPLACE TEMPORARY TABLE t1 (a int, b int) engine= innodb;
+CREATE OR REPLACE TEMPORARY TABLE t1 (a int) engine= innodb;
+CREATE OR REPLACE TEMPORARY TABLE t1 (a int, b int) engine=myisam;
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TEMPORARY TABLE `t1` (
+  `a` int(11) DEFAULT NULL,
+  `b` int(11) DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+DROP TEMPORARY TABLE t1;
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `a` int(11) DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+CREATE OR REPLACE TABLE t2 (a int);
+ERROR HY000: Table 't2' was not locked with LOCK TABLES
+DROP TABLE t1;
+UNLOCK TABLES;
+CREATE OR REPLACE TEMPORARY TABLE t1 (a int) SELECT * from t2;
+SELECT * FROM t1;
+a
+1
+2
+3
+CREATE OR REPLACE TEMPORARY TABLE t1 (b int) SELECT * from t2;
+SELECT * FROM t1;
+b	a
+NULL	1
+NULL	2
+NULL	3
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TEMPORARY TABLE `t1` (
+  `b` int(11) DEFAULT NULL,
+  `a` int(11) DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+DROP TABLE t1;
+CREATE TEMPORARY TABLE t1 AS SELECT a FROM t2;
+CREATE TEMPORARY TABLE IF NOT EXISTS t1(a int, b int) SELECT 1,2 FROM t2;
+Warnings:
+Note	1050	Table 't1' already exists
+create or replace table t1 as select 1;
+show create table t1;
+Table	Create Table
+t1	CREATE TEMPORARY TABLE `t1` (
+  `a` int(11) DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+DROP TABLE t1;
+#
+# Testing with normal tables
+#
+CREATE OR REPLACE TABLE t1 (a int);
+CREATE OR REPLACE TABLE t1 (a int, b int);
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `a` int(11) DEFAULT NULL,
+  `b` int(11) DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+DROP TABLE t1;
+CREATE TABLE t1 (a int) SELECT * from t2;
+SELECT * FROM t1;
+a
+1
+2
+3
+TRUNCATE TABLE t1;
+CREATE TABLE IF NOT EXISTS t1 (a int) SELECT * from t2;
+Warnings:
+Note	1050	Table 't1' already exists
+SELECT * FROM t1;
+a
+DROP TABLE t1;
+CREATE TABLE t1 (i int);
+CREATE OR REPLACE TABLE t1 AS SELECT 1;
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `1` int(1) NOT NULL DEFAULT '0'
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+DROP TABLE t1;
+CREATE OR REPLACE TABLE t1 (a int);
+LOCK TABLES t1 write,t2 write;
+CREATE OR REPLACE TABLE t1 (a int, b int);
+SELECT * FROM t1;
+a	b
+INSERT INTO t1 values(1,1);
+CREATE OR REPLACE TABLE t1 (a int, b int, c int);
+INSERT INTO t1 values(1,1,1);
+CREATE OR REPLACE TABLE t3 (a int);
+ERROR HY000: Table 't3' was not locked with LOCK TABLES
+UNLOCK TABLES;
+DROP TABLE t1;
+CREATE OR REPLACE TABLE t1 (a int);
+LOCK TABLES t1 write,t2 write;
+CREATE OR REPLACE TABLE t1 (a int, b int) select a,1 from t2;
+SELECT * FROM t2;
+a
+1
+2
+3
+SELECT * FROM t1;
+b	a	1
+NULL	1	1
+NULL	2	1
+NULL	3	1
+SELECT * FROM t1;
+b	a	1
+NULL	1	1
+NULL	2	1
+NULL	3	1
+INSERT INTO t1 values(1,1,1);
+CREATE OR REPLACE TABLE t1 (a int, b int, c int, d int);
+INSERT INTO t1 values(1,1,1,1);
+CREATE OR REPLACE TABLE t3 (a int);
+ERROR HY000: Table 't3' was not locked with LOCK TABLES
+UNLOCK TABLES;
+DROP TABLE t1;
+CREATE OR REPLACE TABLE t1 (a int);
+LOCK TABLES t1 write,t2 write, t1 as t1_read read;
+CREATE OR REPLACE TABLE t1 (a int, b int) select a,1 from t2;
+SELECT * FROM t1;
+b	a	1
+NULL	1	1
+NULL	2	1
+NULL	3	1
+SELECT * FROM t2;
+a
+1
+2
+3
+SELECT * FROM t1 as t1_read;
+ERROR HY000: Table 't1_read' was not locked with LOCK TABLES
+DROP TABLE t1;
+UNLOCK TABLES;
+#
+# Test also with InnoDB (transactional engine)
+#
+create table t1 (i int) engine=innodb;
+lock table t1 write;
+create or replace table t1 (j int);
+unlock tables;
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `j` int(11) DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+drop table t1;
+create table t1 (i int) engine=InnoDB;
+lock table t1 write, t2 write;
+create or replace table t1 (j int) engine=innodb;
+unlock tables;
+drop table t1;
+create table t1 (i int) engine=InnoDB;
+create table t3 (i int) engine=InnoDB;
+insert into t3 values(1),(2),(3);
+lock table t1 write, t2 write, t3 write;
+create or replace table t1 (a int, i int) engine=innodb select t2.a,t3.i from t2,t3;
+unlock tables;
+select * from t1 order by a,i;
+a	i
+1	1
+1	2
+1	3
+2	1
+2	2
+2	3
+3	1
+3	2
+3	3
+drop table t1,t3;
+#
+# Testing CREATE .. LIKE
+#
+create or replace table t1 like t2;
+create or replace table t1 like t2;
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `a` int(11) DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+drop table t1;
+create table t1 (b int);
+lock tables t1 write, t2 read;
+create or replace table t1 like t2;
+SELECT * FROM t1;
+a
+INSERT INTO t1 values(1);
+CREATE OR REPLACE TABLE t1 like t2;
+INSERT INTO t1 values(2);
+unlock tables;
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `a` int(11) DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+drop table t1;
+#
+# Test with prepared statements
+#
+prepare stmt1 from 'create or replace table t1 select * from t2';
+execute stmt1;
+select * from t1;
+a
+1
+2
+3
+execute stmt1;
+select * from t1;
+a
+1
+2
+3
+drop table t1;
+execute stmt1;
+select * from t1;
+a
+1
+2
+3
+deallocate prepare stmt1;
+drop table t1;
+#
+# Test with views
+#
+create view t1 as select 1;
+create table if not exists t1 (a int);
+Warnings:
+Note	1050	Table 't1' already exists
+create or replace table t1 (a int);
+ERROR 42S02: Unknown table 'test.t1'
+drop table t1;
+ERROR 42S02: Unknown table 'test.t1'
+drop view t1;
+DROP TABLE t2;
diff --git a/mysql-test/r/partition_exchange.result b/mysql-test/r/partition_exchange.result
index 36499004869..fec08e99c72 100644
--- a/mysql-test/r/partition_exchange.result
+++ b/mysql-test/r/partition_exchange.result
@@ -1088,7 +1088,7 @@ ALTER TABLE t PARTITION BY RANGE (UNIX_TIMESTAMP(event_time) DIV 1)
 (PARTITION p0 VALUES LESS THAN (123456789),
 PARTITION pMAX VALUES LESS THAN MAXVALUE);
 ALTER TABLE t EXCHANGE PARTITION p0 WITH TABLE general_log;
-ERROR HY000: Incorrect usage of PARTITION and log table
+ERROR HY000: You cannot 'ALTER PARTITION' a log table if logging is enabled
 ALTER TABLE general_log ENGINE = CSV;
 SET @@global.general_log = @old_general_log_state;
 DROP TABLE t;
diff --git a/mysql-test/r/ps_ddl.result b/mysql-test/r/ps_ddl.result
index 8284e974574..dec0d12c455 100644
--- a/mysql-test/r/ps_ddl.result
+++ b/mysql-test/r/ps_ddl.result
@@ -1930,7 +1930,7 @@ SUCCESS
 
 execute stmt;
 ERROR 42S01: Table 't2' already exists
-call p_verify_reprepare_count(1);
+call p_verify_reprepare_count(0);
 SUCCESS
 
 execute stmt;
@@ -1946,7 +1946,7 @@ SUCCESS
 
 execute stmt;
 ERROR 42S01: Table 't2' already exists
-call p_verify_reprepare_count(1);
+call p_verify_reprepare_count(0);
 SUCCESS
 
 drop temporary table t2;
@@ -1964,7 +1964,7 @@ drop table t2;
 create view t2 as select 1;
 execute stmt;
 Got one of the listed errors
-call p_verify_reprepare_count(1);
+call p_verify_reprepare_count(0);
 SUCCESS
 
 execute stmt;
diff --git a/mysql-test/suite/archive/discover.result b/mysql-test/suite/archive/discover.result
index c4f4bb4104f..726c8712917 100644
--- a/mysql-test/suite/archive/discover.result
+++ b/mysql-test/suite/archive/discover.result
@@ -135,4 +135,7 @@ select * from t1;
 a
 flush tables;
 create table t1 (a int) engine=archive;
+ERROR 42S01: Table 't1' already exists
+flush tables;
+create table t1 (a int) engine=archive;
 drop table t1;
diff --git a/mysql-test/suite/archive/discover.test b/mysql-test/suite/archive/discover.test
index 8dfe09f7b33..144a5dbdcf9 100644
--- a/mysql-test/suite/archive/discover.test
+++ b/mysql-test/suite/archive/discover.test
@@ -125,6 +125,10 @@ create table t1 (a int) engine=archive;
 select * from t1;
 flush tables;
 remove_file $mysqld_datadir/test/t1.ARZ;
+--error ER_TABLE_EXISTS_ERROR
+create table t1 (a int) engine=archive;
+remove_file $mysqld_datadir/test/t1.frm;
+flush tables;
 create table t1 (a int) engine=archive;
 drop table t1;
 
diff --git a/mysql-test/suite/funcs_1/r/innodb_views.result b/mysql-test/suite/funcs_1/r/innodb_views.result
index 96b6d3171f0..e6d98159b39 100644
--- a/mysql-test/suite/funcs_1/r/innodb_views.result
+++ b/mysql-test/suite/funcs_1/r/innodb_views.result
@@ -7579,7 +7579,7 @@ ERROR 42000: You have an error in your SQL syntax; check the manual that corresp
 SELECT * FROM test.tb2 limit 2' at line 1
 CREATE OR REPLACE TEMPORARY VIEW test.v1 AS
 SELECT * FROM test.tb2 limit 2 ;
-ERROR 42000: You have an error in your SQL syntax; check the manual that corresponds to your MariaDB server version for the right syntax to use near 'TEMPORARY VIEW test.v1 AS
+ERROR 42000: You have an error in your SQL syntax; check the manual that corresponds to your MariaDB server version for the right syntax to use near 'VIEW test.v1 AS
 SELECT * FROM test.tb2 limit 2' at line 1
 Drop view if exists test.v1 ;
 Use test;
diff --git a/mysql-test/suite/funcs_1/r/memory_views.result b/mysql-test/suite/funcs_1/r/memory_views.result
index ddde31b76d1..21990c2bd9d 100644
--- a/mysql-test/suite/funcs_1/r/memory_views.result
+++ b/mysql-test/suite/funcs_1/r/memory_views.result
@@ -7580,7 +7580,7 @@ ERROR 42000: You have an error in your SQL syntax; check the manual that corresp
 SELECT * FROM test.tb2 limit 2' at line 1
 CREATE OR REPLACE TEMPORARY VIEW test.v1 AS
 SELECT * FROM test.tb2 limit 2 ;
-ERROR 42000: You have an error in your SQL syntax; check the manual that corresponds to your MariaDB server version for the right syntax to use near 'TEMPORARY VIEW test.v1 AS
+ERROR 42000: You have an error in your SQL syntax; check the manual that corresponds to your MariaDB server version for the right syntax to use near 'VIEW test.v1 AS
 SELECT * FROM test.tb2 limit 2' at line 1
 Drop view if exists test.v1 ;
 Use test;
diff --git a/mysql-test/suite/rpl/disabled.def b/mysql-test/suite/rpl/disabled.def
index 5cc3916b614..de3091a56e5 100644
--- a/mysql-test/suite/rpl/disabled.def
+++ b/mysql-test/suite/rpl/disabled.def
@@ -10,7 +10,6 @@
 #
 ##############################################################################
 
-rpl_row_create_table      : Bug#11759274 2010-02-27 andrei failed different way than earlier with bug#45576
 rpl_spec_variables        : BUG#11755836 2009-10-27 jasonh rpl_spec_variables fails on PB2 hpux
 rpl_get_master_version_and_clock : Bug#11766137 Jan 05 2011 joro Valgrind warnings rpl_get_master_version_and_clock
 rpl_partition_archive     : MDEV-5077 2013-09-27 svoj Cannot exchange partition with archive table
diff --git a/mysql-test/suite/rpl/r/rpl_row_create_table.result b/mysql-test/suite/rpl/r/rpl_row_create_table.result
index 393e2fdb851..07822a39b46 100644
--- a/mysql-test/suite/rpl/r/rpl_row_create_table.result
+++ b/mysql-test/suite/rpl/r/rpl_row_create_table.result
@@ -1,23 +1,24 @@
-stop slave;
-drop table if exists t1,t2,t3,t4,t5,t6,t7,t8,t9;
-reset master;
-reset slave;
-drop table if exists t1,t2,t3,t4,t5,t6,t7,t8,t9;
-start slave;
-**** Resetting master and slave ****
-include/stop_slave.inc
-RESET SLAVE;
-RESET MASTER;
-include/start_slave.inc
-CREATE TABLE t1 (a INT, b INT);
+include/master-slave.inc
+[connection master]
+include/wait_for_slave_to_stop.inc
+include/wait_for_slave_to_start.inc
+include/rpl_reset.inc
+CREATE TABLE t1 (a INT);
+CREATE OR REPLACE TABLE t1 (a INT, b INT);
 CREATE TABLE t2 (a INT, b INT) ENGINE=Merge;
 CREATE TABLE t3 (a INT, b INT) CHARSET=utf8;
 CREATE TABLE t4 (a INT, b INT) ENGINE=Merge CHARSET=utf8;
-show binlog events from <binlog_start>;
+include/show_binlog_events.inc
 Log_name	Pos	Event_type	Server_id	End_log_pos	Info
-master-bin.000001	#	Query	#	#	use `test`; CREATE TABLE t1 (a INT, b INT)
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; CREATE TABLE t1 (a INT)
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE TABLE t1 (a INT, b INT)
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
 master-bin.000001	#	Query	#	#	use `test`; CREATE TABLE t2 (a INT, b INT) ENGINE=Merge
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
 master-bin.000001	#	Query	#	#	use `test`; CREATE TABLE t3 (a INT, b INT) CHARSET=utf8
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
 master-bin.000001	#	Query	#	#	use `test`; CREATE TABLE t4 (a INT, b INT) ENGINE=Merge CHARSET=utf8
 **** On Master ****
 SHOW CREATE TABLE t1;
@@ -111,15 +112,10 @@ NULL	3	6
 NULL	4	2
 NULL	5	10
 NULL	6	12
-**** Resetting master and slave ****
-include/stop_slave.inc
-RESET SLAVE;
-RESET MASTER;
-include/start_slave.inc
+include/rpl_reset.inc
 CREATE TABLE t7 (UNIQUE(b)) SELECT a,b FROM tt3;
 ERROR 23000: Duplicate entry '2' for key 'b'
-show binlog events from <binlog_start>;
-Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+include/show_binlog_events.inc
 CREATE TABLE t7 (a INT, b INT UNIQUE);
 INSERT INTO t7 SELECT a,b FROM tt3;
 ERROR 23000: Duplicate entry '2' for key 'b'
@@ -128,23 +124,20 @@ a	b
 1	2
 2	4
 3	6
-show binlog events from <binlog_start>;
+include/show_binlog_events.inc
 Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
 master-bin.000001	#	Query	#	#	use `test`; CREATE TABLE t7 (a INT, b INT UNIQUE)
-master-bin.000001	#	Query	#	#	BEGIN
+master-bin.000001	#	Gtid	#	#	BEGIN GTID #-#-#
 master-bin.000001	#	Table_map	#	#	table_id: # (test.t7)
-master-bin.000001	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+master-bin.000001	#	Write_rows_v1	#	#	table_id: # flags: STMT_END_F
 master-bin.000001	#	Query	#	#	COMMIT
 SELECT * FROM t7 ORDER BY a,b;
 a	b
 1	2
 2	4
 3	6
-**** Resetting master and slave ****
-include/stop_slave.inc
-RESET SLAVE;
-RESET MASTER;
-include/start_slave.inc
+include/rpl_reset.inc
 CREATE TEMPORARY TABLE tt4 (a INT, b INT);
 INSERT INTO tt4 VALUES (4,8), (5,10), (6,12);
 BEGIN;
@@ -152,11 +145,11 @@ INSERT INTO t7 SELECT a,b FROM tt4;
 ROLLBACK;
 Warnings:
 Warning	1196	Some non-transactional changed tables couldn't be rolled back
-show binlog events from <binlog_start>;
+include/show_binlog_events.inc
 Log_name	Pos	Event_type	Server_id	End_log_pos	Info
-master-bin.000001	#	Query	#	#	BEGIN
+master-bin.000001	#	Gtid	#	#	BEGIN GTID #-#-#
 master-bin.000001	#	Table_map	#	#	table_id: # (test.t7)
-master-bin.000001	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+master-bin.000001	#	Write_rows_v1	#	#	table_id: # flags: STMT_END_F
 master-bin.000001	#	Query	#	#	COMMIT
 SELECT * FROM t7 ORDER BY a,b;
 a	b
@@ -174,11 +167,7 @@ a	b
 4	8
 5	10
 6	12
-**** Resetting master and slave ****
-include/stop_slave.inc
-RESET SLAVE;
-RESET MASTER;
-include/start_slave.inc
+include/rpl_reset.inc
 CREATE TABLE t8 LIKE t4;
 CREATE TABLE t9 LIKE tt4;
 CREATE TEMPORARY TABLE tt5 LIKE t4;
@@ -197,9 +186,11 @@ Create Table	CREATE TABLE `t9` (
   `a` int(11) DEFAULT NULL,
   `b` int(11) DEFAULT NULL
 ) ENGINE=MyISAM DEFAULT CHARSET=latin1
-show binlog events from <binlog_start>;
+include/show_binlog_events.inc
 Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
 master-bin.000001	#	Query	#	#	use `test`; CREATE TABLE t8 LIKE t4
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
 master-bin.000001	#	Query	#	#	use `test`; CREATE TABLE `t9` (
   `a` int(11) DEFAULT NULL,
   `b` int(11) DEFAULT NULL
@@ -219,15 +210,12 @@ Create Table	CREATE TABLE `t9` (
 ) ENGINE=MEMORY DEFAULT CHARSET=latin1
 DROP TABLE IF EXISTS t1,t2,t3,t4,t5,t6,t7,t8,t9;
 STOP SLAVE;
+include/wait_for_slave_to_stop.inc
 SET GLOBAL storage_engine=@storage_engine;
 START SLAVE;
+include/wait_for_slave_to_start.inc
 ================ BUG#22864 ================
-stop slave;
-drop table if exists t1,t2,t3,t4,t5,t6,t7,t8,t9;
-reset master;
-reset slave;
-drop table if exists t1,t2,t3,t4,t5,t6,t7,t8,t9;
-start slave;
+include/rpl_reset.inc
 SET AUTOCOMMIT=0;
 CREATE TABLE t1 (a INT);
 INSERT INTO t1 VALUES (1),(2),(3);
@@ -270,37 +258,38 @@ a
 1
 2
 3
-show binlog events from <binlog_start>;
+include/show_binlog_events.inc
 Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
 master-bin.000001	#	Query	#	#	use `test`; CREATE TABLE t1 (a INT)
-master-bin.000001	#	Query	#	#	BEGIN
+master-bin.000001	#	Gtid	#	#	BEGIN GTID #-#-#
 master-bin.000001	#	Table_map	#	#	table_id: # (test.t1)
-master-bin.000001	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+master-bin.000001	#	Write_rows_v1	#	#	table_id: # flags: STMT_END_F
 master-bin.000001	#	Query	#	#	COMMIT
-master-bin.000001	#	Query	#	#	BEGIN
+master-bin.000001	#	Gtid	#	#	BEGIN GTID #-#-#
 master-bin.000001	#	Query	#	#	use `test`; CREATE TABLE `t2` (
   `a` int(11) DEFAULT NULL
 ) ENGINE=InnoDB
 master-bin.000001	#	Table_map	#	#	table_id: # (test.t2)
-master-bin.000001	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+master-bin.000001	#	Write_rows_v1	#	#	table_id: # flags: STMT_END_F
 master-bin.000001	#	Xid	#	#	COMMIT /* XID */
-master-bin.000001	#	Query	#	#	BEGIN
+master-bin.000001	#	Gtid	#	#	BEGIN GTID #-#-#
 master-bin.000001	#	Query	#	#	use `test`; CREATE TABLE `t3` (
   `a` int(11) DEFAULT NULL
 ) ENGINE=InnoDB
 master-bin.000001	#	Table_map	#	#	table_id: # (test.t3)
-master-bin.000001	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+master-bin.000001	#	Write_rows_v1	#	#	table_id: # flags: STMT_END_F
 master-bin.000001	#	Xid	#	#	COMMIT /* XID */
-master-bin.000001	#	Query	#	#	BEGIN
+master-bin.000001	#	Gtid	#	#	BEGIN GTID #-#-#
 master-bin.000001	#	Query	#	#	use `test`; CREATE TABLE `t4` (
   `a` int(11) DEFAULT NULL
 ) ENGINE=InnoDB
 master-bin.000001	#	Table_map	#	#	table_id: # (test.t4)
-master-bin.000001	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+master-bin.000001	#	Write_rows_v1	#	#	table_id: # flags: STMT_END_F
 master-bin.000001	#	Xid	#	#	COMMIT /* XID */
-master-bin.000001	#	Query	#	#	BEGIN
+master-bin.000001	#	Gtid	#	#	BEGIN GTID #-#-#
 master-bin.000001	#	Table_map	#	#	table_id: # (test.t1)
-master-bin.000001	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+master-bin.000001	#	Write_rows_v1	#	#	table_id: # flags: STMT_END_F
 master-bin.000001	#	Query	#	#	COMMIT
 SHOW TABLES;
 Tables_in_test
@@ -333,10 +322,7 @@ a
 3
 DROP TABLE IF EXISTS t1,t2,t3,t4;
 SET AUTOCOMMIT=1;
-STOP SLAVE;
-RESET SLAVE;
-RESET MASTER;
-START SLAVE;
+include/rpl_reset.inc
 CREATE TABLE t1 (a INT);
 INSERT INTO t1 VALUES (1),(2),(3);
 CREATE TABLE t2 (a INT) ENGINE=INNODB;
@@ -355,19 +341,21 @@ a
 4
 6
 9
-show binlog events from <binlog_start>;
+include/show_binlog_events.inc
 Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
 master-bin.000001	#	Query	#	#	use `test`; CREATE TABLE t1 (a INT)
-master-bin.000001	#	Query	#	#	BEGIN
+master-bin.000001	#	Gtid	#	#	BEGIN GTID #-#-#
 master-bin.000001	#	Table_map	#	#	table_id: # (test.t1)
-master-bin.000001	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+master-bin.000001	#	Write_rows_v1	#	#	table_id: # flags: STMT_END_F
 master-bin.000001	#	Query	#	#	COMMIT
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
 master-bin.000001	#	Query	#	#	use `test`; CREATE TABLE t2 (a INT) ENGINE=INNODB
-master-bin.000001	#	Query	#	#	BEGIN
+master-bin.000001	#	Gtid	#	#	BEGIN GTID #-#-#
 master-bin.000001	#	Table_map	#	#	table_id: # (test.t2)
-master-bin.000001	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+master-bin.000001	#	Write_rows_v1	#	#	table_id: # flags: STMT_END_F
 master-bin.000001	#	Table_map	#	#	table_id: # (test.t2)
-master-bin.000001	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+master-bin.000001	#	Write_rows_v1	#	#	table_id: # flags: STMT_END_F
 master-bin.000001	#	Xid	#	#	COMMIT /* XID */
 SELECT * FROM t2 ORDER BY a;
 a
@@ -377,11 +365,7 @@ a
 6
 9
 TRUNCATE TABLE t2;
-**** Resetting master and slave ****
-include/stop_slave.inc
-RESET SLAVE;
-RESET MASTER;
-include/start_slave.inc
+include/rpl_reset.inc
 BEGIN;
 INSERT INTO t2 SELECT a*a FROM t1;
 CREATE TEMPORARY TABLE tt2
@@ -394,8 +378,14 @@ Warnings:
 Warning	1196	Some non-transactional changed tables couldn't be rolled back
 SELECT * FROM t2 ORDER BY a;
 a
-show binlog events from <binlog_start>;
+include/show_binlog_events.inc
 Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+master-bin.000001	#	Gtid	#	#	BEGIN GTID #-#-#
+master-bin.000001	#	Table_map	#	#	table_id: # (test.t2)
+master-bin.000001	#	Write_rows_v1	#	#	table_id: # flags: STMT_END_F
+master-bin.000001	#	Table_map	#	#	table_id: # (test.t2)
+master-bin.000001	#	Write_rows_v1	#	#	table_id: # flags: STMT_END_F
+master-bin.000001	#	Query	#	#	ROLLBACK
 SELECT * FROM t2 ORDER BY a;
 a
 DROP TABLE t1,t2;
@@ -412,35 +402,28 @@ a
 1
 2
 DROP TABLE t1;
-stop slave;
-drop table if exists t1,t2,t3,t4,t5,t6,t7,t8,t9;
-reset master;
-reset slave;
-drop table if exists t1,t2,t3,t4,t5,t6,t7,t8,t9;
-start slave;
+include/rpl_reset.inc
 DROP DATABASE IF EXISTS mysqltest1;
 CREATE DATABASE mysqltest1;
 CREATE TABLE mysqltest1.without_select (f1 BIGINT);
 CREATE TABLE mysqltest1.with_select AS SELECT 1 AS f1;
-show binlog events from <binlog_start>;
+include/show_binlog_events.inc
 Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
 master-bin.000001	#	Query	#	#	DROP DATABASE IF EXISTS mysqltest1
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
 master-bin.000001	#	Query	#	#	CREATE DATABASE mysqltest1
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
 master-bin.000001	#	Query	#	#	use `test`; CREATE TABLE mysqltest1.without_select (f1 BIGINT)
-master-bin.000001	#	Query	#	#	BEGIN
+master-bin.000001	#	Gtid	#	#	BEGIN GTID #-#-#
 master-bin.000001	#	Query	#	#	use `test`; CREATE TABLE `mysqltest1`.`with_select` (
   `f1` int(1) NOT NULL DEFAULT '0'
 )
 master-bin.000001	#	Table_map	#	#	table_id: # (mysqltest1.with_select)
-master-bin.000001	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+master-bin.000001	#	Write_rows_v1	#	#	table_id: # flags: STMT_END_F
 master-bin.000001	#	Query	#	#	COMMIT
 DROP DATABASE mysqltest1;
-stop slave;
-drop table if exists t1,t2,t3,t4,t5,t6,t7,t8,t9;
-reset master;
-reset slave;
-drop table if exists t1,t2,t3,t4,t5,t6,t7,t8,t9;
-start slave;
+include/rpl_reset.inc
 CREATE TEMPORARY TABLE t7(c1 INT);
 CREATE TABLE t5(c1 INT);
 CREATE TABLE t4(c1 INT);
@@ -461,4 +444,5 @@ DROP VIEW IF EXISTS bug48506_t1, bug48506_t2, bug48506_t3;
 DROP TEMPORARY TABLES t7;
 DROP TABLES t4, t5;
 DROP TABLES IF EXISTS bug48506_t4;
+include/rpl_end.inc
 end of the tests
diff --git a/mysql-test/suite/rpl/t/rpl_create_if_not_exists.test b/mysql-test/suite/rpl/t/rpl_create_if_not_exists.test
index 72f1201c93c..b27250f908f 100644
--- a/mysql-test/suite/rpl/t/rpl_create_if_not_exists.test
+++ b/mysql-test/suite/rpl/t/rpl_create_if_not_exists.test
@@ -52,6 +52,8 @@ CREATE DATABASE IF NOT EXISTS mysqltest;
 USE mysqltest;
 CREATE TABLE IF NOT EXISTS t(c1 int);
 CREATE TABLE IF NOT EXISTS t1 LIKE t;
+# The following will not be logged because t2 existed and we will not
+# put the data of SELECT into the binary log
 CREATE TABLE IF NOT EXISTS t2 SELECT * FROM t;
 CREATE EVENT IF NOT EXISTS e 
 ON SCHEDULE AT CURRENT_TIMESTAMP + INTERVAL 1 HOUR 
@@ -104,7 +106,7 @@ SELECT * FROM t1;
 SELECT * FROM t2; 
 sync_slave_with_master; 
 
-# In these two statements, t1 and t2 are the base table. The recoreds of t2
+# In these two statements, t1 and t2 are the base table. The records of t2
 # are inserted into it when CREATE TABLE ...  SELECT was executed.  
 SELECT * FROM t1;
 SELECT * FROM t2;
diff --git a/mysql-test/suite/rpl/t/rpl_row_create_table.test b/mysql-test/suite/rpl/t/rpl_row_create_table.test
index ef3c0758643..da73d753dcd 100644
--- a/mysql-test/suite/rpl/t/rpl_row_create_table.test
+++ b/mysql-test/suite/rpl/t/rpl_row_create_table.test
@@ -28,7 +28,8 @@ START SLAVE;
 --source include/rpl_reset.inc
 
 connection master;
-CREATE TABLE t1 (a INT, b INT);
+CREATE TABLE t1 (a INT);
+CREATE OR REPLACE TABLE t1 (a INT, b INT);
 CREATE TABLE t2 (a INT, b INT) ENGINE=Merge;
 CREATE TABLE t3 (a INT, b INT) CHARSET=utf8;
 CREATE TABLE t4 (a INT, b INT) ENGINE=Merge CHARSET=utf8;
diff --git a/mysql-test/t/create.test b/mysql-test/t/create.test
index ebcad5f4af4..8bb7339ce83 100644
--- a/mysql-test/t/create.test
+++ b/mysql-test/t/create.test
@@ -2014,6 +2014,8 @@ create table t1 (a int, b int);
 create table t1 (a int, b int) select 2,2;
 --error ER_TABLE_EXISTS_ERROR
 create table t1 like t2;
+--error ER_LOCK_WAIT_TIMEOUT
+create or replace table t1 (a int, b int) select 2,2;
 disconnect user1;
 connection default;
 select * from t1;
diff --git a/mysql-test/t/create_or_replace-master.opt b/mysql-test/t/create_or_replace-master.opt
new file mode 100644
index 00000000000..e94228f2f33
--- /dev/null
+++ b/mysql-test/t/create_or_replace-master.opt
@@ -0,0 +1 @@
+--log-output=TABLE,FILE --general-log=1 --slow-query-log=1
diff --git a/mysql-test/t/create_or_replace.test b/mysql-test/t/create_or_replace.test
new file mode 100644
index 00000000000..b776be23b08
--- /dev/null
+++ b/mysql-test/t/create_or_replace.test
@@ -0,0 +1,234 @@
+#
+# Check CREATE OR REPLACE ALTER TABLE
+#
+
+--source include/have_innodb.inc
+--disable_warnings
+drop table if exists t1,t2;
+--enable_warnings
+
+#
+# Create help table
+#
+
+CREATE TABLE t2 (a int);
+INSERT INTO t2 VALUES(1),(2),(3);
+
+--echo #
+--echo # Check first syntax and wrong usage
+--echo #
+
+--error ER_WRONG_USAGE
+CREATE OR REPLACE TABLE IF NOT EXISTS t1 (a int);
+--error ER_WRONG_USAGE
+create or replace trigger trg before insert on t1 for each row set @a:=1;
+
+# check that we don't try to create a log table in use
+--error ER_BAD_LOG_STATEMENT
+create or replace table mysql.general_log (a int);
+--error ER_BAD_LOG_STATEMENT
+create or replace table mysql.slow_log (a int);
+
+--echo #
+--echo # Usage when table doesn't exist
+--echo #
+
+CREATE OR REPLACE TABLE t1 (a int);
+--error ER_TABLE_EXISTS_ERROR
+CREATE TABLE t1 (a int);
+DROP TABLE t1;
+CREATE OR REPLACE TEMPORARY TABLE t1 (a int);
+--error ER_TABLE_EXISTS_ERROR
+CREATE TEMPORARY TABLE t1 (a int, b int, c int);
+DROP TEMPORARY TABLE t1;
+
+--echo #
+--echo # Testing with temporary tables
+--echo #
+
+CREATE OR REPLACE TABLE t1 (a int);
+CREATE OR REPLACE TEMPORARY TABLE t1 (a int);
+CREATE OR REPLACE TEMPORARY TABLE t1 (a int, b int);
+SHOW CREATE TABLE t1;
+DROP TEMPORARY TABLE t1;
+SHOW CREATE TABLE t1;
+DROP TABLE t1;
+
+# Test also with InnoDB
+create temporary table t1 (i int) engine=InnoDB;
+create or replace temporary table t1 (a int, b int) engine=InnoDB;
+create or replace temporary table t1 (j int);
+show create table t1;
+drop table t1;
+
+# Using lock tables on normal tables with create or replace on temp tables
+CREATE OR REPLACE TABLE t1 (a int);
+LOCK TABLES t1 write;
+CREATE OR REPLACE TEMPORARY TABLE t1 (a int);
+CREATE OR REPLACE TEMPORARY TABLE t1 (a int, b int);
+CREATE OR REPLACE TEMPORARY TABLE t1 (a int, b int) engine= innodb;
+CREATE OR REPLACE TEMPORARY TABLE t1 (a int) engine= innodb;
+CREATE OR REPLACE TEMPORARY TABLE t1 (a int, b int) engine=myisam;
+SHOW CREATE TABLE t1;
+DROP TEMPORARY TABLE t1;
+SHOW CREATE TABLE t1;
+# Verify that table is still locked
+--error ER_TABLE_NOT_LOCKED
+CREATE OR REPLACE TABLE t2 (a int);
+DROP TABLE t1;
+UNLOCK TABLES;
+
+#
+# Using CREATE SELECT
+#
+
+CREATE OR REPLACE TEMPORARY TABLE t1 (a int) SELECT * from t2;
+SELECT * FROM t1;
+CREATE OR REPLACE TEMPORARY TABLE t1 (b int) SELECT * from t2;
+SELECT * FROM t1;
+SHOW CREATE TABLE t1;
+DROP TABLE t1;
+CREATE TEMPORARY TABLE t1 AS SELECT a FROM t2;
+CREATE TEMPORARY TABLE IF NOT EXISTS t1(a int, b int) SELECT 1,2 FROM t2;
+create or replace table t1 as select 1;
+show create table t1;
+DROP TABLE t1;
+
+
+--echo #
+--echo # Testing with normal tables
+--echo #
+
+CREATE OR REPLACE TABLE t1 (a int);
+CREATE OR REPLACE TABLE t1 (a int, b int);
+SHOW CREATE TABLE t1;
+DROP TABLE t1;
+
+CREATE TABLE t1 (a int) SELECT * from t2;
+SELECT * FROM t1;
+TRUNCATE TABLE t1;
+CREATE TABLE IF NOT EXISTS t1 (a int) SELECT * from t2;
+SELECT * FROM t1;
+DROP TABLE t1;
+
+CREATE TABLE t1 (i int);
+CREATE OR REPLACE TABLE t1 AS SELECT 1;
+SHOW CREATE TABLE t1;
+DROP TABLE t1;
+
+# Using lock tables with CREATE OR REPLACE
+CREATE OR REPLACE TABLE t1 (a int);
+LOCK TABLES t1 write,t2 write;
+CREATE OR REPLACE TABLE t1 (a int, b int);
+# Verify if table is still locked
+SELECT * FROM t1;
+INSERT INTO t1 values(1,1);
+CREATE OR REPLACE TABLE t1 (a int, b int, c int);
+INSERT INTO t1 values(1,1,1);
+--error ER_TABLE_NOT_LOCKED
+CREATE OR REPLACE TABLE t3 (a int);
+UNLOCK TABLES;
+DROP TABLE t1;
+
+# Using lock tables with CREATE OR REPLACE ... SELECT
+CREATE OR REPLACE TABLE t1 (a int);
+LOCK TABLES t1 write,t2 write;
+CREATE OR REPLACE TABLE t1 (a int, b int) select a,1 from t2;
+# Verify if table is still locked
+SELECT * FROM t2;
+SELECT * FROM t1;
+SELECT * FROM t1;
+INSERT INTO t1 values(1,1,1);
+CREATE OR REPLACE TABLE t1 (a int, b int, c int, d int);
+INSERT INTO t1 values(1,1,1,1);
+--error ER_TABLE_NOT_LOCKED
+CREATE OR REPLACE TABLE t3 (a int);
+UNLOCK TABLES;
+DROP TABLE t1;
+
+CREATE OR REPLACE TABLE t1 (a int);
+LOCK TABLES t1 write,t2 write, t1 as t1_read read;
+CREATE OR REPLACE TABLE t1 (a int, b int) select a,1 from t2;
+SELECT * FROM t1;
+SELECT * FROM t2;
+--error ER_TABLE_NOT_LOCKED
+SELECT * FROM t1 as t1_read;
+DROP TABLE t1;
+UNLOCK TABLES;
+
+--echo #
+--echo # Test also with InnoDB (transactional engine)
+--echo #
+
+create table t1 (i int) engine=innodb;
+lock table t1 write;
+create or replace table t1 (j int);
+unlock tables;
+show create table t1;
+drop table t1;
+
+create table t1 (i int) engine=InnoDB;
+lock table t1 write, t2 write;
+create or replace table t1 (j int) engine=innodb;
+unlock tables;
+drop table t1;
+
+create table t1 (i int) engine=InnoDB;
+create table t3 (i int) engine=InnoDB;
+insert into t3 values(1),(2),(3);
+lock table t1 write, t2 write, t3 write;
+create or replace table t1 (a int, i int) engine=innodb select t2.a,t3.i from t2,t3;
+unlock tables;
+select * from t1 order by a,i;
+drop table t1,t3;
+
+--echo #
+--echo # Testing CREATE .. LIKE
+--echo #
+
+create or replace table t1 like t2;
+create or replace table t1 like t2;
+show create table t1;
+drop table t1;
+create table t1 (b int);
+lock tables t1 write, t2 read;
+create or replace table t1 like t2;
+SELECT * FROM t1;
+INSERT INTO t1 values(1);
+CREATE OR REPLACE TABLE t1 like t2;
+INSERT INTO t1 values(2);
+unlock tables;
+show create table t1;
+drop table t1;
+
+--echo #
+--echo # Test with prepared statements
+--echo #
+
+prepare stmt1 from 'create or replace table t1 select * from t2';
+execute stmt1;
+select * from t1;
+execute stmt1;
+select * from t1;
+drop table t1;
+execute stmt1;
+select * from t1;
+deallocate prepare stmt1;
+drop table t1;
+
+--echo #
+--echo # Test with views
+--echo #
+
+create view t1 as select 1;
+create table if not exists t1 (a int);
+--error ER_BAD_TABLE_ERROR
+create or replace table t1 (a int);
+--error ER_BAD_TABLE_ERROR
+drop table t1;
+drop view t1;
+
+#
+# Cleanup
+#
+DROP TABLE t2;
diff --git a/mysql-test/t/partition_exchange.test b/mysql-test/t/partition_exchange.test
index d7dfd6f543e..e538bee16cd 100644
--- a/mysql-test/t/partition_exchange.test
+++ b/mysql-test/t/partition_exchange.test
@@ -439,7 +439,7 @@ CREATE TABLE t LIKE general_log;
 ALTER TABLE t PARTITION BY RANGE (UNIX_TIMESTAMP(event_time) DIV 1)
 (PARTITION p0 VALUES LESS THAN (123456789),
  PARTITION pMAX VALUES LESS THAN MAXVALUE);
---error ER_WRONG_USAGE
+--error ER_BAD_LOG_STATEMENT
 ALTER TABLE t EXCHANGE PARTITION p0 WITH TABLE general_log;
 ALTER TABLE general_log ENGINE = CSV;
 SET @@global.general_log = @old_general_log_state;
diff --git a/mysql-test/t/ps_ddl.test b/mysql-test/t/ps_ddl.test
index c34800976c7..21355ca42b7 100644
--- a/mysql-test/t/ps_ddl.test
+++ b/mysql-test/t/ps_ddl.test
@@ -1610,7 +1610,7 @@ call p_verify_reprepare_count(0);
 # Base table with name of table to be created exists
 --error ER_TABLE_EXISTS_ERROR
 execute stmt;
-call p_verify_reprepare_count(1);
+call p_verify_reprepare_count(0);
 --error ER_TABLE_EXISTS_ERROR
 execute stmt;
 call p_verify_reprepare_count(0);
@@ -1622,7 +1622,7 @@ execute stmt;
 call p_verify_reprepare_count(0);
 --error ER_TABLE_EXISTS_ERROR
 execute stmt;
-call p_verify_reprepare_count(1);
+call p_verify_reprepare_count(0);
 drop temporary table t2;
 --error ER_TABLE_EXISTS_ERROR
 execute stmt;
@@ -1641,7 +1641,7 @@ drop table t2;
 create view t2 as select 1;
 --error ER_TABLE_EXISTS_ERROR,9999
 execute stmt;
-call p_verify_reprepare_count(1);
+call p_verify_reprepare_count(0);
 --error ER_TABLE_EXISTS_ERROR,9999
 execute stmt;
 call p_verify_reprepare_count(0);
diff --git a/sql/handler.h b/sql/handler.h
index 69b9c3e071e..1e22bf05aba 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -32,6 +32,7 @@
 #include "sql_cache.h"
 #include "structs.h"                            /* SHOW_COMP_OPTION */
 #include "sql_array.h"          /* Dynamic_array<> */
+#include "mdl.h"
 
 #include <my_compare.h>
 #include <ft_global.h>
@@ -378,6 +379,7 @@ enum enum_alter_inplace_result {
 #define HA_LEX_CREATE_IF_NOT_EXISTS 2
 #define HA_LEX_CREATE_TABLE_LIKE 4
 #define HA_CREATE_TMP_ALTER     8
+#define HA_LEX_CREATE_REPLACE   16
 #define HA_MAX_REC_LENGTH	65535
 
 /* Table caching type */
@@ -1573,9 +1575,15 @@ struct HA_CREATE_INFO
   ulong avg_row_length;
   ulong used_fields;
   ulong key_block_size;
-  uint stats_sample_pages;		/* number of pages to sample during
-					stats estimation, if used, otherwise 0. */
-  enum_stats_auto_recalc stats_auto_recalc;
+  /*
+    number of pages to sample during
+    stats estimation, if used, otherwise 0.
+  */
+  uint stats_sample_pages;
+  uint null_bits;                       /* NULL bits at start of record */
+  uint options;				/* OR of HA_CREATE_ options */
+  uint merge_insert_method;
+  uint extra_size;                      /* length of extra data segment */
   SQL_I_List<TABLE_LIST> merge_list;
   handlerton *db_type;
   /**
@@ -1588,21 +1596,23 @@ struct HA_CREATE_INFO
     If nothing speficied inherits the value of the original table (if present).
   */
   enum row_type row_type;
-  uint null_bits;                       /* NULL bits at start of record */
-  uint options;				/* OR of HA_CREATE_ options */
-  uint merge_insert_method;
-  uint extra_size;                      /* length of extra data segment */
   enum ha_choice transactional;
-  bool varchar;                         ///< 1 if table has a VARCHAR
   enum ha_storage_media storage_media;  ///< DEFAULT, DISK or MEMORY
   enum ha_choice page_checksum;         ///< If we have page_checksums
   engine_option_value *option_list;     ///< list of table create options
+  enum_stats_auto_recalc stats_auto_recalc;
+  bool varchar;                         ///< 1 if table has a VARCHAR
 
   /* the following three are only for ALTER TABLE, check_if_incompatible_data() */
   ha_table_option_struct *option_struct;           ///< structure with parsed table options
   ha_field_option_struct **fields_option_struct;   ///< array of field option structures
   ha_index_option_struct **indexes_option_struct;  ///< array of index option structures
 
+  /* The following is used to remember the old state for CREATE OR REPLACE */
+  TABLE *table;
+  TABLE_LIST *pos_in_locked_tables;
+  MDL_ticket *mdl_ticket;
+
   bool tmp_table() { return options & HA_LEX_CREATE_TMP_TABLE; }
 };
 
diff --git a/sql/log.cc b/sql/log.cc
index 90305ec227e..9ea1514ef77 100644
--- a/sql/log.cc
+++ b/sql/log.cc
@@ -525,35 +525,57 @@ bool LOGGER::is_log_table_enabled(uint log_table_type)
 }
 
 
-/* Check if a given table is opened log table */
-int check_if_log_table(size_t db_len, const char *db, size_t table_name_len,
-                       const char *table_name, bool check_if_opened)
+/**
+   Check if a given table is opened log table
+
+   @param table             Table to check
+   @param check_if_opened   Only fail if it's a log table in use
+   @param error_msg	    String to put in error message if not ok.
+                            No error message if 0
+   @return 0 ok
+   @return # Type of log file
+ */
+
+int check_if_log_table(const TABLE_LIST *table,
+                       bool check_if_opened,
+                       const char *error_msg)
 {
-  if (db_len == 5 &&
+  int result= 0;
+  if (table->db_length == 5 &&
       !(lower_case_table_names ?
-        my_strcasecmp(system_charset_info, db, "mysql") :
-        strcmp(db, "mysql")))
+        my_strcasecmp(system_charset_info, table->db, "mysql") :
+        strcmp(table->db, "mysql")))
   {
-    if (table_name_len == 11 && !(lower_case_table_names ?
-                                  my_strcasecmp(system_charset_info,
-                                                table_name, "general_log") :
-                                  strcmp(table_name, "general_log")))
+    const char *table_name= table->table_name;
+
+    if (table->table_name_length == 11 &&
+        !(lower_case_table_names ?
+          my_strcasecmp(system_charset_info,
+                        table_name, "general_log") :
+          strcmp(table_name, "general_log")))
     {
-      if (!check_if_opened || logger.is_log_table_enabled(QUERY_LOG_GENERAL))
-        return QUERY_LOG_GENERAL;
-      return 0;
+      result= QUERY_LOG_GENERAL;
+      goto end;
     }
 
-    if (table_name_len == 8 && !(lower_case_table_names ?
+    if (table->table_name_length == 8 && !(lower_case_table_names ?
       my_strcasecmp(system_charset_info, table_name, "slow_log") :
       strcmp(table_name, "slow_log")))
     {
-      if (!check_if_opened || logger.is_log_table_enabled(QUERY_LOG_SLOW))
-        return QUERY_LOG_SLOW;
-      return 0;
+      result= QUERY_LOG_SLOW;
+      goto end;
     }
   }
   return 0;
+
+end:
+  if (!check_if_opened || logger.is_log_table_enabled(result))
+  {
+    if (error_msg)
+      my_error(ER_BAD_LOG_STATEMENT, MYF(0), error_msg);
+    return result;
+  }
+  return 0;
 }
 
 
diff --git a/sql/log.h b/sql/log.h
index 73518d2594f..18e86d9d0f8 100644
--- a/sql/log.h
+++ b/sql/log.h
@@ -833,8 +833,8 @@ public:
 };
 
 
-int check_if_log_table(size_t db_len, const char *db, size_t table_name_len,
-                       const char *table_name, bool check_if_opened);
+int check_if_log_table(const TABLE_LIST *table, bool check_if_opened,
+                       const char *errmsg);
 
 class Log_to_csv_event_handler: public Log_event_handler
 {
diff --git a/sql/slave.cc b/sql/slave.cc
index f6665a6a436..6886a6345ab 100644
--- a/sql/slave.cc
+++ b/sql/slave.cc
@@ -1047,9 +1047,10 @@ static bool sql_slave_killed(rpl_group_info *rgi)
         "documentation for details).";
 
       DBUG_PRINT("info", ("modified_non_trans_table: %d  OPTION_BEGIN: %d  "
-                          "is_in_group: %d",
+                          "OPTION_KEEP_LOG: %d  is_in_group: %d",
                           thd->transaction.all.modified_non_trans_table,
                           test(thd->variables.option_bits & OPTION_BEGIN),
+                          test(thd->variables.option_bits & OPTION_KEEP_LOG),
                           rli->is_in_group()));
 
       if (rli->abort_slave)
diff --git a/sql/sql_alter.cc b/sql/sql_alter.cc
index 01bffaf132f..97b9c127c22 100644
--- a/sql/sql_alter.cc
+++ b/sql/sql_alter.cc
@@ -338,19 +338,8 @@ bool Sql_cmd_discard_import_tablespace::execute(THD *thd)
     it is the case.
     TODO: this design is obsolete and will be removed.
   */
-  int table_kind= check_if_log_table(table_list->db_length, table_list->db,
-                                     table_list->table_name_length,
-                                     table_list->table_name, false);
-
-  if (table_kind)
-  {
-    /* Disable alter of enabled log tables */
-    if (logger.is_log_table_enabled(table_kind))
-    {
-      my_error(ER_BAD_LOG_STATEMENT, MYF(0), "ALTER");
-      return true;
-    }
-  }
+  if (check_if_log_table(table_list, TRUE, "ALTER"))
+    return true;
 
   return
     mysql_discard_or_import_tablespace(thd, table_list,
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index e3ce2adf862..0a38cbbeb1a 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -784,12 +784,18 @@ static void close_open_tables(THD *thd)
                      access the table cache key
 
   @param[in] extra
-                     HA_EXTRA_PREPRE_FOR_DROP if the table is being dropped
-                     HA_EXTRA_PREPARE_FOR_REANME if the table is being renamed
-                     HA_EXTRA_NOT_USED           no drop/rename
-                     In case of drop/reanme the documented behaviour is to
+                     HA_EXTRA_PREPARE_FOR_DROP
+                        - The table is dropped
+                     HA_EXTRA_PREPARE_FOR_RENAME
+                        - The table is renamed
+                     HA_EXTRA_NOT_USED
+                        - The table is marked as closed in the
+                          locked_table_list but kept there so one can call
+                          locked_table_list->reopen_tables() to put it back.
+                          
+                     In case of drop/rename the documented behavior is to
                      implicitly remove the table from LOCK TABLES
-                     list.
+                     list. 
 
   @pre Must be called with an X MDL lock on the table.
 */
@@ -1588,26 +1594,21 @@ TABLE *find_temporary_table(THD *thd,
   thd->temporary_tables list, it's impossible to tell here whether
   we're dealing with an internal or a user temporary table.
 
-  If is_trans is not null, we return the type of the table:
-  either transactional (e.g. innodb) as TRUE or non-transactional
-  (e.g. myisam) as FALSE.
+  @param thd      Thread handler
+  @param table	  Temporary table to be deleted
+  @param is_trans Is set to the type of the table:
+                  transactional (e.g. innodb) as TRUE or non-transactional
+                  (e.g. myisam) as FALSE.
 
   @retval  0  the table was found and dropped successfully.
-  @retval  1  the table was not found in the list of temporary tables
-              of this thread
   @retval -1  the table is in use by a outer query
 */
 
-int drop_temporary_table(THD *thd, TABLE_LIST *table_list, bool *is_trans)
+int drop_temporary_table(THD *thd, TABLE *table, bool *is_trans)
 {
   DBUG_ENTER("drop_temporary_table");
   DBUG_PRINT("tmptable", ("closing table: '%s'.'%s'",
-                          table_list->db, table_list->table_name));
-
-  if (!is_temporary_table(table_list))
-    DBUG_RETURN(1);
-
-  TABLE *table= table_list->table;
+                          table->s->db.str, table->s->table_name.str));
 
   /* Table might be in use by some outer statement. */
   if (table->query_id && table->query_id != thd->query_id)
@@ -1627,10 +1628,10 @@ int drop_temporary_table(THD *thd, TABLE_LIST *table_list, bool *is_trans)
   */
   mysql_lock_remove(thd, thd->lock, table);
   close_temporary_table(thd, table, 1, 1);
-  table_list->table= NULL;
   DBUG_RETURN(0);
 }
 
+
 /*
   unlink from thd->temporary tables and close temporary table
 */
@@ -2611,9 +2612,9 @@ Locked_tables_list::init_locked_tables(THD *thd)
   {
     TABLE_LIST *src_table_list= table->pos_in_table_list;
     char *db, *table_name, *alias;
-    size_t db_len= src_table_list->db_length;
-    size_t table_name_len= src_table_list->table_name_length;
-    size_t alias_len= strlen(src_table_list->alias);
+    size_t db_len=         table->s->db.length;
+    size_t table_name_len= table->s->table_name.length;
+    size_t alias_len=      table->alias.length();
     TABLE_LIST *dst_table_list;
 
     if (! multi_alloc_root(&m_locked_tables_root,
@@ -2623,23 +2624,15 @@ Locked_tables_list::init_locked_tables(THD *thd)
                            &alias, alias_len + 1,
                            NullS))
     {
-      unlock_locked_tables(0);
+      reset();
       return TRUE;
     }
 
-    memcpy(db, src_table_list->db, db_len + 1);
-    memcpy(table_name, src_table_list->table_name, table_name_len + 1);
-    memcpy(alias, src_table_list->alias, alias_len + 1);
-    /**
-      Sic: remember the *actual* table level lock type taken, to
-      acquire the exact same type in reopen_tables().
-      E.g. if the table was locked for write, src_table_list->lock_type is
-      TL_WRITE_DEFAULT, whereas reginfo.lock_type has been updated from
-      thd->update_lock_default.
-    */
+    memcpy(db,         table->s->db.str, db_len + 1);
+    memcpy(table_name, table->s->table_name.str, table_name_len + 1);
+    strmake(alias,     table->alias.ptr(), alias_len);
     dst_table_list->init_one_table(db, db_len, table_name, table_name_len,
-                                   alias,
-                                   src_table_list->table->reginfo.lock_type);
+                                   alias, table->reginfo.lock_type);
     dst_table_list->table= table;
     dst_table_list->mdl_request.ticket= src_table_list->mdl_request.ticket;
 
@@ -2660,7 +2653,7 @@ Locked_tables_list::init_locked_tables(THD *thd)
                                         (m_locked_tables_count+1));
     if (m_reopen_array == NULL)
     {
-      unlock_locked_tables(0);
+      reset();
       return TRUE;
     }
   }
@@ -2681,42 +2674,50 @@ Locked_tables_list::init_locked_tables(THD *thd)
 void
 Locked_tables_list::unlock_locked_tables(THD *thd)
 {
-  if (thd)
+  DBUG_ASSERT(!thd->in_sub_stmt &&
+              !(thd->state_flags & Open_tables_state::BACKUPS_AVAIL));
+  /*
+    Sic: we must be careful to not close open tables if
+    we're not in LOCK TABLES mode: unlock_locked_tables() is
+    sometimes called implicitly, expecting no effect on
+    open tables, e.g. from begin_trans().
+  */
+  if (thd->locked_tables_mode != LTM_LOCK_TABLES)
+    return;
+
+  for (TABLE_LIST *table_list= m_locked_tables;
+       table_list; table_list= table_list->next_global)
   {
-    DBUG_ASSERT(!thd->in_sub_stmt &&
-                !(thd->state_flags & Open_tables_state::BACKUPS_AVAIL));
     /*
-      Sic: we must be careful to not close open tables if
-      we're not in LOCK TABLES mode: unlock_locked_tables() is
-      sometimes called implicitly, expecting no effect on
-      open tables, e.g. from begin_trans().
+      Clear the position in the list, the TABLE object will be
+      returned to the table cache.
     */
-    if (thd->locked_tables_mode != LTM_LOCK_TABLES)
-      return;
+    if (table_list->table)                    // If not closed
+      table_list->table->pos_in_locked_tables= NULL;
+  }
+  thd->leave_locked_tables_mode();
 
-    for (TABLE_LIST *table_list= m_locked_tables;
-         table_list; table_list= table_list->next_global)
-    {
-      /*
-        Clear the position in the list, the TABLE object will be
-        returned to the table cache.
-      */
-      if (table_list->table)                    // If not closed
-        table_list->table->pos_in_locked_tables= NULL;
-    }
-    thd->leave_locked_tables_mode();
+  DBUG_ASSERT(thd->transaction.stmt.is_empty());
+  close_thread_tables(thd);
+
+  /*
+    We rely on the caller to implicitly commit the
+    transaction and release transactional locks.
+  */
 
-    DBUG_ASSERT(thd->transaction.stmt.is_empty());
-    close_thread_tables(thd);
-    /*
-      We rely on the caller to implicitly commit the
-      transaction and release transactional locks.
-    */
-  }
   /*
     After closing tables we can free memory used for storing lock
     request for metadata locks and TABLE_LIST elements.
   */
+  reset();
+}
+
+/*
+  Free memory allocated for storing locks
+*/
+
+void Locked_tables_list::reset()
+{
   free_root(&m_locked_tables_root, MYF(0));
   m_locked_tables= NULL;
   m_locked_tables_last= &m_locked_tables;
@@ -2781,6 +2782,7 @@ void Locked_tables_list::unlink_from_list(THD *thd,
       m_locked_tables_last= table_list->prev_global;
     else
       table_list->next_global->prev_global= table_list->prev_global;
+    m_locked_tables_count--;
   }
 }
 
@@ -2834,6 +2836,7 @@ unlink_all_closed_tables(THD *thd, MYSQL_LOCK *lock, size_t reopen_count)
         m_locked_tables_last= table_list->prev_global;
       else
         table_list->next_global->prev_global= table_list->prev_global;
+      m_locked_tables_count--;
     }
   }
 }
@@ -2908,6 +2911,57 @@ Locked_tables_list::reopen_tables(THD *thd)
   return FALSE;
 }
 
+/**
+  Add back a locked table to the locked list that we just removed from it.
+  This is needed in CREATE OR REPLACE TABLE where we are dropping, creating
+  and re-opening a locked table.
+
+  @return 0  0k
+  @return 1  error
+*/
+
+bool Locked_tables_list::restore_lock(THD *thd, TABLE_LIST *dst_table_list,
+                                      TABLE *table, MYSQL_LOCK *lock)
+{
+  MYSQL_LOCK *merged_lock;
+  DBUG_ENTER("restore_lock");
+  DBUG_ASSERT(!strcmp(dst_table_list->table_name, table->s->table_name.str));
+
+  /* Ensure we have the memory to add the table back */
+  if (!(merged_lock= mysql_lock_merge(thd->lock, lock)))
+    DBUG_RETURN(1);
+  thd->lock= merged_lock;
+
+  /* Link to the new table */
+  dst_table_list->table= table;
+  /*
+    The lock type may have changed (normally it should not as create
+    table will lock the table in write mode
+  */
+  dst_table_list->lock_type= table->reginfo.lock_type;
+  table->pos_in_locked_tables= dst_table_list;
+
+  add_back_last_deleted_lock(dst_table_list);
+
+  DBUG_RETURN(0);
+}
+
+/*
+  Add back the last deleted lock structure.
+  This should be followed by a call to reopen_tables() to
+  open the table.
+*/
+
+void Locked_tables_list::add_back_last_deleted_lock(TABLE_LIST *dst_table_list)
+{
+  /* Link the lock back in the locked tables list */
+  dst_table_list->prev_global= m_locked_tables_last;
+  *m_locked_tables_last= dst_table_list;
+  m_locked_tables_last= &dst_table_list->next_global;
+  dst_table_list->next_global= 0;
+  m_locked_tables_count++;
+}
+
 
 #ifndef DBUG_OFF
 /* Cause a spurious statement reprepare for debug purposes. */
@@ -4045,9 +4099,9 @@ lock_table_names(THD *thd,
   if (mdl_requests.is_empty())
     DBUG_RETURN(FALSE);
 
-  /* Check if CREATE TABLE was used */
-  create_table= (tables_start && tables_start->open_strategy ==
-                 TABLE_LIST::OPEN_IF_EXISTS);
+  /* Check if CREATE TABLE without REPLACE was used */
+  create_table= (thd->lex->sql_command == SQLCOM_CREATE_TABLE &&
+                 !(thd->lex->create_info.options & HA_LEX_CREATE_REPLACE));
 
   if (!(flags & MYSQL_OPEN_SKIP_SCOPED_MDL_LOCK))
   {
@@ -5293,6 +5347,39 @@ bool lock_tables(THD *thd, TABLE_LIST *tables, uint count,
 }
 
 
+/*
+  Restart transaction for tables
+
+  This is used when we had to do an implicit commit after tables are opened
+  and want to restart transactions on tables.
+
+  This is used in case of:
+  LOCK TABLES xx
+  CREATE OR REPLACE TABLE xx;
+*/
+
+bool restart_trans_for_tables(THD *thd, TABLE_LIST *table)
+{
+  DBUG_ENTER("restart_trans_for_tables");
+
+  if (!thd->locked_tables_mode)
+    DBUG_RETURN(FALSE);
+
+  for (; table; table= table->next_global)
+  {
+    if (table->placeholder())
+      continue;
+
+    if (check_lock_and_start_stmt(thd, thd->lex, table))
+    {
+      DBUG_ASSERT(0);                           // Should never happen
+      DBUG_RETURN(TRUE);
+    }
+  }
+  DBUG_RETURN(FALSE);
+}
+
+
 /**
   Prepare statement for reopening of tables and recalculation of set of
   prelocked tables.
diff --git a/sql/sql_base.h b/sql/sql_base.h
index 3e633fad084..61442843a39 100644
--- a/sql/sql_base.h
+++ b/sql/sql_base.h
@@ -248,7 +248,7 @@ void close_thread_table(THD *thd, TABLE **table_ptr);
 bool close_temporary_tables(THD *thd);
 TABLE_LIST *unique_table(THD *thd, TABLE_LIST *table, TABLE_LIST *table_list,
                          bool check_alias);
-int drop_temporary_table(THD *thd, TABLE_LIST *table_list, bool *is_trans);
+int drop_temporary_table(THD *thd, TABLE *table, bool *is_trans);
 void close_temporary_table(THD *thd, TABLE *table, bool free_share,
                            bool delete_table);
 void close_temporary(TABLE *table, bool free_share, bool delete_table);
@@ -486,6 +486,8 @@ inline bool open_and_lock_tables(THD *thd, TABLE_LIST *tables,
 }
 
 
+bool restart_trans_for_tables(THD *thd, TABLE_LIST *table);
+
 /**
   A context of open_tables() function, used to recover
   from a failed open_table() or open_routine() attempt.
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 19c1ac8f4b6..14f58b30f3c 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -1513,8 +1513,9 @@ public:
   void unlock_locked_tables(THD *thd);
   ~Locked_tables_list()
   {
-    unlock_locked_tables(0);
+    reset();
   }
+  void reset();
   bool init_locked_tables(THD *thd);
   TABLE_LIST *locked_tables() { return m_locked_tables; }
   void unlink_from_list(THD *thd, TABLE_LIST *table_list,
@@ -1523,6 +1524,9 @@ public:
                                 MYSQL_LOCK *lock,
                                 size_t reopen_count);
   bool reopen_tables(THD *thd);
+  bool restore_lock(THD *thd, TABLE_LIST *dst_table_list, TABLE *table,
+                    MYSQL_LOCK *lock);
+  void add_back_last_deleted_lock(TABLE_LIST *dst_table_list);
 };
 
 
@@ -3995,6 +3999,8 @@ class select_create: public select_insert {
   MYSQL_LOCK *m_lock;
   /* m_lock or thd->extra_lock */
   MYSQL_LOCK **m_plock;
+  bool       exit_done;
+
 public:
   select_create (TABLE_LIST *table_arg,
 		 HA_CREATE_INFO *create_info_par,
@@ -4006,7 +4012,7 @@ public:
     create_info(create_info_par),
     select_tables(select_tables_arg),
     alter_info(alter_info_arg),
-    m_plock(NULL)
+    m_plock(NULL), exit_done(0)
     {}
   int prepare(List<Item> &list, SELECT_LEX_UNIT *u);
 
diff --git a/sql/sql_db.cc b/sql/sql_db.cc
index 8068901ebec..85e9b1aa852 100644
--- a/sql/sql_db.cc
+++ b/sql/sql_db.cc
@@ -798,14 +798,8 @@ bool mysql_rm_db(THD *thd,char *db,bool if_exists, bool silent)
   if ((my_strcasecmp(system_charset_info, MYSQL_SCHEMA_NAME.str, db) == 0))
   {
     for (table= tables; table; table= table->next_local)
-    {
-      if (check_if_log_table(table->db_length, table->db,
-                             table->table_name_length, table->table_name, true))
-      {
-        my_error(ER_BAD_LOG_STATEMENT, MYF(0), "DROP");
+      if (check_if_log_table(table, TRUE, "DROP"))
         goto exit;
-      }
-    }
   }
 
   /* Lock all tables and stored routines about to be dropped. */
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index 8e3826eff48..719e0bae4ff 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -3916,6 +3916,16 @@ static TABLE *create_table_from_items(THD *thd, HA_CREATE_INFO *create_info,
 
   DEBUG_SYNC(thd,"create_table_select_before_create");
 
+  /* Check if LOCK TABLES + CREATE OR REPLACE of existing normal table*/
+  if (thd->locked_tables_mode && create_table->table &&
+      !create_info->tmp_table())
+  {
+    /* Remember information about the locked table */
+    create_info->pos_in_locked_tables=
+      create_table->table->pos_in_locked_tables;
+    create_info->mdl_ticket= create_table->table->mdl_ticket;
+  }
+
   /*
     Create and lock table.
 
@@ -3932,52 +3942,63 @@ static TABLE *create_table_from_items(THD *thd, HA_CREATE_INFO *create_info,
     TABLE, which is a wrong order. So we keep binary logging disabled when we
     open_table().
   */
+
+  if (!mysql_create_table_no_lock(thd, create_table->db,
+                                  create_table->table_name,
+                                  create_info, alter_info, NULL,
+                                  select_field_count))
   {
-    if (!mysql_create_table_no_lock(thd, create_table->db,
-                                    create_table->table_name,
-                                    create_info, alter_info, NULL,
-                                    select_field_count))
+    DEBUG_SYNC(thd,"create_table_select_before_open");
+
+    /*
+      If we had a temporary table or a table used with LOCK TABLES,
+      it was closed by mysql_create()
+    */
+    create_table->table= 0;
+
+    if (!create_info->tmp_table())
     {
-      DEBUG_SYNC(thd,"create_table_select_before_open");
+      Open_table_context ot_ctx(thd, MYSQL_OPEN_REOPEN);
+      TABLE_LIST::enum_open_strategy save_open_strategy;
 
-      if (!create_info->tmp_table())
-      {
-        Open_table_context ot_ctx(thd, MYSQL_OPEN_REOPEN);
-        /*
-          Here we open the destination table, on which we already have
-          an exclusive metadata lock.
-        */
-        if (open_table(thd, create_table, thd->mem_root, &ot_ctx))
-        {
-          quick_rm_table(thd, create_info->db_type, create_table->db,
-                         table_case_name(create_info, create_table->table_name),
-                         0);
-        }
-        else
-          table= create_table->table;
-      }
-      else
+      /* Force the newly created table to be opened */
+      save_open_strategy= create_table->open_strategy;
+      create_table->open_strategy= TABLE_LIST::OPEN_NORMAL;
+      /*
+        Here we open the destination table, on which we already have
+        an exclusive metadata lock.
+      */
+      if (open_table(thd, create_table, thd->mem_root, &ot_ctx))
       {
-        if (open_temporary_table(thd, create_table))
-        {
-          /*
-            This shouldn't happen as creation of temporary table should make
-            it preparable for open. Anyway we can't drop temporary table if
-            we are unable to find it.
-          */
-          DBUG_ASSERT(0);
-        }
-        else
-          table= create_table->table;
+        quick_rm_table(thd, create_info->db_type, create_table->db,
+                       table_case_name(create_info, create_table->table_name),
+                       0);
       }
+      /* Restore */
+      create_table->open_strategy= save_open_strategy;
     }
-    if (!table)                                   // open failed
+    else
     {
-      if (!thd->is_error())                     // CREATE ... IF NOT EXISTS
-        my_ok(thd);                             //   succeed, but did nothing
-      DBUG_RETURN(0);
+      if (open_temporary_table(thd, create_table))
+      {
+        /*
+          This shouldn't happen as creation of temporary table should make
+          it preparable for open. Anyway we can't drop temporary table if
+          we are unable to find it.
+        */
+        DBUG_ASSERT(0);
+      }
     }
   }
+  else
+    create_table->table= 0;                     // Create failed
+  
+  if (!(table= create_table->table))
+  {
+    if (!thd->is_error())                     // CREATE ... IF NOT EXISTS
+      my_ok(thd);                             //   succeed, but did nothing
+    DBUG_RETURN(0);
+  }
 
   DEBUG_SYNC(thd,"create_table_select_before_lock");
 
@@ -3994,7 +4015,7 @@ static TABLE *create_table_from_items(THD *thd, HA_CREATE_INFO *create_info,
     /* purecov: begin tested */
     /*
       This can happen in innodb when you get a deadlock when using same table
-      in insert and select
+      in insert and select or when you run out of memory.
     */
     my_error(ER_CANT_LOCK, MYF(0), my_errno);
     if (*lock)
@@ -4092,8 +4113,6 @@ select_create::prepare(List<Item> &values, SELECT_LEX_UNIT *u)
     thd->binlog_start_trans_and_stmt();
   }
 
-  DBUG_ASSERT(create_table->table == NULL);
-
   DEBUG_SYNC(thd,"create_table_select_before_check_if_exists");
 
   if (!(table= create_table_from_items(thd, create_info, create_table,
@@ -4236,32 +4255,53 @@ void select_create::send_error(uint errcode,const char *err)
 
 bool select_create::send_eof()
 {
-  bool tmp=select_insert::send_eof();
-  if (tmp)
+  if (select_insert::send_eof())
+  {
     abort_result_set();
-  else
+    return 1;
+  }
+
+  exit_done= 1;                                 // Avoid double calls
+  /*
+    Do an implicit commit at end of statement for non-temporary
+    tables.  This can fail, but we should unlock the table
+    nevertheless.
+  */
+  if (!table->s->tmp_table)
   {
-    /*
-      Do an implicit commit at end of statement for non-temporary
-      tables.  This can fail, but we should unlock the table
-      nevertheless.
-    */
-    if (!table->s->tmp_table)
-    {
-      trans_commit_stmt(thd);
-      trans_commit_implicit(thd);
-    }
+    trans_commit_stmt(thd);
+    trans_commit_implicit(thd);
+  }
 
-    table->file->extra(HA_EXTRA_NO_IGNORE_DUP_KEY);
-    table->file->extra(HA_EXTRA_WRITE_CANNOT_REPLACE);
-    if (m_plock)
+  table->file->extra(HA_EXTRA_NO_IGNORE_DUP_KEY);
+  table->file->extra(HA_EXTRA_WRITE_CANNOT_REPLACE);
+
+  if (m_plock)
+  {
+    MYSQL_LOCK *lock= *m_plock;
+    *m_plock= NULL;
+    m_plock= NULL;
+
+    if (create_info->pos_in_locked_tables)
     {
-      mysql_unlock_tables(thd, *m_plock);
-      *m_plock= NULL;
-      m_plock= NULL;
+      /*
+        If we are under lock tables, we have created a table that was
+        originally locked. We should add back the lock to ensure that
+        all tables in the thd->open_list are locked!
+      */
+      table->mdl_ticket= create_info->mdl_ticket;
+
+      /* The following should never fail, except if out of memory */
+      if (!thd->locked_tables_list.restore_lock(thd,
+                                                create_info->
+                                                pos_in_locked_tables,
+                                                table, lock))
+        return 0;                               // ok
+      /* Fail. Continue without locking the table */
     }
+    mysql_unlock_tables(thd, lock);
   }
-  return tmp;
+  return 0;
 }
 
 
@@ -4269,6 +4309,11 @@ void select_create::abort_result_set()
 {
   DBUG_ENTER("select_create::abort_result_set");
 
+  /* Avoid double calls, could happen in case of out of memory on cleanup */
+  if (exit_done)
+    DBUG_VOID_RETURN;
+  exit_done= 1;
+
   /*
     In select_insert::abort_result_set() we roll back the statement, including
     truncating the transaction cache of the binary log. To do this, we
diff --git a/sql/sql_lex.h b/sql/sql_lex.h
index 06c78ba81a8..479fa81edd7 100644
--- a/sql/sql_lex.h
+++ b/sql/sql_lex.h
@@ -2470,7 +2470,6 @@ struct LEX: public Query_tables_list
   uint8 context_analysis_only;
   bool drop_temporary, local_file, one_shot_set;
   bool check_exists;
-  bool replace;
   bool autocommit;
   bool verbose, no_write_to_binlog;
 
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index f8af7873a50..706e9e1a3a6 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -2832,12 +2832,8 @@ case SQLCOM_PREPARE:
     if ((res= create_table_precheck(thd, select_tables, create_table)))
       goto end_with_restore_list;
 
-#ifndef QQ
     /* Might have been updated in create_table_precheck */
     create_info.alias= create_table->alias;
-#else
-    create_table->alias= (char*) create_info.alias;
-#endif
 
 #ifdef HAVE_READLINK
     /* Fix names if symlinked tables */
@@ -2867,6 +2863,12 @@ case SQLCOM_PREPARE:
       create_info.table_charset= 0;
     }
 
+    /*
+      For CREATE TABLE we should not open the table even if it exists.
+      If the table exists, we should either not create it or replace it
+    */
+    lex->query_tables->open_strategy= TABLE_LIST::OPEN_STUB;
+
 #ifdef WITH_PARTITION_STORAGE_ENGINE
     {
       partition_info *part_info= thd->lex->part_info;
@@ -2958,25 +2960,6 @@ case SQLCOM_PREPARE:
       }
       else
       {
-        /* The table already exists */
-        if (create_table->table)
-        {
-          if (create_info.options & HA_LEX_CREATE_IF_NOT_EXISTS)
-          {
-            push_warning_printf(thd, Sql_condition::WARN_LEVEL_NOTE,
-                                ER_TABLE_EXISTS_ERROR,
-                                ER(ER_TABLE_EXISTS_ERROR),
-                                create_info.alias);
-            my_ok(thd);
-          }
-          else
-          {
-            my_error(ER_TABLE_EXISTS_ERROR, MYF(0), create_info.alias);
-            res= 1;
-          }
-          goto end_with_restore_list;
-        }
-
         /*
           Remove target table from main select and name resolution
           context. This can't be done earlier as it will break view merging in
@@ -2984,9 +2967,8 @@ case SQLCOM_PREPARE:
         */
         lex->unlink_first_table(&link_to_local);
 
-        /* So that CREATE TEMPORARY TABLE gets to binlog at commit/rollback */
-        if (create_info.tmp_table())
-          thd->variables.option_bits|= OPTION_KEEP_LOG;
+        /* Store reference to table in case of LOCK TABLES */
+        create_info.table= create_table->table;
 
         /*
           select_create is currently not re-execution friendly and
@@ -3004,18 +2986,18 @@ case SQLCOM_PREPARE:
             CREATE from SELECT give its SELECT_LEX for SELECT,
             and item_list belong to SELECT
           */
-          res= handle_select(thd, lex, result, 0);
+          if (!(res= handle_select(thd, lex, result, 0)))
+          {
+            if (create_info.tmp_table())
+              thd->variables.option_bits|= OPTION_KEEP_LOG;
+          }
           delete result;
         }
-
         lex->link_first_table_back(create_table, link_to_local);
       }
     }
     else
     {
-      /* So that CREATE TEMPORARY TABLE gets to binlog at commit/rollback */
-      if (create_info.tmp_table())
-        thd->variables.option_bits|= OPTION_KEEP_LOG;
       /* regular create */
       if (create_info.options & HA_LEX_CREATE_TABLE_LIKE)
       {
@@ -3030,7 +3012,12 @@ case SQLCOM_PREPARE:
                                 &create_info, &alter_info);
       }
       if (!res)
+      {
+        /* So that CREATE TEMPORARY TABLE gets to binlog at commit/rollback */
+        if (create_info.tmp_table())
+          thd->variables.option_bits|= OPTION_KEEP_LOG;
         my_ok(thd);
+      }
     }
 
 end_with_restore_list:
@@ -7960,8 +7947,9 @@ bool create_table_precheck(THD *thd, TABLE_LIST *tables,
              (CREATE_ACL | (select_lex->item_list.elements ? INSERT_ACL : 0));
 
   /* CREATE OR REPLACE on not temporary tables require DROP_ACL */
-  if (lex->replace && !lex->create_info.tmp_table())
-    want_priv= DROP_ACL;
+  if ((lex->create_info.options & HA_LEX_CREATE_REPLACE) &&
+      !lex->create_info.tmp_table())
+    want_priv|= DROP_ACL;
                           
   if (check_access(thd, want_priv, create_table->db,
                    &create_table->grant.privilege,
@@ -8031,6 +8019,12 @@ bool create_table_precheck(THD *thd, TABLE_LIST *tables,
   }
   error= FALSE;
 
+  /*
+    For CREATE TABLE we should not open the table even if it exists.
+    If the table exists, we should either not create it or replace it
+  */
+  lex->query_tables->open_strategy= TABLE_LIST::OPEN_STUB;
+
 err:
   DBUG_RETURN(error);
 }
diff --git a/sql/sql_partition_admin.cc b/sql/sql_partition_admin.cc
index 29ca86fa274..b2a8bca72db 100644
--- a/sql/sql_partition_admin.cc
+++ b/sql/sql_partition_admin.cc
@@ -506,13 +506,8 @@ bool Sql_cmd_alter_table_exchange_partition::
 
   /* Don't allow to exchange with log table */
   swap_table_list= table_list->next_local;
-  if (check_if_log_table(swap_table_list->db_length, swap_table_list->db,
-                         swap_table_list->table_name_length,
-                         swap_table_list->table_name, 0))
-  {
-    my_error(ER_WRONG_USAGE, MYF(0), "PARTITION", "log table");
+  if (check_if_log_table(swap_table_list, FALSE, "ALTER PARTITION"))
     DBUG_RETURN(TRUE);
-  }
 
   /*
     Currently no MDL lock that allows both read and write and is upgradeable
diff --git a/sql/sql_rename.cc b/sql/sql_rename.cc
index 6babdd7c636..897aa183b60 100644
--- a/sql/sql_rename.cc
+++ b/sql/sql_rename.cc
@@ -84,12 +84,8 @@ bool mysql_rename_tables(THD *thd, TABLE_LIST *table_list, bool silent)
     for (to_table= 0, ren_table= table_list; ren_table;
          to_table= 1 - to_table, ren_table= ren_table->next_local)
     {
-      int log_table_rename= 0;
-
-      if ((log_table_rename=
-           check_if_log_table(ren_table->db_length, ren_table->db,
-                              ren_table->table_name_length,
-                              ren_table->table_name, 1)))
+      int log_table_rename;
+      if ((log_table_rename= check_if_log_table(ren_table, TRUE, NullS)))
       {
         /*
           as we use log_table_rename as an array index, we need it to start
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 45d9c5dc091..32e4fdc18a1 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -2031,33 +2031,29 @@ bool mysql_rm_table(THD *thd,TABLE_LIST *tables, my_bool if_exists,
   bool error;
   Drop_table_error_handler err_handler;
   TABLE_LIST *table;
-
   DBUG_ENTER("mysql_rm_table");
 
   /* Disable drop of enabled log tables, must be done before name locking */
   for (table= tables; table; table= table->next_local)
   {
-    if (check_if_log_table(table->db_length, table->db,
-                           table->table_name_length, table->table_name, true))
-    {
-      my_error(ER_BAD_LOG_STATEMENT, MYF(0), "DROP");
+    if (check_if_log_table(table, TRUE, "DROP"))
       DBUG_RETURN(true);
-    }
   }
 
-  if (!in_bootstrap)
+  if (!drop_temporary)
   {
-    for (table= tables; table; table= table->next_local)
+    if (!in_bootstrap)
     {
-      LEX_STRING db_name= { table->db, table->db_length };
-      LEX_STRING table_name= { table->table_name, table->table_name_length };
-      if (table->open_type == OT_BASE_ONLY || !find_temporary_table(thd, table))
-        (void) delete_statistics_for_table(thd, &db_name, &table_name);
+      for (table= tables; table; table= table->next_local)
+      {
+        LEX_STRING db_name= { table->db, table->db_length };
+        LEX_STRING table_name= { table->table_name, table->table_name_length };
+        if (table->open_type == OT_BASE_ONLY ||
+            !find_temporary_table(thd, table))
+          (void) delete_statistics_for_table(thd, &db_name, &table_name);
+      }
     }
-  }
 
-  if (!drop_temporary)
-  {
     if (!thd->locked_tables_mode)
     {
       if (lock_table_names(thd, tables, NULL,
@@ -2286,7 +2282,7 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
 
   for (table= tables; table; table= table->next_local)
   {
-    bool is_trans;
+    bool is_trans= 0;
     char *db=table->db;
     size_t db_length= table->db_length;
     handlerton *table_type= 0;
@@ -2311,12 +2307,16 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
       .  1 - a temporary table was not found.
       . -1 - a temporary table is used by an outer statement.
     */
-    if (table->open_type == OT_BASE_ONLY)
+    if (table->open_type == OT_BASE_ONLY || !is_temporary_table(table))
       error= 1;
-    else if ((error= drop_temporary_table(thd, table, &is_trans)) == -1)
+    else
     {
-      DBUG_ASSERT(thd->in_sub_stmt);
-      goto err;
+      if ((error= drop_temporary_table(thd, table->table, &is_trans)) == -1)
+      {
+        DBUG_ASSERT(thd->in_sub_stmt);
+        goto err;
+      }
+      table->table= 0;
     }
 
     if ((drop_temporary && if_exists) || !error)
@@ -4517,12 +4517,13 @@ err:
   way to ensure that concurrent operations won't intervene.
   mysql_create_table() is a wrapper that can be used for this.
 
-  @retval false OK
-  @retval true  error
+  @retval 0 OK
+  @retval 1 error
+  @retval -1 table existed but IF EXISTS was used
 */
 
 static
-bool create_table_impl(THD *thd,
+int create_table_impl(THD *thd,
                        const char *db, const char *table_name,
                        const char *path,
                        HA_CREATE_INFO *create_info,
@@ -4535,7 +4536,7 @@ bool create_table_impl(THD *thd,
 {
   const char	*alias;
   handler	*file= 0;
-  bool		error= TRUE;
+  int		error= 1;
   bool          frm_only= create_table_mode == C_ALTER_TABLE_FRM_ONLY;
   bool          internal_tmp_table= create_table_mode == C_ALTER_TABLE || frm_only;
   DBUG_ENTER("mysql_create_table_no_lock");
@@ -4565,22 +4566,66 @@ bool create_table_impl(THD *thd,
   /* Check if table exists */
   if (create_info->tmp_table())
   {
-    if (find_temporary_table(thd, db, table_name))
+    TABLE *tmp_table;
+    if ((tmp_table= find_temporary_table(thd, db, table_name)))
     {
-      if (create_info->options & HA_LEX_CREATE_IF_NOT_EXISTS)
+      if (create_info->options & HA_LEX_CREATE_REPLACE)
+      {
+        bool is_trans;
+        /*
+          We are using CREATE OR REPLACE on an existing temporary table
+          Remove the old table so that we can re-create it.
+        */
+        if (drop_temporary_table(thd, tmp_table, &is_trans))
+          goto err;
+      }
+      else if (create_info->options & HA_LEX_CREATE_IF_NOT_EXISTS)
         goto warn;
-      my_error(ER_TABLE_EXISTS_ERROR, MYF(0), alias);
-      goto err;
+      else
+      {
+        my_error(ER_TABLE_EXISTS_ERROR, MYF(0), alias);
+        goto err;
+      }
     }
   }
-  else  
+  else
   {
     if (!internal_tmp_table && ha_table_exists(thd, db, table_name))
     {
-      if (create_info->options & HA_LEX_CREATE_IF_NOT_EXISTS)
+      if (create_info->options & HA_LEX_CREATE_REPLACE)
+      {
+        TABLE_LIST table_list;
+        table_list.init_one_table(db, strlen(db), table_name,
+                                  strlen(table_name), table_name,
+                                  TL_WRITE_ALLOW_WRITE);
+        table_list.table= create_info->table;
+
+        if (check_if_log_table(&table_list, TRUE, "CREATE OR REPLACE"))
+          goto err;
+        
+        /*
+          Rollback the empty transaction started in mysql_create_table()
+          call to open_and_lock_tables() when we are using LOCK TABLES.
+        */
+        (void) trans_rollback_stmt(thd);
+        /* Remove normal table without logging */
+        if (mysql_rm_table_no_locks(thd, &table_list, 0, 0, 0, 1))
+          goto err;
+        /*
+          The test of query_tables is to ensure we have any tables in the
+          select part
+        */
+        if (thd->lex->query_tables &&
+            restart_trans_for_tables(thd, thd->lex->query_tables->next_global))
+          goto err;
+      }
+      else if (create_info->options & HA_LEX_CREATE_IF_NOT_EXISTS)
         goto warn;
-      my_error(ER_TABLE_EXISTS_ERROR, MYF(0), table_name);
-      goto err;
+      else
+      {
+        my_error(ER_TABLE_EXISTS_ERROR, MYF(0), table_name);
+        goto err;
+      }
     }
   }
 
@@ -4702,14 +4747,14 @@ bool create_table_impl(THD *thd,
   }
 #endif
 
-  error= FALSE;
+  error= 0;
 err:
   THD_STAGE_INFO(thd, stage_after_create);
   delete file;
   DBUG_RETURN(error);
 
 warn:
-  error= FALSE;
+  error= -1;
   push_warning_printf(thd, Sql_condition::WARN_LEVEL_NOTE,
                       ER_TABLE_EXISTS_ERROR, ER(ER_TABLE_EXISTS_ERROR),
                       alias);
@@ -4720,7 +4765,8 @@ warn:
   Simple wrapper around create_table_impl() to be used
   in various version of CREATE TABLE statement.
 */
-bool mysql_create_table_no_lock(THD *thd,
+
+int mysql_create_table_no_lock(THD *thd,
                                 const char *db, const char *table_name,
                                 HA_CREATE_INFO *create_info,
                                 Alter_info *alter_info, bool *is_trans,
@@ -4728,6 +4774,7 @@ bool mysql_create_table_no_lock(THD *thd,
 {
   KEY *not_used_1;
   uint not_used_2;
+  int res;
   char path[FN_REFLEN + 1];
   LEX_CUSTRING frm= {0,0};
 
@@ -4747,9 +4794,9 @@ bool mysql_create_table_no_lock(THD *thd,
     }
   }
 
-  bool res= create_table_impl(thd, db, table_name, path, create_info,
-                              alter_info, create_table_mode, is_trans,
-                              &not_used_1, &not_used_2, &frm);
+  res= create_table_impl(thd, db, table_name, path, create_info,
+                         alter_info, create_table_mode, is_trans,
+                         &not_used_1, &not_used_2, &frm);
   my_free(const_cast<uchar*>(frm.str));
   return res;
 }
@@ -4772,15 +4819,21 @@ bool mysql_create_table(THD *thd, TABLE_LIST *create_table,
   const char *table_name= create_table->table_name;
   bool is_trans= FALSE;
   int create_table_mode;
+  TABLE_LIST *pos_in_locked_tables= 0;
   DBUG_ENTER("mysql_create_table");
 
+  DBUG_ASSERT(create_table == thd->lex->query_tables);
+
   /* Open or obtain an exclusive metadata lock on table being created  */
   if (open_and_lock_tables(thd, thd->lex->query_tables, FALSE, 0))
   {
     /* is_error() may be 0 if table existed and we generated a warning */
     DBUG_RETURN(thd->is_error());
   }
-
+  /* The following is needed only in case of lock tables */
+  if ((create_info->table= thd->lex->query_tables->table))
+    pos_in_locked_tables= create_info->table->pos_in_locked_tables;
+  
   /* Got lock. */
   DEBUG_SYNC(thd, "locked_table_name");
 
@@ -4791,9 +4844,25 @@ bool mysql_create_table(THD *thd, TABLE_LIST *create_table,
 
   promote_first_timestamp_column(&alter_info->create_list);
   if (mysql_create_table_no_lock(thd, db, table_name, create_info, alter_info,
-                                 &is_trans, create_table_mode))
+                                 &is_trans, create_table_mode) > 0)
     DBUG_RETURN(1);
 
+  /*
+    Check if we are doing CREATE OR REPLACE TABLE under LOCK TABLES
+    on a non temporary table
+  */
+  if (thd->locked_tables_mode && pos_in_locked_tables &&
+      (create_info->options & HA_LEX_CREATE_REPLACE))
+  {
+    /*
+      Add back the deleted table and re-created table as a locked table
+      This should always work as we have a meta lock on the table.
+     */
+    thd->locked_tables_list.add_back_last_deleted_lock(pos_in_locked_tables);
+    if (thd->locked_tables_list.reopen_tables(thd))
+      thd->locked_tables_list.unlink_all_closed_tables(thd, NULL, 0);
+  }
+
   /* In RBR we don't need to log CREATE TEMPORARY TABLE */
   if (thd->is_current_stmt_binlog_format_row() && create_info->tmp_table())
     DBUG_RETURN(0);
@@ -4986,10 +5055,12 @@ mysql_rename_table(handlerton *base, const char *old_db,
     TRUE  error
 */
 
-bool mysql_create_like_table(THD* thd, TABLE_LIST* table, TABLE_LIST* src_table,
+bool mysql_create_like_table(THD* thd, TABLE_LIST* table,
+                             TABLE_LIST* src_table,
                              HA_CREATE_INFO *create_info)
 {
   HA_CREATE_INFO local_create_info;
+  TABLE_LIST *pos_in_locked_tables= 0;
   Alter_info local_alter_info;
   Alter_table_ctx local_alter_ctx; // Not used
   bool res= TRUE;
@@ -4997,7 +5068,6 @@ bool mysql_create_like_table(THD* thd, TABLE_LIST* table, TABLE_LIST* src_table,
   uint not_used;
   DBUG_ENTER("mysql_create_like_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
@@ -5041,7 +5111,9 @@ bool mysql_create_like_table(THD* thd, TABLE_LIST* table, TABLE_LIST* src_table,
   if (src_table->schema_table)
     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;
+  local_create_info.options|= (create_info->options &
+                               (HA_LEX_CREATE_IF_NOT_EXISTS | 
+                                HA_LEX_CREATE_REPLACE));
   /* 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->tmp_table();
@@ -5053,12 +5125,40 @@ bool mysql_create_like_table(THD* thd, TABLE_LIST* table, TABLE_LIST* src_table,
   */
   local_create_info.data_file_name= local_create_info.index_file_name= NULL;
 
-  if ((res= mysql_create_table_no_lock(thd, table->db, table->table_name,
-                                       &local_create_info, &local_alter_info,
-                                       &is_trans, C_ORDINARY_CREATE)))
+  /* The following is needed only in case of lock tables */
+  if ((local_create_info.table= thd->lex->query_tables->table))
+    pos_in_locked_tables= local_create_info.table->pos_in_locked_tables;    
+
+  if ((res= (mysql_create_table_no_lock(thd, table->db, table->table_name,
+                                        &local_create_info, &local_alter_info,
+                                        &is_trans, C_ORDINARY_CREATE) > 0)))
     goto err;
 
   /*
+    Check if we are doing CREATE OR REPLACE TABLE under LOCK TABLES
+    on a non temporary table
+  */
+  if (thd->locked_tables_mode && pos_in_locked_tables &&
+      (create_info->options & HA_LEX_CREATE_REPLACE))
+  {
+    /*
+      Add back the deleted table and re-created table as a locked table
+      This should always work as we have a meta lock on the table.
+     */
+    thd->locked_tables_list.add_back_last_deleted_lock(pos_in_locked_tables);
+    if (thd->locked_tables_list.reopen_tables(thd))
+      thd->locked_tables_list.unlink_all_closed_tables(thd, NULL, 0);
+    else
+    {
+      /*
+        Get pointer to the newly opened table. We need this to ensure we
+        don't reopen the table when doing statment logging below.
+      */
+      table->table= pos_in_locked_tables->table;
+    }
+  }
+
+  /*
     Ensure that we have an exclusive lock on target table if we are creating
     non-temporary table.
   */
@@ -5108,6 +5208,11 @@ bool mysql_create_like_table(THD* thd, TABLE_LIST* table, TABLE_LIST* src_table,
         {
           if (!table->table)
           {
+            TABLE_LIST::enum_open_strategy save_open_strategy;
+            int open_res;
+            /* Force the newly created table to be opened */
+            save_open_strategy= table->open_strategy;
+            table->open_strategy= TABLE_LIST::OPEN_NORMAL;
 
             /*
               In order for store_create_info() to work we need to open
@@ -5117,11 +5222,20 @@ bool mysql_create_like_table(THD* thd, TABLE_LIST* table, TABLE_LIST* src_table,
               lock on this table. The table will be closed by
               close_thread_table() at the end of this branch.
             */
-            if (open_table(thd, table, thd->mem_root, &ot_ctx))
+            open_res= open_table(thd, table, thd->mem_root, &ot_ctx);
+            /* Restore */
+            table->open_strategy= save_open_strategy;
+            if (open_res)
               goto err;
             new_table= TRUE;
           }
-
+        }
+        /*
+          We have to re-test if the table was a view as the view may not
+          have been opened until just above.
+        */
+        if (!table->view)
+        {
           int result __attribute__((unused))=
             store_create_info(thd, table, &query,
                               create_info, FALSE /* show_database */);
@@ -7726,9 +7840,7 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
     it is the case.
     TODO: this design is obsolete and will be removed.
   */
-  int table_kind= check_if_log_table(table_list->db_length, table_list->db,
-                                     table_list->table_name_length,
-                                     table_list->table_name, false);
+  int table_kind= check_if_log_table(table_list, FALSE, NullS);
 
   if (table_kind)
   {
diff --git a/sql/sql_table.h b/sql/sql_table.h
index c42f8aaa39e..c6ba7a581ec 100644
--- a/sql/sql_table.h
+++ b/sql/sql_table.h
@@ -187,11 +187,11 @@ bool mysql_create_table(THD *thd, TABLE_LIST *create_table,
 #define C_ALTER_TABLE_FRM_ONLY   -2
 #define C_ASSISTED_DISCOVERY     -3
 
-bool mysql_create_table_no_lock(THD *thd, const char *db,
-                                const char *table_name,
-                                HA_CREATE_INFO *create_info,
-                                Alter_info *alter_info, bool *is_trans,
-                                int create_table_mode);
+int mysql_create_table_no_lock(THD *thd, const char *db,
+                               const char *table_name,
+                               HA_CREATE_INFO *create_info,
+                               Alter_info *alter_info, bool *is_trans,
+                               int create_table_mode);
 
 handler *mysql_create_frm_image(THD *thd,
                                 const char *db, const char *table_name,
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index 7514b7bec63..f25d3be4d1f 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -1664,7 +1664,7 @@ bool my_yyoverflow(short **a, YYSTYPE **b, ulong *yystacksize);
 %type <num>
         type type_with_opt_collate int_type real_type order_dir lock_option
         udf_type opt_if_exists opt_local opt_table_options table_options
-        table_option opt_if_not_exists opt_no_write_to_binlog
+        table_option opt_if_not_exists create_or_replace opt_no_write_to_binlog
         opt_temporary all_or_any opt_distinct
         opt_ignore_leaves fulltext_options spatial_type union_option
         field_def
@@ -1843,7 +1843,6 @@ bool my_yyoverflow(short **a, YYSTYPE **b, ulong *yystacksize);
         statement sp_suid
         sp_c_chistics sp_a_chistics sp_chistic sp_c_chistic xa
         opt_field_or_var_spec fields_or_vars opt_load_data_set_spec
-        view_replace_or_algorithm view_replace
         view_algorithm view_or_trigger_or_sp_or_event
         definer_tail no_definer_tail
         view_suid view_tail view_list_opt view_list view_select
@@ -2341,25 +2340,29 @@ connection_name:
 /* create a table */
 
 create:
-          CREATE opt_table_options TABLE_SYM opt_if_not_exists table_ident
+          create_or_replace opt_table_options TABLE_SYM opt_if_not_exists table_ident
           {
             LEX *lex= thd->lex;
             lex->sql_command= SQLCOM_CREATE_TABLE;
+            if ($1 && $4)
+            {
+               my_error(ER_WRONG_USAGE, MYF(0), "OR REPLACE", "IF NOT EXISTS");
+               MYSQL_YYABORT;
+            }
             if (!lex->select_lex.add_table_to_list(thd, $5, NULL,
                                                    TL_OPTION_UPDATING,
                                                    TL_WRITE, MDL_EXCLUSIVE))
               MYSQL_YYABORT;
-            /*
-              For CREATE TABLE, an non-existing table is not an error.
-              Instruct open_tables() to just take an MDL lock if the
-              table does not exist.
-            */
-            lex->query_tables->open_strategy= TABLE_LIST::OPEN_IF_EXISTS;
             lex->alter_info.reset();
             lex->col_list.empty();
             lex->change=NullS;
             bzero((char*) &lex->create_info,sizeof(lex->create_info));
-            lex->create_info.options=$2 | $4;
+            /*
+              For CREATE TABLE we should not open the table even if it exists.
+              If the table exists, we should either not create it or replace it
+            */
+            lex->query_tables->open_strategy= TABLE_LIST::OPEN_STUB;
+            lex->create_info.options= ($1 | $2 | $4);
             lex->create_info.default_table_charset= NULL;
             lex->name.str= 0;
             lex->name.length= 0;
@@ -2428,14 +2431,22 @@ create:
             lex->name= $4;
             lex->create_info.options=$3;
           }
-        | CREATE
+        | create_or_replace
           {
-            Lex->create_view_mode= VIEW_CREATE_NEW;
+            Lex->create_view_mode= ($1 == 0 ? VIEW_CREATE_NEW :
+                                    VIEW_CREATE_OR_REPLACE);
             Lex->create_view_algorithm= DTYPE_ALGORITHM_UNDEFINED;
             Lex->create_view_suid= TRUE;
           }
           view_or_trigger_or_sp_or_event
-          {}
+          {
+            if ($1 && Lex->sql_command != SQLCOM_CREATE_VIEW)
+            {
+               my_error(ER_WRONG_USAGE, MYF(0), "OR REPLACE",
+                       "TRIGGERS / SP / EVENT");
+               MYSQL_YYABORT;
+            }
+          }
         | CREATE USER clear_privileges grant_list
           {
             Lex->sql_command = SQLCOM_CREATE_USER;
@@ -5515,6 +5526,17 @@ opt_if_not_exists:
           }
          ;
 
+create_or_replace:
+          CREATE /* empty */
+          {
+            $$= 0;
+          }
+        | CREATE OR_SYM REPLACE
+          {
+            $$= HA_LEX_CREATE_REPLACE;
+          }
+         ;
+
 opt_create_table_options:
           /* empty */
         | create_table_options
@@ -15799,7 +15821,7 @@ view_or_trigger_or_sp_or_event:
           {}
         | no_definer no_definer_tail
           {}
-        | view_replace_or_algorithm definer_opt view_tail
+        | view_algorithm definer_opt view_tail
           {}
         ;
 
@@ -15858,20 +15880,6 @@ definer:
 
 **************************************************************************/
 
-view_replace_or_algorithm:
-          view_replace
-          {}
-        | view_replace view_algorithm
-          {}
-        | view_algorithm
-          {}
-        ;
-
-view_replace:
-          OR_SYM REPLACE
-          { Lex->create_view_mode= VIEW_CREATE_OR_REPLACE; }
-        ;
-
 view_algorithm:
           ALGORITHM_SYM EQ UNDEFINED_SYM
           { Lex->create_view_algorithm= DTYPE_ALGORITHM_UNDEFINED; }
diff --git a/sql/table.h b/sql/table.h
index 0d0f8c49f21..565f4c5ddad 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -1951,7 +1951,7 @@ struct TABLE_LIST
      Indicates that if TABLE_LIST object corresponds to the table/view
      which requires special handling.
   */
-  enum
+  enum enum_open_strategy
   {
     /* Normal open. */
     OPEN_NORMAL= 0,
diff --git a/sql/table_cache.cc b/sql/table_cache.cc
index 6f51ac8276c..6b24f4348ee 100644
--- a/sql/table_cache.cc
+++ b/sql/table_cache.cc
@@ -196,6 +196,7 @@ static void check_unused(THD *thd)
   TABLE *entry;
   TABLE_SHARE *share;
   TDC_iterator tdc_it;
+  DBUG_ENTER("check_unused");
 
   tdc_it.init();
   mysql_mutex_lock(&LOCK_open);
@@ -221,6 +222,7 @@ static void check_unused(THD *thd)
   }
   mysql_mutex_unlock(&LOCK_open);
   tdc_it.deinit();
+  DBUG_VOID_RETURN;
 }
 #else
 #define check_unused(A)