MDEV-19435 buf_fix_count > 0 for corrupted page when it exits the LRU listbb-10.2-MDEV-19435

Problem:
=========
One of the purge thread access the corrupted page and tries to remove from
LRU list. In the mean time, other purge threads are waiting for same page
in buf_wait_for_read(). Assertion(buf_fix_count == 0) fails for the
purge thread which tries to remove the page from LRU list.

Solution:
========
- Set the page id as FIL_NULL to indicate the page is corrupted before removing
the block from LRU list. Acquire hash lock for the particular page id and
wait for the other threads to release buf_fix_count for the block.

- Added the error check for btr_cur_open() in row_search_on_row_ref().

7 files changed, 125 insertions, 4 deletions

diff --git a/mysql-test/suite/encryption/r/load_infile.result b/mysql-test/suite/encryption/r/load_infile.result
new file mode 100644
index 00000000000..287a5324e80
--- /dev/null
+++ b/mysql-test/suite/encryption/r/load_infile.result
@@ -0,0 +1,18 @@
+call mtr.add_suppression("InnoDB: The page \\[page id: space=[0-9][0-9]*, page number=6\\] in file '.*test.t2\\.ibd' cannot be decrypted\\.");
+call mtr.add_suppression("InnoDB:  Error code: .* btr_pcur_open_low  level: 0 called from file: .*");
+CREATE TABLE t1 (pk INT PRIMARY KEY AUTO_INCREMENT, c VARCHAR(256)) ENGINE=INNODB ENCRYPTED=YES;
+select * from t1 into outfile "MYSQLTEST_VARDIR/tmp/t1.outfile";
+select count(*) from t1;
+count(*)
+10
+CREATE TABLE t2 (pk INT, c CHAR(255)) ENGINE=INNODB ENCRYPTED=YES;
+insert into t2 values(1000, "mariadb"), (1001, "server"), (1002, "test");
+insert into t2 values(1003, "mysql"), (1004, "test1"), (1005, "test2");
+insert into t2 select * from t2;
+insert into t2 select * from t2;
+insert into t2 select * from t2;
+insert into t2 select * from t2;
+# Corrupt the pages
+load data infile 'MYSQLTEST_VARDIR/tmp/t1.outfile' into table t2;
+Got one of the listed errors
+drop table t1, t2;
diff --git a/mysql-test/suite/encryption/t/load_infile.test b/mysql-test/suite/encryption/t/load_infile.test
new file mode 100644
index 00000000000..2b90306160d
--- /dev/null
+++ b/mysql-test/suite/encryption/t/load_infile.test
@@ -0,0 +1,61 @@
+-- source include/have_innodb.inc
+-- source include/have_file_key_management_plugin.inc
+
+call mtr.add_suppression("InnoDB: The page \\[page id: space=[0-9][0-9]*, page number=6\\] in file '.*test.t2\\.ibd' cannot be decrypted\\.");
+
+call mtr.add_suppression("InnoDB:  Error code: .* btr_pcur_open_low  level: 0 called from file: .*");
+
+CREATE TABLE t1 (pk INT PRIMARY KEY AUTO_INCREMENT, c VARCHAR(256)) ENGINE=INNODB ENCRYPTED=YES;
+
+--disable_warnings
+--disable_query_log
+begin;
+let $i = 10;
+while ($i)
+{
+INSERT INTO t1 values(NULL, substring(MD5(RAND()), -128));
+dec $i;
+}
+commit;
+--enable_warnings
+--enable_query_log
+
+let INNODB_PAGE_SIZE=`select @@innodb_page_size`;
+let MYSQLD_DATADIR=`select @@datadir`;
+
+--replace_result $MYSQLTEST_VARDIR MYSQLTEST_VARDIR
+eval select * from t1 into outfile "$MYSQLTEST_VARDIR/tmp/t1.outfile";
+
+select count(*) from t1;
+
+CREATE TABLE t2 (pk INT, c CHAR(255)) ENGINE=INNODB ENCRYPTED=YES;
+
+insert into t2 values(1000, "mariadb"), (1001, "server"), (1002, "test");
+insert into t2 values(1003, "mysql"), (1004, "test1"), (1005, "test2");
+insert into t2 select * from t2;
+insert into t2 select * from t2;
+insert into t2 select * from t2;
+insert into t2 select * from t2;
+
+--source include/shutdown_mysqld.inc
+--echo # Corrupt the pages
+
+perl;
+my $ps = $ENV{INNODB_PAGE_SIZE};
+
+my $file = "$ENV{MYSQLD_DATADIR}/test/t2.ibd";
+open(FILE, "+<$file") || die "Unable to open $file";
+binmode FILE;
+seek (FILE, $ENV{INNODB_PAGE_SIZE} * 6, SEEK_SET) or die "seek";
+print FILE "junk";
+close FILE or die "close";
+EOF
+
+--source include/start_mysqld.inc
+
+--replace_result $MYSQLTEST_VARDIR MYSQLTEST_VARDIR
+--error ER_GET_ERRMSG, 192
+eval load data infile '$MYSQLTEST_VARDIR/tmp/t1.outfile' into table t2;
+
+drop table t1, t2;
+--remove_file $MYSQLTEST_VARDIR/tmp/t1.outfile
diff --git a/storage/innobase/buf/buf0buf.cc b/storage/innobase/buf/buf0buf.cc
index 41d9efe25cc..ae90ded5df9 100644
--- a/storage/innobase/buf/buf0buf.cc
+++ b/storage/innobase/buf/buf0buf.cc
@@ -4796,6 +4796,25 @@ evict_from_pool:
 	and block->lock. */
 	buf_wait_for_read(fix_block);
 
