MDEV-14398 When innodb_encryption_rotate_key_age=0 is set, server won't encrypt tablespaces

- Add the encrypted and unencrypted list in fil_system_t to track the
encrypted and unencrypted tablespace.

- Remove the space from encrypted and unencrypted list while
freeing the tablespace.

- Maintain the global encryption status of all tablespaces in
fil_system.crypt_status. There are 3 possible states for
encryption status:
i)   ALL ENCRYPTED
ii)  ALL DECRYPTED
iii) MIXED STATE

- Use the set global innodb_encrypt_tables statement to traverse all spaces
in the space list to encrypt/decrypt when innodb_encryption_rotate_key_age
is zero.

- If innodb_encryption_rotate_key_age is 0 then it avoids the encryption
of already encrypted tablespace.

8 files changed, 526 insertions, 84 deletions

diff --git a/mysql-test/suite/encryption/r/innodb_encrypt_key_rotation_age.result b/mysql-test/suite/encryption/r/innodb_encrypt_key_rotation_age.result
new file mode 100644
index 00000000000..0bcdce7b72c
--- /dev/null
+++ b/mysql-test/suite/encryption/r/innodb_encrypt_key_rotation_age.result
@@ -0,0 +1,96 @@
+CREATE TABLE t1 (f1 INT, f2 VARCHAR(256))engine=innodb;
+INSERT INTO t1 VALUES(1, 'MariaDB'), (2, 'Robot'), (3, 'Science');
+INSERT INTO t1 SELECT * FROM t1;
+CREATE TABLE t2(f1 INT, f2 VARCHAR(256))engine=innodb;
+INSERT INTO t2 SELECT * FROM t1;
+# Restart the 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
+innodb_system
+mysql/innodb_index_stats
+mysql/innodb_table_stats
+mysql/transaction_registry
+test/t1
+test/t2
+# Restart the server with innodb_encryption_rotate_key_age= 0
+create table t3 (f1 int not null)engine=innodb;
+select count(*) from mysql.transaction_registry;
+count(*)
+0
+# 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
+innodb_system
+mysql/innodb_index_stats
+mysql/innodb_table_stats
+mysql/transaction_registry
+test/t1
+test/t2
+test/t3
+# Restart the server with innodb_encryption_rotate_key_age=0
+# and disable encryption
+# Should display all the tables as encrypted
+SELECT NAME FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION <> 0;
+NAME
+innodb_system
+mysql/innodb_index_stats
+mysql/innodb_table_stats
+mysql/transaction_registry
+test/t1
+test/t2
+test/t3
+set global innodb_encrypt_tables = OFF;
+# Wait until encryption threads to decrypt all tablespaces
+SELECT NAME FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION = 0;
+NAME
+innodb_system
+mysql/innodb_index_stats
+mysql/innodb_table_stats
+mysql/transaction_registry
+test/t1
+test/t2
+test/t3
+SELECT NAME FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION <> 0;
+NAME
+# Restart the server with innodb_encryption_rotate_key_age=0
+# and enable encryption
+# Should display all the tables as non-encrypted
+SELECT NAME FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION = 0;
+NAME
+innodb_system
+mysql/innodb_index_stats
+mysql/innodb_table_stats
+mysql/transaction_registry
+test/t1
+test/t2
+test/t3
+set global innodb_encrypt_tables = ON;
+# Wait until encryption threads to decrypt 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
+innodb_system
+mysql/innodb_index_stats
+mysql/innodb_table_stats
+mysql/transaction_registry
+test/t1
+test/t2
+test/t3
+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
+innodb_system
+mysql/innodb_index_stats
+mysql/innodb_table_stats
+mysql/transaction_registry
+test/t1
+test/t2
+test/t3
+DROP TABLE t3, t2, t1;
diff --git a/mysql-test/suite/encryption/t/innodb_encrypt_key_rotation_age.opt b/mysql-test/suite/encryption/t/innodb_encrypt_key_rotation_age.opt
new file mode 100644
index 00000000000..6fa06402377
--- /dev/null
+++ b/mysql-test/suite/encryption/t/innodb_encrypt_key_rotation_age.opt
@@ -0,0 +1,2 @@
+--innodb-tablespaces-encryption
+--innodb_encrypt_tables=ON
diff --git a/mysql-test/suite/encryption/t/innodb_encrypt_key_rotation_age.test b/mysql-test/suite/encryption/t/innodb_encrypt_key_rotation_age.test
new file mode 100644
index 00000000000..a4958cfc9a4
--- /dev/null
+++ b/mysql-test/suite/encryption/t/innodb_encrypt_key_rotation_age.test
@@ -0,0 +1,97 @@
+-- source include/have_innodb.inc
+-- source include/not_embedded.inc
+-- source include/have_example_key_management_plugin.inc
+
+CREATE TABLE t1 (f1 INT, f2 VARCHAR(256))engine=innodb;
+INSERT INTO t1 VALUES(1, 'MariaDB'), (2, 'Robot'), (3, 'Science');
+INSERT INTO t1 SELECT * FROM t1;
+
+CREATE TABLE t2(f1 INT, f2 VARCHAR(256))engine=innodb;
+INSERT INTO t2 SELECT * FROM t1;
+
+--echo # Restart the server with encryption
+
+let $restart_parameters= --innodb_encryption_threads=5 --innodb_encryption_rotate_key_age=16384;
+--source include/restart_mysqld.inc
+
+--echo # Wait until encryption threads have encrypted all tablespaces
+
+--let $tables_count= `select count(*) + 1 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;
+--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 # Restart the server with innodb_encryption_rotate_key_age= 0
+
+let $restart_parameters= --innodb_encryption_threads=1 --innodb_encryption_rotate_key_age=0;
+
+--source include/restart_mysqld.inc
+
+create table t3 (f1 int not null)engine=innodb;
+
+select count(*) from mysql.transaction_registry;
+
+--echo # Wait until encryption threads have encrypted all tablespaces
+
+--let $tables_count= `select count(*) + 1 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;
+--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 # Restart the server with innodb_encryption_rotate_key_age=0
+--echo # and disable encryption
+
+let $restart_parameters= --innodb_encryption_threads=1 --innodb_encryption_rotate_key_age=0 --innodb_encrypt_tables=OFF;
+
+--source include/restart_mysqld.inc
+
+--echo # Should display all the tables as encrypted
+SELECT NAME FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION <> 0;
+
+set global innodb_encrypt_tables = OFF;
+
+--echo # Wait until encryption threads to decrypt all tablespaces
+
+--let $tables_count= `select count(*) + 1 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 # Restart the server with innodb_encryption_rotate_key_age=0
+--echo # and enable encryption
+
+let $restart_parameters= --innodb_encryption_threads=1 --innodb_encryption_rotate_key_age=0 --innodb_encrypt_tables=ON;
+
+--source include/restart_mysqld.inc
+
+--echo # Should display all the tables as non-encrypted
+SELECT NAME FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION = 0;
+
+set global innodb_encrypt_tables = ON;
+
+--echo # Wait until encryption threads to decrypt all tablespaces
+
+--let $tables_count= `select count(*) + 1 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;
+--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;
+
+--let $restart_parameters=
+-- 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;
+DROP TABLE t3, t2, t1;
diff --git a/storage/innobase/fil/fil0crypt.cc b/storage/innobase/fil/fil0crypt.cc
index 7a6d21dba13..7274df2d7d9 100644
--- a/storage/innobase/fil/fil0crypt.cc
+++ b/storage/innobase/fil/fil0crypt.cc
@@ -417,45 +417,6 @@ fil_space_crypt_t::write_page0(
 }
 
 /******************************************************************
-Set crypt data for a tablespace
-@param[in,out]		space		Tablespace
-@param[in,out]		crypt_data	Crypt data to be set
-@return crypt_data in tablespace */
-static
-fil_space_crypt_t*
-fil_space_set_crypt_data(
-	fil_space_t*		space,
-	fil_space_crypt_t*	crypt_data)
-{
-	fil_space_crypt_t* free_crypt_data = NULL;
-	fil_space_crypt_t* ret_crypt_data = NULL;
-
-	/* Provided space is protected using fil_space_acquire()
-	from concurrent operations. */
-	if (space->crypt_data != NULL) {
-		/* There is already crypt data present,
-		merge new crypt_data */
-		fil_space_merge_crypt_data(space->crypt_data,
-						   crypt_data);
-		ret_crypt_data = space->crypt_data;
-		free_crypt_data = crypt_data;
-	} else {
-		space->crypt_data = crypt_data;
-		ret_crypt_data = space->crypt_data;
-	}
-
-	if (free_crypt_data != NULL) {
-		/* there was already crypt data present and the new crypt
-		* data provided as argument to this function has been merged
-		* into that => free new crypt data
-		*/
-		fil_space_destroy_crypt_data(&free_crypt_data);
-	}
-
-	return ret_crypt_data;
-}
-
-/******************************************************************
 Parse a MLOG_FILE_WRITE_CRYPT_DATA log entry
 @param[in]	ptr		Log entry start
 @param[in]	end_ptr		Log entry end
@@ -512,7 +473,18 @@ fil_parse_write_crypt_data(
 		return NULL;
 	}
 
-	fil_space_crypt_t* crypt_data = fil_space_create_crypt_data(encryption, key_id);
+	mutex_enter(&fil_system.mutex);
+
+	/* Update fil_space memory cache with crypt_data */
+	fil_space_t* space = fil_space_get_by_id(space_id);
+
+	if (!space) {
+		mutex_exit(&fil_system.mutex);
+		return ptr + len;
+	}
+
+	fil_space_crypt_t* crypt_data = fil_space_create_crypt_data(
+						encryption, key_id);
 	/* Need to overwrite these as above will initialize fields. */
 	crypt_data->page0_offset = offset;
 	crypt_data->min_key_version = min_key_version;
