Merge 10.4 into 10.5

mysql-test/suite/galera_sr/r/GCF-597.result

@@ -15,7 +15,13 @@ INSERT INTO t1 VALUES (2);
 INSERT INTO t1 VALUES (3);
 INSERT INTO t1 VALUES (4);
 INSERT INTO t1 VALUES (5);
+connect node_1a, 127.0.0.1, root, , test, $NODE_MYPORT_1;
+connection node_1a;
+connect node_2a, 127.0.0.1, root, , test, $NODE_MYPORT_2;
+connection node_2a;
 connection node_2;
 ROLLBACK;
 ERROR 40001: Deadlock found when trying to get lock; try restarting transaction
 DROP TABLE t1;
+disconnect node_1a;
+disconnect node_2a;

mysql-test/suite/galera_sr/r/galera_sr_kill_query.result

@@ -17,17 +17,25 @@ SET TRANSACTION ISOLATION LEVEL REPEATABLE READ;
 SELECT COUNT(*) AS EXPECT_0 FROM t1;
 EXPECT_0
 0
-SELECT COUNT(*) AS EXPECT_0 FROM mysql.wsrep_streaming_log;
-EXPECT_0
-0
 INSERT INTO t1 SELECT 1 FROM ten AS t1, ten AS t2, ten AS t3;
 SELECT COUNT(*) AS EXPECT_1000 FROM t1;
 EXPECT_1000
 1000
-connection node_1a;
+connection node_1;
 SET GLOBAL wsrep_sync_wait=15;
 SELECT COUNT(*) AS EXPECT_1000 FROM t1;
 EXPECT_1000
 1000
+SELECT COUNT(*) AS EXPECT_0 FROM mysql.wsrep_streaming_log;
+EXPECT_0
+0
+connection node_2;
+SELECT COUNT(*) AS EXPECT_1000 FROM t1;
+EXPECT_1000
+1000
+SELECT COUNT(*) AS EXPECT_0 FROM mysql.wsrep_streaming_log;
+EXPECT_0
+0
+connection node_1;
 DROP TABLE t1;
 DROP TABLE ten;

mysql-test/suite/galera_sr/r/galera_sr_kill_slave.result

@@ -2,9 +2,9 @@ connection node_2;
 connection node_1;
 CREATE TABLE t1 (f1 INTEGER PRIMARY KEY) ENGINE=InnoDB;
 connection node_2;
-SELECT COUNT(*) = 0 FROM t1;
-COUNT(*) = 0
-1
+SELECT COUNT(*) AS EXPECT_0 FROM t1;
+EXPECT_0
+0
 connection node_1;
 CREATE TABLE t2 (f1 INTEGER);
 connect node_2a, 127.0.0.1, root, , test, $NODE_MYPORT_2;
@@ -39,16 +39,16 @@ INSERT INTO t1 VALUES (13);
 INSERT INTO t1 VALUES (14);
 INSERT INTO t1 VALUES (15);
 COMMIT;
-SELECT COUNT(*) = 0 FROM mysql.wsrep_streaming_log;
-COUNT(*) = 0
-1
+SELECT COUNT(*) AS EXPECT_0 FROM mysql.wsrep_streaming_log;
+EXPECT_0
+0
 connection node_2;
-SELECT COUNT(*) = 15 FROM t1;
-COUNT(*) = 15
-1
-SELECT COUNT(*) = 0 FROM mysql.wsrep_streaming_log;
-COUNT(*) = 0
-1
+SELECT COUNT(*) AS EXPECT_15 FROM t1;
+EXPECT_15
+15
+SELECT COUNT(*) AS EXPECT_0 FROM mysql.wsrep_streaming_log;
+EXPECT_0
+0
 connection node_1;
 DROP TABLE t1;
 DROP TABLE t2;

mysql-test/suite/galera_sr/t/GCF-597.test

@@ -22,8 +22,21 @@ INSERT INTO t1 VALUES (3);
 INSERT INTO t1 VALUES (4);
 INSERT INTO t1 VALUES (5);
 
+--connect node_1a, 127.0.0.1, root, , test, $NODE_MYPORT_1
+--connection node_1a
+--let $wait_condition = SELECT COUNT(*) = 5 FROM mysql.wsrep_streaming_log
+--source include/wait_condition.inc
+
+--connect node_2a, 127.0.0.1, root, , test, $NODE_MYPORT_2
+--connection node_2a
+--let $wait_condition = SELECT COUNT(*) = 5 FROM mysql.wsrep_streaming_log
+--source include/wait_condition.inc
+
 --connection node_2
 --error ER_LOCK_DEADLOCK
 ROLLBACK;
 
 DROP TABLE t1;
+
+--disconnect node_1a
+--disconnect node_2a

mysql-test/suite/galera_sr/t/galera_sr_kill_query.test

@@ -33,18 +33,27 @@ SET TRANSACTION ISOLATION LEVEL READ UNCOMMITTED;
 
 # Confirm that the kill caused the updates made so far to be removed
 --connection node_2
+--let $wait_condition = SELECT COUNT(*) = 0 FROM t1
+--source include/wait_condition.inc
 SET TRANSACTION ISOLATION LEVEL REPEATABLE READ;
 SELECT COUNT(*) AS EXPECT_0 FROM t1;
-SELECT COUNT(*) AS EXPECT_0 FROM mysql.wsrep_streaming_log;
 
 # Confirm that the transaction can be reissued in its entirety on the slave without a conflict
 
 INSERT INTO t1 SELECT 1 FROM ten AS t1, ten AS t2, ten AS t3;
 SELECT COUNT(*) AS EXPECT_1000 FROM t1;
 
