MDEV-17554 Auto-create new partition for system versioned tables with history partitioned by INTERVAL/LIMIT

:: Syntax change ::

Keyword AUTO enables history partition auto-creation.

Examples:

    CREATE TABLE t1 (x int) WITH SYSTEM VERSIONING
    PARTITION BY SYSTEM_TIME INTERVAL 1 HOUR AUTO;

    CREATE TABLE t1 (x int) WITH SYSTEM VERSIONING
    PARTITION BY SYSTEM_TIME INTERVAL 1 MONTH
    STARTS '2021-01-01 00:00:00' AUTO PARTITIONS 12;

    CREATE TABLE t1 (x int) WITH SYSTEM VERSIONING
    PARTITION BY SYSTEM_TIME LIMIT 1000 AUTO;

Or with explicit partitions:

    CREATE TABLE t1 (x int) WITH SYSTEM VERSIONING
    PARTITION BY SYSTEM_TIME INTERVAL 1 HOUR AUTO
    (PARTITION p0 HISTORY, PARTITION pn CURRENT);

To disable or enable auto-creation one can use ALTER TABLE by adding
or removing AUTO from partitioning specification:

    CREATE TABLE t1 (x int) WITH SYSTEM VERSIONING
    PARTITION BY SYSTEM_TIME INTERVAL 1 HOUR AUTO;

    # Disables auto-creation:
    ALTER TABLE t1 PARTITION BY SYSTEM_TIME INTERVAL 1 HOUR;

    # Enables auto-creation:
    ALTER TABLE t1 PARTITION BY SYSTEM_TIME INTERVAL 1 HOUR AUTO;

If the rest of partitioning specification is identical to CREATE TABLE
no repartitioning will be done (for details see MDEV-27328).

:: Description ::

Before executing history-generating DML command (see the list of commands below)
add N history partitions, so that N would be sufficient for potentially
generated history. N > 1 may be required when history partitions are switched
by INTERVAL and current_timestamp is N times further than the interval
boundary of the last history partition.

If the last history partition equals or exceeds LIMIT records then new history
partition is created and selected as the working partition. According to
MDEV-28411 partitions cannot be switched (or created) while the command is
running. Thus LIMIT does not carry strict limitation and the history partition
size must be planned as LIMIT value plus average number of history one DML
command can generate.

Auto-creation is implemented by synchronous fast_alter_partition_table() call
from the thread of the executed DML command before the command itself is run
(by the fallback and retry mechanism similar to Discovery feature,
see Open_table_context).

The name for newly added partitions are generated like default partition names
with extension of MDEV-22155 (which avoids name clashes by extending assignment
counter to next free-enough gap).

These DML commands can trigger auto-creation:

    DELETE (including multitable DELETE, excluding DELETE HISTORY)
    UPDATE (including multitable UPDATE)
    REPLACE (including REPLACE .. SELECT)
    INSERT .. ON DUPLICATE KEY UPDATE (including INSERT .. SELECT .. ODKU)
    LOAD DATA .. REPLACE

:: Bug fixes ::

MDEV-23642 Locking timeout caused by auto-creation affects original DML

    The reasons for this are:

    - Do not disrupt main business process (the history is auxiliary service);

    - Consequences are non-fatal (history is not lost, but comes into wrong
      partition; fixed by partitioning rebuild);

    - There is more freedom for application to fail in this case or not: it may
      read warning info and find corresponding error number.

    - While non-failing command is easy to handle by an application and fail it,
      the opposite is hard to handle: there is no automatic actions to fix
      failed command and retry, DBA intervention is required and until then
      application is non-functioning.

MDEV-23639 Auto-create does not work under LOCK TABLES or inside triggers

    Don't do tdc_remove_table() for OT_ADD_HISTORY_PARTITION because it is
    not possible in locked tables mode.

    LTM_LOCK_TABLES mode (and LTM_PRELOCKED_UNDER_LOCK_TABLES) works out
    of the box as fast_alter_partition_table() can reopen tables via
    locked_tables_list.

    In LTM_PRELOCKED we reopen and relock table manually.

:: More fixes ::

* some_table_marked_for_reopen flag fix

  some_table_marked_for_reopen affets only reopen of
  m_locked_tables. I.e. Locked_tables_list::reopen_tables() reopens only
  tables from m_locked_tables.

* Unused can_recover_from_failed_open() condition

  Is recover_from_failed_open() can be really used after
  open_and_process_routine()?

:: Reviewed by ::

Sergei Golubchik <serg@mariadb.org>

49 files changed, 3219 insertions, 184 deletions

diff --git a/include/my_base.h b/include/my_base.h
index f96e569cae4..388eab97d44 100644
--- a/include/my_base.h
+++ b/include/my_base.h
@@ -35,7 +35,7 @@
 
 #define HA_OPEN_ABORT_IF_LOCKED		0U	/* default */
 #define HA_OPEN_WAIT_IF_LOCKED		1U
-#define HA_OPEN_IGNORE_IF_LOCKED	2U
+#define HA_OPEN_IGNORE_IF_LOCKED	2U      /* Ignore lock error */
 #define HA_OPEN_TMP_TABLE		4U	/* Table is a temp table */
 #define HA_OPEN_DELAY_KEY_WRITE		8U	/* Don't update index  */
 #define HA_OPEN_ABORT_IF_CRASHED	16U
diff --git a/mysql-test/suite/versioning/common.inc b/mysql-test/suite/versioning/common.inc
index 0c65720fa1a..aff6b9fbc2d 100644
--- a/mysql-test/suite/versioning/common.inc
+++ b/mysql-test/suite/versioning/common.inc
@@ -6,6 +6,7 @@ source include/have_innodb.inc;
 --disable_query_log
 
 set @@session.time_zone='+00:00';
+set @@global.time_zone='+00:00';
 select ifnull(max(transaction_id), 0) into @start_trx_id from mysql.transaction_registry;
 set @test_start=now(6);
 
diff --git a/mysql-test/suite/versioning/common_finish.inc b/mysql-test/suite/versioning/common_finish.inc
index 3c4e7b66ff3..255daa1963a 100644
--- a/mysql-test/suite/versioning/common_finish.inc
+++ b/mysql-test/suite/versioning/common_finish.inc
@@ -6,5 +6,6 @@ drop function if exists current_row;
 drop function if exists check_row;
 drop function if exists current_row_ts;
 drop function if exists check_row_ts;
+set @@global.time_zone= default;
 --enable_warnings
 --enable_query_log
diff --git a/mysql-test/suite/versioning/r/debug.result b/mysql-test/suite/versioning/r/debug.result
index 0e2c67fd9aa..f011bcf4057 100644
--- a/mysql-test/suite/versioning/r/debug.result
+++ b/mysql-test/suite/versioning/r/debug.result
@@ -51,6 +51,8 @@ t4	CREATE TABLE `t4` (
   PERIOD FOR SYSTEM_TIME (`row_start`, `row_end`)
 ) ENGINE=MyISAM DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
 drop table t1, t2, t3, t4;
+set debug_dbug= @old_dbug;
+set global debug_dbug= @old_dbug;
 #
 # MDEV-19525 remove ER_VERS_FIELD_WRONG_TYPE from init_from_binary_frm_image()
 #
@@ -63,4 +65,42 @@ Level	Code	Message
 Warning	4110	`row_start` must be of type TIMESTAMP(6) for system-versioned table `t1`
 Error	1033	Incorrect information in file: './test/t1.frm'
 drop table t1;
-set global debug_dbug=@old_dbug;
+set debug_dbug= @old_dbug;
+#
+# MDEV-17554 Auto-create new partition for system versioned tables
+#            with history partitioned by INTERVAL/LIMIT
+#
+call mtr.add_suppression("need more HISTORY partitions");
+create or replace table t1 (x int) with system versioning
+partition by system_time limit 1 auto partitions 2;
+insert into t1 values (1);
+update t1 set x= x + 1;
+connect con2, localhost, root;
+connect con1, localhost, root;
+# Both threads create partition simultaneously
+connection con1;
+set debug_sync= 'add_history_partition signal s1 wait_for s2';
+update t1 set x= x + 10;
+connection con2;
+set debug_sync= 'now wait_for s1';
+flush tables t1;
+set debug_sync= 'add_history_partition signal s2';
+update t1 set x= x + 20;
+connection con1;
+connection default;
+update t1 set x= x + 2;
+# Second thread skips to reopen 3 times until first thread creates partition
+connection con1;
+set debug_sync= 'add_history_partition SIGNAL s1 WAIT_FOR s2';
+update t1 set x= x + 30;
+connection con2;
+set debug_sync= 'now WAIT_FOR s1';
+set debug_sync= 'reopen_history_partition SIGNAL s2 EXECUTE 3';
+update t1 set x= x + 40;
+connection con1;
+connection default;
+disconnect con1;
+disconnect con2;
+set @@timestamp= default;
+drop tables t1;
+set debug_sync= 'reset';
diff --git a/mysql-test/suite/versioning/r/delete_history.result b/mysql-test/suite/versioning/r/delete_history.result
index 5b4fb5309e7..9e72826dac1 100644
--- a/mysql-test/suite/versioning/r/delete_history.result
+++ b/mysql-test/suite/versioning/r/delete_history.result
@@ -63,8 +63,6 @@ insert into t values (1);
 update t set a= 2;
 update t set a= 3;
 delete history from t;
-Warnings:
-Warning	4114	Versioned table `test`.`t`: last HISTORY partition (`p1`) is out of LIMIT, need more HISTORY partitions
 # The above warning is one command late (MDEV-20345) ^^^
 select * from t for system_time all;
 a
@@ -199,3 +197,21 @@ DELETE HISTORY FROM v1;
 ERROR HY000: The target table v1 of the DELETE is not updatable
 DROP VIEW v1;
 DROP TABLE t1;
