Merge msvensson@bk-internal.mysql.com:/home/bk/mysql-4.1-ndb

into neptunus.(none):/home/magnus/mysql/wl/wl1424_discover_2


sql/ha_ndbcluster.cc:
  Auto merged
sql/ha_ndbcluster.h:
  Auto merged

10 files changed, 405 insertions, 148 deletions

diff --git a/sql/discover.cc b/sql/discover.cc
index 696be193148..1251055c70e 100644
--- a/sql/discover.cc
+++ b/sql/discover.cc
@@ -125,47 +125,4 @@ int writefrm(const char *name, const void *frmdata, uint len)
 
 
 
-/*
-  Try to discover table from handler and 
-  if found, write the frm file to disk.
-  
-  RETURN VALUES:
-   0 : Table existed in handler and created 
-       on disk if so requested
-   1 : Table does not exist
-  >1 : error
-
-*/
 
-int create_table_from_handler(const char *db, 
-			      const char *name,
-			      bool create_if_found)
-{
-  int error= 0;  
-  const void* frmblob = NULL;
-  char path[FN_REFLEN];
-  uint frmlen = 0;
-  DBUG_ENTER("create_table_from_handler");
-  DBUG_PRINT("enter", ("create_if_found: %d", create_if_found));
-
-  if (ha_discover(db, name, &frmblob, &frmlen))
-    DBUG_RETURN(1); // Table does not exist
-
-  // Table exists in handler  
-  if  (create_if_found)
-  {
-    (void)strxnmov(path,FN_REFLEN,mysql_data_home,"/",db,"/",name,NullS);
-    // Save the frm file    
-    error = writefrm(path, frmblob, frmlen);
-  }
-    
-  if (frmblob)
-    my_free((char*) frmblob,MYF(0));
-  DBUG_RETURN(error);  
-}
-
-int table_exists_in_handler(const char *db, 
-			     const char *name)
-{ 
-  return (create_table_from_handler(db, name, false) == 0);
-}
diff --git a/sql/ha_ndbcluster.cc b/sql/ha_ndbcluster.cc
index 09fc6c908a8..d7bd24fe681 100644
--- a/sql/ha_ndbcluster.cc
+++ b/sql/ha_ndbcluster.cc
@@ -32,9 +32,6 @@
 #include <ndbapi/NdbApi.hpp>
 #include <ndbapi/NdbScanFilter.hpp>
 
-#define USE_DISCOVER_ON_STARTUP
-//#define USE_NDB_POOL
-
 // Default value for parallelism
 static const int parallelism= 240;
 
@@ -48,11 +45,13 @@ static const ha_rows autoincrement_prefetch= 32;
 // connectstring to cluster if given by mysqld
 const char *ndbcluster_connectstring= 0;
 
+static const char *ha_ndb_ext=".ndb";
+
 #define NDB_HIDDEN_PRIMARY_KEY_LENGTH 8
 
 
 #define ERR_PRINT(err) \
-  DBUG_PRINT("error", ("Error: %d  message: %s", err.code, err.message))
+  DBUG_PRINT("error", ("%d  message: %s", err.code, err.message))
 
 #define ERR_RETURN(err)		         \
 {				         \
@@ -107,7 +106,9 @@ static const err_code_mapping err_map[]=
   { 893, HA_ERR_FOUND_DUPP_UNIQUE },
   { 721, HA_ERR_TABLE_EXIST },
   { 4244, HA_ERR_TABLE_EXIST },
-  { 241, HA_ERR_OLD_METADATA },
+
+  { 709, HA_ERR_NO_SUCH_TABLE },
+  { 284, HA_ERR_NO_SUCH_TABLE },
 
   { 266, HA_ERR_LOCK_WAIT_TIMEOUT },
   { 274, HA_ERR_LOCK_WAIT_TIMEOUT },
@@ -268,6 +269,7 @@ int ha_ndbcluster::ndb_err(NdbConnection *trans)
     NDBDICT *dict= m_ndb->getDictionary();
     DBUG_PRINT("info", ("invalidateTable %s", m_tabname));
     dict->invalidateTable(m_tabname);
+    table->version=0L;			/* Free when thread is ready */
     break;
   }
   default:
