discovery using sql CREATE TABLE statement

8 files changed, 324 insertions, 197 deletions

diff --git a/sql/handler.cc b/sql/handler.cc
index d58040abaa8..735c2b012e6 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -4317,6 +4317,7 @@ static my_bool discover_handlerton(THD *thd, plugin_ref plugin,
   handlerton *hton= plugin_data(plugin, handlerton *);
   if (hton->state == SHOW_OPTION_YES && hton->discover_table)
   {
+    share->db_plugin= plugin;
     int error= hton->discover_table(hton, thd, share);
     if (error != HA_ERR_NO_SUCH_TABLE)
     {
@@ -4324,6 +4325,7 @@ static my_bool discover_handlerton(THD *thd, plugin_ref plugin,
       {
         DBUG_ASSERT(share->error); // MUST be always set for get_cached_table_share to work
         my_error(ER_GET_ERRNO, MYF(0), error);
+        share->db_plugin= 0;
       }
       else
         share->error= OPEN_FRM_OK;
@@ -4331,6 +4333,7 @@ static my_bool discover_handlerton(THD *thd, plugin_ref plugin,
       status_var_increment(thd->status_var.ha_discover_count);
       return TRUE; // abort the search
     }
+    share->db_plugin= 0;
   }
 
   DBUG_ASSERT(share->error == OPEN_FRM_OPEN_ERROR);
@@ -4342,6 +4345,7 @@ int ha_discover_table(THD *thd, TABLE_SHARE *share)
   DBUG_ENTER("ha_discover_table");
 
   DBUG_ASSERT(share->error == OPEN_FRM_OPEN_ERROR);   // share is not OK yet
+  DBUG_ASSERT(!share->db_plugin);
 
   if (!plugin_foreach(thd, discover_handlerton,
                       MYSQL_STORAGE_ENGINE_PLUGIN, share))
diff --git a/sql/share/errmsg-utf8.txt b/sql/share/errmsg-utf8.txt
index e7f60987cc1..fd74af7d56d 100644
--- a/sql/share/errmsg-utf8.txt
+++ b/sql/share/errmsg-utf8.txt
@@ -6600,3 +6600,6 @@ ER_SLAVE_STARTED
         eng "SLAVE '%.*s' started"
 ER_SLAVE_STOPPED
         eng "SLAVE '%.*s' stopped"
+ER_SQL_DISCOVER_ERROR
+        eng "Engine %s failed to discover table %`-.192s.%`-.192s with '%s'"
+
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index f6a5cc3448a..0a02e674c07 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -1703,13 +1703,23 @@ bool mysql_write_frm(ALTER_PARTITION_PARAM_TYPE *lpt, uint flags)
 #endif
     /* Write shadow frm file */
     lpt->create_info->table_options= lpt->db_options;
-    if ((mysql_create_frm(lpt->thd, shadow_path, lpt->db,
-                          lpt->table_name, lpt->create_info,
-                          lpt->alter_info->create_list, lpt->key_count,
-                          lpt->key_info_buffer, lpt->table->file)) ||
-        lpt->table->file->ha_create_partitioning_metadata(shadow_path, NULL,
-                                                  CHF_CREATE_FLAG,
-                                                  lpt->create_info))
+    LEX_CUSTRING frm= build_frm_image(lpt->thd, lpt->table_name,
+                                      lpt->create_info,
+                                      lpt->alter_info->create_list,
+                                      lpt->key_count, lpt->key_info_buffer,
+                                      lpt->table->file);
+    if (!frm.str)
+    {
+      error= 1;
+      goto end;
+    }
+
+    int error= writefrm(shadow_path, lpt->db, lpt->table_name,
+                        !lpt->create_info->tmp_table(), frm.str, frm.length);
+    my_free(const_cast<uchar*>(frm.str));
+
+    if (error || lpt->table->file->ha_create_partitioning_metadata(shadow_path,
+                                       NULL, CHF_CREATE_FLAG, lpt->create_info))
     {
       mysql_file_delete(key_file_frm, shadow_frm_name, MYF(0));
       error= 1;
@@ -3857,6 +3867,21 @@ mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,
     }
   }
 
+  if (create_info->tmp_table())
+    create_info->table_options|=HA_CREATE_DELAY_KEY_WRITE;
+
+  /* Give warnings for not supported table options */
+#if defined(WITH_ARIA_STORAGE_ENGINE)
+  extern handlerton *maria_hton;
+  if (file->ht != maria_hton)
+#endif
+    if (create_info->transactional)
+      push_warning_printf(thd, MYSQL_ERROR::WARN_LEVEL_WARN,
+                          ER_ILLEGAL_HA_CREATE_OPTION,
+                          ER(ER_ILLEGAL_HA_CREATE_OPTION),
+                          file->engine_name()->str,
+                          "TRANSACTIONAL=1");
+
   if (parse_option_list(thd, &create_info->option_struct,
                           create_info->option_list,
                           file->partition_ht()->table_options, FALSE,
@@ -4047,70 +4072,27 @@ static bool check_if_created_table_can_be_opened(THD *thd,
 #endif
 
 
-/*
-  Create a table
-
-  SYNOPSIS
-    mysql_create_table_no_lock()
-    thd			Thread object
-    db			Database
-    table_name		Table name
-    create_info	        Create information (like MAX_ROWS)
-    fields		List of fields to create
-    keys		List of keys to create
-    internal_tmp_table  Set to 1 if this is an internal temporary table
-			(From ALTER TABLE)
-    select_field_count
-    is_trans            identifies the type of engine where the table
-                        was created: either trans or non-trans.
-
-  DESCRIPTION
-    If one creates a temporary table, this is automatically opened
-
-    Note that this function assumes that caller already have taken
-    exclusive metadata lock on table being created or used some other
-    way to ensure that concurrent operations won't intervene.
-    mysql_create_table() is a wrapper that can be used for this.
-
-    no_log is needed for the case of CREATE ... SELECT,
-    as the logging will be done later in sql_insert.cc
-    select_field_count is also used for CREATE ... SELECT,
-    and must be zero for standard create of table.
-
-  RETURN VALUES
-    FALSE OK
-    TRUE  error
-*/
-
-bool mysql_create_table_no_lock(THD *thd,
+handler *mysql_create_frm_image(THD *thd,
                                 const char *db, const char *table_name,
                                 HA_CREATE_INFO *create_info,
                                 Alter_info *alter_info,
                                 bool internal_tmp_table,
-                                uint select_field_count,
-                                bool *is_trans)
+                                uint select_field_count, LEX_CUSTRING *frm)
 {
-  char		path[FN_REFLEN + 1];
-  uint          path_length;
-  const char	*alias;
   uint		db_options, key_count;
   KEY		*key_info_buffer;
-  handler	*file;
-  bool		error= TRUE;
-  DBUG_ENTER("mysql_create_table_no_lock");
-  DBUG_PRINT("enter", ("db: '%s'  table: '%s'  tmp: %d",
-                       db, table_name, internal_tmp_table));
-
+  handler       *file;
+  DBUG_ENTER("mysql_create_frm_image");
 
   /* Check for duplicate fields and check type of table to create */
   if (!alter_info->create_list.elements)
   {
     my_message(ER_TABLE_MUST_HAVE_COLUMNS, ER(ER_TABLE_MUST_HAVE_COLUMNS),
                MYF(0));
-    DBUG_RETURN(TRUE);
+    DBUG_RETURN(NULL);
   }
   if (check_engine(thd, db, table_name, create_info))
-    DBUG_RETURN(TRUE);
+    DBUG_RETURN(NULL);
 
   set_table_default_charset(thd, create_info, (char*) db);
 
@@ -4119,12 +4101,11 @@ bool mysql_create_table_no_lock(THD *thd,
       create_info->row_type != ROW_TYPE_FIXED &&
       create_info->row_type != ROW_TYPE_DEFAULT)
     db_options|= HA_OPTION_PACK_RECORD;
-  alias= table_case_name(create_info, table_name);
   if (!(file= get_new_handler((TABLE_SHARE*) 0, thd->mem_root,
                               create_info->db_type)))
   {
     mem_alloc_error(sizeof(handler));
-    DBUG_RETURN(TRUE);
+    DBUG_RETURN(NULL);
   }
 #ifdef WITH_PARTITION_STORAGE_ENGINE
   partition_info *part_info= thd->work_part_info;
@@ -4141,7 +4122,7 @@ bool mysql_create_table_no_lock(THD *thd,
     if (!part_info)
     {
       mem_alloc_error(sizeof(partition_info));
-      DBUG_RETURN(TRUE);
+      goto err;
     }
     file->set_auto_partitions(part_info);
     part_info->default_engine_type= create_info->db_type;
@@ -4166,7 +4147,7 @@ bool mysql_create_table_no_lock(THD *thd,
     char *part_syntax_buf;
     uint syntax_len;
     handlerton *engine_type;
-    if (create_info->options & HA_LEX_CREATE_TMP_TABLE)
+    if (create_info->tmp_table())
     {
       my_error(ER_PARTITION_NO_TEMPORARY, MYF(0));
       goto err;
@@ -4238,9 +4219,8 @@ bool mysql_create_table_no_lock(THD *thd,
       delete file;
       create_info->db_type= partition_hton;
       if (!(file= get_ha_partition(part_info)))
-      {
-        DBUG_RETURN(TRUE);
-      }
+        DBUG_RETURN(NULL);
+
       /*
         If we have default number of partitions or subpartitions we
         might require to set-up the part_info object such that it
@@ -4282,65 +4262,90 @@ bool mysql_create_table_no_lock(THD *thd,
                                   engine_type)))
       {
         mem_alloc_error(sizeof(handler));
-        DBUG_RETURN(TRUE);
+        DBUG_RETURN(NULL);
       }
     }
   }
 #endif
 
   if (mysql_prepare_create_table(thd, create_info, alter_info,
-                                 internal_tmp_table,
-                                 &db_options, file,
+                                 internal_tmp_table, &db_options, file,
                                  &key_info_buffer, &key_count,
                                  select_field_count))
     goto err;
+  create_info->table_options=db_options;
 
-      /* Check if table exists */
-  if (create_info->options & HA_LEX_CREATE_TMP_TABLE)
-  {
-    path_length= build_tmptable_filename(thd, path, sizeof(path));
-    create_info->table_options|=HA_CREATE_DELAY_KEY_WRITE;
-  }
-  else  
-  {
-    path_length= build_table_filename(path, sizeof(path) - 1, db, alias, reg_ext,
-                                      internal_tmp_table ? FN_IS_TMP : 0);
-  }
+  *frm= build_frm_image(thd, table_name, create_info,
+                        alter_info->create_list, key_count,
+                        key_info_buffer, file);
 
-  /* Check if table already exists */
-  if ((create_info->options & HA_LEX_CREATE_TMP_TABLE) &&
-      find_temporary_table(thd, db, table_name))
-  {
-    if (create_info->options & HA_LEX_CREATE_IF_NOT_EXISTS)
-      goto warn;
-    my_error(ER_TABLE_EXISTS_ERROR, MYF(0), alias);
-    goto err;
-  }
+  if (frm->str)
+    DBUG_RETURN(file);
 
-  /* Give warnings for not supported table options */
-#if defined(WITH_ARIA_STORAGE_ENGINE)
-  extern handlerton *maria_hton;
-  if (file->ht != maria_hton)
-#endif
-    if (create_info->transactional)
-      push_warning_printf(thd, MYSQL_ERROR::WARN_LEVEL_WARN,
-                          ER_ILLEGAL_HA_CREATE_OPTION,
-                          ER(ER_ILLEGAL_HA_CREATE_OPTION),
-                          file->engine_name()->str,
-                          "TRANSACTIONAL=1");
+err:
+  delete file;
+  DBUG_RETURN(NULL);
+}
 
-  if (!internal_tmp_table && !(create_info->options & HA_LEX_CREATE_TMP_TABLE))
-  {
-    if (ha_table_exists(thd, db, table_name))
-    {
-      if (create_info->options & HA_LEX_CREATE_IF_NOT_EXISTS)
-        goto warn;
-      my_error(ER_TABLE_EXISTS_ERROR, MYF(0), table_name);
-      goto err;
-    }
-  }
 
-  thd_proc_info(thd, "creating table");
+/*
+  Create a table
+
+  SYNOPSIS
+    mysql_create_table_no_lock()
+    thd			Thread object
+    db			Database
+    table_name		Table name
+    create_info	        Create information (like MAX_ROWS)
+    fields		List of fields to create
+    keys		List of keys to create
+    internal_tmp_table  Set to 1 if this is an internal temporary table
+			(From ALTER TABLE)
+    select_field_count
+    is_trans            identifies the type of engine where the table
+                        was created: either trans or non-trans.
+
+  DESCRIPTION
+    If one creates a temporary table, this is automatically opened
+
+    Note that this function assumes that caller already have taken
+    exclusive metadata lock on table being created or used some other
+    way to ensure that concurrent operations won't intervene.
+    mysql_create_table() is a wrapper that can be used for this.
+
+    select_field_count is also used for CREATE ... SELECT,
+    and must be zero for standard create of table.
+
+  RETURN VALUES
+    FALSE OK
+    TRUE  error
+*/
+
+bool mysql_create_table_no_lock(THD *thd,
+                                const char *db, const char *table_name,
+                                HA_CREATE_INFO *create_info,
+                                Alter_info *alter_info,
+                                bool internal_tmp_table,
+                                uint select_field_count,
+                                bool *is_trans)
+{
+  char		path[FN_REFLEN + 1];
+  uint          path_length;
+  const char	*alias;
+  handler	*file;
+  LEX_CUSTRING  frm= {0,0};
+  bool		error= TRUE;
+  DBUG_ENTER("mysql_create_table_no_lock");
+  DBUG_PRINT("enter", ("db: '%s'  table: '%s'  tmp: %d",
+                       db, table_name, internal_tmp_table));
+
+  alias= table_case_name(create_info, table_name);
+
+  file= mysql_create_frm_image(thd, db, table_name, create_info, alter_info,
+                               internal_tmp_table, select_field_count, &frm);
+
+  if (!file)
+    goto err;
 
 #ifdef HAVE_READLINK
   {
@@ -4402,15 +4407,41 @@ bool mysql_create_table_no_lock(THD *thd,
                           "INDEX DIRECTORY");
     create_info->data_file_name= create_info->index_file_name= 0;
   }
-  create_info->table_options=db_options;
 
-  path[path_length - reg_ext_length]= '\0'; // Remove .frm extension
-  if (rea_create_table(thd, path, db, table_name,
-                       create_info, alter_info->create_list,
-                       key_count, key_info_buffer, file))
+  /* Check if table exists */
+  if (create_info->tmp_table())
+  {
+    path_length= build_tmptable_filename(thd, path, sizeof(path));
+    path[path_length - reg_ext_length]= '\0'; // Remove .frm extension
+
+    if (find_temporary_table(thd, db, table_name))
+    {
+      if (create_info->options & HA_LEX_CREATE_IF_NOT_EXISTS)
+        goto warn;
+      my_error(ER_TABLE_EXISTS_ERROR, MYF(0), alias);
+      goto err;
+    }
+  }
+  else  
+  {
+    path_length= build_table_filename(path, sizeof(path) - 1, db, alias, "",
+                                      internal_tmp_table ? FN_IS_TMP : 0);
+
+    if (!internal_tmp_table && ha_table_exists(thd, db, table_name))
+    {
+      if (create_info->options & HA_LEX_CREATE_IF_NOT_EXISTS)
+        goto warn;
+      my_error(ER_TABLE_EXISTS_ERROR, MYF(0), table_name);
+      goto err;
+    }
+  }
+
+  thd_proc_info(thd, "creating table");
+
+  if (rea_create_table(thd, &frm, path, db, table_name, create_info, file))
     goto err;
 
-  if (create_info->options & HA_LEX_CREATE_TMP_TABLE)
+  if (create_info->tmp_table())
   {
     /*
       Open a table (skipping table cache) and add it into
@@ -4431,7 +4462,7 @@ bool mysql_create_table_no_lock(THD *thd,
     thd->thread_specific_used= TRUE;
   }
 #ifdef WITH_PARTITION_STORAGE_ENGINE
-  else if (part_info && create_info->frm_only)
+  else if (thd->work_part_info && create_info->frm_only)
   {
     /*
       For partitioned tables we can't find some problems with table
@@ -4457,6 +4488,7 @@ bool mysql_create_table_no_lock(THD *thd,
   error= FALSE;
 err:
   thd_proc_info(thd, "After create");
+  my_free(const_cast<uchar*>(frm.str));
   delete file;
   DBUG_RETURN(error);
 
@@ -4468,7 +4500,6 @@ warn:
   goto err;
 }
 
-
 /**
   Implementation of SQLCOM_CREATE_TABLE.
 
@@ -4514,7 +4545,7 @@ bool mysql_create_table(THD *thd, TABLE_LIST *create_table,
   if (!result &&
       (!thd->is_current_stmt_binlog_format_row() ||
        (thd->is_current_stmt_binlog_format_row() &&
-        !(create_info->options & HA_LEX_CREATE_TMP_TABLE))))
+        !(create_info->tmp_table()))))
     result= write_bin_log(thd, TRUE, thd->query(), thd->query_length(), is_trans);
 
 end:
@@ -4727,7 +4758,7 @@ bool mysql_create_like_table(THD* thd, TABLE_LIST* table, TABLE_LIST* src_table,
   local_create_info.options|= create_info->options&HA_LEX_CREATE_IF_NOT_EXISTS;
   /* Replace type of source table with one specified in the statement. */
   local_create_info.options&= ~HA_LEX_CREATE_TMP_TABLE;
-  local_create_info.options|= create_info->options & HA_LEX_CREATE_TMP_TABLE;
+  local_create_info.options|= create_info->tmp_table();
   /* Reset auto-increment counter for the new table. */
   local_create_info.auto_increment_value= 0;
   /*
@@ -4745,7 +4776,7 @@ bool mysql_create_like_table(THD* thd, TABLE_LIST* table, TABLE_LIST* src_table,
     Ensure that we have an exclusive lock on target table if we are creating
     non-temporary table.
   */
-  DBUG_ASSERT((create_info->options & HA_LEX_CREATE_TMP_TABLE) ||
+  DBUG_ASSERT((create_info->tmp_table()) ||
               thd->mdl_context.is_lock_owner(MDL_key::TABLE, table->db,
                                              table->table_name,
                                              MDL_EXCLUSIVE));
@@ -4772,7 +4803,7 @@ bool mysql_create_like_table(THD* thd, TABLE_LIST* table, TABLE_LIST* src_table,
            4    temporary temporary Nothing
            ==== ========= ========= ==============================
     */
-    if (!(create_info->options & HA_LEX_CREATE_TMP_TABLE))
+    if (!(create_info->tmp_table()))
     {
       if (src_table->table->s->tmp_table)               // Case 2
       {
@@ -7227,7 +7258,7 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
 
   DBUG_ASSERT(!(mysql_bin_log.is_open() &&
                 thd->is_current_stmt_binlog_format_row() &&
-                (create_info->options & HA_LEX_CREATE_TMP_TABLE)));
+                (create_info->tmp_table())));
   if (write_bin_log(thd, TRUE, thd->query(), thd->query_length()))
     DBUG_RETURN(TRUE);
 
@@ -7852,7 +7883,7 @@ static bool check_engine(THD *thd, const char *db_name,
                        ha_resolve_storage_engine_name(*new_engine),
                        table_name);
   }
-  if (create_info->options & HA_LEX_CREATE_TMP_TABLE &&
+  if (create_info->tmp_table() &&
       ha_check_storage_engine_flag(*new_engine, HTON_TEMPORARY_NOT_SUPPORTED))
   {
     if (create_info->used_fields & HA_CREATE_USED_ENGINE)
diff --git a/sql/sql_table.h b/sql/sql_table.h
index 4dc28aa4933..016260b6b26 100644
--- a/sql/sql_table.h
+++ b/sql/sql_table.h
@@ -25,6 +25,7 @@ struct TABLE_LIST;
 class THD;
 struct TABLE;
 struct handlerton;
+class handler;
 typedef struct st_ha_check_opt HA_CHECK_OPT;
 struct HA_CREATE_INFO;
 typedef struct st_key KEY;
@@ -140,6 +141,12 @@ bool mysql_create_table_no_lock(THD *thd, const char *db,
                                 Alter_info *alter_info,
                                 bool tmp_table, uint select_field_count,
                                 bool *is_trans);
+handler *mysql_create_frm_image(THD *thd,
+                                const char *db, const char *table_name,
+                                HA_CREATE_INFO *create_info,
+                                Alter_info *alter_info,
+                                bool internal_tmp_table,
+                                uint select_field_count, LEX_CUSTRING *frm);
 bool mysql_prepare_alter_table(THD *thd, TABLE *table,
                                HA_CREATE_INFO *create_info,
                                Alter_info *alter_info);
diff --git a/sql/table.cc b/sql/table.cc
index e47c07f3a7e..d6b109a880a 100644
--- a/sql/table.cc
+++ b/sql/table.cc
@@ -693,9 +693,9 @@ err_not_open:
 
 */
 
-bool TABLE_SHARE::init_from_binary_frm_image(THD *thd, bool write,
-                                             const uchar *frm_image,
-                                             size_t frm_length)
+int TABLE_SHARE::init_from_binary_frm_image(THD *thd, bool write,
+                                            const uchar *frm_image,
+                                            size_t frm_length)
 {
   TABLE_SHARE *share= this;
   uint new_frm_ver, field_pack_length, new_field_pack_flag;
@@ -729,10 +729,11 @@ bool TABLE_SHARE::init_from_binary_frm_image(THD *thd, bool write,
   KEY_PART_INFO *first_key_part= NULL;
   uint ext_key_parts= 0;
   uint first_key_parts= 0;
+  plugin_ref se_plugin= 0;
   keyinfo= &first_keyinfo;
   share->ext_key_parts= 0;
   MEM_ROOT **root_ptr, *old_root;
-  DBUG_ENTER("open_binary_frm");
+  DBUG_ENTER("TABLE_SHARE::init_from_binary_frm_image");
 
   root_ptr= my_pthread_getspecific_ptr(MEM_ROOT**, THR_MALLOC);
   old_root= *root_ptr;
@@ -776,15 +777,13 @@ bool TABLE_SHARE::init_from_binary_frm_image(THD *thd, bool write,
   DBUG_PRINT("info", ("default_part_db_type = %u", frm_image[61]));
 #endif
   legacy_db_type= (enum legacy_db_type) (uint) frm_image[3];
-  DBUG_ASSERT(share->db_plugin == NULL);
   /*
     if the storage engine is dynamic, no point in resolving it by its
     dynamically allocated legacy_db_type. We will resolve it later by name.
   */
   if (legacy_db_type > DB_TYPE_UNKNOWN && 
       legacy_db_type < DB_TYPE_FIRST_DYNAMIC)
-    share->db_plugin= ha_lock_engine(NULL, 
-                                     ha_checktype(thd, legacy_db_type, 0, 0));
+    se_plugin= ha_lock_engine(NULL, ha_checktype(thd, legacy_db_type, 0, 0));
   share->db_create_options= db_create_options= uint2korr(frm_image+30);
   share->db_options_in_use= share->db_create_options;
   share->mysql_version= uint4korr(frm_image+51);
@@ -1045,7 +1044,7 @@ bool TABLE_SHARE::init_from_binary_frm_image(THD *thd, bool write,
       name.length= str_db_type_length;
 
       plugin_ref tmp_plugin= ha_resolve_by_name(thd, &name);
-      if (tmp_plugin != NULL && !plugin_equals(tmp_plugin, share->db_plugin))
+      if (tmp_plugin != NULL && !plugin_equals(tmp_plugin, se_plugin))
       {
         if (legacy_db_type > DB_TYPE_UNKNOWN &&
             legacy_db_type < DB_TYPE_FIRST_DYNAMIC &&
@@ -1057,14 +1056,11 @@ bool TABLE_SHARE::init_from_binary_frm_image(THD *thd, bool write,
         }
         /*
           tmp_plugin is locked with a local lock.
-          we unlock the old value of share->db_plugin before
+          we unlock the old value of se_plugin before
           replacing it with a globally locked version of tmp_plugin
         */
-        plugin_unlock(NULL, share->db_plugin);
-        share->db_plugin= my_plugin_lock(NULL, tmp_plugin);
-        DBUG_PRINT("info", ("setting dbtype to '%.*s' (%d)",
-                            str_db_type_length, next_chunk + 2,
-                            ha_legacy_type(share->db_type())));
+        plugin_unlock(NULL, se_plugin);
+        se_plugin= plugin_lock(NULL, tmp_plugin);
       }
 #ifdef WITH_PARTITION_STORAGE_ENGINE
       else if (str_db_type_length == 9 &&
@@ -1073,7 +1069,7 @@ bool TABLE_SHARE::init_from_binary_frm_image(THD *thd, bool write,
         /*
           Use partition handler
           tmp_plugin is locked with a local lock.
-          we unlock the old value of share->db_plugin before
+          we unlock the old value of se_plugin before
           replacing it with a globally locked version of tmp_plugin
         */
         /* Check if the partitioning engine is ready */
@@ -1083,11 +1079,8 @@ bool TABLE_SHARE::init_from_binary_frm_image(THD *thd, bool write,
                    "--skip-partition");
           goto err;
         }
-        plugin_unlock(NULL, share->db_plugin);
-        share->db_plugin= ha_lock_engine(NULL, partition_hton);
-        DBUG_PRINT("info", ("setting dbtype to '%.*s' (%d)",
-                            str_db_type_length, next_chunk + 2,
-                            ha_legacy_type(share->db_type())));
+        plugin_unlock(NULL, se_plugin);
+        se_plugin= ha_lock_engine(NULL, partition_hton);
       }
 #endif
       else if (!tmp_plugin)
@@ -1284,7 +1277,7 @@ bool TABLE_SHARE::init_from_binary_frm_image(THD *thd, bool write,
 
  /* Allocate handler */
   if (!(handler_file= get_new_handler(share, thd->mem_root,
-                                      share->db_type())))
+                                      plugin_data(se_plugin, handlerton *))))
     goto err;
 
   record= share->default_values-1;              /* Fieldstart = 1 */
@@ -1909,6 +1902,8 @@ bool TABLE_SHARE::init_from_binary_frm_image(THD *thd, bool write,
     (void) my_hash_check(&share->name_hash);
 #endif
 
+  DBUG_ASSERT(!share->db_plugin || plugin_equals(share->db_plugin, se_plugin));
+  share->db_plugin= se_plugin;
   share->error= OPEN_FRM_OK;
   thd->status_var.opened_shares++;
   *root_ptr= old_root;
@@ -1918,6 +1913,7 @@ bool TABLE_SHARE::init_from_binary_frm_image(THD *thd, bool write,
   share->error= OPEN_FRM_CORRUPTED;
   share->open_errno= my_errno;
   delete handler_file;
+  plugin_unlock(0, se_plugin);
   my_hash_free(&share->name_hash);
   if (share->ha_data_destroy)
   {
@@ -1936,9 +1932,128 @@ bool TABLE_SHARE::init_from_binary_frm_image(THD *thd, bool write,
     open_table_error(share, OPEN_FRM_CORRUPTED, share->open_errno);
 
   *root_ptr= old_root;
-  DBUG_RETURN(1);
-} /* open_binary_frm */
+  DBUG_RETURN(HA_ERR_NOT_A_TABLE);
+}
+
+
+static bool sql_unusable_for_discovery(THD *thd, const char *sql)
+{
+  LEX *lex= thd->lex;
+  HA_CREATE_INFO *create_info= &lex->create_info;
+
+  // ... not CREATE TABLE
+  if (lex->sql_command != SQLCOM_CREATE_TABLE)
+    return 1;
+  // ... create like
+  if (create_info->options & HA_LEX_CREATE_TABLE_LIKE)
+    return 1;
+  // ... create select
+  if (lex->select_lex.item_list.elements)
+    return 1;
+  // ... temporary
+  if (create_info->options & HA_LEX_CREATE_TMP_TABLE)
+    return 1;
+  // ... if exists
+  if (create_info->options & HA_LEX_CREATE_IF_NOT_EXISTS)
+    return 1;
+
+  // XXX error out or rather ignore the following:
+  // ... partitioning
+  if (lex->part_info)
+    return 1;
+  // ... union
+  if (create_info->used_fields & HA_CREATE_USED_UNION)
+    return 1;
+  // ... index/data directory
+  if (create_info->data_file_name || create_info->index_file_name)
+    return 1;
+  // ... engine
+  if (create_info->used_fields & HA_CREATE_USED_ENGINE)
+    return 1;
+
+  return 0;
+}
+
+int TABLE_SHARE::init_from_sql_statement_string(THD *thd, bool write,
+                                        const char *sql, size_t sql_length)
+{
+  ulonglong saved_mode= thd->variables.sql_mode;
+  CHARSET_INFO *old_cs= thd->variables.character_set_client;
+  Parser_state parser_state;
+  bool error;
+  char *sql_copy;
+  handler *file;
+  LEX *old_lex;
+  Query_arena *arena, backup;
+  LEX tmp_lex;
+  LEX_CUSTRING frm= {0,0};
+
+  DBUG_ENTER("TABLE_SHARE::init_from_sql_statement_string");
+
+  /*
+    Ouch. Parser may *change* the string it's working on.
+    Currently (2013-02-26) it is used to permanently disable
+    conditional comments.
+    Anyway, let's copy the caller's string...
+  */
+  if (!(sql_copy= thd->strmake(sql, sql_length)))
+    DBUG_RETURN(HA_ERR_OUT_OF_MEM);
+
+  if (parser_state.init(thd, sql_copy, sql_length))
+    DBUG_RETURN(HA_ERR_OUT_OF_MEM);
+
+  thd->variables.sql_mode= MODE_NO_ENGINE_SUBSTITUTION | MODE_NO_DIR_IN_CREATE;
+  thd->variables.character_set_client= system_charset_info;
+  tmp_disable_binlog(thd);
+  old_lex= thd->lex;
+  thd->lex= &tmp_lex;
+
+  arena= thd->stmt_arena;
+  if (arena->is_conventional())
+    arena= 0;
+  else
+    thd->set_n_backup_active_arena(arena, &backup);
+
+  lex_start(thd);
 
+  if ((error= parse_sql(thd, & parser_state, NULL)))
+    goto ret;
+
+  if (sql_unusable_for_discovery(thd, sql_copy))
+  {
+    my_error(ER_SQL_DISCOVER_ERROR, MYF(0), plugin_name(db_plugin)->str,
+             db.str, table_name.str, sql_copy);
+    goto ret;
+  }
+
+  thd->lex->create_info.db_type= plugin_data(db_plugin, handlerton *);
+
+  file= mysql_create_frm_image(thd, db.str, table_name.str,
+                               &thd->lex->create_info, &thd->lex->alter_info,
+                               0, 0, &frm);
+  error|= file == 0;
+  delete file;
+
+  if (frm.str)
+    error= init_from_binary_frm_image(thd, write, frm.str, frm.length);
+
+ret:
+  my_free(const_cast<uchar*>(frm.str));
+  lex_end(thd->lex);
+  thd->lex= old_lex;
+  if (arena)
+    thd->restore_active_arena(arena, &backup);
+  reenable_binlog(thd);
+  thd->variables.sql_mode= saved_mode;
+  thd->variables.character_set_client= old_cs;
+  if (thd->is_error() || error)
+  {
+    thd->clear_error();
+    my_error(ER_NO_SUCH_TABLE, MYF(0), db.str, table_name.str);
+    DBUG_RETURN(HA_ERR_NOT_A_TABLE);
+  }
+  DBUG_RETURN(0);
+}
 
 bool TABLE_SHARE::write_frm_image(const uchar *frm, size_t len)
 {
diff --git a/sql/table.h b/sql/table.h
index 562a49bfdae..a10617cc938 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -989,8 +989,10 @@ struct TABLE_SHARE
   uint actual_n_key_parts(THD *thd);
 
   LEX_CUSTRING *frm_image; ///< only during CREATE TABLE (@sa ha_create_table)
-  bool init_from_binary_frm_image(THD *thd, bool write,
-                                  const uchar *frm_image, size_t frm_length);
+  int init_from_binary_frm_image(THD *thd, bool write,
+                                 const uchar *frm_image, size_t frm_length);
+  int init_from_sql_statement_string(THD *thd, bool write,
+                                     const char *sql, size_t sql_length);
   bool write_frm_image(const uchar *frm_image, size_t frm_length);
   bool read_frm_image(const uchar **frm_image, size_t *frm_length);
 };
diff --git a/sql/unireg.cc b/sql/unireg.cc
index a50ac99ed20..aa60a9e49c8 100644
--- a/sql/unireg.cc
+++ b/sql/unireg.cc
@@ -44,8 +44,6 @@ static uint get_interval_id(uint *,List<Create_field> &, Create_field *);
 static bool pack_fields(uchar *, List<Create_field> &, ulong);
 static size_t packed_fields_length(List<Create_field> &);
 static bool make_empty_rec(THD *, uchar *, uint, List<Create_field> &, uint, ulong);
-static LEX_CUSTRING create_frm_image(THD *, const char *, HA_CREATE_INFO *,
-                              List<Create_field> &, uint, KEY *, handler *);
 
 /*
   Create a frm (table definition) file
@@ -67,27 +65,7 @@ static LEX_CUSTRING create_frm_image(THD *, const char *, HA_CREATE_INFO *,
     true   error
 */
 
-bool mysql_create_frm(THD *thd, const char *file_name,
-                      const char *db, const char *table,
-		      HA_CREATE_INFO *create_info,
-		      List<Create_field> &create_fields,
-		      uint keys, KEY *key_info,
-		      handler *db_file)
-{
-  DBUG_ENTER("mysql_create_frm");
-  LEX_CUSTRING frm= create_frm_image(thd, table, create_info,
-                                     create_fields, keys, key_info, db_file);
-  if (!frm.str)
-    DBUG_RETURN(1);
-
-  int error= writefrm(file_name, db, table, !create_info->tmp_table(),
-                      frm.str, frm.length);
-
-  my_free(const_cast<uchar*>(frm.str));
-  DBUG_RETURN(error);
-}
-
-LEX_CUSTRING create_frm_image(THD *thd, const char *table,
+LEX_CUSTRING build_frm_image(THD *thd, const char *table,
                               HA_CREATE_INFO *create_info,
                               List<Create_field> &create_fields,
                               uint keys, KEY *key_info, handler *db_file)
@@ -104,7 +82,7 @@ LEX_CUSTRING create_frm_image(THD *thd, const char *table,
   int error;
   uchar *frm_ptr, *pos;
   LEX_CUSTRING frm= {0,0};
-  DBUG_ENTER("create_frm_image");
+  DBUG_ENTER("build_frm_image");
 
  /* If fixed row records, we need one bit to check for deleted rows */
   if (!(create_info->table_options & HA_OPTION_PACK_RECORD))
@@ -373,37 +351,30 @@ err:
     1  error
 */
 
-int rea_create_table(THD *thd, const char *path,
-                     const char *db, const char *table_name,
-                     HA_CREATE_INFO *create_info,
-                     List<Create_field> &create_fields,
-                     uint keys, KEY *key_info, handler *file)
+int rea_create_table(THD *thd, LEX_CUSTRING *frm,
+                     const char *path, const char *db, const char *table_name,
+                     HA_CREATE_INFO *create_info, handler *file)
 {
   DBUG_ENTER("rea_create_table");
 
-  LEX_CUSTRING frm= create_frm_image(thd, table_name, create_info,
-                                     create_fields, keys, key_info, file);
-  if (!frm.str)
-    DBUG_RETURN(1);
-
   if (thd->variables.keep_files_on_create)
     create_info->options|= HA_CREATE_KEEP_FILES;
 
   if (create_info->frm_only)
   {
-    if (writefrm(path, db, table_name, 1, frm.str, frm.length))
+    if (writefrm(path, db, table_name, 1, frm->str, frm->length))
       goto err_handler;
   }
   else
   {
     // TODO don't write frm for temp tables
     if (create_info->tmp_table() &&
-        writefrm(path, db, table_name, 0, frm.str, frm.length))
+        writefrm(path, db, table_name, 0, frm->str, frm->length))
       goto err_handler;
 
     if (file->ha_create_partitioning_metadata(path, NULL, CHF_CREATE_FLAG,
                                               create_info) ||
-       ha_create_table(thd, path, db, table_name, create_info, &frm))
+       ha_create_table(thd, path, db, table_name, create_info, frm))
     {
       file->ha_create_partitioning_metadata(path, NULL, CHF_DELETE_FLAG,
                                             create_info);
@@ -411,14 +382,12 @@ int rea_create_table(THD *thd, const char *path,
     }
   }
 
-  my_free(const_cast<uchar*>(frm.str));
   DBUG_RETURN(0);
 
 err_handler:
   char frm_name[FN_REFLEN];
   strxmov(frm_name, path, reg_ext, NullS);
   mysql_file_delete(key_file_frm, frm_name, MYF(0));
-  my_free(const_cast<uchar*>(frm.str));
   DBUG_RETURN(1);
 } /* rea_create_table */
 
diff --git a/sql/unireg.h b/sql/unireg.h
index 6972f90c1db..89dae6b2e97 100644
--- a/sql/unireg.h
+++ b/sql/unireg.h
@@ -167,17 +167,13 @@
 #include "sql_list.h"                           /* List<> */
 #include "field.h"                              /* Create_field */
 
-bool mysql_create_frm(THD *thd, const char *file_name,
-                      const char *db, const char *table,
-		      HA_CREATE_INFO *create_info,
-		      List<Create_field> &create_field,
-		      uint key_count,KEY *key_info,handler *db_type);
-int rea_create_table(THD *thd, const char *path,
-                     const char *db, const char *table_name,
-                     HA_CREATE_INFO *create_info,
-  		     List<Create_field> &create_field,
-                     uint key_count,KEY *key_info,
-                     handler *file);
+int rea_create_table(THD *thd, LEX_CUSTRING *frm,
+                     const char *path, const char *db, const char *table_name,
+                     HA_CREATE_INFO *create_info, handler *file);
+LEX_CUSTRING build_frm_image(THD *thd, const char *table,
+                             HA_CREATE_INFO *create_info,
+                             List<Create_field> &create_fields,
+                             uint keys, KEY *key_info, handler *db_file);
 
 #define FRM_HEADER_SIZE 64
 #define FRM_FORMINFO_SIZE 288