WL #2602, #2603, #2604

Added new syntax for partition management


mysql-test/t/partition_error.test:
  New line
sql/ha_partition.cc:
  Added support for DROP PARTITION on-line
sql/ha_partition.h:
  Added support for DROP PARTITION on-line
sql/handler.h:
  Introduced state of partition
  Introduced extra list of temporary partitions
  Removed no_full_parts
  A couple of methods to check for duplicate names of partitions
  Adaptions of default checks to be useful from ALTER TABLE
  partition management
  New method on handler to drop partitions
  New method to check for foreign keys on table
sql/lex.h:
  COALESCE and REORGANISE used by ALTER TABLE x COALESCE/REORGANISE PARTITION
sql/mysql_priv.h:
  Parser needs method to check if partition management command is ongoing
sql/share/errmsg.txt:
  A number of new error messages for partition management
sql/sql_lex.h:
  Adapted the ALTER_INFO data structure for partition management
sql/sql_partition.cc:
  Couple of new routines + adaption of existing for new
  partition management functionality
sql/sql_table.cc:
  bin log writing into separate subroutine to minimise code
  duplication.
  Lots of new code to handle partition management
sql/sql_yacc.yy:
  New syntax for partition management
  Fixed a few errors in the parser part for partitioning

mysql-test/r/partition_mgm_err.result

+CREATE TABLE t1 (a int, b int)
+PARTITION BY RANGE (a)
+(PARTITION x0 VALUES LESS THAN (2),
+PARTITION x1 VALUES LESS THAN (4),
+PARTITION x2 VALUES LESS THAN (6),
+PARTITION x3 VALUES LESS THAN (8),
+PARTITION x4 VALUES LESS THAN (10),
+PARTITION x5 VALUES LESS THAN (12),
+PARTITION x6 VALUES LESS THAN (14),
+PARTITION x7 VALUES LESS THAN (16),
+PARTITION x8 VALUES LESS THAN (18),
+PARTITION x9 VALUES LESS THAN (20));
+ALTER TABLE t1 REORGANISE PARTITION x0,x1 INTO
+(PARTITION x01 VALUES LESS THAN (2),
+PARTITION x11 VALUES LESS THAN (5));
+ERROR HY000: The new partitions cover a bigger range then the reorganised partitions do
+ALTER TABLE t1 DROP PARTITION x0, x1, x2, x3, x3;
+ERROR HY000: Error in list of partitions to change
+ALTER TABLE t1 DROP PARTITION x0, x1, x2, x10;
+ERROR HY000: Error in list of partitions to change
+ALTER TABLE t1 DROP PARTITION x10, x1, x2, x1;
+ERROR HY000: Error in list of partitions to change
+ALTER TABLE t1 DROP PARTITION x10, x1, x2, x3;
+ERROR HY000: Error in list of partitions to change
+ALTER TABLE t1 REORGANISE PARTITION x0,x1,x2,x3,x4,x5,x6,x7,x8,x9,x10 INTO
+(PARTITION x11 VALUES LESS THAN (22));
+ERROR HY000: More partitions to reorganise than there are partitions
+ALTER TABLE t1 REORGANISE PARTITION x0,x1,x2 INTO
+(PARTITION x1 VALUES LESS THAN (6));
+ERROR HY000: All partitions must have unique names in the table
+ALTER TABLE t1 REORGANISE PARTITION x0, x2 INTO
+(PARTITION x11 VALUES LESS THAN (2));
+ERROR HY000: When reorganising a set of partitions they must be in consecutive order
+ALTER TABLE t1 REORGANISE PARTITION x0, x1, x1 INTO
+(PARTITION x11 VALUES LESS THAN (4));
+ERROR HY000: Error in list of partitions to change
+ALTER TABLE t1 REORGANISE PARTITION x0,x1 INTO
+(PARTITION x01 VALUES LESS THAN (5));
+ERROR HY000: The new partitions cover a bigger range then the reorganised partitions do
+ALTER TABLE t1 REORGANISE PARTITION x0,x1 INTO
+(PARTITION x01 VALUES LESS THAN (4),
+PARTITION x11 VALUES LESS THAN (2));
+ERROR HY000: VALUES LESS THAN value must be strictly increasing for each partition
+DROP TABLE t1;
+CREATE TABLE t1 (a int)
+PARTITION BY KEY (a)
+PARTITIONS 2;
+ALTER TABLE t1 ADD PARTITION (PARTITION p1);
+ERROR HY000: All partitions must have unique names in the table
+DROP TABLE t1;
+CREATE TABLE t1 (a int)
+PARTITION BY KEY (a)
+(PARTITION x0, PARTITION x1, PARTITION x2, PARTITION x3, PARTITION x3);
+ERROR HY000: All partitions must have unique names in the table
+CREATE TABLE t1 (a int)
+PARTITION BY RANGE (a)
+SUBPARTITION BY KEY (a)
+SUBPARTITIONS 2
+(PARTITION x0 VALUES LESS THAN (4),
+PARTITION x1 VALUES LESS THAN (8));
+ALTER TABLE t1 ADD PARTITION (PARTITION x2 VALUES LESS THAN (5)
+(SUBPARTITION sp0, SUBPARTITION sp1));
+ERROR HY000: VALUES LESS THAN value must be strictly increasing for each partition
+ALTER TABLE t1 ADD PARTITION (PARTITION x2 VALUES LESS THAN (12)
+(SUBPARTITION sp0, SUBPARTITION sp1, SUBPARTITION sp2));
+ERROR HY000: Trying to Add partition(s) with wrong number of subpartitions
+DROP TABLE t1;
+CREATE TABLE t1 (a int)
+PARTITION BY LIST (a)
+(PARTITION x0 VALUES IN (1,2,3),
+PARTITION x1 VALUES IN (4,5,6));
+ALTER TABLE t1 ADD PARTITION (PARTITION x2 VALUES IN (3,4));
+ERROR HY000: Multiple definition of same constant in list partitioning
+DROP TABLE t1;
+CREATE TABLE t1 (a int);
+ALTER TABLE t1 ADD PARTITION PARTITIONS 1;
+ERROR HY000: Partition management on a not partitioned table is not possible
+ALTER TABLE t1 DROP PARTITION x1;
+ERROR HY000: Partition management on a not partitioned table is not possible
+ALTER TABLE t1 COALESCE PARTITION 1;
+ERROR HY000: Partition management on a not partitioned table is not possible
+DROP TABLE t1;
+CREATE TABLE t1 (a int)
+PARTITION BY KEY (a)
+(PARTITION x0, PARTITION x1);
+ALTER TABLE t1 ADD PARTITION PARTITIONS 0;
+ERROR HY000: At least one partition must be added
+ALTER TABLE t1 ADD PARTITION PARTITIONS 1024;
+ERROR HY000: Too many partitions were defined
+ALTER TABLE t1 DROP PARTITION x0;
+ERROR HY000: DROP PARTITION can only be used on RANGE/LIST partitions
+ALTER TABLE t1 COALESCE PARTITION 1;
+ALTER TABLE t1 COALESCE PARTITION 1;
+ERROR HY000: Cannot remove all partitions, use DROP TABLE instead
+DROP TABLE t1;
+CREATE TABLE t1 (a int)
+PARTITION BY RANGE (a)
+(PARTITION x0 VALUES LESS THAN (4),
+PARTITION x1 VALUES LESS THAN (8));
+ALTER TABLE t1 ADD PARTITION PARTITIONS 1;
+ERROR HY000: For RANGE partitions each partition must be defined
+ALTER TABLE t1 DROP PARTITION x2;
+ERROR HY000: Error in list of partitions to change
+ALTER TABLE t1 COALESCE PARTITION 1;
+ERROR HY000: COALESCE PARTITION can only be used on HASH/KEY partitions
+ALTER TABLE t1 DROP PARTITION x1;
+ALTER TABLE t1 DROP PARTITION x0;
+ERROR HY000: Cannot remove all partitions, use DROP TABLE instead
+DROP TABLE t1;

mysql-test/t/partition_error.test

@@ -726,3 +726,4 @@ partition by list (a)
 partitions 2
 (partition x1 values in 4,
  partition x2 values in (5));
+

mysql-test/t/partition_mgm_err.test

