MDEV-28443: MDEV-15250 causes latch order violation

Problem:
=======
Index page latches must be acquired before undo page latches.
In trx_t::apply_log(), InnoDB acquired undo log page latch
before an index page latch.

Solution:
========
In trx_t::apply_log(), InnoDB should copy the undo log
record and release the undo log block before applying
it on online indexes.

storage/innobase/include/trx0rec.h

@@ -232,8 +232,6 @@ must hold a latch on the index page of the clustered index record.
 @param	v_status	status determine if it is going into this
 			function by purge thread or not.
 			And if we read "after image" of undo log
-@param	undo_block	undo log block which was cached during
-			online dml apply or nullptr
 @retval true if previous version was built, or if it was an insert
 or the table has been rebuilt
 @retval false if the previous version is earlier than purge_view,
@@ -249,8 +247,7 @@ trx_undo_prev_version_build(
 	rec_t		**old_vers,
 	mem_heap_t	*v_heap,
 	dtuple_t	**vrow,
-	ulint		v_status,
-	const buf_block_t *undo_block= nullptr);
+	ulint		v_status);
 
 /** Read from an undo log record a non-virtual column value.
 @param[in,out]	ptr		pointer to remaining part of the undo record

storage/innobase/include/trx0undo.h

@@ -332,14 +332,14 @@ parse the undo log record and store the record type, update vector
 and compiler information */
 class UndorecApplier
 {
-  /** undo log block which was latched */
-  const buf_block_t *block;
+  /** Undo log block page id */
+  page_id_t page_id;
   /** Undo log record pointer */
   trx_undo_rec_t *undo_rec;
   /** Offset of the undo log record within the block */
-  ulint offset;
+  uint16_t offset;
   /** Transaction id of the undo log */
-  trx_id_t trx_id;
+  const trx_id_t trx_id;
   /** Undo log record type */
   ulint type;
   /** compiler information */
@@ -353,21 +353,23 @@ class UndorecApplier
   mtr_t mtr;
 
 public:
-  UndorecApplier(const buf_block_t *block, trx_id_t trx_id)
-                : block(block), trx_id(trx_id)
+  UndorecApplier(page_id_t page_id, trx_id_t trx_id) :
+    page_id(page_id), trx_id(trx_id), heap(mem_heap_create(100))
   {
-    ut_ad(block->page.lock.have_any());
-    heap= mem_heap_create(100);
   }
 
-  /** Assign the undo log block */
-  void assign_block(const buf_block_t *undo_block)
+  /** Assign the next page id */
+  void assign_next(const page_id_t next_page_id)
   {
-    block= undo_block;
+    page_id= next_page_id;
   }
 
   /** Assign the undo log record and offset */
-  void assign_rec(trx_undo_rec_t *rec);
+  inline void assign_rec(const buf_block_t &block, uint16_t offset);
+
+  uint16_t get_offset() const { return offset; }
+
+  page_id_t get_page_id() const { return page_id; }
 
   /** Handle the DML undo log and apply it on online indexes */
   void apply_undo_rec();
@@ -396,7 +398,7 @@ class UndorecApplier
   {
     uint16_t offset= static_cast<uint16_t>(roll_ptr);
     uint32_t page_no= static_cast<uint32_t>(roll_ptr >> 16);
-    return page_no == block->page.id().page_no() && offset == this->offset;
+    return page_no == page_id.page_no() && offset == this->offset;
   }
 
   /** Clear the undo log record information */
@@ -406,7 +408,6 @@ class UndorecApplier
     cmpl_info= 0;
     type= 0;
     update= nullptr;
-    offset= 0;
     mem_heap_empty(heap);
   }
 

storage/innobase/row/row0log.cc

@@ -3829,7 +3829,7 @@ UndorecApplier::get_old_rec(const dtuple_t &tuple, dict_index_t *index,
       return version;
     trx_undo_prev_version_build(*clust_rec, &mtr, version, index,
                                 *offsets, heap, &prev_version, nullptr,
-                                nullptr, 0, block);
+                                nullptr, 0);
     version= prev_version;
   }
   while (version);
