Merge 10.6 into 10.7

40 files changed, 1381 insertions, 1139 deletions

diff --git a/mysql-test/main/mysql_tzinfo_to_sql_symlink.result b/mysql-test/main/mysql_tzinfo_to_sql_symlink.result
index a6be207434f..d5c6518a130 100644
--- a/mysql-test/main/mysql_tzinfo_to_sql_symlink.result
+++ b/mysql-test/main/mysql_tzinfo_to_sql_symlink.result
@@ -7,9 +7,9 @@ CREATE TABLE time_zone_leap_second LIKE mysql.time_zone_leap_second;
 # MDEV-5226 mysql_tzinfo_to_sql errors with tzdata 2013f and above
 #
 # Verbose run
-set @wsrep_is_on=(select sum(SESSION_VALUE='ON') from information_schema.SYSTEM_VARIABLES WHERE VARIABLE_NAME='wsrep_on');
+set @wsrep_is_on=(select coalesce(sum(SESSION_VALUE='ON'), 0) from information_schema.SYSTEM_VARIABLES WHERE VARIABLE_NAME='wsrep_on');
 SELECT concat('%', GROUP_CONCAT(OPTION), '%') INTO @replicate_opt  FROM   (SELECT DISTINCT concat('REPLICATE_', UPPER(ENGINE)) AS OPTION    FROM information_schema.TABLES    WHERE TABLE_SCHEMA=DATABASE()      AND TABLE_NAME IN ('time_zone',                         'time_zone_name',                         'time_zone_transition',                         'time_zone_transition_type',                         'time_zone_leap_second')      AND ENGINE in ('MyISAM',                     'Aria')) AS o ORDER BY OPTION DESC;
-set @wsrep_cannot_replicate_tz=@wsrep_is_on AND (select sum(GLOBAL_VALUE NOT LIKE @replicate_opt) from information_schema.SYSTEM_VARIABLES WHERE VARIABLE_NAME='wsrep_mode');
+set @wsrep_cannot_replicate_tz=@wsrep_is_on AND (select coalesce(sum(GLOBAL_VALUE NOT LIKE @replicate_opt), 0) from information_schema.SYSTEM_VARIABLES WHERE VARIABLE_NAME='wsrep_mode');
 execute immediate if(@wsrep_cannot_replicate_tz, "select ENGINE into @time_zone_engine from information_schema.TABLES where TABLE_SCHEMA=DATABASE() and TABLE_NAME='time_zone'", 'do 0');
 execute immediate if(@wsrep_cannot_replicate_tz, 'ALTER TABLE time_zone ENGINE=InnoDB', 'do 0');
 execute immediate if(@wsrep_cannot_replicate_tz, "select ENGINE into @time_zone_name_engine from information_schema.TABLES where TABLE_SCHEMA=DATABASE() and TABLE_NAME='time_zone_name'", 'do 0');
