MDEV-19509 InnoDB skips the tablespace in rotation list

- If one of the encryption threads already started the initialization
of the tablespace then don't remove the other uninitialized tablespace
from the rotation list.

- If there is a change in innodb_encrypt_tables then
don't remove the processed tablespace from rotation list.

storage/innobase/fil/fil0crypt.cc

@@ -943,14 +943,10 @@ fil_crypt_read_crypt_data(fil_space_t* space)
 	mtr.commit();
 }
 
-/***********************************************************************
-Start encrypting a space
+/** Start encrypting a space
 @param[in,out]		space		Tablespace
-@return true if a recheck is needed */
-static
-bool
-fil_crypt_start_encrypting_space(
-	fil_space_t*	space)
+@return true if a recheck of tablespace is needed by encryption thread. */
+static bool fil_crypt_start_encrypting_space(fil_space_t* space)
 {
 	bool recheck = false;
 
@@ -1408,14 +1404,12 @@ fil_crypt_return_iops(
 	fil_crypt_update_total_stat(state);
 }
 
-/***********************************************************************
-Search for a space needing rotation
-@param[in,out]		key_state		Key state
-@param[in,out]		state			Rotation state
-@param[in,out]		recheck			recheck ? */
-static
-bool
-fil_crypt_find_space_to_rotate(
+/** Search for a space needing rotation
+@param[in,out]	key_state	Key state
+@param[in,out]	state		Rotation state
+@param[in,out]	recheck		recheck of the tablespace is needed or
+				still encryption thread does write page 0 */
+static bool fil_crypt_find_space_to_rotate(
 	key_state_t*		key_state,
 	rotate_thread_t*	state,
 	bool*			recheck)
@@ -1448,7 +1442,9 @@ fil_crypt_find_space_to_rotate(
 	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_system->keyrotate_next(
+					state->space, *recheck,
+					key_state->key_version);
 	}
 
 	while (!state->should_shutdown() && state->space) {
@@ -1470,7 +1466,9 @@ fil_crypt_find_space_to_rotate(
 		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_system->keyrotate_next(
+					state->space, *recheck,
+					key_state->key_version);
 		}
 	}
 

storage/innobase/fil/fil0fil.cc

@@ -5983,28 +5983,32 @@ fil_space_remove_from_keyrotation(fil_space_t* space)
 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.
+@param[in]	prev_space	Previous tablespace or NULL to start
+				from beginning of fil_system->rotation list
+@param[in]	recheck		recheck of the tablespace is needed or
+				still encryption thread does write page0 for it
+@param[in]	key_version	key version of the key state thread
 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*/
+@retval NULL if this was the last */
 fil_space_t*
-fil_space_keyrotate_next(
-	fil_space_t*	prev_space)
+fil_system_t::keyrotate_next(
+	fil_space_t*	prev_space,
+	bool		recheck,
+	uint		key_version)
 {
-	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 one of the encryption threads already started the encryption
+	of the table then don't remove the unencrypted spaces from
+	rotation list
+
+	If there is a change in innodb_encrypt_tables variables value then
+	don't remove the last processed tablespace from the rotation list. */
+	const bool remove = ((!recheck || prev_space->crypt_data)
+			     && (!key_version == !srv_encrypt_tables));
+
+	fil_space_t* space = prev_space;
 
 	if (prev_space == NULL) {
 		space = UT_LIST_GET_FIRST(fil_system->rotation_list);
@@ -6017,22 +6021,17 @@ fil_space_keyrotate_next(
 		/* 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())) {
+		while (space != NULL
+		       && (UT_LIST_GET_LEN(space->chain) == 0
+			   || space->is_stopping())) {
+			space = UT_LIST_GET_NEXT(rotation_list, space);
+		}
 
-		old = space;
-		space = UT_LIST_GET_NEXT(rotation_list, space);
-		fil_space_remove_from_keyrotation(old);
+		if (remove) {
+			fil_space_remove_from_keyrotation(prev_space);
+		}
 	}
 
 	if (space != NULL) {
@@ -6040,7 +6039,6 @@ fil_space_keyrotate_next(
 	}
 
 	mutex_exit(&fil_system->mutex);
-
 	return(space);
 }
 

storage/innobase/include/fil0fil.h

@@ -538,6 +538,25 @@ struct fil_system_t {
 	@return	tablespace
 	@retval	NULL	if the tablespace does not exist or cannot be read */
 	fil_space_t* read_page0(ulint id);
+
+	/** 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      Previous tablespace or NULL to start
+					from beginning of fil_system->rotation
+					list
+	@param[in]      recheck         recheck of the tablespace is needed or
+					still encryption thread does write page0
+					for it
+	@param[in]	key_version	key version of the key state thread
+	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* keyrotate_next(
+		fil_space_t*	prev_space,
+		bool		remove,
+		uint		key_version);
 };
 
 /** The tablespace memory cache. This variable is NULL before the module is
@@ -793,13 +812,15 @@ fil_space_next(
 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.
+@param[in]	prev_space	Previous tablespace or NULL to start
+				from beginning of fil_system->rotation list
+@param[in]	remove		Whether to remove the previous tablespace from
+				the rotation list
 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_keyrotate_next(fil_space_t* prev_space, bool remove)
 	MY_ATTRIBUTE((warn_unused_result));
 
 /** Wrapper with reference-counting for a fil_space_t. */