MDEV-5619: CREATE OR REPLACE does not release MDL_EXCLUSIVE upon failure

mysql-test/r/create_or_replace.result:
  Added test of releasing of metadata locks
mysql-test/t/create_or_replace.test:
  Added test of releasing of metadata locks
sql/handler.h:
  Added marker if table was deleted as part of CREATE OR REPLACE
sql/sql_base.cc:
  Added Locked_tables_list::unlock_locked_table()
sql/sql_class.h:
  New prototypes
sql/sql_insert.cc:
  Unlock metadata locks for deleted table in case of error. Also do unlock tables if this was the only locked table.
sql/sql_table.cc:
  Unlock metadata locks for deleted table in case of error. Also do unlock tables if this was the only locked table.

mysql-test/r/create_or_replace.result

@@ -247,6 +247,69 @@ a	i
 3	3
 drop table t1,t3;
 #
+# Test the meta data locks are freed properly
+#
+create database mysqltest2;
+drop table if exists test.t1,mysqltest2.t2;
+Warnings:
+Note	1051	Unknown table 'test.t1'
+Note	1051	Unknown table 'mysqltest2.t2'
+create table test.t1 (i int);
+create table mysqltest2.t2 like test.t1;
+lock table test.t1 write, mysqltest2.t2 write;
+select * from information_schema.metadata_lock_info;
+THREAD_ID	LOCK_MODE	LOCK_DURATION	LOCK_TYPE	TABLE_SCHEMA	TABLE_NAME
+3	MDL_INTENTION_EXCLUSIVE	MDL_EXPLICIT	Global read lock		
+3	MDL_SHARED_NO_READ_WRITE	MDL_EXPLICIT	Table metadata lock	test	t1
+3	MDL_INTENTION_EXCLUSIVE	MDL_EXPLICIT	Schema metadata lock	mysqltest2	
+3	MDL_INTENTION_EXCLUSIVE	MDL_EXPLICIT	Schema metadata lock	test	
+3	MDL_SHARED_NO_READ_WRITE	MDL_EXPLICIT	Table metadata lock	mysqltest2	t2
+create or replace table test.t1;
+ERROR 42000: A table must have at least 1 column
+show tables;
+Tables_in_test
+t2
+select * from information_schema.metadata_lock_info;
+THREAD_ID	LOCK_MODE	LOCK_DURATION	LOCK_TYPE	TABLE_SCHEMA	TABLE_NAME
+3	MDL_INTENTION_EXCLUSIVE	MDL_EXPLICIT	Global read lock		
+3	MDL_INTENTION_EXCLUSIVE	MDL_EXPLICIT	Schema metadata lock	mysqltest2	
+3	MDL_INTENTION_EXCLUSIVE	MDL_EXPLICIT	Schema metadata lock	test	
+3	MDL_SHARED_NO_READ_WRITE	MDL_EXPLICIT	Table metadata lock	mysqltest2	t2
+create or replace table mysqltest2.t2;
+ERROR 42000: A table must have at least 1 column
+select * from information_schema.metadata_lock_info;
+THREAD_ID	LOCK_MODE	LOCK_DURATION	LOCK_TYPE	TABLE_SCHEMA	TABLE_NAME
+create table t1 (i int);
+drop table t1;
+create table test.t1 (i int);
+create table mysqltest2.t2 like test.t1;
+lock table test.t1 write, mysqltest2.t2 write;
+select * from information_schema.metadata_lock_info;
+THREAD_ID	LOCK_MODE	LOCK_DURATION	LOCK_TYPE	TABLE_SCHEMA	TABLE_NAME
+3	MDL_INTENTION_EXCLUSIVE	MDL_EXPLICIT	Global read lock		
+3	MDL_SHARED_NO_READ_WRITE	MDL_EXPLICIT	Table metadata lock	test	t1
+3	MDL_INTENTION_EXCLUSIVE	MDL_EXPLICIT	Schema metadata lock	mysqltest2	
+3	MDL_INTENTION_EXCLUSIVE	MDL_EXPLICIT	Schema metadata lock	test	
+3	MDL_SHARED_NO_READ_WRITE	MDL_EXPLICIT	Table metadata lock	mysqltest2	t2
+create or replace table test.t1 (a int) select 1 as 'a', 2 as 'a';
+ERROR 42S21: Duplicate column name 'a'
+show tables;
+Tables_in_test
+t2
+select * from information_schema.metadata_lock_info;
+THREAD_ID	LOCK_MODE	LOCK_DURATION	LOCK_TYPE	TABLE_SCHEMA	TABLE_NAME
+3	MDL_INTENTION_EXCLUSIVE	MDL_EXPLICIT	Global read lock		
+3	MDL_INTENTION_EXCLUSIVE	MDL_EXPLICIT	Schema metadata lock	mysqltest2	
+3	MDL_INTENTION_EXCLUSIVE	MDL_EXPLICIT	Schema metadata lock	test	
+3	MDL_SHARED_NO_READ_WRITE	MDL_EXPLICIT	Table metadata lock	mysqltest2	t2
+create or replace table mysqltest2.t2 (a int) select 1 as 'a', 2 as 'a';
+ERROR 42S21: Duplicate column name 'a'
+select * from information_schema.metadata_lock_info;
+THREAD_ID	LOCK_MODE	LOCK_DURATION	LOCK_TYPE	TABLE_SCHEMA	TABLE_NAME
+create table t1 (i int);
+drop table t1;
+drop database mysqltest2;
+#
 # Testing CREATE .. LIKE
 #
 create or replace table t1 like t2;

mysql-test/t/create_or_replace.test

@@ -202,6 +202,42 @@ unlock tables;
 select * from t1 order by a,i;
 drop table t1,t3;
 