+#
+# Simple test for the erroneos create statements using the 
+# partition storage engine
+#
+-- source include/have_partition.inc
+
+#
+# Try faulty DROP PARTITION and COALESCE PARTITION
+#
+CREATE TABLE t1 (a int, b int)
+PARTITION BY RANGE (a)
+(PARTITION x0 VALUES LESS THAN (2),
+ PARTITION x1 VALUES LESS THAN (4),
+ PARTITION x2 VALUES LESS THAN (6),
+ PARTITION x3 VALUES LESS THAN (8),
+ PARTITION x4 VALUES LESS THAN (10),
+ PARTITION x5 VALUES LESS THAN (12),
+ PARTITION x6 VALUES LESS THAN (14),
+ PARTITION x7 VALUES LESS THAN (16),
+ PARTITION x8 VALUES LESS THAN (18),
+ PARTITION x9 VALUES LESS THAN (20));
+
+--error ER_REORG_OUTSIDE_RANGE
+ALTER TABLE t1 REORGANISE PARTITION x0,x1 INTO
+(PARTITION x01 VALUES LESS THAN (2),
+ PARTITION x11 VALUES LESS THAN (5));
+
+--error ER_DROP_PARTITION_NON_EXISTENT
+ALTER TABLE t1 DROP PARTITION x0, x1, x2, x3, x3;
+
+--error ER_DROP_PARTITION_NON_EXISTENT
+ALTER TABLE t1 DROP PARTITION x0, x1, x2, x10;
+
+--error ER_DROP_PARTITION_NON_EXISTENT
+ALTER TABLE t1 DROP PARTITION x10, x1, x2, x1;
+
+--error ER_DROP_PARTITION_NON_EXISTENT
+ALTER TABLE t1 DROP PARTITION x10, x1, x2, x3;
+
+--error ER_REORG_PARTITION_NOT_EXIST
+ALTER TABLE t1 REORGANISE PARTITION x0,x1,x2,x3,x4,x5,x6,x7,x8,x9,x10 INTO
+(PARTITION x11 VALUES LESS THAN (22));
+
+--error ER_SAME_NAME_PARTITION
+ALTER TABLE t1 REORGANISE PARTITION x0,x1,x2 INTO
+(PARTITION x1 VALUES LESS THAN (6));
+
+--error ER_CONSECUTIVE_REORG_PARTITIONS
+ALTER TABLE t1 REORGANISE PARTITION x0, x2 INTO
+(PARTITION x11 VALUES LESS THAN (2));
+
+--error ER_DROP_PARTITION_NON_EXISTENT
+ALTER TABLE t1 REORGANISE PARTITION x0, x1, x1 INTO
+(PARTITION x11 VALUES LESS THAN (4));
+
+--error ER_REORG_OUTSIDE_RANGE
+ALTER TABLE t1 REORGANISE PARTITION x0,x1 INTO
+(PARTITION x01 VALUES LESS THAN (5));
+
+--error ER_RANGE_NOT_INCREASING_ERROR
+ALTER TABLE t1 REORGANISE PARTITION x0,x1 INTO
+(PARTITION x01 VALUES LESS THAN (4),
+ PARTITION x11 VALUES LESS THAN (2));
+
+DROP TABLE t1;
+
+CREATE TABLE t1 (a int)
+PARTITION BY KEY (a)
+PARTITIONS 2;
+
+--error ER_SAME_NAME_PARTITION
+ALTER TABLE t1 ADD PARTITION (PARTITION p1);
+
+DROP TABLE t1;
+
+--error ER_SAME_NAME_PARTITION
+CREATE TABLE t1 (a int)
+PARTITION BY KEY (a)
+(PARTITION x0, PARTITION x1, PARTITION x2, PARTITION x3, PARTITION x3);
+
+CREATE TABLE t1 (a int)
+PARTITION BY RANGE (a)
+SUBPARTITION BY KEY (a)
+SUBPARTITIONS 2
+(PARTITION x0 VALUES LESS THAN (4),
+ PARTITION x1 VALUES LESS THAN (8));
+
+--error ER_RANGE_NOT_INCREASING_ERROR
+ALTER TABLE t1 ADD PARTITION (PARTITION x2 VALUES LESS THAN (5)
+(SUBPARTITION sp0, SUBPARTITION sp1));
+
+--error ER_ADD_PARTITION_SUBPART_ERROR
+ALTER TABLE t1 ADD PARTITION (PARTITION x2 VALUES LESS THAN (12)
+(SUBPARTITION sp0, SUBPARTITION sp1, SUBPARTITION sp2));
+
+DROP TABLE t1;
+
+CREATE TABLE t1 (a int)
+PARTITION BY LIST (a)
+(PARTITION x0 VALUES IN (1,2,3),
+ PARTITION x1 VALUES IN (4,5,6));
+
+--error ER_MULTIPLE_DEF_CONST_IN_LIST_PART_ERROR
+ALTER TABLE t1 ADD PARTITION (PARTITION x2 VALUES IN (3,4));
+
+DROP TABLE t1;
+
+CREATE TABLE t1 (a int);
+
+--error ER_PARTITION_MGMT_ON_NONPARTITIONED
+ALTER TABLE t1 ADD PARTITION PARTITIONS 1;
+
+--error ER_PARTITION_MGMT_ON_NONPARTITIONED
+ALTER TABLE t1 DROP PARTITION x1;
+
+--error ER_PARTITION_MGMT_ON_NONPARTITIONED
+ALTER TABLE t1 COALESCE PARTITION 1;
+
+DROP TABLE t1;
+
+CREATE TABLE t1 (a int)
+PARTITION BY KEY (a)
+(PARTITION x0, PARTITION x1);
+
+--error ER_ADD_PARTITION_NO_NEW_PARTITION
+ALTER TABLE t1 ADD PARTITION PARTITIONS 0;
+
+--error ER_TOO_MANY_PARTITIONS_ERROR
+ALTER TABLE t1 ADD PARTITION PARTITIONS 1024;
+
+--error ER_ONLY_ON_RANGE_LIST_PARTITION
+ALTER TABLE t1 DROP PARTITION x0;
+
+ALTER TABLE t1 COALESCE PARTITION 1;
+
+--error ER_DROP_LAST_PARTITION
+ALTER TABLE t1 COALESCE PARTITION 1;
+
+DROP TABLE t1;
+
+CREATE TABLE t1 (a int)
+PARTITION BY RANGE (a)
+(PARTITION x0 VALUES LESS THAN (4),
+ PARTITION x1 VALUES LESS THAN (8));
+
+--error ER_PARTITIONS_MUST_BE_DEFINED_ERROR
+ALTER TABLE t1 ADD PARTITION PARTITIONS 1;
+
+--error ER_DROP_PARTITION_NON_EXISTENT
+ALTER TABLE t1 DROP PARTITION x2;
+
+--error ER_COALESCE_ONLY_ON_HASH_PARTITION
+ALTER TABLE t1 COALESCE PARTITION 1;
+
+ALTER TABLE t1 DROP PARTITION x1;
+
+--error ER_DROP_LAST_PARTITION
+ALTER TABLE t1 DROP PARTITION x0;
+
+DROP TABLE t1;

sql/ha_partition.cc

@@ -269,6 +269,28 @@ int ha_partition::ha_initialise()
 /****************************************************************************
                 MODULE meta data changes
 ****************************************************************************/
