MDEV-11412 Ensure that table is truly dropped when using DROP TABLE

don't do table discovery on DROP. DROP falls back to "force"
approach when a table isn't found and will try to drop in all
engines anyway. That is, trying to discover in all engines before
the drop is redundant and may be expensive.

6 files changed, 85 insertions, 73 deletions

diff --git a/mysql-test/main/drop_combinations.result b/mysql-test/main/drop_combinations.result
index eb6d70e0704..140665efd19 100644
--- a/mysql-test/main/drop_combinations.result
+++ b/mysql-test/main/drop_combinations.result
@@ -168,8 +168,7 @@ drop table if exists t1,s1,v1,t3,t4;
 Warnings:
 Warning	1017	Can't find file: './test/t1.MYI' (errno: 2 "No such file or directory")
 Note	1965	'test.v1' is a view
-Note	1965	'test.t3' is a view
-Note	1965	'test.t4' is a view
+Note	1051	Unknown table 'test.t3,test.t4'
 drop table if exists s2,v2,t2,t1;
 Warnings:
 Note	1965	'test.v2' is a view
@@ -493,8 +492,7 @@ Warnings:
 Warning	1017	Can't find file: './test/s1.MYI' (errno: 2 "No such file or directory")
 Note	4090	'test.t1' is not a SEQUENCE
 Note	1965	'test.v1' is a view
-Note	1965	'test.t3' is a view
-Note	1965	'test.s4' is a view
+Note	4091	Unknown SEQUENCE: 'test.t3,test.s4'
 drop sequence if exists t2,v2,s2,s1;
 Warnings:
 Note	4090	'test.t2' is not a SEQUENCE
diff --git a/sql/ha_sequence.cc b/sql/ha_sequence.cc
index bf3b5ce2cd0..596f8584041 100644
--- a/sql/ha_sequence.cc
+++ b/sql/ha_sequence.cc
@@ -439,8 +439,7 @@ static int sequence_initialize(void *p)
                                HTON_HIDDEN |
                                HTON_TEMPORARY_NOT_SUPPORTED |
                                HTON_ALTER_NOT_SUPPORTED |
-                               HTON_NO_PARTITION |
-                               HTON_AUTOMATIC_DELETE_TABLE);
+                               HTON_NO_PARTITION);
   DBUG_RETURN(0);
 }
 
diff --git a/sql/handler.cc b/sql/handler.cc
index 2259e7a8e49..e3ce0436a79 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -45,6 +45,7 @@
 #include "sql_audit.h"
 #include "ha_sequence.h"
 #include "rowid_filter.h"
+#include "mysys_err.h"
 
 #ifdef WITH_PARTITION_STORAGE_ENGINE
 #include "ha_partition.h"
