Store and read engine type string in extra block of .frm.

sql/handler.cc:
  Use my_strnncoll instead of my_strcasecmp (it is possible to
  compare non-asciiz strings now).
sql/handler.h:
  extra block size added to HA_CREATE_INFO.

sql/handler.cc

@@ -183,15 +183,18 @@ enum db_type ha_resolve_by_name(const char *name, uint namelen)
   THD *thd= current_thd;
   show_table_alias_st *table_alias;
   handlerton **types;
-  const char *ptr= name;
 
-  if (thd && !my_strcasecmp(&my_charset_latin1, ptr, "DEFAULT"))
+  if (thd && !my_strnncoll(&my_charset_latin1,
+        (const uchar *)name, namelen,
+        (const uchar *)"DEFAULT", 7))
     return (enum db_type) thd->variables.table_type;
 
 retest:
   for (types= sys_table_types; *types; types++)
   {
-    if (!my_strcasecmp(&my_charset_latin1, ptr, (*types)->name))
+    if (!my_strnncoll(&my_charset_latin1,
+          (const uchar *)name, namelen,
+          (const uchar *)(*types)->name, strlen((*types)->name)))
       return (enum db_type) (*types)->db_type;
   }
 
@@ -200,9 +203,12 @@ retest:
   */
   for (table_alias= sys_table_aliases; table_alias->type; table_alias++)
   {
-    if (!my_strcasecmp(&my_charset_latin1, ptr, table_alias->alias))
+    if (!my_strnncoll(&my_charset_latin1,
+          (const uchar *)name, namelen,
+          (const uchar *)table_alias->alias, strlen(table_alias->alias)))
     {
-      ptr= table_alias->type;
+      name= table_alias->type;
+      namelen= strlen(name);
       goto retest;
     }
   }

sql/handler.h

@@ -436,6 +436,7 @@ typedef struct st_ha_create_information
   uint options;				/* OR of HA_CREATE_ options */
   uint raid_type,raid_chunks;
   uint merge_insert_method;
+  uint extra_size;                      /* length of extra data segment */
   bool table_existed;			/* 1 in create if table existed */
   bool frm_only;                        /* 1 if no ha_create_table() */
   bool varchar;                         /* 1 if table has a VARCHAR */

sql/table.cc

@@ -300,6 +300,44 @@ int openfrm(THD *thd, const char *name, const char *alias, uint db_stat,
   }
 #endif
 
+  record_offset= (ulong) (uint2korr(head+6)+
+                          ((uint2korr(head+14) == 0xffff ?
+                            uint4korr(head+47) : uint2korr(head+14))));
+ 
+  if ((n_length= uint2korr(head+55)))
+  {
+    /* Read extra data segment */
+    char *buff, *next_chunk, *buff_end;
+    if (!(next_chunk= buff= my_malloc(n_length, MYF(MY_WME))))
+      goto err;
+    if (my_pread(file, (byte*)buff, n_length, record_offset + share->reclength,
+                 MYF(MY_NABP)))
+    {
+      my_free(buff, MYF(0));
+      goto err;
+    }
+    if (share->db_type == DB_TYPE_FEDERATED_DB)
+    {
+      share->connect_string.length= uint2korr(buff);
+      if (! (share->connect_string.str= strmake_root(&outparam->mem_root,
+              next_chunk + 2, share->connect_string.length)))
+      {
+        my_free(buff, MYF(0));
+        goto err;
+      }
+      next_chunk+= share->connect_string.length + 2;
+    }
+    buff_end= buff + n_length;
+    if (next_chunk + 2 < buff_end)
+    {
+      uint str_db_type_length= uint2korr(next_chunk);
+      share->db_type= ha_resolve_by_name(next_chunk + 2, str_db_type_length);
+      DBUG_PRINT("enter", ("Setting dbtype to: %d - %d - '%.*s'\n", share->db_type,
+            str_db_type_length, str_db_type_length, next_chunk + 2));
+      next_chunk+= str_db_type_length + 2;
+    }
+    my_free(buff, MYF(0));
+  }
   /* Allocate handler */
   if (!(outparam->file= get_new_handler(outparam, share->db_type)))
     goto err;
