MDEV-4955 discover of table non-existance on CREATE

Fix ha_table_exists() to take discovery into account correctly.
It must be able to discover both table existence (when no frm is
found) and table non-existance (when frm was found).

mysql-test/suite/archive/discover.result

@@ -135,7 +135,7 @@ select * from t1;
 a
 flush tables;
 create table t1 (a int) engine=archive;
-ERROR 42S01: Table 't1' already exists
 flush tables;
 create table t1 (a int) engine=archive;
+ERROR 42S01: Table 't1' already exists
 drop table t1;

mysql-test/suite/archive/discover.test

@@ -125,10 +125,10 @@ create table t1 (a int) engine=archive;
 select * from t1;
 flush tables;
 remove_file $mysqld_datadir/test/t1.ARZ;
---error ER_TABLE_EXISTS_ERROR
 create table t1 (a int) engine=archive;
 remove_file $mysqld_datadir/test/t1.frm;
 flush tables;
+--error ER_TABLE_EXISTS_ERROR
 create table t1 (a int) engine=archive;
 drop table t1;
 

sql/handler.cc

@@ -386,12 +386,13 @@ static int ha_finish_errors(void)
 
 static volatile int32 need_full_discover_for_existence= 0;
 static volatile int32 engines_with_discover_table_names= 0;
+static volatile int32 engines_with_discover= 0;
 
 static int full_discover_for_existence(handlerton *, const char *, const char *)
-{ return 1; }
+{ return 0; }
 
 static int ext_based_existence(handlerton *, const char *, const char *)
-{ return 1; }
+{ return 0; }
 
 static int hton_ext_based_table_discovery(handlerton *hton, LEX_STRING *db,
                              MY_DIR *dir, handlerton::discovered_list *result)
@@ -411,6 +412,9 @@ static void update_discovery_counters(handlerton *hton, int val)
 
   if (hton->discover_table_names)
     my_atomic_add32(&engines_with_discover_table_names, val);
+
+  if (hton->discover_table)
+    my_atomic_add32(&engines_with_discover, val);
 }
 
 int ha_finalize_handlerton(st_plugin_int *plugin)
@@ -4787,7 +4791,9 @@ int ha_discover_table(THD *thd, TABLE_SHARE *share)
 
   DBUG_ASSERT(share->error == OPEN_FRM_OPEN_ERROR);   // share is not OK yet
 
