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

- Added new encrypted and unencrypted list in fil_system_t
- Added the space in the list whenever server finds out whether
the tablespace is encrypted or unencrypted.
- Remove the space from list while freeing the tablespace.

7 files changed, 165 insertions, 147 deletions

diff --git a/mysql-test/suite/encryption/r/innodb-key-rotation-disable.result b/mysql-test/suite/encryption/r/innodb-key-rotation-disable.result
index 680db692c17..a33fc347a7d 100644
--- a/mysql-test/suite/encryption/r/innodb-key-rotation-disable.result
+++ b/mysql-test/suite/encryption/r/innodb-key-rotation-disable.result
@@ -1,7 +1,3 @@
-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
 create database enctests;
 use enctests;
 create table t1(a int not null primary key, b char(200)) engine=innodb;
diff --git a/mysql-test/suite/encryption/t/innodb-key-rotation-disable.test b/mysql-test/suite/encryption/t/innodb-key-rotation-disable.test
index 574e0c3becc..0094f33bcf1 100644
--- a/mysql-test/suite/encryption/t/innodb-key-rotation-disable.test
+++ b/mysql-test/suite/encryption/t/innodb-key-rotation-disable.test
@@ -3,9 +3,6 @@
 # not embedded because of restarts
 -- source include/not_embedded.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;
-
 --disable_query_log
 --disable_warnings
 let $encryption = `SELECT @@innodb_encrypt_tables`;
diff --git a/storage/innobase/fil/fil0crypt.cc b/storage/innobase/fil/fil0crypt.cc
index 32fd844ec17..a7625acc4c4 100644
--- a/storage/innobase/fil/fil0crypt.cc
+++ b/storage/innobase/fil/fil0crypt.cc
@@ -444,6 +444,8 @@ fil_space_set_crypt_data(
 	} else {
 		space->crypt_data = crypt_data;
 		ret_crypt_data = space->crypt_data;
+
+		fil_space_add_to_encrypt_or_unencrypt_list(space);
 	}
 
 	if (free_crypt_data != NULL) {
@@ -963,6 +965,8 @@ fil_crypt_read_crypt_data(fil_space_t* space)
 			space->crypt_data = fil_space_read_crypt_data(
 				page_size, block->frame);
 		}
+
+		fil_space_add_to_encrypt_or_unencrypt_list(space);
 		mutex_exit(&fil_system->mutex);
 	}
 	mtr.commit();
@@ -1466,14 +1470,7 @@ fil_crypt_find_space_to_rotate(
 		state->space = NULL;
 	}
 
-	/* If key rotation is enabled (default) we iterate all tablespaces.
-	If key rotation is not enabled we iterate only the tablespaces
-	added to keyrotation list. */
-	if (srv_fil_crypt_rotate_key_age) {
-		state->space = fil_space_next(state->space);
-	} else {
-		state->space = fil_space_keyrotate_next(state->space);
-	}
+	state->space = fil_space_next(state->space);
 
 	while (!state->should_shutdown() && state->space) {
 		/* If there is no crypt data and we have not yet read
@@ -1491,11 +1488,7 @@ fil_crypt_find_space_to_rotate(
 			return true;
 		}
 
-		if (srv_fil_crypt_rotate_key_age) {
-			state->space = fil_space_next(state->space);
-		} else {
-			state->space = fil_space_keyrotate_next(state->space);
-		}
+		state->space = fil_space_next(state->space);
 	}
 
 	/* if we didn't find any space return iops */
@@ -2029,6 +2022,12 @@ fil_crypt_flush_space(
 	}
 
 	mtr.commit();
+
+	if (crypt_data != NULL) {
+		mutex_enter(&fil_system->mutex);
+		fil_space_add_to_encrypt_or_unencrypt_list(space);
+		mutex_exit(&fil_system->mutex);
+	}
 }
 
 /***********************************************************************
diff --git a/storage/innobase/fil/fil0fil.cc b/storage/innobase/fil/fil0fil.cc
index 21a2dddfb9f..d7527b2827b 100644
--- a/storage/innobase/fil/fil0fil.cc
+++ b/storage/innobase/fil/fil0fil.cc
@@ -556,7 +556,17 @@ bool fil_node_t::read_page0(bool first)
 	/* Try to read crypt_data from page 0 if it is not yet read. */
 	if (!space->crypt_data) {
 		space->crypt_data = fil_space_read_crypt_data(page_size, page);
+		fil_space_add_to_encrypt_or_unencrypt_list(space);
 	}
