BUG#20770: DATA DIRECTORY and INDEX DIRECTORY error when ALTER TABLE ADD/DROP/REORGANIZE partition

           Also some error in handling options for subpartitions.


mysql-test/r/partition.result:
  New test cases
mysql-test/t/partition.test:
  New test cases
sql/ha_partition.cc:
  Added partition_element to prepare_new_partition so that we can properly set-up table
  before creating partitions.
sql/ha_partition.h:
  Added partition_element to prepare_new_partition so that we can properly set-up table
  before creating partitions.
sql/sql_yacc.yy:
  Ensure that subpartitions always inherit options from the partition they belong to.
  They can change it afterwards but will use the options as set on partition level
  if set at that level.

mysql-test/r/partition.result

@@ -1082,4 +1082,43 @@ a
 2
 1
 drop table t1;
+create table t1 (a int) engine myisam
+partition by range (a)
+subpartition by hash (a)
+(partition p0 VALUES LESS THAN (1) DATA DIRECTORY = 'hello/master-data/tmpdata' INDEX DIRECTORY = 'hello/master-data/tmpinx'
+(SUBPARTITION subpart00, SUBPARTITION subpart01));
+t1#P#p0#SP#subpart00.MYD
+t1#P#p0#SP#subpart00.MYI
+t1#P#p0#SP#subpart01.MYD
+t1#P#p0#SP#subpart01.MYI
+t1.frm
+t1.par
+t1#P#p0#SP#subpart00.MYD
+t1#P#p0#SP#subpart01.MYD
+t1#P#p0#SP#subpart00.MYI
+t1#P#p0#SP#subpart01.MYI
+ALTER TABLE t1 REORGANIZE PARTITION p0 INTO
+(partition p1 VALUES LESS THAN (1) DATA DIRECTORY = 'hello/master-data/tmpdata' INDEX DIRECTORY = 'hello/master-data/tmpinx'
+(SUBPARTITION subpart10, SUBPARTITION subpart11),
+partition p2 VALUES LESS THAN (2) DATA DIRECTORY = 'hello/master-data/tmpdata' INDEX DIRECTORY = 'hello/master-data/tmpinx'
+(SUBPARTITION subpart20, SUBPARTITION subpart21));
+t1#P#p1#SP#subpart10.MYD
+t1#P#p1#SP#subpart10.MYI
+t1#P#p1#SP#subpart11.MYD
+t1#P#p1#SP#subpart11.MYI
+t1#P#p2#SP#subpart20.MYD
+t1#P#p2#SP#subpart20.MYI
+t1#P#p2#SP#subpart21.MYD
+t1#P#p2#SP#subpart21.MYI
+t1.frm
+t1.par
+t1#P#p1#SP#subpart10.MYD
+t1#P#p1#SP#subpart11.MYD
+t1#P#p2#SP#subpart20.MYD
+t1#P#p2#SP#subpart21.MYD
+t1#P#p1#SP#subpart10.MYI
+t1#P#p1#SP#subpart11.MYI
+t1#P#p2#SP#subpart20.MYI
+t1#P#p2#SP#subpart21.MYI
+drop table t1;
 End of 5.1 tests

mysql-test/t/partition.test

@@ -1261,4 +1261,47 @@ insert into t1 values (1),(2);
 select * from t1 ORDER BY a DESC;
 drop table t1;
 
+#
+# Bug 20770 Partitions: DATA DIRECTORY clause change in reorganize
+#                       doesn't remove old directory
+#
+--disable_query_log
+--exec mkdir $MYSQLTEST_VARDIR/master-data/tmpdata || true
+eval SET @data_dir = 'DATA DIRECTORY = ''$MYSQLTEST_VARDIR/master-data/tmpdata''';
+let $data_directory = `select @data_dir`;
+
+--exec mkdir $MYSQLTEST_VARDIR/master-data/tmpinx || true
+eval SET @inx_dir = 'INDEX DIRECTORY = ''$MYSQLTEST_VARDIR/master-data/tmpinx''';
+let $inx_directory = `select @inx_dir`;
+--enable_query_log
+
+--replace_result $MYSQLTEST_VARDIR "hello"
+eval create table t1 (a int) engine myisam
+partition by range (a)
+subpartition by hash (a)
+(partition p0 VALUES LESS THAN (1) $data_directory $inx_directory
+ (SUBPARTITION subpart00, SUBPARTITION subpart01));
+
+--replace_result $MYSQLTEST_VARDIR "hello"
+--exec ls $MYSQLTEST_VARDIR/master-data/test || true
+--replace_result $MYSQLTEST_VARDIR "hello"
+--exec ls $MYSQLTEST_VARDIR/master-data/tmpdata || true
+--replace_result $MYSQLTEST_VARDIR "hello"
+--exec ls $MYSQLTEST_VARDIR/master-data/tmpinx || true
+--replace_result $MYSQLTEST_VARDIR "hello"
+
+eval ALTER TABLE t1 REORGANIZE PARTITION p0 INTO
+(partition p1 VALUES LESS THAN (1) $data_directory $inx_directory
+ (SUBPARTITION subpart10, SUBPARTITION subpart11),
+ partition p2 VALUES LESS THAN (2) $data_directory $inx_directory
+ (SUBPARTITION subpart20, SUBPARTITION subpart21));
+
+--replace_result $MYSQLTEST_VARDIR "hello"
+--exec ls $MYSQLTEST_VARDIR/master-data/test || true
+--replace_result $MYSQLTEST_VARDIR "hello"
+--exec ls $MYSQLTEST_VARDIR/master-data/tmpdata || true
+--replace_result $MYSQLTEST_VARDIR "hello"
+--exec ls $MYSQLTEST_VARDIR/master-data/tmpinx || true
+
+drop table t1;
 --echo End of 5.1 tests