+--echo #
+--echo # Test the meta data locks are freed properly
+--echo #
+
+create database mysqltest2;
+
+drop table if exists test.t1,mysqltest2.t2;
+create table test.t1 (i int);
+create table mysqltest2.t2 like test.t1;
+lock table test.t1 write, mysqltest2.t2 write;
+select * from information_schema.metadata_lock_info;
+--error ER_TABLE_MUST_HAVE_COLUMNS
+create or replace table test.t1;
+show tables;
+select * from information_schema.metadata_lock_info;
+--error ER_TABLE_MUST_HAVE_COLUMNS
+create or replace table mysqltest2.t2;
+select * from information_schema.metadata_lock_info;
+create table t1 (i int);
+drop table t1;
+
+create table test.t1 (i int);
+create table mysqltest2.t2 like test.t1;
+lock table test.t1 write, mysqltest2.t2 write;
+select * from information_schema.metadata_lock_info;
+--error ER_DUP_FIELDNAME
+create or replace table test.t1 (a int) select 1 as 'a', 2 as 'a';
+show tables;
+select * from information_schema.metadata_lock_info;
+--error ER_DUP_FIELDNAME
+create or replace table mysqltest2.t2 (a int) select 1 as 'a', 2 as 'a';
+select * from information_schema.metadata_lock_info;
+create table t1 (i int);
+drop table t1;
+drop database mysqltest2;
+
 --echo #
 --echo # Testing CREATE .. LIKE
 --echo #

sql/handler.h

@@ -1621,6 +1621,7 @@ struct HA_CREATE_INFO
   TABLE *table;
   TABLE_LIST *pos_in_locked_tables;
   MDL_ticket *mdl_ticket;
+  bool table_was_deleted;
 
   bool tmp_table() { return options & HA_LEX_CREATE_TMP_TABLE; }
 };

sql/sql_base.cc

@@ -2722,6 +2722,38 @@ Locked_tables_list::unlock_locked_tables(THD *thd)
   reset();
 }
 
+
+/**
+  Remove all meta data locks associated with table and release locked
+  table mode if there is no locked tables anymore
+*/
+
+void
+Locked_tables_list::unlock_locked_table(THD *thd, MDL_ticket *mdl_ticket)
+{
+  /*
+    Ensure we are in locked table mode.
+    As this function is only called on error condition it's better
+    to check this condition here than in the caller.
+  */
+  if (thd->locked_tables_mode != LTM_LOCK_TABLES)
+    return;
+
+  if (mdl_ticket)
+  {
+    /*
+      Under LOCK TABLES we may have several instances of table open
+      and locked and therefore have to remove several metadata lock
+      requests associated with them.
+    */
+    thd->mdl_context.release_all_locks_for_name(mdl_ticket);
+  }
+
+  if (thd->lock->table_count == 0)
+    unlock_locked_tables(thd);
+}
+
+
 /*
   Free memory allocated for storing locks
 */

sql/sql_class.h

@@ -1518,6 +1518,7 @@ public:
                    MYF(MY_THREAD_SPECIFIC));
   }
   void unlock_locked_tables(THD *thd);
+  void unlock_locked_table(THD *thd, MDL_ticket *mdl_ticket);
   ~Locked_tables_list()
   {
     reset();

sql/sql_insert.cc

@@ -4343,7 +4343,7 @@ void select_create::abort_result_set()
     of the table succeeded or not, since we need to reset the binary
     log state.
     
-    However if there was an orignal table that was deleted, as part of
+    However if there was an original table that was deleted, as part of
     create or replace table, then we must log the statement.
   */
 
@@ -4357,6 +4357,12 @@ void select_create::abort_result_set()
   /* possible error of writing binary log is ignored deliberately */
   (void) thd->binlog_flush_pending_rows_event(TRUE, TRUE);
 
+  if (create_info->table_was_deleted)
+  {
+    /* Unlock locked table that was dropped by CREATE */
+    thd->locked_tables_list.unlock_locked_table(thd,
+                                                create_info->mdl_ticket);
+  }
   if (m_plock)
   {
     mysql_unlock_tables(thd, *m_plock);

sql/sql_table.cc

@@ -4627,6 +4627,7 @@ int create_table_impl(THD *thd,
         */
         thd->variables.option_bits|= OPTION_KEEP_LOG;
         thd->log_current_statement= 1;
+        create_info->table_was_deleted= 1;
 
         /*
           The test of query_tables is to ensure we have any tables in the
@@ -4838,19 +4839,23 @@ bool mysql_create_table(THD *thd, TABLE_LIST *create_table,
   bool result= 0;
   int create_table_mode;
   TABLE_LIST *pos_in_locked_tables= 0;
+  MDL_ticket *mdl_ticket= 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))
+  if (open_and_lock_tables(thd, create_table, 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))
+  if ((create_info->table= create_table->table))
+  {
     pos_in_locked_tables= create_info->table->pos_in_locked_tables;
+    mdl_ticket= create_table->table->mdl_ticket;
+  }
   
   /* Got lock. */
   DEBUG_SYNC(thd, "locked_table_name");
@@ -4895,9 +4900,19 @@ err:
     DBUG_RETURN(result);
   /* Write log if no error or if we already deleted a table */
   if (!result || thd->log_current_statement)
+  {
+    if (result && create_info->table_was_deleted)
+    {
+      /*
+        Possible locked table was dropped. We should remove meta data locks
+        associated with it and do UNLOCK_TABLES if no more locked tables.
+      */
+      thd->locked_tables_list.unlock_locked_table(thd, mdl_ticket);
+    }    
     if (write_bin_log(thd, result ? FALSE : TRUE, thd->query(),
                       thd->query_length(), is_trans))
       result= 1;
+  }
   DBUG_RETURN(result);
 }