+#
+# MDEV-17554 Auto-create new partition for system versioned tables with history partitioned by INTERVAL/LIMIT
+#
+# Don't auto-create new partition on DELETE HISTORY:
+set timestamp= unix_timestamp('2000-01-01 00:00:00');
+create or replace table t (a int) with system versioning
+partition by system_time interval 1 hour auto;
+set timestamp= unix_timestamp('2000-01-01 10:00:00');
+delete history from t;
+set timestamp= default;
+show create table t;
+Table	Create Table
+t	CREATE TABLE `t` (
+  `a` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME INTERVAL 1 HOUR STARTS TIMESTAMP'2000-01-01 00:00:00' AUTO
+PARTITIONS 2
+drop table t;
diff --git a/mysql-test/suite/versioning/r/not_embedded.result b/mysql-test/suite/versioning/r/not_embedded.result
index f17ab18349c..ed3d6b45a32 100644
--- a/mysql-test/suite/versioning/r/not_embedded.result
+++ b/mysql-test/suite/versioning/r/not_embedded.result
@@ -70,3 +70,26 @@ drop database mysqltest;
 disconnect user1;
 disconnect root;
 connection default;
+#
+# MDEV-25559 Auto-create: infinite loop after interrupted lock wait
+#
+set timestamp= unix_timestamp('2000-01-01 00:00:00');
+create table t (pk int primary key, a int) engine innodb with system versioning
+partition by system_time interval 1 hour auto;
+insert into t values (1, 0);
+begin;
+update t set a= a + 1;
+connect  con1,localhost,root,,;
+set max_statement_time= 1;
+set timestamp= unix_timestamp('2000-01-01 01:00:00');
+update t set a= a + 2;
+connection default;
+set timestamp= unix_timestamp('2000-01-01 01:00:00');
+update t set a= a + 3;
+connection con1;
+ERROR 70100: Query execution was interrupted (max_statement_time exceeded)
+disconnect con1;
+connection default;
+commit;
+drop table t;
+set timestamp= default;
diff --git a/mysql-test/suite/versioning/r/partition,heap.rdiff b/mysql-test/suite/versioning/r/partition,heap.rdiff
index f7b534bb114..779b715f813 100644
--- a/mysql-test/suite/versioning/r/partition,heap.rdiff
+++ b/mysql-test/suite/versioning/r/partition,heap.rdiff
@@ -1,6 +1,6 @@
 --- partition.result
 +++ partition,heap.reject
-@@ -1545,82 +1545,3 @@
+@@ -1582,85 +1582,6 @@
  (PARTITION `p0` HISTORY ENGINE = X,
   PARTITION `pn` CURRENT ENGINE = X)
  drop tables t1, tp1;
@@ -83,3 +83,141 @@
 -(PARTITION `p0` VALUES LESS THAN (10) ENGINE = X,
 - PARTITION `pn` VALUES LESS THAN MAXVALUE ENGINE = X)
 -drop tables t1, tp1;
+ #
+ # MDEV-17554 Auto-create new partition for system versioned tables with history partitioned by INTERVAL/LIMIT
+ #
+@@ -2193,134 +2114,6 @@
+ Warning	4114	Versioned table `test`.`t1`: last HISTORY partition (`p0`) is out of INTERVAL, need more HISTORY partitions
+ drop prepare s;
+ affected rows: 0
+-# Complex table
+-set timestamp= unix_timestamp('2000-01-01 00:00:00');
+-affected rows: 0
+-create or replace table t1 (
+-x int primary key auto_increment,
+-t timestamp(6) default '2001-11-11 11:11:11',
+-b blob(4096) compressed null,
+-c varchar(1033) character set utf8 not null,
+-u int unique,
+-m enum('a', 'b', 'c') not null default 'a' comment 'absolute',
+-i1 tinyint, i2 smallint, i3 bigint,
+-index three(i1, i2, i3),
+-v1 timestamp(6) generated always as (t + interval 1 day),
+-v2 timestamp(6) generated always as (t + interval 1 month) stored,
+-s timestamp(6) as row start,
+-e timestamp(6) as row end,
+-period for system_time (s, e),
+-ps date, pe date,
+-period for app_time (ps, pe),
+-constraint check_constr check (u > -1))
+-with system versioning default charset=ucs2
+-partition by system_time interval 1 hour auto (
+-partition p2 history,
+-partition pn current);
+-affected rows: 0
+-show create table t1;
+-Table	Create Table
+-t1	CREATE TABLE `t1` (
+-  `x` int(11) NOT NULL AUTO_INCREMENT,
+-  `t` timestamp(6) NOT NULL DEFAULT '2001-11-11 11:11:11.000000',
+-  `b` blob /*!100301 COMPRESSED*/ DEFAULT NULL,
+-  `c` varchar(1033) CHARACTER SET utf8mb3 NOT NULL,
+-  `u` int(11) DEFAULT NULL,
+-  `m` enum('a','b','c') NOT NULL DEFAULT 'a' COMMENT 'absolute',
+-  `i1` tinyint(4) DEFAULT NULL,
+-  `i2` smallint(6) DEFAULT NULL,
+-  `i3` bigint(20) DEFAULT NULL,
+-  `v1` timestamp(6) GENERATED ALWAYS AS (`t` + interval 1 day) VIRTUAL,
+-  `v2` timestamp(6) GENERATED ALWAYS AS (`t` + interval 1 month) STORED,
+-  `s` timestamp(6) GENERATED ALWAYS AS ROW START,
+-  `e` timestamp(6) GENERATED ALWAYS AS ROW END,
+-  `ps` date NOT NULL,
+-  `pe` date NOT NULL,
+-  PERIOD FOR `app_time` (`ps`, `pe`),
+-  PRIMARY KEY (`x`,`e`),
+-  UNIQUE KEY `u` (`u`,`e`),
+-  KEY `three` (`i1`,`i2`,`i3`),
+-  PERIOD FOR SYSTEM_TIME (`s`, `e`),
+-  CONSTRAINT `check_constr` CHECK (`u` > -1)
+-) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=ucs2 WITH SYSTEM VERSIONING
+- PARTITION BY SYSTEM_TIME INTERVAL 1 HOUR STARTS TIMESTAMP'2000-01-01 00:00:00' AUTO
+-(PARTITION `p2` HISTORY ENGINE = DEFAULT_ENGINE,
+- PARTITION `pn` CURRENT ENGINE = DEFAULT_ENGINE)
+-affected rows: 1
+-insert into t1 (x, c, u, i1, i2, i3, ps, pe)
+-values (1, 'cc', 0, 1, 2, 3, '1999-01-01', '2000-01-01');
+-affected rows: 1
+-set timestamp= unix_timestamp('2000-01-01 01:00:00');
+-affected rows: 0
+-update t1 set x= x + 8;
+-affected rows: 1
+-info: Rows matched: 1  Changed: 1  Inserted: 1  Warnings: 0
+-show create table t1;
+-Table	Create Table
+-t1	CREATE TABLE `t1` (
+-  `x` int(11) NOT NULL AUTO_INCREMENT,
+-  `t` timestamp(6) NOT NULL DEFAULT '2001-11-11 11:11:11.000000',
+-  `b` blob /*!100301 COMPRESSED*/ DEFAULT NULL,
+-  `c` varchar(1033) CHARACTER SET utf8mb3 NOT NULL,
+-  `u` int(11) DEFAULT NULL,
+-  `m` enum('a','b','c') NOT NULL DEFAULT 'a' COMMENT 'absolute',
+-  `i1` tinyint(4) DEFAULT NULL,
+-  `i2` smallint(6) DEFAULT NULL,
+-  `i3` bigint(20) DEFAULT NULL,
+-  `v1` timestamp(6) GENERATED ALWAYS AS (`t` + interval 1 day) VIRTUAL,
+-  `v2` timestamp(6) GENERATED ALWAYS AS (`t` + interval 1 month) STORED,
+-  `s` timestamp(6) GENERATED ALWAYS AS ROW START,
+-  `e` timestamp(6) GENERATED ALWAYS AS ROW END,
+-  `ps` date NOT NULL,
+-  `pe` date NOT NULL,
+-  PERIOD FOR `app_time` (`ps`, `pe`),
+-  PRIMARY KEY (`x`,`e`),
+-  UNIQUE KEY `u` (`u`,`e`),
+-  KEY `three` (`i1`,`i2`,`i3`),
+-  PERIOD FOR SYSTEM_TIME (`s`, `e`),
+-  CONSTRAINT `check_constr` CHECK (`u` > -1)
+-) ENGINE=DEFAULT_ENGINE AUTO_INCREMENT=10 DEFAULT CHARSET=ucs2 WITH SYSTEM VERSIONING
+- PARTITION BY SYSTEM_TIME INTERVAL 1 HOUR STARTS TIMESTAMP'2000-01-01 00:00:00' AUTO
+-(PARTITION `p2` HISTORY ENGINE = DEFAULT_ENGINE,
+- PARTITION `p1` HISTORY ENGINE = DEFAULT_ENGINE,
+- PARTITION `pn` CURRENT ENGINE = DEFAULT_ENGINE)
+-affected rows: 1
+-set timestamp= unix_timestamp('2000-01-01 02:00:00');
+-affected rows: 0
+-update t1 set x= x - 8;
+-affected rows: 1
+-info: Rows matched: 1  Changed: 1  Inserted: 1  Warnings: 0
+-show create table t1;
+-Table	Create Table
+-t1	CREATE TABLE `t1` (
+-  `x` int(11) NOT NULL AUTO_INCREMENT,
+-  `t` timestamp(6) NOT NULL DEFAULT '2001-11-11 11:11:11.000000',
+-  `b` blob /*!100301 COMPRESSED*/ DEFAULT NULL,
+-  `c` varchar(1033) CHARACTER SET utf8mb3 NOT NULL,
+-  `u` int(11) DEFAULT NULL,
+-  `m` enum('a','b','c') NOT NULL DEFAULT 'a' COMMENT 'absolute',
+-  `i1` tinyint(4) DEFAULT NULL,
+-  `i2` smallint(6) DEFAULT NULL,
+-  `i3` bigint(20) DEFAULT NULL,
+-  `v1` timestamp(6) GENERATED ALWAYS AS (`t` + interval 1 day) VIRTUAL,
+-  `v2` timestamp(6) GENERATED ALWAYS AS (`t` + interval 1 month) STORED,
+-  `s` timestamp(6) GENERATED ALWAYS AS ROW START,
+-  `e` timestamp(6) GENERATED ALWAYS AS ROW END,
+-  `ps` date NOT NULL,
+-  `pe` date NOT NULL,
+-  PERIOD FOR `app_time` (`ps`, `pe`),
+-  PRIMARY KEY (`x`,`e`),
+-  UNIQUE KEY `u` (`u`,`e`),
+-  KEY `three` (`i1`,`i2`,`i3`),
+-  PERIOD FOR SYSTEM_TIME (`s`, `e`),
+-  CONSTRAINT `check_constr` CHECK (`u` > -1)
+-) ENGINE=DEFAULT_ENGINE AUTO_INCREMENT=10 DEFAULT CHARSET=ucs2 WITH SYSTEM VERSIONING
+- PARTITION BY SYSTEM_TIME INTERVAL 1 HOUR STARTS TIMESTAMP'2000-01-01 00:00:00' AUTO
+-(PARTITION `p2` HISTORY ENGINE = DEFAULT_ENGINE,
+- PARTITION `p1` HISTORY ENGINE = DEFAULT_ENGINE,
+- PARTITION `p3` HISTORY ENGINE = DEFAULT_ENGINE,
+- PARTITION `pn` CURRENT ENGINE = DEFAULT_ENGINE)
+-affected rows: 1
+ # INSERT .. ON DUPLICATE KEY UPDATE (ODKU)
+ set timestamp= unix_timestamp('2000-01-01 00:00:00');
+ create or replace table t1 (x int primary key) with system versioning
diff --git a/mysql-test/suite/versioning/r/partition.result b/mysql-test/suite/versioning/r/partition.result
index 3e5ac95c34f..bb8fb980f45 100644
--- a/mysql-test/suite/versioning/r/partition.result
+++ b/mysql-test/suite/versioning/r/partition.result
@@ -237,7 +237,9 @@ select @ts0 = @ts1;
 select @ts2 = @ts3;
 @ts2 = @ts3
 1
-## rotation by LIMIT
+#
+# Rotation by LIMIT
+#
 create or replace table t1 (x int)
 with system versioning
 partition by system_time limit 0 partitions 3;
@@ -276,8 +278,6 @@ x
 5
 6
 insert into t1 values (7), (8);
-Warnings:
-Warning	4114	Versioned table `test`.`t1`: last HISTORY partition (`p1`) is out of LIMIT, need more HISTORY partitions
 ### warn about full partition
 delete from t1;
 Warnings:
@@ -289,7 +289,9 @@ x
 6
 7
 8
-## rotation by INTERVAL
+#
+# Rotation by INTERVAL
+#
 create or replace table t1 (x int)
 with system versioning
 partition by system_time interval 0 second partitions 3;
@@ -349,19 +351,25 @@ delete from t1;
 set timestamp=unix_timestamp('2001-02-04 10:20:50');
 insert t1 values (2);
 delete from t1;
-select subpartition_name,partition_description,table_rows from information_schema.partitions where table_schema='test' and table_name='t1';
+Warnings:
+Warning	4114	Versioned table `test`.`t1`: last HISTORY partition (`p1`) is out of INTERVAL, need more HISTORY partitions
+select subpartition_name, partition_description, table_rows from information_schema.partitions where table_schema='test' and table_name='t1';
 subpartition_name	partition_description	table_rows
 p1sp0	2001-02-04 00:00:00	1
 p1sp1	2001-02-04 00:00:00	1
 pnsp0	CURRENT	0
 pnsp1	CURRENT	0
+select * from t1 partition (p1);
+i
+1
+2
 set timestamp=unix_timestamp('2001-02-04 10:20:55');
 alter table t1 add partition (partition p0 history, partition p2 history);
 set timestamp=unix_timestamp('2001-02-04 10:30:00');
 insert t1 values (4),(5);
 set timestamp=unix_timestamp('2001-02-04 10:30:10');
 update t1 set i=6 where i=5;
-select subpartition_name,partition_description,table_rows from information_schema.partitions where table_schema='test' and table_name='t1';
+select subpartition_name, partition_description, table_rows from information_schema.partitions where table_schema='test' and table_name='t1';
 subpartition_name	partition_description	table_rows
 p1sp0	2001-02-04 00:00:00	1
 p1sp1	2001-02-04 00:00:00	0
@@ -371,6 +379,25 @@ p2sp0	2001-02-06 00:00:00	0
 p2sp1	2001-02-06 00:00:00	0
 pnsp0	CURRENT	0
 pnsp1	CURRENT	2
+select * from t1 partition (p1);
+i
+1
+select * from t1 partition (p0);
+i
+5
+2
+select * from t1 partition (p2);
+i
+alter table t1 rebuild partition p0, p1, p2;
+select * from t1 partition (p1);
+i
+1
+select * from t1 partition (p0);
+i
+5
+2
+select * from t1 partition (p2);
+i
 ## pruning check
 set @ts=(select partition_description from information_schema.partitions
 where table_schema='test' and table_name='t1' and partition_name='p0' limit 1);
@@ -554,12 +581,15 @@ set timestamp= unix_timestamp('2001-01-01 00:00:00');
 create or replace table t1 (i int) with system versioning
 partition by system_time interval 1 day starts '2000-01-01 00:00:00'
 partitions 3;
-# we are warned when we push to present:
 insert into t1 values (0);
 set timestamp= unix_timestamp('2001-01-01 00:00:01');
 update t1 set i= i + 1;
+Warnings:
+Warning	4114	Versioned table `test`.`t1`: last HISTORY partition (`p1`) is out of INTERVAL, need more HISTORY partitions
 set timestamp= unix_timestamp('2001-01-01 00:00:02');
 update t1 set i= i + 1;
+Warnings:
+Warning	4114	Versioned table `test`.`t1`: last HISTORY partition (`p1`) is out of INTERVAL, need more HISTORY partitions
 select *, row_end from t1 partition (p0);
 i	row_end
 select *, row_end from t1 partition (p1);
@@ -590,8 +620,8 @@ i	row_end
 select *, row_end from t1 partition (p1);
 i	row_end
 3	2000-01-04 00:00:01.000000
-set timestamp= unix_timestamp('2000-01-01 00:00:00');
 # and this is how it usually goes:
+set timestamp= unix_timestamp('2000-01-01 00:00:00');
 create or replace table t1 (i int) with system versioning
 partition by system_time interval 1 day
 partitions 3;
@@ -602,8 +632,12 @@ set timestamp= unix_timestamp('2000-01-02 00:00:01');
 update t1 set i= i + 1;
 set timestamp= unix_timestamp('2000-01-03 00:00:01');
 update t1 set i= i + 1;
+Warnings:
+Warning	4114	Versioned table `test`.`t1`: last HISTORY partition (`p1`) is out of INTERVAL, need more HISTORY partitions
 set timestamp= unix_timestamp('2000-01-04 00:00:01');
 update t1 set i= i + 1;
+Warnings:
+Warning	4114	Versioned table `test`.`t1`: last HISTORY partition (`p1`) is out of INTERVAL, need more HISTORY partitions
 alter table t1 add partition (partition p2 history, partition p3 history);
 select *, row_end from t1 partition (p0);
 i	row_end
@@ -634,9 +668,9 @@ select * from t1 partition (pnsp1);
 x
 2
 4
-### warn about full partition
 delete from t1 where x < 3;
 delete from t1;
+### warn about full partition
 delete from t1;
 Warnings:
 Warning	4114	Versioned table `test`.`t1`: last HISTORY partition (`p1`) is out of LIMIT, need more HISTORY partitions
@@ -1117,11 +1151,11 @@ f varchar(45000)
 partition by system_time interval 1 year (partition p1 history,
 partition pn current);
 insert into t1 () values (),(),(),(),(),(),(),(),(),(),(),(),(),(),(),();
-select * into outfile 'load.data' from t1;
-load data infile 'load.data' replace into table t1;
-load data infile 'load.data' replace into table t1;
+select * into outfile 'MDEV-17891.data' from t1;
+load data infile 'MDEV-17891.data' replace into table t1;
+load data infile 'MDEV-17891.data' replace into table t1;
 ERROR HY000: The table 't1' is full
-load data infile 'load.data' replace into table t1;
+load data infile 'MDEV-17891.data' replace into table t1;
 ERROR HY000: The table 't1' is full
 set @@max_heap_table_size= 1048576;
 drop table t1;
@@ -1143,6 +1177,8 @@ insert into t1 values (0);
 update t1 set x= x + 1;
 set timestamp= unix_timestamp('2000-01-01 02:00:01');
 update t1 set x= x + 1;
+Warnings:
+Warning	4114	Versioned table `test`.`t1`: last HISTORY partition (`p1`) is out of INTERVAL, need more HISTORY partitions
 select *, row_start, row_end from t1 for system_time as of '2000-01-01 02:00:00';
 x	row_start	row_end
 1	2000-01-01 00:00:00.000000	2000-01-01 02:00:01.000000
@@ -1192,6 +1228,11 @@ delete from t1 partition (p0, pn);
 ERROR HY000: Not allowed for system-versioned table `test`.`t1`
 delete from t1 partition (p0, p1, pn);
 ERROR HY000: Not allowed for system-versioned table `test`.`t1`
+lock tables t1 write;
+delete from t1 partition (p0, pn);
+ERROR HY000: Not allowed for system-versioned table `test`.`t1`
+delete from t1;
+unlock tables;
 drop table t1;
 set timestamp= default;
 # End of 10.3 tests
@@ -1453,10 +1494,6 @@ select * from t1 partition (pn);
 x
 6
 drop table t1;
-#
-# End of 10.5 tests
-#
-SET GLOBAL innodb_stats_persistent=@save_persistent;
 # End of 10.6 tests
 #
 # MDEV-22166 MIGRATE PARTITION: move out partition into a table
@@ -1822,3 +1859,1047 @@ t1	CREATE TABLE `t1` (
 (PARTITION `p0` VALUES LESS THAN (10) ENGINE = X,
  PARTITION `pn` VALUES LESS THAN MAXVALUE ENGINE = X)
 drop tables t1, tp1;
+#
+# MDEV-17554 Auto-create new partition for system versioned tables
+#            with history partitioned by INTERVAL/LIMIT
+#
+create or replace table t1 (x int) with system versioning
+partition by system_time limit 1 auto;
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME LIMIT 1 AUTO
+PARTITIONS 2
+# Turn off AUTO
+alter table t1 partition by system_time limit 1;
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME LIMIT 1
+PARTITIONS 2
+# Get AUTO back
+alter table t1 partition by system_time limit 1 auto;
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME LIMIT 1 AUTO
+PARTITIONS 2
+insert into t1 values (1);
+create or replace table t2 (y int);
+insert into t2 values (2);
+insert into t1 select * from t2;
+insert into t2 select * from t1;
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME LIMIT 1 AUTO
+PARTITIONS 2
+# Too many partitions error
+set timestamp= unix_timestamp('2000-01-01 00:00:00');
+create or replace table t1 (x int) with system versioning
+partition by system_time interval 1 hour auto;
+set timestamp= unix_timestamp('2001-01-01 00:01:00');
+update t1 set x= x + 1;
+ERROR HY000: Versioned table `test`.`t1`: adding HISTORY partition(s) failed
+show warnings;
+Level	Code	Message
+Warning	1499	Too many partitions (including subpartitions) were defined
+Error	4189	Versioned table `test`.`t1`: adding HISTORY partition(s) failed
+# Auto-create failed error
+set timestamp= unix_timestamp('2000-01-01 00:00:00');
+create or replace table t1 (x int) with system versioning engine innodb
+partition by system_time interval 1 hour auto;
+insert into t1 values (1);
+call mtr.add_suppression("rror number .*(File exists|file operation)");
+call mtr.add_suppression("InnoDB: Cannot create file");
+set timestamp= unix_timestamp('2000-01-01 01:00:00');
+update t1 set x= x + 2;
+ERROR HY000: Got error 184 "Tablespace already exists" from storage engine InnoDB
+show warnings;
+Level	Code	Message
+Error	1030	Got error 184 "Tablespace already exists" from storage engine InnoDB
+Warning	4189	Versioned table `test`.`t1`: adding HISTORY partition(s) failed
+# Partition overflow error and manual fix
+set timestamp= unix_timestamp('2000-01-01 00:00:00');
+create or replace table t1 (x int) with system versioning
+partition by system_time interval 1 hour;
+insert into t1 values (440);
+set timestamp= unix_timestamp('2000-01-01 00:10:00');
+update t1 set x= x + 1;
+# Check how pruning boundaries work
+explain partitions select * from t1 for system_time as of '2000-01-01 00:59:58';
+id	select_type	table	partitions	type	possible_keys	key	key_len	ref	rows	Extra
+1	SIMPLE	t1	p0,pn	#	NULL	NULL	NULL	NULL	#	#
+explain partitions select * from t1 for system_time as of '2000-01-01 00:59:59';
+id	select_type	table	partitions	type	possible_keys	key	key_len	ref	rows	Extra
+1	SIMPLE	t1	p0,pn	#	NULL	NULL	NULL	NULL	#	#
+explain partitions select * from t1 for system_time as of '2000-01-01 01:00:00';
+id	select_type	table	partitions	type	possible_keys	key	key_len	ref	rows	Extra
+1	SIMPLE	t1	p0,pn	#	NULL	NULL	NULL	NULL	#	#
+select * from t1 for system_time as of '2000-01-01 00:09:59';
+x
+440
+set timestamp= unix_timestamp('2000-01-01 02:00:00');
+update t1 set x= x + 1;
+Warnings:
+Warning	4114	Versioned table `test`.`t1`: last HISTORY partition (`p0`) is out of INTERVAL, need more HISTORY partitions
+select * from t1 for system_time as of '2000-01-01 01:00:00';
+x
+441
+select * from t1 partition (p0) order by x;
+x
+440
+441
+# Here is how manual fix works: just add new partitions there
+alter table t1 add partition partitions 3;
+select * from t1 for system_time as of '2000-01-01 01:00:00';
+x
+441
+select * from t1 partition (p0) order by x;
+x
+440
+# Check pruning after ALTER
+explain partitions select * from t1 for system_time as of '2000-01-01 00:59:58';
+id	select_type	table	partitions	type	possible_keys	key	key_len	ref	rows	Extra
+1	SIMPLE	t1	p0,p1,p2,p3,pn	#	NULL	NULL	NULL	NULL	#	#
+explain partitions select * from t1 for system_time as of '2000-01-01 00:59:59';
+id	select_type	table	partitions	type	possible_keys	key	key_len	ref	rows	Extra
+1	SIMPLE	t1	p1,p2,p3,pn	#	NULL	NULL	NULL	NULL	#	#
+explain partitions select * from t1 for system_time as of '2000-01-01 01:00:00';
+id	select_type	table	partitions	type	possible_keys	key	key_len	ref	rows	Extra
+1	SIMPLE	t1	p1,p2,p3,pn	#	NULL	NULL	NULL	NULL	#	#
+drop table t1;
+create or replace table t1 (x int) with system versioning
+partition by system_time interval 3600 second
+starts '2000-01-01 00:00:00' auto partitions 3;
+affected rows: 0
+insert into t1 values (1);
+affected rows: 1
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME INTERVAL 3600 SECOND STARTS TIMESTAMP'2000-01-01 00:00:00' AUTO
+PARTITIONS 3
+affected rows: 1
+set timestamp= unix_timestamp('2000-01-01 02:00:00');
+affected rows: 0
+update t1 set x= x + 1;
+affected rows: 1
+info: Rows matched: 1  Changed: 1  Inserted: 1  Warnings: 0
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME INTERVAL 3600 SECOND STARTS TIMESTAMP'2000-01-01 00:00:00' AUTO
+PARTITIONS 4
+affected rows: 1
+set timestamp= unix_timestamp('2000-01-01 03:00:00');
+affected rows: 0
+update t1 set x= x + 2;
+affected rows: 1
+info: Rows matched: 1  Changed: 1  Inserted: 1  Warnings: 0
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME INTERVAL 3600 SECOND STARTS TIMESTAMP'2000-01-01 00:00:00' AUTO
+PARTITIONS 5
+affected rows: 1
+set timestamp= unix_timestamp('2000-01-01 00:00:00');
+affected rows: 0
+create or replace table t1 (x int) with system versioning
+partition by system_time interval 1 hour auto (
+partition p1 history,
+partition p3 history,
+partition pn current);
+affected rows: 0
+insert into t1 values (1);
+affected rows: 1
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME INTERVAL 1 HOUR STARTS TIMESTAMP'2000-01-01 00:00:00' AUTO
+(PARTITION `p1` HISTORY ENGINE = DEFAULT_ENGINE,
+ PARTITION `p3` HISTORY ENGINE = DEFAULT_ENGINE,
+ PARTITION `pn` CURRENT ENGINE = DEFAULT_ENGINE)
+affected rows: 1
+set timestamp= unix_timestamp('2000-01-01 02:00:00');
+affected rows: 0
+update t1 set x= x + 3;
+affected rows: 1
+info: Rows matched: 1  Changed: 1  Inserted: 1  Warnings: 0
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME INTERVAL 1 HOUR STARTS TIMESTAMP'2000-01-01 00:00:00' AUTO
+(PARTITION `p1` HISTORY ENGINE = DEFAULT_ENGINE,
+ PARTITION `p3` HISTORY ENGINE = DEFAULT_ENGINE,
+ PARTITION `p2` HISTORY ENGINE = DEFAULT_ENGINE,
+ PARTITION `pn` CURRENT ENGINE = DEFAULT_ENGINE)
+affected rows: 1
+set timestamp= unix_timestamp('2000-01-01 03:00:00');
+affected rows: 0
+update t1 set x= x + 4;
+affected rows: 1
+info: Rows matched: 1  Changed: 1  Inserted: 1  Warnings: 0
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME INTERVAL 1 HOUR STARTS TIMESTAMP'2000-01-01 00:00:00' AUTO
+(PARTITION `p1` HISTORY ENGINE = DEFAULT_ENGINE,
+ PARTITION `p3` HISTORY ENGINE = DEFAULT_ENGINE,
+ PARTITION `p2` HISTORY ENGINE = DEFAULT_ENGINE,
+ PARTITION `p4` HISTORY ENGINE = DEFAULT_ENGINE,
+ PARTITION `pn` CURRENT ENGINE = DEFAULT_ENGINE)
+affected rows: 1
+set timestamp= unix_timestamp('2000-01-01 04:00:00');
+affected rows: 0
+lock tables t1 write;
+affected rows: 0
+update t1 set x= x + 5;
+affected rows: 1
+info: Rows matched: 1  Changed: 1  Inserted: 1  Warnings: 0
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME INTERVAL 1 HOUR STARTS TIMESTAMP'2000-01-01 00:00:00' AUTO
+(PARTITION `p1` HISTORY ENGINE = DEFAULT_ENGINE,
+ PARTITION `p3` HISTORY ENGINE = DEFAULT_ENGINE,
+ PARTITION `p2` HISTORY ENGINE = DEFAULT_ENGINE,
+ PARTITION `p4` HISTORY ENGINE = DEFAULT_ENGINE,
+ PARTITION `p5` HISTORY ENGINE = DEFAULT_ENGINE,
+ PARTITION `pn` CURRENT ENGINE = DEFAULT_ENGINE)
+affected rows: 1
+unlock tables;
+affected rows: 0
+set timestamp= default;
+affected rows: 0
+# Couple of more LOCK TABLES cases
+create or replace table t1 (x int) with system versioning
+partition by system_time limit 1 auto;
+affected rows: 0
+lock tables t1 write;
+affected rows: 0
+insert into t1 values (1);
+affected rows: 1
+update t1 set x= x + 1;
+affected rows: 1
+info: Rows matched: 1  Changed: 1  Inserted: 1  Warnings: 0
+update t1 set x= x + 1;
+affected rows: 1
+info: Rows matched: 1  Changed: 1  Inserted: 1  Warnings: 0
+update t1 set x= x + 1;
+affected rows: 1
+info: Rows matched: 1  Changed: 1  Inserted: 1  Warnings: 0
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME LIMIT 1 AUTO
+PARTITIONS 4
+affected rows: 1
+unlock tables;
+affected rows: 0
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME LIMIT 1 AUTO
+PARTITIONS 4
+affected rows: 1
+# Overflow prevention under LOCK TABLES
+create or replace table t1 (x int)
+with system versioning partition by system_time
+limit 10 auto;
+affected rows: 0
+insert into t1 values (1), (2), (3), (4), (5), (6), (7), (8), (9);
+affected rows: 9
+info: Records: 9  Duplicates: 0  Warnings: 0
+update t1 set x= x + 10;
+affected rows: 9
+info: Rows matched: 9  Changed: 9  Inserted: 9  Warnings: 0
+lock tables t1 write;
+affected rows: 0
+update t1 set x= 1 where x = 11;
+affected rows: 1
+info: Rows matched: 1  Changed: 1  Inserted: 1  Warnings: 0
+update t1 set x= 2 where x = 12;
+affected rows: 1
+info: Rows matched: 1  Changed: 1  Inserted: 1  Warnings: 0
+update t1 set x= 3 where x = 13;
+affected rows: 1
+info: Rows matched: 1  Changed: 1  Inserted: 1  Warnings: 0
+unlock tables;
+affected rows: 0
+select count(x) from t1 partition (p0);
+count(x)
+10
+affected rows: 1
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME LIMIT 10 AUTO
+PARTITIONS 3
+affected rows: 1
+drop tables t1;
+affected rows: 0
+# Test VIEW, LOCK TABLES
+set timestamp= unix_timestamp('2000-01-01 00:00:00');
+affected rows: 0
+create or replace table t1 (x int) with system versioning
+partition by system_time interval 1 hour auto;
+affected rows: 0
+create or replace view v1 as select * from t1;
+affected rows: 0
+insert into t1 values (1);
+affected rows: 1
+set timestamp= unix_timestamp('2000-01-01 01:00:00');
+affected rows: 0
+update v1 set x= x + 2;
+affected rows: 1
+info: Rows matched: 1  Changed: 1  Inserted: 1  Warnings: 0
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME INTERVAL 1 HOUR STARTS TIMESTAMP'2000-01-01 00:00:00' AUTO
+PARTITIONS 3
+affected rows: 1
+set timestamp= unix_timestamp('2000-01-01 02:00:00');
+affected rows: 0
+lock tables v1 write;
+affected rows: 0
+update v1 set x= x + 3;
+affected rows: 1
+info: Rows matched: 1  Changed: 1  Inserted: 1  Warnings: 0
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME INTERVAL 1 HOUR STARTS TIMESTAMP'2000-01-01 00:00:00' AUTO
+PARTITIONS 4
+affected rows: 1
+unlock tables;
+affected rows: 0
+drop view v1;
+affected rows: 0
+drop tables t1;
+affected rows: 0
+set timestamp= unix_timestamp('2000-01-01 00:00:00');
+affected rows: 0
+create or replace table t1 (x int) with system versioning
+partition by system_time interval 1 hour auto partitions 3;
+affected rows: 0
+create or replace table t2 (y int) with system versioning
+partition by system_time interval 1 hour auto;
+affected rows: 0
+insert into t1 values (1);
+affected rows: 1
+insert into t2 values (2);
+affected rows: 1
+set timestamp= unix_timestamp('2000-01-01 01:00:00');
+affected rows: 0
+update t1, t2 set x= x + 1, y= y + 1;
+affected rows: 2
+info: Rows matched: 2  Changed: 2  Warnings: 0
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME INTERVAL 1 HOUR STARTS TIMESTAMP'2000-01-01 00:00:00' AUTO
+PARTITIONS 3
+affected rows: 1
+show create table t2;
+Table	Create Table
+t2	CREATE TABLE `t2` (
+  `y` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME INTERVAL 1 HOUR STARTS TIMESTAMP'2000-01-01 00:00:00' AUTO
+PARTITIONS 3
+affected rows: 1
+set timestamp= unix_timestamp('2000-01-01 02:00:00');
+affected rows: 0
+update t1, t2 set x= x + 1, y= y + 1;
+affected rows: 2
+info: Rows matched: 2  Changed: 2  Warnings: 0
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME INTERVAL 1 HOUR STARTS TIMESTAMP'2000-01-01 00:00:00' AUTO
+PARTITIONS 4
+affected rows: 1
+show create table t2;
+Table	Create Table
+t2	CREATE TABLE `t2` (
+  `y` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME INTERVAL 1 HOUR STARTS TIMESTAMP'2000-01-01 00:00:00' AUTO
+PARTITIONS 4
+affected rows: 1
+set timestamp= unix_timestamp('2000-01-01 03:00:00');
+affected rows: 0
+update t1, t2 set t1.x= 0 where t1.x< t2.y;
+affected rows: 1
+info: Rows matched: 1  Changed: 1  Warnings: 0
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME INTERVAL 1 HOUR STARTS TIMESTAMP'2000-01-01 00:00:00' AUTO
+PARTITIONS 5
+affected rows: 1
+show create table t2;
+Table	Create Table
+t2	CREATE TABLE `t2` (
+  `y` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME INTERVAL 1 HOUR STARTS TIMESTAMP'2000-01-01 00:00:00' AUTO
+PARTITIONS ok
+affected rows: 1
+drop tables t1, t2;
+affected rows: 0
+# PS, SP, LOCK TABLES
+set timestamp= unix_timestamp('2000-01-01 00:00:00');
+affected rows: 0
+create or replace table t1 (x int) with system versioning
+partition by system_time interval 1 hour auto;
+affected rows: 0
+insert into t1 values (1);
+affected rows: 1
+set timestamp= unix_timestamp('2000-01-01 01:00:00');
+affected rows: 0
+execute immediate 'update t1 set x= x + 5';
+affected rows: 1
+info: Rows matched: 1  Changed: 1  Inserted: 1  Warnings: 0
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME INTERVAL 1 HOUR STARTS TIMESTAMP'2000-01-01 00:00:00' AUTO
+PARTITIONS 3
+affected rows: 1
+prepare s from 'update t1 set x= x + 6';
+affected rows: 0
+info: Statement prepared
+set timestamp= unix_timestamp('2000-01-01 02:00:00');
+affected rows: 0
+execute s;
+affected rows: 1
+info: Rows matched: 1  Changed: 1  Inserted: 1  Warnings: 0
+execute s;
+affected rows: 1
+info: Rows matched: 1  Changed: 1  Inserted: 1  Warnings: 0
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME INTERVAL 1 HOUR STARTS TIMESTAMP'2000-01-01 00:00:00' AUTO
+PARTITIONS 4
+affected rows: 1
+set timestamp= unix_timestamp('2000-01-01 03:00:00');
+affected rows: 0
+lock tables t1 write;
+affected rows: 0
+execute s;
+affected rows: 1
+info: Rows matched: 1  Changed: 1  Inserted: 1  Warnings: 0
+execute s;
+affected rows: 1
+info: Rows matched: 1  Changed: 1  Inserted: 1  Warnings: 0
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME INTERVAL 1 HOUR STARTS TIMESTAMP'2000-01-01 00:00:00' AUTO
+PARTITIONS 5
+affected rows: 1
+unlock tables;
+affected rows: 0
+drop prepare s;
+affected rows: 0
+create procedure sp() update t1 set x= x + 7;
+affected rows: 0
+set timestamp= unix_timestamp('2000-01-01 04:00:00');
+affected rows: 0
+call sp;
+affected rows: 1
+call sp;
+affected rows: 1
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME INTERVAL 1 HOUR STARTS TIMESTAMP'2000-01-01 00:00:00' AUTO
+PARTITIONS 6
+affected rows: 1
+set timestamp= unix_timestamp('2000-01-01 05:00:00');
+affected rows: 0
+lock tables t1 write;
+affected rows: 0
+call sp;
+affected rows: 1
+call sp;
+affected rows: 1
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME INTERVAL 1 HOUR STARTS TIMESTAMP'2000-01-01 00:00:00' AUTO
+PARTITIONS 7
+affected rows: 1
+unlock tables;
+affected rows: 0
+drop procedure sp;
+affected rows: 0
+set timestamp= unix_timestamp('2001-01-01 00:00:00');
+affected rows: 0
+create or replace table t1 (i int) with system versioning
+partition by system_time interval 1 day starts '2001-01-01 00:00:00';
+affected rows: 0
+insert into t1 values (0);
+affected rows: 1
+set timestamp= unix_timestamp('2001-01-01 00:00:01');
+affected rows: 0
+prepare s from 'update t1 set i= i + 1';
+affected rows: 0
+info: Statement prepared
+execute s;
+affected rows: 1
+info: Rows matched: 1  Changed: 1  Inserted: 1  Warnings: 0
+set timestamp= unix_timestamp('2001-01-02 00:00:01');
+affected rows: 0
+execute s;
+affected rows: 1
+info: Rows matched: 1  Changed: 1  Inserted: 1  Warnings: 1
+Warnings:
+Warning	4114	Versioned table `test`.`t1`: last HISTORY partition (`p0`) is out of INTERVAL, need more HISTORY partitions
+drop prepare s;
+affected rows: 0
+# Complex table
+set timestamp= unix_timestamp('2000-01-01 00:00:00');
+affected rows: 0
+create or replace table t1 (
+x int primary key auto_increment,
+t timestamp(6) default '2001-11-11 11:11:11',
+b blob(4096) compressed null,
+c varchar(1033) character set utf8 not null,
+u int unique,
+m enum('a', 'b', 'c') not null default 'a' comment 'absolute',
+i1 tinyint, i2 smallint, i3 bigint,
+index three(i1, i2, i3),
+v1 timestamp(6) generated always as (t + interval 1 day),
+v2 timestamp(6) generated always as (t + interval 1 month) stored,
+s timestamp(6) as row start,
+e timestamp(6) as row end,
+period for system_time (s, e),
+ps date, pe date,
+period for app_time (ps, pe),
+constraint check_constr check (u > -1))
+with system versioning default charset=ucs2
+partition by system_time interval 1 hour auto (
+partition p2 history,
+partition pn current);
+affected rows: 0
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) NOT NULL AUTO_INCREMENT,
+  `t` timestamp(6) NOT NULL DEFAULT '2001-11-11 11:11:11.000000',
+  `b` blob /*!100301 COMPRESSED*/ DEFAULT NULL,
+  `c` varchar(1033) CHARACTER SET utf8mb3 NOT NULL,
+  `u` int(11) DEFAULT NULL,
+  `m` enum('a','b','c') NOT NULL DEFAULT 'a' COMMENT 'absolute',
+  `i1` tinyint(4) DEFAULT NULL,
+  `i2` smallint(6) DEFAULT NULL,
+  `i3` bigint(20) DEFAULT NULL,
+  `v1` timestamp(6) GENERATED ALWAYS AS (`t` + interval 1 day) VIRTUAL,
+  `v2` timestamp(6) GENERATED ALWAYS AS (`t` + interval 1 month) STORED,
+  `s` timestamp(6) GENERATED ALWAYS AS ROW START,
+  `e` timestamp(6) GENERATED ALWAYS AS ROW END,
+  `ps` date NOT NULL,
+  `pe` date NOT NULL,
+  PERIOD FOR `app_time` (`ps`, `pe`),
+  PRIMARY KEY (`x`,`e`),
+  UNIQUE KEY `u` (`u`,`e`),
+  KEY `three` (`i1`,`i2`,`i3`),
+  PERIOD FOR SYSTEM_TIME (`s`, `e`),
+  CONSTRAINT `check_constr` CHECK (`u` > -1)
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=ucs2 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME INTERVAL 1 HOUR STARTS TIMESTAMP'2000-01-01 00:00:00' AUTO
+(PARTITION `p2` HISTORY ENGINE = DEFAULT_ENGINE,
+ PARTITION `pn` CURRENT ENGINE = DEFAULT_ENGINE)
+affected rows: 1
+insert into t1 (x, c, u, i1, i2, i3, ps, pe)
+values (1, 'cc', 0, 1, 2, 3, '1999-01-01', '2000-01-01');
+affected rows: 1
+set timestamp= unix_timestamp('2000-01-01 01:00:00');
+affected rows: 0
+update t1 set x= x + 8;
+affected rows: 1
+info: Rows matched: 1  Changed: 1  Inserted: 1  Warnings: 0
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) NOT NULL AUTO_INCREMENT,
+  `t` timestamp(6) NOT NULL DEFAULT '2001-11-11 11:11:11.000000',
+  `b` blob /*!100301 COMPRESSED*/ DEFAULT NULL,
+  `c` varchar(1033) CHARACTER SET utf8mb3 NOT NULL,
+  `u` int(11) DEFAULT NULL,
+  `m` enum('a','b','c') NOT NULL DEFAULT 'a' COMMENT 'absolute',
+  `i1` tinyint(4) DEFAULT NULL,
+  `i2` smallint(6) DEFAULT NULL,
+  `i3` bigint(20) DEFAULT NULL,
+  `v1` timestamp(6) GENERATED ALWAYS AS (`t` + interval 1 day) VIRTUAL,
+  `v2` timestamp(6) GENERATED ALWAYS AS (`t` + interval 1 month) STORED,
+  `s` timestamp(6) GENERATED ALWAYS AS ROW START,
+  `e` timestamp(6) GENERATED ALWAYS AS ROW END,
+  `ps` date NOT NULL,
+  `pe` date NOT NULL,
+  PERIOD FOR `app_time` (`ps`, `pe`),
+  PRIMARY KEY (`x`,`e`),
+  UNIQUE KEY `u` (`u`,`e`),
+  KEY `three` (`i1`,`i2`,`i3`),
+  PERIOD FOR SYSTEM_TIME (`s`, `e`),
+  CONSTRAINT `check_constr` CHECK (`u` > -1)
+) ENGINE=DEFAULT_ENGINE AUTO_INCREMENT=10 DEFAULT CHARSET=ucs2 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME INTERVAL 1 HOUR STARTS TIMESTAMP'2000-01-01 00:00:00' AUTO
+(PARTITION `p2` HISTORY ENGINE = DEFAULT_ENGINE,
+ PARTITION `p1` HISTORY ENGINE = DEFAULT_ENGINE,
+ PARTITION `pn` CURRENT ENGINE = DEFAULT_ENGINE)
+affected rows: 1
+set timestamp= unix_timestamp('2000-01-01 02:00:00');
+affected rows: 0
+update t1 set x= x - 8;
+affected rows: 1
+info: Rows matched: 1  Changed: 1  Inserted: 1  Warnings: 0
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) NOT NULL AUTO_INCREMENT,
+  `t` timestamp(6) NOT NULL DEFAULT '2001-11-11 11:11:11.000000',
+  `b` blob /*!100301 COMPRESSED*/ DEFAULT NULL,
+  `c` varchar(1033) CHARACTER SET utf8mb3 NOT NULL,
+  `u` int(11) DEFAULT NULL,
+  `m` enum('a','b','c') NOT NULL DEFAULT 'a' COMMENT 'absolute',
+  `i1` tinyint(4) DEFAULT NULL,
+  `i2` smallint(6) DEFAULT NULL,
+  `i3` bigint(20) DEFAULT NULL,
+  `v1` timestamp(6) GENERATED ALWAYS AS (`t` + interval 1 day) VIRTUAL,
+  `v2` timestamp(6) GENERATED ALWAYS AS (`t` + interval 1 month) STORED,
+  `s` timestamp(6) GENERATED ALWAYS AS ROW START,
+  `e` timestamp(6) GENERATED ALWAYS AS ROW END,
+  `ps` date NOT NULL,
+  `pe` date NOT NULL,
+  PERIOD FOR `app_time` (`ps`, `pe`),
+  PRIMARY KEY (`x`,`e`),
+  UNIQUE KEY `u` (`u`,`e`),
+  KEY `three` (`i1`,`i2`,`i3`),
+  PERIOD FOR SYSTEM_TIME (`s`, `e`),
+  CONSTRAINT `check_constr` CHECK (`u` > -1)
+) ENGINE=DEFAULT_ENGINE AUTO_INCREMENT=10 DEFAULT CHARSET=ucs2 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME INTERVAL 1 HOUR STARTS TIMESTAMP'2000-01-01 00:00:00' AUTO
+(PARTITION `p2` HISTORY ENGINE = DEFAULT_ENGINE,
+ PARTITION `p1` HISTORY ENGINE = DEFAULT_ENGINE,
+ PARTITION `p3` HISTORY ENGINE = DEFAULT_ENGINE,
+ PARTITION `pn` CURRENT ENGINE = DEFAULT_ENGINE)
+affected rows: 1
+# INSERT .. ON DUPLICATE KEY UPDATE (ODKU)
+set timestamp= unix_timestamp('2000-01-01 00:00:00');
+create or replace table t1 (x int primary key) with system versioning
+partition by system_time interval 1 hour auto;
+insert into t1 values (1);
+set timestamp= unix_timestamp('2000-01-01 01:00:00');
+insert into t1 values (1) on duplicate key update x= x + 1;
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) NOT NULL,
+  PRIMARY KEY (`x`)
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME INTERVAL 1 HOUR STARTS TIMESTAMP'2000-01-01 00:00:00' AUTO
+PARTITIONS 3
+# LOAD DATA .. REPLACE
+set timestamp= unix_timestamp('2000-01-01 00:00:00');
+create or replace table t1 (x int primary key) with system versioning
+partition by system_time interval 1 hour auto;
+insert t1 values (1), (2), (3);
+select x into outfile 'MDEV-17554.data' from t1;
+set timestamp= unix_timestamp('2000-01-01 01:00:00');
+load data infile 'MDEV-17554.data' replace into table t1 (x);
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) NOT NULL,
+  PRIMARY KEY (`x`)
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME INTERVAL 1 HOUR STARTS TIMESTAMP'2000-01-01 00:00:00' AUTO
+PARTITIONS 3
+# Concurrent DML
+set timestamp= unix_timestamp('2000-01-01 00:00:00');
+create or replace table t1 (x int) with system versioning
+partition by system_time interval 1 hour auto partitions 3;
+insert into t1 values (1);
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME INTERVAL 1 HOUR STARTS TIMESTAMP'2000-01-01 00:00:00' AUTO
+PARTITIONS 3
+connect con8, localhost, root;
+connect con7, localhost, root;
+connect con6, localhost, root;
+connect con5, localhost, root;
+connect con4, localhost, root;
+connect con3, localhost, root;
+connect con2, localhost, root;
+connect con1, localhost, root;
+set timestamp= unix_timestamp('2000-01-01 02:00:00');
+update t1 set x= x + 10;
+connection con2;
+set timestamp= unix_timestamp('2000-01-01 02:00:00');
+update t1 set x= x + 20;
+connection con3;
+set timestamp= unix_timestamp('2000-01-01 02:00:00');
+update t1 set x= x + 30;
+connection con4;
+set timestamp= unix_timestamp('2000-01-01 02:00:00');
+update t1 set x= x + 40;
+connection con5;
+set timestamp= unix_timestamp('2000-01-01 02:00:00');
+update t1 set x= x + 50;
+connection con6;
+set timestamp= unix_timestamp('2000-01-01 02:00:00');
+update t1 set x= x + 60;
+connection con7;
+set timestamp= unix_timestamp('2000-01-01 02:00:00');
+update t1 set x= x + 70;
+connection con8;
+set timestamp= unix_timestamp('2000-01-01 02:00:00');
+update t1 set x= x + 80;
+connection con1;
+disconnect con1;
+connection con2;
+disconnect con2;
+connection con3;
+disconnect con3;
+connection con4;
+disconnect con4;
+connection con5;
+disconnect con5;
+connection con6;
+disconnect con6;
+connection con7;
+disconnect con7;
+disconnect con8;
+connection default;
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME INTERVAL 1 HOUR STARTS TIMESTAMP'2000-01-01 00:00:00' AUTO
+PARTITIONS 4
+drop tables t1;
+set timestamp= default;
+# Concurrent DML (LIMIT)
+create or replace table t1 (x int) with system versioning engine heap
+partition by system_time limit 1 auto partitions 3;
+insert into t1 values (1);
+update t1 set x= x + N; # (running multithreaded for 3 times)
+drop tables t1;
+# Transaction
+set timestamp= unix_timestamp('2000-01-01 00:00:00');
+create or replace table t1 (x int) with system versioning engine innodb
+partition by system_time interval 1 hour auto;
+insert into t1 values (1);
+set timestamp= unix_timestamp('2000-01-01 01:00:00');
+start transaction;
+update t1 set x= 0;
+connect con1, localhost, root;
+select * from t1;
+x
+1
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) DEFAULT NULL
+) ENGINE=InnoDB DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME INTERVAL 1 HOUR STARTS TIMESTAMP'2000-01-01 00:00:00' AUTO
+PARTITIONS 3
+connection default;
+commit;
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) DEFAULT NULL
+) ENGINE=InnoDB DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME INTERVAL 1 HOUR STARTS TIMESTAMP'2000-01-01 00:00:00' AUTO
+PARTITIONS 3
+set timestamp= unix_timestamp('2000-01-01 02:00:00');
+start transaction;
+update t1 set x= 1;
+connection con1;
+select * from t1;
+x
+0
+connection default;
+rollback;
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) DEFAULT NULL
+) ENGINE=InnoDB DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME INTERVAL 1 HOUR STARTS TIMESTAMP'2000-01-01 00:00:00' AUTO
+PARTITIONS 4
+disconnect con1;
+connection default;
+drop table t1;
+#
+# MDEV-25479 Auto-create: 2nd and further executions of PS or SP fail to create partition
+#
+create table t (a int) with system versioning
+partition by system_time interval 1 hour auto;
+insert into t values (1), (2);
+prepare stmt from "update t set a= a + 1";
+set @@timestamp= @@timestamp + 3601;
+execute stmt;
+set @@timestamp= @@timestamp + 3601;
+execute stmt;
+drop prepare stmt;
+show create table t;
+Table	Create Table
+t	CREATE TABLE `t` (
+  `a` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME INTERVAL 1 HOUR STARTS TIMESTAMP'2000-01-01 02:00:00' AUTO
+PARTITIONS 4
+create procedure sp() update t set a= a + 1;
+set @@timestamp= @@timestamp + 3601;
+call sp();
+set @@timestamp= @@timestamp + 3601;
+call sp();
+drop procedure sp;
+show create table t;
+Table	Create Table
+t	CREATE TABLE `t` (
+  `a` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME INTERVAL 1 HOUR STARTS TIMESTAMP'2000-01-01 02:00:00' AUTO
+PARTITIONS 6
+drop table t;
+#
+# MDEV-23639 Auto-create does not work under LOCK TABLES or inside triggers
+#
+set timestamp= unix_timestamp('2000-01-01 00:00:00');
+create or replace table t1 (x int) with system versioning
+partition by system_time interval 1 hour auto
+partitions 3;
+create table t2 (x int);
+create table t3 (x int);
+insert into t3 values (3);
+create trigger tr after insert on t2 for each row update t1 set x= x + 11;
+create or replace procedure sp() update t1 set x= x + 5;
+create or replace procedure sp2() insert into t2 values (5);
+prepare ps from 'update t1 set x= x + 6';
+prepare ps2 from 'insert into t2 values (6)';
+insert into t1 values (1);
+set timestamp= unix_timestamp('2000-01-01 02:00:00');
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME INTERVAL 1 HOUR STARTS TIMESTAMP'2000-01-01 00:00:00' AUTO
+PARTITIONS 3
+insert into t2 values (2);
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME INTERVAL 1 HOUR STARTS TIMESTAMP'2000-01-01 00:00:00' AUTO
+PARTITIONS 4
+set timestamp= unix_timestamp('2000-01-01 03:00:00');
+call sp;
+call sp;
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME INTERVAL 1 HOUR STARTS TIMESTAMP'2000-01-01 00:00:00' AUTO
+PARTITIONS 5
+set timestamp= unix_timestamp('2000-01-01 04:00:00');
+call sp2;
+call sp2;
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME INTERVAL 1 HOUR STARTS TIMESTAMP'2000-01-01 00:00:00' AUTO
+PARTITIONS 6
+set timestamp= unix_timestamp('2000-01-01 05:00:00');
+execute ps;
+execute ps;
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME INTERVAL 1 HOUR STARTS TIMESTAMP'2000-01-01 00:00:00' AUTO
+PARTITIONS 7
+set timestamp= unix_timestamp('2000-01-01 06:00:00');
+execute ps2;
+execute ps2;
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME INTERVAL 1 HOUR STARTS TIMESTAMP'2000-01-01 00:00:00' AUTO
+PARTITIONS 8
+set timestamp= unix_timestamp('2000-01-01 08:00:00');
+lock tables t1 write, t2 write;
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME INTERVAL 1 HOUR STARTS TIMESTAMP'2000-01-01 00:00:00' AUTO
+PARTITIONS 10
+set timestamp= unix_timestamp('2000-01-01 09:00:00');
+update t1 set x= x + 1;
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME INTERVAL 1 HOUR STARTS TIMESTAMP'2000-01-01 00:00:00' AUTO
+PARTITIONS 11
+set timestamp= unix_timestamp('2000-01-01 10:00:00');
+update t1 set x= x + 2;
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME INTERVAL 1 HOUR STARTS TIMESTAMP'2000-01-01 00:00:00' AUTO
+PARTITIONS 12
+update t2 set x= x + 1;
+set timestamp= unix_timestamp('2000-01-01 11:00:00');
+insert into t2 values (4);
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME INTERVAL 1 HOUR STARTS TIMESTAMP'2000-01-01 00:00:00' AUTO
+PARTITIONS 13
+update t3 set x= x + 1;
+ERROR HY000: Table 't3' was not locked with LOCK TABLES
+set timestamp= unix_timestamp('2000-01-01 12:00:00');
+call sp;
+call sp;
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME INTERVAL 1 HOUR STARTS TIMESTAMP'2000-01-01 00:00:00' AUTO
+PARTITIONS 14
+set timestamp= unix_timestamp('2000-01-01 13:00:00');
+call sp2;
+call sp2;
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME INTERVAL 1 HOUR STARTS TIMESTAMP'2000-01-01 00:00:00' AUTO
+PARTITIONS 15
+set timestamp= unix_timestamp('2000-01-01 14:00:00');
+execute ps;
+execute ps;
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME INTERVAL 1 HOUR STARTS TIMESTAMP'2000-01-01 00:00:00' AUTO
+PARTITIONS 16
+set timestamp= unix_timestamp('2000-01-01 15:00:00');
+execute ps2;
+execute ps2;
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME INTERVAL 1 HOUR STARTS TIMESTAMP'2000-01-01 00:00:00' AUTO
+PARTITIONS 17
+unlock tables;
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) DEFAULT NULL
+) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME INTERVAL 1 HOUR STARTS TIMESTAMP'2000-01-01 00:00:00' AUTO
+PARTITIONS 17
+drop tables t1, t2, t3;
+drop procedure sp;
+drop procedure sp2;
+drop prepare ps;
+drop prepare ps2;
+#
+# MDEV-27456 Assertion `!thd->is_error()' fails in vers_create_partitions upon DML with ER_UNKNOWN_PARTITION
+#
+create table t (a int) with system versioning
+partition by system_time interval 1 minute auto;
+set @@timestamp= @@timestamp + 61;
+select * from t;
+a
+delete from t partition (px);
+ERROR HY000: Unknown partition 'px' in table 't'
+lock tables t write;
+delete from t partition (px);
+ERROR HY000: Unknown partition 'px' in table 't'
+unlock tables;
+drop table t;
+SET GLOBAL innodb_stats_persistent=@save_persistent;
diff --git a/mysql-test/suite/versioning/r/rpl.result b/mysql-test/suite/versioning/r/rpl.result
index 627f3991499..4f0b0fc07ff 100644
--- a/mysql-test/suite/versioning/r/rpl.result
+++ b/mysql-test/suite/versioning/r/rpl.result
@@ -164,4 +164,32 @@ update t1 set i = 0;
 connection slave;
 connection master;
 drop table t1;
