MDEV-5602: CREATE OR REPLACE obtains stricter locks than the connection had before

mysql-test/r/create_or_replace.result:
  Added test case
mysql-test/t/create_or_replace.test:
  Added test case
sql/sql_base.cc:
  Downgrade locks in LOCK TABLES + CREATE TABLE OR REPLACE
sql/sql_table.cc:
  Downgrade locks in LOCK TABLES + CREATE TABLE OR REPLACE

mysql-test/include/have_metadata_lock_info.inc

+if (!`SELECT count(*) FROM information_schema.plugins WHERE
+      (PLUGIN_STATUS = 'ACTIVE') AND PLUGIN_NAME = 'METADATA_LOCK_INFO'`){
+  skip Need archive METADATA_LOCK_INFO plugin;
+}

mysql-test/include/have_metadata_lock_info.opt

+--loose-metadata-lock-info
+--plugin-load-add=$METADATA_LOCK_INFO_SO

mysql-test/r/create_or_replace.result

@@ -324,4 +324,39 @@ ERROR 42S02: 'test.t1' is a view
 drop table t1;
 ERROR 42S02: 'test.t1' is a view
 drop view t1;
+#
+# MDEV-5602 CREATE OR REPLACE obtains stricter locks than the
+# connection had before
+#
+create table t1 (a int);
+lock table t1 write, t2 read;
+select * from information_schema.metadata_lock_info;
+THREAD_ID	LOCK_MODE	LOCK_DURATION	LOCK_TYPE	TABLE_SCHEMA	TABLE_NAME
+#	MDL_INTENTION_EXCLUSIVE	MDL_EXPLICIT	Global read lock		
+#	MDL_INTENTION_EXCLUSIVE	MDL_EXPLICIT	Schema metadata lock	test	
+#	MDL_SHARED_NO_READ_WRITE	MDL_EXPLICIT	Table metadata lock	test	t1
+#	MDL_SHARED_READ	MDL_EXPLICIT	Table metadata lock	test	t2
+create or replace table t1 (i int);
+select * from information_schema.metadata_lock_info;
+THREAD_ID	LOCK_MODE	LOCK_DURATION	LOCK_TYPE	TABLE_SCHEMA	TABLE_NAME
+#	MDL_INTENTION_EXCLUSIVE	MDL_EXPLICIT	Global read lock		
+#	MDL_INTENTION_EXCLUSIVE	MDL_EXPLICIT	Schema metadata lock	test	
+#	MDL_SHARED_NO_READ_WRITE	MDL_EXPLICIT	Table metadata lock	test	t1
+#	MDL_SHARED_READ	MDL_EXPLICIT	Table metadata lock	test	t2
+create or replace table t1 like t2;
+select * from information_schema.metadata_lock_info;
+THREAD_ID	LOCK_MODE	LOCK_DURATION	LOCK_TYPE	TABLE_SCHEMA	TABLE_NAME
+#	MDL_INTENTION_EXCLUSIVE	MDL_EXPLICIT	Global read lock		
+#	MDL_INTENTION_EXCLUSIVE	MDL_EXPLICIT	Schema metadata lock	test	
+#	MDL_SHARED_NO_READ_WRITE	MDL_EXPLICIT	Table metadata lock	test	t1
+#	MDL_SHARED_READ	MDL_EXPLICIT	Table metadata lock	test	t2
+create or replace table t1 select 1 as f1;
+select * from information_schema.metadata_lock_info;
+THREAD_ID	LOCK_MODE	LOCK_DURATION	LOCK_TYPE	TABLE_SCHEMA	TABLE_NAME
+#	MDL_INTENTION_EXCLUSIVE	MDL_EXPLICIT	Global read lock		
+#	MDL_INTENTION_EXCLUSIVE	MDL_EXPLICIT	Schema metadata lock	test	
+#	MDL_SHARED_NO_READ_WRITE	MDL_EXPLICIT	Table metadata lock	test	t1
+#	MDL_SHARED_READ	MDL_EXPLICIT	Table metadata lock	test	t2
+drop table t1;
+unlock tables;
 DROP TABLE t2;

mysql-test/t/create_or_replace.test

@@ -3,6 +3,7 @@
 #
 
 --source include/have_innodb.inc
+--source include/have_metadata_lock_info.inc
 --disable_warnings
 drop table if exists t1,t2,t3;
 --enable_warnings
@@ -265,6 +266,27 @@ create or replace table t1 (a int);
 drop table t1;
 drop view t1;
 
+--echo #
+--echo # MDEV-5602 CREATE OR REPLACE obtains stricter locks than the
+--echo # connection had before
+--echo #
+
+create table t1 (a int);
+lock table t1 write, t2 read;
+--replace_column 1 #
+select * from information_schema.metadata_lock_info;
+create or replace table t1 (i int);
+--replace_column 1 #
+select * from information_schema.metadata_lock_info;
+create or replace table t1 like t2;
+--replace_column 1 #
+select * from information_schema.metadata_lock_info;
+create or replace table t1 select 1 as f1;
+--replace_column 1 #
+select * from information_schema.metadata_lock_info;
+drop table t1;
+unlock tables;
+
 #
 # Cleanup
 #

sql/sql_base.cc

@@ -2948,6 +2948,11 @@ bool Locked_tables_list::restore_lock(THD *thd, TABLE_LIST *dst_table_list,
 
   add_back_last_deleted_lock(dst_table_list);
 
+  table->mdl_ticket->downgrade_lock(table->reginfo.lock_type >=
+                                    TL_WRITE_ALLOW_WRITE ? 
+                                    MDL_SHARED_NO_READ_WRITE :
+                                    MDL_SHARED_READ);
+
   DBUG_RETURN(0);
 }
 

sql/sql_table.cc

@@ -4882,6 +4882,11 @@ bool mysql_create_table(THD *thd, TABLE_LIST *create_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
+    {
+      TABLE *table= pos_in_locked_tables->table;
+      table->mdl_ticket->downgrade_lock(MDL_SHARED_NO_READ_WRITE);
+    }
   }
 
 err:
@@ -5195,17 +5200,20 @@ bool mysql_create_like_table(THD* thd, TABLE_LIST* table,
         don't reopen the table when doing statment logging below.
       */
       table->table= pos_in_locked_tables->table;
+      table->table->mdl_ticket->downgrade_lock(MDL_SHARED_NO_READ_WRITE);
     }
   }
-
-  /*
-    Ensure that we have an exclusive lock on target table if we are creating
-    non-temporary table.
-  */
-  DBUG_ASSERT((create_info->tmp_table()) ||
-              thd->mdl_context.is_lock_owner(MDL_key::TABLE, table->db,
-                                             table->table_name,
-                                             MDL_EXCLUSIVE));
+  else
+  {
+    /*
+      Ensure that we have an exclusive lock on target table if we are creating
+      non-temporary table.
+    */
+    DBUG_ASSERT((create_info->tmp_table()) ||
+                thd->mdl_context.is_lock_owner(MDL_key::TABLE, table->db,
+                                               table->table_name,
+                                               MDL_EXCLUSIVE));
+  }
 
   DEBUG_SYNC(thd, "create_table_like_before_binlog");