Bug#60111 storage type for table not saved in .frm

 - Add new "format section" in extra data segment with additional table and
   column properties. This was originally introduced in 5.1.20 based MySQL Cluster
 - Remove hardcoded STORAGE DISK for table and instead
   output the real storage format used. Keep both TABLESPACE
   and STORAGE inside same version guard.
 - Implement default version of handler::get_tablespace_name() since tablespace
   is now available in share and it's unnecessary for each handler  to implement.
   (the function could actually be removed totally now).
 - Add test for combinations of TABLESPACE  and STORAGE with CREATE TABLE
   and ALTER TABLE
 - Add test to show that 5.5 now can read a .frm file created by MySQL Cluster
   7.0.22. Although it does not yet show the column level attributes, they are read.

7 files changed, 183 insertions, 39 deletions

diff --git a/sql/handler.cc b/sql/handler.cc
index 9ca1f3a20f8..5f9e07164b8 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -3616,6 +3616,22 @@ void handler::get_dynamic_partition_info(PARTITION_STATS *stat_info,
 }
 
 
+char* handler::get_tablespace_name(THD *thd, char *buff, uint buff_len)
+{
+  char *ts= table->s->tablespace;
+  if (!ts)
+    return NULL;
+
+  if (!buff)
+  {
+    buff= my_strdup(ts, MYF(0));
+    return buff;
+  }
+
+  strnmov(buff, ts, buff_len);
+  return buff;
+}
+
 /****************************************************************************
 ** Some general functions that isn't in the handler class
 ****************************************************************************/
diff --git a/sql/handler.h b/sql/handler.h
index 3cd4acee80d..8ff6a9f8f03 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -1646,8 +1646,7 @@ public:
   { return FALSE; }
   virtual char* get_foreign_key_create_info()
   { return(NULL);}  /* gets foreign key create string from InnoDB */
-  virtual char* get_tablespace_name(THD *thd, char *name, uint name_len)
-  { return(NULL);}  /* gets tablespace name from handler */
+  virtual char* get_tablespace_name(THD *thd, char *name, uint name_len);
   /** used in ALTER TABLE; 1 if changing storage engine is allowed */
   virtual bool can_switch_engines() { return 1; }
   /**
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index abef175ab55..c3dda3072f2 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -1379,17 +1379,24 @@ int store_create_info(THD *thd, TABLE_LIST *table_list, String *packet,
   if (!(thd->variables.sql_mode & MODE_NO_TABLE_OPTIONS) && !foreign_db_mode)
   {
     show_table_options= TRUE;
-    /*
-      Get possible table space definitions and append them
-      to the CREATE TABLE statement
-    */
 
-    if ((for_str= file->get_tablespace_name(thd,0,0)))
+    /* TABLESPACE and STORAGE */
+    if (share->tablespace ||
+        share->default_storage_media != HA_SM_DEFAULT)
     {
-      packet->append(STRING_WITH_LEN(" /*!50100 TABLESPACE "));
-      packet->append(for_str, strlen(for_str));
-      packet->append(STRING_WITH_LEN(" STORAGE DISK */"));
-      my_free(for_str);
+      packet->append(STRING_WITH_LEN(" /*!50100"));
+      if (share->tablespace)
+      {
+        packet->append(STRING_WITH_LEN(" TABLESPACE "));
+        packet->append(share->tablespace, strlen(share->tablespace));
+      }
+
+      if (share->default_storage_media == HA_SM_DISK)
+        packet->append(STRING_WITH_LEN(" STORAGE DISK"));
+      if (share->default_storage_media == HA_SM_MEMORY)
+        packet->append(STRING_WITH_LEN(" STORAGE MEMORY"));
+
+      packet->append(STRING_WITH_LEN(" */"));
     }
 
     /*
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 2855bb0ec4a..21b8b5bf0a0 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -5281,17 +5281,12 @@ mysql_prepare_alter_table(THD *thd, TABLE *table,
   if (!(used_fields & HA_CREATE_USED_KEY_BLOCK_SIZE))
     create_info->key_block_size= table->s->key_block_size;
 
-  if (!create_info->tablespace && create_info->storage_media != HA_SM_MEMORY)
-  {
-    char *tablespace= static_cast<char *>(thd->alloc(FN_LEN + 1));
-    /*
-       Regular alter table of disk stored table (no tablespace/storage change)
-       Copy tablespace name
-    */
-    if (tablespace &&
-        (table->file->get_tablespace_name(thd, tablespace, FN_LEN)))
-      create_info->tablespace= tablespace;
-  }
+  if (!create_info->tablespace)
+    create_info->tablespace= table->s->tablespace;
+
+  if (create_info->storage_media == HA_SM_DEFAULT)
+    create_info->storage_media= table->s->default_storage_media;
+
   restore_record(table, s->default_values);     // Empty record for DEFAULT
   Create_field *def;
 
