MDEV-5281 Partitioning issue after upgrade from 10.0.3-1 to 10.0.5-1

merged from 5.6:
Bug#14521864: MYSQL 5.1 TO 5.5 BUGS PARTITIONING
Bug#16589511: MYSQL_UPGRADE FAILS TO WRITE OUT ENTIRE ALTER TABLE ... ALGORITHM= ... STATEMENT
Bug#16274455: CAN NOT ACESS PARTITIONED TABLES WHEN DOWNGRADED FROM 5.6.11 TO 5.6.10

plus minor changes from 5.6, mainly comments

1 files changed, 612 insertions, 40 deletions

diff --git a/sql/ha_partition.cc b/sql/ha_partition.cc
index 3512d1cee87..fd6d89dd7ff 100644
--- a/sql/ha_partition.cc
+++ b/sql/ha_partition.cc
@@ -55,6 +55,8 @@
 #include "sql_table.h"                        // tablename_to_filename
 #include "key.h"
 #include "sql_plugin.h"
+#include "sql_show.h"                        // append_identifier
+#include "sql_admin.h"                       // SQL_ADMIN_MSG_TEXT_SIZE
 
 #include "debug_sync.h"
 
@@ -66,12 +68,15 @@
 #define PAR_NUM_PARTS_OFFSET 8
 /* offset to the engines array */
 #define PAR_ENGINES_OFFSET 12
-#define PARTITION_ENABLED_TABLE_FLAGS (HA_FILE_BASED | HA_REC_NOT_IN_SEQ)
+#define PARTITION_ENABLED_TABLE_FLAGS (HA_FILE_BASED | \
+                                       HA_REC_NOT_IN_SEQ | \
+                                       HA_CAN_REPAIR)
 #define PARTITION_DISABLED_TABLE_FLAGS (HA_CAN_GEOMETRY | \
                                         HA_CAN_FULLTEXT | \
                                         HA_DUPLICATE_POS | \
                                         HA_CAN_SQL_HANDLER | \
-                                        HA_CAN_INSERT_DELAYED)
+                                        HA_CAN_INSERT_DELAYED | \
+                                        HA_READ_BEFORE_WRITE_REMOVAL)
 static const char *ha_par_ext= ".par";
 
 /****************************************************************************
@@ -348,6 +353,7 @@ void ha_partition::init_handler_variables()
   m_rec_length= 0;
   m_last_part= 0;
   m_rec0= 0;
+  m_err_rec= NULL;
   m_curr_key_info[0]= NULL;
   m_curr_key_info[1]= NULL;
   m_part_func_monotonicity_info= NON_MONOTONIC;
@@ -1181,10 +1187,11 @@ int ha_partition::preload_keys(THD *thd, HA_CHECK_OPT *check_opt)
     0                         Success
 */
 
