1 files changed, 317 insertions, 140 deletions

diff --git a/sql/ha_partition.cc b/sql/ha_partition.cc
index b8fedb7b7bb..cd5d7f201a7 100644
--- a/sql/ha_partition.cc
+++ b/sql/ha_partition.cc
@@ -1,5 +1,5 @@
 /*
-   Copyright (c) 2005, 2010, Oracle and/or its affiliates. All rights reserved.
+  Copyright (c) 2005, 2011, Oracle and/or its affiliates. All rights reserved.
 
   This program is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License as published by
@@ -165,8 +165,7 @@ const uint ha_partition::NO_CURRENT_PART_ID= 0xFFFFFFFF;
 */
 
 ha_partition::ha_partition(handlerton *hton, TABLE_SHARE *share)
-  :handler(hton, share), m_part_info(NULL), m_create_handler(FALSE),
-   m_is_sub_partitioned(0)
+  :handler(hton, share)
 {
   DBUG_ENTER("ha_partition::ha_partition(table)");
   init_handler_variables();
@@ -186,15 +185,44 @@ ha_partition::ha_partition(handlerton *hton, TABLE_SHARE *share)
 */
 
 ha_partition::ha_partition(handlerton *hton, partition_info *part_info)
-  :handler(hton, NULL), m_part_info(part_info), m_create_handler(TRUE),
-   m_is_sub_partitioned(m_part_info->is_sub_partitioned())
+  :handler(hton, NULL)
 {
   DBUG_ENTER("ha_partition::ha_partition(part_info)");
+  DBUG_ASSERT(part_info);
   init_handler_variables();
-  DBUG_ASSERT(m_part_info);
+  m_part_info= part_info;
+  m_create_handler= TRUE;
+  m_is_sub_partitioned= m_part_info->is_sub_partitioned();
   DBUG_VOID_RETURN;
 }
 
+/**
+  ha_partition constructor method used by ha_partition::clone()
+
+  @param hton               Handlerton (partition_hton)
+  @param share              Table share object
+  @param part_info_arg      partition_info to use
+  @param clone_arg          ha_partition to clone
+  @param clme_mem_root_arg  MEM_ROOT to use
+
+  @return New partition handler
+*/
+
+ha_partition::ha_partition(handlerton *hton, TABLE_SHARE *share,
+                           partition_info *part_info_arg,
+                           ha_partition *clone_arg,
+                           MEM_ROOT *clone_mem_root_arg)
+  :handler(hton, share)
+{
+  DBUG_ENTER("ha_partition::ha_partition(clone)");
+  init_handler_variables();
+  m_part_info= part_info_arg;
+  m_create_handler= TRUE;
+  m_is_sub_partitioned= m_part_info->is_sub_partitioned();
+  m_is_clone_of= clone_arg;
+  m_clone_mem_root= clone_mem_root_arg;
+  DBUG_VOID_RETURN;
+}
 
 /*
   Initialize handler object
@@ -246,7 +274,6 @@ void ha_partition::init_handler_variables()
   m_rec0= 0;
   m_curr_key_info[0]= NULL;
   m_curr_key_info[1]= NULL;
-  is_clone= FALSE,
   m_part_func_monotonicity_info= NON_MONOTONIC;
   auto_increment_lock= FALSE;
   auto_increment_safe_stmt_log_lock= FALSE;
@@ -254,6 +281,11 @@ void ha_partition::init_handler_variables()
     this allows blackhole to work properly
   */
   m_no_locks= 0;
+  m_part_info= NULL;
+  m_create_handler= FALSE;
+  m_is_sub_partitioned= 0;
+  m_is_clone_of= NULL;
+  m_clone_mem_root= NULL;
 
 #ifdef DONT_HAVE_TO_BE_INITALIZED
   m_start_key.flag= 0;
@@ -361,7 +393,8 @@ bool ha_partition::initialize_partition(MEM_ROOT *mem_root)
     */
     DBUG_RETURN(0);
   }
