MDEV-19435 buf_fix_count > 0 for corrupted page when it exits the LRU listbb-10.3-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().

12 files changed, 162 insertions, 27 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..6e84484b2c3
--- /dev/null
+++ b/mysql-test/suite/encryption/r/load_infile.result
@@ -0,0 +1,8 @@
+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";
+CREATE TABLE t2 (pk INT, c CHAR(255)) ENGINE=INNODB ENCRYPTED=YES;
+INSERT INTO t2 SELECT NULL, 'c' FROM seq_1_to_1000;
+# 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..d0560da611e
--- /dev/null
+++ b/mysql-test/suite/encryption/t/load_infile.test
@@ -0,0 +1,55 @@
+-- source include/have_innodb.inc
+-- source include/have_file_key_management_plugin.inc
+
+--disable_query_log
+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: [0-9][0-9][0-9]* btr_pcur_open_low  level: 0 called from file: .*");
+--enable_query_log
+
+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, REPEAT('x', 32));
+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";
+
+CREATE TABLE t2 (pk INT, c CHAR(255)) ENGINE=INNODB ENCRYPTED=YES;
+--source include/have_sequence.inc
+INSERT INTO t2 SELECT NULL, 'c' FROM seq_1_to_1000;
+
+--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/mysql-test/suite/innodb/t/leaf_page_corrupted_during_recovery.test b/mysql-test/suite/innodb/t/leaf_page_corrupted_during_recovery.test
index f9f2b11471a..b49cbe01a0c 100644
--- a/mysql-test/suite/innodb/t/leaf_page_corrupted_during_recovery.test
+++ b/mysql-test/suite/innodb/t/leaf_page_corrupted_during_recovery.test
@@ -9,6 +9,7 @@ call mtr.add_suppression("\\[ERROR\\] InnoDB: Plugin initialization aborted at s
 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");
+call mtr.add_suppression("InnoDB:  Error code: [0-9][0-9][0-9]* btr_pcur_open_low  level: 0 called from file: .*");
 --enable_query_log
 CREATE TABLE t1 (pk INT PRIMARY KEY, c CHAR(255))ENGINE=InnoDB STATS_PERSISTENT=0;
 
@@ -46,8 +47,6 @@ EOF
 SELECT * FROM t1 WHERE PK = 1;
 
 let $restart_parameters=--innodb-force-recovery=1;
-# Work around MDEV-19435 to avoid crash in row_purge_reset_trx_id()
-let $restart_parameters=--innodb-force-recovery=2;
 --source include/restart_mysqld.inc
 SELECT * FROM t1 WHERE PK = 1;
 --error ER_NOT_KEYFILE
diff --git a/storage/innobase/buf/buf0buf.cc b/storage/innobase/buf/buf0buf.cc
index 9182275c695..2a7ce5235ae 100644
--- a/storage/innobase/buf/buf0buf.cc
+++ b/storage/innobase/buf/buf0buf.cc
@@ -4850,6 +4850,23 @@ 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;
+	}
+
 	mtr_memo_type_t	fix_type;
 
 	switch (rw_latch) {
@@ -5820,14 +5837,18 @@ 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);
+	page_id_t	old_page_id = bpage->id;
 
 	/* First unfix and release lock on the 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);
 	ut_ad(bpage->id.space() == space->id);
 
+	/* buf_fix_count can be greater than zero. Because other thread
+	can wait in buf_page_wait_read() for the page to be read. */
+
+	bpage->id.set_corrupt_id();
 	/* Set BUF_IO_NONE before we remove the block from LRU list */
 	buf_page_set_io_fix(bpage, BUF_IO_NONE);
 
@@ -5844,7 +5865,7 @@ buf_corrupt_page_release(buf_page_t* bpage, const fil_space_t* space)
 	}
 
 	/* After this point bpage can't be referenced. */
-	buf_LRU_free_one_page(bpage);
+	buf_LRU_free_one_page(bpage, old_page_id);
 
 	ut_ad(buf_pool->n_pend_reads > 0);
 	buf_pool->n_pend_reads--;
diff --git a/storage/innobase/buf/buf0lru.cc b/storage/innobase/buf/buf0lru.cc
index c00dd414aa6..6cef80e1ce4 100644
--- a/storage/innobase/buf/buf0lru.cc
+++ b/storage/innobase/buf/buf0lru.cc
@@ -2165,25 +2165,34 @@ buf_LRU_block_free_hashed_page(
 	buf_page_mutex_exit(block);
 }
 
