Merge c-4908e253.1238-1-64736c10.cust.bredbandsbolaget.se:/home/pappa/clean-mysql-5.1-new

into  c-4908e253.1238-1-64736c10.cust.bredbandsbolaget.se:/home/pappa/wl2826

BUILD/SETUP.sh

@@ -71,6 +71,7 @@ pentium_cflags="$check_cpu_cflags"
 pentium64_cflags="$check_cpu_cflags -m64"
 ppc_cflags="$check_cpu_cflags"
 sparc_cflags=""
+error_inject="--with-error-inject "
 
 # be as fast as we can be without losing our ability to backtrace
 fast_cflags="-O3 -fno-omit-frame-pointer"

BUILD/compile-pentium-debug-max

@@ -6,6 +6,6 @@ path=`dirname $0`
 extra_flags="$pentium_cflags $debug_cflags $max_cflags"
 c_warnings="$c_warnings $debug_extra_warnings"
 cxx_warnings="$cxx_warnings $debug_extra_warnings"
-extra_configs="$pentium_configs $debug_configs $max_configs"
+extra_configs="$pentium_configs $debug_configs $max_configs $error_inject"
 
 . "$path/FINISH.sh"

configure.in

@@ -666,6 +666,7 @@ else
   AC_MSG_RESULT([no])
 fi
 
+  
 MYSQL_SYS_LARGEFILE
 
 # Types that must be checked AFTER large file support is checked
@@ -1587,6 +1588,18 @@ then
   DEBUG_OPTIMIZE_CXX=""
 fi
 
+# If we should allow error injection tests
+AC_ARG_WITH(error-inject,
+    [ --with-error-inject    Enable error injection in MySQL Server],
+    [ with_error_inject=$withval ],
+    [ with_error_inject=no ])
+
+if test "$with_error_inject" = "yes"
+then
+  CFLAGS="-DERROR_INJECT_SUPPORT $CFLAGS"
+  CXXFLAGS="-DERROR_INJECT_SUPPORT $CXXFLAGS"
+fi
+
 AC_ARG_WITH(debug,
     [  --with-debug            Add debug code
   --with-debug=full       Add debug code (adds memory checker, very slow)],

include/my_sys.h

@@ -559,7 +559,7 @@ extern File my_register_filename(File fd, const char *FileName,
 				 enum file_type type_of_file,
 				 uint error_message_number, myf MyFlags);
 extern File my_create(const char *FileName,int CreateFlags,
-		      int AccsesFlags, myf MyFlags);
+		      int AccessFlags, myf MyFlags);
 extern int my_close(File Filedes,myf MyFlags);
 extern File my_dup(File file, myf MyFlags);
 extern int my_mkdir(const char *dir, int Flags, myf MyFlags);

mysql-test/r/ndb_partition_key.result

@@ -165,6 +165,19 @@ ENGINE=NDB
 PARTITION BY KEY(c3) PARTITIONS 5;
 ALTER TABLE t1 COALESCE PARTITION 4;
 DROP TABLE t1;