-  else if (get_from_handler_file(table_share->normalized_path.str, mem_root))
+  else if (get_from_handler_file(table_share->normalized_path.str,
+                                 mem_root, false))
   {
     my_message(ER_UNKNOWN_ERROR, "Failed to read from the .par file", MYF(0));
     DBUG_RETURN(1);
@@ -995,6 +1028,10 @@ static int handle_opt_part(THD *thd, HA_CHECK_OPT *check_opt,
 static bool print_admin_msg(THD* thd, const char* msg_type,
                             const char* db_name, const char* table_name,
                             const char* op_name, const char *fmt, ...)
+  ATTRIBUTE_FORMAT(printf, 6, 7);
+static bool print_admin_msg(THD* thd, const char* msg_type,
+                            const char* db_name, const char* table_name,
+                            const char* op_name, const char *fmt, ...)
 {
   va_list args;
   Protocol *protocol= thd->protocol;
@@ -1850,7 +1887,7 @@ uint ha_partition::del_ren_cre_table(const char *from,
     DBUG_RETURN(TRUE);
   }
 
-  if (get_from_handler_file(from, ha_thd()->mem_root))
+  if (get_from_handler_file(from, ha_thd()->mem_root, false))
     DBUG_RETURN(TRUE);
   DBUG_ASSERT(m_file_buffer);
   DBUG_PRINT("enter", ("from: (%s) to: (%s)", from, to));
@@ -2066,18 +2103,16 @@ static uint name_add(char *dest, const char *first_name, const char *sec_name)
 }
 
 
-/*
+/**
   Create the special .par file
 
-  SYNOPSIS
-    create_handler_file()
-    name                      Full path of table name
+  @param name  Full path of table name
 
-  RETURN VALUE
-    >0                        Error code
-    0                         Success
+  @return Operation status
+    @retval FALSE  Error code
+    @retval TRUE   Success
 
-  DESCRIPTION
+  @note
     Method used to create handler file with names of partitions, their
     engine types and the number of partitions.
 */
@@ -2141,19 +2176,22 @@ bool ha_partition::create_handler_file(const char *name)
      Array of engine types        n * 4 bytes where
      n = (m_tot_parts + 3)/4
      Length of name part in bytes 4 bytes
+     (Names in filename format)
      Name part                    m * 4 bytes where
      m = ((length_name_part + 3)/4)*4
 
      All padding bytes are zeroed
   */
-  tot_partition_words= (tot_parts + 3) / 4;
-  tot_name_words= (tot_name_len + 3) / 4;
+  tot_partition_words= (tot_parts + PAR_WORD_SIZE - 1) / PAR_WORD_SIZE;
+  tot_name_words= (tot_name_len + PAR_WORD_SIZE - 1) / PAR_WORD_SIZE;
+  /* 4 static words (tot words, checksum, tot partitions, name length) */
   tot_len_words= 4 + tot_partition_words + tot_name_words;
-  tot_len_byte= 4 * tot_len_words;
+  tot_len_byte= PAR_WORD_SIZE * tot_len_words;
   if (!(file_buffer= (uchar *) my_malloc(tot_len_byte, MYF(MY_ZEROFILL))))
     DBUG_RETURN(TRUE);
-  engine_array= (file_buffer + 12);
-  name_buffer_ptr= (char*) (file_buffer + ((4 + tot_partition_words) * 4));
+  engine_array= (file_buffer + PAR_ENGINES_OFFSET);
+  name_buffer_ptr= (char*) (engine_array + tot_partition_words * PAR_WORD_SIZE
+                            + PAR_WORD_SIZE);
   part_it.rewind();
   for (i= 0; i < no_parts; i++)
   {
@@ -2191,13 +2229,15 @@ bool ha_partition::create_handler_file(const char *name)
   }
   chksum= 0;
   int4store(file_buffer, tot_len_words);
-  int4store(file_buffer + 8, tot_parts);
-  int4store(file_buffer + 12 + (tot_partition_words * 4), tot_name_len);
+  int4store(file_buffer + PAR_NUM_PARTS_OFFSET, tot_parts);
+  int4store(file_buffer + PAR_ENGINES_OFFSET +
+            (tot_partition_words * PAR_WORD_SIZE),
+            tot_name_len);
   for (i= 0; i < tot_len_words; i++)
-    chksum^= uint4korr(file_buffer + 4 * i);
-  int4store(file_buffer + 4, chksum);
+    chksum^= uint4korr(file_buffer + PAR_WORD_SIZE * i);
+  int4store(file_buffer + PAR_CHECKSUM_OFFSET, chksum);
   /*
-    Remove .frm extension and replace with .par
+    Add .par extension to the file name.
     Create and write and close file
     to be used at open, delete_table and rename_table
   */
@@ -2215,14 +2255,9 @@ bool ha_partition::create_handler_file(const char *name)
   DBUG_RETURN(result);
 }
 
-/*
-  Clear handler variables and free some memory
 
-  SYNOPSIS
-    clear_handler_file()
-
-  RETURN VALUE 
-    NONE
+/**
+  Clear handler variables and free some memory
 */
 
 void ha_partition::clear_handler_file()
@@ -2235,16 +2270,15 @@ void ha_partition::clear_handler_file()
   m_engine_array= NULL;
 }
 