+/*
+  This method is used to calculate the partition name, service routine to
+  the del_ren_cre_table method.
+*/
+
+static void create_partition_name(char *out, const char *in1, const char *in2)
+{
+  strxmov(out, in1, "_", in2, NullS);
+}
+
+/*
+  This method is used to calculate the partition name, service routine to
+  the del_ren_cre_table method.
+*/
+
+static void create_subpartition_name(char *out, const char *in1,
+                                     const char *in2, const char *in3)
+{
+  strxmov(out, in1, "_", in2, "_", in3, NullS);
+}
+
+
 /*
   Used to delete a table. By the time delete_table() has been called all
   opened references to this table will have been closed (and your globally
@@ -326,6 +348,12 @@ int ha_partition::rename_table(const char *from, const char *to)
 int ha_partition::create_handler_files(const char *name)
 {
   DBUG_ENTER("ha_partition::create_handler_files()");
+
+  /*
+    We need to update total number of parts since we might write the handler
+    file as part of a partition management command
+  */
+  m_tot_parts= get_tot_partitions(m_part_info);
   if (create_handler_file(name))
   {
     my_error(ER_CANT_CREATE_HANDLER_FILE, MYF(0));
@@ -362,6 +390,49 @@ int ha_partition::create(const char *name, TABLE *table_arg,
   DBUG_RETURN(0);
 }
 
+int ha_partition::drop_partitions(const char *path)
+{
+  List_iterator<partition_element> part_it(m_part_info->partitions);
+  char part_name_buff[FN_REFLEN];
+  uint no_parts= m_part_info->no_parts;
+  uint no_subparts= m_part_info->no_subparts, i= 0;
+  int error= 1;
+  DBUG_ENTER("ha_partition::drop_partitions()");
+
+  do
+  {
+    partition_element *part_elem= part_it++;
+    if (part_elem->part_state == PART_IS_DROPPED)
+    {
+      /*
+        This part is to be dropped, meaning the part or all its subparts.
+      */
+      if (is_sub_partitioned(m_part_info))
+      {
+        List_iterator<partition_element> sub_it(part_elem->subpartitions);
+        uint j= 0, part;
+        do
+        {
+          partition_element *sub_elem= sub_it++;
+          create_subpartition_name(part_name_buff, path,
+                                   part_elem->partition_name,
+                                   sub_elem->partition_name);
+          part= i * no_subparts + j;
+          DBUG_PRINT("info", ("Drop subpartition %s", part_name_buff));
+          error= m_file[part]->delete_table((const char *) part_name_buff);
+        } while (++j < no_subparts);
+      }
+      else
+      {
+        create_partition_name(part_name_buff, path,
+                              part_elem->partition_name);
+        DBUG_PRINT("info", ("Drop partition %s", part_name_buff));
+        error= m_file[i]->delete_table((const char *) part_name_buff);
+      }
+    }
+  } while (++i < no_parts);
+  DBUG_RETURN(error);
+}
 
 void ha_partition::update_create_info(HA_CREATE_INFO *create_info)
 {
@@ -375,16 +446,6 @@ char *ha_partition::update_table_comment(const char *comment)
 }
 
 
-/*
-  This method is used to calculate the partition name, service routine to
-  the del_ren_cre_table method.
-*/
-
-static void create_partition_name(char *out, const char *in1, const char *in2)
-{
-  strxmov(out, in1, "_", in2, NullS);
-}
-
 
 /*
   Common routine to handle delete_table and rename_table.

sql/ha_partition.h

@@ -166,6 +166,7 @@ class ha_partition :public handler
   virtual int create_handler_files(const char *name);
   virtual void update_create_info(HA_CREATE_INFO * create_info);
   virtual char *update_table_comment(const char *comment);
+  virtual int drop_partitions(const char *path);
 private:
   /*
     delete_table, rename_table and create uses very similar logic which
@@ -633,6 +634,11 @@ class ha_partition :public handler
     index scan module.
     (NDB)
   */
+  virtual ulong alter_table_flags(void) const
+  {
+    //return HA_ONLINE_ADD_EMPTY_PARTITION + HA_ONLINE_DROP_PARTITION;
+    return HA_ONLINE_DROP_PARTITION;
+  }
   virtual ulong table_flags() const
   { return m_table_flags; }
   /*

sql/handler.h

@@ -103,6 +103,10 @@
 #define HA_ONLY_WHOLE_INDEX	16	/* Can't use part key searches */
 #define HA_KEYREAD_ONLY         64	/* Support HA_EXTRA_KEYREAD */
 
+/* bits in alter_table_flags */
+#define HA_ONLINE_ADD_EMPTY_PARTITION 1
+#define HA_ONLINE_DROP_PARTITION 2
+
 /* operations for disable/enable indexes */
 #define HA_KEY_SWITCH_NONUNIQ      0
 #define HA_KEY_SWITCH_ALL          1
@@ -399,6 +403,16 @@ enum partition_type {
   LIST_PARTITION
 };
 
+enum partition_state {
+  PART_NORMAL= 0,
+  PART_IS_DROPPED= 1,
+  PART_TO_BE_DROPPED= 2,
+  PART_DROPPING= 3,
+  PART_IS_ADDED= 4,
+  PART_ADDING= 5,
+  PART_ADDED= 6
+};
+
 #define UNDEF_NODEGROUP 65535
 class Item;
 
@@ -415,13 +429,15 @@ class partition_element :public Sql_alloc {
   char* data_file_name;
   char* index_file_name;
   enum db_type engine_type;
+  enum partition_state part_state;
   uint16 nodegroup_id;
   
   partition_element()
   : part_max_rows(0), part_min_rows(0), partition_name(NULL),
     tablespace_name(NULL), range_value(0), part_comment(NULL),
     data_file_name(NULL), index_file_name(NULL),
-    engine_type(DB_TYPE_UNKNOWN), nodegroup_id(UNDEF_NODEGROUP)
+    engine_type(DB_TYPE_UNKNOWN),part_state(PART_NORMAL),
+    nodegroup_id(UNDEF_NODEGROUP)
   {
     subpartitions.empty();
     list_val_list.empty();
@@ -447,6 +463,7 @@ class partition_info :public Sql_alloc {
    * Here comes a set of definitions needed for partitioned table handlers.
    */
   List<partition_element> partitions;
+  List<partition_element> temp_partitions;
 
   List<char> part_field_list;
   List<char> subpart_field_list;
@@ -492,7 +509,6 @@ class partition_info :public Sql_alloc {
   uint part_func_len;
   uint subpart_func_len;
 
-  uint no_full_parts;
   uint no_parts;
   uint no_subparts;
   uint count_curr_parts;
@@ -529,7 +545,7 @@ class partition_info :public Sql_alloc {
     part_result_type(INT_RESULT),
     part_type(NOT_A_PARTITION), subpart_type(NOT_A_PARTITION),
     part_info_len(0), part_func_len(0), subpart_func_len(0),
-    no_full_parts(0), no_parts(0), no_subparts(0),
+    no_parts(0), no_subparts(0),
     count_curr_parts(0), count_curr_subparts(0), part_error_code(0),
     no_list_values(0), no_part_fields(0), no_subpart_fields(0),
     no_full_part_fields(0), linear_hash_mask(0),
@@ -543,6 +559,7 @@ class partition_info :public Sql_alloc {
     all_fields_in_SPF.clear_all();
     some_fields_in_PF.clear_all();
     partitions.empty();
+    temp_partitions.empty();
     part_field_list.empty();
     subpart_field_list.empty();
   }
@@ -634,6 +651,13 @@ typedef struct st_ha_check_opt
 
 
 #ifdef HAVE_PARTITION_DB
+bool is_partition_in_list(char *part_name, List<char> list_part_names);
+bool is_partitions_in_table(partition_info *new_part_info,
+                            partition_info *old_part_info);
+bool set_up_defaults_for_partitioning(partition_info *part_info,
+                                      handler *file,
+                                      ulonglong max_rows,
+                                      uint start_no);
 handler *get_ha_partition(partition_info *part_info);
 int get_parts_for_update(const byte *old_data, byte *new_data,
                          const byte *rec0, partition_info *part_info,
@@ -1138,6 +1162,20 @@ class handler :public Sql_alloc
   virtual char *update_table_comment(const char * comment)
   { return (char*) comment;}
   virtual void append_create_info(String *packet) {}
+  /*
+    SYNOPSIS
+      is_fk_defined_on_table_or_index()
+      index            Index to check if foreign key uses it
+    RETURN VALUE
+       TRUE            Foreign key defined on table or index
+       FALSE           No foreign key defined
+    DESCRIPTION
+      If index == MAX_KEY then a check for table is made and if index <
+      MAX_KEY then a check is made if the table has foreign keys and if
+      a foreign key uses this index (and thus the index cannot be dropped).
+  */
+  virtual bool is_fk_defined_on_table_or_index(uint index)
+  { return FALSE; }
   virtual char* get_foreign_key_create_info()
   { return(NULL);}  /* gets foreign key create string from InnoDB */
   /* used in ALTER TABLE; 1 if changing storage engine is allowed */
@@ -1153,6 +1191,7 @@ class handler :public Sql_alloc
   virtual const char *table_type() const =0;
   virtual const char **bas_ext() const =0;
   virtual ulong table_flags(void) const =0;
+  virtual ulong alter_table_flags(void) const { return 0; }
 #ifdef HAVE_PARTITION_DB
   virtual ulong partition_flags(void) const { return 0;}
   virtual int get_default_no_partitions(ulonglong max_rows) { return 1;}
@@ -1198,6 +1237,19 @@ class handler :public Sql_alloc
   virtual int create(const char *name, TABLE *form, HA_CREATE_INFO *info)=0;
   virtual int create_handler_files(const char *name) { return FALSE;}
 
+  /*
+    SYNOPSIS
+      drop_partitions()
+      path                        Complete path of db and table name
+    RETURN VALUE
+      TRUE                        Failure
+      FALSE                       Success
+    DESCRIPTION
+      Drop a partition, during this operation no other activity is ongoing
+      in this server on the table.
+  */
+  virtual int drop_partitions(const char *path)
+  { return HA_ERR_WRONG_COMMAND; }
   /* lock_count() can be more than one if the table is a MERGE */
   virtual uint lock_count(void) const { return 1; }
   virtual THR_LOCK_DATA **store_lock(THD *thd,

sql/lex.h

@@ -110,6 +110,7 @@ static SYMBOL symbols[] = {
   { "CIPHER",		SYM(CIPHER_SYM)},
   { "CLIENT",		SYM(CLIENT_SYM)},
   { "CLOSE",		SYM(CLOSE_SYM)},
+  { "COALESCE",		SYM(COALESCE)},
   { "COLLATE",		SYM(COLLATE_SYM)},
   { "COLLATION",	SYM(COLLATION_SYM)},
   { "COLUMN",		SYM(COLUMN_SYM)},
@@ -408,6 +409,7 @@ static SYMBOL symbols[] = {
   { "RELEASE",		SYM(RELEASE_SYM)},
   { "RELOAD",		SYM(RELOAD)},
   { "RENAME",		SYM(RENAME)},
+  { "REORGANISE",	SYM(REORGANISE_SYM)},
   { "REPAIR",		SYM(REPAIR)},
   { "REPEATABLE",	SYM(REPEATABLE_SYM)},
   { "REPLACE",		SYM(REPLACE)},
@@ -589,7 +591,6 @@ static SYMBOL sql_functions[] = {
   { "CENTROID",		F_SYM(FUNC_ARG1),0,CREATE_FUNC_GEOM(create_func_centroid)},
   { "CHAR_LENGTH",	F_SYM(FUNC_ARG1),0,CREATE_FUNC(create_func_char_length)},
   { "CHARACTER_LENGTH", F_SYM(FUNC_ARG1),0,CREATE_FUNC(create_func_char_length)},
-  { "COALESCE",		SYM(COALESCE)},
   { "COERCIBILITY",	F_SYM(FUNC_ARG1),0,CREATE_FUNC(create_func_coercibility)},
   { "COMPRESS",		F_SYM(FUNC_ARG1),0,CREATE_FUNC(create_func_compress)},
   { "CONCAT",		SYM(CONCAT)},

sql/mysql_priv.h

@@ -627,6 +627,10 @@ bool check_table_access(THD *thd, ulong want_access, TABLE_LIST *tables,
 			bool no_errors);
 bool check_global_access(THD *thd, ulong want_access);
 
+/*
+  Support routine for SQL parser on partitioning syntax
+*/
+my_bool is_partition_management(LEX *lex);
 /*
   General routine to change field->ptr of a NULL-terminated array of Field
   objects. Useful when needed to call val_int, val_str or similar and the

sql/share/errmsg.txt

@@ -5440,7 +5440,7 @@ ER_PARTITION_ENTRY_ERROR
         eng "Partitioning can not be used stand-alone in query"
         swe "Partitioneringssyntax kan inte anvndas p egen hand i en SQL-frga"
 ER_MIX_HANDLER_ERROR
-        eng "The mix of handlers in the partitions is not allowed in this version in MySQL"
+        eng "The mix of handlers in the partitions is not allowed in this version of MySQL"
         swe "Denna mix av lagringsmotorer r inte tillten i denna version av MySQL"
 ER_PARTITION_NOT_DEFINED_ERROR
         eng "For the partitioned engine it is necessary to define all %s"
@@ -5466,3 +5466,44 @@ ER_UNIQUE_KEY_NEED_ALL_FIELDS_IN_PF
 ER_NO_PARTS_ERROR
         eng "Number of %s = 0 is not an allowed value"
         swe "Antal %s = 0 r inte ett tillten vrde"
+ER_PARTITION_MGMT_ON_NONPARTITIONED
+        eng "Partition management on a not partitioned table is not possible"
+        swe "Partitioneringskommando p en opartitionerad tabell r inte mjligt"
+ER_DROP_PARTITION_NON_EXISTENT
+        eng "Error in list of partitions to change"
+        swe "Fel i listan av partitioner att frndra"
+ER_DROP_LAST_PARTITION
+        eng "Cannot remove all partitions, use DROP TABLE instead"
+        swe "Det r inte tilltet att ta bort alla partitioner, anvnd DROP TABLE istllet"
+ER_COALESCE_ONLY_ON_HASH_PARTITION
+        eng "COALESCE PARTITION can only be used on HASH/KEY partitions"
+        swe "COALESCE PARTITION kan bara anvndas p HASH/KEY partitioner"
+ER_ONLY_ON_RANGE_LIST_PARTITION
+        eng "%s PARTITION can only be used on RANGE/LIST partitions"
+        eng "%s PARTITION kan bara anvndas p RANGE/LIST partitioner"
+ER_ADD_PARTITION_SUBPART_ERROR
+        eng "Trying to Add partition(s) with wrong number of subpartitions"
+        swe "ADD PARTITION med fel antal subpartitioner"
+ER_ADD_PARTITION_NO_NEW_PARTITION
+        eng "At least one partition must be added"
+        swe "tminstone en partition mste lggas till vid ADD PARTITION"
+ER_COALESCE_PARTITION_NO_PARTITION
+        eng "At least one partition must be coalesced"
+        swe "tminstone en partition mste sls ihop vid COALESCE PARTITION"
+ER_REORG_PARTITION_NOT_EXIST
+        eng "More partitions to reorganise than there are partitions"
+        swe "Fler partitioner att reorganisera n det finns partitioner"
+ER_SAME_NAME_PARTITION
+        eng "All partitions must have unique names in the table"
+        swe "Alla partitioner i tabellen mste ha unika namn"
+ER_CONSECUTIVE_REORG_PARTITIONS
+        eng "When reorganising a set of partitions they must be in consecutive order"
+        swe "Nr ett antal partitioner omorganiseras mste de vara i konsekutiv ordning"
+ER_REORG_OUTSIDE_RANGE
+        eng "The new partitions cover a bigger range then the reorganised partitions do"
+        swe "De nya partitionerna tcker ett strre intervall n de omorganiserade partitionerna"
+ER_DROP_PARTITION_FAILURE
+        eng "Drop partition not supported in this version for this handler"
+ER_DROP_PARTITION_WHEN_FK_DEFINED
+        eng "Cannot drop a partition when a foreign key constraint is defined on the table"
+        swe "Kan inte ta bort en partition nr en frmmande nyckel r definierad p tabellen"

sql/sql_lex.h

@@ -650,6 +650,10 @@ typedef class st_select_lex SELECT_LEX;
 #define ALTER_CONVERT          1024
 #define ALTER_FORCE		2048
 #define ALTER_RECREATE          4096
+#define ALTER_ADD_PARTITION     8192
+#define ALTER_DROP_PARTITION    16384
+#define ALTER_COALESCE_PARTITION 32768
+#define ALTER_REORGANISE_PARTITION   65536
 
 typedef struct st_alter_info
 {
@@ -658,9 +662,17 @@ typedef struct st_alter_info
   uint                        flags;
   enum enum_enable_or_disable keys_onoff;
   enum tablespace_op_type     tablespace_op;
+  List<char>                  partition_names;
+  uint                        no_parts;
 
   st_alter_info(){clear();}
-  void clear(){keys_onoff= LEAVE_AS_IS;tablespace_op= NO_TABLESPACE_OP;}
+  void clear()
+  {
+    keys_onoff= LEAVE_AS_IS;
+    tablespace_op= NO_TABLESPACE_OP;
+    no_parts= 0;
+    partition_names.empty();
+  }
   void reset(){drop_list.empty();alter_list.empty();clear();}
 } ALTER_INFO;
 

sql/sql_partition.cc

@@ -89,7 +89,106 @@ uint32 get_partition_id_hash_sub(partition_info *part_info);
 uint32 get_partition_id_key_sub(partition_info *part_info); 
 uint32 get_partition_id_linear_hash_sub(partition_info *part_info); 
 uint32 get_partition_id_linear_key_sub(partition_info *part_info); 
-    
+#endif
+
+
+/*
+  A routine used by the parser to decide whether we are specifying a full
+  partitioning or if only partitions to add or to split.
+  SYNOPSIS
+    is_partition_management()
+    lex                    Reference to the lex object
+  RETURN VALUE
+    TRUE                   Yes, it is part of a management partition command
+    FALSE                  No, not a management partition command
+  DESCRIPTION
+    This needs to be outside of HAVE_PARTITION_DB since it is used from the
+    sql parser that doesn't have any #ifdef's
+*/
+
+my_bool is_partition_management(LEX *lex)
+{
+  return (lex->sql_command == SQLCOM_ALTER_TABLE &&
+          (lex->alter_info.flags == ALTER_ADD_PARTITION ||
+           lex->alter_info.flags == ALTER_REORGANISE_PARTITION));
+}
+
+#ifdef HAVE_PARTITION_DB
+/*
+  A support function to check if a partition name is in a list of strings
+  SYNOPSIS
+    is_partition_in_list()
+    part_name          String searched for
+    list_part_names    A list of names searched in
+  RETURN VALUES
+    TRUE               String found
+    FALSE              String not found
+*/
+
+bool is_partition_in_list(char *part_name,
+                          List<char> list_part_names)
+{
+  List_iterator<char> part_names_it(list_part_names);
+  uint no_names= list_part_names.elements;
+  uint i= 0;
+  do
+  {
+    char *list_name= part_names_it++;
+    if (!(my_strcasecmp(system_charset_info, part_name, list_name)))
+      return TRUE;
+  } while (++i < no_names);
+  return FALSE;
+}
+
+
+/*
+  A support function to check partition names for duplication in a
+  partitioned table
+  SYNOPSIS
+    is_partitions_in_table()
+    new_part_info      New partition info
+    old_part_info      Old partition info
+  RETURN VALUES
+    TRUE               Duplicate names found
+    FALSE              Duplicate names not found
+  DESCRIPTION
+    Can handle that the new and old parts are the same in which case it
+    checks that the list of names in the partitions doesn't contain any
+    duplicated names.
+*/
+
+bool is_partitions_in_table(partition_info *new_part_info,
+                            partition_info *old_part_info)
+{
+  uint no_new_parts= new_part_info->partitions.elements, new_count;
+  uint no_old_parts= old_part_info->partitions.elements, old_count;
+  List_iterator<partition_element> new_parts_it(new_part_info->partitions);
+  bool same_part_info= (new_part_info == old_part_info);
+  DBUG_ENTER("is_partitions_in_table");
+
+  new_count= 0;
+  do
+  {
+    List_iterator<partition_element> old_parts_it(old_part_info->partitions);
+    char *new_name= (new_parts_it++)->partition_name;
+    new_count++;
+    old_count= 0;
+    do
+    {
+      char *old_name= (old_parts_it++)->partition_name;
+      old_count++;
+      if (same_part_info && old_count == new_count)
+        break;
+      if (!(my_strcasecmp(system_charset_info, old_name, new_name)))
+      {
+        DBUG_RETURN(TRUE);
+      }
+    } while (old_count < no_old_parts);
+  } while (new_count < no_new_parts);
+  DBUG_RETURN(FALSE);
+}
+
+
 /*
   A useful routine used by update_row for partition handlers to calculate
   the partition ids of the old and the new record.
@@ -415,7 +514,8 @@ static bool check_list_constants(partition_info *part_info)
 
 #define MAX_PART_NAME_SIZE 8
 
-static char *create_default_partition_names(uint no_parts, bool subpart)
+static char *create_default_partition_names(uint no_parts, uint start_no,
+                                            bool subpart)
 {
   char *ptr= sql_calloc(no_parts*MAX_PART_NAME_SIZE);
   char *move_ptr= ptr;
@@ -426,9 +526,9 @@ static char *create_default_partition_names(uint no_parts, bool subpart)
     do
     {
       if (subpart)
-        my_sprintf(move_ptr, (move_ptr,"sp%u", i));
+        my_sprintf(move_ptr, (move_ptr,"sp%u", (start_no + i)));
       else
-        my_sprintf(move_ptr, (move_ptr,"p%u", i));
+        my_sprintf(move_ptr, (move_ptr,"p%u", (start_no + i)));
       move_ptr+=MAX_PART_NAME_SIZE;
     } while (++i < no_parts);
   }
@@ -462,7 +562,8 @@ static char *create_default_partition_names(uint no_parts, bool subpart)
 */
 
 static bool set_up_default_partitions(partition_info *part_info,
-                                      handler *file, ulonglong max_rows)
+                                      handler *file, ulonglong max_rows,
+                                      uint start_no)
 {
   uint no_parts, i;
   char *default_name;
@@ -482,12 +583,14 @@ static bool set_up_default_partitions(partition_info *part_info,
   if (part_info->no_parts == 0)
     part_info->no_parts= file->get_default_no_partitions(max_rows);
   no_parts= part_info->no_parts;
+  part_info->use_default_partitions= FALSE;
   if (unlikely(no_parts > MAX_PARTITIONS))
   {
     my_error(ER_TOO_MANY_PARTITIONS_ERROR, MYF(0));
     goto end;
   }
   if (unlikely((!(default_name= create_default_partition_names(no_parts,
+                                                               start_no,
                                                                FALSE)))))
     goto end;
   i= 0;
@@ -537,8 +640,8 @@ static bool set_up_default_partitions(partition_info *part_info,
 static bool set_up_default_subpartitions(partition_info *part_info,
                                          handler *file, ulonglong max_rows)
 {
-  uint i, j= 0, no_parts, no_subparts;
-  char *default_name;
+  uint i, j, no_parts, no_subparts;
+  char *default_name, *name_ptr;
   bool result= TRUE;
   partition_element *part_elem;
   List_iterator<partition_element> part_it(part_info->partitions);
@@ -548,26 +651,29 @@ static bool set_up_default_subpartitions(partition_info *part_info,
     part_info->no_subparts= file->get_default_no_partitions(max_rows);
   no_parts= part_info->no_parts;
   no_subparts= part_info->no_subparts;
+  part_info->use_default_subpartitions= FALSE;
   if (unlikely((no_parts * no_subparts) > MAX_PARTITIONS))
   {
     my_error(ER_TOO_MANY_PARTITIONS_ERROR, MYF(0));
     goto end;
   }
   if (unlikely((!(default_name=
-                  create_default_partition_names(no_subparts, TRUE)))))
+             create_default_partition_names(no_subparts, (uint)0, TRUE)))))
     goto end;
   i= 0;
   do
   {
     part_elem= part_it++;
+    j= 0;
+    name_ptr= default_name;
     do
     {
       partition_element *subpart_elem= new partition_element();
       if (likely(subpart_elem != 0))
       {
         subpart_elem->engine_type= DB_TYPE_UNKNOWN;
-        subpart_elem->partition_name= default_name;
-        default_name+= MAX_PART_NAME_SIZE;
+        subpart_elem->partition_name= name_ptr;
+        name_ptr+= MAX_PART_NAME_SIZE;
         part_elem->subpartitions.push_back(subpart_elem);
       }
       else
@@ -598,14 +704,15 @@ static bool set_up_default_subpartitions(partition_info *part_info,
     Support routine for check_partition_info
 */
 
-static bool set_up_defaults_for_partitioning(partition_info *part_info,
-                                             handler *file,
-                                             ulonglong max_rows)
+bool set_up_defaults_for_partitioning(partition_info *part_info,
+                                      handler *file,
+                                      ulonglong max_rows, uint start_no)
 {
   DBUG_ENTER("set_up_defaults_for_partitioning");
 
   if (part_info->use_default_partitions)
-    DBUG_RETURN(set_up_default_partitions(part_info, file, max_rows));
+    DBUG_RETURN(set_up_default_partitions(part_info, file, max_rows,
+                                          start_no));
   if (is_sub_partitioned(part_info) && part_info->use_default_subpartitions)
     DBUG_RETURN(set_up_default_subpartitions(part_info, file, max_rows));
   DBUG_RETURN(FALSE);
@@ -682,7 +789,8 @@ bool check_partition_info(partition_info *part_info,enum db_type eng_type,
     my_error(ER_SUBPARTITION_ERROR, MYF(0));
     goto end;
   }
-  if (unlikely(set_up_defaults_for_partitioning(part_info, file, max_rows)))
+  if (unlikely(set_up_defaults_for_partitioning(part_info, file,
+                                                max_rows, (uint)0)))
     goto end;
   tot_partitions= get_tot_partitions(part_info);
   if (unlikely(tot_partitions > MAX_PARTITIONS))
@@ -690,6 +798,11 @@ bool check_partition_info(partition_info *part_info,enum db_type eng_type,
     my_error(ER_TOO_MANY_PARTITIONS_ERROR, MYF(0));
     goto end;
   }
+  if (unlikely(is_partitions_in_table(part_info, part_info)))
+  {
+    my_error(ER_SAME_NAME_PARTITION, MYF(0));
+    goto end;
+  }
   engine_array= (u_char*)my_malloc(tot_partitions, MYF(MY_WME));
   if (unlikely(!engine_array))
     goto end;
@@ -1524,11 +1637,9 @@ bool fix_partition_func(THD *thd, const char* name, TABLE *table)
   db_name= &db_name_string[home_dir_length];
   tables.db= db_name;
 
-  part_info->no_full_parts= part_info->no_parts;
   if (is_sub_partitioned(part_info))
   {
     DBUG_ASSERT(part_info->subpart_type == HASH_PARTITION);
-    part_info->no_full_parts= part_info->no_parts*part_info->no_subparts;
     /*
      Subpartition is defined. We need to verify that subpartitioning
      function is correct.
@@ -2768,7 +2879,7 @@ void get_partition_set(const TABLE *table, byte *buf, const uint index,
                        const key_range *key_spec, part_id_range *part_spec)
 {
   partition_info *part_info= table->s->part_info;
-  uint no_parts= part_info->no_full_parts, i, part_id;
+  uint no_parts= get_tot_partitions(part_info), i, part_id;
   uint sub_part= no_parts, part_part= no_parts;
   KEY *key_info= NULL;
   bool found_part_field= FALSE;

sql/sql_table.cc

@@ -45,6 +45,64 @@ static bool prepare_blob_field(THD *thd, create_field *sql_field);
 static bool check_engine(THD *thd, const char *table_name,
                          enum db_type *new_engine);                             
 
+/*
+  SYNOPSIS
+    write_bin_log()
+    thd                           Thread object
+    clear_error                   is clear_error to be called
+  RETURN VALUES
+    NONE
+  DESCRIPTION
+    Write the binlog if open, routine used in multiple places in this
+    file
+*/
+
+static void write_bin_log(THD *thd, bool clear_error)
+{
+  if (mysql_bin_log.is_open())
+  {
+    if (clear_error)
+      thd->clear_error();
+    Query_log_event qinfo(thd, thd->query, thd->query_length, FALSE, FALSE);
+    mysql_bin_log.write(&qinfo);
+  }
+}
+
+/*
+  SYNOPSIS
+    abort_and_upgrade_lock()
+    thd                           Thread object
+    table                         Table object
+    db                            Database name
+    table_name                    Table name
+    old_lock_level                Old lock level
+  RETURN VALUES
+    TRUE                          Failure
+    FALSE                         Success
+  DESCRIPTION
+    Remember old lock level (for possible downgrade later on), abort all
+    waiting threads and ensure that all keeping locks currently are
+    completed such that we own the lock exclusively and no other interaction
+    is ongoing.
+*/
+
+static bool abort_and_upgrade_lock(THD *thd, TABLE *table, const char *db,
+                                   const char *table_name,
+                                   uint *old_lock_level)
+{
+  uint flags= RTFC_WAIT_OTHER_THREAD_FLAG | RTFC_CHECK_KILLED_FLAG;
+  DBUG_ENTER("abort_and_upgrade_locks");
+
+  *old_lock_level= table->reginfo.lock_type;
+  mysql_lock_abort(thd, table);
+  VOID(remove_table_from_cache(thd, db, table_name, flags));
+  if (thd->killed)
+  {
+    thd->no_warnings_for_error= 0;
+    DBUG_RETURN(TRUE);
+  }
+  DBUG_RETURN(FALSE);
+}
 
 /*
  Build the path to a file for a table (or the base path that can
@@ -317,13 +375,8 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
   if (some_tables_deleted || tmp_table_deleted || !error)
   {
     query_cache_invalidate3(thd, tables, 0);
-    if (!dont_log_query && mysql_bin_log.is_open())
-    {
-      if (!error)
-        thd->clear_error();
-      Query_log_event qinfo(thd, thd->query, thd->query_length, FALSE, FALSE);
-      mysql_bin_log.write(&qinfo);
-    }
+    if (!dont_log_query)
+      write_bin_log(thd, !error);
   }
 
   unlock_table_names(thd, tables, (TABLE_LIST*) 0);
@@ -1709,12 +1762,8 @@ bool mysql_create_table(THD *thd,const char *db, const char *table_name,
     }
     thd->tmp_table_used= 1;
   }
-  if (!internal_tmp_table && mysql_bin_log.is_open())
-  {
-    thd->clear_error();
-    Query_log_event qinfo(thd, thd->query, thd->query_length, FALSE, FALSE);
-    mysql_bin_log.write(&qinfo);
-  }
+  if (!internal_tmp_table)
+    write_bin_log(thd, TRUE);
   error= FALSE;
 
 end:
@@ -2784,12 +2833,7 @@ bool mysql_create_like_table(THD* thd, TABLE_LIST* table,
   }
 
   // Must be written before unlock
-  if (mysql_bin_log.is_open())
-  {
-    thd->clear_error();
-    Query_log_event qinfo(thd, thd->query, thd->query_length, FALSE, FALSE);
-    mysql_bin_log.write(&qinfo);
-  }
+  write_bin_log(thd, TRUE);
   res= FALSE;
   goto err;
 
@@ -2895,11 +2939,7 @@ mysql_discard_or_import_tablespace(THD *thd,
     error=1;
   if (error)
     goto err;
-  if (mysql_bin_log.is_open())
-  {
-    Query_log_event qinfo(thd, thd->query, thd->query_length, FALSE, FALSE);
-    mysql_bin_log.write(&qinfo);
-  }
+  write_bin_log(thd, FALSE);
 err:
   close_thread_tables(thd);
   thd->tablespace_op=FALSE;
@@ -3301,7 +3341,10 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
   enum db_type old_db_type,new_db_type;
   uint need_copy_table= 0;
 #ifdef HAVE_PARTITION_DB
+  bool online_add_empty_partition= FALSE;
+  bool online_drop_partition= FALSE;
   bool partition_changed= FALSE;
+  enum db_type default_engine_type;
 #endif
   DBUG_ENTER("mysql_alter_table");
 
@@ -3378,66 +3421,421 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
   old_db_type= table->s->db_type;
   if (create_info->db_type == DB_TYPE_DEFAULT)
     create_info->db_type= old_db_type;
+
 #ifdef HAVE_PARTITION_DB
   /*
-   When thd->lex->part_info has a reference to a partition_info the
-   ALTER TABLE contained a definition of a partitioning.
-
-   Case I:
-     If there was a partition before and there is a new one defined.
-     We use the new partitioning. The new partitioning is already
-     defined in the correct variable so no work is needed to
-     accomplish this.
-     We do however need to update partition_changed to ensure that not
-     only the frm file is changed in the ALTER TABLE command.
-
-   Case IIa:
-     There was a partitioning before and there is no new one defined.
-     Also the user has not specified an explicit engine to use.
-
-     We use the old partitioning also for the new table. We do this
-     by assigning the partition_info from the table loaded in
-     open_ltable to the partition_info struct used by mysql_create_table
-     later in this method.
-
-   Case IIb:
-     There was a partitioning before and there is no new one defined.
-     The user has specified an explicit engine to use.
-
-     Since the user has specified an explicit engine to use we override
-     the old partitioning info and create a new table using the specified
-     engine. This is the reason for the extra check if old and new engine
-     is equal.
-     In this case the partition also is changed.
-
-   Case III:
-     There was no partitioning before altering the table, there is
-     partitioning defined in the altered table. Use the new partitioning.
-     No work needed since the partitioning info is already in the
-     correct variable.
-     Also here partition has changed and thus a new table must be
-     created.
-
-   Case IV:
-     There was no partitioning before and no partitioning defined. Obviously
-     no work needed.
-  */
-  if (table->s->part_info)
+   We need to handle both partition management command such as Add Partition
+   and others here as well as an ALTER TABLE that completely changes the
+   partitioning and yet others that don't change anything at all. We start
+   by checking the partition management variants and then check the general
+   change patterns.
+   */
+  if (alter_info->flags & (ALTER_ADD_PARTITION +
+      ALTER_DROP_PARTITION + ALTER_COALESCE_PARTITION +
+      ALTER_REORGANISE_PARTITION))
   {
-    if (!thd->lex->part_info &&
-        create_info->db_type == old_db_type)
-      thd->lex->part_info= table->s->part_info;
+    partition_info *tab_part_info= table->s->part_info;
+    if (!tab_part_info)
+    {
+      my_error(ER_PARTITION_MGMT_ON_NONPARTITIONED, MYF(0));
+      DBUG_RETURN(TRUE);
+    }
+    {
+      List_iterator<partition_element> t_it(tab_part_info->partitions);
+      partition_element *t_part_elem= t_it++;
+      if (is_sub_partitioned(tab_part_info))
+      {
+        List_iterator<partition_element> s_it(t_part_elem->subpartitions);
+        t_part_elem= s_it++;
+      }
+      default_engine_type= t_part_elem->engine_type;
+    }
+    /*
+      We are going to manipulate the partition info on the table object
+      so we need to ensure that the data structure of the table object
+      is freed by setting version to 0.
+    */
+    table->s->version= 0L;
+    if (alter_info->flags == ALTER_ADD_PARTITION)
+    {
+      /*
+        We start by moving the new partitions to the list of temporary
+        partitions. We will then check that the new partitions fit in the
+        partitioning scheme as currently set-up.
+        Partitions are always added at the end in ADD PARTITION.
+      */
+      partition_info *alt_part_info= thd->lex->part_info;
+      uint no_new_partitions= alt_part_info->no_parts;
+      uint no_orig_partitions= tab_part_info->no_parts;
+      uint check_total_partitions= no_new_partitions + no_orig_partitions;
+      uint new_total_partitions= check_total_partitions;
+      /*
+        We allow quite a lot of values to be supplied by defaults, however we
+        must know the number of new partitions in this case.
+      */ 
+      if (no_new_partitions == 0)
+      {
+        my_error(ER_ADD_PARTITION_NO_NEW_PARTITION, MYF(0));
+        DBUG_RETURN(TRUE);
+      }
+      if (is_sub_partitioned(tab_part_info))
+      {
+        if (alt_part_info->no_subparts == 0)
+          alt_part_info->no_subparts= tab_part_info->no_subparts;
+        else if (alt_part_info->no_subparts != tab_part_info->no_subparts)
+        {
+          my_error(ER_ADD_PARTITION_SUBPART_ERROR, MYF(0));
+          DBUG_RETURN(TRUE);
+        }
+        check_total_partitions= new_total_partitions*
+                                alt_part_info->no_subparts;
+      }
+      if (check_total_partitions > MAX_PARTITIONS)
+      {
+        my_error(ER_TOO_MANY_PARTITIONS_ERROR, MYF(0));
+        DBUG_RETURN(TRUE);
+      }
+      alt_part_info->part_type= tab_part_info->part_type;
+      if (set_up_defaults_for_partitioning(alt_part_info,
+                                           table->file,
+                                           (ulonglong)0ULL,
+                                           tab_part_info->no_parts))
+      {
+        DBUG_RETURN(TRUE);
+      }
+      /*
+        Need to concatenate the lists here to make it possible to check the
+        partition info for correctness using check_partition_info
+      */
+      {
+        List_iterator<partition_element> alt_it(alt_part_info->partitions);
+        uint part_count= 0;
+        do
+        {
+          partition_element *part_elem= alt_it++;
+          tab_part_info->partitions.push_back(part_elem);
+          tab_part_info->temp_partitions.push_back(part_elem);
+        } while (++part_count < no_new_partitions);
+        tab_part_info->no_parts+= no_new_partitions;
+      }
+      {
+        List_iterator<partition_element> tab_it(tab_part_info->partitions);
+        partition_element *part_elem= tab_it++;
+        if (is_sub_partitioned(tab_part_info))
+        {
+          List_iterator<partition_element> sub_it(part_elem->subpartitions);
+          part_elem= sub_it++;
+        }
+        if (check_partition_info(tab_part_info, part_elem->engine_type,
+                                 table->file, (ulonglong)0ULL))
+        {
+          DBUG_RETURN(TRUE);
+        }
+      }
+      create_info->db_type= DB_TYPE_PARTITION_DB;
+      thd->lex->part_info= tab_part_info;
+      if (table->file->alter_table_flags() & HA_ONLINE_ADD_EMPTY_PARTITION &&
+          (tab_part_info->part_type == RANGE_PARTITION ||
+           tab_part_info->part_type == LIST_PARTITION))
+      {
+        /*
+          For range and list partitions add partition is simply adding a new
+          empty partition to the table. If the handler support this we will
+          use the simple method of doing this. In this case we need to break
+          out the new partitions from the list again and only keep them in the
+          temporary list. Added partitions are always added at the end.
+        */
+        {
+          List_iterator<partition_element> tab_it(tab_part_info->partitions);
+          uint part_count= 0;
+          do
+          {
+            tab_it++;
+          } while (++part_count < no_orig_partitions);
+          do
+          {
+            tab_it++;
+            tab_it.remove();
+          } while (++part_count < new_total_partitions);
+        }
+        tab_part_info->no_parts-= no_new_partitions;
+        online_add_empty_partition= TRUE;
+      }
+      else
+      {
+        tab_part_info->temp_partitions.empty();
+      }
+    }
+    else if (alter_info->flags == ALTER_DROP_PARTITION)
+    {
+      /*
+        Drop a partition from a range partition and list partitioning is
+        always safe and can be made more or less immediate. It is necessary
+        however to ensure that the partition to be removed is safely removed
+        and that REPAIR TABLE can remove the partition if for some reason the
+        command to drop the partition failed in the middle.
+      */
+      uint part_count= 0;
+      uint no_parts_dropped= alter_info->partition_names.elements;
+      uint no_parts_found= 0;
+      List_iterator<partition_element> part_it(tab_part_info->partitions);
+      if (!(tab_part_info->part_type == RANGE_PARTITION ||
+            tab_part_info->part_type == LIST_PARTITION))
+      {
+        my_error(ER_ONLY_ON_RANGE_LIST_PARTITION, MYF(0), "DROP");
+        DBUG_RETURN(TRUE);
+      }
+      if (no_parts_dropped >= tab_part_info->no_parts)
+      {
+        my_error(ER_DROP_LAST_PARTITION, MYF(0));
+        DBUG_RETURN(TRUE);
+      }
+      do
+      {
+        partition_element *part_elem= part_it++;
+        if (is_partition_in_list(part_elem->partition_name,
+                                 alter_info->partition_names))
+        {
+          /*
+            Remove the partition from the list and put it instead in the
+            list of temporary partitions with a new state.
+          */
+          no_parts_found++;
+          part_elem->part_state= PART_IS_DROPPED;
+        }
+      } while (++part_count < tab_part_info->no_parts);
+      if (no_parts_found != no_parts_dropped)
+      {
+        my_error(ER_DROP_PARTITION_NON_EXISTENT, MYF(0));
+        DBUG_RETURN(TRUE);
+      }
+      if (!(table->file->alter_table_flags() & HA_ONLINE_DROP_PARTITION))
+      {
+        my_error(ER_DROP_PARTITION_FAILURE, MYF(0));
+        DBUG_RETURN(TRUE);
+      }
+      if (table->file->is_fk_defined_on_table_or_index(MAX_KEY))
+      {
+        my_error(ER_DROP_PARTITION_WHEN_FK_DEFINED, MYF(0));
+        DBUG_RETURN(TRUE);
+      }
+      /*
+        This code needs set-up of structures needed by mysql_create_table
+        before it is called and thus we only set a boolean variable to be
+        checked later down in the code when all needed data structures are
+        prepared.
+      */
+      online_drop_partition= TRUE;
+    }
+    else if (alter_info->flags == ALTER_COALESCE_PARTITION)
+    {
+      /*
+        In this version COALESCE PARTITION is implemented by simply removing
+        a partition from the table and using the normal ALTER TABLE code
+        and ensuring that copy to a new table occurs. Later on we can optimise
+        this function for Linear Hash partitions. In that case we can avoid
+        reorganising the entire table. For normal hash partitions it will
+        be a complete reorganise anyways so that can only be made on-line
+        if it still uses a copy table.
+      */
+      uint part_count= 0;
+      uint no_parts_coalesced= alter_info->no_parts;
+      uint no_parts_remain= tab_part_info->no_parts - no_parts_coalesced;
+      List_iterator<partition_element> part_it(tab_part_info->partitions);
+      if (tab_part_info->part_type != HASH_PARTITION)
+      {
+        my_error(ER_COALESCE_ONLY_ON_HASH_PARTITION, MYF(0));
+        DBUG_RETURN(TRUE);
+      }
+      if (no_parts_coalesced == 0)
+      {
+        my_error(ER_COALESCE_PARTITION_NO_PARTITION, MYF(0));
+        DBUG_RETURN(TRUE);
+      }
+      if (no_parts_coalesced >= tab_part_info->no_parts)
+      {
+        my_error(ER_DROP_LAST_PARTITION, MYF(0));
+        DBUG_RETURN(TRUE);
+      }
+      do
+      {
+        part_it++;
+        if (++part_count > no_parts_remain)
+          part_it.remove();
+      } while (part_count < tab_part_info->no_parts);
+      tab_part_info->no_parts= no_parts_remain;
+    }
+    else if (alter_info->flags == ALTER_REORGANISE_PARTITION)
+    {
+      /*
+        Reorganise partitions takes a number of partitions that are next
+        to each other (at least for RANGE PARTITIONS) and then uses those
+        to create a set of new partitions. So data is copied from those
+        partitions into the new set of partitions. Those new partitions
+        can have more values in the LIST value specifications or less both
+        are allowed. The ranges can be different but since they are 
+        changing a set of consecutive partitions they must cover the same
+        range as those changed from.
+        This command can be used on RANGE and LIST partitions.
+      */
+      uint no_parts_reorged= alter_info->partition_names.elements;
+      uint no_parts_new= thd->lex->part_info->partitions.elements;
+      partition_info *alt_part_info= thd->lex->part_info;
+      uint check_total_partitions;
+      if (no_parts_reorged > tab_part_info->no_parts)
+      {
+        my_error(ER_REORG_PARTITION_NOT_EXIST, MYF(0));
+        DBUG_RETURN(TRUE);
+      }
+      if (!(tab_part_info->part_type == RANGE_PARTITION ||
+            tab_part_info->part_type == LIST_PARTITION))
+      {
+        my_error(ER_ONLY_ON_RANGE_LIST_PARTITION, MYF(0), "REORGANISE");
+        DBUG_RETURN(TRUE);
+      }
+      if (is_partitions_in_table(alt_part_info, tab_part_info))
+      {
+        my_error(ER_SAME_NAME_PARTITION, MYF(0));
+        DBUG_RETURN(TRUE);
+      }
+      check_total_partitions= tab_part_info->no_parts + no_parts_new;
+      check_total_partitions-= no_parts_reorged;
+      if (check_total_partitions > MAX_PARTITIONS)
+      {
+        my_error(ER_TOO_MANY_PARTITIONS_ERROR, MYF(0));
+        DBUG_RETURN(TRUE);
+      }
+      {
+        List_iterator<partition_element> tab_it(tab_part_info->partitions);
+        uint part_count= 0;
+        bool found_first= FALSE, found_last= FALSE;
+        uint drop_count= 0;
+        longlong tab_max_range, alt_max_range;
+        do
+        {
+          partition_element *part_elem= tab_it++;
+          if (is_partition_in_list(part_elem->partition_name,
+                                   alter_info->partition_names))
+          {
+            drop_count++;
+            tab_max_range= part_elem->range_value;
+            if (!found_first)
+            {
+              uint alt_part_count= 0;
+              found_first= TRUE;
+              List_iterator<partition_element> alt_it(alt_part_info->partitions);
+              do
+              {
+                partition_element *alt_part_elem= alt_it++;
+                alt_max_range= alt_part_elem->range_value;
+                if (alt_part_count == 0)
+                  tab_it.replace(alt_part_elem);
+                else
+                  tab_it.after(alt_part_elem);
+              } while (++alt_part_count < no_parts_new);
+            }
+            else if (found_last)
+            {
+              my_error(ER_CONSECUTIVE_REORG_PARTITIONS, MYF(0));
+              DBUG_RETURN(TRUE);
+            }
+            else
+              tab_it.remove();
+          }
+          else
+          {
+            if (found_first)
+              found_last= TRUE;
+          }
+        } while (++part_count < tab_part_info->no_parts);
+        if (drop_count != no_parts_reorged)
+        {
+          my_error(ER_DROP_PARTITION_NON_EXISTENT, MYF(0));
+          DBUG_RETURN(TRUE);
+        }
+        if (tab_part_info->part_type == RANGE_PARTITION &&
+            alt_max_range > tab_max_range)
+        {
+          my_error(ER_REORG_OUTSIDE_RANGE, MYF(0));
+          DBUG_RETURN(TRUE);
+        }
+      }
+    }
+    partition_changed= TRUE;
+    create_info->db_type= DB_TYPE_PARTITION_DB;
+    thd->lex->part_info= tab_part_info;
+    if (alter_info->flags == ALTER_ADD_PARTITION ||
+        alter_info->flags == ALTER_REORGANISE_PARTITION)
+    {
+      if (check_partition_info(tab_part_info, default_engine_type,
+                               table->file, (ulonglong)0ULL))
+      {
+        DBUG_RETURN(TRUE);
+      }
+    }
   }
-  if (thd->lex->part_info)
+  else
   {
     /*
-      Need to cater for engine types that can handle partition without
-      using the partition handler.
+     When thd->lex->part_info has a reference to a partition_info the
+     ALTER TABLE contained a definition of a partitioning.
+
+     Case I:
+       If there was a partition before and there is a new one defined.
+       We use the new partitioning. The new partitioning is already
+       defined in the correct variable so no work is needed to
+       accomplish this.
+       We do however need to update partition_changed to ensure that not
+       only the frm file is changed in the ALTER TABLE command.
+
+     Case IIa:
+       There was a partitioning before and there is no new one defined.
+       Also the user has not specified an explicit engine to use.
+
+       We use the old partitioning also for the new table. We do this
+       by assigning the partition_info from the table loaded in
+       open_ltable to the partition_info struct used by mysql_create_table
+       later in this method.
+
+     Case IIb:
+       There was a partitioning before and there is no new one defined.
+       The user has specified an explicit engine to use.
+
+       Since the user has specified an explicit engine to use we override
+       the old partitioning info and create a new table using the specified
+       engine. This is the reason for the extra check if old and new engine
+       is equal.
+       In this case the partition also is changed.
+
+     Case III:
+       There was no partitioning before altering the table, there is
+       partitioning defined in the altered table. Use the new partitioning.
+       No work needed since the partitioning info is already in the
+       correct variable.
+       Also here partition has changed and thus a new table must be
+       created.
+
+     Case IV:
+       There was no partitioning before and no partitioning defined.
+       Obviously no work needed.
     */
-    if (thd->lex->part_info != table->s->part_info)
-      partition_changed= TRUE;
-    thd->lex->part_info->default_engine_type= create_info->db_type;
-    create_info->db_type= DB_TYPE_PARTITION_DB;
+    if (table->s->part_info)
+    {
+      if (!thd->lex->part_info &&
+          create_info->db_type == old_db_type)
+        thd->lex->part_info= table->s->part_info;
+    }
+    if (thd->lex->part_info)
+    {
+      /*
+        Need to cater for engine types that can handle partition without
+        using the partition handler.
+      */
+      if (thd->lex->part_info != table->s->part_info)
+        partition_changed= TRUE;
+      thd->lex->part_info->default_engine_type= create_info->db_type;
+      create_info->db_type= DB_TYPE_PARTITION_DB;
+    }
   }
 #endif
   if (check_engine(thd, new_name, &create_info->db_type))
@@ -3503,12 +3901,7 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
     }
     if (!error)
     {
-      if (mysql_bin_log.is_open())
-      {
-	thd->clear_error();
-	Query_log_event qinfo(thd, thd->query, thd->query_length, FALSE, FALSE);
-	mysql_bin_log.write(&qinfo);
-      }
+      write_bin_log(thd, TRUE);
       if (do_send_ok)
         send_ok(thd);
     }
@@ -3824,6 +4217,100 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
   if (!need_copy_table)
     create_info->frm_only= 1;
 
+  if (partition_changed)
+  {
+    if (online_drop_partition)
+    {
+      /*
+        Now after all checks and setting state on dropped partitions we can
+        start the actual dropping of the partitions.
+        1) Lock table in TL_WRITE_ONLY to ensure all other accesses on table
+           are completed and no new ones are started until we have changed
+           the frm file.
+        2) Write the new frm file where state of dropped partitions is
+           changed to PART_IS_DROPPED
+        3) Perform the actual drop of the partition using the handler of the
+           table.
+        4) Write a new frm file of the table where the partitions are dropped
+           from the table.
+
+      */
+      uint old_lock_type;
+      partition_info *part_info= table->s->part_info;
+      char path[FN_REFLEN+1];
+      uint db_options= 0, key_count, syntax_len;
+      KEY *key_info_buffer;
+      char *part_syntax_buf;
+
+      VOID(pthread_mutex_lock(&LOCK_open));
+      if (abort_and_upgrade_lock(thd, table, db, table_name, &old_lock_type))
+      {
+        DBUG_RETURN(TRUE);
+      }
+      VOID(pthread_mutex_unlock(&LOCK_open));
+      mysql_prepare_table(thd, create_info, &create_list,
+                          &key_list, /*tmp_table*/ 0, &db_options,
+                          table->file, &key_info_buffer, &key_count,
+                          /*select_field_count*/ 0);
+      if (!(part_syntax_buf= generate_partition_syntax(part_info,
+                                                       &syntax_len,
+                                                       TRUE)))
+      {
+        DBUG_RETURN(TRUE);
+      }
+      part_info->part_info_string= part_syntax_buf;
+      part_info->part_info_len= syntax_len;
+      build_table_path(path, sizeof(path), db, table_name, reg_ext);
+      if (mysql_create_frm(thd, path, db, table_name, create_info,
+                           create_list, key_count, key_info_buffer,
+                           table->file))
+      {
+        DBUG_RETURN(TRUE);
+      }
+      thd->lex->part_info= part_info;
+      build_table_path(path, sizeof(path), db, table_name, "");
+      if (table->file->drop_partitions(path))
+      {
+        DBUG_RETURN(TRUE);
+      }
+      {
+        List_iterator<partition_element> part_it(part_info->partitions);
+        uint i= 0, remove_count= 0;
+        do
+        {
+          partition_element *part_elem= part_it++;
+          if (is_partition_in_list(part_elem->partition_name,
+                                   alter_info->partition_names))
+          {
+            part_it.remove();
+            remove_count++;
+          }
+        } while (++i < part_info->no_parts);
+        part_info->no_parts-= remove_count;
+      }
+      if (!(part_syntax_buf= generate_partition_syntax(part_info,
+                                                       &syntax_len,
+                                                       TRUE)))
+      {
+        DBUG_RETURN(TRUE);
+      }
+      part_info->part_info_string= part_syntax_buf;
+      part_info->part_info_len= syntax_len;
+      build_table_path(path, sizeof(path), db, table_name, reg_ext);
+      if (mysql_create_frm(thd, path, db, table_name, create_info,
+                           create_list, key_count, key_info_buffer,
+                           table->file)  ||
+          table->file->create_handler_files(path))
+      {
+        DBUG_RETURN(TRUE);
+      }
+      thd->proc_info="end";
+      write_bin_log(thd, FALSE);
+      send_ok(thd);
+      DBUG_RETURN(FALSE);
+    }
+  }
+
   /*
     Handling of symlinked tables:
     If no rename:
@@ -3949,12 +4436,7 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
       my_free((gptr) new_table,MYF(0));
       goto err;
     }
-    if (mysql_bin_log.is_open())
-    {
-      thd->clear_error();
-      Query_log_event qinfo(thd, thd->query, thd->query_length, FALSE, FALSE);
-      mysql_bin_log.write(&qinfo);
-    }
+    write_bin_log(thd, TRUE);
     goto end_temporary;
   }
 
@@ -4085,15 +4567,14 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
     goto err;
   }
   thd->proc_info="end";
-  if (mysql_bin_log.is_open())
-  {
-    thd->clear_error();
-    Query_log_event qinfo(thd, thd->query, thd->query_length, FALSE, FALSE);
-    mysql_bin_log.write(&qinfo);
-  }
+  write_bin_log(thd, TRUE);
   VOID(pthread_cond_broadcast(&COND_refresh));
   VOID(pthread_mutex_unlock(&LOCK_open));
 #ifdef HAVE_BERKELEY_DB
+  /*
+    TODO RONM: This problem needs to handled for Berkeley DB partitions
+    as well
+  */
   if (old_db_type == DB_TYPE_BERKELEY_DB)
   {
     /*

sql/sql_yacc.yy

@@ -512,6 +512,7 @@ bool my_yyoverflow(short **a, YYSTYPE **b, ulong *yystacksize);
 %token  RELEASE_SYM
 %token  RELOAD
 %token  RENAME
+%token  REORGANISE_SYM
 %token  REPAIR
 %token  REPEATABLE_SYM
 %token  REPEAT_SYM
@@ -2618,7 +2619,15 @@ opt_partitioning:
 
 partitioning:
         PARTITION_SYM
-        { Lex->part_info= new partition_info(); }
+        {
+          LEX *lex= Lex;
+          lex->part_info= new partition_info();
+          if (!lex->part_info)
+          {
+            my_error(ER_OUTOFMEMORY, MYF(0), sizeof(partition_info));
+            YYABORT;
+          }
+        }
         partition
         ;
 
@@ -2633,6 +2642,11 @@ partition_entry:
               partition info string into part_info data structure.
             */
             lex->part_info= new partition_info();
+            if (!lex->part_info)
+            {
+              my_error(ER_OUTOFMEMORY, MYF(0), sizeof(partition_info));
+              YYABORT;
+            }
           }
           else
           {
@@ -2640,10 +2654,12 @@ partition_entry:
             YYABORT;
           }
         }
-        partition {};
+        partition {}
+        ;
 
 partition:
-        BY part_type_def opt_no_parts {} opt_sub_part {} part_defs;
+        BY part_type_def opt_no_parts {} opt_sub_part {} part_defs
+        ;
 
 part_type_def:
         opt_linear KEY_SYM '(' part_field_list ')'
@@ -2660,22 +2676,26 @@ part_type_def:
         part_func {}
         | LIST_SYM
         { Lex->part_info->part_type= LIST_PARTITION; }
-        part_func {};
+        part_func {}
+        ;
 
 opt_linear:
         /* empty */ {}
         | LINEAR_SYM
-        { Lex->part_info->linear_hash_ind= TRUE;};
+        { Lex->part_info->linear_hash_ind= TRUE;}
+        ;
 
 part_field_list:
         part_field_item {}
-        | part_field_list ',' part_field_item {};
+        | part_field_list ',' part_field_item {}
+        ;
 
 part_field_item:
         ident
         {
           Lex->part_info->part_field_list.push_back($1.str);
-        };
+        }
+        ;
 
 part_func:
         '(' remember_name part_func_expr remember_end ')'
@@ -2686,7 +2706,8 @@ part_func:
           lex->part_info->part_expr= $3;
           lex->part_info->part_func_string= $2+1;
           lex->part_info->part_func_len= expr_len;
-        };
+        }
+        ;
 
 sub_part_func:
         '(' remember_name part_func_expr remember_end ')'
@@ -2697,7 +2718,8 @@ sub_part_func:
           lex->part_info->subpart_expr= $3;
           lex->part_info->subpart_func_string= $2+1;
           lex->part_info->subpart_func_len= expr_len;
-        };
+        }
+        ;
 
 
 opt_no_parts:
@@ -2711,7 +2733,8 @@ opt_no_parts:
             YYABORT;
           }
           Lex->part_info->no_parts= no_parts;
-        };
+        }
+        ;
 
 opt_sub_part:
         /* empty */ {}
@@ -2725,15 +2748,18 @@ opt_sub_part:
           lex->part_info->subpart_type= HASH_PARTITION;
           lex->part_info->list_of_subpart_fields= TRUE;
         }
-        opt_no_subparts {};
+        opt_no_subparts {}
+        ;
 
 sub_part_field_list:
         sub_part_field_item {}
-        | sub_part_field_list ',' sub_part_field_item {};
+        | sub_part_field_list ',' sub_part_field_item {}
+        ;
 
 sub_part_field_item:
         ident
-        { Lex->part_info->subpart_field_list.push_back($1.str); };
+        { Lex->part_info->subpart_field_list.push_back($1.str); }
+        ;
 
 part_func_expr:
         bit_expr
@@ -2749,6 +2775,7 @@ part_func_expr:
           }
           $$=$1;
         }
+        ;
 
 opt_no_subparts:
         /* empty */ {}
@@ -2761,7 +2788,8 @@ opt_no_subparts:
             YYABORT;
           }
           Lex->part_info->no_subparts= no_parts;
-        };
+        }
+        ;
 
 part_defs:
         /* empty */