+#
+# MDEV-17554 Auto-create new partition for system versioned tables
+#            with history partitioned by INTERVAL/LIMIT
+#
+set timestamp= unix_timestamp('2000-01-01 00:00:00');
+create or replace table t1 (x int) with system versioning
+partition by system_time interval 1 hour auto;
+insert t1 values ();
+set timestamp= unix_timestamp('2000-01-01 01:00:00');
+delete from t1;
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) DEFAULT NULL
+) ENGINE=ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME INTERVAL 1 HOUR STARTS TIMESTAMP'2000-01-01 00:00:00' AUTO
+PARTITIONS 3
+connection slave;
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `x` int(11) DEFAULT NULL
+) ENGINE=ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME INTERVAL 1 HOUR STARTS TIMESTAMP'2000-01-01 00:00:00' AUTO
+PARTITIONS 3
+connection master;
+drop table t1;
+set timestamp= default;
 include/rpl_end.inc
diff --git a/mysql-test/suite/versioning/r/rpl_mix.result b/mysql-test/suite/versioning/r/rpl_mix.result
index 68eded0faf7..56af76b8fc2 100644
--- a/mysql-test/suite/versioning/r/rpl_mix.result
+++ b/mysql-test/suite/versioning/r/rpl_mix.result
@@ -8,4 +8,59 @@ DELETE HISTORY FROM t1;
 connection slave;
 connection master;
 drop table t1;