@@ -56,9 +56,9 @@ execute immediate if(@wsrep_cannot_replicate_tz, concat('ALTER TABLE time_zone_t
 # MDEV-28263: mariadb-tzinfo-to-sql improve wsrep and binlog cases
 #
 # Run on zoneinfo directory
-set @wsrep_is_on=(select sum(SESSION_VALUE='ON') from information_schema.SYSTEM_VARIABLES WHERE VARIABLE_NAME='wsrep_on');
+set @wsrep_is_on=(select coalesce(sum(SESSION_VALUE='ON'), 0) from information_schema.SYSTEM_VARIABLES WHERE VARIABLE_NAME='wsrep_on');
 SELECT concat('%', GROUP_CONCAT(OPTION), '%') INTO @replicate_opt  FROM   (SELECT DISTINCT concat('REPLICATE_', UPPER(ENGINE)) AS OPTION    FROM information_schema.TABLES    WHERE TABLE_SCHEMA=DATABASE()      AND TABLE_NAME IN ('time_zone',                         'time_zone_name',                         'time_zone_transition',                         'time_zone_transition_type',                         'time_zone_leap_second')      AND ENGINE in ('MyISAM',                     'Aria')) AS o ORDER BY OPTION DESC;
-set @wsrep_cannot_replicate_tz=@wsrep_is_on AND (select sum(GLOBAL_VALUE NOT LIKE @replicate_opt) from information_schema.SYSTEM_VARIABLES WHERE VARIABLE_NAME='wsrep_mode');
+set @wsrep_cannot_replicate_tz=@wsrep_is_on AND (select coalesce(sum(GLOBAL_VALUE NOT LIKE @replicate_opt), 0) from information_schema.SYSTEM_VARIABLES WHERE VARIABLE_NAME='wsrep_mode');
 execute immediate if(@wsrep_cannot_replicate_tz, "select ENGINE into @time_zone_engine from information_schema.TABLES where TABLE_SCHEMA=DATABASE() and TABLE_NAME='time_zone'", 'do 0');
 execute immediate if(@wsrep_cannot_replicate_tz, 'ALTER TABLE time_zone ENGINE=InnoDB', 'do 0');
 execute immediate if(@wsrep_cannot_replicate_tz, "select ENGINE into @time_zone_name_engine from information_schema.TABLES where TABLE_SCHEMA=DATABASE() and TABLE_NAME='time_zone_name'", 'do 0');
@@ -116,9 +116,9 @@ COUNT(*)
 #
 # Run on zoneinfo directory --skip-write-binlog
 #
-set @wsrep_is_on=(select sum(SESSION_VALUE='ON') from information_schema.SYSTEM_VARIABLES WHERE VARIABLE_NAME='wsrep_on');
+set @wsrep_is_on=(select coalesce(sum(SESSION_VALUE='ON'), 0) from information_schema.SYSTEM_VARIABLES WHERE VARIABLE_NAME='wsrep_on');
 SELECT concat('%', GROUP_CONCAT(OPTION), '%') INTO @replicate_opt  FROM   (SELECT DISTINCT concat('REPLICATE_', UPPER(ENGINE)) AS OPTION    FROM information_schema.TABLES    WHERE TABLE_SCHEMA=DATABASE()      AND TABLE_NAME IN ('time_zone',                         'time_zone_name',                         'time_zone_transition',                         'time_zone_transition_type',                         'time_zone_leap_second')      AND ENGINE in ('MyISAM',                     'Aria')) AS o ORDER BY OPTION DESC;
-set @wsrep_cannot_replicate_tz=@wsrep_is_on AND (select sum(GLOBAL_VALUE NOT LIKE @replicate_opt) from information_schema.SYSTEM_VARIABLES WHERE VARIABLE_NAME='wsrep_mode');
+set @wsrep_cannot_replicate_tz=@wsrep_is_on AND (select coalesce(sum(GLOBAL_VALUE NOT LIKE @replicate_opt), 0) from information_schema.SYSTEM_VARIABLES WHERE VARIABLE_NAME='wsrep_mode');
 execute immediate if(@wsrep_is_on, 'SET @save_wsrep_on=@@WSREP_ON, WSREP_ON=OFF', 'do 0');
 SET @save_sql_log_bin=@@SQL_LOG_BIN;
 SET SESSION SQL_LOG_BIN=0;
@@ -188,9 +188,9 @@ TRUNCATE TABLE time_zone_leap_second;
 #
 # Testing with explicit timezonefile
 #
-set @wsrep_is_on=(select sum(SESSION_VALUE='ON') from information_schema.SYSTEM_VARIABLES WHERE VARIABLE_NAME='wsrep_on');
+set @wsrep_is_on=(select coalesce(sum(SESSION_VALUE='ON'), 0) from information_schema.SYSTEM_VARIABLES WHERE VARIABLE_NAME='wsrep_on');
 SELECT concat('%', GROUP_CONCAT(OPTION), '%') INTO @replicate_opt  FROM   (SELECT DISTINCT concat('REPLICATE_', UPPER(ENGINE)) AS OPTION    FROM information_schema.TABLES    WHERE TABLE_SCHEMA=DATABASE()      AND TABLE_NAME IN ('time_zone',                         'time_zone_name',                         'time_zone_transition',                         'time_zone_transition_type',                         'time_zone_leap_second')      AND ENGINE in ('MyISAM',                     'Aria')) AS o ORDER BY OPTION DESC;
-set @wsrep_cannot_replicate_tz=@wsrep_is_on AND (select sum(GLOBAL_VALUE NOT LIKE @replicate_opt) from information_schema.SYSTEM_VARIABLES WHERE VARIABLE_NAME='wsrep_mode');
+set @wsrep_cannot_replicate_tz=@wsrep_is_on AND (select coalesce(sum(GLOBAL_VALUE NOT LIKE @replicate_opt), 0) from information_schema.SYSTEM_VARIABLES WHERE VARIABLE_NAME='wsrep_mode');
 execute immediate if(@wsrep_cannot_replicate_tz, "select ENGINE into @time_zone_engine from information_schema.TABLES where TABLE_SCHEMA=DATABASE() and TABLE_NAME='time_zone'", 'do 0');
 execute immediate if(@wsrep_cannot_replicate_tz, 'ALTER TABLE time_zone ENGINE=InnoDB', 'do 0');
 execute immediate if(@wsrep_cannot_replicate_tz, "select ENGINE into @time_zone_name_engine from information_schema.TABLES where TABLE_SCHEMA=DATABASE() and TABLE_NAME='time_zone_name'", 'do 0');
@@ -252,9 +252,9 @@ TRUNCATE TABLE time_zone_leap_second;
 #
 # Testing with explicit timezonefile --skip-write-binlog
 #
-set @wsrep_is_on=(select sum(SESSION_VALUE='ON') from information_schema.SYSTEM_VARIABLES WHERE VARIABLE_NAME='wsrep_on');
+set @wsrep_is_on=(select coalesce(sum(SESSION_VALUE='ON'), 0) from information_schema.SYSTEM_VARIABLES WHERE VARIABLE_NAME='wsrep_on');
 SELECT concat('%', GROUP_CONCAT(OPTION), '%') INTO @replicate_opt  FROM   (SELECT DISTINCT concat('REPLICATE_', UPPER(ENGINE)) AS OPTION    FROM information_schema.TABLES    WHERE TABLE_SCHEMA=DATABASE()      AND TABLE_NAME IN ('time_zone',                         'time_zone_name',                         'time_zone_transition',                         'time_zone_transition_type',                         'time_zone_leap_second')      AND ENGINE in ('MyISAM',                     'Aria')) AS o ORDER BY OPTION DESC;
-set @wsrep_cannot_replicate_tz=@wsrep_is_on AND (select sum(GLOBAL_VALUE NOT LIKE @replicate_opt) from information_schema.SYSTEM_VARIABLES WHERE VARIABLE_NAME='wsrep_mode');
+set @wsrep_cannot_replicate_tz=@wsrep_is_on AND (select coalesce(sum(GLOBAL_VALUE NOT LIKE @replicate_opt), 0) from information_schema.SYSTEM_VARIABLES WHERE VARIABLE_NAME='wsrep_mode');
 execute immediate if(@wsrep_is_on, 'SET @save_wsrep_on=@@WSREP_ON, WSREP_ON=OFF', 'do 0');
 SET @save_sql_log_bin=@@SQL_LOG_BIN;
 SET SESSION SQL_LOG_BIN=0;
@@ -310,9 +310,9 @@ TRUNCATE TABLE time_zone_leap_second;
 #
 # Testing --leap
 #
-set @wsrep_is_on=(select sum(SESSION_VALUE='ON') from information_schema.SYSTEM_VARIABLES WHERE VARIABLE_NAME='wsrep_on');
+set @wsrep_is_on=(select coalesce(sum(SESSION_VALUE='ON'), 0) from information_schema.SYSTEM_VARIABLES WHERE VARIABLE_NAME='wsrep_on');
 SELECT concat('%', GROUP_CONCAT(OPTION), '%') INTO @replicate_opt  FROM   (SELECT DISTINCT concat('REPLICATE_', UPPER(ENGINE)) AS OPTION    FROM information_schema.TABLES    WHERE TABLE_SCHEMA=DATABASE()      AND TABLE_NAME IN ('time_zone',                         'time_zone_name',                         'time_zone_transition',                         'time_zone_transition_type',                         'time_zone_leap_second')      AND ENGINE in ('MyISAM',                     'Aria')) AS o ORDER BY OPTION DESC;
-set @wsrep_cannot_replicate_tz=@wsrep_is_on AND (select sum(GLOBAL_VALUE NOT LIKE @replicate_opt) from information_schema.SYSTEM_VARIABLES WHERE VARIABLE_NAME='wsrep_mode');
+set @wsrep_cannot_replicate_tz=@wsrep_is_on AND (select coalesce(sum(GLOBAL_VALUE NOT LIKE @replicate_opt), 0) from information_schema.SYSTEM_VARIABLES WHERE VARIABLE_NAME='wsrep_mode');
 execute immediate if(@wsrep_cannot_replicate_tz, "select ENGINE into @time_zone_engine from information_schema.TABLES where TABLE_SCHEMA=DATABASE() and TABLE_NAME='time_zone'", 'do 0');
 execute immediate if(@wsrep_cannot_replicate_tz, 'ALTER TABLE time_zone ENGINE=InnoDB', 'do 0');
 execute immediate if(@wsrep_cannot_replicate_tz, "select ENGINE into @time_zone_name_engine from information_schema.TABLES where TABLE_SCHEMA=DATABASE() and TABLE_NAME='time_zone_name'", 'do 0');
@@ -373,9 +373,9 @@ TRUNCATE TABLE time_zone_leap_second;
 #
 # Testing --skip-write-binlog --leap
 #
-set @wsrep_is_on=(select sum(SESSION_VALUE='ON') from information_schema.SYSTEM_VARIABLES WHERE VARIABLE_NAME='wsrep_on');
+set @wsrep_is_on=(select coalesce(sum(SESSION_VALUE='ON'), 0) from information_schema.SYSTEM_VARIABLES WHERE VARIABLE_NAME='wsrep_on');
 SELECT concat('%', GROUP_CONCAT(OPTION), '%') INTO @replicate_opt  FROM   (SELECT DISTINCT concat('REPLICATE_', UPPER(ENGINE)) AS OPTION    FROM information_schema.TABLES    WHERE TABLE_SCHEMA=DATABASE()      AND TABLE_NAME IN ('time_zone',                         'time_zone_name',                         'time_zone_transition',                         'time_zone_transition_type',                         'time_zone_leap_second')      AND ENGINE in ('MyISAM',                     'Aria')) AS o ORDER BY OPTION DESC;
-set @wsrep_cannot_replicate_tz=@wsrep_is_on AND (select sum(GLOBAL_VALUE NOT LIKE @replicate_opt) from information_schema.SYSTEM_VARIABLES WHERE VARIABLE_NAME='wsrep_mode');
+set @wsrep_cannot_replicate_tz=@wsrep_is_on AND (select coalesce(sum(GLOBAL_VALUE NOT LIKE @replicate_opt), 0) from information_schema.SYSTEM_VARIABLES WHERE VARIABLE_NAME='wsrep_mode');
 execute immediate if(@wsrep_is_on, 'SET @save_wsrep_on=@@WSREP_ON, WSREP_ON=OFF', 'do 0');
 SET @save_sql_log_bin=@@SQL_LOG_BIN;
 SET SESSION SQL_LOG_BIN=0;
@@ -425,9 +425,9 @@ COM_TRUNCATE	1
 #
 # Testing --skip-write-binlog
 #
-set @wsrep_is_on=(select sum(SESSION_VALUE='ON') from information_schema.SYSTEM_VARIABLES WHERE VARIABLE_NAME='wsrep_on');
+set @wsrep_is_on=(select coalesce(sum(SESSION_VALUE='ON'), 0) from information_schema.SYSTEM_VARIABLES WHERE VARIABLE_NAME='wsrep_on');
 SELECT concat('%', GROUP_CONCAT(OPTION), '%') INTO @replicate_opt  FROM   (SELECT DISTINCT concat('REPLICATE_', UPPER(ENGINE)) AS OPTION    FROM information_schema.TABLES    WHERE TABLE_SCHEMA=DATABASE()      AND TABLE_NAME IN ('time_zone',                         'time_zone_name',                         'time_zone_transition',                         'time_zone_transition_type',                         'time_zone_leap_second')      AND ENGINE in ('MyISAM',                     'Aria')) AS o ORDER BY OPTION DESC;
-set @wsrep_cannot_replicate_tz=@wsrep_is_on AND (select sum(GLOBAL_VALUE NOT LIKE @replicate_opt) from information_schema.SYSTEM_VARIABLES WHERE VARIABLE_NAME='wsrep_mode');
+set @wsrep_cannot_replicate_tz=@wsrep_is_on AND (select coalesce(sum(GLOBAL_VALUE NOT LIKE @replicate_opt), 0) from information_schema.SYSTEM_VARIABLES WHERE VARIABLE_NAME='wsrep_mode');
 execute immediate if(@wsrep_is_on, 'SET @save_wsrep_on=@@WSREP_ON, WSREP_ON=OFF', 'do 0');
 SET @save_sql_log_bin=@@SQL_LOG_BIN;
 SET SESSION SQL_LOG_BIN=0;
@@ -447,9 +447,9 @@ UNLOCK TABLES;
 COMMIT;
 SET SESSION SQL_LOG_BIN=@save_sql_log_bin;
 execute immediate if(@wsrep_is_on, 'SET SESSION WSREP_ON=@save_wsrep_on', 'do 0');
-set @wsrep_is_on=(select sum(SESSION_VALUE='ON') from information_schema.SYSTEM_VARIABLES WHERE VARIABLE_NAME='wsrep_on');
+set @wsrep_is_on=(select coalesce(sum(SESSION_VALUE='ON'), 0) from information_schema.SYSTEM_VARIABLES WHERE VARIABLE_NAME='wsrep_on');
 SELECT concat('%', GROUP_CONCAT(OPTION), '%') INTO @replicate_opt  FROM   (SELECT DISTINCT concat('REPLICATE_', UPPER(ENGINE)) AS OPTION    FROM information_schema.TABLES    WHERE TABLE_SCHEMA=DATABASE()      AND TABLE_NAME IN ('time_zone',                         'time_zone_name',                         'time_zone_transition',                         'time_zone_transition_type',                         'time_zone_leap_second')      AND ENGINE in ('MyISAM',                     'Aria')) AS o ORDER BY OPTION DESC;
-set @wsrep_cannot_replicate_tz=@wsrep_is_on AND (select sum(GLOBAL_VALUE NOT LIKE @replicate_opt) from information_schema.SYSTEM_VARIABLES WHERE VARIABLE_NAME='wsrep_mode');
+set @wsrep_cannot_replicate_tz=@wsrep_is_on AND (select coalesce(sum(GLOBAL_VALUE NOT LIKE @replicate_opt), 0) from information_schema.SYSTEM_VARIABLES WHERE VARIABLE_NAME='wsrep_mode');
 execute immediate if(@wsrep_is_on, 'SET @save_wsrep_on=@@WSREP_ON, WSREP_ON=OFF', 'do 0');
 SET @save_sql_log_bin=@@SQL_LOG_BIN;
 SET SESSION SQL_LOG_BIN=0;
@@ -471,9 +471,9 @@ execute immediate if(@wsrep_is_on, 'SET SESSION WSREP_ON=@save_wsrep_on', 'do 0'
 #
 # MDEV-6236 - [PATCH] mysql_tzinfo_to_sql may produce invalid SQL
 #
-set @wsrep_is_on=(select sum(SESSION_VALUE='ON') from information_schema.SYSTEM_VARIABLES WHERE VARIABLE_NAME='wsrep_on');
+set @wsrep_is_on=(select coalesce(sum(SESSION_VALUE='ON'), 0) from information_schema.SYSTEM_VARIABLES WHERE VARIABLE_NAME='wsrep_on');
 SELECT concat('%', GROUP_CONCAT(OPTION), '%') INTO @replicate_opt  FROM   (SELECT DISTINCT concat('REPLICATE_', UPPER(ENGINE)) AS OPTION    FROM information_schema.TABLES    WHERE TABLE_SCHEMA=DATABASE()      AND TABLE_NAME IN ('time_zone',                         'time_zone_name',                         'time_zone_transition',                         'time_zone_transition_type',                         'time_zone_leap_second')      AND ENGINE in ('MyISAM',                     'Aria')) AS o ORDER BY OPTION DESC;
-set @wsrep_cannot_replicate_tz=@wsrep_is_on AND (select sum(GLOBAL_VALUE NOT LIKE @replicate_opt) from information_schema.SYSTEM_VARIABLES WHERE VARIABLE_NAME='wsrep_mode');
+set @wsrep_cannot_replicate_tz=@wsrep_is_on AND (select coalesce(sum(GLOBAL_VALUE NOT LIKE @replicate_opt), 0) from information_schema.SYSTEM_VARIABLES WHERE VARIABLE_NAME='wsrep_mode');
 execute immediate if(@wsrep_cannot_replicate_tz, "select ENGINE into @time_zone_engine from information_schema.TABLES where TABLE_SCHEMA=DATABASE() and TABLE_NAME='time_zone'", 'do 0');
 execute immediate if(@wsrep_cannot_replicate_tz, 'ALTER TABLE time_zone ENGINE=InnoDB', 'do 0');
 execute immediate if(@wsrep_cannot_replicate_tz, "select ENGINE into @time_zone_name_engine from information_schema.TABLES where TABLE_SCHEMA=DATABASE() and TABLE_NAME='time_zone_name'", 'do 0');
diff --git a/mysql-test/main/mysql_tzinfo_to_sql_symlink.test b/mysql-test/main/mysql_tzinfo_to_sql_symlink.test
index 857f24e610f..34df1281c38 100644
--- a/mysql-test/main/mysql_tzinfo_to_sql_symlink.test
+++ b/mysql-test/main/mysql_tzinfo_to_sql_symlink.test
@@ -3,8 +3,6 @@
 --source include/not_windows.inc
 --source include/no_protocol.inc
 
-let $is_embedded=`select version() like '%embedded%'`;
-
 CREATE TABLE time_zone LIKE mysql.time_zone;
 CREATE TABLE time_zone_name LIKE mysql.time_zone_name;
 CREATE TABLE time_zone_transition LIKE mysql.time_zone_transition;
@@ -63,9 +61,6 @@ SELECT COUNT(*) FROM time_zone_transition;
 SELECT COUNT(*) FROM time_zone_transition_type;
 SELECT COUNT(*) FROM time_zone_leap_second;
 
-if ($is_embedded) {
---replace_column 1 0 2 0
-}
 SELECT @wsrep_is_on,  @wsrep_cannot_replicate_tz, @save_wsrep_on, @save_sql_log_bin, @@SQL_LOG_BIN;
 SELECT g.VARIABLE_NAME, g.VARIABLE_VALUE - b.VARIABLE_VALUE AS diff
   FROM information_schema.global_status g
@@ -100,9 +95,6 @@ SELECT COUNT(*) FROM time_zone_transition;
 SELECT COUNT(*) FROM time_zone_transition_type;
 SELECT COUNT(*) FROM time_zone_leap_second;
 
-if ($is_embedded) {
---replace_column 1 0 2 0
-}
 SELECT @wsrep_is_on,  @wsrep_cannot_replicate_tz, @save_wsrep_on, @save_sql_log_bin, @@SQL_LOG_BIN;
 SELECT g.VARIABLE_NAME, g.VARIABLE_VALUE - b.VARIABLE_VALUE AS diff
   FROM information_schema.global_status g
@@ -135,9 +127,6 @@ SELECT COUNT(*) FROM time_zone_transition;
 SELECT COUNT(*) FROM time_zone_transition_type;
 SELECT COUNT(*) FROM time_zone_leap_second;
 
-if ($is_embedded) {
---replace_column 1 0 2 0
-}
 SELECT @wsrep_is_on,  @wsrep_cannot_replicate_tz, @save_wsrep_on, @save_sql_log_bin, @@SQL_LOG_BIN;
 SELECT g.VARIABLE_NAME, g.VARIABLE_VALUE - b.VARIABLE_VALUE AS diff
   FROM information_schema.global_status g
@@ -170,9 +159,6 @@ SELECT COUNT(*) FROM time_zone_transition;
 SELECT COUNT(*) FROM time_zone_transition_type;
 SELECT COUNT(*) FROM time_zone_leap_second;
 
-if ($is_embedded) {
---replace_column 1 0 2 0
-}
 SELECT @wsrep_is_on,  @wsrep_cannot_replicate_tz, @save_wsrep_on, @save_sql_log_bin, @@SQL_LOG_BIN;
 SELECT g.VARIABLE_NAME, g.VARIABLE_VALUE - b.VARIABLE_VALUE AS diff
   FROM information_schema.global_status g
@@ -205,9 +191,6 @@ SELECT COUNT(*) FROM time_zone_transition;
 SELECT COUNT(*) FROM time_zone_transition_type;
 SELECT COUNT(*) FROM time_zone_leap_second;
 
-if ($is_embedded) {
---replace_column 1 0 2 0
-}
 SELECT @wsrep_is_on,  @wsrep_cannot_replicate_tz, @save_wsrep_on, @save_sql_log_bin, @@SQL_LOG_BIN;
 SELECT g.VARIABLE_NAME, g.VARIABLE_VALUE - b.VARIABLE_VALUE AS diff
   FROM information_schema.global_status g
diff --git a/mysql-test/main/strings.result b/mysql-test/main/strings.result
index 2d0eb5e7400..6573e734422 100644
--- a/mysql-test/main/strings.result
+++ b/mysql-test/main/strings.result
@@ -18,3 +18,13 @@ LENGTH(CONCAT_WS(d, ' '))
 1
 1
 DROP TABLE t1;
+#
+# MDEV-28315 ASAN stack-buffer-overflow in String::copy_aligned
+#
+CREATE TABLE t1 (a VARBINARY(128)) CHARACTER SET utf32;
+INSERT INTO t1 VALUES ('South Carolina, Vermont, New Jersey, New Mexico, Wisconsin, Missouri, Delaware');
+CREATE TABLE t2 (b SET('South Carolina', 'Vermont', 'Texas', 'New Mexico', 'Wisconsin', 'Missouri', 'Delaware', 'Wyoming', 'New Jersey', 'Maryland', 'Illinois', 'New York')) CHARACTER SET utf32;
+INSERT INTO t2 SELECT * FROM t1;
+ERROR 01000: Data truncated for column 'b' at row 1
+DROP TABLE t1;
+DROP TABLE t2;
diff --git a/mysql-test/main/strings.test b/mysql-test/main/strings.test
index 083ab7b07a4..3f3f4b572af 100644
--- a/mysql-test/main/strings.test
+++ b/mysql-test/main/strings.test
@@ -24,3 +24,15 @@ CREATE TABLE t1 (d DATE);
 INSERT INTO t1 VALUES ('1920-03-02'),('2020-12-01');
 SELECT LENGTH(CONCAT_WS(d, ' ')) FROM t1;
 DROP TABLE t1;
+
+--echo #
+--echo # MDEV-28315 ASAN stack-buffer-overflow in String::copy_aligned
+--echo #
+
+CREATE TABLE t1 (a VARBINARY(128)) CHARACTER SET utf32;
+INSERT INTO t1 VALUES ('South Carolina, Vermont, New Jersey, New Mexico, Wisconsin, Missouri, Delaware');
+CREATE TABLE t2 (b SET('South Carolina', 'Vermont', 'Texas', 'New Mexico', 'Wisconsin', 'Missouri', 'Delaware', 'Wyoming', 'New Jersey', 'Maryland', 'Illinois', 'New York')) CHARACTER SET utf32;
+--error WARN_DATA_TRUNCATED
+INSERT INTO t2 SELECT * FROM t1;
+DROP TABLE t1;
+DROP TABLE t2;
diff --git a/mysql-test/suite/innodb/r/change_column_collation.result b/mysql-test/suite/innodb/r/change_column_collation.result
index 54eff706276..b1771b2bd00 100644
--- a/mysql-test/suite/innodb/r/change_column_collation.result
+++ b/mysql-test/suite/innodb/r/change_column_collation.result
@@ -40,9 +40,9 @@ INSERT INTO t1 VALUES (4, 'AAA');
 UPDATE t1 set msg = "ddd" where id = 2;
 DELETE FROM t1 WHERE id= 3;
 SET DEBUG_SYNC = 'now SIGNAL go_ahead';
-SET DEBUG_SYNC = 'RESET';
 connection default;
 ERROR 23000: Duplicate entry 'NULL' for key 'msg'
+SET DEBUG_SYNC = 'RESET';
 SELECT * FROM t1;
 id	msg
 4	AAA
diff --git a/mysql-test/suite/innodb/t/change_column_collation.test b/mysql-test/suite/innodb/t/change_column_collation.test
index 7d82aac7ec2..f63d1974c72 100644
--- a/mysql-test/suite/innodb/t/change_column_collation.test
+++ b/mysql-test/suite/innodb/t/change_column_collation.test
@@ -63,13 +63,13 @@ INSERT INTO t1 VALUES (4, 'AAA');
 UPDATE t1 set msg = "ddd" where id = 2;
 DELETE FROM t1 WHERE id= 3;
 SET DEBUG_SYNC = 'now SIGNAL go_ahead';
-SET DEBUG_SYNC = 'RESET';
 
 connection default;
 
 --error ER_DUP_ENTRY
 reap;
 
+SET DEBUG_SYNC = 'RESET';
 SELECT * FROM t1;
 DROP TABLE t1;
 
diff --git a/sql/json_table.cc b/sql/json_table.cc
index 4d07cf0d367..cfc90bb06db 100644
--- a/sql/json_table.cc
+++ b/sql/json_table.cc
@@ -118,10 +118,12 @@ int get_disallowed_table_deps_for_list(MEM_ROOT *mem_root,
   NESTED_JOIN *nested_join;
   List_iterator<TABLE_LIST> li(*join_list);
 
-  long arbitrary_var;
-  long stack_used_up= (available_stack_size(current_thd->thread_stack, &arbitrary_var));
   DBUG_EXECUTE_IF("json_check_min_stack_requirement",
-                  {ALLOCATE_MEM_ON_STACK(my_thread_stack_size-stack_used_up-STACK_MIN_SIZE);});
+                  {
+                    long arbitrary_var;
+                    long stack_used_up= (available_stack_size(current_thd->thread_stack, &arbitrary_var));
+                    ALLOCATE_MEM_ON_STACK(my_thread_stack_size-stack_used_up-STACK_MIN_SIZE);
+                  });
   if (check_stack_overrun(current_thd, STACK_MIN_SIZE , NULL))
     return 1;
 
@@ -1328,10 +1330,12 @@ static void add_extra_deps(List<TABLE_LIST> *join_list, table_map deps)
   TABLE_LIST *table;
   List_iterator<TABLE_LIST> li(*join_list);
 
-  long arbitrary_var;
-  long stack_used_up= (available_stack_size(current_thd->thread_stack, &arbitrary_var));
   DBUG_EXECUTE_IF("json_check_min_stack_requirement",
-                  {ALLOCATE_MEM_ON_STACK(my_thread_stack_size-stack_used_up-STACK_MIN_SIZE);});
+                  {
+                    long arbitrary_var;
+                    long stack_used_up= (available_stack_size(current_thd->thread_stack, &arbitrary_var));
+                    ALLOCATE_MEM_ON_STACK(my_thread_stack_size-stack_used_up-STACK_MIN_SIZE);
+                  });
   if (check_stack_overrun(current_thd, STACK_MIN_SIZE , NULL))
     return;
 
@@ -1423,10 +1427,12 @@ table_map add_table_function_dependencies(List<TABLE_LIST> *join_list,
   table_map res= 0;
   List_iterator<TABLE_LIST> li(*join_list);
 
-  long arbitrary_var;
-  long stack_used_up= (available_stack_size(current_thd->thread_stack, &arbitrary_var));
   DBUG_EXECUTE_IF("json_check_min_stack_requirement",
-                  {ALLOCATE_MEM_ON_STACK(my_thread_stack_size-stack_used_up-STACK_MIN_SIZE);});
+                  {
+                    long arbitrary_var;
+                    long stack_used_up= (available_stack_size(current_thd->thread_stack, &arbitrary_var));
+                    ALLOCATE_MEM_ON_STACK(my_thread_stack_size-stack_used_up-STACK_MIN_SIZE);
+                  });
   if ((res=check_stack_overrun(current_thd, STACK_MIN_SIZE , NULL)))
     return res;
 
diff --git a/sql/sql_string.cc b/sql/sql_string.cc
index 991b65ef4b2..90693f407fb 100644
--- a/sql/sql_string.cc
+++ b/sql/sql_string.cc
@@ -398,7 +398,7 @@ bool String::copy_aligned(const char *str, size_t arg_length, size_t offset,
   DBUG_ASSERT(offset && offset != cs->mbminlen);
 
   size_t aligned_length= arg_length + offset;
-  if (alloc(aligned_length))
+  if (alloc(aligned_length+1))
     return TRUE;
   
   /*
diff --git a/sql/sql_string.h b/sql/sql_string.h
index 795f80c3e08..a8c29eba22c 100644
--- a/sql/sql_string.h
+++ b/sql/sql_string.h
@@ -695,7 +695,7 @@ public:
 
       Note that if arg_length == Alloced_length then we don't allocate.
       This ensures we don't do any extra allocations in protocol and String:int,
-      but the string will not be atomically null terminated if c_ptr() is not
+      but the string will not be automatically null terminated if c_ptr() is not
       called.
     */
     if (arg_length <= Alloced_length && Alloced_length)
diff --git a/sql/tztime.cc b/sql/tztime.cc
index 062e5c0b34c..22e14c8272a 100644
--- a/sql/tztime.cc
+++ b/sql/tztime.cc
@@ -1798,7 +1798,7 @@ end:
   delete thd;
   if (org_thd)
     org_thd->store_globals();			/* purecov: inspected */
-  
+
   default_tz= default_tz_name ? global_system_variables.time_zone
                               : my_tz_SYSTEM;
 
@@ -1873,7 +1873,7 @@ tz_load_from_open_tables(const String *tz_name, TABLE_LIST *tz_tables)
 #ifdef ABBR_ARE_USED
   char chars[MY_MAX(TZ_MAX_CHARS + 1, (2 * (MY_TZNAME_MAX + 1)))];
 #endif
-  /* 
+  /*
     Used as a temporary tz_info until we decide that we actually want to
     allocate and keep the tz info and tz name in tz_storage.
   */
@@ -2026,7 +2026,7 @@ tz_load_from_open_tables(const String *tz_name, TABLE_LIST *tz_tables)
     mysql.time_zone_transition table. Here we additionally need records
     in ascending order by index scan also satisfies us.
   */
-  table= tz_tables->table; 
+  table= tz_tables->table;
   table->field[0]->store((longlong) tzid, TRUE);
   if (table->file->ha_index_init(0, 1))
     goto end;
@@ -2361,7 +2361,7 @@ my_tz_find(THD *thd, const String *name)
 /**
   Convert leap seconds into non-leap
 
-  This function will convert the leap seconds added by the OS to 
+  This function will convert the leap seconds added by the OS to
   non-leap seconds, e.g. 23:59:59, 23:59:60 -> 23:59:59, 00:00:01 ...
   This check is not checking for years on purpose : although it's not a
   complete check this way it doesn't require looking (and having installed)
@@ -2725,11 +2725,20 @@ static const char *trunc_tables_const=
   "TRUNCATE TABLE time_zone_name;\n"
   "TRUNCATE TABLE time_zone_transition;\n"
   "TRUNCATE TABLE time_zone_transition_type;\n";
+
+/*
+   These queries need to return FALSE/0 when the 'wsrep*' variables do not
+   exist at all.
+   Moving the WHERE clause into the sum(...) seems like the obvious solution
+   here, but it does not work in bootstrap mode (see MDEV-28782 and
+   0e4cf497ca11a7298e2bd896cb594bd52085a1d4).
+   Thus we use coalesce(..., 0) instead,
+*/
 static const char *wsrep_is_on=
-  "select sum(SESSION_VALUE='ON')"
+  "select coalesce(sum(SESSION_VALUE='ON'), 0)"
   " from information_schema.SYSTEM_VARIABLES WHERE VARIABLE_NAME='wsrep_on'";
 static const char *wsrep_cannot_replicate_tz=
-  "select sum(GLOBAL_VALUE NOT LIKE @replicate_opt)"
+  "select coalesce(sum(GLOBAL_VALUE NOT LIKE @replicate_opt), 0)"
   " from information_schema.SYSTEM_VARIABLES WHERE VARIABLE_NAME='wsrep_mode'";
 
 int
@@ -2770,7 +2779,7 @@ main(int argc, char **argv)
          " ORDER BY OPTION DESC;\n");
   printf("set @wsrep_cannot_replicate_tz=@wsrep_is_on AND (%s);\n", wsrep_cannot_replicate_tz);
   if (opt_skip_write_binlog)
-    /* If turn off session wsrep if we cannot replicate using galera.
+    /* We turn off session wsrep if we cannot replicate using galera.
        Disable sql_log_bin as the name implies. */
     printf("execute immediate if(@wsrep_is_on, 'SET @save_wsrep_on=@@WSREP_ON, WSREP_ON=OFF', 'do 0');\n"
            "SET @save_sql_log_bin=@@SQL_LOG_BIN;\n"
diff --git a/storage/innobase/btr/btr0btr.cc b/storage/innobase/btr/btr0btr.cc
index 6d3f017be3b..b9521d7e0e5 100644
--- a/storage/innobase/btr/btr0btr.cc
+++ b/storage/innobase/btr/btr0btr.cc
@@ -760,6 +760,9 @@ btr_page_get_father_block(
 	rec_t*	rec
 		= page_rec_get_next(page_get_infimum_rec(buf_block_get_frame(
 								 block)));
+	if (UNIV_UNLIKELY(!rec)) {
+		return nullptr;
+	}
 	btr_cur_position(index, rec, block, cursor);
 	return(btr_page_get_father_node_ptr(offsets, heap, cursor, mtr));
 }
@@ -777,6 +780,9 @@ bool btr_page_get_father(dict_index_t* index, buf_block_t* block, mtr_t* mtr,
 	rec_t*		rec
 		= page_rec_get_next(page_get_infimum_rec(buf_block_get_frame(
 								 block)));
+	if (UNIV_UNLIKELY(!rec)) {
+		return false;
+	}
 	btr_cur_position(index, rec, block, cursor);
 
 	heap = mem_heap_create(100);
@@ -1240,8 +1246,6 @@ btr_write_autoinc(dict_index_t* index, ib_uint64_t autoinc, bool reset)
 static dberr_t btr_page_reorganize_low(page_cur_t *cursor, dict_index_t *index,
                                        mtr_t *mtr)
 {
-  const mtr_log_t log_mode= mtr->set_log_mode(MTR_LOG_NO_REDO);
-
   buf_block_t *const block= cursor->block;
 
   ut_ad(mtr->memo_contains_flagged(block, MTR_MEMO_PAGE_X_FIX));
@@ -1252,15 +1256,20 @@ static dberr_t btr_page_reorganize_low(page_cur_t *cursor, dict_index_t *index,
   ut_ad(index->is_dummy || block->page.id().page_no() != index->page ||
         !page_has_siblings(block->page.frame));
 
+  /* Save the cursor position. */
+  const ulint pos= page_rec_get_n_recs_before(cursor->rec);
+
+  if (UNIV_UNLIKELY(pos == ULINT_UNDEFINED))
+    return DB_CORRUPTION;
+
+  btr_search_drop_page_hash_index(block);
+
   buf_block_t *old= buf_block_alloc();
   /* Copy the old page to temporary space */
   memcpy_aligned<UNIV_PAGE_SIZE_MIN>(old->page.frame, block->page.frame,
                                      srv_page_size);
 
-  btr_search_drop_page_hash_index(block);
-
-  /* Save the cursor position. */
-  const ulint pos= page_rec_get_n_recs_before(cursor->rec);
+  const mtr_log_t log_mode= mtr->set_log_mode(MTR_LOG_NO_REDO);
 
   page_create(block, mtr, index->table->not_redundant());
   if (index->is_spatial())
@@ -1272,10 +1281,13 @@ static dberr_t btr_page_reorganize_low(page_cur_t *cursor, dict_index_t *index,
   /* Copy the records from the temporary space to the recreated page;
   do not copy the lock bits yet */
 
-  if (dberr_t err=
-      page_copy_rec_list_end_no_locks(block, old,
-                                      page_get_infimum_rec(old->page.frame),
-                                      index, mtr))
+  dberr_t err=
+    page_copy_rec_list_end_no_locks(block, old,
+                                    page_get_infimum_rec(old->page.frame),
+                                    index, mtr);
+  mtr->set_log_mode(log_mode);
+
+  if (UNIV_UNLIKELY(err != DB_SUCCESS))
     return err;
 
   /* Copy the PAGE_MAX_TRX_ID or PAGE_ROOT_AUTO_INC. */
@@ -1315,12 +1327,10 @@ static dberr_t btr_page_reorganize_low(page_cur_t *cursor, dict_index_t *index,
   }
 
   /* Restore the cursor position. */
-  if (pos)
-    cursor->rec = page_rec_get_nth(block->page.frame, pos);
-  else
+  if (!pos)
     ut_ad(cursor->rec == page_get_infimum_rec(block->page.frame));
-
-  mtr->set_log_mode(log_mode);
+  else if (!(cursor->rec= page_rec_get_nth(block->page.frame, pos)))
+    return DB_CORRUPTION;
 
   if (block->page.id().page_no() != index->page ||
       fil_page_get_type(old->page.frame) != FIL_PAGE_TYPE_INSTANT)
@@ -1543,13 +1553,16 @@ btr_page_reorganize(
     return btr_page_reorganize_low(cursor, index, mtr);
 
   ulint pos= page_rec_get_n_recs_before(cursor->rec);
+  if (UNIV_UNLIKELY(pos == ULINT_UNDEFINED))
+    return DB_CORRUPTION;
+
   dberr_t err= page_zip_reorganize(cursor->block, index, page_zip_level, mtr,
                                    true);
   if (err == DB_FAIL);
-  else if (pos)
-    cursor->rec= page_rec_get_nth(cursor->block->page.frame, pos);
-  else
+  else if (!pos)
     ut_ad(cursor->rec == page_get_infimum_rec(cursor->block->page.frame));
+  else if (!(cursor->rec= page_rec_get_nth(cursor->block->page.frame, pos)))
+    err= DB_CORRUPTION;
 
   return err;
 }
@@ -1745,7 +1758,6 @@ btr_root_raise_and_insert(
 	dberr_t*	err)	/*!< out: error code */
 {
 	dict_index_t*	index;
-	ulint		new_page_no;
 	rec_t*		rec;
 	dtuple_t*	node_ptr;
 	ulint		level;
@@ -1907,8 +1919,9 @@ btr_root_raise_and_insert(
 		*heap = mem_heap_create(1000);
 	}
 
+	const uint32_t new_page_no = new_block->page.id().page_no();
 	rec = page_rec_get_next(page_get_infimum_rec(new_block->page.frame));
-	new_page_no = new_block->page.id().page_no();
+	ut_ad(rec); /* We just created the page. */
 
 	/* Build the node pointer (= node key and page address) for the
 	child */
@@ -1961,9 +1974,20 @@ btr_root_raise_and_insert(
 		ibuf_reset_free_bits(new_block);
 	}
 
-	if (tuple != NULL) {
+	if (tuple) {
+		ut_ad(dtuple_check_typed(tuple));
 		/* Reposition the cursor to the child node */
-		page_cur_search(new_block, index, tuple, page_cursor);
+		ulint low_match = 0, up_match = 0;
+
+		if (page_cur_search_with_match(new_block, index, tuple,
+					       PAGE_CUR_LE,
+					       &up_match, &low_match,
+					       page_cursor, nullptr)) {
+			if (err) {
+				*err = DB_CORRUPTION;
+			}
+			return nullptr;
+		}
 	} else {
 		/* Set cursor to first record on child node */
 		page_cur_set_before_first(new_block, page_cursor);
@@ -2042,7 +2066,7 @@ btr_page_get_split_rec_to_right(const btr_cur_t* cursor, rec_t** split_rec)
 
 	insert_point = page_rec_get_next(insert_point);
 
-	if (page_rec_is_supremum(insert_point)) {
+	if (!insert_point || page_rec_is_supremum(insert_point)) {
 		insert_point = NULL;
 	} else {
 		insert_point = page_rec_get_next(insert_point);
@@ -2230,19 +2254,19 @@ btr_page_insert_fits(
 	end_rec) will end up on the other half page from tuple when it is
 	inserted. */
 
-	if (split_rec == NULL) {
-		rec = page_rec_get_next(page_get_infimum_rec(page));
+	if (!(end_rec = split_rec)) {
 		end_rec = page_rec_get_next(btr_cur_get_rec(cursor));
-
-	} else if (cmp_dtuple_rec(tuple, split_rec, *offsets) >= 0) {
-
-		rec = page_rec_get_next(page_get_infimum_rec(page));
-		end_rec = split_rec;
-	} else {
+	} else if (cmp_dtuple_rec(tuple, split_rec, *offsets) < 0) {
 		rec = split_rec;
 		end_rec = page_get_supremum_rec(page);
+		goto got_rec;
 	}
 
+	if (!(rec = page_rec_get_next(page_get_infimum_rec(page)))) {
+		return false;
+	}
+
+got_rec:
 	if (total_data + page_dir_calc_reserved_space(total_n_recs)
 	    <= free_space) {
 
@@ -2274,7 +2298,9 @@ btr_page_insert_fits(
 			return(true);
 		}
 
-		rec = page_rec_get_next_const(rec);
+		if (!(rec = page_rec_get_next_const(rec))) {
+			break;
+		}
 	}
 
 	return(false);
@@ -2511,8 +2537,10 @@ btr_page_tuple_smaller(
 	/* Read the first user record in the page. */
 	block = btr_cur_get_block(cursor);
 	page_cur_set_before_first(block, &pcur);
-	page_cur_move_to_next(&pcur);
-	first_rec = page_cur_get_rec(&pcur);
+	if (UNIV_UNLIKELY(!(first_rec = page_cur_move_to_next(&pcur)))) {
+		ut_ad("corrupted page" == 0);
+		return false;
+	}
 
 	*offsets = rec_get_offsets(first_rec, cursor->index, *offsets,
 				   page_is_leaf(block->page.frame)
@@ -2554,6 +2582,7 @@ btr_insert_into_right_sibling(
 					 MTR_MEMO_X_LOCK | MTR_MEMO_SX_LOCK));
 	ut_ad(mtr->memo_contains_flagged(block, MTR_MEMO_PAGE_X_FIX));
 	ut_ad(heap);
+	ut_ad(dtuple_check_typed(tuple));
 
 	if (next_page_no == FIL_NULL || !page_rec_is_supremum(
 			page_rec_get_next(btr_cur_get_rec(cursor)))) {
@@ -2581,9 +2610,13 @@ btr_insert_into_right_sibling(
 		return nullptr;
 	}
 
-	page_cur_search(
-		next_block, cursor->index, tuple, PAGE_CUR_LE,
-		&next_page_cursor);
+	ulint up_match = 0, low_match = 0;
+
+	if (page_cur_search_with_match(next_block, cursor->index, tuple,
+				       PAGE_CUR_LE, &up_match, &low_match,
+				       &next_page_cursor, nullptr)) {
+		return nullptr;
+	}
 
 	max_size = page_get_max_insert_size_after_reorganize(next_page, 1);
 
@@ -2788,6 +2821,7 @@ btr_page_split_and_insert(
 	ulint		n_uniq;
 
 	ut_ad(*err == DB_SUCCESS);
+	ut_ad(dtuple_check_typed(tuple));
 
 	if (cursor->index->is_spatial()) {
 		/* Split rtree page and update parent */
@@ -2855,9 +2889,16 @@ func_start:
 				page_get_infimum_rec(page));
 		} else {
 			split_rec = NULL;
+			goto got_split_rec;
+		}
+
+		if (UNIV_UNLIKELY(!split_rec)) {
+			*err = DB_CORRUPTION;
+			return nullptr;
 		}
 	}
 
+got_split_rec:
 	/* 2. Allocate a new page to the index */
 	const uint16_t page_level = btr_page_get_level(page);
 	new_block = btr_page_alloc(cursor->index, hint_page_no, direction,
@@ -2906,9 +2947,17 @@ func_start:
 			goto insert_empty;
 		}
 	} else if (insert_left) {
-		ut_a(n_iterations > 0);
+		if (UNIV_UNLIKELY(!n_iterations)) {
+corrupted:
+			*err = DB_CORRUPTION;
+			return nullptr;
+		}
 		first_rec = page_rec_get_next(page_get_infimum_rec(page));
+insert_move_limit:
 		move_limit = page_rec_get_next(btr_cur_get_rec(cursor));
+		if (UNIV_UNLIKELY(!first_rec || !move_limit)) {
+			goto corrupted;
+		}
 	} else {
 insert_empty:
 		ut_ad(!split_rec);
@@ -2919,7 +2968,7 @@ insert_empty:
 
 		first_rec = rec_convert_dtuple_to_rec(buf, cursor->index,
 						      tuple, n_ext);
-		move_limit = page_rec_get_next(btr_cur_get_rec(cursor));
+		goto insert_move_limit;
 	}
 
 	/* 4. Do first the modifications in the tree structure */
@@ -3086,7 +3135,14 @@ insert_empty:
 	/* 7. Reposition the cursor for insert and try insertion */
 	page_cursor = btr_cur_get_page_cur(cursor);
 
-	page_cur_search(insert_block, cursor->index, tuple, page_cursor);
+	ulint up_match = 0, low_match = 0;
+
+	if (page_cur_search_with_match(insert_block, cursor->index, tuple,
+				       PAGE_CUR_LE, &up_match, &low_match,
+				       page_cursor, nullptr)) {
+		*err = DB_CORRUPTION;
+		return nullptr;
+	}
 
 	rec = page_cur_tuple_insert(page_cursor, tuple, cursor->index,
 				    offsets, heap, n_ext, mtr);
@@ -3440,8 +3496,8 @@ btr_compress(
 	mtr_t*		mtr)	/*!< in/out: mini-transaction */
 {
 	dict_index_t*	index;
-	buf_block_t*	merge_block;
-	page_t*		merge_page = NULL;
+	buf_block_t*	merge_block = nullptr;
+	page_t*		merge_page = nullptr;
 	page_zip_des_t*	merge_page_zip;
 	ibool		is_left;
 	buf_block_t*	block;
@@ -3499,7 +3555,10 @@ btr_compress(
 
 	if (adjust) {
 		nth_rec = page_rec_get_n_recs_before(btr_cur_get_rec(cursor));
-		ut_ad(nth_rec > 0);
+		if (UNIV_UNLIKELY(!nth_rec || nth_rec == ULINT_UNDEFINED)) {
+			err = DB_CORRUPTION;
+			goto err_exit;
+		}
 	}
 
 	if (left_page_no == FIL_NULL && right_page_no == FIL_NULL) {
@@ -3648,7 +3707,12 @@ cannot_merge:
 		}
 
 		if (adjust) {
-			nth_rec += page_rec_get_n_recs_before(orig_pred);
+			ulint n = page_rec_get_n_recs_before(orig_pred);
+			if (UNIV_UNLIKELY(!n || n == ULINT_UNDEFINED)) {
+				err = DB_CORRUPTION;
+				goto err_exit;
+			}
+			nth_rec += n;
 		}
 	} else {
 		rec_t*		orig_succ;
@@ -3866,16 +3930,20 @@ cannot_merge:
 		ut_ad(leftmost_child
 		      || btr_check_node_ptr(index, merge_block, mtr));
 func_exit:
-		MONITOR_INC(MONITOR_INDEX_MERGE_SUCCESSFUL);
-
 		if (adjust) {
 			ut_ad(nth_rec > 0);
-			btr_cur_position(
-				index,
-				page_rec_get_nth(merge_block->page.frame,
-                                                 nth_rec),
-                                merge_block, cursor);
-                }
+			if (rec_t* nth
+			    = page_rec_get_nth(merge_block->page.frame,
+					       nth_rec)) {
+				btr_cur_position(index, nth,
+						 merge_block, cursor);
+			} else {
+				err = DB_CORRUPTION;
+				goto err_exit;
+			}
+		}
+
+		MONITOR_INC(MONITOR_INDEX_MERGE_SUCCESSFUL);
         } else {
 err_exit:
 		/* We play it safe and reset the free bits. */
@@ -3912,6 +3980,9 @@ btr_discard_only_page_on_level(
 	const trx_id_t max_trx_id = page_get_max_trx_id(block->page.frame);
 	const rec_t* r = page_rec_get_next(
 		page_get_infimum_rec(block->page.frame));
+	/* In the caller we checked that a valid key exists in the page,
+	because we were able to look up a parent page. */
+	ut_ad(r);
 	ut_ad(rec_is_metadata(r, *index) == index->is_instant());
 
 	while (block->page.id().page_no() != dict_index_get_page(index)) {
@@ -4083,14 +4154,17 @@ btr_discard_page(
 
 		ut_d(parent_is_different = page_rec_is_supremum(
 			page_rec_get_next(btr_cur_get_rec(&parent_cursor))));
-		if (!page_is_leaf(merge_block->page.frame)) {
-			rec_t* node_ptr = page_rec_get_next(
-				page_get_infimum_rec(merge_block->page.frame));
+		if (page_is_leaf(merge_block->page.frame)) {
+		} else if (rec_t* node_ptr =
+                           page_rec_get_next(page_get_infimum_rec(
+					   merge_block->page.frame))) {
 			ut_ad(page_rec_is_user_rec(node_ptr));
 			/* We have to mark the leftmost node pointer as the
 			predefined minimum record. */
 			btr_set_min_rec_mark<true>(node_ptr, *merge_block,
 						   mtr);
+		} else {
+			return DB_CORRUPTION;
 		}
 	} else {
 		btr_discard_only_page_on_level(index, block, mtr);
@@ -4550,16 +4624,15 @@ next_field:
 /************************************************************//**
 Checks the size and number of fields in records based on the definition of
 the index.
-@return TRUE if ok */
+@return true if ok */
 static
-ibool
+bool
 btr_index_page_validate(
 /*====================*/
 	buf_block_t*	block,	/*!< in: index page */
 	dict_index_t*	index)	/*!< in: index */
 {
 	page_cur_t	cur;
-	ibool		ret	= TRUE;
 #ifndef DBUG_OFF
 	ulint		nth	= 1;
 #endif /* !DBUG_OFF */
@@ -4576,17 +4649,13 @@ btr_index_page_validate(
 					     page_cur_get_page(&cur), 0))
 			     == 1););
 
-	page_cur_move_to_next(&cur);
-
-	for (;;) {
+	while (page_cur_move_to_next(&cur)) {
 		if (page_cur_is_after_last(&cur)) {
-
-			break;
+			return true;
 		}
 
 		if (!btr_index_rec_validate(cur.rec, index, TRUE)) {
-
-			return(FALSE);
+			break;
 		}
 
 		/* Verify that page_rec_get_nth_const() is correctly
@@ -4598,11 +4667,9 @@ btr_index_page_validate(
 						     cur.rec)));
 				ut_a(nth++ == page_rec_get_n_recs_before(
 					     cur.rec)););
-
-		page_cur_move_to_next(&cur);
 	}
 
-	return(ret);
+	return false;
 }
 
 /************************************************************//**
@@ -4720,14 +4787,16 @@ btr_validate_level(
 		ut_a(!page_zip || page_zip_validate(page_zip, page, index));
 #endif /* UNIV_ZIP_DEBUG */
 		if (page_is_leaf(page)) {
+corrupted:
 			err = DB_CORRUPTION;
 			goto invalid_page;
 		}
 
 		page_cur_set_before_first(block, &cursor);
-		page_cur_move_to_next(&cursor);
+		if (!(node_ptr = page_cur_move_to_next(&cursor))) {
+			goto corrupted;
+		}
 
-		node_ptr = page_cur_get_rec(&cursor);
 		offsets = rec_get_offsets(node_ptr, index, offsets, 0,
 					  ULINT_UNDEFINED, &heap);
 
@@ -4851,12 +4920,16 @@ func_exit:
 		}
 
 		if (!(rec = page_rec_get_prev(page_get_supremum_rec(page)))) {
+broken_links:
 			btr_validate_report1(index, level, block);
 			fputs("InnoDB: broken record links\n", stderr);
 			goto invalid_page;
 		}
-		right_rec = page_rec_get_next(page_get_infimum_rec(
-						      right_page));
+		if (!(right_rec =
+		      page_rec_get_next(page_get_infimum_rec(right_page)))) {
+			goto broken_links;
+		}
+
 		offsets = rec_get_offsets(rec, index, offsets,
 					  page_is_leaf(page)
 					  ? index->n_core_fields : 0,
@@ -4869,7 +4942,7 @@ func_exit:
 		/* For spatial index, we cannot guarantee the key ordering
 		across pages, so skip the record compare verification for
 		now. Will enhanced in special R-Tree index validation scheme */
-		if (!dict_index_is_spatial(index)
+		if (!index->is_spatial()
 		    && cmp_rec_rec(rec, right_rec,
 				   offsets, offsets2, index) >= 0) {
 
@@ -4887,19 +4960,26 @@ func_exit:
 			fputs("InnoDB: record ", stderr);
 			rec = page_rec_get_next(
 				page_get_infimum_rec(right_page));
-			rec_print(stderr, rec, index);
+			if (rec) {
+				rec_print(stderr, rec, index);
+			}
 			putc('\n', stderr);
-                        err = DB_CORRUPTION;
+			err = DB_CORRUPTION;
 		}
 	}
 
-	if (level > 0 && left_page_no == FIL_NULL
-            && !(REC_INFO_MIN_REC_FLAG & rec_get_info_bits(
-			 page_rec_get_next(page_get_infimum_rec(page)),
-			 page_is_comp(page)))) {
-		btr_validate_report1(index, level, block);
-		ib::error() << "Missing REC_INFO_MIN_REC_FLAG";
+	if (!level || left_page_no != FIL_NULL) {
+	} else if (const rec_t* first =
+		   page_rec_get_next_const(page_get_infimum_rec(page))) {
+		if (!(REC_INFO_MIN_REC_FLAG
+		      & rec_get_info_bits(first, page_is_comp(page)))) {
+			btr_validate_report1(index, level, block);
+			ib::error() << "Missing REC_INFO_MIN_REC_FLAG";
+			err = DB_CORRUPTION;
+		}
+	} else {
 		err = DB_CORRUPTION;
+		goto node_ptr_fails;
 	}
 
 	/* Similarly skip the father node check for spatial index for now,
@@ -4908,15 +4988,17 @@ func_exit:
 	in parent level and linked pages in the child level.
 	2) Search parent from root is very costly for R-tree.
 	We will add special validation mechanism for R-tree later (WL #7520) */
-	if (!dict_index_is_spatial(index)
-	    && block->page.id().page_no() != dict_index_get_page(index)) {
-
+	if (!index->is_spatial()
+	    && block->page.id().page_no() != index->page) {
 		/* Check father node pointers */
-		rec_t*	node_ptr;
+		rec_t*	node_ptr
+			= page_rec_get_next(page_get_infimum_rec(page));
+		if (!node_ptr) {
+			err = DB_CORRUPTION;
+			goto node_ptr_fails;
+		}
 
-		btr_cur_position(
-			index, page_rec_get_next(page_get_infimum_rec(page)),
-			block, &node_cur);
+		btr_cur_position(index, node_ptr, block, &node_cur);
 		offsets = btr_page_get_father_node_ptr_for_validate(
 			offsets, heap, &node_cur, &mtr);
 
@@ -4965,17 +5047,15 @@ func_exit:
 			goto node_ptr_fails;
 		}
 
-		if (!page_is_leaf(page)) {
+		if (page_is_leaf(page)) {
+		} else if (const rec_t* first_rec =
+			   page_rec_get_next(page_get_infimum_rec(page))) {
 			node_ptr_tuple = dict_index_build_node_ptr(
-				index,
-				page_rec_get_next(page_get_infimum_rec(page)),
+				index, first_rec,
 				0, heap, btr_page_get_level(page));
 
 			if (cmp_dtuple_rec(node_ptr_tuple, node_ptr,
 					   offsets)) {
-				const rec_t* first_rec = page_rec_get_next(
-					page_get_infimum_rec(page));
-
 				btr_validate_report1(index, level, block);
 
 				ib::error() << "Node ptrs differ on levels > 0";
@@ -4985,14 +5065,17 @@ func_exit:
 				fputs("InnoDB: first rec ", stderr);
 				rec_print(stderr, first_rec, index);
 				putc('\n', stderr);
-	                        err = DB_CORRUPTION;
+				err = DB_CORRUPTION;
 				goto node_ptr_fails;
 			}
+		} else {
+			err = DB_CORRUPTION;
+			goto node_ptr_fails;
 		}
 
 		if (left_page_no == FIL_NULL) {
 			if (page_has_prev(father_page)
-                            || node_ptr != page_rec_get_next(
+			    || node_ptr != page_rec_get_next(
 				     page_get_infimum_rec(father_page))) {
 				err = DB_CORRUPTION;
 				goto node_ptr_fails;
@@ -5006,11 +5089,8 @@ func_exit:
 				err = DB_CORRUPTION;
 				goto node_ptr_fails;
 			}
-		} else {
-			const rec_t*	right_node_ptr;
-
-			right_node_ptr = page_rec_get_next(node_ptr);
-
+		} else if (const rec_t* right_node_ptr
+			   = page_rec_get_next(node_ptr)) {
 			if (!lockout && rightmost_child) {
 
 				/* To obey latch order of tree blocks,
@@ -5041,11 +5121,12 @@ func_exit:
 			}
 
 			btr_cur_position(
-				index, page_rec_get_next(
-					page_get_infimum_rec(
-						buf_block_get_frame(
-							right_block))),
+				index,
+				page_get_infimum_rec(right_block->page.frame),
 				right_block, &right_node_cur);
+			if (!page_cur_move_to_next(&right_node_cur.page_cur)) {
+				goto node_pointer_corrupted;
+			}
 
 			offsets = btr_page_get_father_node_ptr_for_validate(
 					offsets, heap, &right_node_cur, &mtr);
@@ -5055,6 +5136,7 @@ func_exit:
 
 				if (btr_cur_get_rec(&right_node_cur)
 				    != right_node_ptr) {
+node_pointer_corrupted:
 					err = DB_CORRUPTION;
 					fputs("InnoDB: node pointer to"
 					      " the right page is wrong\n",
@@ -5092,6 +5174,8 @@ func_exit:
 							     block);
 				}
 			}
+		} else {
+			err = DB_CORRUPTION;
 		}
 	}
 
diff --git a/storage/innobase/btr/btr0bulk.cc b/storage/innobase/btr/btr0bulk.cc
index c56df29cbbc..c64bda46cbb 100644
--- a/storage/innobase/btr/btr0bulk.cc
+++ b/storage/innobase/btr/btr0bulk.cc
@@ -1199,6 +1199,9 @@ err_exit:
 
 		first_rec = page_rec_get_next(
 			page_get_infimum_rec(last_block->page.frame));
+		/* Because this index tree is being created by this thread,
+		we assume that it cannot be corrupted. */
+		ut_ad(first_rec);
 		ut_ad(page_rec_is_user_rec(first_rec));
 
 		/* Copy last page to root page. */
diff --git a/storage/innobase/btr/btr0cur.cc b/storage/innobase/btr/btr0cur.cc
index a21a812cdc9..d11a3229e7c 100644
--- a/storage/innobase/btr/btr0cur.cc
+++ b/storage/innobase/btr/btr0cur.cc
@@ -403,15 +403,13 @@ unreadable:
 	ut_ad(page_cur_is_before_first(&cur.page_cur));
 	ut_ad(page_is_leaf(cur.page_cur.block->page.frame));
 
-	page_cur_move_to_next(&cur.page_cur);
-
-	const rec_t* rec = cur.page_cur.rec;
+	const rec_t* rec = page_cur_move_to_next(&cur.page_cur);
 	const ulint comp = dict_table_is_comp(index->table);
-	const ulint info_bits = rec_get_info_bits(rec, comp);
+	const ulint info_bits = rec ? rec_get_info_bits(rec, comp) : 0;
 
 	if (page_rec_is_supremum(rec)
 	    || !(info_bits & REC_INFO_MIN_REC_FLAG)) {
-		if (!index->is_instant()) {
+		if (rec && !index->is_instant()) {
 			/* The FIL_PAGE_TYPE_INSTANT and PAGE_INSTANT may be
 			assigned even if instant ADD COLUMN was not
 			committed. Changes to these page header fields are not
@@ -884,6 +882,21 @@ btr_cur_latch_for_root_leaf(
 	return(RW_NO_LATCH); /* avoid compiler warnings */
 }
 
+/** @return whether the distance between two records is at most the
+specified value */
+static bool
+page_rec_distance_is_at_most(const rec_t *left, const rec_t *right, ulint val)
+{
+  do
+  {
+    if (left == right)
+      return true;
+    left= page_rec_get_next_const(left);
+  }
+  while (left && val--);
+  return false;
+}
+
 /** Detects whether the modifying record might need a modifying tree structure.
 @param[in]	index		index
 @param[in]	page		page
@@ -1914,22 +1927,28 @@ retry_page_get:
 #ifdef BTR_CUR_HASH_ADAPT
 	} else if (height == 0 && btr_search_enabled
 		   && !(tuple->info_bits & REC_INFO_MIN_REC_FLAG)
-		   && !dict_index_is_spatial(index)) {
+		   && index->is_btree()) {
 		/* The adaptive hash index is only used when searching
 		for leaf pages (height==0), but not in r-trees.
 		We only need the byte prefix comparison for the purpose
 		of updating the adaptive hash index. */
-		page_cur_search_with_match_bytes(
+		if (page_cur_search_with_match_bytes(
 			block, index, tuple, page_mode, &up_match, &up_bytes,
-			&low_match, &low_bytes, page_cursor);
+			&low_match, &low_bytes, page_cursor)) {
+			err = DB_CORRUPTION;
+			goto func_exit;
+		}
 #endif /* BTR_CUR_HASH_ADAPT */
 	} else {
 		/* Search for complete index fields. */
 		up_bytes = low_bytes = 0;
-		page_cur_search_with_match(
+		if (page_cur_search_with_match(
 			block, index, tuple, page_mode, &up_match,
 			&low_match, page_cursor,
-			need_path ? cursor->rtr_info : NULL);
+			need_path ? cursor->rtr_info : nullptr)) {
+			err = DB_CORRUPTION;
+			goto func_exit;
+		}
 	}
 
 	/* If this is the desired level, leave the loop */
@@ -2097,6 +2116,11 @@ need_opposite_intention:
 
 			ut_ad(upper_rw_latch == RW_X_LATCH);
 
+			if (UNIV_UNLIKELY(!first_rec)) {
+			corrupted:
+				err = DB_CORRUPTION;
+				goto func_exit;
+			}
 			if (node_ptr == first_rec
 			    || page_rec_is_last(node_ptr, page)) {
 				detected_same_key_root = true;
@@ -2131,8 +2155,7 @@ need_opposite_intention:
 						detected_same_key_root = true;
 					}
 				} else {
-					err = DB_CORRUPTION;
-					goto func_exit;
+					goto corrupted;
 				}
 			}
 		}
@@ -2241,11 +2264,14 @@ need_opposite_intention:
 					? cursor->rtr_info : NULL;
 
 				for (ulint i = 0; i < n_blocks; i++) {
-					page_cur_search_with_match(
+					if (page_cur_search_with_match(
 						tree_blocks[i], index, tuple,
 						page_mode, &up_match,
 						&low_match, page_cursor,
-						rtr_info);
+						rtr_info)) {
+						err = DB_CORRUPTION;
+						goto func_exit;
+					}
 				}
 
 				goto search_loop;
@@ -2678,13 +2704,11 @@ btr_cur_open_at_index_side(
 
 		ut_ad(height > 0);
 
-		if (from_left) {
-			page_cur_move_to_next(page_cursor);
-		} else {
-			if (!page_cur_move_to_prev(page_cursor)) {
-				err = DB_CORRUPTION;
-				goto exit_loop;
-			}
+		if (from_left
+		    ? !page_cur_move_to_next(page_cursor)
+		    : !page_cur_move_to_prev(page_cursor)) {
+			err = DB_CORRUPTION;
+			goto exit_loop;
 		}
 
 		height--;
@@ -3117,7 +3141,7 @@ btr_cur_insert_if_possible(
 
 /*************************************************************//**
 For an insert, checks the locks and does the undo logging if desired.
-@return DB_SUCCESS, DB_WAIT_LOCK, DB_FAIL, or error number */
+@return DB_SUCCESS, DB_LOCK_WAIT, DB_FAIL, or error number */
 UNIV_INLINE MY_ATTRIBUTE((warn_unused_result, nonnull(2,3,5,6)))
 dberr_t
 btr_cur_ins_lock_and_undo(
@@ -3268,7 +3292,7 @@ It is assumed that mtr holds an x-latch on the page. The operation does
 not succeed if there is too little space on the page. If there is just
 one record on the page, the insert will always succeed; this is to
 prevent trying to split a page with just one record.
-@return DB_SUCCESS, DB_WAIT_LOCK, DB_FAIL, or error number */
+@return DB_SUCCESS, DB_LOCK_WAIT, DB_FAIL, or error number */
 dberr_t
 btr_cur_optimistic_insert(
 /*======================*/
@@ -3736,7 +3760,7 @@ func_exit:
 
 /*************************************************************//**
 For an update, checks the locks and does the undo logging.
-@return DB_SUCCESS, DB_WAIT_LOCK, or error number */
+@return DB_SUCCESS, DB_LOCK_WAIT, or error number */
 UNIV_INLINE MY_ATTRIBUTE((warn_unused_result))
 dberr_t
 btr_cur_upd_lock_and_undo(
@@ -4682,8 +4706,7 @@ any_extern:
 	rec = btr_cur_insert_if_possible(cursor, new_entry, offsets, heap,
 					 0/*n_ext*/, mtr);
 	if (UNIV_UNLIKELY(!rec)) {
-		err = DB_CORRUPTION;
-		goto func_exit;
+		goto corrupted;
 	}
 
 	if (UNIV_UNLIKELY(update->is_metadata())) {
@@ -4701,8 +4724,11 @@ any_extern:
 						   block->page.id());
 	}
 
-	page_cur_move_to_next(page_cursor);
 	ut_ad(err == DB_SUCCESS);
+	if (!page_cur_move_to_next(page_cursor)) {
+corrupted:
+		err = DB_CORRUPTION;
+	}
 
 func_exit:
 	if (!(flags & BTR_KEEP_IBUF_BITMAP)
@@ -5454,6 +5480,10 @@ btr_cur_optimistic_delete(
 		dict_index_t* index = cursor->index;
 		const rec_t* first_rec = page_rec_get_next_const(
 			page_get_infimum_rec(block->page.frame));
+		if (UNIV_UNLIKELY(!first_rec)) {
+			err = DB_CORRUPTION;
+			goto func_exit;
+		}
 		ut_ad(!index->is_instant()
 		      || rec_is_metadata(first_rec, *index));
 		const bool is_metadata = rec_is_metadata(rec, *index);
@@ -5683,6 +5713,10 @@ btr_cur_pessimistic_delete(
 
 			const rec_t* first_rec = page_rec_get_next_const(
 				page_get_infimum_rec(page));
+			if (UNIV_UNLIKELY(!first_rec)) {
+				*err = DB_CORRUPTION;
+				goto err_exit;
+			}
 			ut_ad(!index->is_instant()
 			      || rec_is_metadata(first_rec, *index));
 			if (is_metadata || !index->is_instant()
@@ -5729,7 +5763,11 @@ discard_page:
 			goto return_after_reservations;
 		}
 
-		next_rec = page_rec_get_next(rec);
+		if (UNIV_UNLIKELY(!(next_rec = page_rec_get_next(rec)))) {
+			ut_ad(!ret);
+			*err = DB_CORRUPTION;
+			goto err_exit;
+		}
 
 		if (!page_has_prev(page)) {
 			/* If we delete the leftmost node pointer on a
@@ -5996,9 +6034,10 @@ public:
     if (dtuple_get_n_fields(&m_tuple) > 0)
     {
       m_up_bytes= m_low_bytes= 0;
-      page_cur_search_with_match(m_block, index(), &m_tuple, m_page_mode,
-                                 &m_up_match, &m_low_match, &m_page_cur,
-                                 nullptr);
+      if (page_cur_search_with_match(m_block, index(), &m_tuple, m_page_mode,
+                                     &m_up_match, &m_low_match, &m_page_cur,
+                                     nullptr))
+        return false;
       m_nth_rec= page_rec_get_n_recs_before(page_cur_get_rec(&m_page_cur));
     }
     else if (left)
@@ -6006,7 +6045,8 @@ public:
       page_cur_set_before_first(m_block, &m_page_cur);
       if (level)
       {
-        page_cur_move_to_next(&m_page_cur);
+        if (!page_cur_move_to_next(&m_page_cur))
+          return false;
         m_nth_rec= 1;
       }
       else
diff --git a/storage/innobase/btr/btr0defragment.cc b/storage/innobase/btr/btr0defragment.cc
index 5a278fbd9a2..0c1959bb554 100644
--- a/storage/innobase/btr/btr0defragment.cc
+++ b/storage/innobase/btr/btr0defragment.cc
@@ -259,9 +259,10 @@ btr_defragment_calc_n_recs_for_size(
 
 	const ulint n_core = page_is_leaf(page) ? index->n_core_fields : 0;
 	page_cur_set_before_first(block, &cur);
-	page_cur_move_to_next(&cur);
-	while (page_cur_get_rec(&cur) != page_get_supremum_rec(page)) {
-		rec_t* cur_rec = page_cur_get_rec(&cur);
+	while (rec_t* cur_rec = page_cur_move_to_next(&cur)) {
+		if (page_rec_is_supremum(cur_rec)) {
+			break;
+		}
 		offsets = rec_get_offsets(cur_rec, index, offsets, n_core,
 					  ULINT_UNDEFINED, &heap);
 		ulint rec_size = rec_offs_size(offsets);
@@ -271,7 +272,6 @@ btr_defragment_calc_n_recs_for_size(
 			break;
 		}
 		n_recs ++;
-		page_cur_move_to_next(&cur);
 	}
 	*n_recs_size = size;
 	if (UNIV_LIKELY_NULL(heap)) {
@@ -356,8 +356,9 @@ btr_defragment_merge_pages(
 	target_n_recs = n_recs_to_move;
 	dberr_t err;
 	while (n_recs_to_move > 0) {
-		rec = page_rec_get_nth(from_page,
-					n_recs_to_move + 1);
+		if (!(rec = page_rec_get_nth(from_page, n_recs_to_move + 1))) {
+			return nullptr;
+		}
 		orig_pred = page_copy_rec_list_start(
 			to_block, from_block, rec, index, mtr, &err);
 		if (orig_pred)
@@ -439,14 +440,17 @@ btr_defragment_merge_pages(
 			}
 			rec = page_rec_get_next(
 				page_get_infimum_rec(from_page));
+			if (!rec) {
+				return nullptr;
+			}
 			node_ptr = dict_index_build_node_ptr(
 				index, rec, page_get_page_no(from_page),
 				heap, level);
 			if (btr_insert_on_non_leaf_level(0, index, level+1,
-                                                         node_ptr, mtr)
-                            != DB_SUCCESS) {
+							 node_ptr, mtr)
+			    != DB_SUCCESS) {
 				return nullptr;
-                        }
+			}
 		}
 		to_block = from_block;
 	}
diff --git a/storage/innobase/btr/btr0pcur.cc b/storage/innobase/btr/btr0pcur.cc
index c45ce3be89e..eb7f452749f 100644
--- a/storage/innobase/btr/btr0pcur.cc
+++ b/storage/innobase/btr/btr0pcur.cc
@@ -176,10 +176,16 @@ before_first:
 	} else if (page_rec_is_infimum_low(offs)) {
 		rec = page_rec_get_next(rec);
 
+		if (UNIV_UNLIKELY(!rec)) {
+			ut_ad("corrupted page" == 0);
+			goto before_first;
+		}
+
 		if (rec_is_metadata(rec, *index)) {
 			ut_ad(!page_has_prev(block->page.frame));
 			rec = page_rec_get_next(rec);
-			if (page_rec_is_supremum(rec)) {
+			ut_ad(rec);
+			if (!rec || page_rec_is_supremum(rec)) {
 				goto before_first;
 			}
 		}
diff --git a/storage/innobase/btr/btr0sea.cc b/storage/innobase/btr/btr0sea.cc
index 53cc464f18f..137a8b10d9f 100644
--- a/storage/innobase/btr/btr0sea.cc
+++ b/storage/innobase/btr/btr0sea.cc
@@ -788,8 +788,20 @@ btr_search_check_guess(
 
 	rec = btr_cur_get_rec(cursor);
 
-	ut_ad(page_rec_is_user_rec(rec));
-	ut_ad(page_rec_is_leaf(rec));
+	if (UNIV_UNLIKELY(!page_rec_is_user_rec(rec)
+			  || !page_rec_is_leaf(rec))) {
+		ut_ad("corrupted index" == 0);
+		return false;
+	} else if (cursor->index->table->not_redundant()) {
+		switch (rec_get_status(rec)) {
+		case REC_STATUS_INSTANT:
+		case REC_STATUS_ORDINARY:
+			break;
+		default:
+			ut_ad("corrupted index" == 0);
+			return false;
+		}
+	}
 
 	match = 0;
 
@@ -847,6 +859,17 @@ btr_search_check_guess(
 			goto exit_func;
 		}
 
+		if (cursor->index->table->not_redundant()) {
+			switch (rec_get_status(prev_rec)) {
+			case REC_STATUS_INSTANT:
+			case REC_STATUS_ORDINARY:
+				break;
+			default:
+				ut_ad("corrupted index" == 0);
+				goto exit_func;
+			}
+		}
+
 		offsets = rec_get_offsets(prev_rec, cursor->index, offsets,
 					  cursor->index->n_core_fields,
 					  n_unique, &heap);
@@ -862,15 +885,31 @@ btr_search_check_guess(
 
 		const rec_t* next_rec = page_rec_get_next(rec);
 
+		if (UNIV_UNLIKELY(!next_rec)) {
+			ut_ad("corrupted index" == 0);
+			goto exit_func;
+		}
+
 		if (page_rec_is_supremum(next_rec)) {
 			if (!page_has_next(page_align(next_rec))) {
 				cursor->up_match = 0;
-				success = TRUE;
+				success = true;
 			}
 
 			goto exit_func;
 		}
 
+		if (cursor->index->table->not_redundant()) {
+			switch (rec_get_status(next_rec)) {
+			case REC_STATUS_INSTANT:
+			case REC_STATUS_ORDINARY:
+				break;
+			default:
+				ut_ad("corrupted index" == 0);
+				goto exit_func;
+			}
+		}
+
 		offsets = rec_get_offsets(next_rec, cursor->index, offsets,
 					  cursor->index->n_core_fields,
 					  n_unique, &heap);
@@ -1245,14 +1284,7 @@ void btr_search_drop_page_hash_index(buf_block_t* block)
 {
 	ulint			n_fields;
 	ulint			n_bytes;
-	const page_t*		page;
 	const rec_t*		rec;
-	ulint			fold;
-	ulint			prev_fold;
-	ulint			n_cached;
-	ulint			n_recs;
-	ulint*			folds;
-	ulint			i;
 	mem_heap_t*		heap;
 	rec_offs*		offsets;
 
@@ -1323,33 +1355,50 @@ retry:
 
 	ut_a(n_fields > 0 || n_bytes > 0);
 
-	page = block->page.frame;
-	n_recs = page_get_n_recs(page);
+	const page_t* const page = block->page.frame;
+	ulint n_recs = page_get_n_recs(page);
+	if (!n_recs) {
+		ut_ad("corrupted adaptive hash index" == 0);
+		return;
+	}
 
 	/* Calculate and cache fold values into an array for fast deletion
 	from the hash index */
 
-	folds = (ulint*) ut_malloc_nokey(n_recs * sizeof(ulint));
-
-	n_cached = 0;
-
 	rec = page_get_infimum_rec(page);
 	rec = page_rec_get_next_low(rec, page_is_comp(page));
-	if (rec_is_metadata(rec, *index)) {
+
+	ulint* folds;
+	ulint n_cached = 0;
+	ulint prev_fold = 0;
+
+	if (rec && rec_is_metadata(rec, *index)) {
 		rec = page_rec_get_next_low(rec, page_is_comp(page));
+		if (!--n_recs) {
+			/* The page only contains the hidden metadata record
+			for instant ALTER TABLE that the adaptive hash index
+			never points to. */
+			folds = nullptr;
+			goto all_deleted;
+		}
 	}
 
-	prev_fold = 0;
-
-	heap = NULL;
-	offsets = NULL;
+	folds = (ulint*) ut_malloc_nokey(n_recs * sizeof(ulint));
+	heap = nullptr;
+	offsets = nullptr;
 
-	while (!page_rec_is_supremum(rec)) {
+	while (rec) {
+		if (n_cached >= n_recs) {
+			ut_ad(page_rec_is_supremum(rec));
+			break;
+		}
+		ut_ad(page_rec_is_user_rec(rec));
 		offsets = rec_get_offsets(
 			rec, index, offsets, index->n_core_fields,
 			btr_search_get_n_fields(n_fields, n_bytes),
 			&heap);
-		fold = rec_fold(rec, offsets, n_fields, n_bytes, index_id);
+		const ulint fold = rec_fold(rec, offsets, n_fields, n_bytes,
+					    index_id);
 
 		if (fold == prev_fold && prev_fold != 0) {
 
@@ -1358,11 +1407,13 @@ retry:
 
 		/* Remove all hash nodes pointing to this page from the
 		hash chain */
+		folds[n_cached++] = fold;
 
-		folds[n_cached] = fold;
-		n_cached++;
 next_rec:
 		rec = page_rec_get_next_low(rec, page_rec_is_comp(rec));
+		if (!rec || page_rec_is_supremum(rec)) {
+			break;
+		}
 		prev_fold = fold;
 	}
 
@@ -1370,6 +1421,7 @@ next_rec:
 		mem_heap_free(heap);
 	}
 
+all_deleted:
 	if (!is_freed) {
 		part->latch.wr_lock(SRW_LOCK_CALL);
 
@@ -1394,7 +1446,7 @@ next_rec:
 		goto retry;
 	}
 
-	for (i = 0; i < n_cached; i++) {
+	for (ulint i = 0; i < n_cached; i++) {
 		ha_remove_all_nodes_to_page(&part->table, part->heap,
 					    folds[i], page);
 	}
@@ -1408,7 +1460,7 @@ next_rec:
 		}
 	}
 
-	block->index = NULL;
+	block->index = nullptr;
 
 	MONITOR_INC(MONITOR_ADAPTIVE_HASH_PAGE_REMOVED);
 	MONITOR_INC_VALUE(MONITOR_ADAPTIVE_HASH_ROW_REMOVED, n_cached);
@@ -1472,7 +1524,6 @@ btr_search_build_page_hash_index(
 	bool		left_side)
 {
 	const rec_t*	rec;
-	const rec_t*	next_rec;
 	ulint		fold;
 	ulint		next_fold;
 	ulint		n_cached;
@@ -1538,10 +1589,11 @@ btr_search_build_page_hash_index(
 	}
 
 	rec = page_rec_get_next_const(page_get_infimum_rec(page));
+        if (!rec) return;
 
 	if (rec_is_metadata(rec, *index)) {
 		rec = page_rec_get_next_const(rec);
-		if (!--n_recs) return;
+		if (!rec || !--n_recs) return;
 	}
 
 	/* Calculate and cache fold values and corresponding records into
@@ -1571,9 +1623,7 @@ btr_search_build_page_hash_index(
 		n_cached++;
 	}
 
-	for (;;) {
-		next_rec = page_rec_get_next_const(rec);
-
+	while (const rec_t* next_rec = page_rec_get_next_const(rec)) {
 		if (page_rec_is_supremum(next_rec)) {
 
 			if (!left_side) {
@@ -1906,12 +1956,15 @@ void btr_search_update_hash_node_on_insert(btr_cur_t *cursor,
 	    && (cursor->n_fields == block->curr_n_fields)
 	    && (cursor->n_bytes == block->curr_n_bytes)
 	    && !block->curr_left_side) {
-
-		if (ha_search_and_update_if_found(
-			&btr_search_sys.get_part(*cursor->index)->table,
-			cursor->fold, rec, block,
-			page_rec_get_next(rec))) {
-			MONITOR_INC(MONITOR_ADAPTIVE_HASH_ROW_UPDATED);
+		if (const rec_t *new_rec = page_rec_get_next_const(rec)) {
+			if (ha_search_and_update_if_found(
+				&btr_search_sys.get_part(*cursor->index)
+				->table,
+				cursor->fold, rec, block, new_rec)) {
+				MONITOR_INC(MONITOR_ADAPTIVE_HASH_ROW_UPDATED);
+			}
+		} else {
+			ut_ad("corrupted page" == 0);
 		}
 
 func_exit:
@@ -1976,6 +2029,7 @@ void btr_search_update_hash_on_insert(btr_cur_t *cursor,
 
 	if (index != cursor->index) {
 		ut_ad(index->id == cursor->index->id);
+drop:
 		btr_search_drop_page_hash_index(block);
 		return;
 	}
@@ -1988,7 +2042,9 @@ void btr_search_update_hash_on_insert(btr_cur_t *cursor,
 	const bool left_side = block->curr_left_side;
 
 	ins_rec = page_rec_get_next_const(rec);
+	if (UNIV_UNLIKELY(!ins_rec)) goto drop;
 	next_rec = page_rec_get_next_const(ins_rec);
+	if (UNIV_UNLIKELY(!next_rec)) goto drop;
 
 	offsets = rec_get_offsets(ins_rec, index, offsets,
 				  index->n_core_fields,
diff --git a/storage/innobase/dict/dict0stats.cc b/storage/innobase/dict/dict0stats.cc
index 31d2a73466e..c32c14748a4 100644
--- a/storage/innobase/dict/dict0stats.cc
+++ b/storage/innobase/dict/dict0stats.cc
@@ -1239,7 +1239,7 @@ btr_estimate_number_of_different_key_vals(dict_index_t* index,
 		const ulint n_core = page_is_leaf(page)
 			? index->n_core_fields : 0;
 
-		if (!page_rec_is_supremum(rec)) {
+		if (rec && !page_rec_is_supremum(rec)) {
 			not_empty_flag = 1;
 			offsets_rec = rec_get_offsets(rec, index, offsets_rec,
 						      n_core,
@@ -1254,7 +1254,7 @@ btr_estimate_number_of_different_key_vals(dict_index_t* index,
 		while (!page_rec_is_supremum(rec)) {
 			ulint	matched_fields;
 			rec_t*	next_rec = page_rec_get_next(rec);
-			if (page_rec_is_supremum(next_rec)) {
+			if (!next_rec || page_rec_is_supremum(next_rec)) {
 				total_external_size +=
 					btr_rec_get_externally_stored_len(
 						rec, offsets_rec);
@@ -1630,12 +1630,11 @@ dict_stats_analyze_index_level(
 
 	if (btr_pcur_open_at_index_side(
 		    true, index, BTR_SEARCH_TREE_ALREADY_S_LATCHED,
-		    &pcur, true, level, mtr) != DB_SUCCESS) {
+		    &pcur, true, level, mtr) != DB_SUCCESS
+	    || !btr_pcur_move_to_next_on_page(&pcur)) {
 		goto func_exit;
 	}
 
-	btr_pcur_move_to_next_on_page(&pcur);
-
 	page = btr_pcur_get_page(&pcur);
 
 	/* The page must not be empty, except when
@@ -1877,6 +1876,36 @@ func_exit:
 	mem_heap_free(heap);
 }
 
+
+/************************************************************//**
+Gets the pointer to the next non delete-marked record on the page.
+If all subsequent records are delete-marked, then this function
+will return the supremum record.
+@return pointer to next non delete-marked record or pointer to supremum */
+static
+const rec_t*
+page_rec_get_next_non_del_marked(
+/*=============================*/
+	const rec_t*	rec)	/*!< in: pointer to record */
+{
+  const page_t *const page= page_align(rec);
+
+  if (page_is_comp(page))
+  {
+    for (rec= page_rec_get_next_low(rec, TRUE);
+         rec && rec_get_deleted_flag(rec, TRUE);
+         rec= page_rec_get_next_low(rec, TRUE));
+    return rec ? rec : page + PAGE_NEW_SUPREMUM;
+  }
+  else
+  {
+    for (rec= page_rec_get_next_low(rec, FALSE);
+         rec && rec_get_deleted_flag(rec, FALSE);
+         rec= page_rec_get_next_low(rec, FALSE));
+    return rec ? rec : page + PAGE_OLD_SUPREMUM;
+  }
+}
+
 /** Scan a page, reading records from left to right and counting the number
 of distinct records (looking only at the first n_prefix
 columns) and the number of external pages pointed by records from this page.
@@ -1934,7 +1963,7 @@ dict_stats_scan_page(
 
 	rec = get_next(page_get_infimum_rec(page));
 
-	if (page_rec_is_supremum(rec)) {
+	if (!rec || page_rec_is_supremum(rec)) {
 		/* the page is empty or contains only delete-marked records */
 		*n_diff = 0;
 		*out_rec = NULL;
@@ -1953,7 +1982,7 @@ dict_stats_scan_page(
 
 	*n_diff = 1;
 
-	while (!page_rec_is_supremum(next_rec)) {
+	while (next_rec && !page_rec_is_supremum(next_rec)) {
 
 		ulint	matched_fields;
 
@@ -2242,12 +2271,11 @@ dict_stats_analyze_index_for_n_prefix(
 	if (btr_pcur_open_at_index_side(true, index,
 					BTR_SEARCH_TREE_ALREADY_S_LATCHED,
 					&pcur, true, n_diff_data->level, mtr)
-	    != DB_SUCCESS) {
+	    != DB_SUCCESS
+	    || !btr_pcur_move_to_next_on_page(&pcur)) {
 		return;
 	}
 
-	btr_pcur_move_to_next_on_page(&pcur);
-
 	page = btr_pcur_get_page(&pcur);
 
 	const rec_t*	first_rec = btr_pcur_get_rec(&pcur);
diff --git a/storage/innobase/gis/gis0rtree.cc b/storage/innobase/gis/gis0rtree.cc
index 3dd7fade872..59ad44ec093 100644
--- a/storage/innobase/gis/gis0rtree.cc
+++ b/storage/innobase/gis/gis0rtree.cc
@@ -405,21 +405,24 @@ update_mbr:
 		}
 
 		/* Insert the new rec. */
-		page_cur_search_with_match(block, index, node_ptr,
-					   PAGE_CUR_LE , &up_match, &low_match,
-					   btr_cur_get_page_cur(cursor), NULL);
+		if (page_cur_search_with_match(block, index, node_ptr,
+					       PAGE_CUR_LE,
+					       &up_match, &low_match,
+					       btr_cur_get_page_cur(cursor),
+					       NULL)) {
+			goto err_exit;
+		}
 
 		err = btr_cur_optimistic_insert(flags, cursor, &insert_offsets,
 						&heap, node_ptr, &insert_rec,
 						&dummy_big_rec, 0, NULL, mtr);
 
-		if (!ins_suc && err == DB_SUCCESS) {
-			ins_suc = true;
-		}
-
 		/* If optimistic insert fail, try reorganize the page
 		and insert again. */
-		if (err != DB_SUCCESS && ins_suc) {
+		if (err == DB_SUCCESS) {
+			ins_suc = true;
+		} else if (ins_suc) {
+			ut_ad(err == DB_FAIL);
 			err = btr_page_reorganize(btr_cur_get_page_cur(cursor),
 						  index, mtr);
 			if (err == DB_SUCCESS) {
@@ -431,6 +434,7 @@ update_mbr:
 
 			/* Will do pessimistic insert */
 			if (err != DB_SUCCESS) {
+				ut_ad(err == DB_FAIL);
 				ins_suc = false;
 			}
 		}
@@ -462,10 +466,14 @@ update_mbr:
 			cur2_pno = btr_node_ptr_get_child_page_no(cur2_rec, offsets2);
 			if ((del_page_no != cur2_pno)
 			    || (cur2_rec == insert_rec)) {
-				cur2_rec = page_rec_get_next(
-					page_get_infimum_rec(page));
+				cur2_rec = page_get_infimum_rec(page);
+
+				while ((cur2_rec
+					= page_rec_get_next(cur2_rec))) {
+					if (page_rec_is_supremum(cur2_rec)) {
+						break;
+					}
 
-				while (!page_rec_is_supremum(cur2_rec)) {
 					offsets2 = rec_get_offsets(cur2_rec, index,
 								   NULL,
 								   n_core,
@@ -480,10 +488,7 @@ update_mbr:
 							break;
 						}
 					}
-					cur2_rec = page_rec_get_next(cur2_rec);
 				}
-
-				ut_ad(!page_rec_is_supremum(cur2_rec));
 			}
 
 			rec_info = rec_get_info_bits(cur2_rec,
@@ -529,7 +534,7 @@ update_mbr:
 	      || (REC_INFO_MIN_REC_FLAG & rec_get_info_bits(
 			  page_rec_get_next(page_get_infimum_rec(page)),
 			  page_is_comp(page))));
-
+err_exit:
 	mem_heap_free(heap);
 }
 
@@ -556,11 +561,10 @@ rtr_adjust_upper_level(
 	rec_offs*	offsets;
 	mem_heap_t*	heap;
 	ulint		level;
-	dtuple_t*	node_ptr_upper;
+	dtuple_t*	node_ptr_upper = nullptr;
 	page_cur_t*	page_cursor;
 	lock_prdt_t	prdt;
 	lock_prdt_t	new_prdt;
-	dberr_t		err;
 	big_rec_t*	dummy_big_rec;
 	rec_t*		rec;
 
@@ -597,29 +601,32 @@ rtr_adjust_upper_level(
 		}
 	}
 
-	/* Insert the node for the new page. */
-	node_ptr_upper = rtr_index_build_node_ptr(
-		index, new_mbr,
-		page_rec_get_next(page_get_infimum_rec(new_block->page.frame)),
-		new_page_no, heap);
-
-	ulint	up_match = 0;
-	ulint	low_match = 0;
-
-	buf_block_t*	father_block = btr_cur_get_block(&cursor);
-
-	page_cur_search_with_match(
-		father_block, index, node_ptr_upper,
-		PAGE_CUR_LE , &up_match, &low_match,
-		btr_cur_get_page_cur(&cursor), NULL);
+	dberr_t err;
 
-	err = btr_cur_optimistic_insert(
-		flags
-		| BTR_NO_LOCKING_FLAG
-		| BTR_KEEP_SYS_FLAG
-		| BTR_NO_UNDO_LOG_FLAG,
-		&cursor, &offsets, &heap,
-		node_ptr_upper, &rec, &dummy_big_rec, 0, NULL, mtr);
+	if (const rec_t* first = page_rec_get_next_const(
+		    page_get_infimum_rec(new_block->page.frame))) {
+		/* Insert the node for the new page. */
+		node_ptr_upper = rtr_index_build_node_ptr(
+			index, new_mbr, first, new_page_no, heap);
+		ulint	up_match = 0, low_match = 0;
+		err = page_cur_search_with_match(btr_cur_get_block(&cursor),
+						 index, node_ptr_upper,
+						 PAGE_CUR_LE,
+						 &up_match, &low_match,
+						 btr_cur_get_page_cur(&cursor),
+						 NULL)
+			? DB_CORRUPTION
+			: btr_cur_optimistic_insert(flags
+						    | BTR_NO_LOCKING_FLAG
+						    | BTR_KEEP_SYS_FLAG
+						    | BTR_NO_UNDO_LOG_FLAG,
+						    &cursor, &offsets, &heap,
+						    node_ptr_upper, &rec,
+						    &dummy_big_rec, 0, NULL,
+						    mtr);
+	} else {
+		err = DB_CORRUPTION;
+	}
 
 	if (err == DB_FAIL) {
 		cursor.rtr_info = sea_cur->rtr_info;
@@ -638,23 +645,27 @@ rtr_adjust_upper_level(
 						 node_ptr_upper, &rec,
 						 &dummy_big_rec, 0, NULL, mtr);
 		cursor.rtr_info = NULL;
-		ut_a(err == DB_SUCCESS);
-
 		mem_heap_free(new_heap);
 	}
 
-	prdt.data = static_cast<void*>(mbr);
-	prdt.op = 0;
-	new_prdt.data = static_cast<void*>(new_mbr);
-	new_prdt.op = 0;
+	if (err == DB_SUCCESS) {
+		prdt.data = static_cast<void*>(mbr);
+		prdt.op = 0;
+		new_prdt.data = static_cast<void*>(new_mbr);
+		new_prdt.op = 0;
 
-	lock_prdt_update_parent(block, new_block, &prdt, &new_prdt,
-				page_cursor->block->page.id());
+		lock_prdt_update_parent(block, new_block, &prdt, &new_prdt,
+					page_cursor->block->page.id());
+	}
 
 	mem_heap_free(heap);
 
 	ut_ad(block->zip_size() == index->table->space->zip_size());
 
+	if (err != DB_SUCCESS) {
+		return err;
+	}
+
 	const uint32_t next_page_no = btr_page_get_next(block->page.frame);
 
 	if (next_page_no == FIL_NULL) {
@@ -765,13 +776,15 @@ rtr_split_page_move_rec_list(
 				&new_page_cursor,
 				index, cur_split_node->key, offsets, mtr);
 
-			ut_a(rec);
+			if (UNIV_UNLIKELY
+			    (!rec
+			     || !page_cur_move_to_next(&new_page_cursor))) {
+				return DB_CORRUPTION;
+			}
 
 			lock_rec_restore_from_page_infimum(
 				*new_block, rec, block->page.id());
 
-			page_cur_move_to_next(&new_page_cursor);
-
 			rec_move[moved].new_rec = rec;
 			rec_move[moved].old_rec = cur_split_node->key;
 			rec_move[moved].moved = false;
@@ -913,6 +926,11 @@ func_start:
 	if (!page_has_prev(page) && !page_is_leaf(page)) {
 		first_rec = page_rec_get_next(
 			page_get_infimum_rec(buf_block_get_frame(block)));
+		if (UNIV_UNLIKELY(!first_rec)) {
+corrupted:
+			*err = DB_CORRUPTION;
+			return nullptr;
+		}
 	}
 
 	/* Initial split nodes array. */
@@ -1098,9 +1116,13 @@ func_start:
 
 	/* Reposition the cursor for insert and try insertion */
 	page_cursor = btr_cur_get_page_cur(cursor);
+	ulint up_match = 0, low_match = 0;
 
-	page_cur_search(insert_block, cursor->index, tuple,
-			PAGE_CUR_LE, page_cursor);
+	if (page_cur_search_with_match(insert_block, cursor->index, tuple,
+				       PAGE_CUR_LE, &up_match, &low_match,
+				       page_cursor, nullptr)) {
+		goto corrupted;
+	}
 
 	/* It's possible that the new record is too big to be inserted into
 	the page, and it'll need the second round split in this case.
@@ -1154,7 +1176,7 @@ after_insert:
 	the first ssn value from it after restart server. */
 
 	root_block = btr_root_block_get(cursor->index, RW_SX_LATCH, mtr, err);
-        if (UNIV_UNLIKELY(!root_block)) {
+	if (UNIV_UNLIKELY(!root_block)) {
 		return nullptr;
 	}
 
@@ -1180,6 +1202,9 @@ after_insert:
 
 		rec_t* i_rec = page_rec_get_next(page_get_infimum_rec(
 			buf_block_get_frame(block)));
+		if (UNIV_UNLIKELY(!i_rec)) {
+			goto corrupted;
+		}
 		btr_cur_position(cursor->index, i_rec, block, cursor);
 
 		goto func_start;
@@ -1299,8 +1324,8 @@ rtr_page_copy_rec_list_end_no_locks(
 
 	page_cur_position(rec, block, &cur1);
 
-	if (page_cur_is_before_first(&cur1)) {
-		page_cur_move_to_next(&cur1);
+	if (page_cur_is_before_first(&cur1) && !page_cur_move_to_next(&cur1)) {
+		return DB_CORRUPTION;
 	}
 
 	ut_a(page_is_comp(new_page) == page_rec_is_comp(rec));
@@ -1309,6 +1334,9 @@ rtr_page_copy_rec_list_end_no_locks(
 
 	cur_rec = page_rec_get_next(
 		page_get_infimum_rec(buf_block_get_frame(new_block)));
+	if (UNIV_UNLIKELY(!cur_rec)) {
+		return DB_CORRUPTION;
+	}
 	page_cur_position(cur_rec, new_block, &page_cur);
 
 	/* Copy records from the original page to the new page */
@@ -1318,6 +1346,9 @@ rtr_page_copy_rec_list_end_no_locks(
 
 		if (page_rec_is_infimum(cur_rec)) {
 			cur_rec = page_rec_get_next(cur_rec);
+			if (UNIV_UNLIKELY(!cur_rec)) {
+				return DB_CORRUPTION;
+			}
 		}
 
 		offsets1 = rec_get_offsets(cur1_rec, index, offsets1, n_core,
@@ -1336,8 +1367,7 @@ rtr_page_copy_rec_list_end_no_locks(
 				goto move_to_prev;
 			} else if (cmp > 0) {
 				/* Skip small recs. */
-				page_cur_move_to_next(&page_cur);
-				cur_rec = page_cur_get_rec(&page_cur);
+				cur_rec = page_cur_move_to_next(&page_cur);
 			} else if (n_core) {
 				if (rec_get_deleted_flag(cur1_rec,
 					dict_table_is_comp(index->table))) {
@@ -1380,7 +1410,9 @@ move_to_prev:
 		rec_move[moved].moved = false;
 		moved++;
 next:
-		page_cur_move_to_next(&cur1);
+		if (UNIV_UNLIKELY(!page_cur_move_to_next(&cur1))) {
+			return DB_CORRUPTION;
+		}
 	}
 
 	*num_moved = moved;
@@ -1419,10 +1451,15 @@ rtr_page_copy_rec_list_start_no_locks(
 	rec_offs_init(offsets_2);
 
 	page_cur_set_before_first(block, &cur1);
-	page_cur_move_to_next(&cur1);
+	if (UNIV_UNLIKELY(!page_cur_move_to_next(&cur1))) {
+		return DB_CORRUPTION;
+	}
 
 	cur_rec = page_rec_get_next(
 		page_get_infimum_rec(buf_block_get_frame(new_block)));
+	if (UNIV_UNLIKELY(!cur_rec)) {
+		return DB_CORRUPTION;
+	}
 	page_cur_position(cur_rec, new_block, &page_cur);
 
 	while (page_cur_get_rec(&cur1) != rec) {
@@ -1431,6 +1468,9 @@ rtr_page_copy_rec_list_start_no_locks(
 
 		if (page_rec_is_infimum(cur_rec)) {
 			cur_rec = page_rec_get_next(cur_rec);
+			if (UNIV_UNLIKELY(!cur_rec)) {
+				return DB_CORRUPTION;
+			}
 		}
 
 		offsets1 = rec_get_offsets(cur1_rec, index, offsets1, n_core,
@@ -1449,8 +1489,7 @@ rtr_page_copy_rec_list_start_no_locks(
 				goto move_to_prev;
 			} else if (cmp > 0) {
 				/* Skip small recs. */
-				page_cur_move_to_next(&page_cur);
-				cur_rec = page_cur_get_rec(&page_cur);
+				cur_rec = page_cur_move_to_next(&page_cur);
 			} else if (n_core) {
 				if (rec_get_deleted_flag(
 					cur1_rec,
@@ -1472,13 +1511,14 @@ rtr_page_copy_rec_list_start_no_locks(
 		if (page_rec_is_supremum(cur_rec)) {
 move_to_prev:
 			cur_rec = page_cur_move_to_prev(&page_cur);
-			if (UNIV_UNLIKELY(!cur_rec)) {
-				return DB_CORRUPTION;
-			}
 		} else {
 			cur_rec = page_cur_get_rec(&page_cur);
 		}
 
+		if (UNIV_UNLIKELY(!cur_rec)) {
+			return DB_CORRUPTION;
+		}
+
 		offsets1 = rec_get_offsets(cur1_rec, index, offsets1, n_core,
 					   ULINT_UNDEFINED, &heap);
 
@@ -1493,7 +1533,9 @@ move_to_prev:
 		rec_move[moved].moved = false;
 		moved++;
 next:
-		page_cur_move_to_next(&cur1);
+		if (UNIV_UNLIKELY(!page_cur_move_to_next(&cur1))) {
+			return DB_CORRUPTION;
+		}
 	}
 
 	*num_moved = moved;
@@ -1559,7 +1601,7 @@ rtr_merge_and_update_mbr(
 	rtr_mbr_t		new_mbr;
 
 	if (rtr_merge_mbr_changed(cursor, cursor2, offsets, offsets2,
-                                  &new_mbr)) {
+				  &new_mbr)) {
 		rtr_update_mbr_field(cursor, offsets, cursor2, child_page,
 				     &new_mbr, NULL, mtr);
 	} else {
@@ -1604,10 +1646,9 @@ rtr_check_same_block(
 {
 	ulint		page_no = childb->page.id().page_no();
 	rec_offs*	offsets;
-	rec_t*		rec = page_rec_get_next(page_get_infimum_rec(
-				buf_block_get_frame(parentb)));
+	rec_t*		rec = page_get_infimum_rec(parentb->page.frame);
 
-	while (!page_rec_is_supremum(rec)) {
+	while ((rec = page_rec_get_next(rec)) && !page_rec_is_supremum(rec)) {
 		offsets = rec_get_offsets(
 			rec, index, NULL, 0, ULINT_UNDEFINED, &heap);
 
@@ -1615,8 +1656,6 @@ rtr_check_same_block(
 			btr_cur_position(index, rec, parentb, cursor);
 			return(true);
 		}
-
-		rec = page_rec_get_next(rec);
 	}
 
 	return(false);
@@ -1826,9 +1865,9 @@ err_exit:
 
 	/* Scan records in root page and calculate area. */
 	double	area = 0;
-	for (const rec_t* rec = page_rec_get_next(
+	for (const rec_t* rec = page_rec_get_next_const(
 		     page_get_infimum_rec(block->page.frame));
-	     !page_rec_is_supremum(rec);
+	     rec && !page_rec_is_supremum(rec);
 	     rec = page_rec_get_next_const(rec)) {
 		rtr_mbr_t	mbr;
 		double		rec_area;
diff --git a/storage/innobase/gis/gis0sea.cc b/storage/innobase/gis/gis0sea.cc
index 9eb56f1d46c..9a378d531d9 100644
--- a/storage/innobase/gis/gis0sea.cc
+++ b/storage/innobase/gis/gis0sea.cc
@@ -301,18 +301,17 @@ rtr_pcur_getnext_from_path(
 		page_cursor->rec = NULL;
 
 		if (mode == PAGE_CUR_RTREE_LOCATE) {
-			if (level == target_level && level == 0) {
-				ulint	low_match;
+			if (target_level == 0 && level == 0) {
+				ulint	low_match = 0, up_match = 0;
 
 				found = false;
 
-				low_match = page_cur_search(
-					block, index, tuple,
-					PAGE_CUR_LE,
-					btr_cur_get_page_cur(btr_cur));
-
-				if (low_match == dtuple_get_n_fields_cmp(
-							tuple)) {
+				if (!page_cur_search_with_match(
+					block, index, tuple, PAGE_CUR_LE,
+					&up_match, &low_match,
+					btr_cur_get_page_cur(btr_cur), nullptr)
+				    && low_match
+				    == dtuple_get_n_fields_cmp(tuple)) {
 					rec_t*	rec = btr_cur_get_rec(btr_cur);
 
 					if (!rec_get_deleted_flag(rec,
@@ -803,6 +802,9 @@ rtr_page_get_father_block(
 {
 	rec_t*  rec = page_rec_get_next(
 		page_get_infimum_rec(buf_block_get_frame(block)));
+	if (!rec) {
+		return nullptr;
+	}
 	btr_cur_position(index, rec, block, cursor);
 
 	return(rtr_page_get_father_node_ptr(offsets, heap, sea_cur,
@@ -1238,11 +1240,14 @@ rtr_cur_restore_position(
 	ut_ad(r_cursor == node->cursor);
 
 search_again:
+	ulint up_match = 0, low_match = 0;
+
 	block = buf_page_get_gen(
 		page_id_t(index->table->space_id, page_no),
 		zip_size, RW_X_LATCH, NULL, BUF_GET, mtr);
 
 	if (!block) {
+corrupted:
 		ret = false;
 		goto func_exit;
 	}
@@ -1251,8 +1256,13 @@ search_again:
 	page = buf_block_get_frame(block);
 	page_ssn = page_get_ssn_id(page);
 
-	if (page_cur_search(block, index, tuple, PAGE_CUR_LE, page_cursor)
-	    == r_cursor->old_n_fields) {
+	if (page_cur_search_with_match(block, index, tuple, PAGE_CUR_LE,
+				       &up_match, &low_match, page_cursor,
+				       nullptr)) {
+		goto corrupted;
+	}
+
+	if (low_match == r_cursor->old_n_fields) {
 		const rec_t*	rec;
 		const rec_offs*	offsets1;
 		const rec_offs*	offsets2;
@@ -1609,13 +1619,20 @@ rtr_cur_search_with_match(
 		mode = PAGE_CUR_WITHIN;
 	}
 
-	rec = page_dir_slot_get_rec(page_dir_get_nth_slot(page, 0));
+	rec = page_dir_slot_get_rec_validate(page_dir_get_nth_slot(page, 0));
+
+	if (UNIV_UNLIKELY(!rec)) {
+		return false;
+	}
 
 	last_rec = rec;
 	best_rec = rec;
 
 	if (page_rec_is_infimum(rec)) {
 		rec = page_rec_get_next_const(rec);
+		if (UNIV_UNLIKELY(!rec)) {
+			return false;
+		}
 	}
 
 	/* Check insert tuple size is larger than first rec, and try to
@@ -1813,7 +1830,7 @@ rtr_cur_search_with_match(
 	}
 
 	/* All records on page are searched */
-	if (page_rec_is_supremum(rec)) {
+	if (rec && page_rec_is_supremum(rec)) {
 		if (!n_core) {
 			if (!found) {
 				/* No match case, if it is for insertion,
diff --git a/storage/innobase/handler/handler0alter.cc b/storage/innobase/handler/handler0alter.cc
index c0f5d98645d..466e7c6cdf0 100644
--- a/storage/innobase/handler/handler0alter.cc
+++ b/storage/innobase/handler/handler0alter.cc
@@ -2132,11 +2132,14 @@ non_empty:
     return false;
   }
   rec= page_rec_get_next(btr_pcur_get_rec(&pcur));
+  if (UNIV_UNLIKELY(!rec))
+    goto non_empty;
   if (rec_is_metadata(rec, *clust_index))
     btr_pcur_get_page_cur(&pcur)->rec= rec;
 scan_leaf:
   cur= btr_pcur_get_page_cur(&pcur);
-  page_cur_move_to_next(cur);
+  if (UNIV_UNLIKELY(!page_cur_move_to_next(cur)))
+    goto non_empty;
 next_page:
   if (next_page)
   {
@@ -2154,7 +2157,8 @@ next_page:
       goto non_empty;
     btr_leaf_page_release(page_cur_get_block(cur), BTR_SEARCH_LEAF, &mtr);
     page_cur_set_before_first(block, cur);
-    page_cur_move_to_next(cur);
+    if (UNIV_UNLIKELY(!page_cur_move_to_next(cur)))
+      goto non_empty;
   }
 
   rec= page_cur_get_rec(cur);
@@ -6029,13 +6033,17 @@ func_exit:
 		return false;
 	}
 	ut_ad(btr_pcur_is_before_first_on_page(&pcur));
-	btr_pcur_move_to_next_on_page(&pcur);
 
 	buf_block_t* block = btr_pcur_get_block(&pcur);
 	ut_ad(page_is_leaf(block->page.frame));
 	ut_ad(!page_has_prev(block->page.frame));
 	ut_ad(!buf_block_get_page_zip(block));
-	const rec_t* rec = btr_pcur_get_rec(&pcur);
+	const rec_t* rec = btr_pcur_move_to_next_on_page(&pcur);
+	if (UNIV_UNLIKELY(!rec)) {
+		err = DB_CORRUPTION;
+		goto func_exit;
+	}
+
 	que_thr_t* thr = pars_complete_graph_for_exec(
 		NULL, trx, ctx->heap, NULL);
 	const bool is_root = block->page.id().page_no() == index->page;
@@ -6111,10 +6119,13 @@ func_exit:
 			&offsets, &offsets_heap, ctx->heap,
 			&big_rec, update, UPD_NODE_NO_ORD_CHANGE,
 			thr, trx->id, &mtr);
+		if (err == DB_SUCCESS) {
+			offsets = rec_get_offsets(
+				btr_pcur_get_rec(&pcur), index, offsets,
+				index->n_core_fields, ULINT_UNDEFINED,
+				&offsets_heap);
+		}
 
-		offsets = rec_get_offsets(
-			btr_pcur_get_rec(&pcur), index, offsets,
-			index->n_core_fields, ULINT_UNDEFINED, &offsets_heap);
 		if (big_rec) {
 			if (err == DB_SUCCESS) {
 				err = btr_store_big_rec_extern_fields(
diff --git a/storage/innobase/ibuf/ibuf0ibuf.cc b/storage/innobase/ibuf/ibuf0ibuf.cc
index 3bd1deced5c..0a6e951dbaa 100644
--- a/storage/innobase/ibuf/ibuf0ibuf.cc
+++ b/storage/innobase/ibuf/ibuf0ibuf.cc
@@ -929,7 +929,7 @@ ibuf_page_low(
 			zip_size, RW_NO_LATCH, nullptr, BUF_GET, &local_mtr);
 
 		ret = block
-                        && ibuf_bitmap_page_get_bits_low(
+			&& ibuf_bitmap_page_get_bits_low(
 			block->page.frame, page_id, zip_size,
 			MTR_MEMO_BUF_FIX, &local_mtr, IBUF_BITMAP_IBUF);
 
@@ -2047,7 +2047,7 @@ corruption:
 
 	if (page_rec_is_infimum(rec)) {
 		rec = page_rec_get_next_const(rec);
-		if (page_rec_is_supremum(rec)) {
+		if (!rec || page_rec_is_supremum(rec)) {
 			return 0;
 		}
 	}
@@ -2101,7 +2101,7 @@ corruption:
 	sum_volumes = 0;
 	volume_for_page = 0;
 
-	while (*n_stored < limit) {
+	while (*n_stored < limit && rec) {
 		if (page_rec_is_supremum(rec)) {
 			/* When no more records available, mark this with
 			another 'impossible' pair of space id, page no */
@@ -2861,10 +2861,10 @@ corruption:
 	static_assert(FIL_PAGE_NEXT % 4 == 0, "alignment");
 	static_assert(FIL_PAGE_OFFSET % 4 == 0, "alignment");
 
-        if (UNIV_UNLIKELY(memcmp_aligned<4>(prev_page + FIL_PAGE_NEXT,
-                                            page + FIL_PAGE_OFFSET, 4))) {
+	if (UNIV_UNLIKELY(memcmp_aligned<4>(prev_page + FIL_PAGE_NEXT,
+					    page + FIL_PAGE_OFFSET, 4))) {
 		return srv_page_size;
-        }
+	}
 
 	rec = page_rec_get_prev_const(page_get_supremum_rec(prev_page));
 
@@ -2908,6 +2908,9 @@ count_later:
 
 	for (; !page_rec_is_supremum(rec);
 	     rec = page_rec_get_next_const(rec)) {
+		if (UNIV_UNLIKELY(!rec)) {
+			return srv_page_size;
+		}
 		if (page_no != ibuf_rec_get_page_no(mtr, rec)
 		    || space != ibuf_rec_get_space(mtr, rec)) {
 
@@ -2940,24 +2943,22 @@ count_later:
 	static_assert(FIL_PAGE_PREV % 4 == 0, "alignment");
 	static_assert(FIL_PAGE_OFFSET % 4 == 0, "alignment");
 
-        if (UNIV_UNLIKELY(memcmp_aligned<4>(next_page + FIL_PAGE_PREV,
-                                            page + FIL_PAGE_OFFSET, 4))) {
+	if (UNIV_UNLIKELY(memcmp_aligned<4>(next_page + FIL_PAGE_PREV,
+					    page + FIL_PAGE_OFFSET, 4))) {
 		return 0;
-        }
+	}
 
 	rec = page_get_infimum_rec(next_page);
 	rec = page_rec_get_next_const(rec);
 
-	for (;; rec = page_rec_get_next_const(rec)) {
-		ut_ad(page_align(rec) == next_page);
-
-		if (page_rec_is_supremum(rec)) {
-
+	for (; ; rec = page_rec_get_next_const(rec)) {
+		if (!rec || page_rec_is_supremum(rec)) {
 			/* We give up */
-
 			return(srv_page_size);
 		}
 
+		ut_ad(page_align(rec) == next_page);
+
 		if (page_no != ibuf_rec_get_page_no(mtr, rec)
 		    || space != ibuf_rec_get_space(mtr, rec)) {
 
@@ -3697,7 +3698,6 @@ ibuf_insert_to_index_page(
 	mtr_t*		mtr)	/*!< in: mtr */
 {
 	page_cur_t	page_cur;
-	ulint		low_match;
 	page_t*		page		= buf_block_get_frame(block);
 	rec_t*		rec;
 	rec_offs*	offsets;
@@ -3726,7 +3726,7 @@ ibuf_insert_to_index_page(
 
 	rec = page_rec_get_next(page_get_infimum_rec(page));
 
-	if (page_rec_is_supremum(rec)) {
+	if (!rec || page_rec_is_supremum(rec)) {
 		return DB_CORRUPTION;
 	}
 
@@ -3734,8 +3734,15 @@ ibuf_insert_to_index_page(
 		return DB_CORRUPTION;
 	}
 
+	ulint up_match = 0, low_match = 0;
+
+	if (page_cur_search_with_match(block, index, entry, PAGE_CUR_LE,
+				       &up_match, &low_match, &page_cur,
+				       nullptr)) {
+		return DB_CORRUPTION;
+	}
+
 	dberr_t err = DB_SUCCESS;
-	low_match = page_cur_search(block, index, entry, &page_cur);
 
 	heap = mem_heap_create(
 		sizeof(upd_t)
@@ -3824,7 +3831,7 @@ ibuf_insert_to_index_page(
 		}
 	} else {
 		offsets = NULL;
-        }
+	}
 
 	err = ibuf_insert_to_index_page_low(entry, block, index,
 					    &offsets, heap, mtr, &page_cur);
@@ -3847,14 +3854,15 @@ ibuf_set_del_mark(
 	mtr_t*			mtr)	/*!< in: mtr */
 {
 	page_cur_t	page_cur;
-	ulint		low_match;
+	ulint		up_match = 0, low_match = 0;
 
 	ut_ad(ibuf_inside(mtr));
 	ut_ad(dtuple_check_typed(entry));
 
-	low_match = page_cur_search(block, index, entry, &page_cur);
-
-	if (low_match == dtuple_get_n_fields(entry)) {
+	if (!page_cur_search_with_match(block, index, entry, PAGE_CUR_LE,
+					&up_match, &low_match, &page_cur,
+					nullptr)
+	    && low_match == dtuple_get_n_fields(entry)) {
 		rec_t* rec = page_cur_get_rec(&page_cur);
 
 		/* Delete mark the old index record. According to a
@@ -3903,16 +3911,17 @@ ibuf_delete(
 				before latching any further pages */
 {
 	page_cur_t	page_cur;
-	ulint		low_match;
+	ulint		up_match = 0, low_match = 0;
 
 	ut_ad(ibuf_inside(mtr));
 	ut_ad(dtuple_check_typed(entry));
 	ut_ad(!index->is_spatial());
 	ut_ad(!index->is_clust());
 
-	low_match = page_cur_search(block, index, entry, &page_cur);
-
-	if (low_match == dtuple_get_n_fields(entry)) {
+	if (!page_cur_search_with_match(block, index, entry, PAGE_CUR_LE,
+					&up_match, &low_match, &page_cur,
+					nullptr)
+	    && low_match == dtuple_get_n_fields(entry)) {
 		page_zip_des_t*	page_zip= buf_block_get_page_zip(block);
 		page_t*		page	= buf_block_get_frame(block);
 		rec_t*		rec	= page_cur_get_rec(&page_cur);
@@ -3976,8 +3985,6 @@ ibuf_delete(
 		if (UNIV_LIKELY_NULL(heap)) {
 			mem_heap_free(heap);
 		}
-	} else {
-		/* The record must have been purged already. */
 	}
 }
 
@@ -4016,9 +4023,6 @@ ibuf_restore_pos(
 		rec_print_old(stderr, btr_pcur_get_rec(pcur));
 		rec_print_old(stderr, pcur->old_rec);
 		dtuple_print(stderr, search_tuple);
-
-		rec_print_old(stderr,
-			      page_rec_get_next(btr_pcur_get_rec(pcur)));
 	}
 
 	ibuf_btr_pcur_commit_specify_mtr(pcur, mtr);
@@ -4644,7 +4648,7 @@ dberr_t ibuf_check_bitmap_on_import(const trx_t* trx, fil_space_t* space)
 				buf_block_t* block = buf_page_get(
 					page_id_t(space->id, curr_page),
 					zip_size, RW_S_LATCH, &mtr);
-	                        page_t*	page = buf_block_get_frame(block);
+				page_t*	page = buf_block_get_frame(block);
 				ut_ad(buf_is_zeroes(span<const byte>(
 							    page,
 							    physical_size)));
diff --git a/storage/innobase/include/btr0cur.h b/storage/innobase/include/btr0cur.h
index 922f4bbc429..bbf7ab40d7e 100644
--- a/storage/innobase/include/btr0cur.h
+++ b/storage/innobase/include/btr0cur.h
@@ -222,7 +222,7 @@ It is assumed that mtr holds an x-latch on the page. The operation does
 not succeed if there is too little space on the page. If there is just
 one record on the page, the insert will always succeed; this is to
 prevent trying to split a page with just one record.
-@return DB_SUCCESS, DB_WAIT_LOCK, DB_FAIL, or error number */
+@return DB_SUCCESS, DB_LOCK_WAIT, DB_FAIL, or error number */
 dberr_t
 btr_cur_optimistic_insert(
 /*======================*/
diff --git a/storage/innobase/include/btr0pcur.h b/storage/innobase/include/btr0pcur.h
index f2a1f4220da..2d8f6d1024f 100644
--- a/storage/innobase/include/btr0pcur.h
+++ b/storage/innobase/include/btr0pcur.h
@@ -324,10 +324,11 @@ static inline bool btr_pcur_is_before_first_in_tree(btr_pcur_t* cursor);
 Checks if the persistent cursor is after the last user record in
 the index tree. */
 static inline bool btr_pcur_is_after_last_in_tree(btr_pcur_t* cursor);
+MY_ATTRIBUTE((nonnull, warn_unused_result))
 /*********************************************************//**
 Moves the persistent cursor to the next record on the same page. */
 UNIV_INLINE
-void
+rec_t*
 btr_pcur_move_to_next_on_page(
 /*==========================*/
 	btr_pcur_t*	cursor);/*!< in/out: persistent cursor */
@@ -513,8 +514,7 @@ btr_pcur_open_on_user_rec(
     return DB_SUCCESS;
   if (dberr_t err= btr_pcur_move_to_next_page(cursor, mtr))
     return err;
-  btr_pcur_move_to_next_on_page(cursor);
-  return DB_SUCCESS;
+  return btr_pcur_move_to_next_on_page(cursor) ? DB_SUCCESS : DB_CORRUPTION;
 }
 
 #include "btr0pcur.inl"
diff --git a/storage/innobase/include/btr0pcur.inl b/storage/innobase/include/btr0pcur.inl
index b21de209760..2b573068c01 100644
--- a/storage/innobase/include/btr0pcur.inl
+++ b/storage/innobase/include/btr0pcur.inl
@@ -155,7 +155,7 @@ static inline bool btr_pcur_is_after_last_in_tree(btr_pcur_t* cursor)
 /*********************************************************//**
 Moves the persistent cursor to the next record on the same page. */
 UNIV_INLINE
-void
+rec_t*
 btr_pcur_move_to_next_on_page(
 /*==========================*/
 	btr_pcur_t*	cursor)	/*!< in/out: persistent cursor */
@@ -163,9 +163,8 @@ btr_pcur_move_to_next_on_page(
 	ut_ad(cursor->pos_state == BTR_PCUR_IS_POSITIONED);
 	ut_ad(cursor->latch_mode != BTR_NO_LATCHES);
 
-	page_cur_move_to_next(btr_pcur_get_page_cur(cursor));
-
 	cursor->old_stored = false;
+	return page_cur_move_to_next(btr_pcur_get_page_cur(cursor));
 }
 
 /*********************************************************//**
@@ -204,8 +203,8 @@ loop:
 		    || btr_pcur_move_to_next_page(cursor, mtr) != DB_SUCCESS) {
 			return(FALSE);
 		}
-	} else {
-		btr_pcur_move_to_next_on_page(cursor);
+	} else if (UNIV_UNLIKELY(!btr_pcur_move_to_next_on_page(cursor))) {
+		return false;
 	}
 
 	if (btr_pcur_is_on_user_rec(cursor)) {
@@ -228,20 +227,16 @@ btr_pcur_move_to_next(
 				function may release the page latch */
 	mtr_t*		mtr)	/*!< in: mtr */
 {
-	ut_ad(cursor->pos_state == BTR_PCUR_IS_POSITIONED);
-	ut_ad(cursor->latch_mode != BTR_NO_LATCHES);
+  ut_ad(cursor->pos_state == BTR_PCUR_IS_POSITIONED);
+  ut_ad(cursor->latch_mode != BTR_NO_LATCHES);
 
-	cursor->old_stored = false;
+  cursor->old_stored= false;
 
-	if (btr_pcur_is_after_last_on_page(cursor)) {
-		if (btr_pcur_is_after_last_in_tree(cursor)
-		    || btr_pcur_move_to_next_page(cursor, mtr) != DB_SUCCESS) {
-			return(FALSE);
-		}
-	} else {
-		btr_pcur_move_to_next_on_page(cursor);
-	}
-	return(TRUE);
+  if (btr_pcur_is_after_last_on_page(cursor))
+    return !btr_pcur_is_after_last_in_tree(cursor) &&
+      btr_pcur_move_to_next_page(cursor, mtr) == DB_SUCCESS;
+  else
+    return !!btr_pcur_move_to_next_on_page(cursor);
 }
 
 /**************************************************************//**
diff --git a/storage/innobase/include/gis0rtree.inl b/storage/innobase/include/gis0rtree.inl
index 2564df7e0d6..5101eeb6f7a 100644
--- a/storage/innobase/include/gis0rtree.inl
+++ b/storage/innobase/include/gis0rtree.inl
@@ -57,6 +57,9 @@ rtr_page_cal_mbr(
 	page = buf_block_get_frame(block);
 
 	rec = page_rec_get_next(page_get_infimum_rec(page));
+	if (UNIV_UNLIKELY(!rec)) {
+		return;
+	}
 	offsets = rec_get_offsets(rec, index, offsets, page_is_leaf(page)
 				  ? index->n_fields : 0,
 				  ULINT_UNDEFINED, &heap);
diff --git a/storage/innobase/include/page0cur.h b/storage/innobase/include/page0cur.h
index 11677513dd8..80542299482 100644
--- a/storage/innobase/include/page0cur.h
+++ b/storage/innobase/include/page0cur.h
@@ -110,21 +110,6 @@ page_cur_position(
 	const buf_block_t*	block,	/*!< in: buffer block containing
 					the record */
 	page_cur_t*		cur);	/*!< out: page cursor */
-/**********************************************************//**
-Moves the cursor to the next record on page. */
-UNIV_INLINE
-void
-page_cur_move_to_next(
-/*==================*/
-	page_cur_t*	cur);	/*!< in/out: cursor; must not be after last */
-MY_ATTRIBUTE((nonnull, warn_unused_result))
-/**********************************************************//**
-Moves the cursor to the previous record on page. */
-UNIV_INLINE
-rec_t*
-page_cur_move_to_prev(
-/*==================*/
-	page_cur_t*	cur);	/*!< in/out: cursor; not before first */
 
 /***********************************************************//**
 Inserts a record next to page cursor. Returns pointer to inserted record if
@@ -166,20 +151,20 @@ page_cur_insert_rec_low(
 
 /***********************************************************//**
 Inserts a record next to page cursor on a compressed and uncompressed
-page. Returns pointer to inserted record if succeed, i.e.,
-enough space available, NULL otherwise.
-The cursor stays at the same position.
+page.
 
 IMPORTANT: The caller will have to update IBUF_BITMAP_FREE
 if this is a compressed leaf page in a secondary index.
 This has to be done either within the same mini-transaction,
 or by invoking ibuf_reset_free_bits() before mtr_commit().
 
-@return pointer to record if succeed, NULL otherwise */
+@return pointer to inserted record
+@return nullptr on failure */
 rec_t*
 page_cur_insert_rec_zip(
 /*====================*/
-	page_cur_t*	cursor,	/*!< in/out: page cursor */
+	page_cur_t*	cursor,	/*!< in/out: page cursor,
+				logical position unchanged  */
 	dict_index_t*	index,	/*!< in: record descriptor */
 	const rec_t*	rec,	/*!< in: pointer to a physical record */
 	rec_offs*	offsets,/*!< in/out: rec_get_offsets(rec, index) */
@@ -248,39 +233,10 @@ page_cur_delete_rec() for a ROW_FORMAT=COMPACT or DYNAMIC page.
 bool page_apply_delete_dynamic(const buf_block_t &block, ulint prev,
                                size_t hdr_size, size_t data_size);
 
-/** Search the right position for a page cursor.
-@param[in] block buffer block
-@param[in] index index tree
-@param[in] tuple data tuple
-@param[in] mode PAGE_CUR_L, PAGE_CUR_LE, PAGE_CUR_G, or PAGE_CUR_GE
-@param[out] cursor page cursor
-@return number of matched fields on the left */
-UNIV_INLINE
-ulint
-page_cur_search(
-	const buf_block_t*	block,
-	const dict_index_t*	index,
-	const dtuple_t*		tuple,
-	page_cur_mode_t		mode,
-	page_cur_t*		cursor);
-
-/** Search the right position for a page cursor.
-@param[in] block buffer block
-@param[in] index index tree
-@param[in] tuple data tuple
-@param[out] cursor page cursor
-@return number of matched fields on the left */
-UNIV_INLINE
-ulint
-page_cur_search(
-	const buf_block_t*	block,
-	const dict_index_t*	index,
-	const dtuple_t*		tuple,
-	page_cur_t*		cursor);
-
+MY_ATTRIBUTE((warn_unused_result))
 /****************************************************************//**
 Searches the right position for a page cursor. */
-void
+bool
 page_cur_search_with_match(
 /*=======================*/
 	const buf_block_t*	block,	/*!< in: buffer block */
@@ -298,6 +254,7 @@ page_cur_search_with_match(
 	page_cur_t*		cursor,	/*!< out: page cursor */
 	rtr_info_t*		rtr_info);/*!< in/out: rtree search stack */
 #ifdef BTR_CUR_HASH_ADAPT
+MY_ATTRIBUTE((warn_unused_result))
 /** Search the right position for a page cursor.
 @param[in]	block			buffer block
 @param[in]	index			index tree
@@ -312,7 +269,7 @@ lower limit record
 @param[in,out]	ilow_matched_bytes	already matched bytes in the
 first partially matched field in the lower limit record
 @param[out]	cursor			page cursor */
-void
+bool
 page_cur_search_with_match_bytes(
 	const buf_block_t*	block,
 	const dict_index_t*	index,
@@ -342,6 +299,19 @@ struct page_cur_t{
 	buf_block_t*	block;	/*!< pointer to the block containing rec */
 };
 
+
+MY_ATTRIBUTE((nonnull, warn_unused_result))
+inline rec_t *page_cur_move_to_next(page_cur_t *cur)
+{
+  return cur->rec= page_rec_get_next(cur->rec);
+}
+
+MY_ATTRIBUTE((nonnull, warn_unused_result))
+inline rec_t *page_cur_move_to_prev(page_cur_t *cur)
+{
+  return cur->rec= page_rec_get_prev(cur->rec);
+}
+
 #include "page0cur.inl"
 
 #endif
diff --git a/storage/innobase/include/page0cur.inl b/storage/innobase/include/page0cur.inl
index e604d85f13f..8f69dc22878 100644
--- a/storage/innobase/include/page0cur.inl
+++ b/storage/innobase/include/page0cur.inl
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1994, 2014, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2015, 2021, MariaDB Corporation.
+Copyright (c) 2015, 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
@@ -149,75 +149,6 @@ page_cur_position(
 	cur->block = (buf_block_t*) block;
 }
 
-/**********************************************************//**
-Moves the cursor to the next record on page. */
-UNIV_INLINE
-void
-page_cur_move_to_next(
-/*==================*/
-	page_cur_t*	cur)	/*!< in/out: cursor; must not be after last */
-{
-	ut_ad(!page_cur_is_after_last(cur));
-
-	cur->rec = page_rec_get_next(cur->rec);
-}
-
-/**********************************************************//**
-Moves the cursor to the previous record on page. */
-UNIV_INLINE
-rec_t*
-page_cur_move_to_prev(
-/*==================*/
-	page_cur_t*	cur)	/*!< in/out: page cursor, not before first */
-{
-	ut_ad(!page_cur_is_before_first(cur));
-
-	return cur->rec = page_rec_get_prev(cur->rec);
-}
-
-/** Search the right position for a page cursor.
-@param[in] block buffer block
-@param[in] index index tree
-@param[in] tuple data tuple
-@param[in] mode PAGE_CUR_L, PAGE_CUR_LE, PAGE_CUR_G, or PAGE_CUR_GE
-@param[out] cursor page cursor
-@return number of matched fields on the left */
-UNIV_INLINE
-ulint
-page_cur_search(
-	const buf_block_t*	block,
-	const dict_index_t*	index,
-	const dtuple_t*		tuple,
-	page_cur_mode_t		mode,
-	page_cur_t*		cursor)
-{
-	ulint		low_match = 0;
-	ulint		up_match = 0;
-
-	ut_ad(dtuple_check_typed(tuple));
-
-	page_cur_search_with_match(block, index, tuple, mode,
-				   &up_match, &low_match, cursor, NULL);
-	return(low_match);
-}
-
-/** Search the right position for a page cursor.
-@param[in] block buffer block
-@param[in] index index tree
-@param[in] tuple data tuple
-@param[out] cursor page cursor
-@return number of matched fields on the left */
-UNIV_INLINE
-ulint
-page_cur_search(
-	const buf_block_t*	block,
-	const dict_index_t*	index,
-	const dtuple_t*		tuple,
-	page_cur_t*		cursor)
-{
-	return(page_cur_search(block, index, tuple, PAGE_CUR_LE, cursor));
-}
-
 /***********************************************************//**
 Inserts a record next to page cursor. Returns pointer to inserted record if
 succeed, i.e., enough space available, NULL otherwise. The cursor stays at
diff --git a/storage/innobase/include/page0page.h b/storage/innobase/include/page0page.h
index 4787ce36c7a..b0e2eb98d01 100644
--- a/storage/innobase/include/page0page.h
+++ b/storage/innobase/include/page0page.h
@@ -534,7 +534,8 @@ inline void page_header_reset_last_insert(buf_block_t *block, mtr_t *mtr)
 /************************************************************//**
 Returns the nth record of the record list.
 This is the inverse function of page_rec_get_n_recs_before().
-@return nth record */
+@return nth record
+@retval nullptr on corrupted page */
 const rec_t*
 page_rec_get_nth_const(
 /*===================*/
@@ -544,14 +545,12 @@ page_rec_get_nth_const(
 /************************************************************//**
 Returns the nth record of the record list.
 This is the inverse function of page_rec_get_n_recs_before().
-@return nth record */
-UNIV_INLINE
-rec_t*
-page_rec_get_nth(
-/*=============*/
-	page_t*	page,	/*< in: page */
-	ulint	nth)	/*!< in: nth record */
-	MY_ATTRIBUTE((nonnull, warn_unused_result));
+@return nth record
+@retval nullptr on corrupted page */
+inline rec_t *page_rec_get_nth(page_t* page, ulint nth)
+{
+  return const_cast<rec_t*>(page_rec_get_nth_const(page, nth));
+}
 
 /************************************************************//**
 Returns the middle record of the records on the page. If there is an
@@ -592,15 +591,11 @@ page_get_n_recs(
 /*============*/
 	const page_t*	page);	/*!< in: index page */
 
-/***************************************************************//**
-Returns the number of records before the given record in chain.
-The number includes infimum and supremum records.
-This is the inverse function of page_rec_get_nth().
-@return number of records */
-ulint
-page_rec_get_n_recs_before(
-/*=======================*/
-	const rec_t*	rec);	/*!< in: the physical record */
+/** Return the number of preceding records in an index page.
+@param rec index record
+@return number of preceding records, including the infimum pseudo-record
+@retval ULINT_UNDEFINED on corrupted page */
+ulint page_rec_get_n_recs_before(const rec_t *rec);
 /*************************************************************//**
 Gets the number of records in the heap.
 @return number of user records */
@@ -649,6 +644,23 @@ inline const rec_t *page_dir_slot_get_rec(const page_dir_slot_t *slot)
 {
   return page_dir_slot_get_rec(const_cast<rec_t*>(slot));
 }
+
+inline rec_t *page_dir_slot_get_rec_validate(page_dir_slot_t *slot)
+{
+  const size_t s= mach_read_from_2(my_assume_aligned<2>(slot));
+  page_t *page= page_align(slot);
+
+  return UNIV_LIKELY(s >= PAGE_NEW_INFIMUM &&
+                     s <= page_header_get_field(page, PAGE_HEAP_TOP))
+    ? page + s
+    : nullptr;
+}
+inline const rec_t *page_dir_slot_get_rec_validate(const page_dir_slot_t *slot)
+{
+  return page_dir_slot_get_rec_validate(const_cast<rec_t*>(slot));
+}
+
+
 /***************************************************************//**
 Gets the number of records owned by a directory slot.
 @return number of records */
@@ -753,20 +765,9 @@ page_rec_get_next_const(
 /*====================*/
 	const rec_t*	rec);	/*!< in: pointer to record */
 /************************************************************//**
-Gets the pointer to the next non delete-marked record on the page.
-If all subsequent records are delete-marked, then this function
-will return the supremum record.
-@return pointer to next non delete-marked record or pointer to supremum */
-UNIV_INLINE
-const rec_t*
-page_rec_get_next_non_del_marked(
-/*=============================*/
-	const rec_t*	rec);	/*!< in: pointer to record */
-/************************************************************//**
 Gets the pointer to the previous record.
 @return pointer to previous record
 @retval nullptr on error */
-UNIV_INLINE
 const rec_t*
 page_rec_get_prev_const(
 /*====================*/
@@ -816,22 +817,6 @@ page_rec_is_last(
 	MY_ATTRIBUTE((warn_unused_result));
 
 /************************************************************//**
-true if distance between the records (measured in number of times we have to
-move to the next record) is at most the specified value
-@param[in]	left_rec	lefter record
-@param[in]	right_rec	righter record
-@param[in]	val		specified value to compare
-@return true if the distance is smaller than the value */
-UNIV_INLINE
-bool
-page_rec_distance_is_at_most(
-/*=========================*/
-	const rec_t*	left_rec,
-	const rec_t*	right_rec,
-	ulint		val)
-	MY_ATTRIBUTE((warn_unused_result));
-
-/************************************************************//**
 true if the record is the second last user record on a page.
 @return true if the second last user record */
 UNIV_INLINE
@@ -1129,9 +1114,7 @@ page_find_rec_with_heap_no(
 @param[in]	page	index tree leaf page
 @return the last record, not delete-marked
 @retval infimum record if all records are delete-marked */
-const rec_t*
-page_find_rec_max_not_deleted(
-	const page_t*	page);
+const rec_t *page_find_rec_max_not_deleted(const page_t *page);
 
 #endif /* !UNIV_INNOCHECKSUM */
 
diff --git a/storage/innobase/include/page0page.inl b/storage/innobase/include/page0page.inl
index 76bc62e5eb2..61c1b96ff79 100644
--- a/storage/innobase/include/page0page.inl
+++ b/storage/innobase/include/page0page.inl
@@ -202,10 +202,10 @@ page_rec_is_second(
 	const rec_t*	rec,	/*!< in: record */
 	const page_t*	page)	/*!< in: page */
 {
-	ut_ad(page_get_n_recs(page) > 1);
-
-	return(page_rec_get_next_const(
-		page_rec_get_next_const(page_get_infimum_rec(page))) == rec);
+  ut_ad(page_get_n_recs(page) > 1);
+  if (const rec_t *first= page_rec_get_next_const(page_get_infimum_rec(page)))
+    return page_rec_get_next_const(first) == rec;
+  return false;
 }
 
 /************************************************************//**
@@ -224,26 +224,6 @@ page_rec_is_last(
 }
 
 /************************************************************//**
-true if distance between the records (measured in number of times we have to
-move to the next record) is at most the specified value */
-UNIV_INLINE
-bool
-page_rec_distance_is_at_most(
-/*=========================*/
-	const rec_t*	left_rec,
-	const rec_t*	right_rec,
-	ulint		val)
-{
-	for (ulint i = 0; i <= val; i++) {
-		if (left_rec == right_rec) {
-			return (true);
-		}
-		left_rec = page_rec_get_next_const(left_rec);
-	}
-	return (false);
-}
-
-/************************************************************//**
 true if the record is the second last user record on a page.
 @return true if the second last user record */
 UNIV_INLINE
@@ -253,25 +233,12 @@ page_rec_is_second_last(
 	const rec_t*	rec,	/*!< in: record */
 	const page_t*	page)	/*!< in: page */
 {
-	ut_ad(page_get_n_recs(page) > 1);
-	ut_ad(!page_rec_is_last(rec, page));
-
-	return(page_rec_get_next_const(
-		page_rec_get_next_const(rec)) == page_get_supremum_rec(page));
-}
+  ut_ad(page_get_n_recs(page) > 1);
+  ut_ad(!page_rec_is_last(rec, page));
 
-/************************************************************//**
-Returns the nth record of the record list.
-This is the inverse function of page_rec_get_n_recs_before().
-@return nth record */
-UNIV_INLINE
-rec_t*
-page_rec_get_nth(
-/*=============*/
-	page_t*	page,	/*!< in: page */
-	ulint	nth)	/*!< in: nth record */
-{
-	return((rec_t*) page_rec_get_nth_const(page, nth));
+  if (const rec_t *next= page_rec_get_next_const(rec))
+    return page_rec_is_supremum(page_rec_get_next_const(next));
+  return false;
 }
 
 /************************************************************//**
@@ -421,36 +388,19 @@ page_rec_get_next_low(
 	const rec_t*	rec,	/*!< in: pointer to record */
 	ulint		comp)	/*!< in: nonzero=compact page layout */
 {
-	ulint		offs;
-	const page_t*	page;
-
-	ut_ad(page_rec_check(rec));
-
-	page = page_align(rec);
-
-	offs = rec_get_next_offs(rec, comp);
-
-	if (offs >= srv_page_size) {
-		fprintf(stderr,
-			"InnoDB: Next record offset is nonsensical %lu"
-			" in record at offset %lu\n"
-			"InnoDB: rec address %p, space id %lu, page %lu\n",
-			(ulong) offs, (ulong) page_offset(rec),
-			(void*) rec,
-			(ulong) page_get_space_id(page),
-			(ulong) page_get_page_no(page));
-		ut_error;
-	} else if (offs == 0) {
-
-		return(NULL);
-	}
-
-	ut_ad(page_rec_is_infimum(rec)
-	      || (!page_is_leaf(page) && !page_has_prev(page))
-	      || !(rec_get_info_bits(page + offs, comp)
-		   & REC_INFO_MIN_REC_FLAG));
-
-	return(page + offs);
+  const page_t *page= page_align(rec);
+  ut_ad(page_rec_check(rec));
+  ulint offs= rec_get_next_offs(rec, comp);
+  if (!offs)
+    return nullptr;
+  if (UNIV_UNLIKELY(offs < (comp ? PAGE_NEW_SUPREMUM : PAGE_OLD_SUPREMUM)))
+    return nullptr;
+  if (UNIV_UNLIKELY(offs > page_header_get_field(page, PAGE_HEAP_TOP)))
+    return nullptr;
+  ut_ad(page_rec_is_infimum(rec) ||
+        (!page_is_leaf(page) && !page_has_prev(page)) ||
+        !(rec_get_info_bits(page + offs, comp) & REC_INFO_MIN_REC_FLAG));
+  return page + offs;
 }
 
 /************************************************************//**
@@ -476,78 +426,6 @@ page_rec_get_next_const(
 {
 	return(page_rec_get_next_low(rec, page_rec_is_comp(rec)));
 }
-
-/************************************************************//**
-Gets the pointer to the next non delete-marked record on the page.
-If all subsequent records are delete-marked, then this function
-will return the supremum record.
-@return pointer to next non delete-marked record or pointer to supremum */
-UNIV_INLINE
-const rec_t*
-page_rec_get_next_non_del_marked(
-/*=============================*/
-	const rec_t*	rec)	/*!< in: pointer to record */
-{
-	const rec_t*	r;
-	ulint		page_is_compact = page_rec_is_comp(rec);
-
-	for (r = page_rec_get_next_const(rec);
-	     !page_rec_is_supremum(r)
-	     && rec_get_deleted_flag(r, page_is_compact);
-	     r = page_rec_get_next_const(r)) {
-		/* noop */
-	}
-
-	return(r);
-}
-
-/************************************************************//**
-Gets the pointer to the previous record.
-@return pointer to previous record
-@retval nullptr on error */
-UNIV_INLINE
-const rec_t*
-page_rec_get_prev_const(
-/*====================*/
-	const rec_t*	rec)	/*!< in: pointer to record, must not be page
-				infimum */
-{
-	const page_dir_slot_t*	slot;
-	ulint			slot_no;
-	const rec_t*		rec2;
-	const rec_t*		prev_rec = NULL;
-	const page_t*		page;
-
-	ut_ad(page_rec_check(rec));
-
-	page = page_align(rec);
-
-	ut_ad(!page_rec_is_infimum(rec));
-
-	slot_no = page_dir_find_owner_slot(rec);
-
-	if (UNIV_UNLIKELY(!slot_no || slot_no == ULINT_UNDEFINED)) {
-		return nullptr;
-	}
-
-	slot = page_dir_get_nth_slot(page, slot_no - 1);
-
-	rec2 = page_dir_slot_get_rec(slot);
-
-	if (page_is_comp(page)) {
-		while (rec2 && rec != rec2) {
-			prev_rec = rec2;
-			rec2 = page_rec_get_next_low(rec2, TRUE);
-		}
-	} else {
-		while (rec2 && rec != rec2) {
-			prev_rec = rec2;
-			rec2 = page_rec_get_next_low(rec2, FALSE);
-		}
-	}
-
-	return(prev_rec);
-}
 #endif /* UNIV_INNOCHECKSUM */
 
 /************************************************************//**
diff --git a/storage/innobase/include/rem0rec.h b/storage/innobase/include/rem0rec.h
index 7faf0ca06bd..dcf26e57e7e 100644
--- a/storage/innobase/include/rem0rec.h
+++ b/storage/innobase/include/rem0rec.h
@@ -141,28 +141,7 @@ constexpr rec_offs REC_OFFS_EXTERNAL= REC_OFFS_COMPACT >> 1;
 /** Default value flag in offsets returned by rec_get_offsets() */
 constexpr rec_offs REC_OFFS_DEFAULT= REC_OFFS_COMPACT >> 2;
 constexpr rec_offs REC_OFFS_MASK= REC_OFFS_DEFAULT - 1;
-/******************************************************//**
-The following function is used to get the pointer of the next chained record
-on the same page.
-@return pointer to the next chained record, or NULL if none */
-UNIV_INLINE
-const rec_t*
-rec_get_next_ptr_const(
-/*===================*/
-	const rec_t*	rec,	/*!< in: physical record */
-	ulint		comp)	/*!< in: nonzero=compact page format */
-	MY_ATTRIBUTE((warn_unused_result));
-/******************************************************//**
-The following function is used to get the pointer of the next chained record
-on the same page.
-@return pointer to the next chained record, or NULL if none */
-UNIV_INLINE
-rec_t*
-rec_get_next_ptr(
-/*=============*/
-	rec_t*	rec,	/*!< in: physical record */
-	ulint	comp)	/*!< in: nonzero=compact page format */
-	MY_ATTRIBUTE((warn_unused_result));
+
 /******************************************************//**
 The following function is used to get the offset of the
 next chained record on the same page.
diff --git a/storage/innobase/include/rem0rec.inl b/storage/innobase/include/rem0rec.inl
index 30c72a7415a..46c209cbdec 100644
--- a/storage/innobase/include/rem0rec.inl
+++ b/storage/innobase/include/rem0rec.inl
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1994, 2019, 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
@@ -204,76 +204,6 @@ rec_set_bit_field_2(
 }
 
 /******************************************************//**
-The following function is used to get the pointer of the next chained record
-on the same page.
-@return pointer to the next chained record, or NULL if none */
-UNIV_INLINE
-const rec_t*
-rec_get_next_ptr_const(
-/*===================*/
-	const rec_t*	rec,	/*!< in: physical record */
-	ulint		comp)	/*!< in: nonzero=compact page format */
-{
-	ulint	field_value;
-
-	compile_time_assert(REC_NEXT_MASK == 0xFFFFUL);
-	compile_time_assert(REC_NEXT_SHIFT == 0);
-
-	field_value = mach_read_from_2(rec - REC_NEXT);
-
-	if (field_value == 0) {
-
-		return(NULL);
-	}
-
-	if (comp) {
-#if UNIV_PAGE_SIZE_MAX <= 32768
-		/* Note that for 64 KiB pages, field_value can 'wrap around'
-		and the debug assertion is not valid */
-
-		/* In the following assertion, field_value is interpreted
-		as signed 16-bit integer in 2's complement arithmetics.
-		If all platforms defined int16_t in the standard headers,
-		the expression could be written simpler as
-		(int16_t) field_value + ut_align_offset(...) < srv_page_size
-		*/
-		ut_ad((field_value >= 32768
-		       ? field_value - 65536
-		       : field_value)
-		      + ut_align_offset(rec, srv_page_size)
-		      < srv_page_size);
-#endif
-		/* There must be at least REC_N_NEW_EXTRA_BYTES + 1
-		between each record. */
-		ut_ad((field_value > REC_N_NEW_EXTRA_BYTES
-		       && field_value < 32768)
-		      || field_value < (uint16) -REC_N_NEW_EXTRA_BYTES);
-
-		return((byte*) ut_align_down(rec, srv_page_size)
-		       + ut_align_offset(rec + field_value, srv_page_size));
-	} else {
-		ut_ad(field_value < srv_page_size);
-
-		return((byte*) ut_align_down(rec, srv_page_size)
-		       + field_value);
-	}
-}
-
-/******************************************************//**
-The following function is used to get the pointer of the next chained record
-on the same page.
-@return pointer to the next chained record, or NULL if none */
-UNIV_INLINE
-rec_t*
-rec_get_next_ptr(
-/*=============*/
-	rec_t*	rec,	/*!< in: physical record */
-	ulint	comp)	/*!< in: nonzero=compact page format */
-{
-	return(const_cast<rec_t*>(rec_get_next_ptr_const(rec, comp)));
-}
-
-/******************************************************//**
 The following function is used to get the offset of the next chained record
 on the same page.
 @return the page offset of the next chained record, or 0 if none */
diff --git a/storage/innobase/lock/lock0lock.cc b/storage/innobase/lock/lock0lock.cc
index d5905e7b523..76ad538ea0e 100644
--- a/storage/innobase/lock/lock0lock.cc
+++ b/storage/innobase/lock/lock0lock.cc
@@ -2378,10 +2378,12 @@ lock_move_reorganize_page(
 
 	lock_trx->mutex_unlock();
 
-        if (new_heap_no == PAGE_HEAP_NO_SUPREMUM)
+        if (!rec1 || !rec2)
         {
-           ut_ad(old_heap_no == PAGE_HEAP_NO_SUPREMUM);
-           break;
+          ut_ad(!rec1 == !rec2);
+          ut_ad(new_heap_no == PAGE_HEAP_NO_SUPREMUM);
+          ut_ad(old_heap_no == PAGE_HEAP_NO_SUPREMUM);
+          break;
         }
       }
 
@@ -2450,6 +2452,12 @@ lock_move_rec_list_end(
                                     FALSE);
       }
 
+      if (UNIV_UNLIKELY(!rec1 || !rec2))
+      {
+        ut_ad("corrupted page" == 0);
+        return;
+      }
+
       /* Copy lock requests on user records to new page and
       reset the lock bits on the old */
       for (;;)
@@ -2463,28 +2471,37 @@ lock_move_rec_list_end(
         if (comp)
         {
           rec1_heap_no= rec_get_heap_no_new(rec1);
-          if (rec1_heap_no == PAGE_HEAP_NO_SUPREMUM)
+          if (!(rec1= page_rec_get_next_low(rec1, TRUE)))
+          {
+            ut_ad(rec1_heap_no == PAGE_HEAP_NO_SUPREMUM);
             break;
-
+          }
           rec2_heap_no= rec_get_heap_no_new(rec2);
-          rec1= page_rec_get_next_low(rec1, TRUE);
           rec2= page_rec_get_next_low(rec2, TRUE);
         }
         else
         {
+          ut_d(const rec_t *old1= rec1);
           rec1_heap_no= rec_get_heap_no_old(rec1);
-
-          if (rec1_heap_no == PAGE_HEAP_NO_SUPREMUM)
+          if (!(rec1= page_rec_get_next_low(rec1, FALSE)))
+          {
+            ut_ad(rec1_heap_no == PAGE_HEAP_NO_SUPREMUM);
             break;
-          rec2_heap_no= rec_get_heap_no_old(rec2);
+          }
 
-          ut_ad(rec_get_data_size_old(rec1) == rec_get_data_size_old(rec2));
-          ut_ad(!memcmp(rec1, rec2, rec_get_data_size_old(rec1)));
+          ut_ad(rec_get_data_size_old(old1) == rec_get_data_size_old(rec2));
+          ut_ad(!memcmp(old1, rec2, rec_get_data_size_old(old1)));
 
-          rec1= page_rec_get_next_low(rec1, FALSE);
+          rec2_heap_no= rec_get_heap_no_old(rec2);
           rec2= page_rec_get_next_low(rec2, FALSE);
         }
 
+        if (UNIV_UNLIKELY(!rec2))
+        {
+          ut_ad("corrupted page" == 0);
+          return;
+        }
+
         trx_t *lock_trx= lock->trx;
         lock_trx->mutex_lock();
 
@@ -2577,6 +2594,12 @@ lock_move_rec_list_start(
 
       while (rec1 != rec)
       {
+        if (UNIV_UNLIKELY(!rec1 || !rec2))
+        {
+          ut_ad("corrupted page" == 0);
+          return;
+        }
+
         ut_ad(page_rec_is_metadata(rec1) == page_rec_is_metadata(rec2));
         ut_d(const rec_t* const prev= rec1);
 
@@ -2880,11 +2903,15 @@ void lock_update_merge_left(const buf_block_t& left, const rec_t *orig_pred,
   ut_ad(left.page.frame == page_align(orig_pred));
 
   const page_id_t l{left.page.id()};
+  const rec_t *left_next_rec= page_rec_get_next_const(orig_pred);
+  if (UNIV_UNLIKELY(!left_next_rec))
+  {
+    ut_ad("corrupted page" == 0);
+    return;
+  }
 
   /* This would likely be too large for a memory transaction. */
   LockMultiGuard g{lock_sys.rec_hash, l, right};
-  const rec_t *left_next_rec= page_rec_get_next_const(orig_pred);
-
   if (!page_rec_is_supremum(left_next_rec))
   {
     /* Inherit the locks on the supremum of the left page to the
@@ -3031,12 +3058,18 @@ lock_update_insert(
 
 	if (page_rec_is_comp(rec)) {
 		receiver_heap_no = rec_get_heap_no_new(rec);
-		donator_heap_no = rec_get_heap_no_new(
-			page_rec_get_next_low(rec, TRUE));
+		rec = page_rec_get_next_low(rec, TRUE);
+		if (UNIV_UNLIKELY(!rec)) {
+			return;
+		}
+		donator_heap_no = rec_get_heap_no_new(rec);
 	} else {
 		receiver_heap_no = rec_get_heap_no_old(rec);
-		donator_heap_no = rec_get_heap_no_old(
-			page_rec_get_next_low(rec, FALSE));
+		rec = page_rec_get_next_low(rec, FALSE);
+		if (UNIV_UNLIKELY(!rec)) {
+			return;
+		}
+		donator_heap_no = rec_get_heap_no_old(rec);
 	}
 
 	lock_rec_inherit_to_gap_if_gap_lock(
@@ -4952,13 +4985,15 @@ lock_rec_insert_check_and_lock(
   ut_ad(page_is_leaf(block->page.frame));
   ut_ad(!index->table->is_temporary());
 
+  const rec_t *next_rec= page_rec_get_next_const(rec);
+  if (UNIV_UNLIKELY(!next_rec || rec_is_metadata(next_rec, *index)))
+    return DB_CORRUPTION;
+
   dberr_t err= DB_SUCCESS;
   bool inherit_in= *inherit;
   trx_t *trx= thr_get_trx(thr);
-  const rec_t *next_rec= page_rec_get_next_const(rec);
   ulint heap_no= page_rec_get_heap_no(next_rec);
   const page_id_t id{block->page.id()};
-  ut_ad(!rec_is_metadata(next_rec, *index));
 
   {
     LockGuard g{lock_sys.rec_hash, id};
@@ -6126,6 +6161,7 @@ namespace Deadlock
       for (trx_t *next= cycle;;)
       {
         next= next->lock.wait_trx;
+        l++;
         const undo_no_t next_weight= TRX_WEIGHT(next) |
           (next->mysql_thd &&
 #ifdef WITH_WSREP
@@ -6332,11 +6368,16 @@ void lock_update_split_and_merge(
 
   const page_id_t l{left_block->page.id()};
   const page_id_t r{right_block->page.id()};
+  const rec_t *left_next_rec= page_rec_get_next_const(orig_pred);
+  if (UNIV_UNLIKELY(!left_next_rec))
+  {
+    ut_ad("corrupted page" == 0);
+    return;
+  }
+  ut_ad(!page_rec_is_metadata(left_next_rec));
 
   /* This would likely be too large for a memory transaction. */
   LockMultiGuard g{lock_sys.rec_hash, l, r};
-  const rec_t *left_next_rec= page_rec_get_next_const(orig_pred);
-  ut_ad(!page_rec_is_metadata(left_next_rec));
 
   /* Inherit the locks on the supremum of the left page to the
   first record which was moved from the right page */
diff --git a/storage/innobase/log/log0recv.cc b/storage/innobase/log/log0recv.cc
index 68a051d8600..61f7d3961d2 100644
--- a/storage/innobase/log/log0recv.cc
+++ b/storage/innobase/log/log0recv.cc
@@ -727,7 +727,7 @@ static struct
   @param id   tablespace id
   @return tablespace whose creation was deferred
   @retval nullptr if no such tablespace was found */
-  const item *find(uint32_t id)
+  item *find(uint32_t id)
   {
     mysql_mutex_assert_owner(&recv_sys.mutex);
     auto it= defers.find(id);
@@ -1195,11 +1195,11 @@ inline size_t recv_sys_t::files_size()
 @param[in]	name		file name
 @param[in]	len		length of the file name
 @param[in]	space_id	the tablespace ID
-@param[in]	deleted		whether this is a FILE_DELETE record
+@param[in]	ftype		FILE_MODIFY, FILE_DELETE, or FILE_RENAME
 @param[in]	lsn		lsn of the redo log
 @param[in]	store		whether the redo log has to be stored */
 static void fil_name_process(const char *name, ulint len, uint32_t space_id,
-                             bool deleted, lsn_t lsn, const store_t *store)
+                             mfile_type_t ftype, lsn_t lsn, store_t store)
 {
 	if (srv_operation == SRV_OPERATION_BACKUP
 	    || srv_operation == SRV_OPERATION_BACKUP_NO_DEFER) {
@@ -1214,6 +1214,7 @@ static void fil_name_process(const char *name, ulint len, uint32_t space_id,
 	further checks can ensure that a FILE_MODIFY record was
 	scanned before applying any page records for the space_id. */
 
+	const bool deleted{ftype == FILE_DELETE};
 	const file_name_t fname(std::string(name, len), deleted);
 	std::pair<recv_spaces_t::iterator,bool> p = recv_spaces.emplace(
 		space_id, fname);
@@ -1226,6 +1227,9 @@ static void fil_name_process(const char *name, ulint len, uint32_t space_id,
 			d->deleted = true;
 			goto got_deleted;
 		}
+		if (ftype == FILE_RENAME) {
+			d->file_name= fname.name;
+		}
 		goto reload;
 	}
 
@@ -1250,7 +1254,7 @@ reload:
 		the space_id. If not, ignore the file after displaying
 		a note. Abort if there are multiple files with the
 		same space_id. */
-		switch (fil_ibd_load(space_id, name, space)) {
+		switch (fil_ibd_load(space_id, fname.name.c_str(), space)) {
 		case FIL_LOAD_OK:
 			ut_ad(space != NULL);
 
@@ -1310,10 +1314,11 @@ same_space:
 			break;
 
 		case FIL_LOAD_DEFER:
-			/** Skip the deferred spaces
+			/* Skip the deferred spaces
 			when lsn is already processed */
-			if (*store != store_t::STORE_IF_EXISTS) {
-				deferred_spaces.add(space_id, name, lsn);
+			if (store != store_t::STORE_IF_EXISTS) {
+				deferred_spaces.add(
+					space_id, fname.name.c_str(), lsn);
 			}
 			break;
 		case FIL_LOAD_INVALID:
@@ -2644,28 +2649,29 @@ same_page:
         if (fnend - fn < 4 || memcmp(fnend - 4, DOT_IBD, 4))
           goto file_rec_error;
 
-        const char saved_end= fn[rlen];
-        const_cast<char&>(fn[rlen])= '\0';
-        fil_name_process(const_cast<char*>(fn), fnend - fn, space_id,
-                         (b & 0xf0) == FILE_DELETE, start_lsn,
-                         store);
-        if (fn2)
-          fil_name_process(const_cast<char*>(fn2), fn2end - fn2, space_id,
-                           false, start_lsn, store);
+        fil_name_process(fn, fnend - fn, space_id,
+                         (b & 0xf0) == FILE_DELETE ? FILE_DELETE : FILE_MODIFY,
+                         start_lsn, *store);
+
         if ((b & 0xf0) < FILE_CHECKPOINT && log_file_op)
           log_file_op(space_id, b & 0xf0,
                       l, static_cast<ulint>(fnend - fn),
                       reinterpret_cast<const byte*>(fn2),
                       fn2 ? static_cast<ulint>(fn2end - fn2) : 0);
-        const_cast<char&>(fn[rlen])= saved_end;
 
-        if (fn2 && apply)
+        if (fn2)
         {
-          const size_t len= fn2end - fn2;
-          auto r= renamed_spaces.emplace(space_id, std::string{fn2, len});
-          if (!r.second)
-            r.first->second= std::string{fn2, len};
+          fil_name_process(fn2, fn2end - fn2, space_id,
+                           FILE_RENAME, start_lsn, *store);
+          if (apply)
+          {
+            const size_t len= fn2end - fn2;
+            auto r= renamed_spaces.emplace(space_id, std::string{fn2, len});
+            if (!r.second)
+              r.first->second= std::string{fn2, len};
+          }
         }
+
         if (is_corrupt_fs())
           return true;
       }
diff --git a/storage/innobase/page/page0cur.cc b/storage/innobase/page/page0cur.cc
index e608abaa361..5f7192675dc 100644
--- a/storage/innobase/page/page0cur.cc
+++ b/storage/innobase/page/page0cur.cc
@@ -88,7 +88,10 @@ page_cur_try_search_shortcut(
 		goto exit_func;
 	}
 
-	next_rec = page_rec_get_next_const(rec);
+	if (!(next_rec = page_rec_get_next_const(rec))) {
+		goto exit_func;
+	}
+
 	if (!page_rec_is_supremum(next_rec)) {
 		offsets = rec_get_offsets(next_rec, index, offsets,
 					  index->n_core_fields,
@@ -179,7 +182,10 @@ page_cur_try_search_shortcut_bytes(
 		goto exit_func;
 	}
 
-	next_rec = page_rec_get_next_const(rec);
+	if (!(next_rec = page_rec_get_next_const(rec))) {
+		goto exit_func;
+	}
+
 	if (!page_rec_is_supremum(next_rec)) {
 		offsets = rec_get_offsets(next_rec, index, offsets,
 					  index->n_core_fields,
@@ -267,7 +273,7 @@ page_cur_rec_field_extends(
 
 /****************************************************************//**
 Searches the right position for a page cursor. */
-void
+bool
 page_cur_search_with_match(
 /*=======================*/
 	const buf_block_t*	block,	/*!< in: buffer block */
@@ -289,7 +295,6 @@ page_cur_search_with_match(
 	ulint		low;
 	ulint		mid;
 	const page_t*	page;
-	const page_dir_slot_t* slot;
 	const rec_t*	up_rec;
 	const rec_t*	low_rec;
 	const rec_t*	mid_rec;
@@ -335,7 +340,7 @@ page_cur_search_with_match(
 	    && page_cur_try_search_shortcut(
 		    block, index, tuple,
 		    iup_matched_fields, ilow_matched_fields, cursor)) {
-		return;
+		return false;
 	}
 # ifdef PAGE_CUR_DBG
 	if (mode == PAGE_CUR_DBG) {
@@ -352,10 +357,9 @@ page_cur_search_with_match(
 		if (mode == PAGE_CUR_RTREE_INSERT && n_core) {
 			mode = PAGE_CUR_LE;
 		} else {
-			rtr_cur_search_with_match(
+			return rtr_cur_search_with_match(
 				block, (dict_index_t*)index, tuple, mode,
 				cursor, rtr_info);
-			return;
 		}
 	}
 
@@ -386,9 +390,11 @@ page_cur_search_with_match(
 
 	while (up - low > 1) {
 		mid = (low + up) / 2;
-		slot = page_dir_get_nth_slot(page, mid);
-		mid_rec = page_dir_slot_get_rec(slot);
-
+		const page_dir_slot_t* slot = page_dir_get_nth_slot(page, mid);
+		if (UNIV_UNLIKELY(!(mid_rec
+				    = page_dir_slot_get_rec_validate(slot)))) {
+			goto corrupted;
+		}
 		cur_matched_fields = std::min(low_matched_fields,
 					      up_matched_fields);
 
@@ -431,18 +437,30 @@ up_slot_match:
 		}
 	}
 
-	slot = page_dir_get_nth_slot(page, low);
-	low_rec = page_dir_slot_get_rec(slot);
-	slot = page_dir_get_nth_slot(page, up);
-	up_rec = page_dir_slot_get_rec(slot);
+	low_rec = page_dir_slot_get_rec_validate(
+		page_dir_get_nth_slot(page, low));
+	up_rec = page_dir_slot_get_rec_validate(
+		page_dir_get_nth_slot(page, up));
+	if (UNIV_UNLIKELY(!low_rec || !up_rec)) {
+corrupted:
+		if (UNIV_LIKELY_NULL(heap)) {
+			mem_heap_free(heap);
+		}
+		return true;
+	}
 
 	/* Perform linear search until the upper and lower records come to
 	distance 1 of each other. */
 
-	while (page_rec_get_next_const(low_rec) != up_rec) {
-
-		mid_rec = page_rec_get_next_const(low_rec);
-
+	for (;;) {
+		if (const rec_t* next = page_rec_get_next_const(low_rec)) {
+			if (next == up_rec) {
+				break;
+			}
+			mid_rec = next;
+		} else {
+			goto corrupted;
+		}
 		cur_matched_fields = std::min(low_matched_fields,
 					      up_matched_fields);
 
@@ -512,6 +530,8 @@ up_rec_match:
 	if (UNIV_LIKELY_NULL(heap)) {
 		mem_heap_free(heap);
 	}
+
+	return false;
 }
 
 #ifdef BTR_CUR_HASH_ADAPT
@@ -529,7 +549,7 @@ lower limit record
 @param[in,out]	ilow_matched_bytes	already matched bytes in the
 first partially matched field in the lower limit record
 @param[out]	cursor			page cursor */
-void
+bool
 page_cur_search_with_match_bytes(
 	const buf_block_t*	block,
 	const dict_index_t*	index,
@@ -543,9 +563,7 @@ page_cur_search_with_match_bytes(
 {
 	ulint		up;
 	ulint		low;
-	ulint		mid;
 	const page_t*	page;
-	const page_dir_slot_t* slot;
 	const rec_t*	up_rec;
 	const rec_t*	low_rec;
 	const rec_t*	mid_rec;
@@ -594,7 +612,7 @@ page_cur_search_with_match_bytes(
 		    iup_matched_fields, iup_matched_bytes,
 		    ilow_matched_fields, ilow_matched_bytes,
 		    cursor)) {
-		return;
+		return false;
 	}
 # ifdef PAGE_CUR_DBG
 	if (mode == PAGE_CUR_DBG) {
@@ -632,9 +650,12 @@ page_cur_search_with_match_bytes(
 	const ulint n_core = page_is_leaf(page) ? index->n_core_fields : 0;
 
 	while (up - low > 1) {
-		mid = (low + up) / 2;
-		slot = page_dir_get_nth_slot(page, mid);
-		mid_rec = page_dir_slot_get_rec(slot);
+		const ulint mid = (low + up) / 2;
+		mid_rec = page_dir_slot_get_rec_validate(
+			page_dir_get_nth_slot(page, mid));
+		if (UNIV_UNLIKELY(!mid_rec)) {
+			goto corrupted;
+		}
 
 		ut_pair_min(&cur_matched_fields, &cur_matched_bytes,
 			    low_matched_fields, low_matched_bytes,
@@ -681,18 +702,30 @@ up_slot_match:
 		}
 	}
 
-	slot = page_dir_get_nth_slot(page, low);
-	low_rec = page_dir_slot_get_rec(slot);
-	slot = page_dir_get_nth_slot(page, up);
-	up_rec = page_dir_slot_get_rec(slot);
+	low_rec = page_dir_slot_get_rec_validate(
+		page_dir_get_nth_slot(page, low));
+	up_rec = page_dir_slot_get_rec_validate(
+		page_dir_get_nth_slot(page, up));
+	if (UNIV_UNLIKELY(!low_rec || !up_rec)) {
+corrupted:
+		if (UNIV_LIKELY_NULL(heap)) {
+			mem_heap_free(heap);
+		}
+		return true;
+	}
 
 	/* Perform linear search until the upper and lower records come to
 	distance 1 of each other. */
 
-	while (page_rec_get_next_const(low_rec) != up_rec) {
-
-		mid_rec = page_rec_get_next_const(low_rec);
-
+	for (;;) {
+		if (const rec_t* next = page_rec_get_next_const(low_rec)) {
+			if (next == up_rec) {
+				break;
+			}
+			mid_rec = next;
+		} else {
+			goto corrupted;
+		}
 		ut_pair_min(&cur_matched_fields, &cur_matched_bytes,
 			    low_matched_fields, low_matched_bytes,
 			    up_matched_fields, up_matched_bytes);
@@ -761,6 +794,7 @@ up_rec_match:
 	if (UNIV_LIKELY_NULL(heap)) {
 		mem_heap_free(heap);
 	}
+	return false;
 }
 #endif /* BTR_CUR_HASH_ADAPT */
 
@@ -773,17 +807,12 @@ page_cur_open_on_rnd_user_rec(
 	buf_block_t*	block,	/*!< in: page */
 	page_cur_t*	cursor)	/*!< out: page cursor */
 {
-	const ulint	n_recs = page_get_n_recs(block->page.frame);
-
-	page_cur_set_before_first(block, cursor);
-
-	if (UNIV_UNLIKELY(n_recs == 0)) {
-
-		return;
-	}
-
-	cursor->rec = page_rec_get_nth(block->page.frame,
-				       ut_rnd_interval(n_recs) + 1);
+  cursor->block= block;
+  if (const ulint n_recs= page_get_n_recs(block->page.frame))
+    if ((cursor->rec= page_rec_get_nth(block->page.frame,
+                                       ut_rnd_interval(n_recs) + 1)))
+      return;
+  cursor->rec= page_get_infimum_rec(block->page.frame);
 }
 
 /**
@@ -802,7 +831,7 @@ static void page_rec_set_n_owned(rec_t *rec, ulint n_owned, bool comp)
 Split a directory slot which owns too many records.
 @param[in,out]  block   index page
 @param[in,out]  slot    the slot that needs to be split */
-static void page_dir_split_slot(const buf_block_t &block,
+static bool page_dir_split_slot(const buf_block_t &block,
                                 page_dir_slot_t *slot)
 {
   ut_ad(slot <= &block.page.frame[srv_page_size - PAGE_EMPTY_DIR_START]);
@@ -815,10 +844,17 @@ static void page_dir_split_slot(const buf_block_t &block,
                 PAGE_DIR_SLOT_MIN_N_OWNED, "compatibility");
 
   /* Find a record approximately in the middle. */
-  const rec_t *rec= page_dir_slot_get_rec(slot + PAGE_DIR_SLOT_SIZE);
+  const rec_t *rec= page_dir_slot_get_rec_validate(slot + PAGE_DIR_SLOT_SIZE);
 
   for (ulint i= n_owned / 2; i--; )
+  {
+    if (UNIV_UNLIKELY(!rec))
+      return true;
     rec= page_rec_get_next_const(rec);
+  }
+
+  if (UNIV_UNLIKELY(!rec))
+    return true;
 
   /* Add a directory slot immediately below this one. */
   constexpr uint16_t n_slots_f= PAGE_N_DIR_SLOTS + PAGE_HEADER;
@@ -828,7 +864,10 @@ static void page_dir_split_slot(const buf_block_t &block,
   page_dir_slot_t *last_slot= static_cast<page_dir_slot_t*>
     (block.page.frame + srv_page_size - (PAGE_DIR + PAGE_DIR_SLOT_SIZE) -
      n_slots * PAGE_DIR_SLOT_SIZE);
-  ut_ad(slot >= last_slot);
+
+  if (UNIV_UNLIKELY(slot < last_slot))
+    return true;
+
   memmove_aligned<2>(last_slot, last_slot + PAGE_DIR_SLOT_SIZE,
                      slot - last_slot);
 
@@ -841,6 +880,7 @@ static void page_dir_split_slot(const buf_block_t &block,
   page_rec_set_n_owned(page_dir_slot_get_rec(slot), half_owned, comp);
   page_rec_set_n_owned(page_dir_slot_get_rec(slot - PAGE_DIR_SLOT_SIZE),
                        n_owned - half_owned, comp);
+  return false;
 }
 
 /**
@@ -866,6 +906,10 @@ static void page_zip_dir_split_slot(buf_block_t *block, ulint s, mtr_t* mtr)
 
   const rec_t *rec= page_dir_slot_get_rec(slot + PAGE_DIR_SLOT_SIZE);
 
+  /* We do not try to prevent crash on corruption here.
+  For ROW_FORMAT=COMPRESSED pages, the next-record links should
+  be validated in page_zip_decompress(). Corruption should only
+  be possible here if the buffer pool was corrupted later. */
   for (ulint i= n_owned / 2; i--; )
     rec= page_rec_get_next_const(rec);
 
@@ -951,8 +995,12 @@ static void page_zip_dir_balance_slot(buf_block_t *block, ulint s, mtr_t *mtr)
 
 	/* Transfer one record to the underfilled slot */
 	page_rec_set_n_owned<true>(block, slot_rec, 0, true, mtr);
-	rec_t* new_rec = rec_get_next_ptr(slot_rec, TRUE);
-	page_rec_set_n_owned<true>(block, new_rec,
+	const rec_t* new_rec = page_rec_get_next_low(slot_rec, TRUE);
+	/* We do not try to prevent crash on corruption here.
+	For ROW_FORMAT=COMPRESSED pages, the next-record links should
+	be validated in page_zip_decompress(). Corruption should only
+	be possible here if the buffer pool was corrupted later. */
+	page_rec_set_n_owned<true>(block, const_cast<rec_t*>(new_rec),
 				   PAGE_DIR_SLOT_MIN_N_OWNED,
 				   true, mtr);
 	mach_write_to_2(slot, page_offset(new_rec));
@@ -1013,18 +1061,27 @@ static void page_dir_balance_slot(const buf_block_t &block, ulint s)
 	}
 
 	/* Transfer one record to the underfilled slot */
-	rec_t* new_rec;
+	const rec_t* new_rec;
 
 	if (comp) {
+		if (UNIV_UNLIKELY(!(new_rec =
+				    page_rec_get_next_low(slot_rec, true)))) {
+			ut_ad("corrupted page" == 0);
+			return;
+		}
 		page_rec_set_n_owned(slot_rec, 0, true);
-		new_rec = rec_get_next_ptr(slot_rec, TRUE);
-		page_rec_set_n_owned(new_rec, PAGE_DIR_SLOT_MIN_N_OWNED, true);
+		page_rec_set_n_owned(const_cast<rec_t*>(new_rec),
+				     PAGE_DIR_SLOT_MIN_N_OWNED, true);
 		page_rec_set_n_owned(up_rec, up_n_owned - 1, true);
 	} else {
+		if (UNIV_UNLIKELY(!(new_rec =
+				    page_rec_get_next_low(slot_rec, false)))) {
+			ut_ad("corrupted page" == 0);
+			return;
+		}
 		page_rec_set_n_owned(slot_rec, 0, false);
-		new_rec = rec_get_next_ptr(slot_rec, FALSE);
-		page_rec_set_n_owned(new_rec, PAGE_DIR_SLOT_MIN_N_OWNED,
-				     false);
+		page_rec_set_n_owned(const_cast<rec_t*>(new_rec),
+				     PAGE_DIR_SLOT_MIN_N_OWNED, false);
 		page_rec_set_n_owned(up_rec, up_n_owned - 1, false);
 	}
 
@@ -1281,6 +1338,20 @@ inline void mtr_t::page_insert(const buf_block_t &block, bool reuse,
   m_last_offset= FIL_PAGE_TYPE;
 }
 
+/** Report page directory corruption.
+@param block  index page
+@param index  index tree
+*/
+ATTRIBUTE_COLD
+static void page_cur_directory_corrupted(const buf_block_t &block,
+                                         const dict_index_t &index)
+{
+  ib::error() << "Directory of " << block.page.id()
+              << " of index " << index.name
+              << " in table " << index.table->name
+              << " is corrupted";
+}
+
 /***********************************************************//**
 Inserts a record next to page cursor on an uncompressed page.
 @return pointer to record
@@ -1623,9 +1694,14 @@ copied:
   {
     const ulint owner= page_dir_find_owner_slot(next_rec);
     if (UNIV_UNLIKELY(owner == ULINT_UNDEFINED))
+    {
+      page_cur_directory_corrupted(*block, *index);
+      return nullptr;
+    }
+
+    if (page_dir_split_slot(*block, page_dir_get_nth_slot(block->page.frame,
+                                                          owner)))
       return nullptr;
-    page_dir_split_slot(*block,
-                        page_dir_get_nth_slot(block->page.frame, owner));
   }
 
   rec_offs_make_valid(insert_buf + extra_size, index,
@@ -1693,20 +1769,20 @@ static inline void page_zip_dir_add_slot(buf_block_t *block,
 
 /***********************************************************//**
 Inserts a record next to page cursor on a compressed and uncompressed
-page. Returns pointer to inserted record if succeed, i.e.,
-enough space available, NULL otherwise.
-The cursor stays at the same position.
+page.
 
 IMPORTANT: The caller will have to update IBUF_BITMAP_FREE
 if this is a compressed leaf page in a secondary index.
 This has to be done either within the same mini-transaction,
 or by invoking ibuf_reset_free_bits() before mtr_commit().
 
-@return pointer to record if succeed, NULL otherwise */
+@return pointer to inserted record
+@return nullptr on failure */
 rec_t*
 page_cur_insert_rec_zip(
 /*====================*/
-	page_cur_t*	cursor,	/*!< in/out: page cursor */
+	page_cur_t*	cursor,	/*!< in/out: page cursor,
+				logical position unchanged  */
 	dict_index_t*	index,	/*!< in: record descriptor */
 	const rec_t*	rec,	/*!< in: pointer to a physical record */
 	rec_offs*	offsets,/*!< in/out: rec_get_offsets(rec, index) */
@@ -1786,6 +1862,9 @@ page_cur_insert_rec_zip(
     {
       ulint pos= page_rec_get_n_recs_before(cursor->rec);
 
+      if (UNIV_UNLIKELY(pos == ULINT_UNDEFINED))
+        return nullptr;
+
       switch (page_zip_reorganize(cursor->block, index, level, mtr, true)) {
       case DB_FAIL:
         ut_ad(cursor->rec == cursor_rec);
@@ -1796,10 +1875,13 @@ page_cur_insert_rec_zip(
         return nullptr;
       }
 
-      if (pos)
-        cursor->rec= page_rec_get_nth(page, pos);
-      else
-        ut_ad(cursor->rec == page_get_infimum_rec(page));
+      if (!pos)
+        ut_ad(cursor->rec == page + PAGE_NEW_INFIMUM);
+      else if (!(cursor->rec= page_rec_get_nth(page, pos)))
+      {
+        cursor->rec= page + PAGE_NEW_SUPREMUM;
+        return nullptr;
+      }
 
       ut_ad(!page_header_get_ptr(page, PAGE_FREE));
 
@@ -1816,16 +1898,21 @@ page_cur_insert_rec_zip(
     if (insert_rec)
     {
       ulint pos= page_rec_get_n_recs_before(insert_rec);
-      ut_ad(pos > 0);
+      if (UNIV_UNLIKELY(!pos || pos == ULINT_UNDEFINED))
+        return nullptr;
 
       /* We are writing entire page images to the log.  Reduce the redo
       log volume by reorganizing the page at the same time. */
       switch (page_zip_reorganize(cursor->block, index, level, mtr)) {
       case DB_SUCCESS:
         /* The page was reorganized: Seek to pos. */
-        cursor->rec= pos > 1
-          ? page_rec_get_nth(page, pos - 1)
-          : page + PAGE_NEW_INFIMUM;
+        if (pos <= 1)
+          cursor->rec= page + PAGE_NEW_INFIMUM;
+        else if (!(cursor->rec= page_rec_get_nth(page, pos - 1)))
+        {
+          cursor->rec= page + PAGE_NEW_INFIMUM;
+          return nullptr;
+        }
         insert_rec= page + rec_get_next_offs(cursor->rec, 1);
         rec_offs_make_valid(insert_rec, index, page_is_leaf(page), offsets);
         break;
@@ -1891,19 +1978,25 @@ too_small:
 
     byte *const free_rec_ptr= page + free_rec;
     heap_no= rec_get_heap_no_new(free_rec_ptr);
-    int16_t next_rec= mach_read_from_2(free_rec_ptr - REC_NEXT);
+    int16_t next_free= mach_read_from_2(free_rec_ptr - REC_NEXT);
     /* With innodb_page_size=64k, int16_t would be unsafe to use here,
     but that cannot be used with ROW_FORMAT=COMPRESSED. */
     static_assert(UNIV_ZIP_SIZE_SHIFT_MAX == 14, "compatibility");
-    if (next_rec)
+    if (next_free)
     {
-      next_rec= static_cast<int16_t>(next_rec + free_rec);
-      ut_ad(int{PAGE_NEW_SUPREMUM_END + REC_N_NEW_EXTRA_BYTES} <= next_rec);
-      ut_ad(static_cast<uint16_t>(next_rec) < srv_page_size);
+      next_free= static_cast<int16_t>(next_free + free_rec);
+      if (UNIV_UNLIKELY(int{PAGE_NEW_SUPREMUM_END + REC_N_NEW_EXTRA_BYTES} >
+                        next_free ||
+                        uint16_t(next_free) >= srv_page_size))
+      {
+        if (UNIV_LIKELY_NULL(heap))
+          mem_heap_free(heap);
+        return nullptr;
+      }
     }
 
     byte *hdr= my_assume_aligned<4>(&page_zip->data[page_free_f]);
-    mach_write_to_2(hdr, static_cast<uint16_t>(next_rec));
+    mach_write_to_2(hdr, static_cast<uint16_t>(next_free));
     const byte *const garbage= my_assume_aligned<2>(page_free + 2);
     ut_ad(mach_read_from_2(garbage) >= rec_size);
     mach_write_to_2(my_assume_aligned<2>(hdr + 2),
@@ -1959,18 +2052,20 @@ use_heap:
     page_zip_dir_add_slot(cursor->block, index, mtr);
   }
 
+  /* next record after current before the insertion */
+  const rec_t *next_rec = page_rec_get_next_low(cursor->rec, TRUE);
+  if (UNIV_UNLIKELY(!next_rec ||
+                    rec_get_status(next_rec) == REC_STATUS_INFIMUM ||
+                    rec_get_status(cursor->rec) > REC_STATUS_INFIMUM))
+    return nullptr;
+
   /* 3. Create the record */
   byte *insert_rec= rec_copy(insert_buf, rec, offsets);
   rec_offs_make_valid(insert_rec, index, page_is_leaf(page), offsets);
 
   /* 4. Insert the record in the linked list of records */
   ut_ad(cursor->rec != insert_rec);
-
-  /* next record after current before the insertion */
-  const rec_t* next_rec = page_rec_get_next_low(cursor->rec, TRUE);
-  ut_ad(rec_get_status(cursor->rec) <= REC_STATUS_INFIMUM);
   ut_ad(rec_get_status(insert_rec) < REC_STATUS_INFIMUM);
-  ut_ad(rec_get_status(next_rec) != REC_STATUS_INFIMUM);
 
   mach_write_to_2(insert_rec - REC_NEXT, static_cast<uint16_t>
                   (next_rec - insert_rec));
@@ -2037,8 +2132,9 @@ inc_dir:
   /* 7. It remains to update the owner record. */
   ulint n_owned;
 
-  while (!(n_owned = rec_get_n_owned_new(next_rec)))
-    next_rec= page_rec_get_next_low(next_rec, true);
+  while (!(n_owned= rec_get_n_owned_new(next_rec)))
+    if (!(next_rec= page_rec_get_next_low(next_rec, true)))
+      return nullptr;
 
   rec_set_bit_field_1(const_cast<rec_t*>(next_rec), n_owned + 1,
                       REC_NEW_N_OWNED, REC_N_OWNED_MASK, REC_N_OWNED_SHIFT);
@@ -2052,7 +2148,10 @@ inc_dir:
   {
     const ulint owner= page_dir_find_owner_slot(next_rec);
     if (UNIV_UNLIKELY(owner == ULINT_UNDEFINED))
+    {
+      page_cur_directory_corrupted(*cursor->block, *index);
       return nullptr;
+    }
     page_zip_dir_split_slot(cursor->block, owner, mtr);
   }
 
@@ -2185,8 +2284,8 @@ page_cur_delete_rec(
 		However, this could also be a call in
 		btr_cur_pessimistic_update() to delete the only
 		record in the page and to insert another one. */
-		page_cur_move_to_next(cursor);
-		ut_ad(page_cur_is_after_last(cursor));
+		ut_ad(page_rec_is_supremum(page_rec_get_next(cursor->rec)));
+		page_cur_set_after_last(block, cursor);
 		page_create_empty(page_cur_get_block(cursor),
 				  const_cast<dict_index_t*>(index), mtr);
 		return;
@@ -2197,6 +2296,7 @@ page_cur_delete_rec(
 
 	if (UNIV_UNLIKELY(!cur_slot_no || cur_slot_no == ULINT_UNDEFINED)) {
 		/* Avoid crashing due to a corrupted page. */
+		page_cur_directory_corrupted(*block, *index);
 		return;
 	}
 
@@ -2220,11 +2320,16 @@ page_cur_delete_rec(
 
 	while (current_rec != rec) {
 		prev_rec = rec;
-		rec = page_rec_get_next(rec);
+		if (!(rec = page_rec_get_next(rec))) {
+			/* Avoid crashing due to a corrupted page. */
+			return;
+                }
 	}
 
-	page_cur_move_to_next(cursor);
-	next_rec = cursor->rec;
+	if (!(next_rec = page_cur_move_to_next(cursor))) {
+		/* Avoid crashing due to a corrupted page. */
+		return;
+	}
 
 	/* Remove the record from the linked list of records */
 	/* If the deleted record is pointed to by a dir slot, update the
@@ -2546,7 +2651,7 @@ inc_dir:
   mach_write_to_2(page_n_recs, mach_read_from_2(page_n_recs) + 1);
 
   if (UNIV_UNLIKELY(n_owned == PAGE_DIR_SLOT_MAX_N_OWNED))
-    page_dir_split_slot(block, owner_slot);
+    return page_dir_split_slot(block, owner_slot);
   ut_ad(page_simple_validate_old(page));
   return false;
 }
@@ -2771,7 +2876,7 @@ inc_dir:
   mach_write_to_2(page_n_recs, mach_read_from_2(page_n_recs) + 1);
 
   if (UNIV_UNLIKELY(n_owned == PAGE_DIR_SLOT_MAX_N_OWNED))
-    page_dir_split_slot(block, owner_slot);
+    return page_dir_split_slot(block, owner_slot);
   ut_ad(page_simple_validate_new(page));
   return false;
 }
diff --git a/storage/innobase/page/page0page.cc b/storage/innobase/page/page0page.cc
index 22d3983072b..28b63dc4209 100644
--- a/storage/innobase/page/page0page.cc
+++ b/storage/innobase/page/page0page.cc
@@ -99,43 +99,25 @@ page_dir_find_owner_slot(
 
 	if (page_is_comp(page)) {
 		while (rec_get_n_owned_new(r) == 0) {
-			r = rec_get_next_ptr_const(r, TRUE);
-			ut_ad(r >= page + PAGE_NEW_SUPREMUM);
-			ut_ad(r < page + (srv_page_size - PAGE_DIR));
+			r = page_rec_get_next_low(r, true);
+			if (UNIV_UNLIKELY(r < page + PAGE_NEW_SUPREMUM
+					  || r >= slot)) {
+				return ULINT_UNDEFINED;
+			}
 		}
 	} else {
 		while (rec_get_n_owned_old(r) == 0) {
-			r = rec_get_next_ptr_const(r, FALSE);
-			ut_ad(r >= page + PAGE_OLD_SUPREMUM);
-			ut_ad(r < page + (srv_page_size - PAGE_DIR));
+			r = page_rec_get_next_low(r, false);
+			if (UNIV_UNLIKELY(r < page + PAGE_OLD_SUPREMUM
+					  || r >= slot)) {
+				return ULINT_UNDEFINED;
+			}
 		}
 	}
 
-	uint16 rec_offs_bytes = mach_encode_2(ulint(r - page));
-
-	while (UNIV_LIKELY(*(uint16*) slot != rec_offs_bytes)) {
-
+	while (UNIV_LIKELY(*(uint16*) slot
+			   != mach_encode_2(ulint(r - page)))) {
 		if (UNIV_UNLIKELY(slot == first_slot)) {
-			ib::error() << "Probable data corruption on page "
-				<< page_get_page_no(page)
-				<< ". Original record on that page;";
-
-			if (page_is_comp(page)) {
-				fputs("(compact record)", stderr);
-			} else {
-				rec_print_old(stderr, rec);
-			}
-
-			ib::error() << "Cannot find the dir slot for this"
-				" record on that page;";
-
-			if (page_is_comp(page)) {
-				fputs("(compact record)", stderr);
-			} else {
-				rec_print_old(stderr, page
-					      + mach_decode_2(rec_offs_bytes));
-			}
-
 			return ULINT_UNDEFINED;
 		}
 
@@ -478,9 +460,8 @@ page_copy_rec_list_end_no_locks(
 
 	page_cur_position(rec, block, &cur1);
 
-	if (page_cur_is_before_first(&cur1)) {
-
-		page_cur_move_to_next(&cur1);
+	if (page_cur_is_before_first(&cur1) && !page_cur_move_to_next(&cur1)) {
+		return DB_CORRUPTION;
 	}
 
 	if (UNIV_UNLIKELY(page_is_comp(new_page) != page_rec_is_comp(rec)
@@ -504,12 +485,10 @@ page_copy_rec_list_end_no_locks(
 					  ULINT_UNDEFINED, &heap);
 		ins_rec = page_cur_insert_rec_low(&cur2, index,
 						  cur1.rec, offsets, mtr);
-		if (UNIV_UNLIKELY(!ins_rec)) {
+		if (UNIV_UNLIKELY(!ins_rec || !page_cur_move_to_next(&cur1))) {
 			err = DB_CORRUPTION;
 			break;
 		}
-
-		page_cur_move_to_next(&cur1);
 		ut_ad(!(rec_get_info_bits(cur1.rec, page_is_comp(new_page))
 			& REC_INFO_MIN_REC_FLAG));
 		cur2.rec = ins_rec;
@@ -550,10 +529,13 @@ page_copy_rec_list_end(
 	rec_t*		ret		= page_rec_get_next(
 		page_get_infimum_rec(new_page));
 	ulint		num_moved	= 0;
-	rtr_rec_move_t*	rec_move	= NULL;
-	mem_heap_t*	heap		= NULL;
 	ut_ad(page_align(rec) == page);
 
+	if (UNIV_UNLIKELY(!ret)) {
+		*err = DB_CORRUPTION;
+		return nullptr;
+	}
+
 #ifdef UNIV_ZIP_DEBUG
 	if (new_page_zip) {
 		page_zip_des_t*	page_zip = buf_block_get_page_zip(block);
@@ -579,15 +561,15 @@ page_copy_rec_list_end(
 	alignas(2) byte h[PAGE_N_DIRECTION + 2 - PAGE_LAST_INSERT];
 	memcpy_aligned<2>(h, PAGE_HEADER + PAGE_LAST_INSERT + new_page,
 			  sizeof h);
+	mem_heap_t* heap = nullptr;
+	rtr_rec_move_t* rec_move = nullptr;
 
 	if (index->is_spatial()) {
 		ulint	max_to_move = page_get_n_recs(
 			buf_block_get_frame(block));
 		heap = mem_heap_create(256);
-
-		rec_move = static_cast<rtr_rec_move_t*>(
+		rec_move= static_cast<rtr_rec_move_t*>(
 			mem_heap_alloc(heap, max_to_move * sizeof *rec_move));
-
 		/* For spatial index, we need to insert recs one by one
 		to keep recs ordered. */
 		*err = rtr_page_copy_rec_list_end_no_locks(new_block,
@@ -600,6 +582,10 @@ page_copy_rec_list_end(
 		*err = page_copy_rec_list_end_no_locks(new_block, block, rec,
 						       index, mtr);
 		if (UNIV_UNLIKELY(*err != DB_SUCCESS)) {
+err_exit:
+			if (UNIV_LIKELY_NULL(heap)) {
+				mem_heap_free(heap);
+			}
 			return nullptr;
 		}
 		if (was_empty) {
@@ -640,7 +626,11 @@ page_copy_rec_list_end(
 			have at least one predecessor (the predefined
 			infimum record, or a freshly copied record
 			that is smaller than "ret"). */
-			ut_a(ret_pos > 0);
+			if (UNIV_UNLIKELY(!ret_pos
+					  || ret_pos == ULINT_UNDEFINED)) {
+				*err = DB_CORRUPTION;
+				goto err_exit;
+			}
 
 			*err = page_zip_reorganize(new_block, index,
 						   page_zip_level, mtr);
@@ -653,14 +643,12 @@ page_copy_rec_list_end(
 				ut_ad(page_validate(new_page, index));
 				/* fall through */
 			default:
-				if (heap) {
-					mem_heap_free(heap);
-				}
-				return nullptr;
+				goto err_exit;
 			case DB_SUCCESS:
 				/* The page was reorganized:
 				Seek to ret_pos. */
 				ret = page_rec_get_nth(new_page, ret_pos);
+				ut_ad(ret);
 			}
 		}
 	}
@@ -668,13 +656,13 @@ page_copy_rec_list_end(
 	/* Update the lock table and possible hash index */
 
 	if (!index->has_locking()) {
-	} else if (rec_move && dict_index_is_spatial(index)) {
+	} else if (UNIV_LIKELY_NULL(rec_move)) {
 		lock_rtr_move_rec_list(new_block, block, rec_move, num_moved);
 	} else {
 		lock_move_rec_list_end(new_block, block, rec);
 	}
 
-	if (heap) {
+	if (UNIV_LIKELY_NULL(heap)) {
 		mem_heap_free(heap);
 	}
 
@@ -721,8 +709,9 @@ page_copy_rec_list_start(
 	rec_offs_init(offsets_);
 
 	if (UNIV_UNLIKELY(!ret)) {
+corrupted:
 		*err = DB_CORRUPTION;
-		return ret;
+		return nullptr;
 	}
 
 	/* Here, "ret" may be pointing to a user record or the
@@ -733,15 +722,17 @@ page_copy_rec_list_start(
 		return(ret);
 	}
 
+	page_cur_set_before_first(block, &cur1);
+	if (UNIV_UNLIKELY(!page_cur_move_to_next(&cur1))) {
+		goto corrupted;
+	}
+
 	mtr_log_t	log_mode = MTR_LOG_NONE;
 
 	if (new_page_zip) {
 		log_mode = mtr_set_log_mode(mtr, MTR_LOG_NONE);
 	}
 
-	page_cur_set_before_first(block, &cur1);
-	page_cur_move_to_next(&cur1);
-
 	page_cur_position(ret, new_block, &cur2);
 
 	const ulint n_core = page_rec_is_leaf(rec) ? index->n_core_fields : 0;
@@ -775,12 +766,12 @@ page_copy_rec_list_start(
 			cur2.rec = page_cur_insert_rec_low(&cur2, index,
 							   cur1.rec, offsets,
 							   mtr);
-			if (UNIV_UNLIKELY(!cur2.rec)) {
+			if (UNIV_UNLIKELY(!cur2.rec
+					  || !page_cur_move_to_next(&cur1))) {
 				*err = DB_CORRUPTION;
 				return nullptr;
 			}
 
-			page_cur_move_to_next(&cur1);
 			ut_ad(!(rec_get_info_bits(cur1.rec,
 						  page_is_comp(new_page))
 				& REC_INFO_MIN_REC_FLAG));
@@ -821,11 +812,17 @@ zip_reorganize:
 			the predefined infimum record, then it would
 			still be the infimum, and we would have
 			ret_pos == 0. */
+			if (UNIV_UNLIKELY(!ret_pos
+					  || ret_pos == ULINT_UNDEFINED)) {
+				*err = DB_CORRUPTION;
+				return nullptr;
+			}
 			*err = page_zip_reorganize(new_block, index,
 						   page_zip_level, mtr);
 			switch (*err) {
 			case DB_SUCCESS:
 				ret = page_rec_get_nth(new_page, ret_pos);
+				ut_ad(ret);
 				break;
 			case DB_FAIL:
 				if (UNIV_UNLIKELY
@@ -936,7 +933,7 @@ page_delete_rec_list_end(
       page_cur_position(rec, block, &cur);
       offsets= rec_get_offsets(rec, index, offsets, n_core,
 			       ULINT_UNDEFINED, &heap);
-      rec= rec_get_next_ptr(rec, TRUE);
+      rec= const_cast<rec_t*>(page_rec_get_next_low(rec, true));
 #ifdef UNIV_ZIP_DEBUG
       ut_a(page_zip_validate(&block->page.zip, page, index));
 #endif /* UNIV_ZIP_DEBUG */
@@ -981,12 +978,15 @@ page_delete_rec_list_end(
       if (scrub)
         mtr->memset(block, page_offset(rec2), rec_offs_data_size(offsets), 0);
 
-      rec2 = page_rec_get_next(rec2);
+      rec2= page_rec_get_next(rec2);
     }
-    while (!page_rec_is_supremum(rec2));
+    while (rec2 && !page_rec_is_supremum(rec2));
 
     if (UNIV_LIKELY_NULL(heap))
       mem_heap_free(heap);
+
+    if (UNIV_UNLIKELY(!rec))
+      return DB_CORRUPTION;
   }
 
   ut_ad(size < srv_page_size);
@@ -1000,13 +1000,15 @@ page_delete_rec_list_end(
       while (!(n_owned= rec_get_n_owned_new(owner_rec)))
       {
         count++;
-	owner_rec= rec_get_next_ptr_const(owner_rec, TRUE);
+	if (!(owner_rec= page_rec_get_next_low(owner_rec, true)))
+          return DB_CORRUPTION;
       }
     else
       while (!(n_owned= rec_get_n_owned_old(owner_rec)))
       {
         count++;
-	owner_rec= rec_get_next_ptr_const(owner_rec, FALSE);
+	if (!(owner_rec= page_rec_get_next_low(owner_rec, false)))
+          return DB_CORRUPTION;
       }
 
     ut_ad(n_owned > count);
@@ -1133,7 +1135,10 @@ page_delete_rec_list_start(
 	}
 
 	page_cur_set_before_first(block, &cur1);
-	page_cur_move_to_next(&cur1);
+	if (UNIV_UNLIKELY(!page_cur_move_to_next(&cur1))) {
+		ut_ad("corrupted page" == 0);
+		return;
+	}
 
 	const ulint	n_core = page_rec_is_leaf(rec)
 		? index->n_core_fields : 0;
@@ -1153,7 +1158,8 @@ page_delete_rec_list_start(
 /************************************************************//**
 Returns the nth record of the record list.
 This is the inverse function of page_rec_get_n_recs_before().
-@return nth record */
+@return nth record
+@retval nullptr on corrupted page */
 const rec_t*
 page_rec_get_nth_const(
 /*===================*/
@@ -1172,7 +1178,6 @@ page_rec_get_nth_const(
 	ut_ad(nth < srv_page_size / (REC_N_NEW_EXTRA_BYTES + 1));
 
 	for (i = 0;; i++) {
-
 		slot = page_dir_get_nth_slot(page, i);
 		n_owned = page_dir_slot_get_n_owned(slot);
 
@@ -1183,87 +1188,154 @@ page_rec_get_nth_const(
 		}
 	}
 
-	ut_ad(i > 0);
-	slot = page_dir_get_nth_slot(page, i - 1);
-	rec = page_dir_slot_get_rec(slot);
+	if (UNIV_UNLIKELY(!i)) {
+		return nullptr;
+	}
+	rec = page_dir_slot_get_rec(slot + 2);
 
 	if (page_is_comp(page)) {
 		do {
 			rec = page_rec_get_next_low(rec, TRUE);
-			ut_ad(rec);
-		} while (nth--);
+		} while (rec && nth--);
 	} else {
 		do {
 			rec = page_rec_get_next_low(rec, FALSE);
-			ut_ad(rec);
-		} while (nth--);
+		} while (rec && nth--);
 	}
 
 	return(rec);
 }
 
-/***************************************************************//**
-Returns the number of records before the given record in chain.
-The number includes infimum and supremum records.
-@return number of records */
-ulint
-page_rec_get_n_recs_before(
-/*=======================*/
-	const rec_t*	rec)	/*!< in: the physical record */
+
+/************************************************************//**
+Gets the pointer to the previous record.
+@return pointer to previous record
+@retval nullptr on error */
+const rec_t*
+page_rec_get_prev_const(
+/*====================*/
+	const rec_t*	rec)	/*!< in: pointer to record, must not be page
+				infimum */
 {
-	const page_dir_slot_t*	slot;
-	const rec_t*		slot_rec;
-	const page_t*		page;
-	ulint			i;
-	lint			n	= 0;
+	const rec_t*		rec2;
+	const rec_t*		prev_rec = NULL;
 
 	ut_ad(page_rec_check(rec));
 
-	page = page_align(rec);
-	if (page_is_comp(page)) {
-		while (rec_get_n_owned_new(rec) == 0) {
+	const page_t* const page = page_align(rec);
 
-			rec = rec_get_next_ptr_const(rec, TRUE);
-			n--;
-		}
+	ut_ad(!page_rec_is_infimum(rec));
+
+	ulint slot_no = page_dir_find_owner_slot(rec);
 
-		for (i = 0; ; i++) {
-			slot = page_dir_get_nth_slot(page, i);
-			slot_rec = page_dir_slot_get_rec(slot);
+	if (UNIV_UNLIKELY(!slot_no || slot_no == ULINT_UNDEFINED)) {
+		return nullptr;
+	}
 
-			n += lint(rec_get_n_owned_new(slot_rec));
+	const page_dir_slot_t* slot = page_dir_get_nth_slot(page, slot_no - 1);
 
-			if (rec == slot_rec) {
+	if (UNIV_UNLIKELY(!(rec2 = page_dir_slot_get_rec_validate(slot)))) {
+		return nullptr;
+	}
 
+	if (page_is_comp(page)) {
+		while (rec2 && rec != rec2) {
+			prev_rec = rec2;
+			ulint offs = rec_get_next_offs(rec2, TRUE);
+			if (offs < PAGE_NEW_INFIMUM
+			    || offs > page_header_get_field(page,
+							    PAGE_HEAP_TOP)) {
+				return nullptr;
+			}
+			rec2 = page + offs;
+		}
+		switch (rec_get_status(prev_rec)) {
+		case REC_STATUS_INSTANT:
+		case REC_STATUS_ORDINARY:
+			if (!page_is_leaf(page)) {
+				return nullptr;
+			}
+			break;
+		case REC_STATUS_INFIMUM:
+			break;
+		case REC_STATUS_NODE_PTR:
+			if (!page_is_leaf(page)) {
 				break;
 			}
+			/* fall through */
+		default:
+			return nullptr;
 		}
 	} else {
-		while (rec_get_n_owned_old(rec) == 0) {
-
-			rec = rec_get_next_ptr_const(rec, FALSE);
-			n--;
+		while (rec2 && rec != rec2) {
+			prev_rec = rec2;
+			ulint offs = rec_get_next_offs(rec2, FALSE);
+			if (offs < PAGE_OLD_INFIMUM
+			    || offs > page_header_get_field(page,
+							    PAGE_HEAP_TOP)) {
+				return nullptr;
+			}
+			rec2 = page + offs;
 		}
+	}
+
+	return(prev_rec);
+}
 
-		for (i = 0; ; i++) {
-			slot = page_dir_get_nth_slot(page, i);
-			slot_rec = page_dir_slot_get_rec(slot);
+/** Return the number of preceding records in an index page.
+@param rec index record
+@return number of preceding records, including the infimum pseudo-record
+@retval ULINT_UNDEFINED on corrupted page */
+ulint page_rec_get_n_recs_before(const rec_t *rec)
+{
+  const page_t *const page= page_align(rec);
+  const page_dir_slot_t *slot = page_dir_get_nth_slot(page, 0);
+  const page_dir_slot_t *const end_slot= slot - 2 * page_dir_get_n_slots(page);
 
-			n += lint(rec_get_n_owned_old(slot_rec));
+  lint n= 0;
 
-			if (rec == slot_rec) {
+  ut_ad(page_rec_check(rec));
 
-				break;
-			}
-		}
-	}
+  if (page_is_comp(page))
+  {
+    for (; rec_get_n_owned_new(rec) == 0; n--)
+      if (UNIV_UNLIKELY(!(rec= page_rec_get_next_low(rec, true))))
+        return ULINT_UNDEFINED;
 
-	n--;
+    do
+    {
+      const rec_t *slot_rec= page_dir_slot_get_rec_validate(slot);
+      if (UNIV_UNLIKELY(!slot_rec))
+        break;
+      n+= lint(rec_get_n_owned_new(slot_rec));
+
+      if (rec == slot_rec)
+        goto found;
+    }
+    while ((slot-= 2) > end_slot);
+  }
+  else
+  {
+    for (; rec_get_n_owned_old(rec) == 0; n--)
+      if (UNIV_UNLIKELY(!(rec= page_rec_get_next_low(rec, false))))
+        return ULINT_UNDEFINED;
+
+    do
+    {
+      const rec_t *slot_rec= page_dir_slot_get_rec_validate(slot);
+      if (UNIV_UNLIKELY(!slot_rec))
+        break;
+      n+= lint(rec_get_n_owned_old(slot_rec));
 
-	ut_ad(n >= 0);
-	ut_ad((ulong) n < srv_page_size / (REC_N_NEW_EXTRA_BYTES + 1));
+      if (rec == slot_rec)
+        goto found;
+    }
+    while ((slot-= 2) > end_slot);
+  }
 
-	return((ulint) n);
+  return ULINT_UNDEFINED;
+found:
+  return --n < 0 ? ULINT_UNDEFINED : ulint(n);
 }
 
 /************************************************************//**
@@ -1783,15 +1855,14 @@ page_simple_validate_new(
 	slot_no = 0;
 	slot = page_dir_get_nth_slot(page, slot_no);
 
-	rec = page_get_infimum_rec(page);
+	rec = page + PAGE_NEW_INFIMUM;
 
 	for (;;) {
-		if (UNIV_UNLIKELY(rec > rec_heap_top)) {
-
+		if (UNIV_UNLIKELY(rec < page + PAGE_NEW_INFIMUM
+				  || rec > rec_heap_top)) {
 			ib::error() << "Record " << page_offset(rec)
-				<< " is above rec heap top "
+				<< " is out of bounds: "
 				<< page_offset(rec_heap_top);
-
 			goto func_exit;
 		}
 
@@ -2247,14 +2318,21 @@ wrong_page_type:
 		}
 
 next_rec:
-		if (page_rec_is_supremum(rec)) {
+		old_rec = rec;
+		rec = page_rec_get_next_const(rec);
+
+		if (UNIV_UNLIKELY(!rec != page_rec_is_supremum(old_rec))) {
+			ib::error() << "supremum is not last record: " << offs;
+			ret = FALSE;
+		}
+
+		if (!rec) {
+			rec = old_rec; /* supremum */
 			break;
 		}
 
 		count++;
 		own_count++;
-		old_rec = rec;
-		rec = page_rec_get_next_const(rec);
 
 		if (page_rec_is_infimum(old_rec)
 		    && page_rec_is_user_rec(rec)) {
@@ -2409,37 +2487,36 @@ page_find_rec_with_heap_no(
 @param[in]	page	index tree leaf page
 @return the last record, not delete-marked
 @retval infimum record if all records are delete-marked */
-const rec_t*
-page_find_rec_max_not_deleted(
-	const page_t*	page)
+const rec_t *page_find_rec_max_not_deleted(const page_t *page)
 {
-	const rec_t*	rec = page_get_infimum_rec(page);
-	const rec_t*	prev_rec = NULL; // remove warning
+  ut_ad(page_is_leaf(page));
 
-	/* Because the page infimum is never delete-marked
-	and never the metadata pseudo-record (MIN_REC_FLAG)),
-	prev_rec will always be assigned to it first. */
-	ut_ad(!rec_get_info_bits(rec, page_rec_is_comp(rec)));
-	ut_ad(page_is_leaf(page));
-
-	if (page_is_comp(page)) {
-		do {
-			if (!(rec[-REC_NEW_INFO_BITS]
-			      & (REC_INFO_DELETED_FLAG
-				 | REC_INFO_MIN_REC_FLAG))) {
-				prev_rec = rec;
-			}
-			rec = page_rec_get_next_low(rec, true);
-		} while (rec != page + PAGE_NEW_SUPREMUM);
-	} else {
-		do {
-			if (!(rec[-REC_OLD_INFO_BITS]
-			      & (REC_INFO_DELETED_FLAG
-				 | REC_INFO_MIN_REC_FLAG))) {
-				prev_rec = rec;
-			}
-			rec = page_rec_get_next_low(rec, false);
-		} while (rec != page + PAGE_OLD_SUPREMUM);
-	}
-	return(prev_rec);
+  if (page_is_comp(page))
+  {
+    const rec_t *rec= page + PAGE_NEW_INFIMUM;
+    const rec_t *prev_rec= rec;
+    do
+    {
+      if (!(rec[-REC_NEW_INFO_BITS] &
+            (REC_INFO_DELETED_FLAG | REC_INFO_MIN_REC_FLAG)))
+        prev_rec= rec;
+      if (!(rec= page_rec_get_next_low(rec, true)))
+        return page + PAGE_NEW_INFIMUM;
+    } while (rec != page + PAGE_NEW_SUPREMUM);
+    return prev_rec;
+  }
+  else
+  {
+    const rec_t *rec= page + PAGE_OLD_INFIMUM;
+    const rec_t *prev_rec= rec;
+    do
+    {
+      if (!(rec[-REC_OLD_INFO_BITS] &
+            (REC_INFO_DELETED_FLAG | REC_INFO_MIN_REC_FLAG)))
+        prev_rec= rec;
+      if (!(rec= page_rec_get_next_low(rec, false)))
+        return page + PAGE_OLD_INFIMUM;
+    } while (rec != page + PAGE_OLD_SUPREMUM);
+    return prev_rec;
+  }
 }
diff --git a/storage/innobase/rem/rem0rec.cc b/storage/innobase/rem/rem0rec.cc
index bd572372aca..9948f2e4d68 100644
--- a/storage/innobase/rem/rem0rec.cc
+++ b/storage/innobase/rem/rem0rec.cc
@@ -864,19 +864,19 @@ rec_get_offsets_func(
 			ut_ad(!n_core);
 			n = dict_index_get_n_unique_in_tree_nonleaf(index) + 1;
 			break;
+		default:
+			ut_ad("corrupted record header" == 0);
+			/* fall through */
 		case REC_STATUS_INFIMUM:
 		case REC_STATUS_SUPREMUM:
 			/* infimum or supremum record */
 			ut_ad(rec_get_heap_no_new(rec)
 			      == ulint(rec_get_status(rec)
-                                       == REC_STATUS_INFIMUM
-                                       ? PAGE_HEAP_NO_INFIMUM
-                                       : PAGE_HEAP_NO_SUPREMUM));
+				       == REC_STATUS_INFIMUM
+				       ? PAGE_HEAP_NO_INFIMUM
+				       : PAGE_HEAP_NO_SUPREMUM));
 			n = 1;
 			break;
-		default:
-			ut_error;
-			return(NULL);
 		}
 	} else {
 		n = rec_get_n_fields_old(rec);
diff --git a/storage/innobase/row/row0import.cc b/storage/innobase/row/row0import.cc
index 8824be11917..5468f3aed34 100644
--- a/storage/innobase/row/row0import.cc
+++ b/storage/innobase/row/row0import.cc
@@ -253,19 +253,17 @@ public:
 	}
 
 	/** Position the cursor on the first user record. */
-	void	open(buf_block_t* block) UNIV_NOTHROW
+	rec_t* open(buf_block_t* block) noexcept
+		MY_ATTRIBUTE((warn_unused_result))
 	{
 		page_cur_set_before_first(block, &m_cur);
-
-		if (!end()) {
-			next();
-		}
+		return next();
 	}
 
 	/** Move to the next record. */
-	void	next() UNIV_NOTHROW
+	rec_t* next() noexcept MY_ATTRIBUTE((warn_unused_result))
 	{
-		page_cur_move_to_next(&m_cur);
+		return page_cur_move_to_next(&m_cur);
 	}
 
 	/**
@@ -1520,6 +1518,8 @@ inline bool IndexPurge::open() noexcept
     return false;
 
   rec_t *rec= page_rec_get_next(btr_pcur_get_rec(&m_pcur));
+  if (!rec)
+    return false;
   if (rec_is_metadata(rec, *m_index))
     /* Skip the metadata pseudo-record. */
     btr_pcur_get_page_cur(&m_pcur)->rec= rec;
@@ -1531,7 +1531,9 @@ Position the cursor on the next record.
 @return DB_SUCCESS or error code */
 dberr_t IndexPurge::next() noexcept
 {
-	btr_pcur_move_to_next_on_page(&m_pcur);
+	if (UNIV_UNLIKELY(!btr_pcur_move_to_next_on_page(&m_pcur))) {
+		return DB_CORRUPTION;
+	}
 
 	/* When switching pages, commit the mini-transaction
 	in order to release the latch on the old page. */
@@ -1569,8 +1571,8 @@ dberr_t IndexPurge::next() noexcept
 								     &m_mtr)) {
 				return err;
 			}
-		} else {
-			btr_pcur_move_to_next_on_page(&m_pcur);
+		} else if (!btr_pcur_move_to_next_on_page(&m_pcur)) {
+			return DB_CORRUPTION;
 		}
 	} while (!btr_pcur_is_on_user_rec(&m_pcur));
 
@@ -1787,7 +1789,9 @@ PageConverter::update_records(
 
 	/* This will also position the cursor on the first user record. */
 
-	m_rec_iter.open(block);
+	if (!m_rec_iter.open(block)) {
+		return DB_CORRUPTION;
+	}
 
 	while (!m_rec_iter.end()) {
 		rec_t*	rec = m_rec_iter.current();
@@ -1818,17 +1822,19 @@ PageConverter::update_records(
 		optimistic delete. */
 
 		if (deleted) {
+			++m_index->m_stats.m_n_deleted;
 			/* A successful purge will move the cursor to the
 			next record. */
 
-			if (!purge()) {
-				m_rec_iter.next();
+			if (purge()) {
+				continue;
 			}
-
-			++m_index->m_stats.m_n_deleted;
 		} else {
 			++m_index->m_stats.m_n_rows;
-			m_rec_iter.next();
+		}
+
+		if (!m_rec_iter.next()) {
+			return DB_CORRUPTION;
 		}
 	}
 
@@ -3167,6 +3173,8 @@ static dberr_t handle_instant_metadata(dict_table_t *table,
     while (btr_page_get_level(page.get()) != 0)
     {
       const rec_t *rec= page_rec_get_next(page_get_infimum_rec(page.get()));
+      if (!rec)
+        return DB_CORRUPTION;
 
       /* Relax the assertion in rec_init_offsets(). */
       ut_ad(!index->in_instant_init);
@@ -3191,18 +3199,22 @@ static dberr_t handle_instant_metadata(dict_table_t *table,
         return err;
     }
 
-    const auto *rec= page_rec_get_next(page_get_infimum_rec(page.get()));
+    const auto *rec= page_rec_get_next_const(page_get_infimum_rec(page.get()));
     const auto comp= dict_table_is_comp(index->table);
-    const auto info_bits= rec_get_info_bits(rec, comp);
 
-    if (page_rec_is_supremum(rec) || !(info_bits & REC_INFO_MIN_REC_FLAG))
+    if (!rec || page_rec_is_supremum(rec))
     {
+    corrupted_metadata:
       ib::error() << "Table " << index->table->name
                   << " is missing instant ALTER metadata";
       index->table->corrupted= true;
       return DB_CORRUPTION;
     }
 
+    const auto info_bits= rec_get_info_bits(rec, comp);
+    if (!(info_bits & REC_INFO_MIN_REC_FLAG))
+      goto corrupted_metadata;
+
     if ((info_bits & ~REC_INFO_DELETED_FLAG) != REC_INFO_MIN_REC_FLAG ||
         (comp && rec_get_status(rec) != REC_STATUS_INSTANT))
     {
diff --git a/storage/innobase/row/row0ins.cc b/storage/innobase/row/row0ins.cc
index c491030070c..b428a7bc43e 100644
--- a/storage/innobase/row/row0ins.cc
+++ b/storage/innobase/row/row0ins.cc
@@ -2229,7 +2229,9 @@ row_ins_duplicate_error_in_clust_online(
 		}
 	}
 
-	rec = page_rec_get_next_const(btr_cur_get_rec(cursor));
+	if (!(rec = page_rec_get_next_const(btr_cur_get_rec(cursor)))) {
+		return DB_CORRUPTION;
+	}
 
 	if (cursor->up_match >= n_uniq && !page_rec_is_supremum(rec)) {
 		*offsets = rec_get_offsets(rec, cursor->index, *offsets,
@@ -2349,11 +2351,13 @@ duplicate:
 		}
 	}
 
+	err = DB_SUCCESS;
+
 	if (cursor->up_match >= n_unique) {
 
 		rec = page_rec_get_next(btr_cur_get_rec(cursor));
 
-		if (!page_rec_is_supremum(rec)) {
+		if (rec && !page_rec_is_supremum(rec)) {
 			offsets = rec_get_offsets(rec, cursor->index, offsets,
 						  cursor->index->n_core_fields,
 						  ULINT_UNDEFINED, &heap);
@@ -2378,24 +2382,23 @@ duplicate:
 			}
 
 			switch (err) {
+			default:
+				break;
 			case DB_SUCCESS_LOCKED_REC:
+				err = DB_SUCCESS;
+				/* fall through */
 			case DB_SUCCESS:
-				break;
-			default:
-				goto func_exit;
-			}
-
-			if (row_ins_dupl_error_with_rec(
-				    rec, entry, cursor->index, offsets)) {
-				goto duplicate;
+				if (row_ins_dupl_error_with_rec(
+					    rec, entry, cursor->index,
+					    offsets)) {
+					goto duplicate;
+				}
 			}
 		}
 
 		/* This should never happen */
-		ut_error;
+		err = DB_CORRUPTION;
 	}
-
-	err = DB_SUCCESS;
 func_exit:
 	if (UNIV_LIKELY_NULL(heap)) {
 		mem_heap_free(heap);
diff --git a/storage/innobase/row/row0merge.cc b/storage/innobase/row/row0merge.cc
index 9312a114990..c3d86bce53f 100644
--- a/storage/innobase/row/row0merge.cc
+++ b/storage/innobase/row/row0merge.cc
@@ -2142,10 +2142,14 @@ corrupted_metadata:
 
 		mem_heap_empty(row_heap);
 
-		page_cur_move_to_next(cur);
-
 		stage->n_pk_recs_inc();
 
+		if (!page_cur_move_to_next(cur)) {
+corrupted_rec:
+			err = DB_CORRUPTION;
+			goto err_exit;
+		}
+
 		if (page_cur_is_after_last(cur)) {
 
 			stage->inc();
@@ -2251,9 +2255,10 @@ end_of_index:
 				btr_leaf_page_release(page_cur_get_block(cur),
 						      BTR_SEARCH_LEAF, &mtr);
 				page_cur_set_before_first(block, cur);
-				page_cur_move_to_next(cur);
-
-				ut_ad(!page_cur_is_after_last(cur));
+				if (!page_cur_move_to_next(cur)
+				    || page_cur_is_after_last(cur)) {
+					goto corrupted_rec;
+				}
 			}
 		}
 
diff --git a/storage/innobase/row/row0sel.cc b/storage/innobase/row/row0sel.cc
index ebd4ddce130..ec4d09115f6 100644
--- a/storage/innobase/row/row0sel.cc
+++ b/storage/innobase/row/row0sel.cc
@@ -1804,7 +1804,11 @@ rec_loop:
 		search result set, resulting in the phantom problem. */
 
 		if (!node->read_view) {
-			rec_t*	next_rec = page_rec_get_next(rec);
+			const rec_t* next_rec = page_rec_get_next_const(rec);
+			if (UNIV_UNLIKELY(!next_rec)) {
+				err = DB_CORRUPTION;
+				goto lock_wait_or_error;
+			}
 			unsigned lock_type;
 
 			offsets = rec_get_offsets(next_rec, index, offsets,
@@ -3425,11 +3429,13 @@ Row_sel_get_clust_rec_for_mysql::operator()(
 				rec, sec_index, true,
 				sec_index->n_fields, heap);
 			page_cur_t     page_cursor;
-
-		        ulint		low_match = page_cur_search(
-						block, sec_index, tuple,
-						PAGE_CUR_LE, &page_cursor);
-
+			ulint up_match = 0, low_match = 0;
+			ut_ad(!page_cur_search_with_match(block, sec_index,
+							  tuple, PAGE_CUR_LE,
+							  &up_match,
+							  &low_match,
+							  &page_cursor,
+							  nullptr));
 			ut_ad(low_match < dtuple_get_n_fields_cmp(tuple));
 			mem_heap_free(heap);
 			clust_rec = NULL;
@@ -4770,6 +4776,7 @@ wait_table_again:
 					   	 pcur, 0, &mtr);
 
 		if (err != DB_SUCCESS) {
+page_corrupted:
 			rec = NULL;
 			goto page_read_error;
 		}
@@ -4787,6 +4794,10 @@ wait_table_again:
 			/* Try to place a gap lock on the next index record
 			to prevent phantoms in ORDER BY ... DESC queries */
 			const rec_t*	next_rec = page_rec_get_next_const(rec);
+			if (UNIV_UNLIKELY(!next_rec)) {
+				err = DB_CORRUPTION;
+				goto page_corrupted;
+			}
 
 			offsets = rec_get_offsets(next_rec, index, offsets,
 						  index->n_core_fields,
@@ -5792,8 +5803,8 @@ next_rec_after_check:
 				if (err != DB_SUCCESS) {
 					goto lock_wait_or_error;
 				}
-			} else {
-				btr_pcur_move_to_next_on_page(pcur);
+			} else if (!btr_pcur_move_to_next_on_page(pcur)) {
+				goto corrupted;
 			}
 
 			goto rec_loop;
@@ -5803,6 +5814,7 @@ next_rec_after_check:
 			goto rec_loop;
 		}
 		if (UNIV_UNLIKELY(!btr_pcur_get_rec(pcur))) {
+corrupted:
 			err = DB_CORRUPTION;
 			goto normal_return;
 		}