MDEV-8233: InnoDB: Assertion failure in fil_page_decompress with encrypted tables

Analysis: Problem was that used compression method needs to be stored
to the page.

Fixed by storing compression method after key_version to the page.

mysql-test/suite/encryption/t/innodb_encryption-page-compression.test

@@ -12,12 +12,156 @@ let $innodb_file_per_table_orig = `SELECT @@innodb_file_per_table`;
 
 SET GLOBAL innodb_file_format = `Barracuda`;
 SET GLOBAL innodb_file_per_table = ON;
+SET GLOBAL innodb_encryption_threads = 4;
+SET GLOBAL innodb_encrypt_tables = on;
+
 # zlib
 set global innodb_compression_algorithm = 1;
 
+create table innodb_normal (c1 int, b char(20)) engine=innodb;
+show warnings;
+create table innodb_page_compressed1 (c1 int, b char(20)) engine=innodb page_compressed=1 page_compression_level=1;
+show warnings;
+show create table innodb_page_compressed1;
+create table innodb_page_compressed2 (c1 int, b char(20)) engine=innodb page_compressed=1 page_compression_level=2;
+show warnings;
+show create table innodb_page_compressed2;
+create table innodb_page_compressed3 (c1 int, b char(20)) engine=innodb page_compressed=1 page_compression_level=3;
+show warnings;
+show create table innodb_page_compressed3;
+create table innodb_page_compressed4 (c1 int, b char(20)) engine=innodb page_compressed=1 page_compression_level=4;
+show warnings;
+show create table innodb_page_compressed4;
+create table innodb_page_compressed5 (c1 int, b char(20)) engine=innodb page_compressed=1 page_compression_level=5;
+show warnings;
+show create table innodb_page_compressed5;
+create table innodb_page_compressed6 (c1 int, b char(20)) engine=innodb page_compressed=1 page_compression_level=6;
+show warnings;
+show create table innodb_page_compressed6;
+create table innodb_page_compressed7 (c1 int, b char(20)) engine=innodb page_compressed=1 page_compression_level=7;
+show warnings;
+show create table innodb_page_compressed7;
+create table innodb_page_compressed8 (c1 int, b char(20)) engine=innodb page_compressed=1 page_compression_level=8;
+show warnings;
+show create table innodb_page_compressed8;
+create table innodb_page_compressed9 (c1 int, b char(20)) engine=innodb page_compressed=1 page_compression_level=9;
+show warnings;
+show create table innodb_page_compressed9;
+delimiter //;
+create procedure innodb_insert_proc (repeat_count int)
+begin
+  declare current_num int;
+  set current_num = 0;
+  while current_num < repeat_count do
+    insert into innodb_normal values(current_num,'testing..');
+    set current_num = current_num + 1;
+  end while;
+end//
+delimiter ;//
+commit;
+
+set autocommit=0;
+call innodb_insert_proc(5000);
+commit;
+set autocommit=1;
+select count(*) from innodb_normal;
+insert into innodb_page_compressed1 select * from innodb_normal;
+insert into innodb_page_compressed2 select * from innodb_normal;
+insert into innodb_page_compressed3 select * from innodb_normal;
+insert into innodb_page_compressed4 select * from innodb_normal;
+insert into innodb_page_compressed5 select * from innodb_normal;
+insert into innodb_page_compressed6 select * from innodb_normal;
+insert into innodb_page_compressed7 select * from innodb_normal;
+insert into innodb_page_compressed8 select * from innodb_normal;
+insert into innodb_page_compressed9 select * from innodb_normal;
+commit;
+select count(*) from innodb_page_compressed1 where c1 < 500000;
+select count(*) from innodb_page_compressed2 where c1 < 500000;
+select count(*) from innodb_page_compressed3 where c1 < 500000;
+select count(*) from innodb_page_compressed4 where c1 < 500000;
+select count(*) from innodb_page_compressed5 where c1 < 500000;
+select count(*) from innodb_page_compressed6 where c1 < 500000;
+select count(*) from innodb_page_compressed7 where c1 < 500000;
+select count(*) from innodb_page_compressed8 where c1 < 500000;
+select count(*) from innodb_page_compressed9 where c1 < 500000;
+
+SELECT variable_value >= 0 FROM information_schema.global_status WHERE LOWER(variable_name) = 'innodb_num_pages_encrypted';
+SELECT variable_value >= 0 FROM information_schema.global_status WHERE LOWER(variable_name) = 'innodb_num_pages_decrypted';
+SELECT variable_value >= 0 FROM information_schema.global_status WHERE LOWER(variable_name) = 'innodb_num_pages_page_compressed';
+SELECT variable_value >= 0 FROM information_schema.global_status WHERE LOWER(variable_name) = 'innodb_num_pages_page_decompressed';
+
+
+--source include/restart_mysqld.inc
+
+select count(*) from innodb_page_compressed1 where c1 < 500000;
+select count(*) from innodb_page_compressed2 where c1 < 500000;
+select count(*) from innodb_page_compressed3 where c1 < 500000;
+select count(*) from innodb_page_compressed4 where c1 < 500000;
+select count(*) from innodb_page_compressed5 where c1 < 500000;
+select count(*) from innodb_page_compressed6 where c1 < 500000;
+select count(*) from innodb_page_compressed7 where c1 < 500000;
+select count(*) from innodb_page_compressed8 where c1 < 500000;
+select count(*) from innodb_page_compressed9 where c1 < 500000;
+update innodb_page_compressed1 set c1 = c1 + 1;
+update innodb_page_compressed2 set c1 = c1 + 1;
+update innodb_page_compressed3 set c1 = c1 + 1;
+update innodb_page_compressed4 set c1 = c1 + 1;
+update innodb_page_compressed5 set c1 = c1 + 1;
+update innodb_page_compressed6 set c1 = c1 + 1;
+update innodb_page_compressed7 set c1 = c1 + 1;
+update innodb_page_compressed8 set c1 = c1 + 1;
+update innodb_page_compressed9 set c1 = c1 + 1;
+
+SELECT variable_value >= 0 FROM information_schema.global_status WHERE LOWER(variable_name) = 'innodb_num_pages_encrypted';
+SELECT variable_value >= 0 FROM information_schema.global_status WHERE LOWER(variable_name) = 'innodb_num_pages_decrypted';
+SELECT variable_value >= 0 FROM information_schema.global_status WHERE LOWER(variable_name) = 'innodb_num_pages_page_compressed';
+SELECT variable_value >= 0 FROM information_schema.global_status WHERE LOWER(variable_name) = 'innodb_num_pages_page_decompressed';
+
+SET GLOBAL innodb_encryption_threads = 4;
+SET GLOBAL innodb_encrypt_tables = off;
+
+select count(*) from innodb_page_compressed1 where c1 < 500000;
+select count(*) from innodb_page_compressed2 where c1 < 500000;
+select count(*) from innodb_page_compressed3 where c1 < 500000;
+select count(*) from innodb_page_compressed4 where c1 < 500000;
+select count(*) from innodb_page_compressed5 where c1 < 500000;
+select count(*) from innodb_page_compressed6 where c1 < 500000;
+select count(*) from innodb_page_compressed7 where c1 < 500000;
+select count(*) from innodb_page_compressed8 where c1 < 500000;
+select count(*) from innodb_page_compressed9 where c1 < 500000;
+update innodb_page_compressed1 set c1 = c1 + 1;
+update innodb_page_compressed2 set c1 = c1 + 1;
+update innodb_page_compressed3 set c1 = c1 + 1;
+update innodb_page_compressed4 set c1 = c1 + 1;
+update innodb_page_compressed5 set c1 = c1 + 1;
+update innodb_page_compressed6 set c1 = c1 + 1;
+update innodb_page_compressed7 set c1 = c1 + 1;
+update innodb_page_compressed8 set c1 = c1 + 1;
+update innodb_page_compressed9 set c1 = c1 + 1;
+
+SELECT variable_value >= 0 FROM information_schema.global_status WHERE LOWER(variable_name) = 'innodb_num_pages_encrypted';
+SELECT variable_value >= 0 FROM information_schema.global_status WHERE LOWER(variable_name) = 'innodb_num_pages_decrypted';
+SELECT variable_value >= 0 FROM information_schema.global_status WHERE LOWER(variable_name) = 'innodb_num_pages_page_compressed';
+SELECT variable_value >= 0 FROM information_schema.global_status WHERE LOWER(variable_name) = 'innodb_num_pages_page_decompressed';
+
+drop procedure innodb_insert_proc;
+drop table innodb_normal;
+drop table innodb_page_compressed1;
+drop table innodb_page_compressed2;
+drop table innodb_page_compressed3;
+drop table innodb_page_compressed4;
+drop table innodb_page_compressed5;
+drop table innodb_page_compressed6;
+drop table innodb_page_compressed7;
+drop table innodb_page_compressed8;
+drop table innodb_page_compressed9;
+
 # reset system
 --disable_query_log
 EVAL SET GLOBAL innodb_compression_algorithm = $innodb_compression_algorithm_orig;
 EVAL SET GLOBAL innodb_file_per_table = $innodb_file_per_table_orig;
 EVAL SET GLOBAL innodb_file_format = $innodb_file_format_orig;
 --enable_query_log
