MDEV-12353: Introduce mtr_t::zmemcpy()

Exclusively write MLOG_ZIP_WRITE_STRING records by mtr_t::zmemcpy().

storage/innobase/include/mtr0log.h

@@ -144,6 +144,18 @@ void mtr_t::memcpy(buf_block_t *b, ulint offset, const void *str, ulint len)
   memcpy(*b, offset, len);
 }
 
+/** Write a byte string to a page.
+@param[in,out]  b       ROW_FORMAT=COMPRESSED index page
+@param[in]      ofs     byte offset from b->zip.data
+@param[in]      str     the data to write
+@param[in]      len     length of the data to write */
+inline
+void mtr_t::zmemcpy(buf_page_t *b, ulint offset, const void *str, ulint len)
+{
+  ::memcpy(b->zip.data + offset, str, len);
+  zmemcpy(*b, offset, len);
+}
+
 /** Writes a log record about an operation.
 @param[in]	type		redo log record type
 @param[in]	space_id	tablespace identifier

storage/innobase/include/mtr0mtr.h

@@ -432,6 +432,19 @@ struct mtr_t {
   @param[in]      len     length of the data to write */
   inline void memcpy(buf_block_t *b, ulint offset, const void *str, ulint len);
 
+  /** Write a byte string to a ROW_FORMAT=COMPRESSED page.
+  @param[in]      b       ROW_FORMAT=COMPRESSED index page
+  @param[in]      ofs     byte offset from b.zip.data
+  @param[in]      len     length of the data to write */
+  void zmemcpy(const buf_page_t &b, ulint offset, ulint len);
+
+  /** Write a byte string to a ROW_FORMAT=COMPRESSED page.
+  @param[in,out]  b       ROW_FORMAT=COMPRESSED index page
+  @param[in]      ofs     byte offset from b->zip.data
+  @param[in]      str     the data to write
+  @param[in]      len     length of the data to write */
+  inline void zmemcpy(buf_page_t *b, ulint offset, const void *str, ulint len);
+
   /** Initialize a string of bytes.
   @param[in,out]        b       buffer page
   @param[in]            ofs     byte offset from b->frame

storage/innobase/include/page0zip.ic

@@ -332,35 +332,14 @@ page_zip_write_header(
 	buf_block_t*	block,	/*!< in/out: compressed page */
 	const byte*	str,	/*!< in: address on the uncompressed page */
 	ulint		length,	/*!< in: length of the data */
-	mtr_t*		mtr)	/*!< in: mini-transaction, or NULL */
+	mtr_t*		mtr)	/*!< in/out: mini-transaction */
 {
 	ut_ad(page_align(str) == block->frame);
-
 	const uint16_t pos = page_offset(str);
 
 	ut_ad(pos < PAGE_DATA);
 	ut_ad(pos + length < PAGE_DATA);
-
-	page_zip_des_t* page_zip = &block->page.zip;
-
-	ut_ad(page_zip_simple_validate(page_zip));
-	UNIV_MEM_ASSERT_RW(page_zip->data, page_zip_get_size(page_zip));
-
-	memcpy(page_zip->data + pos, str, length);
-
-	/* The following would fail in page_cur_insert_rec_zip(). */
-	/* ut_ad(page_zip_validate(page_zip, str - pos)); */
-
-	if (byte* log_ptr = mlog_open(mtr, 11 + 2 + 2 + length)) {
-		log_ptr = mlog_write_initial_log_record_low(
-			MLOG_ZIP_WRITE_STRING,
-			block->page.id.space(), block->page.id.page_no(),
-			log_ptr, mtr);
-		mach_write_to_2(log_ptr, pos);
-		mach_write_to_2(log_ptr + 2, length);
-		memcpy(log_ptr + 4, str, length);
-		mlog_close(mtr, log_ptr + 4 + length);
-	}
+	mtr->zmemcpy(&block->page, pos, str, length);
 }
 
 /**********************************************************************//**

storage/innobase/mtr/mtr0log.cc

@@ -297,6 +297,35 @@ void mtr_t::memcpy(const buf_block_t &b, ulint ofs, ulint len)
   mlog_catenate_string(this, b.frame + ofs, len);
 }
 
+/** Write a byte string to a ROW_FORMAT=COMPRESSED page.
+@param[in]      b       ROW_FORMAT=COMPRESSED index page
+@param[in]      ofs     byte offset from b.zip.data
+@param[in]      len     length of the data to write */
+void mtr_t::zmemcpy(const buf_page_t &b, ulint offset, ulint len)
+{
+  ut_ad(page_zip_simple_validate(&b.zip));
+  ut_ad(len);
+  ut_ad(offset + len <= page_zip_get_size(&b.zip));
+  ut_ad(mach_read_from_2(b.zip.data + FIL_PAGE_TYPE) == FIL_PAGE_INDEX ||
+        mach_read_from_2(b.zip.data + FIL_PAGE_TYPE) == FIL_PAGE_RTREE);
+
+  set_modified();
+  if (get_log_mode() != MTR_LOG_ALL)
+  {
+    ut_ad(get_log_mode() == MTR_LOG_NONE ||
+          get_log_mode() == MTR_LOG_NO_REDO);
+    return;
+  }
+
+  byte *l= get_log()->open(11 + 2 + 2);
+  l= mlog_write_initial_log_record_low(MLOG_ZIP_WRITE_STRING, b.id.space(),
+                                       b.id.page_no(), l, this);
+  mach_write_to_2(l, offset);
+  mach_write_to_2(l + 2, len);
+  mlog_close(this, l + 4);
+  mlog_catenate_string(this, b.zip.data + offset, len);
+}
+
 /********************************************************//**
 Parses a log record written by mtr_t::memcpy().
 @return parsed record end, NULL if not a complete record */

storage/innobase/page/page0cur.cc

@@ -1542,17 +1542,7 @@ static inline void page_zip_dir_add_slot(buf_block_t *block,
     {
       memmove(dst, externs, len);
       /* TODO: write MEMMOVE record */
-      if (byte* log_ptr = mlog_open(mtr, 11 + 2 + 2))
-      {
-        log_ptr= mlog_write_initial_log_record_low(MLOG_ZIP_WRITE_STRING,
-                                                   block->page.id.space(),
-                                                   block->page.id.page_no(),
-                                                   log_ptr, mtr);
-        mach_write_to_2(log_ptr, dst - page_zip->data);
-        mach_write_to_2(log_ptr + 2, len);
-        mlog_close(mtr, log_ptr + 4);
-        mlog_catenate_string(mtr, dst, len);
-      }
+      mtr->zmemcpy(block->page, dst - page_zip->data, len);
     }
   }
   else
@@ -1569,17 +1559,7 @@ static inline void page_zip_dir_add_slot(buf_block_t *block,
     byte* dst = stored - PAGE_ZIP_DIR_SLOT_SIZE;
     memmove(dst, stored, len);
     /* TODO: write MEMMOVE record */
-    if (byte* log_ptr = mlog_open(mtr, 11 + 2 + 2))
-    {
-      log_ptr= mlog_write_initial_log_record_low(MLOG_ZIP_WRITE_STRING,
-                                                 block->page.id.space(),
-                                                 block->page.id.page_no(),
-                                                 log_ptr, mtr);
-      mach_write_to_2(log_ptr, dst - page_zip->data);
-      mach_write_to_2(log_ptr + 2, len);
-      mlog_close(mtr, log_ptr + 4);
-      mlog_catenate_string(mtr, dst, len);
-    }
+    mtr->zmemcpy(block->page, dst - page_zip->data, len);
   }
 }