MDEV-15380 Index for versioned table gets corrupt after partitioning and DELETE

In a test case Update occurs between Search and Delete/Update. This corrupts rowid
which Search saves for Delete/Update. Patch prevents this by using of
HA_EXTRA_REMEMBER_POS and HA_EXTRA_RESTORE_POS in a partition code.

This situation possibly occurs only with system versioning table and partition.
MyISAM and Aria engines are affected.

fix by midenok
Closes #705

mysql-test/suite/versioning/r/partition.result

@@ -483,6 +483,24 @@ delete from t1 where a is not null;
 create or replace table t1 (i int) with system versioning partition by system_time limit 10 (partition p0 history, partition pn current);
 select * from t1 partition (p0) for system_time all;
 ERROR HY000: SYSTEM_TIME partitions in table `t1` does not support historical query
+# MDEV-15380 Index for versioned table gets corrupt after partitioning and DELETE
+create or replace table t1 (pk int primary key)
+engine=myisam
+with system versioning
+partition by key() partitions 3;
+set timestamp=1523466002.799571;
+insert into t1 values (11),(12);
+set timestamp=1523466004.169435;
+delete from t1 where pk in (11, 12);
+Same test but for Aria storage engine
+create or replace table t1 (pk int primary key)
+engine=aria
+with system versioning
+partition by key() partitions 3;
+set timestamp=1523466002.799571;
+insert into t1 values (11),(12);
+set timestamp=1523466004.169435;
+delete from t1 where pk in (11, 12);
 # Test cleanup
 drop database test;
 create database test;

mysql-test/suite/versioning/t/partition.test

@@ -430,6 +430,25 @@ create or replace table t1 (i int) with system versioning partition by system_ti
 --error ER_VERS_QUERY_IN_PARTITION
 select * from t1 partition (p0) for system_time all;
 
+--echo # MDEV-15380 Index for versioned table gets corrupt after partitioning and DELETE
+create or replace table t1 (pk int primary key)
+  engine=myisam
+  with system versioning
+  partition by key() partitions 3;
+set timestamp=1523466002.799571;
+insert into t1 values (11),(12);
+set timestamp=1523466004.169435;
+delete from t1 where pk in (11, 12);
+--echo Same test but for Aria storage engine
+create or replace table t1 (pk int primary key)
+  engine=aria
+  with system versioning
+  partition by key() partitions 3;
+set timestamp=1523466002.799571;
+insert into t1 values (11),(12);
+set timestamp=1523466004.169435;
+delete from t1 where pk in (11, 12);
+
 --echo # Test cleanup
 drop database test;
 create database test;

sql/ha_partition.cc

@@ -8687,6 +8687,16 @@ err_handler:
      HA_EXTRA_NO_READCHECK=5                 No readcheck on update
      HA_EXTRA_READCHECK=6                    Use readcheck (def)
 
+  HA_EXTRA_REMEMBER_POS:
+  HA_EXTRA_RESTORE_POS:
+    System versioning needs this for MyISAM and Aria tables.
+    On DELETE using PRIMARY KEY:
+    1) handler::ha_index_read_map() saves rowid used for row delete/update
+    2) handler::ha_update_row() can rewrite saved rowid
+    3) handler::ha_delete_row()/handler::ha_update_row() expects saved but got
+       different rowid and operation fails
+    Using those flags prevents harmful side effect of 2)
+
   4) Operations only used by temporary tables for query processing
   ----------------------------------------------------------------
   HA_EXTRA_RESET_STATE:
@@ -8746,8 +8756,6 @@ err_handler:
     Only used MyISAM, only used internally in MyISAM handler, never called
     from server level.
   HA_EXTRA_KEYREAD_CHANGE_POS:
-  HA_EXTRA_REMEMBER_POS:
-  HA_EXTRA_RESTORE_POS:
   HA_EXTRA_PRELOAD_BUFFER_SIZE:
   HA_EXTRA_CHANGE_KEY_TO_DUP:
   HA_EXTRA_CHANGE_KEY_TO_UNIQUE:
@@ -8830,6 +8838,8 @@ int ha_partition::extra(enum ha_extra_function operation)
   case HA_EXTRA_PREPARE_FOR_DROP:
   case HA_EXTRA_FLUSH_CACHE:
   case HA_EXTRA_PREPARE_FOR_ALTER_TABLE:
+  case HA_EXTRA_REMEMBER_POS:
+  case HA_EXTRA_RESTORE_POS:
   {
     DBUG_RETURN(loop_extra(operation));
   }

sql/sql_delete.cc

@@ -254,7 +254,12 @@ int TABLE::delete_row()
 
   store_record(this, record[1]);
   vers_update_end();
-  return file->ha_update_row(record[1], record[0]);
+  int res;
+  if ((res= file->extra(HA_EXTRA_REMEMBER_POS)))
+    return res;
+  if ((res= file->ha_update_row(record[1], record[0])))
+    return res;
+  return file->extra(HA_EXTRA_RESTORE_POS);
 }