MDEV-15566: System tablespace does not easily key rotate to unencryptedbb-10.1-MDEV-15566

Problem was that key rotation from encrypted to unecrypted was skipped
when encryption is disabled.

fil_crypt_needs_rotation
	If encryption is disabled and there is tablespaces using
	default encryption (e.g. system tablespace) that are still
	encrypted state we need to rotate them from encrypted state
	to unencrypted state.

8 files changed, 204 insertions, 12 deletions

diff --git a/mysql-test/suite/encryption/r/innodb-remove-encryption.result b/mysql-test/suite/encryption/r/innodb-remove-encryption.result
new file mode 100644
index 00000000000..5d21cc83cdb
--- /dev/null
+++ b/mysql-test/suite/encryption/r/innodb-remove-encryption.result
@@ -0,0 +1,45 @@
+set global innodb_file_per_table=OFF;
+set global innodb_file_format='Barracuda';
+call mtr.add_suppression("mysqld: file-key-management-filename is not set");
+call mtr.add_suppression("Plugin 'file_key_management' init function returned error.");
+call mtr.add_suppression("Plugin 'file_key_management' registration as a ENCRYPTION failed.");
+flush tables;
+create table t1(a int not null primary key, b char(200)) engine=innodb;
+insert into t1 values (1, 'secretdata');
+
+# Restart server with encryption
+# Wait until encryption threads have encrypted all tablespaces
+SELECT NAME FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION = 0;
+NAME
+SELECT NAME FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION <> 0;
+NAME
+mysql/innodb_table_stats
+mysql/innodb_index_stats
+./ibdata1
+# Success!
+SELECT * from t1;
+a	b
+1	secretdata
+# Now turn off encryption and wait for threads to decrypt all tablespaces
+SET GLOBAL innodb_encrypt_tables = off;
+SELECT NAME FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION = 0;
+NAME
+mysql/innodb_table_stats
+mysql/innodb_index_stats
+./ibdata1
+SELECT NAME FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION <> 0;
+NAME
+# Success!
+
+# Restart server with no encryption setup, there should be no crashes
+SELECT NAME FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION = 0;
+NAME
+mysql/innodb_table_stats
+mysql/innodb_index_stats
+./ibdata1
+SELECT NAME FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION <> 0;
+NAME
+SELECT * from t1;
+a	b
+1	secretdata
+DROP TABLE t1;
diff --git a/mysql-test/suite/encryption/t/innodb-remove-encryption.test b/mysql-test/suite/encryption/t/innodb-remove-encryption.test
new file mode 100644
index 00000000000..784a5629d2f
--- /dev/null
+++ b/mysql-test/suite/encryption/t/innodb-remove-encryption.test
@@ -0,0 +1,63 @@
+--source include/have_innodb.inc
+# Test uses restart
+--source include/not_embedded.inc
+--source filekeys_plugin.inc
+
+#
+# MDEV-15566: System tablespace does not easily key rotate to unencrypted
+#
+
+--disable_warnings
+set global innodb_file_per_table=OFF;
+set global innodb_file_format='Barracuda';
+--enable_warnings
+
+call mtr.add_suppression("mysqld: file-key-management-filename is not set");
+call mtr.add_suppression("Plugin 'file_key_management' init function returned error.");
+call mtr.add_suppression("Plugin 'file_key_management' registration as a ENCRYPTION failed.");
+flush tables;
+
+create table t1(a int not null primary key, b char(200)) engine=innodb;
+insert into t1 values (1, 'secretdata');
+
+--echo
+--echo # Restart server with encryption
+-- let $restart_parameters=--plugin-load-add=$FILE_KEY_MANAGEMENT_SO --loose-file-key-management --loose-file-key-management-filename=$MYSQL_TEST_DIR/std_data/keys.txt --file-key-management-encryption-algorithm=aes_cbc --innodb-encrypt-tables=ON --innodb-encryption-threads=4 --innodb-tablespaces-encryption --innodb-encryption-rotate-key-age=15
+-- source include/restart_mysqld.inc
+
+--echo # Wait until encryption threads have encrypted all tablespaces
+
+--let $tables_count= `select count(*) from information_schema.tables where engine = 'InnoDB'`
+--let $wait_timeout= 600
+--let $wait_condition=SELECT COUNT(*) = $tables_count FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION <> 0 AND ROTATING_OR_FLUSHING = 0;
+--source include/wait_condition.inc
+
+SELECT NAME FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION = 0;
+SELECT NAME FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION <> 0;
+
+--echo # Success!
+
+SELECT * from t1;
+
+--echo # Now turn off encryption and wait for threads to decrypt all tablespaces
+SET GLOBAL innodb_encrypt_tables = off;
+
+--let $wait_condition=SELECT COUNT(*) = $tables_count FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION = 0 AND ROTATING_OR_FLUSHING = 0;
+--source include/wait_condition.inc
+
+SELECT NAME FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION = 0;
+SELECT NAME FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION <> 0;
+
+--echo # Success!
+
+--echo
+--echo # Restart server with no encryption setup, there should be no crashes
+--let $restart_parameters=--skip-file-key-management --innodb-encrypt-tables=OFF --innodb-encryption-threads=0 --innodb-tablespaces-encryption
+-- source include/restart_mysqld.inc
+
+SELECT NAME FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION = 0;
+SELECT NAME FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION <> 0;
+
+SELECT * from t1;
+DROP TABLE t1;
+
diff --git a/storage/innobase/fil/fil0crypt.cc b/storage/innobase/fil/fil0crypt.cc
index 3095503cfc5..299ccafb589 100644
--- a/storage/innobase/fil/fil0crypt.cc
+++ b/storage/innobase/fil/fil0crypt.cc
@@ -1,6 +1,6 @@
 /*****************************************************************************
 Copyright (C) 2013, 2015, Google Inc. All Rights Reserved.
-Copyright (c) 2014, 2017, MariaDB Corporation. All Rights Reserved.
+Copyright (c) 2014, 2018, MariaDB Corporation. All Rights Reserved.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -114,6 +114,9 @@ extern my_bool srv_background_scrub_data_compressed;
 
 /***********************************************************************
 Check if a key needs rotation given a key_state
+@param[in]	type			Encryption type
+					CRYPT_SCHEME_UNENCRYPTED or
+					CRYPT_SCHEME_1
 @param[in]	encrypt_mode		Encryption mode
 @param[in]	key_version		Current key version
 @param[in]	latest_key_version	Latest key version
@@ -121,6 +124,7 @@ Check if a key needs rotation given a key_state
 @return true if key needs rotation, false if not */
 static bool
 fil_crypt_needs_rotation(
+	uint			type,
 	fil_encryption_t	encrypt_mode,
 	uint			key_version,
 	uint			latest_key_version,
@@ -187,7 +191,9 @@ fil_crypt_get_latest_key_version(
 
 	if (crypt_data->is_key_found()) {
 
-		if (fil_crypt_needs_rotation(crypt_data->encryption,
+		if (fil_crypt_needs_rotation(
+				crypt_data->type,
+				crypt_data->encryption,
 				crypt_data->min_key_version,
 				key_version,
 				srv_fil_crypt_rotate_key_age)) {
@@ -963,6 +969,9 @@ fil_crypt_get_key_state(
 
 /***********************************************************************
 Check if a key needs rotation given a key_state
+@param[in]	type			Encryption type
+					CRYPT_SCHEME_UNENCRYPTED or
+					CRYPT_SCHEME_1
 @param[in]	encrypt_mode		Encryption mode
 @param[in]	key_version		Current key version
 @param[in]	latest_key_version	Latest key version
@@ -970,6 +979,7 @@ Check if a key needs rotation given a key_state
 @return true if key needs rotation, false if not */
 static bool
 fil_crypt_needs_rotation(
+	uint			type,
 	fil_encryption_t	encrypt_mode,
 	uint			key_version,
 	uint			latest_key_version,
@@ -993,6 +1003,13 @@ fil_crypt_needs_rotation(
 		return false;
 	}
 
+	if (srv_encrypt_tables == 0 &&
+	    encrypt_mode == FIL_ENCRYPTION_DEFAULT &&
+	    type == CRYPT_SCHEME_1) {
+		/* This is rotation encrypted => unencrypted */
+		return true;
+	}
+
 	/* this is rotation encrypted => encrypted,
 	* only reencrypt if key is sufficiently old */
 	if (key_version + rotate_key_age < latest_key_version) {
@@ -1272,9 +1289,11 @@ fil_crypt_space_needs_rotation(
 		}
 
 		bool need_key_rotation = fil_crypt_needs_rotation(
+			crypt_data->type,
 			crypt_data->encryption,
 			crypt_data->min_key_version,
-			key_state->key_version, key_state->rotate_key_age);
+			key_state->key_version,
+			key_state->rotate_key_age);
 
 		crypt_data->rotate_state.scrubbing.is_active =
 			btr_scrub_start_space(space->id, &state->scrub_data);
@@ -1862,9 +1881,11 @@ fil_crypt_rotate_page(
 			ut_ad(kv == 0);
 			ut_ad(page_get_space_id(frame) == 0);
 		} else if (fil_crypt_needs_rotation(
-				   crypt_data->encryption,
-				   kv, key_state->key_version,
-				   key_state->rotate_key_age)) {
+				crypt_data->type,
+				crypt_data->encryption,
+				kv,
+				key_state->key_version,
+				key_state->rotate_key_age)) {
 
 			modified = true;
 
diff --git a/storage/innobase/fil/fil0fil.cc b/storage/innobase/fil/fil0fil.cc
index 0cdbee09548..9983f3b64da 100644
--- a/storage/innobase/fil/fil0fil.cc
+++ b/storage/innobase/fil/fil0fil.cc
@@ -7054,6 +7054,19 @@ fil_space_acquire_for_io(ulint id)
 	return(space);
 }
 
+/** Prepare a tablespace for reading or writing a block,
+when it could be dropped concurrently.
+@param[in]	space	tablespace
+*/
+UNIV_INTERN
+void
+fil_space_prepare_for_io(fil_space_t* space)
+{
+	mutex_enter(&fil_system->mutex);
+	space->n_pending_ios++;
+	mutex_exit(&fil_system->mutex);
+}
+
 /** Release a tablespace acquired with fil_space_acquire_for_io().
 @param[in,out]	space	tablespace to release  */
 UNIV_INTERN
diff --git a/storage/innobase/include/fil0fil.h b/storage/innobase/include/fil0fil.h
index 312b09e1f2d..b2c7ff21377 100644
--- a/storage/innobase/include/fil0fil.h
+++ b/storage/innobase/include/fil0fil.h
@@ -700,6 +700,14 @@ UNIV_INTERN
 fil_space_t*
 fil_space_acquire_for_io(ulint id);
 
+/** Prepare a tablespace for reading or writing a block,
+when it could be dropped concurrently.
+@param[in]	space	tablespace
+*/
+UNIV_INTERN
+void
+fil_space_prepare_for_io(fil_space_t* space);
+
 /** Release a tablespace acquired with fil_space_acquire_for_io().
 @param[in,out]	space	tablespace to release  */
 UNIV_INTERN
diff --git a/storage/xtradb/fil/fil0crypt.cc b/storage/xtradb/fil/fil0crypt.cc
index 3095503cfc5..299ccafb589 100644
--- a/storage/xtradb/fil/fil0crypt.cc
+++ b/storage/xtradb/fil/fil0crypt.cc
@@ -1,6 +1,6 @@
 /*****************************************************************************
 Copyright (C) 2013, 2015, Google Inc. All Rights Reserved.
-Copyright (c) 2014, 2017, MariaDB Corporation. All Rights Reserved.
+Copyright (c) 2014, 2018, MariaDB Corporation. All Rights Reserved.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -114,6 +114,9 @@ extern my_bool srv_background_scrub_data_compressed;
 
 /***********************************************************************
 Check if a key needs rotation given a key_state
+@param[in]	type			Encryption type
+					CRYPT_SCHEME_UNENCRYPTED or
+					CRYPT_SCHEME_1
 @param[in]	encrypt_mode		Encryption mode
 @param[in]	key_version		Current key version
 @param[in]	latest_key_version	Latest key version
@@ -121,6 +124,7 @@ Check if a key needs rotation given a key_state
 @return true if key needs rotation, false if not */
 static bool
 fil_crypt_needs_rotation(
+	uint			type,
 	fil_encryption_t	encrypt_mode,
 	uint			key_version,
 	uint			latest_key_version,
@@ -187,7 +191,9 @@ fil_crypt_get_latest_key_version(
 
 	if (crypt_data->is_key_found()) {
 
-		if (fil_crypt_needs_rotation(crypt_data->encryption,
+		if (fil_crypt_needs_rotation(
+				crypt_data->type,
+				crypt_data->encryption,
 				crypt_data->min_key_version,
 				key_version,
 				srv_fil_crypt_rotate_key_age)) {
@@ -963,6 +969,9 @@ fil_crypt_get_key_state(
 
 /***********************************************************************
 Check if a key needs rotation given a key_state
+@param[in]	type			Encryption type
+					CRYPT_SCHEME_UNENCRYPTED or
+					CRYPT_SCHEME_1
 @param[in]	encrypt_mode		Encryption mode
 @param[in]	key_version		Current key version
 @param[in]	latest_key_version	Latest key version
@@ -970,6 +979,7 @@ Check if a key needs rotation given a key_state
 @return true if key needs rotation, false if not */
 static bool
 fil_crypt_needs_rotation(
+	uint			type,
 	fil_encryption_t	encrypt_mode,
 	uint			key_version,
 	uint			latest_key_version,
@@ -993,6 +1003,13 @@ fil_crypt_needs_rotation(
 		return false;
 	}
 
+	if (srv_encrypt_tables == 0 &&
+	    encrypt_mode == FIL_ENCRYPTION_DEFAULT &&
+	    type == CRYPT_SCHEME_1) {
+		/* This is rotation encrypted => unencrypted */
+		return true;
+	}
+
 	/* this is rotation encrypted => encrypted,
 	* only reencrypt if key is sufficiently old */
 	if (key_version + rotate_key_age < latest_key_version) {
@@ -1272,9 +1289,11 @@ fil_crypt_space_needs_rotation(
 		}
 
 		bool need_key_rotation = fil_crypt_needs_rotation(
+			crypt_data->type,
 			crypt_data->encryption,
 			crypt_data->min_key_version,
-			key_state->key_version, key_state->rotate_key_age);
+			key_state->key_version,
+			key_state->rotate_key_age);
 
 		crypt_data->rotate_state.scrubbing.is_active =
 			btr_scrub_start_space(space->id, &state->scrub_data);
@@ -1862,9 +1881,11 @@ fil_crypt_rotate_page(
 			ut_ad(kv == 0);
 			ut_ad(page_get_space_id(frame) == 0);
 		} else if (fil_crypt_needs_rotation(
-				   crypt_data->encryption,
-				   kv, key_state->key_version,
-				   key_state->rotate_key_age)) {
+				crypt_data->type,
+				crypt_data->encryption,
+				kv,
+				key_state->key_version,
+				key_state->rotate_key_age)) {
 
 			modified = true;
 
diff --git a/storage/xtradb/fil/fil0fil.cc b/storage/xtradb/fil/fil0fil.cc
index e3a5a351edf..ce3a38eb9ce 100644
--- a/storage/xtradb/fil/fil0fil.cc
+++ b/storage/xtradb/fil/fil0fil.cc
@@ -7434,6 +7434,19 @@ fil_space_acquire_for_io(ulint id)
 	return(space);
 }
 
+/** Prepare a tablespace for reading or writing a block,
+when it could be dropped concurrently.
+@param[in]	space	tablespace
+*/
+UNIV_INTERN
+void
+fil_space_prepare_for_io(fil_space_t* space)
+{
+	mutex_enter(&fil_system->mutex);
+	space->n_pending_ios++;
+	mutex_exit(&fil_system->mutex);
+}
+
 /** Release a tablespace acquired with fil_space_acquire_for_io().
 @param[in,out]	space	tablespace to release  */
 UNIV_INTERN
diff --git a/storage/xtradb/include/fil0fil.h b/storage/xtradb/include/fil0fil.h
index 8c3bf7d2b06..8f89d281d47 100644
--- a/storage/xtradb/include/fil0fil.h
+++ b/storage/xtradb/include/fil0fil.h
@@ -706,6 +706,14 @@ UNIV_INTERN
 fil_space_t*
 fil_space_acquire_for_io(ulint id);
 
+/** Prepare a tablespace for reading or writing a block,
+when it could be dropped concurrently.
+@param[in]	space	tablespace
+*/
+UNIV_INTERN
+void
+fil_space_prepare_for_io(fil_space_t* space);
+
 /** Release a tablespace acquired with fil_space_acquire_for_io().
 @param[in,out]	space	tablespace to release  */
 UNIV_INTERN