@@ -2785,11 +2813,13 @@ part_defs:
           }
           part_info->count_curr_subparts= 0;
           part_info->count_curr_parts= 0;
-        };
+        }
+        ;
 
 part_def_list:
         part_definition {}
-        | part_def_list ',' part_definition {};
+        | part_def_list ',' part_definition {}
+        ;
 
 part_definition:
         PARTITION_SYM
@@ -2812,47 +2842,61 @@ part_definition:
         part_name {}
         opt_part_values {}
         opt_part_options {}
-        opt_sub_partition {};
+        opt_sub_partition {}
+        ;
 
 part_name:
         ident_or_text
-        { Lex->part_info->curr_part_elem->partition_name= $1.str; };
+        { Lex->part_info->curr_part_elem->partition_name= $1.str; }
+        ;
 
 opt_part_values:
         /* empty */
         {
           LEX *lex= Lex;
-          if (lex->part_info->part_type == RANGE_PARTITION)
+          if (!is_partition_management(lex))
           {
-            my_error(ER_PARTITION_REQUIRES_VALUES_ERROR, MYF(0),
-            "RANGE", "LESS THAN");
-            YYABORT;
-          }
-          if (lex->part_info->part_type == LIST_PARTITION)
-          {
-            my_error(ER_PARTITION_REQUIRES_VALUES_ERROR, MYF(0),
-            "LIST", "IN");
-            YYABORT;
+            if (lex->part_info->part_type == RANGE_PARTITION)
+            {
+              my_error(ER_PARTITION_REQUIRES_VALUES_ERROR, MYF(0),
+              "RANGE", "LESS THAN");
+              YYABORT;
+            }
+            if (lex->part_info->part_type == LIST_PARTITION)
+            {
+              my_error(ER_PARTITION_REQUIRES_VALUES_ERROR, MYF(0),
+              "LIST", "IN");
+              YYABORT;
+            }
           }
         }
         | VALUES LESS_SYM THAN_SYM part_func_max
         {
-          if (Lex->part_info->part_type != RANGE_PARTITION)
+          LEX *lex= Lex;
+          if (!is_partition_management(lex))
           {
-            my_error(ER_PARTITION_WRONG_VALUES_ERROR, MYF(0),
-            "RANGE", "LESS THAN");
-            YYABORT;
+            if (Lex->part_info->part_type != RANGE_PARTITION)
+            {
+              my_error(ER_PARTITION_WRONG_VALUES_ERROR, MYF(0),
+              "RANGE", "LESS THAN");
+              YYABORT;
+            }
           }
         }
         | VALUES IN_SYM '(' part_list_func ')'
         {
-          if (Lex->part_info->part_type != LIST_PARTITION)
+          LEX *lex= Lex;
+          if (!is_partition_management(lex))
           {
-            my_error(ER_PARTITION_WRONG_VALUES_ERROR, MYF(0),
-            "LIST", "IN");
-            YYABORT;
+            if (Lex->part_info->part_type != LIST_PARTITION)
+            {
+              my_error(ER_PARTITION_WRONG_VALUES_ERROR, MYF(0),
+              "LIST", "IN");
+              YYABORT;
+            }
           }
-        };
+        }
+        ;
 
 part_func_max:
         MAX_VALUE_SYM