@@ -308,7 +310,8 @@ bool ha_ndbcluster::get_error_message(int error,
 
 /*
   Check if type is supported by NDB.
-  TODO Use this once, not in every operation
+  TODO Use this once in open(), not in every operation
+
 */
 
 static inline bool ndb_supported_type(enum_field_types type)
@@ -635,7 +638,7 @@ int ha_ndbcluster::get_metadata(const char *path)
     DBUG_PRINT("error",
                ("Wrong number of columns, ndb: %d mysql: %d", 
                 ndb_columns, mysql_columns));
-    DBUG_RETURN(HA_ERR_OLD_METADATA);
+    DBUG_RETURN(3);
   }
   
   /*
@@ -659,7 +662,7 @@ int ha_ndbcluster::get_metadata(const char *path)
 		memcmp(pack_data, tab->getFrmData(), pack_length)));      
     DBUG_DUMP("pack_data", (char*)pack_data, pack_length);
     DBUG_DUMP("frm", (char*)tab->getFrmData(), tab->getFrmLength());
-    error= HA_ERR_OLD_METADATA;
+    error= 2;
   }
   my_free((char*)data, MYF(0));
   my_free((char*)pack_data, MYF(0));
@@ -2628,7 +2631,7 @@ int ha_ndbcluster::reset()
 
 
 const char **ha_ndbcluster::bas_ext() const
-{ static const char *ext[1]= { NullS }; return ext; }
+{ static const char *ext[]= { ".ndb", NullS }; return ext; }
 
 
 /*
@@ -2814,18 +2817,26 @@ int ha_ndbcluster::external_lock(THD *thd, int lock_type)
         thd->transaction.stmt.ndb_tid= 0;
       }
     }
-    if (m_active_trans)
-      DBUG_PRINT("warning", ("m_active_trans != NULL"));
+    /*
+      This is the place to make sure this handler instance
+      no longer are connected to the active transaction.
+
+      And since the handler is no longer part of the transaction 
+      it can't have open cursors, ops or blobs pending.
+    */
+    m_active_trans= NULL;    
+
     if (m_active_cursor)
       DBUG_PRINT("warning", ("m_active_cursor != NULL"));
+    m_active_cursor= NULL;
+
     if (blobs_pending)
       DBUG_PRINT("warning", ("blobs_pending != 0"));
+    blobs_pending= 0;
+    
     if (ops_pending)
       DBUG_PRINT("warning", ("ops_pending != 0L"));
-    m_active_trans= NULL;
-    m_active_cursor= NULL;
     ops_pending= 0;
-    blobs_pending= 0;
   }
   DBUG_RETURN(error);
 }
@@ -3136,12 +3147,24 @@ int ha_ndbcluster::create(const char *name,
   const void *data, *pack_data;
   const char **key_names= form->keynames.type_names;
   char name2[FN_HEADLEN];
+  bool create_from_engine= (info->table_options & HA_CREATE_FROM_ENGINE);
    
   DBUG_ENTER("create");
   DBUG_PRINT("enter", ("name: %s", name));
   fn_format(name2, name, "", "",2);       // Remove the .frm extension
   set_dbname(name2);
-  set_tabname(name2);  
+  set_tabname(name2);    
+
+  if (create_from_engine)
+  {
+    /*
+      Table alreay exists in NDB and frm file has been created by 
+      caller.
+      Do Ndb specific stuff, such as create a .ndb file
+    */
+    my_errno= write_ndb_file();
+    DBUG_RETURN(my_errno);
+  }
 
   DBUG_PRINT("table", ("name: %s", m_tabname));  
   tab.setName(m_tabname);
@@ -3182,16 +3205,12 @@ int ha_ndbcluster::create(const char *name,
     tab.addColumn(col);
   }
   
-  my_errno= 0;
-  if (check_ndb_connection())
-  {
-    my_errno= HA_ERR_NO_CONNECTION;
+  if ((my_errno= check_ndb_connection()))
     DBUG_RETURN(my_errno);
-  }
   
   // Create the table in NDB     
   NDBDICT *dict= m_ndb->getDictionary();
-  if (dict->createTable(tab)) 
+  if (dict->createTable(tab) != 0) 
   {
     const NdbError err= dict->getNdbError();
     ERR_PRINT(err);
@@ -3204,6 +3223,9 @@ int ha_ndbcluster::create(const char *name,
   // Create secondary indexes
   my_errno= build_index_list(form, ILBP_CREATE);
 
+  if (!my_errno)
+    my_errno= write_ndb_file();
+
   DBUG_RETURN(my_errno);
 }
 
@@ -3280,14 +3302,16 @@ int ha_ndbcluster::rename_table(const char *from, const char *to)
   set_tabname(from);
   set_tabname(to, new_tabname);
 
-  if (check_ndb_connection()) {
-    my_errno= HA_ERR_NO_CONNECTION;
-    DBUG_RETURN(my_errno);
-  }
+  if (check_ndb_connection())
+    DBUG_RETURN(my_errno= HA_ERR_NO_CONNECTION);
+
 
   int result= alter_table_name(m_tabname, new_tabname);
   if (result == 0)
+  {
     set_tabname(to);
+    handler::rename_table(from, to);
+  }
   
   DBUG_RETURN(result);
 }
@@ -3331,6 +3355,8 @@ int ha_ndbcluster::delete_table(const char *name)
   
   if (check_ndb_connection())
     DBUG_RETURN(HA_ERR_NO_CONNECTION);
+
+  handler::delete_table(name);
   DBUG_RETURN(drop_table());
 }
 