+#
+# MDEV-25347 DML events for auto-partitioned tables are written into binary log twice
+#
+flush binary logs;
+create table t1 (a int) with system versioning
+partition by system_time limit 1 auto;
+insert into t1 values (1);
+update t1 set a= a + 1;
+update t1 set a= a + 2;
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `a` int(11) DEFAULT NULL
+) ENGINE=ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME LIMIT 1 AUTO
+PARTITIONS 3
+select * from t1;
+a
+4
+include/show_binlog_events.inc
+Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+master-bin.000002	#	Binlog_checkpoint	#	#	master-bin.000002
+master-bin.000002	#	Gtid	#	#	GTID #-#-#
+master-bin.000002	#	Query	#	#	use `test`; create table t1 (a int) with system versioning
+partition by system_time limit 1 auto
+master-bin.000002	#	Gtid	#	#	BEGIN GTID #-#-#
+master-bin.000002	#	Annotate_rows	#	#	insert into t1 values (1)
+master-bin.000002	#	Table_map	#	#	table_id: # (test.t1)
+master-bin.000002	#	Write_rows_v1	#	#	table_id: # flags: STMT_END_F
+master-bin.000002	#	Query	#	#	COMMIT
+master-bin.000002	#	Gtid	#	#	BEGIN GTID #-#-#
+master-bin.000002	#	Annotate_rows	#	#	update t1 set a= a + 1
+master-bin.000002	#	Table_map	#	#	table_id: # (test.t1)
+master-bin.000002	#	Update_rows_v1	#	#	table_id: #
+master-bin.000002	#	Write_rows_v1	#	#	table_id: # flags: STMT_END_F
+master-bin.000002	#	Query	#	#	COMMIT
+master-bin.000002	#	Gtid	#	#	BEGIN GTID #-#-#
+master-bin.000002	#	Annotate_rows	#	#	update t1 set a= a + 2
+master-bin.000002	#	Table_map	#	#	table_id: # (test.t1)
+master-bin.000002	#	Update_rows_v1	#	#	table_id: #
+master-bin.000002	#	Write_rows_v1	#	#	table_id: # flags: STMT_END_F
+master-bin.000002	#	Query	#	#	COMMIT
+connection slave;
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `a` int(11) DEFAULT NULL
+) ENGINE=ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME LIMIT 1 AUTO
+PARTITIONS 3
+select * from t1;
+a
+4
+connection master;
+drop table t1;
 include/rpl_end.inc
diff --git a/mysql-test/suite/versioning/r/rpl_row.result b/mysql-test/suite/versioning/r/rpl_row.result
index 03ac8dc9eb8..124993423e7 100644
--- a/mysql-test/suite/versioning/r/rpl_row.result
+++ b/mysql-test/suite/versioning/r/rpl_row.result
@@ -11,4 +11,59 @@ connection slave;
 connection master;
 drop table t1;
 set binlog_row_image= @old_row_image;
+#
+# MDEV-25347 DML events for auto-partitioned tables are written into binary log twice
+#
+flush binary logs;
+create table t1 (a int) with system versioning
+partition by system_time limit 1 auto;
+insert into t1 values (1);
+update t1 set a= a + 1;
+update t1 set a= a + 2;
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `a` int(11) DEFAULT NULL
+) ENGINE=ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME LIMIT 1 AUTO
+PARTITIONS 3
+select * from t1;
+a
+4
+include/show_binlog_events.inc
+Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+master-bin.000002	#	Binlog_checkpoint	#	#	master-bin.000002
+master-bin.000002	#	Gtid	#	#	GTID #-#-#
+master-bin.000002	#	Query	#	#	use `test`; create table t1 (a int) with system versioning
+partition by system_time limit 1 auto
+master-bin.000002	#	Gtid	#	#	BEGIN GTID #-#-#
+master-bin.000002	#	Annotate_rows	#	#	insert into t1 values (1)
+master-bin.000002	#	Table_map	#	#	table_id: # (test.t1)
+master-bin.000002	#	Write_rows_v1	#	#	table_id: # flags: STMT_END_F
+master-bin.000002	#	Query	#	#	COMMIT
+master-bin.000002	#	Gtid	#	#	BEGIN GTID #-#-#
+master-bin.000002	#	Annotate_rows	#	#	update t1 set a= a + 1
+master-bin.000002	#	Table_map	#	#	table_id: # (test.t1)
+master-bin.000002	#	Update_rows_v1	#	#	table_id: #
+master-bin.000002	#	Write_rows_v1	#	#	table_id: # flags: STMT_END_F
+master-bin.000002	#	Query	#	#	COMMIT
+master-bin.000002	#	Gtid	#	#	BEGIN GTID #-#-#
+master-bin.000002	#	Annotate_rows	#	#	update t1 set a= a + 2
+master-bin.000002	#	Table_map	#	#	table_id: # (test.t1)
+master-bin.000002	#	Update_rows_v1	#	#	table_id: #
+master-bin.000002	#	Write_rows_v1	#	#	table_id: # flags: STMT_END_F
+master-bin.000002	#	Query	#	#	COMMIT
+connection slave;
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `a` int(11) DEFAULT NULL
+) ENGINE=ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME LIMIT 1 AUTO
+PARTITIONS 3
+select * from t1;
+a
+4
+connection master;
+drop table t1;
 include/rpl_end.inc
diff --git a/mysql-test/suite/versioning/r/rpl_stmt.result b/mysql-test/suite/versioning/r/rpl_stmt.result
new file mode 100644
index 00000000000..1a5087dfa74
--- /dev/null
+++ b/mysql-test/suite/versioning/r/rpl_stmt.result
@@ -0,0 +1,50 @@
+include/master-slave.inc
+[connection master]
+#
+# MDEV-25347 DML events for auto-partitioned tables are written into binary log twice
+#
+flush binary logs;
+create table t1 (a int) with system versioning
+partition by system_time limit 1 auto;
+insert into t1 values (1);
+update t1 set a= a + 1;
+update t1 set a= a + 2;
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `a` int(11) DEFAULT NULL
+) ENGINE=ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME LIMIT 1 AUTO
+PARTITIONS 3
+select * from t1;
+a
+4
+include/show_binlog_events.inc
+Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+master-bin.000002	#	Binlog_checkpoint	#	#	master-bin.000002
+master-bin.000002	#	Gtid	#	#	GTID #-#-#
+master-bin.000002	#	Query	#	#	use `test`; create table t1 (a int) with system versioning
+partition by system_time limit 1 auto
+master-bin.000002	#	Gtid	#	#	BEGIN GTID #-#-#
+master-bin.000002	#	Query	#	#	use `test`; insert into t1 values (1)
+master-bin.000002	#	Query	#	#	COMMIT
+master-bin.000002	#	Gtid	#	#	BEGIN GTID #-#-#
+master-bin.000002	#	Query	#	#	use `test`; update t1 set a= a + 1
+master-bin.000002	#	Query	#	#	COMMIT
+master-bin.000002	#	Gtid	#	#	BEGIN GTID #-#-#
+master-bin.000002	#	Query	#	#	use `test`; update t1 set a= a + 2
+master-bin.000002	#	Query	#	#	COMMIT
+connection slave;
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `a` int(11) DEFAULT NULL
+) ENGINE=ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+ PARTITION BY SYSTEM_TIME LIMIT 1 AUTO
+PARTITIONS 3
+select * from t1;
+a
+4
+connection master;
+drop table t1;
+include/rpl_end.inc
diff --git a/mysql-test/suite/versioning/r/update-big.result b/mysql-test/suite/versioning/r/update-big.result
index 89297fe5d89..506d1d23c5d 100644
--- a/mysql-test/suite/versioning/r/update-big.result
+++ b/mysql-test/suite/versioning/r/update-big.result
@@ -22,4 +22,78 @@ update t1 set a= 1 where a = 5;
 update t1 set a= 8 where a = 1;
 update t1 set a= 4 where a = 8;
 update t1 set a= 6;
+disconnect con1;
 drop table t1;
+call mtr.add_suppression("need more HISTORY partitions");
+#
+# MDEV-23642 Locking timeout caused by auto-creation affects original DML
+#
+set timestamp= unix_timestamp('2000-01-01 00:00:00');
+create or replace table t1 (x int primary key) with system versioning engine innodb
+partition by system_time interval 1 hour auto;
+insert into t1 values (1);
+start transaction;
+insert into t1 values (2);
+connect con1, localhost, root;
+set lock_wait_timeout= 1;
+set innodb_lock_wait_timeout= 1;
+set timestamp= unix_timestamp('2000-01-01 01:00:01');
+update t1 set x= x + 122 where x = 1;
+Warnings:
+Error	1205	Lock wait timeout exceeded; try restarting transaction
+Warning	4114	Versioned table `test`.`t1`: last HISTORY partition (`p0`) is out of INTERVAL, need more HISTORY partitions
+disconnect con1;
+connection default;
+select * from t1;
+x
+2
+123
+drop table t1;
+set timestamp= default;
+#
+# MDEV-25339 Assertion `thd->transaction.stmt.is_empty() || thd->in_sub_stmt' failed
+#
+create or replace table t1 (x int) with system versioning engine innodb
+partition by system_time interval 1 hour auto;
+start transaction;
+insert into t1 values (1);
+select * from t1;
+x
+1
+connect con1, localhost, root;
+set lock_wait_timeout= 1;
+set innodb_lock_wait_timeout= 1;
+update t1 set x= x + 111;
+ERROR HY000: Lock wait timeout exceeded; try restarting transaction
+select * from t1;
+x
+disconnect con1;
+connection default;
+drop table t1;
+#
+# MDEV-25482 Auto-create: Server hangs after a failed attempt to create partition
+#
+set timestamp= default;
+create table t (pk int primary key, a int) engine=InnoDB
+with system versioning partition by system_time interval 1 hour auto;
+insert into t values (1,1),(2,2),(3,3),(4,4),(5,5);
+start transaction;
+update t set a= 20 where pk = 2;
+connect  con1,localhost,root,,;
+set lock_wait_timeout= 1;
+set @@timestamp= @@timestamp+3601;
+update t set a= 40 where pk = 4;
+Warnings:
+Error	1205	Lock wait timeout exceeded; try restarting transaction
+Warning	4114	Versioned table `test`.`t`: last HISTORY partition (`p0`) is out of INTERVAL, need more HISTORY partitions
+update t set a= 400 where pk = 4;
+Warnings:
+Error	1205	Lock wait timeout exceeded; try restarting transaction
+Warning	4114	Versioned table `test`.`t`: last HISTORY partition (`p0`) is out of INTERVAL, need more HISTORY partitions
+disconnect con1;
+connection default;
+select * from t where pk = 4;
+pk	a
+4	400
+rollback;
+drop tables t;
diff --git a/mysql-test/suite/versioning/t/debug.test b/mysql-test/suite/versioning/t/debug.test
index be58b67546d..94917f01389 100644
--- a/mysql-test/suite/versioning/t/debug.test
+++ b/mysql-test/suite/versioning/t/debug.test
@@ -1,4 +1,5 @@
 --source include/have_debug.inc
+--source include/have_partition.inc
 
 --let $datadir=`select @@datadir`
 
@@ -33,7 +34,8 @@ show create table t3;
 create table t4 (a int);
 show create table t4;
 drop table t1, t2, t3, t4;
-
+set debug_dbug= @old_dbug;
+set global debug_dbug= @old_dbug;
 
 --echo #
 --echo # MDEV-19525 remove ER_VERS_FIELD_WRONG_TYPE from init_from_binary_frm_image()
@@ -46,4 +48,61 @@ show create table t1;
 --replace_result $datadir ./
 show warnings;
 drop table t1;
-set global debug_dbug=@old_dbug;
+set debug_dbug= @old_dbug;
+
+
+--echo #
+--echo # MDEV-17554 Auto-create new partition for system versioned tables
+--echo #            with history partitioned by INTERVAL/LIMIT
+--echo #
+call mtr.add_suppression("need more HISTORY partitions");
+create or replace table t1 (x int) with system versioning
+partition by system_time limit 1 auto partitions 2;
+insert into t1 values (1);
+update t1 set x= x + 1;
+
+--connect con2, localhost, root
+--connect con1, localhost, root
+
+--disable_warnings
+--echo # Both threads create partition simultaneously
+--connection con1
+set debug_sync= 'add_history_partition signal s1 wait_for s2';
+send update t1 set x= x + 10;
+--connection con2
+set debug_sync= 'now wait_for s1';
+flush tables t1;
+set debug_sync= 'add_history_partition signal s2';
+update t1 set x= x + 20;
+--connection con1
+reap;
+--connection default
+# 1 or 2 history rows may be created depending on which UPDATE finishes first (MDEV-28459)
+# select partition_name, table_rows from information_schema.partitions
+# where table_name = 't1';
+
+# Fill empty partition for next UPDATE to trigger auto-create
+update t1 set x= x + 2;
+
+--echo # Second thread skips to reopen 3 times until first thread creates partition
+--connection con1
+set debug_sync= 'add_history_partition SIGNAL s1 WAIT_FOR s2';
+send update t1 set x= x + 30;
+--connection con2
+set debug_sync= 'now WAIT_FOR s1';
+set debug_sync= 'reopen_history_partition SIGNAL s2 EXECUTE 3';
+update t1 set x= x + 40;
+--connection con1
+reap;
+--connection default
+# Same here (MDEV-28459)
+# select partition_name, table_rows from information_schema.partitions
+# where table_name = 't1';
+--enable_warnings
+
+--disconnect con1
+--disconnect con2
+set @@timestamp= default;
+
+drop tables t1;
+set debug_sync= 'reset';
diff --git a/mysql-test/suite/versioning/t/delete_history.test b/mysql-test/suite/versioning/t/delete_history.test
index 624f57cd2b5..b9d957f461c 100644
--- a/mysql-test/suite/versioning/t/delete_history.test
+++ b/mysql-test/suite/versioning/t/delete_history.test
@@ -204,4 +204,18 @@ DELETE HISTORY FROM v1;
 DROP VIEW v1;
 DROP TABLE t1;
 
+--echo #
+--echo # MDEV-17554 Auto-create new partition for system versioned tables with history partitioned by INTERVAL/LIMIT
+--echo #
+--echo # Don't auto-create new partition on DELETE HISTORY:
+set timestamp= unix_timestamp('2000-01-01 00:00:00');
+create or replace table t (a int) with system versioning
+partition by system_time interval 1 hour auto;
+set timestamp= unix_timestamp('2000-01-01 10:00:00');
+delete history from t;
+set timestamp= default;
+--replace_result $default_engine DEFAULT_ENGINE
+show create table t;
+drop table t;
+
 --source suite/versioning/common_finish.inc
diff --git a/mysql-test/suite/versioning/t/load_data.test b/mysql-test/suite/versioning/t/load_data.test
index 3bac2942a81..d4a1fa1a736 100644
--- a/mysql-test/suite/versioning/t/load_data.test
+++ b/mysql-test/suite/versioning/t/load_data.test
@@ -1,6 +1,7 @@
 #
 # MDEV-15330 Server crash or assertion `table->insert_values' failure in write_record upon LOAD DATA
 #
+--let $datadir= `select @@datadir`
 CREATE TABLE t1 (a INT, b INT, c INT, vc INT AS (c), UNIQUE(a), UNIQUE(b)) WITH SYSTEM VERSIONING;
 INSERT IGNORE INTO t1 (a,b,c) VALUES (1,2,3);
 
@@ -12,3 +13,4 @@ LOAD DATA INFILE '15330.data' REPLACE INTO TABLE t1 (a,b,c);
 
 # Cleanup
 DROP TABLE t1;
+--remove_file $datadir/test/15330.data
diff --git a/mysql-test/suite/versioning/t/not_embedded.test b/mysql-test/suite/versioning/t/not_embedded.test
index 2afae013e70..80e936b02f3 100644
--- a/mysql-test/suite/versioning/t/not_embedded.test
+++ b/mysql-test/suite/versioning/t/not_embedded.test
@@ -1,5 +1,6 @@
 --source include/not_embedded.inc
 --source include/have_innodb.inc
+--source include/have_partition.inc
 
 --echo # 
 --echo # SYSTEM_VERSIONING_ASOF sysvar
@@ -77,3 +78,30 @@ drop database mysqltest;
 --disconnect user1
 --disconnect root
 --connection default
+
+--echo #
+--echo # MDEV-25559 Auto-create: infinite loop after interrupted lock wait
+--echo #
+
+set timestamp= unix_timestamp('2000-01-01 00:00:00');
+create table t (pk int primary key, a int) engine innodb with system versioning
+partition by system_time interval 1 hour auto;
+insert into t values (1, 0);
+begin;
+update t set a= a + 1;
+--connect (con1,localhost,root,,)
+set max_statement_time= 1;
+set timestamp= unix_timestamp('2000-01-01 01:00:00');
+send update t set a= a + 2;
+--connection default
+set timestamp= unix_timestamp('2000-01-01 01:00:00');
+send update t set a= a + 3;
+--connection con1
+--error ER_STATEMENT_TIMEOUT
+reap;
+--disconnect con1
+--connection default
+reap;
+commit;
+drop table t;
+set timestamp= default;
diff --git a/mysql-test/suite/versioning/t/partition.test b/mysql-test/suite/versioning/t/partition.test
index 5e209bb7969..de7536f0e6c 100644
--- a/mysql-test/suite/versioning/t/partition.test
+++ b/mysql-test/suite/versioning/t/partition.test
@@ -10,6 +10,7 @@ call mtr.add_suppression("need more HISTORY partitions");
 --enable_prepare_warnings
 
 set system_versioning_alter_history=keep;
+--let $datadir= `select @@datadir`
 --echo # Check conventional partitioning on temporal tables
 
 --replace_result $sys_datatype_expl SYS_DATATYPE
@@ -218,7 +219,9 @@ prepare stmt from @str; execute stmt; drop prepare stmt;
 select @ts0 = @ts1;
 select @ts2 = @ts3;
 
---echo ## rotation by LIMIT
+--echo #
+--echo # Rotation by LIMIT
+--echo #
 --error ER_PART_WRONG_VALUE
 create or replace table t1 (x int)
 with system versioning
@@ -246,8 +249,9 @@ insert into t1 values (7), (8);
 delete from t1;
 select * from t1 partition (p1) order by x;
 
-
---echo ## rotation by INTERVAL
+--echo #
+--echo # Rotation by INTERVAL
+--echo #
 --error ER_PART_WRONG_VALUE
 create or replace table t1 (x int)
 with system versioning
@@ -291,7 +295,8 @@ insert t1 values (1); delete from t1;
 set timestamp=unix_timestamp('2001-02-04 10:20:50');
 insert t1 values (2); delete from t1;
 
-select subpartition_name,partition_description,table_rows from information_schema.partitions where table_schema='test' and table_name='t1';
+select subpartition_name, partition_description, table_rows from information_schema.partitions where table_schema='test' and table_name='t1';
+select * from t1 partition (p1);
 
 set timestamp=unix_timestamp('2001-02-04 10:20:55');
 alter table t1 add partition (partition p0 history, partition p2 history);
@@ -300,7 +305,15 @@ insert t1 values (4),(5);
 set timestamp=unix_timestamp('2001-02-04 10:30:10');
 update t1 set i=6 where i=5;
 