-/*
+
+/**
   Create underlying handler objects
 
-  SYNOPSIS
-    create_handlers()
-    mem_root		Allocate memory through this
+  @param mem_root  Allocate memory through this
 
-  RETURN VALUE
-    TRUE                  Error
-    FALSE                 Success
+  @return Operation status
+    @retval TRUE   Error
+    @retval FALSE  Success
 */
 
 bool ha_partition::create_handlers(MEM_ROOT *mem_root)
@@ -2282,6 +2316,7 @@ bool ha_partition::create_handlers(MEM_ROOT *mem_root)
   DBUG_RETURN(FALSE);
 }
 
+
 /*
   Create underlying handler objects from partition info
 
@@ -2353,100 +2388,164 @@ error_end:
 }
 
 
-/*
-  Get info about partition engines and their names from the .par file
+/**
+  Read the .par file to get the partitions engines and names
 
-  SYNOPSIS
-    get_from_handler_file()
-    name                        Full path of table name
-    mem_root			Allocate memory through this
+  @param name  Name of table file (without extention)
 
-  RETURN VALUE
-    TRUE                        Error
-    FALSE                       Success
+  @return Operation status
+    @retval true   Failure
+    @retval false  Success
 
-  DESCRIPTION
-    Open handler file to get partition names, engine types and number of
-    partitions.
+  @note On success, m_file_buffer is allocated and must be
+  freed by the caller. m_name_buffer_ptr and m_tot_parts is also set.
 */
 
-bool ha_partition::get_from_handler_file(const char *name, MEM_ROOT *mem_root)
+bool ha_partition::read_par_file(const char *name)
 {
-  char buff[FN_REFLEN], *address_tot_name_len;
+  char buff[FN_REFLEN], *tot_name_len_offset;
   File file;
-  char *file_buffer, *name_buffer_ptr;
-  handlerton **engine_array;
+  char *file_buffer;
   uint i, len_bytes, len_words, tot_partition_words, tot_name_words, chksum;
-  DBUG_ENTER("ha_partition::get_from_handler_file");
+  DBUG_ENTER("ha_partition::read_par_file");
   DBUG_PRINT("enter", ("table name: '%s'", name));
 
   if (m_file_buffer)
-    DBUG_RETURN(FALSE);
+    DBUG_RETURN(false);
   fn_format(buff, name, "", ha_par_ext, MY_APPEND_EXT);
 
   /* Following could be done with my_stat to read in whole file */
   if ((file= my_open(buff, O_RDONLY | O_SHARE, MYF(0))) < 0)
-    DBUG_RETURN(TRUE);
-  if (my_read(file, (uchar *) & buff[0], 8, MYF(MY_NABP)))
+    DBUG_RETURN(true);
+  if (my_read(file, (uchar *) & buff[0], PAR_WORD_SIZE, MYF(MY_NABP)))
     goto err1;
   len_words= uint4korr(buff);
-  len_bytes= 4 * len_words;
+  len_bytes= PAR_WORD_SIZE * len_words;
+  if (my_seek(file, 0, MY_SEEK_SET, MYF(0)) == MY_FILEPOS_ERROR)
+    goto err1;
   if (!(file_buffer= (char*) my_malloc(len_bytes, MYF(0))))
     goto err1;
-  VOID(my_seek(file, 0, MY_SEEK_SET, MYF(0)));
   if (my_read(file, (uchar *) file_buffer, len_bytes, MYF(MY_NABP)))
     goto err2;
 
   chksum= 0;
   for (i= 0; i < len_words; i++)
-    chksum ^= uint4korr((file_buffer) + 4 * i);
+    chksum ^= uint4korr((file_buffer) + PAR_WORD_SIZE * i);
   if (chksum)
     goto err2;
-  m_tot_parts= uint4korr((file_buffer) + 8);
+  m_tot_parts= uint4korr((file_buffer) + PAR_NUM_PARTS_OFFSET);
   DBUG_PRINT("info", ("No of parts = %u", m_tot_parts));
-  tot_partition_words= (m_tot_parts + 3) / 4;
+  tot_partition_words= (m_tot_parts + PAR_WORD_SIZE - 1) / PAR_WORD_SIZE;
+
+  tot_name_len_offset= file_buffer + PAR_ENGINES_OFFSET +
+                       PAR_WORD_SIZE * tot_partition_words;
+  tot_name_words= (uint4korr(tot_name_len_offset) + PAR_WORD_SIZE - 1) /
+                  PAR_WORD_SIZE;
+  /*
+    Verify the total length = tot size word, checksum word, num parts word +
+    engines array + name length word + name array.
+  */
+  if (len_words != (tot_partition_words + tot_name_words + 4))
+    goto err2;
+  VOID(my_close(file, MYF(0)));
+  m_file_buffer= file_buffer;          // Will be freed in clear_handler_file()
+  m_name_buffer_ptr= tot_name_len_offset + PAR_WORD_SIZE;
+
+  DBUG_RETURN(false);
+
+err2:
+  my_free(file_buffer, MYF(0));
+err1:
+  VOID(my_close(file, MYF(0)));
+  DBUG_RETURN(true);
+}
+
+
+/**
+  Setup m_engine_array
+
+  @param mem_root  MEM_ROOT to use for allocating new handlers
+
+  @return Operation status
+    @retval false  Success
+    @retval true   Failure
+*/
+
+bool ha_partition::setup_engine_array(MEM_ROOT *mem_root)
+{
+  uint i;
+  uchar *buff;
+  handlerton **engine_array;
+
+  DBUG_ASSERT(!m_file);
+  DBUG_ENTER("ha_partition::setup_engine_array");
   engine_array= (handlerton **) my_alloca(m_tot_parts * sizeof(handlerton*));
+  if (!engine_array)
+    DBUG_RETURN(true);
+
+  buff= (uchar *) (m_file_buffer + PAR_ENGINES_OFFSET);
   for (i= 0; i < m_tot_parts; i++)
   {
     engine_array[i]= ha_resolve_by_legacy_type(ha_thd(),
                                                (enum legacy_db_type)
-                                               *(uchar *) ((file_buffer) +
-                                                           12 + i));
+                                                 *(buff + i));
     if (!engine_array[i])
-      goto err3;
+      goto err;
   }
