MDEV-19541 InnoDB crashes when trying to recover a corrupted pagebb-10.2-MDEV-19541

- Don't apply redo log for the corrupted page when innodb_force_recovery > 0.

- Don't mark the tablespace as corrupted when innodb_force_recovery > 0 or
recovery is in progress.

- Allow the table to be dropped when index root page is
corrupted when innodb_force_recovery > 0.

10 files changed, 162 insertions, 27 deletions

diff --git a/mysql-test/suite/encryption/r/corrupted_during_recovery.result b/mysql-test/suite/encryption/r/corrupted_during_recovery.result
index 41c0d7d75a8..356dce64f8d 100644
--- a/mysql-test/suite/encryption/r/corrupted_during_recovery.result
+++ b/mysql-test/suite/encryption/r/corrupted_during_recovery.result
@@ -9,14 +9,11 @@ INSERT INTO t2 VALUES(2);
 SELECT * FROM t1;
 ERROR 42000: Unknown storage engine 'InnoDB'
 SELECT * FROM t1;
-a
-1
-2
+ERROR 42S02: Table 'test.t1' doesn't exist in engine
 SELECT * FROM t2;
 a
 2
-CHECK TABLE t1,t2;
+CHECK TABLE t2;
 Table	Op	Msg_type	Msg_text
-test.t1	check	status	OK
 test.t2	check	status	OK
 DROP TABLE t1, t2;
diff --git a/mysql-test/suite/encryption/t/corrupted_during_recovery.test b/mysql-test/suite/encryption/t/corrupted_during_recovery.test
index 5784d5775c6..1ca3419820b 100644
--- a/mysql-test/suite/encryption/t/corrupted_during_recovery.test
+++ b/mysql-test/suite/encryption/t/corrupted_during_recovery.test
@@ -50,11 +50,16 @@ EOF
 --source include/start_mysqld.inc
 --error ER_UNKNOWN_STORAGE_ENGINE
 SELECT * FROM t1;
+--disable_query_log
+call mtr.add_suppression("InnoDB: Encrypted page \\[page id: space=[1-9][0-9]*, page number=3\\] in file .*test.t[1].ibd looks corrupted; key_version=1786080875");
+call mtr.add_suppression("InnoDB: Table `test`\\.`t1` is corrupted. Please drop the table and recreate.");
+--enable_query_log
 let $restart_parameters=--innodb_force_recovery=1;
 --source include/restart_mysqld.inc
 
+--error ER_NO_SUCH_TABLE_IN_ENGINE
 SELECT * FROM t1;
 SELECT * FROM t2;
-CHECK TABLE t1,t2;
+CHECK TABLE t2;
 
 DROP TABLE t1, t2;
diff --git a/mysql-test/suite/innodb/r/corrupted_during_recovery.result b/mysql-test/suite/innodb/r/corrupted_during_recovery.result
index 788f17e3284..ee4db08fc85 100644
--- a/mysql-test/suite/innodb/r/corrupted_during_recovery.result
+++ b/mysql-test/suite/innodb/r/corrupted_during_recovery.result
@@ -9,14 +9,11 @@ INSERT INTO t2 VALUES(1);
 SELECT * FROM t1;
 ERROR 42000: Unknown storage engine 'InnoDB'
 SELECT * FROM t1;
-a
-0
-2
+ERROR 42S02: Table 'test.t1' doesn't exist in engine
 SELECT * FROM t2;
 a
 1
-CHECK TABLE t1,t2;
+CHECK TABLE t2;
 Table	Op	Msg_type	Msg_text
-test.t1	check	status	OK
 test.t2	check	status	OK
 DROP TABLE t1, t2;