@@ -2873,17 +2917,20 @@ part_func_max:
             yyerror(ER(ER_PARTITION_MAXVALUE_ERROR));
             YYABORT;
           }
-        };
+        }
+        ;
 
 part_range_func:
         '(' part_bit_expr ')' 
         {
           Lex->part_info->curr_part_elem->range_value= $2;
-        };
+        }
+        ;
 
 part_list_func:
         part_list_item {}
-        | part_list_func ',' part_list_item {};
+        | part_list_func ',' part_list_item {}
+        ;
 
 part_list_item:
         part_bit_expr
@@ -2896,7 +2943,8 @@ part_list_item:
           }
           *value_ptr= $1;
           Lex->part_info->curr_part_elem->list_val_list.push_back(value_ptr);
-        };
+        }
+        ;
 
 part_bit_expr:
         bit_expr
@@ -2926,6 +2974,7 @@ part_bit_expr:
           item_value= part_expr->val_int();
           $$= item_value; 
         }
+        ;
 
 opt_sub_partition:
         /* empty */ {}
@@ -2947,11 +2996,13 @@ opt_sub_partition:
             part_info->no_subparts= part_info->count_curr_subparts;
           }
           part_info->count_curr_subparts= 0;
-        };
+        }
+        ;
 
 sub_part_list:
         sub_part_definition {}
