Merge 10.4 into 10.5

86 files changed, 789 insertions, 629 deletions

diff --git a/README.md b/README.md
index a95f4cf64eb..f17a882d66e 100644
--- a/README.md
+++ b/README.md
@@ -1,7 +1,6 @@
 Code status:
 ------------
 
-* [![Travis CI status](https://secure.travis-ci.org/MariaDB/server.png?branch=10.5)](https://travis-ci.org/MariaDB/server) travis-ci.org (10.5 branch)
 * [![Appveyor CI status](https://ci.appveyor.com/api/projects/status/4u6pexmtpuf8jq66?svg=true)](https://ci.appveyor.com/project/rasmushoj/server) ci.appveyor.com
 
 ## MariaDB: The open source relational database 
@@ -78,5 +77,3 @@ https://mariadb.org/about/security-policy/
 
 The code for MariaDB, including all revision history, can be found at:
 https://github.com/MariaDB/server
-
-***************************************************************************
diff --git a/client/mysqltest.cc b/client/mysqltest.cc
index 4c4e747e69d..3a15972acaf 100644
--- a/client/mysqltest.cc
+++ b/client/mysqltest.cc
@@ -1,5 +1,5 @@
 /* Copyright (c) 2000, 2013, Oracle and/or its affiliates.
-   Copyright (c) 2009, 2020, MariaDB
+   Copyright (c) 2009, 2021, MariaDB
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -91,6 +91,8 @@ static my_bool non_blocking_api_enabled= 0;
 
 #define QUERY_PRINT_ORIGINAL_FLAG 4
 
+#define CLOSED_CONNECTION "-closed_connection-"
+
 #ifndef HAVE_SETENV
 static int setenv(const char *name, const char *value, int overwrite);
 #endif
@@ -5583,11 +5585,13 @@ void do_close_connection(struct st_command *command)
   my_free(con->name);
 
   /*
-    When the connection is closed set name to "-closed_connection-"
+    When the connection is closed set name to CLOSED_CONNECTION
     to make it possible to reuse the connection name.
   */
-  if (!(con->name = my_strdup(PSI_NOT_INSTRUMENTED, "-closed_connection-", MYF(MY_WME))))
+  if (!(con->name = my_strdup(PSI_NOT_INSTRUMENTED, CLOSED_CONNECTION,
+                              MYF(MY_WME))))
     die("Out of memory");
+  con->name_len= sizeof(CLOSED_CONNECTION)-1;
 
   if (con == cur_con)
   {
@@ -5990,7 +5994,7 @@ void do_connect(struct st_command *command)
     con_slot= next_con;
   else
   {
-    if (!(con_slot= find_connection_by_name("-closed_connection-")))
+    if (!(con_slot= find_connection_by_name(CLOSED_CONNECTION)))
       die("Connection limit exhausted, you can have max %d connections",
           opt_max_connections);
     my_free(con_slot->name);
@@ -8603,7 +8607,7 @@ void run_query(struct st_connection *cn, struct st_command *command, int flags)
   log_file.flush();
   dynstr_set(&ds_res, 0);
 
-  if (view_protocol_enabled &&
+  if (view_protocol_enabled && mysql &&
       complete_query &&
       match_re(&view_re, query))
   {
@@ -8649,7 +8653,7 @@ void run_query(struct st_connection *cn, struct st_command *command, int flags)
     dynstr_free(&query_str);
   }
 
-  if (sp_protocol_enabled &&
+  if (sp_protocol_enabled && mysql &&
       complete_query &&
       match_re(&sp_re, query))
   {
@@ -9012,7 +9016,7 @@ static void dump_backtrace(void)
   struct st_connection *conn= cur_con;
 
   fprintf(stderr, "read_command_buf (%p): ", read_command_buf);
-  my_safe_print_str(read_command_buf, sizeof(read_command_buf));
+  fprintf(stderr, "%.*s\n", (int)read_command_buflen, read_command_buf);
   fputc('\n', stderr);
 
   if (conn)
diff --git a/cmake/do_abi_check.cmake b/cmake/do_abi_check.cmake
index d775d310383..02bf9fbe1fb 100644
--- a/cmake/do_abi_check.cmake
+++ b/cmake/do_abi_check.cmake
@@ -75,7 +75,9 @@ FOREACH(file ${ABI_HEADERS})
   FILE(REMOVE ${tmpfile})
   EXECUTE_PROCESS(
     COMMAND diff -w ${file}.pp ${abi_check_out} RESULT_VARIABLE result)
-  IF(NOT ${result} EQUAL 0)
+  IF(result MATCHES "No such file or directory")
+    MESSAGE("Command 'diff' not found. ABI check for ${file} skipped.")
+  ELSEIF(NOT result EQUAL 0)
     IF(ABI_UPDATE)
       EXECUTE_PROCESS(COMMAND mv -v ${abi_check_out} ${file}.pp)
     ELSE(ABI_UPDATE)
@@ -85,4 +87,3 @@ FOREACH(file ${ABI_HEADERS})
   ENDIF()
   FILE(REMOVE ${abi_check_out})
 ENDFOREACH()
-
diff --git a/cmake/os/OpenBSD.cmake b/cmake/os/OpenBSD.cmake
new file mode 100644
index 00000000000..c8b91944275
--- /dev/null
+++ b/cmake/os/OpenBSD.cmake
@@ -0,0 +1,22 @@
+# Copyright (C) 2012 Monty Program Ab, 2021 Brad Smith
+# 
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; version 2 of the License.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1335  USA 
+
+# This file includes OpenBSD specific options and quirks, related to system checks
+
+# Find libexecinfo (library that contains backtrace_symbols etc)
+FIND_LIBRARY(EXECINFO NAMES execinfo)
+IF(EXECINFO)
+ SET(LIBEXECINFO ${EXECINFO})
+ENDIF()
diff --git a/extra/innochecksum.cc b/extra/innochecksum.cc
index 833e66341c7..b2aade70d68 100644
--- a/extra/innochecksum.cc
+++ b/extra/innochecksum.cc
@@ -1,6 +1,6 @@
 /*
    Copyright (c) 2005, 2016, Oracle and/or its affiliates. All rights reserved.
-   Copyright (c) 2014, 2020, MariaDB Corporation.
+   Copyright (c) 2014, 2021, MariaDB Corporation.
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -111,13 +111,10 @@ static ulong			write_check;
 struct innodb_page_type {
 	int n_undo_state_active;
 	int n_undo_state_cached;
-	int n_undo_state_to_free;
 	int n_undo_state_to_purge;
 	int n_undo_state_prepared;
 	int n_undo_state_other;
-	int n_undo_insert;
-	int n_undo_update;
-	int n_undo_other;
+	int n_undo;
 	int n_fil_page_index;
 	int n_fil_page_undo_log;
 	int n_fil_page_inode;
@@ -937,21 +934,7 @@ parse_page(
 			fprintf(file, "#::%llu\t\t|\t\tUndo log page\t\t\t|",
 				cur_page_num);
 		}
-		if (undo_page_type == TRX_UNDO_INSERT) {
-			page_type.n_undo_insert++;
-			if (page_type_dump) {
-				fprintf(file, "\t%s",
-					"Insert Undo log page");
-			}
-
-		} else if (undo_page_type == TRX_UNDO_UPDATE) {
-			page_type.n_undo_update++;
-			if (page_type_dump) {
-				fprintf(file, "\t%s",
-					"Update undo log page");
-			}
-		}
-
+		page_type.n_undo++;
 		undo_page_type = mach_read_from_2(page + TRX_UNDO_SEG_HDR +
 						  TRX_UNDO_STATE);
 		switch (undo_page_type) {
@@ -971,14 +954,6 @@ parse_page(
 				}
 				break;
 
-			case TRX_UNDO_TO_FREE:
-				page_type.n_undo_state_to_free++;
-				if (page_type_dump) {
-					fprintf(file, ", %s", "Insert undo "
-						"segment that can be freed");
-				}
-				break;
-
 			case TRX_UNDO_TO_PURGE:
 				page_type.n_undo_state_to_purge++;
 				if (page_type_dump) {
@@ -1202,15 +1177,11 @@ print_summary(
 
 	fprintf(fil_out, "\n===============================================\n");
 	fprintf(fil_out, "Additional information:\n");
-	fprintf(fil_out, "Undo page type: %d insert, %d update, %d other\n",
-		page_type.n_undo_insert,
-		page_type.n_undo_update,
-		page_type.n_undo_other);
-	fprintf(fil_out, "Undo page state: %d active, %d cached, %d to_free, %d"
+	fprintf(fil_out, "Undo page type: %d\n", page_type.n_undo);
+	fprintf(fil_out, "Undo page state: %d active, %d cached, %d"
 		" to_purge, %d prepared, %d other\n",
 		page_type.n_undo_state_active,
 		page_type.n_undo_state_cached,
-		page_type.n_undo_state_to_free,
 		page_type.n_undo_state_to_purge,
 		page_type.n_undo_state_prepared,
 		page_type.n_undo_state_other);
diff --git a/extra/wolfssl/CMakeLists.txt b/extra/wolfssl/CMakeLists.txt
index c99fb155dd6..908e27734b9 100644
--- a/extra/wolfssl/CMakeLists.txt
+++ b/extra/wolfssl/CMakeLists.txt
@@ -134,6 +134,7 @@ IF(WOLFSSL_X86_64_BUILD)
     SET(USE_INTEL_SPEEDUP 1)
     LIST(APPEND WOLFCRYPT_SOURCES
       ${WOLFCRYPT_SRCDIR}/aes_asm.S
+      ${WOLFCRYPT_SRCDIR}/aes_gcm_asm.S
       ${WOLFCRYPT_SRCDIR}/sha512_asm.S
       ${WOLFCRYPT_SRCDIR}/sha256_asm.S)
     ADD_DEFINITIONS(-maes -msse4.2 -mpclmul)
diff --git a/extra/wolfssl/user_settings.h.in b/extra/wolfssl/user_settings.h.in
index e381e87ce71..4adb27142d9 100644
--- a/extra/wolfssl/user_settings.h.in
+++ b/extra/wolfssl/user_settings.h.in
@@ -17,10 +17,11 @@
 #define WC_RSA_BLINDING
 #define HAVE_TLS_EXTENSIONS
 #define HAVE_AES_ECB
+#define HAVE_AESGCM
 #define WOLFSSL_AES_COUNTER
 #define NO_WOLFSSL_STUB
 #define OPENSSL_ALL
-#undef WOLFSSL_ALLOW_TLSV10 /* see https://github.com/wolfSSL/wolfssl/issues/2960 */
+#define WOLFSSL_ALLOW_TLSV10
 #define NO_OLD_TIMEVAL_NAME
 /*
   FP_MAX_BITS is set high solely to satisfy ssl_8k_key.test
diff --git a/include/mysql/service_my_crypt.h b/include/mysql/service_my_crypt.h
index 930d12a7dd1..2a232117ca1 100644
--- a/include/mysql/service_my_crypt.h
+++ b/include/mysql/service_my_crypt.h
@@ -45,7 +45,7 @@ extern "C" {
 /* The max key length of all supported algorithms */
 #define MY_AES_MAX_KEY_LENGTH 32
 
-#define MY_AES_CTX_SIZE 640
+#define MY_AES_CTX_SIZE 656
 
 enum my_aes_mode {
     MY_AES_ECB, MY_AES_CBC
diff --git a/libmariadb b/libmariadb
-Subproject 180c543704d627a50a52aaf60e24ca14e0ec468
+Subproject 802ce584a26fdc0ba67fcf35e277bf3c7440956
diff --git a/mysql-test/main/connect-abstract.cnf b/mysql-test/main/connect-abstract.cnf
index 5798c4f2f2a..ed7dbd838f0 100644
--- a/mysql-test/main/connect-abstract.cnf
+++ b/mysql-test/main/connect-abstract.cnf
@@ -1,9 +1,6 @@
 
 !include include/default_my.cnf
 
-[mysqld.1]
-socket=       @ENV.ABSTRACT_SOCKET
-
 # Using @OPT.port here for uniqueness
 [ENV]
 ABSTRACT_SOCKET= @mtr-test-abstract-socket-@OPT.port
diff --git a/mysql-test/main/connect-abstract.result b/mysql-test/main/connect-abstract.result
index 68a9674dfaa..8f7c125196a 100644
--- a/mysql-test/main/connect-abstract.result
+++ b/mysql-test/main/connect-abstract.result
@@ -1,5 +1,3 @@
-connect con1,localhost,root,,test,,$ABSTRACT_SOCKET;
 select 1;
 1
 1
-disconnect con1;
diff --git a/mysql-test/main/connect-abstract.test b/mysql-test/main/connect-abstract.test
index 0f212fe5a0d..09bc607e0e8 100644
--- a/mysql-test/main/connect-abstract.test
+++ b/mysql-test/main/connect-abstract.test
@@ -1,6 +1,9 @@
 --source include/linux.inc
 --source include/not_embedded.inc
 
+let $restart_parameters=--socket=$ABSTRACT_SOCKET
+--source include/kill_mysqld.inc
+--source include/start_mysqld.inc
+
 connect(con1,localhost,root,,test,,$ABSTRACT_SOCKET);
 select 1;
-disconnect con1;
diff --git a/mysql-test/main/failed_auth_unixsocket.result b/mysql-test/main/failed_auth_unixsocket.result
index 32357a5a7a9..7e8b8fe70b9 100644
--- a/mysql-test/main/failed_auth_unixsocket.result
+++ b/mysql-test/main/failed_auth_unixsocket.result
@@ -1,7 +1,10 @@
-update mysql.global_priv set priv=json_insert(priv, '$.plugin', 'unix_socket')  where user='root';
+create table global_priv_backup select * from mysql.global_priv;
+update mysql.global_priv set priv=json_insert(priv, '$.plugin', 'unix_socket');
+delete from mysql.global_priv where user != 'root';
 flush privileges;
 connect(localhost,USER,,test,MASTER_PORT,MASTER_SOCKET);
 ERROR 28000: Access denied for user 'USER'@'localhost'
 ERROR 28000: Access denied for user 'USER'@'localhost'
-update mysql.global_priv set priv=json_compact(json_remove(priv, '$.plugin'))  where user='root';
+replace mysql.global_priv select * from global_priv_backup;
 flush privileges;
+drop table global_priv_backup;
diff --git a/mysql-test/main/failed_auth_unixsocket.test b/mysql-test/main/failed_auth_unixsocket.test
index e163a0c230f..fe80d947036 100644
--- a/mysql-test/main/failed_auth_unixsocket.test
+++ b/mysql-test/main/failed_auth_unixsocket.test
@@ -4,7 +4,9 @@
 # MDEV-3909 remote user enumeration
 # unix_socket tests
 #
-update mysql.global_priv set priv=json_insert(priv, '$.plugin', 'unix_socket')  where user='root';
+create table global_priv_backup select * from mysql.global_priv;
+update mysql.global_priv set priv=json_insert(priv, '$.plugin', 'unix_socket');
+delete from mysql.global_priv where user != 'root';
 flush privileges;
 
 # Make sure that the replace works, even if $USER is 'user' or something else
@@ -22,5 +24,6 @@ connect (fail,localhost,$USER);
 --error ER_ACCESS_DENIED_NO_PASSWORD_ERROR
 change_user $USER;
 
-update mysql.global_priv set priv=json_compact(json_remove(priv, '$.plugin'))  where user='root';
+replace mysql.global_priv select * from global_priv_backup;
 flush privileges;
+drop table global_priv_backup;
diff --git a/mysql-test/main/password_expiration_unix_socket.result b/mysql-test/main/password_expiration_unix_socket.result
index 5feee17f205..b37b8868edf 100644
--- a/mysql-test/main/password_expiration_unix_socket.result
+++ b/mysql-test/main/password_expiration_unix_socket.result
@@ -1,8 +1,8 @@
 #
 # A password cannot expire, if there is no password
 #
-create user USER identified via unix_socket;
-alter user USER password expire;
+create user 'USER' identified via unix_socket;
+alter user 'USER' password expire;
 1
 1
-drop user USER;
+drop user 'USER';
diff --git a/mysql-test/main/password_expiration_unix_socket.test b/mysql-test/main/password_expiration_unix_socket.test
index f2579aaf18f..d936d65bc21 100644
--- a/mysql-test/main/password_expiration_unix_socket.test
+++ b/mysql-test/main/password_expiration_unix_socket.test
@@ -9,16 +9,16 @@
 --echo # A password cannot expire, if there is no password
 --echo #
 
---let $replace=create user $USER
---replace_result $replace "create user USER"
---eval create user $USER identified via unix_socket
+--let $replace=create user '$USER'
+--replace_result $replace "create user 'USER'"
+--eval create user '$USER' identified via unix_socket
 
---let $replace=alter user $USER
---replace_result $replace "alter user USER"
---eval alter user $USER password expire
+--let $replace=alter user '$USER'
+--replace_result $replace "alter user 'USER'"
+--eval alter user '$USER' password expire
 
 --exec $MYSQL -u $USER -e 'select 1'
 
---let $replace=drop user $USER
---replace_result $replace "drop user USER"
---eval drop user $USER
+--let $replace=drop user '$USER'
+--replace_result $replace "drop user 'USER'"
+--eval drop user '$USER'
diff --git a/mysql-test/main/win.result b/mysql-test/main/win.result
index 20cea67b2de..c2b3c30011f 100644
--- a/mysql-test/main/win.result
+++ b/mysql-test/main/win.result
@@ -3893,6 +3893,25 @@ id	rn
 1	1
 drop table t1;
 #
+# MDEV-25630: Crash with window function in left expr of IN subquery
+#
+CREATE TABLE t1 (i int);
+INSERT INTO t1 VALUES (1),(2),(3);
+SELECT lag(i) over (ORDER BY i) IN ( SELECT 1 FROM t1 a) FROM t1;
+lag(i) over (ORDER BY i) IN ( SELECT 1 FROM t1 a)
+NULL
+1
+0
+DROP TABLE t1;
+CREATE TABLE t1 (i int);
+INSERT INTO t1 VALUES (1),(2),(3);
+SELECT sum(i) over () IN ( SELECT 1 FROM t1 a) FROM t1;
+sum(i) over () IN ( SELECT 1 FROM t1 a)
+0
+0
+0
+DROP TABLE t1;
+#
 # End of 10.2 tests
 #
 #
diff --git a/mysql-test/main/win.test b/mysql-test/main/win.test
index 0bdfe48a82b..778c685b680 100644
--- a/mysql-test/main/win.test
+++ b/mysql-test/main/win.test
@@ -2542,6 +2542,20 @@ order by rn desc;
 drop table t1;
 
 --echo #
+--echo # MDEV-25630: Crash with window function in left expr of IN subquery
+--echo #
+
+CREATE TABLE t1 (i int);
+INSERT INTO t1 VALUES (1),(2),(3);
+SELECT lag(i) over (ORDER BY i) IN ( SELECT 1 FROM t1 a) FROM t1;
+DROP TABLE t1;
+
+CREATE TABLE t1 (i int);
+INSERT INTO t1 VALUES (1),(2),(3);
+SELECT sum(i) over () IN ( SELECT 1 FROM t1 a) FROM t1;
+DROP TABLE t1;
+
+--echo #
 --echo # End of 10.2 tests
 --echo #
 
diff --git a/mysql-test/main/wolfssl.opt b/mysql-test/main/wolfssl.opt
new file mode 100644
index 00000000000..812dba7bcbd
--- /dev/null
+++ b/mysql-test/main/wolfssl.opt
@@ -0,0 +1 @@
+--ssl_cipher=ECDHE-RSA-AES256-GCM-SHA384
\ No newline at end of file
diff --git a/mysql-test/main/wolfssl.test b/mysql-test/main/wolfssl.test
new file mode 100644
index 00000000000..d9afc43901f
--- /dev/null
+++ b/mysql-test/main/wolfssl.test
@@ -0,0 +1,6 @@
+#
+# Various tests that require WolfSSL
+#
+--source include/have_ssl_communication.inc
+--source include/not_embedded.inc
+SELECT @@ssl_cipher;
diff --git a/mysql-test/suite.pm b/mysql-test/suite.pm
index faab55e9741..86f2d6c0c18 100644
--- a/mysql-test/suite.pm
+++ b/mysql-test/suite.pm
@@ -80,8 +80,6 @@ sub skip_combinations {
   $skip{'main/ssl_verify_ip.test'} = 'x509v3 support required'
     unless $openssl_ver ge "1.0.2";
 
-  $skip{'main/tls_version1.test'} = 'https://github.com/wolfSSL/wolfssl/issues/2960'
-    if $ssl_lib =~ /WolfSSL 4.4.0/;
 
   %skip;
 }
diff --git a/mysql-test/suite/encryption/r/innodb-key-rotation-disable.result b/mysql-test/suite/encryption/r/innodb-key-rotation-disable.result
index 02304fbda17..4e816bea43b 100644
--- a/mysql-test/suite/encryption/r/innodb-key-rotation-disable.result
+++ b/mysql-test/suite/encryption/r/innodb-key-rotation-disable.result
@@ -1,7 +1,3 @@
-SELECT NAME FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION <> 0;
-NAME
-SELECT NAME FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION = 0;
-NAME
 SET GLOBAL innodb_file_per_table = ON;
 set global innodb_compression_algorithm = 1;
 create database enctests;
diff --git a/mysql-test/suite/encryption/r/key_version_rotation.result b/mysql-test/suite/encryption/r/key_version_rotation.result
new file mode 100644
index 00000000000..afc0a0dd122
--- /dev/null
+++ b/mysql-test/suite/encryption/r/key_version_rotation.result
@@ -0,0 +1,20 @@
+create table t1(f1 int not null)engine=innodb;
+create table t2(f1 int not null)engine=innodb;
+insert into t1 select * from seq_1_to_100;
+insert into t2 select * from seq_1_to_100;
+# restart: --innodb_encrypt_tables=0 --innodb_encryption_threads=1 --innodb_encryption_rotate_key_age=9
+# Enable encryption
+set global innodb_encrypt_tables=ON;
+# Create a new table and it is added to rotation list
+create table t3(f1 int not null)engine=innodb;
+insert into t3 select * from seq_1_to_100;
+# Increase the version and it should set rotation
+# variable for the encryption plugin
+set global debug_key_management_version=10;
+select  @@debug_key_management_version;
+@@debug_key_management_version
+10
+# Decrease the key version and Disable the encryption
+set global debug_key_management_version=1;
+set global innodb_encrypt_tables=off;
+DROP TABLE t1, t2, t3;
diff --git a/mysql-test/suite/encryption/r/tempfiles_encrypted.result b/mysql-test/suite/encryption/r/tempfiles_encrypted.result
index dfc4e2a7b4c..977bb0101fd 100644
--- a/mysql-test/suite/encryption/r/tempfiles_encrypted.result
+++ b/mysql-test/suite/encryption/r/tempfiles_encrypted.result
@@ -3899,6 +3899,25 @@ id	rn
 1	1
 drop table t1;
 #
+# MDEV-25630: Crash with window function in left expr of IN subquery
+#
+CREATE TABLE t1 (i int);
+INSERT INTO t1 VALUES (1),(2),(3);
+SELECT lag(i) over (ORDER BY i) IN ( SELECT 1 FROM t1 a) FROM t1;
+lag(i) over (ORDER BY i) IN ( SELECT 1 FROM t1 a)
+NULL
+1
+0
+DROP TABLE t1;
+CREATE TABLE t1 (i int);
+INSERT INTO t1 VALUES (1),(2),(3);
+SELECT sum(i) over () IN ( SELECT 1 FROM t1 a) FROM t1;
+sum(i) over () IN ( SELECT 1 FROM t1 a)
+0
+0
+0
+DROP TABLE t1;
+#
 # End of 10.2 tests
 #
 #
diff --git a/mysql-test/suite/encryption/t/innodb-key-rotation-disable.test b/mysql-test/suite/encryption/t/innodb-key-rotation-disable.test
index dffabaf97f1..96b62f7c05b 100644
--- a/mysql-test/suite/encryption/t/innodb-key-rotation-disable.test
+++ b/mysql-test/suite/encryption/t/innodb-key-rotation-disable.test
@@ -3,9 +3,6 @@
 # not embedded because of restarts
 -- source include/not_embedded.inc
 
-SELECT NAME FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION <> 0;
-SELECT NAME FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION = 0;
-
 let $encryption = `SELECT @@innodb_encrypt_tables`;
 SET GLOBAL innodb_file_per_table = ON;
 # zlib
diff --git a/mysql-test/suite/encryption/t/key_version_rotation.opt b/mysql-test/suite/encryption/t/key_version_rotation.opt
new file mode 100644
index 00000000000..d7933f0f943
--- /dev/null
+++ b/mysql-test/suite/encryption/t/key_version_rotation.opt
@@ -0,0 +1,2 @@
+--innodb-tablespaces-encryption
+--plugin-load-add=$DEBUG_KEY_MANAGEMENT_SO
diff --git a/mysql-test/suite/encryption/t/key_version_rotation.test b/mysql-test/suite/encryption/t/key_version_rotation.test
new file mode 100644
index 00000000000..d36d47251a1
--- /dev/null
+++ b/mysql-test/suite/encryption/t/key_version_rotation.test
@@ -0,0 +1,41 @@
+--source include/have_innodb.inc
+--source include/have_debug.inc
+--source include/have_sequence.inc
+
+create table t1(f1 int not null)engine=innodb;
+create table t2(f1 int not null)engine=innodb;
+insert into t1 select * from seq_1_to_100;
+insert into t2 select * from seq_1_to_100;
+
+let $restart_parameters=--innodb_encrypt_tables=0 --innodb_encryption_threads=1 --innodb_encryption_rotate_key_age=9;
+--source include/restart_mysqld.inc
+
+--echo # Enable encryption
+
+set global innodb_encrypt_tables=ON;
+--let $tables_count= `select count(*) from information_schema.tables where engine = 'InnoDB'`
+--let $wait_timeout= 600
+--let $wait_condition=SELECT COUNT(*) >= $tables_count FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION <> 0;
+--source include/wait_condition.inc
+--echo # Create a new table and it is added to rotation list
+create table t3(f1 int not null)engine=innodb;
+insert into t3 select * from seq_1_to_100;
+
+--echo # Increase the version and it should set rotation
+--echo # variable for the encryption plugin
+
+set global debug_key_management_version=10;
+select  @@debug_key_management_version;
+--let $tables_count= `select count(*) from information_schema.tables where engine = 'InnoDB'`
+--let $wait_timeout= 600
+--let $wait_condition=SELECT COUNT(*) >= $tables_count FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION <> 0;
+--source include/wait_condition.inc
+
+--echo # Decrease the key version and Disable the encryption
+set global debug_key_management_version=1;
+set global innodb_encrypt_tables=off;
+
+--let $wait_timeout= 600
+--let $wait_condition=SELECT COUNT(*) >= $tables_count FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION = 0;
+--source include/wait_condition.inc
+DROP TABLE t1, t2, t3;
diff --git a/mysql-test/suite/gcol/r/innodb_virtual_index.result b/mysql-test/suite/gcol/r/innodb_virtual_index.result
index 70c9d10a68b..2e9b762500d 100644
--- a/mysql-test/suite/gcol/r/innodb_virtual_index.result
+++ b/mysql-test/suite/gcol/r/innodb_virtual_index.result
@@ -262,3 +262,37 @@ CHECK TABLE t1;
 Table	Op	Msg_type	Msg_text
 test.t1	check	status	OK
 DROP TABLE t1;
+#
+# MDEV-25872 InnoDB: Assertion failure in row_merge_read_clustered_index
+#	upon ALTER on table with indexed virtual columns
+#
+CREATE TABLE t1 (
+id BIGINT AUTO_INCREMENT PRIMARY KEY,
+a INT,
+va INT ZEROFILL AS (a) VIRTUAL,
+b TIMESTAMP,
+c CHAR(204),
+vc CHAR(8),
+KEY(vc,c(64),b,va)
+) ENGINE=InnoDB CHARACTER SET utf32;
+INSERT INTO t1 (id) SELECT NULL FROM seq_1_to_75;
+INSERT IGNORE INTO t1 (id, a) VALUES (NULL, -1);
+Warnings:
+Warning	1264	Out of range value for column 'va' at row 1
+ALTER TABLE t1 FORCE;
+ERROR 22003: Out of range value for column 'va' at row 1
+DROP TABLE t1;
+#
+# MDEV-24713 Assertion `dict_table_is_comp(index->table)' failed
+#		in row_merge_buf_add()
+#
+CREATE TABLE t1 (id INT PRIMARY KEY, a CHAR(3),
+b CHAR(8) AS (a) VIRTUAL, KEY(b))
+ROW_FORMAT=REDUNDANT ENGINE=InnoDB
+CHARACTER SET utf8;
+INSERT INTO t1 (id,a) VALUES (1,'foo');
+OPTIMIZE TABLE t1;
+Table	Op	Msg_type	Msg_text
+test.t1	optimize	note	Table does not support optimize, doing recreate + analyze instead
+test.t1	optimize	status	OK
+DROP TABLE t1;
diff --git a/mysql-test/suite/gcol/t/innodb_virtual_index.opt b/mysql-test/suite/gcol/t/innodb_virtual_index.opt
new file mode 100644
index 00000000000..c3f4a891cca
--- /dev/null
+++ b/mysql-test/suite/gcol/t/innodb_virtual_index.opt
@@ -0,0 +1 @@
+--innodb_sort_buffer_size=64k
diff --git a/mysql-test/suite/gcol/t/innodb_virtual_index.test b/mysql-test/suite/gcol/t/innodb_virtual_index.test
index 353841840dc..94c3b7f9204 100644
--- a/mysql-test/suite/gcol/t/innodb_virtual_index.test
+++ b/mysql-test/suite/gcol/t/innodb_virtual_index.test
@@ -1,4 +1,5 @@
 --source include/have_innodb.inc
+--source include/have_sequence.inc
 
 # Ensure that the history list length will actually be decremented by purge.
 SET @saved_frequency = @@GLOBAL.innodb_purge_rseg_truncate_frequency;
@@ -281,3 +282,35 @@ ROLLBACK;
 SELECT * FROM t1;
 CHECK TABLE t1;
 DROP TABLE t1;
+
+--echo #
+--echo # MDEV-25872 InnoDB: Assertion failure in row_merge_read_clustered_index
+--echo #	upon ALTER on table with indexed virtual columns
+--echo #
+
+CREATE TABLE t1 (
+  id BIGINT AUTO_INCREMENT PRIMARY KEY,
+  a INT,
+  va INT ZEROFILL AS (a) VIRTUAL,
+  b TIMESTAMP,
+  c CHAR(204),
+  vc CHAR(8),
+  KEY(vc,c(64),b,va)
+) ENGINE=InnoDB CHARACTER SET utf32;
+INSERT INTO t1 (id) SELECT NULL FROM seq_1_to_75;
+INSERT IGNORE INTO t1 (id, a) VALUES (NULL, -1);
+--error ER_WARN_DATA_OUT_OF_RANGE
+ALTER TABLE t1 FORCE;
+DROP TABLE t1;
+
+--echo #
+--echo # MDEV-24713 Assertion `dict_table_is_comp(index->table)' failed
+--echo #		in row_merge_buf_add()
+--echo #
+CREATE TABLE t1 (id INT PRIMARY KEY, a CHAR(3),
+		 b CHAR(8) AS (a) VIRTUAL, KEY(b))
+		 ROW_FORMAT=REDUNDANT ENGINE=InnoDB
+		 CHARACTER SET utf8;
+INSERT INTO t1 (id,a) VALUES (1,'foo');
+OPTIMIZE TABLE t1;
+DROP TABLE t1;
diff --git a/mysql-test/suite/innodb/r/innodb-alter.result b/mysql-test/suite/innodb/r/innodb-alter.result
index 59abdb01d82..e191c8048fb 100644
--- a/mysql-test/suite/innodb/r/innodb-alter.result
+++ b/mysql-test/suite/innodb/r/innodb-alter.result
@@ -1052,10 +1052,13 @@ a
 10
 DROP TABLE t1;
 CREATE TABLE t1 (a INT NOT NULL DEFAULT 0) ENGINE=InnoDB;
+SET @save_allowed = @@GLOBAL.innodb_instant_alter_column_allowed;
+SET GLOBAL innodb_instant_alter_column_allowed=never;
 iNSERT INTO t1 VALUES (10);
 ALTER TABLE t1 ADD b DATE NOT NULL DEFAULT if(unix_timestamp()>1,TIMESTAMP'2001-01-01 10:20:30',0);
 affected rows: 0
 info: Records: 0  Duplicates: 0  Warnings: 0
+SET GLOBAL innodb_instant_alter_column_allowed=@save_allowed;
 SELECT * FROM t1;
 a	b
 10	2001-01-01
diff --git a/mysql-test/suite/innodb/t/innodb-alter.test b/mysql-test/suite/innodb/t/innodb-alter.test
index 536bf4bff26..555a138b0dc 100644
--- a/mysql-test/suite/innodb/t/innodb-alter.test
+++ b/mysql-test/suite/innodb/t/innodb-alter.test
@@ -656,10 +656,13 @@ DROP TABLE t1;
 
 # DATETIME-to-DATE truncation is OK
 CREATE TABLE t1 (a INT NOT NULL DEFAULT 0) ENGINE=InnoDB;
+SET @save_allowed = @@GLOBAL.innodb_instant_alter_column_allowed;
+SET GLOBAL innodb_instant_alter_column_allowed=never;
 iNSERT INTO t1 VALUES (10);
 --enable_info
 ALTER TABLE t1 ADD b DATE NOT NULL DEFAULT if(unix_timestamp()>1,TIMESTAMP'2001-01-01 10:20:30',0);
 --disable_info
+SET GLOBAL innodb_instant_alter_column_allowed=@save_allowed;
 SELECT * FROM t1;
 DROP TABLE t1;
 
diff --git a/mysql-test/suite/innodb_fts/r/misc_debug.result b/mysql-test/suite/innodb_fts/r/misc_debug.result
index 41f8d08ea7d..9f13822ecda 100644
--- a/mysql-test/suite/innodb_fts/r/misc_debug.result
+++ b/mysql-test/suite/innodb_fts/r/misc_debug.result
@@ -31,19 +31,17 @@ DROP TABLE t2, t1;
 #
 CREATE TABLE t1(a INT, b TEXT, c TEXT, FULLTEXT INDEX(b)) ENGINE=InnoDB;
 connect con1,localhost,root,,test;
-SET DEBUG_SYNC='innodb_inplace_alter_table_enter SIGNAL s1 WAIT_FOR g1';
+SET DEBUG_DBUG="+d,innodb_OOM_inplace_alter";
 SET DEBUG_SYNC='innodb_commit_inplace_alter_table_enter SIGNAL s2 WAIT_FOR g2';
 ALTER TABLE t1 ADD FULLTEXT(c);
 connection default;
-SET DEBUG_SYNC='now WAIT_FOR s1';
-KILL QUERY @id;
-SET DEBUG_SYNC='now SIGNAL g1 WAIT_FOR s2';
+SET DEBUG_SYNC='now WAIT_FOR s2';
 START TRANSACTION;
 SELECT * FROM t1;
 a	b	c
-SET DEBUG_SYNC='now SIGNAL s2';
+SET DEBUG_SYNC='now SIGNAL g2';
 connection con1;
-ERROR 70100: Query execution was interrupted
+ERROR HY000: Out of memory.
 disconnect con1;
 connection default;
 SET DEBUG_SYNC=RESET;
diff --git a/mysql-test/suite/innodb_fts/t/misc_debug.test b/mysql-test/suite/innodb_fts/t/misc_debug.test
index 90cb84976ce..c8542152d4d 100644
--- a/mysql-test/suite/innodb_fts/t/misc_debug.test
+++ b/mysql-test/suite/innodb_fts/t/misc_debug.test
@@ -60,20 +60,16 @@ DROP TABLE t2, t1;
 --echo #
 CREATE TABLE t1(a INT, b TEXT, c TEXT, FULLTEXT INDEX(b)) ENGINE=InnoDB;
 connect(con1,localhost,root,,test);
-let $ID= `SELECT @id := CONNECTION_ID()`;
-SET DEBUG_SYNC='innodb_inplace_alter_table_enter SIGNAL s1 WAIT_FOR g1';
+SET DEBUG_DBUG="+d,innodb_OOM_inplace_alter";
 SET DEBUG_SYNC='innodb_commit_inplace_alter_table_enter SIGNAL s2 WAIT_FOR g2';
 send ALTER TABLE t1 ADD FULLTEXT(c);
 connection default;
-SET DEBUG_SYNC='now WAIT_FOR s1';
-let $ignore= `SELECT @id := $ID`;
-KILL QUERY @id;
-SET DEBUG_SYNC='now SIGNAL g1 WAIT_FOR s2';
+SET DEBUG_SYNC='now WAIT_FOR s2';
 START TRANSACTION;
 SELECT * FROM t1;
-SET DEBUG_SYNC='now SIGNAL s2';
+SET DEBUG_SYNC='now SIGNAL g2';
 connection con1;
---error ER_QUERY_INTERRUPTED
+--error ER_OUT_OF_RESOURCES
 reap;
 disconnect con1;
 connection default;
diff --git a/mysql-test/suite/innodb_zip/r/innochecksum_3.result b/mysql-test/suite/innodb_zip/r/innochecksum_3.result
index cfe1e4852ba..edc71c801c0 100644
--- a/mysql-test/suite/innodb_zip/r/innochecksum_3.result
+++ b/mysql-test/suite/innodb_zip/r/innochecksum_3.result
@@ -112,8 +112,8 @@ File::tab#.ibd
 
 ===============================================
 Additional information:
-Undo page type: # insert, # update, # other
-Undo page state: # active, # cached, # to_free, # to_purge, # prepared, # other
+Undo page type: #
+Undo page state: # active, # cached, # to_purge, # prepared, # other
 index_id	#pages		#leaf_pages	#recs_per_page	#bytes_per_page
 #		#		#		#		#
 #		#		#		#		#
@@ -147,8 +147,8 @@ File::tab#.ibd
 
 ===============================================
 Additional information:
-Undo page type: # insert, # update, # other
-Undo page state: # active, # cached, # to_free, # to_purge, # prepared, # other
+Undo page type: #
+Undo page state: # active, # cached, # to_purge, # prepared, # other
 index_id	#pages		#leaf_pages	#recs_per_page	#bytes_per_page
 #		#		#		#		#
 #		#		#		#		#
diff --git a/mysql-test/suite/mariabackup/auth_plugin_win.result b/mysql-test/suite/mariabackup/auth_plugin_win.result
index 7a623be147f..caf5d8df87d 100644
--- a/mysql-test/suite/mariabackup/auth_plugin_win.result
+++ b/mysql-test/suite/mariabackup/auth_plugin_win.result
@@ -1,5 +1,5 @@
 INSTALL SONAME 'auth_named_pipe';
 CREATE USER 'USERNAME' IDENTIFIED WITH named_pipe;
-GRANT ALL PRIVILEGES ON *.* to USERNAME;
+GRANT ALL PRIVILEGES ON *.* to 'USERNAME';
 DROP USER 'USERNAME';
 UNINSTALL SONAME 'auth_named_pipe';
diff --git a/mysql-test/suite/mariabackup/auth_plugin_win.test b/mysql-test/suite/mariabackup/auth_plugin_win.test
index 9c8cd5ad411..70ae74b7028 100644
--- a/mysql-test/suite/mariabackup/auth_plugin_win.test
+++ b/mysql-test/suite/mariabackup/auth_plugin_win.test
@@ -18,7 +18,7 @@ INSTALL SONAME 'auth_named_pipe';
 --replace_result $USERNAME USERNAME
 eval CREATE USER '$USERNAME' IDENTIFIED WITH named_pipe;
 --replace_result $USERNAME USERNAME
-eval GRANT ALL PRIVILEGES ON *.* to $USERNAME;
+eval GRANT ALL PRIVILEGES ON *.* to '$USERNAME';
 
 let $targetdir=$MYSQLTEST_VARDIR/tmp/backup;
 --disable_result_log
diff --git a/mysql-test/suite/plugins/r/multiauth.result b/mysql-test/suite/plugins/r/multiauth.result
index aed46ac8964..8ae45d1fb9f 100644
--- a/mysql-test/suite/plugins/r/multiauth.result
+++ b/mysql-test/suite/plugins/r/multiauth.result
@@ -1,5 +1,5 @@
 install soname 'auth_ed25519';
-create user USER         identified via unix_socket OR mysql_native_password as password("GOOD");
+create user 'USER'         identified via unix_socket OR mysql_native_password as password("GOOD");
 create user mysqltest1 identified via unix_socket OR mysql_native_password as password("good");
 show create user mysqltest1;
 CREATE USER for mysqltest1@%
@@ -14,8 +14,8 @@ user()	current_user()	database()
 mysqltest1@localhost	mysqltest1@%	test
 # name does not match, password bad = failure
 mysqltest: Could not open connection 'default': 1045 Access denied for user 'mysqltest1'@'localhost' (using password: YES)
-drop user USER, mysqltest1;
-create user USER         identified via mysql_native_password as password("GOOD") OR unix_socket;
+drop user 'USER', mysqltest1;
+create user 'USER'         identified via mysql_native_password as password("GOOD") OR unix_socket;
 create user mysqltest1 identified via mysql_native_password as password("good") OR unix_socket;
 show create user mysqltest1;
 CREATE USER for mysqltest1@%
@@ -30,8 +30,8 @@ user()	current_user()	database()
 mysqltest1@localhost	mysqltest1@%	test
 # name does not match, password bad = failure
 mysqltest: Could not open connection 'default': 1698 Access denied for user 'mysqltest1'@'localhost'
-drop user USER, mysqltest1;
-create user USER         identified via unix_socket OR ed25519 as password("GOOD");
+drop user 'USER', mysqltest1;
+create user 'USER'         identified via unix_socket OR ed25519 as password("GOOD");
 create user mysqltest1 identified via unix_socket OR ed25519 as password("good");
 show create user mysqltest1;
 CREATE USER for mysqltest1@%
@@ -46,8 +46,8 @@ user()	current_user()	database()
 mysqltest1@localhost	mysqltest1@%	test
 # name does not match, password bad = failure
 mysqltest: Could not open connection 'default': 1045 Access denied for user 'mysqltest1'@'localhost' (using password: YES)
-drop user USER, mysqltest1;
-create user USER         identified via ed25519 as password("GOOD") OR unix_socket;
+drop user 'USER', mysqltest1;
+create user 'USER'         identified via ed25519 as password("GOOD") OR unix_socket;
 create user mysqltest1 identified via ed25519 as password("good") OR unix_socket;
 show create user mysqltest1;
 CREATE USER for mysqltest1@%
@@ -62,8 +62,8 @@ user()	current_user()	database()
 mysqltest1@localhost	mysqltest1@%	test
 # name does not match, password bad = failure
 mysqltest: Could not open connection 'default': 1698 Access denied for user 'mysqltest1'@'localhost'
-drop user USER, mysqltest1;
-create user USER         identified via ed25519 as password("GOOD") OR unix_socket OR mysql_native_password as password("works");
+drop user 'USER', mysqltest1;
+create user 'USER'         identified via ed25519 as password("GOOD") OR unix_socket OR mysql_native_password as password("works");
 create user mysqltest1 identified via ed25519 as password("good") OR unix_socket OR mysql_native_password as password("works");
 show create user mysqltest1;
 CREATE USER for mysqltest1@%
@@ -82,7 +82,7 @@ user()	current_user()	database()
 mysqltest1@localhost	mysqltest1@%	test
 # name does not match, password bad = failure
 mysqltest: Could not open connection 'default': 1045 Access denied for user 'mysqltest1'@'localhost' (using password: YES)
-drop user USER, mysqltest1;
+drop user 'USER', mysqltest1;
 create user mysqltest1 identified via mysql_native_password as password("good") OR mysql_native_password as password("works");
 show create user mysqltest1;
 CREATE USER for mysqltest1@%
@@ -157,7 +157,7 @@ drop user mysqltest1;
 create user mysqltest1 identified via ed25519 as password("good") OR unix_socket OR mysql_native_password as password("works");
 ERROR HY000: Column count of mysql.user is wrong. Expected 3, found 47. Created with MariaDB XX.YY.ZZ, now running XX.YY.ZZ. Please use mariadb-upgrade to fix this error
 # switching back from mysql.user to mysql.global_priv
-create user USER         identified via mysql_native_password as '1234567890123456789012345678901234567890a' OR unix_socket;
+create user 'USER'         identified via mysql_native_password as '1234567890123456789012345678901234567890a' OR unix_socket;
 create user mysqltest1 identified via mysql_native_password as '1234567890123456789012345678901234567890a' OR unix_socket;
 update mysql.global_priv set priv=replace(priv, '1234567890123456789012345678901234567890a', 'invalid password');
 flush privileges;
@@ -175,7 +175,7 @@ set password for mysqltest1 = password('bla');
 select user(), current_user(), database();
 user()	current_user()	database()
 mysqltest1@localhost	mysqltest1@%	test
-drop user USER, mysqltest1;
+drop user 'USER', mysqltest1;
 create user mysqltest1 identified via ed25519 as password("good");
 show create user mysqltest1;
 CREATE USER for mysqltest1@%
diff --git a/mysql-test/suite/plugins/r/unix_socket.result b/mysql-test/suite/plugins/r/unix_socket.result
index a725583f201..b663d341221 100644
--- a/mysql-test/suite/plugins/r/unix_socket.result
+++ b/mysql-test/suite/plugins/r/unix_socket.result
@@ -1,7 +1,7 @@
 #
 # with named user
 #
-create user USER identified via unix_socket;
+create user 'USER' identified via unix_socket;
 #
 # name match = ok
 #
@@ -11,7 +11,7 @@ USER@localhost	USER@%	test
 #
 # name does not match = failure
 #
-drop user USER;
+drop user 'USER';
 #
 # and now with anonymous user
 #
diff --git a/mysql-test/suite/plugins/t/multiauth.test b/mysql-test/suite/plugins/t/multiauth.test
index 66f29bdbfec..b1795705691 100644
--- a/mysql-test/suite/plugins/t/multiauth.test
+++ b/mysql-test/suite/plugins/t/multiauth.test
@@ -26,13 +26,13 @@ install soname 'auth_ed25519';
 select user(), current_user(), database();
 EOF
 
---let $creplace=create user $USER
---let $dreplace=drop user $USER
+--let $creplace=create user '$USER'
+--let $dreplace=drop user '$USER'
 
 #
 # socket,password
 #
---replace_result $creplace "create user USER"
+--replace_result $creplace "create user 'USER'"
 eval $creplace         identified via unix_socket OR mysql_native_password as password("GOOD");
 create user mysqltest1 identified via unix_socket OR mysql_native_password as password("good");
 show create user mysqltest1;
@@ -43,13 +43,13 @@ show create user mysqltest1;
 --echo # name does not match, password bad = failure
 --error 1
 --exec $try_auth -u mysqltest1 -pbad
---replace_result $dreplace "drop user USER"
+--replace_result $dreplace "drop user 'USER'"
 eval $dreplace, mysqltest1;
 
 #
 # password,socket
 #
---replace_result $creplace "create user USER"
+--replace_result $creplace "create user 'USER'"
 eval $creplace         identified via mysql_native_password as password("GOOD") OR unix_socket;
 create user mysqltest1 identified via mysql_native_password as password("good") OR unix_socket;
 show create user mysqltest1;
@@ -60,13 +60,13 @@ show create user mysqltest1;
 --echo # name does not match, password bad = failure
 --error 1
 --exec $try_auth -u mysqltest1 -pbad
---replace_result $dreplace "drop user USER"
+--replace_result $dreplace "drop user 'USER'"
 eval $dreplace, mysqltest1;
 
 #
 # socket,ed25519
 #
---replace_result $creplace "create user USER"
+--replace_result $creplace "create user 'USER'"
 eval $creplace         identified via unix_socket OR ed25519 as password("GOOD");
 create user mysqltest1 identified via unix_socket OR ed25519 as password("good");
 show create user mysqltest1;
@@ -77,13 +77,13 @@ show create user mysqltest1;
 --echo # name does not match, password bad = failure
 --error 1
 --exec $try_auth -u mysqltest1 -pbad
---replace_result $dreplace "drop user USER"
+--replace_result $dreplace "drop user 'USER'"
 eval $dreplace, mysqltest1;
 
 #
 # ed25519,socket
 #
---replace_result $creplace "create user USER"
+--replace_result $creplace "create user 'USER'"
 eval $creplace         identified via ed25519 as password("GOOD") OR unix_socket;
 create user mysqltest1 identified via ed25519 as password("good") OR unix_socket;
 show create user mysqltest1;
@@ -94,13 +94,13 @@ show create user mysqltest1;
 --echo # name does not match, password bad = failure
 --error 1
 --exec $try_auth -u mysqltest1 -pbad
---replace_result $dreplace "drop user USER"
+--replace_result $dreplace "drop user 'USER'"
 eval $dreplace, mysqltest1;
 
 #
 # ed25519,socket,password
 #
---replace_result $creplace "create user USER"
+--replace_result $creplace "create user 'USER'"
 eval $creplace         identified via ed25519 as password("GOOD") OR unix_socket OR mysql_native_password as password("works");
 create user mysqltest1 identified via ed25519 as password("good") OR unix_socket OR mysql_native_password as password("works");
 show create user mysqltest1;
@@ -113,7 +113,7 @@ show create user mysqltest1;
 --echo # name does not match, password bad = failure
 --error 1
 --exec $try_auth -u mysqltest1 -pbad
---replace_result $dreplace "drop user USER"
+--replace_result $dreplace "drop user 'USER'"
 eval $dreplace, mysqltest1;
 
 #
@@ -162,7 +162,7 @@ create user mysqltest1 identified via ed25519 as password("good") OR unix_socket
 #
 # invalid password,socket
 #
---replace_result $creplace "create user USER"
+--replace_result $creplace "create user 'USER'"
 eval $creplace         identified via mysql_native_password as '1234567890123456789012345678901234567890a' OR unix_socket;
 create user mysqltest1 identified via mysql_native_password as '1234567890123456789012345678901234567890a' OR unix_socket;
 update mysql.global_priv set priv=replace(priv, '1234567890123456789012345678901234567890a', 'invalid password');
@@ -176,7 +176,7 @@ show create user mysqltest1;
 --echo # SET PASSWORD helps
 set password for mysqltest1 = password('bla');
 --exec $try_auth -u mysqltest1 -pbla
---replace_result $dreplace "drop user USER"
+--replace_result $dreplace "drop user 'USER'"
 eval $dreplace, mysqltest1;
 
 #
diff --git a/mysql-test/suite/plugins/t/unix_socket.test b/mysql-test/suite/plugins/t/unix_socket.test
index 9bb56aae290..be2afb0ca66 100644
--- a/mysql-test/suite/plugins/t/unix_socket.test
+++ b/mysql-test/suite/plugins/t/unix_socket.test
@@ -4,9 +4,9 @@
 --echo # with named user
 --echo #
 
---let $replace=create user $USER
---replace_result $replace "create user USER"
-eval create user $USER identified via unix_socket;
+--let $replace=create user '$USER'
+--replace_result $replace "create user 'USER'"
+eval create user '$USER' identified via unix_socket;
 
 --write_file $MYSQLTEST_VARDIR/tmp/peercred_test.txt
 --let $replace1=$USER@localhost
@@ -26,9 +26,9 @@ EOF
 --error 1
 --exec $MYSQL_TEST -u foobar < $MYSQLTEST_VARDIR/tmp/peercred_test.txt
 
---let $replace=drop user $USER
---replace_result $replace "drop user USER"
-eval drop user $USER;
+--let $replace=drop user '$USER'
+--replace_result $replace "drop user 'USER'"
+eval drop user '$USER';
 
 --echo #
 --echo # and now with anonymous user
diff --git a/mysys/mf_iocache.c b/mysys/mf_iocache.c
index 2d6031fa8cd..3eaf15803b6 100644
--- a/mysys/mf_iocache.c
+++ b/mysys/mf_iocache.c
@@ -1349,7 +1349,7 @@ static void copy_to_read_buffer(IO_CACHE *write_cache,
 
 static int _my_b_seq_read(IO_CACHE *info, uchar *Buffer, size_t Count)
 {
-  size_t length, diff_length, left_length= 0, save_count, max_length;
+  size_t length, diff_length, save_count, max_length;
   my_off_t pos_in_file;
   save_count=Count;
 
@@ -1400,7 +1400,6 @@ static int _my_b_seq_read(IO_CACHE *info, uchar *Buffer, size_t Count)
       */
       goto read_append_buffer;
     }
-    left_length+=length;
     diff_length=0;
   }
 
diff --git a/scripts/wsrep_sst_common.sh b/scripts/wsrep_sst_common.sh
index 05944ef6035..4dedecb439f 100644
--- a/scripts/wsrep_sst_common.sh
+++ b/scripts/wsrep_sst_common.sh
@@ -1190,7 +1190,6 @@ trim_string()
 check_pid()
 {
     local pid_file="$1"
-    local remove=${2:-0}
     if [ -r "$pid_file" ]; then
         local pid=$(cat "$pid_file" 2>/dev/null)
         if [ -n "$pid" ]; then
@@ -1201,6 +1200,7 @@ check_pid()
                 fi
             fi
         fi
+        local remove=${2:-0}
         if [ $remove -eq 1 ]; then
             rm -f "$pid_file"
         fi
@@ -1223,7 +1223,7 @@ check_pid()
 #
 cleanup_pid()
 {
-    local pid="$1"
+    local pid=$1
     local pid_file="${2:-}"
     local config="${3:-}"
 
@@ -1241,8 +1241,9 @@ cleanup_pid()
                            round=8
                            force=1
                            kill -9 $pid >/dev/null 2>&1
+                           sleep 0.5
                        else
-                           return 1;
+                           return 1
                        fi
                    fi
                 done
@@ -1254,7 +1255,7 @@ cleanup_pid()
     fi
 
     [ -n "$pid_file" ] && [ -f "$pid_file" ] && rm -f "$pid_file"
-    [ -n "$config" ]   && [ -f "$config"   ] && rm -f "$config"
+    [ -n "$config" ]   && [ -f "$config" ]   && rm -f "$config"
 
     return 0
 }
diff --git a/scripts/wsrep_sst_mariabackup.sh b/scripts/wsrep_sst_mariabackup.sh
index 7f97d9e8dea..339a8fcf4a5 100644
--- a/scripts/wsrep_sst_mariabackup.sh
+++ b/scripts/wsrep_sst_mariabackup.sh
@@ -741,15 +741,15 @@ recv_joiner()
         fi
 
         # check donor supplied secret
-        SECRET=$(grep -- "$SECRET_TAG " "$MAGIC_FILE" 2>/dev/null | cut -d ' ' -f 2)
+        SECRET=$(grep -F -- "$SECRET_TAG " "$MAGIC_FILE" 2>/dev/null | cut -d ' ' -f 2)
         if [ "$SECRET" != "$MY_SECRET" ]; then
             wsrep_log_error "Donor does not know my secret!"
             wsrep_log_info "Donor:'$SECRET', my:'$MY_SECRET'"
             exit 32
         fi
 
-        # remove secret from magic file
-        grep -v -- "$SECRET_TAG " "$MAGIC_FILE" > "$MAGIC_FILE.new"
+        # remove secret from the magic file
+        grep -v -F -- "$SECRET_TAG " "$MAGIC_FILE" > "$MAGIC_FILE.new"
         mv "$MAGIC_FILE.new" "$MAGIC_FILE"
     fi
 }
diff --git a/scripts/wsrep_sst_rsync.sh b/scripts/wsrep_sst_rsync.sh
index 19a4d19fded..fc9f5017937 100644
--- a/scripts/wsrep_sst_rsync.sh
+++ b/scripts/wsrep_sst_rsync.sh
@@ -68,6 +68,8 @@ cleanup_joiner()
     if [ "$WSREP_SST_OPT_ROLE" = 'joiner' ]; then
         wsrep_cleanup_progress_file
     fi
+
+    [ -f "$SST_PID" ] && rm -f "$SST_PID"
 }
 
 check_pid_and_port()
@@ -281,6 +283,7 @@ then
     *)
         wsrep_log_error "Unrecognized ssl-mode option: '$SSLMODE'"
         exit 22 # EINVAL
+        ;;
     esac
     if [ -z "$CAFILE_OPT" ]; then
         wsrep_log_error "Can't have ssl-mode='$SSLMODE' without CA file"
@@ -426,7 +429,7 @@ EOF
             exit 255 # unknown error
         fi
 
-        # second, we transfer InnoDB log files
+        # second, we transfer InnoDB and Aria log files
         rsync ${STUNNEL:+--rsh="$STUNNEL"} \
               --owner --group --perms --links --specials \
               --ignore-times --inplace --dirs --delete --quiet \
@@ -499,7 +502,22 @@ elif [ "$WSREP_SST_OPT_ROLE" = 'joiner' ]
 then
     check_sockets_utils
 
-    # give some time for lingering stunnel from previous SST to complete
+    SST_PID="$WSREP_SST_OPT_DATA/wsrep_rsync_sst.pid"
+
+    # give some time for previous SST to complete:
+    check_round=0
+    while check_pid "$SST_PID" 0
+    do
+        wsrep_log_info "previous SST is not completed, waiting for it to exit"
+        check_round=$(( check_round + 1 ))
+        if [ $check_round -eq 10 ]; then
+            wsrep_log_error "previous SST script still running."
+            exit 114 # EALREADY
+        fi
+        sleep 1
+    done
+
+    # give some time for stunnel from the previous SST to complete:
     check_round=0
     while check_pid "$STUNNEL_PID" 1
     do
@@ -516,7 +534,7 @@ then
     RSYNC_PID="$WSREP_SST_OPT_DATA/$MODULE.pid"
     RSYNC_CONF="$WSREP_SST_OPT_DATA/$MODULE.conf"
 
-    # give some time for lingering rsync from previous SST to complete
+    # give some time for rsync from the previous SST to complete:
     check_round=0
     while check_pid "$RSYNC_PID" 1
     do
@@ -583,12 +601,14 @@ EOF
         RSYNC_ADDR="*"
     fi
 
+    echo $$ > "$SST_PID"
+
     if [ -z "$STUNNEL" ]
     then
         rsync --daemon --no-detach --port "$RSYNC_PORT" --config "$RSYNC_CONF" $RSYNC_EXTRA_ARGS &
         RSYNC_REAL_PID=$!
-        TRANSFER_REAL_PID="$RSYNC_REAL_PID"
-        TRANSFER_PID=$RSYNC_PID
+        TRANSFER_REAL_PID=$RSYNC_REAL_PID
+        TRANSFER_PID="$RSYNC_PID"
     else
         # Let's check if the path to the config file contains a space?
         if [ "${RSYNC_CONF#* }" = "$RSYNC_CONF" ]; then
@@ -631,8 +651,8 @@ EOF
         fi
         stunnel "$STUNNEL_CONF" &
         STUNNEL_REAL_PID=$!
-        TRANSFER_REAL_PID="$STUNNEL_REAL_PID"
-        TRANSFER_PID=$STUNNEL_PID
+        TRANSFER_REAL_PID=$STUNNEL_REAL_PID
+        TRANSFER_PID="$STUNNEL_PID"
     fi
 
     if [ "${SSLMODE#VERIFY}" != "$SSLMODE" ]
@@ -691,35 +711,49 @@ EOF
 
             # Clean up old binlog files first
             rm -f "$BINLOG_FILENAME".[0-9]*
-            [ -f "$binlog_index" ] && rm "$binlog_index"
+            [ -f "$binlog_index" ] && rm -f "$binlog_index"
+
+            # Create a temporary file:
+            tmpdir=$(parse_cnf '--mysqld|sst' 'tmpdir')
+            if [ -z "$tmpdir" ]; then
+               tmpfile="$(mktemp)"
+            else
+               tmpfile=$(mktemp "--tmpdir=$tmpdir")
+            fi
 
             wsrep_log_info "Extracting binlog files:"
-            tar -xvf "$BINLOG_TAR_FILE" >> _binlog_tmp_files_$!
+            if ! tar -xvf "$BINLOG_TAR_FILE" > "$tmpfile"; then
+                wsrep_log_error "Error unpacking tar file with binlog files"
+                rm -f "$tmpfile"
+                exit 32
+            fi
+
+            # Rebuild binlog index:
             while read bin_file; do
                 echo "$BINLOG_DIRNAME/$bin_file" >> "$binlog_index"
-            done < _binlog_tmp_files_$!
-            rm -f _binlog_tmp_files_$!
+            done < "$tmpfile"
+            rm -f "$tmpfile"
 
             cd "$OLD_PWD"
         fi
     fi
 
-    if [ -r "$MAGIC_FILE" ]
-    then
-        # check donor supplied secret
-        SECRET=$(grep -F -- "$SECRET_TAG " "$MAGIC_FILE" 2>/dev/null | cut -d ' ' -f 2)
-        if [ "$SECRET" != "$MY_SECRET" ]; then
-            wsrep_log_error "Donor does not know my secret!"
-            wsrep_log_info "Donor:'$SECRET', my:'$MY_SECRET'"
-            exit 32
+    if [ -r "$MAGIC_FILE" ]; then
+        if [ -n "$MY_SECRET" ]; then
+            # check donor supplied secret
+            SECRET=$(grep -F -- "$SECRET_TAG " "$MAGIC_FILE" 2>/dev/null | cut -d ' ' -f 2)
+            if [ "$SECRET" != "$MY_SECRET" ]; then
+                wsrep_log_error "Donor does not know my secret!"
+                wsrep_log_info "Donor:'$SECRET', my:'$MY_SECRET'"
+                exit 32
+            fi
+            # remove secret from the magic file, and output
+            # the UUID:seqno & wsrep_gtid_domain_id:
+            grep -v -F -- "$SECRET_TAG " "$MAGIC_FILE"
+        else
+            # Output the UUID:seqno and wsrep_gtid_domain_id:
+            cat "$MAGIC_FILE"
         fi
-
-        # remove secret from magic file
-        grep -v -F -- "$SECRET_TAG " "$MAGIC_FILE" > "$MAGIC_FILE.new"
-
-        mv "$MAGIC_FILE.new" "$MAGIC_FILE"
-        # UUID:seqno & wsrep_gtid_domain_id is received here.
-        cat "$MAGIC_FILE" # Output : UUID:seqno wsrep_gtid_domain_id
     else
         # this message should cause joiner to abort
         echo "rsync process ended without creating '$MAGIC_FILE'"
diff --git a/sql/event_scheduler.cc b/sql/event_scheduler.cc
index 78802a5e109..9015c1b2655 100644
--- a/sql/event_scheduler.cc
+++ b/sql/event_scheduler.cc
@@ -448,7 +448,7 @@ Event_scheduler::start(int *err_no)
     scheduler_thd= NULL;
     deinit_event_thread(new_thd);
 
-    delete scheduler_param_value;
+    my_free(scheduler_param_value);
     ret= true;
   }
 
diff --git a/sql/item.cc b/sql/item.cc
index 2658c443f4d..6e45c6a8086 100644
--- a/sql/item.cc
+++ b/sql/item.cc
@@ -8544,6 +8544,7 @@ Item_cache_wrapper::Item_cache_wrapper(THD *thd, Item *item_arg):
   with_field= orig_item->with_field;
   name= item_arg->name;
   m_with_subquery= orig_item->with_subquery();
+  with_window_func= orig_item->with_window_func;
 
   if ((expr_value= orig_item->get_cache(thd)))
     expr_value->setup(thd, orig_item);
diff --git a/sql/item_cmpfunc.cc b/sql/item_cmpfunc.cc
index 97c41373d12..389f012a016 100644
--- a/sql/item_cmpfunc.cc
+++ b/sql/item_cmpfunc.cc
@@ -1,5 +1,5 @@
 /* Copyright (c) 2000, 2013, Oracle and/or its affiliates.
-   Copyright (c) 2009, 2020, MariaDB
+   Copyright (c) 2009, 2021, MariaDB
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -1391,6 +1391,9 @@ bool Item_in_optimizer::fix_fields(THD *thd, Item **ref)
     maybe_null=1;
   m_with_subquery= true;
   join_with_sum_func(args[1]);
+  with_window_func= args[0]->with_window_func;
+  // The subquery cannot have window functions aggregated in this select
+  DBUG_ASSERT(!args[1]->with_window_func);
   with_field= with_field || args[1]->with_field;
   with_param= args[0]->with_param || args[1]->with_param; 
   used_tables_and_const_cache_join(args[1]);
diff --git a/sql/mdl.cc b/sql/mdl.cc
index e83ef151554..0349d147023 100644
--- a/sql/mdl.cc
+++ b/sql/mdl.cc
@@ -241,6 +241,35 @@ private:
   static const uint MAX_SEARCH_DEPTH= 32;
 };
 
+#ifndef DBUG_OFF
+
+/*
+  Print a list of all locks to DBUG trace to help with debugging
+*/
+
+static int mdl_dbug_print_lock(MDL_ticket *mdl_ticket, void *arg, bool granted)
+{
+  String *tmp= (String*) arg;
+  char buffer[128];
+  MDL_key *mdl_key= mdl_ticket->get_key();
+  size_t length;
+  length= my_snprintf(buffer, sizeof(buffer)-1,
+                      "\nname: %s  db: %.*s  key_name: %.*s (%s)",
+                      mdl_ticket->get_type_name()->str,
+                      (int) mdl_key->db_name_length(), mdl_key->db_name(),
+                      (int) mdl_key->name_length(),    mdl_key->name(),
+                      granted ? "granted" : "waiting");
+  tmp->append(buffer, length);
+  return 0;
+}
+
+const char *mdl_dbug_print_locks()
+{
+  static String tmp;
+  mdl_iterate(mdl_dbug_print_lock, (void*) &tmp);
+  return tmp.c_ptr();
+}
+#endif /* DBUG_OFF */
 
 /**
   Enter a node of a wait-for graph. After
@@ -2368,7 +2397,9 @@ MDL_context::acquire_lock(MDL_request *mdl_request, double lock_wait_timeout)
     switch (wait_status)
     {
     case MDL_wait::VICTIM:
-      mdl_dbug_print_locks();
+      DBUG_LOCK_FILE;
+      DBUG_PRINT("mdl_locks", ("%s", mdl_dbug_print_locks()));
+      DBUG_UNLOCK_FILE;
       my_error(ER_LOCK_DEADLOCK, MYF(0));
       break;
     case MDL_wait::TIMEOUT:
@@ -3247,34 +3278,3 @@ void MDL_ticket::wsrep_report(bool debug) const
               psi_stage->m_name);
 }
 #endif /* WITH_WSREP */
-
-
-#ifndef DBUG_OFF
-
-/*
-  Print a list of all locks to DBUG trace to help with debugging
-*/
-
-static int mdl_dbug_print_lock(MDL_ticket *mdl_ticket, void *arg, bool granted)
-{
-  String *tmp= (String*) arg;
-  char buffer[128];
-  MDL_key *mdl_key= mdl_ticket->get_key();
-  size_t length;
-  length= my_snprintf(buffer, sizeof(buffer)-1,
-                      "\nname: %s  db: %.*s  key_name: %.*s (%s)",
-                      mdl_ticket->get_type_name()->str,
-                      (int) mdl_key->db_name_length(), mdl_key->db_name(),
-                      (int) mdl_key->name_length(),    mdl_key->name(),
-                      granted ? "granted" : "waiting");
-  tmp->append(buffer, length);
-  return 0;
-}
-
-void mdl_dbug_print_locks()
-{
-  String tmp;
-  mdl_iterate(mdl_dbug_print_lock, (void*) &tmp);
-  DBUG_PRINT("mdl_locks", ("%s", tmp.c_ptr()));
-}
-#endif /* DBUG_OFF */
diff --git a/sql/mdl.h b/sql/mdl.h
index 52c48768329..31ac4e81377 100644
--- a/sql/mdl.h
+++ b/sql/mdl.h
@@ -1130,9 +1130,4 @@ typedef int (*mdl_iterator_callback)(MDL_ticket *ticket, void *arg,
                                      bool granted);
 extern MYSQL_PLUGIN_IMPORT
 int mdl_iterate(mdl_iterator_callback callback, void *arg);
-#ifndef DBUG_OFF
-void mdl_dbug_print_locks();
-#else
-  static inline void mdl_dbug_print_locks() {}
-#endif /* DBUG_OFF */
 #endif /* MDL_H */
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index 80017da803d..8800fbf5b6d 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -4935,7 +4935,7 @@ make_join_statistics(JOIN *join, List<TABLE_LIST> &tables_list,
   int error= 0;
   TABLE *UNINIT_VAR(table); /* inited in all loops */
   uint i,table_count,const_count,key;
-  table_map found_const_table_map, all_table_map, found_ref, refs;
+  table_map found_const_table_map, all_table_map;
   key_map const_ref, eq_part;
   bool has_expensive_keyparts;
   TABLE **table_vector;
@@ -5201,7 +5201,6 @@ make_join_statistics(JOIN *join, List<TABLE_LIST> &tables_list,
   {
     ref_changed = 0;
   more_const_tables_found:
-    found_ref=0;
 
     /*
       We only have to loop from stat_vector + const_count as
@@ -5291,7 +5290,6 @@ make_join_statistics(JOIN *join, List<TABLE_LIST> &tables_list,
 	  key=keyuse->key;
 	  s->keys.set_bit(key);               // TODO: remove this ?
 
-	  refs=0;
           const_ref.clear_all();
 	  eq_part.clear_all();
           has_expensive_keyparts= false;
@@ -5307,8 +5305,6 @@ make_join_statistics(JOIN *join, List<TABLE_LIST> &tables_list,
                 if (keyuse->val->is_expensive())
                   has_expensive_keyparts= true;
               }
-	      else
-		refs|=keyuse->used_tables;
 	      eq_part.set_bit(keyuse->keypart);
 	    }
 	    keyuse++;
@@ -5360,8 +5356,6 @@ make_join_statistics(JOIN *join, List<TABLE_LIST> &tables_list,
 		  found_const_table_map|= table->map;
 	        break;
 	      }
-	      else
-	        found_ref|= refs;      // Table is const if all refs are const
 	    }
             else if (base_const_ref == base_eq_part)
               s->const_keys.set_bit(key);
@@ -28324,7 +28318,6 @@ static bool get_range_limit_read_cost(const JOIN_TAB *tab,
     */
     if (tab)
     {
-      key_part_map const_parts= 0;
       key_part_map map= 1;
       uint kp;
       /* Find how many key parts would be used by ref(const) */
@@ -28332,7 +28325,6 @@ static bool get_range_limit_read_cost(const JOIN_TAB *tab,
       {
         if (!(table->const_key_parts[keynr] & map))
           break;
-        const_parts |= map;
       }
       
       if (kp > 0)
diff --git a/storage/innobase/fil/fil0crypt.cc b/storage/innobase/fil/fil0crypt.cc
index ca66544ffa6..c4882304206 100644
--- a/storage/innobase/fil/fil0crypt.cc
+++ b/storage/innobase/fil/fil0crypt.cc
@@ -55,6 +55,9 @@ UNIV_INTERN uint srv_n_fil_crypt_threads_started = 0;
 /** At this age or older a space/page will be rotated */
 UNIV_INTERN uint srv_fil_crypt_rotate_key_age;
 
+/** Whether the encryption plugin does key rotation */
+static bool srv_encrypt_rotate;
+
 /** Event to signal FROM the key rotation threads. */
 static os_event_t fil_crypt_event;
 
@@ -128,6 +131,14 @@ fil_space_crypt_t::key_get_latest_version(void)
 
 	if (is_key_found()) {
 		key_version = encryption_key_get_latest_version(key_id);
+		/* InnoDB does dirty read of srv_fil_crypt_rotate_key_age.
+		It doesn't matter because srv_encrypt_rotate
+		can be set to true only once */
+		if (!srv_encrypt_rotate
+		    && key_version > srv_fil_crypt_rotate_key_age) {
+			srv_encrypt_rotate = true;
+		}
+
 		srv_stats.n_key_requests.inc();
 		key_found = key_version;
 	}
@@ -1453,47 +1464,65 @@ inline bool fil_space_t::acquire_if_not_stopped()
   return UNIV_LIKELY(!(n & CLOSING)) || prepare(true);
 }
 
-/** Return the next tablespace from rotation_list.
-@param space   previous tablespace (NULL to start from the start)
+bool fil_crypt_must_default_encrypt()
+{
+  return !srv_fil_crypt_rotate_key_age || !srv_encrypt_rotate;
+}
+
+/** Return the next tablespace from default_encrypt_tables list.
+@param space   previous tablespace (nullptr to start from the start)
 @param recheck whether the removal condition needs to be rechecked after
 the encryption parameters were changed
 @param encrypt expected state of innodb_encrypt_tables
 @return the next tablespace to process (n_pending_ops incremented)
 @retval fil_system.temp_space if there is no work to do
 @retval nullptr upon reaching the end of the iteration */
-inline fil_space_t *fil_system_t::keyrotate_next(fil_space_t *space,
-                                                 bool recheck, bool encrypt)
+inline fil_space_t *fil_system_t::default_encrypt_next(fil_space_t *space,
+                                                       bool recheck,
+                                                       bool encrypt)
 {
   ut_ad(mutex_own(&mutex));
 
   sized_ilist<fil_space_t, rotation_list_tag_t>::iterator it=
-    space && space->is_in_rotation_list ? space : rotation_list.begin();
+    space && space->is_in_default_encrypt
+    ? space
+    : default_encrypt_tables.begin();
   const sized_ilist<fil_space_t, rotation_list_tag_t>::iterator end=
-    rotation_list.end();
+    default_encrypt_tables.end();
 
   if (space)
   {
     const bool released= !space->release();
 
-    if (space->is_in_rotation_list)
+    if (space->is_in_default_encrypt)
     {
       while (++it != end &&
              (!UT_LIST_GET_LEN(it->chain) || it->is_stopping()));
 
-      /* If one of the encryption threads already started the encryption
-      of the table then don't remove the unencrypted spaces from rotation list
+      /* If one of the encryption threads already started
+      the encryption of the table then don't remove the
+      unencrypted spaces from default encrypt list.
 
-      If there is a change in innodb_encrypt_tables variables value then
-      don't remove the last processed tablespace from the rotation list. */
+      If there is a change in innodb_encrypt_tables variables
+      value then don't remove the last processed tablespace
+      from the default encrypt list. */
       if (released && (!recheck || space->crypt_data) &&
           !encrypt == !srv_encrypt_tables)
       {
-        ut_a(!rotation_list.empty());
-        rotation_list.remove(*space);
-        space->is_in_rotation_list= false;
+        ut_a(!default_encrypt_tables.empty());
+        default_encrypt_tables.remove(*space);
+        space->is_in_default_encrypt= false;
       }
     }
   }
+  else while (it != end &&
+	      (!UT_LIST_GET_LEN(it->chain) || it->is_stopping()))
+  {
+    /* Find the next suitable default encrypt table if
+    beginning of default_encrypt_tables list has been scheduled
+    to be deleted */
+    it++;
+  }
 
   if (it == end)
     return temp_space;
@@ -1524,8 +1553,8 @@ inline fil_space_t *fil_space_t::next(fil_space_t *space, bool recheck,
 {
   mutex_enter(&fil_system.mutex);
 
-  if (!srv_fil_crypt_rotate_key_age)
-    space= fil_system.keyrotate_next(space, recheck, encrypt);
+  if (fil_crypt_must_default_encrypt())
+    space= fil_system.default_encrypt_next(space, recheck, encrypt);
   else
   {
     if (!space)
@@ -2271,9 +2300,9 @@ fil_crypt_set_thread_cnt(
 	}
 }
 
-/** Initialize the tablespace rotation_list
+/** Initialize the tablespace default_encrypt_tables
 if innodb_encryption_rotate_key_age=0. */
-static void fil_crypt_rotation_list_fill()
+static void fil_crypt_default_encrypt_tables_fill()
 {
 	ut_ad(mutex_own(&fil_system.mutex));
 
@@ -2281,7 +2310,7 @@ static void fil_crypt_rotation_list_fill()
 	     space != NULL;
 	     space = UT_LIST_GET_NEXT(space_list, space)) {
 		if (space->purpose != FIL_TYPE_TABLESPACE
-		    || space->is_in_rotation_list
+		    || space->is_in_default_encrypt
 		    || UT_LIST_GET_LEN(space->chain) == 0
 		    || !space->acquire_if_not_stopped()) {
 			continue;
@@ -2312,8 +2341,8 @@ static void fil_crypt_rotation_list_fill()
 			}
 		}
 
-		fil_system.rotation_list.push_back(*space);
-		space->is_in_rotation_list = true;
+		fil_system.default_encrypt_tables.push_back(*space);
+		space->is_in_default_encrypt = true;
 next:
 		space->release();
 	}
@@ -2330,7 +2359,7 @@ fil_crypt_set_rotate_key_age(
 	mutex_enter(&fil_system.mutex);
 	srv_fil_crypt_rotate_key_age = val;
 	if (val == 0) {
-		fil_crypt_rotation_list_fill();
+		fil_crypt_default_encrypt_tables_fill();
 	}
 	mutex_exit(&fil_system.mutex);
 	os_event_set(fil_crypt_threads_event);
@@ -2357,8 +2386,8 @@ void fil_crypt_set_encrypt_tables(ulong val)
 
 	srv_encrypt_tables = val;
 
-	if (srv_fil_crypt_rotate_key_age == 0) {
-		fil_crypt_rotation_list_fill();
+	if (fil_crypt_must_default_encrypt()) {
+		fil_crypt_default_encrypt_tables_fill();
 	}
 
 	mutex_exit(&fil_system.mutex);
diff --git a/storage/innobase/fil/fil0fil.cc b/storage/innobase/fil/fil0fil.cc
index db802155b3d..407e6e96982 100644
--- a/storage/innobase/fil/fil0fil.cc
+++ b/storage/innobase/fil/fil0fil.cc
@@ -776,10 +776,10 @@ std::vector<pfs_os_file_t> fil_system_t::detach(fil_space_t *space,
     unflushed_spaces.remove(*space);
   }
 
-  if (space->is_in_rotation_list)
+  if (space->is_in_default_encrypt)
   {
-    space->is_in_rotation_list= false;
-    rotation_list.remove(*space);
+    space->is_in_default_encrypt= false;
+    default_encrypt_tables.remove(*space);
   }
   UT_LIST_REMOVE(space_list, space);
   if (space == sys_space)
@@ -1001,20 +1001,25 @@ fil_space_t *fil_space_t::create(const char *name, ulint id, ulint flags,
 		fil_system.max_assigned_id = id;
 	}
 
+	const bool rotate =
+		(purpose == FIL_TYPE_TABLESPACE
+		 && (mode == FIL_ENCRYPTION_ON
+		     || mode == FIL_ENCRYPTION_OFF || srv_encrypt_tables)
+		 && fil_crypt_must_default_encrypt());
+
 	/* Inform key rotation that there could be something
 	to do */
-	if (purpose == FIL_TYPE_TABLESPACE
-	    && !srv_fil_crypt_rotate_key_age && fil_crypt_threads_event &&
-	    (mode == FIL_ENCRYPTION_ON || mode == FIL_ENCRYPTION_OFF
-	     || srv_encrypt_tables)) {
+	if (rotate) {
 		/* Key rotation is not enabled, need to inform background
 		encryption threads. */
-		fil_system.rotation_list.push_back(*space);
-		space->is_in_rotation_list = true;
-		mutex_exit(&fil_system.mutex);
+		fil_system.default_encrypt_tables.push_back(*space);
+		space->is_in_default_encrypt = true;
+	}
+
+	mutex_exit(&fil_system.mutex);
+
+	if (rotate && srv_n_fil_crypt_threads_started) {
 		os_event_set(fil_crypt_threads_event);
-	} else {
-		mutex_exit(&fil_system.mutex);
 	}
 
 	return(space);
diff --git a/storage/innobase/fsp/fsp0fsp.cc b/storage/innobase/fsp/fsp0fsp.cc
index ceaab79b1c4..3d5a7edd947 100644
--- a/storage/innobase/fsp/fsp0fsp.cc
+++ b/storage/innobase/fsp/fsp0fsp.cc
@@ -2323,7 +2323,6 @@ fsp_reserve_free_extents(
 	uint32_t	n_pages)
 {
 	ulint		reserve;
-	size_t		total_reserved = 0;
 
 	ut_ad(mtr);
 	*n_reserved = n_ext;
@@ -2409,8 +2408,7 @@ try_again:
 		return(true);
 	}
 try_to_extend:
-	if (ulint n = fsp_try_extend_data_file(space, header, mtr)) {
-		total_reserved += n;
+	if (fsp_try_extend_data_file(space, header, mtr)) {
 		goto try_again;
 	}
 
diff --git a/storage/innobase/include/fil0crypt.h b/storage/innobase/include/fil0crypt.h
index 49567a49a78..872053dcbcb 100644
--- a/storage/innobase/include/fil0crypt.h
+++ b/storage/innobase/include/fil0crypt.h
@@ -446,4 +446,10 @@ encrypted, or corrupted.
 bool fil_space_verify_crypt_checksum(const byte* page, ulint zip_size)
 	MY_ATTRIBUTE((warn_unused_result));
 
+/** Add the tablespace to the rotation list if
+innodb_encrypt_rotate_key_age is 0 or encryption plugin does
+not do key version rotation
+@return whether the tablespace should be added to rotation list */
+bool fil_crypt_must_default_encrypt();
+
 #endif /* fil0crypt_h */
diff --git a/storage/innobase/include/fil0fil.h b/storage/innobase/include/fil0fil.h
index 5fef6c6faea..57b103518d5 100644
--- a/storage/innobase/include/fil0fil.h
+++ b/storage/innobase/include/fil0fil.h
@@ -405,7 +405,7 @@ public:
 	bool is_in_unflushed_spaces;
 
 	/** Checks that this tablespace needs key rotation. */
-	bool is_in_rotation_list;
+	bool is_in_default_encrypt;
 
 	/** True if the device this filespace is on supports atomic writes */
 	bool		atomic_write_supported;
@@ -1404,24 +1404,25 @@ public:
 					record has been written since
 					the latest redo log checkpoint.
 					Protected only by log_sys.mutex. */
-	ilist<fil_space_t, rotation_list_tag_t> rotation_list;
-					/*!< list of all file spaces needing
-					key rotation.*/
+
+	/** List of all file spaces need key rotation */
+	ilist<fil_space_t, rotation_list_tag_t> default_encrypt_tables;
 
 	bool		space_id_reuse_warned;
 					/*!< whether fil_space_t::create()
 					has issued a warning about
 					potential space_id reuse */
 
-  /** Return the next tablespace from rotation_list.
-  @param space   previous tablespace (NULL to start from the start)
+  /** Return the next tablespace from default_encrypt_tables list.
+  @param space   previous tablespace (nullptr to start from the start)
   @param recheck whether the removal condition needs to be rechecked after
   the encryption parameters were changed
   @param encrypt expected state of innodb_encrypt_tables
   @return the next tablespace to process (n_pending_ops incremented)
-  @retval NULL if this was the last */
-  inline fil_space_t* keyrotate_next(fil_space_t *space, bool recheck,
-                                     bool encrypt);
+  @retval fil_system.temp_space if there is no work to do
+  @retval nullptr upon reaching the end of the iteration */
+  inline fil_space_t* default_encrypt_next(fil_space_t *space, bool recheck,
+                                           bool encrypt);
 
   /** Extend all open data files to the recovered size */
   ATTRIBUTE_COLD void extend_to_recv_size();
diff --git a/storage/innobase/include/trx0purge.h b/storage/innobase/include/trx0purge.h
index 38e9e7207a2..ef94207617d 100644
--- a/storage/innobase/include/trx0purge.h
+++ b/storage/innobase/include/trx0purge.h
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1996, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, 2020, MariaDB Corporation.
+Copyright (c) 2017, 2021, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -60,20 +60,17 @@ public:
 	typedef trx_rsegs_t::iterator iterator;
 	typedef trx_rsegs_t::const_iterator const_iterator;
 
-	/** Default constructor */
 	TrxUndoRsegs() {}
+
 	/** Constructor */
 	TrxUndoRsegs(trx_rseg_t& rseg)
-		: m_commit(rseg.last_commit), m_rsegs(1, &rseg) {}
+		: trx_no(rseg.last_trx_no()), m_rsegs(1, &rseg) {}
 	/** Constructor */
 	TrxUndoRsegs(trx_id_t trx_no, trx_rseg_t& rseg)
-		: m_commit(trx_no << 1), m_rsegs(1, &rseg) {}
-
-	/** @return the transaction commit identifier */
-	trx_id_t trx_no() const { return m_commit >> 1; }
+		: trx_no(trx_no), m_rsegs(1, &rseg) {}
 
 	bool operator!=(const TrxUndoRsegs& other) const
-	{ return m_commit != other.m_commit; }
+	{ return trx_no != other.trx_no; }
 	bool empty() const { return m_rsegs.empty(); }
 	void erase(iterator& it) { m_rsegs.erase(it); }
 	iterator begin() { return(m_rsegs.begin()); }
@@ -87,14 +84,14 @@ public:
 	@return true if elem1 > elem2 else false.*/
 	bool operator()(const TrxUndoRsegs& lhs, const TrxUndoRsegs& rhs)
 	{
-		return(lhs.m_commit > rhs.m_commit);
+		return(lhs.trx_no > rhs.trx_no);
 	}
 
+	/** Copy of trx_rseg_t::last_trx_no() */
+	trx_id_t trx_no= 0;
 private:
-	/** Copy trx_rseg_t::last_commit */
-	trx_id_t		m_commit;
 	/** Rollback segments of a transaction, scheduled for purge. */
-	trx_rsegs_t		m_rsegs;
+	trx_rsegs_t m_rsegs{};
 };
 
 typedef std::priority_queue<
@@ -146,17 +143,13 @@ public:
 	{
 		bool operator<=(const iterator& other) const
 		{
-			if (commit < other.commit) return true;
-			if (commit > other.commit) return false;
+			if (trx_no < other.trx_no) return true;
+			if (trx_no > other.trx_no) return false;
 			return undo_no <= other.undo_no;
 		}
 
-		/** @return the commit number of the transaction */
-		trx_id_t trx_no() const { return commit >> 1; }
-		void reset_trx_no(trx_id_t trx_no) { commit = trx_no << 1; }
-
-		/** 2 * trx_t::no + old_insert of the committed transaction */
-		trx_id_t	commit;
+		/** trx_t::no of the committed transaction */
+		trx_id_t	trx_no;
 		/** The record number within the committed transaction's undo
 		log, increasing, purged from from 0 onwards */
 		undo_no_t	undo_no;
diff --git a/storage/innobase/include/trx0rseg.h b/storage/innobase/include/trx0rseg.h
index a3ccc90b630..7e4511b8f30 100644
--- a/storage/innobase/include/trx0rseg.h
+++ b/storage/innobase/include/trx0rseg.h
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1996, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, 2020, MariaDB Corporation.
+Copyright (c) 2017, 2021, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -65,9 +65,8 @@ trx_rseg_header_create(
 	buf_block_t*	sys_header,
 	mtr_t*		mtr);
 
-/** Initialize the rollback segments in memory at database startup. */
-void
-trx_rseg_array_init();
+/** Initialize or recover the rollback segments at startup. */
+dberr_t trx_rseg_array_init();
 
 /** Free a rollback segment in memory. */
 void
@@ -119,21 +118,13 @@ struct trx_rseg_t {
 	/** List of undo log segments cached for fast reuse */
 	UT_LIST_BASE_NODE_T(trx_undo_t)	undo_cached;
 
-	/** List of recovered old insert_undo logs of incomplete
-	transactions (to roll back or XA COMMIT & purge) */
-	UT_LIST_BASE_NODE_T(trx_undo_t) old_insert_list;
-
 	/*--------------------------------------------------------*/
 
-	/** Page number of the last not yet purged log header in the history
-	list; FIL_NULL if all list purged */
-	uint32_t			last_page_no;
-
-	/** Byte offset of the last not yet purged log header */
-	uint16_t			last_offset;
+  /** Last not yet purged undo log header; FIL_NULL if all purged */
+  uint32_t last_page_no;
 
-	/** trx_t::no * 2 + old_insert of the last not yet purged log */
-	trx_id_t			last_commit;
+  /** trx_t::no | last_offset << 48 */
+  uint64_t last_commit_and_offset;
 
 	/** Whether the log segment needs purge */
 	bool				needs_purge;
@@ -145,13 +136,17 @@ struct trx_rseg_t {
 	UNDO-tablespace marked for truncate. */
 	bool				skip_allocation;
 
-	/** @return the commit ID of the last committed transaction */
-	trx_id_t last_trx_no() const { return last_commit >> 1; }
-
-	void set_last_trx_no(trx_id_t trx_no, bool is_update)
-	{
-		last_commit = trx_no << 1 | trx_id_t(is_update);
-	}
+  /** @return the commit ID of the last committed transaction */
+  trx_id_t last_trx_no() const
+  { return last_commit_and_offset & ((1ULL << 48) - 1); }
+  /** @return header offset of the last committed transaction */
+  uint16_t last_offset() const
+  { return static_cast<uint16_t>(last_commit_and_offset >> 48); }
+
+  void set_last_commit(uint16_t last_offset, trx_id_t trx_no)
+  {
+    last_commit_and_offset= static_cast<uint64_t>(last_offset) << 48 | trx_no;
+  }
 
 	/** @return whether the rollback segment is persistent */
 	bool is_persistent() const
diff --git a/storage/innobase/include/trx0trx.h b/storage/innobase/include/trx0trx.h
index f5df6209acf..09132e7cb98 100644
--- a/storage/innobase/include/trx0trx.h
+++ b/storage/innobase/include/trx0trx.h
@@ -79,7 +79,7 @@ void trx_free_at_shutdown(trx_t *trx);
 void trx_disconnect_prepared(trx_t *trx);
 
 /** Initialize (resurrect) transactions at startup. */
-void trx_lists_init_at_db_start();
+dberr_t trx_lists_init_at_db_start();
 
 /*************************************************************//**
 Starts the transaction if it is not yet started. */
@@ -621,10 +621,6 @@ struct trx_undo_ptr_t {
 					yet */
 	trx_undo_t*	undo;		/*!< pointer to the undo log, or
 					NULL if nothing logged yet */
-	trx_undo_t*     old_insert;	/*!< pointer to recovered
-					insert undo log, or NULL if no
-					INSERT transactions were
-					recovered from old-format undo logs */
 };
 
 /** An instance of temporary rollback segment. */
@@ -945,13 +941,6 @@ public:
 		return(has_logged_persistent() || rsegs.m_noredo.undo);
 	}
 
-	/** @return whether any undo log has been generated or
-	recovered */
-	bool has_logged_or_recovered() const
-	{
-		return(has_logged() || rsegs.m_redo.old_insert);
-	}
-
 	/** @return rollback segment for modifying temporary tables */
 	trx_rseg_t* get_temp_rseg()
 	{
diff --git a/storage/innobase/include/trx0types.h b/storage/innobase/include/trx0types.h
index 219783b59e1..99a9c66c839 100644
--- a/storage/innobase/include/trx0types.h
+++ b/storage/innobase/include/trx0types.h
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1996, 2014, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, 2019, MariaDB Corporation.
+Copyright (c) 2017, 2021, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -93,8 +93,6 @@ enum trx_dict_op_t {
 struct trx_t;
 /** The locks and state of an active transaction */
 struct trx_lock_t;
-/** Signal */
-struct trx_sig_t;
 /** Rollback segment */
 struct trx_rseg_t;
 /** Transaction undo log */
diff --git a/storage/innobase/include/trx0undo.h b/storage/innobase/include/trx0undo.h
index e032c53263a..319ea4ee8d9 100644
--- a/storage/innobase/include/trx0undo.h
+++ b/storage/innobase/include/trx0undo.h
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1996, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, 2020, MariaDB Corporation.
+Copyright (c) 2017, 2021, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -241,16 +241,14 @@ trx_undo_set_state_at_finish(
 @param[in]	rollback	false=XA PREPARE, true=XA ROLLBACK
 @param[in,out]	mtr		mini-transaction */
 void trx_undo_set_state_at_prepare(trx_t *trx, trx_undo_t *undo, bool rollback,
-				   mtr_t *mtr)
+                                   mtr_t *mtr)
   MY_ATTRIBUTE((nonnull));
 
-/** Free an old insert or temporary undo log after commit or rollback.
+/** Free temporary undo log after commit or rollback.
 The information is not needed after a commit or rollback, therefore
 the data can be discarded.
-@param[in,out]	undo	undo log
-@param[in]	is_temp	whether this is temporary undo log */
-void
-trx_undo_commit_cleanup(trx_undo_t* undo, bool is_temp);
+@param undo     temporary undo log */
+void trx_undo_commit_cleanup(trx_undo_t *undo);
 
 /** At shutdown, frees the undo logs of a transaction. */
 void
@@ -261,26 +259,21 @@ trx_undo_free_at_shutdown(trx_t *trx);
 @param[in]	id		rollback segment slot
 @param[in]	page_no		undo log segment page number
 @param[in,out]	max_trx_id	the largest observed transaction ID
-@return	size of the undo log in pages */
-uint32_t
+@return	the undo log
+@retval nullptr on error */
+trx_undo_t *
 trx_undo_mem_create_at_db_start(trx_rseg_t *rseg, ulint id, uint32_t page_no,
                                 trx_id_t &max_trx_id);
 
 #endif /* !UNIV_INNOCHECKSUM */
 
-/* Types of an undo log segment */
-#define	TRX_UNDO_INSERT		1	/* contains undo entries for inserts */
-#define	TRX_UNDO_UPDATE		2	/* contains undo entries for updates
-					and delete markings: in short,
-					modifys (the name 'UPDATE' is a
-					historical relic) */
+/** the only rollback segment type since MariaDB 10.3.1 */
+constexpr uint16_t TRX_UNDO_UPDATE= 2;
 /* TRX_UNDO_STATE values of an undo log segment */
 /** contains an undo log of an active transaction */
 constexpr uint16_t TRX_UNDO_ACTIVE = 1;
 /** cached for quick reuse */
 constexpr uint16_t TRX_UNDO_CACHED = 2;
-/** old_insert undo segment that can be freed */
-constexpr uint16_t TRX_UNDO_TO_FREE = 3;
 /** can be freed in purge when all undo data in it is removed */
 constexpr uint16_t TRX_UNDO_TO_PURGE = 4;
 /** contains an undo log of a prepared transaction */
@@ -343,7 +336,8 @@ struct trx_undo_t {
 /** Transaction undo log page header offsets */
 /* @{ */
 #define	TRX_UNDO_PAGE_TYPE	0	/*!< unused; 0 (before MariaDB 10.3.1:
-					TRX_UNDO_INSERT or TRX_UNDO_UPDATE) */
+					1=TRX_UNDO_INSERT or
+					2=TRX_UNDO_UPDATE) */
 #define	TRX_UNDO_PAGE_START	2	/*!< Byte offset where the undo log
 					records for the LATEST transaction
 					start on this page (remember that
diff --git a/storage/innobase/lock/lock0lock.cc b/storage/innobase/lock/lock0lock.cc
index e124607617c..8dc2d7c585a 100644
--- a/storage/innobase/lock/lock0lock.cc
+++ b/storage/innobase/lock/lock0lock.cc
@@ -4476,7 +4476,7 @@ lock_print_info_summary(
 		"Purge done for trx's n:o < " TRX_ID_FMT
 		" undo n:o < " TRX_ID_FMT " state: %s\n"
 		"History list length %u\n",
-		purge_sys.tail.trx_no(),
+		purge_sys.tail.trx_no,
 		purge_sys.tail.undo_no,
 		purge_sys.enabled()
 		? (purge_sys.running() ? "running"
diff --git a/storage/innobase/row/row0ftsort.cc b/storage/innobase/row/row0ftsort.cc
index 562395e3e80..5ec73d9677c 100644
--- a/storage/innobase/row/row0ftsort.cc
+++ b/storage/innobase/row/row0ftsort.cc
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 2010, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2015, 2020, MariaDB Corporation.
+Copyright (c) 2015, 2021, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -755,7 +755,6 @@ void fts_parallel_tokenization(
 	row_merge_block_t**	crypt_block;
 	pfs_os_file_t		tmpfd[FTS_NUM_AUX_INDEX];
 	ulint			mycount[FTS_NUM_AUX_INDEX];
-	ib_uint64_t		total_rec = 0;
 	ulint			num_doc_processed = 0;
 	doc_id_t		last_doc_id = 0;
 	mem_heap_t*		blob_heap = NULL;
@@ -1023,7 +1022,6 @@ exit:
 			goto func_exit;
 		}
 
-		total_rec += merge_file[i]->n_rec;
 		row_merge_file_destroy_low(tmpfd[i]);
 	}
 
diff --git a/storage/innobase/row/row0merge.cc b/storage/innobase/row/row0merge.cc
index 6b071d3c3d7..417bf6a4a1f 100644
--- a/storage/innobase/row/row0merge.cc
+++ b/storage/innobase/row/row0merge.cc
@@ -517,7 +517,9 @@ row_merge_buf_add(
 	DBUG_ENTER("row_merge_buf_add");
 
 	if (buf->n_tuples >= buf->max_tuples) {
-		DBUG_RETURN(0);
+error:
+		n_row_added = 0;
+		goto end;
 	}
 
 	DBUG_EXECUTE_IF(
@@ -683,7 +685,10 @@ row_merge_buf_add(
 				continue;
 			}
 
-			if (field->len != UNIV_SQL_NULL
+			/* innobase_get_computed_value() sets the
+			length of the virtual column field. */
+			if (v_col == NULL
+			    && field->len != UNIV_SQL_NULL
 			    && col->mtype == DATA_MYSQL
 			    && col->len != field->len) {
 				if (conv_heap != NULL) {
@@ -839,11 +844,6 @@ end:
         if (vcol_storage.innobase_record)
 		innobase_free_row_for_vcol(&vcol_storage);
 	DBUG_RETURN(n_row_added);
-
-error:
-        if (vcol_storage.innobase_record)
-		innobase_free_row_for_vcol(&vcol_storage);
-        DBUG_RETURN(0);
 }
 
 /*************************************************************//**
@@ -2667,16 +2667,18 @@ write_buffers:
 						new_table, psort_info, row, ext,
 						&doc_id, conv_heap,
 						&err, &v_heap, eval_table, trx)))) {
-					/* An empty buffer should have enough
-					room for at least one record. */
-					ut_error;
-				}
-
-				if (err != DB_SUCCESS) {
-					break;
+                                        /* An empty buffer should have enough
+                                        room for at least one record. */
+					ut_ad(err == DB_COMPUTE_VALUE_FAILED
+					      || err == DB_OUT_OF_MEMORY
+					      || err == DB_TOO_BIG_RECORD);
+				} else if (err == DB_SUCCESS) {
+					file->n_rec += rows_added;
+					continue;
 				}
 
-				file->n_rec += rows_added;
+				trx->error_key_num = i;
+				break;
 			}
 		}
 
diff --git a/storage/innobase/row/row0undo.cc b/storage/innobase/row/row0undo.cc
index a02f0467431..3ac8e434f35 100644
--- a/storage/innobase/row/row0undo.cc
+++ b/storage/innobase/row/row0undo.cc
@@ -284,23 +284,13 @@ static bool row_undo_rec_get(undo_node_t* node)
 	}
 
 	trx_undo_t*	undo	= NULL;
-	trx_undo_t*	insert	= trx->rsegs.m_redo.old_insert;
 	trx_undo_t*	update	= trx->rsegs.m_redo.undo;
 	trx_undo_t*	temp	= trx->rsegs.m_noredo.undo;
 	const undo_no_t	limit	= trx->roll_limit;
 
-	ut_ad(!insert || !update || insert->empty() || update->empty()
-	      || insert->top_undo_no != update->top_undo_no);
-	ut_ad(!insert || !temp || insert->empty() || temp->empty()
-	      || insert->top_undo_no != temp->top_undo_no);
 	ut_ad(!update || !temp || update->empty() || temp->empty()
 	      || update->top_undo_no != temp->top_undo_no);
 
-	if (UNIV_LIKELY_NULL(insert)
-	    && !insert->empty() && limit <= insert->top_undo_no) {
-		undo = insert;
-	}
-
 	if (update && !update->empty() && update->top_undo_no >= limit) {
 		if (!undo) {
 			undo = update;
@@ -369,19 +359,16 @@ static bool row_undo_rec_get(undo_node_t* node)
 		MDEV-12288 removed the insert_undo log. There is no
 		instant ADD COLUMN for temporary tables. Therefore,
 		this record can only be present in the main undo log. */
-		ut_ad(undo == update);
 		/* fall through */
 	case TRX_UNDO_RENAME_TABLE:
-		ut_ad(undo == insert || undo == update);
+		ut_ad(undo == update);
 		/* fall through */
 	case TRX_UNDO_INSERT_REC:
-		ut_ad(undo == insert || undo == update || undo == temp);
 		node->roll_ptr |= 1ULL << ROLL_PTR_INSERT_FLAG_POS;
 		node->state = undo == temp
 			? UNDO_INSERT_TEMPORARY : UNDO_INSERT_PERSISTENT;
 		break;
 	default:
-		ut_ad(undo == update || undo == temp);
 		node->state = undo == temp
 			? UNDO_UPDATE_TEMPORARY : UNDO_UPDATE_PERSISTENT;
 		break;
diff --git a/storage/innobase/srv/srv0start.cc b/storage/innobase/srv/srv0start.cc
index 261e50285b7..c4bf7ccb68b 100644
--- a/storage/innobase/srv/srv0start.cc
+++ b/storage/innobase/srv/srv0start.cc
@@ -1059,6 +1059,11 @@ dberr_t srv_start(bool create_new_db)
 	      || srv_operation == SRV_OPERATION_RESTORE
 	      || srv_operation == SRV_OPERATION_RESTORE_EXPORT);
 
+	if (srv_force_recovery) {
+		ib::info() << "!!! innodb_force_recovery is set to "
+			<< srv_force_recovery << " !!!";
+	}
+
 	if (srv_force_recovery == SRV_FORCE_NO_LOG_REDO) {
 		srv_read_only_mode = true;
 	}
@@ -1414,7 +1419,11 @@ file_checked:
 		All the remaining rollback segments will be created later,
 		after the double write buffer has been created. */
 		trx_sys_create_sys_pages();
-		trx_lists_init_at_db_start();
+		err = trx_lists_init_at_db_start();
+
+		if (err != DB_SUCCESS) {
+			return(srv_init_abort(err));
+		}
 
 		err = dict_create();
 
@@ -1476,7 +1485,10 @@ file_checked:
 			if (srv_operation == SRV_OPERATION_RESTORE) {
 				break;
 			}
-			trx_lists_init_at_db_start();
+			err = trx_lists_init_at_db_start();
+			if (err != DB_SUCCESS) {
+				return srv_init_abort(err);
+			}
 			break;
 		case SRV_OPERATION_RESTORE_DELTA:
 		case SRV_OPERATION_BACKUP:
@@ -1899,11 +1911,6 @@ skip_monitors:
 			   << "; transaction id " << trx_sys.get_max_trx_id();
 	}
 
-	if (srv_force_recovery > 0) {
-		ib::info() << "!!! innodb_force_recovery is set to "
-			<< srv_force_recovery << " !!!";
-	}
-
 	if (srv_force_recovery == 0) {
 		/* In the change buffer we may have even bigger tablespace
 		id's, because we may have dropped those tablespaces, but
diff --git a/storage/innobase/trx/trx0purge.cc b/storage/innobase/trx/trx0purge.cc
index c0375f25fa6..0a1d8ba7532 100644
--- a/storage/innobase/trx/trx0purge.cc
+++ b/storage/innobase/trx/trx0purge.cc
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1996, 2017, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, 2020, MariaDB Corporation.
+Copyright (c) 2017, 2021, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -89,7 +89,7 @@ inline bool TrxUndoRsegsIterator::set_next()
 		number shouldn't increase. Undo the increment of
 		expected commit done by caller assuming rollback
 		segments from given transaction are done. */
-		purge_sys.tail.commit = (*m_iter)->last_commit;
+		purge_sys.tail.trx_no = (*m_iter)->last_trx_no();
 	} else if (!purge_sys.purge_queue.empty()) {
 		m_rsegs = purge_sys.purge_queue.top();
 		purge_sys.purge_queue.pop();
@@ -110,17 +110,17 @@ inline bool TrxUndoRsegsIterator::set_next()
 	mutex_enter(&purge_sys.rseg->mutex);
 
 	ut_a(purge_sys.rseg->last_page_no != FIL_NULL);
-	ut_ad(purge_sys.rseg->last_trx_no() == m_rsegs.trx_no());
+	ut_ad(purge_sys.rseg->last_trx_no() == m_rsegs.trx_no);
 
 	/* We assume in purge of externally stored fields that space id is
 	in the range of UNDO tablespace space ids */
 	ut_ad(purge_sys.rseg->space->id == TRX_SYS_SPACE
 	      || srv_is_undo_tablespace(purge_sys.rseg->space->id));
 
-	ut_a(purge_sys.tail.commit <= purge_sys.rseg->last_commit);
+	ut_a(purge_sys.tail.trx_no <= purge_sys.rseg->last_trx_no());
 
-	purge_sys.tail.commit = purge_sys.rseg->last_commit;
-	purge_sys.hdr_offset = purge_sys.rseg->last_offset;
+	purge_sys.tail.trx_no = purge_sys.rseg->last_trx_no();
+	purge_sys.hdr_offset = purge_sys.rseg->last_offset();
 	purge_sys.hdr_page_no = purge_sys.rseg->last_page_no;
 
 	mutex_exit(&purge_sys.rseg->mutex);
@@ -211,8 +211,7 @@ trx_purge_add_undo_to_history(const trx_t* trx, trx_undo_t*& undo, mtr_t* mtr)
 {
 	DBUG_PRINT("trx", ("commit(" TRX_ID_FMT "," TRX_ID_FMT ")",
 			   trx->id, trx_id_t{trx->rw_trx_hash_element->no}));
-	ut_ad(undo == trx->rsegs.m_redo.undo
-	      || undo == trx->rsegs.m_redo.old_insert);
+	ut_ad(undo == trx->rsegs.m_redo.undo);
 	trx_rseg_t*	rseg		= trx->rsegs.m_redo.rseg;
 	ut_ad(undo->rseg == rseg);
 	buf_block_t*	rseg_header	= trx_rsegf_get(
@@ -303,19 +302,13 @@ trx_purge_add_undo_to_history(const trx_t* trx, trx_undo_t*& undo, mtr_t* mtr)
 	mtr->write<8,mtr_t::MAYBE_NOP>(*undo_page,
 				       undo_header + TRX_UNDO_TRX_NO,
 				       trx->rw_trx_hash_element->no);
-	/* This is needed for upgrading old undo log pages from
-	before MariaDB 10.3.1. */
-	if (UNIV_UNLIKELY(!mach_read_from_2(undo_header
-					    + TRX_UNDO_NEEDS_PURGE))) {
-		mtr->write<2>(*undo_page, undo_header + TRX_UNDO_NEEDS_PURGE,
-			      1U);
-	}
+	mtr->write<2,mtr_t::MAYBE_NOP>(*undo_page, undo_header
+				       + TRX_UNDO_NEEDS_PURGE, 1U);
 
 	if (rseg->last_page_no == FIL_NULL) {
 		rseg->last_page_no = undo->hdr_page_no;
-		rseg->last_offset = undo->hdr_offset;
-		rseg->set_last_trx_no(trx->rw_trx_hash_element->no,
-				      undo == trx->rsegs.m_redo.undo);
+		rseg->set_last_commit(undo->hdr_offset,
+				      trx->rw_trx_hash_element->no);
 		rseg->needs_purge = true;
 	}
 
@@ -462,8 +455,8 @@ func_exit:
 	undo_trx_no = mach_read_from_8(block->frame + hdr_addr.boffset
 				       + TRX_UNDO_TRX_NO);
 
-	if (undo_trx_no >= limit.trx_no()) {
-		if (undo_trx_no == limit.trx_no()) {
+	if (undo_trx_no >= limit.trx_no) {
+		if (undo_trx_no == limit.trx_no) {
 			trx_undo_truncate_start(
 				&rseg, hdr_addr.page,
 				hdr_addr.boffset, limit.undo_no);
@@ -554,12 +547,12 @@ function is called, the caller must not have any latches on undo log pages!
 static void trx_purge_truncate_history()
 {
 	ut_ad(purge_sys.head <= purge_sys.tail);
-	purge_sys_t::iterator& head = purge_sys.head.commit
+	purge_sys_t::iterator& head = purge_sys.head.trx_no
 		? purge_sys.head : purge_sys.tail;
 
-	if (head.trx_no() >= purge_sys.low_limit_no()) {
+	if (head.trx_no >= purge_sys.low_limit_no()) {
 		/* This is sometimes necessary. TODO: find out why. */
-		head.reset_trx_no(purge_sys.low_limit_no());
+		head.trx_no = purge_sys.low_limit_no();
 		head.undo_no = 0;
 	}
 
@@ -649,7 +642,7 @@ not_free:
 				     undo;
 				     undo = UT_LIST_GET_NEXT(undo_list,
 							     undo)) {
-					if (head.trx_no() < undo->trx_id) {
+					if (head.trx_no < undo->trx_id) {
 						goto not_free;
 					} else {
 						cached += undo->size;
@@ -728,7 +721,6 @@ not_free:
 			free the existing structure. There can't be
 			any active transactions. */
 			ut_a(UT_LIST_GET_LEN(rseg->undo_list) == 0);
-			ut_a(UT_LIST_GET_LEN(rseg->old_insert_list) == 0);
 
 			trx_undo_t*	next_undo;
 
@@ -746,8 +738,6 @@ not_free:
 				     &trx_undo_t::undo_list);
 			UT_LIST_INIT(rseg->undo_cached,
 				     &trx_undo_t::undo_list);
-			UT_LIST_INIT(rseg->old_insert_list,
-				     &trx_undo_t::undo_list);
 
 			/* These were written by trx_rseg_header_create(). */
 			ut_ad(!mach_read_from_4(TRX_RSEG + TRX_RSEG_FORMAT
@@ -760,8 +750,7 @@ not_free:
 			rseg->curr_size = 1;
 			rseg->trx_ref_count = 0;
 			rseg->last_page_no = FIL_NULL;
-			rseg->last_offset = 0;
-			rseg->last_commit = 0;
+			rseg->last_commit_and_offset = 0;
 			rseg->needs_purge = false;
 		}
 
@@ -836,7 +825,7 @@ static void trx_purge_rseg_get_next_history_log(
 
 	ut_a(purge_sys.rseg->last_page_no != FIL_NULL);
 
-	purge_sys.tail.commit = purge_sys.rseg->last_commit + 1;
+	purge_sys.tail.trx_no = purge_sys.rseg->last_trx_no() + 1;
 	purge_sys.tail.undo_no = 0;
 	purge_sys.next_stored = false;
 
@@ -847,7 +836,7 @@ static void trx_purge_rseg_get_next_history_log(
 			  purge_sys.rseg->last_page_no), &mtr);
 
 	const trx_ulogf_t* log_hdr = undo_page->frame
-		+ purge_sys.rseg->last_offset;
+		+ purge_sys.rseg->last_offset();
 
 	/* Increase the purge page count by one for every handled log */
 
@@ -881,17 +870,15 @@ static void trx_purge_rseg_get_next_history_log(
 		+ prev_log_addr.boffset;
 
 	trx_no = mach_read_from_8(log_hdr + TRX_UNDO_TRX_NO);
-	unsigned purge = mach_read_from_2(log_hdr + TRX_UNDO_NEEDS_PURGE);
-	ut_ad(purge <= 1);
+	ut_ad(mach_read_from_2(log_hdr + TRX_UNDO_NEEDS_PURGE) <= 1);
 
 	mtr_commit(&mtr);
 
 	mutex_enter(&purge_sys.rseg->mutex);
 
 	purge_sys.rseg->last_page_no = prev_log_addr.page;
-	purge_sys.rseg->last_offset = prev_log_addr.boffset;
-	purge_sys.rseg->set_last_trx_no(trx_no, purge != 0);
-	purge_sys.rseg->needs_purge = purge != 0;
+	purge_sys.rseg->set_last_commit(prev_log_addr.boffset, trx_no);
+	purge_sys.rseg->needs_purge = log_hdr[TRX_UNDO_NEEDS_PURGE + 1] != 0;
 
 	/* Purge can also produce events, however these are already ordered
 	in the rollback segment and any user generated event will be greater
@@ -908,15 +895,13 @@ static void trx_purge_rseg_get_next_history_log(
 }
 
 /** Position the purge sys "iterator" on the undo record to use for purging. */
-static
-void
-trx_purge_read_undo_rec()
+static void trx_purge_read_undo_rec()
 {
 	uint16_t	offset;
 	uint32_t	page_no;
 	ib_uint64_t	undo_no;
 
-	purge_sys.hdr_offset = purge_sys.rseg->last_offset;
+	purge_sys.hdr_offset = purge_sys.rseg->last_offset();
 	page_no = purge_sys.hdr_page_no = purge_sys.rseg->last_page_no;
 
 	if (purge_sys.rseg->needs_purge) {
@@ -983,7 +968,7 @@ trx_purge_get_next_rec(
 	mtr_t		mtr;
 
 	ut_ad(purge_sys.next_stored);
-	ut_ad(purge_sys.tail.trx_no() < purge_sys.low_limit_no());
+	ut_ad(purge_sys.tail.trx_no < purge_sys.low_limit_no());
 
 	const ulint space = purge_sys.rseg->space->id;
 	const uint32_t page_no = purge_sys.page_no;
@@ -1073,7 +1058,7 @@ trx_purge_fetch_next_rec(
 		}
 	}
 
-	if (purge_sys.tail.trx_no() >= purge_sys.low_limit_no()) {
+	if (purge_sys.tail.trx_no >= purge_sys.low_limit_no()) {
 
 		return(NULL);
 	}
diff --git a/storage/innobase/trx/trx0roll.cc b/storage/innobase/trx/trx0roll.cc
index ee307d3fba6..23aa950a14a 100644
--- a/storage/innobase/trx/trx0roll.cc
+++ b/storage/innobase/trx/trx0roll.cc
@@ -71,12 +71,6 @@ inline bool trx_t::rollback_finish()
   ut_a(!srv_undo_sources);
   ut_ad(srv_fast_shutdown);
   ut_d(in_rollback= false);
-  if (trx_undo_t *&undo= rsegs.m_redo.old_insert)
-  {
-    UT_LIST_REMOVE(rsegs.m_redo.rseg->old_insert_list, undo);
-    ut_free(undo);
-    undo= nullptr;
-  }
   if (trx_undo_t *&undo= rsegs.m_redo.undo)
   {
     UT_LIST_REMOVE(rsegs.m_redo.rseg->undo_list, undo);
@@ -119,7 +113,7 @@ inline void trx_t::rollback_low(trx_savept_t *savept)
 
   error_state = DB_SUCCESS;
 
-  if (has_logged_or_recovered())
+  if (has_logged())
   {
     ut_ad(rsegs.m_redo.rseg || rsegs.m_noredo.rseg);
     que_thr_t *thr= pars_complete_graph_for_exec(roll_node, this, heap,
@@ -231,7 +225,7 @@ dberr_t trx_rollback_for_mysql(trx_t* trx)
 	case TRX_STATE_PREPARED:
 	case TRX_STATE_PREPARED_RECOVERED:
 		ut_ad(!trx->is_autocommit_non_locking());
-		if (trx->rsegs.m_redo.undo || trx->rsegs.m_redo.old_insert) {
+		if (trx->rsegs.m_redo.undo) {
 			/* The XA ROLLBACK of a XA PREPARE transaction
 			will consist of multiple mini-transactions.
 
@@ -247,11 +241,7 @@ dberr_t trx_rollback_for_mysql(trx_t* trx)
 			killed, and finally, the transaction would be
 			recovered in XA PREPARE state, with some of
 			the actions already having been rolled back. */
-			ut_ad(!trx->rsegs.m_redo.undo
-			      || trx->rsegs.m_redo.undo->rseg
-			      == trx->rsegs.m_redo.rseg);
-			ut_ad(!trx->rsegs.m_redo.old_insert
-			      || trx->rsegs.m_redo.old_insert->rseg
+			ut_ad(trx->rsegs.m_redo.undo->rseg
 			      == trx->rsegs.m_redo.rseg);
 			mtr_t		mtr;
 			mtr.start();
@@ -260,10 +250,6 @@ dberr_t trx_rollback_for_mysql(trx_t* trx)
 				trx_undo_set_state_at_prepare(trx, undo, true,
 							      &mtr);
 			}
-			if (trx_undo_t* undo = trx->rsegs.m_redo.old_insert) {
-				trx_undo_set_state_at_prepare(trx, undo, true,
-							      &mtr);
-			}
 			mutex_exit(&trx->rsegs.m_redo.rseg->mutex);
 			/* Write the redo log for the XA ROLLBACK
 			state change to the global buffer. It is
diff --git a/storage/innobase/trx/trx0rseg.cc b/storage/innobase/trx/trx0rseg.cc
index eea5654052e..bdb25b85ef4 100644
--- a/storage/innobase/trx/trx0rseg.cc
+++ b/storage/innobase/trx/trx0rseg.cc
@@ -367,7 +367,6 @@ trx_rseg_mem_free(trx_rseg_t* rseg)
 
 	/* There can't be any active transactions. */
 	ut_a(UT_LIST_GET_LEN(rseg->undo_list) == 0);
-	ut_a(UT_LIST_GET_LEN(rseg->old_insert_list) == 0);
 
 	for (undo = UT_LIST_GET_FIRST(rseg->undo_cached);
 	     undo != NULL;
@@ -407,7 +406,6 @@ trx_rseg_mem_create(ulint id, fil_space_t* space, uint32_t page_no)
 		     &rseg->mutex);
 
 	UT_LIST_INIT(rseg->undo_list, &trx_undo_t::undo_list);
-	UT_LIST_INIT(rseg->old_insert_list, &trx_undo_t::undo_list);
 	UT_LIST_INIT(rseg->undo_cached, &trx_undo_t::undo_list);
 
 	return(rseg);
@@ -417,34 +415,36 @@ trx_rseg_mem_create(ulint id, fil_space_t* space, uint32_t page_no)
 @param[in,out]  rseg            rollback segment
 @param[in,out]  max_trx_id      maximum observed transaction identifier
 @param[in]      rseg_header     rollback segment header
-@return the combined size of undo log segments in pages */
-static uint32_t trx_undo_lists_init(trx_rseg_t *rseg, trx_id_t &max_trx_id,
-				    const buf_block_t *rseg_header)
+@return error code */
+static dberr_t trx_undo_lists_init(trx_rseg_t *rseg, trx_id_t &max_trx_id,
+                                   const buf_block_t *rseg_header)
 {
   ut_ad(srv_force_recovery < SRV_FORCE_NO_UNDO_LOG_SCAN);
 
-  uint32_t size= 0;
-
   for (ulint i= 0; i < TRX_RSEG_N_SLOTS; i++)
   {
     uint32_t page_no= trx_rsegf_get_nth_undo(rseg_header, i);
     if (page_no != FIL_NULL)
     {
-      size+= trx_undo_mem_create_at_db_start(rseg, i, page_no, max_trx_id);
+      const trx_undo_t *undo= trx_undo_mem_create_at_db_start(rseg, i, page_no,
+                                                              max_trx_id);
+      if (!undo)
+        return DB_CORRUPTION;
+      rseg->curr_size+= undo->size;
       MONITOR_INC(MONITOR_NUM_UNDO_SLOT_USED);
     }
   }
 
-  return size;
+  return DB_SUCCESS;
 }
 
 /** Restore the state of a persistent rollback segment.
 @param[in,out]	rseg		persistent rollback segment
 @param[in,out]	max_trx_id	maximum observed transaction identifier
-@param[in,out]	mtr		mini-transaction */
-static
-void
-trx_rseg_mem_restore(trx_rseg_t* rseg, trx_id_t& max_trx_id, mtr_t* mtr)
+@param[in,out]	mtr		mini-transaction
+@return error code */
+static dberr_t trx_rseg_mem_restore(trx_rseg_t *rseg, trx_id_t &max_trx_id,
+                                    mtr_t *mtr)
 {
 	buf_block_t* rseg_hdr = trx_rsegf_get_new(
 		rseg->space->id, rseg->page_no, mtr);
@@ -489,14 +489,17 @@ trx_rseg_mem_restore(trx_rseg_t* rseg, trx_id_t& max_trx_id, mtr_t* mtr)
 		/* mariabackup --prepare only deals with
 		the redo log and the data files, not with
 		transactions or the data dictionary. */
-		return;
+		return DB_SUCCESS;
 	}
 
 	/* Initialize the undo log lists according to the rseg header */
 
 	rseg->curr_size = mach_read_from_4(TRX_RSEG + TRX_RSEG_HISTORY_SIZE
 					   + rseg_hdr->frame)
-		+ 1 + trx_undo_lists_init(rseg, max_trx_id, rseg_hdr);
+		+ 1;
+	if (dberr_t err = trx_undo_lists_init(rseg, max_trx_id, rseg_hdr)) {
+		return err;
+	}
 
 	if (auto len = flst_get_len(TRX_RSEG + TRX_RSEG_HISTORY
 				    + rseg_hdr->frame)) {
@@ -509,7 +512,6 @@ trx_rseg_mem_restore(trx_rseg_t* rseg, trx_id_t& max_trx_id, mtr_t* mtr)
 			node_addr.boffset - TRX_UNDO_HISTORY_NODE);
 
 		rseg->last_page_no = node_addr.page;
-		rseg->last_offset = node_addr.boffset;
 
 		const buf_block_t* block = trx_undo_page_get(
 			page_id_t(rseg->space->id, node_addr.page), mtr);
@@ -524,11 +526,12 @@ trx_rseg_mem_restore(trx_rseg_t* rseg, trx_id_t& max_trx_id, mtr_t* mtr)
 		if (id > max_trx_id) {
 			max_trx_id = id;
 		}
+
+		rseg->set_last_commit(node_addr.boffset, id);
 		unsigned purge = mach_read_from_2(block->frame
 						  + node_addr.boffset
 						  + TRX_UNDO_NEEDS_PURGE);
 		ut_ad(purge <= 1);
-		rseg->set_last_trx_no(id, purge != 0);
 		rseg->needs_purge = purge != 0;
 
 		if (rseg->last_page_no != FIL_NULL) {
@@ -538,6 +541,8 @@ trx_rseg_mem_restore(trx_rseg_t* rseg, trx_id_t& max_trx_id, mtr_t* mtr)
 			purge_sys.purge_queue.push(*rseg);
 		}
 	}
+
+	return DB_SUCCESS;
 }
 
 /** Read binlog metadata from the TRX_SYS page, in case we are upgrading
@@ -561,9 +566,8 @@ static void trx_rseg_init_binlog_info(const page_t* page)
 #endif
 }
 
-/** Initialize the rollback segments in memory at database startup. */
-void
-trx_rseg_array_init()
+/** Initialize or recover the rollback segments at startup. */
+dberr_t trx_rseg_array_init()
 {
 	trx_id_t max_trx_id = 0;
 
@@ -575,9 +579,9 @@ trx_rseg_array_init()
 	wsrep_sys_xid.null();
 	bool wsrep_xid_in_rseg_found = false;
 #endif
+	mtr_t mtr;
 
 	for (ulint rseg_id = 0; rseg_id < TRX_SYS_N_RSEGS; rseg_id++) {
-		mtr_t mtr;
 		mtr.start();
 		if (const buf_block_t* sys = trx_sysf_get(&mtr, false)) {
 			if (rseg_id == 0) {
@@ -605,7 +609,11 @@ trx_rseg_array_init()
 				ut_ad(rseg->id == rseg_id);
 				ut_ad(!trx_sys.rseg_array[rseg_id]);
 				trx_sys.rseg_array[rseg_id] = rseg;
-				trx_rseg_mem_restore(rseg, max_trx_id, &mtr);
+				if (dberr_t err = trx_rseg_mem_restore(
+					    rseg, max_trx_id, &mtr)) {
+					mtr.commit();
+					return err;
+				}
 #ifdef WITH_WSREP
 				if (!wsrep_sys_xid.is_null() &&
 				    !wsrep_sys_xid.eq(&trx_sys.recovered_wsrep_xid)) {
@@ -632,7 +640,6 @@ trx_rseg_array_init()
 		If no rollback segment has a WSREP XID set,
 		we must copy the XID found in TRX_SYS page
 		to rollback segments. */
-		mtr_t mtr;
 		mtr.start();
 
 		if (!wsrep_xid_in_rseg_found) {
@@ -648,6 +655,7 @@ trx_rseg_array_init()
 #endif
 
 	trx_sys.init_max_trx_id(max_trx_id + 1);
+	return DB_SUCCESS;
 }
 
 /** Create a persistent rollback segment.
diff --git a/storage/innobase/trx/trx0trx.cc b/storage/innobase/trx/trx0trx.cc
index b4974314699..bd3b9641e1e 100644
--- a/storage/innobase/trx/trx0trx.cc
+++ b/storage/innobase/trx/trx0trx.cc
@@ -658,8 +658,7 @@ trx_resurrect_table_locks(
 
 static void trx_resurrect(trx_undo_t *undo, trx_rseg_t *rseg,
                           time_t start_time, ulonglong start_time_micro,
-                          uint64_t *rows_to_undo,
-                          bool is_old_insert)
+                          uint64_t *rows_to_undo)
 {
   trx_state_t state;
   /*
@@ -682,8 +681,6 @@ static void trx_resurrect(trx_undo_t *undo, trx_rseg_t *rseg,
     state= TRX_STATE_PREPARED;
     break;
   default:
-    if (is_old_insert && srv_force_recovery < SRV_FORCE_NO_TRX_UNDO)
-      trx_undo_commit_cleanup(undo, false);
     return;
   }
 
@@ -692,11 +689,7 @@ static void trx_resurrect(trx_undo_t *undo, trx_rseg_t *rseg,
   ut_d(trx->start_file= __FILE__);
   ut_d(trx->start_line= __LINE__);
 
-  if (is_old_insert)
-    trx->rsegs.m_redo.old_insert= undo;
-  else
-    trx->rsegs.m_redo.undo= undo;
-
+  trx->rsegs.m_redo.undo= undo;
   trx->undo_no= undo->top_undo_no + 1;
   trx->rsegs.m_redo.rseg= rseg;
   /*
@@ -727,8 +720,7 @@ static void trx_resurrect(trx_undo_t *undo, trx_rseg_t *rseg,
 
 
 /** Initialize (resurrect) transactions at startup. */
-void
-trx_lists_init_at_db_start()
+dberr_t trx_lists_init_at_db_start()
 {
 	ut_a(srv_is_being_started);
 	ut_ad(!srv_was_started);
@@ -737,16 +729,18 @@ trx_lists_init_at_db_start()
 		/* mariabackup --prepare only deals with
 		the redo log and the data files, not with
 		transactions or the data dictionary. */
-		trx_rseg_array_init();
-		return;
+		return trx_rseg_array_init();
 	}
 
 	if (srv_force_recovery >= SRV_FORCE_NO_UNDO_LOG_SCAN) {
-		return;
+		return DB_SUCCESS;
 	}
 
 	purge_sys.create();
-	trx_rseg_array_init();
+	if (dberr_t err = trx_rseg_array_init()) {
+		ib::info() << "Retry with innodb_force_recovery=5";
+		return err;
+	}
 
 	/* Look from the rollback segments if there exist undo logs for
 	transactions. */
@@ -764,17 +758,6 @@ trx_lists_init_at_db_start()
 		if (rseg == NULL) {
 			continue;
 		}
-
-		/* Resurrect transactions that were doing inserts
-		using the old separate insert_undo log. */
-		undo = UT_LIST_GET_FIRST(rseg->old_insert_list);
-		while (undo) {
-			trx_undo_t* next = UT_LIST_GET_NEXT(undo_list, undo);
-			trx_resurrect(undo, rseg, start_time, start_time_micro,
-				      &rows_to_undo, true);
-			undo = next;
-		}
-
 		/* Ressurrect other transactions. */
 		for (undo = UT_LIST_GET_FIRST(rseg->undo_list);
 		     undo != NULL;
@@ -782,8 +765,7 @@ trx_lists_init_at_db_start()
 			trx_t *trx = trx_sys.find(0, undo->trx_id, false);
 			if (!trx) {
 				trx_resurrect(undo, rseg, start_time,
-					      start_time_micro,
-					      &rows_to_undo, false);
+					      start_time_micro, &rows_to_undo);
 			} else {
 				ut_ad(trx_state_eq(trx, TRX_STATE_ACTIVE) ||
 				      trx_state_eq(trx, TRX_STATE_PREPARED));
@@ -808,16 +790,16 @@ trx_lists_init_at_db_start()
 		}
 	}
 
-	if (trx_sys.rw_trx_hash.size()) {
-
-		ib::info() << trx_sys.rw_trx_hash.size()
+	if (const auto size = trx_sys.rw_trx_hash.size()) {
+		ib::info() << size
 			<< " transaction(s) which must be rolled back or"
 			" cleaned up in total " << rows_to_undo
 			<< " row operations to undo";
-
 		ib::info() << "Trx id counter is " << trx_sys.get_max_trx_id();
 	}
+
 	purge_sys.clone_oldest_view();
+	return DB_SUCCESS;
 }
 
 /** Assign a persistent rollback segment in a round-robin fashion,
@@ -1090,30 +1072,22 @@ trx_write_serialisation_history(
 	trx_rseg_t*	rseg = trx->rsegs.m_redo.rseg;
 	if (!rseg) {
 		ut_ad(!trx->rsegs.m_redo.undo);
-		ut_ad(!trx->rsegs.m_redo.old_insert);
 		return;
 	}
 
 	trx_undo_t*& undo = trx->rsegs.m_redo.undo;
-	trx_undo_t*& old_insert = trx->rsegs.m_redo.old_insert;
 
-	if (!undo && !old_insert) {
+	if (!undo) {
 		return;
 	}
 
 	ut_ad(!trx->read_only);
 	ut_ad(!undo || undo->rseg == rseg);
-	ut_ad(!old_insert || old_insert->rseg == rseg);
 	mutex_enter(&rseg->mutex);
 
 	/* Assign the transaction serialisation number and add any
 	undo log to the purge queue. */
 	trx_serialise(trx);
-
-	if (UNIV_LIKELY_NULL(old_insert)) {
-		UT_LIST_REMOVE(rseg->old_insert_list, old_insert);
-		trx_purge_add_undo_to_history(trx, old_insert, mtr);
-	}
 	if (undo) {
 		UT_LIST_REMOVE(rseg->undo_list, undo);
 		trx_purge_add_undo_to_history(trx, undo, mtr);
@@ -1381,23 +1355,14 @@ inline void trx_t::commit_in_memory(const mtr_t *mtr)
     ut_ad(rseg->trx_ref_count > 0);
     --rseg->trx_ref_count;
     mutex_exit(&rseg->mutex);
-
-    if (trx_undo_t *&insert= rsegs.m_redo.old_insert)
-    {
-      ut_ad(insert->rseg == rseg);
-      trx_undo_commit_cleanup(insert, false);
-      insert= nullptr;
-    }
   }
 
-  ut_ad(!rsegs.m_redo.old_insert);
-
   if (mtr)
   {
     if (trx_undo_t *&undo= rsegs.m_noredo.undo)
     {
       ut_ad(undo->rseg == rsegs.m_noredo.rseg);
-      trx_undo_commit_cleanup(undo, true);
+      trx_undo_commit_cleanup(undo);
       undo= nullptr;
     }
 
@@ -1480,7 +1445,7 @@ void trx_t::commit_low(mtr_t *mtr)
 {
   ut_ad(!mtr || mtr->is_active());
   ut_d(bool aborted = in_rollback && error_state == DB_DEADLOCK);
-  ut_ad(!mtr == (aborted || !has_logged_or_recovered()));
+  ut_ad(!mtr == (aborted || !has_logged()));
   ut_ad(!mtr || !aborted);
 
   /* undo_no is non-zero if we're doing the final commit. */
@@ -1534,7 +1499,7 @@ void trx_t::commit()
   mtr_t *mtr= nullptr;
   mtr_t local_mtr;
 
-  if (has_logged_or_recovered())
+  if (has_logged())
   {
     mtr= &local_mtr;
     local_mtr.start();
@@ -1919,11 +1884,8 @@ trx_weight_ge(
 /** Prepare a transaction.
 @return	log sequence number that makes the XA PREPARE durable
 @retval	0	if no changes needed to be made durable */
-static
-lsn_t
-trx_prepare_low(trx_t* trx)
+static lsn_t trx_prepare_low(trx_t *trx)
 {
-	ut_ad(!trx->rsegs.m_redo.old_insert);
 	ut_ad(!trx->is_recovered);
 
 	mtr_t	mtr;
diff --git a/storage/innobase/trx/trx0undo.cc b/storage/innobase/trx/trx0undo.cc
index 3dd84c9c598..36656d5cb17 100644
--- a/storage/innobase/trx/trx0undo.cc
+++ b/storage/innobase/trx/trx0undo.cc
@@ -34,6 +34,7 @@ Created 3/26/1996 Heikki Tuuri
 #include "trx0purge.h"
 #include "trx0rec.h"
 #include "trx0rseg.h"
+#include "log.h"
 
 /* How should the old versions in the history list be managed?
    ----------------------------------------------------------
@@ -434,7 +435,7 @@ static uint16_t trx_undo_header_create(buf_block_t *undo_page, trx_id_t trx_id,
   repurposed after upgrading to MariaDB 10.3. */
   byte *undo_type= my_assume_aligned<2>
     (TRX_UNDO_PAGE_HDR + TRX_UNDO_PAGE_TYPE + undo_page->frame);
-  ut_ad(mach_read_from_2(undo_type) <= TRX_UNDO_UPDATE);
+  ut_ad(mach_read_from_2(undo_type) <= 2);
   mtr->write<2,mtr_t::MAYBE_NOP>(*undo_page, undo_type, 0U);
   byte *start= my_assume_aligned<4>(TRX_UNDO_PAGE_HDR + TRX_UNDO_PAGE_START +
                                     undo_page->frame);
@@ -790,27 +791,22 @@ done:
 }
 
 /** Frees an undo log segment which is not in the history list.
-@param[in]	undo	undo log
-@param[in]	noredo	whether the undo tablespace is redo logged */
-static void trx_undo_seg_free(const trx_undo_t* undo, bool noredo)
+@param undo	temporary undo log */
+static void trx_undo_seg_free(const trx_undo_t *undo)
 {
 	ut_ad(undo->id < TRX_RSEG_N_SLOTS);
 
 	trx_rseg_t* const rseg = undo->rseg;
 	bool		finished;
 	mtr_t		mtr;
+	ut_ad(rseg->space == fil_system.temp_space);
 
 	do {
 		mtr.start();
-
-		if (noredo) {
-			mtr.set_log_mode(MTR_LOG_NO_REDO);
-		}
-
-		mutex_enter(&rseg->mutex);
+		mtr.set_log_mode(MTR_LOG_NO_REDO);
 
 		buf_block_t* block = trx_undo_page_get(
-			page_id_t(rseg->space->id, undo->hdr_page_no), &mtr);
+			page_id_t(SRV_TMP_SPACE_ID, undo->hdr_page_no), &mtr);
 
 		fseg_header_t* file_seg = TRX_UNDO_SEG_HDR
 			+ TRX_UNDO_FSEG_HEADER + block->frame;
@@ -822,12 +818,12 @@ static void trx_undo_seg_free(const trx_undo_t* undo, bool noredo)
 			buf_block_t* rseg_header = trx_rsegf_get(
 				rseg->space, rseg->page_no, &mtr);
 			compile_time_assert(FIL_NULL == 0xffffffff);
-			mtr.memset(rseg_header, TRX_RSEG + TRX_RSEG_UNDO_SLOTS
-				   + undo->id * TRX_RSEG_SLOT_SIZE, 4, 0xff);
+			memset(TRX_RSEG + TRX_RSEG_UNDO_SLOTS
+			       + undo->id * TRX_RSEG_SLOT_SIZE +
+			       rseg_header->frame, 0xff, 4);
 			MONITOR_DEC(MONITOR_NUM_UNDO_SLOT_USED);
 		}
 
-		mutex_exit(&rseg->mutex);
 		mtr.commit();
 	} while (!finished);
 }
@@ -839,8 +835,9 @@ static void trx_undo_seg_free(const trx_undo_t* undo, bool noredo)
 @param[in]	id		rollback segment slot
 @param[in]	page_no		undo log segment page number
 @param[in,out]	max_trx_id	the largest observed transaction ID
-@return	size of the undo log in pages */
-uint32_t
+@return	the undo log
+@retval nullptr on error */
+trx_undo_t *
 trx_undo_mem_create_at_db_start(trx_rseg_t *rseg, ulint id, uint32_t page_no,
                                 trx_id_t &max_trx_id)
 {
@@ -852,16 +849,56 @@ trx_undo_mem_create_at_db_start(trx_rseg_t *rseg, ulint id, uint32_t page_no,
 	mtr.start();
 	const buf_block_t* block = trx_undo_page_get(
 		page_id_t(rseg->space->id, page_no), &mtr);
-	const ulint type = mach_read_from_2(
-		TRX_UNDO_PAGE_HDR + TRX_UNDO_PAGE_TYPE + block->frame);
-	ut_ad(type == 0 || type == TRX_UNDO_INSERT || type == TRX_UNDO_UPDATE);
+	const uint16_t type = mach_read_from_2(TRX_UNDO_PAGE_HDR
+					       + TRX_UNDO_PAGE_TYPE
+					       + block->frame);
+	switch (type) {
+	case 0:
+	case 2: /* TRX_UNDO_UPDATE */
+		break;
+	case 1: /* TRX_UNDO_INSERT */
+		sql_print_error("InnoDB: upgrade from older version than"
+				" MariaDB 10.3 requires clean shutdown");
+		goto corrupted;
+	default:
+		sql_print_error("InnoDB: unsupported undo header type %u",
+				type);
+	corrupted:
+		mtr.commit();
+		return nullptr;
+	}
 
-	uint16_t state = mach_read_from_2(TRX_UNDO_SEG_HDR + TRX_UNDO_STATE
-					  + block->frame);
 	uint16_t offset = mach_read_from_2(TRX_UNDO_SEG_HDR + TRX_UNDO_LAST_LOG
 					   + block->frame);
+	if (offset < TRX_UNDO_SEG_HDR + TRX_UNDO_SEG_HDR_SIZE ||
+	    offset >= srv_page_size - TRX_UNDO_LOG_OLD_HDR_SIZE) {
+		sql_print_error("InnoDB: invalid undo header offset %u",
+				offset);
+		goto corrupted;
+	}
 
-	const trx_ulogf_t*	undo_header = block->frame + offset;
+	const trx_ulogf_t* const undo_header = block->frame + offset;
+	uint16_t state = mach_read_from_2(TRX_UNDO_SEG_HDR + TRX_UNDO_STATE
+					  + block->frame);
+	switch (state) {
+	case TRX_UNDO_ACTIVE:
+	case TRX_UNDO_PREPARED:
+		break;
+	default:
+		sql_print_error("InnoDB: unsupported undo header state %u",
+				state);
+		goto corrupted;
+	case TRX_UNDO_TO_PURGE:
+	case TRX_UNDO_CACHED:
+		trx_id_t id = mach_read_from_8(TRX_UNDO_TRX_NO + undo_header);
+		if (id >> 48) {
+			sql_print_error("InnoDB: corrupted TRX_NO %llx", id);
+			goto corrupted;
+		}
+		if (id > max_trx_id) {
+			max_trx_id = id;
+		}
+	}
 
 	/* Read X/Open XA transaction identification if it exists, or
 	set it to NULL. */
@@ -873,6 +910,10 @@ trx_undo_mem_create_at_db_start(trx_rseg_t *rseg, ulint id, uint32_t page_no,
 	}
 
 	trx_id_t trx_id = mach_read_from_8(undo_header + TRX_UNDO_TRX_ID);
+	if (trx_id >> 48) {
+		sql_print_error("InnoDB: corrupted TRX_ID %llx", trx_id);
+		goto corrupted;
+	}
 	if (trx_id > max_trx_id) {
 		max_trx_id = trx_id;
 	}
@@ -881,60 +922,45 @@ trx_undo_mem_create_at_db_start(trx_rseg_t *rseg, ulint id, uint32_t page_no,
 	trx_undo_t* undo = trx_undo_mem_create(
 		rseg, id, trx_id, &xid, page_no, offset);
 	mutex_exit(&rseg->mutex);
+	if (!undo) {
+		return undo;
+	}
 
 	undo->dict_operation = undo_header[TRX_UNDO_DICT_TRANS];
 	undo->table_id = mach_read_from_8(undo_header + TRX_UNDO_TABLE_ID);
 	undo->size = flst_get_len(TRX_UNDO_SEG_HDR + TRX_UNDO_PAGE_LIST
 				  + block->frame);
 
-	if (UNIV_UNLIKELY(state == TRX_UNDO_TO_FREE)) {
-		/* This is an old-format insert_undo log segment that
-		is being freed. The page list is inconsistent. */
-		ut_ad(type == TRX_UNDO_INSERT);
-		state = TRX_UNDO_TO_PURGE;
-	} else {
-		if (state == TRX_UNDO_TO_PURGE
-		    || state == TRX_UNDO_CACHED) {
-			trx_id_t id = mach_read_from_8(TRX_UNDO_TRX_NO
-						       + undo_header);
-			if (id > max_trx_id) {
-				max_trx_id = id;
-			}
-		}
-
-		fil_addr_t	last_addr = flst_get_last(
-			TRX_UNDO_SEG_HDR + TRX_UNDO_PAGE_LIST + block->frame);
+	fil_addr_t	last_addr = flst_get_last(
+		TRX_UNDO_SEG_HDR + TRX_UNDO_PAGE_LIST + block->frame);
 
-		undo->last_page_no = last_addr.page;
-		undo->top_page_no = last_addr.page;
+	undo->last_page_no = last_addr.page;
+	undo->top_page_no = last_addr.page;
 
-		const buf_block_t* last = trx_undo_page_get(
-			page_id_t(rseg->space->id, undo->last_page_no), &mtr);
+	const buf_block_t* last = trx_undo_page_get(
+		page_id_t(rseg->space->id, undo->last_page_no), &mtr);
 
-		if (const trx_undo_rec_t* rec = trx_undo_page_get_last_rec(
-			    last, page_no, offset)) {
-			undo->top_offset = uint16_t(rec - last->frame);
-			undo->top_undo_no = trx_undo_rec_get_undo_no(rec);
-			ut_ad(!undo->empty());
-		} else {
-			undo->top_undo_no = IB_ID_MAX;
-			ut_ad(undo->empty());
-		}
+	if (const trx_undo_rec_t* rec = trx_undo_page_get_last_rec(
+		    last, page_no, offset)) {
+		undo->top_offset = static_cast<uint16_t>(rec - last->frame);
+		undo->top_undo_no = trx_undo_rec_get_undo_no(rec);
+		ut_ad(!undo->empty());
+	} else {
+		undo->top_undo_no = IB_ID_MAX;
+		ut_ad(undo->empty());
 	}
 
 	undo->state = state;
 
 	if (state != TRX_UNDO_CACHED) {
-		UT_LIST_ADD_LAST(type == TRX_UNDO_INSERT
-				 ? rseg->old_insert_list
-				 : rseg->undo_list, undo);
+		UT_LIST_ADD_LAST(rseg->undo_list, undo);
 	} else {
 		UT_LIST_ADD_LAST(rseg->undo_cached, undo);
 		MONITOR_INC(MONITOR_NUM_UNDO_SLOT_CACHED);
 	}
 
 	mtr.commit();
-	return undo->size;
+	return undo;
 }
 
 /********************************************************************//**
@@ -1304,22 +1330,18 @@ void trx_undo_set_state_at_prepare(trx_t *trx, trx_undo_t *undo, bool rollback,
 	trx_undo_write_xid(block, offset, undo->xid, mtr);
 }
 
-/** Free an old insert or temporary undo log after commit or rollback.
+/** Free temporary undo log after commit or rollback.
 The information is not needed after a commit or rollback, therefore
 the data can be discarded.
-@param[in,out]	undo	undo log
-@param[in]	is_temp	whether this is temporary undo log */
-void
-trx_undo_commit_cleanup(trx_undo_t* undo, bool is_temp)
+@param undo     temporary undo log */
+void trx_undo_commit_cleanup(trx_undo_t *undo)
 {
 	trx_rseg_t*	rseg	= undo->rseg;
-	ut_ad(is_temp == !rseg->is_persistent());
-	ut_ad(!is_temp || 0 == UT_LIST_GET_LEN(rseg->old_insert_list));
+	ut_ad(rseg->space == fil_system.temp_space);
 
 	mutex_enter(&rseg->mutex);
 
-	UT_LIST_REMOVE(is_temp ? rseg->undo_list : rseg->old_insert_list,
-		       undo);
+	UT_LIST_REMOVE(rseg->undo_list, undo);
 
 	if (undo->state == TRX_UNDO_CACHED) {
 		UT_LIST_ADD_FIRST(rseg->undo_cached, undo);
@@ -1328,9 +1350,7 @@ trx_undo_commit_cleanup(trx_undo_t* undo, bool is_temp)
 		ut_ad(undo->state == TRX_UNDO_TO_PURGE);
 
 		/* Delete first the undo log segment in the file */
-		mutex_exit(&rseg->mutex);
-		trx_undo_seg_free(undo, is_temp);
-		mutex_enter(&rseg->mutex);
+		trx_undo_seg_free(undo);
 
 		ut_ad(rseg->curr_size > undo->size);
 		rseg->curr_size -= undo->size;
@@ -1342,15 +1362,13 @@ trx_undo_commit_cleanup(trx_undo_t* undo, bool is_temp)
 }
 
 /** At shutdown, frees the undo logs of a transaction. */
-void
-trx_undo_free_at_shutdown(trx_t *trx)
+void trx_undo_free_at_shutdown(trx_t *trx)
 {
 	if (trx_undo_t*& undo = trx->rsegs.m_redo.undo) {
 		switch (undo->state) {
 		case TRX_UNDO_PREPARED:
 			break;
 		case TRX_UNDO_CACHED:
-		case TRX_UNDO_TO_FREE:
 		case TRX_UNDO_TO_PURGE:
 			ut_ad(trx_state_eq(trx,
 					   TRX_STATE_COMMITTED_IN_MEMORY));
@@ -1371,34 +1389,6 @@ trx_undo_free_at_shutdown(trx_t *trx)
 		ut_free(undo);
 		undo = NULL;
 	}
-
-	if (trx_undo_t*& undo = trx->rsegs.m_redo.old_insert) {
-		switch (undo->state) {
-		case TRX_UNDO_PREPARED:
-			break;
-		case TRX_UNDO_CACHED:
-		case TRX_UNDO_TO_FREE:
-		case TRX_UNDO_TO_PURGE:
-			ut_ad(trx_state_eq(trx,
-					   TRX_STATE_COMMITTED_IN_MEMORY));
-			/* fall through */
-		case TRX_UNDO_ACTIVE:
-			/* trx_t::commit_state() assigns
-			trx->state = TRX_STATE_COMMITTED_IN_MEMORY. */
-			ut_a(!srv_was_started
-			     || srv_read_only_mode
-			     || srv_force_recovery >= SRV_FORCE_NO_TRX_UNDO
-			     || srv_fast_shutdown);
-			break;
-		default:
-			ut_error;
-		}
-
-		UT_LIST_REMOVE(trx->rsegs.m_redo.rseg->old_insert_list, undo);
-		ut_free(undo);
-		undo = NULL;
-	}
-
 	if (trx_undo_t*& undo = trx->rsegs.m_noredo.undo) {
 		ut_a(undo->state == TRX_UNDO_PREPARED);
 
diff --git a/storage/maria/ma_check.c b/storage/maria/ma_check.c
index 5b37cac74d5..2b65cd36418 100644
--- a/storage/maria/ma_check.c
+++ b/storage/maria/ma_check.c
@@ -4286,7 +4286,7 @@ int maria_repair_parallel(HA_CHECK *param, register MARIA_HA *info,
 			const char * name, my_bool rep_quick)
 {
   int got_error;
-  uint i,key, total_key_length, istep;
+  uint i,key, istep;
   ha_rows start_records;
   my_off_t new_header_length,del;
   File new_file;
@@ -4448,7 +4448,9 @@ int maria_repair_parallel(HA_CHECK *param, register MARIA_HA *info,
     _ma_check_print_error(param,"Not enough memory for key!");
     goto err;
   }
-  total_key_length=0;
+#ifdef USING_SECOND_APPROACH
+  uint total_key_length=0;
+#endif
   rec_per_key_part= param->new_rec_per_key_part;
   share->state.state.records=share->state.state.del=share->state.split=0;
   share->state.state.empty=0;
@@ -4517,7 +4519,9 @@ int maria_repair_parallel(HA_CHECK *param, register MARIA_HA *info,
       if (keyseg->flag & HA_NULL_PART)
         sort_param[i].key_length++;
     }
+#ifdef USING_SECOND_APPROACH
     total_key_length+=sort_param[i].key_length;
+#endif
 
     if (sort_param[i].keyinfo->flag & HA_FULLTEXT)
     {
diff --git a/storage/myisam/mi_check.c b/storage/myisam/mi_check.c
index b97c9656ab1..16184a1b166 100644
--- a/storage/myisam/mi_check.c
+++ b/storage/myisam/mi_check.c
@@ -2617,7 +2617,7 @@ int mi_repair_parallel(HA_CHECK *param, register MI_INFO *info,
 			const char * name, int rep_quick)
 {
   int got_error;
-  uint i,key, total_key_length, istep;
+  uint i,key, istep;
   ulong rec_length;
   ha_rows start_records;
   my_off_t new_header_length,del;
@@ -2803,7 +2803,9 @@ int mi_repair_parallel(HA_CHECK *param, register MI_INFO *info,
     mi_check_print_error(param,"Not enough memory for key!");
     goto err;
   }
-  total_key_length=0;
+#ifdef USING_SECOND_APPROACH
+  uint total_key_length=0;
+#endif
   rec_per_key_part= param->rec_per_key_part;
   info->state->records=info->state->del=share->state.split=0;
   info->state->empty=0;
@@ -2872,7 +2874,9 @@ int mi_repair_parallel(HA_CHECK *param, register MI_INFO *info,
       if (keyseg->flag & HA_NULL_PART)
         sort_param[i].key_length++;
     }
+#ifdef USING_SECOND_APPROACH
     total_key_length+=sort_param[i].key_length;
+#endif
 
     if (sort_param[i].keyinfo->flag & HA_FULLTEXT)
     {
diff --git a/storage/spider/mysql-test/spider/bg/suite.pm b/storage/spider/mysql-test/spider/bg/suite.pm
index f106147deb6..171fa6c4f01 100644
--- a/storage/spider/mysql-test/spider/bg/suite.pm
+++ b/storage/spider/mysql-test/spider/bg/suite.pm
@@ -4,7 +4,6 @@ package My::Suite::Spider;
 
 return "No Spider engine" unless $ENV{HA_SPIDER_SO};
 return "Not run for embedded server" if $::opt_embedded_server;
-return "Test needs --big-test" unless $::opt_big_test;
 
 sub is_default { 1 }
 
diff --git a/storage/spider/mysql-test/spider/bugfix/suite.pm b/storage/spider/mysql-test/spider/bugfix/suite.pm
index f106147deb6..171fa6c4f01 100644
--- a/storage/spider/mysql-test/spider/bugfix/suite.pm
+++ b/storage/spider/mysql-test/spider/bugfix/suite.pm
@@ -4,7 +4,6 @@ package My::Suite::Spider;
 
 return "No Spider engine" unless $ENV{HA_SPIDER_SO};
 return "Not run for embedded server" if $::opt_embedded_server;
-return "Test needs --big-test" unless $::opt_big_test;
 
 sub is_default { 1 }
 
diff --git a/storage/spider/mysql-test/spider/bugfix/t/mdev_20100.test b/storage/spider/mysql-test/spider/bugfix/t/mdev_20100.test
index b72facd11a6..3f522b4f6f5 100644
--- a/storage/spider/mysql-test/spider/bugfix/t/mdev_20100.test
+++ b/storage/spider/mysql-test/spider/bugfix/t/mdev_20100.test
@@ -22,7 +22,9 @@ USE auto_test_remote;
 --connection child2_1
 --disable_query_log
 echo CHILD2_1_CREATE_TABLES;
+--disable_ps_protocol
 eval $CHILD2_1_CREATE_TABLES;
+--enable_ps_protocol
 --enable_query_log
 TRUNCATE TABLE mysql.general_log;
 
@@ -66,7 +68,9 @@ SELECT a, b, c FROM tbl_a PARTITION (pt2,pt3);
 
 --connection child2_1
 eval $CHILD2_1_SELECT_ARGUMENT1;
+--disable_ps_protocol
 eval $CHILD2_1_SELECT_TABLES;
+--enable_ps_protocol
 
 --echo
 --echo deinit
diff --git a/storage/spider/mysql-test/spider/bugfix/t/mdev_21884.test b/storage/spider/mysql-test/spider/bugfix/t/mdev_21884.test
index be4aa6a6330..16836cfce28 100644
--- a/storage/spider/mysql-test/spider/bugfix/t/mdev_21884.test
+++ b/storage/spider/mysql-test/spider/bugfix/t/mdev_21884.test
@@ -22,7 +22,9 @@ USE auto_test_remote;
 --connection child2_1
 --disable_query_log
 echo CHILD2_1_CREATE_TABLES;
+--disable_ps_protocol
 eval $CHILD2_1_CREATE_TABLES;
+--enable_ps_protocol
 --enable_query_log
 TRUNCATE TABLE mysql.general_log;
 
@@ -78,7 +80,9 @@ SELECT STRAIGHT_JOIN b.a, b.b FROM tb_l a, tbl_a b WHERE a.a = b.a;
 --connection child2_1
 SET NAMES utf8;
 eval $CHILD2_1_SELECT_ARGUMENT1;
+--disable_ps_protocol
 eval $CHILD2_1_SELECT_TABLES;
+--enable_ps_protocol
 
 --echo
 --echo deinit
diff --git a/storage/spider/mysql-test/spider/bugfix/t/wrapper_mariadb.test b/storage/spider/mysql-test/spider/bugfix/t/wrapper_mariadb.test
index 0102155b5ab..4f980140a31 100644
--- a/storage/spider/mysql-test/spider/bugfix/t/wrapper_mariadb.test
+++ b/storage/spider/mysql-test/spider/bugfix/t/wrapper_mariadb.test
@@ -49,6 +49,8 @@ TRUNCATE TABLE mysql.general_log;
 SELECT * FROM tbl_a ORDER BY pkey;
 
 --connection child2_1
+# in --ps a query is logged differently in a general log
+replace_regex /order by t0.`pkey`/order by `pkey`/;
 eval $CHILD2_1_SELECT_ARGUMENT1;
 eval $CHILD2_1_SELECT_TABLES;
 
diff --git a/storage/spider/mysql-test/spider/feature/suite.pm b/storage/spider/mysql-test/spider/feature/suite.pm
index f106147deb6..171fa6c4f01 100644
--- a/storage/spider/mysql-test/spider/feature/suite.pm
+++ b/storage/spider/mysql-test/spider/feature/suite.pm
@@ -4,7 +4,6 @@ package My::Suite::Spider;
 
 return "No Spider engine" unless $ENV{HA_SPIDER_SO};
 return "Not run for embedded server" if $::opt_embedded_server;
-return "Test needs --big-test" unless $::opt_big_test;
 
 sub is_default { 1 }
 
diff --git a/storage/spider/mysql-test/spider/handler/suite.pm b/storage/spider/mysql-test/spider/handler/suite.pm
index b023e5206ef..af267d047a4 100644
--- a/storage/spider/mysql-test/spider/handler/suite.pm
+++ b/storage/spider/mysql-test/spider/handler/suite.pm
@@ -4,7 +4,6 @@ package My::Suite::Spider;
 
 return "No Spider engine" unless $ENV{HA_SPIDER_SO};
 return "Not run for embedded server" if $::opt_embedded_server;
-return "Test needs --big-test" unless $::opt_big_test;
 
 sub is_default { 1 }
 
diff --git a/storage/spider/mysql-test/spider/regression/e1121/suite.pm b/storage/spider/mysql-test/spider/regression/e1121/suite.pm
index f106147deb6..171fa6c4f01 100644
--- a/storage/spider/mysql-test/spider/regression/e1121/suite.pm
+++ b/storage/spider/mysql-test/spider/regression/e1121/suite.pm
@@ -4,7 +4,6 @@ package My::Suite::Spider;
 
 return "No Spider engine" unless $ENV{HA_SPIDER_SO};
 return "Not run for embedded server" if $::opt_embedded_server;
-return "Test needs --big-test" unless $::opt_big_test;
 
 sub is_default { 1 }
 
diff --git a/storage/spider/mysql-test/spider/regression/e1121/t/direct_join_by_pkey_key.test b/storage/spider/mysql-test/spider/regression/e1121/t/direct_join_by_pkey_key.test
index e915a21fd4a..39b5b5535bb 100644
--- a/storage/spider/mysql-test/spider/regression/e1121/t/direct_join_by_pkey_key.test
+++ b/storage/spider/mysql-test/spider/regression/e1121/t/direct_join_by_pkey_key.test
@@ -21,8 +21,10 @@ USE auto_test_remote;
 
 --connection child2_1
 --disable_query_log
+--disable_ps_protocol
 echo CHILD2_1_CREATE_TABLES;
 eval $CHILD2_1_CREATE_TABLES;
+--enable_ps_protocol
 --enable_query_log
 TRUNCATE TABLE mysql.general_log;
 
@@ -62,8 +64,10 @@ TRUNCATE TABLE mysql.general_log;
 SELECT a.val, a.akey FROM tbl_a a, tbl_b b WHERE a.akey = b.akey AND b.bkey = 5;
 
 --connection child2_1
+--disable_ps_protocol
 eval $CHILD2_1_SELECT_ARGUMENT1;
 eval $CHILD2_1_SELECT_TABLES;
+--enable_ps_protocol
 
 --echo
 --echo deinit
diff --git a/storage/spider/mysql-test/spider/regression/e1121/t/direct_join_by_pkey_pkey.test b/storage/spider/mysql-test/spider/regression/e1121/t/direct_join_by_pkey_pkey.test
index dcd6e3a4535..652f7d15177 100644
--- a/storage/spider/mysql-test/spider/regression/e1121/t/direct_join_by_pkey_pkey.test
+++ b/storage/spider/mysql-test/spider/regression/e1121/t/direct_join_by_pkey_pkey.test
@@ -22,7 +22,9 @@ USE auto_test_remote;
 --connection child2_1
 --disable_query_log
 echo CHILD2_1_CREATE_TABLES;
+--disable_ps_protocol
 eval $CHILD2_1_CREATE_TABLES;
+--enable_ps_protocol
 --enable_query_log
 TRUNCATE TABLE mysql.general_log;
 
@@ -62,8 +64,10 @@ TRUNCATE TABLE mysql.general_log;
 SELECT a.val, a.akey FROM tbl_a a, tbl_b b WHERE a.akey = b.akey AND b.bkey = 5;
 
 --connection child2_1
+--disable_ps_protocol
 eval $CHILD2_1_SELECT_ARGUMENT1;
 eval $CHILD2_1_SELECT_TABLES;
+--enable_ps_protocol
 
 --echo
 --echo deinit
diff --git a/storage/spider/mysql-test/spider/regression/e112122/suite.pm b/storage/spider/mysql-test/spider/regression/e112122/suite.pm
index f106147deb6..171fa6c4f01 100644
--- a/storage/spider/mysql-test/spider/regression/e112122/suite.pm
+++ b/storage/spider/mysql-test/spider/regression/e112122/suite.pm
@@ -4,7 +4,6 @@ package My::Suite::Spider;
 
 return "No Spider engine" unless $ENV{HA_SPIDER_SO};
 return "Not run for embedded server" if $::opt_embedded_server;
-return "Test needs --big-test" unless $::opt_big_test;
 
 sub is_default { 1 }
 
diff --git a/storage/spider/mysql-test/spider/suite.pm b/storage/spider/mysql-test/spider/suite.pm
index f106147deb6..171fa6c4f01 100644
--- a/storage/spider/mysql-test/spider/suite.pm
+++ b/storage/spider/mysql-test/spider/suite.pm
@@ -4,7 +4,6 @@ package My::Suite::Spider;
 
 return "No Spider engine" unless $ENV{HA_SPIDER_SO};
 return "Not run for embedded server" if $::opt_embedded_server;
-return "Test needs --big-test" unless $::opt_big_test;
 
 sub is_default { 1 }
 
diff --git a/storage/spider/mysql-test/spider/t/partition_mrr.test b/storage/spider/mysql-test/spider/t/partition_mrr.test
index e7fedce33e6..2816d65cadb 100644
--- a/storage/spider/mysql-test/spider/t/partition_mrr.test
+++ b/storage/spider/mysql-test/spider/t/partition_mrr.test
@@ -1,3 +1,4 @@
+--source include/no_protocol.inc
 --source ../include/partition_mrr_init.inc
 if (!$HAVE_PARTITION)
 {