@@ -3983,7 +3983,7 @@ void UndorecApplier::log_update(const dtuple_t &tuple,
         heap, rec_offs_size(offsets)), match_rec, offsets);
     trx_undo_prev_version_build(rec, &mtr, match_rec, clust_index,
                                 offsets, heap, &prev_version, nullptr,
-                                nullptr, 0, block);
+                                nullptr, 0);
 
     prev_offsets= rec_get_offsets(prev_version, clust_index, prev_offsets,
                                   clust_index->n_core_fields,

storage/innobase/trx/trx0rec.cc

@@ -2230,15 +2230,12 @@ trx_undo_report_row_operation(
 
 /** Copy an undo record to heap.
 @param[in]	roll_ptr	roll pointer to a record that exists
-@param[in,out]	heap		memory heap where copied
-@param[in]	undo_block	undo log block which was cached during
-				DML online log apply */
+@param[in,out]	heap		memory heap where copied */
 static
 trx_undo_rec_t*
 trx_undo_get_undo_rec_low(
 	roll_ptr_t		roll_ptr,
-	mem_heap_t*		heap,
-	const buf_block_t*	undo_block= nullptr)
+	mem_heap_t*		heap)
 {
 	trx_undo_rec_t*	undo_rec;
 	ulint		rseg_id;
@@ -2254,21 +2251,15 @@ trx_undo_get_undo_rec_low(
 	trx_rseg_t* rseg = &trx_sys.rseg_array[rseg_id];
 	ut_ad(rseg->is_persistent());
 
-	if (undo_block
-            && undo_block->page.id().page_no() == page_no) {
-		undo_rec = trx_undo_rec_copy(
-			undo_block->page.frame + offset, heap);
-	} else {
-		mtr.start();
+	mtr.start();
 
-		buf_block_t *undo_page = trx_undo_page_get_s_latched(
-			page_id_t(rseg->space->id, page_no), &mtr);
+	buf_block_t *undo_page = trx_undo_page_get_s_latched(
+		page_id_t(rseg->space->id, page_no), &mtr);
 
-		undo_rec = trx_undo_rec_copy(
-			undo_page->page.frame + offset, heap);
+	undo_rec = trx_undo_rec_copy(
+		undo_page->page.frame + offset, heap);
 
-		mtr.commit();
-	}
+	mtr.commit();
 
 	return(undo_rec);
 }
@@ -2281,8 +2272,6 @@ trx_undo_get_undo_rec_low(
 				undo log of this transaction
 @param[in]	name		table name
 @param[out]	undo_rec	own: copy of the record
-@param[in]	undo_block	undo log block which was cached during
-				DML online log apply
 @retval true if the undo log has been
 truncated and we cannot fetch the old version
 @retval false if the undo log record is available
@@ -2294,15 +2283,13 @@ trx_undo_get_undo_rec(
 	mem_heap_t*		heap,
 	trx_id_t		trx_id,
 	const table_name_t&	name,
-	trx_undo_rec_t**	undo_rec,
-	const buf_block_t*	undo_block)
+	trx_undo_rec_t**	undo_rec)
 {
 	purge_sys.latch.rd_lock(SRW_LOCK_CALL);
 
 	bool missing_history = purge_sys.changes_visible(trx_id, name);
 	if (!missing_history) {
-		*undo_rec = trx_undo_get_undo_rec_low(
-				roll_ptr, heap, undo_block);
+		*undo_rec = trx_undo_get_undo_rec_low(roll_ptr, heap);
 	}
 
 	purge_sys.latch.rd_unlock();
@@ -2356,8 +2343,7 @@ trx_undo_prev_version_build(
 	rec_t		**old_vers,
 	mem_heap_t	*v_heap,
 	dtuple_t	**vrow,
-	ulint		v_status,
-	const buf_block_t*undo_block)
+	ulint		v_status)
 {
 	trx_undo_rec_t*	undo_rec	= NULL;
 	dtuple_t*	entry;
@@ -2396,7 +2382,7 @@ trx_undo_prev_version_build(
 
 	if (trx_undo_get_undo_rec(
 		    roll_ptr, heap, rec_trx_id, index->table->name,
-		    &undo_rec, undo_block)) {
+		    &undo_rec)) {
 		if (v_status & TRX_UNDO_PREV_IN_PURGE) {
 			/* We are fetching the record being purged */
 			undo_rec = trx_undo_get_undo_rec_low(roll_ptr, heap);

storage/innobase/trx/trx0undo.cc

@@ -290,10 +290,12 @@ trx_undo_get_first_rec(const fil_space_t &space, uint32_t page_no,
                                               mtr);
 }
 
-void UndorecApplier::assign_rec(trx_undo_rec_t *rec)
+inline void UndorecApplier::assign_rec(const buf_block_t &block,
+                                       uint16_t offset)
 {
-  this->undo_rec= rec;
-  this->offset= page_offset(rec);
+  ut_ad(block.page.lock.have_s());
+  this->offset= offset;
+  this->undo_rec= trx_undo_rec_copy(block.page.frame + offset, heap);
 }
 
 void UndorecApplier::apply_undo_rec()
@@ -355,12 +357,11 @@ ATTRIBUTE_COLD void trx_t::apply_log()
   page_id_t page_id{rsegs.m_redo.rseg->space->id, undo->hdr_page_no};
   page_id_t next_page_id(page_id);
   mtr_t mtr;
-  mem_heap_t *heap= mem_heap_create(100);
   mtr.start();
   buf_block_t *block= buf_page_get(page_id, 0, RW_S_LATCH, &mtr);
   ut_ad(block);
 
-  UndorecApplier log_applier(block, id);
+  UndorecApplier log_applier(page_id, id);
 
   for (;;)
   {
@@ -368,9 +369,12 @@ ATTRIBUTE_COLD void trx_t::apply_log()
                                                      undo->hdr_offset);
     while (rec)
     {
-      log_applier.assign_rec(rec);
+      log_applier.assign_rec(*block, page_offset(rec));
+      mtr.commit();
       log_applier.apply_undo_rec();
-      rec= trx_undo_page_get_next_rec(block, page_offset(rec),
+      mtr.start();
+      block= buf_page_get(log_applier.get_page_id(), 0, RW_S_LATCH, &mtr);
+      rec= trx_undo_page_get_next_rec(block, log_applier.get_offset(),
                                       page_id.page_no(), undo->hdr_offset);
     }
 
@@ -382,12 +386,11 @@ ATTRIBUTE_COLD void trx_t::apply_log()
     next_page_id.set_page_no(next);
     mtr.commit();
     mtr.start();
-    block= buf_page_get(next_page_id, 0, RW_S_LATCH, &mtr);
-    log_applier.assign_block(block);
+    block= buf_page_get_gen(next_page_id, 0, RW_S_LATCH, block, BUF_GET, &mtr);
+    log_applier.assign_next(next_page_id);
     ut_ad(block);
   }
   mtr.commit();
-  mem_heap_free(heap);
   apply_online_log= false;
 }