-        | sub_part_list ',' sub_part_definition {};
+        | sub_part_list ',' sub_part_definition {}
+        ;
 
 sub_part_definition:
         SUBPARTITION_SYM
@@ -2970,19 +3021,23 @@ sub_part_definition:
           part_info->count_curr_subparts++;
           p_elem->engine_type= DB_TYPE_UNKNOWN;
         }
-        sub_name opt_part_options {};
+        sub_name opt_part_options {}
+        ;
 
 sub_name:
         ident_or_text
-        { Lex->part_info->curr_part_elem->partition_name= $1.str; };
+        { Lex->part_info->curr_part_elem->partition_name= $1.str; }
+        ;
 
 opt_part_options:
        /* empty */ {}
-       | opt_part_option_list {};
+       | opt_part_option_list {}
+       ;
 
 opt_part_option_list:
        opt_part_option_list opt_part_option {}
-       | opt_part_option {};
+       | opt_part_option {}
+       ;
 
 opt_part_option:
         TABLESPACE opt_equal ident_or_text
@@ -3000,7 +3055,8 @@ opt_part_option:
         | INDEX_SYM DIRECTORY_SYM opt_equal TEXT_STRING_sys
         { Lex->part_info->curr_part_elem->index_file_name= $4.str; }
         | COMMENT_SYM opt_equal TEXT_STRING_sys