+
+	if (first
+	    || (space->purpose == FIL_TYPE_TABLESPACE
+		&& this == UT_LIST_GET_FIRST(space->chain)
+		&& srv_startup_is_before_trx_rollback_phase
+		&& !undo::Truncate::was_tablespace_truncated(space->id))) {
+		fil_space_add_to_encrypt_or_unencrypt_list(space);
+	}
+
 	ut_free(buf2);
 
 	if (!fsp_flags_is_valid(flags, space->id)) {
@@ -1236,6 +1246,82 @@ fil_node_close_to_free(
 	}
 }
 
+/** Remove the space from encrypted list.
+@param[in]	space	space to be removed from encrypted list. */
+static void fil_space_remove_from_encrypted_list(fil_space_t* space)
+{
+	space->set_encrypted_list(false);
+	UT_LIST_REMOVE(fil_system->encrypted_spaces, space);
+}
+
+/** Remove the space from unencrypted list.
+@param[in]	space	space to be removed from unencrypted list. */
+static void fil_space_remove_from_unencrypted_list(fil_space_t* space)
+{
+	space->set_unencrypted_list(false);
+	UT_LIST_REMOVE(fil_system->unencrypted_spaces, space);
+}
+
+/** Remove the space from encrypted list or unencrypted list.
+@param[in]	space	space to be removed. */
+void fil_space_remove_from_encrypt_or_unencrypt_list(fil_space_t* space)
+{
+	if (space->is_in_encrypted_list()) {
+		fil_space_remove_from_encrypted_list(space);
+	} else if (space->is_in_unencrypted_list()) {
+		fil_space_remove_from_unencrypted_list(space);
+	} else {
+		ut_ad(space->size == 0);
+	}
+}
+
+/** Add the space to the encrypted list. Remove the space from
+unencrypted list if it is unencrypted.
+@param[in]	space	space to be added in encrypted list. */
+static void fil_space_add_encrypted_list(fil_space_t* space)
+{
+	if (space->is_in_encrypted_list()) {
+		return;
+	}
+
+	if (space->is_in_unencrypted_list()) {
+		fil_space_remove_from_unencrypted_list(space);
+	}
+
+	UT_LIST_ADD_LAST(fil_system->encrypted_spaces, space);
+	space->set_encrypted_list(true);
+}
+
+/** Add the space to the unencrypted list. Remove the space from
+encrypted list if it is encrypted.
+@param[in]	space	space to be added in unencrypted list. */
+static void fil_space_add_unencrypted_list(fil_space_t* space)
+{
+	if (space->is_in_unencrypted_list()) {
+		return;
+	}
+
+	if (space->is_in_encrypted_list()) {
+		fil_space_remove_from_encrypted_list(space);
+	}
+
+	UT_LIST_ADD_LAST(fil_system->encrypted_spaces, space);
+	space->set_encrypted_list(true);
+}
+
+/** Add the space to encrypted or unencrypted list. Remove the space
+from the unencrypted or encrypted list if it is present.
+@param[in]	space	space to be added. */
+void fil_space_add_to_encrypt_or_unencrypt_list(fil_space_t* space)
+{
+	if (space->crypt_data == NULL
+	    || space->crypt_data->min_key_version == 0) {
+		fil_space_add_unencrypted_list(space);
+	} else {
+		fil_space_add_encrypted_list(space);
+	}
+}
+
 /** Detach a space object from the tablespace memory cache.
 Closes the files in the chain but does not delete them.
 There must not be any pending i/o's or flushes on the files.
@@ -1247,6 +1333,10 @@ fil_space_detach(
 {
 	ut_ad(mutex_own(&fil_system->mutex));
 
+	if (space->purpose == FIL_TYPE_TABLESPACE) {
+		fil_space_remove_from_encrypt_or_unencrypt_list(space);
+	}
+
 	HASH_DELETE(fil_space_t, hash, fil_system->spaces, space->id, space);
 
 	fil_space_t*	fnamespace = fil_space_get_by_name(space->name);
@@ -1264,12 +1354,6 @@ fil_space_detach(
 		UT_LIST_REMOVE(fil_system->unflushed_spaces, space);
 	}
 
-	if (space->is_in_rotation_list) {
-		space->is_in_rotation_list = false;
-
-		UT_LIST_REMOVE(fil_system->rotation_list, space);
-	}
-
 	UT_LIST_REMOVE(fil_system->space_list, space);
 
 	ut_a(space->magic_n == FIL_SPACE_MAGIC_N);
@@ -1465,6 +1549,10 @@ fil_space_create(
 
 	rw_lock_create(fil_space_latch_key, &space->latch, SYNC_FSP);
 
+	if (crypt_data != NULL) {
+		fil_space_add_to_encrypt_or_unencrypt_list(space);
+	}
+
 	if (space->purpose == FIL_TYPE_TEMPORARY) {
 		ut_d(space->latch.set_temp_fsp());
 		/* SysTablespace::open_or_create() would pass
@@ -1488,23 +1576,7 @@ fil_space_create(
 		fil_system->max_assigned_id = id;
 	}
 
-	/* Inform key rotation that there could be something
-	to do */
-	if (purpose == FIL_TYPE_TABLESPACE
-	    && !srv_fil_crypt_rotate_key_age && fil_crypt_threads_event &&
-	    (mode == FIL_ENCRYPTION_ON || mode == FIL_ENCRYPTION_OFF ||
-		    srv_encrypt_tables)) {
-		/* Key rotation is not enabled, need to inform background
-		encryption threads. */
-		UT_LIST_ADD_LAST(fil_system->rotation_list, space);
-		space->is_in_rotation_list = true;
-		mutex_exit(&fil_system->mutex);
-		mutex_enter(&fil_crypt_threads_mutex);
-		os_event_set(fil_crypt_threads_event);
-		mutex_exit(&fil_crypt_threads_mutex);
-	} else {
-		mutex_exit(&fil_system->mutex);
-	}
+	mutex_exit(&fil_system->mutex);
 
 	return(space);
 }
