MDEV-27557 InnoDB unnecessarily commits mtr during secondary index search to preserve clustered index latching order

New function to release latches till savepoint was added in mtr_t. As
there is no longer need to limit MDEV-20605 fix usage for locking reads
only, the limitation is removed.

storage/innobase/include/mtr0mtr.h

@@ -97,6 +97,15 @@ struct mtr_t {
   /** Commit the mini-transaction. */
   void commit();
 
+  /** Release latches till savepoint. To simplify the code only
+  MTR_MEMO_S_LOCK and MTR_MEMO_PAGE_S_FIX slot types are allowed to be
+  released, otherwise it would be neccesary to add one more argument in the
+  function to point out what slot types are allowed for rollback, and this
+  would be overengineering as currently the function is used only in one place
+  in the code.
+  @param savepoint   savepoint, can be obtained with get_savepoint */
+  void rollback_to_savepoint(ulint savepoint);
+
   /** Commit a mini-transaction that is shrinking a tablespace.
   @param space   tablespace that is being shrunk */
   ATTRIBUTE_COLD void commit_shrink(fil_space_t &space);

storage/innobase/mtr/mtr0mtr.cc

@@ -300,6 +300,50 @@ struct ReleaseAll {
   }
 };
 
+/** Stops iteration is savepoint is reached */
+template <typename Functor> struct TillSavepoint
+{
+
+  /** Constructor
+  @param[in] functor functor which is called if savepoint is not reached
+  @param[in] savepoint savepoint value to rollback
+  @param[in] used current position in slots container */
+  TillSavepoint(const Functor &functor, ulint savepoint, ulint used)
+      : functor(functor),
+        m_slots_count((used - savepoint) / sizeof(mtr_memo_slot_t))
+  {
+    ut_ad(savepoint);
+    ut_ad(used >= savepoint);
+  }
+
+  /** @return true if savepoint is not reached, false otherwise */
+  bool operator()(mtr_memo_slot_t *slot)
+  {
+#ifdef UNIV_DEBUG
+    /** This check is added because the code is invoked only from
+    row_search_mvcc() to release latches acquired during clustered index search
+    for secondary index record. To make it more universal we could add one more
+    member in this functor for debug build to pass only certain slot types,
+    but this is currently not necessary. */
+    switch (slot->type)
+    {
+    case MTR_MEMO_S_LOCK:
+    case MTR_MEMO_PAGE_S_FIX:
+      break;
+    default:
+      ut_a(false);
+    }
+#endif
+    return m_slots_count-- && functor(slot);
+  }
+
+private:
+  /** functor to invoke */
+  const Functor &functor;
+  /** slots count left till savepoint */
+  ulint m_slots_count;
+};
+
 #ifdef UNIV_DEBUG
 /** Check that all slots have been handled. */
 struct DebugCheck {
@@ -468,6 +512,21 @@ void mtr_t::commit()
   release_resources();
 }
 