diff --git a/sql/table.cc b/sql/table.cc
index 0ce7d7427d3..7bdfa65d94b 100644
--- a/sql/table.cc
+++ b/sql/table.cc
@@ -750,6 +750,9 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
   const char **interval_array;
   enum legacy_db_type legacy_db_type;
   my_bitmap_map *bitmaps;
+  uchar *extra_segment_buff= 0;
+  const uint format_section_header_size= 8;
+  uchar *format_section_fields= 0;
   DBUG_ENTER("open_binary_frm");
 
   new_field_pack_flag= head[27];
@@ -942,27 +945,27 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
   if ((n_length= uint4korr(head+55)))
   {
     /* Read extra data segment */
-    uchar *buff, *next_chunk, *buff_end;
+    uchar *next_chunk, *buff_end;
     DBUG_PRINT("info", ("extra segment size is %u bytes", n_length));
-    if (!(next_chunk= buff= (uchar*) my_malloc(n_length, MYF(MY_WME))))
+    if (!(extra_segment_buff= (uchar*) my_malloc(n_length, MYF(MY_WME))))
       goto err;
-    if (mysql_file_pread(file, buff, n_length, record_offset + share->reclength,
+    next_chunk= extra_segment_buff;
+    if (mysql_file_pread(file, extra_segment_buff,
+                         n_length, record_offset + share->reclength,
                          MYF(MY_NABP)))
     {
-      my_free(buff);
       goto err;
     }
-    share->connect_string.length= uint2korr(buff);
+    share->connect_string.length= uint2korr(next_chunk);
     if (!(share->connect_string.str= strmake_root(&share->mem_root,
                                                   (char*) next_chunk + 2,
                                                   share->connect_string.
                                                   length)))
     {
-      my_free(buff);
       goto err;
     }
     next_chunk+= share->connect_string.length + 2;
-    buff_end= buff + n_length;
+    buff_end= extra_segment_buff + n_length;
     if (next_chunk + 2 < buff_end)
     {
       uint str_db_type_length= uint2korr(next_chunk);
@@ -979,7 +982,6 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
                 plugin_data(tmp_plugin, handlerton *)))
         {
           /* bad file, legacy_db_type did not match the name */
-          my_free(buff);
           goto err;
         }
         /*
@@ -1009,7 +1011,6 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
           error= 8;
           my_error(ER_OPTION_PREVENTS_STATEMENT, MYF(0),
                    "--skip-partition");
-          my_free(buff);
           goto err;
         }
         plugin_unlock(NULL, share->db_plugin);
@@ -1025,7 +1026,6 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
         error= 8;
         name.str[name.length]=0;
         my_error(ER_UNKNOWN_STORAGE_ENGINE, MYF(0), name.str);
-        my_free(buff);
         goto err;
         /* purecov: end */
       }
@@ -1042,7 +1042,6 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
               memdup_root(&share->mem_root, next_chunk + 4,
                           partition_info_str_len + 1)))
         {
-          my_free(buff);
           goto err;
         }
       }
@@ -1050,7 +1049,6 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
       if (partition_info_str_len)
       {
         DBUG_PRINT("info", ("WITH_PARTITION_STORAGE_ENGINE is not defined"));
-        my_free(buff);
         goto err;
       }
 #endif
@@ -1088,7 +1086,6 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
         {
           DBUG_PRINT("error",
                      ("fulltext key uses parser that is not defined in .frm"));
-          my_free(buff);
           goto err;
         }
         parser_name.str= (char*) next_chunk;