@@ -1841,10 +1913,13 @@ fil_init(
 
 	UT_LIST_INIT(fil_system->LRU, &fil_node_t::LRU);
 	UT_LIST_INIT(fil_system->space_list, &fil_space_t::space_list);
-	UT_LIST_INIT(fil_system->rotation_list, &fil_space_t::rotation_list);
 	UT_LIST_INIT(fil_system->unflushed_spaces,
 		     &fil_space_t::unflushed_spaces);
 	UT_LIST_INIT(fil_system->named_spaces, &fil_space_t::named_spaces);
+	UT_LIST_INIT(fil_system->encrypted_spaces,
+		     &fil_space_t::encrypted_spaces);
+	UT_LIST_INIT(fil_system->unencrypted_spaces,
+		     &fil_space_t::unencrypted_spaces);
 
 	fil_system->max_n_open = max_n_open;
 
@@ -3789,6 +3864,16 @@ fil_ibd_create(
 		file->block_size = block_size;
 		space->punch_hole = punch_hole;
 
+		if (space->purpose == FIL_TYPE_TABLESPACE) {
+			mutex_enter(&fil_system->mutex);
+
+			if (crypt_data == NULL) {
+				fil_space_add_to_encrypt_or_unencrypt_list(space);
+			}
+
+			mutex_exit(&fil_system->mutex);
+		}
+
 		err = DB_SUCCESS;
 	}
 
@@ -6067,90 +6152,6 @@ fil_space_next(fil_space_t* prev_space)
 	return(space);
 }
 