-  address_tot_name_len= file_buffer + 12 + 4 * tot_partition_words;
-  tot_name_words= (uint4korr(address_tot_name_len) + 3) / 4;
-  if (len_words != (tot_partition_words + tot_name_words + 4))
-    goto err3;
-  name_buffer_ptr= file_buffer + 16 + 4 * tot_partition_words;
-  VOID(my_close(file, MYF(0)));
-  m_file_buffer= file_buffer;          // Will be freed in clear_handler_file()
-  m_name_buffer_ptr= name_buffer_ptr;
-  
   if (!(m_engine_array= (plugin_ref*)
                 my_malloc(m_tot_parts * sizeof(plugin_ref), MYF(MY_WME))))
-    goto err3;
+    goto err;
 
   for (i= 0; i < m_tot_parts; i++)
     m_engine_array[i]= ha_lock_engine(NULL, engine_array[i]);
 
   my_afree((gptr) engine_array);
     
-  if (!m_file && create_handlers(mem_root))
+  if (create_handlers(mem_root))
   {
     clear_handler_file();
-    DBUG_RETURN(TRUE);
+    DBUG_RETURN(true);
   }
-  DBUG_RETURN(FALSE);
 
-err3:
+  DBUG_RETURN(false);
+
+err:
   my_afree((gptr) engine_array);
-err2:
-  my_free(file_buffer, MYF(0));
-err1:
-  VOID(my_close(file, MYF(0)));
-  DBUG_RETURN(TRUE);
+  DBUG_RETURN(true);
+}
+
+
+/**
+  Get info about partition engines and their names from the .par file
+
+  @param name      Full path of table name
+  @param mem_root  Allocate memory through this
+  @param is_clone  If it is a clone, don't create new handlers
+
+  @return Operation status
+    @retval true   Error
+    @retval false  Success
+
+  @note Open handler file to get partition names, engine types and number of
+  partitions.
+*/
+
+bool ha_partition::get_from_handler_file(const char *name, MEM_ROOT *mem_root,
+                                         bool is_clone)
+{
+  DBUG_ENTER("ha_partition::get_from_handler_file");
+  DBUG_PRINT("enter", ("table name: '%s'", name));
+
+  if (m_file_buffer)
+    DBUG_RETURN(false);
+
+  if (read_par_file(name))
+    DBUG_RETURN(true);
+
+  if (!is_clone && setup_engine_array(mem_root))
+    DBUG_RETURN(true);
+
+  DBUG_RETURN(false);
 }
 
 
