MDEV-19541: Avoid infinite loop of reading a corrupted page

row_search_mvcc(): Duplicate the logic of btr_pcur_move_to_next()
so that an infinite loop can be avoided when advancing to the next
page fails due to a corrupted page.

mysql-test/suite/encryption/r/innodb-force-corrupt.result

+call mtr.add_suppression("InnoDB: Table `test`\\.`t[13]` (has an unreadable root page|is corrupted)");
 call mtr.add_suppression("InnoDB: Encrypted page \\[page id: space=\\d+, page number=[36]\\] in file .*test.t[123]\\.ibd looks corrupted; key_version=3221342974");
-call mtr.add_suppression("InnoDB: Table `test`\\.`t[13]` is corrupted");
+call mtr.add_suppression("\\[ERROR\\] InnoDB: We detected index corruption in an InnoDB type table");
+call mtr.add_suppression("\\[ERROR\\] mysqld: Index for table 't2' is corrupt; try to repair it");
 SET GLOBAL innodb_file_per_table = ON;
 set global innodb_compression_algorithm = 1;
 # Create and populate tables to be corrupted
@@ -17,7 +19,7 @@ COMMIT;
 SELECT * FROM t1;
 ERROR 42S02: Table 'test.t1' doesn't exist in engine
 SELECT * FROM t2;
-ERROR HY000: Got error 192 'Table encrypted but decryption failed. This could be because correct encryption management plugin is not loaded, used encryption key is not available or encryption method does not match.' from InnoDB
+Got one of the listed errors
 SELECT * FROM t3;
 ERROR 42S02: Table 'test.t3' doesn't exist in engine
 # Restore the original tables

mysql-test/suite/encryption/t/innodb-force-corrupt.test

@@ -7,8 +7,10 @@
 # Don't test under embedded
 -- source include/not_embedded.inc
 
+call mtr.add_suppression("InnoDB: Table `test`\\.`t[13]` (has an unreadable root page|is corrupted)");
 call mtr.add_suppression("InnoDB: Encrypted page \\[page id: space=\\d+, page number=[36]\\] in file .*test.t[123]\\.ibd looks corrupted; key_version=3221342974");
-call mtr.add_suppression("InnoDB: Table `test`\\.`t[13]` is corrupted");
+call mtr.add_suppression("\\[ERROR\\] InnoDB: We detected index corruption in an InnoDB type table");
+call mtr.add_suppression("\\[ERROR\\] mysqld: Index for table 't2' is corrupt; try to repair it");
 
 SET GLOBAL innodb_file_per_table = ON;
 set global innodb_compression_algorithm = 1;
@@ -68,7 +70,7 @@ EOF
 
 --error ER_NO_SUCH_TABLE_IN_ENGINE
 SELECT * FROM t1;
---error ER_GET_ERRMSG
+--error ER_GET_ERRMSG,ER_NOT_KEYFILE
 SELECT * FROM t2;
 --error ER_NO_SUCH_TABLE_IN_ENGINE
 SELECT * FROM t3;

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

@@ -16,5 +16,5 @@ SELECT * FROM t1 WHERE PK = 1;
 pk	c
 1	sql
 SELECT * FROM t1 WHERE pk = 12;
-ERROR HY000: Got error 192 'Table encrypted but decryption failed. This could be because correct encryption management plugin is not loaded, used encryption key is not available or encryption method does not match.' from InnoDB
+ERROR HY000: Index for table 't1' is corrupt; try to repair it
 DROP TABLE t1;

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

