Worklog#1563 - Support of on-line CREATE/DROP INDEX.

This is to enable table handlers to implement online create/drop index.
It consists of some parts:
- New default handler methods in handler.h
- Split of mysql_alter_table. It decides if only one kind of
  alteration is to be done (e.g. only create indexes or only drop
  indexes etc.) It then calls the specialized new handler method if
  the handler implements it. Otherwise it calls real_alter_table.
- The parser sets flags for each alter operation detected in a
  command. These are used by mysql_alter_table for the decision.
- mysql_prepare_table is pulled out of mysql_create_table. This is
  also used by mysql_create_index to prepare the key structure array
  for the handler. It is also used by mysql_create_index and
  mysql_drop_index to prepare a call to mysql_create_frm.
- mysql_create_frm is pulled out of rea_create_table for use by
  mysql_create_index and mysql_drop_index after the index is
  created/dropped.
Thanks to Antony who supplied most of the changes.


sql/handler.h:
  Worklog#1563 - Support of on-line CREATE/DROP INDEX.
  New virtual handler methods with default implementation and return flags.
sql/mysql_priv.h:
  Worklog#1563 - Support of on-line CREATE/DROP INDEX.
  New function prototypes. mysql_prepare_table is pulled out of
  mysql_create_table. It prepares a table structure, which is used by 
  mysql_create_table, mysql_create_index and mysql_drop_index.
  real_alter_table is pulled out of mysql_alter_table. The latter only
  looks if the requested operation can be done by specialized functions
  or else calls real_alter_table, which does the real thing.
  mysql_add_column and mysql_drop_colunm are for future use. They simply
  call real_alter_table. mysql_create_frm is pulled out of
  rea_create_table. It is called also from mysql_create_index and
  mysql_drop_index after successful create/drop index.
  mysql_alter_table prototype is extended by the new alter_flags argument.
sql/sql_base.cc:
  Worklog#1563 - Support of on-line CREATE/DROP INDEX.
  mysql_create_index and mysql_drop_index moved to sql_table.cc.
sql/sql_lex.h:
  Worklog#1563 - Support of on-line CREATE/DROP INDEX.
  Definitions for the new alter_flags.
sql/sql_parse.cc:
  Worklog#1563 - Support of on-line CREATE/DROP INDEX.
  Extend the calls to mysql_alter_table by the new alter_flags argument.
sql/sql_table.cc:
  Worklog#1563 - Support of on-line CREATE/DROP INDEX.
  mysql_prepare_table is pulled out of mysql_create_table.
  mysql_create_index and mysql_drop_index are changed so that they use
  the new handler functions if the handler implements them.
  mysql_add_column and mysql_drop_column are for future use. They simply
  call real_alter_table. mysql_alter_table only decides which function
  to use for the requested operation. real_alter_table implements most
  of what mysql_alter_table did before.
sql/sql_yacc.yy:
  Worklog#1563 - Support of on-line CREATE/DROP INDEX.
  Set the alter_flags where appropriate.
sql/unireg.cc:
  Worklog#1563 - Support of on-line CREATE/DROP INDEX.
  mysql_create_frm is pulled out of rea_create_table.

sql/handler.h

@@ -89,6 +89,20 @@
 #define HA_NOT_READ_PREFIX_LAST	32	/* No support for index_read_last() */
 #define HA_KEY_READ_ONLY	64	/* Support HA_EXTRA_KEYREAD */
 
