Merge branch 10.6 into 10.7

libmariadb

@@ -1 +1 @@
-Subproject commit 380ee32375bb36b68796c1c3eb09285f03fea5f5
+Subproject commit 72b40bfaa869f3fe84242471dda989d13983d84c

mysql-test/main/ps_missed_cmds_bin_prot.result

@@ -104,32 +104,11 @@ HANDLER t1 CLOSE;
 DROP TABLE t1;
 # Test case 10: Check that the statements 'HELP'
 # is supported by prepared statements
-HELP `ALTER SERVER`;
+INSERT INTO mysql.help_topic VALUES (0, 'Tamagotchi', 0, 'This digital pet is not a KB article', 'no example', 'https://tamagotchi.com/');
+HELP `Tamagotchi`;
 name	description	example
-ALTER SERVER	Syntax
+Tamagotchi	This digital pet is not a KB article	no example
-------
+DELETE FROM mysql.help_topic WHERE help_topic_id = 0;
-
-ALTER SERVER server_name
-  OPTIONS (option [, option] ...)
-
-Description
------------
-
-Alters the server information for server_name, adjusting the specified options
-as per the CREATE SERVER command. The corresponding fields in the
-mysql.servers table are updated accordingly. This statement requires the SUPER
-privilege or, from MariaDB 10.5.2, the FEDERATED ADMIN privilege.
-
-ALTER SERVER is not written to the binary log, irrespective of the binary log
-format being used. From MariaDB 10.1.13, Galera replicates the CREATE SERVER,
-ALTER SERVER and DROP SERVER statements.
-
-Examples
---------
-
-ALTER SERVER s OPTIONS (USER 'sally');
-
-URL: mariadb.com/kb/en/alter-server/	
 # Test case 11: Check that the statements CREATE/ALTER/DROP PROCEDURE
 # are supported by prepared statements
 CREATE PROCEDURE p1() SET @a=1;

mysql-test/main/ps_missed_cmds_bin_prot.test

@@ -128,7 +128,10 @@ DROP TABLE t1;
 
 --echo # Test case 10: Check that the statements 'HELP'
 --echo # is supported by prepared statements
-HELP `ALTER SERVER`;
+# avoid existing articles that may get updated.
+INSERT INTO mysql.help_topic VALUES (0, 'Tamagotchi', 0, 'This digital pet is not a KB article', 'no example', 'https://tamagotchi.com/');
+HELP `Tamagotchi`;
+DELETE FROM mysql.help_topic WHERE help_topic_id = 0;
 
 --echo # Test case 11: Check that the statements CREATE/ALTER/DROP PROCEDURE
 --echo # are supported by prepared statements

mysql-test/suite/gcol/r/innodb_virtual_purge.result

@@ -24,7 +24,7 @@ COMMIT;
 UPDATE t1 SET a=1;
 connection default;
 InnoDB		0 transactions not purged
-CHECK TABLE t1;
+CHECK TABLE t1 EXTENDED;
 Table	Op	Msg_type	Msg_text
 test.t1	check	status	OK
 SELECT b1 FROM t1;
@@ -123,7 +123,7 @@ COMMIT;
 disconnect con1;
 connection default;
 InnoDB		0 transactions not purged
-CHECK TABLE t1;
+CHECK TABLE t1 EXTENDED;
 Table	Op	Msg_type	Msg_text
 test.t1	check	status	OK
 SELECT b1 FROM t1;
@@ -134,7 +134,7 @@ SELECT * FROM t1;
 a	b	b1	a1	a4	b3
 100	10	10	100	90	100
 100	10	10	100	90	100
-CHECK TABLE t2;
+CHECK TABLE t2 EXTENDED;
 Table	Op	Msg_type	Msg_text
 test.t2	check	status	OK
 DROP TABLE t2, t1, t0;

mysql-test/suite/gcol/t/innodb_virtual_purge.test

@@ -38,7 +38,7 @@ UPDATE t1 SET a=1;
 connection default;
 --source ../../innodb/include/wait_all_purged.inc
 
-CHECK TABLE t1;
+CHECK TABLE t1 EXTENDED;
 SELECT b1 FROM t1;
 
 
@@ -123,11 +123,11 @@ disconnect con1;
 connection default;
 --source ../../innodb/include/wait_all_purged.inc
 
-CHECK TABLE t1;
+CHECK TABLE t1 EXTENDED;
 SELECT b1 FROM t1;
 
 SELECT * FROM t1;
-CHECK TABLE t2;
+CHECK TABLE t2 EXTENDED;
 DROP TABLE t2, t1, t0;
 
 CREATE TABLE t1 (a VARCHAR(30), b INT, a2 VARCHAR(30) GENERATED ALWAYS AS (a) VIRTUAL);

mysql-test/suite/innodb/r/deadlock_wait_lock_race.result

@@ -0,0 +1,31 @@
+connect suspend_purge,localhost,root,,;
+START TRANSACTION WITH CONSISTENT SNAPSHOT;
+connection default;
+CREATE TABLE t (a int PRIMARY KEY, b int) engine = InnoDB;
+CREATE TABLE t2 (a int PRIMARY KEY) engine = InnoDB;
+INSERT INTO t VALUES (10, 10), (20, 20), (30, 30);
+INSERT INTO t2 VALUES (10), (20), (30);
+BEGIN;
+UPDATE t2 SET a = a + 100;
+SELECT * FROM t WHERE a = 20 FOR UPDATE;
+a	b
+20	20
+connect con_2,localhost,root,,;
+SET TRANSACTION ISOLATION LEVEL READ COMMITTED;
+BEGIN;
+SET DEBUG_SYNC = 'lock_trx_handle_wait_before_unlocked_wait_lock_check SIGNAL upd_locked WAIT_FOR upd_cont';
+UPDATE t SET b = 100;
+connection default;
+SET DEBUG_SYNC="now WAIT_FOR upd_locked";
+SET DEBUG_SYNC="lock_wait_before_suspend SIGNAL upd_cont";
+SELECT * FROM t WHERE a = 10 FOR UPDATE;
+connection con_2;
+ERROR 40001: Deadlock found when trying to get lock; try restarting transaction
+disconnect con_2;
+connection default;
+a	b
+10	10
+SET DEBUG_SYNC = 'RESET';
+DROP TABLE t;
+DROP TABLE t2;
+disconnect suspend_purge;

mysql-test/suite/innodb/r/deadlock_wait_thr_race.result

@@ -0,0 +1,37 @@
+connect suspend_purge,localhost,root,,;
+START TRANSACTION WITH CONSISTENT SNAPSHOT;
+connection default;
+CREATE TABLE t (a int PRIMARY KEY, b int) engine = InnoDB;
+CREATE TABLE t2 (a int PRIMARY KEY) engine = InnoDB;
+INSERT INTO t VALUES (10, 10), (20, 20), (30, 30);
+INSERT INTO t2 VALUES (10), (20), (30);
+BEGIN;
+UPDATE t2 SET a = a + 100;
+SELECT * FROM t WHERE a = 20 FOR UPDATE;
+a	b
+20	20
+connect con_2,localhost,root,,;
+SET TRANSACTION ISOLATION LEVEL READ COMMITTED;
+SET DEBUG_SYNC = 'lock_trx_handle_wait_enter SIGNAL upd_locked WAIT_FOR upd_cont';
+SET DEBUG_SYNC = 'trx_t_release_locks_enter SIGNAL sel_cont WAIT_FOR upd_cont_2';
+BEGIN;
+UPDATE t SET b = 100;
+connection default;
+SET DEBUG_SYNC="now WAIT_FOR upd_locked";
+SET DEBUG_SYNC="deadlock_report_before_lock_releasing SIGNAL upd_cont WAIT_FOR sel_cont";
+SET DEBUG_SYNC="lock_wait_before_suspend SIGNAL sel_before_suspend";
+SELECT * FROM t WHERE a = 10 FOR UPDATE;;
+connect con_3,localhost,root,,;
+SET DEBUG_SYNC="now WAIT_FOR sel_before_suspend";
+SET DEBUG_SYNC="now SIGNAL upd_cont_2";
+disconnect con_3;
+connection con_2;
+ERROR 40001: Deadlock found when trying to get lock; try restarting transaction
+disconnect con_2;
+connection default;
+a	b
+10	10
+SET DEBUG_SYNC = 'RESET';
+DROP TABLE t;
+DROP TABLE t2;
+disconnect suspend_purge;

mysql-test/suite/innodb/r/trx_id_future.result

@@ -6,13 +6,9 @@ SET GLOBAL innodb_purge_rseg_truncate_frequency=1;
 CREATE TABLE t1(a INT) row_format=redundant engine=innoDB;
 INSERT INTO t1 VALUES(1);
 InnoDB		0 transactions not purged
-NOT FOUND /\[Warning\] InnoDB: A transaction id in a record of table `test`\.`t1` is newer than the system-wide maximum/ in mysqld.1.err
 call mtr.add_suppression("\\[Warning\\] InnoDB: A transaction id in a record of table `test`\\.`t1` is newer than the system-wide maximum");
-SET @save_count = @@max_error_count;
+call mtr.add_suppression("\\[ERROR\\] InnoDB: We detected index corruption");
-SET max_error_count = 1;
+call mtr.add_suppression("Index for table 't1' is corrupt; try to repair it");
 SELECT * FROM t1;
-a
+ERROR HY000: Index for table 't1' is corrupt; try to repair it
-Warnings:
-Warning	1642	InnoDB: Transaction id in a record of table `test`.`t1` is newer than system-wide maximum.
-SET max_error_count = @save_count;
 DROP TABLE t1;

mysql-test/suite/innodb/t/deadlock_wait_lock_race.test

@@ -0,0 +1,62 @@
+--source include/have_innodb.inc
+--source include/have_debug_sync.inc
+--source include/count_sessions.inc
+
+--connect(suspend_purge,localhost,root,,)
+# Purge can cause deadlock in the test, requesting page's RW_X_LATCH for trx
+# ids reseting, after trx 2 acqured RW_S_LATCH and suspended in debug sync point
+# lock_trx_handle_wait_enter, waiting for upd_cont signal, which must be
+# emitted after the last SELECT in this test. The last SELECT will hang waiting
+# for purge RW_X_LATCH releasing, and trx 2 will be rolled back by timeout.
+START TRANSACTION WITH CONSISTENT SNAPSHOT;
+
+--connection default
+CREATE TABLE t (a int PRIMARY KEY, b int) engine = InnoDB;
+CREATE TABLE t2 (a int PRIMARY KEY) engine = InnoDB;
+
+INSERT INTO t VALUES (10, 10), (20, 20), (30, 30);
+INSERT INTO t2 VALUES (10), (20), (30);
+
+BEGIN; # trx 1
+# The following update is necessary to increase the transaction weight, which is
+# calculated as the number of locks + the number of undo records during deadlock
+# report. Victim's transaction should have minimum weight. We need trx 2 to be
+# choosen as victim, that's why we need to increase the current transaction
+# weight.
+UPDATE t2 SET a = a + 100;
+SELECT * FROM t WHERE a = 20 FOR UPDATE;
+
+--connect(con_2,localhost,root,,)
+# RC is necessary to do semi-consistent read
+SET TRANSACTION ISOLATION LEVEL READ COMMITTED;
+BEGIN; # trx 2
+# The first time it will be hit on trying to lock (20,20), the second hit
+# will be on (30,30).
+SET DEBUG_SYNC = 'lock_trx_handle_wait_before_unlocked_wait_lock_check SIGNAL upd_locked WAIT_FOR upd_cont';
+# We must not modify primary key fields to cause rr_sequential() read record
+# function choosing in mysql_update(), i.e. both query_plan.using_filesort and
+# query_plan.using_io_buffer must be false during init_read_record() call.
+--send UPDATE t SET b = 100
+
+--connection default
+SET DEBUG_SYNC="now WAIT_FOR upd_locked";
+SET DEBUG_SYNC="lock_wait_before_suspend SIGNAL upd_cont";
+--send SELECT * FROM t WHERE a = 10 FOR UPDATE
+
+--connection con_2
+# If the bug is not fixed, lock_trx_handle_wait() wrongly returns DB_SUCCESS
+# instead of DB_DEADLOCK, row_search_mvcc() of trx 2 behaves so as if (20,20)
+# was locked. Some debug assertion must crash the server. If the bug is fixed,
+# trx 2 must just be rolled back by deadlock detector.
+--error ER_LOCK_DEADLOCK
+--reap
+
+--disconnect con_2
+
+--connection default
+--reap
+SET DEBUG_SYNC = 'RESET';
+DROP TABLE t;
+DROP TABLE t2;
+--disconnect suspend_purge
+--source include/wait_until_count_sessions.inc