-/******************************************************************//**
-Remove one page from LRU list and put it to free list */
+/** Remove one page from LRU list and put it to free list.
+@param[in,out]	bpage		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
+@param[in]	old_page_id	old page number before setting corruption id. */
 void
 buf_LRU_free_one_page(
-/*==================*/
-	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 */
+	buf_page_t*	bpage,
+	page_id_t	old_page_id)
 {
 	buf_pool_t*	buf_pool = buf_pool_from_bpage(bpage);
 
-	rw_lock_t*	hash_lock = buf_page_hash_lock_get(buf_pool, bpage->id);
+	rw_lock_t*	hash_lock = buf_page_hash_lock_get(buf_pool, old_page_id);
 	BPageMutex*	block_mutex = buf_page_get_mutex(bpage);
 
 	ut_ad(buf_pool_mutex_own(buf_pool));
 
 	rw_lock_x_lock(hash_lock);
+
+	while (buf_block_get_fix(bpage) > 0) {
+		/* Wait for other threads to release the fix count
+		before releasing the bpage from LRU list. */
+	}
+
 	mutex_enter(block_mutex);
 
+	bpage->id = old_page_id;
+
 	if (buf_LRU_block_remove_hashed(bpage, true)) {
 		buf_LRU_block_free_hashed_page((buf_block_t*) bpage);
 	}
diff --git a/storage/innobase/buf/buf0rea.cc b/storage/innobase/buf/buf0rea.cc
index 76eb01c8e91..9dd5857df17 100644
--- a/storage/innobase/buf/buf0rea.cc
+++ b/storage/innobase/buf/buf0rea.cc
@@ -63,13 +63,14 @@ buf_read_page_handle_error(
 	buf_pool_t*	buf_pool = buf_pool_from_bpage(bpage);
 	const bool	uncompressed = (buf_page_get_state(bpage)
 					== BUF_BLOCK_FILE_PAGE);
+	const page_id_t	old_page_id = bpage->id;
 
 	/* First unfix and release lock on the 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);
 
+	bpage->id.set_corrupt_id();
 	/* Set BUF_IO_NONE before we remove the block from LRU list */
 	buf_page_set_io_fix(bpage, BUF_IO_NONE);
 
@@ -82,7 +83,7 @@ buf_read_page_handle_error(
 	mutex_exit(buf_page_get_mutex(bpage));
 
 	/* remove the block from LRU list */
-	buf_LRU_free_one_page(bpage);
+	buf_LRU_free_one_page(bpage, old_page_id);
 
 	ut_ad(buf_pool->n_pend_reads > 0);
 	buf_pool->n_pend_reads--;
diff --git a/storage/innobase/include/buf0buf.h b/storage/innobase/include/buf0buf.h
index e7593cf482d..1af5455ed43 100644
--- a/storage/innobase/include/buf0buf.h
+++ b/storage/innobase/include/buf0buf.h
@@ -1435,7 +1435,7 @@ public:
 	page_size_t	size;
 
 	/** Count of how manyfold this block is currently bufferfixed. */
-	ib_uint32_t	buf_fix_count;
+	int32		buf_fix_count;
 
 	/** type of pending I/O operation; also protected by
 	buf_pool->mutex for writes only */
diff --git a/storage/innobase/include/buf0buf.ic b/storage/innobase/include/buf0buf.ic
index eccf723faa6..0525616b05a 100644
--- a/storage/innobase/include/buf0buf.ic
+++ b/storage/innobase/include/buf0buf.ic
@@ -953,7 +953,9 @@ ulint
 buf_block_fix(
 	buf_page_t*	bpage)
 {
-	return uint32(my_atomic_add32((int32*) &bpage->buf_fix_count, 1) + 1);
+	return uint32(my_atomic_add32_explicit(
+				(int32*) &bpage->buf_fix_count, 1,
+				MY_MEMORY_ORDER_RELAXED) + 1);
 }
 
 /** Increments the bufferfix count.
@@ -967,6 +969,29 @@ buf_block_fix(
 	return(buf_block_fix(&block->page));
 }
 
+/** Get the bufferfix count.
+@param[in]	bpage	block to bufferfix
+@return the count */
+UNIV_INLINE
+ulint
+buf_block_get_fix(
+	buf_page_t*	bpage)
+{
+	return my_atomic_load32_explicit(&bpage->buf_fix_count,
+					 MY_MEMORY_ORDER_RELAXED);
+}
+
+/** Get the bufferfix count.
+@param[in]	bpage	block to bufferfix
+@return the count */
+UNIV_INLINE
+ulint
+buf_block_get_fix(
+	buf_block_t*	block)
+{
+	return(buf_block_get_fix(&block->page));
+}
+
 /*******************************************************************//**
 Increments the bufferfix count. */
 UNIV_INLINE
@@ -1001,8 +1026,9 @@ ulint
 buf_block_unfix(
 	buf_page_t*	bpage)
 {
-	uint32	count = uint32(my_atomic_add32((int32*) &bpage->buf_fix_count,
-					       -1));
+	uint32	count = uint32(my_atomic_add32_explicit(
+					(int32*) &bpage->buf_fix_count,
+					-1, MY_MEMORY_ORDER_RELAXED));
 	ut_ad(count != 0);
 	return count - 1;
 }
diff --git a/storage/innobase/include/buf0lru.h b/storage/innobase/include/buf0lru.h
index 6583629be8c..2568e1f15cd 100644
--- a/storage/innobase/include/buf0lru.h
+++ b/storage/innobase/include/buf0lru.h
@@ -33,6 +33,7 @@ Created 11/5/1995 Heikki Tuuri
 // Forward declaration
 struct trx_t;
 struct fil_space_t;
+class page_id_t;
 
 /******************************************************************//**
 Returns TRUE if less than 25 % of the buffer pool is available. This can be
@@ -202,14 +203,15 @@ void
 buf_LRU_stat_update(void);
 /*=====================*/
 
-/******************************************************************//**
-Remove one page from LRU list and put it to free list */
+/** Remove one page from LRU list and put it to free list
+@param[in,out]	bpage		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
+@param[in]	old_page_id	old page number before setting corruption id. */
 void
 buf_LRU_free_one_page(
-/*==================*/
-	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 */
+	buf_page_t*	bpage,
+	page_id_t	old_page_id)
 	MY_ATTRIBUTE((nonnull));
 
 /******************************************************************//**
diff --git a/storage/innobase/include/buf0types.h b/storage/innobase/include/buf0types.h
index bc74b59f3b1..bd5e26df47b 100644
--- a/storage/innobase/include/buf0types.h
+++ b/storage/innobase/include/buf0types.h
@@ -175,6 +175,12 @@ public:
 		ut_ad(page_no <= 0xFFFFFFFFU);
 	}
 
+	/** Set the FIL_NULL for the space and page_no */
+	void set_corrupt_id()
+	{
+		m_space = m_page_no = ULINT32_UNDEFINED;
+	}
+
 private:
 
 	/** Tablespace id. */
diff --git a/storage/innobase/include/log0recv.h b/storage/innobase/include/log0recv.h
index 927e04ce04e..4c36d805395 100644
--- a/storage/innobase/include/log0recv.h
+++ b/storage/innobase/include/log0recv.h
@@ -36,6 +36,8 @@ Created 9/20/1997 Heikki Tuuri
 #include <list>
 #include <vector>
 
+class page_id_t;
+
 /** Is recv_writer_thread active? */
 extern bool	recv_writer_thread_active;
 
diff --git a/storage/innobase/row/row0row.cc b/storage/innobase/row/row0row.cc
index f3b40ce0900..a2eb6ea6cf4 100644
--- a/storage/innobase/row/row0row.cc
+++ b/storage/innobase/row/row0row.cc
@@ -1033,9 +1033,12 @@ row_search_on_row_ref(
 	if (UNIV_UNLIKELY(ref->info_bits != 0)) {
 		ut_ad(ref->info_bits == REC_INFO_METADATA);
 		ut_ad(ref->n_fields <= index->n_uniq);
-		btr_pcur_open_at_index_side(true, index, mode, pcur, true, 0,
-					    mtr);
-		btr_pcur_move_to_next_user_rec(pcur, mtr);
+		if (btr_pcur_open_at_index_side(
+			    true, index, mode, pcur, true, 0, mtr)
+		    != DB_SUCCESS
+		    || !btr_pcur_move_to_next_user_rec(pcur, mtr)) {
+			return FALSE;
+		}
 		/* We do not necessarily have index->is_instant() here,
 		because we could be executing a rollback of an
 		instant ADD COLUMN operation. The function
@@ -1046,7 +1049,10 @@ row_search_on_row_ref(
 			& REC_INFO_MIN_REC_FLAG;
 	} else {
 		ut_a(ref->n_fields == index->n_uniq);
-		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);