MDEV-12353: Reduce log volume by an UNDO_INIT record

We introduce an EXTENDED log record for initializing an undo log page.
The size of the record will be 2 bytes plus the optional page identifier.
The entire undo page will be initialized, except the space that is
already reserved for TRX_UNDO_SEG_HDR in trx_undo_seg_create().

mtr_t::undo_create(): Write the UNDO_INIT record.

trx_undo_page_init(): Initialize the undo page corresponding to the
UNDO_INIT record. Unlike the former MLOG_UNDO_INIT record, we will
initialize almost the entire page, including initializing the
TRX_UNDO_PAGE_NODE to an empty list node, so that the subsequent call
to flst_init() will avoid writing log for the undo page.

6 files changed, 95 insertions, 39 deletions

diff --git a/storage/innobase/include/mtr0log.h b/storage/innobase/include/mtr0log.h
index 9d084634fe6..67b93556b52 100644
--- a/storage/innobase/include/mtr0log.h
+++ b/storage/innobase/include/mtr0log.h
@@ -521,18 +521,33 @@ inline void mtr_t::free(const page_id_t id)
     m_log.close(log_write<FREE_PAGE>(id, nullptr));
 }
 
-/** Partly initialize a B-tree page.
-@param block    B-tree page
-@param comp     false=ROW_FORMAT=REDUNDANT, true=COMPACT or DYNAMIC */
-inline void mtr_t::page_create(const buf_block_t &block, bool comp)
+/** Write an EXTENDED log record.
+@param block  buffer pool page
+@param type	extended record subtype; @see mrec_ext_t */
+inline void mtr_t::log_write_extended(const buf_block_t &block, byte type)
 {
   set_modified();
   if (m_log_mode != MTR_LOG_ALL)
     return;
   byte *l= log_write<EXTENDED>(block.page.id, &block.page, 1, true);
-  static_assert(false == INIT_ROW_FORMAT_REDUNDANT, "encoding");
-  static_assert(true == INIT_ROW_FORMAT_DYNAMIC, "encoding");
-  *l++= comp;
+  *l++= type;
   m_log.close(l);
   m_last_offset= FIL_PAGE_TYPE;
 }
+
+/** Write log for partly initializing a B-tree or R-tree page.
+@param block    B-tree page
+@param comp     false=ROW_FORMAT=REDUNDANT, true=COMPACT or DYNAMIC */
+inline void mtr_t::page_create(const buf_block_t &block, bool comp)
+{
+  static_assert(false == INIT_ROW_FORMAT_REDUNDANT, "encoding");
+  static_assert(true == INIT_ROW_FORMAT_DYNAMIC, "encoding");
+  log_write_extended(block, comp);
+}
+
+/** Write log for initializing an undo log page.
+@param block    undo page */
+inline void mtr_t::undo_create(const buf_block_t &block)
+{
+  log_write_extended(block, UNDO_INIT);
+}
diff --git a/storage/innobase/include/mtr0mtr.h b/storage/innobase/include/mtr0mtr.h
index 49537faa030..0f471b1186c 100644
--- a/storage/innobase/include/mtr0mtr.h
+++ b/storage/innobase/include/mtr0mtr.h
@@ -490,10 +490,13 @@ struct mtr_t {
   /** Free a page.
   @param id      page identifier */
   inline void free(const page_id_t id);
-  /** Partly initialize a B-tree page.
+  /** Write log for partly initializing a B-tree or R-tree page.
   @param block    B-tree page
   @param comp     false=ROW_FORMAT=REDUNDANT, true=COMPACT or DYNAMIC */
   inline void page_create(const buf_block_t &block, bool comp);
+  /** Write log for initializing an undo log page.
+  @param block    undo page */
+  inline void undo_create(const buf_block_t &block);
 
   /** Write a log record about a file operation.
   @param type           file operation
@@ -526,6 +529,11 @@ private:
   inline byte *log_write(const page_id_t id, const buf_page_t *bpage,
                          size_t len= 0, bool alloc= false, size_t offset= 0);
 
+  /** Write an EXTENDED log record.
+  @param block  buffer pool page
+  @param type	extended record subtype; @see mrec_ext_t */
+  inline void log_write_extended(const buf_block_t &block, byte type);
+
   /** Prepare to write the mini-transaction log to the redo log buffer.
   @return number of bytes to write in finish_write() */
   inline ulint prepare_write();
diff --git a/storage/innobase/include/mtr0types.h b/storage/innobase/include/mtr0types.h
index 4100347e39a..2098cbd8e0e 100644
--- a/storage/innobase/include/mtr0types.h
+++ b/storage/innobase/include/mtr0types.h
@@ -254,7 +254,11 @@ enum mrec_ext_t
   /** Partly initialize a ROW_FORMAT=COMPACT or DYNAMIC index page,
   including writing the "infimum" and "supremum" pseudo-records.
   The current byte offset will be reset to FIL_PAGE_TYPE. */
-  INIT_ROW_FORMAT_DYNAMIC= 1
+  INIT_ROW_FORMAT_DYNAMIC= 1,
+  /** Initialize an undo log page.
+  This is roughly (not exactly) equivalent to the old MLOG_UNDO_INIT record.
+  The current byte offset will be reset to FIL_PAGE_TYPE. */
+  UNDO_INIT= 2
 };
 
 
