Added support for key_block_size for key and table level (WL#602)

Added support for key_block_size to MyISAM.
Simplify interface to 'new Key' to make it easier to add new key options.
mysqld option --new is used to define where key options are printed.
(In 5.3 we should move all key options to after key part definition to avoid problem with reserved names)
Fixed some compiler warnings and a memory leak in ssl


include/my_base.h:
  Added flag to check if block size for key was secified
include/my_sys.h:
  Added new support function to round up to a power of two
include/myisam.h:
  Rename block_size -> block_size_index to avoid confusion with 'block_size'
include/violite.h:
  Added function to free memory after new_VioSSLAcceptorFd
  (Will rename all vio_ssl functions in a future changeset)
mysql-test/mysql-test-run.pl:
  Don't print port number info when we use --extern
mysql-test/r/myisam.result:
  Added test for key_block_size
mysql-test/t/myisam.test:
  Added test for key_block_size
mysys/mf_keycache.c:
  Simplify code
mysys/my_bit.c:
  Added new support function to round up to a power of two
sql/ha_myisam.cc:
  Tell MyISAM to use the specified key_block_size
  MyISAM also updates the global key_block_size from the used values.
sql/handler.cc:
  Added 'default_key_create_info' to be used as 'dummy' argument to 'new Key'
sql/handler.h:
  Added KEY_CREATE_INFO, to be used as for general options for KEY's
sql/item_func.h:
  Removed compiler warning
sql/lex.h:
  Added new symbol
sql/mysqld.cc:
  Fixed memory leak in ssl (new_VioSSLAcceptorFd)
sql/sql_class.h:
  Change 'new Key' to use KEY_CREATE_INFO instead of 'algoritm', parser, key_page_size.
  This makes it easier to add new key options in the future.
sql/sql_lex.h:
  Added key create options
sql/sql_parse.cc:
  Use new interface to 'new Key'
sql/sql_show.cc:
  Added support for key_block_size
  If --new is used, key options are printed after the key part definition.
sql/sql_table.cc:
  Use new interface to 'new Key'
  Add support for key_block_size
sql/sql_yacc.yy:
  Add support for key_block_size
  Allow key options before and after key_parts (In future they should be always after the key_part defintion)
  Use new interface to 'new Key'
sql/structs.h:
  Added block_size to keys
sql/table.cc:
  Remmeber and read key_block_size for table and key level
sql/table.h:
  Added default key_block_size for table
sql/unireg.cc:
  Remember key_block_size for key
storage/myisam/ft_eval.c:
  Set block_length to 0 to get default key page size
storage/myisam/ft_test1.c:
  Set block_length to 0 to get default key page size
storage/myisam/mi_check.c:
  block_size -> block_size_index
storage/myisam/mi_create.c:
  Added support for block size per key.
  Block size is rounded up to next power of two and enforced between MIN and MAX KEY_BLOCK_LENGTH.
  Align start of key block to start at an even offset of max_key_block_length to ensure key cache works good if all key pages are of same size.
storage/myisam/mi_open.c:
  block_size -> block_size_index
storage/myisam/mi_page.c:
  block_size -> block_size_index
storage/myisam/mi_test1.c:
  Set block_length to 0 to get default key page size
storage/myisam/mi_test2.c:
  Set block_length to 0 to get default key page size
storage/myisam/mi_test3.c:
  Set block_length to 0 to get default key page size
storage/myisam/myisamdef.h:
  block_size -> block_size_index to avoid confusion with 'block_size' in MySQL
  Added block_size as argument to MI_BLOCK_SIZE
  Added missing prototypes to get rid of compiler warnings
storage/myisam/myisampack.c:
  Removed compiler warning
  block_size -> block_size_index
vio/viosslfactories.c:
  Fixed memory leak in ssl (new_VioSSLAcceptorFd)
mysql-test/r/myisam-system.result:
  New BitKeeper file ``mysql-test/r/myisam-system.result''
mysql-test/t/myisam-system.test:
  New BitKeeper file ``mysql-test/t/myisam-system.test''

16 files changed, 232 insertions, 89 deletions

diff --git a/sql/ha_myisam.cc b/sql/ha_myisam.cc
index ec39ee00efc..cb7aa889734 100644
--- a/sql/ha_myisam.cc
+++ b/sql/ha_myisam.cc
@@ -350,6 +350,7 @@ int ha_myisam::open(const char *name, int mode, uint test_if_locked)
     if (table->key_info[i].flags & HA_USES_PARSER)
       file->s->keyinfo[i].parser=
         (struct st_mysql_ftparser *)parser->plugin->info;
+    table->key_info[i].block_size= file->s->keyinfo[i].block_length;
   }
   return (0);
 }