-select subpartition_name,partition_description,table_rows from information_schema.partitions where table_schema='test' and table_name='t1';
+select subpartition_name, partition_description, table_rows from information_schema.partitions where table_schema='test' and table_name='t1';
+select * from t1 partition (p1);
+select * from t1 partition (p0);
+select * from t1 partition (p2);
+
+alter table t1 rebuild partition p0, p1, p2;
+select * from t1 partition (p1);
+select * from t1 partition (p0);
+select * from t1 partition (p2);
 
 --echo ## pruning check
 set @ts=(select partition_description from information_schema.partitions
@@ -431,7 +444,6 @@ create or replace table t1 (i int) with system versioning
 partition by system_time interval 1 day starts '2000-01-01 00:00:00'
 partitions 3;
 
---echo # we are warned when we push to present:
 insert into t1 values (0);
 set timestamp= unix_timestamp('2001-01-01 00:00:01');
 update t1 set i= i + 1;
@@ -460,8 +472,8 @@ update t1 set i= i + 1;
 select *, row_end from t1 partition (p0);
 select *, row_end from t1 partition (p1);
 
-set timestamp= unix_timestamp('2000-01-01 00:00:00');
 --echo # and this is how it usually goes:
+set timestamp= unix_timestamp('2000-01-01 00:00:00');
 create or replace table t1 (i int) with system versioning
 partition by system_time interval 1 day
 partitions 3;
@@ -477,7 +489,6 @@ set timestamp= unix_timestamp('2000-01-04 00:00:01');
 update t1 set i= i + 1;
 
 alter table t1 add partition (partition p2 history, partition p3 history);
-
 select *, row_end from t1 partition (p0);
 select *, row_end from t1 partition (p1);
 select *, row_end from t1 partition (p2);
@@ -496,9 +507,9 @@ insert into t1 (x) values (1), (2), (3), (4), (5);
 select * from t1 partition (pnsp0);
 select * from t1 partition (pnsp1);
 
---echo ### warn about full partition
 delete from t1 where x < 3;
 delete from t1;
+--echo ### warn about full partition
 delete from t1;
 select * from t1 partition (p0sp0);
 select * from t1 partition (p0sp1);
@@ -987,16 +998,15 @@ create or replace table t1 (
 
 insert into t1 () values (),(),(),(),(),(),(),(),(),(),(),(),(),(),(),();
 
-select * into outfile 'load.data' from t1;
-load data infile 'load.data' replace into table t1;
+select * into outfile 'MDEV-17891.data' from t1;
+load data infile 'MDEV-17891.data' replace into table t1;
 --error ER_RECORD_FILE_FULL
-load data infile 'load.data' replace into table t1;
+load data infile 'MDEV-17891.data' replace into table t1;
 --error ER_RECORD_FILE_FULL
-load data infile 'load.data' replace into table t1;
+load data infile 'MDEV-17891.data' replace into table t1;
 
 # Cleanup
---let $datadir= `select @@datadir`
---remove_file $datadir/test/load.data
+--remove_file $datadir/test/MDEV-17891.data
 eval set @@max_heap_table_size= $max_heap_table_size_orig;
 drop table t1;
 
@@ -1072,6 +1082,11 @@ insert t1 values (1);
 delete from t1 partition (p0, pn);
 --error ER_VERS_NOT_ALLOWED
 delete from t1 partition (p0, p1, pn);
+lock tables t1 write;
+--error ER_VERS_NOT_ALLOWED
+delete from t1 partition (p0, pn);
+delete from t1;
+unlock tables;
 drop table t1;
 set timestamp= default;
 --echo # End of 10.3 tests
@@ -1155,8 +1170,6 @@ show create table t1;
 
 drop tables t1;
 
---disable_prepare_warnings
-
 --echo #
 --echo # MDEV-27328 Change of SYSTEM_TIME partitioning options is not possible without data copy
 --echo #
@@ -1227,11 +1240,6 @@ select * from t1 partition (p4);
 select * from t1 partition (pn);
 
 drop table t1;
---echo #
---echo # End of 10.5 tests
---echo #
-
-SET GLOBAL innodb_stats_persistent=@save_persistent;
 
 --echo # End of 10.6 tests
 
@@ -1434,4 +1442,740 @@ show create table t1;
 drop tables t1, tp1;
 }
 
+--echo #
+--echo # MDEV-17554 Auto-create new partition for system versioned tables
+--echo #            with history partitioned by INTERVAL/LIMIT
+--echo #
+create or replace table t1 (x int) with system versioning
+partition by system_time limit 1 auto;
+--replace_result $default_engine DEFAULT_ENGINE
+show create table t1;
+--echo # Turn off AUTO
+alter table t1 partition by system_time limit 1;
+--replace_result $default_engine DEFAULT_ENGINE
+show create table t1;
+--echo # Get AUTO back
+alter table t1 partition by system_time limit 1 auto;
+--replace_result $default_engine DEFAULT_ENGINE
+show create table t1;
+
+insert into t1 values (1);
+
+create or replace table t2 (y int);
+insert into t2 values (2);
+
+insert into t1 select * from t2;
+insert into t2 select * from t1;
+--replace_result $default_engine DEFAULT_ENGINE
+show create table t1;
+
+--echo # Too many partitions error
+set timestamp= unix_timestamp('2000-01-01 00:00:00');
+create or replace table t1 (x int) with system versioning
+partition by system_time interval 1 hour auto;
+set timestamp= unix_timestamp('2001-01-01 00:01:00');
+--error ER_VERS_HIST_PART_FAILED
+update t1 set x= x + 1;
+show warnings;
+
+--echo # Auto-create failed error
+set timestamp= unix_timestamp('2000-01-01 00:00:00');
+create or replace table t1 (x int) with system versioning engine innodb
+partition by system_time interval 1 hour auto;
+
+insert into t1 values (1);
+
+call mtr.add_suppression("rror number .*(File exists|file operation)");
+call mtr.add_suppression("InnoDB: Cannot create file");
+
+--let $datadir= `select @@datadir`
+--let $dummy= $datadir/test/t1#P#p1.ibd
+--write_file $dummy
+EOF
+
+set timestamp= unix_timestamp('2000-01-01 01:00:00');
+--error ER_GET_ERRNO
+update t1 set x= x + 2;
+show warnings;
+--remove_file $dummy
+
+--echo # Partition overflow error and manual fix
+set timestamp= unix_timestamp('2000-01-01 00:00:00');
+create or replace table t1 (x int) with system versioning
+partition by system_time interval 1 hour;
+
+insert into t1 values (440);
+set timestamp= unix_timestamp('2000-01-01 00:10:00');
+update t1 set x= x + 1;
+
+--echo # Check how pruning boundaries work
+--replace_column 5 # 10 # 11 #
+explain partitions select * from t1 for system_time as of '2000-01-01 00:59:58';
+--replace_column 5 # 10 # 11 #
+explain partitions select * from t1 for system_time as of '2000-01-01 00:59:59';
+--replace_column 5 # 10 # 11 #
+explain partitions select * from t1 for system_time as of '2000-01-01 01:00:00';
+select * from t1 for system_time as of '2000-01-01 00:09:59';
+
+set timestamp= unix_timestamp('2000-01-01 02:00:00');
+update t1 set x= x + 1;
+
+select * from t1 for system_time as of '2000-01-01 01:00:00';
+select * from t1 partition (p0) order by x;
+
+--echo # Here is how manual fix works: just add new partitions there
+alter table t1 add partition partitions 3;
+select * from t1 for system_time as of '2000-01-01 01:00:00';
+select * from t1 partition (p0) order by x;
+
+--echo # Check pruning after ALTER
+--replace_column 5 # 10 # 11 #
+explain partitions select * from t1 for system_time as of '2000-01-01 00:59:58';
+--replace_column 5 # 10 # 11 #
+explain partitions select * from t1 for system_time as of '2000-01-01 00:59:59';
+--replace_column 5 # 10 # 11 #
+explain partitions select * from t1 for system_time as of '2000-01-01 01:00:00';
+
+drop table t1;
+
+--enable_info
+create or replace table t1 (x int) with system versioning
+partition by system_time interval 3600 second
+starts '2000-01-01 00:00:00' auto partitions 3;
+
+insert into t1 values (1);
+--replace_result $default_engine DEFAULT_ENGINE
+show create table t1;
+
+set timestamp= unix_timestamp('2000-01-01 02:00:00');
+update t1 set x= x + 1;
+--replace_result $default_engine DEFAULT_ENGINE
+show create table t1;
+
+set timestamp= unix_timestamp('2000-01-01 03:00:00');
+update t1 set x= x + 2;
+--replace_result $default_engine DEFAULT_ENGINE
+show create table t1;
+
+set timestamp= unix_timestamp('2000-01-01 00:00:00');
+create or replace table t1 (x int) with system versioning
+partition by system_time interval 1 hour auto (
+    partition p1 history,
+    partition p3 history,
+    partition pn current);
+
+insert into t1 values (1);
+--replace_result $default_engine DEFAULT_ENGINE
+show create table t1;
+
+set timestamp= unix_timestamp('2000-01-01 02:00:00');
+update t1 set x= x + 3;
+--replace_result $default_engine DEFAULT_ENGINE
+show create table t1;
+
+set timestamp= unix_timestamp('2000-01-01 03:00:00');
+update t1 set x= x + 4;
+--replace_result $default_engine DEFAULT_ENGINE
+show create table t1;
+
+set timestamp= unix_timestamp('2000-01-01 04:00:00');
+lock tables t1 write;
+update t1 set x= x + 5;
+--replace_result $default_engine DEFAULT_ENGINE
+show create table t1;
+unlock tables;
+set timestamp= default;
+
+--echo # Couple of more LOCK TABLES cases
+create or replace table t1 (x int) with system versioning
+partition by system_time limit 1 auto;
+lock tables t1 write;
+insert into t1 values (1);
+update t1 set x= x + 1;
+update t1 set x= x + 1;
+update t1 set x= x + 1;
+--replace_result $default_engine DEFAULT_ENGINE
+show create table t1;
+unlock tables;
+--replace_result $default_engine DEFAULT_ENGINE
+show create table t1;
+
+--echo # Overflow prevention under LOCK TABLES
+create or replace table t1 (x int)
+with system versioning partition by system_time
+limit 10 auto;
+
+insert into t1 values (1), (2), (3), (4), (5), (6), (7), (8), (9);
+update t1 set x= x + 10;
+
+lock tables t1 write;
+update t1 set x= 1 where x = 11;
+update t1 set x= 2 where x = 12;
+update t1 set x= 3 where x = 13;
+unlock tables;
+
+select count(x) from t1 partition (p0);
+--replace_result $default_engine DEFAULT_ENGINE
+show create table t1;
+drop tables t1;
+
+--echo # Test VIEW, LOCK TABLES
+set timestamp= unix_timestamp('2000-01-01 00:00:00');
+create or replace table t1 (x int) with system versioning
+partition by system_time interval 1 hour auto;
+create or replace view v1 as select * from t1;
+
+insert into t1 values (1);
+set timestamp= unix_timestamp('2000-01-01 01:00:00');
+update v1 set x= x + 2;
+--replace_result $default_engine DEFAULT_ENGINE
+show create table t1;
+
+set timestamp= unix_timestamp('2000-01-01 02:00:00');
+lock tables v1 write;
+update v1 set x= x + 3;
+--replace_result $default_engine DEFAULT_ENGINE
+show create table t1;
+unlock tables;
+
+drop view v1;
+drop tables t1;
+
+set timestamp= unix_timestamp('2000-01-01 00:00:00');
+create or replace table t1 (x int) with system versioning
+partition by system_time interval 1 hour auto partitions 3;
+
+create or replace table t2 (y int) with system versioning
+partition by system_time interval 1 hour auto;
+
+insert into t1 values (1);
+insert into t2 values (2);
+
+set timestamp= unix_timestamp('2000-01-01 01:00:00');
+update t1, t2 set x= x + 1, y= y + 1;
+--replace_result $default_engine DEFAULT_ENGINE
+show create table t1;
+--replace_result $default_engine DEFAULT_ENGINE
+show create table t2;
+
+set timestamp= unix_timestamp('2000-01-01 02:00:00');
+update t1, t2 set x= x + 1, y= y + 1;
+--replace_result $default_engine DEFAULT_ENGINE
+show create table t1;
+--replace_result $default_engine DEFAULT_ENGINE
+show create table t2;
+
+set timestamp= unix_timestamp('2000-01-01 03:00:00');
+update t1, t2 set t1.x= 0 where t1.x< t2.y;
+--replace_result $default_engine DEFAULT_ENGINE
+show create table t1;
+# Multiupdate_prelocking_strategy::handle_end() is processed after table open.
+# For PS it is possible to skip unneeded auto-creation because the above happens at
+# prepare stage and auto-creation is done at execute stage.
+--replace_result $default_engine DEFAULT_ENGINE 'PARTITIONS 4' 'PARTITIONS ok' 'PARTITIONS 5' 'PARTITIONS ok'
+show create table t2;
+
+drop tables t1, t2;
+
+--echo # PS, SP, LOCK TABLES
+set timestamp= unix_timestamp('2000-01-01 00:00:00');
+create or replace table t1 (x int) with system versioning
+partition by system_time interval 1 hour auto;
+
+insert into t1 values (1);
+
+set timestamp= unix_timestamp('2000-01-01 01:00:00');
+execute immediate 'update t1 set x= x + 5';
+--replace_result $default_engine DEFAULT_ENGINE
+show create table t1;
+
+prepare s from 'update t1 set x= x + 6';
+set timestamp= unix_timestamp('2000-01-01 02:00:00');
+execute s; execute s;
+--replace_result $default_engine DEFAULT_ENGINE
+show create table t1;
+
+set timestamp= unix_timestamp('2000-01-01 03:00:00');
+lock tables t1 write;
+execute s; execute s;
+--replace_result $default_engine DEFAULT_ENGINE
+show create table t1;
+unlock tables;
+drop prepare s;
+
+create procedure sp() update t1 set x= x + 7;
+set timestamp= unix_timestamp('2000-01-01 04:00:00');
+call sp; call sp;
+--replace_result $default_engine DEFAULT_ENGINE
+show create table t1;
+set timestamp= unix_timestamp('2000-01-01 05:00:00');
+lock tables t1 write;
+call sp; call sp;
+--replace_result $default_engine DEFAULT_ENGINE
+show create table t1;
+unlock tables;
+drop procedure sp;
+
+set timestamp= unix_timestamp('2001-01-01 00:00:00');
+create or replace table t1 (i int) with system versioning
+partition by system_time interval 1 day starts '2001-01-01 00:00:00';
+insert into t1 values (0);
+set timestamp= unix_timestamp('2001-01-01 00:00:01');
+prepare s from 'update t1 set i= i + 1';
+execute s;
+set timestamp= unix_timestamp('2001-01-02 00:00:01');
+execute s;
+drop prepare s;
+
+# Because of blobs:
+if (!$MTR_COMBINATION_HEAP)
+{
+--echo # Complex table
+set timestamp= unix_timestamp('2000-01-01 00:00:00');
+create or replace table t1 (
+  x int primary key auto_increment,
+  t timestamp(6) default '2001-11-11 11:11:11',
+  b blob(4096) compressed null,
+  c varchar(1033) character set utf8 not null,
+  u int unique,
+  m enum('a', 'b', 'c') not null default 'a' comment 'absolute',
+  i1 tinyint, i2 smallint, i3 bigint,
+  index three(i1, i2, i3),
+  v1 timestamp(6) generated always as (t + interval 1 day),
+  v2 timestamp(6) generated always as (t + interval 1 month) stored,
+  s timestamp(6) as row start,
+  e timestamp(6) as row end,
+  period for system_time (s, e),
+  ps date, pe date,
+  period for app_time (ps, pe),
+  constraint check_constr check (u > -1))
+with system versioning default charset=ucs2
+partition by system_time interval 1 hour auto (
+    partition p2 history,
+    partition pn current);
+--replace_result $default_engine DEFAULT_ENGINE
+show create table t1;
+
+insert into t1 (x, c, u, i1, i2, i3, ps, pe)
+values (1, 'cc', 0, 1, 2, 3, '1999-01-01', '2000-01-01');
+set timestamp= unix_timestamp('2000-01-01 01:00:00');
+update t1 set x= x + 8;
+--replace_result $default_engine DEFAULT_ENGINE
+show create table t1;
+
+set timestamp= unix_timestamp('2000-01-01 02:00:00');
+update t1 set x= x - 8;
+--replace_result $default_engine DEFAULT_ENGINE
+show create table t1;
+}
+--disable_info
+
+--echo # INSERT .. ON DUPLICATE KEY UPDATE (ODKU)
+set timestamp= unix_timestamp('2000-01-01 00:00:00');
+create or replace table t1 (x int primary key) with system versioning
+partition by system_time interval 1 hour auto;
+insert into t1 values (1);
+set timestamp= unix_timestamp('2000-01-01 01:00:00');
+insert into t1 values (1) on duplicate key update x= x + 1;
+--replace_result $default_engine DEFAULT_ENGINE
+show create table t1;
+
+--echo # LOAD DATA .. REPLACE
+set timestamp= unix_timestamp('2000-01-01 00:00:00');
+create or replace table t1 (x int primary key) with system versioning
+partition by system_time interval 1 hour auto;
+
+insert t1 values (1), (2), (3);
+select x into outfile 'MDEV-17554.data' from t1;
+
+set timestamp= unix_timestamp('2000-01-01 01:00:00');
+load data infile 'MDEV-17554.data' replace into table t1 (x);
+--replace_result $default_engine DEFAULT_ENGINE
+show create table t1;
+--remove_file $datadir/test/MDEV-17554.data
+
+--echo # Concurrent DML
+set timestamp= unix_timestamp('2000-01-01 00:00:00');
+create or replace table t1 (x int) with system versioning
+partition by system_time interval 1 hour auto partitions 3;
+
+insert into t1 values (1);
+--replace_result $default_engine DEFAULT_ENGINE
+show create table t1;
+
+--connect con8, localhost, root
+--connect con7, localhost, root
+--connect con6, localhost, root
+--connect con5, localhost, root
+--connect con4, localhost, root
+--connect con3, localhost, root
+--connect con2, localhost, root
+--connect con1, localhost, root
+set timestamp= unix_timestamp('2000-01-01 02:00:00');
+send update t1 set x= x + 10;
+--connection con2
+set timestamp= unix_timestamp('2000-01-01 02:00:00');
+send update t1 set x= x + 20;
+--connection con3
+set timestamp= unix_timestamp('2000-01-01 02:00:00');
+send update t1 set x= x + 30;
+--connection con4
+set timestamp= unix_timestamp('2000-01-01 02:00:00');
+send update t1 set x= x + 40;
+--connection con5
+set timestamp= unix_timestamp('2000-01-01 02:00:00');
+send update t1 set x= x + 50;
+--connection con6
+set timestamp= unix_timestamp('2000-01-01 02:00:00');
+send update t1 set x= x + 60;
+--connection con7
+set timestamp= unix_timestamp('2000-01-01 02:00:00');
+send update t1 set x= x + 70;
+--connection con8
+set timestamp= unix_timestamp('2000-01-01 02:00:00');
+update t1 set x= x + 80;
+--connection con1
+reap;
+--disconnect con1
+--connection con2
+reap;
+--disconnect con2
+--connection con3
+reap;
+--disconnect con3
+--connection con4
+reap;
+--disconnect con4
+--connection con5
+reap;
+--disconnect con5
+--connection con6
+reap;
+--disconnect con6
+--connection con7
+reap;
+--disconnect con7
+--disconnect con8
+--connection default
+--replace_result $default_engine DEFAULT_ENGINE
+show create table t1;
+
+drop tables t1;
+set timestamp= default;
+
+--echo # Concurrent DML (LIMIT)
+create or replace table t1 (x int) with system versioning engine heap
+partition by system_time limit 1 auto partitions 3;
+
+insert into t1 values (1);
+
+--let $max_loop= 3
+# For more intensity use
+# --let $max_loop= 30
+--echo update t1 set x= x + N; # (running multithreaded for $max_loop times)
+--disable_query_log
+--disable_result_log
+--connect con9, localhost, root
+--connect con10, localhost, root
+--connect con11, localhost, root
+--connect con12, localhost, root
+--connect con13, localhost, root
+--connect con14, localhost, root
+--connect con15, localhost, root
+--connect con16, localhost, root
+--connect con17, localhost, root
+--connect con18, localhost, root
+--connect con19, localhost, root
+--connect con20, localhost, root
+--connect con8, localhost, root
+--connect con7, localhost, root
+--connect con6, localhost, root
+--connect con5, localhost, root
+--connect con4, localhost, root
+--connect con3, localhost, root
+--connect con2, localhost, root
+--connect con1, localhost, root
+--let $i= 0
+while ($i < $max_loop)
+{
+  --connection con1
+  send update t1 set x= x + 100;
+  --connection con2
+  send update t1 set x= x + 200;
+  --connection con3
+  send update t1 set x= x + 300;
+  --connection con4
+  send update t1 set x= x + 400;
+  --connection con5
+  send update t1 set x= x + 500;
+  --connection con6
+  send update t1 set x= x + 600;
+  --connection con7
+  send update t1 set x= x + 700;
+  --connection con8
+  send update t1 set x= x + 800;
+  --connection con9
+  send update t1 set x= x + 900;
+  --connection con10
+  send update t1 set x= x + 1000;
+  --connection con11
+  send update t1 set x= x + 1100;
+  --connection con12
+  send update t1 set x= x + 1200;
+  --connection con13
+  send update t1 set x= x + 1300;
+  --connection con14
+  send update t1 set x= x + 1400;
+  --connection con15
+  send update t1 set x= x + 1500;
+  --connection con16
+  send update t1 set x= x + 1600;
+  --connection con17
+  send update t1 set x= x + 1700;
+  --connection con18
+  send update t1 set x= x + 1800;
+  --connection con19
+  send update t1 set x= x + 1900;
+  --connection con20
+  send update t1 set x= x + 2000;
+  --connection con1
+  reap;
+  --connection con2
+  reap;
+  --connection con3
+  reap;
+  --connection con4
+  reap;
+  --connection con5
+  reap;
+  --connection con6
+  reap;
+  --connection con7
+  reap;
+  --connection con8
+  reap;
+  --connection con9
+  reap;
+  --connection con10
+  reap;
+  --connection con11
+  reap;
+  --connection con12
+  reap;
+  --connection con13
+  reap;
+  --connection con14
+  reap;
+  --connection con15
+  reap;
+  --connection con16
+  reap;
+  --connection con17
+  reap;
+  --connection con18
+  reap;
+  --connection con19
+  reap;
+  --connection con20
+  reap;
+  --inc $i
+}
+
+--disconnect con1
+--disconnect con2
+--disconnect con3
+--disconnect con4
+--disconnect con5
+--disconnect con6
+--disconnect con7
+--disconnect con8
+--disconnect con9
+--disconnect con10
+--disconnect con11
+--disconnect con12
+--disconnect con13
+--disconnect con14
+--disconnect con15
+--disconnect con16
+--disconnect con17
+--disconnect con18
+--disconnect con19
+--disconnect con20
+
+--connection default
+# Result is undeterministic under LIMIT concurrency (MDEV-28459)
+# show create table t1;
+
+--enable_query_log
+--enable_result_log
+
+drop tables t1;
+
+--echo # Transaction
+set timestamp= unix_timestamp('2000-01-01 00:00:00');
+create or replace table t1 (x int) with system versioning engine innodb
+partition by system_time interval 1 hour auto;
+
+insert into t1 values (1);
+set timestamp= unix_timestamp('2000-01-01 01:00:00');
+start transaction;
+update t1 set x= 0;
+--connect con1, localhost, root
+select * from t1;
+show create table t1;
+--connection default
+commit;
+show create table t1;
+
+set timestamp= unix_timestamp('2000-01-01 02:00:00');
+start transaction;
+update t1 set x= 1;
+--connection con1
+select * from t1;
+--connection default
+rollback;
+show create table t1;
+--disconnect con1
+--connection default
+drop table t1;
+
+--echo #
+--echo # MDEV-25479 Auto-create: 2nd and further executions of PS or SP fail to create partition
+--echo #
+create table t (a int) with system versioning
+  partition by system_time interval 1 hour auto;
+
+insert into t values (1), (2);
+prepare stmt from "update t set a= a + 1";
+set @@timestamp= @@timestamp + 3601;
+execute stmt;
+set @@timestamp= @@timestamp + 3601;
+execute stmt;
+drop prepare stmt;
+--replace_result $default_engine DEFAULT_ENGINE
+show create table t;
+
+create procedure sp() update t set a= a + 1;
+set @@timestamp= @@timestamp + 3601;
+call sp();
+set @@timestamp= @@timestamp + 3601;
+call sp();
+drop procedure sp;
+--replace_result $default_engine DEFAULT_ENGINE
+show create table t;
+
+# Cleanup
+drop table t;
+
+--echo #
+--echo # MDEV-23639 Auto-create does not work under LOCK TABLES or inside triggers
+--echo #
+set timestamp= unix_timestamp('2000-01-01 00:00:00');
+create or replace table t1 (x int) with system versioning
+partition by system_time interval 1 hour auto
+partitions 3;
+
+create table t2 (x int);
+create table t3 (x int);
+insert into t3 values (3);
+
+create trigger tr after insert on t2 for each row update t1 set x= x + 11;
+create or replace procedure sp() update t1 set x= x + 5;
+create or replace procedure sp2() insert into t2 values (5);
+prepare ps from 'update t1 set x= x + 6';
+prepare ps2 from 'insert into t2 values (6)';
+
+insert into t1 values (1);
+set timestamp= unix_timestamp('2000-01-01 02:00:00');
+--replace_result $default_engine DEFAULT_ENGINE
+show create table t1;
+insert into t2 values (2);
+--replace_result $default_engine DEFAULT_ENGINE
+show create table t1;
+
+set timestamp= unix_timestamp('2000-01-01 03:00:00');
+call sp; call sp;
+--replace_result $default_engine DEFAULT_ENGINE
+show create table t1;
+set timestamp= unix_timestamp('2000-01-01 04:00:00');
+call sp2; call sp2;
+--replace_result $default_engine DEFAULT_ENGINE
+show create table t1;
+
+set timestamp= unix_timestamp('2000-01-01 05:00:00');
+execute ps; execute ps;
+--replace_result $default_engine DEFAULT_ENGINE
+show create table t1;
+set timestamp= unix_timestamp('2000-01-01 06:00:00');
+execute ps2; execute ps2;
+--replace_result $default_engine DEFAULT_ENGINE
+show create table t1;
+
+set timestamp= unix_timestamp('2000-01-01 08:00:00');
+lock tables t1 write, t2 write;
+--replace_result $default_engine DEFAULT_ENGINE
+show create table t1;
+set timestamp= unix_timestamp('2000-01-01 09:00:00');
+update t1 set x= x + 1;
+--replace_result $default_engine DEFAULT_ENGINE
+show create table t1;
+set timestamp= unix_timestamp('2000-01-01 10:00:00');
+update t1 set x= x + 2;
+--replace_result $default_engine DEFAULT_ENGINE
+show create table t1;
+update t2 set x= x + 1;
+set timestamp= unix_timestamp('2000-01-01 11:00:00');
+insert into t2 values (4);
+--replace_result $default_engine DEFAULT_ENGINE
+show create table t1;
+--error ER_TABLE_NOT_LOCKED
+update t3 set x= x + 1;
+
+set timestamp= unix_timestamp('2000-01-01 12:00:00');
+call sp; call sp;
+--replace_result $default_engine DEFAULT_ENGINE
+show create table t1;
+set timestamp= unix_timestamp('2000-01-01 13:00:00');
+call sp2; call sp2;
+--replace_result $default_engine DEFAULT_ENGINE
+show create table t1;
+
+set timestamp= unix_timestamp('2000-01-01 14:00:00');
+execute ps; execute ps;
+--replace_result $default_engine DEFAULT_ENGINE
+show create table t1;
+set timestamp= unix_timestamp('2000-01-01 15:00:00');
+execute ps2; execute ps2;
+--replace_result $default_engine DEFAULT_ENGINE
+show create table t1;
+
+unlock tables;
+--replace_result $default_engine DEFAULT_ENGINE
+show create table t1;
+
+# Cleanup
+drop tables t1, t2, t3;
+drop procedure sp;
+drop procedure sp2;
+drop prepare ps;
+drop prepare ps2;
+
+--echo #
+--echo # MDEV-27456 Assertion `!thd->is_error()' fails in vers_create_partitions upon DML with ER_UNKNOWN_PARTITION
+--echo #
+create table t (a int) with system versioning
+partition by system_time interval 1 minute auto;
+set @@timestamp= @@timestamp + 61;
+select * from t;
+--error ER_UNKNOWN_PARTITION
+delete from t partition (px);
+lock tables t write;
+--error ER_UNKNOWN_PARTITION
+delete from t partition (px);
+unlock tables;
+# cleanup
+drop table t;
+
+--disable_prepare_warnings
+SET GLOBAL innodb_stats_persistent=@save_persistent;
 --source suite/versioning/common_finish.inc
diff --git a/mysql-test/suite/versioning/t/rpl.test b/mysql-test/suite/versioning/t/rpl.test
index b5be68feece..7ce293d7eed 100644
--- a/mysql-test/suite/versioning/t/rpl.test
+++ b/mysql-test/suite/versioning/t/rpl.test
@@ -133,4 +133,23 @@ sync_slave_with_master;
 connection master;
 drop table t1;
 
+--echo #
+--echo # MDEV-17554 Auto-create new partition for system versioned tables
+--echo #            with history partitioned by INTERVAL/LIMIT
+--echo #
+set timestamp= unix_timestamp('2000-01-01 00:00:00');
+create or replace table t1 (x int) with system versioning
+partition by system_time interval 1 hour auto;
+insert t1 values ();
+set timestamp= unix_timestamp('2000-01-01 01:00:00');
+delete from t1;
+--replace_result InnoDB ENGINE MyISAM ENGINE MEMORY ENGINE
+show create table t1;
+--sync_slave_with_master
+--replace_result InnoDB ENGINE MyISAM ENGINE MEMORY ENGINE
+show create table t1;
+--connection master
+drop table t1;
+set timestamp= default;
+
 --source include/rpl_end.inc
diff --git a/mysql-test/suite/versioning/t/rpl_common.inc b/mysql-test/suite/versioning/t/rpl_common.inc
new file mode 100644
index 00000000000..c37e58f0f96
--- /dev/null
+++ b/mysql-test/suite/versioning/t/rpl_common.inc
@@ -0,0 +1,29 @@
+--source include/have_partition.inc
+
+--echo #
+--echo # MDEV-25347 DML events for auto-partitioned tables are written into binary log twice
+--echo #
+flush binary logs;
+create table t1 (a int) with system versioning
+partition by system_time limit 1 auto;
+
+insert into t1 values (1);
+update t1 set a= a + 1;
+update t1 set a= a + 2;
+--replace_result InnoDB ENGINE MyISAM ENGINE MEMORY ENGINE
+show create table t1;
+select * from t1;
+
+--let $binlog_file=master-bin.000002
+--source include/show_binlog_events.inc
+
+--sync_slave_with_master
+--replace_result InnoDB ENGINE MyISAM ENGINE MEMORY ENGINE
+show create table t1;
+
+select * from t1;
+--connection master
+# cleanup
+drop table t1;
+
+--source include/rpl_end.inc
diff --git a/mysql-test/suite/versioning/t/rpl_mix.test b/mysql-test/suite/versioning/t/rpl_mix.test
index 64025c74625..bf4e0b93422 100644
--- a/mysql-test/suite/versioning/t/rpl_mix.test
+++ b/mysql-test/suite/versioning/t/rpl_mix.test
@@ -16,4 +16,4 @@ DELETE HISTORY FROM t1;
 connection master;
 drop table t1;
 
---source include/rpl_end.inc
+--source rpl_common.inc
diff --git a/mysql-test/suite/versioning/t/rpl_row.test b/mysql-test/suite/versioning/t/rpl_row.test
index 17ce2dfdcf8..a9b12e52a41 100644
--- a/mysql-test/suite/versioning/t/rpl_row.test
+++ b/mysql-test/suite/versioning/t/rpl_row.test
@@ -15,4 +15,4 @@ update t1 set i = 0;
 drop table t1;
 set binlog_row_image= @old_row_image;
 
---source include/rpl_end.inc
+--source rpl_common.inc
diff --git a/mysql-test/suite/versioning/t/rpl_stmt.test b/mysql-test/suite/versioning/t/rpl_stmt.test
new file mode 100644
index 00000000000..dca4e8cb4a2
--- /dev/null
+++ b/mysql-test/suite/versioning/t/rpl_stmt.test
@@ -0,0 +1,4 @@
+--source include/have_binlog_format_statement.inc
+--source include/master-slave.inc
+
+--source rpl_common.inc
diff --git a/mysql-test/suite/versioning/t/update-big.test b/mysql-test/suite/versioning/t/update-big.test
index 175bfc79a48..98767cefaad 100644
--- a/mysql-test/suite/versioning/t/update-big.test
+++ b/mysql-test/suite/versioning/t/update-big.test
@@ -1,6 +1,7 @@
 source include/big_test.inc;
 source suite/versioning/engines.inc;
 source suite/versioning/common.inc;
+source include/have_partition.inc;
 
 --echo #
 --echo # MDEV-15458 Segfault in heap_scan() upon UPDATE after ADD SYSTEM VERSIONING
@@ -29,6 +30,79 @@ update t1 set a= 8 where a = 1;
 update t1 set a= 4 where a = 8;
 update t1 set a= 6;
 
+--disconnect con1
 drop table t1;
 
+call mtr.add_suppression("need more HISTORY partitions");
+
+--echo #
+--echo # MDEV-23642 Locking timeout caused by auto-creation affects original DML
+--echo #
+set timestamp= unix_timestamp('2000-01-01 00:00:00');
+create or replace table t1 (x int primary key) with system versioning engine innodb
+partition by system_time interval 1 hour auto;
+
+insert into t1 values (1);
+start transaction;
+insert into t1 values (2);
+
+--connect con1, localhost, root
+set lock_wait_timeout= 1;
+set innodb_lock_wait_timeout= 1;
+set timestamp= unix_timestamp('2000-01-01 01:00:01');
+update t1 set x= x + 122 where x = 1;
+--disconnect con1
+--connection default
+select * from t1;
+
+# cleanup
+drop table t1;
+set timestamp= default;
+
+--echo #
+--echo # MDEV-25339 Assertion `thd->transaction.stmt.is_empty() || thd->in_sub_stmt' failed
+--echo #
+create or replace table t1 (x int) with system versioning engine innodb
+partition by system_time interval 1 hour auto;
+start transaction;
+insert into t1 values (1);
+select * from t1;
+
+--connect con1, localhost, root
+set lock_wait_timeout= 1;
+set innodb_lock_wait_timeout= 1;
+--error ER_LOCK_WAIT_TIMEOUT
+update t1 set x= x + 111;
+select * from t1;
+
+# cleanup
+--disconnect con1
+--connection default
+drop table t1;
+
+--echo #
+--echo # MDEV-25482 Auto-create: Server hangs after a failed attempt to create partition
+--echo #
+set timestamp= default;
+create table t (pk int primary key, a int) engine=InnoDB
+  with system versioning partition by system_time interval 1 hour auto;
+
+insert into t values (1,1),(2,2),(3,3),(4,4),(5,5);
+
+start transaction;
+update t set a= 20 where pk = 2;
+
+--connect (con1,localhost,root,,)
+set lock_wait_timeout= 1;
+set @@timestamp= @@timestamp+3601;
+update t set a= 40 where pk = 4;
+update t set a= 400 where pk = 4;
+
+# Cleanup
+--disconnect con1
+--connection default
+select * from t where pk = 4;
+rollback;
+drop tables t;
+
 source suite/versioning/common_finish.inc;
diff --git a/sql/ha_partition.cc b/sql/ha_partition.cc
index 3a9d1eccaf5..34c65f2e45d 100644
--- a/sql/ha_partition.cc
+++ b/sql/ha_partition.cc
@@ -4193,19 +4193,9 @@ int ha_partition::external_lock(THD *thd, int lock_type)
       (void) (*file)->ha_external_lock(thd, lock_type);
     } while (*(++file));
   }
