Fixes for CREATE_OR_REPLACE

- MDEV-5587 Server crashes in Locked_tables_list::restore_lock on CREATE OR REPLACE .. SELECT under LOCK
- MDEV-5586 Assertion `share->tdc.all_tables.is_empty() || remove_type != TDC_RT_REMOVE_ALL' fails in tdc_remove_table
- MDEV-5588 Strange error on CREATE OR REPLACE table over an existing view

mysql-test/r/create_or_replace.result:
  Added test cases
mysql-test/r/lowercase_view.result:
  New error message
mysql-test/r/merge.result:
  New error message
mysql-test/r/multi_update.result:
  New error message
mysql-test/r/subselect.result:
  New error message
mysql-test/r/subselect_exists_to_in.result:
  New error message
mysql-test/r/subselect_no_mat.result:
  New error message
mysql-test/r/subselect_no_opts.result:
  New error message
mysql-test/r/subselect_no_scache.result:
  New error message
mysql-test/r/subselect_no_semijoin.result:
  New error message
mysql-test/r/view.result:
  New error message
mysql-test/suite/funcs_1/r/myisam_views-big.result:
  New error message
mysql-test/t/create_or_replace.test:
  New tests
mysql-test/t/view.test:
  New error message
sql/share/errmsg-utf8.txt:
  Added new error message
sql/sql_base.cc:
  Updated error message
  Do an automatic UNLOCK TABLES if we don't have any locked tables (safety fix)
sql/sql_db.cc:
  Updated arguments
sql/sql_load.cc:
  New error message
sql/sql_parse.cc:
  Check that we are not using a table we are dropping and re-creating
sql/sql_table.cc:
  Added parameter to mysql_rm_table_no_locks() to not automaticly do UNLOCK TABLES
  Added better error message if trying to drop a view with DROP TABLE
  Don't try to create something we select from
sql/sql_table.h:
  Updated prototypes

7 files changed, 64 insertions, 19 deletions

diff --git a/sql/share/errmsg-utf8.txt b/sql/share/errmsg-utf8.txt
index 2d061fc314c..cc231599037 100644
--- a/sql/share/errmsg-utf8.txt
+++ b/sql/share/errmsg-utf8.txt
@@ -2140,13 +2140,9 @@ ER_INSERT_INFO
         spa "Registros: %ld  Duplicados: %ld  Peligros: %ld"
         swe "Rader: %ld  Dubletter: %ld  Varningar: %ld"
         ukr "Записів: %ld  Дублікатів: %ld  Застережень: %ld"
-ER_UPDATE_TABLE_USED  
-        eng "You can't specify target table '%-.192s' for update in FROM clause"
-        ger "Die Verwendung der zu aktualisierenden Zieltabelle '%-.192s' ist in der FROM-Klausel nicht zulässig."
-        jpn "FROM句にある表 '%-.192s' はUPDATEの対象にできません。"
-        rus "Не допускается указание таблицы '%-.192s' в списке таблиц FROM для внесения в нее изменений"
-        swe "INSERT-table '%-.192s' får inte finnas i FROM tabell-listan"
-        ukr "Таблиця '%-.192s' що змінюється не дозволена у переліку таблиць FROM"
+ER_UPDATE_TABLE_USED
+        eng "Table '%-.192s' is specified twice, both as a target for '%s' and as a separate source for data"
+        swe "Table '%-.192s' är använd två gånger. Både för '%s' och för att hämta data"
 ER_NO_SUCH_THREAD  
         cze "Neznámá identifikace threadu: %lu"
         dan "Ukendt tråd id: %lu"
@@ -7089,3 +7085,5 @@ ER_CHANGE_SLAVE_PARALLEL_THREADS_ACTIVE
 	eng "Cannot change @@slave_parallel_threads while another change is in progress"
 ER_PRIOR_COMMIT_FAILED
 	eng "Commit failed due to failure of an earlier commit on which this one depends"
+ER_IT_IS_A_VIEW 42S02
+        eng "'%-.192s' is a view"
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index 0a38cbbeb1a..e5105b03614 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -1483,7 +1483,7 @@ void update_non_unique_table_error(TABLE_LIST *update,
       return;
     }
   }
-  my_error(ER_UPDATE_TABLE_USED, MYF(0), update->alias);
+  my_error(ER_UPDATE_TABLE_USED, MYF(0), update->alias, operation);
 }
 
 
@@ -2839,6 +2839,10 @@ unlink_all_closed_tables(THD *thd, MYSQL_LOCK *lock, size_t reopen_count)
       m_locked_tables_count--;
     }
   }
+
+  /* If no tables left, do an automatic UNLOCK TABLES */
+  if (thd->lock && thd->lock->table_count == 0)
+    unlock_locked_tables(thd);
 }
 
 
diff --git a/sql/sql_db.cc b/sql/sql_db.cc
index 85e9b1aa852..fcde2b4acbd 100644
--- a/sql/sql_db.cc
+++ b/sql/sql_db.cc
@@ -829,7 +829,7 @@ bool mysql_rm_db(THD *thd,char *db,bool if_exists, bool silent)
   thd->push_internal_handler(&err_handler);
   if (!thd->killed &&
       !(tables &&
-        mysql_rm_table_no_locks(thd, tables, true, false, true, true)))
+        mysql_rm_table_no_locks(thd, tables, true, false, true, true, false)))
   {
     /*
       We temporarily disable the binary log while dropping the objects
diff --git a/sql/sql_load.cc b/sql/sql_load.cc
index 281d1de7877..a0959bdd278 100644
--- a/sql/sql_load.cc
+++ b/sql/sql_load.cc
@@ -258,7 +258,8 @@ int mysql_load(THD *thd,sql_exchange *ex,TABLE_LIST *table_list,
   */
   if (unique_table(thd, table_list, table_list->next_global, 0))
   {
-    my_error(ER_UPDATE_TABLE_USED, MYF(0), table_list->table_name);
+    my_error(ER_UPDATE_TABLE_USED, MYF(0), table_list->table_name,
+             "LOAD DATA");
     DBUG_RETURN(TRUE);
   }
 
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 706e9e1a3a6..0d0f1a2634f 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -2957,8 +2957,24 @@ case SQLCOM_PREPARE:
         /* Got error or warning. Set res to 1 if error */
         if (!(res= thd->is_error()))
           my_ok(thd);                           // CREATE ... IF NOT EXISTS
+        goto end_with_restore_list;
+      }
+
+      /* Ensure we don't try to create something from which we select from */
+      if ((create_info.options & HA_LEX_CREATE_REPLACE) &&
+          !create_info.tmp_table())
+      {
+        TABLE_LIST *duplicate;
+        if ((duplicate= unique_table(thd, lex->query_tables,
+                                     lex->query_tables->next_global,
+                                     0)))
+        {
+          update_non_unique_table_error(lex->query_tables, "CREATE",
+                                        duplicate);
+          res= TRUE;
+          goto end_with_restore_list;
+        }
       }
-      else
       {
         /*
           Remove target table from main select and name resolution
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 32e4fdc18a1..817823063eb 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -2103,7 +2103,7 @@ bool mysql_rm_table(THD *thd,TABLE_LIST *tables, my_bool if_exists,
   /* mark for close and remove all cached entries */
   thd->push_internal_handler(&err_handler);
   error= mysql_rm_table_no_locks(thd, tables, if_exists, drop_temporary,
-                                 false, false);
+                                 false, false, false);
   thd->pop_internal_handler();
 
   if (error)
@@ -2168,6 +2168,8 @@ static uint32 comment_length(THD *thd, uint32 comment_pos,
   @param  drop_view       Allow to delete VIEW .frm
   @param  dont_log_query  Don't write query to log files. This will also not
                           generate warnings if the handler files doesn't exists
+  @param  dont_free_locks Don't do automatic UNLOCK TABLE if no more locked
+                          tables
 
   @retval  0  ok
   @retval  1  Error
@@ -2190,7 +2192,8 @@ static uint32 comment_length(THD *thd, uint32 comment_pos,
 
 int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
                             bool drop_temporary, bool drop_view,
-                            bool dont_log_query)
+                            bool dont_log_query,
+                            bool dont_free_locks)
 {
   TABLE_LIST *table;
   char path[FN_REFLEN + 1], wrong_tables_buff[160], *alias= NULL;
@@ -2204,6 +2207,8 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
   bool trans_tmp_table_deleted= 0, non_trans_tmp_table_deleted= 0;
   bool non_tmp_table_deleted= 0;
   bool is_drop_tmp_if_exists_added= 0;
+  bool one_table= tables->next_local == 0;
+  bool was_view= 0;
   String built_query;
   String built_trans_tmp_query, built_non_trans_tmp_query;
   DBUG_ENTER("mysql_rm_table_no_locks");
@@ -2413,7 +2418,7 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
     DEBUG_SYNC(thd, "rm_table_no_locks_before_delete_table");
     error= 0;
     if ((drop_temporary || !ha_table_exists(thd, db, alias, &table_type) ||
-         (!drop_view && table_type == view_pseudo_hton)))
+         (!drop_view && (was_view= (table_type == view_pseudo_hton)))))
     {
       /*
         One of the following cases happened:
@@ -2544,7 +2549,10 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
 err:
   if (wrong_tables.length())
   {
-    if (!foreign_key_error)
+    if (one_table && was_view)
+      my_printf_error(ER_IT_IS_A_VIEW, ER(ER_IT_IS_A_VIEW), MYF(0),
+                      wrong_tables.c_ptr_safe());
+    else if (!foreign_key_error)
       my_printf_error(ER_BAD_TABLE_ERROR, ER(ER_BAD_TABLE_ERROR), MYF(0),
                       wrong_tables.c_ptr_safe());
     else
@@ -2610,7 +2618,8 @@ err:
     */
     if (thd->locked_tables_mode)
     {
-      if (thd->lock && thd->lock->table_count == 0 && non_temp_tables_count > 0)
+      if (thd->lock && thd->lock->table_count == 0 &&
+          non_temp_tables_count > 0 && !dont_free_locks)
       {
         thd->locked_tables_list.unlock_locked_tables(thd);
         goto end;
@@ -4608,8 +4617,8 @@ int create_table_impl(THD *thd,
           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))
+        /* Remove normal table without logging. Keep tables locked */
+        if (mysql_rm_table_no_locks(thd, &table_list, 0, 0, 0, 1, 1))
           goto err;
         /*
           The test of query_tables is to ensure we have any tables in the
@@ -4618,6 +4627,11 @@ int create_table_impl(THD *thd,
         if (thd->lex->query_tables &&
             restart_trans_for_tables(thd, thd->lex->query_tables->next_global))
           goto err;
+        /*
+          We have to log this query, even if it failed later to ensure the
+          drop is done.
+        */
+        thd->variables.option_bits|= OPTION_KEEP_LOG;
       }
       else if (create_info->options & HA_LEX_CREATE_IF_NOT_EXISTS)
         goto warn;
@@ -5084,6 +5098,18 @@ bool mysql_create_like_table(THD* thd, TABLE_LIST* table,
     res= thd->is_error();
     goto err;
   }
+  /* Ensure we don't try to create something from which we select from */
+  if ((create_info->options & HA_LEX_CREATE_REPLACE) &&
+      !create_info->tmp_table())
+  {
+    TABLE_LIST *duplicate;
+    if ((duplicate= unique_table(thd, table, src_table, 0)))
+    {
+      update_non_unique_table_error(src_table, "CREATE", duplicate);
+      goto err;
+    }
+  }
+
   src_table->table->use_all_columns();
 
   DEBUG_SYNC(thd, "create_table_like_after_open");
diff --git a/sql/sql_table.h b/sql/sql_table.h
index c6ba7a581ec..fc7eb775bef 100644
--- a/sql/sql_table.h
+++ b/sql/sql_table.h
@@ -238,7 +238,7 @@ bool mysql_rm_table(THD *thd,TABLE_LIST *tables, my_bool if_exists,
                     my_bool drop_temporary);
 int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
                             bool drop_temporary, bool drop_view,
-                            bool log_query);
+                            bool log_query, bool dont_free_locks);
 bool quick_rm_table(THD *thd, handlerton *base, const char *db,
                     const char *table_name, uint flags);
 void close_cached_table(THD *thd, TABLE *table);