Manual merge of mysql-5.1-bugteam into mysql-trunk-merge.

18 files changed, 239 insertions, 101 deletions

diff --git a/sql/item.h b/sql/item.h
index 24d800300e7..060958338e5 100644
--- a/sql/item.h
+++ b/sql/item.h
@@ -965,6 +965,15 @@ public:
   virtual Item *equal_fields_propagator(uchar * arg) { return this; }
   virtual bool set_no_const_sub(uchar *arg) { return FALSE; }
   virtual Item *replace_equal_field(uchar * arg) { return this; }
+  /*
+    Check if an expression value depends on the current timezone. Used by
+    partitioning code to reject timezone-dependent expressions in a
+    (sub)partitioning function.
+  */
+  virtual bool is_timezone_dependent_processor(uchar *bool_arg)
+  {
+    return FALSE;
+  }
 
   /*
     For SP local variable returns pointer to Item representing its
diff --git a/sql/item_func.h b/sql/item_func.h
index f22bc0c301c..e3690232904 100644
--- a/sql/item_func.h
+++ b/sql/item_func.h
@@ -192,6 +192,29 @@ public:
     null_value=1;
     return 0.0;
   }
+  bool has_timestamp_args()
+  {
+    DBUG_ASSERT(fixed == TRUE);
+    for (uint i= 0; i < arg_count; i++)
+    {
+      if (args[i]->type() == Item::FIELD_ITEM &&
+          args[i]->field_type() == MYSQL_TYPE_TIMESTAMP)
+        return TRUE;
+    }
+    return FALSE;
+  }
+  /*
+    We assume the result of any function that has a TIMESTAMP argument to be
+    timezone-dependent, since a TIMESTAMP value in both numeric and string
+    contexts is interpreted according to the current timezone.
+    The only exception is UNIX_TIMESTAMP() which returns the internal
+    representation of a TIMESTAMP argument verbatim, and thus does not depend on
+    the timezone.
+   */
+  virtual bool is_timezone_dependent_processor(uchar *bool_arg)
+  {
+    return has_timestamp_args();
+  }
 };
 
 
diff --git a/sql/item_timefunc.h b/sql/item_timefunc.h
index cdd74c8c601..b25812299f0 100644
--- a/sql/item_timefunc.h
+++ b/sql/item_timefunc.h
@@ -326,6 +326,16 @@ public:
   Item_func_unix_timestamp(Item *a) :Item_int_func(a) {}
   longlong val_int();
   const char *func_name() const { return "unix_timestamp"; }
