MDEV-23199 page_compression flag is missing for full_crc32 tablespace

Problem:
======
Making the tablespace as page_compressed doesn't do table rebuild.
It does change only the FSP_SPACE_FLAGS.

During recovery:
1) InnoDB encounters FILE_CREATE redo log and opens the tablespace
with old FSP_SPACE_FLAGS value.
2) Only parsing of redo log has been finished. Now InnoDB tries to
load the table. If the existing tablespace flags doesn't match
with table flags then InnoDB should read page0. But in
fsp_flags_try_adjust(), skips the page read for full_crc32 format.
3) After that, InnoDB tries to open the clustered index and it
leads to failure of page validation.

Fix:
===
While parsing the redo log record, track FSP_SPACE_FLAGS in
recv_spaces for the respective space id. Assign the flags for
the tablespace when it is loaded.

recv_parse_set_size_and_flags(): Parse the redo log to set the
tablespace recovery size and flags.

fil_space_set_recv_size_and_flags(): Changed from
fil_space_set_recv_size(). To set the recovery size and flags of
the tablespace.

Introduce flags variable in file_name_t to maintain the tablespace
flag which we encountered during parsing of redo log.

is_flags_full_crc32_equal(), is_flags_non_full_crc32_equal(): Rename
the variable page_ssize and space_page_ssize with fcrc32_psize and
non_fcrc32_psize.

4 files changed, 87 insertions, 60 deletions