diff --git a/storage/innobase/include/trx0undo.h b/storage/innobase/include/trx0undo.h
index b68f34ff4da..37421930b9b 100644
--- a/storage/innobase/include/trx0undo.h
+++ b/storage/innobase/include/trx0undo.h
@@ -156,6 +156,13 @@ trx_undo_get_first_rec(const fil_space_t &space, uint32_t page_no,
                        uint16_t offset, ulint mode, buf_block_t*& block,
                        mtr_t *mtr);
 
+/** Initialize an undo log page.
+NOTE: This corresponds to a redo log record and must not be changed!
+@see mtr_t::undo_create()
+@param[in,out]	block	undo log page */
+void trx_undo_page_init(const buf_block_t &block)
+	MY_ATTRIBUTE((nonnull));
+
 /** Allocate an undo log page.
 @param[in,out]	undo	undo log
 @param[in,out]	mtr	mini-transaction that does not hold any page latch
diff --git a/storage/innobase/log/log0recv.cc b/storage/innobase/log/log0recv.cc
index b38c6045379..346f0e09b03 100644
--- a/storage/innobase/log/log0recv.cc
+++ b/storage/innobase/log/log0recv.cc
@@ -44,10 +44,8 @@ Created 9/20/1997 Heikki Tuuri
 #include "buf0flu.h"
 #include "mtr0mtr.h"
 #include "mtr0log.h"
-#include "page0cur.h"
-#include "page0zip.h"
-#include "btr0btr.h"
-#include "btr0cur.h"
+#include "page0page.h"
+#include "trx0undo.h"
 #include "ibuf0ibuf.h"
 #include "trx0undo.h"
 #include "trx0rec.h"
@@ -259,10 +257,20 @@ public:
             !srv_force_recovery)
           goto record_corrupted;
         static_assert(INIT_ROW_FORMAT_REDUNDANT == 0, "compatiblity");
-        static_assert(INIT_ROW_FORMAT_DYNAMIC == 1, "compatibilit");
-        if (UNIV_UNLIKELY(rlen != 1 || *l > INIT_ROW_FORMAT_DYNAMIC))
+        static_assert(INIT_ROW_FORMAT_DYNAMIC == 1, "compatibility");
+        if (UNIV_UNLIKELY(rlen != 1))
+          goto record_corrupted;
+        switch (*l) {
+        default:
           goto record_corrupted;
-        page_create_low(&block, *l != INIT_ROW_FORMAT_REDUNDANT);
+        case INIT_ROW_FORMAT_REDUNDANT:
+        case INIT_ROW_FORMAT_DYNAMIC:
+          page_create_low(&block, *l != INIT_ROW_FORMAT_REDUNDANT);
+          break;
+        case UNDO_INIT:
+          trx_undo_page_init(block);
+          break;
+        }
         last_offset= FIL_PAGE_TYPE;
         goto next_after_applying;
       case WRITE:
diff --git a/storage/innobase/trx/trx0undo.cc b/storage/innobase/trx/trx0undo.cc
index afe6440c5b7..17b50fca56c 100644
--- a/storage/innobase/trx/trx0undo.cc
+++ b/storage/innobase/trx/trx0undo.cc
@@ -292,25 +292,37 @@ trx_undo_get_first_rec(const fil_space_t &space, uint32_t page_no,
 
 /*============== UNDO LOG FILE COPY CREATION AND FREEING ==================*/
 
