MDEV-23639 Auto-create does not work under LOCK TABLES or inside triggersbb-10.6-midenok-MDEV-17554

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.

7 files changed, 465 insertions, 126 deletions

diff --git a/mysql-test/suite/versioning/r/partition.result b/mysql-test/suite/versioning/r/partition.result
index 80d20f03c2b..522f37bf614 100644
--- a/mysql-test/suite/versioning/r/partition.result
+++ b/mysql-test/suite/versioning/r/partition.result
@@ -1506,16 +1506,14 @@ 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: 1
-Warnings:
-Warning	4114	Versioned table `test`.`t1`: last HISTORY partition (`p1`) is out of LIMIT, need more HISTORY partitions
+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 3
+PARTITIONS 4
 affected rows: 1
 unlock tables;
 affected rows: 0
@@ -1525,7 +1523,7 @@ 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 3
+PARTITIONS 4
 affected rows: 1
 # Overflow prevention under LOCK TABLES
 create or replace table t1 (x int)
@@ -2115,3 +2113,167 @@ t	CREATE TABLE `t` (
  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;
diff --git a/mysql-test/suite/versioning/t/partition.test b/mysql-test/suite/versioning/t/partition.test
index 4b69a024e95..150a7f1ba65 100644
--- a/mysql-test/suite/versioning/t/partition.test
+++ b/mysql-test/suite/versioning/t/partition.test
@@ -1552,4 +1552,97 @@ 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;
+
 --source suite/versioning/common_finish.inc
diff --git a/sql/lock.cc b/sql/lock.cc
index d62a8d49979..77c98510781 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/partition_info.cc b/sql/partition_info.cc
index 06773d03381..e89e8870832 100644
--- a/sql/partition_info.cc
+++ b/sql/partition_info.cc
@@ -910,15 +910,6 @@ bool partition_info::vers_set_hist_part(THD *thd, uint *create_count)
     }
   }
 
-  /*
-     When hist_part is almost full LOCK TABLES may overflow the partition as we
-     can't add new partitions under LOCK TABLES. Reserve one more for this case.
-  */
-  if (auto_hist && *create_count == 0 &&
-      thd->lex->sql_command == SQLCOM_LOCK_TABLES &&
-      vers_info->hist_part->id + 1 == vers_info->now_part->id)
-    ++*create_count;
-
   return false;
 }
 
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index 5db2d67ac87..c575078cb6d 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,
@@ -1622,26 +1621,48 @@ bool is_locked_view(THD *thd, TABLE_LIST *t)
 }
 
 
-bool TABLE::vers_need_hist_part(const THD *thd, const TABLE_LIST *table_list) const
-{
 #ifdef WITH_PARTITION_STORAGE_ENGINE
-  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)
+/**
+  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)
-          break;
-        return true;
+          return false;
+        break;
       case SQLCOM_LOAD:
         if (thd->lex->duplicates != DUP_REPLACE)
-          break;
-        return true;
+          return false;
+        break;
       case SQLCOM_LOCK_TABLES:
       case SQLCOM_DELETE:
       case SQLCOM_UPDATE:
@@ -1649,35 +1670,87 @@ bool TABLE::vers_need_hist_part(const THD *thd, const TABLE_LIST *table_list) co
       case SQLCOM_REPLACE_SELECT:
       case SQLCOM_DELETE_MULTI:
       case SQLCOM_UPDATE_MULTI:
-        return true;
-      default:;
+        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;
     }
-    /*
-      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)
+  TABLE *table= this;
+
+  /*
+      NOTE: The semantics of vers_set_hist_part() is double: even when we
+      don't need auto-create, we need to update part_info->hist_part.
+  */
+  uint *create_count= table_list->vers_skip_create ?
+    NULL : &ot_ctx->vers_create_count;
+  table_list->vers_skip_create= true;
+  if (table->part_info->vers_set_hist_part(thd, create_count))
+  {
+    MYSQL_UNBIND_TABLE(table->file);
+    tc_release_table(table);
+    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)
     {
-      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:
-        return true;
-      default:;
-        break;
-      }
+      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 first 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= false;
+      ot_ctx->vers_create_count= 0;
+      action= Open_table_context::OT_REOPEN_TABLES;
+      table_arg= NULL;
+    }
+    mysql_mutex_unlock(&table->s->LOCK_share);
+    if (!thd->locked_tables_mode)
+    {
+      MYSQL_UNBIND_TABLE(table->file);
+      tc_release_table(table);
+    }
+    ot_ctx->request_backoff_action(action, table_arg);
+    return true;
   }