@@ -1368,7 +1369,7 @@ void ha_myisam::info(uint flag)
     sortkey= info.sortkey;
     ref_length= info.reflength;
     share->db_options_in_use= info.options;
-    block_size= myisam_block_size;
+    block_size= myisam_block_size;		/* record block size */
 
     /* Update share */
     if (share->tmp_table == NO_TMP_TABLE)
@@ -1501,6 +1502,8 @@ int ha_myisam::create(const char *name, register TABLE *table_arg,
     keydef[i].key_alg= pos->algorithm == HA_KEY_ALG_UNDEF ? 
       (pos->flags & HA_SPATIAL ? HA_KEY_ALG_RTREE : HA_KEY_ALG_BTREE) :
       pos->algorithm;
+    keydef[i].block_length= pos->block_size;
+
     keydef[i].seg=keyseg;
     keydef[i].keysegs=pos->key_parts;
     for (j=0 ; j < pos->key_parts ; j++)
diff --git a/sql/handler.cc b/sql/handler.cc
index 808dd0841c5..5de25b969a8 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -47,6 +47,8 @@ extern handlerton *sys_table_types[];
 
 #define BITMAP_STACKBUF_SIZE (128/8)
 
+KEY_CREATE_INFO default_key_create_info= { HA_KEY_ALG_UNDEF, 0 };
+
 /* static functions defined in this file */
 
 static handler *create_default(TABLE_SHARE *table);
diff --git a/sql/handler.h b/sql/handler.h
index 090ef1f9f30..8029930ce07 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -280,6 +280,7 @@ enum enum_binlog_command {
 #define HA_CREATE_USED_COMMENT          (1L << 16)
 #define HA_CREATE_USED_PASSWORD         (1L << 17)
 #define HA_CREATE_USED_CONNECTION       (1L << 18)
+#define HA_CREATE_USED_KEY_BLOCK_SIZE   (1L << 19)
 
 typedef ulonglong my_xid; // this line is the same as in log_event.h
 #define MYSQL_XID_PREFIX "MySQLXid"
@@ -653,6 +654,7 @@ typedef struct st_ha_create_information
   ulong table_options;
   ulong avg_row_length;
   ulong used_fields;
+  ulong key_block_size;
   SQL_LIST merge_list;
   handlerton *db_type;
   enum row_type row_type;
@@ -666,6 +668,15 @@ typedef struct st_ha_create_information
   bool store_on_disk;                   /* 1 if table stored on disk */
 } HA_CREATE_INFO;
 
+
+typedef struct st_key_create_information
+{
+  enum ha_key_alg algorithm;
+  ulong block_size;
+  LEX_STRING parser_name;
+} KEY_CREATE_INFO;
+
+
 /*
   Class for maintaining hooks used inside operations on tables such
   as: create table functions, delete table functions, and alter table
@@ -700,6 +711,7 @@ private:
 
 typedef struct st_savepoint SAVEPOINT;
 extern ulong savepoint_alloc_size;
+extern KEY_CREATE_INFO default_key_create_info;
 
 /* Forward declaration for condition pushdown to storage engine */
 typedef class Item COND;
diff --git a/sql/item_func.h b/sql/item_func.h
index ed5924e8fe1..a91d93be8c6 100644
--- a/sql/item_func.h
+++ b/sql/item_func.h
@@ -284,6 +284,7 @@ class Item_func_connection_id :public Item_int_func
   longlong value;
 
 public:
+  Item_func_connection_id() {}
   const char *func_name() const { return "connection_id"; }
   void fix_length_and_dec();
   bool fix_fields(THD *thd, Item **ref);
diff --git a/sql/lex.h b/sql/lex.h
index 171f7a48980..555a68dc388 100644
--- a/sql/lex.h
+++ b/sql/lex.h
@@ -275,6 +275,7 @@ static SYMBOL symbols[] = {
   { "JOIN",		SYM(JOIN_SYM)},
   { "KEY",		SYM(KEY_SYM)},
   { "KEYS",		SYM(KEYS)},
+  { "KEY_BLOCK_SIZE",	SYM(KEY_BLOCK_SIZE)},
   { "KILL",		SYM(KILL_SYM)},
   { "LANGUAGE",         SYM(LANGUAGE_SYM)},
   { "LAST",		SYM(LAST_SYM)},
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index 4de1e95d0f3..df91b0dbe64 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -742,6 +742,7 @@ static void clean_up_mutexes(void);
 static void wait_for_signal_thread_to_end(void);
 static int test_if_case_insensitive(const char *dir_name);
 static void create_pid_file();
+static void end_ssl();
 
 #ifndef EMBEDDED_LIBRARY
 /****************************************************************************
@@ -1217,10 +1218,7 @@ void clean_up(bool print_message)
 #endif
   delete binlog_filter;
   delete rpl_filter;
-#ifdef HAVE_OPENSSL
-  if (ssl_acceptor_fd)
-    my_free((gptr) ssl_acceptor_fd, MYF(MY_ALLOW_ZERO_PTR));
-#endif /* HAVE_OPENSSL */
+  end_ssl();
 #ifdef USE_REGEX
   my_regex_end();
 #endif
@@ -2968,6 +2966,18 @@ static void init_ssl()
 }
 
 
+static void end_ssl()
+{
+#ifdef HAVE_OPENSSL
+  if (ssl_acceptor_fd)
+  {
+    free_vio_ssl_acceptor_fd(ssl_acceptor_fd);
+    ssl_acceptor_fd= 0;
+  }
+#endif /* HAVE_OPENSSL */
+}
+
+
 static int init_server_components()
 {
   DBUG_ENTER("init_server_components");
diff --git a/sql/sql_class.h b/sql/sql_class.h
index fdb70b6c991..937457abc64 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -112,17 +112,16 @@ class Key :public Sql_alloc {
 public:
   enum Keytype { PRIMARY, UNIQUE, MULTIPLE, FULLTEXT, SPATIAL, FOREIGN_KEY};
   enum Keytype type;
-  enum ha_key_alg algorithm;
+  KEY_CREATE_INFO key_info;
   List<key_part_spec> columns;
   const char *name;
   bool generated;
-  LEX_STRING *parser_name;
 
-  Key(enum Keytype type_par, const char *name_arg, enum ha_key_alg alg_par,
-      bool generated_arg, List<key_part_spec> &cols,
-      LEX_STRING *parser_arg= 0)
-    :type(type_par), algorithm(alg_par), columns(cols), name(name_arg),
-    generated(generated_arg), parser_name(parser_arg)
+  Key(enum Keytype type_par, const char *name_arg,
+      KEY_CREATE_INFO *key_info_arg,
+      bool generated_arg, List<key_part_spec> &cols)
+    :type(type_par), key_info(*key_info_arg), columns(cols), name(name_arg),
+    generated(generated_arg)
   {}
   ~Key() {}
   /* Equality comparison of keys (ignoring name) */
@@ -144,7 +143,7 @@ public:
   foreign_key(const char *name_arg, List<key_part_spec> &cols,
 	      Table_ident *table,   List<key_part_spec> &ref_cols,
 	      uint delete_opt_arg, uint update_opt_arg, uint match_opt_arg)
-    :Key(FOREIGN_KEY, name_arg, HA_KEY_ALG_UNDEF, 0, cols),
+    :Key(FOREIGN_KEY, name_arg, &default_key_create_info, 0, cols),
     ref_table(table), ref_columns(cols),
     delete_opt(delete_opt_arg), update_opt(update_opt_arg),
     match_opt(match_opt_arg)
diff --git a/sql/sql_lex.h b/sql/sql_lex.h
index 47fbc685bab..a7f9aa8ac5e 100644
--- a/sql/sql_lex.h
+++ b/sql/sql_lex.h
@@ -840,6 +840,7 @@ typedef struct st_lex
   udf_func udf;
   HA_CHECK_OPT   check_opt;			// check/repair options
   HA_CREATE_INFO create_info;
+  KEY_CREATE_INFO key_info;
   LEX_MASTER_INFO mi;				// used by CHANGE MASTER
   USER_RESOURCES mqh;
   ulong type;
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index ebf4b3fed66..72ce9e8e9a2 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -5973,14 +5973,16 @@ bool add_field_to_list(THD *thd, char *field_name, enum_field_types type,
   if (type_modifier & PRI_KEY_FLAG)
   {
     lex->col_list.push_back(new key_part_spec(field_name,0));
-    lex->key_list.push_back(new Key(Key::PRIMARY, NullS, HA_KEY_ALG_UNDEF,
+    lex->key_list.push_back(new Key(Key::PRIMARY, NullS,
+                                    &default_key_create_info,
 				    0, lex->col_list));
     lex->col_list.empty();
   }
   if (type_modifier & (UNIQUE_FLAG | UNIQUE_KEY_FLAG))
   {
     lex->col_list.push_back(new key_part_spec(field_name,0));
-    lex->key_list.push_back(new Key(Key::UNIQUE, NullS, HA_KEY_ALG_UNDEF, 0,
+    lex->key_list.push_back(new Key(Key::UNIQUE, NullS,
+                                    &default_key_create_info, 0,
 				    lex->col_list));
     lex->col_list.empty();
   }
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index 5e54040f0ae..aecb18b7d0f 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -42,6 +42,9 @@ static TYPELIB grant_types = { sizeof(grant_names)/sizeof(char **),
                                grant_names, NULL};
 #endif
 
+static void store_key_options(THD *thd, String *packet, TABLE *table,
+                              KEY *key_info);
+
 /***************************************************************************
 ** List all table types supported 
 ***************************************************************************/
@@ -929,15 +932,12 @@ store_create_info(THD *thd, TABLE_LIST *table_list, String *packet,
   handler *file= table->file;
   TABLE_SHARE *share= table->s;
   HA_CREATE_INFO create_info;
-  my_bool foreign_db_mode=    (thd->variables.sql_mode & (MODE_POSTGRESQL |
-							  MODE_ORACLE |
-							  MODE_MSSQL |
-							  MODE_DB2 |
-							  MODE_MAXDB |
-							  MODE_ANSI)) != 0;
-  my_bool limited_mysql_mode= (thd->variables.sql_mode &
-			       (MODE_NO_FIELD_OPTIONS | MODE_MYSQL323 |
-				MODE_MYSQL40)) != 0;
+  bool foreign_db_mode=  (thd->variables.sql_mode & (MODE_POSTGRESQL |
+                                                     MODE_ORACLE |
+                                                     MODE_MSSQL |
+                                                     MODE_DB2 |
+                                                     MODE_MAXDB |
+                                                     MODE_ANSI)) != 0;
   DBUG_ENTER("store_create_info");
   DBUG_PRINT("enter",("table: %s", table->s->table_name.str));
 
@@ -1100,22 +1100,12 @@ store_create_info(THD *thd, TABLE_LIST *table_list, String *packet,
     if (!found_primary)
      append_identifier(thd, packet, key_info->name, strlen(key_info->name));
 
-    if (!(thd->variables.sql_mode & MODE_NO_KEY_OPTIONS) &&
-	!limited_mysql_mode && !foreign_db_mode)
-    {
-      if (key_info->algorithm == HA_KEY_ALG_BTREE)
-        packet->append(STRING_WITH_LEN(" USING BTREE"));
-
-      if (key_info->algorithm == HA_KEY_ALG_HASH)
-        packet->append(STRING_WITH_LEN(" USING HASH"));
-
-      // +BAR: send USING only in non-default case: non-spatial rtree
-      if ((key_info->algorithm == HA_KEY_ALG_RTREE) &&
-	  !(key_info->flags & HA_SPATIAL))
-        packet->append(STRING_WITH_LEN(" USING RTREE"));
+#if MYSQL_VERSION_ID < 50300
+    /* Key options moved to after key parts in 5.3.0 */
+    if (!thd->variables.new_mode)
+      store_key_options(thd, packet, table, key_info);
+#endif
 
-      // No need to send USING FULLTEXT, it is sent as FULLTEXT KEY
-    }
     packet->append(STRING_WITH_LEN(" ("));
 
     for (uint j=0 ; j < key_info->key_parts ; j++,key_part++)
@@ -1140,6 +1130,10 @@ store_create_info(THD *thd, TABLE_LIST *table_list, String *packet,
       }
     }
     packet->append(')');
+#if MYSQL_VERSION_ID < 50300
+    if (thd->variables.new_mode)
+#endif
+      store_key_options(thd, packet, table, key_info);
     if (key_info->parser)
     {
       packet->append(" WITH PARSER ", 13);
@@ -1252,6 +1246,12 @@ store_create_info(THD *thd, TABLE_LIST *table_list, String *packet,
       packet->append(STRING_WITH_LEN(" ROW_FORMAT="));
       packet->append(ha_row_type[(uint) share->row_type]);
     }
+    if (table->s->key_block_size)
+    {
+      packet->append(STRING_WITH_LEN(" KEY_BLOCK_SIZE="));
+      end= longlong10_to_str(table->s->key_block_size, buff, 10);
+      packet->append(buff, (uint) (end - buff));
+    }
     table->file->append_create_info(packet);
     if (share->comment && share->comment[0])
     {
@@ -1286,6 +1286,47 @@ store_create_info(THD *thd, TABLE_LIST *table_list, String *packet,
   DBUG_RETURN(0);
 }
 
+
+static void store_key_options(THD *thd, String *packet, TABLE *table,
+                              KEY *key_info)
+{
+  bool limited_mysql_mode= (thd->variables.sql_mode &
+                            (MODE_NO_FIELD_OPTIONS | MODE_MYSQL323 |
+                             MODE_MYSQL40)) != 0;
+  bool foreign_db_mode=  (thd->variables.sql_mode & (MODE_POSTGRESQL |
+                                                     MODE_ORACLE |
+                                                     MODE_MSSQL |
+                                                     MODE_DB2 |
+                                                     MODE_MAXDB |
+                                                     MODE_ANSI)) != 0;
+  char *end, buff[32];
+
+  if (!(thd->variables.sql_mode & MODE_NO_KEY_OPTIONS) &&
+      !limited_mysql_mode && !foreign_db_mode)
+  {
+
+    if (key_info->algorithm == HA_KEY_ALG_BTREE)
+      packet->append(STRING_WITH_LEN(" USING BTREE"));
+
+    if (key_info->algorithm == HA_KEY_ALG_HASH)
+      packet->append(STRING_WITH_LEN(" USING HASH"));
+
+    /* send USING only in non-default case: non-spatial rtree */
+    if ((key_info->algorithm == HA_KEY_ALG_RTREE) &&
+        !(key_info->flags & HA_SPATIAL))
+      packet->append(STRING_WITH_LEN(" USING RTREE"));
+
+    if ((key_info->flags & HA_USES_BLOCK_SIZE) &&
+        table->s->key_block_size != key_info->block_size)
+    {
+      packet->append(STRING_WITH_LEN(" KEY_BLOCK_SIZE="));
+      end= longlong10_to_str(key_info->block_size, buff, 10);
+      packet->append(buff, (uint) (end - buff));
+    }
+  }
+}
+
+
 void
 view_store_options(THD *thd, TABLE_LIST *table, String *buff)
 {
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 412682fc3b0..6225cda3e1c 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -231,10 +231,9 @@ static int mysql_copy_key_list(List<Key> *orig_key,
       }
     }
     if (!(temp_key= new Key(prep_key->type, prep_key->name,
-                            prep_key->algorithm,
+                            &prep_key->key_info,
                             prep_key->generated,
-                            prep_columns,
-                            prep_key->parser_name)))
+                            prep_columns)))
     {
       mem_alloc_error(sizeof(Key));
       DBUG_RETURN(TRUE);
@@ -2495,14 +2494,16 @@ static int mysql_prepare_table(THD *thd, HA_CREATE_INFO *create_info,
 	break;
     }
 
-    switch(key->type){
+    switch (key->type) {
     case Key::MULTIPLE:
 	key_info->flags= 0;
 	break;
     case Key::FULLTEXT:
 	key_info->flags= HA_FULLTEXT;
-	if ((key_info->parser_name= key->parser_name))
+	if ((key_info->parser_name= &key->key_info.parser_name)->str)
           key_info->flags|= HA_USES_PARSER;
+        else
+          key_info->parser_name= 0;
 	break;
     case Key::SPATIAL:
 #ifdef HAVE_SPATIAL
@@ -2526,7 +2527,7 @@ static int mysql_prepare_table(THD *thd, HA_CREATE_INFO *create_info,
     key_info->key_parts=(uint8) key->columns.elements;
     key_info->key_part=key_part_info;
     key_info->usable_key_parts= key_number;
-    key_info->algorithm=key->algorithm;
+    key_info->algorithm= key->key_info.algorithm;
 
     if (key->type == Key::FULLTEXT)
     {
@@ -2572,6 +2573,18 @@ static int mysql_prepare_table(THD *thd, HA_CREATE_INFO *create_info,
 #endif
     }
 
+    /* Take block size from key part or table part */
+    /*
+      TODO: Add warning if block size changes. We can't do it here, as
+      this may depend on the size of the key
+    */
+    key_info->block_size= (key->key_info.block_size ?
+                           key->key_info.block_size :
+                           create_info->key_block_size);
+
+    if (key_info->block_size)
+      key_info->flags|= HA_USES_BLOCK_SIZE;
+
     List_iterator<key_part_spec> cols(key->columns), cols2(key->columns);
     CHARSET_INFO *ft_key_charset=0;  // for FULLTEXT
     for (uint column_nr=0 ; (column=cols++) ; column_nr++)
@@ -5142,6 +5155,8 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
     create_info->avg_row_length= table->s->avg_row_length;
   if (!(used_fields & HA_CREATE_USED_DEFAULT_CHARSET))
     create_info->default_table_charset= table->s->table_charset;
+  if (!(used_fields & HA_CREATE_USED_KEY_BLOCK_SIZE))
+    create_info->key_block_size= table->s->key_block_size;
 
   restore_record(table, s->default_values);     // Empty record for DEFAULT
   List_iterator<Alter_drop> drop_it(alter_info->drop_list);
@@ -5344,6 +5359,16 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
 					    key_part_length));
     }
     if (key_parts.elements)
+    {
+      KEY_CREATE_INFO key_create_info;
+      bzero((char*) &key_create_info, sizeof(key_create_info));
+
+      key_create_info.algorithm= key_info->algorithm;
+      if (key_info->flags & HA_USES_BLOCK_SIZE)
+        key_create_info.block_size= key_info->block_size;
+      if (key_info->flags & HA_USES_PARSER)
+        key_create_info.parser_name= *key_info->parser_name;
+
       key_list.push_back(new Key(key_info->flags & HA_SPATIAL ? Key::SPATIAL :
 				 (key_info->flags & HA_NOSAME ?
 				 (!my_strcasecmp(system_charset_info,
@@ -5352,11 +5377,10 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
 				  (key_info->flags & HA_FULLTEXT ?
 				   Key::FULLTEXT : Key::MULTIPLE)),
 				 key_name,
-				 key_info->algorithm,
+                                 &key_create_info,
                                  test(key_info->flags & HA_GENERATED_KEY),
-				 key_parts,
-                                 key_info->flags & HA_USES_PARSER ?
-                                 &key_info->parser->name : 0));
+				 key_parts));
+    }
   }
   {
     Key *key;
@@ -5452,9 +5476,8 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
       while ((prep_col= prep_col_it++))
         prep_columns.push_back(new key_part_spec(*prep_col));
       prepared_key_list.push_back(new Key(prep_key->type, prep_key->name,
-                                          prep_key->algorithm,
-                                          prep_key->generated, prep_columns,
-                                          prep_key->parser_name));
+                                          &prep_key->key_info,
+                                          prep_key->generated, prep_columns));
     }
 
     /* Create the prepared information. */
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index ab87c78d17c..75a588a2bc4 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -364,6 +364,7 @@ bool my_yyoverflow(short **a, YYSTYPE **b, ulong *yystacksize);
 %token  JOIN_SYM
 %token  KEYS
 %token  KEY_SYM
+%token  KEY_BLOCK_SIZE
 %token  KILL_SYM
 %token  LANGUAGE_SYM
 %token  LAST_INSERT_ID
@@ -730,7 +731,7 @@ bool my_yyoverflow(short **a, YYSTYPE **b, ulong *yystacksize);
         sp_opt_label BIN_NUM label_ident TEXT_STRING_filesystem
 
 %type <lex_str_ptr>
-	opt_table_alias opt_fulltext_parser
+	opt_table_alias
 
 %type <table>
 	table_ident table_ident_nodb references xid
@@ -795,7 +796,7 @@ bool my_yyoverflow(short **a, YYSTYPE **b, ulong *yystacksize);
 	key_type opt_unique_or_fulltext constraint_key_type
 
 %type <key_alg>
-	key_alg opt_btree_or_rtree
+	opt_btree_or_rtree
 
 %type <string_list>
 	key_usage_list using_list
@@ -886,6 +887,7 @@ bool my_yyoverflow(short **a, YYSTYPE **b, ulong *yystacksize);
         view_suid view_tail view_list_opt view_list view_select
         view_check_option trigger_tail sp_tail
         install uninstall partition_entry binlog_base64_event
+	init_key_options key_options key_opts key_opt
 END_OF_INPUT
 
 %type <NONE> call sp_proc_stmts sp_proc_stmts1 sp_proc_stmt
@@ -1220,11 +1222,13 @@ create:
 	}
 	create2
 	  { Lex->current_select= &Lex->select_lex; }
-	| CREATE opt_unique_or_fulltext INDEX_SYM ident key_alg ON table_ident
+	| CREATE opt_unique_or_fulltext INDEX_SYM ident key_alg ON
+	  table_ident
 	  {
 	    LEX *lex=Lex;
 	    lex->sql_command= SQLCOM_CREATE_INDEX;
-	    if (!lex->current_select->add_table_to_list(lex->thd, $7, NULL,
+	    if (!lex->current_select->add_table_to_list(lex->thd, $7,
+							NULL,
 							TL_OPTION_UPDATING))
 	      YYABORT;
 	    lex->create_list.empty();
@@ -1232,15 +1236,16 @@ create:
 	    lex->col_list.empty();
 	    lex->change=NullS;
 	  }
-	   '(' key_list ')' opt_fulltext_parser
+	   '(' key_list ')' key_options
 	  {
 	    LEX *lex=Lex;
-	    if ($2 != Key::FULLTEXT && $12)
+	    if ($2 != Key::FULLTEXT && lex->key_info.parser_name.str)
 	    {
 	      yyerror(ER(ER_SYNTAX_ERROR));
 	      YYABORT;
 	    }
-	    lex->key_list.push_back(new Key($2,$4.str,$5,0,lex->col_list,$12));
+	    lex->key_list.push_back(new Key($2, $4.str, &lex->key_info, 0,
+					   lex->col_list));
 	    lex->col_list.empty();
 	  }
 	| CREATE DATABASE opt_if_not_exists ident
@@ -3890,6 +3895,11 @@ create_table_option:
         | STORAGE_SYM DISK_SYM {Lex->create_info.store_on_disk= TRUE;}
         | STORAGE_SYM MEMORY_SYM {Lex->create_info.store_on_disk= FALSE;}
 	| CONNECTION_SYM opt_equal TEXT_STRING_sys { Lex->create_info.connect_string.str= $3.str; Lex->create_info.connect_string.length= $3.length;  Lex->create_info.used_fields|= HA_CREATE_USED_CONNECTION; }
+	| KEY_BLOCK_SIZE opt_equal ulong_num
+	  {
+	    Lex->create_info.used_fields|= HA_CREATE_USED_KEY_BLOCK_SIZE;
+	    Lex->create_info.key_block_size= $3;
+	  }
         ;
 
 default_charset:
@@ -3983,23 +3993,25 @@ column_def:
 	;
 
 key_def:
-	key_type opt_ident key_alg '(' key_list ')' opt_fulltext_parser
+	key_type opt_ident key_alg '(' key_list ')' key_options
 	  {
 	    LEX *lex=Lex;
-	    if ($1 != Key::FULLTEXT && $7)
+	    if ($1 != Key::FULLTEXT && lex->key_info.parser_name.str)
 	    {
 	      yyerror(ER(ER_SYNTAX_ERROR));
 	      YYABORT;
 	    }
-	    lex->key_list.push_back(new Key($1,$2, $3, 0, lex->col_list, $7));
+	    lex->key_list.push_back(new Key($1,$2, &lex->key_info, 0,
+					   lex->col_list));
 	    lex->col_list.empty();		/* Alloced by sql_alloc */
 	  }
-	| opt_constraint constraint_key_type opt_ident key_alg '(' key_list ')'
+	| opt_constraint constraint_key_type opt_ident key_alg
+	  '(' key_list ')' key_options
 	  {
 	    LEX *lex=Lex;
-	    const char *key_name= $3 ? $3:$1;
-	    lex->key_list.push_back(new Key($2, key_name, $4, 0,
-				    lex->col_list));
+	    const char *key_name= $3 ? $3 : $1;
+	    lex->key_list.push_back(new Key($2, key_name, &lex->key_info, 0,
+					    lex->col_list));
 	    lex->col_list.empty();		/* Alloced by sql_alloc */
 	  }
 	| opt_constraint FOREIGN KEY_SYM opt_ident '(' key_list ')' references
@@ -4012,7 +4024,7 @@ key_def:
 				    lex->fk_update_opt,
 				    lex->fk_match_option));
 	    lex->key_list.push_back(new Key(Key::MULTIPLE, $4 ? $4 : $1,
-					    HA_KEY_ALG_UNDEF, 1,
+					    &default_key_create_info, 1,
 					    lex->col_list));
 	    lex->col_list.empty();		/* Alloced by sql_alloc */
 
@@ -4029,20 +4041,6 @@ key_def:
 	  }
 	;
 
-opt_fulltext_parser:
-        /* empty */               { $$= (LEX_STRING *)0; }
-	| WITH PARSER_SYM IDENT_sys
-          {
-            if (plugin_is_ready(&$3, MYSQL_FTPARSER_PLUGIN))
-              $$= (LEX_STRING *)sql_memdup(&$3, sizeof(LEX_STRING));
-            else
-            {
-              my_error(ER_FUNCTION_NOT_DEFINED, MYF(0), $3.str);
-              YYABORT;
-            }
-          }
-        ;
-
 opt_check_constraint:
 	/* empty */
 	| check_constraint
@@ -4516,10 +4514,50 @@ opt_unique_or_fulltext:
 	  }
         ;
 
+init_key_options:
+	{
+	  Lex->key_info= default_key_create_info;
+	}
+	;
+
+/*
+  For now, key_alg initializies lex->key_info.
+  In the future, when all key options are after key definition,
+  we can remove key_alg and move init_key_options to key_options
+*/
+
 key_alg:
-	/* empty */		   { $$= HA_KEY_ALG_UNDEF; }
-	| USING opt_btree_or_rtree { $$= $2; }
-	| TYPE_SYM opt_btree_or_rtree  { $$= $2; };
+	/* empty */ init_key_options
+	| init_key_options key_opts
+	;
+
+key_options:
+	/* empty */ {}
+	| key_opts
+	;
+
+key_opts:
+	key_opt
+	| key_opts key_opt
+	;
+	
+key_opt:
+	USING opt_btree_or_rtree       { Lex->key_info.algorithm= $2; }
+	| TYPE_SYM opt_btree_or_rtree  { Lex->key_info.algorithm= $2; }
+	| KEY_BLOCK_SIZE opt_equal ulong_num
+	  { Lex->key_info.block_size= $3; }
+	| WITH PARSER_SYM IDENT_sys
+          {
+            if (plugin_is_ready(&$3, MYSQL_FTPARSER_PLUGIN))
+              Lex->key_info.parser_name= $3;
+            else
+            {
+              my_error(ER_FUNCTION_NOT_DEFINED, MYF(0), $3.str);
+              YYABORT;
+            }
+          }
+        ;
+
 
 opt_btree_or_rtree:
 	BTREE_SYM	{ $$= HA_KEY_ALG_BTREE; }
@@ -9348,7 +9386,7 @@ keyword_sp:
 	| ISSUER_SYM		{}
 	| INNOBASE_SYM		{}
 	| INSERT_METHOD		{}
-	| RELAY_THREAD		{}
+	| KEY_BLOCK_SIZE	{}
 	| LAST_SYM		{}
 	| LEAVES                {}
 	| LESS_SYM		{}
@@ -9435,6 +9473,7 @@ keyword_sp:
         | REDUNDANT_SYM         {}
 	| RELAY_LOG_FILE_SYM	{}
 	| RELAY_LOG_POS_SYM	{}
+	| RELAY_THREAD		{}
 	| RELOAD		{}
 	| REORGANIZE_SYM	{}
 	| REPEATABLE_SYM	{}
diff --git a/sql/structs.h b/sql/structs.h
index e369d8ed7e8..72237887514 100644
--- a/sql/structs.h
+++ b/sql/structs.h
@@ -86,6 +86,7 @@ typedef struct st_key {
   uint	key_parts;			/* How many key_parts */
   uint  extra_length;
   uint	usable_key_parts;		/* Should normally be = key_parts */
+  uint  block_size;
   enum  ha_key_alg algorithm;
   /*
     Note that parser is used when the table is opened for use, and
diff --git a/sql/table.cc b/sql/table.cc
index 6ba66569f5c..bacb703a28c 100644
--- a/sql/table.cc
+++ b/sql/table.cc
@@ -535,6 +535,7 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
       keyinfo->key_length= (uint) uint2korr(strpos+2);
       keyinfo->key_parts=  (uint) strpos[4];
       keyinfo->algorithm=  (enum ha_key_alg) strpos[5];
+      keyinfo->block_size= uint2korr(strpos+6);
       strpos+=8;
     }
     else
@@ -706,6 +707,7 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
     }
     my_free(buff, MYF(0));
   }
+  share->key_block_size= uint2korr(head+62);
 
   error=4;
   extra_rec_buf_length= uint2korr(head+59);
@@ -2065,6 +2067,11 @@ File create_frm(THD *thd, const char *name, const char *db,
     tmp= MYSQL_VERSION_ID;          // Store to avoid warning from int4store
     int4store(fileinfo+51, tmp);
     int4store(fileinfo+55, create_info->extra_size);
+    /*
+      59-60 is reserved for extra_rec_buf_length,
+      61 for default_part_db_type
+    */
+    int2store(fileinfo+62, create_info->key_block_size);
     bzero(fill,IO_SIZE);
     for (; length > IO_SIZE ; length-= IO_SIZE)
     {
diff --git a/sql/table.h b/sql/table.h
index 85d49444b29..5fd9cd28585 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -160,6 +160,7 @@ typedef struct st_table_share
   uint ref_count;                       /* How many TABLE objects uses this */
   uint open_count;			/* Number of tables in open list */
   uint blob_ptr_size;			/* 4 or 8 */
+  uint key_block_size;			/* create key_block_size, if used */
   uint null_bytes, last_null_bit_pos;
   uint fields;				/* Number of fields */
   uint rec_buff_length;                 /* Size of table->record[] buffer */
diff --git a/sql/unireg.cc b/sql/unireg.cc
index bbb4d970d37..eb38e6c0592 100644
--- a/sql/unireg.cc
+++ b/sql/unireg.cc
@@ -338,9 +338,9 @@ err_handler:
 
 	/* Pack screens to a screen for save in a form-file */
 
-static uchar * pack_screens(List<create_field> &create_fields,
-			    uint *info_length, uint *screens,
-			    bool small_file)
+static uchar *pack_screens(List<create_field> &create_fields,
+                           uint *info_length, uint *screens,
+                           bool small_file)
 {
   reg1 uint i;
   uint row,start_row,end_row,fields_on_screen;
@@ -431,7 +431,7 @@ static uint pack_keys(uchar *keybuff, uint key_count, KEY *keyinfo,
     int2store(pos+2,key->key_length);
     pos[4]= (uchar) key->key_parts;
     pos[5]= (uchar) key->algorithm;
-    pos[6]=pos[7]=0;				// For the future
+    int2store(pos+6, key->block_size);
     pos+=8;
     key_parts+=key->key_parts;
     DBUG_PRINT("loop",("flags: %d  key_parts: %d at 0x%lx",