+CREATE TABLE t1 (a int primary key)
+ENGINE=NDB
+PARTITION BY KEY(a);
+ALTER TABLE t1 OPTIMIZE PARTITION p0;
+ERROR HY000: Table storage engine for 't1' doesn't have this option
+ALTER TABLE t1 CHECK PARTITION p0;
+ERROR HY000: Table storage engine for 't1' doesn't have this option
+ALTER TABLE t1 REPAIR PARTITION p0;
+ERROR HY000: Table storage engine for 't1' doesn't have this option
+ALTER TABLE t1 ANALYZE PARTITION p0;
+ERROR HY000: Table storage engine for 't1' doesn't have this option
+ALTER TABLE t1 REBUILD PARTITION p0;
+ERROR HY000: Table storage engine for 't1' doesn't have this option
 CREATE TABLE t1 (
 c1 MEDIUMINT NOT NULL AUTO_INCREMENT,
 c2 TEXT NOT NULL,

mysql-test/t/ndb_partition_key.test

@@ -154,6 +154,24 @@ ALTER TABLE t1 COALESCE PARTITION 4;
 
 DROP TABLE t1;
 
+#
+# Bug 16822: OPTIMIZE TABLE hangs test
+#
+CREATE TABLE t1 (a int primary key)
+ENGINE=NDB
+PARTITION BY KEY(a);
+--error 1031
+ALTER TABLE t1 OPTIMIZE PARTITION p0;
+--error 1031
+ALTER TABLE t1 CHECK PARTITION p0;
+--error 1031
+ALTER TABLE t1 REPAIR PARTITION p0;
+--error 1031
+ALTER TABLE t1 ANALYZE PARTITION p0;
+--error 1031
+ALTER TABLE t1 REBUILD PARTITION p0;
+DROP TABLE t1;
+
 #
 # BUG 16806: ALTER TABLE fails
 #

sql/ha_ndbcluster.cc

@@ -4687,7 +4687,9 @@ int ha_ndbcluster::create(const char *name,
   DBUG_RETURN(my_errno);
 }
 
-int ha_ndbcluster::create_handler_files(const char *file) 
+int ha_ndbcluster::create_handler_files(const char *file,
+                                        const char *old_name,
+                                        bool rename_flag) 
 { 
   const char *name;
   Ndb* ndb;
@@ -4698,6 +4700,10 @@ int ha_ndbcluster::create_handler_files(const char *file)
 
   DBUG_ENTER("create_handler_files");
 
+  if (rename_flag)
+  {
+    DBUG_RETURN(FALSE);
+  }
   if (!(ndb= get_ndb()))
     DBUG_RETURN(HA_ERR_NO_CONNECTION);
 

sql/ha_ndbcluster.h

@@ -611,7 +611,8 @@ class ha_ndbcluster: public handler
   int rename_table(const char *from, const char *to);
   int delete_table(const char *name);
   int create(const char *name, TABLE *form, HA_CREATE_INFO *info);
-  int create_handler_files(const char *file);
+  int create_handler_files(const char *file, const char *old_name,
+                           bool rename_flag);
   int get_default_no_partitions(ulonglong max_rows);
   bool get_no_parts(const char *name, uint *no_parts);
   void set_auto_partitions(partition_info *part_info);

sql/ha_partition.cc

@@ -403,88 +403,6 @@ int ha_partition::ha_initialise()
 /****************************************************************************
                 MODULE meta data changes
 ****************************************************************************/
-/*
-  Create partition names
-
-  SYNOPSIS
-    create_partition_name()
-    out:out                   Created partition name string
-    in1                       First part
-    in2                       Second part
-    name_variant              Normal, temporary or renamed partition name
-
-  RETURN VALUE
-    NONE
-
-  DESCRIPTION
-    This method is used to calculate the partition name, service routine to
-    the del_ren_cre_table method.
-*/
-
-#define NORMAL_PART_NAME 0
-#define TEMP_PART_NAME 1
-#define RENAMED_PART_NAME 2
-static void create_partition_name(char *out, const char *in1,
-                                  const char *in2, uint name_variant,
-                                  bool translate)
-{
-  char transl_part_name[FN_REFLEN];
-  const char *transl_part;
-
-  if (translate)
-  {
-    tablename_to_filename(in2, transl_part_name, FN_REFLEN);
-    transl_part= transl_part_name;
-  }
-  else
-    transl_part= in2;
-  if (name_variant == NORMAL_PART_NAME)
-    strxmov(out, in1, "#P#", transl_part, NullS);
-  else if (name_variant == TEMP_PART_NAME)
-    strxmov(out, in1, "#P#", transl_part, "#TMP#", NullS);
-  else if (name_variant == RENAMED_PART_NAME)
-    strxmov(out, in1, "#P#", transl_part, "#REN#", NullS);
-}
-
-/*
-  Create subpartition name
-
-  SYNOPSIS
-    create_subpartition_name()
-    out:out                   Created partition name string
-    in1                       First part
-    in2                       Second part
-    in3                       Third part
-    name_variant              Normal, temporary or renamed partition name
-
-  RETURN VALUE
-    NONE
-
-  DESCRIPTION
-  This method is used to calculate the subpartition 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,
-                                     uint name_variant)
-{
-  char transl_part_name[FN_REFLEN], transl_subpart_name[FN_REFLEN];
-
-  tablename_to_filename(in2, transl_part_name, FN_REFLEN);
-  tablename_to_filename(in3, transl_subpart_name, FN_REFLEN);
-  if (name_variant == NORMAL_PART_NAME)
-    strxmov(out, in1, "#P#", transl_part_name,
-            "#SP#", transl_subpart_name, NullS);
-  else if (name_variant == TEMP_PART_NAME)
-    strxmov(out, in1, "#P#", transl_part_name,
-            "#SP#", transl_subpart_name, "#TMP#", NullS);
-  else if (name_variant == RENAMED_PART_NAME)
-    strxmov(out, in1, "#P#", transl_part_name,
-            "#SP#", transl_subpart_name, "#REN#", NullS);
-}
-
-
 /*
   Delete a table
 
@@ -575,7 +493,9 @@ int ha_partition::rename_table(const char *from, const char *to)
     and types of engines in the partitions.
 */
 
-int ha_partition::create_handler_files(const char *name)
+int ha_partition::create_handler_files(const char *path,
+                                       const char *old_path,
+                                       bool rename_flag)
 {
   DBUG_ENTER("ha_partition::create_handler_files()");
 
@@ -583,10 +503,26 @@ int ha_partition::create_handler_files(const char *name)
     We need to update total number of parts since we might write the handler
     file as part of a partition management command
   */
-  if (create_handler_file(name))
+  if (rename_flag)
   {
-    my_error(ER_CANT_CREATE_HANDLER_FILE, MYF(0));
+    char name[FN_REFLEN];
-    DBUG_RETURN(1);
+    char old_name[FN_REFLEN];
+
+    strxmov(name, path, ha_par_ext, NullS);
+    strxmov(old_name, old_path, ha_par_ext, NullS);
+    if (my_delete(name, MYF(MY_WME)) ||
+        my_rename(old_name, name, MYF(MY_WME)))
+    {
+      DBUG_RETURN(TRUE);
+    }
+  }
+  else
+  {
+    if (create_handler_file(path))
+    {
+      my_error(ER_CANT_CREATE_HANDLER_FILE, MYF(0));
+      DBUG_RETURN(1);
+    }
   }
   DBUG_RETURN(0);
 }
@@ -652,45 +588,26 @@ int ha_partition::create(const char *name, TABLE *table_arg,
 int ha_partition::drop_partitions(const char *path)
 {
   List_iterator<partition_element> part_it(m_part_info->partitions);
-  List_iterator<partition_element> temp_it(m_part_info->temp_partitions);
   char part_name_buff[FN_REFLEN];
   uint no_parts= m_part_info->partitions.elements;
   uint part_count= 0;
   uint no_subparts= m_part_info->no_subparts;
   uint i= 0;
   uint name_variant;
-  int  error= 1;
+  int  ret_error;
-  bool reorged_parts= (m_reorged_parts > 0);
+  int  error= 0;
-  bool temp_partitions= (m_part_info->temp_partitions.elements > 0);
   DBUG_ENTER("ha_partition::drop_partitions");
 
-  if (temp_partitions)
-    no_parts= m_part_info->temp_partitions.elements;
   do
   {
-    partition_element *part_elem;
+    partition_element *part_elem= part_it++;
-    if (temp_partitions)
+    if (part_elem->part_state == PART_TO_BE_DROPPED)
-    {
-      /*
-        We need to remove the reorganised partitions that were put in the
-        temp_partitions-list.
-      */
-      part_elem= temp_it++;
-      DBUG_ASSERT(part_elem->part_state == PART_TO_BE_DROPPED);
-    }
-    else
-      part_elem= part_it++;
-    if (part_elem->part_state == PART_TO_BE_DROPPED ||
-        part_elem->part_state == PART_IS_CHANGED)
     {
       handler *file;
       /*
         This part is to be dropped, meaning the part or all its subparts.
       */
       name_variant= NORMAL_PART_NAME;
-      if (part_elem->part_state == PART_IS_CHANGED ||
-          (part_elem->part_state == PART_TO_BE_DROPPED && temp_partitions))
-        name_variant= RENAMED_PART_NAME;
       if (m_is_sub_partitioned)
       {
         List_iterator<partition_element> sub_it(part_elem->subpartitions);
@@ -702,12 +619,10 @@ int ha_partition::drop_partitions(const char *path)
           create_subpartition_name(part_name_buff, path,
                                    part_elem->partition_name,
                                    sub_elem->partition_name, name_variant);
-          if (reorged_parts)
+          file= m_file[part];
-            file= m_reorged_file[part_count++];
-          else
-            file= m_file[part];
           DBUG_PRINT("info", ("Drop subpartition %s", part_name_buff));
-          error= file->delete_table((const char *) part_name_buff);
+          if ((ret_error= file->delete_table((const char *) part_name_buff)))
+            error= ret_error;
         } while (++j < no_subparts);
       }
       else
@@ -715,12 +630,10 @@ int ha_partition::drop_partitions(const char *path)
         create_partition_name(part_name_buff, path,
                               part_elem->partition_name, name_variant,
                               TRUE);
-        if (reorged_parts)
+        file= m_file[i];
-          file= m_reorged_file[part_count++];
-        else
-          file= m_file[i];
         DBUG_PRINT("info", ("Drop partition %s", part_name_buff));
-        error= file->delete_table((const char *) part_name_buff);
+        if ((ret_error= file->delete_table((const char *) part_name_buff)))
+          error= ret_error;
       }
       if (part_elem->part_state == PART_IS_CHANGED)
         part_elem->part_state= PART_NORMAL;
@@ -762,7 +675,8 @@ int ha_partition::rename_partitions(const char *path)
   uint no_subparts= m_part_info->no_subparts;
   uint i= 0;
   uint j= 0;
-  int error= 1;
+  int error= 0;
+  int ret_error;
   uint temp_partitions= m_part_info->temp_partitions.elements;
   handler *file;
   partition_element *part_elem, *sub_elem;
@@ -770,6 +684,14 @@ int ha_partition::rename_partitions(const char *path)
 
   if (temp_partitions)
   {
+    /*
+      These are the reorganised partitions that have already been copied.
+      We delete the partitions and log the delete by inactivating the
+      delete log entry in the table log. We only need to synchronise
+      these writes before moving to the next loop since there is no
+      interaction among reorganised partitions, they cannot have the
+      same name.
+    */
     do
     {
       part_elem= temp_it++;
@@ -780,39 +702,59 @@ int ha_partition::rename_partitions(const char *path)
         {
           sub_elem= sub_it++;
           file= m_reorged_file[part_count++];
-          create_subpartition_name(part_name_buff, path,
-                                   part_elem->partition_name,
-                                   sub_elem->partition_name,
-                                   RENAMED_PART_NAME);
           create_subpartition_name(norm_name_buff, path,
                                    part_elem->partition_name,
                                    sub_elem->partition_name,
                                    NORMAL_PART_NAME);
-          DBUG_PRINT("info", ("Rename subpartition from %s to %s",
+          DBUG_PRINT("info", ("Delete subpartition %s", norm_name_buff));
-                     norm_name_buff, part_name_buff));
+          if ((ret_error= file->delete_table((const char *) norm_name_buff)))
-          error= file->rename_table((const char *) norm_name_buff,
+            error= ret_error;
-                                    (const char *) part_name_buff);
+          else if (deactivate_ddl_log_entry(sub_elem->log_entry->entry_pos))
+            error= 1;
+          else
+            sub_elem->log_entry= NULL; /* Indicate success */
         } while (++j < no_subparts);
       }
       else
       {
         file= m_reorged_file[part_count++];
-        create_partition_name(part_name_buff, path,
-                              part_elem->partition_name, RENAMED_PART_NAME,
-                              TRUE);
         create_partition_name(norm_name_buff, path,
                               part_elem->partition_name, NORMAL_PART_NAME,
                               TRUE);
-        DBUG_PRINT("info", ("Rename partition from %s to %s",
+        DBUG_PRINT("info", ("Delete partition %s", norm_name_buff));
-                   norm_name_buff, part_name_buff));
+        if ((ret_error= file->delete_table((const char *) norm_name_buff)))
-        error= file->rename_table((const char *) norm_name_buff,
+          error= ret_error;
-                                  (const char *) part_name_buff);
+        else if (deactivate_ddl_log_entry(part_elem->log_entry->entry_pos))
+          error= 1;
+        else
+          part_elem->log_entry= NULL; /* Indicate success */
       }
     } while (++i < temp_partitions);
+    VOID(sync_ddl_log());
   }
   i= 0;
   do
   {
+    /*
+       When state is PART_IS_CHANGED it means that we have created a new
+       TEMP partition that is to be renamed to normal partition name and
+       we are to delete the old partition with currently the normal name.
+       
+       We perform this operation by
+       1) Delete old partition with normal partition name
+       2) Signal this in table log entry
+       3) Synch table log to ensure we have consistency in crashes
+       4) Rename temporary partition name to normal partition name
+       5) Signal this to table log entry
+       It is not necessary to synch the last state since a new rename
+       should not corrupt things if there was no temporary partition.
+
+       The only other parts we need to cater for are new parts that
+       replace reorganised parts. The reorganised parts were deleted
+       by the code above that goes through the temp_partitions list.
+       Thus the synch above makes it safe to simply perform step 4 and 5
+       for those entries.
+    */
     part_elem= part_it++;
     if (part_elem->part_state == PART_IS_CHANGED ||
         (part_elem->part_state == PART_IS_ADDED && temp_partitions))
@@ -834,14 +776,12 @@ int ha_partition::rename_partitions(const char *path)
           if (part_elem->part_state == PART_IS_CHANGED)
           {
             file= m_reorged_file[part_count++];
-            create_subpartition_name(part_name_buff, path,
+            DBUG_PRINT("info", ("Delete subpartition %s", norm_name_buff));
-                                     part_elem->partition_name,
+            if ((ret_error= file->delete_table((const char *) norm_name_buff)))
-                                     sub_elem->partition_name,
+              error= ret_error;
-                                     RENAMED_PART_NAME);
+            else if (deactivate_ddl_log_entry(sub_elem->log_entry->entry_pos))
-            DBUG_PRINT("info", ("Rename subpartition from %s to %s",
+              error= 1;
-                       norm_name_buff, part_name_buff));
+            VOID(sync_ddl_log());
-            error= file->rename_table((const char *) norm_name_buff,
-                                      (const char *) part_name_buff);
           }
           file= m_new_file[part];
           create_subpartition_name(part_name_buff, path,
@@ -850,8 +790,13 @@ int ha_partition::rename_partitions(const char *path)
                                    TEMP_PART_NAME);
           DBUG_PRINT("info", ("Rename subpartition from %s to %s",
                      part_name_buff, norm_name_buff));
-          error= file->rename_table((const char *) part_name_buff,
+          if ((ret_error= file->rename_table((const char *) part_name_buff,
-                                    (const char *) norm_name_buff);
+                                             (const char *) norm_name_buff)))
+            error= ret_error;
+          else if (deactivate_ddl_log_entry(sub_elem->log_entry->entry_pos))
+            error= 1;
+          else
+            sub_elem->log_entry= NULL;
         } while (++j < no_subparts);
       }
       else
@@ -862,13 +807,12 @@ int ha_partition::rename_partitions(const char *path)
         if (part_elem->part_state == PART_IS_CHANGED)
         {
           file= m_reorged_file[part_count++];
-          create_partition_name(part_name_buff, path,
+          DBUG_PRINT("info", ("Delete partition %s", norm_name_buff));
-                                part_elem->partition_name, RENAMED_PART_NAME,
+          if ((ret_error= file->delete_table((const char *) norm_name_buff)))
-                                TRUE);
+            error= ret_error;
-          DBUG_PRINT("info", ("Rename partition from %s to %s",
+          else if (deactivate_ddl_log_entry(part_elem->log_entry->entry_pos))
-                     norm_name_buff, part_name_buff));
+            error= 1;
-          error= file->rename_table((const char *) norm_name_buff,
+          VOID(sync_ddl_log());
-                                    (const char *) part_name_buff);
         }
         file= m_new_file[i];
         create_partition_name(part_name_buff, path,
@@ -876,11 +820,17 @@ int ha_partition::rename_partitions(const char *path)
                               TRUE);
         DBUG_PRINT("info", ("Rename partition from %s to %s",
                    part_name_buff, norm_name_buff));
-        error= file->rename_table((const char *) part_name_buff,
+        if ((ret_error= file->rename_table((const char *) part_name_buff,
-                                  (const char *) norm_name_buff);
+                                           (const char *) norm_name_buff)))
+          error= ret_error;
+        else if (deactivate_ddl_log_entry(part_elem->log_entry->entry_pos))
+          error= 1;
+        else
+          part_elem->log_entry= NULL;
       }
     }
   } while (++i < no_parts);