+
+/*
+   Bits in index_ddl_flags(KEY *wanted_index)
+   for what ddl you can do with index
+   If none is set, the wanted type of index is not supported
+   by the handler at all. See WorkLog 1563.
+*/
+#define HA_DDL_SUPPORT   1 /* Supported by handler */
+#define HA_DDL_WITH_LOCK 2 /* Can create/drop with locked table */
+#define HA_DDL_ONLINE    4 /* Can create/drop without lock */
+
+/* Return value for ddl methods */
+#define HA_DDL_NOT_IMPLEMENTED -1
+
 /*
   Parameters for open() (in register form->filestat)
   HA_GET_INFO does an implicit HA_ABORT_IF_LOCKED
@@ -354,6 +368,20 @@ class handler :public Sql_alloc
   {
     return (HA_READ_NEXT | HA_READ_PREV | HA_READ_ORDER | HA_KEY_READ_ONLY);
   }
+  virtual ulong index_ddl_flags(KEY *wanted_index) const
+  {
+    return (HA_DDL_SUPPORT);
+  }
+  virtual int add_index(TABLE *table, KEY *key_info, uint num_of_keys)
+  {
+    my_error(ER_NOT_SUPPORTED_YET, MYF(0), "online add index");
+    return (HA_DDL_NOT_IMPLEMENTED);
+  }
+  virtual int drop_index(TABLE *table, uint *key_num, uint num_of_keys)
+  {
+    my_error(ER_NOT_SUPPORTED_YET, MYF(0), "online drop index");
+    return (HA_DDL_NOT_IMPLEMENTED);
+  }
   virtual uint max_record_length() const =0;
   virtual uint max_keys() const =0;
   virtual uint max_key_parts() const =0;

sql/mysql_priv.h

@@ -493,6 +493,11 @@ int mysql_handle_derived(LEX *lex);
 Field *create_tmp_field(THD *thd, TABLE *table,Item *item, Item::Type type,
 			Item ***copy_func, Field **from_field,
 			bool group,bool modify_item);
+int mysql_prepare_table(THD *thd, HA_CREATE_INFO *create_info,
+		       List<create_field> &fields,
+		       List<Key> &keys, uint &db_options, 
+		       handler *file, KEY *&key_info_buffer,
+		       uint &key_count, int select_field_count);
 int mysql_create_table(THD *thd,const char *db, const char *table_name,
 		       HA_CREATE_INFO *create_info,
 		       List<create_field> &fields, List<Key> &keys,
@@ -509,11 +514,23 @@ int mysql_alter_table(THD *thd, char *new_db, char *new_name,
 		      List<create_field> &fields,
 		      List<Key> &keys,List<Alter_drop> &drop_list,
 		      List<Alter_column> &alter_list,
-                      uint order_num, ORDER *order,
+                      uint order_num, ORDER *order, int alter_flags,
 		      enum enum_duplicates handle_duplicates,
 		      enum enum_enable_or_disable keys_onoff=LEAVE_AS_IS,
 		      enum tablespace_op_type tablespace_op=NO_TABLESPACE_OP,
 		      bool simple_alter=0);
+int real_alter_table(THD *thd, char *new_db, char *new_name,
+		     HA_CREATE_INFO *create_info,
+		     TABLE_LIST *table_list,
+                     TABLE *table,
+		     List<create_field> &fields,
+		     List<Key> &keys,List<Alter_drop> &drop_list,
+		     List<Alter_column> &alter_list,
+                     uint order_num, ORDER *order, int alter_flags,
+		     enum enum_duplicates handle_duplicates,
+		     enum enum_enable_or_disable keys_onoff=LEAVE_AS_IS,
+		     enum tablespace_op_type tablespace_op=NO_TABLESPACE_OP,
+		     bool simple_alter=0);
 int mysql_create_like_table(THD *thd, TABLE_LIST *table,
                             HA_CREATE_INFO *create_info,
                             Table_ident *src_table);
@@ -525,6 +542,10 @@ bool mysql_rename_table(enum db_type base,
 int mysql_create_index(THD *thd, TABLE_LIST *table_list, List<Key> &keys);
 int mysql_drop_index(THD *thd, TABLE_LIST *table_list,
 		     List<Alter_drop> &drop_list);
+int mysql_add_column(THD *thd, TABLE_LIST *table_list,
+		     List<create_field> &fields);
+int mysql_drop_column(THD *thd, TABLE_LIST *table_list,
+		      List<Alter_drop> &drop_list);
 int mysql_update(THD *thd,TABLE_LIST *tables,List<Item> &fields,
 		 List<Item> &values,COND *conds,
                  uint order_num, ORDER *order, ha_rows limit,
@@ -927,6 +948,9 @@ void unlock_table_names(THD *thd, TABLE_LIST *table_list,
 
 void unireg_init(ulong options);
 void unireg_end(void);
+int mysql_create_frm(THD *thd, my_string file_name,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, my_string file_name,HA_CREATE_INFO *create_info,
 		     List<create_field> &create_field,
 		     uint key_count,KEY *key_info);

sql/sql_base.cc

@@ -2497,45 +2497,6 @@ static void mysql_rm_tmp_tables(void)
 }
 
 
-/*
-  CREATE INDEX and DROP INDEX are implemented by calling ALTER TABLE with
-  the proper arguments.  This isn't very fast but it should work for most
-  cases.
-  One should normally create all indexes with CREATE TABLE or ALTER TABLE.
-*/
-
-int mysql_create_index(THD *thd, TABLE_LIST *table_list, List<Key> &keys)
-{
-  List<create_field> fields;
-  List<Alter_drop> drop;
-  List<Alter_column> alter;
-  HA_CREATE_INFO create_info;
-  DBUG_ENTER("mysql_create_index");
-  bzero((char*) &create_info,sizeof(create_info));
-  create_info.db_type=DB_TYPE_DEFAULT;
-  create_info.default_table_charset= thd->variables.collation_database;
-  DBUG_RETURN(mysql_alter_table(thd,table_list->db,table_list->real_name,
-				&create_info, table_list,
-				fields, keys, drop, alter, 0, (ORDER*)0,
-				DUP_ERROR));
-}
-
-
-int mysql_drop_index(THD *thd, TABLE_LIST *table_list, List<Alter_drop> &drop)
-{
-  List<create_field> fields;
-  List<Key> keys;
-  List<Alter_column> alter;
-  HA_CREATE_INFO create_info;
-  DBUG_ENTER("mysql_drop_index");
-  bzero((char*) &create_info,sizeof(create_info));
-  create_info.db_type=DB_TYPE_DEFAULT;
-  create_info.default_table_charset= thd->variables.collation_database;
-  DBUG_RETURN(mysql_alter_table(thd,table_list->db,table_list->real_name,
-				&create_info, table_list,
-				fields, keys, drop, alter, 0, (ORDER*)0,
-				DUP_ERROR));
-}
 
 /*****************************************************************************
 	unireg support functions

sql/sql_lex.h

@@ -508,6 +508,14 @@ class st_select_lex: public st_select_lex_node
 };
 typedef class st_select_lex SELECT_LEX;
 
+#define ALTER_ADD_COLUMN	1
+#define ALTER_DROP_COLUMN	2
+#define ALTER_CHANGE_COLUMN	4
+#define ALTER_ADD_INDEX		8
+#define ALTER_DROP_INDEX	16
+#define ALTER_RENAME		32
+#define ALTER_ORDER		64
+#define ALTER_OPTIONS		128
 
 /* The state of the lex parsing. This is saved in the THD struct */
 