@@ -3499,7 +3525,7 @@ int ha_ndbcluster::open(const char *name, int mode, uint test_if_locked)
     free_share(m_share); m_share= 0;
     DBUG_RETURN(HA_ERR_NO_CONNECTION);
   }
-
+  
   DBUG_RETURN(get_metadata(name));
 }
 
@@ -3524,13 +3550,7 @@ Thd_ndb* ha_ndbcluster::seize_thd_ndb()
   Thd_ndb *thd_ndb;
   DBUG_ENTER("seize_thd_ndb");
 
-#ifdef USE_NDB_POOL
-  // Seize from pool
-  ndb= Ndb::seize();
-  xxxxxxxxxxxxxx error
-#else
   thd_ndb= new Thd_ndb();
-#endif
   thd_ndb->ndb->getDictionary()->set_local_table_data_size(sizeof(Ndb_table_local_info));
   if (thd_ndb->ndb->init(max_transactions) != 0)
   {
@@ -3551,46 +3571,45 @@ Thd_ndb* ha_ndbcluster::seize_thd_ndb()
 void ha_ndbcluster::release_thd_ndb(Thd_ndb* thd_ndb)
 {
   DBUG_ENTER("release_thd_ndb");
-#ifdef USE_NDB_POOL
-  // Release to  pool
-  Ndb::release(ndb);
-  xxxxxxxxxxxx error
-#else
   delete thd_ndb;
-#endif
   DBUG_VOID_RETURN;
 }
 
 
 /*
-  If this thread already has a Ndb object allocated
+  If this thread already has a Thd_ndb object allocated
   in current THD, reuse it. Otherwise
-  seize a Ndb object, assign it to current THD and use it.
- 
-  Having a Ndb object also means that a connection to 
-  NDB cluster has been opened. The connection is 
-  checked.
+  seize a Thd_ndb object, assign it to current THD and use it.
  
 */
 
-int ha_ndbcluster::check_ndb_connection()
+Ndb* check_ndb_in_thd(THD* thd)
 {
-  THD *thd= current_thd;
+  DBUG_ENTER("check_ndb_in_thd");
   Thd_ndb *thd_ndb= (Thd_ndb*)thd->transaction.thd_ndb;
-  DBUG_ENTER("check_ndb_connection");
   
   if (!thd_ndb)
   {
-    thd_ndb= seize_thd_ndb();
-    if (!thd_ndb)
-      DBUG_RETURN(2);
+    if (!(thd_ndb= ha_ndbcluster::seize_thd_ndb()))
+      DBUG_RETURN(NULL);
     thd->transaction.thd_ndb= thd_ndb;
   }
-  m_ndb= thd_ndb->ndb;
+  DBUG_RETURN(thd_ndb->ndb);
+}
+
+
+int ha_ndbcluster::check_ndb_connection()
+{
+  THD* thd= current_thd;
+  DBUG_ENTER("check_ndb_connection");
+  
+  if (!(m_ndb= check_ndb_in_thd(thd)))
+    DBUG_RETURN(HA_ERR_NO_CONNECTION);
   m_ndb->setDatabaseName(m_dbname);
   DBUG_RETURN(0);
 }
 
+
 void ndbcluster_close_connection(THD *thd)
 {
   Thd_ndb *thd_ndb= (Thd_ndb*)thd->transaction.thd_ndb;
@@ -3608,28 +3627,29 @@ void ndbcluster_close_connection(THD *thd)
   Try to discover one table from NDB
  */
 
-int ndbcluster_discover(const char *dbname, const char *name,
+int ndbcluster_discover(THD* thd, const char *db, const char *name,
 			const void** frmblob, uint* frmlen)
 {
   uint len;
   const void* data;
   const NDBTAB* tab;
+  Ndb* ndb;
   DBUG_ENTER("ndbcluster_discover");
-  DBUG_PRINT("enter", ("db: %s, name: %s", dbname, name)); 
-
-  Ndb ndb(g_ndb_cluster_connection, dbname);
-  ndb.getDictionary()->set_local_table_data_size(sizeof(Ndb_table_local_info));
+  DBUG_PRINT("enter", ("db: %s, name: %s", db, name)); 
 
-  if (ndb.init())
-    ERR_RETURN(ndb.getNdbError());
+  if (!(ndb= check_ndb_in_thd(thd)))
+    DBUG_RETURN(HA_ERR_NO_CONNECTION);  
+  ndb->setDatabaseName(db);
 
-  if (ndb.waitUntilReady(0))
-    ERR_RETURN(ndb.getNdbError());
-  
-  if (!(tab= ndb.getDictionary()->getTable(name)))
-  {
-    DBUG_PRINT("info", ("Table %s not found", name));
-    DBUG_RETURN(1);
+  NDBDICT* dict= ndb->getDictionary();
+  dict->set_local_table_data_size(sizeof(Ndb_table_local_info));
+  dict->invalidateTable(name);
+  if (!(tab= dict->getTable(name)))
+  {    
+    const NdbError err= dict->getNdbError();
+    if (err.code == 709)
+      DBUG_RETURN(1);
+    ERR_RETURN(err);
   }
   
   DBUG_PRINT("info", ("Found table %s", tab->getName()));
@@ -3651,41 +3671,159 @@ int ndbcluster_discover(const char *dbname, const char *name,
   DBUG_RETURN(0);
 }
 
-
-#ifdef USE_DISCOVER_ON_STARTUP
 /*
-  Dicover tables from NDB Cluster
-  - fetch a list of tables from NDB 
-  - store the frm file for each table on disk 
-   - if the table has an attached frm file
-   - if the database of the table exists
-*/
+  Check if a table exists in NDB
+   
+ */
 
-int ndb_discover_tables()
+int ndbcluster_table_exists(THD* thd, const char *db, const char *name)
+{
+  uint len;
+  const void* data;
+  const NDBTAB* tab;
+  Ndb* ndb;
+  DBUG_ENTER("ndbcluster_table_exists");
+  DBUG_PRINT("enter", ("db: %s, name: %s", db, name)); 
+
+  if (!(ndb= check_ndb_in_thd(thd)))
+    DBUG_RETURN(HA_ERR_NO_CONNECTION);  
+  ndb->setDatabaseName(db);
+
+  NDBDICT* dict= ndb->getDictionary();
+  dict->set_local_table_data_size(sizeof(Ndb_table_local_info));
+  dict->invalidateTable(name);
+  if (!(tab= dict->getTable(name)))
+  {    
+    const NdbError err= dict->getNdbError();
+    if (err.code == 709)
+      DBUG_RETURN(0);
+    ERR_RETURN(err);
+  }
+  
+  DBUG_PRINT("info", ("Found table %s", tab->getName()));
+  DBUG_RETURN(1);
+}
+
+
+
+extern "C" byte* ndb_tables_get_key(const char *entry, uint *length,
+				   my_bool not_used __attribute__((unused)))
+{
+  *length= strlen(entry);
+  return (byte*) entry;
+}
+
+
+int ndbcluster_find_files(THD *thd,const char *db,const char *path,
+			  const char *wild, bool dir)
 {
   uint i;
   NdbDictionary::Dictionary::List list;
-  NdbDictionary::Dictionary* dict;
-  char  path[FN_REFLEN];
-  DBUG_ENTER("ndb_discover_tables");
-  
+  Ndb* ndb;
+  char name[FN_REFLEN];
+  HASH ndb_tables;
+  DBUG_ENTER("ndbcluster_list_tables");
+  DBUG_PRINT("enter", ("db: %s", db));
+
+  if (!(ndb= check_ndb_in_thd(thd)))
+    DBUG_RETURN(HA_ERR_NO_CONNECTION);
+
+  if (dir)
+    DBUG_RETURN(0); // Discover of databases not yet discovered
+
+  if (hash_init(&ndb_tables, system_charset_info,32,0,0,
+		(hash_get_key)ndb_tables_get_key,0,0))
+  {
+    DBUG_PRINT("info", ("Failed to init HASH ndb_tables"));
+    DBUG_RETURN(-1);
+  }
+
   /* List tables in NDB Cluster kernel    */  
-  dict= g_ndb->getDictionary();
+  NDBDICT *dict= ndb->getDictionary();
   if (dict->listObjects(list, 
 			NdbDictionary::Object::UserTable) != 0)
     ERR_RETURN(dict->getNdbError());
-  
+
   for (i= 0 ; i < list.count ; i++)
   {
     NdbDictionary::Dictionary::List::Element& t= list.elements[i];
+    DBUG_PRINT("discover", ("%d, %s/%s", t.id, t.database, t.name));     
+    if (my_hash_insert(&ndb_tables, (byte*)thd->strdup(t.name)))
+       continue;
+
+    // Only discover files that fullfill wildcard
+    if (wild)
+    {
+      if (lower_case_table_names)
+      {
+	if (wild_case_compare(files_charset_info, t.name, wild))
+	  continue;
+      }
+      else if (wild_compare(t.name,wild,0))
+	continue;
+    }
 
-    DBUG_PRINT("discover", ("%d: %s/%s", t.id, t.database, t.name));     
-    if (create_table_from_handler(t.database, t.name, true))
-      DBUG_PRINT("info", ("Could not discover %s/%s", t.database, t.name));
+    // Discover the file if it does not already exists on disk    
+    (void)strxnmov(name, FN_REFLEN, 
+		   mysql_data_home,"/",t.database,"/",t.name,reg_ext,NullS);
+    DBUG_PRINT("discover", ("Check access for %s", name));
+    if (access(name, F_OK))
+    {	
+      DBUG_PRINT("discover", ("Table %s need disocver", name));
+      pthread_mutex_lock(&LOCK_open);
+      ha_create_table_from_engine(thd, t.database, t.name, true);
+      pthread_mutex_unlock(&LOCK_open);      
+    }
   }
-  DBUG_RETURN(0);  
+
+  /*
+    Find all .ndb files in current dir and check 
+    if they still exists in NDB
+  */
+  char *ext;
+  MY_DIR *dirp;
+  FILEINFO *file;
+  
+  if (!(dirp= my_dir(path,MYF(MY_WME | (dir ? MY_WANT_STAT : 0)))))
+    DBUG_RETURN(-1);
+
+  for (i= 0; i < (uint)dirp->number_off_files; i++)
+  {
+    file= dirp->dir_entry+i;
+    {
+      ext= fn_ext(file->name);
+      if(!my_strcasecmp(system_charset_info, ext, ha_ndb_ext))
+      {
+	DBUG_PRINT("discover", ("Found file: %s", file->name));
+	*ext= 0;
+
+	if (hash_search(&ndb_tables, file->name, strlen(file->name)))
+	  continue;
+
+	DBUG_PRINT("discover", ("File didn't exist in ndb_tables list"));
+
+	// Verify that handler agrees table is gone.
+	if (ndbcluster_table_exists(thd, db, file->name) == 0)
+	{
+	  DBUG_PRINT("discover", ("Remove table %s/%s",db, file->name ));
+	  // Delete the table and all related files
+	  TABLE_LIST table_list;
+	  bzero((char*) &table_list,sizeof(table_list));
+	  table_list.db= (char*) db;
+	  table_list.real_name=(char*)file->name;
+	  (void)mysql_rm_table_part2_with_lock(thd, &table_list, 
+					       /* if_exists */ true, 
+					       /* drop_temporary */ false, 
+					       /* dont_log_query*/ true);
+	}
+      }      
+    }
+  }
+
+  hash_free(&ndb_tables);
+  my_dirend(dirp);
+  DBUG_RETURN(0);    
 }
-#endif
 
 
 /*
@@ -3737,10 +3875,6 @@ bool ndbcluster_init()
   pthread_mutex_init(&ndbcluster_mutex,MY_MUTEX_INIT_FAST);
 
   ndbcluster_inited= 1;
-#ifdef USE_DISCOVER_ON_STARTUP
-  if (ndb_discover_tables() != 0)
-    DBUG_RETURN(TRUE);    
-#endif
   DBUG_RETURN(false);
 }
 
@@ -3754,7 +3888,6 @@ bool ndbcluster_init()
 bool ndbcluster_end()
 {
   DBUG_ENTER("ndbcluster_end");
-
   if(g_ndb)
     delete g_ndb;
   g_ndb= NULL;
@@ -3764,9 +3897,6 @@ bool ndbcluster_end()
   if (!ndbcluster_inited)
     DBUG_RETURN(0);
   hash_free(&ndbcluster_open_tables);
-#ifdef USE_NDB_POOL
-  ndb_pool_release();
-#endif
   pthread_mutex_destroy(&ndbcluster_mutex);
   ndbcluster_inited= 0;
   DBUG_RETURN(0);
@@ -4125,4 +4255,30 @@ ndb_get_table_statistics(Ndb* ndb, const char * table,
   DBUG_RETURN(-1);
 }
 
+/*
+  Create a .ndb file to serve as a placeholder indicating 
+  that the table with this name is a ndb table
+*/
+
+int ha_ndbcluster::write_ndb_file()
+{
+  File file;
+  bool error=1;
+  char path[FN_REFLEN];
+  
+  DBUG_ENTER("write_ndb_file");
+  DBUG_PRINT("enter", ("db: %s, name: %s", m_dbname, m_tabname));
+
+  (void)strxnmov(path, FN_REFLEN, 
+		 mysql_data_home,"/",m_dbname,"/",m_tabname,ha_ndb_ext,NullS);
+
+  if ((file=my_create(path, CREATE_MODE,O_RDWR | O_TRUNC,MYF(MY_WME))) >= 0)
+  {
+    // It's an empty file
+    error=0;
+    my_close(file,MYF(0));
+  }
+  DBUG_RETURN(error);
+}
+
 #endif /* HAVE_NDBCLUSTER_DB */
diff --git a/sql/ha_ndbcluster.h b/sql/ha_ndbcluster.h
index eb4556a606b..217ba84b00a 100644
--- a/sql/ha_ndbcluster.h
+++ b/sql/ha_ndbcluster.h
@@ -226,6 +226,8 @@ class ha_ndbcluster: public handler
   int ndb_err(NdbConnection*);
   bool uses_blob_value(bool all_fields);
 
+  int write_ndb_file();
+
  private:
   int check_ndb_connection();
 
@@ -276,8 +278,11 @@ int ndbcluster_rollback(THD *thd, void* ndb_transaction);
 
 void ndbcluster_close_connection(THD *thd);
 
-int ndbcluster_discover(const char* dbname, const char* name,
+int ndbcluster_discover(THD* thd, const char* dbname, const char* name,
 			const void** frmblob, uint* frmlen);
+int ndbcluster_find_files(THD *thd,const char *db,const char *path,
+			  const char *wild, bool dir);
+int ndbcluster_table_exists(THD* thd, const char *db, const char *name);
 int ndbcluster_drop_database(const char* path);
 
 void ndbcluster_print_error(int error, const NdbOperation *error_op);
diff --git a/sql/handler.cc b/sql/handler.cc
index 859c7124566..dc2f213640a 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -1119,6 +1119,16 @@ void handler::print_error(int error, myf errflag)
   case HA_ERR_NO_REFERENCED_ROW:
     textno=ER_NO_REFERENCED_ROW;
     break;
+  case HA_ERR_NO_SUCH_TABLE:
+  {
+    char *db;
+    char buff[FN_REFLEN];
+    uint length=dirname_part(buff,table->path);
+    buff[length-1]=0;
+    db=buff+dirname_length(buff);
+    my_error(ER_NO_SUCH_TABLE,MYF(0),db,table->table_name);
+    break;
+  }
   default:
     {
       /* The error was "unknown" to this function.
@@ -1265,6 +1275,71 @@ int ha_create_table(const char *name, HA_CREATE_INFO *create_info,
   DBUG_RETURN(error != 0);
 }
 
+/*
+  Try to discover table from engine and 
+  if found, write the frm file to disk.
+  
+  RETURN VALUES:
+   0 : Table existed in engine and created 
+       on disk if so requested
+   1 : Table does not exist
+  >1 : error
+
+*/
+
+int ha_create_table_from_engine(THD* thd, 
+				const char *db, 
+				const char *name,
+				bool create_if_found)
+{
+  int error= 0;  
+  const void* frmblob = NULL;
+  uint frmlen = 0;
+  char path[FN_REFLEN];
+  HA_CREATE_INFO create_info;
+  TABLE table;
+  DBUG_ENTER("ha_create_table_from_engine");
+  DBUG_PRINT("enter", ("db: %s, name: %s", db, name));
+  DBUG_PRINT("enter", ("create_if_found: %d", create_if_found));
+
+  bzero((char*) &create_info,sizeof(create_info));
+
+  if ((error= ha_discover(thd, db, name, &frmblob, &frmlen)))
+    DBUG_RETURN(error); 
+
+  // Table exists in handler  
+  if  (create_if_found)
+  {
+    (void)strxnmov(path,FN_REFLEN,mysql_data_home,"/",db,"/",name,NullS);
+    // Save the frm file    
+    if ((error = writefrm(path, frmblob, frmlen)))
+      goto err_end;
+
+    if (openfrm(path,"",0,(uint) READ_ALL, 0, &table))
+      DBUG_RETURN(1);
+
+    update_create_info_from_table(&create_info, &table);
+    create_info.table_options|= HA_CREATE_FROM_ENGINE;
+
+    if (lower_case_table_names == 2 &&
+	!(table.file->table_flags() & HA_FILE_BASED))
+    {
+      /* Ensure that handler gets name in lower case */
+      strmov(path, name);
+      my_casedn_str(files_charset_info, path);
+      name= path;
+    }
+    
+    error=table.file->create(path,&table,&create_info);
+    VOID(closefrm(&table));
+  }
+
+err_end:
+  if (frmblob)
+    my_free((char*) frmblob,MYF(0));
+  DBUG_RETURN(error);  
+}
+
 static int NEAR_F delete_file(const char *name,const char *ext,int extflag)
 {
   char buff[FN_REFLEN];
@@ -1372,15 +1447,15 @@ int ha_change_key_cache(KEY_CACHE *old_key_cache,
   Try to discover one table from handler(s)
 */
 
-int ha_discover(const char* dbname, const char* name,
-               const void** frmblob, uint* frmlen)
+int ha_discover(THD* thd, const char* db, const char* name,
+		const void** frmblob, uint* frmlen)
 {
   int error= 1; // Table does not exist in any handler
   DBUG_ENTER("ha_discover");
-  DBUG_PRINT("enter", ("db: %s, name: %s", dbname, name));
+  DBUG_PRINT("enter", ("db: %s, name: %s", db, name));
 #ifdef HAVE_NDBCLUSTER_DB
   if (have_ndbcluster == SHOW_OPTION_YES)
-    error= ndbcluster_discover(dbname, name, frmblob, frmlen);
+    error= ndbcluster_discover(thd, db, name, frmblob, frmlen);
 #endif
   if (!error)
     statistic_increment(ha_discover_count,&LOCK_status);
@@ -1389,6 +1464,52 @@ int ha_discover(const char* dbname, const char* name,
 
 
 /*
+  Call this function in order to give the handler the possiblity 
+  to ask engine if there are any new tables that should be written to disk 
+  or any dropped tables that need to be removed from disk
+*/
+
+int
+ha_find_files(THD *thd,const char *db,const char *path,
+	      const char *wild, bool dir)
+{
+  int error= 0;
+  DBUG_ENTER("ha_find_files");
+  DBUG_PRINT("enter", ("db: %s, path: %s, wild: %s, dir: %d", 
+		       db, path, wild, dir));
+#ifdef HAVE_NDBCLUSTER_DB
+  if (have_ndbcluster == SHOW_OPTION_YES)
+    error= ndbcluster_find_files(thd, db, path, wild, dir);
+#endif
+  DBUG_RETURN(error);
+  
+  
+}
+
+/*
+  Ask handler if the table exists in engine
+
+  RETURN
+    0                   Table does not exist
+    1                   Table exists
+    #                   Error code
+
+ */
+int ha_table_exists(THD* thd, const char* db, const char* name)
+{
+  int error= 2;
+  DBUG_ENTER("ha_table_exists");
+  DBUG_PRINT("enter", ("db: %s, name: %s", db, name));
+#ifdef HAVE_NDBCLUSTER_DB
+  if (have_ndbcluster == SHOW_OPTION_YES)
+    error= ndbcluster_table_exists(thd, db, name);
+#endif
+  DBUG_RETURN(error);
+}
+
+
+
+/*
   Read first row between two ranges.
   Store ranges for future calls to read_range_next
 
diff --git a/sql/handler.h b/sql/handler.h
index 0b7e9c04381..fea03c5080e 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -540,6 +540,8 @@ void ha_close_connection(THD* thd);
 enum db_type ha_checktype(enum db_type database_type);
 int ha_create_table(const char *name, HA_CREATE_INFO *create_info,
 		    bool update_create_info);
+int ha_create_table_from_engine(THD* thd, const char *db, const char *name,
+				bool create_if_found);
 int ha_delete_table(enum db_type db_type, const char *path);
 void ha_drop_database(char* path);
 int ha_init_key_cache(const char *name, KEY_CACHE *key_cache);
@@ -561,5 +563,10 @@ bool ha_flush_logs(void);
 int ha_enable_transaction(THD *thd, bool on);
 int ha_change_key_cache(KEY_CACHE *old_key_cache,
 			KEY_CACHE *new_key_cache);
-int ha_discover(const char* dbname, const char* name,
+int ha_discover(THD* thd, const char* dbname, const char* name,
 		const void** frmblob, uint* frmlen);
+int ha_find_files(THD *thd,const char *db,const char *path,
+		  const char *wild, bool dir);
+int ha_table_exists(THD* thd, const char* db, const char* name);
+
+
diff --git a/sql/mysql_priv.h b/sql/mysql_priv.h
index e47807dd36e..6742c248d93 100644
--- a/sql/mysql_priv.h
+++ b/sql/mysql_priv.h
@@ -1003,8 +1003,6 @@ int openfrm(const char *name,const char *alias,uint filestat,uint prgflag,
 	    uint ha_open_flags, TABLE *outparam);
 int readfrm(const char *name, const void** data, uint* length);
 int writefrm(const char* name, const void* data, uint len);
-int create_table_from_handler(const char *db, const char *name,
-			      bool create_if_found);
 int closefrm(TABLE *table);
 db_type get_table_type(const char *name);
 int read_string(File file, gptr *to, uint length);
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index 75eb5753e1e..32ef0081fa0 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -1343,7 +1343,7 @@ static int open_unireg_entry(THD *thd, TABLE *entry, const char *db,
       */
       if (discover_retry_count++ != 0)
        goto err;
-      if (create_table_from_handler(db, name, true) != 0)
+      if (ha_create_table_from_engine(thd, db, name, true) != 0)
        goto err;
 
       thd->clear_error(); // Clear error message
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index 57c5f01d0bf..850e18db748 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -374,6 +374,10 @@ mysql_find_files(THD *thd,List<char> *files, const char *db,const char *path,
 
   if (wild && !wild[0])
     wild=0;
+
+  if (ha_find_files(thd,db,path,wild,dir))
+    DBUG_RETURN(-1);
+
   bzero((char*) &table_list,sizeof(table_list));
 
   if (!(dirp = my_dir(path,MYF(MY_WME | (dir ? MY_WANT_STAT : 0)))))
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 65e26346834..3dbda7a45c5 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -222,7 +222,8 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
       strxmov(path, mysql_data_home, "/", db, "/", alias, reg_ext, NullS);
       (void) unpack_filename(path,path);
     }
-    if (drop_temporary || access(path,F_OK))
+    if (drop_temporary || 
+	(access(path,F_OK) && ha_create_table_from_engine(thd,db,alias,true)))
     {
       if (if_exists)
 	push_warning_printf(thd, MYSQL_ERROR::WARN_LEVEL_NOTE,
@@ -1243,8 +1244,8 @@ int mysql_create_table(THD *thd,const char *db, const char *table_name,
   {
     bool create_if_not_exists =
       create_info->options & HA_LEX_CREATE_IF_NOT_EXISTS;
-    if (!create_table_from_handler(db, table_name,
-                                 create_if_not_exists))
+    if (!ha_create_table_from_engine(thd, db, table_name,
+				     create_if_not_exists))
     {
       DBUG_PRINT("info", ("Table already existed in handler"));
 
diff --git a/sql/table.cc b/sql/table.cc
index 8b018d61e5a..7b64ef7a48c 100644
--- a/sql/table.cc
+++ b/sql/table.cc
@@ -705,6 +705,14 @@ int openfrm(const char *name, const char *alias, uint db_stat, uint prgflag,
       outparam->crashed=((err == HA_ERR_CRASHED_ON_USAGE) &&
 			 outparam->file->auto_repair() &&
 			 !(ha_open_flags & HA_OPEN_FOR_REPAIR));
+
+      if (err==HA_ERR_NO_SUCH_TABLE)
+      {
+	/* The table did not exists in storage engine, use same error message
+	   as if the .frm file didn't exist */
+	error= 1;
+	my_errno= ENOENT;
+      }
       goto err_not_open; /* purecov: inspected */
     }
   }