+  bool check_partition_func_processor(uchar *int_arg) {return FALSE;}
+  /*
+    UNIX_TIMESTAMP() depends on the current timezone
+    (and thus may not be used as a partitioning function)
+    when its argument is NOT of the TIMESTAMP type.
+  */
+  bool is_timezone_dependent_processor(uchar *int_arg)
+  {
+    return !has_timestamp_args();
+  }
   void fix_length_and_dec()
   {
     decimals=0;
diff --git a/sql/log.cc b/sql/log.cc
index 7578d391782..5215dcb9a84 100644
--- a/sql/log.cc
+++ b/sql/log.cc
@@ -4511,6 +4511,9 @@ bool general_log_write(THD *thd, enum enum_server_command command,
 
 void MYSQL_BIN_LOG::rotate_and_purge(uint flags)
 {
+#ifdef HAVE_REPLICATION
+  bool check_purge= false;
+#endif
   if (!(flags & RP_LOCK_LOG_IS_ALREADY_LOCKED))
     pthread_mutex_lock(&LOCK_log);
   if ((flags & RP_FORCE_ROTATE) ||
@@ -4518,16 +4521,24 @@ void MYSQL_BIN_LOG::rotate_and_purge(uint flags)
   {
     new_file_without_locking();
 #ifdef HAVE_REPLICATION
-    if (expire_logs_days)
-    {
-      time_t purge_time= my_time(0) - expire_logs_days*24*60*60;
-      if (purge_time >= 0)
-        purge_logs_before_date(purge_time);
-    }
+    check_purge= true;
 #endif
   }
   if (!(flags & RP_LOCK_LOG_IS_ALREADY_LOCKED))
     pthread_mutex_unlock(&LOCK_log);
+
+#ifdef HAVE_REPLICATION
+  /*
+    NOTE: Run purge_logs wo/ holding LOCK_log
+          as it otherwise will deadlock in ndbcluster_binlog_index_purge_file
+  */
+  if (check_purge && expire_logs_days)
+  {
+    time_t purge_time= my_time(0) - expire_logs_days*24*60*60;
+    if (purge_time >= 0)
+      purge_logs_before_date(purge_time);
+  }
+#endif
 }
 
 uint MYSQL_BIN_LOG::next_file_id()
diff --git a/sql/log_event.cc b/sql/log_event.cc
index d1f78f4c49a..8fda9d828bd 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -4039,6 +4039,7 @@ uint Load_log_event::get_query_buffer_length()
   return
     5 + db_len + 3 +                        // "use DB; "
     18 + fname_len + 2 +                    // "LOAD DATA INFILE 'file''"
+    11 +                                    // "CONCURRENT "
     7 +					    // LOCAL
     9 +                                     // " REPLACE or IGNORE "
     13 + table_name_len*2 +                 // "INTO TABLE `table`"
@@ -4066,6 +4067,9 @@ void Load_log_event::print_query(bool need_db, const char *cs, char *buf,
 
   pos= strmov(pos, "LOAD DATA ");
 
+  if (thd->lex->lock_option == TL_WRITE_CONCURRENT_INSERT)
+    pos= strmov(pos, "CONCURRENT ");
+
   if (fn_start)
     *fn_start= pos;
 
diff --git a/sql/log_event.h b/sql/log_event.h
index cd5e659c910..1fdd7a05968 100644
--- a/sql/log_event.h
+++ b/sql/log_event.h
@@ -1787,7 +1787,7 @@ private:
 
   @verbatim
    (1)    USE db;
-   (2)    LOAD DATA [LOCAL] INFILE 'file_name'
+   (2)    LOAD DATA [CONCURRENT] [LOCAL] INFILE 'file_name'
    (3)    [REPLACE | IGNORE]
    (4)    INTO TABLE 'table_name'
    (5)    [FIELDS
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index 79d930cafe5..560db3356d6 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -1312,7 +1312,6 @@ void clean_up(bool print_message)
   lex_free();				/* Free some memory */
   item_create_cleanup();
   set_var_free();
-  free_charsets();
   if (!opt_noacl)
   {
 #ifdef HAVE_DLOPEN
diff --git a/sql/rpl_rli.cc b/sql/rpl_rli.cc
index 269ae0e6d68..4bbafa0253a 100644
--- a/sql/rpl_rli.cc
+++ b/sql/rpl_rli.cc
@@ -1029,7 +1029,7 @@ err:
      false - condition not met
 */
 
-bool Relay_log_info::is_until_satisfied(my_off_t master_beg_pos)
+bool Relay_log_info::is_until_satisfied(THD *thd, Log_event *ev)
 {
   const char *log_name;
   ulonglong log_pos;
@@ -1039,8 +1039,12 @@ bool Relay_log_info::is_until_satisfied(my_off_t master_beg_pos)
 
   if (until_condition == UNTIL_MASTER_POS)
   {
+    if (ev && ev->server_id == (uint32) ::server_id && !replicate_same_server_id)
+      DBUG_RETURN(FALSE);
     log_name= group_master_log_name;
-    log_pos= master_beg_pos;
+    log_pos= (!ev)? group_master_log_pos :
+      ((thd->options & OPTION_BEGIN || !ev->log_pos) ?
+       group_master_log_pos : ev->log_pos - ev->data_written);
   }
   else
   { /* until_condition == UNTIL_RELAY_POS */
diff --git a/sql/rpl_rli.h b/sql/rpl_rli.h
index fd36d18adae..60b55533926 100644
--- a/sql/rpl_rli.h
+++ b/sql/rpl_rli.h
@@ -316,7 +316,7 @@ public:
   void close_temporary_tables();
 
   /* Check if UNTIL condition is satisfied. See slave.cc for more. */
-  bool is_until_satisfied(my_off_t master_beg_pos);
+  bool is_until_satisfied(THD *thd, Log_event *ev);
   inline ulonglong until_pos()
   {
     return ((until_condition == UNTIL_MASTER_POS) ? group_master_log_pos :
diff --git a/sql/share/errmsg-utf8.txt b/sql/share/errmsg-utf8.txt
index 4260efdeb56..03d7ff30f7e 100644
--- a/sql/share/errmsg-utf8.txt
+++ b/sql/share/errmsg-utf8.txt
@@ -5690,8 +5690,8 @@ ER_PARTITION_WRONG_NO_SUBPART_ERROR
         eng "Wrong number of subpartitions defined, mismatch with previous setting"
         ger "Falsche Anzahl von Unterpartitionen definiert, stimmt nicht mit vorherigen Einstellungen überein"
         swe "Antal subpartitioner definierade och antal subpartitioner är inte lika"
-ER_CONST_EXPR_IN_PARTITION_FUNC_ERROR
-        eng "Constant/Random expression in (sub)partitioning function is not allowed"
+ER_WRONG_EXPR_IN_PARTITION_FUNC_ERROR
+        eng "Constant, random or timezone-dependent expressions in (sub)partitioning function are not allowed"
         ger "Konstante oder Random-Ausdrücke in (Unter-)Partitionsfunktionen sind nicht erlaubt"
         swe "Konstanta uttryck eller slumpmässiga uttryck är inte tillåtna (sub)partitioneringsfunktioner"
 ER_NO_CONST_EXPR_IN_RANGE_OR_LIST_ERROR
diff --git a/sql/share/errmsg.txt b/sql/share/errmsg.txt
index 034f987e0e7..ea2ce2b58b0 100644
--- a/sql/share/errmsg.txt
+++ b/sql/share/errmsg.txt
@@ -5690,8 +5690,8 @@ ER_PARTITION_WRONG_NO_SUBPART_ERROR
         eng "Wrong number of subpartitions defined, mismatch with previous setting"
         ger "Falsche Anzahl von Unterpartitionen definiert, stimmt nicht mit vorherigen Einstellungen überein"
         swe "Antal subpartitioner definierade och antal subpartitioner är inte lika"
-ER_CONST_EXPR_IN_PARTITION_FUNC_ERROR
-        eng "Constant/Random expression in (sub)partitioning function is not allowed"
+ER_WRONG_EXPR_IN_PARTITION_FUNC_ERROR
+        eng "Constant, random or timezone-dependent expressions in (sub)partitioning function are not allowed"
         ger "Konstante oder Random-Ausdrücke in (Unter-)Partitionsfunktionen sind nicht erlaubt"
         swe "Konstanta uttryck eller slumpmässiga uttryck är inte tillåtna (sub)partitioneringsfunktioner"
 ER_NO_CONST_EXPR_IN_RANGE_OR_LIST_ERROR
diff --git a/sql/slave.cc b/sql/slave.cc
index 8be17860c61..27f87d18800 100644
--- a/sql/slave.cc
+++ b/sql/slave.cc
@@ -2559,9 +2559,7 @@ static int exec_relay_log_event(THD* thd, Relay_log_info* rli)
       hits the UNTIL barrier.
     */
     if (rli->until_condition != Relay_log_info::UNTIL_NONE &&
-        rli->is_until_satisfied((rli->is_in_group() || !ev->log_pos) ?
-                                rli->group_master_log_pos :
-                                ev->log_pos - ev->data_written))
+        rli->is_until_satisfied(thd, ev))
     {
       char buf[22];
       sql_print_information("Slave SQL thread stopped because it reached its"
@@ -3348,7 +3346,7 @@ log '%s' at position %s, relay log '%s' position: %s", RPL_LOG_NAME,
   */
   pthread_mutex_lock(&rli->data_lock);
   if (rli->until_condition != Relay_log_info::UNTIL_NONE &&
-      rli->is_until_satisfied(rli->group_master_log_pos))
+      rli->is_until_satisfied(thd, NULL))
   {
     char buf[22];
     sql_print_information("Slave SQL thread stopped because it reached its"
diff --git a/sql/sql_partition.cc b/sql/sql_partition.cc
index c0961f84feb..3ecc1c28288 100644
--- a/sql/sql_partition.cc
+++ b/sql/sql_partition.cc
@@ -947,6 +947,8 @@ int check_signed_flag(partition_info *part_info)
     table                The table object
     part_info            Reference to partitioning data structure
     is_sub_part          Is the table subpartitioned as well
+    is_create_table_ind  Indicator of whether openfrm was called as part of
+                         CREATE or ALTER TABLE
 
   RETURN VALUE
     TRUE                 An error occurred, something was wrong with the
@@ -970,7 +972,7 @@ int check_signed_flag(partition_info *part_info)
 */
 
 static bool fix_fields_part_func(THD *thd, Item* func_expr, TABLE *table,
-                                 bool is_sub_part)
+                          bool is_sub_part, bool is_create_table_ind)
 {
   partition_info *part_info= table->part_info;
   uint dir_length, home_dir_length;
@@ -1074,10 +1076,31 @@ static bool fix_fields_part_func(THD *thd, Item* func_expr, TABLE *table,
   thd->where= save_where;
   if (unlikely(func_expr->const_item()))
   {
-    my_error(ER_CONST_EXPR_IN_PARTITION_FUNC_ERROR, MYF(0));
+    my_error(ER_WRONG_EXPR_IN_PARTITION_FUNC_ERROR, MYF(0));
     clear_field_flag(table);
     goto end;
   }
+
+  /*
+    We don't allow creating partitions with timezone-dependent expressions as
+    a (sub)partitioning function, but we want to allow such expressions when
+    opening existing tables for easier maintenance. This exception should be
+    deprecated at some point in future so that we always throw an error.
+  */
+  if (func_expr->walk(&Item::is_timezone_dependent_processor,
+                      0, NULL))
+  {
+    if (is_create_table_ind)
+    {
+      my_error(ER_WRONG_EXPR_IN_PARTITION_FUNC_ERROR, MYF(0));
+      goto end;
+    }
+    else
+      push_warning(thd, MYSQL_ERROR::WARN_LEVEL_WARN,
+                   ER_WRONG_EXPR_IN_PARTITION_FUNC_ERROR,
+                   ER(ER_WRONG_EXPR_IN_PARTITION_FUNC_ERROR));
+  }
+
   if ((!is_sub_part) && (error= check_signed_flag(part_info)))
     goto end;
   result= set_up_field_array(table, is_sub_part);
@@ -1685,7 +1708,7 @@ bool fix_partition_func(THD *thd, TABLE *table,
     else
     {
       if (unlikely(fix_fields_part_func(thd, part_info->subpart_expr,
-                                        table, TRUE)))
+                                        table, TRUE, is_create_table_ind)))
         goto end;
       if (unlikely(part_info->subpart_expr->result_type() != INT_RESULT))
       {
@@ -1713,7 +1736,7 @@ bool fix_partition_func(THD *thd, TABLE *table,
     else
     {
       if (unlikely(fix_fields_part_func(thd, part_info->part_expr,
-                                        table, FALSE)))
+                                        table, FALSE, is_create_table_ind)))
         goto end;
       if (unlikely(part_info->part_expr->result_type() != INT_RESULT))
       {
@@ -1736,7 +1759,7 @@ bool fix_partition_func(THD *thd, TABLE *table,
     else
     {
       if (unlikely(fix_fields_part_func(thd, part_info->part_expr,
-                                        table, FALSE)))
+                                        table, FALSE, is_create_table_ind)))
         goto end;
     }
     part_info->fixed= TRUE;
diff --git a/sql/sql_partition.h b/sql/sql_partition.h
index 6e1bf8b5728..7ac1415c158 100644
--- a/sql/sql_partition.h
+++ b/sql/sql_partition.h
@@ -98,9 +98,6 @@ uint32 get_list_array_idx_for_endpoint(partition_info *part_info,
 uint32 get_partition_id_range_for_endpoint(partition_info *part_info,
                                            bool left_endpoint,
                                            bool include_endpoint);
-bool fix_fields_part_func(THD *thd, Item* func_expr, TABLE *table,
-                          bool is_sub_part, bool is_field_to_be_setup);
-
 bool check_part_func_fields(Field **ptr, bool ok_with_charsets);
 bool field_is_partition_charset(Field *field);
 Item* convert_charset_partition_constant(Item *item, CHARSET_INFO *cs);
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index 082ef0006cf..47426d28eb5 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -1640,6 +1640,11 @@ JOIN::reinit()
   if (join_tab_save)
     memcpy(join_tab, join_tab_save, sizeof(JOIN_TAB) * tables);
 
+  /* need to reset ref access state (see join_read_key) */
+  if (join_tab)
+    for (uint i= 0; i < tables; i++)
+      join_tab[i].ref.key_err= TRUE;
+
   if (tmp_join)
     restore_tmp();
 
diff --git a/sql/sql_update.cc b/sql/sql_update.cc
index e7b4eb22e78..18fcde90f61 100644
--- a/sql/sql_update.cc
+++ b/sql/sql_update.cc
@@ -830,7 +830,7 @@ int mysql_update(THD *thd,
 
   if (error < 0)
   {
-    char buff[STRING_BUFFER_USUAL_SIZE];
+    char buff[MYSQL_ERRMSG_SIZE];
     my_snprintf(buff, sizeof(buff), ER(ER_UPDATE_INFO), (ulong) found,
                 (ulong) updated,
                 (ulong) thd->warning_info->statement_warn_count());
diff --git a/sql/sql_view.cc b/sql/sql_view.cc
index 9abbadb8c6b..e9729d9e091 100644
--- a/sql/sql_view.cc
+++ b/sql/sql_view.cc
@@ -1772,7 +1772,7 @@ bool check_key_in_view(THD *thd, TABLE_LIST *view)
       if (!fld->item->fixed && fld->item->fix_fields(thd, &fld->item))
       {
         thd->mark_used_columns= save_mark_used_columns;
-        return TRUE;
+        DBUG_RETURN(TRUE);
       }
     }
     thd->mark_used_columns= save_mark_used_columns;
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index e0b319b562b..eba59b10879 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -598,6 +598,36 @@ Item* handle_sql2003_note184_exception(THD *thd, Item* left, bool equal,
   DBUG_RETURN(result);
 }
 
+
+static bool add_create_index_prepare (LEX *lex, Table_ident *table)
+{
+  lex->sql_command= SQLCOM_CREATE_INDEX;
+  if (!lex->current_select->add_table_to_list(lex->thd, table, NULL,
+                                              TL_OPTION_UPDATING))
+    return TRUE;
+  lex->alter_info.reset();
+  lex->alter_info.flags= ALTER_ADD_INDEX;
+  lex->col_list.empty();
+  lex->change= NullS;
+  return FALSE;
+}
+
+
+static bool add_create_index (LEX *lex, Key::Keytype type,
+                              const LEX_STRING &name,
+                              KEY_CREATE_INFO *info= NULL, bool generated= 0)
+{
+  Key *key;
+  key= new Key(type, name, info ? info : &lex->key_create_info, generated, 
+               lex->col_list);
+  if (key == NULL)
+    return TRUE;
+
+  lex->alter_info.key_list.push_back(key);
+  lex->col_list.empty();
+  return FALSE;
+}
+
 %}
 %union {
   int  num;
@@ -1350,7 +1380,7 @@ bool my_yyoverflow(short **a, YYSTYPE **b, ulong *yystacksize);
         option_type opt_var_type opt_var_ident_type
 
 %type <key_type>
-        key_type opt_unique_or_fulltext constraint_key_type
+        normal_key_type opt_unique constraint_key_type fulltext spatial
 
 %type <key_alg>
         btree_or_rtree
@@ -1451,7 +1481,10 @@ 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 sf_tail udf_tail event_tail
         install uninstall partition_entry binlog_base64_event
-        init_key_options key_options key_opts key_opt key_using_alg
+        init_key_options normal_key_options normal_key_opts all_key_opt 
+        spatial_key_options fulltext_key_options normal_key_opt 
+        fulltext_key_opt spatial_key_opt fulltext_key_opts spatial_key_opts
+        key_using_alg
         part_column_list
         server_def server_options_list server_option
         definer_opt no_definer definer
@@ -1916,35 +1949,37 @@ create:
                                   $5->table.str);
             }
           }
-        | CREATE opt_unique_or_fulltext INDEX_SYM ident key_alg ON
+        | CREATE opt_unique INDEX_SYM ident key_alg ON table_ident
+          {
+            if (add_create_index_prepare(Lex, $7))
+              MYSQL_YYABORT;
+          }
+          '(' key_list ')' normal_key_options
+          {
+            if (add_create_index(Lex, $2, $4))
+              MYSQL_YYABORT;
+          }
+        | CREATE fulltext INDEX_SYM ident init_key_options ON
           table_ident
           {
-            LEX *lex=Lex;
-            lex->sql_command= SQLCOM_CREATE_INDEX;
-            if (!lex->current_select->add_table_to_list(lex->thd, $7,
-                                                        NULL,
-                                                        TL_OPTION_UPDATING))
+            if (add_create_index_prepare(Lex, $7))
               MYSQL_YYABORT;
-            lex->alter_info.reset();
-            lex->alter_info.flags= ALTER_ADD_INDEX;
-            lex->col_list.empty();
-            lex->change=NullS;
           }
-          '(' key_list ')' key_options
+          '(' key_list ')' fulltext_key_options
           {
-            LEX *lex=Lex;
-            Key *key;
-            if ($2 != Key::FULLTEXT && lex->key_create_info.parser_name.str)
-            {
-              my_parse_error(ER(ER_SYNTAX_ERROR));
+            if (add_create_index(Lex, $2, $4))
               MYSQL_YYABORT;
-            }
-            key= new Key($2, $4, &lex->key_create_info, 0,
-                         lex->col_list);
-            if (key == NULL)
+          }
+        | CREATE spatial INDEX_SYM ident init_key_options ON
+          table_ident
+          {
+            if (add_create_index_prepare(Lex, $7))
+              MYSQL_YYABORT;
+          }
+          '(' key_list ')' spatial_key_options
+          {
+            if (add_create_index(Lex, $2, $4))
               MYSQL_YYABORT;
-            lex->alter_info.key_list.push_back(key);
-            lex->col_list.empty();
           }
         | CREATE DATABASE opt_if_not_exists ident
           {
@@ -4355,7 +4390,7 @@ part_func_expr:
             lex->safe_to_cache_query= 1;
             if (not_corr_func)
             {
-              my_parse_error(ER(ER_CONST_EXPR_IN_PARTITION_FUNC_ERROR));
+              my_parse_error(ER(ER_WRONG_EXPR_IN_PARTITION_FUNC_ERROR));
               MYSQL_YYABORT;
             }
             $$=$1;
@@ -5125,31 +5160,28 @@ column_def:
         ;
 
 key_def:
-          key_type opt_ident key_alg '(' key_list ')' key_options
+          normal_key_type opt_ident key_alg '(' key_list ')' normal_key_options
           {
-            LEX *lex=Lex;
-            if ($1 != Key::FULLTEXT && lex->key_create_info.parser_name.str)
-            {
-              my_parse_error(ER(ER_SYNTAX_ERROR));
+            if (add_create_index (Lex, $1, $2))
               MYSQL_YYABORT;
-            }
-            Key *key= new Key($1, $2, &lex->key_create_info, 0,
-                              lex->col_list);
-            if (key == NULL)
+          }
+        | fulltext opt_key_or_index opt_ident init_key_options 
+            '(' key_list ')' fulltext_key_options
+          {
+            if (add_create_index (Lex, $1, $3))
+              MYSQL_YYABORT;
+          }
+        | spatial opt_key_or_index opt_ident init_key_options 
+            '(' key_list ')' spatial_key_options
+          {
+            if (add_create_index (Lex, $1, $3))
               MYSQL_YYABORT;
-            lex->alter_info.key_list.push_back(key);
-            lex->col_list.empty(); /* Alloced by sql_alloc */
           }
         | opt_constraint constraint_key_type opt_ident key_alg
-          '(' key_list ')' key_options
+          '(' key_list ')' normal_key_options
           {
-            LEX *lex=Lex;
-            Key *key= new Key($2, $3.str ? $3 : $1, &lex->key_create_info, 0,
-                              lex->col_list);
-            if (key == NULL)
+            if (add_create_index (Lex, $2, $3.str ? $3 : $1))
               MYSQL_YYABORT;
-            lex->alter_info.key_list.push_back(key);
-            lex->col_list.empty(); /* Alloced by sql_alloc */
           }
         | opt_constraint FOREIGN KEY_SYM opt_ident '(' key_list ')' references
           {
@@ -5163,13 +5195,9 @@ key_def:
             if (key == NULL)
               MYSQL_YYABORT;
             lex->alter_info.key_list.push_back(key);
-            key= new Key(Key::MULTIPLE, $1.str ? $1 : $4,
-                         &default_key_create_info, 1,
-                         lex->col_list);
-            if (key == NULL)
+            if (add_create_index (lex, Key::MULTIPLE, $1.str ? $1 : $4,
+                                  &default_key_create_info, 1))
               MYSQL_YYABORT;
-            lex->alter_info.key_list.push_back(key);
-            lex->col_list.empty(); /* Alloced by sql_alloc */
             /* Only used for ALTER TABLE. Ignored otherwise. */
             lex->alter_info.flags|= ALTER_FOREIGN_KEY;
           }
@@ -5738,19 +5766,8 @@ delete_option:
         | SET DEFAULT   { $$= (int) Foreign_key::FK_OPTION_DEFAULT;  }
         ;
 
-key_type:
+normal_key_type:
           key_or_index { $$= Key::MULTIPLE; }
-        | FULLTEXT_SYM opt_key_or_index { $$= Key::FULLTEXT; }
-        | SPATIAL_SYM opt_key_or_index
-          {
-#ifdef HAVE_SPATIAL
-            $$= Key::SPATIAL;
-#else
-            my_error(ER_FEATURE_DISABLED, MYF(0),
-                     sym_group_geom.name, sym_group_geom.needed_define);
-            MYSQL_YYABORT;
-#endif
-          }
         ;
 
 constraint_key_type:
@@ -5774,11 +5791,17 @@ keys_or_index:
         | INDEXES {}
         ;
 
-opt_unique_or_fulltext:
+opt_unique:
           /* empty */  { $$= Key::MULTIPLE; }
         | UNIQUE_SYM   { $$= Key::UNIQUE; }
-        | FULLTEXT_SYM { $$= Key::FULLTEXT;}
-        | SPATIAL_SYM
+        ;
+
+fulltext:
+          FULLTEXT_SYM { $$= Key::FULLTEXT;}
+        ;
+
+spatial:
+          SPATIAL_SYM
           {
 #ifdef HAVE_SPATIAL
             $$= Key::SPATIAL;
@@ -5807,14 +5830,34 @@ key_alg:
         | init_key_options key_using_alg
         ;
 
-key_options:
+normal_key_options:
+          /* empty */ {}
+        | normal_key_opts
+        ;
+
+fulltext_key_options:
           /* empty */ {}
-        | key_opts
+        | fulltext_key_opts
+        ;
+
+spatial_key_options:
+          /* empty */ {}
+        | spatial_key_opts
+        ;
+
+normal_key_opts:
+          normal_key_opt
+        | normal_key_opts normal_key_opt
+        ;
+
+spatial_key_opts:
+          spatial_key_opt
+        | spatial_key_opts spatial_key_opt
         ;
 
-key_opts:
-          key_opt
-        | key_opts key_opt
+fulltext_key_opts:
+          fulltext_key_opt
+        | fulltext_key_opts fulltext_key_opt
         ;
 
 key_using_alg:
@@ -5822,10 +5865,22 @@ key_using_alg:
         | TYPE_SYM btree_or_rtree  { Lex->key_create_info.algorithm= $2; }
         ;
 
-key_opt:
-          key_using_alg
-        | KEY_BLOCK_SIZE opt_equal ulong_num
+all_key_opt:
+          KEY_BLOCK_SIZE opt_equal ulong_num
           { Lex->key_create_info.block_size= $3; }
+        ;
+
+normal_key_opt:
+          all_key_opt
+        | key_using_alg
+        ;
+
+spatial_key_opt:
+          all_key_opt
+        ;
+
+fulltext_key_opt:
+          all_key_opt
         | WITH PARSER_SYM IDENT_sys
           {
             if (plugin_is_ready(&$3, MYSQL_FTPARSER_PLUGIN))