---connection node_1a
+--connection node_1
 SET GLOBAL wsrep_sync_wait=15;
+--let $wait_condition = SELECT COUNT(*) = 1000 FROM t1;
+--source include/wait_condition.inc
+SELECT COUNT(*) AS EXPECT_1000 FROM t1;
+SELECT COUNT(*) AS EXPECT_0 FROM mysql.wsrep_streaming_log;
+
+--connection node_2
 SELECT COUNT(*) AS EXPECT_1000 FROM t1;
+SELECT COUNT(*) AS EXPECT_0 FROM mysql.wsrep_streaming_log;
 
+--connection node_1
 DROP TABLE t1;
 DROP TABLE ten;

mysql-test/suite/galera_sr/t/galera_sr_kill_slave.test

@@ -13,12 +13,17 @@ CREATE TABLE t1 (f1 INTEGER PRIMARY KEY) ENGINE=InnoDB;
 # Block node #2's applier before table t1's inserts have come into play
 
 --connection node_2
-SELECT COUNT(*) = 0 FROM t1;
+--let $wait_condition = SELECT COUNT(*) = 1 FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_NAME = 't1'
+--source include/wait_condition.inc
+SELECT COUNT(*) AS EXPECT_0 FROM t1;
 
 --connection node_1
 CREATE TABLE t2 (f1 INTEGER);
 
 --connect node_2a, 127.0.0.1, root, , test, $NODE_MYPORT_2
+--let $wait_condition = SELECT COUNT(*) = 1 FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_NAME = 't2'
+--source include/wait_condition.inc
+
 LOCK TABLE t2 WRITE;
 
 --connection node_1
@@ -39,11 +44,8 @@ INSERT INTO t1 VALUES (3);
 INSERT INTO t1 VALUES (4);
 INSERT INTO t1 VALUES (5);
 
---sleep 2
-
 --connection node_2
 --source include/kill_galera.inc
---sleep 1
 
 --connection node_1
 INSERT INTO t1 VALUES (6);
@@ -54,8 +56,6 @@ INSERT INTO t1 VALUES (10);
 
 --connection node_2
 --source include/start_mysqld.inc
---sleep 1
-
 --source include/wait_until_connected_again.inc
 --source include/galera_wait_ready.inc
 
@@ -67,12 +67,17 @@ INSERT INTO t1 VALUES (14);
 INSERT INTO t1 VALUES (15);
 COMMIT;
 
-SELECT COUNT(*) = 0 FROM mysql.wsrep_streaming_log;
+--let $wait_condition = SELECT COUNT(*) = 0 FROM mysql.wsrep_streaming_log
+--source include/wait_condition.inc
+
+SELECT COUNT(*) AS EXPECT_0 FROM mysql.wsrep_streaming_log;
 
 --connection node_2
---sleep 5
-SELECT COUNT(*) = 15 FROM t1;
-SELECT COUNT(*) = 0 FROM mysql.wsrep_streaming_log;
+--let $wait_condition = SELECT COUNT(*) = 15 FROM t1
+--source include/wait_condition.inc
+
+SELECT COUNT(*) AS EXPECT_15 FROM t1;
+SELECT COUNT(*) AS EXPECT_0 FROM mysql.wsrep_streaming_log;
 
 --connection node_1
 

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

