[PATCH 4/5] vinyl: do not use space_vtab::commit_alter for preparing new indexes

[Vladimir Davydov <vdavydov.dev@gmail.com>]

Currently, space_vtab::commit_alter is called before WAL write so we can
use it for preparing new indexes in vinyl. However, this is going to
change soon, because actually space_vtab::commit_alter should be called
after WAL write, like index_vtab::commit_drop or commit_create. Calling
it before WAL write may result in use-after-free in memtx (see #3289).
Besides, using this function for iterating over space indexes just feels
wrong, as we have index methods invoked by AlterSpaceOp descendants for
this.

So let's move all the operations performed by vinyl_space_commit_alter
somewhere else. Fortunately, it's easy to do without damaging code
readability or efficiency:

 - Update of vy_lsm::pk can be done by vinyl_space_swap_index and
   vinyl_build_secondary_key.

 - vy_lsm->check_is_unique can be computed by vinyl_engine_create_space
   and then set by vinyl_space_swap_index.

 - Call to key_def_update_optionality is implied by key_def_swap, which
   is already called by vinyl_space_swap_index, hence it can be removed.

Needed for #3289
---
 src/box/vinyl.c  | 79 ++++++++++++++++++++++++--------------------------------
 src/box/vy_lsm.h | 17 ++++++++++++
 2 files changed, 51 insertions(+), 45 deletions(-)

diff --git a/src/box/vinyl.c b/src/box/vinyl.c
index d503704e..01755d7d 100644
--- a/src/box/vinyl.c
+++ b/src/box/vinyl.c
@@ -643,6 +643,27 @@ vinyl_engine_create_space(struct engine *engine, struct space_def *def,
 
 	/* Format is now referenced by the space. */
 	tuple_format_unref(format);
+
+	/*
+	 * Check if there are unique indexes that are contained
+	 * by other unique indexes. For them, we can skip check
+	 * for duplicates on INSERT. Prefer indexes with higher
+	 * ids for uniqueness check optimization as they are
+	 * likelier to have a "colder" cache.
+	 */
+	for (int i = space->index_count - 1; i >= 0; i--) {
+		struct vy_lsm *lsm = vy_lsm(space->index[i]);
+		if (!lsm->check_is_unique)
+			continue;
+		for (int j = 0; j < (int)space->index_count; j++) {
+			struct vy_lsm *other = vy_lsm(space->index[j]);
+			if (other != lsm && other->check_is_unique &&
+			    key_def_contains(lsm->key_def, other->key_def)) {
+				lsm->check_is_unique = false;
+				break;
+			}
+		}
+	}
 	return space;
 }
 
@@ -1035,49 +1056,7 @@ static void
 vinyl_space_commit_alter(struct space *old_space, struct space *new_space)
 {
 	(void)old_space;
-	if (new_space == NULL || new_space->index_count == 0)
-		return; /* space drop */
-
-	struct tuple_format *new_format = new_space->format;
-	struct vy_lsm *pk = vy_lsm(new_space->index[0]);
-	assert(pk->pk == NULL);
-
-	pk->check_is_unique = true;
-	key_def_update_optionality(pk->key_def, new_format->min_field_count);
-	key_def_update_optionality(pk->cmp_def, new_format->min_field_count);
-
-	for (uint32_t i = 1; i < new_space->index_count; ++i) {
-		struct vy_lsm *lsm = vy_lsm(new_space->index[i]);
-		vy_lsm_unref(lsm->pk);
-		vy_lsm_ref(pk);
-		lsm->pk = pk;
-		lsm->check_is_unique = lsm->opts.is_unique;
-		key_def_update_optionality(lsm->key_def,
-					   new_format->min_field_count);
-		key_def_update_optionality(lsm->cmp_def,
-					   new_format->min_field_count);
-	}
-
-	/*
-	 * Check if there are unique indexes that are contained
-	 * by other unique indexes. For them, we can skip check
-	 * for duplicates on INSERT. Prefer indexes with higher
-	 * ids for uniqueness check optimization as they are
-	 * likelier to have a "colder" cache.
-	 */
-	for (int i = new_space->index_count - 1; i >= 0; i--) {
-		struct vy_lsm *lsm = vy_lsm(new_space->index[i]);
-		if (!lsm->check_is_unique)
-			continue;
-		for (int j = 0; j < (int)new_space->index_count; j++) {
-			struct vy_lsm *other = vy_lsm(new_space->index[j]);
-			if (other != lsm && other->check_is_unique &&
-			    key_def_contains(lsm->key_def, other->key_def)) {
-				lsm->check_is_unique = false;
-				break;
-			}
-		}
-	}
+	(void)new_space;
 }
 
 static void
@@ -1094,6 +1073,8 @@ vinyl_space_swap_index(struct space *old_space, struct space *new_space,
 	generic_space_swap_index(old_space, new_space,
 				 old_index_id, new_index_id);
 
+	SWAP(old_lsm, new_lsm);
+	SWAP(old_lsm->check_is_unique, new_lsm->check_is_unique);
 	SWAP(old_lsm->mem_format, new_lsm->mem_format);
 	SWAP(old_lsm->mem_format_with_colmask,
 	     new_lsm->mem_format_with_colmask);
@@ -1101,6 +1082,10 @@ vinyl_space_swap_index(struct space *old_space, struct space *new_space,
 	SWAP(old_lsm->opts, new_lsm->opts);
 	key_def_swap(old_lsm->key_def, new_lsm->key_def);
 	key_def_swap(old_lsm->cmp_def, new_lsm->cmp_def);
+
+	/* Update pointer to the primary key. */
+	vy_lsm_update_pk(old_lsm, vy_lsm(old_space->index_map[0]));
+	vy_lsm_update_pk(new_lsm, vy_lsm(new_space->index_map[0]));
 }
 
 static int
@@ -1121,7 +1106,6 @@ vinyl_space_build_secondary_key(struct space *old_space,
 				struct index *new_index)
 {
 	(void)old_space;
-	(void)new_space;
 	/*
 	 * Unlike Memtx, Vinyl does not need building of a secondary index.
 	 * This is true because of two things:
@@ -1142,7 +1126,12 @@ vinyl_space_build_secondary_key(struct space *old_space,
 	 *   Engine::buildSecondaryKey(old_space, new_space, new_index_arg);
 	 *  but aware of three cases mentioned above.
 	 */
-	return vinyl_index_open(new_index);
+	if (vinyl_index_open(new_index) != 0)
+		return -1;
+
+	/* Set pointer to the primary key for the new index. */
+	vy_lsm_update_pk(vy_lsm(new_index), vy_lsm(space_index(new_space, 0)));
+	return 0;
 }
 
 static size_t
diff --git a/src/box/vy_lsm.h b/src/box/vy_lsm.h
index 201b5a56..2ef190ba 100644
--- a/src/box/vy_lsm.h
+++ b/src/box/vy_lsm.h
@@ -332,6 +332,23 @@ vy_lsm_unref(struct vy_lsm *lsm)
 }
 
 /**
+ * Update pointer to the primary key for an LSM tree.
+ * If called for an LSM tree corresponding to a primary
+ * index, this function does nothing.
+ */
+static inline void
+vy_lsm_update_pk(struct vy_lsm *lsm, struct vy_lsm *pk)
+{
+	if (lsm->index_id == 0) {
+		assert(lsm->pk == NULL);
+		return;
+	}
+	vy_lsm_unref(lsm->pk);
+	vy_lsm_ref(pk);
+	lsm->pk = pk;
+}
+
+/**
  * Create a new LSM tree.
  *
  * This function is called when an LSM tree is created
-- 
2.11.0