-        { Lex->part_info->curr_part_elem->part_comment= $3.str; };
+        { Lex->part_info->curr_part_elem->part_comment= $3.str; }
+        ;
 
 /*
  End of partition parser part
@@ -3850,8 +3906,92 @@ alter_commands:
         | alter_list
         opt_partitioning
         | partitioning
+/*
+  This part was added for release 5.1 by Mikael Ronstrm.
+  From here we insert a number of commands to manage the partitions of a
+  partitioned table such as adding partitions, dropping partitions,
+  reorganising partitions in various manners. In future releases the list
+  will be longer and also include moving partitions to a
+  new table and so forth.
+*/
+        | add_partition_rule
+        | DROP PARTITION_SYM alt_part_name_list
+          {
+	    Lex->alter_info.flags|= ALTER_DROP_PARTITION;
+          }
+        | COALESCE PARTITION_SYM ulong_num
+          {
+            LEX *lex= Lex;
+	    lex->alter_info.flags|= ALTER_COALESCE_PARTITION;
+	    lex->alter_info.no_parts= $3;
+          }
+        | reorg_partition_rule
         ;
 
+add_partition_rule:
+        ADD PARTITION_SYM
+        {
+          LEX *lex= Lex;
+          lex->part_info= new partition_info();
+          if (!lex->part_info)
+          {
+            my_error(ER_OUTOFMEMORY, MYF(0), sizeof(partition_info));
+            YYABORT;
+          }
+	  lex->alter_info.flags|= ALTER_ADD_PARTITION;
+        }
+        add_part_extra
+        {}
+        ;
+
+add_part_extra:
+        | '(' part_def_list ')'
+        {
+          LEX *lex= Lex;
+          lex->part_info->no_parts= lex->part_info->count_curr_parts;
+        }
+        | PARTITIONS_SYM ulong_num
+        {
+          LEX *lex= Lex;
+          lex->part_info->no_parts= $2;
+        }
+        ;
+
+reorg_partition_rule:
+        REORGANISE_SYM PARTITION_SYM
+        {
+          LEX *lex= Lex;
+          lex->part_info= new partition_info();
+          if (!lex->part_info)
+          {
+            my_error(ER_OUTOFMEMORY, MYF(0), sizeof(partition_info));
+            YYABORT;
+          }
+	  lex->alter_info.flags|= ALTER_REORGANISE_PARTITION;
+        }
+        alt_part_name_list INTO '(' part_def_list ')'
+        {
+          LEX *lex= Lex;
+          lex->part_info->no_parts= lex->part_info->count_curr_parts;
+        }
+        ;
+
+alt_part_name_list:
+        alt_part_name_item {}
+        | alt_part_name_list ',' alt_part_name_item {}
+        ;
+
+alt_part_name_item:
+        ident
+        {
+          Lex->alter_info.partition_names.push_back($1.str);
+        }
+        ;
+
+/*
+  End of management of partition commands
+*/
+
 alter_list:
         alter_list_item
 	| alter_list ',' alter_list_item
@@ -7872,6 +8012,7 @@ keyword_sp:
 	| CHANGED		{}
 	| CIPHER_SYM		{}
 	| CLIENT_SYM		{}
+	| COALESCE		{}
 	| COLLATION_SYM		{}
         | COLUMNS               {}
 	| COMMITTED_SYM		{}
@@ -8007,6 +8148,7 @@ keyword_sp:
 	| RELAY_LOG_FILE_SYM	{}
 	| RELAY_LOG_POS_SYM	{}
 	| RELOAD		{}
+	| REORGANISE_SYM	{}
 	| REPEATABLE_SYM	{}
 	| REPLICATION		{}
 	| RESOURCES		{}