Bug #28430 Failure in replication of innodb partitioned tables on row/mixed format.

In the ha_partition::position() we don't calculate the number
of the partition of the record, but use m_last_part value instead,
relying on that it's previously set by some other call like ::write_row().
Delete_rows_log_event::do_exec_row() calls find_and_fetch_row(),
where we used position() + rnd_pos() call for the InnoDB-based PARTITION-ed
table as there HA_PRIMARY_KEY_REQUIRED_FOR_POSITION enabled.

fixed by introducing new handler::rnd_pos_by_record() method to be
used for random record-based positioning


sql/ha_partition.cc:
  Bug #28430 Failure in replication of innodb partitioned tables on row/mixed format.
  
  ha_partition::rnd_pos_by_record() implemented
sql/ha_partition.h:
  Bug #28430 Failure in replication of innodb partitioned tables on row/mixed format.
  
  ha_partition::rnd_pos_by_record() declared
sql/handler.h:
  Bug #28430 Failure in replication of innodb partitioned tables on row/mixed format.
  
  handler::rnd_pos_by_record() introduced
sql/log_event.cc:
  Bug #28430 Failure in replication of innodb partitioned tables on row/mixed format.
  
  handler::rnd_pos_by_record used instead of position() + rnd_pos() call

sql/ha_partition.cc

@@ -3235,14 +3235,9 @@ int ha_partition::rnd_next(uchar *buf)
 
 void ha_partition::position(const uchar *record)
 {
-  handler *file;
+  handler *file= m_file[m_last_part];
   DBUG_ENTER("ha_partition::position");
 
-  if (unlikely(get_part_for_delete(record, m_rec0, m_part_info, &m_last_part)))
-    m_last_part= 0;
-
-  file= m_file[m_last_part];
-
   file->position(record);
   int2store(ref, m_last_part);
   memcpy((ref + PARTITION_BYTES_IN_POS), file->ref,
@@ -3257,6 +3252,7 @@ void ha_partition::position(const uchar *record)
   DBUG_VOID_RETURN;
 }
 
+
 /*
   Read row using position
 
@@ -3292,6 +3288,36 @@ int ha_partition::rnd_pos(uchar * buf, uchar *pos)
 }
 
 
+/*
+  Read row using position using given record to find
+
+  SYNOPSIS
+    rnd_pos_by_record()
+    record             Current record in MySQL Row Format
+
+  RETURN VALUE
+    >0                 Error code
+    0                  Success
+
+  DESCRIPTION
+    this works as position()+rnd_pos() functions, but does some extra work,
+    calculating m_last_part - the partition to where the 'record'
+    should go.
+
+    called from replication (log_event.cc)
+*/
+
+int ha_partition::rnd_pos_by_record(uchar *record)
+{
+  DBUG_ENTER("ha_partition::rnd_pos_by_record");
+
+  if (unlikely(get_part_for_delete(record, m_rec0, m_part_info, &m_last_part)))
+    DBUG_RETURN(1);
+
+  DBUG_RETURN(handler::rnd_pos_by_record(record));
+}
+
+
 /****************************************************************************
                 MODULE index scan
 ****************************************************************************/

sql/ha_partition.h

@@ -353,6 +353,7 @@ class ha_partition :public handler
   virtual int rnd_end();
   virtual int rnd_next(uchar * buf);
   virtual int rnd_pos(uchar * buf, uchar * pos);
+  virtual int rnd_pos_by_record(uchar *record);
   virtual void position(const uchar * record);
 
   /*

sql/handler.h

@@ -1343,6 +1343,17 @@ class handler :public Sql_alloc
   virtual int ft_read(uchar *buf) { return HA_ERR_WRONG_COMMAND; }
   virtual int rnd_next(uchar *buf)=0;
   virtual int rnd_pos(uchar * buf, uchar *pos)=0;
+  /*
+    one has to use this method when to find
+    random position by record as the plain
+    position() call doesn't work for some
+    handlers for random position
+  */
+  virtual int rnd_pos_by_record(uchar *record)
+    {
+      position(record);
+      return rnd_pos(record, ref);
+    }
   virtual int read_first_row(uchar *buf, uint primary_key);
   /*
     The following function is only needed for tables that may be temporary

sql/log_event.cc

@@ -7509,8 +7509,7 @@ static int find_and_fetch_row(TABLE *table, uchar *key)
                                  table->s->reclength) == 0);
 
     */
-    table->file->position(table->record[0]);
-    int error= table->file->rnd_pos(table->record[0], table->file->ref);
+    int error= table->file->rnd_pos_by_record(table->record[0]);
     /*
       rnd_pos() returns the record in table->record[0], so we have to
       move it to table->record[1].