-/** Initialize the fields in an undo log segment page.
-@param[in,out]	undo_block	undo log segment page
-@param[in,out]	mtr		mini-transaction */
-static void trx_undo_page_init(const buf_block_t *undo_block, mtr_t *mtr)
+/** Initialize an undo log page.
+NOTE: This corresponds to a redo log record and must not be changed!
+@see mtr_t::undo_create()
+@param[in,out]	block	undo log page */
+void trx_undo_page_init(const buf_block_t &block)
 {
-  static_assert(FIL_PAGE_TYPE_ALLOCATED == 0, "compatibility");
-  /* FIXME: FIL_PAGE_TYPE should be FIL_PAGE_TYPE_ALLOCATED here! */
-  ut_ad(mach_read_from_2(FIL_PAGE_TYPE + undo_block->frame) < 0x100);
-  mtr->write<1>(*undo_block, FIL_PAGE_TYPE + 1 + undo_block->frame,
-                FIL_PAGE_UNDO_LOG);
-  compile_time_assert(TRX_UNDO_PAGE_TYPE == 0);
-  compile_time_assert(TRX_UNDO_PAGE_START == 2);
-  compile_time_assert(TRX_UNDO_PAGE_NODE == TRX_UNDO_PAGE_FREE + 2);
-
-  alignas(4) byte hdr[6];
-  mach_write_to_4(hdr, TRX_UNDO_PAGE_HDR + TRX_UNDO_PAGE_HDR_SIZE);
-  memcpy_aligned<2>(hdr + 4, hdr + 2, 2);
-  static_assert(TRX_UNDO_PAGE_FREE == 4, "compatibility");
-  mtr->memcpy(*undo_block, undo_block->frame + TRX_UNDO_PAGE_HDR, hdr, 6);
+  mach_write_to_2(my_assume_aligned<2>(FIL_PAGE_TYPE + block.frame),
+                  FIL_PAGE_UNDO_LOG);
+  static_assert(TRX_UNDO_PAGE_HDR == FIL_PAGE_DATA, "compatibility");
+  memset_aligned<2>(TRX_UNDO_PAGE_HDR + TRX_UNDO_PAGE_TYPE + block.frame,
+                    0, 2);
+  mach_write_to_2(my_assume_aligned<2>
+                  (TRX_UNDO_PAGE_HDR + TRX_UNDO_PAGE_START + block.frame),
+                  TRX_UNDO_PAGE_HDR + TRX_UNDO_PAGE_HDR_SIZE);
+  memcpy_aligned<2>(TRX_UNDO_PAGE_HDR + TRX_UNDO_PAGE_FREE + block.frame,
+                    TRX_UNDO_PAGE_HDR + TRX_UNDO_PAGE_START + block.frame, 2);
+  /* The following corresponds to flst_zero_both(), but without writing log. */
+  memset_aligned<4>(TRX_UNDO_PAGE_HDR + TRX_UNDO_PAGE_NODE + FLST_PREV +
+                    FIL_ADDR_PAGE + block.frame, 0xff, 4);
+  memset_aligned<2>(TRX_UNDO_PAGE_HDR + TRX_UNDO_PAGE_NODE + FLST_PREV +
+                    FIL_ADDR_BYTE + block.frame, 0, 2);
+  memset_aligned<2>(TRX_UNDO_PAGE_HDR + TRX_UNDO_PAGE_NODE + FLST_NEXT +
+                    FIL_ADDR_PAGE + block.frame, 0xff, 4);
+  memset_aligned<2>(TRX_UNDO_PAGE_HDR + TRX_UNDO_PAGE_NODE + FLST_NEXT +
+                    FIL_ADDR_BYTE + block.frame, 0, 2);
+  static_assert(TRX_UNDO_PAGE_NODE + FLST_NEXT + FIL_ADDR_BYTE + 2 ==
+                TRX_UNDO_PAGE_HDR_SIZE, "compatibility");
+  /* Preserve TRX_UNDO_SEG_HDR, but clear the rest of the page. */
+  memset_aligned<2>(TRX_UNDO_SEG_HDR + TRX_UNDO_SEG_HDR_SIZE + block.frame, 0,
+                    srv_page_size - (TRX_UNDO_SEG_HDR + TRX_UNDO_SEG_HDR_SIZE +
+                                     FIL_PAGE_DATA_END));
 }
 
 /** Look for a free slot for an undo log segment.
@@ -383,11 +395,12 @@ trx_undo_seg_create(fil_space_t *space, buf_block_t *rseg_hdr, ulint *id,
 
 	buf_block_dbg_add_level(block, SYNC_TRX_UNDO_PAGE);
 
-	trx_undo_page_init(block, mtr);
+	mtr->undo_create(*block);
+	trx_undo_page_init(*block);
 
-	mtr->write<2,mtr_t::OPT>(*block, TRX_UNDO_PAGE_HDR + TRX_UNDO_PAGE_FREE
-				 + block->frame,
-				 TRX_UNDO_SEG_HDR + TRX_UNDO_SEG_HDR_SIZE);
+	mtr->write<2>(*block, TRX_UNDO_PAGE_HDR + TRX_UNDO_PAGE_FREE
+		      + block->frame,
+		      TRX_UNDO_SEG_HDR + TRX_UNDO_SEG_HDR_SIZE);
 	mtr->write<2,mtr_t::OPT>(*block, TRX_UNDO_SEG_HDR + TRX_UNDO_LAST_LOG
 				 + block->frame, 0U);
 
@@ -565,7 +578,8 @@ buf_block_t* trx_undo_add_page(trx_undo_t* undo, mtr_t* mtr)
 	buf_block_dbg_add_level(new_block, SYNC_TRX_UNDO_PAGE);
 	undo->last_page_no = new_block->page.id.page_no();
 
-	trx_undo_page_init(new_block, mtr);
+	mtr->undo_create(*new_block);
+	trx_undo_page_init(*new_block);
 
 	flst_add_last(header_block, TRX_UNDO_SEG_HDR + TRX_UNDO_PAGE_LIST,
 		      new_block, TRX_UNDO_PAGE_HDR + TRX_UNDO_PAGE_NODE, mtr);