@@ -578,6 +586,7 @@ typedef struct st_lex
   uint grant, grant_tot_col, which_columns;
   uint fk_delete_opt, fk_update_opt, fk_match_option;
   uint slave_thd_opt;
+  uint alter_flags;
   uint8 describe;
   bool drop_if_exists, drop_temporary, local_file;
   bool in_comment, ignore_space, verbose, simple_alter, no_write_to_binlog;

sql/sql_parse.cc

@@ -2355,6 +2355,7 @@ mysql_execute_command(THD *thd)
 			       lex->key_list, lex->drop_list, lex->alter_list,
 			       select_lex->order_list.elements,
                                (ORDER *) select_lex->order_list.first,
+			       lex->alter_flags,
 			       lex->duplicates,
 			       lex->alter_keys_onoff,
                                lex->tablespace_op,
@@ -2503,7 +2504,7 @@ mysql_execute_command(THD *thd)
       res= mysql_alter_table(thd, NullS, NullS, &create_info,
 			     tables, lex->create_list,
 			     lex->key_list, lex->drop_list, lex->alter_list,
-                             0, (ORDER *) 0,
+                             0, (ORDER *) 0, 0,
 			     DUP_ERROR);
     }
     else

sql/sql_table.cc

@@ -385,84 +385,42 @@ void check_duplicates_in_interval(const char *set_or_name,
 }
 
 /*
-  Create a table
+  Preparation for table creation
 
   SYNOPSIS
-    mysql_create_table()
+    mysql_prepare_table()
     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
-    tmp_table		Set to 1 if this is an internal temporary table
-			(From ALTER TABLE)
-    no_log		Don't log the query to binary log.
 
   DESCRIPTION
-    If one creates a temporary table, this is automaticly opened
-
-    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.
+    Prepares the table and key structures for table creation.
 
   RETURN VALUES
     0	ok
     -1	error
 */
 