@@ -7,6 +7,8 @@ call mtr.add_suppression("InnoDB: Background Page read failed to read or decrypt
 call mtr.add_suppression("\\[ERROR\\] InnoDB: Failed to read file '.*test.t1\\.ibd' at offset 19: Page read from tablespace is corrupted\\.");
 call mtr.add_suppression("\\[ERROR\\] InnoDB: Plugin initialization aborted at srv0start\\.cc.* with error Data structure corruption");
 call mtr.add_suppression("\\[ERROR\\] Plugin 'InnoDB' (init function|registration)");
+call mtr.add_suppression("\\[ERROR\\] InnoDB: We detected index corruption");
+call mtr.add_suppression("\\[ERROR\\] mysqld: Index for table 't1' is corrupt; try to repair it");
 --enable_query_log
 CREATE TABLE t1 (pk INT PRIMARY KEY, c CHAR(255))ENGINE=InnoDB STATS_PERSISTENT=0;
 
@@ -19,13 +21,15 @@ INSERT INTO t1 VALUES(1, 'sql'), (2, 'server'), (3, 'mariadb'),
 
 --source include/restart_mysqld.inc
 
-SELECT COUNT(*) FROM t1;
-UPDATE t1 SET c='best8' WHERE pk=12;
-
 let INNODB_PAGE_SIZE=`select @@innodb_page_size`;
 let MYSQLD_DATADIR=`select @@datadir`;
 
---source include/kill_mysqld.inc
+SELECT COUNT(*) FROM t1;
+--source ../include/no_checkpoint_start.inc
+UPDATE t1 SET c='best8' WHERE pk=12;
+
+--let CLEANUP_IF_CHECKPOINT=DROP TABLE t1;
+--source ../include/no_checkpoint_end.inc
 --echo # Corrupt the pages
 
 perl;
@@ -44,7 +48,9 @@ SELECT * FROM t1 WHERE PK = 1;
 let $restart_parameters=--innodb-force-recovery=1;
 --source include/restart_mysqld.inc
 SELECT * FROM t1 WHERE PK = 1;
---error ER_GET_ERRMSG
+--error ER_NOT_KEYFILE
 SELECT * FROM t1 WHERE pk = 12;
 
 DROP TABLE t1;
+let $restart_parameters=;
+--source include/restart_mysqld.inc

storage/innobase/row/row0sel.cc

@@ -5636,36 +5636,48 @@ row_search_mvcc(
 	}
 
 	if (moves_up) {
-		bool	 move;
-
-		if (spatial_search) {
-			move = rtr_pcur_move_to_next(
-				search_tuple, mode, pcur, 0, &mtr);
-		} else {
-			move = btr_pcur_move_to_next(pcur, &mtr);
-		}
-
-		if (!move) {
-not_moved:
-			if (!spatial_search) {
-				btr_pcur_store_position(pcur, &mtr);
+		if (UNIV_UNLIKELY(spatial_search)) {
+			if (rtr_pcur_move_to_next(
+				    search_tuple, mode, pcur, 0, &mtr)) {
+				goto rec_loop;
 			}
-
-			if (match_mode != 0) {
-				err = DB_RECORD_NOT_FOUND;
+		} else {
+			const buf_block_t* block = btr_pcur_get_block(pcur);
+			/* This is based on btr_pcur_move_to_next(),
+			but avoids infinite read loop of a corrupted page. */
+			ut_ad(pcur->pos_state == BTR_PCUR_IS_POSITIONED);
+			ut_ad(pcur->latch_mode != BTR_NO_LATCHES);
+			pcur->old_stored = false;
+			if (btr_pcur_is_after_last_on_page(pcur)) {
+				if (btr_pcur_is_after_last_in_tree(pcur,
+								   &mtr)) {
+					goto not_moved;
+				}
+				btr_pcur_move_to_next_page(pcur, &mtr);
+				if (UNIV_UNLIKELY(btr_pcur_get_block(pcur)
+						  == block)) {
+					err = DB_CORRUPTION;
+					goto lock_wait_or_error;
+				}
 			} else {
-				err = DB_END_OF_INDEX;
+				btr_pcur_move_to_next_on_page(pcur);
 			}
 
-			goto normal_return;
+			goto rec_loop;
 		}
 	} else {
-		if (UNIV_UNLIKELY(!btr_pcur_move_to_prev(pcur, &mtr))) {
-			goto not_moved;
+		if (btr_pcur_move_to_prev(pcur, &mtr)) {
+			goto rec_loop;
 		}
 	}
 
-	goto rec_loop;
+not_moved:
+	if (!spatial_search) {
+		btr_pcur_store_position(pcur, &mtr);
+	}
+
+	err = match_mode ? DB_RECORD_NOT_FOUND : DB_END_OF_INDEX;
+	goto normal_return;
 
 lock_wait_or_error:
 	/* Reset the old and new "did semi-consistent read" flags. */