MDEV-12353: Introduce an EXTENDED record subtype TRIM_PAGES

For undo log truncation, commit 055a3334
repurposed the MLOG_FILE_CREATE2 record with a nonzero page size
to indicate that an undo tablespace will be shrunk in size.
In commit 7ae21b18 the
MLOG_FILE_CREATE2 record was replaced by a FILE_CREATE record.

Now that the redo log encoding was changed, there is no actual need
to write a file name in the log record; it suffices to write the
page identifier of the first page that is not part of the file.

This TRIM_PAGES record could allow us to shrink any data files in the
future. For now, it will be limited to undo tablespaces.

mtr_t::log_file_op(): Remove the parameter first_page_no, because
it would always be 0 for file operations.

mtr_t::trim_pages(): Replaces fil_truncate_log().

mtr_t::log_write(): Avoid same_page encoding if !bpage&&!m_last.

fil_op_replay_rename(): Remove the constant parameter first_page_no=0.

storage/innobase/fil/fil0fil.cc

@@ -1814,23 +1814,19 @@ fil_create_directory_for_tablename(
 
 /** Write a log record about a file operation.
 @param type           file operation
-@param space_id       tablespace identifier
 @param first_page_no  first page number in the file
 @param path           file path
 @param new_path       new file path for type=FILE_RENAME */
-inline void mtr_t::log_file_op(mfile_type_t type,
-			       ulint space_id, ulint first_page_no,
+inline void mtr_t::log_file_op(mfile_type_t type, ulint space_id,
 			       const char *path, const char *new_path)
 {
-  ut_ad(first_page_no == 0 || type == FILE_CREATE);
   ut_ad((new_path != nullptr) == (type == FILE_RENAME));
   ut_ad(!(byte(type) & 15));
 
   /* fil_name_parse() requires that there be at least one path
   separator and that the file path end with ".ibd". */
   ut_ad(strchr(path, OS_PATH_SEPARATOR) != NULL);
-  ut_ad(first_page_no /* trimming an undo tablespace */ ||
-        !strcmp(&path[strlen(path) - strlen(DOT_IBD)], DOT_IBD));
+  ut_ad(!strcmp(&path[strlen(path) - strlen(DOT_IBD)], DOT_IBD));
 
   set_modified();
   if (m_log_mode != MTR_LOG_ALL)
@@ -1841,10 +1837,10 @@ inline void mtr_t::log_file_op(mfile_type_t type,
   const size_t new_len= type == FILE_RENAME ? 1 + strlen(new_path) : 0;
   ut_ad(len > 0);
   byte *const log_ptr= m_log.open(1 + 3/*length*/ + 5/*space_id*/ +
-                                  5/*first_page_no*/);
+                                  1/*page_no=0*/);
   byte *end= log_ptr + 1;
   end= mlog_encode_varint(end, space_id);
-  end= mlog_encode_varint(end, first_page_no);
+  *end++= 0;
   if (UNIV_LIKELY(end + len + new_len >= &log_ptr[16]))
   {
     *log_ptr= type;
@@ -1855,7 +1851,7 @@ inline void mtr_t::log_file_op(mfile_type_t type,
       total_len++;
     end= mlog_encode_varint(log_ptr + 1, total_len);
     end= mlog_encode_varint(end, space_id);
-    end= mlog_encode_varint(end, first_page_no);
+    *end++= 0;
   }
   else
   {
@@ -1877,7 +1873,6 @@ inline void mtr_t::log_file_op(mfile_type_t type,
 
 /** Write redo log for renaming a file.
 @param[in]	space_id	tablespace id
-@param[in]	first_page_no	first page number in the file
 @param[in]	old_name	tablespace file name
 @param[in]	new_name	tablespace file name after renaming
 @param[in,out]	mtr		mini-transaction */
@@ -1885,13 +1880,12 @@ static
 void
 fil_name_write_rename_low(
 	ulint		space_id,
-	ulint		first_page_no,
 	const char*	old_name,
 	const char*	new_name,
 	mtr_t*		mtr)
 {
   ut_ad(!is_predefined_tablespace(space_id));
-  mtr->log_file_op(FILE_RENAME, space_id, first_page_no, old_name, new_name);
+  mtr->log_file_op(FILE_RENAME, space_id, old_name, new_name);
 }
 
 /** Write redo log for renaming a file.
@@ -1906,46 +1900,28 @@ fil_name_write_rename(
 {
 	mtr_t	mtr;
 	mtr.start();
-	fil_name_write_rename_low(space_id, 0, old_name, new_name, &mtr);
+	fil_name_write_rename_low(space_id, old_name, new_name, &mtr);
 	mtr.commit();
 	log_write_up_to(mtr.commit_lsn(), true);
 }
 
 /** Write FILE_MODIFY for a file.
 @param[in]	space_id	tablespace id
-@param[in]	first_page_no	first page number in the file
 @param[in]	name		tablespace file name
 @param[in,out]	mtr		mini-transaction */
 static
 void
 fil_name_write(
 	ulint		space_id,
-	ulint		first_page_no,
 	const char*	name,
 	mtr_t*		mtr)
 {
   ut_ad(!is_predefined_tablespace(space_id));
-  mtr->log_file_op(FILE_MODIFY, space_id, first_page_no, name);
-}
-/** Write FILE_MODIFY for a file.
-@param[in]	space		tablespace
-@param[in]	first_page_no	first page number in the file
-@param[in]	file		tablespace file
-@param[in,out]	mtr		mini-transaction */
-static
-void
-fil_name_write(
-	const fil_space_t*	space,
-	ulint			first_page_no,
-	const fil_node_t*	file,
-	mtr_t*			mtr)
-{
-	fil_name_write(space->id, first_page_no, file->name, mtr);
+  mtr->log_file_op(FILE_MODIFY, space_id, name);
 }
 
 /** Replay a file rename operation if possible.
 @param[in]	space_id	tablespace identifier
-@param[in]	first_page_no	first page number in the file
 @param[in]	name		old file name
 @param[in]	new_name	new file name
 @return	whether the operation was successfully applied
@@ -1954,12 +1930,9 @@ name was successfully renamed to new_name)  */
 bool
 fil_op_replay_rename(
 	ulint		space_id,
-	ulint		first_page_no,
 	const char*	name,
 	const char*	new_name)
 {
-	ut_ad(first_page_no == 0);
-
 	/* In order to replay the rename, the following must hold:
 	* The new name is not already used.
 	* A tablespace exists with the old name.
@@ -2341,7 +2314,7 @@ fil_delete_tablespace(
 		mtr_t		mtr;
 
 		mtr.start();
-		mtr.log_file_op(FILE_DELETE, id, 0, path);
+		mtr.log_file_op(FILE_DELETE, id, path);
 		mtr.commit();
 		/* Even if we got killed shortly after deleting the
 		tablespace file, the record must have already been
@@ -2420,17 +2393,6 @@ fil_space_t* fil_truncate_prepare(ulint space_id)
 	return space;
 }
 
-/** Write log about an undo tablespace truncate operation. */
-void fil_truncate_log(fil_space_t* space, ulint size, mtr_t* mtr)
-{
-  /* Write a record with the new size, so that recovery and
-  backup will ignore any preceding redo log records for writing
-  pages that are after the new end of the tablespace. */
-  ut_ad(UT_LIST_GET_LEN(space->chain) == 1);
-  const fil_node_t *file= UT_LIST_GET_FIRST(space->chain);
-  mtr->log_file_op(FILE_CREATE, space->id, size, file->name);
-}
-
 /*******************************************************************//**
 Allocates and builds a file name from a path, a table or tablespace name
 and a suffix. The string must be freed by caller with ut_free().
@@ -2921,7 +2883,7 @@ fil_ibd_create(
 					      false, true);
 		mtr_t mtr;
 		mtr.start();
-		mtr.log_file_op(FILE_CREATE, space_id, 0, node->name);
+		mtr.log_file_op(FILE_CREATE, space_id, node->name);
 		mtr.commit();
 
 		node->find_metadata(file);
@@ -4499,7 +4461,7 @@ fil_mtr_rename_log(
 		}
 
 		fil_name_write_rename_low(
-			old_table->space_id, 0, old_path, tmp_path, mtr);
+			old_table->space_id, old_path, tmp_path, mtr);
 
 		ut_free(tmp_path);
 	}
@@ -4522,7 +4484,7 @@ fil_mtr_rename_log(
 		}
 
 		fil_name_write_rename_low(
-			new_table->space_id, 0, new_path, old_path, mtr);
+			new_table->space_id, new_path, old_path, mtr);
 		ut_free(old_path);
 	}
 
@@ -4562,7 +4524,7 @@ fil_names_write(
 	mtr_t*			mtr)
 {
 	ut_ad(UT_LIST_GET_LEN(space->chain) == 1);
-	fil_name_write(space, 0, UT_LIST_GET_FIRST(space->chain), mtr);
+	fil_name_write(space->id, UT_LIST_GET_FIRST(space->chain)->name, mtr);
 }
 
 /** Note that a non-predefined persistent tablespace has been modified
@@ -4602,7 +4564,7 @@ void fil_names_dirty_and_write(fil_space_t* space)
 				char bogus_name[] = "./test/bogus file.ibd";
 				os_normalize_path(bogus_name);
 				fil_name_write(
-					SRV_SPACE_ID_UPPER_BOUND, 0,
+					SRV_SPACE_ID_UPPER_BOUND,
 					bogus_name, &mtr);
 			});
 

storage/innobase/include/fil0fil.h

@@ -1162,7 +1162,6 @@ fil_create_directory_for_tablename(
 				'databasename/tablename' format */
 /** Replay a file rename operation if possible.
 @param[in]	space_id	tablespace identifier
-@param[in]	first_page_no	first page number in the file
 @param[in]	name		old file name
 @param[in]	new_name	new file name
 @return	whether the operation was successfully applied
@@ -1171,7 +1170,6 @@ name was successfully renamed to new_name)  */
 bool
 fil_op_replay_rename(
 	ulint		space_id,
-	ulint		first_page_no,
 	const char*	name,
 	const char*	new_name)
 	MY_ATTRIBUTE((warn_unused_result));
@@ -1204,10 +1202,6 @@ fil_delete_tablespace(
 @retval	NULL if the tablespace does not exist */
 fil_space_t* fil_truncate_prepare(ulint space_id);
 
-/** Write log about an undo tablespace truncate operation. */
-void fil_truncate_log(fil_space_t* space, ulint size, mtr_t* mtr)
-	MY_ATTRIBUTE((nonnull));
-
 /*******************************************************************//**
 Closes a single-table tablespace. The tablespace must be cached in the
 memory cache. Free all pages used by the tablespace.

storage/innobase/include/mtr0log.h

@@ -393,10 +393,10 @@ inline byte *mtr_t::log_write(const page_id_t id, const buf_page_t *bpage,
   if (!have_len)
     max_len= 1 + 5 + 5;
   else if (!have_offset)
-    max_len= m_last == bpage
+    max_len= bpage && m_last == bpage
       ? 1 + 3
       : 1 + 3 + 5 + 5;
-  else if (m_last == bpage && m_last_offset <= offset)
+  else if (bpage && m_last == bpage && m_last_offset <= offset)
   {
     /* Encode the offset relative from m_last_offset. */
     offset-= m_last_offset;
@@ -631,3 +631,14 @@ inline void mtr_t::undo_append(const buf_block_t &block,
   }
   m_last_offset= FIL_PAGE_TYPE;
 }
+
+/** Trim the end of a tablespace.
+@param id       first page identifier that will not be in the file */
+inline void mtr_t::trim_pages(const page_id_t id)
+{
+  if (m_log_mode != MTR_LOG_ALL)
+    return;
+  byte *l= log_write<EXTENDED>(id, nullptr, 1, true);
+  *l++= TRIM_PAGES;
+  m_log.close(l);
+}

storage/innobase/include/mtr0mtr.h

@@ -556,15 +556,17 @@ struct mtr_t {
   @param len      length of the undo record, in bytes */
   inline void undo_append(const buf_block_t &block,
                           const void *data, size_t len);
+  /** Trim the end of a tablespace.
+  @param id       first page identifier that will not be in the file */
+  inline void trim_pages(const page_id_t id);
 
   /** Write a log record about a file operation.
   @param type           file operation
   @param space_id       tablespace identifier
-  @param first_page_no  first page number in the file
   @param path           file path
   @param new_path       new file path for type=FILE_RENAME */
   inline void log_file_op(mfile_type_t type, ulint space_id,
-                          ulint first_page_no, const char *path,
+                          const char *path,
                           const char *new_path= nullptr);
 
 private:

storage/innobase/include/mtr0types.h

@@ -285,7 +285,9 @@ enum mrec_ext_t
   and include the total size of the record being deleted.
   The current byte offset will be reset to FIL_PAGE_TYPE.
   This is similar to the old MLOG_COMP_REC_DELETE record. */
-  DELETE_ROW_FORMAT_DYNAMIC= 9
+  DELETE_ROW_FORMAT_DYNAMIC= 9,
+  /** Truncate a data file. */
+  TRIM_PAGES= 10
 };
 
 

storage/innobase/log/log0recv.cc

@@ -1977,14 +1977,31 @@ bool recv_sys_t::parse(lsn_t checkpoint_lsn, store_t store, bool apply)
         last_offset= 1; /* the next record must not be same_page  */
         goto free_or_init_page;
       case INIT_PAGE:
-      free_or_init_page:
         last_offset= FIL_PAGE_TYPE;
+      free_or_init_page:
         if (UNIV_UNLIKELY(rlen != 0))
           goto record_corrupted;
         break;
       case EXTENDED:
         if (UNIV_UNLIKELY(!rlen))
           goto record_corrupted;
+        if (rlen == 1 && *l == TRIM_PAGES)
+        {
+#if 0 /* For now, we can only truncate an undo log tablespace */
+          if (UNIV_UNLIKELY(!space_id || !page_no))
+            goto record_corrupted;
+#else
+          if (!srv_is_undo_tablespace(space_id) ||
+              page_no != SRV_UNDO_TABLESPACE_SIZE_IN_PAGES)
+            goto record_corrupted;
+          static_assert(UT_ARR_SIZE(truncated_undo_spaces) ==
+                        TRX_SYS_MAX_UNDO_SPACES, "compatibility");
+          truncated_undo_spaces[space_id - srv_undo_space_id_start]=
+            { recovered_lsn, page_no };
+#endif
+          last_offset= 1; /* the next record must not be same_page  */
+          continue;
+        }
         last_offset= FIL_PAGE_TYPE;
         break;
       case RESERVED:
@@ -2145,7 +2162,7 @@ bool recv_sys_t::parse(lsn_t checkpoint_lsn, store_t store, bool apply)
         }
         /* fall through */
       case FILE_CREATE:
-        if (UNIV_UNLIKELY(space_id == 0))
+        if (UNIV_UNLIKELY(!space_id || page_no))
           goto file_rec_error;
         /* There is no terminating NUL character. Names must end in .ibd.
         For FILE_RENAME, there is a NUL between the two file names. */
@@ -2167,24 +2184,6 @@ bool recv_sys_t::parse(lsn_t checkpoint_lsn, store_t store, bool apply)
             goto file_rec_error;
         }
 
-        if (page_no)
-        {
-          if (UNIV_UNLIKELY((b & 0xf0) != FILE_CREATE))
-            goto file_rec_error;
-          /* truncating an undo log tablespace */
-          ut_ad(fnend - fn >= 7);
-          ut_ad(!memcmp(fnend - 7, "undo", 4));
-          ut_d(char n[4]; char *end; memcpy(n, fnend - 3, 3); n[3]= 0);
-          ut_ad(strtoul(n, &end, 10) <= 127);
-          ut_ad(end == &n[3]);
-          ut_ad(page_no == SRV_UNDO_TABLESPACE_SIZE_IN_PAGES);
-          ut_ad(srv_is_undo_tablespace(space_id));
-          static_assert(UT_ARR_SIZE(truncated_undo_spaces) ==
-                        TRX_SYS_MAX_UNDO_SPACES, "compatibility");
-          truncated_undo_spaces[space_id - srv_undo_space_id_start]=
-            { recovered_lsn, page_no };
-          continue;
-        }
         if (is_predefined_tablespace(space_id))
           goto file_rec_error;
         if (fnend - fn < 4 || memcmp(fnend - 4, DOT_IBD, 4))
@@ -2204,7 +2203,7 @@ bool recv_sys_t::parse(lsn_t checkpoint_lsn, store_t store, bool apply)
                       fn2 ? static_cast<ulint>(fn2end - fn2) : 0);
 
         if (!fn2 || !apply);
-        else if (!fil_op_replay_rename(space_id, 0, fn, fn2))
+        else if (!fil_op_replay_rename(space_id, fn, fn2))
           found_corrupt_fs= true;
         const_cast<char&>(fn[rlen])= saved_end;
         if (UNIV_UNLIKELY(found_corrupt_fs))

storage/innobase/trx/trx0purge.cc

@@ -687,7 +687,7 @@ static void trx_purge_truncate_history()
 		const ulint size = SRV_UNDO_TABLESPACE_SIZE_IN_PAGES;
 		mtr.start();
 		mtr_x_lock_space(purge_sys.truncate.current, &mtr);
-		fil_truncate_log(purge_sys.truncate.current, size, &mtr);
+		mtr.trim_pages(page_id_t(space.id, size));
 		fsp_header_init(purge_sys.truncate.current, size, &mtr);
 		mutex_enter(&fil_system.mutex);
 		purge_sys.truncate.current->size = file->size = size;