MDEV-27183 optimize std::map lookup in during crash recoverybb-10.5-MDEV-27183-map-lookup-recovery

This is a low hanging fruit. Before this patch std::map::emplace() was
a ~50% of the whole recv_sys_t::parse() operation in by test.
After the fix it's only ~20%.

recv_sys_t::parse() recv_sys_t::pages is a collection of all pages
to recovery. Often, there are multiple changes for a single page.
Often, they go in a row and for such cases let's avoid
lookup in a std::map. cached_pages_it serves as a cache
of size 1.

recv_sys_t::add(): replace page_id argument with a std::map::iterator

2 files changed, 13 insertions, 11 deletions

diff --git a/storage/innobase/include/log0recv.h b/storage/innobase/include/log0recv.h
index fcc7e989885..9159ba00859 100644
--- a/storage/innobase/include/log0recv.h
+++ b/storage/innobase/include/log0recv.h
@@ -332,12 +332,12 @@ public:
   bool is_initialised() const { return last_stored_lsn != 0; }
 
   /** Register a redo log snippet for a page.
-  @param page_id  page identifier
+  @param it       page iterator
   @param start_lsn start LSN of the mini-transaction
   @param lsn      @see mtr_t::commit_lsn()
   @param l        redo log snippet @see log_t::FORMAT_10_5
   @param len      length of l, in bytes */
-  inline void add(const page_id_t page_id, lsn_t start_lsn, lsn_t lsn,
+  inline void add(map::iterator it, lsn_t start_lsn, lsn_t lsn,
                   const byte *l, size_t len);
 
   /** Parse and register one mini-transaction in log_t::FORMAT_10_5.
diff --git a/storage/innobase/log/log0recv.cc b/storage/innobase/log/log0recv.cc
index eb34fd8ede9..3e8986d4651 100644
--- a/storage/innobase/log/log0recv.cc
+++ b/storage/innobase/log/log0recv.cc
@@ -1664,20 +1664,17 @@ inline void page_recv_t::will_not_read()
 
 
 /** Register a redo log snippet for a page.
-@param page_id  page identifier
+@param it       page iterator
 @param start_lsn start LSN of the mini-transaction
 @param lsn      @see mtr_t::commit_lsn()
 @param recs     redo log snippet @see log_t::FORMAT_10_5
 @param len      length of l, in bytes */
-inline void recv_sys_t::add(const page_id_t page_id,
-                            lsn_t start_lsn, lsn_t lsn, const byte *l,
-                            size_t len)
+inline void recv_sys_t::add(map::iterator it, lsn_t start_lsn, lsn_t lsn,
+                            const byte *l, size_t len)
 {
   ut_ad(mutex_own(&mutex));
-  std::pair<map::iterator, bool> p= pages.emplace(map::value_type
-                                                  (page_id, page_recv_t()));
-  page_recv_t& recs= p.first->second;
-  ut_ad(p.second == recs.log.empty());
+  page_id_t page_id = it->first;
+  page_recv_t &recs= it->second;
 
   switch (*l & 0x70) {
   case FREE_PAGE: case INIT_PAGE:
@@ -1769,6 +1766,7 @@ bool recv_sys_t::parse(lsn_t checkpoint_lsn, store_t *store, bool apply)
 loop:
   const byte *const log= buf + recovered_offset;
   const lsn_t start_lsn= recovered_lsn;
+  map::iterator cached_pages_it = pages.end();
 
   /* Check that the entire mini-transaction is included within the buffer */
   const byte *l;
@@ -2092,8 +2090,12 @@ same_page:
         /* fall through */
       case STORE_YES:
         if (!mlog_init.will_avoid_read(id, start_lsn))
-          add(id, start_lsn, end_lsn, recs,
+        {
+          if (cached_pages_it == pages.end() || cached_pages_it->first != id)
+            cached_pages_it= pages.emplace(id, page_recv_t()).first;
+          add(cached_pages_it, start_lsn, end_lsn, recs,
               static_cast<size_t>(l + rlen - recs));
+        }
         continue;
       case STORE_NO:
         if (!is_init)