@@ -1099,7 +1096,6 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
         if (! keyinfo->parser)
         {
           my_error(ER_PLUGIN_IS_NOT_LOADED, MYF(0), parser_name.str);
-          my_free(buff);
           goto err;
         }
       }
@@ -1111,19 +1107,68 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
       {
           DBUG_PRINT("error",
                      ("long table comment is not defined in .frm"));
-          my_free(buff);
           goto err;
       }
       share->comment.length = uint2korr(next_chunk);
       if (! (share->comment.str= strmake_root(&share->mem_root,
              (char*)next_chunk + 2, share->comment.length)))
       {
-          my_free(buff);
           goto err;
       }
       next_chunk+= 2 + share->comment.length;
     }
-    my_free(buff);
+
+    if (next_chunk + format_section_header_size < buff_end)
+    {
+      /*
+        New extra data segment called "format section" with additional
+        table and column properties introduced by MySQL Cluster
+        based on 5.1.20
+
+        Table properties:
+        TABLESPACE <ts> and STORAGE [DISK|MEMORY]
+
+        Column properties:
+        COLUMN_FORMAT [DYNAMIC|FIXED] and STORAGE [DISK|MEMORY]
+      */
+      DBUG_PRINT("info", ("Found format section"));
+
+      /* header */
+      const uint format_section_length= uint2korr(next_chunk);
+      const uint format_section_flags= uint4korr(next_chunk+2);
+      /* 2 bytes unused */
+
+      if (next_chunk + format_section_length > buff_end)
+      {
+        DBUG_PRINT("error", ("format section length too long: %u",
+                             format_section_length));
+        goto err;
+      }
+      DBUG_PRINT("info", ("format_section_length: %u, format_section_flags: %u",
+                          format_section_length, format_section_flags));
+
+      share->default_storage_media=
+        (enum ha_storage_media) (format_section_flags & 0x7);
+
+      /* tablespace */
+      const char *tablespace=
+        (const char*)next_chunk + format_section_header_size;
+      const uint tablespace_length= strlen(tablespace);
+      if (tablespace_length &&
+          !(share->tablespace= strmake_root(&share->mem_root,
+                                            tablespace, tablespace_length+1)))
+      {
+        goto err;
+      }
+      DBUG_PRINT("info", ("tablespace: '%s'",
+                          share->tablespace ? share->tablespace : "<null>"));
+
+      /* pointer to format section for fields */
+      format_section_fields=
+        next_chunk + format_section_header_size + tablespace_length + 1;
+
+      next_chunk+= format_section_length;
+    }
   }
   share->key_block_size= uint2korr(head+62);
 
@@ -1438,6 +1483,18 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
         error= 8; 
         goto err;
       }
+
+    if (format_section_fields)
+    {
+      const uchar field_flags= format_section_fields[i];
+      const uchar field_storage= (field_flags & STORAGE_TYPE_MASK);
+      const uchar field_column_format=
+        ((field_flags >> COLUMN_FORMAT_SHIFT)& STORAGE_TYPE_MASK);
+      DBUG_PRINT("debug", ("field flags: %u, storage: %u, column_format: %u",
+                           field_flags, field_storage, field_column_format));
+      (void)field_storage; /* Reserved by and used in MySQL Cluster */
+      (void)field_column_format; /* Reserved by and used in MySQL Cluster */
+    }
   }
   *field_ptr=0;					// End marker
 
@@ -1705,6 +1762,7 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
   if (use_hash)
     (void) my_hash_check(&share->name_hash);
 #endif
+  my_free(extra_segment_buff);
   DBUG_RETURN (0);
 
  err:
@@ -1712,6 +1770,7 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
   share->open_errno= my_errno;
   share->errarg= errarg;
   my_free(disk_buff);
+  my_free(extra_segment_buff);
   delete crypted;
   delete handler_file;
   my_hash_free(&share->name_hash);
@@ -2687,6 +2746,8 @@ void update_create_info_from_table(HA_CREATE_INFO *create_info, TABLE *table)
   create_info->default_table_charset= share->table_charset;
   create_info->table_charset= 0;
   create_info->comment= share->comment;