-static int handle_opt_part(THD *thd, HA_CHECK_OPT *check_opt,
-                           handler *file, uint flag)
+int ha_partition::handle_opt_part(THD *thd, HA_CHECK_OPT *check_opt,
+                                  uint part_id, uint flag)
 {
   int error;
+  handler *file= m_file[part_id];
   DBUG_ENTER("handle_opt_part");
   DBUG_PRINT("enter", ("flag = %u", flag));
 
@@ -1193,9 +1200,27 @@ static int handle_opt_part(THD *thd, HA_CHECK_OPT *check_opt,
   else if (flag == ANALYZE_PARTS)
     error= file->ha_analyze(thd, check_opt);
   else if (flag == CHECK_PARTS)
+  {
     error= file->ha_check(thd, check_opt);
+    if (!error ||
+        error == HA_ADMIN_ALREADY_DONE ||
+        error == HA_ADMIN_NOT_IMPLEMENTED)
+    {
+      if (check_opt->flags & (T_MEDIUM | T_EXTEND))
+        error= check_misplaced_rows(part_id, false);
+    }
+  }
   else if (flag == REPAIR_PARTS)
+  {
     error= file->ha_repair(thd, check_opt);
+    if (!error ||
+        error == HA_ADMIN_ALREADY_DONE ||
+        error == HA_ADMIN_NOT_IMPLEMENTED)
+    {
+      if (check_opt->flags & (T_MEDIUM | T_EXTEND))
+        error= check_misplaced_rows(part_id, true);
+    }
+  }
   else if (flag == ASSIGN_KEYCACHE_PARTS)
     error= file->assign_to_keycache(thd, check_opt);
   else if (flag == PRELOAD_KEYS_PARTS)
@@ -1216,30 +1241,38 @@ static int handle_opt_part(THD *thd, HA_CHECK_OPT *check_opt,
    (modelled after mi_check_print_msg)
    TODO: move this into the handler, or rewrite mysql_admin_table.
 */
-static bool print_admin_msg(THD* thd, const char* msg_type,
+static bool print_admin_msg(THD* thd, uint len,
+                            const char* msg_type,
                             const char* db_name, String &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,
+  ATTRIBUTE_FORMAT(printf, 7, 8);
+static bool print_admin_msg(THD* thd, uint len,
+                            const char* msg_type,
                             const char* db_name, String &table_name,
                             const char* op_name, const char *fmt, ...)
 {
   va_list args;
   Protocol *protocol= thd->protocol;
-  uint length, msg_length;
-  char msgbuf[MYSQL_ERRMSG_SIZE];
-  char name[SAFE_NAME_LEN*2+2];
-
+  uint length;
+  uint msg_length;
+  char name[NAME_LEN*2+2];
+  char *msgbuf;
+  bool error= true;
+
+  if (!(msgbuf= (char*) my_malloc(len, MYF(0))))
+    return true;
   va_start(args, fmt);
-  msg_length= my_vsnprintf(msgbuf, sizeof(msgbuf), fmt, args);
+  msg_length= my_vsnprintf(msgbuf, len, fmt, args);
   va_end(args);
-  msgbuf[sizeof(msgbuf) - 1] = 0; // healthy paranoia
+  if (msg_length >= (len - 1))
+    goto err;
+  msgbuf[len - 1] = 0; // healthy paranoia
 
 
   if (!thd->vio_ok())
   {
-    sql_print_error(fmt, args);
-    return TRUE;
+    sql_print_error("%s", msgbuf);
+    goto err;
   }
 
   length=(uint) (strxmov(name, db_name, ".", table_name.c_ptr_safe(), NullS) - name);
@@ -1262,9 +1295,12 @@ static bool print_admin_msg(THD* thd, const char* msg_type,
   {
     sql_print_error("Failed on my_net_write, writing to stderr instead: %s\n",
                     msgbuf);
-    return TRUE;
+    goto err;
   }
-  return FALSE;
+  error= false;
+err:
+  my_free(msgbuf);
+  return error;
 }
 
 
@@ -1314,14 +1350,15 @@ int ha_partition::handle_opt_partitions(THD *thd, HA_CHECK_OPT *check_opt,
           part= i * num_subparts + j;
           DBUG_PRINT("info", ("Optimize subpartition %u (%s)",
                      part, sub_elem->partition_name));
-          if ((error= handle_opt_part(thd, check_opt, m_file[part], flag)))
+          if ((error= handle_opt_part(thd, check_opt, part, flag)))
           {
             /* print a line which partition the error belongs to */
             if (error != HA_ADMIN_NOT_IMPLEMENTED &&
                 error != HA_ADMIN_ALREADY_DONE &&
                 error != HA_ADMIN_TRY_ALTER)
             {
-              print_admin_msg(thd, "error", table_share->db.str, table->alias,
+	      print_admin_msg(thd, MYSQL_ERRMSG_SIZE, "error",
+                              table_share->db.str, table->alias,
                               opt_op_name[flag],
                               "Subpartition %s returned error", 
                               sub_elem->partition_name);
@@ -1340,14 +1377,15 @@ int ha_partition::handle_opt_partitions(THD *thd, HA_CHECK_OPT *check_opt,
       {
         DBUG_PRINT("info", ("Optimize partition %u (%s)", i,
                             part_elem->partition_name));
-        if ((error= handle_opt_part(thd, check_opt, m_file[i], flag)))
+        if ((error= handle_opt_part(thd, check_opt, i, flag)))
         {
           /* print a line which partition the error belongs to */
           if (error != HA_ADMIN_NOT_IMPLEMENTED &&
               error != HA_ADMIN_ALREADY_DONE &&
               error != HA_ADMIN_TRY_ALTER)
           {
-            print_admin_msg(thd, "error", table_share->db.str, table->alias,
+	    print_admin_msg(thd, MYSQL_ERRMSG_SIZE, "error",
+                            table_share->db.str, table->alias,
                             opt_op_name[flag], "Partition %s returned error", 
                             part_elem->partition_name);
           }
@@ -2025,8 +2063,7 @@ void ha_partition::update_create_info(HA_CREATE_INFO *create_info)
   my_bool from_alter = (create_info->data_file_name == (const char*) -1);
   create_info->data_file_name= create_info->index_file_name = NULL;
 
-  create_info->connect_string.str= NULL;
-  create_info->connect_string.length= 0;
+  create_info->connect_string= null_lex_str;
 
   /*
     We do not need to update the individual partition DATA DIRECTORY settings
@@ -2984,7 +3021,7 @@ bool ha_partition::setup_engine_array(MEM_ROOT *mem_root)
     }
   }
 
-  my_afree((void*) engine_array);
+  my_afree(engine_array);
     
   if (create_handlers(mem_root))
   {
@@ -2995,7 +3032,7 @@ bool ha_partition::setup_engine_array(MEM_ROOT *mem_root)
   DBUG_RETURN(false);
 
 err:
-  my_afree((void*) engine_array);
+  my_afree(engine_array);
   DBUG_RETURN(true);
 }
 
@@ -4137,7 +4174,7 @@ exit:
 
     Called from sql_select.cc, sql_acl.cc, sql_update.cc, and sql_insert.cc.
     new_data is always record[0]
-    old_data is normally record[1] but may be anything
+    old_data is always record[1]
 */
 
 int ha_partition::update_row(const uchar *old_data, uchar *new_data)
@@ -4147,6 +4184,7 @@ int ha_partition::update_row(const uchar *old_data, uchar *new_data)
   int error= 0;
   longlong func_value;
   DBUG_ENTER("ha_partition::update_row");
+  m_err_rec= NULL;
 
   // Need to read partition-related columns, to locate the row's partition:
   DBUG_ASSERT(bitmap_is_subset(&m_part_info->full_part_field_set,
@@ -4164,6 +4202,30 @@ int ha_partition::update_row(const uchar *old_data, uchar *new_data)
     error= HA_ERR_NOT_IN_LOCK_PARTITIONS;
     goto exit;
   }
+
+  /*
+    The protocol for updating a row is:
+    1) position the handler (cursor) on the row to be updated,
+       either through the last read row (rnd or index) or by rnd_pos.
+    2) call update_row with both old and new full records as arguments.
+
+    This means that m_last_part should already be set to actual partition
+    where the row was read from. And if that is not the same as the
+    calculated part_id we found a misplaced row, we return an error to
+    notify the user that something is broken in the row distribution
+    between partitions! Since we don't check all rows on read, we return an
+    error instead of correcting m_last_part, to make the user aware of the
+    problem!
+
+    Notice that HA_READ_BEFORE_WRITE_REMOVAL does not require this protocol,
+    so this is not supported for this engine.
+  */
+  if (old_part_id != m_last_part)
+  {
+    m_err_rec= old_data;
+    DBUG_RETURN(HA_ERR_ROW_IN_WRONG_PARTITION);
+  }
+
   m_last_part= new_part_id;
   start_part_bulk_insert(thd, new_part_id);
   if (new_part_id == old_part_id)
@@ -4267,6 +4329,7 @@ int ha_partition::delete_row(const uchar *buf)
   int error;
   THD *thd= ha_thd();
   DBUG_ENTER("ha_partition::delete_row");
+  m_err_rec= NULL;
 
   DBUG_ASSERT(bitmap_is_subset(&m_part_info->full_part_field_set,
                                table->read_set));
@@ -4274,12 +4337,39 @@ int ha_partition::delete_row(const uchar *buf)
   {
     DBUG_RETURN(error);
   }
-  m_last_part= part_id;
   /* Should never call delete_row on a partition which is not read */
   DBUG_ASSERT(bitmap_is_set(&(m_part_info->read_partitions), part_id));
   DBUG_ASSERT(bitmap_is_set(&(m_part_info->lock_partitions), part_id));
   if (!bitmap_is_set(&(m_part_info->lock_partitions), part_id))
     DBUG_RETURN(HA_ERR_NOT_IN_LOCK_PARTITIONS);
+
+  /*
+    The protocol for deleting a row is:
+    1) position the handler (cursor) on the row to be deleted,
+       either through the last read row (rnd or index) or by rnd_pos.
+    2) call delete_row with the full record as argument.
+
+    This means that m_last_part should already be set to actual partition
+    where the row was read from. And if that is not the same as the
+    calculated part_id we found a misplaced row, we return an error to
+    notify the user that something is broken in the row distribution
+    between partitions! Since we don't check all rows on read, we return an
+    error instead of forwarding the delete to the correct (m_last_part)
+    partition!
+
+    Notice that HA_READ_BEFORE_WRITE_REMOVAL does not require this protocol,
+    so this is not supported for this engine.
+
+    TODO: change the assert in InnoDB into an error instead and make this one
+    an assert instead and remove the get_part_for_delete()!
+  */
+  if (part_id != m_last_part)
+  {
+    m_err_rec= buf;
+    DBUG_RETURN(HA_ERR_ROW_IN_WRONG_PARTITION);
+  }
+
+  m_last_part= part_id;
   tmp_disable_binlog(thd);
   error= m_file[part_id]->ha_delete_row(buf);
   reenable_binlog(thd);
@@ -4402,7 +4492,7 @@ int ha_partition::truncate_partition(Alter_info *alter_info, bool *binlog_stmt)
       {
         List_iterator<partition_element>
                                     subpart_it(part_elem->subpartitions);
-        partition_element *sub_elem __attribute__((unused));
+        partition_element *sub_elem;
         uint j= 0, part;
         do
         {
@@ -7498,7 +7588,6 @@ double ha_partition::read_time(uint index, uint ranges, ha_rows rows)
 /**
   Number of rows in table. see handler.h
 
-
   @return Number of records in the table (after pruning!)
 */
 
@@ -7582,10 +7671,89 @@ uint32 ha_partition::calculate_key_hash_value(Field **field_array)
 {
   ulong nr1= 1;
   ulong nr2= 4;
+  bool use_51_hash;
+  use_51_hash= test((*field_array)->table->part_info->key_algorithm ==
+                    partition_info::KEY_ALGORITHM_51);
 
   do
   {
     Field *field= *field_array;
+    if (use_51_hash)
+    {
+      switch (field->real_type()) {
+      case MYSQL_TYPE_TINY:
+      case MYSQL_TYPE_SHORT:
+      case MYSQL_TYPE_LONG:
+      case MYSQL_TYPE_FLOAT:
+      case MYSQL_TYPE_DOUBLE:
+      case MYSQL_TYPE_NEWDECIMAL:
+      case MYSQL_TYPE_TIMESTAMP:
+      case MYSQL_TYPE_LONGLONG:
+      case MYSQL_TYPE_INT24:
+      case MYSQL_TYPE_TIME:
+      case MYSQL_TYPE_DATETIME:
+      case MYSQL_TYPE_YEAR:
+      case MYSQL_TYPE_NEWDATE:
+        {
+          if (field->is_null())
+          {
+            nr1^= (nr1 << 1) | 1;
+            continue;
+          }
+          /* Force this to my_hash_sort_bin, which was used in 5.1! */
+          uint len= field->pack_length();
+          my_charset_bin.coll->hash_sort(&my_charset_bin, field->ptr, len,
+                                         &nr1, &nr2);
+          /* Done with this field, continue with next one. */
+          continue;
+        }
+      case MYSQL_TYPE_STRING:
+      case MYSQL_TYPE_VARCHAR:
+      case MYSQL_TYPE_BIT:
+        /* Not affected, same in 5.1 and 5.5 */
+        break;
+      /*
+        ENUM/SET uses my_hash_sort_simple in 5.1 (i.e. my_charset_latin1)
+        and my_hash_sort_bin in 5.5!
+      */
+      case MYSQL_TYPE_ENUM:
+      case MYSQL_TYPE_SET:
+        {
+          if (field->is_null())
+          {
+            nr1^= (nr1 << 1) | 1;
+            continue;
+          }
+          /* Force this to my_hash_sort_bin, which was used in 5.1! */
+          uint len= field->pack_length();
+          my_charset_latin1.coll->hash_sort(&my_charset_latin1, field->ptr,
+                                            len, &nr1, &nr2);
+          continue;
+        }
+      /* New types in mysql-5.6. */
+      case MYSQL_TYPE_DATETIME2:
+      case MYSQL_TYPE_TIME2:
+      case MYSQL_TYPE_TIMESTAMP2:
+        /* Not affected, 5.6+ only! */
+        break;
+
+      /* These types should not be allowed for partitioning! */
+      case MYSQL_TYPE_NULL:
+      case MYSQL_TYPE_DECIMAL:
+      case MYSQL_TYPE_DATE:
+      case MYSQL_TYPE_TINY_BLOB:
+      case MYSQL_TYPE_MEDIUM_BLOB:
+      case MYSQL_TYPE_LONG_BLOB:
+      case MYSQL_TYPE_BLOB:
+      case MYSQL_TYPE_VAR_STRING:
+      case MYSQL_TYPE_GEOMETRY:
+        /* fall through. */
+      default:
+        DBUG_ASSERT(0);                    // New type?
+        /* Fall through for default hashing (5.5). */
+      }
+      /* fall through, use collation based hashing. */
+    }
     field->hash(&nr1, &nr2);
   } while (*(++field_array));
   return (uint32) nr1;
@@ -7641,6 +7809,57 @@ enum row_type ha_partition::get_row_type() const
 }
 
 
+void ha_partition::append_row_to_str(String &str)
+{
+  const uchar *rec;
+  bool is_rec0= !m_err_rec || m_err_rec == table->record[0];
+  if (is_rec0)
+    rec= table->record[0];
+  else
+    rec= m_err_rec;
+  // If PK, use full PK instead of full part field array!
+  if (table->s->primary_key != MAX_KEY)
+  {
+    KEY *key= table->key_info + table->s->primary_key;
+    KEY_PART_INFO *key_part=     key->key_part;
+    KEY_PART_INFO *key_part_end= key_part + key->user_defined_key_parts;
+    if (!is_rec0)
+      set_key_field_ptr(key, rec, table->record[0]);
+    for (; key_part != key_part_end; key_part++)
+    {
+      Field *field= key_part->field;
+      str.append(" ");
+      str.append(field->field_name);
+      str.append(":");
+      field_unpack(&str, field, rec, 0, false);
+    }
+    if (!is_rec0)
+      set_key_field_ptr(key, table->record[0], rec);
+  }
+  else
+  {
+    Field **field_ptr;
+    if (!is_rec0)
+      set_field_ptr(m_part_info->full_part_field_array, rec,
+                    table->record[0]);
+    /* No primary key, use full partition field array. */
+    for (field_ptr= m_part_info->full_part_field_array;
+         *field_ptr;
+         field_ptr++)
+    {
+      Field *field= *field_ptr;
+      str.append(" ");
+      str.append(field->field_name);
+      str.append(":");
+      field_unpack(&str, field, rec, 0, false);
+    }
+    if (!is_rec0)
+      set_field_ptr(m_part_info->full_part_field_array, table->record[0],
+                    rec);
+  }
+}
+
+
 void ha_partition::print_error(int error, myf errflag)
 {
   THD *thd= ha_thd();
@@ -7652,21 +7871,62 @@ void ha_partition::print_error(int error, myf errflag)
   if ((error == HA_ERR_NO_PARTITION_FOUND) &&
       ! (thd->lex->alter_info.flags & Alter_info::ALTER_TRUNCATE_PARTITION))
     m_part_info->print_no_partition_found(table);
-  else
-  {
-    /* In case m_file has not been initialized, like in bug#42438 */
-    if (m_file)
+  else if (error == HA_ERR_ROW_IN_WRONG_PARTITION)
+  {
+    /* Should only happen on DELETE or UPDATE! */
+    DBUG_ASSERT(thd_sql_command(thd) == SQLCOM_DELETE ||
+                thd_sql_command(thd) == SQLCOM_DELETE_MULTI ||
+                thd_sql_command(thd) == SQLCOM_UPDATE ||
+                thd_sql_command(thd) == SQLCOM_UPDATE_MULTI);
+    DBUG_ASSERT(m_err_rec);
+    if (m_err_rec)
     {
-      if (m_last_part >= m_tot_parts)
+      uint max_length;
+      char buf[MAX_KEY_LENGTH];
+      String str(buf,sizeof(buf),system_charset_info);
+      uint32 part_id;
+      str.length(0);
+      str.append("(");
+      str.append_ulonglong(m_last_part);
+      str.append(" != ");
+      if (get_part_for_delete(m_err_rec, m_rec0, m_part_info, &part_id))
+        str.append("?");
+      else
+        str.append_ulonglong(part_id);
+      str.append(")");
+      append_row_to_str(str);
+
+      /* Log this error, so the DBA can notice it and fix it! */
+      sql_print_error("Table '%-192s' corrupted: row in wrong partition: %s\n"
+                      "Please REPAIR the table!",
+                      table->s->table_name.str,
+                      str.c_ptr_safe());
+
+      max_length= (MYSQL_ERRMSG_SIZE - (uint) strlen(ER(ER_ROW_IN_WRONG_PARTITION)));
+      if (str.length() >= max_length)
       {
-        DBUG_ASSERT(0);
-        m_last_part= 0;
+        str.length(max_length-4);
+        str.append(STRING_WITH_LEN("..."));
       }
-      m_file[m_last_part]->print_error(error, errflag);
+      my_error(ER_ROW_IN_WRONG_PARTITION, MYF(0), str.c_ptr_safe());
+      m_err_rec= NULL;
+      DBUG_VOID_RETURN;
     }
-    else
-      handler::print_error(error, errflag);
+    /* fall through to generic error handling. */
+  }
+
+  /* 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);
   DBUG_VOID_RETURN;
 }
 
@@ -7804,6 +8064,13 @@ ha_partition::check_if_supported_inplace_alter(TABLE *altered_table,
 #endif
 
   DBUG_ENTER("ha_partition::check_if_supported_inplace_alter");
+  /*
+    Support inplace change of KEY () -> KEY ALGORITHM = N ().
+    Any other change would set partition_changed in
+    prep_alter_part_table() in mysql_alter_table().
+  */
+  if (ha_alter_info->alter_info->flags == Alter_info::ALTER_PARTITION)
+    DBUG_RETURN(HA_ALTER_INPLACE_NO_LOCK);
 
 #ifndef PARTITION_SUPPORTS_INPLACE_ALTER
   /*
@@ -7861,6 +8128,13 @@ bool ha_partition::prepare_inplace_alter_table(TABLE *altered_table,
 
   DBUG_ENTER("ha_partition::prepare_inplace_alter_table");
 
+  /*
+    Changing to similar partitioning, only update metadata.
+    Non allowed changes would be catched in prep_alter_part_table().
+  */
+  if (ha_alter_info->alter_info->flags == Alter_info::ALTER_PARTITION)
+    DBUG_RETURN(false);
+
   part_inplace_ctx=
     static_cast<class ha_partition_inplace_ctx*>(ha_alter_info->handler_ctx);
 
@@ -7887,6 +8161,13 @@ bool ha_partition::inplace_alter_table(TABLE *altered_table,
 
   DBUG_ENTER("ha_partition::inplace_alter_table");
 
+  /*
+    Changing to similar partitioning, only update metadata.
+    Non allowed changes would be catched in prep_alter_part_table().
+  */
+  if (ha_alter_info->alter_info->flags == Alter_info::ALTER_PARTITION)
+    DBUG_RETURN(false);
+
   part_inplace_ctx=
     static_cast<class ha_partition_inplace_ctx*>(ha_alter_info->handler_ctx);
 
@@ -7920,6 +8201,13 @@ bool ha_partition::commit_inplace_alter_table(TABLE *altered_table,
 
   DBUG_ENTER("ha_partition::commit_inplace_alter_table");
 
+  /*
+    Changing to similar partitioning, only update metadata.
+    Non allowed changes would be catched in prep_alter_part_table().
+  */
+  if (ha_alter_info->alter_info->flags == Alter_info::ALTER_PARTITION)
+    DBUG_RETURN(false);
+
   part_inplace_ctx=
     static_cast<class ha_partition_inplace_ctx*>(ha_alter_info->handler_ctx);
 
@@ -8236,7 +8524,7 @@ void ha_partition::get_auto_increment(ulonglong offset, ulonglong increment,
       /* Only nb_desired_values = 1 makes sense */
       (*file)->get_auto_increment(offset, increment, 1,
                                  &first_value_part, &nb_reserved_values_part);
-      if (first_value_part == ~(ulonglong)(0)) // error in one partition
+      if (first_value_part == ULONGLONG_MAX) // error in one partition
       {
         *first_value= first_value_part;
         /* log that the error was between table/partition handler */
@@ -8456,6 +8744,290 @@ int ha_partition::indexes_are_disabled(void)
 }
 
 
+/**
+  Check/fix misplaced rows.
+
+  @param read_part_id  Partition to check/fix.
+  @param repair        If true, move misplaced rows to correct partition.
+
+  @return Operation status.
+    @retval 0     Success
+    @retval != 0  Error
+*/
+
+int ha_partition::check_misplaced_rows(uint read_part_id, bool repair)
+{
+  int result= 0;
+  uint32 correct_part_id;
+  longlong func_value;
+  longlong num_misplaced_rows= 0;
+
+  DBUG_ENTER("ha_partition::check_misplaced_rows");
+
+  DBUG_ASSERT(m_file);
+
+  if (repair)
+  {
+    /* We must read the full row, if we need to move it! */
+    bitmap_set_all(table->read_set);
+    bitmap_set_all(table->write_set);
+  }
+  else
+  {
+    /* Only need to read the partitioning fields. */
+    bitmap_union(table->read_set, &m_part_info->full_part_field_set);
+  }
+
+  if ((result= m_file[read_part_id]->ha_rnd_init(1)))
+    DBUG_RETURN(result);
+
+  while (true)
+  {
+    if ((result= m_file[read_part_id]->ha_rnd_next(m_rec0)))
+    {
+      if (result == HA_ERR_RECORD_DELETED)
+        continue;
+      if (result != HA_ERR_END_OF_FILE)
+        break;
+
+      if (num_misplaced_rows > 0)
+      {
+	print_admin_msg(ha_thd(), MYSQL_ERRMSG_SIZE, "warning",
+                        table_share->db.str, table->alias,
+                        opt_op_name[REPAIR_PARTS],
+                        "Moved %lld misplaced rows",
+                        num_misplaced_rows);
+      }
+      /* End-of-file reached, all rows are now OK, reset result and break. */
+      result= 0;
+      break;
+    }
+
+    result= m_part_info->get_partition_id(m_part_info, &correct_part_id,
+                                          &func_value);
+    if (result)
+      break;
+
+    if (correct_part_id != read_part_id)
+    {
+      num_misplaced_rows++;
+      if (!repair)
+      {
+        /* Check. */
+	print_admin_msg(ha_thd(), MYSQL_ERRMSG_SIZE, "error",
+                        table_share->db.str, table->alias,
+                        opt_op_name[CHECK_PARTS],
+                        "Found a misplaced row");
+        /* Break on first misplaced row! */
+        result= HA_ADMIN_NEEDS_UPGRADE;
+        break;
+      }
+      else
+      {
+        DBUG_PRINT("info", ("Moving row from partition %d to %d",
+                            read_part_id, correct_part_id));
+
+        /*
+          Insert row into correct partition. Notice that there are no commit
+          for every N row, so the repair will be one large transaction!
+        */
+        if ((result= m_file[correct_part_id]->ha_write_row(m_rec0)))
+        {
+          /*
+            We have failed to insert a row, it might have been a duplicate!
+          */
+          char buf[MAX_KEY_LENGTH];
+          String str(buf,sizeof(buf),system_charset_info);
+          str.length(0);
+          if (result == HA_ERR_FOUND_DUPP_KEY)
+          {
+            str.append("Duplicate key found, "
+                       "please update or delete the record:\n");
+            result= HA_ADMIN_CORRUPT;
+          }
+          m_err_rec= NULL;
+          append_row_to_str(str);
+
+          /*
+            If the engine supports transactions, the failure will be
+            rollbacked.
+          */
+          if (!m_file[correct_part_id]->has_transactions())
+          {
+            /* Log this error, so the DBA can notice it and fix it! */
+            sql_print_error("Table '%-192s' failed to move/insert a row"
+                            " from part %d into part %d:\n%s",
+                            table->s->table_name.str,
+                            read_part_id,
+                            correct_part_id,
+                            str.c_ptr_safe());
+          }
+	  print_admin_msg(ha_thd(), MYSQL_ERRMSG_SIZE, "error",
+                          table_share->db.str, table->alias,
+                          opt_op_name[REPAIR_PARTS],
+                          "Failed to move/insert a row"
+                          " from part %d into part %d:\n%s",
+                          read_part_id,
+                          correct_part_id,
+                          str.c_ptr_safe());
+          break;
+        }
+
+        /* Delete row from wrong partition. */
+        if ((result= m_file[read_part_id]->ha_delete_row(m_rec0)))
+        {
+          if (m_file[correct_part_id]->has_transactions())
+            break;
+          /*
+            We have introduced a duplicate, since we failed to remove it
+            from the wrong partition.
+          */
+          char buf[MAX_KEY_LENGTH];
+          String str(buf,sizeof(buf),system_charset_info);
+          str.length(0);
+          m_err_rec= NULL;
+          append_row_to_str(str);
+
+          /* Log this error, so the DBA can notice it and fix it! */
+          sql_print_error("Table '%-192s': Delete from part %d failed with"
+                          " error %d. But it was already inserted into"
+                          " part %d, when moving the misplaced row!"
+                          "\nPlease manually fix the duplicate row:\n%s",
+                          table->s->table_name.str,
+                          read_part_id,
+                          result,
+                          correct_part_id,
+                          str.c_ptr_safe());
+          break;
+        }
+      }
+    }
+  }
+
+  int tmp_result= m_file[read_part_id]->ha_rnd_end();
+  DBUG_RETURN(result ? result : tmp_result);
+}
+
+
+#define KEY_PARTITIONING_CHANGED_STR \
+  "KEY () partitioning changed, please run:\n" \
+  "ALTER TABLE %s.%s ALGORITHM = INPLACE %s"
+
+int ha_partition::check_for_upgrade(HA_CHECK_OPT *check_opt)
+{
+  int error= HA_ADMIN_NEEDS_CHECK;
+  DBUG_ENTER("ha_partition::check_for_upgrade");
+
+  /*
+    This is called even without FOR UPGRADE,
+    if the .frm version is lower than the current version.
+    In that case return that it needs checking!
+  */
+  if (!(check_opt->sql_flags & TT_FOR_UPGRADE))
+    DBUG_RETURN(error);
+
+  /*
+    Partitions will be checked for during their ha_check!
+
+    Check if KEY (sub)partitioning was used and any field's hash calculation
+    differs from 5.1, see bug#14521864.
+  */
+  if (table->s->mysql_version < 50503 &&              // 5.1 table (<5.5.3)
+      ((m_part_info->part_type == HASH_PARTITION &&   // KEY partitioned
+        m_part_info->list_of_part_fields) ||
+       (m_is_sub_partitioned &&                       // KEY subpartitioned
+        m_part_info->list_of_subpart_fields)))
+  {
+    Field **field;
+    if (m_is_sub_partitioned)
+    {
+      field= m_part_info->subpart_field_array;
+    }
+    else
+    {
+      field= m_part_info->part_field_array;
+    }
+    for (; *field; field++)
+    {
+      switch ((*field)->real_type()) {
+      case MYSQL_TYPE_TINY:
+      case MYSQL_TYPE_SHORT:
+      case MYSQL_TYPE_LONG:
+      case MYSQL_TYPE_FLOAT:
+      case MYSQL_TYPE_DOUBLE:
+      case MYSQL_TYPE_NEWDECIMAL:
+      case MYSQL_TYPE_TIMESTAMP:
+      case MYSQL_TYPE_LONGLONG:
+      case MYSQL_TYPE_INT24:
+      case MYSQL_TYPE_TIME:
+      case MYSQL_TYPE_DATETIME:
+      case MYSQL_TYPE_YEAR:
+      case MYSQL_TYPE_NEWDATE:
+      case MYSQL_TYPE_ENUM:
+      case MYSQL_TYPE_SET:
+        {
+          THD *thd= ha_thd();
+          char *part_buf;
+          String db_name, table_name;
+          uint part_buf_len;
+          bool skip_generation= false;
+          partition_info::enum_key_algorithm old_algorithm;
+          old_algorithm= m_part_info->key_algorithm;
+          error= HA_ADMIN_FAILED;
+          append_identifier(ha_thd(), &db_name, table_share->db.str,
+                            table_share->db.length);
+          append_identifier(ha_thd(), &table_name, table_share->table_name.str,
+                            table_share->table_name.length);
+          if (m_part_info->key_algorithm != partition_info::KEY_ALGORITHM_NONE)
+          {
+            /*
+              Only possible when someone tampered with .frm files,
+              like during tests :)
+            */
+            skip_generation= true;
+          }
+          m_part_info->key_algorithm= partition_info::KEY_ALGORITHM_51;
+          if (skip_generation ||
+              !(part_buf= generate_partition_syntax(m_part_info,
+                                                    &part_buf_len,
+                                                    true,
+                                                    true,
+                                                    NULL,
+                                                    NULL,
+                                                    NULL)) ||
+	      print_admin_msg(thd, SQL_ADMIN_MSG_TEXT_SIZE + 1, "error",
+	                      table_share->db.str,
+	                      table->alias,
+                              opt_op_name[CHECK_PARTS],
+                              KEY_PARTITIONING_CHANGED_STR,
+                              db_name.c_ptr_safe(),
+                              table_name.c_ptr_safe(),
+                              part_buf))
+	  {
+	    /* Error creating admin message (too long string?). */
+	    print_admin_msg(thd, MYSQL_ERRMSG_SIZE, "error",
+                            table_share->db.str, table->alias,
+                            opt_op_name[CHECK_PARTS],
+                            KEY_PARTITIONING_CHANGED_STR,
+                            db_name.c_ptr_safe(), table_name.c_ptr_safe(),
+                            "<old partition clause>, but add ALGORITHM = 1"
+                            " between 'KEY' and '(' to change the metadata"
+                            " without the need of a full table rebuild.");
+          }
+          m_part_info->key_algorithm= old_algorithm;
+          DBUG_RETURN(error);
+        }
+      default:
+        /* Not affected! */
+        ;
+      }
+    }
+  }
+
+  DBUG_RETURN(error);
+}
+
+
 struct st_mysql_storage_engine partition_storage_engine=
 { MYSQL_HANDLERTON_INTERFACE_VERSION };