+	if (fix_block->page.id != page_id) {
+
+		buf_block_unfix(fix_block);
+
+#ifdef UNIV_DEBUG
+		if (!fsp_is_system_temporary(page_id.space())) {
+			rw_lock_s_unlock(&fix_block->debug_latch);
+		}
+#endif /* UNIV_DEBUG */
+
+		if (err) {
+			*err = DB_PAGE_CORRUPTED;
+		}
+
+		return NULL;
+	}
+
+	ut_ad(fix_block->page.id == page_id);
+
 	mtr_memo_type_t	fix_type;
 
 	switch (rw_latch) {
@@ -5760,7 +5779,9 @@ buf_mark_space_corrupt(buf_page_t* bpage)
 	buf_pool_mutex_enter(buf_pool);
 	mutex_enter(buf_page_get_mutex(bpage));
 	ut_ad(buf_page_get_io_fix(bpage) == BUF_IO_READ);
-	ut_ad(bpage->buf_fix_count == 0);
+
+	/* buf_fix_count can be greater than zero. Because other thread
+	can wait in buf_page_wait_read() for the page to be read. */
 
 	/* Set BUF_IO_NONE before we remove the block from LRU list */
 	buf_page_set_io_fix(bpage, BUF_IO_NONE);
diff --git a/storage/innobase/buf/buf0lru.cc b/storage/innobase/buf/buf0lru.cc
index 25a64fc81cd..e0902f5d00b 100644
--- a/storage/innobase/buf/buf0lru.cc
+++ b/storage/innobase/buf/buf0lru.cc
@@ -2171,7 +2171,7 @@ Remove one page from LRU list and put it to free list */
 void
 buf_LRU_free_one_page(
 /*==================*/
-	buf_page_t*	bpage)	/*!< in/out: block, must contain a file page and
+	buf_page_t*	bpage)  /*!< in/out: block, must contain a file page and
 				be in a state where it can be freed; there
 				may or may not be a hash index to the page */
 {
@@ -2182,7 +2182,19 @@ buf_LRU_free_one_page(
 
 	ut_ad(buf_pool_mutex_own(buf_pool));
 
+	page_id_t old_page_id = bpage->id;
+
+	bpage->id.set_corrupt_id();
+
 	rw_lock_x_lock(hash_lock);
+
+	while (bpage->buf_fix_count > 0) {
+		/* Wait for other threads to release the fix count
+		before releasing the bpage from LRU list. */
+	}
+
+	bpage->id = old_page_id;
+
 	mutex_enter(block_mutex);
 
 	if (buf_LRU_block_remove_hashed(bpage, true)) {
diff --git a/storage/innobase/include/buf0buf.h b/storage/innobase/include/buf0buf.h
index 58314bed71a..c483a8b7f25 100644
--- a/storage/innobase/include/buf0buf.h
+++ b/storage/innobase/include/buf0buf.h
@@ -265,6 +265,12 @@ public:
 		ut_ad(page_no <= 0xFFFFFFFFU);
 	}
 
+	/** Set the FIL_NULL for the space and page_no */
+	inline void set_corrupt_id()
+	{
+		m_space = m_page_no = FIL_NULL;
+	}
+
 private:
 
 	/** Tablespace id. */
diff --git a/storage/innobase/include/buf0lru.h b/storage/innobase/include/buf0lru.h
index 4a7570bf7ff..771104607eb 100644
--- a/storage/innobase/include/buf0lru.h
+++ b/storage/innobase/include/buf0lru.h
@@ -206,7 +206,7 @@ Remove one page from LRU list and put it to free list */
 void
 buf_LRU_free_one_page(
 /*==================*/
-	buf_page_t*	bpage)	/*!< in/out: block, must contain a file page and
+	buf_page_t*	bpage) /*!< in/out: block, must contain a file page and
 				be in a state where it can be freed; there
 				may or may not be a hash index to the page */
 	MY_ATTRIBUTE((nonnull));
diff --git a/storage/innobase/row/row0row.cc b/storage/innobase/row/row0row.cc
index 3bb351eed8f..3e65dc1d28b 100644
--- a/storage/innobase/row/row0row.cc
+++ b/storage/innobase/row/row0row.cc
@@ -980,7 +980,10 @@ row_search_on_row_ref(
 
 	ut_a(dtuple_get_n_fields(ref) == dict_index_get_n_unique(index));
 
-	btr_pcur_open(index, ref, PAGE_CUR_LE, mode, pcur, mtr);
+	if (btr_pcur_open(index, ref, PAGE_CUR_LE, mode, pcur, mtr)
+	    != DB_SUCCESS) {
+		return FALSE;
+	}
 
 	low_match = btr_pcur_get_low_match(pcur);