-  if (lock_type == F_WRLCK)
-  {
-    if (m_part_info->part_expr)
-      m_part_info->part_expr->walk(&Item::register_field_in_read_map, 1, 0);
-    if (m_part_info->part_type == VERSIONING_PARTITION &&
-      /* TODO: MDEV-20345 exclude more inapproriate commands like INSERT
-         These commands may be excluded because working history partition is needed
-         only for versioned DML. */
-      thd->lex->sql_command != SQLCOM_SELECT &&
-      thd->lex->sql_command != SQLCOM_INSERT_SELECT &&
-      (error= m_part_info->vers_set_hist_part(thd)))
-      goto err_handler;
-  }
+  if (lock_type == F_WRLCK && m_part_info->part_expr)
+    m_part_info->part_expr->walk(&Item::register_field_in_read_map, 1, 0);
+
   DBUG_RETURN(0);
 
 err_handler:
@@ -4349,11 +4339,6 @@ int ha_partition::start_stmt(THD *thd, thr_lock_type lock_type)
   {
     if (m_part_info->part_expr)
       m_part_info->part_expr->walk(&Item::register_field_in_read_map, 1, 0);
-    if (m_part_info->part_type == VERSIONING_PARTITION &&
-      // TODO: MDEV-20345 (see above)
-      thd->lex->sql_command != SQLCOM_SELECT &&
-      thd->lex->sql_command != SQLCOM_INSERT_SELECT)
-      error= m_part_info->vers_set_hist_part(thd);
   }
   DBUG_RETURN(error);
 }
diff --git a/sql/ha_partition.h b/sql/ha_partition.h
index 29763447e6e..fbf34efe39f 100644
--- a/sql/ha_partition.h
+++ b/sql/ha_partition.h
@@ -1605,6 +1605,13 @@ public:
   }
 
   bool partition_engine() override { return 1;}
+
+  /**
+     Get the number of records in part_elem and its subpartitions, if any.
+     Also sets read_partitions bit for each partition id it uses (that is needed
+     for vers_set_hist_part() because it is called before read_partitions bitmap
+     is initialized).
+  */
   ha_rows part_records(partition_element *part_elem)
   {
     DBUG_ASSERT(m_part_info);
@@ -1616,7 +1623,7 @@ public:
     for (; part_id < part_id_end; ++part_id)
     {
       handler *file= m_file[part_id];
-      DBUG_ASSERT(bitmap_is_set(&(m_part_info->read_partitions), part_id));
+      bitmap_set_bit(&(m_part_info->read_partitions), part_id);
       file->info(HA_STATUS_VARIABLE | HA_STATUS_NO_LOCK | HA_STATUS_OPEN);
       part_recs+= file->stats.records;
     }
diff --git a/sql/handler.cc b/sql/handler.cc
index 991c0d5aae5..fb1f9ba8a76 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -1651,9 +1651,10 @@ int ha_commit_trans(THD *thd, bool all)
   DBUG_ASSERT(thd->transaction->stmt.ha_list == NULL ||
               trans == &thd->transaction->stmt);
 
+  DBUG_ASSERT(!thd->in_sub_stmt);
+
   if (thd->in_sub_stmt)
   {
-    DBUG_ASSERT(0);
     /*
       Since we don't support nested statement transactions in 5.0,
       we can't commit or rollback stmt transactions while we are inside
diff --git a/sql/handler.h b/sql/handler.h
index 750250a390f..cdab18cecbc 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -841,6 +841,8 @@ typedef bool Log_func(THD*, TABLE*, bool, const uchar*, const uchar*);
 #define ALTER_PARTITION_TABLE_REORG (1ULL << 12)
 #define ALTER_PARTITION_CONVERT_IN  (1ULL << 13)
 #define ALTER_PARTITION_CONVERT_OUT (1ULL << 14)
+// Set for vers_add_auto_hist_parts() operation
+#define ALTER_PARTITION_AUTO_HIST   (1ULL << 15)
 
 /*
   This is master database for most of system tables. However there
@@ -2090,7 +2092,6 @@ struct Vers_parse_info: public Table_period_info
 
   Table_period_info::start_end_t as_row;
 
-protected:
   friend struct Table_scope_and_contents_source_st;
   void set_start(const LEX_CSTRING field_name)
   {
@@ -2102,6 +2103,8 @@ protected:
     as_row.end= field_name;
     period.end= field_name;
   }
+
+protected:
   bool is_start(const char *name) const;
   bool is_end(const char *name) const;
   bool is_start(const Create_field &f) const;
@@ -4202,6 +4205,8 @@ public:
   */
   virtual uint lock_count(void) const { return 1; }
   /**
+    Get the lock(s) for the table and perform conversion of locks if needed.
+
     Is not invoked for non-transactional temporary tables.
 
     @note store_lock() can return more than one lock if the table is MERGE
diff --git a/sql/lock.cc b/sql/lock.cc
index ef8e93cdfcc..6b4cc656a8c 100644
--- a/sql/lock.cc
+++ b/sql/lock.cc
@@ -654,7 +654,7 @@ bool mysql_lock_abort_for_thread(THD *thd, TABLE *table)
   a and b are freed with my_free()
 */
 
-MYSQL_LOCK *mysql_lock_merge(MYSQL_LOCK *a,MYSQL_LOCK *b)
+MYSQL_LOCK *mysql_lock_merge(MYSQL_LOCK *a, MYSQL_LOCK *b, THD *thd)
 {
   MYSQL_LOCK *sql_lock;
   TABLE **table, **end_table;
@@ -662,16 +662,28 @@ MYSQL_LOCK *mysql_lock_merge(MYSQL_LOCK *a,MYSQL_LOCK *b)
   DBUG_PRINT("enter", ("a->lock_count: %u  b->lock_count: %u",
                        a->lock_count, b->lock_count));
 
-  if (!(sql_lock= (MYSQL_LOCK*)
-	my_malloc(key_memory_MYSQL_LOCK, sizeof(*sql_lock) +
-		  sizeof(THR_LOCK_DATA*)*((a->lock_count+b->lock_count)*2) +
-		  sizeof(TABLE*)*(a->table_count+b->table_count),MYF(MY_WME))))
-    DBUG_RETURN(0);				// Fatal error
+  const size_t lock_size= sizeof(*sql_lock) +
+    sizeof(THR_LOCK_DATA *) * ((a->lock_count + b->lock_count) * 2) +
+    sizeof(TABLE *) * (a->table_count + b->table_count);
+  if (thd)
+  {
+    sql_lock= (MYSQL_LOCK *) thd->alloc(lock_size);
+    if (!sql_lock)
+      DBUG_RETURN(0);
+    sql_lock->flags= GET_LOCK_ON_THD;
+  }
+  else
+  {
+    sql_lock= (MYSQL_LOCK *)
+      my_malloc(key_memory_MYSQL_LOCK, lock_size, MYF(MY_WME));
+    if (!sql_lock)
+      DBUG_RETURN(0);
+    sql_lock->flags= 0;
+  }
   sql_lock->lock_count=a->lock_count+b->lock_count;
   sql_lock->table_count=a->table_count+b->table_count;
   sql_lock->locks=(THR_LOCK_DATA**) (sql_lock+1);
   sql_lock->table=(TABLE**) (sql_lock->locks+sql_lock->lock_count*2);
-  sql_lock->flags= 0;
   memcpy(sql_lock->locks,a->locks,a->lock_count*sizeof(*a->locks));
   memcpy(sql_lock->locks+a->lock_count,b->locks,
 	 b->lock_count*sizeof(*b->locks));
@@ -705,8 +717,10 @@ MYSQL_LOCK *mysql_lock_merge(MYSQL_LOCK *a,MYSQL_LOCK *b)
                   a->lock_count, b->lock_count);
 
   /* Delete old, not needed locks */
-  my_free(a);
-  my_free(b);
+  if (!(a->flags & GET_LOCK_ON_THD))
+    my_free(a);
+  if (!(b->flags & GET_LOCK_ON_THD))
+    my_free(b);
   DBUG_RETURN(sql_lock);
 }
 
diff --git a/sql/lock.h b/sql/lock.h
index 0b23ddd3846..85a93b9a7e3 100644
--- a/sql/lock.h
+++ b/sql/lock.h
@@ -34,7 +34,7 @@ int mysql_unlock_read_tables(THD *thd, MYSQL_LOCK *sql_lock);
 int mysql_unlock_some_tables(THD *thd, TABLE **table,uint count, uint flag);
 int mysql_lock_remove(THD *thd, MYSQL_LOCK *locked,TABLE *table);
 bool mysql_lock_abort_for_thread(THD *thd, TABLE *table);
-MYSQL_LOCK *mysql_lock_merge(MYSQL_LOCK *a,MYSQL_LOCK *b);
+MYSQL_LOCK *mysql_lock_merge(MYSQL_LOCK *a, MYSQL_LOCK *b, THD *thd= NULL);
 /* Lock based on name */
 bool lock_schema_name(THD *thd, const char *db);
 /* Lock based on stored routine name */
diff --git a/sql/log_event.cc b/sql/log_event.cc
index b3943760720..6f6de1ec282 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -4142,3 +4142,16 @@ bool copy_event_cache_to_file_and_reinit(IO_CACHE *cache, FILE *file)
   return (my_b_copy_all_to_file(cache, file) ||
           reinit_io_cache(cache, WRITE_CACHE, 0, FALSE, TRUE));
 }
+
+#if defined(MYSQL_SERVER) && defined(HAVE_REPLICATION)
+int Log_event::apply_event(rpl_group_info* rgi)
+{
+  int res;
+  THD_STAGE_INFO(thd, stage_apply_event);
+  rgi->current_event= this;
+  res= do_apply_event(rgi);
+  rgi->current_event= NULL;
+  THD_STAGE_INFO(thd, stage_after_apply_event);
+  return res;
+}
+#endif
diff --git a/sql/log_event.h b/sql/log_event.h
index 405b9a15003..517c5870445 100644
--- a/sql/log_event.h
+++ b/sql/log_event.h
@@ -1559,14 +1559,7 @@ public:
 
      @see do_apply_event
    */
-  int apply_event(rpl_group_info *rgi)
-  {
-    int res;
-    THD_STAGE_INFO(thd, stage_apply_event);
-    res= do_apply_event(rgi);
-    THD_STAGE_INFO(thd, stage_after_apply_event);
-    return res;
-  }
+  int apply_event(rpl_group_info *rgi);
 
 
   /**
diff --git a/sql/log_event_server.cc b/sql/log_event_server.cc
index 6969c409ef4..69ce44fe211 100644
--- a/sql/log_event_server.cc
+++ b/sql/log_event_server.cc
@@ -5817,6 +5817,18 @@ int Rows_log_event::do_apply_event(rpl_group_info *rgi)
         lex->query_tables_last= &tables->next_global;
       }
     }
+
+    /*
+      It is needed to set_time():
+      1) it continues the property that "Time" in SHOW PROCESSLIST shows how
+      much slave is behind
+      2) it will be needed when we allow replication from a table with no
+      TIMESTAMP column to a table with one.
+      So we call set_time(), like in SBR. Presently it changes nothing.
+      3) vers_set_hist_part() requires proper query time.
+    */
+    thd->set_time(when, when_sec_part);
+
     if (unlikely(open_and_lock_tables(thd, rgi->tables_to_lock, FALSE, 0)))
     {
 #ifdef WITH_WSREP
@@ -5993,16 +6005,6 @@ int Rows_log_event::do_apply_event(rpl_group_info *rgi)
       which tested replicate-* rules).
     */
 
-    /*
-      It's not needed to set_time() but
-      1) it continues the property that "Time" in SHOW PROCESSLIST shows how
-      much slave is behind
-      2) it will be needed when we allow replication from a table with no
-      TIMESTAMP column to a table with one.
-      So we call set_time(), like in SBR. Presently it changes nothing.
-    */
-    thd->set_time(when, when_sec_part);
-
      if (m_width == table->s->fields && bitmap_is_set_all(&m_cols))
       set_flags(COMPLETE_ROWS_F);
 
diff --git a/sql/partition_info.cc b/sql/partition_info.cc
index 4342b403acd..7d33f042431 100644
--- a/sql/partition_info.cc
+++ b/sql/partition_info.cc
@@ -38,6 +38,8 @@
 
 #ifdef WITH_PARTITION_STORAGE_ENGINE
 #include "ha_partition.h"
+#include "sql_table.h"
+#include "transaction.h"
 
 
 partition_info *partition_info::get_clone(THD *thd, bool empty_data_and_index_file)
@@ -325,13 +327,11 @@ bool partition_info::set_partition_bitmaps_from_table(TABLE_LIST *table_list)
     The external routine needing this code is check_partition_info
 */
 
