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.

mysql-test/suite/innodb/r/table_flags.result

@@ -201,3 +201,13 @@ CREATE TABLE t1(f1 INT, f2 VARCHAR(1), KEY k1(f2),
 FULLTEXT KEY(f2),
 FOREIGN KEY (f2) REFERENCES t1(f3))ENGINE=InnoDB;
 ERROR HY000: Can't create table `test`.`t1` (errno: 150 "Foreign key constraint is incorrectly formed")
+#
+# MDEV-23199 page_compression flag is missing
+#	for full_crc32 tablespace
+#
+CREATE TABLE t1(f1 BIGINT PRIMARY KEY)ENGINE=InnoDB;
+INSERT INTO t1 VALUES(1);
+ALTER TABLE t1 PAGE_COMPRESSED = 1;
+INSERT INTO t1 VALUES(2);
+# restart
+DROP TABLE t1;

mysql-test/suite/innodb/t/table_flags.test

@@ -16,6 +16,7 @@ call mtr.add_suppression("InnoDB: Operating system error number .* in a file ope
 call mtr.add_suppression("InnoDB: The error means the system cannot find the path specified");
 call mtr.add_suppression("InnoDB: If you are installing InnoDB, remember that you must create directories yourself");
 call mtr.add_suppression("InnoDB: adjusting FSP_SPACE_FLAGS of file ");
+call mtr.add_suppression("InnoDB: Parent table of FTS auxiliary table .* not found.");
 FLUSH TABLES;
 --enable_query_log
 
@@ -237,3 +238,16 @@ call mtr.add_suppression("ERROR HY000: Can't create table `test`.`t1`");
 CREATE TABLE t1(f1 INT, f2 VARCHAR(1), KEY k1(f2),
 		FULLTEXT KEY(f2),
 		FOREIGN KEY (f2) REFERENCES t1(f3))ENGINE=InnoDB;
+
+--echo #
+--echo # MDEV-23199 page_compression flag is missing
+--echo #	for full_crc32 tablespace
+--echo #
+
+CREATE TABLE t1(f1 BIGINT PRIMARY KEY)ENGINE=InnoDB;
+INSERT INTO t1 VALUES(1);
+ALTER TABLE t1 PAGE_COMPRESSED = 1;
+INSERT INTO t1 VALUES(2);
+let $shutdown_timeout = 0;
+--source include/restart_mysqld.inc
+DROP TABLE t1;

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;
 		}
 

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);
 }
 
 /*******************************************************************//**

storage/innobase/include/fil0fil.h

@@ -311,7 +311,7 @@ struct fil_space_t
 	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 @@ struct fil_space_t
 	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 @@ struct fil_space_t
 			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.

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 @@ void recv_apply_hashed_log_recs(bool last_batch)
 	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);