@@ -2493,13 +2592,13 @@ void ha_data_partition_destroy(void *ha_data)
 
 int ha_partition::open(const char *name, int mode, uint test_if_locked)
 {
-  char *name_buffer_ptr= m_name_buffer_ptr;
-  int error;
+  char *name_buffer_ptr;
+  int error= HA_ERR_INITIALIZATION;
   uint alloc_len;
   handler **file;
   char name_buff[FN_REFLEN];
   bool is_not_tmp_table= (table_share->tmp_table == NO_TMP_TABLE);
-  ulonglong check_table_flags= 0;
+  ulonglong check_table_flags;
   DBUG_ENTER("ha_partition::open");
 
   DBUG_ASSERT(table->s == table_share);
@@ -2507,8 +2606,9 @@ int ha_partition::open(const char *name, int mode, uint test_if_locked)
   m_mode= mode;
   m_open_test_lock= test_if_locked;
   m_part_field_array= m_part_info->full_part_field_array;
-  if (get_from_handler_file(name, &table->mem_root))
-    DBUG_RETURN(1);
+  if (get_from_handler_file(name, &table->mem_root, test(m_is_clone_of)))
+    DBUG_RETURN(error);
+  name_buffer_ptr= m_name_buffer_ptr;
   m_start_key.length= 0;
   m_rec0= table->record[0];
   m_rec_length= table_share->reclength;
@@ -2518,7 +2618,7 @@ int ha_partition::open(const char *name, int mode, uint test_if_locked)
   {
     if (!(m_ordered_rec_buffer= (uchar*)my_malloc(alloc_len, MYF(MY_WME))))
     {
-      DBUG_RETURN(1);
+      DBUG_RETURN(error);
     }
     {
       /*
@@ -2541,48 +2641,84 @@ int ha_partition::open(const char *name, int mode, uint test_if_locked)
 
   /* Initialize the bitmap we use to minimize ha_start_bulk_insert calls */
   if (bitmap_init(&m_bulk_insert_started, NULL, m_tot_parts + 1, FALSE))
-    DBUG_RETURN(1);
+    DBUG_RETURN(error);
   bitmap_clear_all(&m_bulk_insert_started);
   /* Initialize the bitmap we use to determine what partitions are used */
-  if (!is_clone)
+  if (!m_is_clone_of)
   {
+    DBUG_ASSERT(!m_clone_mem_root);
     if (bitmap_init(&(m_part_info->used_partitions), NULL, m_tot_parts, TRUE))
     {
       bitmap_free(&m_bulk_insert_started);
-      DBUG_RETURN(1);
+      DBUG_RETURN(error);
     }
     bitmap_set_all(&(m_part_info->used_partitions));
   }
 
+  if (m_is_clone_of)
+  {
+    uint i;
+    DBUG_ASSERT(m_clone_mem_root);
+    /* Allocate an array of handler pointers for the partitions handlers. */
+    alloc_len= (m_tot_parts + 1) * sizeof(handler*);
+    if (!(m_file= (handler **) alloc_root(m_clone_mem_root, alloc_len)))
+      goto err_alloc;
+    memset(m_file, 0, alloc_len);
+    /*
+      Populate them by cloning the original partitions. This also opens them.
+      Note that file->ref is allocated too.
+    */
+    file= m_is_clone_of->m_file;
+    for (i= 0; i < m_tot_parts; i++)
+    {
+      create_partition_name(name_buff, name, name_buffer_ptr, NORMAL_PART_NAME,
+                            FALSE);
+      if (!(m_file[i]= file[i]->clone(name_buff, m_clone_mem_root)))
+      {
+        error= HA_ERR_INITIALIZATION;
+        file= &m_file[i];
+        goto err_handler;
+      }
+      name_buffer_ptr+= strlen(name_buffer_ptr) + 1;
+    }
+  }
+  else
+  {
+   file= m_file;
+   do
+   {
+      create_partition_name(name_buff, name, name_buffer_ptr, NORMAL_PART_NAME,
+                            FALSE);
+      if ((error= (*file)->ha_open(table, name_buff, mode, test_if_locked)))
+        goto err_handler;
+      m_no_locks+= (*file)->lock_count();
+      name_buffer_ptr+= strlen(name_buffer_ptr) + 1;
+    } while (*(++file));
+  }
+  
   file= m_file;
-  do
+  ref_length= (*file)->ref_length;
+  check_table_flags= (((*file)->ha_table_flags() &
+                       ~(PARTITION_DISABLED_TABLE_FLAGS)) |
+                      (PARTITION_ENABLED_TABLE_FLAGS));
+  while (*(++file))
   {
-    create_partition_name(name_buff, name, name_buffer_ptr, NORMAL_PART_NAME,
-                          FALSE);
-    if ((error= (*file)->ha_open(table, (const char*) name_buff, mode,
-                                 test_if_locked)))
-      goto err_handler;
-    m_no_locks+= (*file)->lock_count();
-    name_buffer_ptr+= strlen(name_buffer_ptr) + 1;
+    /* MyISAM can have smaller ref_length for partitions with MAX_ROWS set */
     set_if_bigger(ref_length, ((*file)->ref_length));
     /*
       Verify that all partitions have the same set of table flags.
       Mask all flags that partitioning enables/disables.
     */
-    if (!check_table_flags)
-    {
-      check_table_flags= (((*file)->ha_table_flags() &
-                           ~(PARTITION_DISABLED_TABLE_FLAGS)) |
-                          (PARTITION_ENABLED_TABLE_FLAGS));
-    }
-    else if (check_table_flags != (((*file)->ha_table_flags() &
-                                    ~(PARTITION_DISABLED_TABLE_FLAGS)) |
-                                   (PARTITION_ENABLED_TABLE_FLAGS)))
+    if (check_table_flags != (((*file)->ha_table_flags() &
+                               ~(PARTITION_DISABLED_TABLE_FLAGS)) |
+                              (PARTITION_ENABLED_TABLE_FLAGS)))
     {
       error= HA_ERR_INITIALIZATION;
+      /* set file to last handler, so all of them is closed */
+      file = &m_file[m_tot_parts - 1];
       goto err_handler;
     }
-  } while (*(++file));
+  }
   key_used_on_scan= m_file[0]->key_used_on_scan;
   implicit_emptied= m_file[0]->implicit_emptied;
   /*
@@ -2591,6 +2727,7 @@ int ha_partition::open(const char *name, int mode, uint test_if_locked)
   */
   ref_length+= PARTITION_BYTES_IN_POS;
   m_ref_length= ref_length;
+
   /*
     Release buffer read from .par file. It will not be reused again after
     being opened once.
@@ -2648,25 +2785,54 @@ err_handler:
   DEBUG_SYNC(ha_thd(), "partition_open_error");
   while (file-- != m_file)
     (*file)->close();
+err_alloc:
   bitmap_free(&m_bulk_insert_started);
-  if (!is_clone)
+  if (!m_is_clone_of)
     bitmap_free(&(m_part_info->used_partitions));
 
   DBUG_RETURN(error);
 }
 
-handler *ha_partition::clone(MEM_ROOT *mem_root)
+
+/**
+  Clone the open and locked partitioning handler.
+
+  @param  mem_root  MEM_ROOT to use.
+
+  @return Pointer to the successfully created clone or NULL
+
+  @details
+  This function creates a new ha_partition handler as a clone/copy. The
+  original (this) must already be opened and locked. The clone will use
+  the originals m_part_info.
+  It also allocates memory for ref + ref_dup.
+  In ha_partition::open() it will clone its original handlers partitions
+  which will allocate then on the correct MEM_ROOT and also open them.
+*/
+
+handler *ha_partition::clone(const char *name, MEM_ROOT *mem_root)
 {
-  handler *new_handler= get_new_handler(table->s, mem_root,
-                                        table->s->db_type());
-  ((ha_partition*)new_handler)->m_part_info= m_part_info;
-  ((ha_partition*)new_handler)->is_clone= TRUE;
-  if (new_handler && !new_handler->ha_open(table,
-                                           table->s->normalized_path.str,
-                                           table->db_stat,
-                                           HA_OPEN_IGNORE_IF_LOCKED))
-    return new_handler;
-  return NULL;
+  ha_partition *new_handler;
+
+  DBUG_ENTER("ha_partition::clone");
+  new_handler= new (mem_root) ha_partition(ht, table_share, m_part_info,
+                                           this, mem_root);
+  /*
+    Allocate new_handler->ref here because otherwise ha_open will allocate it
+    on this->table->mem_root and we will not be able to reclaim that memory 
+    when the clone handler object is destroyed.
+  */
+  if (new_handler &&
+      !(new_handler->ref= (uchar*) alloc_root(mem_root,
+                                              ALIGN_SIZE(m_ref_length)*2)))
+    new_handler= NULL;
+
+  if (new_handler &&
+      new_handler->ha_open(table, name,
+                           table->db_stat, HA_OPEN_IGNORE_IF_LOCKED))
+    new_handler= NULL;
+
+  DBUG_RETURN((handler*) new_handler);
 }
 
 
@@ -2697,7 +2863,7 @@ int ha_partition::close(void)
   DBUG_ASSERT(table->s == table_share);
   delete_queue(&m_queue);
   bitmap_free(&m_bulk_insert_started);
-  if (!is_clone)
+  if (!m_is_clone_of)
     bitmap_free(&(m_part_info->used_partitions));
   file= m_file;
 
@@ -3243,15 +3409,19 @@ int ha_partition::update_row(const uchar *old_data, uchar *new_data)
 
 exit:
   /*
-    if updating an auto_increment column, update
+    If updating an auto_increment column, update
     table_share->ha_data->next_auto_inc_val if needed.
     (not to be used if auto_increment on secondary field in a multi-column
     index)
     mysql_update does not set table->next_number_field, so we use
     table->found_next_number_field instead.
+    Also checking that the field is marked in the write set.
   */
-  if (table->found_next_number_field && new_data == table->record[0] &&
-      !table->s->next_number_keypart)
+  if (table->found_next_number_field &&
+      new_data == table->record[0] &&
+      !table->s->next_number_keypart &&
+      bitmap_is_set(table->write_set,
+                    table->found_next_number_field->field_index))
   {
     HA_DATA_PARTITION *ha_data= (HA_DATA_PARTITION*) table_share->ha_data;
     if (!ha_data->auto_inc_initialized)
@@ -3797,19 +3967,16 @@ end_dont_reset_start_part:
 void ha_partition::position(const uchar *record)
 {
   handler *file= m_file[m_last_part];
+  uint pad_length;
   DBUG_ENTER("ha_partition::position");
 
   file->position(record);
   int2store(ref, m_last_part);
-  memcpy((ref + PARTITION_BYTES_IN_POS), file->ref,
-	 (ref_length - PARTITION_BYTES_IN_POS));
-
-#ifdef SUPPORTING_PARTITION_OVER_DIFFERENT_ENGINES
-#ifdef HAVE_purify
-  bzero(ref + PARTITION_BYTES_IN_POS + ref_length,
-        max_ref_length-ref_length);
-#endif /* HAVE_purify */
-#endif
+  memcpy((ref + PARTITION_BYTES_IN_POS), file->ref, file->ref_length);
+  pad_length= m_ref_length - PARTITION_BYTES_IN_POS - file->ref_length;
+  if (pad_length)
+    memset((ref + PARTITION_BYTES_IN_POS + file->ref_length), 0, pad_length);
+
   DBUG_VOID_RETURN;
 }
 
@@ -3817,6 +3984,7 @@ void ha_partition::position(const uchar *record)
 void ha_partition::column_bitmaps_signal()
 {
     handler::column_bitmaps_signal();
+    /* Must read all partition fields to make position() call possible */
     bitmap_union(table->read_set, &m_part_info->full_part_field_set);
 }
  
@@ -4319,6 +4487,8 @@ int ha_partition::index_read_idx_map(uchar *buf, uint index,
           break;
       }
     }
+    if (part <= m_part_spec.end_part)
+      m_last_part= part;
   }
   else
   {
@@ -6238,7 +6408,14 @@ void ha_partition::print_error(int error, myf errflag)
   {
     /* In case m_file has not been initialized, like in bug#42438 */
     if (m_file)
+    {
+      if (m_last_part >= m_tot_parts)
+      {
+        DBUG_ASSERT(0);
+        m_last_part= 0;
+      }
       m_file[m_last_part]->print_error(error, errflag);
+    }
     else
       handler::print_error(error, errflag);
   }