+  VOID(sync_ddl_log());
   DBUG_RETURN(error);
 }
 
@@ -1872,8 +1822,8 @@ bool ha_partition::create_handler_file(const char *name)
   {
     part_elem= part_it++;
     if (part_elem->part_state != PART_NORMAL &&
-        part_elem->part_state != PART_IS_ADDED &&
+        part_elem->part_state != PART_TO_BE_ADDED &&
-        part_elem->part_state != PART_IS_CHANGED)
+        part_elem->part_state != PART_CHANGED)
       continue;
     tablename_to_filename(part_elem->partition_name, part_name,
                           FN_REFLEN);
@@ -1924,8 +1874,8 @@ bool ha_partition::create_handler_file(const char *name)
   {
     part_elem= part_it++;
     if (part_elem->part_state != PART_NORMAL &&
-        part_elem->part_state != PART_IS_ADDED &&
+        part_elem->part_state != PART_TO_BE_ADDED &&
-        part_elem->part_state != PART_IS_CHANGED)
+        part_elem->part_state != PART_CHANGED)
       continue;
     if (!m_is_sub_partitioned)
     {

sql/ha_partition.h

@@ -179,7 +179,8 @@ public:
   virtual int rename_table(const char *from, const char *to);
   virtual int create(const char *name, TABLE *form,
 		     HA_CREATE_INFO *create_info);
-  virtual int create_handler_files(const char *name);
+  virtual int create_handler_files(const char *name,
+                                   const char *old_name, bool rename_flag);
   virtual void update_create_info(HA_CREATE_INFO *create_info);
   virtual char *update_table_comment(const char *comment);
   virtual int change_partitions(HA_CREATE_INFO *create_info,

sql/handler.h

@@ -632,6 +632,7 @@ typedef struct {
 
 #define UNDEF_NODEGROUP 65535
 class Item;
+struct st_table_log_memory_entry;
 
 class partition_info;
 
@@ -639,7 +640,6 @@ struct st_partition_iter;
 #define NOT_A_PARTITION_ID ((uint32)-1)
 
 
-
 typedef struct st_ha_create_information
 {
   CHARSET_INFO *table_charset, *default_table_charset;
@@ -1378,7 +1378,11 @@ public:
   virtual void drop_table(const char *name);
   
   virtual int create(const char *name, TABLE *form, HA_CREATE_INFO *info)=0;
-  virtual int create_handler_files(const char *name) { return FALSE;}
+  virtual int create_handler_files(const char *name, const char *old_name,
+                                   bool rename_flag)
+  {
+    return FALSE;
+  }
 
   virtual int change_partitions(HA_CREATE_INFO *create_info,
                                 const char *path,
@@ -1386,7 +1390,7 @@ public:
                                 ulonglong *deleted,
                                 const void *pack_frm_data,
                                 uint pack_frm_len)
-  { return HA_ERR_WRONG_COMMAND; }
+  { print_error(HA_ERR_WRONG_COMMAND, MYF(0)); return TRUE; }
   virtual int drop_partitions(const char *path)
   { return HA_ERR_WRONG_COMMAND; }
   virtual int rename_partitions(const char *path)

sql/mysql_priv.h

@@ -617,6 +617,101 @@ struct Query_cache_query_flags
 #define query_cache_invalidate_by_MyISAM_filename_ref NULL
 #endif /*HAVE_QUERY_CACHE*/
 
+/*
+  Error injector Macros to enable easy testing of recovery after failures
+  in various error cases.
+*/
+#ifndef ERROR_INJECT_SUPPORT
+
+#define ERROR_INJECT(x) 0
+#define ERROR_INJECT_ACTION(x,action) 0
+#define ERROR_INJECT_CRASH(x) 0
+#define ERROR_INJECT_VALUE(x) 0
+#define ERROR_INJECT_VALUE_ACTION(x,action) 0
+#define ERROR_INJECT_VALUE_CRASH(x) 0
+#define SET_ERROR_INJECT_VALUE(x)
+
+#else
+
+inline bool
+my_error_inject_name(const char *dbug_str)
+{
+  const char *extra_str= "-d,";
+  char total_str[200];
+  if (_db_strict_keyword_ (dbug_str))
+  {
+    strxmov(total_str, extra_str, dbug_str, NullS);
+    DBUG_SET(total_str);
+    return 1;
+  }
+  return 0;
+}
+
+
+inline bool
+my_error_inject(int value)
+{
+  THD *thd= current_thd;
+  if (thd->error_inject_value == (uint)value)
+  {
+    thd->error_inject_value= 0;
+    return 1;
+  }
+  return 0;
+}
+
+/*
+  ERROR INJECT MODULE:
+  --------------------
+  These macros are used to insert macros from the application code.
+  The event that activates those error injections can be activated
+  from SQL by using:
+  SET SESSION dbug=+d,code;
+
+  After the error has been injected, the macros will automatically
+  remove the debug code, thus similar to using:
+  SET SESSION dbug=-d,code
+  from SQL.
+
+  ERROR_INJECT_CRASH will inject a crash of the MySQL Server if code
+  is set when macro is called. ERROR_INJECT_CRASH can be used in
+  if-statements, it will always return FALSE unless of course it
+  crashes in which case it doesn't return at all.
+
+  ERROR_INJECT_ACTION will inject the action specified in the action
+  parameter of the macro, before performing the action the code will
+  be removed such that no more events occur. ERROR_INJECT_ACTION
+  can also be used in if-statements and always returns FALSE.
+  ERROR_INJECT can be used in a normal if-statement, where the action
+  part is performed in the if-block. The macro returns TRUE if the
+  error was activated and otherwise returns FALSE. If activated the
+  code is removed.
+
+  Sometimes it is necessary to perform error inject actions as a serie
+  of events. In this case one can use one variable on the THD object.
+  Thus one sets this value by using e.g. SET_ERROR_INJECT_VALUE(100).
+  Then one can later test for it by using ERROR_INJECT_CRASH_VALUE,
+  ERROR_INJECT_ACTION_VALUE and ERROR_INJECT_VALUE. This have the same
+  behaviour as the above described macros except that they use the
+  error inject value instead of a code used by DBUG macros.
+*/
+#define SET_ERROR_INJECT_VALUE(x) \
+  current_thd->error_inject_value= (x)
+#define ERROR_INJECT_CRASH(code) \
+  DBUG_EVALUATE_IF(code, (abort(), 0), 0)
+#define ERROR_INJECT_ACTION(code, action) \
+  (my_error_inject_name(code) ? ((action), 0) : 0)
+#define ERROR_INJECT(code) \
+  my_error_inject_name(code)
+#define ERROR_INJECT_VALUE(value) \
+  my_error_inject(value)
+#define ERROR_INJECT_VALUE_ACTION(value,action) \
+  (my_error_inject(value) ? (action) : 0)
+#define ERROR_INJECT_VALUE_CRASH(value) \
+  (my_error_inject(value) ? abort() : 0)
+
+#endif
+
 uint build_table_path(char *buff, size_t bufflen, const char *db,
                       const char *table, const char *ext);
 void write_bin_log(THD *thd, bool clear_error,
@@ -1093,6 +1188,16 @@ uint prep_alter_part_table(THD *thd, TABLE *table, ALTER_INFO *alter_info,
 bool remove_table_from_cache(THD *thd, const char *db, const char *table,
                              uint flags);
 
+#define NORMAL_PART_NAME 0
+#define TEMP_PART_NAME 1
+#define RENAMED_PART_NAME 2
+void create_partition_name(char *out, const char *in1,
+                           const char *in2, uint name_variant,
+                           bool translate);
+void create_subpartition_name(char *out, const char *in1,
+                              const char *in2, const char *in3,
+                              uint name_variant);
+
 typedef struct st_lock_param_type
 {
   ulonglong copied;
@@ -1112,14 +1217,97 @@ typedef struct st_lock_param_type
   uint key_count;
   uint db_options;
   uint pack_frm_len;
+  partition_info *part_info;
 } ALTER_PARTITION_PARAM_TYPE;
 
 void mem_alloc_error(size_t size);
-#define WFRM_INITIAL_WRITE 1
+
-#define WFRM_CREATE_HANDLER_FILES 2
+enum ddl_log_entry_code
+{
+  /*
+    DDL_LOG_EXECUTE_CODE:
+      This is a code that indicates that this is a log entry to
+      be executed, from this entry a linked list of log entries
+      can be found and executed.
+    DDL_LOG_ENTRY_CODE:
+      An entry to be executed in a linked list from an execute log
+      entry.
+    DDL_IGNORE_LOG_ENTRY_CODE:
+      An entry that is to be ignored
+  */
+  DDL_LOG_EXECUTE_CODE = 'e',
+  DDL_LOG_ENTRY_CODE = 'l',
+  DDL_IGNORE_LOG_ENTRY_CODE = 'i'
+};
+
+enum ddl_log_action_code
+{
+  /*
+    The type of action that a DDL_LOG_ENTRY_CODE entry is to
+    perform.
+    DDL_LOG_DELETE_ACTION:
+      Delete an entity
+    DDL_LOG_RENAME_ACTION:
+      Rename an entity
+    DDL_LOG_REPLACE_ACTION:
+      Rename an entity after removing the previous entry with the
+      new name, that is replace this entry.
+  */
+  DDL_LOG_DELETE_ACTION = 'd',
+  DDL_LOG_RENAME_ACTION = 'r',
+  DDL_LOG_REPLACE_ACTION = 's'
+};
+
+
+typedef struct st_ddl_log_entry
+{
+  const char *name;
+  const char *from_name;
+  const char *handler_name;
+  uint next_entry;
+  uint entry_pos;
+  enum ddl_log_entry_code entry_type;
+  enum ddl_log_action_code action_type;
+  /*
+    Most actions have only one phase. REPLACE does however have two
+    phases. The first phase removes the file with the new name if
+    there was one there before and the second phase renames the
+    old name to the new name.
+  */
+  char phase;
+} DDL_LOG_ENTRY;
+
+typedef struct st_ddl_log_memory_entry
+{
+  uint entry_pos;
+  struct st_ddl_log_memory_entry *next_log_entry;
+  struct st_ddl_log_memory_entry *prev_log_entry;
+  struct st_ddl_log_memory_entry *next_active_log_entry;
+} DDL_LOG_MEMORY_ENTRY;
+
+
+
+#define DDL_LOG_HANDLER_TYPE_LEN 32
+
+bool write_ddl_log_entry(DDL_LOG_ENTRY *ddl_log_entry,
+                           DDL_LOG_MEMORY_ENTRY **active_entry);
+bool write_execute_ddl_log_entry(uint first_entry,
+                                   bool complete,
+                                   DDL_LOG_MEMORY_ENTRY **active_entry);
+bool deactivate_ddl_log_entry(uint entry_no);
+void release_ddl_log_memory_entry(DDL_LOG_MEMORY_ENTRY *log_entry);
+bool sync_ddl_log();
+void release_ddl_log();
+void execute_ddl_log_recovery();
+bool execute_ddl_log_entry(uint first_entry);
+void lock_global_ddl_log();
+void unlock_global_ddl_log();
+
+#define WFRM_WRITE_SHADOW 1
+#define WFRM_INSTALL_SHADOW 2
 #define WFRM_PACK_FRM 4
 bool mysql_write_frm(ALTER_PARTITION_PARAM_TYPE *lpt, uint flags);
-bool abort_and_upgrade_lock(ALTER_PARTITION_PARAM_TYPE *lpt);
+void abort_and_upgrade_lock(ALTER_PARTITION_PARAM_TYPE *lpt);
 void close_open_tables_and_downgrade(ALTER_PARTITION_PARAM_TYPE *lpt);
 void mysql_wait_completed_table(ALTER_PARTITION_PARAM_TYPE *lpt, TABLE *my_table);
 
@@ -1290,6 +1478,9 @@ extern ulong delayed_insert_timeout;
 extern ulong delayed_insert_limit, delayed_queue_size;
 extern ulong delayed_insert_threads, delayed_insert_writes;
 extern ulong delayed_rows_in_use,delayed_insert_errors;
+#ifdef ERROR_INJECT_SUPPORT
+extern ulong error_inject_value;
+#endif
 extern ulong slave_open_temp_tables;
 extern ulong query_cache_size, query_cache_min_res_unit;
 extern ulong slow_launch_threads, slow_launch_time;

sql/mysqld.cc

@@ -3665,6 +3665,7 @@ we force server id to 2, but this MySQL server will not act as a slave.");
       unireg_abort(1);
     }
   }
+  execute_ddl_log_recovery();
 
   create_shutdown_thread();
   create_maintenance_thread();
@@ -3695,6 +3696,7 @@ we force server id to 2, but this MySQL server will not act as a slave.");
   /* (void) pthread_attr_destroy(&connection_attrib); */
   
   DBUG_PRINT("quit",("Exiting main thread"));
+  release_ddl_log();
 
 #ifndef __WIN__
 #ifdef EXTRA_DEBUG2

sql/partition_element.h

@@ -36,6 +36,8 @@ enum partition_state {
   PART_IS_ADDED= 8
 };
 
+struct st_ddl_log_memory_entry;
+
 class partition_element :public Sql_alloc {
 public:
   List<partition_element> subpartitions;
@@ -44,6 +46,7 @@ public:
   ulonglong part_min_rows;
   char *partition_name;
   char *tablespace_name;
+  struct st_ddl_log_memory_entry *log_entry;
   longlong range_value;
   char* part_comment;
   char* data_file_name;
@@ -55,7 +58,8 @@ public:
 
   partition_element()
   : part_max_rows(0), part_min_rows(0), partition_name(NULL),
-    tablespace_name(NULL), range_value(0), part_comment(NULL),
+    tablespace_name(NULL), log_entry(NULL),
+    range_value(0), part_comment(NULL),
     data_file_name(NULL), index_file_name(NULL),
     engine_type(NULL),part_state(PART_NORMAL),
     nodegroup_id(UNDEF_NODEGROUP), has_null_value(FALSE)

sql/partition_info.h

@@ -28,7 +28,7 @@ typedef int (*get_part_id_func)(partition_info *part_info,
                                  longlong *func_value);
 typedef uint32 (*get_subpart_id_func)(partition_info *part_info);
 
-
+struct st_ddl_log_memory_entry;
 
 class partition_info : public Sql_alloc
 {
@@ -76,7 +76,11 @@ public:
   Item *subpart_expr;
 
   Item *item_free_list;
-  
+
+  struct st_ddl_log_memory_entry *first_log_entry;
+  struct st_ddl_log_memory_entry *exec_log_entry;
+  struct st_ddl_log_memory_entry *frm_log_entry;
+
   /* 
     A bitmap of partitions used by the current query. 
     Usage pattern:
@@ -191,6 +195,7 @@ public:
     part_field_array(NULL), subpart_field_array(NULL),
     full_part_field_array(NULL),
     part_expr(NULL), subpart_expr(NULL), item_free_list(NULL),
+    first_log_entry(NULL), exec_log_entry(NULL), frm_log_entry(NULL),
     list_array(NULL),
     part_info_string(NULL),
     part_func_string(NULL), subpart_func_string(NULL),

sql/share/errmsg.txt

@@ -5826,3 +5826,5 @@ ER_NDB_CANT_SWITCH_BINLOG_FORMAT
 	eng "The NDB cluster engine does not support changing the binlog format on the fly yet"
 ER_PARTITION_NO_TEMPORARY
 	eng "Cannot create temporary table with partitions"
+ER_DDL_LOG_ERROR
+        eng "Error in DDL log"

sql/sql_base.cc

@@ -6156,23 +6156,17 @@ bool is_equal(const LEX_STRING *a, const LEX_STRING *b)
     old_lock_level                Old lock level
 */
 
-bool abort_and_upgrade_lock(ALTER_PARTITION_PARAM_TYPE *lpt)
+void abort_and_upgrade_lock(ALTER_PARTITION_PARAM_TYPE *lpt)
 {
   uint flags= RTFC_WAIT_OTHER_THREAD_FLAG | RTFC_CHECK_KILLED_FLAG;
-  int error= FALSE;
   DBUG_ENTER("abort_and_upgrade_locks");
 
   lpt->old_lock_type= lpt->table->reginfo.lock_type;
   VOID(pthread_mutex_lock(&LOCK_open));
   mysql_lock_abort(lpt->thd, lpt->table, TRUE);
   VOID(remove_table_from_cache(lpt->thd, lpt->db, lpt->table_name, flags));
-  if (lpt->thd->killed)
-  {
-    lpt->thd->no_warnings_for_error= 0;
-    error= TRUE;
-  }
   VOID(pthread_mutex_unlock(&LOCK_open));
-  DBUG_RETURN(error);
+  DBUG_VOID_RETURN;
 }
 
 

sql/sql_class.cc

@@ -223,6 +223,9 @@ THD::THD()
   cuted_fields= sent_row_count= 0L;
   limit_found_rows= 0;
   statement_id_counter= 0UL;
+#ifdef ERROR_INJECT_SUPPORT
+  error_inject_value= 0UL;
+#endif
   // Must be reset to handle error with THD's created for init of mysqld
   lex->current_select= 0;
   start_time=(time_t) 0;

sql/sql_class.h

@@ -1119,6 +1119,9 @@ public:
   query_id_t query_id, warn_id;
   ulong      thread_id, col_access;
 
+#ifdef ERROR_INJECT_SUPPORT
+  ulong      error_inject_value;
+#endif
   /* Statement id is thread-wide. This counter is used to generate ids */
   ulong      statement_id_counter;
   ulong	     rand_saved_seed1, rand_saved_seed2;

sql/sql_yacc.yy

@@ -3557,75 +3557,14 @@ part_definition:
           LEX *lex= Lex;
           partition_info *part_info= lex->part_info;
           partition_element *p_elem= new partition_element();
-          uint part_id= part_info->partitions.elements +
+          uint part_id= part_info->partitions.elements;
-                        part_info->temp_partitions.elements;
-          enum partition_state part_state;
 
-          if (part_info->part_state)
+          if (!p_elem || part_info->partitions.push_back(p_elem))
-            part_state= (enum partition_state)part_info->part_state[part_id];
-          else
-            part_state= PART_NORMAL;
-          switch (part_state)
           {
-            case PART_TO_BE_DROPPED:
+            mem_alloc_error(sizeof(partition_element));
-            /*
+            YYABORT;
-              This part is currently removed so we keep it in a
-              temporary list for REPAIR TABLE to be able to handle
-              failures during drop partition process.
-            */
-            case PART_TO_BE_ADDED:
-            /*
-              This part is currently being added so we keep it in a
-              temporary list for REPAIR TABLE to be able to handle
-              failures during add partition process.
-            */
-              if (!p_elem || part_info->temp_partitions.push_back(p_elem))
-              {
-                mem_alloc_error(sizeof(partition_element));
-                YYABORT;
-              }
-              break;
-            case PART_IS_ADDED:
-            /*
-              Part has been added and is now a normal partition
-            */
-            case PART_TO_BE_REORGED:
-            /*
-              This part is currently reorganised, it is still however
-              used so we keep it in the list of partitions. We do
-              however need the state to be able to handle REPAIR TABLE
-              after failures in the reorganisation process.
-            */
-            case PART_REORGED_DROPPED:
-            /*
-              This part is currently reorganised as part of a
-              COALESCE PARTITION and it will be dropped without a new
-              replacement partition after completing the reorganisation.
-            */
-            case PART_CHANGED:
-            /*
-              This part is currently split or merged as part of ADD
-              PARTITION for a hash partition or as part of COALESCE
-              PARTITION for a hash partitioned table.
-            */
-            case PART_IS_CHANGED:
-            /*
-              This part has been split or merged as part of ADD
-              PARTITION for a hash partition or as part of COALESCE
-              PARTITION for a hash partitioned table.
-            */
-            case PART_NORMAL:
-              if (!p_elem || part_info->partitions.push_back(p_elem))
-              {
-                mem_alloc_error(sizeof(partition_element));
-                YYABORT;
-              }
-              break;
-            default:
-              mem_alloc_error((part_id * 1000) + part_state);
-              YYABORT;
           }
-          p_elem->part_state= part_state;
+          p_elem->part_state= PART_NORMAL;
           part_info->curr_part_elem= p_elem;
           part_info->current_partition= p_elem;
           part_info->use_default_partitions= FALSE;

sql/table.cc

@@ -667,36 +667,16 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
 #endif
       next_chunk+= 5 + partition_info_len;
     }
-    if (share->mysql_version > 50105 && next_chunk + 5 < buff_end)
+#if 1
+    if (share->mysql_version == 50106 ||
+        share->mysql_version == 50107)
     {
       /*
-        Partition state was introduced to support partition management in version 5.1.5
+         Partition state array was here in version 5.1.6, this code makes
-      */
+         it possible to load a 5.1.6 table in later versions. Can most
-      uint32 part_state_len= uint4korr(next_chunk);
+         likely be removed at some point in time.
-#ifdef WITH_PARTITION_STORAGE_ENGINE
+       */
-      if ((share->part_state_len= part_state_len))
+      next_chunk+= 4;
-        if (!(share->part_state=
-              (uchar*) memdup_root(&share->mem_root, next_chunk + 4,
-                                   part_state_len)))
-        {
-          my_free(buff, MYF(0));
-          goto err;
-        }
-#else
-      if (part_state_len)
-      {
-        DBUG_PRINT("info", ("WITH_PARTITION_STORAGE_ENGINE is not defined"));
-        my_free(buff, MYF(0));
-        goto err;
-      }
-#endif
-      next_chunk+= 4 + part_state_len;
-    }
-#ifdef WITH_PARTITION_STORAGE_ENGINE
-    else
-    {
-      share->part_state_len= 0;
-      share->part_state= NULL;
     }
 #endif
     keyinfo= share->key_info;

sql/unireg.cc

@@ -136,7 +136,6 @@ bool mysql_create_frm(THD *thd, const char *file_name,
   if (part_info)
   {
     create_info->extra_size+= part_info->part_info_len;
-    create_info->extra_size+= part_info->part_state_len;
   }
 #endif
 
@@ -209,12 +208,6 @@ bool mysql_create_frm(THD *thd, const char *file_name,
         my_write(file, (const byte*)part_info->part_info_string,
                  part_info->part_info_len + 1, MYF_RW))
       goto err;
-    DBUG_PRINT("info", ("Part state len = %d", part_info->part_state_len));
-    int4store(buff, part_info->part_state_len);
-    if (my_write(file, (const byte*)buff, 4, MYF_RW) ||
-        my_write(file, (const byte*)part_info->part_state,
-                 part_info->part_state_len, MYF_RW))
-      goto err;
   }
   else
 #endif
@@ -330,7 +323,7 @@ int rea_create_table(THD *thd, const char *path,
 
   // Make sure mysql_create_frm din't remove extension
   DBUG_ASSERT(*fn_rext(frm_name));
-  if (file->create_handler_files(path))
+  if (file->create_handler_files(path, NULL, FALSE))
     goto err_handler;
   if (!create_info->frm_only && ha_create_table(thd, path, db, table_name,
                                                 create_info,0))