[PATCH 08/12] alter: introduce preparation phase

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

We may yield while rebuilding a vinyl index hence we must not modify
the old space before we are done building all indexes, otherwise the
user might see an inconsistent state while a space is being altered.
Currently, this requirement doesn't hold - we both modify the old space
and build the new indexes in AlterSpaceOp::alter. Let's introduce
AlterSpaceOp::prepare method that will perform all yielding operations
and forbid yielding in AlterSpaceOp::alter.

Actually, we introduce two new methods AlterSpaceOp::prepare and abort.
The latter is called if preparation phase failed and we need to undo the
effect of AlterSpaceOp::prepare for all operations that have performed
it. AlterSpaceOp::rollback is now only called in case of WAL error.
AlterSpaceOp::alter may not fail anymore, not that it needs to anyway.

Here's the list of AlterSpaceOp descendants that now use 'prepare'
instead of 'alter': CheckSpaceFormat, CreateIndex, DropIndex, and
RebuildIndex.

While we are at it, let's also add some comments to AlterSpaceOp
methods.
---
 src/box/alter.cc | 104 ++++++++++++++++++++++++++++++++++++++++++++++---------
 1 file changed, 88 insertions(+), 16 deletions(-)

diff --git a/src/box/alter.cc b/src/box/alter.cc
index 9dd8d8d5..f5996850 100644
--- a/src/box/alter.cc
+++ b/src/box/alter.cc
@@ -614,12 +614,48 @@ struct alter_space;
 class AlterSpaceOp {
 public:
 	AlterSpaceOp(struct alter_space *alter);
+
+	/** Link in alter_space::ops. */
 	struct rlist link;
+	/**
+	 * Called before creating the new space. Used to update
+	 * the space definition and/or key list that will be used
+	 * for creating the new space. Must not yield or fail.
+	 */
 	virtual void alter_def(struct alter_space * /* alter */) {}
+	/**
+	 * Called after creating a new space. Used for performing
+	 * long-lasting operations, such as index rebuild or format
+	 * check. May yield. May throw an exception. Must not modify
+	 * the old space.
+	 */
+	virtual void prepare(struct alter_space * /* alter */) {}
+	/**
+	 * Called if the preparation phase failed after this
+	 * operation has been successfully prepared. Supposed
+	 * to undo the effect of AlterSpaceOp::prepare.
+	 */
+	virtual void abort(struct alter_space * /* alter */) {}
+	/**
+	 * Called after all registered operations have completed
+	 * the preparation phase. Used to propagate the old space
+	 * state to the new space (e.g. move unchanged indexes).
+	 * Must not yield or fail.
+	 */
 	virtual void alter(struct alter_space * /* alter */) {}
+	/**
+	 * Called after the change has been successfully written
+	 * to WAL. Must not fail.
+	 */
 	virtual void commit(struct alter_space * /* alter */,
 			    int64_t /* signature */) {}
+	/**
+	 * Called in case a WAL error occurred. It is supposed to undo
+	 * the effect of AlterSpaceOp::prepare and AlterSpaceOp::alter.
+	 * Must not fail.
+	 */
 	virtual void rollback(struct alter_space * /* alter */) {}
+
 	virtual ~AlterSpaceOp() {}
 
 	void *operator new(size_t size)
@@ -657,6 +693,12 @@ struct alter_space {
 	/** New space. */
 	struct space *new_space;
 	/**
+	 * Space used as the source when building a new index.
+	 * Initially it is set to old_space, but may be reset
+	 * to new_space if the primary key is recreated.
+	 */
+	struct space *source_space;
+	/**
 	 * Assigned to the new primary key definition if we're
 	 * rebuilding the primary key, i.e. changing its key parts
 	 * substantially.
@@ -677,6 +719,7 @@ alter_space_new(struct space *old_space)
 		region_calloc_object_xc(&fiber()->gc, struct alter_space);
 	rlist_create(&alter->ops);
 	alter->old_space = old_space;
+	alter->source_space = old_space;
 	alter->space_def = space_def_dup_xc(alter->old_space->def);
 	if (old_space->format != NULL)
 		alter->new_min_field_count = old_space->format->min_field_count;
@@ -836,12 +879,12 @@ alter_space_do(struct txn *txn, struct alter_space *alter)
 	       sizeof(alter->old_space->access));
 
 	/*
-	 * Change the new space: build the new index, rename,
-	 * change the fixed field count.
+	 * Build new indexes, check if tuples conform to
+	 * the new space format.
 	 */
 	try {
 		rlist_foreach_entry(op, &alter->ops, link)
-			op->alter(alter);
+			op->prepare(alter);
 	} catch (Exception *e) {
 		/*
 		 * Undo space changes from the last successful
@@ -853,7 +896,7 @@ alter_space_do(struct txn *txn, struct alter_space *alter)
 		while (op != rlist_first_entry(&alter->ops,
 					       class AlterSpaceOp, link)) {
 			op = rlist_prev_entry(op, link);
-			op->rollback(alter);
+			op->abort(alter);
 		}
 		throw;
 	}
@@ -863,6 +906,10 @@ alter_space_do(struct txn *txn, struct alter_space *alter)
 	 * this point.
 	 */
 
+	/* Move old indexes, update space format. */
+	rlist_foreach_entry(op, &alter->ops, link)
+		op->alter(alter);
+
 	/* Rebuild index maps once for all indexes. */
 	space_fill_index_map(alter->old_space);
 	space_fill_index_map(alter->new_space);
@@ -900,11 +947,11 @@ class CheckSpaceFormat: public AlterSpaceOp
 public:
 	CheckSpaceFormat(struct alter_space *alter)
 		:AlterSpaceOp(alter) {}
-	virtual void alter(struct alter_space *alter);
+	virtual void prepare(struct alter_space *alter);
 };
 
 void
-CheckSpaceFormat::alter(struct alter_space *alter)
+CheckSpaceFormat::prepare(struct alter_space *alter)
 {
 	struct space *new_space = alter->new_space;
 	struct space *old_space = alter->old_space;
@@ -987,7 +1034,7 @@ public:
 	/** A reference to the definition of the dropped index. */
 	struct index_def *old_index_def;
 	virtual void alter_def(struct alter_space *alter);
-	virtual void alter(struct alter_space *alter);
+	virtual void prepare(struct alter_space *alter);
 	virtual void commit(struct alter_space *alter, int64_t lsn);
 };
 
@@ -1003,7 +1050,7 @@ DropIndex::alter_def(struct alter_space * /* alter */)
 
 /* Do the drop. */
 void
-DropIndex::alter(struct alter_space *alter)
+DropIndex::prepare(struct alter_space *alter)
 {
 	if (old_index_def->iid == 0)
 		space_drop_primary_key(alter->new_space);
@@ -1151,7 +1198,8 @@ public:
 	/** New index index_def. */
 	struct index_def *new_index_def;
 	virtual void alter_def(struct alter_space *alter);
-	virtual void alter(struct alter_space *alter);
+	virtual void prepare(struct alter_space *alter);
+	virtual void abort(struct alter_space *alter);
 	virtual void commit(struct alter_space *alter, int64_t lsn);
 	virtual void rollback(struct alter_space *alter);
 	virtual ~CreateIndex();
@@ -1175,7 +1223,7 @@ CreateIndex::alter_def(struct alter_space *alter)
  * they are fully enabled at all times.
  */
 void
-CreateIndex::alter(struct alter_space *alter)
+CreateIndex::prepare(struct alter_space *alter)
 {
 	if (new_index_def->iid == 0) {
 		/*
@@ -1189,6 +1237,12 @@ CreateIndex::alter(struct alter_space *alter)
 		 * all keys.
 		 */
 		space_add_primary_key_xc(alter->new_space);
+		/**
+		 * The primary index is recreated hence the
+		 * old data (if any) is discarded. Use the
+		 * new space for building secondary indexes.
+		 */
+		alter->source_space = alter->new_space;
 		return;
 	}
 	/**
@@ -1197,11 +1251,20 @@ CreateIndex::alter(struct alter_space *alter)
 	struct index *new_index = space_index(alter->new_space,
 					      new_index_def->iid);
 	assert(new_index != NULL);
-	space_build_index_xc(alter->new_space, new_index,
+	space_build_index_xc(alter->source_space, new_index,
 			     alter->new_space->format);
 }
 
 void
+CreateIndex::abort(struct alter_space *alter)
+{
+	struct index *new_index = space_index(alter->new_space,
+					      new_index_def->iid);
+	assert(new_index != NULL);
+	index_abort_create(new_index);
+}
+
+void
 CreateIndex::commit(struct alter_space *alter, int64_t signature)
 {
 	struct index *new_index = space_index(alter->new_space,
@@ -1249,7 +1312,8 @@ public:
 	/** Old index index_def. */
 	struct index_def *old_index_def;
 	virtual void alter_def(struct alter_space *alter);
-	virtual void alter(struct alter_space *alter);
+	virtual void prepare(struct alter_space *alter);
+	virtual void abort(struct alter_space *alter);
 	virtual void commit(struct alter_space *alter, int64_t signature);
 	virtual void rollback(struct alter_space *alter);
 	virtual ~RebuildIndex();
@@ -1264,15 +1328,23 @@ RebuildIndex::alter_def(struct alter_space *alter)
 }
 
 void
-RebuildIndex::alter(struct alter_space *alter)
+RebuildIndex::prepare(struct alter_space *alter)
 {
 	/* Get the new index and build it.  */
 	struct index *new_index = space_index(alter->new_space,
 					      new_index_def->iid);
 	assert(new_index != NULL);
-	space_build_index_xc(new_index_def->iid != 0 ?
-			     alter->new_space : alter->old_space,
-			     new_index, alter->new_space->format);
+	space_build_index_xc(alter->source_space, new_index,
+			     alter->new_space->format);
+}
+
+void
+RebuildIndex::abort(struct alter_space *alter)
+{
+	struct index *new_index = space_index(alter->new_space,
+					      new_index_def->iid);
+	assert(new_index != NULL);
+	index_abort_create(new_index);
 }
 
 void
-- 
2.11.0