MDEV-24569 Assertion `mach_read_from_4(frame + 4U) == block.page.id().page_no()' failed in log_phys_t::applybb-10.5-MDEV-24569

Even though the page is freed, InnoDB change buffer merges the buffered
operation after reading the page from disk. During recovery, InnoDB
encounters the redo log even after freed redo log.

Before merging the buffered operation, InnoDB should check whether
the page is freed or not in allocation bitmap page. If it is freed then
reset the bits in change buffer bitmap page.

3 files changed, 50 insertions, 27 deletions

diff --git a/storage/innobase/fsp/fsp0fsp.cc b/storage/innobase/fsp/fsp0fsp.cc
index 5fed394b647..d80ff52001e 100644
--- a/storage/innobase/fsp/fsp0fsp.cc
+++ b/storage/innobase/fsp/fsp0fsp.cc
@@ -404,7 +404,7 @@ xdes_get_descriptor_const(
 	page_no_t		offset,
 	mtr_t*			mtr)
 {
-	ut_ad(mtr->memo_contains(space->latch, MTR_MEMO_S_LOCK));
+	ut_ad(mtr->memo_contains(space->latch, MTR_MEMO_SX_LOCK));
 	ut_ad(offset < space->free_limit);
 	ut_ad(offset < space->size_in_header);
 
@@ -2569,7 +2569,7 @@ fseg_page_is_free(fil_space_t* space, unsigned page)
 							  page);
 
 	mtr.start();
-	mtr_s_lock_space(space, &mtr);
+	mtr_sx_lock_space(space, &mtr);
 
 	if (page >= space->free_limit || page >= space->size_in_header) {
 		is_free = true;
diff --git a/storage/innobase/ibuf/ibuf0ibuf.cc b/storage/innobase/ibuf/ibuf0ibuf.cc
index 33c7df5947b..10008b53dba 100644
--- a/storage/innobase/ibuf/ibuf0ibuf.cc
+++ b/storage/innobase/ibuf/ibuf0ibuf.cc
@@ -4164,6 +4164,31 @@ bool ibuf_page_exists(const page_id_t id, ulint zip_size)
 	return bitmap_bits;
 }
 
+/** Reset the bits in the bitmap page for the given block and
+page id.
+@param block	X-latched page to apply changes or NULL to discard
+@param page_id	page identifier
+@param zip_size ROW_FORMAT=COMPRESSED page size, or 0
+@param mtr	mini-transaction */
+void ibuf_reset_bitmap(buf_block_t *block,
+                       page_id_t page_id,
+                       ulint zip_size,
+		       mtr_t *mtr)
+{
+ buf_block_t* bitmap = ibuf_bitmap_get_map_page(page_id, zip_size, mtr);
+ if (!bitmap)
+   return;
+
+ const ulint physical_size = zip_size ? zip_size : srv_page_size;
+ /* FIXME: update the bitmap byte only once! */
+ ibuf_bitmap_page_set_bits<IBUF_BITMAP_BUFFERED>(
+   bitmap, page_id, physical_size, false, mtr);
+
+ if (block != NULL)
+   ibuf_bitmap_page_set_bits<IBUF_BITMAP_FREE>(
+	bitmap, page_id, physical_size, ibuf_index_page_calc_free(block), mtr);
+}
+
 /** When an index page is read from a disk to the buffer pool, this function
 applies any buffered operations to the page and deletes the entries from the
 insert buffer. If the page is not read, but created in the buffer pool, this
@@ -4225,6 +4250,14 @@ void ibuf_merge_or_delete_for_page(buf_block_t *block, const page_id_t page_id,
 
 		ibuf_mtr_commit(&mtr);
 
+		if (bitmap_bits && fseg_page_is_free(
+				space, page_id.page_no())) {
+			ibuf_mtr_start(&mtr);
+			ibuf_reset_bitmap(block, page_id, zip_size, &mtr);
+			ibuf_mtr_commit(&mtr);
+			bitmap_bits = 0;
+		}
+
 		if (!bitmap_bits) {
 			/* No changes are buffered for this page. */
 			space->release();
@@ -4446,18 +4479,8 @@ loop:
 	}
 
 reset_bit:
-	if (!space) {
-	} else if (buf_block_t* bitmap = ibuf_bitmap_get_map_page(
-			   page_id, zip_size, &mtr)) {
-		/* FIXME: update the bitmap byte only once! */
-		ibuf_bitmap_page_set_bits<IBUF_BITMAP_BUFFERED>(
-			bitmap, page_id, physical_size, false, &mtr);
-
-		if (block != NULL) {
-			ibuf_bitmap_page_set_bits<IBUF_BITMAP_FREE>(
-				bitmap, page_id, physical_size,
-				ibuf_index_page_calc_free(block), &mtr);
-		}
+	if (space) {
+		ibuf_reset_bitmap(block, page_id, zip_size, &mtr);
 	}
 
 	ibuf_mtr_commit(&mtr);
diff --git a/storage/innobase/include/mtr0mtr.h b/storage/innobase/include/mtr0mtr.h
index d65a830dea2..4779647c395 100644
--- a/storage/innobase/include/mtr0mtr.h
+++ b/storage/innobase/include/mtr0mtr.h
@@ -65,8 +65,8 @@ savepoint. */
 /** Push an object to an mtr memo stack. */
 #define mtr_memo_push(m, o, t)	(m)->memo_push(o, t)
 
-#define mtr_s_lock_space(s, m)	(m)->s_lock_space((s), __FILE__, __LINE__)
 #define mtr_x_lock_space(s, m)	(m)->x_lock_space((s), __FILE__, __LINE__)
+#define mtr_sx_lock_space(s, m) (m)->sx_lock_space((s), __FILE__, __LINE__)
 
 #define mtr_s_lock_index(i, m)	(m)->s_lock(&(i)->lock, __FILE__, __LINE__)
 #define mtr_x_lock_index(i, m)	(m)->x_lock(&(i)->lock, __FILE__, __LINE__)
@@ -252,18 +252,6 @@ struct mtr_t {
 		memo_push(lock, MTR_MEMO_SX_LOCK);
 	}
 
-	/** Acquire a tablespace S-latch.
-	@param[in]	space	tablespace
-	@param[in]	file	file name from where called
-	@param[in]	line	line number in file */
-	void s_lock_space(fil_space_t* space, const char* file, unsigned line)
-	{
-		ut_ad(space->purpose == FIL_TYPE_TEMPORARY
-		      || space->purpose == FIL_TYPE_IMPORT
-		      || space->purpose == FIL_TYPE_TABLESPACE);
-		s_lock(&space->latch, file, line);
-	}
-
 	/** Acquire a tablespace X-latch.
 	@param[in]	space	tablespace
 	@param[in]	file	file name from where called
@@ -277,6 +265,18 @@ struct mtr_t {
 		rw_lock_x_lock_inline(&space->latch, 0, file, line);
 	}
 
+ /** Acquire a tablespace SX-latch.
+ @param[in]	space	tablespace
+ @param[in]	file	file name from where called
+ @param[in]	line	line number in file */
+ void sx_lock_space(fil_space_t* space, const char* file, unsigned line)
+ {
+   ut_ad(space->purpose == FIL_TYPE_TEMPORARY
+         || space->purpose == FIL_TYPE_IMPORT
+	 || space->purpose == FIL_TYPE_TABLESPACE);
+   sx_lock(&space->latch, file, line);
+ }
+
 	/** Release an object in the memo stack.
 	@param object	object
 	@param type	object type