mysql-test/suite/innodb/t/deadlock_wait_thr_race.test

@@ -0,0 +1,66 @@
+--source include/have_innodb.inc
+--source include/have_debug_sync.inc
+--source include/count_sessions.inc
+
+--connect(suspend_purge,localhost,root,,)
+# Purge can cause deadlock in the test, requesting page's RW_X_LATCH for trx
+# ids reseting, after trx 2 acqured RW_S_LATCH and suspended in debug sync point
+# lock_trx_handle_wait_enter, waiting for upd_cont signal, which must be
+# emitted after the last SELECT in this test. The last SELECT will hang waiting
+# for purge RW_X_LATCH releasing, and trx 2 will be rolled back by timeout.
+START TRANSACTION WITH CONSISTENT SNAPSHOT;
+
+--connection default
+CREATE TABLE t (a int PRIMARY KEY, b int) engine = InnoDB;
+CREATE TABLE t2 (a int PRIMARY KEY) engine = InnoDB;
+
+INSERT INTO t VALUES (10, 10), (20, 20), (30, 30);
+INSERT INTO t2 VALUES (10), (20), (30);
+
+BEGIN; # trx 1
+# The following update is necessary to increase the transaction weight, which is
+# calculated as the number of locks + the number of undo records during deadlock
+# report. Victim's transaction should have minimum weight. We need trx 2 to be
+# choosen as victim, that's why we need to increase the current transaction
+# weight.
+UPDATE t2 SET a = a + 100;
+SELECT * FROM t WHERE a = 20 FOR UPDATE;
+
+--connect(con_2,localhost,root,,)
+# RC is necessary to do semi-consistent read
+SET TRANSACTION ISOLATION LEVEL READ COMMITTED;
+# It will be hit on trying to lock (20,20).
+SET DEBUG_SYNC = 'lock_trx_handle_wait_enter SIGNAL upd_locked WAIT_FOR upd_cont';
+SET DEBUG_SYNC = 'trx_t_release_locks_enter SIGNAL sel_cont WAIT_FOR upd_cont_2';
+BEGIN; # trx 2
+# We must not modify primary key fields to cause rr_sequential() read record
+# function choosing in mysql_update(), i.e. both query_plan.using_filesort and
+# query_plan.using_io_buffer must be false during init_read_record() call.
+# The following UPDATE will be chosen as deadlock victim and rolled back.
+--send UPDATE t SET b = 100
+
+--connection default
+SET DEBUG_SYNC="now WAIT_FOR upd_locked";
+SET DEBUG_SYNC="deadlock_report_before_lock_releasing SIGNAL upd_cont WAIT_FOR sel_cont";
+SET DEBUG_SYNC="lock_wait_before_suspend SIGNAL sel_before_suspend";
+# If the bug is not fixed, the following SELECT will crash, as the above UPDATE
+# will reset trx->lock.wait_thr during rollback
+--send SELECT * FROM t WHERE a = 10 FOR UPDATE;
+
+--connect(con_3,localhost,root,,)
+SET DEBUG_SYNC="now WAIT_FOR sel_before_suspend";
+SET DEBUG_SYNC="now SIGNAL upd_cont_2";
+--disconnect con_3
+
+--connection con_2
+--error ER_LOCK_DEADLOCK
+--reap
+--disconnect con_2
+
+--connection default
+--reap
+SET DEBUG_SYNC = 'RESET';
+DROP TABLE t;
+DROP TABLE t2;
+--disconnect suspend_purge
+--source include/wait_until_count_sessions.inc

mysql-test/suite/innodb/t/trx_id_future.test

@@ -57,19 +57,11 @@ syswrite(FILE, $page, $ps)==$ps || die "Unable to write $file\n";
 close(FILE) || die "Unable to close $file";
 EOF
 
-# Debug assertions would fail due to the injected corruption.
---let $restart_parameters= --loose-skip-debug-assert
 --source include/start_mysqld.inc
-
-let SEARCH_FILE= $MYSQLTEST_VARDIR/log/mysqld.1.err;
-let SEARCH_PATTERN= \[Warning\] InnoDB: A transaction id in a record of table `test`\.`t1` is newer than the system-wide maximum;
---source include/search_pattern_in_file.inc
-
 call mtr.add_suppression("\\[Warning\\] InnoDB: A transaction id in a record of table `test`\\.`t1` is newer than the system-wide maximum");
+call mtr.add_suppression("\\[ERROR\\] InnoDB: We detected index corruption");
+call mtr.add_suppression("Index for table 't1' is corrupt; try to repair it");
 
-# A debug assertion would cause a duplicated message to be output.
+--error ER_NOT_KEYFILE
-SET @save_count = @@max_error_count;
-SET max_error_count = 1;
 SELECT * FROM t1;
-SET max_error_count = @save_count;
 DROP TABLE t1;

mysys/my_addr_resolve.c

@@ -324,7 +324,7 @@ int my_addr_resolve(void *ptr, my_addr_loc *loc)
       for the base program. This is depending on if the compilation is
       done with PIE or not.
     */
