MDEV-30648 btr_estimate_n_rows_in_range() accesses unfixed, unlatched page

The issue is caused by MDEV-30400 fix.

There are two cursors in btr_estimate_n_rows_in_range() - p1 and p2, but
both share the same mtr. Each cursor contains mtr savepoint for the
previously fetched block to release it then the current block is
fetched.

Before MDEV-30400 the block was released with
mtr_t::release_block_at_savepoint(), it just unfixed a block and
released its page patch. In MDEV-30400 it was replaced with
mtr_t::rollback_to_savepoint(), which does the same as the former
mtr_t::release_block_at_savepoint(ulint begin, ulint end) but also
erases the corresponding slots from mtr memo, what invalidates any
stored mtr's memo savepoints, greater or equal to "begin".

The idea of the fix is to get rid of savepoints at all in
btr_estimate_n_rows_in_range() and
btr_estimate_n_rows_in_range_on_level(). As
mtr_t::rollback_to_savepoint() erases elements from mtr_t::m_memo, we
know what element of mtr_t::m_memo can be deleted on the certain case,
so there is no need to store savepoints.

See also the following slides for details:
https://docs.google.com/presentation/d/1RFYBo7EUhM22ab3GOYctv3j_3yC0vHtBY9auObZec8U

Reviewed by: Marko Mäkelä

Author: vlad-lesin

storage/innobase/btr/btr0cur.cc

@@ -4990,8 +4990,6 @@ class btr_est_cur_t
   page_id_t m_page_id;
   /** Current block */
   buf_block_t *m_block;
-  /** mtr savepoint of the current block */
-  ulint m_savepoint;
   /** Page search mode, can differ from m_mode for non-leaf pages, see c-tor
   comments for details */
   page_cur_mode_t m_page_mode;
@@ -5050,17 +5048,17 @@ class btr_est_cur_t
   bool fetch_child(ulint level, mtr_t &mtr, const buf_block_t *right_parent)
   {
     buf_block_t *parent_block= m_block;
-    ulint parent_savepoint= m_savepoint;
 
     m_block= btr_block_get(*index(), m_page_id.page_no(), RW_S_LATCH, !level,
                            &mtr, nullptr);
     if (!m_block)
       return false;
 
     if (parent_block && parent_block != right_parent)
-      mtr.rollback_to_savepoint(parent_savepoint, parent_savepoint + 1);
-
-    m_savepoint= mtr.get_savepoint() - 1;
+    {
+      ut_ad(mtr.get_savepoint() >= 2);
+      mtr.rollback_to_savepoint(1, 2);
+    }
 
     return level == ULINT_UNDEFINED ||
       btr_page_get_level(m_block->page.frame) == level;
@@ -5122,10 +5120,10 @@ class btr_est_cur_t
     return true;
   }
 
-  /** Gets page id of the current record child.
+  /** Read page id of the current record child.
   @param offsets offsets array.
   @param heap heap for offsets array */
-  void get_child(rec_offs **offsets, mem_heap_t **heap)
+  void read_child_page_id(rec_offs **offsets, mem_heap_t **heap)
   {
     const rec_t *node_ptr= page_cur_get_rec(&m_page_cur);
 
@@ -5195,11 +5193,7 @@ class btr_est_cur_t
   /** Copies block pointer and savepoint from another btr_est_cur_t in the case
   if both left and right border cursors point to the same block.
   @param o reference to the other btr_est_cur_t object. */
-  void set_block(const btr_est_cur_t &o)
-  {
-    m_block= o.m_block;
-    m_savepoint= o.m_savepoint;
-  }
+  void set_block(const btr_est_cur_t &o) { m_block= o.m_block; }
 
   /** @return current record number. */
   ulint nth_rec() const { return m_nth_rec; }
@@ -5238,7 +5232,6 @@ static ha_rows btr_estimate_n_rows_in_range_on_level(
   pages before reaching right_page_no, then we estimate the average from the
   pages scanned so far. */
   static constexpr uint n_pages_read_limit= 9;
-  ulint savepoint= 0;
   buf_block_t *block= nullptr;
   const dict_index_t *index= left_cur.index();
 
@@ -5268,19 +5261,18 @@ static ha_rows btr_estimate_n_rows_in_range_on_level(
   {
     page_t *page;
     buf_block_t *prev_block= block;
-    ulint prev_savepoint= savepoint;
-
-    savepoint= mtr.get_savepoint();
 
     /* Fetch the page. */
     block= btr_block_get(*index, page_id.page_no(), RW_S_LATCH, !level, &mtr,
                          nullptr);
 
     if (prev_block)
     {
-      mtr.rollback_to_savepoint(prev_savepoint, prev_savepoint + 1);
-      if (block)
-        savepoint--;
+      ulint savepoint = mtr.get_savepoint();
+      /* Index s-lock, p1, p2 latches, can also be p1 and p2 parent latch if
+      they are not diverged */
+      ut_ad(savepoint >= 3);
+      mtr.rollback_to_savepoint(savepoint - 2, savepoint - 1);
     }
 
     if (!block || btr_page_get_level(buf_block_get_frame(block)) != level)
@@ -5311,8 +5303,8 @@ static ha_rows btr_estimate_n_rows_in_range_on_level(
 
   if (block)
   {
-    ut_ad(block == mtr.at_savepoint(savepoint));
-    mtr.rollback_to_savepoint(savepoint, savepoint + 1);
+    ut_ad(block == mtr.at_savepoint(mtr.get_savepoint() - 1));
+    mtr.rollback_to_savepoint(mtr.get_savepoint() - 1);
   }
 
   return (n_rows);
@@ -5321,8 +5313,8 @@ static ha_rows btr_estimate_n_rows_in_range_on_level(
 
   if (block)
   {
-    ut_ad(block == mtr.at_savepoint(savepoint));
-    mtr.rollback_to_savepoint(savepoint, savepoint + 1);
+    ut_ad(block == mtr.at_savepoint(mtr.get_savepoint() - 1));
+    mtr.rollback_to_savepoint(mtr.get_savepoint() - 1);
   }
 
   is_n_rows_exact= false;
@@ -5516,8 +5508,12 @@ ha_rows btr_estimate_n_rows_in_range(dict_index_t *index,
   {
     ut_ad(height > 0);
     height--;
-    p1.get_child(&offsets, &heap);
-    p2.get_child(&offsets, &heap);
+    ut_ad(mtr.memo_contains(p1.index()->lock, MTR_MEMO_S_LOCK));
+    ut_ad(mtr.memo_contains_flagged(p1.block(), MTR_MEMO_PAGE_S_FIX));
+    p1.read_child_page_id(&offsets, &heap);
+    ut_ad(mtr.memo_contains(p2.index()->lock, MTR_MEMO_S_LOCK));
+    ut_ad(mtr.memo_contains_flagged(p2.block(), MTR_MEMO_PAGE_S_FIX));
+    p2.read_child_page_id(&offsets, &heap);
     goto search_loop;
   }