diff --git a/mysql-test/suite/innodb/r/leaf_page_corrupted_during_recovery.result b/mysql-test/suite/innodb/r/leaf_page_corrupted_during_recovery.result
new file mode 100644
index 00000000000..a7b8dcddde5
--- /dev/null
+++ b/mysql-test/suite/innodb/r/leaf_page_corrupted_during_recovery.result
@@ -0,0 +1,17 @@
+CREATE TABLE t1 (pk INT PRIMARY KEY, c CHAR(255))ENGINE=InnoDB STATS_PERSISTENT=0;
+SET GLOBAL INNODB_LIMIT_OPTIMISTIC_INSERT_DEBUG = 2;
+INSERT INTO t1 VALUES(1, "sql"), (2, "server"), (3, "mariadb"),
+(4, "mariadb"), (5, "test1"), (6, "test2"), (7, "test3"),
+(8, "test4"), (9, "test5"), (10, "test6"), (11, "test7"),
+(12, "test8");
+SELECT COUNT(*) FROM t1;
+COUNT(*)
+12
+# Kill the server
+# Corrupt the pages
+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
+DROP TABLE t1;
diff --git a/mysql-test/suite/innodb/t/corrupted_during_recovery.test b/mysql-test/suite/innodb/t/corrupted_during_recovery.test
index fbfb1bbe5d5..dad08645085 100644
--- a/mysql-test/suite/innodb/t/corrupted_during_recovery.test
+++ b/mysql-test/suite/innodb/t/corrupted_during_recovery.test
@@ -6,6 +6,8 @@ call mtr.add_suppression("Plugin 'InnoDB' init function returned error");
 call mtr.add_suppression("Plugin 'InnoDB' registration as a STORAGE ENGINE failed");
 call mtr.add_suppression("InnoDB: Database page corruption on disk or a failed file read of tablespace test/t1 page");
 call mtr.add_suppression("InnoDB: Failed to read file '.*test.t1\\.ibd' at offset 3: Page read from tablespace is corrupted.");
+call mtr.add_suppression("InnoDB: Background Page read failed to read or decrypt \\[page id: space=\\d+, page number=3\\]");
+call mtr.add_suppression("InnoDB: Table `test`.`t1` is corrupted. Please drop the table and recreate.");
 --enable_query_log
 
 let INNODB_PAGE_SIZE=`select @@innodb_page_size`;
@@ -53,8 +55,10 @@ EOF
 SELECT * FROM t1;
 let $restart_parameters=--innodb_force_recovery=1;
 --source include/restart_mysqld.inc
+
+--error ER_NO_SUCH_TABLE_IN_ENGINE
 SELECT * FROM t1;
 SELECT * FROM t2;
-CHECK TABLE t1,t2;
+CHECK TABLE t2;
 
 DROP TABLE t1, t2;
diff --git a/mysql-test/suite/innodb/t/leaf_page_corrupted_during_recovery.opt b/mysql-test/suite/innodb/t/leaf_page_corrupted_during_recovery.opt
new file mode 100644
index 00000000000..5bbc22dfc99
--- /dev/null
+++ b/mysql-test/suite/innodb/t/leaf_page_corrupted_during_recovery.opt
@@ -0,0 +1,2 @@
+--innodb_doublewrite=0
+--innodb_buffer_pool_load_at_startup=0
diff --git a/mysql-test/suite/innodb/t/leaf_page_corrupted_during_recovery.test b/mysql-test/suite/innodb/t/leaf_page_corrupted_during_recovery.test
new file mode 100644
index 00000000000..f3eb9d83059
--- /dev/null
+++ b/mysql-test/suite/innodb/t/leaf_page_corrupted_during_recovery.test
@@ -0,0 +1,45 @@
+--source include/have_innodb.inc
+
+--disable_query_log
+call mtr.add_suppression("InnoDB: Database page corruption on disk or a failed file read of tablespace test/t1 page .*");
+call mtr.add_suppression("InnoDB: Background Page read failed to read or decrypt \\[page id: space=\\d+, page number=19\\]");
+--enable_query_log
+CREATE TABLE t1 (pk INT PRIMARY KEY, c CHAR(255))ENGINE=InnoDB STATS_PERSISTENT=0;
+
+SET GLOBAL INNODB_LIMIT_OPTIMISTIC_INSERT_DEBUG = 2;
+
+INSERT INTO t1 VALUES(1, "sql"), (2, "server"), (3, "mariadb"),
+	(4, "mariadb"), (5, "test1"), (6, "test2"), (7, "test3"),
+	(8, "test4"), (9, "test5"), (10, "test6"), (11, "test7"),
+	(12, "test8");
+
+--source include/restart_mysqld.inc
+
+SELECT COUNT(*) FROM t1;
+
+let INNODB_PAGE_SIZE=`select @@innodb_page_size`;
+let MYSQLD_DATADIR=`select @@datadir`;
+
+--source include/kill_mysqld.inc
+--echo # Corrupt the pages
+
+perl;
+my $ps = $ENV{INNODB_PAGE_SIZE};
+
+my $file = "$ENV{MYSQLD_DATADIR}/test/t1.ibd";
+open(FILE, "+<$file") || die "Unable to open $file";
+binmode FILE;
+seek (FILE, $ENV{INNODB_PAGE_SIZE} * 19, SEEK_SET) or die "seek";
+print FILE "junk";
+close FILE or die "close";
+EOF
+
+let $restart_paramters=--innodb_force_recovery=1;
+--source include/start_mysqld.inc
+
+SELECT * FROM t1 WHERE PK = 1;
+
+--error ER_GET_ERRMSG
+SELECT * FROM t1 WHERE pk = 12;
+
+DROP TABLE t1;
diff --git a/storage/innobase/buf/buf0buf.cc b/storage/innobase/buf/buf0buf.cc
index 41d9efe25cc..97f21fa1d3b 100644
--- a/storage/innobase/buf/buf0buf.cc
+++ b/storage/innobase/buf/buf0buf.cc
@@ -4378,6 +4378,14 @@ loop:
 				return (NULL);
 			}
 