sql/ha_partition.cc

@@ -1125,13 +1125,15 @@ int ha_partition::handle_opt_partitions(THD *thd, HA_CHECK_OPT *check_opt,
 
 int ha_partition::prepare_new_partition(TABLE *table,
                                         HA_CREATE_INFO *create_info,
-                                        handler *file, const char *part_name)
+                                        handler *file, const char *part_name,
+                                        partition_element *p_elem)
 {
   int error;
   bool create_flag= FALSE;
   bool open_flag= FALSE;
   DBUG_ENTER("prepare_new_partition");
 
+  set_up_table_before_create(table, part_name, create_info, 0, p_elem);
   if ((error= file->create(part_name, table, create_info)))
     goto error;
   create_flag= TRUE;
@@ -1420,7 +1422,8 @@ int ha_partition::change_partitions(HA_CREATE_INFO *create_info,
           DBUG_PRINT("info", ("Add subpartition %s", part_name_buff));
           if ((error= prepare_new_partition(table, create_info,
                                             new_file_array[part],
-                                            (const char *)part_name_buff)))
+                                            (const char *)part_name_buff,
+                                            sub_elem)))
           {
             cleanup_new_partition(part_count);
             DBUG_RETURN(error);
@@ -1436,7 +1439,8 @@ int ha_partition::change_partitions(HA_CREATE_INFO *create_info,
         DBUG_PRINT("info", ("Add partition %s", part_name_buff));
         if ((error= prepare_new_partition(table, create_info,
                                           new_file_array[i],
-                                          (const char *)part_name_buff)))
+                                          (const char *)part_name_buff,
+                                          part_elem)))
         {
           cleanup_new_partition(part_count);
           DBUG_RETURN(error);
@@ -1648,7 +1652,7 @@ uint ha_partition::del_ren_cre_table(const char *from,
       error= (*file)->delete_table((const char*) from_buff);
     else
     {
-      set_up_table_before_create(table_arg, from_buff, create_info, i);
+      set_up_table_before_create(table_arg, from_buff, create_info, i, NULL);
       error= (*file)->create(from_buff, table_arg, create_info);
     }
     name_buffer_ptr= strend(name_buffer_ptr) + 1;
@@ -1724,12 +1728,15 @@ partition_element *ha_partition::find_partition_element(uint part_id)
 void ha_partition::set_up_table_before_create(TABLE *table,
                    const char *partition_name_with_path, 
                    HA_CREATE_INFO *info,
-                   uint part_id)
+                   uint part_id,
+                   partition_element *part_elem)
 {
-  partition_element *part_elem= find_partition_element(part_id);
-
   if (!part_elem)
-    return;                                     // Fatal error
+  {
+    part_elem= find_partition_element(part_id);
+    if (!part_elem)
+      return;                                   // Fatal error
+  }
   table->s->max_rows= part_elem->part_max_rows;
   table->s->min_rows= part_elem->part_min_rows;
   const char *partition_name= strrchr(partition_name_with_path, FN_LIBCHAR);

sql/ha_partition.h

@@ -202,7 +202,8 @@ class ha_partition :public handler
   int copy_partitions(ulonglong *copied, ulonglong *deleted);
   void cleanup_new_partition(uint part_count);
   int prepare_new_partition(TABLE *table, HA_CREATE_INFO *create_info,
-                            handler *file, const char *part_name);
+                            handler *file, const char *part_name,
+                            partition_element *p_elem);
   /*
     delete_table, rename_table and create uses very similar logic which
     is packed into this routine.
@@ -222,7 +223,8 @@ class ha_partition :public handler
   void set_up_table_before_create(TABLE *table_arg,
                                   const char *partition_name_with_path,
                                   HA_CREATE_INFO *info,
-                                  uint part_id);
+                                  uint part_id,
+                                  partition_element *p_elem);
   partition_element *find_partition_element(uint part_id);
 
 public:

sql/sql_yacc.yy

@@ -3716,14 +3716,15 @@ sub_part_definition:
         {
           LEX *lex= Lex;
           partition_info *part_info= lex->part_info;
-          partition_element *p_elem= new partition_element();
-          if (!p_elem ||
-           part_info->current_partition->subpartitions.push_back(p_elem))
+          partition_element *curr_part= part_info->current_partition;
+          partition_element *sub_p_elem= new partition_element(curr_part);
+          if (!sub_p_elem ||
+           curr_part->subpartitions.push_back(sub_p_elem))
           {
             mem_alloc_error(sizeof(partition_element));
             YYABORT;
           }
-          part_info->curr_part_elem= p_elem;
+          part_info->curr_part_elem= sub_p_elem;
           part_info->use_default_subpartitions= FALSE;
           part_info->use_default_no_subpartitions= FALSE;
           part_info->count_curr_subparts++;