+  create_info->storage_media= share->default_storage_media;
+  create_info->tablespace= share->tablespace;
 
   DBUG_VOID_RETURN;
 }
diff --git a/sql/table.h b/sql/table.h
index 39e9ef2e0dd..bd8c66173f4 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -609,6 +609,8 @@ struct TABLE_SHARE
   }
   enum row_type row_type;		/* How rows are stored */
   enum tmp_table_type tmp_table;
+  enum ha_storage_media default_storage_media;
+  char *tablespace;
 
   uint ref_count;                       /* How many TABLE objects uses this */
   uint blob_ptr_size;			/* 4 or 8 */
diff --git a/sql/unireg.cc b/sql/unireg.cc
index 6433b8bc7c2..5a44c5e68f5 100644
--- a/sql/unireg.cc
+++ b/sql/unireg.cc
@@ -124,6 +124,9 @@ bool mysql_create_frm(THD *thd, const char *file_name,
 #endif
   Pack_header_error_handler pack_header_error_handler;
   int error;
+  const uint format_section_header_size= 8;
+  uint format_section_length;
+  uint tablespace_length= 0;
   DBUG_ENTER("mysql_create_frm");
 
   DBUG_ASSERT(*fn_rext((char*)file_name)); // Check .frm extension
@@ -256,6 +259,18 @@ bool mysql_create_frm(THD *thd, const char *file_name,
     forminfo[46]=(uchar) create_info->comment.length;
   }
 
+  /*
+    Add room in extra segment for "format section" with additional
+    table and column properties
+  */
+  if (create_info->tablespace)
+    tablespace_length= strlen(create_info->tablespace);
+  format_section_length=
+    format_section_header_size +
+    tablespace_length + 1 +
+    create_fields.elements;
+  create_info->extra_size+= format_section_length;
+
   if ((file=create_frm(thd, file_name, db, table, reclength, fileinfo,
 		       create_info, keys, key_info)) < 0)
   {
@@ -353,6 +368,55 @@ bool mysql_create_frm(THD *thd, const char *file_name,
       goto err;
   }
 
+  /* "Format section" with additional table and column properties */
+  {
+    uchar *ptr, *format_section_buff;
+    if (!(format_section_buff=(uchar*) my_malloc(format_section_length,
+                                                 MYF(MY_WME))))
+      goto err;
+    ptr= format_section_buff;
+
+    /* header */
+    const uint format_section_flags=
+      create_info->storage_media; // 3 bits
+    const uint format_section_unused= 0;
+    int2store(ptr+0, format_section_length);
+    int4store(ptr+2, format_section_flags);
+    int2store(ptr+6, format_section_unused);
+    ptr+= format_section_header_size;
+
+    /* tablespace name */
+    if (tablespace_length > 0)
+      memcpy(ptr, create_info->tablespace, tablespace_length);
+    ptr+= tablespace_length;
+    *ptr= 0; /* tablespace string terminating zero */
+    ptr++;
+
+    /* column properties  */
+    Create_field *field;
+    List_iterator<Create_field> it(create_fields);
+    while ((field=it++))
+    {
+      const uchar field_storage= 0; /* Used in MySQL Cluster */
+      const uchar field_column_format= 0; /* Used in MySQL Cluster */
+      const uchar field_flags=
+        field_storage + (field_column_format << COLUMN_FORMAT_SHIFT);
+      *ptr= field_flags;
+      ptr++;
+    }
+    DBUG_ASSERT(format_section_buff + format_section_length == ptr);
+
+    if (mysql_file_write(file, format_section_buff,
+                         format_section_length, MYF_RW))
+    {
+      my_free(format_section_buff);
+      goto err;
+    }
+    DBUG_PRINT("info", ("wrote format section, length: %u",
+                        format_section_length));
+    my_free(format_section_buff);
+  }
+
   mysql_file_seek(file, filepos, MY_SEEK_SET, MYF(0));
   if (mysql_file_write(file, forminfo, 288, MYF_RW) ||
       mysql_file_write(file, screen_buff, info_length, MYF_RW) ||