-#define MAX_PART_NAME_SIZE 8
-
 char *partition_info::create_default_partition_names(THD *thd, uint part_no,
                                                      uint num_parts_arg,
                                                      uint start_no)
 {
-  char *ptr= (char*) thd->calloc(num_parts_arg * MAX_PART_NAME_SIZE);
+  char *ptr= (char*) thd->calloc(num_parts_arg * MAX_PART_NAME_SIZE + 1);
   char *move_ptr= ptr;
   uint i= 0;
   DBUG_ENTER("create_default_partition_names");
@@ -818,15 +818,16 @@ bool partition_info::has_unique_name(partition_element *element)
     vers_info->interval   Limit by fixed time interval
     vers_info->hist_part  (out) Working history partition
 */
-int partition_info::vers_set_hist_part(THD *thd)
+bool partition_info::vers_set_hist_part(THD *thd, uint *create_count)
 {
-  if (table->pos_in_table_list &&
-      table->pos_in_table_list->partition_names)
-  {
-    return HA_ERR_PARTITION_LIST;
-  }
+  DBUG_ASSERT(!thd->lex->last_table() ||
+              !thd->lex->last_table()->vers_conditions.delete_history);
+
+  const bool auto_hist= create_count && vers_info->auto_hist;
+
   if (vers_info->limit)
   {
+    DBUG_ASSERT(!vers_info->interval.is_set());
     ha_partition *hp= (ha_partition*)(table->file);
     partition_element *next= NULL;
     List_iterator<partition_element> it(partitions);
@@ -847,22 +848,22 @@ int partition_info::vers_set_hist_part(THD *thd)
     {
       if (next == vers_info->now_part)
       {
-        my_error(WARN_VERS_PART_FULL, MYF(ME_WARNING|ME_ERROR_LOG),
-                table->s->db.str, table->s->table_name.str,
-                vers_info->hist_part->partition_name, "LIMIT");
+        if (auto_hist)
+          *create_count= 1;
+        else
+          my_error(WARN_VERS_PART_FULL, MYF(ME_WARNING|ME_ERROR_LOG),
+                  table->s->db.str, table->s->table_name.str,
+                  vers_info->hist_part->partition_name, "LIMIT");
       }
       else
         vers_info->hist_part= next;
     }
-    return 0;
   }
-
-  if (vers_info->interval.is_set())
+  else if (vers_info->interval.is_set() &&
+           vers_info->hist_part->range_value <= thd->query_start())
   {
-    if (vers_info->hist_part->range_value > thd->query_start())
-      return 0;
-
     partition_element *next= NULL;
+    bool error= true;
     List_iterator<partition_element> it(partitions);
     while (next != vers_info->hist_part)
       next= it++;
@@ -871,10 +872,166 @@ int partition_info::vers_set_hist_part(THD *thd)
     {
       vers_info->hist_part= next;
       if (next->range_value > thd->query_start())
-        return 0;
+      {
+        error= false;
+        break;
+      }
+    }
+    if (error)
+    {
+      if (auto_hist)
+      {
+        *create_count= 0;
+        const my_time_t hist_end= (my_time_t) vers_info->hist_part->range_value;
+        DBUG_ASSERT(thd->query_start() >= hist_end);
+        MYSQL_TIME h0, q0;
+        my_tz_OFFSET0->gmt_sec_to_TIME(&h0, hist_end);
+        my_tz_OFFSET0->gmt_sec_to_TIME(&q0, thd->query_start());
+        longlong q= pack_time(&q0);
+        longlong h= pack_time(&h0);
+        while (h <= q)
+        {
+          if (date_add_interval(thd, &h0, vers_info->interval.type,
+                                vers_info->interval.step))
+            return true;
+          h= pack_time(&h0);
+          ++*create_count;
+          if (*create_count == MAX_PARTITIONS - 2)
+          {
+            my_error(ER_TOO_MANY_PARTITIONS_ERROR, MYF(ME_WARNING));
+            my_error(ER_VERS_HIST_PART_FAILED, MYF(0),
+                     table->s->db.str, table->s->table_name.str);
+            return true;
+          }
+        }
+      }
+      else
+      {
+        my_error(WARN_VERS_PART_FULL, MYF(ME_WARNING|ME_ERROR_LOG),
+                table->s->db.str, table->s->table_name.str,
+                vers_info->hist_part->partition_name, "INTERVAL");
+      }
     }
   }
-  return 0;
+
+  return false;
+}
+
+
+/**
+  @brief Run fast_alter_partition_table() to add new history partitions
+         for tables requiring them.
+*/
+bool vers_create_partitions(THD *thd, TABLE_LIST* tl, uint num_parts)
+{
+  bool result= true;
+  HA_CREATE_INFO create_info;
+  Alter_info alter_info;
+  partition_info *save_part_info= thd->work_part_info;
+  Query_tables_list save_query_tables;
+  Reprepare_observer *save_reprepare_observer= thd->m_reprepare_observer;
+  bool save_no_write_to_binlog= thd->lex->no_write_to_binlog;
+  thd->m_reprepare_observer= NULL;
+  thd->lex->reset_n_backup_query_tables_list(&save_query_tables);
+  thd->lex->no_write_to_binlog= true;
+  TABLE *table= tl->table;
+
+  DBUG_ASSERT(!thd->is_error());
+
+  {
+    DBUG_ASSERT(table->s->get_table_ref_type() == TABLE_REF_BASE_TABLE);
+    DBUG_ASSERT(table->versioned());
+    DBUG_ASSERT(table->part_info);
+    DBUG_ASSERT(table->part_info->vers_info);
+    alter_info.reset();
+    alter_info.partition_flags= ALTER_PARTITION_ADD|ALTER_PARTITION_AUTO_HIST;
+    create_info.init();
+    create_info.alter_info= &alter_info;
+    Alter_table_ctx alter_ctx(thd, tl, 1, &table->s->db, &table->s->table_name);
+
+    MDL_REQUEST_INIT(&tl->mdl_request, MDL_key::TABLE, tl->db.str,
+                    tl->table_name.str, MDL_SHARED_NO_WRITE, MDL_TRANSACTION);
+    if (thd->mdl_context.acquire_lock(&tl->mdl_request,
+                                      thd->variables.lock_wait_timeout))
+      goto exit;
+    table->mdl_ticket= tl->mdl_request.ticket;
+
+    create_info.db_type= table->s->db_type();
+    create_info.options|= HA_VERSIONED_TABLE;
+    DBUG_ASSERT(create_info.db_type);
+
+    create_info.vers_info.set_start(table->s->vers_start_field()->field_name);
+    create_info.vers_info.set_end(table->s->vers_end_field()->field_name);
+
+    partition_info *part_info= new partition_info();
+    if (unlikely(!part_info))
+    {
+      my_error(ER_OUT_OF_RESOURCES, MYF(0));
+      goto exit;
+    }
+    part_info->use_default_num_partitions= false;
+    part_info->use_default_num_subpartitions= false;
+    part_info->num_parts= num_parts;
+    part_info->num_subparts= table->part_info->num_subparts;
+    part_info->subpart_type= table->part_info->subpart_type;
+    if (unlikely(part_info->vers_init_info(thd)))
+    {
+      my_error(ER_OUT_OF_RESOURCES, MYF(0));
+      goto exit;
+    }
+
+    thd->work_part_info= part_info;
+    if (part_info->set_up_defaults_for_partitioning(thd, table->file, NULL,
+                                    table->part_info->next_part_no(num_parts)))
+    {
+      my_error(ER_VERS_HIST_PART_FAILED, MYF(ME_WARNING),
+               tl->db.str, tl->table_name.str);
+      goto exit;
+    }
+    bool partition_changed= false;
+    bool fast_alter_partition= false;
+    if (prep_alter_part_table(thd, table, &alter_info, &create_info,
+                              &partition_changed, &fast_alter_partition))
+    {
+      my_error(ER_VERS_HIST_PART_FAILED, MYF(ME_WARNING),
+               tl->db.str, tl->table_name.str);
+      goto exit;
+    }
+    if (!fast_alter_partition)
+    {
+      my_error(ER_VERS_HIST_PART_FAILED, MYF(ME_WARNING),
+               tl->db.str, tl->table_name.str);
+      goto exit;
+    }
+    DBUG_ASSERT(partition_changed);
+    if (mysql_prepare_alter_table(thd, table, &create_info, &alter_info,
+                                  &alter_ctx))
+    {
+      my_error(ER_VERS_HIST_PART_FAILED, MYF(ME_WARNING),
+               tl->db.str, tl->table_name.str);
+      goto exit;
+    }
+
+    if (fast_alter_partition_table(thd, table, &alter_info, &alter_ctx,
+                                   &create_info, tl))
+    {
+      my_error(ER_VERS_HIST_PART_FAILED, MYF(ME_WARNING),
+               tl->db.str, tl->table_name.str);
+      goto exit;
+    }
+  }
+
+  result= false;
+  // NOTE: we have to return DA_EMPTY for new command
+  DBUG_ASSERT(thd->get_stmt_da()->is_ok());
+  thd->get_stmt_da()->reset_diagnostics_area();
+
+exit:
+  thd->work_part_info= save_part_info;
+  thd->m_reprepare_observer= save_reprepare_observer;
+  thd->lex->restore_backup_query_tables_list(&save_query_tables);
+  thd->lex->no_write_to_binlog= save_no_write_to_binlog;
+  return result;
 }
 
 
@@ -2649,13 +2806,14 @@ bool partition_info::vers_init_info(THD * thd)
 
 bool partition_info::vers_set_interval(THD* thd, Item* interval,
                                        interval_type int_type, Item* starts,
-                                       const char *table_name)
+                                       bool auto_hist, const char *table_name)
 {
   DBUG_ASSERT(part_type == VERSIONING_PARTITION);
 
   MYSQL_TIME ltime;
   uint err;
   vers_info->interval.type= int_type;
+  vers_info->auto_hist= auto_hist;
 
   /* 1. assign INTERVAL to interval.step */
   if (interval->fix_fields_if_needed_for_scalar(thd, &interval))
@@ -2737,6 +2895,23 @@ interval_starts_error:
 }
 
 
+bool partition_info::vers_set_limit(ulonglong limit, bool auto_hist,
+                                    const char *table_name)
+{
+  DBUG_ASSERT(part_type == VERSIONING_PARTITION);
+
+  if (limit < 1)
+  {
+    my_error(ER_PART_WRONG_VALUE, MYF(0), table_name, "LIMIT");
+    return true;
+  }
+
+  vers_info->limit= limit;
+  vers_info->auto_hist= auto_hist;
+  return !limit;
+}
+
+
 bool partition_info::error_if_requires_values() const
 {
   switch (part_type) {
diff --git a/sql/partition_info.h b/sql/partition_info.h
index 525eedd310f..a3e36a64ffa 100644
--- a/sql/partition_info.h
+++ b/sql/partition_info.h
@@ -36,11 +36,11 @@ struct st_ddl_log_memory_entry;
 
 #define MAX_PART_NAME_SIZE 8
 
-
 struct Vers_part_info : public Sql_alloc
 {
   Vers_part_info() :
     limit(0),
+    auto_hist(false),
     now_part(NULL),
     hist_part(NULL)
   {
@@ -49,6 +49,7 @@ struct Vers_part_info : public Sql_alloc
   Vers_part_info(const Vers_part_info &src) :
     interval(src.interval),
     limit(src.limit),
+    auto_hist(src.auto_hist),
     now_part(NULL),
     hist_part(NULL)
   {
@@ -81,9 +82,10 @@ struct Vers_part_info : public Sql_alloc
     my_time_t start;
     INTERVAL step;
     enum interval_type type;
-    bool is_set() { return type < INTERVAL_LAST; }
+    bool is_set() const { return type < INTERVAL_LAST; }
   } interval;
   ulonglong limit;
+  bool auto_hist;
   partition_element *now_part;
   partition_element *hist_part;
 };
@@ -402,14 +404,9 @@ public:
   bool vers_init_info(THD *thd);
   bool vers_set_interval(THD *thd, Item *interval,
                          interval_type int_type, Item *starts,
-                         const char *table_name);
-  bool vers_set_limit(ulonglong limit)
-  {
-    DBUG_ASSERT(part_type == VERSIONING_PARTITION);
-    vers_info->limit= limit;
-    return !limit;
-  }
-  int vers_set_hist_part(THD *thd);
+                         bool auto_part, const char *table_name);
+  bool vers_set_limit(ulonglong limit, bool auto_part, const char *table_name);
+  bool vers_set_hist_part(THD* thd, uint *create_count);
   bool vers_fix_field_list(THD *thd);
   void vers_update_el_ids();
   partition_element *get_partition(uint part_id)
@@ -428,6 +425,7 @@ public:
 
 uint32 get_next_partition_id_range(struct st_partition_iter* part_iter);
 bool check_partition_dirs(partition_info *part_info);
+bool vers_create_partitions(THD* thd, TABLE_LIST* tl, uint num_parts);
 
 /* Initialize the iterator to return a single partition with given part_id */
 
@@ -483,11 +481,6 @@ bool partition_info::vers_fix_field_list(THD * thd)
 }
 
 
-/**
-  @brief Update partition_element's id
-
-  @returns true on error; false on success
-*/
 inline
 void partition_info::vers_update_el_ids()
 {
diff --git a/sql/rpl_rli.h b/sql/rpl_rli.h
index 80ee143a8e8..0fd9070426b 100644
--- a/sql/rpl_rli.h
+++ b/sql/rpl_rli.h
@@ -695,6 +695,8 @@ struct rpl_group_info
   */
   uint64 gtid_sub_id;
   rpl_gtid current_gtid;
+  /* Currently applied event or NULL */
+  Log_event *current_event;
   uint64 commit_id;
   /*
     This is used to keep transaction commit order.
diff --git a/sql/share/errmsg-utf8.txt b/sql/share/errmsg-utf8.txt
index 94cfd41c799..388a3aa74c1 100644
--- a/sql/share/errmsg-utf8.txt
+++ b/sql/share/errmsg-utf8.txt
@@ -10061,3 +10061,5 @@ ER_SF_OUT_INOUT_ARG_NOT_ALLOWED
 	eng "OUT or INOUT argument %d for function %s is not allowed here"
 ER_INCONSISTENT_SLAVE_TEMP_TABLE
         eng "Replicated query '%s' table `%s.%s` can not be temporary"
+ER_VERS_HIST_PART_FAILED
+        eng "Versioned table %`s.%`s: adding HISTORY partition(s) failed"
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index cb674c62ea2..73bfa7290e5 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -63,7 +63,6 @@
 #include "wsrep_trans_observer.h"
 #endif /* WITH_WSREP */
 
-
 bool
 No_such_table_error_handler::handle_condition(THD *,
                                               uint sql_errno,
@@ -948,7 +947,7 @@ void close_thread_table(THD *thd, TABLE **table_ptr)
   DBUG_PRINT("tcache", ("table: '%s'.'%s' %p", table->s->db.str,
                         table->s->table_name.str, table));
   DBUG_ASSERT(!table->file->keyread_enabled());
-  DBUG_ASSERT(!table->file || table->file->inited == handler::NONE);
+  DBUG_ASSERT(table->file->inited == handler::NONE);
 
   /*
     The metadata lock must be released after giving back
@@ -962,11 +961,8 @@ void close_thread_table(THD *thd, TABLE **table_ptr)
   table->vcol_cleanup_expr(thd);
   table->mdl_ticket= NULL;
 
-  if (table->file)
-  {
-    table->file->update_global_table_stats();
-    table->file->update_global_index_stats();
-  }
+  table->file->update_global_table_stats();
+  table->file->update_global_index_stats();
 
   /*
     This look is needed to allow THD::notify_shared_lock() to
@@ -1644,6 +1640,136 @@ bool is_locked_view(THD *thd, TABLE_LIST *t)
 }
 
 
+#ifdef WITH_PARTITION_STORAGE_ENGINE
+/**
+  Switch part_info->hist_part and request partition creation if needed.
+
+  @retval true  Error or partition creation was requested.
+  @retval false No error
+*/
+bool TABLE::vers_switch_partition(THD *thd, TABLE_LIST *table_list,
+                                  Open_table_context *ot_ctx)
+{
+  if (!part_info || part_info->part_type != VERSIONING_PARTITION ||
+      table_list->vers_conditions.delete_history ||
+      thd->stmt_arena->is_stmt_prepare() ||
+      table_list->lock_type < TL_WRITE_ALLOW_WRITE ||
+      table_list->mdl_request.type < MDL_SHARED_WRITE ||
+      table_list->mdl_request.type == MDL_EXCLUSIVE)
+  {
+    return false;
+  }
+
+  /*
+    NOTE: we need this condition of prelocking_placeholder because we cannot do
+    auto-create after the transaction is started. Auto-create does
+    close_tables_for_reopen() and that is not possible under started transaction.
+    Also the transaction may not be cancelled at that moment: f.ex. trigger
+    after insert is run when some data is already written.
+
+    We must do auto-creation for PRELOCK_ROUTINE tables at the initial
+    open_tables() no matter what initiating sql_command is.
+  */
+  if (table_list->prelocking_placeholder != TABLE_LIST::PRELOCK_ROUTINE)
+  {
+    switch (thd->lex->sql_command)
+    {
+      case SQLCOM_INSERT:
+        if (thd->lex->duplicates != DUP_UPDATE)
+          return false;
+        break;
+      case SQLCOM_LOAD:
+        if (thd->lex->duplicates != DUP_REPLACE)
+          return false;
+        break;
+      case SQLCOM_LOCK_TABLES:
+      case SQLCOM_DELETE:
+      case SQLCOM_UPDATE:
+      case SQLCOM_REPLACE:
+      case SQLCOM_REPLACE_SELECT:
+      case SQLCOM_DELETE_MULTI:
+      case SQLCOM_UPDATE_MULTI:
+        break;
+      default:
+        /*
+          TODO: make row events set thd->lex->sql_command appropriately.
+
+          Sergei Golubchik: f.ex. currently row events increment
+          thd->status_var.com_stat[] each event for its own SQLCOM_xxx, it won't be
+          needed if they'll just set thd->lex->sql_command.
+        */
+        if (thd->rgi_slave && thd->rgi_slave->current_event &&
+            thd->lex->sql_command == SQLCOM_END)
+        {
+          switch (thd->rgi_slave->current_event->get_type_code())
+          {
+          case UPDATE_ROWS_EVENT:
+          case UPDATE_ROWS_EVENT_V1:
+          case DELETE_ROWS_EVENT:
+          case DELETE_ROWS_EVENT_V1:
+            break;
+          default:;
+            return false;
+          }
+        }
+        break;
+    }
+  }
+
+  if (table_list->partition_names)
+  {
+    my_error(ER_VERS_NOT_ALLOWED, MYF(0), s->db.str, s->table_name.str);
+    return true;
+  }
+
+  TABLE *table= this;
+
+  /*
+      NOTE: The semantics of vers_set_hist_part() is twofold: even when we
+      don't need auto-create, we need to update part_info->hist_part.
+  */
+  uint *create_count= (table_list->vers_skip_create == thd->query_id) ?
+    NULL : &ot_ctx->vers_create_count;
+  table_list->vers_skip_create= thd->query_id;
+  if (table->part_info->vers_set_hist_part(thd, create_count))
+    return true;
+  if (ot_ctx->vers_create_count)
+  {
+    Open_table_context::enum_open_table_action action;
+    TABLE_LIST *table_arg;
+    mysql_mutex_lock(&table->s->LOCK_share);
+    if (!table->s->vers_skip_auto_create)
+    {
+      table->s->vers_skip_auto_create= true;
+      action= Open_table_context::OT_ADD_HISTORY_PARTITION;
+      table_arg= table_list;
+    }
+    else
+    {
+      /*
+          NOTE: this may repeat multiple times until creating thread acquires
+          MDL_EXCLUSIVE. Since auto-creation is rare operation this is acceptable.
+          We could suspend this thread on cond-var but we must first exit
+          MDL_SHARED_WRITE and we cannot store cond-var into TABLE_SHARE
+          because it is already released and there is no guarantee that it will
+          be same instance if we acquire it again.
+      */
+      table_list->vers_skip_create= 0;
+      ot_ctx->vers_create_count= 0;
+      action= Open_table_context::OT_REOPEN_TABLES;
+      table_arg= NULL;
+      DEBUG_SYNC(thd, "reopen_history_partition");
+    }
+    mysql_mutex_unlock(&table->s->LOCK_share);
+    ot_ctx->request_backoff_action(action, table_arg);
+    return true;
+  }
+
+  return false;
+}
+#endif /* WITH_PARTITION_STORAGE_ENGINE */
+
+
 /**
   Open a base table.
 
@@ -1797,6 +1923,9 @@ bool open_table(THD *thd, TABLE_LIST *table_list, Open_table_context *ot_ctx)
       DBUG_PRINT("info",("Using locked table"));
 #ifdef WITH_PARTITION_STORAGE_ENGINE
       part_names_error= set_partitions_as_used(table_list, table);
+      if (!part_names_error
+          && table->vers_switch_partition(thd, table_list, ot_ctx))
+        DBUG_RETURN(true);
 #endif
       goto reset;
     }
@@ -2053,6 +2182,16 @@ retry_share:
     from_share= true;
   }
 
+#ifdef WITH_PARTITION_STORAGE_ENGINE
+  if (!part_names_error &&
+      table->vers_switch_partition(thd, table_list, ot_ctx))
+  {
+    MYSQL_UNBIND_TABLE(table->file);
+    tc_release_table(table);
+    DBUG_RETURN(true);
+  }
+#endif /* WITH_PARTITION_STORAGE_ENGINE */
+
   if (!(flags & MYSQL_OPEN_HAS_MDL_LOCK) &&
       table->s->table_category < TABLE_CATEGORY_INFORMATION)
   {
@@ -2130,6 +2269,7 @@ retry_share:
 
   table->init(thd, table_list);
 
+  DBUG_ASSERT(table != thd->open_tables);
   table->next= thd->open_tables;		/* Link into simple list */
   thd->set_open_tables(table);
 
@@ -2598,7 +2738,7 @@ unlink_all_closed_tables(THD *thd, MYSQL_LOCK *lock, size_t reopen_count)
   This is only needed when LOCK TABLES is active
 */
 
-void Locked_tables_list::mark_table_for_reopen(THD *thd, TABLE *table)
+void Locked_tables_list::mark_table_for_reopen(TABLE *table)
 {
   TABLE_SHARE *share= table->s;
 
@@ -2611,11 +2751,13 @@ void Locked_tables_list::mark_table_for_reopen(THD *thd, TABLE *table)
       close_all_tables_for_name().
     */
     if (table_list->table && table_list->table->s == share)
+    {
       table_list->table->internal_set_needs_reopen(true);
+      some_table_marked_for_reopen= 1;
+    }
   }
   /* This is needed in the case where lock tables where not used */
   table->internal_set_needs_reopen(true);
-  some_table_marked_for_reopen= 1;
 }
 
 
@@ -3072,7 +3214,8 @@ Open_table_context::Open_table_context(THD *thd, uint flags)
    m_flags(flags),
    m_action(OT_NO_ACTION),
    m_has_locks(thd->mdl_context.has_locks()),
-   m_has_protection_against_grl(0)
+   m_has_protection_against_grl(0),
+   vers_create_count(0)
 {}
 
 
@@ -3152,13 +3295,15 @@ request_backoff_action(enum_open_table_action action_arg,
   */
   if (table)
   {
-    DBUG_ASSERT(action_arg == OT_DISCOVER || action_arg == OT_REPAIR);
+    DBUG_ASSERT(action_arg == OT_DISCOVER || action_arg == OT_REPAIR ||
+                action_arg == OT_ADD_HISTORY_PARTITION);
     m_failed_table= (TABLE_LIST*) m_thd->alloc(sizeof(TABLE_LIST));
     if (m_failed_table == NULL)
       return TRUE;
     m_failed_table->init_one_table(&table->db, &table->table_name, &table->alias, TL_WRITE);
     m_failed_table->open_strategy= table->open_strategy;
     m_failed_table->mdl_request.set_type(MDL_EXCLUSIVE);
+    m_failed_table->vers_skip_create= table->vers_skip_create;
   }
   m_action= action_arg;
   return FALSE;
@@ -3219,13 +3364,50 @@ Open_table_context::recover_from_failed_open()
       break;
     case OT_DISCOVER:
     case OT_REPAIR:
-      if ((result= lock_table_names(m_thd, m_thd->lex->create_info,
-                                    m_failed_table, NULL,
-                                    get_timeout(), 0)))
+    case OT_ADD_HISTORY_PARTITION:
+      DEBUG_SYNC(m_thd, "add_history_partition");
+      if (!m_thd->locked_tables_mode)
+        result= lock_table_names(m_thd, m_thd->lex->create_info, m_failed_table,
+                                NULL, get_timeout(), 0);
+      else
+      {
+        DBUG_ASSERT(!result);
+        DBUG_ASSERT(m_action == OT_ADD_HISTORY_PARTITION);
+      }
+      /*
+         We are now under MDL_EXCLUSIVE mode. Other threads have no table share
+         acquired: they are blocked either at open_table_get_mdl_lock() in
+         open_table() or at lock_table_names() here.
+      */
+      if (result)
+      {
+        if (m_action == OT_ADD_HISTORY_PARTITION)
+        {
+          TABLE_SHARE *share= tdc_acquire_share(m_thd, m_failed_table,
+                                                GTS_TABLE, NULL);
+          if (share)
+          {
+            share->vers_skip_auto_create= false;
+            tdc_release_share(share);
+          }
+          if (m_thd->get_stmt_da()->sql_errno() == ER_LOCK_WAIT_TIMEOUT)
+          {
+            // MDEV-23642 Locking timeout caused by auto-creation affects original DML
+            m_thd->clear_error();
+            vers_create_count= 0;
+            result= false;
+          }
+        }
         break;
+      }
 
-      tdc_remove_table(m_thd, m_failed_table->db.str,
-                       m_failed_table->table_name.str);
+      /*
+         We don't need to remove share under OT_ADD_HISTORY_PARTITION.
+         Moreover fast_alter_partition_table() works with TABLE instance.
+      */
+      if (m_action != OT_ADD_HISTORY_PARTITION)
+        tdc_remove_table(m_thd, m_failed_table->db.str,
+                        m_failed_table->table_name.str);
 
       switch (m_action)
       {
@@ -3253,6 +3435,72 @@ Open_table_context::recover_from_failed_open()
         case OT_REPAIR:
           result= auto_repair_table(m_thd, m_failed_table);
           break;
+        case OT_ADD_HISTORY_PARTITION:
+#ifdef WITH_PARTITION_STORAGE_ENGINE
+        {
+          result= false;
+          TABLE *table= open_ltable(m_thd, m_failed_table, TL_WRITE,
+                    MYSQL_OPEN_HAS_MDL_LOCK | MYSQL_OPEN_IGNORE_LOGGING_FORMAT);
+          if (table == NULL)
+          {
+            m_thd->clear_error();
+            break;
+          }
+
+          DBUG_ASSERT(vers_create_count);
+          result= vers_create_partitions(m_thd, m_failed_table, vers_create_count);
+          vers_create_count= 0;
+          if (!m_thd->transaction->stmt.is_empty())
+            trans_commit_stmt(m_thd);
+          DBUG_ASSERT(!result ||
+                      !m_thd->locked_tables_mode ||
+                      m_thd->lock->lock_count);
+          if (result)
+            break;
+          if (!m_thd->locked_tables_mode)
+          {
+            /*
+              alter_partition_lock_handling() does mysql_lock_remove() but
+              does not clear thd->lock completely.
+            */
+            DBUG_ASSERT(m_thd->lock->lock_count == 0);
+            if (!(m_thd->lock->flags & GET_LOCK_ON_THD))
+              my_free(m_thd->lock);
+            m_thd->lock= NULL;
+          }
+          else if (m_thd->locked_tables_mode == LTM_PRELOCKED)
+          {
+            MYSQL_LOCK *lock;
+            MYSQL_LOCK *merged_lock;
+
+            /*
+              In LTM_LOCK_TABLES table was reopened via locked_tables_list,
+              but not in prelocked environment where we have to reopen
+              the table manually.
+            */
+            Open_table_context ot_ctx(m_thd, MYSQL_OPEN_REOPEN);
+            if (open_table(m_thd, m_failed_table, &ot_ctx))
+            {
+              result= true;
+              break;
+            }
+            TABLE *table= m_failed_table->table;
+            table->reginfo.lock_type= m_thd->update_lock_default;
+            m_thd->in_lock_tables= 1;
+            lock= mysql_lock_tables(m_thd, &table, 1,
+                                    MYSQL_OPEN_REOPEN | MYSQL_LOCK_USE_MALLOC);
+            m_thd->in_lock_tables= 0;
+            if (lock == NULL ||
+                !(merged_lock= mysql_lock_merge(m_thd->lock, lock, m_thd)))
+            {
+              result= true;
+              break;
+            }
+            m_thd->lock= merged_lock;
+          }
+          break;
+        }
+#endif /* WITH_PARTITION_STORAGE_ENGINE */
         case OT_BACKOFF_AND_RETRY:
         case OT_REOPEN_TABLES:
         case OT_NO_ACTION:
@@ -4227,6 +4475,7 @@ bool open_tables(THD *thd, const DDL_options_st &options,
     }
 
   thd->current_tablenr= 0;
+
 restart:
   /*
     Close HANDLER tables which are marked for flush or against which there
@@ -4307,6 +4556,9 @@ restart:
     /*
       For every table in the list of tables to open, try to find or open
       a table.
+
+      NOTE: there can be duplicates in the list. F.ex. table specified in
+      LOCK TABLES and prelocked via another table (like when used in a trigger).
     */
     for (tables= *table_to_open; tables;
          table_to_open= &tables->next_global, tables= tables->next_global)
@@ -4401,6 +4653,8 @@ restart:
         {
           if (ot_ctx.can_recover_from_failed_open())
           {
+            // FIXME: is this really used?
+            DBUG_ASSERT(0);
             close_tables_for_reopen(thd, start,
                                     ot_ctx.start_of_statement_svp());
             if (ot_ctx.recover_from_failed_open())
@@ -5174,16 +5428,14 @@ TABLE *open_ltable(THD *thd, TABLE_LIST *table_list, thr_lock_type lock_type,
   if (table_list->table)
     DBUG_RETURN(table_list->table);
 
-  /* should not be used in a prelocked_mode context, see NOTE above */
-  DBUG_ASSERT(thd->locked_tables_mode < LTM_PRELOCKED);
-
   THD_STAGE_INFO(thd, stage_opening_tables);
   thd->current_tablenr= 0;
   /* open_ltable can be used only for BASIC TABLEs */
   table_list->required_type= TABLE_TYPE_NORMAL;
 
   /* This function can't properly handle requests for such metadata locks. */
-  DBUG_ASSERT(table_list->mdl_request.type < MDL_SHARED_UPGRADABLE);
+  DBUG_ASSERT(lock_flags & MYSQL_OPEN_HAS_MDL_LOCK  ||
+              table_list->mdl_request.type < MDL_SHARED_UPGRADABLE);
 
   while ((error= open_table(thd, table_list, &ot_ctx)) &&
          ot_ctx.can_recover_from_failed_open())
diff --git a/sql/sql_base.h b/sql/sql_base.h
index c593d85b5e7..06d75596827 100644
--- a/sql/sql_base.h
+++ b/sql/sql_base.h
@@ -533,7 +533,8 @@ public:
     OT_BACKOFF_AND_RETRY,
     OT_REOPEN_TABLES,
     OT_DISCOVER,
-    OT_REPAIR
+    OT_REPAIR,
+    OT_ADD_HISTORY_PARTITION
   };
   Open_table_context(THD *thd, uint flags);
 
@@ -606,6 +607,9 @@ private:
     protection against global read lock.
   */
   mdl_bitmap_t m_has_protection_against_grl;
+
+public:
+  uint vers_create_count;
 };
 
 
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index 27875e3837b..1d6409db25d 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -845,7 +845,7 @@ THD::THD(my_thread_id id, bool is_wsrep_applier)
   wsrep_info[sizeof(wsrep_info) - 1] = '\0'; /* make sure it is 0-terminated */
 #endif
   /* Call to init() below requires fully initialized Open_tables_state. */
-  reset_open_tables_state(this);
+  reset_open_tables_state();
 
   init();
   debug_sync_init_thread(this);
@@ -4589,7 +4589,7 @@ void THD::reset_n_backup_open_tables_state(Open_tables_backup *backup)
   DBUG_ENTER("reset_n_backup_open_tables_state");
   backup->set_open_tables_state(this);
   backup->mdl_system_tables_svp= mdl_context.mdl_savepoint();
-  reset_open_tables_state(this);
+  reset_open_tables_state();
   state_flags|= Open_tables_state::BACKUPS_AVAIL;
   DBUG_VOID_RETURN;
 }
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 01ec676e132..a2439e96564 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -1600,6 +1600,10 @@ enum enum_locked_tables_mode
   LTM_NONE= 0,
   LTM_LOCK_TABLES,
   LTM_PRELOCKED,
+  /*
+     TODO: remove LTM_PRELOCKED_UNDER_LOCK_TABLES: it is never used apart from
+     LTM_LOCK_TABLES.
+  */
   LTM_PRELOCKED_UNDER_LOCK_TABLES,
   LTM_always_last
 };
@@ -1782,7 +1786,7 @@ public:
     *this= *state;
   }
 
-  void reset_open_tables_state(THD *thd)
+  void reset_open_tables_state()
   {
     open_tables= 0;
     temporary_tables= 0;
@@ -2130,7 +2134,7 @@ public:
   bool restore_lock(THD *thd, TABLE_LIST *dst_table_list, TABLE *table,
                     MYSQL_LOCK *lock);
   void add_back_last_deleted_lock(TABLE_LIST *dst_table_list);
-  void mark_table_for_reopen(THD *thd, TABLE *table);
+  void mark_table_for_reopen(TABLE *table);
 };
 
 
diff --git a/sql/sql_lex.cc b/sql/sql_lex.cc
index 48d86981932..18d54bf0d69 100644
--- a/sql/sql_lex.cc
+++ b/sql/sql_lex.cc
@@ -9782,7 +9782,11 @@ bool LEX::part_values_history(THD *thd)
   }
   else
   {
-    part_info->vers_init_info(thd);
+    if (unlikely(part_info->vers_init_info(thd)))
+    {
+      my_error(ER_OUT_OF_RESOURCES, MYF(0));
+      return true;
+    }
     elem->id= UINT_MAX32;
   }
   DBUG_ASSERT(part_info->vers_info);
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index f841af1fea6..0ac082b1812 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -84,6 +84,7 @@
 #include "events.h"
 #include "sql_trigger.h"
 #include "transaction.h"
+#include "sql_alter.h"
 #include "sql_audit.h"
 #include "sql_prepare.h"
 #include "sql_cte.h"
@@ -6071,7 +6072,6 @@ finish:
   /* Free tables. Set stage 'closing tables' */
   close_thread_tables_for_query(thd);
 
-
 #ifndef DBUG_OFF
   if (lex->sql_command != SQLCOM_SET_OPTION && ! thd->in_sub_stmt)
     DEBUG_SYNC(thd, "execute_command_after_close_tables");
@@ -7584,6 +7584,7 @@ void THD::reset_for_next_command(bool do_clear_error)
       global_system_variables.auto_increment_increment;
   }
 #endif /* WITH_WSREP */
+
   query_start_sec_part_used= 0;
   is_fatal_error= time_zone_used= 0;
   variables.option_bits&= ~OPTION_BINLOG_THIS_STMT;
diff --git a/sql/sql_partition.cc b/sql/sql_partition.cc
index ff4fad6906a..d2cdbdaf122 100644
--- a/sql/sql_partition.cc
+++ b/sql/sql_partition.cc
@@ -2595,11 +2595,17 @@ char *generate_partition_syntax(THD *thd, partition_info *part_info,
         err+= str.append('\'');
       }
     }