@@ -521,17 +493,20 @@ fil_parse_write_crypt_data(
 	memcpy(crypt_data->iv, ptr, len);
 	ptr += len;
 
-	/* update fil_space memory cache with crypt_data */
-	if (fil_space_t* space = fil_space_acquire_silent(space_id)) {
-		crypt_data = fil_space_set_crypt_data(space, crypt_data);
-		space->release();
-		/* Check is used key found from encryption plugin */
-		if (crypt_data->should_encrypt()
-		    && !crypt_data->is_key_found()) {
-			*err = DB_DECRYPTION_FAILED;
-		}
-	} else {
+	if (space->crypt_data) {
+		fil_space_merge_crypt_data(space->crypt_data, crypt_data);
 		fil_space_destroy_crypt_data(&crypt_data);
+		crypt_data = space->crypt_data;
+	} else {
+		space->crypt_data = crypt_data;
+		space->crypt_enlist();
+	}
+
+	mutex_exit(&fil_system.mutex);
+
+	/* Check is used key found from encryption plugin */
+	if (crypt_data->should_encrypt() && !crypt_data->is_key_found()) {
+		*err = DB_DECRYPTION_FAILED;
 	}
 
 	return ptr;
@@ -1138,7 +1113,7 @@ fil_crypt_needs_rotation(
 
 	if (crypt_data->encryption == FIL_ENCRYPTION_DEFAULT
 	    && crypt_data->type == CRYPT_SCHEME_1
-	    && srv_encrypt_tables == 0 ) {
+	    && srv_encrypt_tables == 0) {
 		/* This is rotation encrypted => unencrypted */
 		return true;
 	}
@@ -1149,6 +1124,10 @@ fil_crypt_needs_rotation(
 		return true;
 	}
 
+	if (rotate_key_age == 0) {
+		return false;
+	}
+
 	return false;
 }
 
@@ -1225,7 +1204,8 @@ fil_crypt_start_encrypting_space(
 	* crypt data in page 0 */
 
 	/* 1 - create crypt data */
-	crypt_data = fil_space_create_crypt_data(FIL_ENCRYPTION_DEFAULT, FIL_DEFAULT_ENCRYPTION_KEY);
+	crypt_data = fil_space_create_crypt_data(
+			FIL_ENCRYPTION_DEFAULT, FIL_DEFAULT_ENCRYPTION_KEY);
 
 	if (crypt_data == NULL) {
 		mutex_exit(&fil_crypt_threads_mutex);
@@ -1238,9 +1218,10 @@ fil_crypt_start_encrypting_space(
 	crypt_data->rotate_state.starting = true;
 	crypt_data->rotate_state.active_threads = 1;
 
-	mutex_enter(&crypt_data->mutex);
-	crypt_data = fil_space_set_crypt_data(space, crypt_data);
-	mutex_exit(&crypt_data->mutex);
+	mutex_enter(&fil_system.mutex);
+	space->crypt_data = crypt_data;
+	space->crypt_enlist();
+	mutex_exit(&fil_system.mutex);
 
 	fil_crypt_start_converting = true;
 	mutex_exit(&fil_crypt_threads_mutex);
@@ -1690,6 +1671,9 @@ fil_crypt_find_space_to_rotate(
 	added to keyrotation list. */
 	if (srv_fil_crypt_rotate_key_age) {
 		state->space = fil_space_next(state->space);
+	} else if (fil_system.is_stable_crypt_status(srv_encrypt_tables)) {
+		fil_crypt_return_iops(state);
+		return false;
 	} else {
 		state->space = fil_space_keyrotate_next(state->space);
 	}
@@ -1712,6 +1696,10 @@ fil_crypt_find_space_to_rotate(
 
 		if (srv_fil_crypt_rotate_key_age) {
 			state->space = fil_space_next(state->space);
+		} else if (fil_system.is_stable_crypt_status(
+					srv_encrypt_tables)) {
+			fil_crypt_return_iops(state);
+			return false;
 		} else {
 			state->space = fil_space_keyrotate_next(state->space);
 		}
@@ -2248,6 +2236,13 @@ fil_crypt_flush_space(
 	}
 
 	mtr.commit();
+
+	if (crypt_data != NULL) {
+		mutex_enter(&fil_system.mutex);
+		space->crypt_enlist();
+		mutex_exit(&fil_system.mutex);
+	}
+
 }
 
 /***********************************************************************
@@ -2539,6 +2534,23 @@ fil_crypt_set_encrypt_tables(
 {
 	srv_encrypt_tables = val;
 	os_event_set(fil_crypt_threads_event);
+
+	if (srv_fil_crypt_rotate_key_age == 0) {
+		mutex_enter(&fil_system.mutex);
+
+		for (fil_space_t* space = UT_LIST_GET_FIRST(fil_system.space_list);
+		     space != NULL;
+		     space = UT_LIST_GET_NEXT(space_list, space)) {
+			if (space->purpose != FIL_TYPE_TABLESPACE
+			    || space->is_in_rotation_list()) {
+				continue;
+			}
+
+			UT_LIST_ADD_LAST(fil_system.rotation_list, space);
+		}
+
+		mutex_exit(&fil_system.mutex);
+	}
 }
 
 /*********************************************************************
diff --git a/storage/innobase/fil/fil0fil.cc b/storage/innobase/fil/fil0fil.cc
index 639c8dcb23f..65661334d6b 100644
--- a/storage/innobase/fil/fil0fil.cc
+++ b/storage/innobase/fil/fil0fil.cc
@@ -572,6 +572,11 @@ invalid:
 		space->crypt_data = fil_space_read_crypt_data(
 			fil_space_t::zip_size(flags), page);
 	}
+
+	if (first) {
+		space->crypt_enlist();
+	}
+
 	ut_free(buf2);
 
 	if (UNIV_UNLIKELY(space_id != space->id)) {
@@ -1210,6 +1215,11 @@ fil_space_detach(
 {
 	ut_ad(mutex_own(&fil_system.mutex));
 
+	if (space->purpose == FIL_TYPE_TABLESPACE
+	    || space->purpose == FIL_TYPE_IMPORT) {
+		space->crypt_delist();
+	}
+
 	HASH_DELETE(fil_space_t, hash, fil_system.spaces, space->id, space);
 
 	if (space->is_in_unflushed_spaces()) {
@@ -1431,6 +1441,10 @@ fil_space_create(
 		fil_system.max_assigned_id = id;
 	}
 
+	if (crypt_data) {
+		space->crypt_enlist();
+	}
+
 	/* Inform key rotation that there could be something
 	to do */
 	if (purpose == FIL_TYPE_TABLESPACE
@@ -3162,6 +3176,12 @@ err_exit:
 		file->block_size = block_size;
 		space->punch_hole = punch_hole;
 
+		if (space->purpose == FIL_TYPE_TABLESPACE && !crypt_data) {
+			mutex_enter(&fil_system.mutex);
+			space->crypt_enlist();
+			mutex_exit(&fil_system.mutex);
+		}
+
 		*err = DB_SUCCESS;
 	}
 
@@ -5226,8 +5246,7 @@ fil_node_should_punch_hole(
 	return (node->space->punch_hole);
 }
 
-/**
-Set punch hole to tablespace to given value.
+/** Set punch hole to tablespace to given value.
 @param[in]	node		File node
 @param[in]	val		value to be set. */
 void
@@ -5238,20 +5257,108 @@ fil_space_set_punch_hole(
 	node->space->punch_hole = val;
 }
 
-/** Checks that this tablespace in a list of unflushed tablespaces.
-@return true if in a list */
-bool fil_space_t::is_in_unflushed_spaces() const {
-	ut_ad(mutex_own(&fil_system.mutex));
+/** Set the encryption status of all tablespaces after
+unencrypted spaces or encrypted spaces has been changed.
+@param[in]	encrypted	whether the tablespace is encrypted
+@param[in]	remove		whether the tablespace is being removed. */
+inline void fil_system_t::crypt_update(bool encrypted, bool remove)
+{
+	ut_ad(is_initialised());
+	ut_ad(this == &fil_system);
+
+	if (srv_read_only_mode) {
+		return;
+	}
+
+	ut_ad(mutex_own(&mutex));
+	ut_ad(srv_operation == SRV_OPERATION_NORMAL);
 
-	return fil_system.unflushed_spaces.start == this
-	       || unflushed_spaces.next || unflushed_spaces.prev;
+	crypt_status_t status = crypt_status;
+
+	switch (status) {
+	case CRYPT_ENCRYPTED:
+		if (!remove && !encrypted) {
+			status = CRYPT_MIXED;
+		}
+		break;
+	case CRYPT_DECRYPTED:
+		if (!remove && encrypted) {
+			status = CRYPT_MIXED;
+		}
+		break;
+	case CRYPT_MIXED:
+		ulint n_encrypted = UT_LIST_GET_LEN(encrypted_spaces);
+		ulint n_unencrypted = UT_LIST_GET_LEN(unencrypted_spaces);
+		/* Space list include redo log and temp spaces,
+		which are not part of key rotation. */
+		DBUG_ASSERT(UT_LIST_GET_LEN(space_list) >= 2);
+		ulint n_total = UT_LIST_GET_LEN(space_list)
+					- (1 + !!(temp_space));
+		DBUG_ASSERT(n_total + !temp_space
+			    >= n_encrypted + n_unencrypted);
+
+		if (n_total == n_encrypted) {
+			status = CRYPT_ENCRYPTED;
+		} else if (n_total == n_unencrypted) {
+			status = CRYPT_DECRYPTED;
+		}
+	}
+
+	crypt_status = status;
 }
 
-/** Checks that this tablespace needs key rotation.
-@return true if in a rotation list */
-bool fil_space_t::is_in_rotation_list() const {
+/** Add space to encrypted or unencrypted list. */
+void fil_space_t::crypt_enlist()
+{
 	ut_ad(mutex_own(&fil_system.mutex));
 
-	return fil_system.rotation_list.start == this || rotation_list.next
-	       || rotation_list.prev;
+	if (srv_operation != SRV_OPERATION_NORMAL) {
+		return;
+	}
+
+	if (!crypt_data || !crypt_data->min_key_version) {
+		if (add_if_not_in_unencrypted_spaces()) {
+			remove_if_in_encrypted_spaces();
+			fil_system.crypt_update(false, false);
+		}
+	} else {
+		if (add_if_not_in_encrypted_spaces()) {
+			remove_if_in_unencrypted_spaces();
+			fil_system.crypt_update(true, false);
+		}
+	}
+}
+
+/** Remove the space from encrypted or unencrypted list. */
+inline void fil_space_t::crypt_delist()
+{
+	if (srv_operation != SRV_OPERATION_NORMAL) {
+		return;
+	}
+
+	if (remove_if_in_encrypted_spaces()) {
+		ut_ad(!is_in_unencrypted_spaces());
+
+		if (srv_shutdown_state == SRV_SHUTDOWN_LAST_PHASE) {
+			return;
+		}
+
+		ut_ad(fil_system.crypt_status
+				!= fil_system_t::CRYPT_DECRYPTED);
+		return fil_system.crypt_update(true, true);
+	}
+
+	if (remove_if_in_unencrypted_spaces()) {
+		ut_ad(!is_in_unencrypted_spaces());
+
+		if (srv_shutdown_state == SRV_SHUTDOWN_LAST_PHASE) {
+			return;
+		}
+
+		ut_ad(fil_system.crypt_status
+				!= fil_system_t::CRYPT_ENCRYPTED);
+		return fil_system.crypt_update(false, true);
+	}
+
+	ut_ad(size == 0);
 }
diff --git a/storage/innobase/handler/ha_innodb.cc b/storage/innobase/handler/ha_innodb.cc
index 9e6d1a35536..a9ca572a6c9 100644
--- a/storage/innobase/handler/ha_innodb.cc
+++ b/storage/innobase/handler/ha_innodb.cc
@@ -21322,16 +21322,6 @@ innodb_encrypt_tables_validate(
 		return 1;
 	}
 
-	if (!srv_fil_crypt_rotate_key_age) {
-		const char *msg = (encrypt_tables ? "enable" : "disable");
-		push_warning_printf(thd, Sql_condition::WARN_LEVEL_WARN,
-				    HA_ERR_UNSUPPORTED,
-				    "InnoDB: cannot %s encryption, "
-				    "innodb_encryption_rotate_key_age=0"
-				    " i.e. key rotation disabled", msg);
-		return 1;
-	}
-
 	return 0;
 }
 
diff --git a/storage/innobase/include/fil0fil.h b/storage/innobase/include/fil0fil.h
index bde5a85ebf9..61a9b258edf 100644
--- a/storage/innobase/include/fil0fil.h
+++ b/storage/innobase/include/fil0fil.h
@@ -150,16 +150,16 @@ struct fil_space_t {
 	UT_LIST_NODE_T(fil_space_t) named_spaces;
 				/*!< list of spaces for which MLOG_FILE_NAME
 				records have been issued */
-	/** Checks that this tablespace in a list of unflushed tablespaces.
-	@return true if in a list */
-	bool is_in_unflushed_spaces() const;
 	UT_LIST_NODE_T(fil_space_t) space_list;
 				/*!< list of all spaces */
 	/** other tablespaces needing key rotation */
 	UT_LIST_NODE_T(fil_space_t) rotation_list;
-	/** Checks that this tablespace needs key rotation.
-	@return true if in a rotation list */
-	bool is_in_rotation_list() const;
+
+	/** List of all encrypted spaces. Protected by fil_system.mutex. */
+	UT_LIST_NODE_T(fil_space_t) encrypted_spaces;
+
+	/** List of all unencrypted spaces. Protected by fil_system.mutex. */
+	UT_LIST_NODE_T(fil_space_t) unencrypted_spaces;
 
 	/** MariaDB encryption data */
 	fil_space_crypt_t* crypt_data;
@@ -241,7 +241,34 @@ struct fil_space_t {
 	bool open();
 	/** Close each file. Only invoked on fil_system.temp_space. */
 	void close();
+	/** Add the space to encrypted or unencrypted list.*/
+	void crypt_enlist();
+	/** Remove the space from encrypted or unencrypted list. */
+	void crypt_delist();
+
+	/** @return whether this is in fil_system.rotation list */
+	bool is_in_rotation_list() const;
+	/** @return whether this is in fil_system.unflushed_spaces */
+	inline bool is_in_unflushed_spaces() const;
+	/** @return whether this is in fil_system.encrypted_spaces */
+	inline bool is_in_encrypted_spaces() const;
+	/** @return whether this is in fil_system.unencrypted_spaces */
+	inline bool is_in_unencrypted_spaces() const;
 
+private:
+	/** Remove this from fil_system.encrypted_spaces if listed.
+	@return whether this tablespace was listed and removed */
+	inline bool remove_if_in_encrypted_spaces();
+	/** Remove this from fil_system.unencrypted_spaces if listed.
+	@return whether this tablespace was listed and removed */
+	inline bool remove_if_in_unencrypted_spaces();
+	/** Add this to fil_system.encrypted_spaces if not listed.
+	@return whether we added this tablespace */
+	inline bool add_if_not_in_encrypted_spaces();
+	/** Add this to fil_system.unencrypted_spaces if not listed.
+	@return whether we added this tablespace */
+	inline bool add_if_not_in_unencrypted_spaces();
+public:
 	/** Acquire a tablespace reference. */
 	void acquire() { n_pending_ops++; }
 	/** Release a tablespace reference. */
@@ -837,6 +864,9 @@ struct fil_system_t {
     UT_LIST_INIT(rotation_list, &fil_space_t::rotation_list);
     UT_LIST_INIT(unflushed_spaces, &fil_space_t::unflushed_spaces);
     UT_LIST_INIT(named_spaces, &fil_space_t::named_spaces);
+    UT_LIST_INIT(encrypted_spaces, &fil_space_t::encrypted_spaces);
+    UT_LIST_INIT(unencrypted_spaces, &fil_space_t::unencrypted_spaces);
+    crypt_status = CRYPT_DECRYPTED;
   }
 
   bool is_initialised() const { return m_initialised; }
@@ -851,6 +881,14 @@ struct fil_system_t {
   /** Close the file system interface at shutdown */
   void close();
 
+  /**
+    Set the encryption status of all tablespaces after
+    unencrypted_spaces or encrypted_spaces has been changed.
+
+    @param[in]	encrypted	whether the tablespace is encrypted
+    @param[in]	remove		whether the tablespace is being removed. */
+  inline void crypt_update(bool encrypted, bool remove);
+
 private:
   bool m_initialised;
 public:
@@ -895,16 +933,111 @@ public:
 	UT_LIST_BASE_NODE_T(fil_space_t) rotation_list;
 					/*!< list of all file spaces needing
 					key rotation.*/
+	/** List of all encrypted spaces */
+	UT_LIST_BASE_NODE_T(fil_space_t) encrypted_spaces;
+
+	/** List of all unencrypted spaces */
+	UT_LIST_BASE_NODE_T(fil_space_t) unencrypted_spaces;
 
 	bool		space_id_reuse_warned;
 					/*!< whether fil_space_create()
 					has issued a warning about
 					potential space_id reuse */
+
+	/** Global encryption status. */
+	enum crypt_status_t {
+		/** All tablespaces are in encrypted state */
+		CRYPT_ENCRYPTED = 8,
+		/** All tablespaces are in unencrypted state */
+		CRYPT_DECRYPTED,
+		/** Some are unencrypted, some are encrypted */
+		CRYPT_MIXED
+	} crypt_status;
+
+	/** @return whether crypt_status has reached a stable state */
+	bool is_stable_crypt_status(bool encrypted)
+	{
+		mutex_enter(&mutex);
+		bool stable = crypt_status == (encrypted
+					       ? CRYPT_ENCRYPTED
+					       : CRYPT_DECRYPTED);
+		mutex_exit(&mutex);
+
+		return stable;
+	}
 };
 
 /** The tablespace memory cache. */
 extern fil_system_t	fil_system;
 
+/** @return whether this is in fil_system.unflushed_spaces */
+inline bool fil_space_t::is_in_unflushed_spaces() const
+{
+	ut_ad(mutex_own(&fil_system.mutex));
+	return fil_system.unflushed_spaces.start == this
+		|| unflushed_spaces.next || unflushed_spaces.prev;
+}
+
+/** @return whether this is in fil_system.rotation_list */
+inline bool fil_space_t::is_in_rotation_list() const
+{
+	ut_ad(mutex_own(&fil_system.mutex));
+	return fil_system.rotation_list.start == this
+		|| rotation_list.next || rotation_list.prev;
+}
+
+/** @return whether this is in fil_system.encrypted_spaces */
+inline bool fil_space_t::is_in_encrypted_spaces() const
+{
+	ut_ad(mutex_own(&fil_system.mutex));
+	return fil_system.encrypted_spaces.start == this
+		|| encrypted_spaces.next || encrypted_spaces.prev;
+}
+
+/** @return whether this is in fil_system.unencrypted_spaces */
+inline bool fil_space_t::is_in_unencrypted_spaces() const
+{
+	ut_ad(mutex_own(&fil_system.mutex));
+	return fil_system.unencrypted_spaces.start == this
+		|| unencrypted_spaces.next || unencrypted_spaces.prev;
+}
+
+/** Remove this from fil_system.encrypted_spaces if listed.
+  @return whether this tablespace was listed and removed */
+inline bool fil_space_t::remove_if_in_encrypted_spaces()
+{
+	bool remove = is_in_encrypted_spaces();
+	if (remove) UT_LIST_REMOVE(fil_system.encrypted_spaces, this);
+	return remove;
+}
+
+/** Remove this from fil_system.unencrypted_spaces if listed.
+  @return whether this tablespace was listed and removed */
+inline bool fil_space_t::remove_if_in_unencrypted_spaces()
+{
+	bool remove = is_in_unencrypted_spaces();
+	if (remove) UT_LIST_REMOVE(fil_system.unencrypted_spaces, this);
+	return remove;
+}
+
+/** Add this to fil_system.encrypted_spaces if not listed.
+  @return whether we added this tablespace */
+inline bool fil_space_t::add_if_not_in_encrypted_spaces()
+{
+	bool add = !is_in_encrypted_spaces();
+	if (add) UT_LIST_ADD_LAST(fil_system.encrypted_spaces, this);
+	return add;
+}
+
+/** Remove this from fil_system.unencrypted_spaces if listed.
+  @return whether we added this tablespace */
+inline bool fil_space_t::add_if_not_in_unencrypted_spaces()
+{
+	bool add = !is_in_unencrypted_spaces();
+	if (add) UT_LIST_ADD_LAST(fil_system.unencrypted_spaces, this);
+	return add;
+}
+
 #include "fil0crypt.h"
 
 /** Returns the latch of a file space.
diff --git a/storage/innobase/srv/srv0start.cc b/storage/innobase/srv/srv0start.cc
index e82d486c206..b02385b954a 100644
--- a/storage/innobase/srv/srv0start.cc
+++ b/storage/innobase/srv/srv0start.cc
@@ -678,6 +678,7 @@ static bool srv_undo_tablespace_open(const char* name, ulint space_id,
 	if (create_new_db) {
 		space->size = file->size = ulint(size >> srv_page_size_shift);
 		space->size_in_header = SRV_UNDO_TABLESPACE_SIZE_IN_PAGES;
+		space->crypt_enlist();
 	} else {
 		success = file->read_page0(true);
 		if (!success) {
@@ -1586,6 +1587,10 @@ dberr_t srv_start(bool create_new_db)
 		return(srv_init_abort(err));
 	}
 
+	mutex_enter(&fil_system.mutex);
+	fil_system.sys_space->crypt_enlist();
+	mutex_exit(&fil_system.mutex);
+
 	dirnamelen = strlen(srv_log_group_home_dir);
 	ut_a(dirnamelen < (sizeof logfilename) - 10 - sizeof "ib_logfile");
 	memcpy(logfilename, srv_log_group_home_dir, dirnamelen);