@@ -322,9 +360,6 @@ int openfrm(THD *thd, const char *name, const char *alias, uint db_stat,
     goto err;                                   /* purecov: inspected */
   share->default_values= (byte *) record;
 
-  record_offset= (ulong) (uint2korr(head+6)+
-                          ((uint2korr(head+14) == 0xffff ?
-                            uint4korr(head+47) : uint2korr(head+14))));
   if (my_pread(file,(byte*) record, (uint) share->reclength,
                record_offset, MYF(MY_NABP)))
     goto err; /* purecov: inspected */
@@ -342,20 +377,7 @@ int openfrm(THD *thd, const char *name, const char *alias, uint db_stat,
     else
       outparam->record[1]= outparam->record[0];   // Safety
   }
-
+ 
-  if ((n_length= uint2korr(head+55)))
-  {
-    /* Read extra block information */
-    char *buff;
-    if (!(buff= alloc_root(&outparam->mem_root, n_length)))
-      goto err;
-    if (my_pread(file, (byte*)buff, n_length, record_offset + share->reclength,
-                 MYF(MY_NABP)))
-      goto err;
-    share->connect_string.length= uint2korr(buff);
-    share->connect_string.str= buff+2;
-  }
-
 #ifdef HAVE_purify
   /*
     We need this because when we read var-length rows, we are not updating
@@ -1371,15 +1393,10 @@ File create_frm(THD *thd, my_string name, const char *db,
   ulong length;
   char fill[IO_SIZE];
   int create_flags= O_RDWR | O_TRUNC;
-  uint extra_size;
 
   if (create_info->options & HA_LEX_CREATE_TMP_TABLE)
     create_flags|= O_EXCL | O_NOFOLLOW;
 
-  extra_size= 0;
-  if (create_info->connect_string.length)
-    extra_size= 2+create_info->connect_string.length;
-
 #if SIZEOF_OFF_T > 4
   /* Fix this when we have new .frm files;  Current limit is 4G rows (QQ) */
   if (create_info->max_rows > ~(ulong) 0)
@@ -1407,7 +1424,8 @@ File create_frm(THD *thd, my_string name, const char *db,
     fileinfo[4]=1;
     int2store(fileinfo+6,IO_SIZE);		/* Next block starts here */
     key_length=keys*(7+NAME_LEN+MAX_REF_PARTS*9)+16;
-    length= next_io_size((ulong) (IO_SIZE+key_length+reclength+extra_size));
+    length= next_io_size((ulong) (IO_SIZE+key_length+reclength+
+                                  create_info->extra_size));
     int4store(fileinfo+10,length);
     tmp_key_length= (key_length < 0xffff) ? key_length : 0xffff;
     int2store(fileinfo+14,tmp_key_length);
@@ -1429,7 +1447,7 @@ File create_frm(THD *thd, my_string name, const char *db,
     int4store(fileinfo+47, key_length);
     tmp= MYSQL_VERSION_ID;          // Store to avoid warning from int4store
     int4store(fileinfo+51, tmp);
-    int2store(fileinfo+55, extra_size);
+    int2store(fileinfo+55, create_info->extra_size);
     bzero(fill,IO_SIZE);
     for (; length > IO_SIZE ; length-= IO_SIZE)
     {

sql/unireg.cc

@@ -75,6 +75,7 @@ bool mysql_create_frm(THD *thd, my_string file_name,
 		      uint keys, KEY *key_info,
 		      handler *db_file)
 {
+  LEX_STRING str_db_type;
   uint reclength,info_length,screens,key_info_length,maxlength;
   ulong key_buff_length;
   File file;
@@ -82,6 +83,7 @@ bool mysql_create_frm(THD *thd, my_string file_name,
   uchar fileinfo[64],forminfo[288],*keybuff;
   TYPELIB formnames;
   uchar *screen_buff;
+  char buff[2];
   DBUG_ENTER("mysql_create_frm");
 
   formnames.type_names=0;
@@ -116,6 +118,13 @@ bool mysql_create_frm(THD *thd, my_string file_name,
   }
   reclength=uint2korr(forminfo+266);
 
+  /* Calculate extra data segment length */
+  str_db_type.str= (char *)ha_get_storage_engine(create_info->db_type);
+  str_db_type.length= strlen(str_db_type.str);
+  create_info->extra_size= 2 + str_db_type.length;
+  if (create_info->db_type == DB_TYPE_FEDERATED_DB)
+    create_info->extra_size+= create_info->connect_string.length + 2;
+
   if ((file=create_frm(thd, file_name, db, table, reclength, fileinfo,
 		       create_info, keys)) < 0)
   {
@@ -149,16 +158,21 @@ bool mysql_create_frm(THD *thd, my_string file_name,
   if (make_empty_rec(thd,file,create_info->db_type,create_info->table_options,
 		     create_fields,reclength, data_offset))
     goto err;
-  if (create_info->connect_string.length)
+
+  if (create_info->db_type == DB_TYPE_FEDERATED_DB)
   {
-    char buff[2];
+    int2store(buff, create_info->connect_string.length);
-    int2store(buff,create_info->connect_string.length);
     if (my_write(file, (const byte*)buff, sizeof(buff), MYF(MY_NABP)) ||
         my_write(file, (const byte*)create_info->connect_string.str,
                  create_info->connect_string.length, MYF(MY_NABP)))
       goto err;
   }
-
+  int2store(buff, str_db_type.length);
+  if (my_write(file, (const byte*)buff, sizeof(buff), MYF(MY_NABP)) ||
+      my_write(file, (const byte*)str_db_type.str,
+               str_db_type.length, MYF(MY_NABP)))
+    goto err;
+ 
   VOID(my_seek(file,filepos,MY_SEEK_SET,MYF(0)));
   if (my_write(file,(byte*) forminfo,288,MYF_RW) ||
       my_write(file,(byte*) screen_buff,info_length,MYF_RW) ||