-    if (vers_info->limit)
+    else if (vers_info->limit)
     {
       err+= str.append(STRING_WITH_LEN("LIMIT "));
       err+= str.append_ulonglong(vers_info->limit);
     }
+    if (vers_info->auto_hist)
+    {
+      DBUG_ASSERT(vers_info->interval.is_set() ||
+                  vers_info->limit);
+      err+= str.append(STRING_WITH_LEN(" AUTO"));
+    }
   }
   else if (part_info->part_expr)
   {
@@ -5349,7 +5355,9 @@ that are reorganised.
               now_part= el;
             }
           }
-          if (*fast_alter_table && tab_part_info->vers_info->interval.is_set())
+          if (*fast_alter_table &&
+              !(alter_info->partition_flags & ALTER_PARTITION_AUTO_HIST) &&
+              tab_part_info->vers_info->interval.is_set())
           {
             partition_element *hist_part= tab_part_info->vers_info->hist_part;
             if (hist_part->range_value <= thd->query_start())
@@ -6085,7 +6093,7 @@ err:
                                records are added
 */
 
-static bool mysql_change_partitions(ALTER_PARTITION_PARAM_TYPE *lpt)
+static bool mysql_change_partitions(ALTER_PARTITION_PARAM_TYPE *lpt, bool copy_data)
 {
   char path[FN_REFLEN+1];
   int error;
@@ -6095,7 +6103,7 @@ static bool mysql_change_partitions(ALTER_PARTITION_PARAM_TYPE *lpt)
 
   build_table_filename(path, sizeof(path) - 1, lpt->db.str, lpt->table_name.str, "", 0);
 
-  if(mysql_trans_prepare_alter_copy_data(thd))
+  if(copy_data && mysql_trans_prepare_alter_copy_data(thd))
     DBUG_RETURN(TRUE);
 
   /* TODO: test if bulk_insert would increase the performance */
@@ -6109,7 +6117,9 @@ static bool mysql_change_partitions(ALTER_PARTITION_PARAM_TYPE *lpt)
     file->print_error(error, MYF(error != ER_OUTOFMEMORY ? 0 : ME_FATAL));
   }
 
-  if (mysql_trans_commit_alter_copy_data(thd))
+  DBUG_ASSERT(copy_data || (!lpt->copied && !lpt->deleted));
+
+  if (copy_data && mysql_trans_commit_alter_copy_data(thd))
     error= 1;                                /* The error has been reported */
 
   DBUG_RETURN(MY_TEST(error));
@@ -7392,7 +7402,8 @@ uint fast_alter_partition_table(THD *thd, TABLE *table,
     thd->variables.option_bits|= OPTION_IF_EXISTS;
 
   if (table->file->alter_table_flags(alter_info->flags) &
-        HA_PARTITION_ONE_PHASE)
+        HA_PARTITION_ONE_PHASE &&
+      !(alter_info->partition_flags & ALTER_PARTITION_AUTO_HIST))
   {
     /*
       In the case where the engine supports one phase online partition
@@ -7434,7 +7445,7 @@ uint fast_alter_partition_table(THD *thd, TABLE *table,
       2) Perform the change within the handler
     */
     if (mysql_write_frm(lpt, WFRM_WRITE_SHADOW) ||
-        mysql_change_partitions(lpt))
+        mysql_change_partitions(lpt, true))
     {
       goto err;
     }
@@ -7627,9 +7638,14 @@ uint fast_alter_partition_table(THD *thd, TABLE *table,
         ERROR_INJECT("convert_partition_11"))
       goto err;
   }
+  /*
+    TODO: would be good if adding new empty VERSIONING partitions would always
+    go this way, auto or not.
+  */
   else if ((alter_info->partition_flags & ALTER_PARTITION_ADD) &&
            (part_info->part_type == RANGE_PARTITION ||
-            part_info->part_type == LIST_PARTITION))
+            part_info->part_type == LIST_PARTITION ||
+            alter_info->partition_flags & ALTER_PARTITION_AUTO_HIST))
   {
     DBUG_ASSERT(!(alter_info->partition_flags & ALTER_PARTITION_CONVERT_IN));
     /*
@@ -7670,7 +7686,7 @@ uint fast_alter_partition_table(THD *thd, TABLE *table,
         ERROR_INJECT("add_partition_3") ||
         write_log_add_change_partition(lpt) ||
         ERROR_INJECT("add_partition_4") ||
-        mysql_change_partitions(lpt) ||
+        mysql_change_partitions(lpt, false) ||
         ERROR_INJECT("add_partition_5") ||
         alter_close_table(lpt) ||
         ERROR_INJECT("add_partition_6") ||
@@ -7757,7 +7773,7 @@ uint fast_alter_partition_table(THD *thd, TABLE *table,
         ERROR_INJECT("change_partition_2") ||
         write_log_add_change_partition(lpt) ||
         ERROR_INJECT("change_partition_3") ||
-        mysql_change_partitions(lpt) ||
+        mysql_change_partitions(lpt, true) ||
         ERROR_INJECT("change_partition_4") ||
         wait_while_table_is_used(thd, table, HA_EXTRA_NOT_USED) ||
         ERROR_INJECT("change_partition_5") ||
diff --git a/sql/sql_truncate.cc b/sql/sql_truncate.cc
index c6af72c5979..a0ef89ff0f5 100644
--- a/sql/sql_truncate.cc
+++ b/sql/sql_truncate.cc
@@ -511,7 +511,7 @@ bool Sql_cmd_truncate_table::truncate_table(THD *thd, TABLE_LIST *table_ref)
            (HTON_REQUIRES_CLOSE_AFTER_TRUNCATE |
             HTON_TRUNCATE_REQUIRES_EXCLUSIVE_USE)))
       {
-        thd->locked_tables_list.mark_table_for_reopen(thd, table_ref->table);
+        thd->locked_tables_list.mark_table_for_reopen(table_ref->table);
         if (unlikely(thd->locked_tables_list.reopen_tables(thd, false)))
           thd->locked_tables_list.unlink_all_closed_tables(thd, NULL, 0);
       }
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index b86f49e7f2e..826c470511f 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -1498,6 +1498,8 @@ bool my_yyoverflow(short **a, YYSTYPE **b, size_t *yystacksize);
         condition_number
         opt_versioning_interval_start
 
+%type <num> opt_vers_auto_part
+
 %type <item_param> param_marker
 
 %type <item_num>
@@ -5114,24 +5116,20 @@ server_part_option:
 
 opt_versioning_rotation:
          /* empty */ {}
-       | INTERVAL_SYM expr interval opt_versioning_interval_start
+       | INTERVAL_SYM expr interval opt_versioning_interval_start opt_vers_auto_part
          {
            partition_info *part_info= Lex->part_info;
            const char *table_name= Lex->create_last_non_select_table->table_name.str;
-           if (unlikely(part_info->vers_set_interval(thd, $2, $3, $4, table_name)))
+           if (unlikely(part_info->vers_set_interval(thd, $2, $3, $4, $5, table_name)))
              MYSQL_YYABORT;
          }
-       | LIMIT ulonglong_num
-       {
-         partition_info *part_info= Lex->part_info;
-         if (unlikely(part_info->vers_set_limit($2)))
+       | LIMIT ulonglong_num opt_vers_auto_part
          {
-           my_error(ER_PART_WRONG_VALUE, MYF(0),
-                    Lex->create_last_non_select_table->table_name.str,
-                    "LIMIT");
-           MYSQL_YYABORT;
+           partition_info *part_info= Lex->part_info;
+           const char *table_name= Lex->create_last_non_select_table->table_name.str;
+           if (unlikely(part_info->vers_set_limit($2, $3, table_name)))
+             MYSQL_YYABORT;
          }
-       }
        ;
 
 
@@ -5146,6 +5144,16 @@ opt_versioning_interval_start:
          }
        ;
 
+opt_vers_auto_part:
+         /* empty */
+         {
+           $$= 0;
+         }
+       | AUTO_SYM
+         {
+           $$= 1;
+         }
+       ;
 /*
  End of partition parser part
 */
diff --git a/sql/table.cc b/sql/table.cc
index b9f0e7596aa..efa7ae3a5b6 100644
--- a/sql/table.cc
+++ b/sql/table.cc
@@ -10270,5 +10270,5 @@ void TABLE::mark_table_for_reopen()
 {
   THD *thd= in_use;
   DBUG_ASSERT(thd);
-  thd->locked_tables_list.mark_table_for_reopen(thd, this);
+  thd->locked_tables_list.mark_table_for_reopen(this);
 }
diff --git a/sql/table.h b/sql/table.h
index dd40b1760b4..426e204e2fb 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -64,6 +64,7 @@ class derived_handler;
 class Pushdown_derived;
 struct Name_resolution_context;
 class Table_function_json_table;
+class Open_table_context;
 
 /*
   Used to identify NESTED_JOIN structures within a join (applicable only to
@@ -914,6 +915,13 @@ struct TABLE_SHARE
   vers_kind_t versioned;
   period_info_t vers;
   period_info_t period;
+  /*
+      Protect multiple threads from repeating partition auto-create over
+      single share.
+
+      TODO: remove it when partitioning metadata will be in TABLE_SHARE.
+  */
+  bool          vers_skip_auto_create;
 
   bool init_period_from_extra2(period_info_t *period, const uchar *data,
                                const uchar *end);
@@ -1778,6 +1786,10 @@ public:
 
   ulonglong vers_start_id() const;
   ulonglong vers_end_id() const;
+#ifdef WITH_PARTITION_STORAGE_ENGINE
+  bool vers_switch_partition(THD *thd, TABLE_LIST *table_list,
+                             Open_table_context *ot_ctx);
+#endif
 
   int update_generated_fields();
   int period_make_insert(Item *src, Field *dst);
@@ -2572,6 +2584,13 @@ struct TABLE_LIST
   bool          merged;
   bool          merged_for_insert;
   bool          sequence;  /* Part of NEXTVAL/CURVAL/LASTVAL */
+  /*
+      Protect single thread from repeating partition auto-create over
+      multiple share instances (as the share is closed on backoff action).
+
+      Skips auto-create only for one given query id.
+  */
+  query_id_t    vers_skip_create;
 
   /*
     Items created by create_view_field and collected to change them in case
diff --git a/sql/transaction.cc b/sql/transaction.cc
index bf8b95353fd..123f9283d23 100644
--- a/sql/transaction.cc
+++ b/sql/transaction.cc
@@ -463,7 +463,7 @@ bool trans_commit_stmt(THD *thd)
     a savepoint for each nested statement, and release the
     savepoint when statement has succeeded.
   */
-  DBUG_ASSERT(! thd->in_sub_stmt);
+  DBUG_ASSERT(!(thd->in_sub_stmt));
 
   thd->merge_unsafe_rollback_flags();