WL #2602, #2603, #2604

Added new syntax for partition management

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,6 +89,105 @@ 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
@@ -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,
+bool set_up_defaults_for_partitioning(partition_info *part_info,
                                       handler *file,
-                                             ulonglong max_rows)
+                                      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,7 +3421,361 @@ 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
+  /*
+   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))
+  {
+    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);
+      }
+    }
+  }
+  else
+  {
     /*
      When thd->lex->part_info has a reference to a partition_info the
      ALTER TABLE contained a definition of a partitioning.
@@ -3419,8 +3816,8 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
        created.
 
      Case IV:
-     There was no partitioning before and no partitioning defined. Obviously
-     no work needed.
+       There was no partitioning before and no partitioning defined.
+       Obviously no work needed.
     */
     if (table->s->part_info)
     {
@@ -3439,6 +3836,7 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
       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))
     DBUG_RETURN(TRUE);
@@ -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,16 +2842,20 @@ 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 (!is_partition_management(lex))
+          {
             if (lex->part_info->part_type == RANGE_PARTITION)
             {
               my_error(ER_PARTITION_REQUIRES_VALUES_ERROR, MYF(0),
@@ -2835,7 +2869,11 @@ opt_part_values:
               YYABORT;
             }
           }
+        }
         | VALUES LESS_SYM THAN_SYM part_func_max
+        {
+          LEX *lex= Lex;
+          if (!is_partition_management(lex))
           {
             if (Lex->part_info->part_type != RANGE_PARTITION)
             {
@@ -2844,7 +2882,11 @@ opt_part_values:
               YYABORT;
             }
           }
+        }
         | VALUES IN_SYM '(' part_list_func ')'
+        {
+          LEX *lex= Lex;
+          if (!is_partition_management(lex))
           {
             if (Lex->part_info->part_type != LIST_PARTITION)
             {
@@ -2852,7 +2894,9 @@ opt_part_values:
               "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		{}