MDEV-4441 DROP DATABASE with a newly created ARCHIVE table does not work

1. DROP DATABASE should use ha_discover_table_names(), not look at .frm files.
2. filename_to_tablename() also encodes temp file names #sql- -> #mysql50##sql
3. no special treatment for #sql- files, no TABLE_LIST::internal_tmp_table
4. discover also table file names, that start from #

7 files changed, 66 insertions, 112 deletions

diff --git a/sql/discover.cc b/sql/discover.cc
index 4224e8ce0b0..cc0dece031a 100644
--- a/sql/discover.cc
+++ b/sql/discover.cc
@@ -198,10 +198,10 @@ int extension_based_table_discovery(MY_DIR *dirp, const char *ext_meta,
   end= cur + dirp->number_of_files;
   while (cur < end)
   {
-    char *octothorp= strrchr(cur->name, '#');
+    char *octothorp= strrchr(cur->name + 1, '#');
     char *ext= strchr(octothorp ? octothorp : cur->name, FN_EXTCHAR);
 
-    if (ext && octothorp != cur->name)
+    if (ext)
     {
       size_t len= (octothorp ? octothorp : ext) - cur->name;
       if (from != cur &&
@@ -239,13 +239,6 @@ int extension_based_table_discovery(MY_DIR *dirp, const char *ext_meta,
   simplified version of extension_based_table_discovery(), that does not
   modify the list of files. It cannot be called many times for the same
   directory listing, otherwise it'll produce duplicate results.
-
-  @note
-  For backward compatibility reasons, this will find tables with names,
-  starting from '#', as long as they don't start from '#sql-'.
-  These names are invalid since 5.0, and the compex discovery function
-  will ignore them. Anyone still having these files, should disable
-  discovering engines, and rename these invalid table files.
 */
 int ext_table_discovery_simple(MY_DIR *dirp,
                                handlerton::discovered_list *result)
@@ -259,7 +252,7 @@ int ext_table_discovery_simple(MY_DIR *dirp,
   {
     char *ext= strrchr(cur->name, FN_EXTCHAR);
 
-    if (ext && !is_prefix(cur->name, tmp_file_prefix))
+    if (ext)
     {
       if (my_strnncoll(cs, (uchar*)ext, strlen(ext),
                            (uchar*)reg_ext, reg_ext_length) == 0)
diff --git a/sql/handler.cc b/sql/handler.cc
index 660697cd74b..7a6edc967ba 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -4634,12 +4634,12 @@ static my_bool discover_names(THD *thd, plugin_ref plugin,
 }
 
 int ha_discover_table_names(THD *thd, LEX_STRING *db, MY_DIR *dirp,
-                            Discovered_table_list *result)
+                            Discovered_table_list *result, bool reusable)
 {
   int error;
   DBUG_ENTER("ha_discover_table_names");
 
-  if (engines_with_discover_table_names == 0)
+  if (engines_with_discover_table_names == 0 && !reusable)
   {
     error= ext_table_discovery_simple(dirp, result);
     result->sort();
diff --git a/sql/handler.h b/sql/handler.h
index c6a044085b0..bfc83a52582 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -3163,7 +3163,7 @@ public:
 
 int ha_discover_table(THD *thd, TABLE_SHARE *share);
 int ha_discover_table_names(THD *thd, LEX_STRING *db, MY_DIR *dirp,
-                            Discovered_table_list *result);
+                            Discovered_table_list *result, bool reusable);
 bool ha_table_exists(THD *thd, const char *db, const char *table_name,
                      handlerton **hton= 0);
 #endif
diff --git a/sql/sql_db.cc b/sql/sql_db.cc
index d4f8431beec..bc9247ad8e3 100644
--- a/sql/sql_db.cc
+++ b/sql/sql_db.cc
@@ -48,13 +48,12 @@
 
 #define MAX_DROP_TABLE_Q_LEN      1024
 
-const char *del_exts[]= {".frm", ".BAK", ".TMD",".opt", NullS};
+const char *del_exts[]= {".BAK", ".TMD",".opt", NullS};
 static TYPELIB deletable_extentions=
 {array_elements(del_exts)-1,"del_exts", del_exts, NULL};
 
-static bool find_db_tables_and_rm_known_files(THD *, MY_DIR *, const char *,
-                                              const char *, TABLE_LIST **,
-                                              bool *);
+static bool find_db_tables_and_rm_known_files(THD *, MY_DIR *, char *,
+                                              const char *, TABLE_LIST **);
 
 long mysql_rm_arc_files(THD *thd, MY_DIR *dirp, const char *org_path);
 static my_bool rm_dir_w_symlink(const char *org_path, my_bool send_error);
@@ -757,7 +756,6 @@ bool mysql_rm_db(THD *thd,char *db,bool if_exists, bool silent)
   char	path[FN_REFLEN + 16];
   MY_DIR *dirp;
   uint length;
-  bool found_other_files= false;
   TABLE_LIST *tables= NULL;
   TABLE_LIST *table;
   Drop_table_error_handler err_handler;
@@ -789,8 +787,7 @@ bool mysql_rm_db(THD *thd,char *db,bool if_exists, bool silent)
     }
   }
 
-  if (find_db_tables_and_rm_known_files(thd, dirp, db, path, &tables,
-                                        &found_other_files))
+  if (find_db_tables_and_rm_known_files(thd, dirp, db, path, &tables))
     goto exit;
 
   /*
@@ -871,10 +868,7 @@ bool mysql_rm_db(THD *thd,char *db,bool if_exists, bool silent)
       If the directory is a symbolic link, remove the link first, then
       remove the directory the symbolic link pointed at
     */
-    if (found_other_files)
-      my_error(ER_DB_DROP_RMDIR, MYF(0), path, EEXIST);
-    else
-      error= rm_dir_w_symlink(path, true);
+    error= rm_dir_w_symlink(path, true);
   }
   thd->pop_internal_handler();
 
@@ -985,18 +979,58 @@ exit:
 
 
 static bool find_db_tables_and_rm_known_files(THD *thd, MY_DIR *dirp,
-                                              const char *db,
+                                              char *dbname,
                                               const char *path,
-                                              TABLE_LIST **tables,
-                                              bool *found_other_files)
+                                              TABLE_LIST **tables)
 {
   char filePath[FN_REFLEN];
+  LEX_STRING db= { dbname, strlen(dbname) };
   TABLE_LIST *tot_list=0, **tot_list_next_local, **tot_list_next_global;
   DBUG_ENTER("find_db_tables_and_rm_known_files");
   DBUG_PRINT("enter",("path: %s", path));
 
+  /* first, get the list of tables */
+  Dynamic_array<LEX_STRING*> files(dirp->number_of_files);
+  Discovered_table_list tl(thd, &files, &null_lex_str);
+  if (ha_discover_table_names(thd, &db, dirp, &tl, true))
+    DBUG_RETURN(1);
+
+  /* Now put the tables in the list */
   tot_list_next_local= tot_list_next_global= &tot_list;
 
+  for (size_t idx=0; idx < files.elements(); idx++)
+  {
+    LEX_STRING *table= files.at(idx);
+
+    /* Drop the table nicely */
+    TABLE_LIST *table_list=(TABLE_LIST*)thd->calloc(sizeof(*table_list));
+
+    if (!table_list)
+      DBUG_RETURN(true);
+    table_list->db= db.str;
+    table_list->db_length= db.length;
+    table_list->table_name= table->str;
+    table_list->table_name_length= table->length;
+    table_list->open_type= OT_BASE_ONLY;
+
+    /* To be able to correctly look up the table in the table cache. */
+    if (lower_case_table_names)
+      table_list->table_name_length= my_casedn_str(files_charset_info,
+                                                   table_list->table_name);
+
+    table_list->alias= table_list->table_name;	// If lower_case_table_names=2
+    table_list->mdl_request.init(MDL_key::TABLE, table_list->db,
+                                 table_list->table_name, MDL_EXCLUSIVE,
+                                 MDL_TRANSACTION);
+    /* Link into list */
+    (*tot_list_next_local)= table_list;
+    (*tot_list_next_global)= table_list;
+    tot_list_next_local= &table_list->next_local;
+    tot_list_next_global= &table_list->next_global;
+  }
+  *tables= tot_list;
+
+  /* and at last delete all non-table files */
   for (uint idx=0 ;
        idx < (uint) dirp->number_of_files && !thd->killed ;
        idx++)
@@ -1021,59 +1055,12 @@ static bool find_db_tables_and_rm_known_files(THD *thd, MY_DIR *dirp,
 	DBUG_PRINT("my",("Archive subdir found: %s", newpath));
 	if ((mysql_rm_arc_files(thd, new_dirp, newpath)) < 0)
 	  DBUG_RETURN(true);
-	continue;
       }
-      *found_other_files= true;
       continue;
     }
     if (!(extension= strrchr(file->name, '.')))
       extension= strend(file->name);
-    if (find_type(extension, &deletable_extentions, FIND_TYPE_NO_PREFIX) <= 0)
-    {
-      if (find_type(extension, ha_known_exts(), FIND_TYPE_NO_PREFIX) <= 0)
-	*found_other_files= true;
-      continue;
-    }
-    /* just for safety we use files_charset_info */
-    if (db && !my_strcasecmp(files_charset_info,
-                             extension, reg_ext))
-    {
-      /* Drop the table nicely */
-      *extension= 0;			// Remove extension
-      TABLE_LIST *table_list=(TABLE_LIST*)
-                              thd->calloc(sizeof(*table_list) + 
-                                          strlen(db) + 1 +
-                                          MYSQL50_TABLE_NAME_PREFIX_LENGTH + 
-                                          strlen(file->name) + 1);
-
-      if (!table_list)
-        DBUG_RETURN(true);
-      table_list->db= (char*) (table_list+1);
-      table_list->db_length= strmov(table_list->db, db) - table_list->db;
-      table_list->table_name= table_list->db + table_list->db_length + 1;
-      table_list->table_name_length= filename_to_tablename(file->name,
-                                       table_list->table_name,
-                                       MYSQL50_TABLE_NAME_PREFIX_LENGTH +
-                                       strlen(file->name) + 1);
-      table_list->open_type= OT_BASE_ONLY;
-
-      /* To be able to correctly look up the table in the table cache. */
-      if (lower_case_table_names)
-        table_list->table_name_length= my_casedn_str(files_charset_info,
-                                                     table_list->table_name);
-
-      table_list->alias= table_list->table_name;	// If lower_case_table_names=2
-      table_list->internal_tmp_table= is_prefix(file->name, tmp_file_prefix);
-      table_list->mdl_request.init(MDL_key::TABLE, table_list->db,
-                                   table_list->table_name, MDL_EXCLUSIVE,
-                                   MDL_TRANSACTION);
-      /* Link into list */
-      (*tot_list_next_local)= table_list;
-      (*tot_list_next_global)= table_list;
-      tot_list_next_local= &table_list->next_local;
-      tot_list_next_global= &table_list->next_global;
-    }
-    else
+    if (find_type(extension, &deletable_extentions, FIND_TYPE_NO_PREFIX) > 0)
     {
       strxmov(filePath, path, "/", file->name, NullS);
       /*
@@ -1088,7 +1075,7 @@ static bool find_db_tables_and_rm_known_files(THD *thd, MY_DIR *dirp,
       }
     }
   }
-  *tables= tot_list;
+
   DBUG_RETURN(false);
 }
 
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index ae0360b7ed0..b7b2574c31d 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -831,7 +831,7 @@ find_files(THD *thd, Dynamic_array<LEX_STRING*> *files, LEX_STRING *db,
   }
   else
   {
-    if (ha_discover_table_names(thd, db, dirp, &tl))
+    if (ha_discover_table_names(thd, db, dirp, &tl, false))
       goto err;
   }
 
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 20e19484aa7..b16264d1145 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -391,31 +391,14 @@ uint filename_to_tablename(const char *from, char *to, uint to_length
   DBUG_ENTER("filename_to_tablename");
   DBUG_PRINT("enter", ("from '%s'", from));
 
-  if (!strncmp(from, tmp_file_prefix, tmp_file_prefix_length))
+  res= strconvert(&my_charset_filename, from,
+                  system_charset_info,  to, to_length, &errors);
+  if (errors) // Old 5.0 name
   {
-    /* Temporary table name. */
-    res= (strnmov(to, from, to_length) - to);
-  }
-  else
-  {
-    res= strconvert(&my_charset_filename, from,
-                    system_charset_info,  to, to_length, &errors);
-    if (errors) // Old 5.0 name
-    {
-      res= (strxnmov(to, to_length, MYSQL50_TABLE_NAME_PREFIX,  from, NullS) -
-            to);
-#ifndef DBUG_OFF
-      if (!stay_quiet) {
-#endif /* DBUG_OFF */
-        sql_print_error("Invalid (old?) table or database name '%s'", from);
-#ifndef DBUG_OFF
-      }
-#endif /* DBUG_OFF */
-      /*
-        TODO: add a stored procedure for fix table and database names,
-        and mention its name in error log.
-      */
-    }
+    res= (strxnmov(to, to_length, MYSQL50_TABLE_NAME_PREFIX,  from, NullS) -
+          to);
+    if (IF_DBUG(!stay_quiet,0))
+      sql_print_error("Invalid (old?) table or database name '%s'", from);
   }
 
   DBUG_PRINT("exit", ("to '%s'", to));
@@ -2230,9 +2213,7 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
       alias= (lower_case_table_names == 2) ? table->alias : table->table_name;
       /* remove .frm file and engine files */
       path_length= build_table_filename(path, sizeof(path) - 1, db, alias,
-                                        reg_ext,
-                                        table->internal_tmp_table ?
-                                        FN_IS_TMP : 0);
+                                        reg_ext, 0);
 
       /*
         This handles the case where a "DROP" was executed and a regular
@@ -2266,8 +2247,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 (!table->internal_tmp_table &&
-        (drop_temporary || !ha_table_exists(thd, db, alias, &table_type) ||
+    if ((drop_temporary || !ha_table_exists(thd, db, alias, &table_type) ||
          (!drop_view && table_type == view_pseudo_hton)))
     {
       /*
@@ -2276,10 +2256,6 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
           . "DROP" but table was not found on disk and table can't be
             created from engine.
           . ./sql/datadict.cc +32 /Alfranio - TODO: We need to test this.
-
-        Table->internal_tmp_table is set when one of the #sql-xxx files
-        was left in the datadir after a crash during ALTER TABLE.
-        See Bug#30152.
       */
       if (if_exists)
 	push_warning_printf(thd, MYSQL_ERROR::WARN_LEVEL_NOTE,
@@ -2294,7 +2270,6 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
     else
     {
       char *end;
-
       /*
         It could happen that table's share in the table_def_cache
         is the only thing that keeps the engine plugin loaded
diff --git a/sql/table.h b/sql/table.h
index af79bffd437..78d1cd01db9 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -1983,7 +1983,6 @@ struct TABLE_LIST
   /* For transactional locking. */
   int           lock_timeout;           /* NOWAIT or WAIT [X]               */
   bool          lock_transactional;     /* If transactional lock requested. */
-  bool          internal_tmp_table;
   /** TRUE if an alias for this table was specified in the SQL. */
   bool          is_alias;
   /** TRUE if the table is referred to in the statement using a fully