-    addr_offset= info.dli_fbase;
+    addr_offset= (void*) info.dli_fbase;
 #ifndef __PIE__
     if (strcmp(info.dli_fname, my_progname) == 0 &&
         addr_resolve((void*) my_addr_resolve, loc) == 0 &&

storage/columnstore/columnstore

@@ -1 +1 @@
-Subproject commit ac0741e82d71ea8faf6865c4c6604f02449cfa27
+Subproject commit 1071ce954807104d25c0951f422e83d5e17406fe

storage/innobase/CMakeLists.txt

@@ -328,7 +328,6 @@ SET(INNOBASE_SOURCES
 	include/row0row.h
 	include/row0row.inl
 	include/row0sel.h
-	include/row0sel.inl
 	include/row0types.h
 	include/row0uins.h
 	include/row0umod.h

storage/innobase/btr/btr0btr.cc

@@ -5276,11 +5276,6 @@ btr_validate_index(
 	dict_index_t*	index,	/*!< in: index */
 	const trx_t*	trx)	/*!< in: transaction or NULL */
 {
-  /* Full Text index are implemented by auxiliary tables, not the B-tree */
-  if (index->online_status != ONLINE_INDEX_COMPLETE ||
-      (index->type & (DICT_FTS | DICT_CORRUPT)))
-    return DB_SUCCESS;
-
   const bool lockout= index->is_spatial();
 
   mtr_t mtr;

storage/innobase/dict/dict0stats.cc

@@ -33,6 +33,7 @@ Created Jan 06, 2010 Vasil Dimov
 #include <mysql_com.h>
 #include "log.h"
 #include "btr0btr.h"
+#include "que0que.h"
 
 #include <algorithm>
 #include <map>

storage/innobase/fts/fts0ast.cc

@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 2007, 2020, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2018, MariaDB Corporation.
+Copyright (c) 2018, 2022, 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
@@ -28,6 +28,7 @@ Created 2007/3/16 Sunny Bains.
 #include "fts0ast.h"
 #include "fts0pars.h"
 #include "fts0fts.h"
+#include "trx0trx.h"
 
 /* The FTS ast visit pass. */
 enum fts_ast_visit_pass_t {

storage/innobase/handler/ha_innodb.cc

@@ -3657,7 +3657,7 @@ ha_innobase::init_table_handle_for_HANDLER(void)
 	innobase_register_trx(ht, m_user_thd, m_prebuilt->trx);
 
 	/* We did the necessary inits in this function, no need to repeat them
-	in row_search_for_mysql */
+	in row_search_mvcc() */
 
 	m_prebuilt->sql_stat_start = FALSE;
 
@@ -7395,7 +7395,7 @@ ha_innobase::build_template(
 			/* We must at least fetch all primary key cols. Note
 			that if the clustered index was internally generated
 			by InnoDB on the row id (no primary key was
-			defined), then row_search_for_mysql() will always
+			defined), then row_search_mvcc() will always
 			retrieve the row id to a special buffer in the
 			m_prebuilt struct. */
 
@@ -8927,7 +8927,7 @@ statement issued by the user. We also increment trx->n_mysql_tables_in_use.
 instructions to m_prebuilt->template of the table handle instance in
 ::index_read. The template is used to save CPU time in large joins.
 
-  3) In row_search_for_mysql, if m_prebuilt->sql_stat_start is true, we
+  3) In row_search_mvcc(), if m_prebuilt->sql_stat_start is true, we
 allocate a new consistent read view for the trx if it does not yet have one,
 or in the case of a locking read, set an InnoDB 'intention' table level
 lock on the table.
@@ -9229,7 +9229,7 @@ ha_innobase::change_active_index(
 		}
 
 		/* The caller seems to ignore this.  Thus, we must check
-		this again in row_search_for_mysql(). */
+		this again in row_search_mvcc(). */
 		DBUG_RETURN(convert_error_code_to_mysql(DB_MISSING_HISTORY,
 				0, NULL));
 	}
@@ -9829,7 +9829,7 @@ next_record:
 
 		int	error;
 
-		switch (dberr_t ret = row_search_for_mysql(buf, PAGE_CUR_GE,
+		switch (dberr_t ret = row_search_mvcc(buf, PAGE_CUR_GE,
 						      m_prebuilt,
 						      ROW_SEL_EXACT, 0)) {
 		case DB_SUCCESS:
@@ -15167,8 +15167,10 @@ ha_innobase::check(
 
 	DBUG_ENTER("ha_innobase::check");
 	DBUG_ASSERT(thd == ha_thd());
+	DBUG_ASSERT(thd == m_user_thd);
 	ut_a(m_prebuilt->trx->magic_n == TRX_MAGIC_N);
 	ut_a(m_prebuilt->trx == thd_to_trx(thd));
+	ut_ad(m_prebuilt->trx->mysql_thd == thd);
 
 	if (m_prebuilt->mysql_template == NULL) {
 		/* Build the template; we will use a dummy template
@@ -15178,7 +15180,6 @@ ha_innobase::check(
 	}
 
 	if (!m_prebuilt->table->space) {
-
 		ib_senderrf(
 			thd,
 			IB_LOG_LEVEL_ERROR,
@@ -15186,10 +15187,7 @@ ha_innobase::check(
 			table->s->table_name.str);
 
 		DBUG_RETURN(HA_ADMIN_CORRUPT);
-
+	} else if (!m_prebuilt->table->is_readable()) {
-	} else if (!m_prebuilt->table->is_readable() &&
-		   !m_prebuilt->table->space) {
-
 		ib_senderrf(
 			thd, IB_LOG_LEVEL_ERROR,
 			ER_TABLESPACE_MISSING,
@@ -15210,6 +15208,9 @@ ha_innobase::check(
 		? TRX_ISO_READ_UNCOMMITTED
 		: TRX_ISO_REPEATABLE_READ;
 
+	trx_start_if_not_started(m_prebuilt->trx, false);
+	m_prebuilt->trx->read_view.open(m_prebuilt->trx);
+
 	for (dict_index_t* index
 	     = dict_table_get_first_index(m_prebuilt->table);
 	     index;
@@ -15218,18 +15219,16 @@ ha_innobase::check(
 		if (!index->is_committed()) {
 			continue;
 		}
+		if (index->type & DICT_FTS) {
+			/* We do not check any FULLTEXT INDEX. */
+			continue;
+		}
 
-		if (!(check_opt->flags & T_QUICK)
+		if ((check_opt->flags & T_QUICK) || index->is_corrupted()) {
-		    && !index->is_corrupted()) {
+		} else if (btr_validate_index(index, m_prebuilt->trx)
-
+			   != DB_SUCCESS) {
-			dberr_t err = btr_validate_index(
-					index, m_prebuilt->trx);
-
-			if (err != DB_SUCCESS) {
 			is_ok = false;
-
+			push_warning_printf(thd,
-				push_warning_printf(
-					thd,
 					    Sql_condition::WARN_LEVEL_WARN,
 					    ER_NOT_KEYFILE,
 					    "InnoDB: The B-tree of"
@@ -15237,7 +15236,6 @@ ha_innobase::check(
 					    index->name());
 			continue;
 		}
-		}
 
 		/* Instead of invoking change_active_index(), set up
 		a dummy template for non-locking reads, disabling
@@ -15258,7 +15256,7 @@ ha_innobase::check(
 		if (UNIV_UNLIKELY(!m_prebuilt->index_usable)) {
 			if (index->is_corrupted()) {
 				push_warning_printf(
-					m_user_thd,
+					thd,
 					Sql_condition::WARN_LEVEL_WARN,
 					HA_ERR_INDEX_CORRUPT,
 					"InnoDB: Index %s is marked as"
@@ -15267,7 +15265,7 @@ ha_innobase::check(
 				is_ok = false;
 			} else {
 				push_warning_printf(
-					m_user_thd,
+					thd,
 					Sql_condition::WARN_LEVEL_WARN,
 					HA_ERR_TABLE_DEF_CHANGED,
 					"InnoDB: Insufficient history for"
@@ -15280,18 +15278,22 @@ ha_innobase::check(
 		m_prebuilt->sql_stat_start = TRUE;
 		m_prebuilt->template_type = ROW_MYSQL_DUMMY_TEMPLATE;
 		m_prebuilt->n_template = 0;
-		m_prebuilt->need_to_access_clustered = FALSE;
+		m_prebuilt->read_just_key = 0;
+		m_prebuilt->autoinc_error = DB_SUCCESS;
+		m_prebuilt->need_to_access_clustered =
+			!!(check_opt->flags & T_EXTEND);
 
 		dtuple_set_n_fields(m_prebuilt->search_tuple, 0);
 
 		m_prebuilt->select_lock_type = LOCK_NONE;
 
 		/* Scan this index. */
-		if (dict_index_is_spatial(index)) {
+		if (index->is_spatial()) {
 			ret = row_count_rtree_recs(m_prebuilt, &n_rows);
+		} else if (index->type & DICT_FTS) {
+			ret = DB_SUCCESS;
 		} else {
-			ret = row_scan_index_for_mysql(
+			ret = row_check_index(m_prebuilt, &n_rows);
-				m_prebuilt, index, &n_rows);
 		}
 
 		DBUG_EXECUTE_IF(
@@ -15300,11 +15302,18 @@ ha_innobase::check(
 				ret = DB_CORRUPTION;
 			});
 
-		if (ret == DB_INTERRUPTED || thd_killed(m_user_thd)) {
+		if (ret == DB_INTERRUPTED || thd_killed(thd)) {
 			/* Do not report error since this could happen
 			during shutdown */
 			break;
 		}
+
+		if (ret == DB_SUCCESS
+		    && m_prebuilt->autoinc_error != DB_MISSING_HISTORY) {
+			/* See if any non-fatal errors were reported. */
+			ret = m_prebuilt->autoinc_error;
+		}
+
 		if (ret != DB_SUCCESS) {
 			/* Assume some kind of corruption. */
 			push_warning_printf(

storage/innobase/include/read0types.h

@@ -121,19 +121,6 @@ loop:
   inline void snapshot(trx_t *trx);
 
 
-  /**
-    Check whether transaction id is valid.
-    @param[in] id transaction id to check
-    @param[in] name table name
-
-    @todo changes_visible() was an unfortunate choice for this check.
-    It should be moved towards the functions that load trx id like
-    trx_read_trx_id(). No need to issue a warning, error log message should
-    be enough. Although statement should ideally fail if it sees corrupt
-    data.
-  */
-  static void check_trx_id_sanity(trx_id_t id, const table_name_t &name);
-
   /**
     Check whether the changes by id are visible.
     @param[in] id transaction id to check against the view
@@ -149,26 +136,6 @@ loop:
            !std::binary_search(m_ids.begin(), m_ids.end(), id);
   }
 
-  /**
-    Check whether the changes by id are visible.
-    @param[in] id transaction id to check against the view
-    @param[in] name table name
-    @return whether the view sees the modifications of id.
-  */
-  bool changes_visible(trx_id_t id, const table_name_t &name) const
-  MY_ATTRIBUTE((warn_unused_result))
-  {
-    if (id >= m_low_limit_id)
-    {
-      check_trx_id_sanity(id, name);
-      return false;
-    }
-    return id < m_up_limit_id ||
-           m_ids.empty() ||
-           !std::binary_search(m_ids.begin(), m_ids.end(), id);
-  }
-
-
   /**
     @param id transaction to check
     @return true if view sees transaction id
@@ -180,6 +147,13 @@ loop:
 
   /** @return the low limit id */
   trx_id_t low_limit_id() const { return m_low_limit_id; }
+
+  /** Clamp the low limit id for purge_sys.end_view */
+  void clamp_low_limit_id(trx_id_t limit)
+  {
+    if (m_low_limit_id > limit)
+      m_low_limit_id= limit;
+  }
 };
 
 
@@ -250,7 +224,6 @@ public:
   */
   void set_creator_trx_id(trx_id_t id)
   {
-    ut_ad(id > 0);
     ut_ad(m_creator_trx_id == 0);
     m_creator_trx_id= id;
   }
@@ -275,8 +248,6 @@ public:
     A wrapper around ReadViewBase::changes_visible().
     Intended to be called by the ReadView owner thread.
   */
-  bool changes_visible(trx_id_t id, const table_name_t &name) const
-  { return id == m_creator_trx_id || ReadViewBase::changes_visible(id, name); }
   bool changes_visible(trx_id_t id) const
   { return id == m_creator_trx_id || ReadViewBase::changes_visible(id); }
 

storage/innobase/include/row0mysql.h

@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 2000, 2017, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, 2021, MariaDB Corporation.
+Copyright (c) 2017, 2022, 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
@@ -263,7 +263,7 @@ row_update_for_mysql(
 
 /** This can only be used when the current transaction is at
 READ COMMITTED or READ UNCOMMITTED isolation level.
-Before calling this function row_search_for_mysql() must have
+Before calling this function row_search_mvcc() must have
 initialized prebuilt->new_rec_locks to store the information which new
 record locks really were set. This function removes a newly set
 clustered index record lock under prebuilt->pcur or
@@ -382,22 +382,6 @@ row_rename_table_for_mysql(
 					FOREIGN KEY constraints */
 	MY_ATTRIBUTE((nonnull, warn_unused_result));
 
-/*********************************************************************//**
-Scans an index for either COOUNT(*) or CHECK TABLE.
-If CHECK TABLE; Checks that the index contains entries in an ascending order,
-unique constraint is not broken, and calculates the number of index entries
-in the read view of the current transaction.
-@return DB_SUCCESS or other error */
-dberr_t
-row_scan_index_for_mysql(
-/*=====================*/
-	row_prebuilt_t*		prebuilt,	/*!< in: prebuilt struct
-						in MySQL handle */
-	const dict_index_t*	index,		/*!< in: index */
-	ulint*			n_rows)		/*!< out: number of entries
-						seen in the consistent read */
-	MY_ATTRIBUTE((warn_unused_result));
-
 /* A struct describing a place for an individual column in the MySQL
 row format which is presented to the table handler in ha_innobase.
 This template struct is used to speed up row transformations between
@@ -606,7 +590,7 @@ struct row_prebuilt_t {
 					ROW_READ_TRY_SEMI_CONSISTENT and
 					to simply skip the row.	 If
 					the row matches, the next call to
-					row_search_for_mysql() will lock
+					row_search_mvcc() will lock
 					the row.
 					This eliminates lock waits in some
 					cases; note that this breaks
@@ -615,7 +599,7 @@ struct row_prebuilt_t {
 					the session is using READ
 					COMMITTED or READ UNCOMMITTED
 					isolation level, set in
-					row_search_for_mysql() if we set a new
+					row_search_mvcc() if we set a new
 					record lock on the secondary
 					or clustered index; this is
 					used in row_unlock_for_mysql()
@@ -847,7 +831,7 @@ innobase_rename_vc_templ(
 #define ROW_MYSQL_REC_FIELDS	1
 #define ROW_MYSQL_NO_TEMPLATE	2
 #define ROW_MYSQL_DUMMY_TEMPLATE 3	/* dummy template used in
-					row_scan_and_check_index */
+					row_check_index() */
 
 /* Values for hint_need_to_fetch_extra_cols */
 #define ROW_RETRIEVE_PRIMARY_KEY	1

storage/innobase/include/row0sel.h

@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1997, 2017, Oracle and/or its affiliates.
-Copyright (c) 2017, MariaDB Corporation.
+Copyright (c) 2017, 2022, 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
@@ -24,8 +24,7 @@ Select
 Created 12/19/1997 Heikki Tuuri
 *******************************************************/
 
-#ifndef row0sel_h
+#pragma once
-#define row0sel_h
 
 #include "data0data.h"
 #include "que0types.h"
@@ -58,15 +57,6 @@ void
 sel_col_prefetch_buf_free(
 /*======================*/
 	sel_buf_t*	prefetch_buf);	/*!< in, own: prefetch buffer */
-/*********************************************************************//**
-Gets the plan node for the nth table in a join.
-@return plan node */
-UNIV_INLINE
-plan_t*
-sel_node_get_nth_plan(
-/*==================*/
-	sel_node_t*	node,	/*!< in: select node */
-	ulint		i);	/*!< in: get ith plan node */
 /**********************************************************************//**
 Performs a select step. This is a high-level function used in SQL execution
 graphs.
@@ -76,14 +66,6 @@ row_sel_step(
 /*=========*/
 	que_thr_t*	thr);	/*!< in: query thread */
 /**********************************************************************//**
-Performs an execution step of an open or close cursor statement node.
-@return query thread to run next or NULL */
-UNIV_INLINE
-que_thr_t*
-open_step(
-/*======*/
-	que_thr_t*	thr);	/*!< in: query thread */
-/**********************************************************************//**
 Performs a fetch for a cursor.
 @return query thread to run next or NULL */
 que_thr_t*
@@ -136,37 +118,7 @@ row_sel_convert_mysql_key_to_innobase(
 	ulint		key_len);	/*!< in: MySQL key value length */
 
 
-/** Searches for rows in the database. This is used in the interface to
+/** Search for rows in the database using cursor.
-MySQL. This function opens a cursor, and also implements fetch next
-and fetch prev. NOTE that if we do a search with a full key value
-from a unique index (ROW_SEL_EXACT), then we will not store the cursor
-position and fetch next or fetch prev must not be tried to the cursor!
-
-@param[out]	buf		buffer for the fetched row in MySQL format
-@param[in]	mode		search mode PAGE_CUR_L
-@param[in,out]	prebuilt	prebuilt struct for the table handler;
-				this contains the info to search_tuple,
-				index; if search tuple contains 0 field then
-				we position the cursor at start or the end of
-				index, depending on 'mode'
-@param[in]	match_mode	0 or ROW_SEL_EXACT or ROW_SEL_EXACT_PREFIX
-@param[in]	direction	0 or ROW_SEL_NEXT or ROW_SEL_PREV;
-				Note: if this is != 0, then prebuilt must has a
-				pcur with stored position! In opening of a
-				cursor 'direction' should be 0.
-@return DB_SUCCESS, DB_RECORD_NOT_FOUND, DB_END_OF_INDEX, DB_DEADLOCK,
-DB_LOCK_TABLE_FULL, DB_CORRUPTION, or DB_TOO_BIG_RECORD */
-UNIV_INLINE
-dberr_t
-row_search_for_mysql(
-	byte*		buf,
-	page_cur_mode_t	mode,
-	row_prebuilt_t*	prebuilt,
-	ulint		match_mode,
-	ulint		direction)
-	MY_ATTRIBUTE((warn_unused_result));
-
-/** Searches for rows in the database using cursor.
 Function is mainly used for tables that are shared across connections and
 so it employs technique that can help re-construct the rows that
 transaction is suppose to see.
@@ -184,7 +136,8 @@ It also has optimization such as pre-caching the rows, using AHI, etc.
 				Note: if this is != 0, then prebuilt must has a
 				pcur with stored position! In opening of a
 				cursor 'direction' should be 0.
-@return DB_SUCCESS or error code */
+@return DB_SUCCESS, DB_RECORD_NOT_FOUND, DB_END_OF_INDEX, DB_DEADLOCK,
+DB_LOCK_TABLE_FULL, DB_CORRUPTION, or DB_TOO_BIG_RECORD */
 dberr_t
 row_search_mvcc(
 	byte*		buf,
@@ -210,6 +163,21 @@ row_count_rtree_recs(
 	ulint*		n_rows);	/*!< out: number of entries
 					seen in the consistent read */
 
+/**
+Check the index records in CHECK TABLE.
+The index must contain entries in an ascending order,
+unique constraint must not be violated by duplicated keys,
+and the number of index entries is counted in according to the
+current read view.
+
+@param prebuilt    index and transaction
+@param n_rows      number of records counted
+
+@return error code
+@retval DB_SUCCESS  if no error was found */
+dberr_t row_check_index(row_prebuilt_t *prebuilt, ulint *n_rows)
+  MY_ATTRIBUTE((nonnull, warn_unused_result));
+
 /** Read the max AUTOINC value from an index.
 @param[in] index	index starting with an AUTO_INCREMENT column
 @return	the largest AUTO_INCREMENT value
@@ -382,6 +350,17 @@ struct sel_node_t{
 					fetches */
 };
 
+/**
+Get the plan node for a table in a join.
+@param node  query graph node for SELECT
+@param i     plan node element
+@return ith plan node */
+inline plan_t *sel_node_get_nth_plan(sel_node_t *node, ulint i)
+{
+  ut_ad(i < node->n_tables);
+  return &node->plans[i];
+}
+
 /** Fetch statement node */
 struct fetch_node_t{
 	que_common_t	common;		/*!< type: QUE_NODE_FETCH */
@@ -476,7 +455,3 @@ row_sel_field_store_in_mysql_format_func(
 #endif /* UNIV_DEBUG */
         const byte*     data,   /*!< in: data to store */
         ulint           len);    /*!< in: length of the data */
-
-#include "row0sel.inl"
-
-#endif

storage/innobase/include/row0sel.inl

@@ -1,138 +0,0 @@
-/*****************************************************************************
-
-Copyright (c) 1997, 2014, Oracle and/or its affiliates. All Rights Reserved.
-
-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 Street, Fifth Floor, Boston, MA 02110-1335 USA
-
-*****************************************************************************/
-
-/**************************************************//**
-@file include/row0sel.ic
-Select
-
-Created 12/19/1997 Heikki Tuuri
-*******************************************************/
-
-#include "que0que.h"
-
-/*********************************************************************//**
-Gets the plan node for the nth table in a join.
-@return plan node */
-UNIV_INLINE
-plan_t*
-sel_node_get_nth_plan(
-/*==================*/
-	sel_node_t*	node,	/*!< in: select node */
-	ulint		i)	/*!< in: get ith plan node */
-{
-	ut_ad(i < node->n_tables);
-
-	return(node->plans + i);
-}
-
-/*********************************************************************//**
-Resets the cursor defined by sel_node to the SEL_NODE_OPEN state, which means
-that it will start fetching from the start of the result set again, regardless
-of where it was before, and it will set intention locks on the tables. */
-UNIV_INLINE
-void
-sel_node_reset_cursor(
-/*==================*/
-	sel_node_t*	node)	/*!< in: select node */
-{
-	node->state = SEL_NODE_OPEN;
-}
-
-/**********************************************************************//**
-Performs an execution step of an open or close cursor statement node.
-@return query thread to run next or NULL */
-UNIV_INLINE
-que_thr_t*
-open_step(
-/*======*/
-	que_thr_t*	thr)	/*!< in: query thread */
-{
-	sel_node_t*	sel_node;
-	open_node_t*	node;
-	ulint		err;
-
-	ut_ad(thr);
-
-	node = (open_node_t*) thr->run_node;
-	ut_ad(que_node_get_type(node) == QUE_NODE_OPEN);
-
-	sel_node = node->cursor_def;
-
-	err = DB_SUCCESS;
-
-	if (node->op_type == ROW_SEL_OPEN_CURSOR) {
-
-		/*		if (sel_node->state == SEL_NODE_CLOSED) { */
-
-		sel_node_reset_cursor(sel_node);
-		/*		} else {
-		err = DB_ERROR;
-		} */
-	} else {
-		if (sel_node->state != SEL_NODE_CLOSED) {
-
-			sel_node->state = SEL_NODE_CLOSED;
-		} else {
-			err = DB_ERROR;
-		}
-	}
-
-	if (err != DB_SUCCESS) {
-		/* SQL error detected */
-		fprintf(stderr, "SQL error %lu\n", (ulong) err);
-
-		ut_error;
-	}
-
-	thr->run_node = que_node_get_parent(node);
-
-	return(thr);
-}
-
-
-/** Searches for rows in the database. This is used in the interface to
-MySQL. This function opens a cursor, and also implements fetch next
-and fetch prev. NOTE that if we do a search with a full key value
-from a unique index (ROW_SEL_EXACT), then we will not store the cursor
-position and fetch next or fetch prev must not be tried to the cursor!
-
-@param[out]	buf		buffer for the fetched row in MySQL format
-@param[in]	mode		search mode PAGE_CUR_L
-@param[in,out]	prebuilt	prebuilt struct for the table handler;
-				this contains the info to search_tuple,
-				index; if search tuple contains 0 field then
-				we position the cursor at start or the end of
-				index, depending on 'mode'
-@param[in]	match_mode	0 or ROW_SEL_EXACT or ROW_SEL_EXACT_PREFIX
-@param[in]	direction	0 or ROW_SEL_NEXT or ROW_SEL_PREV;
-				Note: if this is != 0, then prebuilt must has a
-				pcur with stored position! In opening of a
-				cursor 'direction' should be 0.
-@return DB_SUCCESS, DB_RECORD_NOT_FOUND, DB_END_OF_INDEX, DB_DEADLOCK,
-DB_LOCK_TABLE_FULL, DB_CORRUPTION, or DB_TOO_BIG_RECORD */
-UNIV_INLINE
-dberr_t
-row_search_for_mysql(
-	byte*			buf,
-	page_cur_mode_t		mode,
-	row_prebuilt_t*		prebuilt,
-	ulint			match_mode,
-	ulint			direction)
-{
-	return(row_search_mvcc(buf, mode, prebuilt, match_mode, direction));
-}

storage/innobase/include/row0upd.h

@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1996, 2018, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, 2020, MariaDB Corporation.
+Copyright (c) 2017, 2022, 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
@@ -118,14 +118,6 @@ row_upd_changes_field_size_or_external(
 	dict_index_t*	index,	/*!< in: index */
 	const rec_offs*	offsets,/*!< in: rec_get_offsets(rec, index) */
 	const upd_t*	update);/*!< in: update vector */
-/***********************************************************//**
-Returns true if row update contains disowned external fields.
-@return true if the update contains disowned external fields. */
-bool
-row_upd_changes_disowned_external(
-/*==============================*/
-	const upd_t*	update)	/*!< in: update vector */
-	MY_ATTRIBUTE((nonnull, warn_unused_result));
 
 /***************************************************************//**
 Builds an update vector from those fields which in a secondary index entry

storage/innobase/include/row0vers.h

@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1997, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, 2019, MariaDB Corporation.
+Copyright (c) 2017, 2022, 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
@@ -55,7 +55,7 @@ row_vers_impl_x_locked(
 	const rec_offs*	offsets);
 
 /** Finds out if a version of the record, where the version >= the current
-purge view, should have ientry as its secondary index entry. We check
+purge_sys.view, should have ientry as its secondary index entry. We check
 if there is any not delete marked version of the record where the trx
 id >= purge view, and the secondary index entry == ientry; exactly in
 this case we return TRUE.
@@ -85,7 +85,9 @@ row_vers_old_has_index_entry(
 Constructs the version of a clustered index record which a consistent
 read should see. We assume that the trx id stored in rec is such that
 the consistent read should not see rec in its present version.
-@return DB_SUCCESS or DB_MISSING_HISTORY */
+@return error code
+@retval DB_SUCCESS if a previous version was fetched
+@retval DB_MISSING_HISTORY if the history is missing (a sign of corruption) */
 dberr_t
 row_vers_build_for_consistent_read(
 /*===============================*/

storage/innobase/include/trx0purge.h

@@ -24,8 +24,7 @@ Purge old versions
 Created 3/26/1996 Heikki Tuuri
 *******************************************************/
 
-#ifndef trx0purge_h
+#pragma once
-#define trx0purge_h
 
 #include "trx0sys.h"
 #include "que0types.h"
@@ -123,7 +122,8 @@ public:
   /** latch protecting view, m_enabled */
   alignas(CPU_LEVEL1_DCACHE_LINESIZE) mutable srw_spin_lock latch;
 private:
-  /** The purge will not remove undo logs which are >= this view */
+  /** Read view at the start of a purge batch. Any encountered index records
+  that are older than view will be removed. */
   ReadViewBase view;
   /** whether purge is enabled; protected by latch and std::atomic */
   std::atomic<bool> m_enabled;
@@ -133,6 +133,12 @@ private:
   Atomic_counter<uint32_t> m_SYS_paused;
   /** number of stop_FTS() calls without resume_FTS() */
   Atomic_counter<uint32_t> m_FTS_paused;
+
+  /** latch protecting end_view */
+  alignas(CPU_LEVEL1_DCACHE_LINESIZE) srw_spin_lock_low end_latch;
+  /** Read view at the end of a purge batch (copied from view). Any undo pages
+  containing records older than end_view may be freed. */
+  ReadViewBase end_view;
 public:
 	que_t*		query;		/*!< The query graph which will do the
 					parallelized purge operation */
@@ -261,28 +267,56 @@ public:
   /** check stop_SYS() */
   void check_stop_FTS() { if (must_wait_FTS()) wait_FTS(); }
 
-  /** A wrapper around ReadView::changes_visible(). */
+  /** Determine if the history of a transaction is purgeable.
-  bool changes_visible(trx_id_t id, const table_name_t &name) const
+  @param trx_id  transaction identifier
-  {
+  @return whether the history is purgeable */
-    return view.changes_visible(id, name);
+  TRANSACTIONAL_TARGET bool is_purgeable(trx_id_t trx_id) const;
-  }
+
   /** A wrapper around ReadView::low_limit_no(). */
   trx_id_t low_limit_no() const
   {
-    /* Other callers than purge_coordinator_callback() must be holding
+    /* This function may only be called by purge_coordinator_callback().
-    purge_sys.latch here. The purge coordinator task may call this
+
-    without holding any latch, because it is the only thread that may
+    The purge coordinator task may call this without holding any latch,
-    modify purge_sys.view. */
+    because it is the only thread that may modify purge_sys.view.
+
+    Any other threads that access purge_sys.view must hold purge_sys.latch,
+    typically via purge_sys_t::view_guard. */
     return view.low_limit_no();
   }
   /** A wrapper around trx_sys_t::clone_oldest_view(). */
+  template<bool also_end_view= false>
   void clone_oldest_view()
   {
     latch.wr_lock(SRW_LOCK_CALL);
     trx_sys.clone_oldest_view(&view);
+    if (also_end_view)
+      (end_view= view).
+        clamp_low_limit_id(head.trx_no ? head.trx_no : tail.trx_no);
     latch.wr_unlock();
   }
 
+  /** Update end_view at the end of a purge batch. */
+  inline void clone_end_view();
+
+  struct view_guard
+  {
+    inline view_guard();
+    inline ~view_guard();
+
+    /** @return purge_sys.view */
+    inline const ReadViewBase &view() const;
+  };
+
+  struct end_view_guard
+  {
+    inline end_view_guard();
+    inline ~end_view_guard();
+
+    /** @return purge_sys.end_view */
+    inline const ReadViewBase &view() const;
+  };
+
   /** Stop the purge thread and check n_ref_count of all auxiliary
   and common table associated with the fts table.
   @param	table		parent FTS table
@@ -294,4 +328,20 @@ public:
 /** The global data structure coordinating a purge */
 extern purge_sys_t	purge_sys;
 
-#endif /* trx0purge_h */
+purge_sys_t::view_guard::view_guard()
+{ purge_sys.latch.rd_lock(SRW_LOCK_CALL); }
+
+purge_sys_t::view_guard::~view_guard()
+{ purge_sys.latch.rd_unlock(); }
+
+const ReadViewBase &purge_sys_t::view_guard::view() const
+{ return purge_sys.view; }
+
+purge_sys_t::end_view_guard::end_view_guard()
+{ purge_sys.end_latch.rd_lock(); }
+
+purge_sys_t::end_view_guard::~end_view_guard()
+{ purge_sys.end_latch.rd_unlock(); }
+
+const ReadViewBase &purge_sys_t::end_view_guard::view() const
+{ return purge_sys.end_view; }

storage/innobase/include/trx0rec.h

@@ -181,17 +181,17 @@ trx_undo_report_row_operation(
 is being called purge view and we would like to get the purge record
 even it is in the purge view (in normal case, it will return without
 fetching the purge record */
-#define		TRX_UNDO_PREV_IN_PURGE		0x1
+static constexpr ulint TRX_UNDO_PREV_IN_PURGE = 1;
 
 /** This tells trx_undo_prev_version_build() to fetch the old value in
 the undo log (which is the after image for an update) */
-#define		TRX_UNDO_GET_OLD_V_VALUE	0x2
+static constexpr ulint TRX_UNDO_GET_OLD_V_VALUE = 2;
+
+/** indicate a call from row_vers_old_has_index_entry() */
+static constexpr ulint TRX_UNDO_CHECK_PURGEABILITY = 4;
 
 /** Build a previous version of a clustered index record. The caller
 must hold a latch on the index page of the clustered index record.
-@param	index_rec	clustered index record in the index tree
-@param	index_mtr	mtr which contains the latch to index_rec page
-			and purge_view
 @param	rec		version of a clustered index record
 @param	index		clustered index
 @param	offsets		rec_get_offsets(rec, index)
@@ -210,14 +210,12 @@ must hold a latch on the index page of the clustered index record.
 @param	v_status	status determine if it is going into this
 			function by purge thread or not.
 			And if we read "after image" of undo log
-@retval true if previous version was built, or if it was an insert
+@return error code
-or the table has been rebuilt
+@retval DB_SUCCESS if previous version was successfully built,
-@retval false if the previous version is earlier than purge_view,
+or if it was an insert or the undo record refers to the table before rebuild
-or being purged, which means that it may have been removed */
+@retval DB_MISSING_HISTORY if the history is missing */
-bool
+dberr_t
 trx_undo_prev_version_build(
-	const rec_t	*index_rec,
-	mtr_t		*index_mtr,
 	const rec_t 	*rec,
 	dict_index_t	*index,
 	rec_offs	*offsets,

storage/innobase/lock/lock0lock.cc

@@ -44,6 +44,7 @@ Created 5/7/1996 Heikki Tuuri
 #include "row0vers.h"
 #include "pars0pars.h"
 #include "srv0mon.h"
+#include "que0que.h"
 
 #include <set>
 
@@ -1788,8 +1789,8 @@ dberr_t lock_wait(que_thr_t *thr)
      wait_lock->un_member.tab_lock.table->id <= DICT_FIELDS_ID);
   thd_wait_begin(trx->mysql_thd, (type_mode & LOCK_TABLE)
                  ? THD_WAIT_TABLE_LOCK : THD_WAIT_ROW_LOCK);
-  trx->error_state= DB_SUCCESS;
 
+  int err= 0;
   mysql_mutex_lock(&lock_sys.wait_mutex);
   if (trx->lock.wait_lock)
   {
@@ -1811,25 +1812,24 @@ dberr_t lock_wait(que_thr_t *thr)
   if (row_lock_wait)
     lock_sys.wait_start();
 
-  trx->error_state= DB_SUCCESS;
-
 #ifdef HAVE_REPLICATION
   if (rpl)
     lock_wait_rpl_report(trx);
 #endif
 
+  if (trx->error_state != DB_SUCCESS)
+    goto check_trx_error;
+
   while (trx->lock.wait_lock)
   {
-    int err;
+    DEBUG_SYNC_C("lock_wait_before_suspend");
 
     if (no_timeout)
-    {
       my_cond_wait(&trx->lock.cond, &lock_sys.wait_mutex.m_mutex);
-      err= 0;
-    }
     else
       err= my_cond_timedwait(&trx->lock.cond, &lock_sys.wait_mutex.m_mutex,
                              &abstime);
+check_trx_error:
     switch (trx->error_state) {
     case DB_DEADLOCK:
     case DB_INTERRUPTED:
@@ -1875,17 +1875,19 @@ end_wait:
 
 
 /** Resume a lock wait */
-static void lock_wait_end(trx_t *trx)
+template <bool from_deadlock= false>
+void lock_wait_end(trx_t *trx)
 {
   mysql_mutex_assert_owner(&lock_sys.wait_mutex);
   ut_ad(trx->mutex_is_owner());
   ut_d(const auto state= trx->state);
-  ut_ad(state == TRX_STATE_ACTIVE || state == TRX_STATE_PREPARED);
+  ut_ad(state == TRX_STATE_COMMITTED_IN_MEMORY || state == TRX_STATE_ACTIVE ||
-  ut_ad(trx->lock.wait_thr);
+        state == TRX_STATE_PREPARED);
+  ut_ad(from_deadlock || trx->lock.wait_thr);
 
   if (trx->lock.was_chosen_as_deadlock_victim)
   {
-    ut_ad(state == TRX_STATE_ACTIVE);
+    ut_ad(from_deadlock || state == TRX_STATE_ACTIVE);
     trx->error_state= DB_DEADLOCK;
   }
 
@@ -5672,13 +5674,16 @@ static void lock_release_autoinc_locks(trx_t *trx)
 }
 
 /** Cancel a waiting lock request and release possibly waiting transactions */
-static void lock_cancel_waiting_and_release(lock_t *lock)
+template <bool from_deadlock= false>
+void lock_cancel_waiting_and_release(lock_t *lock)
 {
   lock_sys.assert_locked(*lock);
   mysql_mutex_assert_owner(&lock_sys.wait_mutex);
   trx_t *trx= lock->trx;
   trx->mutex_lock();
-  ut_ad(trx->state == TRX_STATE_ACTIVE);
+  ut_d(const auto trx_state= trx->state);
+  ut_ad(trx_state == TRX_STATE_COMMITTED_IN_MEMORY ||
+        trx_state == TRX_STATE_ACTIVE);
 
   if (!lock->is_table())
     lock_rec_dequeue_from_page(lock, true);
@@ -5697,7 +5702,8 @@ static void lock_cancel_waiting_and_release(lock_t *lock)
   /* Reset the wait flag and the back pointer to lock in trx. */
   lock_reset_lock_and_trx_wait(lock);
 
-  lock_wait_end(trx);
+  lock_wait_end<from_deadlock>(trx);
+
   trx->mutex_unlock();
 }
 
@@ -5868,6 +5874,7 @@ lock_unlock_table_autoinc(
 
 /** Handle a pending lock wait (DB_LOCK_WAIT) in a semi-consistent read
 while holding a clustered index leaf page latch.
+
 @param trx           transaction that is or was waiting for a lock
 @retval DB_SUCCESS   if the lock was granted
 @retval DB_DEADLOCK  if the transaction must be aborted due to a deadlock
@@ -5878,8 +5885,13 @@ dberr_t lock_trx_handle_wait(trx_t *trx)
   DEBUG_SYNC_C("lock_trx_handle_wait_enter");
   if (trx->lock.was_chosen_as_deadlock_victim)
     return DB_DEADLOCK;
+  DEBUG_SYNC_C("lock_trx_handle_wait_before_unlocked_wait_lock_check");
+  /* trx->lock.was_chosen_as_deadlock_victim must always be set before
+  trx->lock.wait_lock if the transaction was chosen as deadlock victim,
+  the function must not return DB_SUCCESS if
+  trx->lock.was_chosen_as_deadlock_victim is set. */
   if (!trx->lock.wait_lock)
-    return DB_SUCCESS;
+    return trx->lock.was_chosen_as_deadlock_victim ? DB_DEADLOCK : DB_SUCCESS;
   dberr_t err= DB_SUCCESS;
   mysql_mutex_lock(&lock_sys.wait_mutex);
   if (trx->lock.was_chosen_as_deadlock_victim)
@@ -6282,8 +6294,11 @@ namespace Deadlock
 
       ut_ad(victim->state == TRX_STATE_ACTIVE);
 
+      /* victim->lock.was_chosen_as_deadlock_victim must always be set before
+      releasing waiting locks and reseting trx->lock.wait_lock */
       victim->lock.was_chosen_as_deadlock_victim= true;
-      lock_cancel_waiting_and_release(victim->lock.wait_lock);
+      DEBUG_SYNC_C("deadlock_report_before_lock_releasing");
+      lock_cancel_waiting_and_release<true>(victim->lock.wait_lock);
 #ifdef WITH_WSREP
       if (victim->is_wsrep() && wsrep_thd_is_SR(victim->mysql_thd))
         wsrep_handle_SR_rollback(trx->mysql_thd, victim->mysql_thd);

storage/innobase/que/que0que.cc

@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1996, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, 2021, MariaDB Corporation.
+Copyright (c) 2017, 2022, 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
@@ -566,6 +566,27 @@ que_node_type_string(
 }
 #endif /* DBUG_TRACE */
 
+
+/**********************************************************************//**
+Performs an execution step of an open or close cursor statement node.
+@param thr query thread */
+static void open_step(que_thr_t *thr)
+{
+  open_node_t *node= static_cast<open_node_t*>(thr->run_node);
+  ut_ad(que_node_get_type(node) == QUE_NODE_OPEN);
+  sel_node_t *sel_node= node->cursor_def;
+
+  if (node->op_type == ROW_SEL_OPEN_CURSOR)
+    sel_node->state= SEL_NODE_OPEN;
+  else
+  {
+    ut_ad(sel_node->state != SEL_NODE_CLOSED);
+    sel_node->state= SEL_NODE_CLOSED;
+  }
+
+  thr->run_node= que_node_get_parent(node);
+}
+
 /**********************************************************************//**
 Performs an execution step on a query thread.
 @return query thread to run next: it may differ from the input
@@ -636,7 +657,7 @@ que_thr_step(
 	} else if (type == QUE_NODE_FETCH) {
 		thr = fetch_step(thr);
 	} else if (type == QUE_NODE_OPEN) {
-		thr = open_step(thr);
+		open_step(thr);
 	} else if (type == QUE_NODE_FUNC) {
 		proc_eval_step(thr);
 

storage/innobase/row/row0log.cc

@@ -3842,9 +3842,8 @@ UndorecApplier::get_old_rec(const dtuple_t &tuple, dict_index_t *index,
     ut_ad(len == DATA_ROLL_PTR_LEN);
     if (is_same(roll_ptr))
       return version;
-    trx_undo_prev_version_build(*clust_rec, &mtr, version, index,
+    trx_undo_prev_version_build(version, index, *offsets, heap, &prev_version,
-                                *offsets, heap, &prev_version, nullptr,
+                                nullptr, nullptr, 0);
-                                nullptr, 0);
     version= prev_version;
   }
   while (version);
@@ -4014,9 +4013,8 @@ void UndorecApplier::log_update(const dtuple_t &tuple,
     if (match_rec == rec)
       copy_rec= rec_copy(mem_heap_alloc(
         heap, rec_offs_size(offsets)), match_rec, offsets);
-    trx_undo_prev_version_build(rec, &mtr, match_rec, clust_index,
+    trx_undo_prev_version_build(match_rec, clust_index, offsets, heap,
-                                offsets, heap, &prev_version, nullptr,
+                                &prev_version, nullptr, nullptr, 0);
-                                nullptr, 0);
 
     prev_offsets= rec_get_offsets(prev_version, clust_index, prev_offsets,
                                   clust_index->n_core_fields,

storage/innobase/row/row0merge.cc

@@ -2295,8 +2295,14 @@ end_of_index:
 			ut_ad(trx->read_view.is_open());
 			ut_ad(rec_trx_id != trx->id);
 
-			if (!trx->read_view.changes_visible(
+			if (!trx->read_view.changes_visible(rec_trx_id)) {
-				    rec_trx_id, old_table->name)) {
+				if (rec_trx_id
+				    >= trx->read_view.low_limit_id()
+				    && rec_trx_id
+				    >= trx_sys.get_max_trx_id()) {
+					goto corrupted_rec;
+				}
+
 				rec_t*	old_vers;
 
 				row_vers_build_for_consistent_read(
@@ -4617,9 +4623,7 @@ row_merge_is_index_usable(
 	       && (index->table->is_temporary() || index->table->no_rollback()
 		   || index->trx_id == 0
 		   || !trx->read_view.is_open()
-		   || trx->read_view.changes_visible(
+		   || trx->read_view.changes_visible(index->trx_id)));
-			   index->trx_id,
-			   index->table->name)));
 }
 
 /** Build indexes on a table by reading a clustered index, creating a temporary

storage/innobase/row/row0mysql.cc

@@ -1766,7 +1766,7 @@ error:
 
 /** This can only be used when the current transaction is at
 READ COMMITTED or READ UNCOMMITTED isolation level.
-Before calling this function row_search_for_mysql() must have
+Before calling this function row_search_mvcc() must have
 initialized prebuilt->new_rec_locks to store the information which new
 record locks really were set. This function removes a newly set
 clustered index record lock under prebuilt->pcur or
@@ -2937,182 +2937,3 @@ funct_exit:
 
 	return(err);
 }
-
-/*********************************************************************//**
-Scans an index for either COUNT(*) or CHECK TABLE.
-If CHECK TABLE; Checks that the index contains entries in an ascending order,
-unique constraint is not broken, and calculates the number of index entries
-in the read view of the current transaction.
-@return DB_SUCCESS or other error */
-dberr_t
-row_scan_index_for_mysql(
-/*=====================*/
-	row_prebuilt_t*		prebuilt,	/*!< in: prebuilt struct
-						in MySQL handle */
-	const dict_index_t*	index,		/*!< in: index */
-	ulint*			n_rows)		/*!< out: number of entries
-						seen in the consistent read */
-{
-	dtuple_t*	prev_entry	= NULL;
-	ulint		matched_fields;
-	byte*		buf;
-	dberr_t		ret;
-	rec_t*		rec;
-	int		cmp;
-	ibool		contains_null;
-	ulint		i;
-	ulint		cnt;
-	mem_heap_t*	heap		= NULL;
-	rec_offs	offsets_[REC_OFFS_NORMAL_SIZE];
-	rec_offs*	offsets;
-	rec_offs_init(offsets_);
-
-	*n_rows = 0;
-
-	/* Don't support RTree Leaf level scan */
-	ut_ad(!dict_index_is_spatial(index));
-
-	if (dict_index_is_clust(index)) {
-		/* The clustered index of a table is always available.
-		During online ALTER TABLE that rebuilds the table, the
-		clustered index in the old table will have
-		index->online_log pointing to the new table. All
-		indexes of the old table will remain valid and the new
-		table will be unaccessible to MySQL until the
-		completion of the ALTER TABLE. */
-	} else if (dict_index_is_online_ddl(index)
-		   || (index->type & DICT_FTS)) {
-		/* Full Text index are implemented by auxiliary tables,
-		not the B-tree. We also skip secondary indexes that are
-		being created online. */
-		return(DB_SUCCESS);
-	}
-
-	ulint bufsize = std::max<ulint>(srv_page_size,
-					prebuilt->mysql_row_len);
-	buf = static_cast<byte*>(ut_malloc_nokey(bufsize));
-	heap = mem_heap_create(100);
-
-	cnt = 1000;
-
-	ret = row_search_for_mysql(buf, PAGE_CUR_G, prebuilt, 0, 0);
-loop:
-	/* Check thd->killed every 1,000 scanned rows */
-	if (--cnt == 0) {
-		if (trx_is_interrupted(prebuilt->trx)) {
-			ret = DB_INTERRUPTED;
-			goto func_exit;
-		}
-		cnt = 1000;
-	}
-
-	switch (ret) {
-	case DB_SUCCESS:
-		break;
-	case DB_DEADLOCK:
-	case DB_LOCK_TABLE_FULL:
-	case DB_LOCK_WAIT_TIMEOUT:
-	case DB_INTERRUPTED:
-		goto func_exit;
-	default:
-		ib::warn() << "CHECK TABLE on index " << index->name << " of"
-			" table " << index->table->name << " returned " << ret;
-		/* (this error is ignored by CHECK TABLE) */
-		/* fall through */
-	case DB_END_OF_INDEX:
-		ret = DB_SUCCESS;
-func_exit:
-		ut_free(buf);
-		mem_heap_free(heap);
-
-		return(ret);
-	}
-
-	*n_rows = *n_rows + 1;
-
-	/* else this code is doing handler::check() for CHECK TABLE */
-
-	/* row_search... returns the index record in buf, record origin offset
-	within buf stored in the first 4 bytes, because we have built a dummy
-	template */
-
-	rec = buf + mach_read_from_4(buf);
-
-	offsets = rec_get_offsets(rec, index, offsets_, index->n_core_fields,
-				  ULINT_UNDEFINED, &heap);
-
-	if (prev_entry != NULL) {
-		matched_fields = 0;
-
-		cmp = cmp_dtuple_rec_with_match(prev_entry, rec, offsets,
-						&matched_fields);
-		contains_null = FALSE;
-
-		/* In a unique secondary index we allow equal key values if
-		they contain SQL NULLs */
-
-		for (i = 0;
-		     i < dict_index_get_n_ordering_defined_by_user(index);
-		     i++) {
-			if (UNIV_SQL_NULL == dfield_get_len(
-				    dtuple_get_nth_field(prev_entry, i))) {
-
-				contains_null = TRUE;
-				break;
-			}
-		}
-
-		const char* msg;
-
-		if (cmp > 0) {
-			ret = DB_INDEX_CORRUPT;
-			msg = "index records in a wrong order in ";
-not_ok:
-			ib::error()
-				<< msg << index->name
-				<< " of table " << index->table->name
-				<< ": " << *prev_entry << ", "
-				<< rec_offsets_print(rec, offsets);
-			/* Continue reading */
-		} else if (dict_index_is_unique(index)
-			   && !contains_null
-			   && matched_fields
-			   >= dict_index_get_n_ordering_defined_by_user(
-				   index)) {
-			ret = DB_DUPLICATE_KEY;
-			msg = "duplicate key in ";
-			goto not_ok;
-		}
-	}
-
-	{
-		mem_heap_t*	tmp_heap = NULL;
-
-		/* Empty the heap on each round.  But preserve offsets[]
-		for the row_rec_to_index_entry() call, by copying them
-		into a separate memory heap when needed. */
-		if (UNIV_UNLIKELY(offsets != offsets_)) {
-			ulint	size = rec_offs_get_n_alloc(offsets)
-				* sizeof *offsets;
-
-			tmp_heap = mem_heap_create(size);
-
-			offsets = static_cast<rec_offs*>(
-				mem_heap_dup(tmp_heap, offsets, size));
-		}
-
-		mem_heap_empty(heap);
-
-		prev_entry = row_rec_to_index_entry(
-			rec, index, offsets, heap);
-
-		if (UNIV_LIKELY_NULL(tmp_heap)) {
-			mem_heap_free(tmp_heap);
-		}
-	}
-
-	ret = row_search_for_mysql(
-		buf, PAGE_CUR_G, prebuilt, 0, ROW_SEL_NEXT);
-
-	goto loop;
-}

storage/innobase/row/row0umod.cc

@@ -216,26 +216,23 @@ static ulint row_trx_id_offset(const rec_t* rec, const dict_index_t* index)
 }
 
 /** Determine if rollback must execute a purge-like operation.
-@param[in,out]	node	row undo
+@param node   row undo
-@param[in,out]	mtr	mini-transaction
 @return	whether the record should be purged */
-static bool row_undo_mod_must_purge(undo_node_t* node, mtr_t* mtr)
+static bool row_undo_mod_must_purge(const undo_node_t &node)
 {
-	ut_ad(node->rec_type == TRX_UNDO_UPD_DEL_REC);
+  ut_ad(node.rec_type == TRX_UNDO_UPD_DEL_REC);
-	ut_ad(!node->table->is_temporary());
+  ut_ad(!node.table->is_temporary());
 
-	btr_cur_t* btr_cur = btr_pcur_get_btr_cur(&node->pcur);
+  const btr_cur_t &btr_cur= node.pcur.btr_cur;
-	ut_ad(btr_cur->index->is_primary());
+  ut_ad(btr_cur.index->is_primary());
   DEBUG_SYNC_C("rollback_purge_clust");
 
-	if (!purge_sys.changes_visible(node->new_trx_id, node->table->name)) {
+  if (!purge_sys.is_purgeable(node.new_trx_id))
     return false;
-	}
 
-	const rec_t* rec = btr_cur_get_rec(btr_cur);
+  const rec_t *rec= btr_cur_get_rec(&btr_cur);
-
+  return trx_read_trx_id(rec + row_trx_id_offset(rec, btr_cur.index)) ==
-	return trx_read_trx_id(rec + row_trx_id_offset(rec, btr_cur->index))
+    node.new_trx_id;
-		== node->new_trx_id;
 }
 
 /***********************************************************//**
@@ -251,7 +248,6 @@ row_undo_mod_clust(
 {
 	btr_pcur_t*	pcur;
 	mtr_t		mtr;
-	bool		have_latch = false;
 	dberr_t		err;
 	dict_index_t*	index;
 
@@ -347,9 +343,7 @@ row_undo_mod_clust(
 			btr_pcur_commit_specify_mtr(pcur, &mtr);
 		} else {
 			index->set_modified(mtr);
-			have_latch = true;
+			if (!row_undo_mod_must_purge(*node)) {
-			purge_sys.latch.rd_lock(SRW_LOCK_CALL);
-			if (!row_undo_mod_must_purge(node, &mtr)) {
 				goto mtr_commit_exit;
 			}
 			err = btr_cur_optimistic_delete(&pcur->btr_cur, 0,
@@ -358,9 +352,7 @@ row_undo_mod_clust(
 				goto mtr_commit_exit;
 			}
 			err = DB_SUCCESS;
-			purge_sys.latch.rd_unlock();
 			btr_pcur_commit_specify_mtr(pcur, &mtr);
-			have_latch = false;
 		}
 
 		mtr.start();
@@ -376,9 +368,7 @@ row_undo_mod_clust(
 		if (index->table->is_temporary()) {
 			mtr.set_log_mode(MTR_LOG_NO_REDO);
 		} else {
-			have_latch = true;
+			if (!row_undo_mod_must_purge(*node)) {
-			purge_sys.latch.rd_lock(SRW_LOCK_CALL);
-			if (!row_undo_mod_must_purge(node, &mtr)) {
 				goto mtr_commit_exit;
 			}
 			index->set_modified(mtr);
@@ -400,17 +390,12 @@ row_undo_mod_clust(
 
 		mtr.start();
 		if (pcur->restore_position(BTR_MODIFY_LEAF, &mtr)
-		    != btr_pcur_t::SAME_ALL) {
+		    != btr_pcur_t::SAME_ALL
-			goto mtr_commit_exit;
+		    || !purge_sys.is_purgeable(node->new_trx_id)) {
-		}
-		rec_t* rec = btr_pcur_get_rec(pcur);
-		have_latch = true;
-		purge_sys.latch.rd_lock(SRW_LOCK_CALL);
-		if (!purge_sys.changes_visible(node->new_trx_id,
-					       node->table->name)) {
 			goto mtr_commit_exit;
 		}
 
+		rec_t* rec = btr_pcur_get_rec(pcur);
 		ulint trx_id_offset = index->trx_id_offset;
 		ulint trx_id_pos = index->n_uniq ? index->n_uniq : 1;
 		/* Reserve enough offsets for the PRIMARY KEY and
@@ -477,10 +462,6 @@ row_undo_mod_clust(
 	}
 
 mtr_commit_exit:
-	if (have_latch) {
-		purge_sys.latch.rd_unlock();
-	}
-
 	btr_pcur_commit_specify_mtr(pcur, &mtr);
 
 func_exit:

storage/innobase/row/row0upd.cc

@@ -469,46 +469,6 @@ row_upd_changes_field_size_or_external(
 	return(FALSE);
 }
 
-/***********************************************************//**
-Returns true if row update contains disowned external fields.
-@return true if the update contains disowned external fields. */
-bool
-row_upd_changes_disowned_external(
-/*==============================*/
-	const upd_t*	update)	/*!< in: update vector */
-{
-	const upd_field_t*	upd_field;
-	const dfield_t*		new_val;
-	ulint			new_len;
-	ulint                   n_fields;
-	ulint			i;
-
-	n_fields = upd_get_n_fields(update);
-
-	for (i = 0; i < n_fields; i++) {
-		const byte*	field_ref;
-
-		upd_field = upd_get_nth_field(update, i);
-		new_val = &(upd_field->new_val);
-		new_len = dfield_get_len(new_val);
-
-		if (!dfield_is_ext(new_val)) {
-			continue;
-		}
-
-		ut_ad(new_len >= BTR_EXTERN_FIELD_REF_SIZE);
-
-		field_ref = static_cast<const byte*>(dfield_get_data(new_val))
-			    + new_len - BTR_EXTERN_FIELD_REF_SIZE;
-
-		if (field_ref[BTR_EXTERN_LEN] & BTR_EXTERN_OWNER_FLAG) {
-			return(true);
-		}
-	}
-
-	return(false);
-}
-
 /***************************************************************//**
 Builds an update vector from those fields which in a secondary index entry
 differ from a record that has the equal ordering fields. NOTE: we compare

storage/innobase/row/row0vers.cc

@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1997, 2017, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, 2021, MariaDB Corporation.
+Copyright (c) 2017, 2022, 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
@@ -104,6 +104,9 @@ row_vers_impl_x_locked_low(
 	DBUG_ENTER("row_vers_impl_x_locked_low");
 
 	ut_ad(rec_offs_validate(rec, index, offsets));
+	ut_ad(mtr->memo_contains_page_flagged(clust_rec,
+					      MTR_MEMO_PAGE_S_FIX
+					      | MTR_MEMO_PAGE_X_FIX));
 
 	if (ulint trx_id_offset = clust_index->trx_id_offset) {
 		trx_id = mach_read_from_6(clust_rec + trx_id_offset);
@@ -190,7 +193,7 @@ row_vers_impl_x_locked_low(
 		heap = mem_heap_create(1024);
 
 		trx_undo_prev_version_build(
-			clust_rec, mtr, version, clust_index, clust_offsets,
+			version, clust_index, clust_offsets,
 			heap, &prev_version, NULL,
 			dict_index_has_virtual(index) ? &vrow : NULL, 0);
 
@@ -527,6 +530,10 @@ row_vers_build_cur_vrow_low(
 			 = DATA_MISSING;
 	}
 
+	ut_ad(mtr->memo_contains_page_flagged(rec,
+					      MTR_MEMO_PAGE_S_FIX
+					      | MTR_MEMO_PAGE_X_FIX));
+
 	version = rec;
 
 	/* If this is called by purge thread, set TRX_UNDO_PREV_IN_PURGE
@@ -543,7 +550,7 @@ row_vers_build_cur_vrow_low(
 			version, clust_index, clust_offsets);
 
 		trx_undo_prev_version_build(
-			rec, mtr, version, clust_index, clust_offsets,
+			version, clust_index, clust_offsets,
 			heap, &prev_version, NULL, vrow, status);
 
 		if (heap2) {
@@ -643,6 +650,10 @@ row_vers_vc_matches_cluster(
 	/* First compare non-virtual columns (primary keys) */
 	ut_ad(index->n_fields == n_fields);
 	ut_ad(n_fields == dtuple_get_n_fields(icentry));
+	ut_ad(mtr->memo_contains_page_flagged(rec,
+					      MTR_MEMO_PAGE_S_FIX
+					      | MTR_MEMO_PAGE_X_FIX));
+
 	{
 		const dfield_t* a = ientry->fields;
 		const dfield_t* b = icentry->fields;
@@ -684,7 +695,7 @@ row_vers_vc_matches_cluster(
 		ut_ad(roll_ptr != 0);
 
 		trx_undo_prev_version_build(
-			rec, mtr, version, clust_index, clust_offsets,
+			version, clust_index, clust_offsets,
 			heap, &prev_version, NULL, vrow,
 			TRX_UNDO_PREV_IN_PURGE | TRX_UNDO_GET_OLD_V_VALUE);
 
@@ -834,7 +845,7 @@ row_vers_build_cur_vrow(
 }
 
 /** Finds out if a version of the record, where the version >= the current
-purge view, should have ientry as its secondary index entry. We check
+purge_sys.view, should have ientry as its secondary index entry. We check
 if there is any not delete marked version of the record where the trx
 id >= purge view, and the secondary index entry == ientry; exactly in
 this case we return TRUE.
@@ -1016,11 +1027,12 @@ unsafe_to_purge:
 		heap = mem_heap_create(1024);
 		vrow = NULL;
 
-		trx_undo_prev_version_build(rec, mtr, version,
+		trx_undo_prev_version_build(version,
 					    clust_index, clust_offsets,
-					    heap, &prev_version, NULL,
+					    heap, &prev_version, nullptr,
 					    dict_index_has_virtual(index)
-						? &vrow : NULL, 0);
+					    ? &vrow : nullptr,
+					    TRX_UNDO_CHECK_PURGEABILITY);
 		mem_heap_free(heap2); /* free version and clust_offsets */
 
 		if (!prev_version) {
@@ -1099,7 +1111,9 @@ unsafe_to_purge:
 Constructs the version of a clustered index record which a consistent
 read should see. We assume that the trx id stored in rec is such that
 the consistent read should not see rec in its present version.
-@return DB_SUCCESS or DB_MISSING_HISTORY */
+@return error code
+@retval DB_SUCCESS if a previous version was fetched
+@retval DB_MISSING_HISTORY if the history is missing (a sign of corruption) */
 dberr_t
 row_vers_build_for_consistent_read(
 /*===============================*/
@@ -1139,7 +1153,7 @@ row_vers_build_for_consistent_read(
 
 	trx_id = row_get_rec_trx_id(rec, index, *offsets);
 
-	ut_ad(!view->changes_visible(trx_id, index->table->name));
+	ut_ad(!view->changes_visible(trx_id));
 
 	ut_ad(!vrow || !(*vrow));
 
@@ -1157,12 +1171,10 @@ row_vers_build_for_consistent_read(
 		/* If purge can't see the record then we can't rely on
 		the UNDO log record. */
 
-		bool	purge_sees = trx_undo_prev_version_build(
+		err = trx_undo_prev_version_build(
-			rec, mtr, version, index, *offsets, heap,
+			version, index, *offsets, heap,
 			&prev_version, NULL, vrow, 0);
 
-		err  = (purge_sees) ? DB_SUCCESS : DB_MISSING_HISTORY;
-
 		if (prev_heap != NULL) {
 			mem_heap_free(prev_heap);
 		}
@@ -1184,7 +1196,7 @@ row_vers_build_for_consistent_read(
 
 		trx_id = row_get_rec_trx_id(prev_version, index, *offsets);
 
-		if (view->changes_visible(trx_id, index->table->name)) {
+		if (view->changes_visible(trx_id)) {
 
 			/* The view already sees this version: we can copy
 			it to in_heap and return */
@@ -1201,8 +1213,11 @@ row_vers_build_for_consistent_read(
 				dtuple_dup_v_fld(*vrow, in_heap);
 			}
 			break;
+		} else if (trx_id >= view->low_limit_id()
+			   && trx_id >= trx_sys.get_max_trx_id()) {
+			err = DB_CORRUPTION;
+			break;
 		}
-
 		version = prev_version;
 	}
 
@@ -1319,10 +1334,9 @@ committed_version_trx:
 		heap2 = heap;
 		heap = mem_heap_create(1024);
 
-		if (!trx_undo_prev_version_build(rec, mtr, version, index,
+		if (trx_undo_prev_version_build(version, index, *offsets, heap,
-						 *offsets, heap,
+						&prev_version, in_heap, vrow,
-						 &prev_version,
+						0) != DB_SUCCESS) {
-						 in_heap, vrow, 0)) {
 			mem_heap_free(heap);
 			heap = heap2;
 			heap2 = NULL;

storage/innobase/trx/trx0purge.cc

@@ -42,10 +42,6 @@ Created 3/26/1996 Heikki Tuuri
 
 #include <unordered_map>
 
-#ifdef UNIV_PFS_RWLOCK
-extern mysql_pfs_key_t trx_purge_latch_key;
-#endif /* UNIV_PFS_RWLOCK */
-
 /** Maximum allowable purge history length.  <=0 means 'infinite'. */
 ulong		srv_max_purge_lag = 0;
 
@@ -184,6 +180,7 @@ void purge_sys_t::create()
   hdr_page_no= 0;
   hdr_offset= 0;
   latch.SRW_LOCK_INIT(trx_purge_latch_key);
+  end_latch.init();
   mysql_mutex_init(purge_sys_pq_mutex_key, &pq_mutex, nullptr);
   truncate.current= NULL;
   truncate.last= NULL;
@@ -205,11 +202,40 @@ void purge_sys_t::close()
   trx->state= TRX_STATE_NOT_STARTED;
   trx->free();
   latch.destroy();
+  end_latch.destroy();
   mysql_mutex_destroy(&pq_mutex);
   mem_heap_free(heap);
   heap= nullptr;
 }
 
+/** Determine if the history of a transaction is purgeable.
+@param trx_id  transaction identifier
+@return whether the history is purgeable */
+TRANSACTIONAL_TARGET bool purge_sys_t::is_purgeable(trx_id_t trx_id) const
+{
+  bool purgeable;
+#if !defined SUX_LOCK_GENERIC && !defined NO_ELISION
+  purgeable= false;
+  if (xbegin())
+  {
+    if (!latch.is_write_locked())
+    {
+      purgeable= view.changes_visible(trx_id);
+      xend();
+    }
+    else
+      xabort();
+  }
+  else
+#endif
+  {
+    latch.rd_lock(SRW_LOCK_CALL);
+    purgeable= view.changes_visible(trx_id);
+    latch.rd_unlock();
+  }
+  return purgeable;
+}
+
 /*================ UNDO LOG HISTORY LIST =============================*/
 
 /** Prepend the history list with an undo log.
@@ -1199,7 +1225,6 @@ trx_purge_attach_undo_recs(ulint n_purge_threads)
 
 	i = 0;
 
-	const ulint		batch_size = srv_purge_batch_size;
 	std::unordered_map<table_id_t, purge_node_t*> table_id_map;
 	mem_heap_empty(purge_sys.heap);
 
@@ -1251,7 +1276,7 @@ trx_purge_attach_undo_recs(ulint n_purge_threads)
 
 		node->undo_recs.push(purge_rec);
 
-		if (n_pages_handled >= batch_size) {
+		if (n_pages_handled >= srv_purge_batch_size) {
 			break;
 		}
 	}
@@ -1303,7 +1328,7 @@ extern tpool::waitable_task purge_worker_task;
 /** Wait for pending purge jobs to complete. */
 static void trx_purge_wait_for_workers_to_complete()
 {
-  bool notify_wait = purge_worker_task.is_running();
+  const bool notify_wait{purge_worker_task.is_running()};
 
   if (notify_wait)
     tpool::tpool_wait_begin();
@@ -1318,12 +1343,33 @@ static void trx_purge_wait_for_workers_to_complete()
   ut_ad(srv_get_task_queue_length() == 0);
 }
 
+/** Update end_view at the end of a purge batch. */
+TRANSACTIONAL_INLINE void purge_sys_t::clone_end_view()
+{
+  /* This is only invoked only by the purge coordinator,
+  which is the only thread that can modify our inputs head, tail, view.
+  Therefore, we only need to protect end_view from concurrent reads. */
+
+  /* Limit the end_view similar to what trx_purge_truncate_history() does. */
+  const trx_id_t trx_no= head.trx_no ? head.trx_no : tail.trx_no;
+#ifdef SUX_LOCK_GENERIC
+  end_latch.wr_lock();
+#else
+  transactional_lock_guard<srw_spin_lock_low> g(end_latch);
+#endif
+  end_view= view;
+  end_view.clamp_low_limit_id(trx_no);
+#ifdef SUX_LOCK_GENERIC
+  end_latch.wr_unlock();
+#endif
+}
+
 /**
 Run a purge batch.
 @param n_tasks   number of purge tasks to submit to the queue
 @param truncate  whether to truncate the history at the end of the batch
 @return number of undo log pages handled in the batch */
-ulint trx_purge(ulint n_tasks, bool truncate)
+TRANSACTIONAL_TARGET ulint trx_purge(ulint n_tasks, bool truncate)
 {
 	que_thr_t*	thr = NULL;
 	ulint		n_pages_handled;
@@ -1357,6 +1403,8 @@ ulint trx_purge(ulint n_tasks, bool truncate)
 
 	trx_purge_wait_for_workers_to_complete();
 
+	purge_sys.clone_end_view();
+
 	if (truncate) {
 		trx_purge_truncate_history();
 	}

storage/innobase/trx/trx0rec.cc

@@ -2076,51 +2076,49 @@ trx_undo_get_undo_rec_low(
   return undo_rec;
 }
 
-/** Copy an undo record to heap.
+/** Copy an undo record to heap, to check if a secondary index record
-@param[in]	roll_ptr	roll pointer to record
+can be safely purged.
-@param[in,out]	heap		memory heap where copied
+@param trx_id   DB_TRX_ID corresponding to roll_ptr
-@param[in]	trx_id		id of the trx that generated
+@param name     table name
-				the roll pointer: it points to an
+@param roll_ptr	DB_ROLL_PTR pointing to the undo log record
-				undo log of this transaction
+@param heap     memory heap for allocation
-@param[in]	name		table name
+@return copy of the record
-@param[out]	undo_rec	own: copy of the record
+@retval nullptr if the version is visible to purge_sys.view */
-@retval true if the undo log has been
+static trx_undo_rec_t *trx_undo_get_rec_if_purgeable(trx_id_t trx_id,
-truncated and we cannot fetch the old version
-@retval false if the undo log record is available
-NOTE: the caller must have latches on the clustered index page. */
-static MY_ATTRIBUTE((warn_unused_result))
-bool
-trx_undo_get_undo_rec(
-	roll_ptr_t		roll_ptr,
-	mem_heap_t*		heap,
-	trx_id_t		trx_id,
                                                      const table_name_t &name,
-	trx_undo_rec_t**	undo_rec)
+                                                     roll_ptr_t roll_ptr,
+                                                     mem_heap_t* heap)
 {
-	purge_sys.latch.rd_lock(SRW_LOCK_CALL);
+  {
-
+    purge_sys_t::view_guard check;
-	bool missing_history = purge_sys.changes_visible(trx_id, name);
+    if (!check.view().changes_visible(trx_id))
-	if (!missing_history) {
+      return trx_undo_get_undo_rec_low(roll_ptr, heap);
-		*undo_rec = trx_undo_get_undo_rec_low(roll_ptr, heap);
+  }
-		missing_history = !*undo_rec;
+  return nullptr;
 }
 
-	purge_sys.latch.rd_unlock();
+/** Copy an undo record to heap.
-
+@param trx_id   DB_TRX_ID corresponding to roll_ptr
-	return missing_history;
+@param name     table name
+@param roll_ptr	DB_ROLL_PTR pointing to the undo log record
+@param heap     memory heap for allocation
+@return copy of the record
+@retval nullptr if the undo log is not available */
+static trx_undo_rec_t *trx_undo_get_undo_rec(trx_id_t trx_id,
+                                             const table_name_t &name,
+                                             roll_ptr_t roll_ptr,
+                                             mem_heap_t *heap)
+{
+  {
+    purge_sys_t::end_view_guard check;
+    if (!check.view().changes_visible(trx_id))
+      return trx_undo_get_undo_rec_low(roll_ptr, heap);
+  }
+  return nullptr;
 }
-
-#ifdef UNIV_DEBUG
-#define ATTRIB_USED_ONLY_IN_DEBUG
-#else /* UNIV_DEBUG */
-#define ATTRIB_USED_ONLY_IN_DEBUG	MY_ATTRIBUTE((unused))
-#endif /* UNIV_DEBUG */
 
 /** Build a previous version of a clustered index record. The caller
 must hold a latch on the index page of the clustered index record.
-@param	index_rec	clustered index record in the index tree
-@param	index_mtr	mtr which contains the latch to index_rec page
-			and purge_view
 @param	rec		version of a clustered index record
 @param	index		clustered index
 @param	offsets		rec_get_offsets(rec, index)
@@ -2141,14 +2139,13 @@ must hold a latch on the index page of the clustered index record.
 			And if we read "after image" of undo log
 @param	undo_block	undo log block which was cached during
 			online dml apply or nullptr
-@retval true if previous version was built, or if it was an insert
+@return error code
-or the table has been rebuilt
+@retval DB_SUCCESS if previous version was successfully built,
-@retval false if the previous version is earlier than purge_view,
+or if it was an insert or the undo record refers to the table before rebuild
-or being purged, which means that it may have been removed */
+@retval DB_MISSING_HISTORY if the history is missing */
-bool
+TRANSACTIONAL_TARGET
+dberr_t
 trx_undo_prev_version_build(
-	const rec_t	*index_rec ATTRIB_USED_ONLY_IN_DEBUG,
-	mtr_t		*index_mtr ATTRIB_USED_ONLY_IN_DEBUG,
 	const rec_t 	*rec,
 	dict_index_t	*index,
 	rec_offs	*offsets,
@@ -2158,7 +2155,6 @@ trx_undo_prev_version_build(
 	dtuple_t	**vrow,
 	ulint		v_status)
 {
-	trx_undo_rec_t*	undo_rec	= NULL;
 	dtuple_t*	entry;
 	trx_id_t	rec_trx_id;
 	ulint		type;
@@ -2173,11 +2169,7 @@ trx_undo_prev_version_build(
 	byte*		buf;
 
 	ut_ad(!index->table->is_temporary());
-	ut_ad(index_mtr->memo_contains_page_flagged(index_rec,
-						    MTR_MEMO_PAGE_S_FIX
-						    | MTR_MEMO_PAGE_X_FIX));
 	ut_ad(rec_offs_validate(rec, index, offsets));
-	ut_a(index->is_primary());
 
 	roll_ptr = row_get_rec_roll_ptr(rec, index, offsets);
 
@@ -2185,27 +2177,20 @@ trx_undo_prev_version_build(
 
 	if (trx_undo_roll_ptr_is_insert(roll_ptr)) {
 		/* The record rec is the first inserted version */
-		return(true);
+		return DB_SUCCESS;
 	}
 
 	rec_trx_id = row_get_rec_trx_id(rec, index, offsets);
 
 	ut_ad(!index->table->skip_alter_undo);
 
-	if (trx_undo_get_undo_rec(
+	trx_undo_rec_t*	undo_rec = v_status == TRX_UNDO_CHECK_PURGEABILITY
-		    roll_ptr, heap, rec_trx_id, index->table->name,
+		? trx_undo_get_rec_if_purgeable(rec_trx_id, index->table->name,
-		    &undo_rec)) {
+						roll_ptr, heap)
-		if (v_status & TRX_UNDO_PREV_IN_PURGE) {
+		: trx_undo_get_undo_rec(rec_trx_id, index->table->name,
-			/* We are fetching the record being purged */
+					roll_ptr, heap);
-			undo_rec = trx_undo_get_undo_rec_low(roll_ptr, heap);
 	if (!undo_rec) {
-				return false;
+		return DB_MISSING_HISTORY;
-			}
-		} else {
-			/* The undo record may already have been purged,
-			during purge or semi-consistent read. */
-			return(false);
-		}
 	}
 
 	const byte *ptr =
@@ -2216,7 +2201,7 @@ trx_undo_prev_version_build(
 		/* The table should have been rebuilt, but purge has
 		not yet removed the undo log records for the
 		now-dropped old table (table_id). */
-		return(true);
+		return DB_SUCCESS;
 	}
 
 	ptr = trx_undo_update_rec_get_sys_cols(ptr, &trx_id, &roll_ptr,
@@ -2264,24 +2249,9 @@ trx_undo_prev_version_build(
 		delete-marked record by trx_id, no transactions need to access
 		the BLOB. */
 
-		/* the row_upd_changes_disowned_external(update) call could be
+		if (update->info_bits & REC_INFO_DELETED_FLAG
-		omitted, but the synchronization on purge_sys.latch is likely
+		    && purge_sys.is_purgeable(trx_id)) {
-		more expensive. */
+			return DB_SUCCESS;
-
-		if ((update->info_bits & REC_INFO_DELETED_FLAG)
-		    && row_upd_changes_disowned_external(update)) {
-			purge_sys.latch.rd_lock(SRW_LOCK_CALL);
-
-			bool missing_extern = purge_sys.changes_visible(
-				trx_id,	index->table->name);
-
-			purge_sys.latch.rd_unlock();
-
-			if (missing_extern) {
-				/* treat as a fresh insert, not to
-				cause assertion error at the caller. */
-				return(true);
-			}
 		}
 
 		/* We have to set the appropriate extern storage bits in the
@@ -2296,8 +2266,8 @@ trx_undo_prev_version_build(
 		following call is safe. */
 		if (!row_upd_index_replace_new_col_vals(entry, *index, update,
 							heap)) {
-			ut_a(v_status & TRX_UNDO_PREV_IN_PURGE);
+			return (v_status & TRX_UNDO_PREV_IN_PURGE)
-			return false;
+				? DB_MISSING_HISTORY : DB_CORRUPTION;
 		}
 
 		/* Get number of externally stored columns in updated record */
@@ -2394,7 +2364,7 @@ trx_undo_prev_version_build(
 				     v_status & TRX_UNDO_PREV_IN_PURGE);
 	}
 
-	return(true);
+	return DB_SUCCESS;
 }
 
 /** Read virtual column value from undo log

storage/innobase/trx/trx0sys.cc

@@ -44,40 +44,6 @@ Created 3/26/1996 Heikki Tuuri
 /** The transaction system */
 trx_sys_t		trx_sys;
 
-/** Check whether transaction id is valid.
-@param[in]	id              transaction id to check
-@param[in]      name            table name */
-void
-ReadViewBase::check_trx_id_sanity(
-	trx_id_t		id,
-	const table_name_t&	name)
-{
-	if (id >= trx_sys.get_max_trx_id()) {
-
-		ib::warn() << "A transaction id"
-			   << " in a record of table "
-			   << name
-			   << " is newer than the"
-			   << " system-wide maximum.";
-		ut_ad(0);
-		THD *thd = current_thd;
-		if (thd != NULL) {
-			char    table_name[MAX_FULL_NAME_LEN + 1];
-
-			innobase_format_name(
-				table_name, sizeof(table_name),
-				name.m_name);
-
-			push_warning_printf(thd, Sql_condition::WARN_LEVEL_WARN,
-					    ER_SIGNAL_WARN,
-					    "InnoDB: Transaction id"
-					    " in a record of table"
-					    " %s is newer than system-wide"
-					    " maximum.", table_name);
-		}
-	}
-}
-
 #ifdef UNIV_DEBUG
 /* Flag to control TRX_RSEG_N_SLOTS behavior debugging. */
 uint	trx_rseg_n_slots_debug = 0;

storage/innobase/trx/trx0trx.cc

@@ -485,6 +485,7 @@ TRANSACTIONAL_INLINE inline void trx_t::commit_state()
 /** Release any explicit locks of a committing transaction. */
 inline void trx_t::release_locks()
 {
+  DEBUG_SYNC_C("trx_t_release_locks_enter");
   DBUG_ASSERT(state == TRX_STATE_COMMITTED_IN_MEMORY);
   DBUG_ASSERT(!is_referenced());
 
@@ -785,7 +786,7 @@ corrupted:
 		ib::info() << "Trx id counter is " << trx_sys.get_max_trx_id();
 	}
 
-	purge_sys.clone_oldest_view();
+	purge_sys.clone_oldest_view<true>();
 	return DB_SUCCESS;
 }
 
@@ -2176,6 +2177,7 @@ trx_set_rw_mode(
 	ut_ad(trx->rsegs.m_redo.rseg != 0);
 
 	trx_sys.register_rw(trx);
+	ut_ad(trx->id);
 
 	/* So that we can see our own changes. */
 	if (trx->read_view.is_open()) {