-#endif
+
   return false;
 }
+#endif /* WITH_PARTITION_STORAGE_ENGINE */
 
 
 /**
@@ -1832,14 +1905,8 @@ 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 (table->vers_need_hist_part(thd, table_list))
-      {
-        /*
-          New partitions are not auto-created under LOCK TABLES (TODO: fix it)
-          but rotation can still happen.
-        */
-        (void) table->part_info->vers_set_hist_part(thd, NULL);
-      }
+      if (table->vers_switch_partition(thd, table_list, ot_ctx))
+        DBUG_RETURN(true);
 #endif
       goto reset;
     }
@@ -2095,54 +2162,10 @@ retry_share:
     tc_add_table(thd, table);
   }
 
-  if (table->vers_need_hist_part(thd, table_list))
-  {
-    /*
-       NOTE: The semantics of vers_set_hist_part() is double: even when we
-       don't need auto-create, we need to update part_info->hist_part.
-    */
-    uint *create_count= table_list->vers_skip_create ?
-      NULL : &ot_ctx->vers_create_count;
-    table_list->vers_skip_create= true;
-    if (table->part_info->vers_set_hist_part(thd, create_count))
-    {
-      MYSQL_UNBIND_TABLE(table->file);
-      tc_release_table(table);
-      DBUG_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 first 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= false;
-        ot_ctx->vers_create_count= 0;
-        action= Open_table_context::OT_REOPEN_TABLES;
-        table_arg= NULL;
-      }
-      mysql_mutex_unlock(&table->s->LOCK_share);
-      MYSQL_UNBIND_TABLE(table->file);
-      tc_release_table(table);
-      ot_ctx->request_backoff_action(action, table_arg);
-      DBUG_RETURN(TRUE);
-    }
-  }
+#ifdef WITH_PARTITION_STORAGE_ENGINE
+  if (table->vers_switch_partition(thd, table_list, ot_ctx))
+    DBUG_RETURN(true);
+#endif /* WITH_PARTITION_STORAGE_ENGINE */
 
   if (!(flags & MYSQL_OPEN_HAS_MDL_LOCK) &&
       table->s->table_category < TABLE_CATEGORY_INFORMATION)
@@ -3291,8 +3314,14 @@ Open_table_context::recover_from_failed_open()
     case OT_DISCOVER:
     case OT_REPAIR:
     case OT_ADD_HISTORY_PARTITION:
-      result= lock_table_names(m_thd, m_thd->lex->create_info, m_failed_table,
-                               NULL, get_timeout(), 0);
+      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
@@ -3320,8 +3349,13 @@ Open_table_context::recover_from_failed_open()
         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)
       {
@@ -3361,12 +3395,56 @@ Open_table_context::recover_from_failed_open()
           }
 
           DBUG_ASSERT(vers_create_count);
-          TABLE_LIST *tl= m_failed_table;
-          result= vers_create_partitions(m_thd, tl, 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);
-          close_tables_for_reopen(m_thd, &tl, start_of_statement_svp());
-          vers_create_count= 0;
+          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;
         }
         case OT_BACKOFF_AND_RETRY:
@@ -5309,9 +5387,6 @@ 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 */
diff --git a/sql/table.h b/sql/table.h
index 113e8a7dc64..8ace358e0da 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
@@ -1768,7 +1769,10 @@ public:
 
   ulonglong vers_start_id() const;
   ulonglong vers_end_id() const;
-  bool vers_need_hist_part(const THD *thd, const TABLE_LIST *table_list) 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);