@@ -736,6 +736,17 @@ t2	CREATE TABLE `t2` (
 CREATE TABLE t2 (f1 INT NOT NULL)ENGINE=InnoDB;
 ERROR 42S01: Table 't2' already exists
 DROP TABLE t2, t1;
+#
+# MDEV-23685 SIGSEGV on ADD FOREIGN KEY after failed attempt
+# to create unique key on virtual column
+#
+CREATE TABLE t1 (pk INT PRIMARY KEY, a INT, b INT AS (a)) ENGINE=InnODB;
+INSERT INTO t1 (pk,a) VALUES (1,10),(2,10);
+ALTER TABLE t1 ADD UNIQUE INDEX ind9 (b), LOCK=SHARED;
+ERROR 23000: Duplicate entry '10' for key 'ind9'
+SET FOREIGN_KEY_CHECKS= 0;
+ALTER TABLE t1 ADD FOREIGN KEY (a) REFERENCES t1 (pk);
+DROP TABLE t1;
 # End of 10.2 tests
 CREATE TABLE t1 (a GEOMETRY, INDEX(a(8)),
 FOREIGN KEY (a) REFERENCES x (xx)) ENGINE=InnoDB;

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

@@ -720,6 +720,19 @@ SHOW CREATE TABLE t2;
 CREATE TABLE t2 (f1 INT NOT NULL)ENGINE=InnoDB;
 DROP TABLE t2, t1;
 
+--echo #
+--echo # MDEV-23685 SIGSEGV on ADD FOREIGN KEY after failed attempt
+--echo # to create unique key on virtual column
+--echo #
+CREATE TABLE t1 (pk INT PRIMARY KEY, a INT, b INT AS (a)) ENGINE=InnODB;
+
+INSERT INTO t1 (pk,a) VALUES (1,10),(2,10);
+--error ER_DUP_ENTRY
+ALTER TABLE t1 ADD UNIQUE INDEX ind9 (b), LOCK=SHARED;
+SET FOREIGN_KEY_CHECKS= 0;
+ALTER TABLE t1 ADD FOREIGN KEY (a) REFERENCES t1 (pk);
+DROP TABLE t1;
+
 --echo # End of 10.2 tests
 
 # MDEV-21792 Server aborts upon attempt to create foreign key on spatial field

storage/innobase/btr/btr0btr.cc

@@ -1025,8 +1025,7 @@ btr_create(
 	if (UNIV_UNLIKELY(type & DICT_IBUF)) {
 		/* Allocate first the ibuf header page */
 		buf_block_t*	ibuf_hdr_block = fseg_create(
-			space, 0,
-			IBUF_HEADER + IBUF_TREE_SEG_HEADER, mtr);
+			space, IBUF_HEADER + IBUF_TREE_SEG_HEADER, mtr);
 
 		if (ibuf_hdr_block == NULL) {
 			return(FIL_NULL);
@@ -1056,8 +1055,8 @@ btr_create(
 
 		flst_init(block, PAGE_HEADER + PAGE_BTR_IBUF_FREE_LIST, mtr);
 	} else {
-		block = fseg_create(space, 0,
-				    PAGE_HEADER + PAGE_BTR_SEG_TOP, mtr);
+		block = fseg_create(space, PAGE_HEADER + PAGE_BTR_SEG_TOP,
+				    mtr);
 
 		if (block == NULL) {
 			return(FIL_NULL);
@@ -1065,8 +1064,8 @@ btr_create(
 
 		buf_block_dbg_add_level(block, SYNC_TREE_NODE_NEW);
 
-		if (!fseg_create(space, block->page.id().page_no(),
-				 PAGE_HEADER + PAGE_BTR_SEG_LEAF, mtr)) {
+		if (!fseg_create(space, PAGE_HEADER + PAGE_BTR_SEG_LEAF, mtr,
+				 false, block)) {
 			/* Not enough space for new segment, free root
 			segment before return. */
 			btr_free_root(block, mtr);

storage/innobase/buf/buf0buf.cc

@@ -3823,6 +3823,7 @@ buf_page_create(fil_space_t *space, uint32_t offset,
   ut_ad(page_id.space() != 0 || !zip_size);
 
   space->free_page(offset, false);
+loop:
   buf_block_t *free_block= buf_LRU_get_free_block(false);
   free_block->initialise(page_id, zip_size, 1);
 
@@ -3837,35 +3838,71 @@ buf_page_create(fil_space_t *space, uint32_t offset,
       !buf_pool.watch_is_sentinel(block->page))
   {
 #ifdef BTR_CUR_HASH_ADAPT
-    const bool drop_hash_entry= block->page.state() == BUF_BLOCK_FILE_PAGE &&
-      UNIV_LIKELY_NULL(block->index);
-    if (UNIV_UNLIKELY(drop_hash_entry))
+    const dict_index_t *drop_hash_entry= nullptr;
+#endif
+    switch (block->page.state()) {
+    default:
+      ut_ad(0);
+      break;
+    case BUF_BLOCK_FILE_PAGE:
+      buf_block_buf_fix_inc(block, __FILE__, __LINE__);
+      {
+        const auto num_fix_count= mtr->get_fix_count(block) + 1;
+        while (block->page.io_fix() != BUF_IO_NONE ||
+               num_fix_count != block->page.buf_fix_count())
+        {
+          mutex_exit(&buf_pool.mutex);
+          os_thread_yield();
+          mutex_enter(&buf_pool.mutex);
+        }
+      }
       rw_lock_x_lock(&block->lock);
-#endif /* BTR_CUR_HASH_ADAPT */
-
-    /* Page can be found in buf_pool */
+#ifdef BTR_CUR_HASH_ADAPT
+      drop_hash_entry= block->index;
+#endif
       buf_LRU_block_free_non_file_page(free_block);
+      break;
+    case BUF_BLOCK_ZIP_PAGE:
+      page_hash_latch *hash_lock= buf_pool.page_hash.lock_get(fold);
+      hash_lock->write_lock();
+      if (block->page.io_fix() != BUF_IO_NONE)
+      {
+        hash_lock->write_unlock();
+        buf_LRU_block_free_non_file_page(block);
+        mutex_exit(&buf_pool.mutex);
+        goto loop;
+      }
+
+      rw_lock_x_lock(&free_block->lock);
+      buf_relocate(&block->page, &free_block->page);
+
+      if (block->page.oldest_modification() > 0)
+        buf_flush_relocate_on_flush_list(&block->page, &free_block->page);
+#ifdef UNIV_DEBUG
+      else
+        UT_LIST_REMOVE(buf_pool.zip_clean, &block->page);
+#endif
+
+      free_block->page.set_state(BUF_BLOCK_FILE_PAGE);
+      free_block->lock_hash_val= lock_rec_hash(
+        page_id.space(), page_id.page_no());
+      buf_unzip_LRU_add_block(free_block, FALSE);
+      hash_lock->write_unlock();
+      buf_page_free_descriptor(&block->page);
+      block= free_block;
+      buf_block_buf_fix_inc(block, __FILE__, __LINE__);
+      break;
+    }
+
     mutex_exit(&buf_pool.mutex);
 
 #ifdef BTR_CUR_HASH_ADAPT
-    if (UNIV_UNLIKELY(drop_hash_entry))
-    {
+    if (drop_hash_entry)
       btr_search_drop_page_hash_index(block);
-      rw_lock_x_unlock(&block->lock);
-    }
 #endif /* BTR_CUR_HASH_ADAPT */
 
-    if (!recv_recovery_is_on())
-      /* FIXME: Remove the redundant lookup and avoid
-      the unnecessary invocation of buf_zip_decompress().
-      We may have to convert buf_page_t to buf_block_t,
-      but we are going to initialize the page. */
-      return buf_page_get_gen(page_id, zip_size, RW_NO_LATCH,
-                              block, BUF_GET_POSSIBLY_FREED,
-                              __FILE__, __LINE__, mtr);
-    mutex_exit(&recv_sys.mutex);
-    block= buf_page_get_with_no_latch(page_id, zip_size, mtr);
-    mutex_enter(&recv_sys.mutex);
+    mtr_memo_push(mtr, block, MTR_MEMO_PAGE_X_FIX);
+
     return block;
   }
 
@@ -3889,13 +3926,13 @@ buf_page_create(fil_space_t *space, uint32_t offset,
   ut_d(block->page.in_page_hash= true);
   HASH_INSERT(buf_page_t, hash, &buf_pool.page_hash, fold, &block->page);
 
+  rw_lock_x_lock(&block->lock);
   if (UNIV_UNLIKELY(zip_size))
   {
     /* Prevent race conditions during buf_buddy_alloc(), which may
     release and reacquire buf_pool.mutex, by IO-fixing and X-latching
     the block. */
     block->page.set_io_fix(BUF_IO_READ);
-    rw_lock_x_lock(&block->lock);
     hash_lock->write_unlock();
 
     /* buf_pool.mutex may be released and reacquired by
@@ -3911,14 +3948,13 @@ buf_page_create(fil_space_t *space, uint32_t offset,
     buf_unzip_LRU_add_block(block, FALSE);
 
     block->page.set_io_fix(BUF_IO_NONE);
-    rw_lock_x_unlock(&block->lock);
   }
   else
     hash_lock->write_unlock();
 
   mutex_exit(&buf_pool.mutex);
 
-  mtr->memo_push(block, MTR_MEMO_BUF_FIX);
+  mtr->memo_push(block, MTR_MEMO_PAGE_X_FIX);
   block->page.set_accessed();
   buf_pool.stat.n_pages_created++;
 

storage/innobase/buf/buf0dblwr.cc

@@ -166,9 +166,9 @@ buf_dblwr_create()
 		}
 	}
 
-	block2 = fseg_create(fil_system.sys_space, TRX_SYS_PAGE_NO,
-			     TRX_SYS_DOUBLEWRITE
-			     + TRX_SYS_DOUBLEWRITE_FSEG, &mtr);
+	block2 = fseg_create(fil_system.sys_space,
+			     TRX_SYS_DOUBLEWRITE + TRX_SYS_DOUBLEWRITE_FSEG,
+			     &mtr, false, trx_sys_block);
 
 	if (block2 == NULL) {
 too_small:

storage/innobase/dict/dict0boot.cc

@@ -134,7 +134,7 @@ dict_hdr_create(
 
 	/* Create the dictionary header file block in a new, allocated file
 	segment in the system tablespace */
-	block = fseg_create(fil_system.sys_space, 0,
+	block = fseg_create(fil_system.sys_space,
 			    DICT_HDR + DICT_HDR_FSEG_HEADER, mtr);
 
 	ut_a(block->page.id() == page_id_t(DICT_HDR_SPACE, DICT_HDR_PAGE_NO));

storage/innobase/dict/dict0dict.cc

@@ -5027,7 +5027,11 @@ dict_foreign_qualify_index(
 		return(false);
 	}
 
-	if (index->type & (DICT_SPATIAL | DICT_FTS)) {
+	if (index->type & (DICT_SPATIAL | DICT_FTS | DICT_CORRUPT)) {
+		return false;
+	}
+
+	if (index->online_status >= ONLINE_INDEX_ABORTED) {
 		return false;
 	}
 

storage/innobase/fsp/fsp0fsp.cc

@@ -101,7 +101,6 @@ minimize file space fragmentation.
 @param[in]	direction		if the new page is needed because of
 an index page split, and records are inserted there in order, into which
 direction they go alphabetically: FSP_DOWN, FSP_UP, FSP_NO_DIR
-@param[in]	rw_latch		RW_SX_LATCH, RW_X_LATCH
 @param[in,out]	mtr			mini-transaction
 @param[in,out]	init_mtr		mtr or another mini-transaction in
 which the page should be initialized.
@@ -114,7 +113,6 @@ fseg_alloc_free_page_low(
 	buf_block_t*		iblock,
 	ulint			hint,
 	byte			direction,
-	rw_lock_type_t		rw_latch,
 #ifdef UNIV_DEBUG
 	bool			has_done_reservation,
 	/*!< whether the space has already been reserved */
@@ -147,7 +145,8 @@ template<bool free>
 inline void xdes_set_free(const buf_block_t &block, xdes_t *descr,
                           ulint offset, mtr_t *mtr)
 {
-  ut_ad(mtr->memo_contains_flagged(&block, MTR_MEMO_PAGE_SX_FIX));
+  ut_ad(mtr->memo_contains_flagged(&block, MTR_MEMO_PAGE_SX_FIX |
+                                   MTR_MEMO_PAGE_X_FIX));
   ut_ad(offset < FSP_EXTENT_SIZE);
   ut_ad(page_align(descr) == block.frame);
   compile_time_assert(XDES_BITS_PER_PAGE == 2);
@@ -218,7 +217,8 @@ inline void xdes_set_state(const buf_block_t &block, xdes_t *descr,
   ut_ad(descr && mtr);
   ut_ad(state >= XDES_FREE);
   ut_ad(state <= XDES_FSEG);
-  ut_ad(mtr->memo_contains_flagged(&block, MTR_MEMO_PAGE_SX_FIX));
+  ut_ad(mtr->memo_contains_flagged(&block, MTR_MEMO_PAGE_SX_FIX |
+                                   MTR_MEMO_PAGE_X_FIX));
   ut_ad(page_align(descr) == block.frame);
   ut_ad(mach_read_from_4(descr + XDES_STATE) <= XDES_FSEG);
   mtr->write<1>(block, XDES_STATE + 3 + descr, state);
@@ -245,7 +245,8 @@ xdes_get_state(
 Inits an extent descriptor to the free and clean state. */
 inline void xdes_init(const buf_block_t &block, xdes_t *descr, mtr_t *mtr)
 {
-  ut_ad(mtr->memo_contains_flagged(&block, MTR_MEMO_PAGE_SX_FIX));
+  ut_ad(mtr->memo_contains_flagged(&block, MTR_MEMO_PAGE_SX_FIX |
+                                   MTR_MEMO_PAGE_X_FIX));
   mtr->memset(&block, uint16_t(descr - block.frame) + XDES_BITMAP,
               XDES_SIZE - XDES_BITMAP, 0xff);
   xdes_set_state(block, descr, XDES_FREE, mtr);
@@ -320,7 +321,8 @@ xdes_get_descriptor_with_space_hdr(
 	ulint	size;
 	ulint	descr_page_no;
 	ut_ad(mtr->memo_contains(*space));
-	ut_ad(mtr->memo_contains_flagged(header, MTR_MEMO_PAGE_SX_FIX));
+	ut_ad(mtr->memo_contains_flagged(header, MTR_MEMO_PAGE_SX_FIX
+					 | MTR_MEMO_PAGE_X_FIX));
 	/* Read free limit and space size */
 	limit = mach_read_from_4(FSP_HEADER_OFFSET + FSP_FREE_LIMIT
 				 + header->frame);
@@ -553,9 +555,7 @@ void fsp_header_init(fil_space_t* space, ulint size, mtr_t* mtr)
 
 	mtr_x_lock_space(space, mtr);
 
-	const auto savepoint = mtr->get_savepoint();
 	buf_block_t* block = buf_page_create(space, 0, zip_size, mtr);
-	mtr->sx_latch_at_savepoint(savepoint, block);
 	buf_block_dbg_add_level(block, SYNC_FSP_PAGE);
 
 	space->size_in_header = size;
@@ -874,11 +874,9 @@ fsp_fill_free_list(
 			pages should be ignored. */
 
 			if (i > 0) {
-				const auto savepoint = mtr->get_savepoint();
 				block= buf_page_create(
 					space, static_cast<uint32_t>(i),
 					zip_size, mtr);
-				mtr->sx_latch_at_savepoint(savepoint, block);
 
 				buf_block_dbg_add_level(block, SYNC_FSP_PAGE);
 				fsp_init_file_page(space, block, mtr);
@@ -888,13 +886,11 @@ fsp_fill_free_list(
 			}
 
 			if (space->purpose != FIL_TYPE_TEMPORARY) {
-				const auto savepoint = mtr->get_savepoint();
 				block = buf_page_create(
 					space,
 					static_cast<uint32_t>(
 						i + FSP_IBUF_BITMAP_OFFSET),
 					zip_size, mtr);
-				mtr->sx_latch_at_savepoint(savepoint, block);
 				buf_block_dbg_add_level(block, SYNC_FSP_PAGE);
 
 				fsp_init_file_page(space, block, mtr);
@@ -1040,46 +1036,22 @@ fsp_alloc_from_free_frag(buf_block_t *header, buf_block_t *xdes, xdes_t *descr,
 /** Gets a buffer block for an allocated page.
 @param[in,out]	space		tablespace
 @param[in]	offset		page number of the allocated page
-@param[in]	rw_latch	RW_SX_LATCH, RW_X_LATCH
 @param[in,out]	mtr		mini-transaction
 @return block, initialized */
 static
 buf_block_t*
-fsp_page_create(
-	fil_space_t*		space,
-	page_no_t		offset,
-	rw_lock_type_t		rw_latch,
-	mtr_t*			mtr)
+fsp_page_create(fil_space_t *space, page_no_t offset, mtr_t *mtr)
 {
-	buf_block_t*	block = buf_page_create(
-					space, static_cast<uint32_t>(offset),
+  buf_block_t *block= buf_page_create(space, static_cast<uint32_t>(offset),
 				      space->zip_size(), mtr);
-
-	/* The latch may already have been acquired, so we cannot invoke
-	mtr_t::x_latch_at_savepoint() or mtr_t::sx_latch_at_savepoint(). */
-	mtr_memo_type_t memo;
-
-	if (rw_latch == RW_X_LATCH) {
-		rw_lock_x_lock(&block->lock);
-		memo = MTR_MEMO_PAGE_X_FIX;
-	} else {
-		ut_ad(rw_latch == RW_SX_LATCH);
-		rw_lock_sx_lock(&block->lock);
-		memo = MTR_MEMO_PAGE_SX_FIX;
-	}
-
-	mtr_memo_push(mtr, block, memo);
-	buf_block_buf_fix_inc(block, __FILE__, __LINE__);
   fsp_init_file_page(space, block, mtr);
-
-	return(block);
+  return block;
 }
 
 /** Allocates a single free page from a space.
 The page is marked as used.
 @param[in,out]	space		tablespace
 @param[in]	hint		hint of which page would be desirable
-@param[in]	rw_latch	RW_SX_LATCH, RW_X_LATCH
 @param[in,out]	mtr		mini-transaction
 @param[in,out]	init_mtr	mini-transaction in which the page should be
 initialized (may be the same as mtr)
@@ -1089,7 +1061,6 @@ buf_block_t*
 fsp_alloc_free_page(
 	fil_space_t*		space,
 	ulint			hint,
-	rw_lock_type_t		rw_latch,
 	mtr_t*			mtr,
 	mtr_t*			init_mtr)
 {
@@ -1184,7 +1155,7 @@ fsp_alloc_free_page(
 	}
 
 	fsp_alloc_from_free_frag(block, xdes, descr, free, mtr);
-	return fsp_page_create(space, page_no, rw_latch, init_mtr);
+	return fsp_page_create(space, page_no, init_mtr);
 }
 
 /** Frees a single page of a space.
@@ -1376,13 +1347,13 @@ bool
 fsp_alloc_seg_inode_page(fil_space_t *space, buf_block_t *header, mtr_t *mtr)
 {
   ut_ad(header->page.id().space() == space->id);
-  buf_block_t *block= fsp_alloc_free_page(space, 0, RW_SX_LATCH, mtr, mtr);
+  buf_block_t *block= fsp_alloc_free_page(space, 0, mtr, mtr);
 
   if (!block)
     return false;
 
   buf_block_dbg_add_level(block, SYNC_FSP_PAGE);
-  ut_ad(rw_lock_get_sx_lock_count(&block->lock) == 1);
+  ut_ad(rw_lock_get_x_lock_count(&block->lock) == 1);
 
   mtr->write<2>(*block, block->frame + FIL_PAGE_TYPE, FIL_PAGE_INODE);
 
@@ -1650,48 +1621,40 @@ static ulint fseg_get_n_frag_pages(const fseg_inode_t *inode)
 	return(count);
 }
 
-/**********************************************************************//**
-Creates a new segment.
-@return the block where the segment header is placed, x-latched, NULL
-if could not create segment because of lack of space */
+/** Create a new segment.
+@param space                tablespace
+@param byte_offset          byte offset of the created segment header
+@param mtr                  mini-transaction
+@param has_done_reservation whether fsp_reserve_free_extents() was invoked
+@param block                block where segment header is placed,
+                            or NULL to allocate an additional page for that
+@return the block where the segment header is placed, x-latched
+@retval NULL if could not create segment because of lack of space */
 buf_block_t*
-fseg_create(
-	fil_space_t* space, /*!< in,out: tablespace */
-	ulint	page,	/*!< in: page where the segment header is placed: if
-			this is != 0, the page must belong to another segment,
-			if this is 0, a new page will be allocated and it
-			will belong to the created segment */
-	ulint	byte_offset, /*!< in: byte offset of the created segment header
-			on the page */
-	mtr_t*	mtr,
-   	bool	has_done_reservation) /*!< in: whether the caller
-			has already done the reservation for the pages with
-			fsp_reserve_free_extents (at least 2 extents: one for
-			the inode and the other for the segment) then there is
-			no need to do the check for this individual
-			operation */
+fseg_create(fil_space_t *space, ulint byte_offset, mtr_t *mtr,
+            bool has_done_reservation, buf_block_t *block)
 {
 	fseg_inode_t*	inode;
 	ib_id_t		seg_id;
-	buf_block_t*	block	= 0; /* remove warning */
 	ulint		n_reserved;
 
 	DBUG_ENTER("fseg_create");
 
 	ut_ad(mtr);
+	ut_ad(byte_offset >= FIL_PAGE_DATA);
 	ut_ad(byte_offset + FSEG_HEADER_SIZE
 	      <= srv_page_size - FIL_PAGE_DATA_END);
 
 	mtr_x_lock_space(space, mtr);
 	ut_d(space->modify_check(*mtr));
 
-	if (page != 0) {
-		block = buf_page_get(page_id_t(space->id, page),
-				     space->zip_size(),
-				     RW_SX_LATCH, mtr);
+	if (block) {
+		ut_ad(block->page.id().space() == space->id);
+
 		if (!space->full_crc32()) {
-			fil_block_check_type(*block, space->id == TRX_SYS_SPACE
-					     && page == TRX_SYS_PAGE_NO
+			fil_block_check_type(*block, block->page.id()
+					     == page_id_t(TRX_SYS_SPACE,
+							  TRX_SYS_PAGE_NO)
 					     ? FIL_PAGE_TYPE_TRX_SYS
 					     : FIL_PAGE_TYPE_SYS,
 					     mtr);
@@ -1734,10 +1697,9 @@ fseg_create(
 	mtr->memset(iblock, uint16_t(inode - iblock->frame) + FSEG_FRAG_ARR,
 		    FSEG_FRAG_SLOT_SIZE * FSEG_FRAG_ARR_N_SLOTS, 0xff);
 
-	if (page == 0) {
+	if (!block) {
 		block = fseg_alloc_free_page_low(space,
 						 inode, iblock, 0, FSP_UP,
-						 RW_SX_LATCH,
 #ifdef UNIV_DEBUG
 						 has_done_reservation,
 #endif /* UNIV_DEBUG */
@@ -1752,7 +1714,7 @@ fseg_create(
 			goto funct_exit;
 		}
 
-		ut_ad(rw_lock_get_sx_lock_count(&block->lock) == 1);
+		ut_ad(rw_lock_get_x_lock_count(&block->lock) == 1);
 		ut_ad(!fil_page_get_type(block->frame));
 		mtr->write<1>(*block, FIL_PAGE_TYPE + 1 + block->frame,
 			      FIL_PAGE_TYPE_SYS);
@@ -1957,7 +1919,6 @@ minimize file space fragmentation.
 @param[in]	direction		if the new page is needed because of
 an index page split, and records are inserted there in order, into which
 direction they go alphabetically: FSP_DOWN, FSP_UP, FSP_NO_DIR
-@param[in]	rw_latch		RW_SX_LATCH, RW_X_LATCH
 @param[in,out]	mtr			mini-transaction
 @param[in,out]	init_mtr		mtr or another mini-transaction in
 which the page should be initialized.
@@ -1970,7 +1931,6 @@ fseg_alloc_free_page_low(
 	buf_block_t*		iblock,
 	ulint			hint,
 	byte			direction,
-	rw_lock_type_t		rw_latch,
 #ifdef UNIV_DEBUG
 	bool			has_done_reservation,
 	/*!< whether the space has already been reserved */
@@ -2111,11 +2071,11 @@ fseg_alloc_free_page_low(
 		/* 6. We allocate an individual page from the space
 		===================================================*/
 		buf_block_t* block = fsp_alloc_free_page(
-			space, hint, rw_latch, mtr, init_mtr);
+			space, hint, mtr, init_mtr);
 
-		ut_ad(!has_done_reservation || block != NULL);
+		ut_ad(!has_done_reservation || block);
 
-		if (block != NULL) {
+		if (block) {
 			/* Put the page in the fragment page array of the
 			segment */
 			n = fseg_find_free_frag_page_slot(seg_inode);
@@ -2192,7 +2152,7 @@ fseg_alloc_free_page_low(
 				    xdes, mtr);
 	}
 
-	return fsp_page_create(space, ret_page, rw_latch, init_mtr);
+	return fsp_page_create(space, ret_page, init_mtr);
 }
 
 /**********************************************************************//**
@@ -2243,7 +2203,6 @@ fseg_alloc_free_page_general(
 
 	block = fseg_alloc_free_page_low(space,
 					 inode, iblock, hint, direction,
-					 RW_X_LATCH,
 #ifdef UNIV_DEBUG
 					 has_done_reservation,
 #endif /* UNIV_DEBUG */

storage/innobase/include/fsp0fsp.h

@@ -362,26 +362,18 @@ fsp_header_init_fields(
 void fsp_header_init(fil_space_t* space, ulint size, mtr_t* mtr)
 	MY_ATTRIBUTE((nonnull));
 
-/**********************************************************************//**
-Creates a new segment.
-@return the block where the segment header is placed, x-latched, NULL
-if could not create segment because of lack of space */
+/** Create a new segment.
+@param space                tablespace
+@param byte_offset          byte offset of the created segment header
+@param mtr                  mini-transaction
+@param has_done_reservation whether fsp_reserve_free_extents() was invoked
+@param block                block where segment header is placed,
+                            or NULL to allocate an additional page for that
+@return the block where the segment header is placed, x-latched
+@retval NULL if could not create segment because of lack of space */
 buf_block_t*
-fseg_create(
-	fil_space_t* space, /*!< in,out: tablespace */
-	ulint	page,	/*!< in: page where the segment header is placed: if
-			this is != 0, the page must belong to another segment,
-			if this is 0, a new page will be allocated and it
-			will belong to the created segment */
-	ulint	byte_offset, /*!< in: byte offset of the created segment header
-			on the page */
-	mtr_t*	mtr,
-   	bool	has_done_reservation = false); /*!< in: whether the caller
-			has already done the reservation for the pages with
-			fsp_reserve_free_extents (at least 2 extents: one for
-			the inode and the other for the segment) then there is
-			no need to do the check for this individual
-			operation */
+fseg_create(fil_space_t *space, ulint byte_offset, mtr_t *mtr,
+            bool has_done_reservation= false, buf_block_t *block= NULL);
 
 /** Calculate the number of pages reserved by a segment,
 and how many pages are currently used.

storage/innobase/include/mtr0mtr.h

@@ -591,6 +591,11 @@ struct mtr_t {
     m_freed_pages->add_value(id.page_no());
   }
 
+  /** Determine the added buffer fix count of a block.
+  @param block block to be checked
+  @return number of buffer count added by this mtr */
+  uint32_t get_fix_count(const buf_block_t *block) const;
+
 private:
   /** Log a write of a byte string to a page.
   @param block   buffer page

storage/innobase/log/log0recv.cc

@@ -2592,7 +2592,6 @@ inline buf_block_t *recv_sys_t::recover_low(const page_id_t page_id,
       ut_ad(&recs == &p->second);
       i.created= true;
       buf_block_dbg_add_level(block, SYNC_NO_ORDER_CHECK);
-      mtr.x_latch_at_savepoint(0, block);
       recv_recover_page(block, mtr, p, space, &i);
       ut_ad(mtr.has_committed());
       recs.log.clear();

storage/innobase/mtr/mtr0mtr.cc

@@ -702,6 +702,31 @@ inline lsn_t mtr_t::finish_write(ulint len)
 	return start_lsn;
 }
 
+/** Find buffer fix count of the given block acquired by the
+mini-transaction */
+struct FindBlock
+{
+  int32_t num_fix;
+  const buf_block_t *const block;
+
+  FindBlock(const buf_block_t *block_buf): num_fix(0), block(block_buf) {}
+
+  bool operator()(const mtr_memo_slot_t* slot)
+  {
+    if (slot->object == block)
+      num_fix++;
+    return true;
+  }
+};
+
+uint32_t mtr_t::get_fix_count(const buf_block_t *block) const
+{
+  Iterate<FindBlock> iteration((FindBlock(block)));
+  if (m_memo.for_each_block(iteration))
+    return iteration.functor.num_fix;
+  return 0;
+}
+
 #ifdef UNIV_DEBUG
 /** Check if we are holding an rw-latch in this mini-transaction
 @param lock   latch to search for

storage/innobase/row/row0sel.cc

@@ -3178,15 +3178,41 @@ row_sel_build_prev_vers_for_mysql(
 	return(err);
 }
 
-/** Helper class to cache clust_rec and old_ver */
+/** Helper class to cache clust_rec and old_vers */
 class Row_sel_get_clust_rec_for_mysql
 {
   const rec_t *cached_clust_rec;
   rec_t *cached_old_vers;
+  lsn_t cached_lsn;
+  page_id_t cached_page_id;
+
+#ifdef UNIV_DEBUG
+  void check_eq(const dict_index_t *index, const rec_offs *offsets) const
+  {
+    rec_offs vers_offs[REC_OFFS_HEADER_SIZE + MAX_REF_PARTS];
+    rec_offs_init(vers_offs);
+    mem_heap_t *heap= nullptr;
+
+    ut_ad(rec_offs_validate(cached_clust_rec, index, offsets));
+    ut_ad(index->first_user_field() <= rec_offs_n_fields(offsets));
+    ut_ad(vers_offs == rec_get_offsets(cached_old_vers, index, vers_offs, true,
+                                       index->db_trx_id(), &heap));
+    ut_ad(!heap);
+    for (auto n= index->db_trx_id(); n--; )
+    {
+      const dict_col_t *col= dict_index_get_nth_col(index, n);
+      ulint len1, len2;
+      const byte *b1= rec_get_nth_field(cached_clust_rec, offsets, n, &len1);
+      const byte *b2= rec_get_nth_field(cached_old_vers, vers_offs, n, &len2);
+      ut_ad(!cmp_data_data(col->mtype, col->prtype, b1, len1, b2, len2));
+    }
+  }
+#endif
 
 public:
   Row_sel_get_clust_rec_for_mysql() :
-	cached_clust_rec(NULL), cached_old_vers(NULL) {}
+    cached_clust_rec(NULL), cached_old_vers(NULL), cached_lsn(0),
+    cached_page_id(page_id_t(0,0)) {}
 
   dberr_t operator()(row_prebuilt_t *prebuilt, dict_index_t *sec_index,
                      const rec_t *rec, que_thr_t *thr, const rec_t **out_rec,
@@ -3391,8 +3417,15 @@ Row_sel_get_clust_rec_for_mysql::operator()(
 		    && !lock_clust_rec_cons_read_sees(
 			    clust_rec, clust_index, *offsets,
 			    &trx->read_view)) {
+			const buf_page_t& bpage = btr_pcur_get_block(
+				prebuilt->clust_pcur)->page;
+
+			const lsn_t lsn = mach_read_from_8(
+				page_align(clust_rec) + FIL_PAGE_LSN);
 
-			if (clust_rec != cached_clust_rec) {
+			if (lsn != cached_lsn
+			    || bpage.id() != cached_page_id
+			    || clust_rec != cached_clust_rec) {
 				/* The following call returns 'offsets' associated with
 				'old_vers' */
 				err = row_sel_build_prev_vers_for_mysql(
@@ -3404,6 +3437,8 @@ Row_sel_get_clust_rec_for_mysql::operator()(
 
 					goto err_exit;
 				}
+				cached_lsn = lsn;
+				cached_page_id = bpage.id();
 				cached_clust_rec = clust_rec;
 				cached_old_vers = old_vers;
 			} else {
@@ -3414,7 +3449,8 @@ Row_sel_get_clust_rec_for_mysql::operator()(
 				version of clust_rec and its old version
 				old_vers. Re-calculate the offsets for old_vers. */
 
-				if (old_vers != NULL) {
+				if (old_vers) {
+					ut_d(check_eq(clust_index, *offsets));
 					*offsets = rec_get_offsets(
 						old_vers, clust_index, *offsets,
 						true, ULINT_UNDEFINED, offset_heap);

storage/innobase/trx/trx0rseg.cc

@@ -316,7 +316,7 @@ trx_rseg_header_create(
 	ut_ad(!sys_header == (space == fil_system.temp_space));
 
 	/* Allocate a new file segment for the rollback segment */
-	block = fseg_create(space, 0, TRX_RSEG + TRX_RSEG_FSEG_HEADER, mtr);
+	block = fseg_create(space, TRX_RSEG + TRX_RSEG_FSEG_HEADER, mtr);
 
 	if (block == NULL) {
 		/* No space left */

storage/innobase/trx/trx0sys.cc

@@ -158,7 +158,7 @@ trx_sysf_create(
 	compile_time_assert(TRX_SYS_SPACE == 0);
 
 	/* Create the trx sys file block in a new allocated file segment */
-	block = fseg_create(fil_system.sys_space, 0,
+	block = fseg_create(fil_system.sys_space,
 			    TRX_SYS + TRX_SYS_FSEG_HEADER,
 			    mtr);
 	buf_block_dbg_add_level(block, SYNC_TRX_SYS_HEADER);

storage/innobase/trx/trx0undo.cc

@@ -383,7 +383,7 @@ trx_undo_seg_create(fil_space_t *space, buf_block_t *rseg_hdr, ulint *id,
 	}
 
 	/* Allocate a new file segment for the undo log */
-	block = fseg_create(space, 0, TRX_UNDO_SEG_HDR + TRX_UNDO_FSEG_HEADER,
+	block = fseg_create(space, TRX_UNDO_SEG_HDR + TRX_UNDO_FSEG_HEADER,
 			    mtr, true);
 
 	space->release_free_extents(n_reserved);