-  if (share->db_plugin)
+  if (!engines_with_discover)
+    found= FALSE;
+  else if (share->db_plugin)
     found= discover_handlerton(thd, share->db_plugin, share);
   else
     found= plugin_foreach(thd, discover_handlerton,
@@ -4811,6 +4817,7 @@ struct st_discover_existence_args
   size_t  path_len;
   const char *db, *table_name;
   handlerton *hton;
+  bool frm_exists;
 };
 
 static my_bool discover_existence(THD *thd, plugin_ref plugin,
@@ -4819,7 +4826,7 @@ static my_bool discover_existence(THD *thd, plugin_ref plugin,
   st_discover_existence_args *args= (st_discover_existence_args*)arg;
   handlerton *ht= plugin_hton(plugin);
   if (ht->state != SHOW_OPTION_YES || !ht->discover_table_existence)
-    return FALSE;
+    return args->frm_exists;
 
   args->hton= ht;
 
@@ -4874,40 +4881,36 @@ class Table_exists_error_handler : public Internal_error_handler
   If the 'hton' is not NULL, it's set to the handlerton of the storage engine
   of this table, or to view_pseudo_hton if the frm belongs to a view.
 
+  This function takes discovery correctly into account. If frm is found,
+  it discovers the table to make sure it really exists in the engine.
+  If no frm is found it discovers the table, in case it still exists in
+  the engine.
+
+  While it tries to cut corners (don't open .frm if no discovering engine is
+  enabled, no full discovery if all discovering engines support
+  discover_table_existence, etc), it still *may* be quite expensive
+  and must be used sparingly.
 
   @retval true    Table exists (even if the error occurred, like bad frm)
   @retval false   Table does not exist (one can do CREATE TABLE table_name)
+
+  @note if frm exists and the table in engine doesn't, *hton will be set,
+        but the return value will be false.
+
+  @note if frm file exists, but the table cannot be opened (engine not
+        loaded, frm is invalid), the return value will be true, but
+        *hton will be NULL.
 */
 bool ha_table_exists(THD *thd, const char *db, const char *table_name,
                      handlerton **hton)
 {
+  handlerton *dummy;
   DBUG_ENTER("ha_table_exists");
 
   if (hton)
     *hton= 0;
-
-  if (need_full_discover_for_existence)
-  {
-    TABLE_LIST table;
-    uint flags = GTS_TABLE | GTS_VIEW;
-
-    if (!hton)
-      flags|= GTS_NOLOCK;
-
-    Table_exists_error_handler no_such_table_handler;
-    thd->push_internal_handler(&no_such_table_handler);
-    TABLE_SHARE *share= tdc_acquire_share(thd, db, table_name, flags);
-    thd->pop_internal_handler();
-
-    if (hton && share)
-    {
-      *hton= share->db_type();
-      tdc_release_share(share);
-    }
-
-    // the table doesn't exist if we've caught ER_NO_SUCH_TABLE and nothing else
-    DBUG_RETURN(!no_such_table_handler.safely_trapped_errors());
-  }
+  else if (engines_with_discover)
+    hton= &dummy;
 
   TABLE_SHARE *share= tdc_lock_share(db, table_name);
   if (share)
@@ -4921,22 +4924,29 @@ bool ha_table_exists(THD *thd, const char *db, const char *table_name,
   char path[FN_REFLEN + 1];
   size_t path_len = build_table_filename(path, sizeof(path) - 1,
                                          db, table_name, "", 0);
+  st_discover_existence_args args= {path, path_len, db, table_name, 0, true};
 
   if (file_ext_exists(path, path_len, reg_ext))
   {
+    bool exists= true;
     if (hton)
     {
       enum legacy_db_type db_type;
       if (dd_frm_type(thd, path, &db_type) != FRMTYPE_VIEW)
-        *hton= ha_resolve_by_legacy_type(thd, db_type);
+      {
+        handlerton *ht= ha_resolve_by_legacy_type(thd, db_type);
+        if ((*hton= ht))
+          // verify that the table really exists
+          exists= discover_existence(thd,
+                             plugin_int_to_ref(hton2plugin[ht->slot]), &args);
+      }
       else
         *hton= view_pseudo_hton;
     }
-    DBUG_RETURN(TRUE);
+    DBUG_RETURN(exists);
   }
 
-  st_discover_existence_args args= {path, path_len, db, table_name, 0};
-
+  args.frm_exists= false;
   if (plugin_foreach(thd, discover_existence, MYSQL_STORAGE_ENGINE_PLUGIN,
                      &args))
   {
@@ -4945,6 +4955,30 @@ bool ha_table_exists(THD *thd, const char *db, const char *table_name,
     DBUG_RETURN(TRUE);
   }
 
+
+  if (need_full_discover_for_existence)
+  {
+    TABLE_LIST table;
+    uint flags = GTS_TABLE | GTS_VIEW;
+
+    if (!hton)
+      flags|= GTS_NOLOCK;
+
+    Table_exists_error_handler no_such_table_handler;
+    thd->push_internal_handler(&no_such_table_handler);
+    TABLE_SHARE *share= tdc_acquire_share(thd, db, table_name, flags);
+    thd->pop_internal_handler();
+
+    if (hton && share)
+    {
+      *hton= share->db_type();
+      tdc_release_share(share);
+    }
+
+    // the table doesn't exist if we've caught ER_NO_SUCH_TABLE and nothing else
+    DBUG_RETURN(!no_such_table_handler.safely_trapped_errors());
+  }
+
   DBUG_RETURN(FALSE);
 }
 

sql/sql_table.cc

@@ -2389,7 +2389,7 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
         This handles the case where a "DROP" was executed and a regular
         table "may be" dropped as drop_temporary is FALSE and error is
         TRUE. If the error was FALSE a temporary table was dropped and
-        regardless of the status of drop_tempoary a "DROP TEMPORARY"
+        regardless of the status of drop_temporary a "DROP TEMPORARY"
         must be used.
       */
       if (!dont_log_query)
@@ -2417,15 +2417,15 @@ 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 && (was_view= (table_type == view_pseudo_hton)))))
+    if (drop_temporary ||
+        (ha_table_exists(thd, db, alias, &table_type) == 0 && table_type == 0) ||
+        (!drop_view && (was_view= (table_type == view_pseudo_hton))))
     {
       /*
         One of the following cases happened:
           . "DROP TEMPORARY" but a temporary table was not found.
-          . "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.
+          . "DROP" but table was not found
+          . "DROP TABLE" statement, but it's a view. 
       */
       if (if_exists)
       {

storage/test_sql_discovery/mysql-test/archive/discover.rdiff

---- suite/archive/discover.result	2013-04-08 00:06:37.000000000 +0200
-+++ /usr/home/serg/Abk/mysql/10.0-serg/storage/test_sql_discovery/mysql-test/archive/discover.reject	2013-04-08 00:07:02.000000000 +0200
-@@ -42,6 +42,7 @@
- t1	BASE TABLE
- t2	BASE TABLE
- t1.ARZ
-+t1.frm
- t2.ARZ
- t2.frm
- #
-@@ -60,6 +61,7 @@
- flush tables;
- rename table t2 to t0;
- t0.ARZ
-+t0.frm
- t1.ARZ
- t1.frm
- #
-@@ -77,6 +79,7 @@
- flush tables;
- drop table t1;
- t0.ARZ
-+t0.frm
- #
- # discover of table non-existance on drop
- #
-@@ -86,7 +89,7 @@
- drop table t0;
- show status like 'Handler_discover';
- Variable_name	Value
--Handler_discover	6
-+Handler_discover	7
- #
- # Bug#45377: ARCHIVE tables aren't discoverable after OPTIMIZE
- #

storage/test_sql_discovery/mysql-test/main/r/plugin.rdiff

---- r/plugin.result	2013-02-21 19:46:59.000000000 +0100
-+++ r/plugin.reject	2013-02-27 11:13:22.000000000 +0100
-@@ -71,6 +71,8 @@
- SELECT * FROM t2;
- ERROR 42000: Unknown storage engine 'EXAMPLE'
- DROP TABLE t2;
-+Warnings:
-+Error	1286	Unknown storage engine 'EXAMPLE'
- UNINSTALL PLUGIN EXAMPLE;
- ERROR 42000: PLUGIN EXAMPLE does not exist
- UNINSTALL PLUGIN non_exist;