-int mysql_create_table(THD *thd,const char *db, const char *table_name,
-		       HA_CREATE_INFO *create_info,
+int mysql_prepare_table(THD *thd, HA_CREATE_INFO *create_info,
 		       List<create_field> &fields,
-		       List<Key> &keys,bool tmp_table,bool no_log,
-                       uint select_field_count)
+		       List<Key> &keys, bool tmp_table, uint &db_options, 
+		       handler *file, KEY *&key_info_buffer,
+		       uint &key_count, int select_field_count)
 {
-  char		path[FN_REFLEN];
-  const char	*key_name, *alias;
+  const char	*key_name;
   create_field	*sql_field,*dup_field;
-  int		error= -1;
-  uint		db_options,field,null_fields,blob_columns;
+  uint		field,null_fields,blob_columns;
   ulong		pos;
-  KEY	        *key_info,*key_info_buffer;
+  KEY		*key_info;
   KEY_PART_INFO *key_part_info;
-  int		auto_increment=0;
-  handler	*file;
   int           field_no,dup_no;
-  enum db_type	new_db_type;
-  DBUG_ENTER("mysql_create_table");
+  int		select_field_pos,auto_increment=0;
+  DBUG_ENTER("mysql_prepare_table");
 
-  /* Check for duplicate fields and check type of table to create */
-  if (!fields.elements)
-  {
-    my_error(ER_TABLE_MUST_HAVE_COLUMNS,MYF(0));
-    DBUG_RETURN(-1);
-  }
   List_iterator<create_field> it(fields),it2(fields);
-  int select_field_pos=fields.elements - select_field_count;
+  select_field_pos=fields.elements - select_field_count;
   null_fields=blob_columns=0;
-  if ((new_db_type= ha_checktype(create_info->db_type)) !=
-      create_info->db_type)
-  {
-    create_info->db_type= new_db_type;
-    push_warning_printf(thd, MYSQL_ERROR::WARN_LEVEL_WARN,
-			ER_WARN_USING_OTHER_HANDLER,
-			ER(ER_WARN_USING_OTHER_HANDLER),
-			ha_get_storage_engine(new_db_type),
-			table_name);
-  }
-  db_options=create_info->table_options;
-  if (create_info->row_type == ROW_TYPE_DYNAMIC)
-    db_options|=HA_OPTION_PACK_RECORD;
-  alias= table_case_name(create_info, table_name);
-  file=get_new_handler((TABLE*) 0, create_info->db_type);
-
-  if ((create_info->options & HA_LEX_CREATE_TMP_TABLE) &&
-      (file->table_flags() & HA_NO_TEMP_TABLES))
-  {
-    my_error(ER_ILLEGAL_HA,MYF(0),table_name);
-    DBUG_RETURN(-1);
-  }
 
   for (field_no=0; (sql_field=it++) ; field_no++)
   {
@@ -649,13 +607,14 @@ int mysql_create_table(THD *thd,const char *db, const char *table_name,
   /* Create keys */
 
   List_iterator<Key> key_iterator(keys);
-  uint key_parts=0, key_count=0, fk_key_count=0;
+  uint key_parts=0, fk_key_count=0;
   List<Key> keys_in_order;			// Add new keys here
   bool primary_key=0,unique_key=0;
   Key *key;
   uint tmp, key_number;
 
   /* Calculate number of key segements */
+  key_count=0;
 
   while ((key=key_iterator++))
   {
@@ -1029,6 +988,87 @@ int mysql_create_table(THD *thd,const char *db, const char *table_name,
   /* Sort keys in optimized order */
   qsort((gptr) key_info_buffer, key_count, sizeof(KEY), (qsort_cmp) sort_keys);
 
+  DBUG_RETURN(0);
+}
+
+/*
+  Create a table
+
+  SYNOPSIS
+    mysql_create_table()
+    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
+    tmp_table		Set to 1 if this is an internal temporary table
+			(From ALTER TABLE)
+    no_log		Don't log the query to binary log.
+
+  DESCRIPTION
+    If one creates a temporary table, this is automaticly opened
+
+    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
+    0	ok
+    -1	error
+*/
+
+int mysql_create_table(THD *thd,const char *db, const char *table_name,
+		       HA_CREATE_INFO *create_info,
+		       List<create_field> &fields,
+		       List<Key> &keys,bool tmp_table,bool no_log,
+                       uint select_field_count)
+{
+  char		path[FN_REFLEN];
+  const char	*alias;
+  int		error= -1;
+  uint		db_options, key_count;
+  KEY	        *key_info_buffer;
+  handler	*file;
+  enum db_type	new_db_type;
+  DBUG_ENTER("mysql_create_table");
+
+  /* Check for duplicate fields and check type of table to create */
+  if (!fields.elements)
+  {
+    my_error(ER_TABLE_MUST_HAVE_COLUMNS,MYF(0));
+    DBUG_RETURN(-1);
+  }
+  if ((new_db_type= ha_checktype(create_info->db_type)) !=
+      create_info->db_type)
+  {
+    create_info->db_type= new_db_type;
+    push_warning_printf(thd, MYSQL_ERROR::WARN_LEVEL_WARN,
+			ER_WARN_USING_OTHER_HANDLER,
+			ER(ER_WARN_USING_OTHER_HANDLER),
+			ha_get_storage_engine(new_db_type),
+			table_name);
+  }
+  db_options=create_info->table_options;
+  if (create_info->row_type == ROW_TYPE_DYNAMIC)
+    db_options|=HA_OPTION_PACK_RECORD;
+  alias= table_case_name(create_info, table_name);
+  file=get_new_handler((TABLE*) 0, create_info->db_type);
+
+  if ((create_info->options & HA_LEX_CREATE_TMP_TABLE) &&
+      (file->table_flags() & HA_NO_TEMP_TABLES))
+  {
+    my_error(ER_ILLEGAL_HA,MYF(0),table_name);
+    DBUG_RETURN(-1);
+  }
+
+  if (mysql_prepare_table(thd, create_info, fields,
+  			  keys, tmp_table, db_options, file, 
+			  key_info_buffer, key_count,
+			  select_field_count))
+    DBUG_RETURN(-1);
+
       /* Check if table exists */
   if (create_info->options & HA_LEX_CREATE_TMP_TABLE)
   {
@@ -2064,19 +2104,311 @@ int mysql_discard_or_import_tablespace(THD *thd,
   DBUG_RETURN(error);
 }
 
+/*
+  CREATE INDEX and DROP INDEX are implemented by calling ALTER TABLE with
+  the proper arguments.  This isn't very fast but it should work for most
+  cases.
+  One should normally create all indexes with CREATE TABLE or ALTER TABLE.
+*/
+
+int mysql_create_index(THD *thd, TABLE_LIST *table_list, List<Key> &keys)
+{
+  List<create_field> fields;
+  List<Alter_drop>   drop;
+  List<Alter_column> alter;
+  HA_CREATE_INFO     create_info;
+  int                rc;
+  uint               idx;
+  uint	             db_options;
+  uint               key_count;
+  TABLE              *table;
+  Field              **f_ptr;
+  KEY                *key_info_buffer;
+  char		     path[FN_REFLEN];
+  DBUG_ENTER("mysql_create_index");
+
+  /*
+      Try to use online generation of index.
+      This requires that all indexes can be created online.
+      Otherwise, the old alter table procedure is executed.
+
+      Open the table to have access to the correct table handler.
+   */
+  if (!(table=open_ltable(thd,table_list,TL_WRITE_ALLOW_READ)))
+    DBUG_RETURN(-1);
+
+  /*
+      The add_index method takes an array of KEY structs for the new indexes.
+      Preparing a new table structure generates this array.
+      It needs a list with all fields of the table, which does not need to
+      be correct in every respect. The field names are important.
+   */
+  for (f_ptr= table->field; *f_ptr; f_ptr++)
+  {
+    create_field *c_fld= new create_field(*f_ptr, *f_ptr);
+    c_fld->unireg_check= Field::NONE; /*avoid multiple auto_increments*/
+    fields.push_back(c_fld);
+  }
+  bzero((char*) &create_info,sizeof(create_info));
+  create_info.db_type=DB_TYPE_DEFAULT;
+  create_info.default_table_charset= thd->variables.collation_database;
+  db_options= 0;
+  if (mysql_prepare_table(thd, &create_info, fields,
+  			  keys, /*tmp_table*/ 0, db_options, table->file, 
+			  key_info_buffer, key_count,
+			  /*select_field_count*/ 0))
+    DBUG_RETURN(-1);
+
+  /*
+      Check if all keys can be generated with the add_index method.
+      If anyone cannot, then take the old way.
+   */
+  for (idx=0; idx< key_count; idx++)
+  {
+    DBUG_PRINT("info", ("creating index %s", key_info_buffer[idx].name));
+    if (!(table->file->index_ddl_flags(key_info_buffer+idx)&
+          (HA_DDL_ONLINE| HA_DDL_WITH_LOCK)))
+      break ;
+  }
+  if ((idx < key_count)|| (key_count<= 0))
+  {
+    /* Re-initialize the create_info, which was changed by prepare table. */
+    bzero((char*) &create_info,sizeof(create_info));
+    create_info.db_type=DB_TYPE_DEFAULT;
+    create_info.default_table_charset= thd->variables.collation_database;
+    /* Cleanup the fields list. We do not want to create existing fields. */
+    fields.delete_elements();
+    if (real_alter_table(thd, table_list->db, table_list->real_name,
+                         &create_info, table_list, table,
+                         fields, keys, drop, alter, 0, (ORDER*)0,
+                         ALTER_ADD_INDEX, DUP_ERROR))
+      /*don't need to free((gptr) key_info_buffer);*/
+      DBUG_RETURN(-1);
+  }
+  else
+  {
+    if (table->file->add_index(table, key_info_buffer, key_count)||
+        ((void) sprintf(path, "%s/%s/%s%s", mysql_data_home, table_list->db,
+                        (lower_case_table_names == 2)? table_list->alias:
+                        table_list->real_name, reg_ext), 0)||
+        ! unpack_filename(path, path)||
+        mysql_create_frm(thd, path, &create_info,
+                         fields, key_count, key_info_buffer, table->file))
+      /*don't need to free((gptr) key_info_buffer);*/
+      DBUG_RETURN(-1);
+  }
+
+  /*don't need to free((gptr) key_info_buffer);*/
+  DBUG_RETURN(0);
+}
+
+
+int mysql_drop_index(THD *thd, TABLE_LIST *table_list, List<Alter_drop> &drop)
+{
+  List<create_field> fields;
+  List<Key>          keys;
+  List<Alter_column> alter;
+  HA_CREATE_INFO     create_info;
+  uint               idx;
+  uint	             db_options;
+  uint               key_count;
+  uint               *key_numbers;
+  TABLE              *table;
+  Field              **f_ptr;
+  KEY                *key_info;
+  KEY                *key_info_buffer;
+  char		     path[FN_REFLEN];
+  DBUG_ENTER("mysql_drop_index");
+
+  /*
+      Try to use online generation of index.
+      This requires that all indexes can be created online.
+      Otherwise, the old alter table procedure is executed.
+
+      Open the table to have access to the correct table handler.
+   */
+  if (!(table=open_ltable(thd,table_list,TL_WRITE_ALLOW_READ)))
+    DBUG_RETURN(-1);
+
+  /*
+      The drop_index method takes an array of key numbers.
+      It cannot get more entries than keys in the table.
+   */
+  key_numbers= (uint*) thd->alloc(sizeof(uint*)*table->keys);
+  key_count= 0;
+
+  /*
+      Get the number of each key and check if it can be created online.
+   */
+  List_iterator<Alter_drop> drop_it(drop);
+  Alter_drop *drop_key;
+  while ((drop_key= drop_it++))
+  {
+    /* Find the key in the table. */
+    key_info=table->key_info;
+    for (idx=0; idx< table->keys; idx++, key_info++)
+    {
+      if (!my_strcasecmp(system_charset_info, key_info->name, drop_key->name))
+	break;
+    }
+    if (idx>= table->keys)
+    {
+      my_error(ER_CANT_DROP_FIELD_OR_KEY, MYF(0), drop_key->name);
+      /*don't need to free((gptr) key_numbers);*/
+      DBUG_RETURN(-1);
+    }
+    /*
+        Check if the key can be generated with the add_index method.
+        If anyone cannot, then take the old way.
+     */
+    DBUG_PRINT("info", ("dropping index %s", table->key_info[idx].name));
+    if (!(table->file->index_ddl_flags(table->key_info+idx)&
+          (HA_DDL_ONLINE| HA_DDL_WITH_LOCK)))
+      break ;
+    key_numbers[key_count++]= idx;
+  }
+
+  bzero((char*) &create_info,sizeof(create_info));
+  create_info.db_type=DB_TYPE_DEFAULT;
+  create_info.default_table_charset= thd->variables.collation_database;
+
+  if ((drop_key)|| (drop.elements<= 0))
+  {
+    if (real_alter_table(thd, table_list->db, table_list->real_name,
+                         &create_info, table_list, table,
+                         fields, keys, drop, alter, 0, (ORDER*)0,
+                         ALTER_DROP_INDEX, DUP_ERROR))
+      /*don't need to free((gptr) key_numbers);*/
+      DBUG_RETURN(-1);
+  }
+  else
+  {
+    db_options= 0;
+    if (table->file->drop_index(table, key_numbers, key_count)||
+        mysql_prepare_table(thd, &create_info, fields,
+                            keys, /*tmp_table*/ 0, db_options, table->file, 
+                            key_info_buffer, key_count,
+                            /*select_field_count*/ 0)||
+        ((void) sprintf(path, "%s/%s/%s%s", mysql_data_home, table_list->db,
+                     (lower_case_table_names == 2)? table_list->alias:
+                        table_list->real_name, reg_ext), 0)||
+        ! unpack_filename(path, path)||
+        mysql_create_frm(thd, path, &create_info,
+                         fields, key_count, key_info_buffer, table->file))
+      /*don't need to free((gptr) key_numbers);*/
+      DBUG_RETURN(-1);
+  }
+
+  /*don't need to free((gptr) key_numbers);*/
+  DBUG_RETURN(0);
+}
+
+int mysql_add_column(THD *thd, TABLE_LIST *table_list, 
+		     List<create_field> &fields)
+{
+  List<Alter_drop> drop;
+  List<Key> keys;
+  List<Alter_column> alter;
+  HA_CREATE_INFO create_info;
+  DBUG_ENTER("mysql_add_column");
+  bzero((char*) &create_info,sizeof(create_info));
+  create_info.db_type=DB_TYPE_DEFAULT;
+  create_info.default_table_charset= thd->variables.collation_database;
+  TABLE *table;
+  if (!(table=open_ltable(thd,table_list,TL_WRITE_ALLOW_READ)))
+    DBUG_RETURN(-1);
+
+  DBUG_RETURN(real_alter_table(thd,table_list->db,table_list->real_name,
+			       &create_info, table_list, table,
+			       fields, keys, drop, alter, 0, (ORDER*)0,
+			       ALTER_ADD_COLUMN, DUP_ERROR));
+}
+
+int mysql_drop_column(THD *thd, TABLE_LIST *table_list, List<Alter_drop> &drop)
+{
+  List<create_field> fields;
+  List<Key> keys;
+  List<Alter_column> alter;
+  HA_CREATE_INFO create_info;
+  DBUG_ENTER("mysql_drop_column");
+  bzero((char*) &create_info,sizeof(create_info));
+  create_info.db_type=DB_TYPE_DEFAULT;
+  create_info.default_table_charset= thd->variables.collation_database;
+  TABLE *table;
+  if (!(table=open_ltable(thd,table_list,TL_WRITE_ALLOW_READ)))
+    DBUG_RETURN(-1);
+
+  DBUG_RETURN(real_alter_table(thd,table_list->db,table_list->real_name,
+			       &create_info, table_list, table,
+			       fields, keys, drop, alter, 0, (ORDER*)0,
+			       ALTER_DROP_COLUMN, DUP_ERROR));
+}
+
 int mysql_alter_table(THD *thd,char *new_db, char *new_name,
 		      HA_CREATE_INFO *create_info,
 		      TABLE_LIST *table_list,
 		      List<create_field> &fields,
 		      List<Key> &keys,List<Alter_drop> &drop_list,
 		      List<Alter_column> &alter_list,
-                      uint order_num, ORDER *order,
+                      uint order_num, ORDER *order, int alter_flags,
 		      enum enum_duplicates handle_duplicates,
 	              enum enum_enable_or_disable keys_onoff,
 		      enum tablespace_op_type tablespace_op,
                       bool simple_alter)
 {
-  TABLE *table,*new_table;
+  DBUG_ENTER("mysql_alter_table");
+
+  /* !!!!!!!! WARNING: This comment must be removed after a decision !!!!!!!!!
+     I'm not sure if the next two commands are at the right place here.
+     I guess that closing all is necessary before table dropping which is
+     part of alter table, but may be harmful before online DDLs.
+     So I would put both behind the DDL branches right before open_ltable.
+     !!!!!!!! WARNING: This comment must be removed after a decision !!!!!! */
+  mysql_ha_closeall(thd, table_list);
+
+  /* DISCARD/IMPORT TABLESPACE is always alone in an ALTER TABLE */
+  if (tablespace_op != NO_TABLESPACE_OP)
+    DBUG_RETURN(mysql_discard_or_import_tablespace(thd,table_list,
+						   tablespace_op));
+
+  if (alter_flags == ALTER_ADD_INDEX)
+    DBUG_RETURN(mysql_create_index(thd, table_list, keys));
+    
+  if (alter_flags == ALTER_DROP_INDEX)
+    DBUG_RETURN(mysql_drop_index(thd, table_list, drop_list));
+
+  if (alter_flags == ALTER_ADD_COLUMN)
+    DBUG_RETURN(mysql_add_column(thd, table_list, fields));
+  
+  if (alter_flags == ALTER_DROP_COLUMN)
+    DBUG_RETURN(mysql_drop_column(thd, table_list, drop_list));
+
+  TABLE *table;
+  if (!(table=open_ltable(thd,table_list,TL_WRITE_ALLOW_READ)))
+    DBUG_RETURN(-1);
+
+  DBUG_RETURN(real_alter_table(thd, new_db, new_name,
+  				create_info, table_list, table, fields,
+				keys, drop_list, alter_list,
+				order_num, order, alter_flags,
+				handle_duplicates, keys_onoff,
+				tablespace_op, simple_alter));
+}
+
+int real_alter_table(THD *thd,char *new_db, char *new_name,
+		     HA_CREATE_INFO *create_info,
+		     TABLE_LIST *table_list,
+                     TABLE *table,
+		     List<create_field> &fields,
+		     List<Key> &keys,List<Alter_drop> &drop_list,
+		     List<Alter_column> &alter_list,
+                     uint order_num, ORDER *order, int alter_flags,
+		     enum enum_duplicates handle_duplicates,
+	             enum enum_enable_or_disable keys_onoff,
+		     enum tablespace_op_type tablespace_op,
+                     bool simple_alter)
+{
+  TABLE *new_table;
   int error;
   char tmp_name[80],old_name[32],new_name_buff[FN_REFLEN];
   char new_alias_buff[FN_REFLEN], *table_name, *db, *new_alias, *alias;
@@ -2086,7 +2418,7 @@ int mysql_alter_table(THD *thd,char *new_db, char *new_name,
   ulonglong next_insert_id;
   uint save_time_stamp,db_create_options, used_fields;
   enum db_type old_db_type,new_db_type;
-  DBUG_ENTER("mysql_alter_table");
+  DBUG_ENTER("real_alter_table");
 
   thd->proc_info="init";
   table_name=table_list->real_name;
@@ -2099,15 +2431,6 @@ int mysql_alter_table(THD *thd,char *new_db, char *new_name,
   }
   used_fields=create_info->used_fields;
 
-  mysql_ha_closeall(thd, table_list);
-
-  /* DISCARD/IMPORT TABLESPACE is always alone in an ALTER TABLE */
-  if (tablespace_op != NO_TABLESPACE_OP)
-    DBUG_RETURN(mysql_discard_or_import_tablespace(thd,table_list,
-						   tablespace_op));
-  if (!(table=open_ltable(thd,table_list,TL_WRITE_ALLOW_READ)))
-    DBUG_RETURN(-1);
-
   /* Check that we are not trying to rename to an existing table */
   if (new_name)
   {

sql/sql_yacc.yy

@@ -1441,10 +1441,29 @@ attribute:
 	| DEFAULT signed_literal { Lex->default_value=$2; }
 	| AUTO_INC	  { Lex->type|= AUTO_INCREMENT_FLAG | NOT_NULL_FLAG; }
 	| SERIAL_SYM DEFAULT VALUE_SYM
-	  { Lex->type|= AUTO_INCREMENT_FLAG | NOT_NULL_FLAG | UNIQUE_FLAG; }
-	| opt_primary KEY_SYM { Lex->type|= PRI_KEY_FLAG | NOT_NULL_FLAG; }
-	| UNIQUE_SYM	  { Lex->type|= UNIQUE_FLAG; }
-	| UNIQUE_SYM KEY_SYM { Lex->type|= UNIQUE_KEY_FLAG; }
+	  { 
+	    LEX *lex=Lex;
+	    lex->type|= AUTO_INCREMENT_FLAG | NOT_NULL_FLAG | UNIQUE_FLAG; 
+	    lex->alter_flags|= ALTER_ADD_INDEX; 
+	  }
+	| opt_primary KEY_SYM 
+	  {
+	    LEX *lex=Lex;
+	    lex->type|= PRI_KEY_FLAG | NOT_NULL_FLAG; 
+	    lex->alter_flags|= ALTER_ADD_INDEX; 
+	  }
+	| UNIQUE_SYM	  
+	  {
+	    LEX *lex=Lex;
+	    lex->type|= UNIQUE_FLAG; 
+	    lex->alter_flags|= ALTER_ADD_INDEX; 
+	  }
+	| UNIQUE_SYM KEY_SYM 
+	  {
+	    LEX *lex=Lex;
+	    lex->type|= UNIQUE_KEY_FLAG; 
+	    lex->alter_flags|= ALTER_ADD_INDEX; 
+	  }
 	| COMMENT_SYM TEXT_STRING_sys { Lex->comment= &$2; }
 	| COLLATE_SYM collation_name
 	  {
@@ -1697,6 +1716,7 @@ alter:
           lex->alter_keys_onoff=LEAVE_AS_IS;
 	  lex->tablespace_op=NO_TABLESPACE_OP;
           lex->simple_alter=1;
+	  lex->alter_flags=0;
 	}
 	alter_list
 	{}
@@ -1715,16 +1735,28 @@ alter_list:
 	| alter_list ',' alter_list_item;
 
 add_column:
-	ADD opt_column { Lex->change=0; };
+	ADD opt_column 
+	{
+	  LEX *lex=Lex;
+	  lex->change=0; 
+	  lex->alter_flags|= ALTER_ADD_COLUMN; 
+	};
 
 alter_list_item:
 	add_column column_def opt_place { Lex->simple_alter=0; }
-	| ADD key_def { Lex->simple_alter=0; }
+	| ADD key_def 
+	  { 
+	    LEX *lex=Lex;
+	    lex->simple_alter=0; 
+	    lex->alter_flags|= ALTER_ADD_INDEX; 
+	  }
 	| add_column '(' field_list ')'      { Lex->simple_alter=0; }
 	| CHANGE opt_column field_ident
 	  {
 	     LEX *lex=Lex;
-	     lex->change= $3.str; lex->simple_alter=0;
+	     lex->change= $3.str; 
+	     lex->simple_alter=0;
+	     lex->alter_flags|= ALTER_CHANGE_COLUMN;
 	  }
           field_spec opt_place
         | MODIFY_SYM opt_column field_ident
@@ -1735,6 +1767,7 @@ alter_list_item:
 	    lex->comment=0;
 	    lex->charset= NULL;
             lex->simple_alter=0;
+	    lex->alter_flags|= ALTER_CHANGE_COLUMN;
           }
           type opt_attribute
           {
@@ -1752,7 +1785,9 @@ alter_list_item:
 	  {
 	    LEX *lex=Lex;
 	    lex->drop_list.push_back(new Alter_drop(Alter_drop::COLUMN,
-					    $3.str)); lex->simple_alter=0;
+        					    $3.str)); 
+	    lex->simple_alter=0;
+	    lex->alter_flags|= ALTER_DROP_COLUMN;
 	  }
 	| DROP FOREIGN KEY_SYM opt_ident { Lex->simple_alter=0; }
 	| DROP PRIMARY_SYM KEY_SYM
@@ -1761,6 +1796,7 @@ alter_list_item:
 	    lex->drop_list.push_back(new Alter_drop(Alter_drop::KEY,
 						    primary_key_name));
 	    lex->simple_alter=0;
+	    lex->alter_flags|= ALTER_DROP_INDEX;
 	  }
 	| DROP key_or_index field_ident
 	  {
@@ -1768,6 +1804,7 @@ alter_list_item:
 	    lex->drop_list.push_back(new Alter_drop(Alter_drop::KEY,
 						    $3.str));
 	    lex->simple_alter=0;
+	    lex->alter_flags|= ALTER_DROP_INDEX;
 	  }
 	| DISABLE_SYM KEYS { Lex->alter_keys_onoff=DISABLE; }
 	| ENABLE_SYM KEYS  { Lex->alter_keys_onoff=ENABLE; }
@@ -1776,21 +1813,34 @@ alter_list_item:
 	    LEX *lex=Lex;
 	    lex->alter_list.push_back(new Alter_column($3.str,$6));
 	    lex->simple_alter=0;
+	    lex->alter_flags|= ALTER_CHANGE_COLUMN;
 	  }
 	| ALTER opt_column field_ident DROP DEFAULT
 	  {
 	    LEX *lex=Lex;
 	    lex->alter_list.push_back(new Alter_column($3.str,(Item*) 0));
 	    lex->simple_alter=0;
+	    lex->alter_flags|= ALTER_CHANGE_COLUMN;
 	  }
 	| RENAME opt_to table_ident
 	  {
 	    LEX *lex=Lex;
 	    lex->select_lex.db=$3->db.str;
 	    lex->name= $3->table.str;
+	    lex->alter_flags|= ALTER_RENAME;
 	  }
-        | create_table_options_space_separated { Lex->simple_alter=0; }
-	| order_clause         { Lex->simple_alter=0; };
+        | create_table_options_space_separated 
+	  {
+	    LEX *lex=Lex;
+	    lex->simple_alter=0; 
+	    lex->alter_flags|= ALTER_OPTIONS;
+	  }
+	| order_clause         
+	  {
+	    LEX *lex=Lex;
+	    lex->simple_alter=0; 
+	    lex->alter_flags|= ALTER_ORDER;
+	  };
 
 opt_column:
 	/* empty */	{}

sql/unireg.cc

@@ -47,10 +47,11 @@ static bool make_empty_rec(int file, enum db_type table_type,
 			   uint reclength,uint null_fields);
 
 
-int rea_create_table(THD *thd, my_string file_name,
+int mysql_create_frm(THD *thd, my_string file_name,
 		     HA_CREATE_INFO *create_info,
 		     List<create_field> &create_fields,
-		     uint keys, KEY *key_info)
+		     uint keys, KEY *key_info,
+		     handler *db_file)
 {
   uint reclength,info_length,screens,key_info_length,maxlength,null_fields;
   File file;
@@ -58,13 +59,13 @@ int rea_create_table(THD *thd, my_string file_name,
   uchar fileinfo[64],forminfo[288],*keybuff;
   TYPELIB formnames;
   uchar *screen_buff;
-  handler *db_file;
   DBUG_ENTER("rea_create_table");
 
   formnames.type_names=0;
   if (!(screen_buff=pack_screens(create_fields,&info_length,&screens,0)))
     DBUG_RETURN(1);
-  db_file=get_new_handler((TABLE*) 0, create_info->db_type);
+  if (db_file == NULL)
+    db_file=get_new_handler((TABLE*) 0, create_info->db_type);
   if (pack_header(forminfo, create_info->db_type,create_fields,info_length,
 		  screens, create_info->table_options, db_file))
   {
@@ -155,8 +156,7 @@ int rea_create_table(THD *thd, my_string file_name,
   if (opt_sync_frm && !(create_info->options & HA_LEX_CREATE_TMP_TABLE) &&
       my_sync(file, MYF(MY_WME)))
     goto err2;
-  if (my_close(file,MYF(MY_WME)) ||
-      ha_create_table(file_name,create_info,0))
+  if (my_close(file,MYF(MY_WME)))
     goto err3;
   DBUG_RETURN(0);
 
@@ -166,6 +166,23 @@ int rea_create_table(THD *thd, my_string file_name,
 err2:
   VOID(my_close(file,MYF(MY_WME)));
 err3:
+  DBUG_RETURN(1);
+} /* mysql_create_frm */
+
+int rea_create_table(THD *thd, my_string file_name,
+		     HA_CREATE_INFO *create_info,
+		     List<create_field> &create_fields,
+		     uint keys, KEY *key_info)
+{
+  DBUG_ENTER("rea_create_table");
+
+  if (mysql_create_frm(thd, file_name, create_info,
+  		       create_fields, keys, key_info, NULL) ||
+      ha_create_table(file_name,create_info,0))
+    goto err;
+  DBUG_RETURN(0);
+
+err:
   my_delete(file_name,MYF(0));
   DBUG_RETURN(1);
 } /* rea_create_table */