-/**
-Remove space from key rotation list if there are no more
-pending operations.
-@param[in,out]	space		Tablespace */
-static
-void
-fil_space_remove_from_keyrotation(fil_space_t* space)
-{
-	ut_ad(mutex_own(&fil_system->mutex));
-	ut_ad(space);
-
-	if (space->n_pending_ops == 0 && space->is_in_rotation_list) {
-		space->is_in_rotation_list = false;
-		ut_a(UT_LIST_GET_LEN(fil_system->rotation_list) > 0);
-		UT_LIST_REMOVE(fil_system->rotation_list, space);
-	}
-}
-
-
-/** Return the next fil_space_t from key rotation list.
-Once started, the caller must keep calling this until it returns NULL.
-fil_space_acquire() and fil_space_release() are invoked here which
-blocks a concurrent operation from dropping the tablespace.
-@param[in]	prev_space	Pointer to the previous fil_space_t.
-If NULL, use the first fil_space_t on fil_system->space_list.
-@return pointer to the next fil_space_t.
-@retval NULL if this was the last*/
-fil_space_t*
-fil_space_keyrotate_next(
-	fil_space_t*	prev_space)
-{
-	fil_space_t* space = prev_space;
-	fil_space_t* old   = NULL;
-
-	mutex_enter(&fil_system->mutex);
-
-	if (UT_LIST_GET_LEN(fil_system->rotation_list) == 0) {
-		if (space) {
-			ut_ad(space->n_pending_ops > 0);
-			space->n_pending_ops--;
-			fil_space_remove_from_keyrotation(space);
-		}
-		mutex_exit(&fil_system->mutex);
-		return(NULL);
-	}
-
-	if (prev_space == NULL) {
-		space = UT_LIST_GET_FIRST(fil_system->rotation_list);
-
-		/* We can trust that space is not NULL because we
-		checked list length above */
-	} else {
-		ut_ad(space->n_pending_ops > 0);
-
-		/* Move on to the next fil_space_t */
-		space->n_pending_ops--;
-
-		old = space;
-		space = UT_LIST_GET_NEXT(rotation_list, space);
-
-		fil_space_remove_from_keyrotation(old);
-	}
-
-	/* Skip spaces that are being created by fil_ibd_create(),
-	or dropped. Note that rotation_list contains only
-	space->purpose == FIL_TYPE_TABLESPACE. */
-	while (space != NULL
-	       && (UT_LIST_GET_LEN(space->chain) == 0
-		   || space->is_stopping())) {
-
-		old = space;
-		space = UT_LIST_GET_NEXT(rotation_list, space);
-		fil_space_remove_from_keyrotation(old);
-	}
-
-	if (space != NULL) {
-		space->n_pending_ops++;
-	}
-
-	mutex_exit(&fil_system->mutex);
-
-	return(space);
-}
-
 /** Determine the block size of the data file.
 @param[in]	space		tablespace
 @param[in]	offset		page number
diff --git a/storage/innobase/fsp/fsp0sysspace.cc b/storage/innobase/fsp/fsp0sysspace.cc
index b856bf3da74..a156603c9bc 100644
--- a/storage/innobase/fsp/fsp0sysspace.cc
+++ b/storage/innobase/fsp/fsp0sysspace.cc
@@ -933,6 +933,10 @@ SysTablespace::open_or_create(
 
 		space->add(it->m_filepath, OS_FILE_CLOSED, it->m_size,
 			   it->m_type != SRV_NOT_RAW, true, max_size);
+
+		if (space->purpose == FIL_TYPE_TABLESPACE) {
+			fil_space_add_to_encrypt_or_unencrypt_list(space);
+		}
 	}
 
 	return(err);
diff --git a/storage/innobase/include/fil0fil.h b/storage/innobase/include/fil0fil.h
index 162f10f08bb..c31c774042c 100644
--- a/storage/innobase/include/fil0fil.h
+++ b/storage/innobase/include/fil0fil.h
@@ -164,10 +164,18 @@ struct fil_space_t {
 				unflushed_spaces */
 	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;