+
+SET GLOBAL innodb_encryption_threads = 4;
+SET GLOBAL innodb_encrypt_tables = on;

storage/innobase/buf/buf0dblwr.cc

@@ -744,6 +744,14 @@ buf_dblwr_check_page_lsn(
 /*=====================*/
 	const page_t*	page)		/*!< in: page to check */
 {
+	ibool page_compressed = (mach_read_from_2(page+FIL_PAGE_TYPE) == FIL_PAGE_PAGE_COMPRESSED);
+	uint key_version = mach_read_from_4(page + FIL_PAGE_FILE_FLUSH_LSN_OR_KEY_VERSION);
+
+	/* Ignore page compressed or encrypted pages */
+	if (page_compressed || key_version) {
+		return;
+	}
+
 	if (memcmp(page + (FIL_PAGE_LSN + 4),
 		   page + (UNIV_PAGE_SIZE
 			   - FIL_PAGE_END_LSN_OLD_CHKSUM + 4),

storage/innobase/fil/fil0crypt.cc

@@ -618,6 +618,7 @@ fil_space_encrypt(
 	}
 
 	ibool page_compressed = (mach_read_from_2(src_frame+FIL_PAGE_TYPE) == FIL_PAGE_PAGE_COMPRESSED);
+	ulint page_comp_method = mach_read_from_8(src_frame+FIL_PAGE_FILE_FLUSH_LSN_OR_KEY_VERSION);
 
 	/* FIL page header is not encrypted */
 	memcpy(dst_frame, src_frame, FIL_PAGE_DATA);
@@ -698,6 +699,9 @@ fil_space_encrypt(
 		// store the post-encryption checksum after the key-version
 		mach_write_to_4(dst_frame + FIL_PAGE_FILE_FLUSH_LSN_OR_KEY_VERSION + 4,
 			        checksum);
+	} else {
+		mach_write_to_4(dst_frame + FIL_PAGE_FILE_FLUSH_LSN_OR_KEY_VERSION + 4,
+			page_comp_method);
 	}
 
 	srv_stats.pages_encrypted.inc();
@@ -744,6 +748,7 @@ fil_space_decrypt(
 	ulint page_type = mach_read_from_2(src_frame+FIL_PAGE_TYPE);
 	uint key_version = mach_read_from_4(src_frame + FIL_PAGE_FILE_FLUSH_LSN_OR_KEY_VERSION);
 	bool page_compressed = (page_type == FIL_PAGE_PAGE_COMPRESSED);
+	ulint page_comp_method = mach_read_from_4(src_frame + FIL_PAGE_FILE_FLUSH_LSN_OR_KEY_VERSION + 4);
 
 	if (key_version == ENCRYPTION_KEY_NOT_ENCRYPTED) {
 		//TODO: is this really needed ?
@@ -806,6 +811,8 @@ fil_space_decrypt(
 
 		// clear key-version & crypt-checksum from dst
 		memset(dst_frame + FIL_PAGE_FILE_FLUSH_LSN_OR_KEY_VERSION, 0, 8);
+	} else {
+		mach_write_to_8(dst_frame + FIL_PAGE_FILE_FLUSH_LSN_OR_KEY_VERSION, page_comp_method);
 	}
 
 	srv_stats.pages_decrypted.inc();

storage/innobase/fil/fil0fil.cc

@@ -5594,17 +5594,19 @@ fil_space_get_block_size(
 	ulint	len)
 {
 	ulint block_size = 512;
+	ut_ad(!mutex_own(&fil_system->mutex));
+
+	mutex_enter(&fil_system->mutex);
 	fil_space_t* space = fil_space_get_space(space_id);
 
 	if (space) {
-		mutex_enter(&fil_system->mutex);
 		fil_node_t* node = fil_space_get_node(space, space_id, &block_offset, 0, len);
-		mutex_exit(&fil_system->mutex);
 
 		if (node) {
 			block_size = node->file_block_size;
 		}
 	}
+	mutex_exit(&fil_system->mutex);
 
 	return block_size;
 }

storage/xtradb/buf/buf0dblwr.cc

@@ -743,6 +743,14 @@ buf_dblwr_check_page_lsn(
 /*=====================*/
 	const page_t*	page)		/*!< in: page to check */
 {
+	ibool page_compressed = (mach_read_from_2(page+FIL_PAGE_TYPE) == FIL_PAGE_PAGE_COMPRESSED);
+	uint key_version = mach_read_from_4(page + FIL_PAGE_FILE_FLUSH_LSN_OR_KEY_VERSION);
+
+	/* Ignore page compressed or encrypted pages */
+	if (page_compressed || key_version) {
+		return;
+	}
+
 	if (memcmp(page + (FIL_PAGE_LSN + 4),
 		   page + (UNIV_PAGE_SIZE
 			   - FIL_PAGE_END_LSN_OLD_CHKSUM + 4),

storage/xtradb/fil/fil0crypt.cc

@@ -618,6 +618,7 @@ fil_space_encrypt(
 	}
 
 	ibool page_compressed = (mach_read_from_2(src_frame+FIL_PAGE_TYPE) == FIL_PAGE_PAGE_COMPRESSED);
+	ulint page_comp_method = mach_read_from_8(src_frame+FIL_PAGE_FILE_FLUSH_LSN_OR_KEY_VERSION);
 
 	/* FIL page header is not encrypted */
 	memcpy(dst_frame, src_frame, FIL_PAGE_DATA);
@@ -698,6 +699,9 @@ fil_space_encrypt(
 		// store the post-encryption checksum after the key-version
 		mach_write_to_4(dst_frame + FIL_PAGE_FILE_FLUSH_LSN_OR_KEY_VERSION + 4,
 			        checksum);
+	} else {
+		mach_write_to_4(dst_frame + FIL_PAGE_FILE_FLUSH_LSN_OR_KEY_VERSION + 4,
+			page_comp_method);
 	}
 
 	srv_stats.pages_encrypted.inc();
@@ -744,6 +748,7 @@ fil_space_decrypt(
 	ulint page_type = mach_read_from_2(src_frame+FIL_PAGE_TYPE);
 	uint key_version = mach_read_from_4(src_frame + FIL_PAGE_FILE_FLUSH_LSN_OR_KEY_VERSION);
 	bool page_compressed = (page_type == FIL_PAGE_PAGE_COMPRESSED);
+	ulint page_comp_method = mach_read_from_4(src_frame + FIL_PAGE_FILE_FLUSH_LSN_OR_KEY_VERSION + 4);
 
 	if (key_version == ENCRYPTION_KEY_NOT_ENCRYPTED) {
 		//TODO: is this really needed ?
@@ -806,6 +811,8 @@ fil_space_decrypt(
 
 		// clear key-version & crypt-checksum from dst
 		memset(dst_frame + FIL_PAGE_FILE_FLUSH_LSN_OR_KEY_VERSION, 0, 8);
+	} else {
+		mach_write_to_8(dst_frame + FIL_PAGE_FILE_FLUSH_LSN_OR_KEY_VERSION, page_comp_method);
 	}
 
 	srv_stats.pages_decrypted.inc();

storage/xtradb/fil/fil0fil.cc

@@ -5623,17 +5623,19 @@ fil_space_get_block_size(
 	ulint	len)
 {
 	ulint block_size = 512;
+	ut_ad(!mutex_own(&fil_system->mutex));
+
+	mutex_enter(&fil_system->mutex);
 	fil_space_t* space = fil_space_get_space(space_id);
 
 	if (space) {
-		mutex_enter(&fil_system->mutex);
 		fil_node_t* node = fil_space_get_node(space, space_id, &block_offset, 0, len);
-		mutex_exit(&fil_system->mutex);
 
 		if (node) {
 			block_size = node->file_block_size;
 		}
 	}
+	mutex_exit(&fil_system->mutex);
 
 	return block_size;
 }