+			/* Ignore the corrupted pages only and don't
+			mark the space as corrupted when
+			innodb_force_recovery is set or recovery is on. */
+			if (local_err == DB_PAGE_CORRUPTED
+			    && (srv_force_recovery || recv_recovery_is_on())) {
+				return NULL;
+			}
+
 			/* Try to set table as corrupted instead of
 			asserting. */
 			if (page_id.space() != TRX_SYS_SPACE &&
@@ -5743,18 +5751,33 @@ buf_page_monitor(
 	MONITOR_INC_NOCHECK(counter);
 }
 
-/********************************************************************//**
-Mark a table with the specified space pointed by bpage->id.space() corrupted.
-Also remove the bpage from LRU list.
-@param[in,out]		bpage			Block */
+/** Mark a table corrupted.
+@param[in]	bpage	Corrupted page. */
 static
 void
 buf_mark_space_corrupt(buf_page_t* bpage)
 {
+	/* If block is not encrypted find the table with specified
+	space id, and mark it corrupted. Encrypted tables
+	are marked unusable later e.g. in ::open(). */
+	if (!bpage->encrypted) {
+		dict_set_corrupted_by_space(bpage->id.space());
+	} else {
+		dict_set_encrypted_by_space(bpage->id.space());
+	}
+}
+
+/** Mark a table corrupted.
+@param[in]	bpage	Corrupted page
+@param[in]	space	Corrupted page belongs to tablespace
+Also remove the bpage from LRU list. */
+static
+void
+buf_corrupt_page_release(buf_page_t* bpage, const fil_space_t* space)
+{
 	buf_pool_t*	buf_pool = buf_pool_from_bpage(bpage);
 	const ibool	uncompressed = (buf_page_get_state(bpage)
 					== BUF_BLOCK_FILE_PAGE);
-	uint32_t	space = bpage->id.space();
 
 	/* First unfix and release lock on the bpage */
 	buf_pool_mutex_enter(buf_pool);
@@ -5773,13 +5796,12 @@ buf_mark_space_corrupt(buf_page_t* bpage)
 
 	mutex_exit(buf_page_get_mutex(bpage));
 
-	/* If block is not encrypted find the table with specified
-	space id, and mark it corrupted. Encrypted tables
-	are marked unusable later e.g. in ::open(). */
-	if (!bpage->encrypted) {
-		dict_set_corrupted_by_space(space);
-	} else {
-		dict_set_encrypted_by_space(space);
+	/* Don't mark the tablespace as corrupted when
+	innodb_force_recovery is enabled or recovery is
+	in progress. */
+	if (!recv_recovery_is_on()
+	    || !srv_force_recovery) {
+		buf_mark_space_corrupt(bpage);
 	}
 
 	/* After this point bpage can't be referenced. */
@@ -5981,7 +6003,7 @@ database_corrupted:
 				"buf_page_import_corrupt_failure",
 				if (!is_predefined_tablespace(
 					    bpage->id.space())) {
-					buf_mark_space_corrupt(bpage);
+					buf_corrupt_page_release(bpage, space);
 					ib::info() << "Simulated IMPORT "
 						"corruption";
 					fil_space_release_for_io(space);
@@ -6015,7 +6037,7 @@ database_corrupted:
 					<< FORCE_RECOVERY_MSG;
 			}
 
-			if (srv_force_recovery < SRV_FORCE_IGNORE_CORRUPT) {
+			if (!srv_force_recovery) {
 
 				/* If page space id is larger than TRX_SYS_SPACE
 				(0), we will attempt to mark the corresponding
@@ -6025,7 +6047,7 @@ database_corrupted:
 						" a corrupt database page.";
 				}
 
-				buf_mark_space_corrupt(bpage);
+				buf_corrupt_page_release(bpage, space);
 				fil_space_release_for_io(space);
 				return(err);
 			}
@@ -6034,6 +6056,18 @@ database_corrupted:
 		DBUG_EXECUTE_IF("buf_page_import_corrupt_failure",
 				page_not_corrupt: bpage = bpage; );
 
+		if (err == DB_PAGE_CORRUPTED
+		    || err == DB_DECRYPTION_FAILED) {
+			buf_corrupt_page_release(bpage, space);
+
+			if (recv_recovery_is_on()) {
+				recv_recover_corrupt_page(bpage);
+			}
+
+			fil_space_release_for_io(space);
+			return err;
+		}
+
 		if (recv_recovery_is_on()) {
 			recv_recover_page(bpage);
 		}
diff --git a/storage/innobase/include/log0recv.h b/storage/innobase/include/log0recv.h
index 0b04e1642cf..9bb54960058 100644
--- a/storage/innobase/include/log0recv.h
+++ b/storage/innobase/include/log0recv.h
@@ -49,6 +49,11 @@ dberr_t
 recv_find_max_checkpoint(ulint* max_field)
 	MY_ATTRIBUTE((nonnull, warn_unused_result));
 
+/** Reduces recv_sys->n_addrs for the corrupted page.
+This function should called when srv_force_recovery > 0.
+@param[in]	bpage	buffer pool page */
+void recv_recover_corrupt_page(buf_page_t* bpage);
+
 /** Apply any buffered redo log to a page that was just read from a data file.
 @param[in,out]	bpage	buffer pool page */
 ATTRIBUTE_COLD void recv_recover_page(buf_page_t* bpage);
diff --git a/storage/innobase/log/log0recv.cc b/storage/innobase/log/log0recv.cc
index 9484901c90f..c3377ea284e 100644
--- a/storage/innobase/log/log0recv.cc
+++ b/storage/innobase/log/log0recv.cc
@@ -2212,6 +2212,35 @@ skip_log:
 	}
 }
 
+/** Reduces recv_sys->n_addrs for the corrupted page.
+This function should called when srv_force_recovery > 0.
+@param[in]	bpage	buffer pool page */
+void recv_recover_corrupt_page(buf_page_t* bpage)
+{
+	ut_ad(srv_force_recovery);
+	mutex_enter(&recv_sys->mutex);
+
+	if (!recv_sys->apply_log_recs) {
+		mutex_exit(&recv_sys->mutex);
+		return;
+	}
+
+	recv_addr_t* recv_addr = recv_get_fil_addr_struct(
+			bpage->id.space(), bpage->id.page_no());
+
+	switch (recv_addr->state) {
+	case RECV_BEING_PROCESSED:
+	case RECV_PROCESSED:
+		break;
+	case RECV_WILL_NOT_READ:
+		ut_ad(0);
+	default:
+		recv_sys->n_addrs--;
+	}
+
+	mutex_exit(&recv_sys->mutex);
+}
+
 /** Apply any buffered redo log to a page that was just read from a data file.
 @param[in,out]	bpage	buffer pool page */
 void recv_recover_page(buf_page_t* bpage)