diff --git a/storage/innobase/dict/dict0load.cc b/storage/innobase/dict/dict0load.cc
index 8094e4d9c08..41abe2d0a48 100644
--- a/storage/innobase/dict/dict0load.cc
+++ b/storage/innobase/dict/dict0load.cc
@@ -2710,8 +2710,9 @@ dict_get_and_save_data_dir_path(
 		if (table->data_dir_path == NULL) {
 			/* Since we did not set the table data_dir_path,
 			unset the flag.  This does not change SYS_DATAFILES
-			or SYS_TABLES or FSP_FLAGS on the header page of the
-			tablespace, but it makes dict_table_t consistent. */
+			or SYS_TABLES or FSP_SPACE_FLAGS on the header page
+			of the tablespace, but it makes dict_table_t
+			consistent. */
 			table->flags &= ~DICT_TF_MASK_DATA_DIR;
 		}
 
diff --git a/storage/innobase/fil/fil0fil.cc b/storage/innobase/fil/fil0fil.cc
index 72639d16fd6..97f0a578fe9 100644
--- a/storage/innobase/fil/fil0fil.cc
+++ b/storage/innobase/fil/fil0fil.cc
@@ -1403,22 +1403,18 @@ fil_space_get_space(
 	return(space);
 }
 
-/** Set the recovered size of a tablespace in pages.
-@param id	tablespace ID
-@param size	recovered size in pages */
 UNIV_INTERN
-void
-fil_space_set_recv_size(ulint id, ulint size)
+void fil_space_set_recv_size_and_flags(ulint id, ulint size, uint32_t flags)
 {
-	mutex_enter(&fil_system.mutex);
-	ut_ad(size);
-	ut_ad(id < SRV_LOG_SPACE_FIRST_ID);
-
-	if (fil_space_t* space = fil_space_get_space(id)) {
-		space->recv_size = size;
-	}
+  mutex_enter(&fil_system.mutex);
+  ut_ad(id < SRV_LOG_SPACE_FIRST_ID);
 
-	mutex_exit(&fil_system.mutex);
+  if (fil_space_t* space= fil_space_get_space(id))
+  {
+    if (!size) space->recv_size= size;
+    if (flags != FSP_FLAGS_FCRC32_MASK_MARKER) space->flags= flags;
+  }
+  mutex_exit(&fil_system.mutex);
 }
 
 /*******************************************************************//**
diff --git a/storage/innobase/include/fil0fil.h b/storage/innobase/include/fil0fil.h
index d0af38bd80f..3001817a78c 100644
--- a/storage/innobase/include/fil0fil.h
+++ b/storage/innobase/include/fil0fil.h
@@ -311,7 +311,7 @@ public:
 	ulint		flags;
 
 	/** Determine if full_crc32 is used for a data file
-	@param[in]	flags	tablespace flags (FSP_FLAGS)
+	@param[in]	flags	tablespace flags (FSP_SPACE_FLAGS)
 	@return whether the full_crc32 algorithm is active */
 	static bool full_crc32(ulint flags) {
 		return flags & FSP_FLAGS_FCRC32_MASK_MARKER;
@@ -430,23 +430,23 @@ public:
 	static bool is_flags_full_crc32_equal(ulint flags, ulint expected)
 	{
 		ut_ad(full_crc32(flags));
-		ulint page_ssize = FSP_FLAGS_FCRC32_GET_PAGE_SSIZE(flags);
+		ulint fcrc32_psize = FSP_FLAGS_FCRC32_GET_PAGE_SSIZE(flags);
 
 		if (full_crc32(expected)) {
 			/* The data file may have been created with a
 			different innodb_compression_algorithm. But
 			we only support one innodb_page_size for all files. */
-			return page_ssize
-				== FSP_FLAGS_FCRC32_GET_PAGE_SSIZE(expected);
+			return fcrc32_psize
+			       == FSP_FLAGS_FCRC32_GET_PAGE_SSIZE(expected);
 		}
 
-		ulint space_page_ssize = FSP_FLAGS_GET_PAGE_SSIZE(expected);
+		ulint non_fcrc32_psize = FSP_FLAGS_GET_PAGE_SSIZE(expected);
 
-		if (page_ssize == 5) {
-			if (space_page_ssize) {
+		if (!non_fcrc32_psize) {
+			if (fcrc32_psize != 5) {
 				return false;
 			}
-		} else if (space_page_ssize != page_ssize) {
+		} else if (fcrc32_psize != non_fcrc32_psize) {
 			return false;
 		}
 
@@ -464,15 +464,15 @@ public:
 			return false;
 		}
 
-		ulint page_ssize = FSP_FLAGS_GET_PAGE_SSIZE(flags);
-		ulint space_page_ssize = FSP_FLAGS_FCRC32_GET_PAGE_SSIZE(
+		ulint non_fcrc32_psize = FSP_FLAGS_GET_PAGE_SSIZE(flags);
+		ulint fcrc32_psize = FSP_FLAGS_FCRC32_GET_PAGE_SSIZE(
 			expected);
 
-		if (page_ssize) {
-			if (space_page_ssize != 5) {
+		if (!non_fcrc32_psize) {
+			if (fcrc32_psize != 5) {
 				return false;
 			}
-		} else if (space_page_ssize != page_ssize) {
+		} else if (fcrc32_psize != non_fcrc32_psize) {
 			return false;
 		}
 
@@ -1060,11 +1060,12 @@ fil_space_free(
 	bool		x_latched);
 
 /** Set the recovered size of a tablespace in pages.
-@param id	tablespace ID
-@param size	recovered size in pages */
+@param	id	tablespace ID
+@param	size	recovered size in pages
+@param	flags	tablespace flags */
 UNIV_INTERN
-void
-fil_space_set_recv_size(ulint id, ulint size);
+void fil_space_set_recv_size_and_flags(ulint id, ulint size, uint32_t flags);
+
 /*******************************************************************//**
 Returns the size of the space in pages. The tablespace must be cached in the
 memory cache.
diff --git a/storage/innobase/log/log0recv.cc b/storage/innobase/log/log0recv.cc
index c9035ee8b79..4698b9fa441 100644
--- a/storage/innobase/log/log0recv.cc
+++ b/storage/innobase/log/log0recv.cc
@@ -144,17 +144,21 @@ struct file_name_t {
 	fil_status	status;
 
 	/** FSP_SIZE of tablespace */
-	ulint		size;
+	ulint		size = 0;
 
 	/** the log sequence number of the last observed MLOG_INDEX_LOAD
 	record for the tablespace */
-	lsn_t		enable_lsn;
+	lsn_t		enable_lsn = 0;
+
+	/** Dummy flags before they have been read from the .ibd file */
+	static constexpr uint32_t initial_flags = FSP_FLAGS_FCRC32_MASK_MARKER;
+	/** FSP_SPACE_FLAGS of tablespace */
+	uint32_t	flags = initial_flags;
 
 	/** Constructor */
 	file_name_t(std::string name_, bool deleted)
 		: name(std::move(name_)), space(NULL),
-		status(deleted ? DELETED: NORMAL),
-		size(0), enable_lsn(0) {}
+		status(deleted ? DELETED: NORMAL) {}
 
 	/** Report a MLOG_INDEX_LOAD operation, meaning that
 	mlog_init for any earlier LSN must be skipped.
@@ -440,14 +444,18 @@ fil_name_process(
 		case FIL_LOAD_OK:
 			ut_ad(space != NULL);
 
-			if (f.space == NULL || f.space == space) {
-
-				if (f.size && f.space == NULL) {
-					fil_space_set_recv_size(space->id, f.size);
+			if (!f.space) {
+				if (f.size
+				    || f.flags != f.initial_flags) {
+					fil_space_set_recv_size_and_flags(
+						space->id, f.size, f.flags);
 				}
 
-				f.name = fname.name;
 				f.space = space;
+				goto same_space;
+			} else if (f.space == space) {
+same_space:
+				f.name = fname.name;
 				f.status = file_name_t::NORMAL;
 			} else {
 				ib::error() << "Tablespace " << space_id
@@ -2441,6 +2449,44 @@ apply:
 	mutex_exit(&recv_sys.mutex);
 }
 
+/** Parse the redo log to set the space recovery size and flags
+@param[in]	ptr	pointer to parsing redo buffer
+@param[in]	end_ptr	end of the parsing redo buffer
+@param[in]	space	tablespace id */
+static
+void recv_parse_set_size_and_flags(const byte *ptr, byte *end_ptr,
+                                   ulint space)
+{
+  switch (const uint16_t offset= mach_read_from_2(ptr))
+  {
+  default:
+    break;
+  case FSP_HEADER_OFFSET + FSP_SIZE:
+  case FSP_HEADER_OFFSET + FSP_SPACE_FLAGS:
+    ptr += 2;
+    ulint val= mach_parse_compressed(&ptr, end_ptr);
+    recv_spaces_t::iterator it= recv_spaces.find(space);
+
+    ut_ad(!recv_sys.mlog_checkpoint_lsn || space == TRX_SYS_SPACE ||
+          srv_is_undo_tablespace(space) || it != recv_spaces.end());
+
+    if (offset == FSP_HEADER_OFFSET + FSP_SIZE)
+      fil_space_set_recv_size_and_flags(
+         space, val, FSP_FLAGS_FCRC32_MASK_MARKER);
+    else
+      fil_space_set_recv_size_and_flags(
+         space, 0, static_cast<uint32_t>(val));
+
+    if (it == recv_spaces.end() || it->second.space)
+      return;
+
+    if (offset == FSP_HEADER_OFFSET + FSP_SIZE)
+      it->second.size= val;
+    else
+      it->second.flags= static_cast<uint32_t>(val);
+  }
+}
+
 /** Tries to parse a single log record.
 @param[out]	type		log record type
 @param[in]	ptr		pointer to a buffer
@@ -2527,25 +2573,8 @@ recv_parse_log_rec(
 		return(0);
 	}
 
-	if (*page_no == 0 && *type == MLOG_4BYTES
-	    && apply
-	    && mach_read_from_2(old_ptr) == FSP_HEADER_OFFSET + FSP_SIZE) {
-		old_ptr += 2;
-
-		ulint size = mach_parse_compressed(&old_ptr, end_ptr);
-
-		recv_spaces_t::iterator it = recv_spaces.find(*space);
-
-		ut_ad(!recv_sys.mlog_checkpoint_lsn
-		      || *space == TRX_SYS_SPACE
-		      || srv_is_undo_tablespace(*space)
-		      || it != recv_spaces.end());
-
-		if (it != recv_spaces.end() && !it->second.space) {
-			it->second.size = size;
-		}
-
-		fil_space_set_recv_size(*space, size);
+	if (*page_no == 0 && *type == MLOG_4BYTES && apply) {
+		recv_parse_set_size_and_flags(old_ptr, end_ptr, *space);
 	}
 
 	return ulint(new_ptr - ptr);