+/** Release latches till savepoint. To simplify the code only
+MTR_MEMO_S_LOCK and MTR_MEMO_PAGE_S_FIX slot types are allowed to be
+released, otherwise it would be neccesary to add one more argument in the
+function to point out what slot types are allowed for rollback, and this
+would be overengineering as corrently the function is used only in one place
+in the code.
+@param savepoint   savepoint, can be obtained with get_savepoint */
+void mtr_t::rollback_to_savepoint(ulint savepoint)
+{
+  Iterate<TillSavepoint<ReleaseLatches>> iteration(
+      TillSavepoint<ReleaseLatches>(ReleaseLatches(), savepoint,
+                                    get_savepoint()));
+  m_memo.for_each_block_in_reverse(iteration);
+}
+
 /** Shrink a tablespace. */
 struct Shrink
 {

storage/innobase/row/row0sel.cc

@@ -3588,14 +3588,12 @@ record with the same ordering prefix in in the B-tree index
 @param[in] latch_mode latch mode wished in restoration
 @param[in] pcur cursor whose position has been stored
 @param[in] moves_up true if the cursor moves up in the index
-@param[in] mtr mtr; CAUTION: may commit mtr temporarily!
+@param[in,out] mtr mtr; CAUTION: may commit mtr temporarily!
-@param[in] select_lock_type select lock type
 @return true if we may need to process the record the cursor is now
 positioned on (i.e. we should not go to the next record yet) */
 static bool sel_restore_position_for_mysql(bool *same_user_rec,
                                            ulint latch_mode, btr_pcur_t *pcur,
-                                           bool moves_up, mtr_t *mtr,
+                                           bool moves_up, mtr_t *mtr)
-                                           lock_mode select_lock_type)
 {
 	auto status = btr_pcur_restore_position(latch_mode, pcur, mtr);
 
@@ -3618,8 +3616,7 @@ static bool sel_restore_position_for_mysql(bool *same_user_rec,
 	switch (pcur->rel_pos) {
 	case BTR_PCUR_ON:
 		if (!*same_user_rec && moves_up) {
-			if (status == btr_pcur_t::SAME_UNIQ
+			if (status == btr_pcur_t::SAME_UNIQ)
-			    && select_lock_type != LOCK_NONE)
 			  return true;
 next:
 			if (btr_pcur_move_to_next(pcur, mtr)
@@ -4303,7 +4300,7 @@ row_search_mvcc(
 	const rec_t*	clust_rec;
 	Row_sel_get_clust_rec_for_mysql row_sel_get_clust_rec_for_mysql;
 	ibool		unique_search			= FALSE;
-	ibool		mtr_has_extra_clust_latch	= FALSE;
+	ulint		mtr_extra_clust_savepoint	= 0;
 	bool		moves_up			= false;
 	/* if the returned record was locked and we did a semi-consistent
 	read (fetch the newest committed version), then this is set to
@@ -4673,7 +4670,7 @@ wait_table_again:
 
 		bool	need_to_process = sel_restore_position_for_mysql(
 			&same_user_rec, BTR_SEARCH_LEAF,
-			pcur, moves_up, &mtr, prebuilt->select_lock_type);
+			pcur, moves_up, &mtr);
 
 		if (UNIV_UNLIKELY(need_to_process)) {
 			if (UNIV_UNLIKELY(prebuilt->row_read_type
@@ -5355,7 +5352,7 @@ requires_clust_rec:
 		/* It was a non-clustered index and we must fetch also the
 		clustered index record */
 
-		mtr_has_extra_clust_latch = TRUE;
+		mtr_extra_clust_savepoint = mtr.get_savepoint();
 
 		ut_ad(!vrow);
 		/* The following call returns 'offsets' associated with
@@ -5643,27 +5640,15 @@ next_rec:
 		/* No need to do store restore for R-tree */
 		mtr.commit();
 		mtr.start();
-		mtr_has_extra_clust_latch = FALSE;
+		mtr_extra_clust_savepoint = 0;
-	} else if (mtr_has_extra_clust_latch) {
+	} else if (mtr_extra_clust_savepoint) {
-		/* If we have extra cluster latch, we must commit
+		/* We must release any clustered index latches
-		mtr if we are moving to the next non-clustered
+		if we are moving to the next non-clustered
 		index record, because we could break the latching
 		order if we would access a different clustered
 		index page right away without releasing the previous. */
-
+		mtr.rollback_to_savepoint(mtr_extra_clust_savepoint);
-		btr_pcur_store_position(pcur, &mtr);
+		mtr_extra_clust_savepoint = 0;
-		mtr.commit();
-		mtr_has_extra_clust_latch = FALSE;
-
-		mtr.start();
-
-		if (sel_restore_position_for_mysql(&same_user_rec,
-						   BTR_SEARCH_LEAF,
-						   pcur, moves_up, &mtr,
-						   prebuilt->select_lock_type)
-		    ) {
-			goto rec_loop;
-		}
 	}
 
 	if (moves_up) {
@@ -5723,7 +5708,7 @@ page_read_error:
 
 lock_table_wait:
 	mtr.commit();
-	mtr_has_extra_clust_latch = FALSE;
+	mtr_extra_clust_savepoint = 0;
 
 	trx->error_state = err;
 
@@ -5752,7 +5737,7 @@ lock_table_wait:
 		if (!dict_index_is_spatial(index)) {
 			sel_restore_position_for_mysql(
 				&same_user_rec, BTR_SEARCH_LEAF, pcur,
-				moves_up, &mtr, prebuilt->select_lock_type);
+				moves_up, &mtr);
 		}
 
 		if (trx->isolation_level <= TRX_ISO_READ_COMMITTED