@@ -2771,18 +2772,20 @@ int ha_delete_table(THD *thd, handlerton *hton, const char *path,
       TABLE dummy_table;
       TABLE_SHARE dummy_share;
       handler *file= get_new_handler(nullptr, thd->mem_root, hton);
-      bzero((char*) &dummy_table, sizeof(dummy_table));
-      bzero((char*) &dummy_share, sizeof(dummy_share));
-      dummy_share.path.str= (char*) path;
-      dummy_share.path.length= strlen(path);
-      dummy_share.normalized_path= dummy_share.path;
-      dummy_share.db= *db;
-      dummy_share.table_name= *alias;
-      dummy_table.s= &dummy_share;
-      dummy_table.alias.set(alias->str, alias->length, table_alias_charset);
-      file->change_table_ptr(&dummy_table, &dummy_share);
-      file->print_error(error, MYF(intercept ? ME_WARNING : 0));
-      delete file;
+      if (file) {
+        bzero((char*) &dummy_table, sizeof(dummy_table));
+        bzero((char*) &dummy_share, sizeof(dummy_share));
+        dummy_share.path.str= (char*) path;
+        dummy_share.path.length= strlen(path);
+        dummy_share.normalized_path= dummy_share.path;
+        dummy_share.db= *db;
+        dummy_share.table_name= *alias;
+        dummy_table.s= &dummy_share;
+        dummy_table.alias.set(alias->str, alias->length, table_alias_charset);
+        file->change_table_ptr(&dummy_table, &dummy_share);
+        file->print_error(error, MYF(intercept ? ME_WARNING : 0));
+        delete file;
+      }
     }
     if (intercept)
     {
@@ -4455,10 +4458,6 @@ int handler::delete_table(const char *name)
   bool some_file_deleted= 0;
   DBUG_ENTER("handler::delete_table");
 
-  // For discovery tables, it's ok if first file doesn't exists
-  if (ht->discover_table)
-    saved_error= 0;
-
   for (const char **ext= bas_ext(); *ext ; ext++)
   {
     int err= mysql_file_delete_with_symlink(key_file_misc, name, *ext, MYF(0));
@@ -4522,7 +4521,9 @@ void handler::drop_table(const char *name)
 
 bool non_existing_table_error(int error)
 {
-  return (error == ENOENT || error == HA_ERR_NO_SUCH_TABLE ||
+  return (error == ENOENT ||
+          (error == EE_DELETE && my_errno == ENOENT) ||
+          error == HA_ERR_NO_SUCH_TABLE ||
           error == HA_ERR_UNSUPPORTED ||
           error == ER_NO_SUCH_TABLE ||
           error == ER_NO_SUCH_TABLE_IN_ENGINE ||
@@ -5002,12 +5003,11 @@ static my_bool delete_table_force(THD *thd, plugin_ref plugin, void *arg)
 
   /*
     We have to ignore HEAP tables as these may not have been created yet
-    We also remove engines that is using discovery (as these will recrate
-    any missing .frm if needed) and tables marked with
+    We also remove engines  marked with
     HTON_AUTOMATIC_DELETE_TABLE as for these we can't check if the table
     ever existed.
   */
-  if (!hton->discover_table && hton->db_type != DB_TYPE_HEAP &&
+  if (hton->db_type != DB_TYPE_HEAP &&
       !(hton->flags & HTON_AUTOMATIC_DELETE_TABLE))
   {
     int error;
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index b7e9b50cfbe..2237d7bc70d 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -2246,7 +2246,7 @@ 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 was_view= 0, was_table= 0, is_sequence, log_if_exists= if_exists;
+  bool was_view= 0, was_table= 0, log_if_exists= if_exists;
   const char *object_to_drop= (drop_sequence) ? "SEQUENCE" : "TABLE";
   String normal_tables;
   String built_trans_tmp_query, built_non_trans_tmp_query;
@@ -2306,13 +2306,14 @@ 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= 0, frm_was_deleted= 0, temporary_table_was_dropped= 0;
+    bool is_trans= 0, temporary_table_was_dropped= 0;
     bool table_creation_was_logged= 0;
-    bool local_non_tmp_error= 0, frm_exists= 0, wrong_drop_sequence= 0;
+    bool local_non_tmp_error= 0, wrong_drop_sequence= 0;
     bool table_dropped= 0;
     const LEX_CSTRING db= table->db;
     const LEX_CSTRING table_name= table->table_name;
-    handlerton *table_type= 0;
+    handlerton *hton= 0;
+    Table_type table_type;
     size_t path_length= 0;
     char *path_end= 0;
 
@@ -2410,13 +2411,32 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
     if (drop_temporary)
     {
       /* "DROP TEMPORARY" but a temporary table was not found */
+      unknown_tables.append(&db);
+      unknown_tables.append('.');
+      unknown_tables.append(&table_name);
+      unknown_tables.append(',');
       error= ENOENT;
+      not_found_errors++;
+      continue;
+    }
+
+    {
+      char engine_buf[NAME_CHAR_LEN + 1];
+      LEX_CSTRING engine= { engine_buf, 0 };
+
+      table_type= dd_frm_type(thd, path, &engine);
+      if (table_type == TABLE_TYPE_NORMAL || table_type == TABLE_TYPE_SEQUENCE)
+      {
+        plugin_ref p= plugin_lock_by_name(thd, &engine,
+                                           MYSQL_STORAGE_ENGINE_PLUGIN);
+        hton= p ? plugin_hton(p) : NULL;
+      }
+      // note that for TABLE_TYPE_VIEW and TABLE_TYPE_UNKNOWN hton == NULL
     }
-    else if (((frm_exists= ha_table_exists(thd, &db, &alias, &table_type,
-                                           &is_sequence)) == 0 &&
-              table_type == 0) ||
-             (!drop_view && (was_view= (table_type == view_pseudo_hton))) ||
-             (drop_sequence && !is_sequence))
+
+    was_view= table_type == TABLE_TYPE_VIEW;
+    if ((table_type == TABLE_TYPE_UNKNOWN) || (was_view && !drop_view) ||
+        (table_type != TABLE_TYPE_SEQUENCE && drop_sequence))
     {
       /*
         One of the following cases happened:
@@ -2424,34 +2444,19 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
           . "DROP TABLE" statement, but it's a view.
           . "DROP SEQUENCE", but it's not a sequence
       */
-      wrong_drop_sequence= drop_sequence && table_type;
+      wrong_drop_sequence= drop_sequence && hton;
       was_table|= wrong_drop_sequence;
       local_non_tmp_error= 1;
-      error= -1;
-      if ((!frm_exists && !table_type))       // no .frm
-        error= ENOENT;
+      error= table_type == TABLE_TYPE_UNKNOWN ? ENOENT : -1;
     }
     else
     {
-      if (WSREP(thd) && !wsrep_should_replicate_ddl(thd, table_type->db_type))
+      if (WSREP(thd) && hton && !wsrep_should_replicate_ddl(thd, hton->db_type))
       {
         error= 1;
         goto err;
       }
 
-      /*
-        It could happen that table's share in the table definition cache
-        is the only thing that keeps the engine plugin loaded
-        (if it is uninstalled and waits for the ref counter to drop to 0).
-
-        In this case, the tdc_remove_table() below will release and unload
-        the plugin. And ha_delete_table() will get a dangling pointer.
-
-        Let's lock the plugin till the end of the statement.
-      */
-      if (table_type && table_type != view_pseudo_hton)
-        ha_lock_engine(thd, table_type);
-
       if (thd->locked_tables_mode == LTM_LOCK_TABLES ||
           thd->locked_tables_mode == LTM_PRELOCKED_UNDER_LOCK_TABLES)
       {
@@ -2474,13 +2479,12 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
       // Remove extension for delete
       *path_end= '\0';
 
-      if (table_type && table_type != view_pseudo_hton &&
-          table_type->flags & HTON_TABLE_MAY_NOT_EXIST_ON_SLAVE)
+      if (hton && hton->flags & HTON_TABLE_MAY_NOT_EXIST_ON_SLAVE)
         log_if_exists= 1;
 
       thd->replication_flags= 0;
-      error= ha_delete_table(thd, table_type, path, &db,
-                             &table_name, !dont_log_query);
+      bool enoent_warning= !dont_log_query && !(hton && hton->discover_table);
+      error= ha_delete_table(thd, hton, path, &db, &table_name, enoent_warning);
 
       if (!error)
         table_dropped= 1;
@@ -2508,8 +2512,7 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
       {
         int frm_delete_error= 0;
         /* Delete the table definition file */
-        if (table_type && table_type != view_pseudo_hton &&
-            (table_type->discover_table || error))
+        if (hton && (hton->discover_table || error))
         {
           /*
             Table type is using discovery and may not need a .frm file
@@ -2527,7 +2530,6 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
           frm_delete_error= my_errno;
           DBUG_ASSERT(frm_delete_error);
         }
-        frm_was_deleted= 1;                     // We tried to delete .frm
 
         if (frm_delete_error)
         {
@@ -2548,10 +2550,10 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
       scan all engines try to drop the table from there.
       This is to ensure we don't have any partial table files left.
     */
-    if (non_existing_table_error(error) && !drop_temporary &&
-        table_type != view_pseudo_hton && !wrong_drop_sequence)
+    if (non_existing_table_error(error) && !wrong_drop_sequence)
     {
       int ferror= 0;
+      DBUG_ASSERT(!was_view);
 
       /* Remove extension for delete */
       *path_end= '\0';
@@ -2567,14 +2569,10 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
       {
         ferror= 0;                              // Ignore table not found
 
-        /* Delete the table definition file */
-        if (!frm_was_deleted)
-        {
-          strmov(path_end, reg_ext);
-          if (mysql_file_delete(key_file_frm, path,
-                                MYF(MY_WME | MY_IGNORE_ENOENT)))
-            ferror= my_errno;
-        }
+        /* Delete the frm file again (just in case it was rediscovered) */
+        strmov(path_end, reg_ext);
+        if (mysql_file_delete(key_file_frm, path, MYF(MY_WME|MY_IGNORE_ENOENT)))
+          ferror= my_errno;
       }
       if (!error)
         error= ferror;
@@ -2642,7 +2640,7 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
       mysql_audit_drop_table(thd, table);
     }
 
-    if (!dont_log_query && !drop_temporary &&
+    if (!dont_log_query &&
         (!error || table_dropped || non_existing_table_error(error)))
     {
       non_tmp_table_deleted|= (if_exists || table_dropped);
diff --git a/storage/sequence/sequence.cc b/storage/sequence/sequence.cc
index ddebb1a1a8d..c8f3e76b873 100644
--- a/storage/sequence/sequence.cc
+++ b/storage/sequence/sequence.cc
@@ -497,19 +497,27 @@ int ha_seq_group_by_handler::next_row()
   Initialize the interface between the sequence engine and MariaDB
 *****************************************************************************/
 
+static int drop_table(handlerton *hton, const char *path)
+{
+  const char *name= strrchr(path, FN_LIBCHAR)+1;
+  ulonglong from, to, step;
+  if (parse_table_name(name, strlen(name), &from, &to, &step))
+    return ENOENT;
+  return 0;
+}
+
 static int init(void *p)
 {
   handlerton *hton= (handlerton *)p;
   sequence_hton= hton;
   hton->create= create_handler;
-  hton->drop_table= [](handlerton *, const char*) { return 0; };
+  hton->drop_table= drop_table;
   hton->discover_table= discover_table;
   hton->discover_table_existence= discover_table_existence;
   hton->commit= hton->rollback= dummy_commit_rollback;
   hton->savepoint_set= hton->savepoint_rollback= hton->savepoint_release=
     dummy_savepoint;
   hton->create_group_by= create_group_by_handler;
-  hton->flags= HTON_AUTOMATIC_DELETE_TABLE;
   return 0;
 }
 
diff --git a/storage/test_sql_discovery/test_sql_discovery.cc b/storage/test_sql_discovery/test_sql_discovery.cc
index 9e7a22368fc..0758d5f503f 100644
--- a/storage/test_sql_discovery/test_sql_discovery.cc
+++ b/storage/test_sql_discovery/test_sql_discovery.cc
@@ -147,11 +147,20 @@ static int discover_table(handlerton *hton, THD* thd, TABLE_SHARE *share)
                                                sql, strlen(sql));
 }
 
+static int drop_table(handlerton *hton, const char *path)
+{
+  const char *name= strrchr(path, FN_LIBCHAR)+1;
+  const char *sql= THDVAR(current_thd, statement);
+  return !sql || strncmp(sql, name, strlen(name)) || sql[strlen(name)] != ':'
+    ? ENOENT : 0;
+}
+
 static int init(void *p)
 {
   handlerton *hton = (handlerton *)p;
   hton->create = create_handler;
   hton->discover_table = discover_table;
+  hton->drop_table= drop_table;
   return 0;
 }