MDEV-25546 LIMIT partitioning does not respect ROLLBACK

vers_info->hist_part retained stale value after ROLLBACK. The
algorithm in vers_set_hist_part() continued iteration from that value.

The simplest solution is to process partitions each time from start
for LIMIT in vers_set_hist_part().

3 files changed, 86 insertions, 5 deletions

diff --git a/mysql-test/suite/versioning/r/partition.result b/mysql-test/suite/versioning/r/partition.result
index d903315414d..70416360697 100644
--- a/mysql-test/suite/versioning/r/partition.result
+++ b/mysql-test/suite/versioning/r/partition.result
@@ -1,3 +1,5 @@
+set @save_persistent=@@global.innodb_stats_persistent;
+set global innodb_stats_persistent= 0;
 set system_versioning_alter_history=keep;
 # Check conventional partitioning on temporal tables
 create or replace table t1 (
@@ -799,4 +801,43 @@ delete from t1 partition (p0, p1, pn);
 ERROR HY000: Not allowed for system-versioned table `test`.`t1`
 drop table t1;
 set timestamp= default;
+#
+# MDEV-25546 LIMIT partitioning does not respect ROLLBACK
+#
+create or replace table t1 (pk int primary key)
+with system versioning engine innodb
+partition by system_time limit 100 (
+partition p0 history,
+partition p1 history,
+partition pn current);
+insert into t1 select seq from seq_1_to_90;
+start transaction;
+replace into t1 select seq from seq_1_to_80;
+replace into t1 select seq from seq_1_to_70;
+replace into t1 select seq from seq_1_to_60;
+select partition_name, table_rows
+from information_schema.partitions
+where table_name = 't1';
+partition_name	table_rows
+p0	150
+p1	60
+pn	90
+rollback;
+select partition_name, table_rows
+from information_schema.partitions
+where table_name = 't1';
+partition_name	table_rows
+p0	0
+p1	0
+pn	90
+replace into t1 select seq from seq_1_to_10;
+select partition_name, table_rows
+from information_schema.partitions
+where table_name = 't1';
+partition_name	table_rows
+p0	10
+p1	0
+pn	90
+drop table t1;
 # End of 10.3 tests
+set global innodb_stats_persistent= @save_persistent;
diff --git a/mysql-test/suite/versioning/t/partition.test b/mysql-test/suite/versioning/t/partition.test
index c5531d3c7f6..a687cd9b09f 100644
--- a/mysql-test/suite/versioning/t/partition.test
+++ b/mysql-test/suite/versioning/t/partition.test
@@ -1,6 +1,10 @@
 -- source include/have_partition.inc
 -- source suite/versioning/common.inc
 -- source suite/versioning/engines.inc
+-- source include/have_sequence.inc
+
+set @save_persistent=@@global.innodb_stats_persistent;
+set global innodb_stats_persistent= 0;
 
 set system_versioning_alter_history=keep;
 --echo # Check conventional partitioning on temporal tables
@@ -782,6 +786,44 @@ delete from t1 partition (p0, pn);
 delete from t1 partition (p0, p1, pn);
 drop table t1;
 set timestamp= default;
+
+--echo #
+--echo # MDEV-25546 LIMIT partitioning does not respect ROLLBACK
+--echo #
+create or replace table t1 (pk int primary key)
+with system versioning engine innodb
+partition by system_time limit 100 (
+  partition p0 history,
+  partition p1 history,
+  partition pn current);
+insert into t1 select seq from seq_1_to_90;
+
+start transaction;
+# Puts 80 rows into p0
+replace into t1 select seq from seq_1_to_80;
+# Puts another 70 rows into p0
+replace into t1 select seq from seq_1_to_70;
+# Puts 60 rows into p1
+replace into t1 select seq from seq_1_to_60;
+
+select partition_name, table_rows
+from information_schema.partitions
+where table_name = 't1';
+rollback;
+
+select partition_name, table_rows
+from information_schema.partitions
+where table_name = 't1';
+
+# Should put 10 rows into the empty partition p0
+replace into t1 select seq from seq_1_to_10;
+select partition_name, table_rows
+from information_schema.partitions
+where table_name = 't1';
+ # Cleanup
+drop table t1;
 --echo # End of 10.3 tests
 
+set global innodb_stats_persistent= @save_persistent;
+
 --source suite/versioning/common_finish.inc
diff --git a/sql/partition_info.cc b/sql/partition_info.cc
index edcdd6d2b37..f523415f6cc 100644
--- a/sql/partition_info.cc
+++ b/sql/partition_info.cc
@@ -843,12 +843,10 @@ int partition_info::vers_set_hist_part(THD *thd)
   if (vers_info->limit)
   {
     ha_partition *hp= (ha_partition*)(table->file);
-    partition_element *next= NULL;
+    partition_element *next;
     List_iterator<partition_element> it(partitions);
-    while (next != vers_info->hist_part)
-      next= it++;
-    DBUG_ASSERT(bitmap_is_set(&read_partitions, next->id));
-    ha_rows records= hp->part_records(next);
+    ha_rows records= 0;
+    vers_info->hist_part= partitions.head();
     while ((next= it++) != vers_info->now_part)
     {
       DBUG_ASSERT(bitmap_is_set(&read_partitions, next->id));