-	/** whether this tablespace needs key rotation */
-	bool		is_in_rotation_list;
+
+	/** 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;
+
+	/** Whether the space is in encrypted list. */
+	bool in_encrypted_list;
+
+	/** Whether the space is in unencrypted list. */
+	bool in_unencrypted_list;
 
 	/** MariaDB encryption data */
 	fil_space_crypt_t* crypt_data;
@@ -207,6 +215,18 @@ struct fil_space_t {
 	fil_node_t* add(const char* name, pfs_os_file_t handle,
 			ulint size, bool is_raw, bool atomic_write,
 			ulint max_pages = ULINT_MAX);
+
+	/** Set the encrypted list flag. */
+	void set_encrypted_list(bool flag) { in_encrypted_list = flag; }
+
+	/** Set the unencrypted list flag. */
+	void set_unencrypted_list(bool flag) { in_unencrypted_list = flag; }
+
+	/** @retval whehter the space is in encrypted list. */
+	bool is_in_encrypted_list() { return in_encrypted_list; }
+
+	/** @retval whether the space is in unencrypted list. */
+	bool is_in_unencrypted_list() { return in_unencrypted_list; }
 };
 
 /** Value of fil_space_t::magic_n */
@@ -522,9 +542,13 @@ struct fil_system_t {
 					record has been written since
 					the latest redo log checkpoint.
 					Protected only by log_sys->mutex. */
-	UT_LIST_BASE_NODE_T(fil_space_t) rotation_list;
-					/*!< list of all file spaces needing
-					key rotation.*/
+	/** list of all encrypted space.
+	Protected only by fil_system->mutex. */
+	UT_LIST_BASE_NODE_T(fil_space_t) encrypted_spaces;
+
+	/** list of all unencrypted spaces.
+	Protected only by fil_system->mutex. */
+	UT_LIST_BASE_NODE_T(fil_space_t) unencrypted_spaces;
 
 	ibool		space_id_reuse_warned;
 					/* !< TRUE if fil_space_create()
@@ -781,18 +805,14 @@ fil_space_next(
 	fil_space_t*	prev_space)
 	MY_ATTRIBUTE((warn_unused_result));
 
-/** Return the next fil_space_t from key rotation list.
-Once started, the caller must keep calling this until it returns NULL.
-fil_space_acquire() and fil_space_release() are invoked here which
-blocks a concurrent operation from dropping the tablespace.
-@param[in,out]	prev_space	Pointer to the previous fil_space_t.
-If NULL, use the first fil_space_t on fil_system->space_list.
-@return pointer to the next fil_space_t.
-@retval NULL if this was the last*/
-fil_space_t*
-fil_space_keyrotate_next(
-	fil_space_t*	prev_space)
-	MY_ATTRIBUTE((warn_unused_result));
+/** Add the space to encrypted or unencrypted list. Remove the space
+from the unencrypted or encrypted list if it is present.
+@param[in]	space	space to be added */
+void fil_space_add_to_encrypt_or_unencrypt_list(fil_space_t* space);
+
+/** Remove the space from encrypted list or unencrypted list.
+@param[in]	space	space to be removed. */
+void fil_space_remove_from_encrypt_or_unencrypt_list(fil_space_t* space);
 
 /** Wrapper with reference-counting for a fil_space_t. */
 class FilSpace
diff --git a/storage/innobase/srv/srv0start.cc b/storage/innobase/srv/srv0start.cc
index ae890750841..46bbe1840d9 100644
--- a/storage/innobase/srv/srv0start.cc
+++ b/storage/innobase/srv/srv0start.cc
@@ -676,6 +676,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;
+		fil_space_add_to_encrypt_or_unencrypt_list(space);
 	} else {
 		success = file->read_page0(true);
 		if (!success) {