[PATCH v2 0/8] Allow to build indexes for vinyl spaces

[PATCH v2 0/8] Allow to build indexes for vinyl spaces

[Vladimir Davydov]

This patch set implements the ability to build secondary indexes for
non-empty vinyl spaces. For implementation details, see patch 8.

https://github.com/tarantool/tarantool/issues/1653
https://github.com/tarantool/tarantool/commits/vy-allow-to-build-secondary-indexes

Changes in v2:
 - Split the patch into logical parts.
 - Address comments by Konstantin and Vladislav (refactoring mostly).

v1: https://www.freelists.org/post/tarantool-patches/PATCH-vinyl-allow-to-build-secondary-index-for-nonempty-space

Vladimir Davydov (8):
  vinyl: allocate key parts in vy_recovery_do_create_lsm
  vinyl: update recovery context with records written during recovery
  vinyl: log new index before WAL write on DDL
  vinyl: bump mem version after committing statement
  vinyl: allow to commit statements to mem in arbitrary order
  vinyl: relax limitation imposed on run min/max lsn
  vinyl: factor out vy_check_is_unique_secondary
  vinyl: allow to build secondary index for non-empty space

 src/box/vinyl.c                  | 610 +++++++++++++++++++++++++++++------
 src/box/vy_log.c                 | 216 +++++++++++--
 src/box/vy_log.h                 |  51 ++-
 src/box/vy_lsm.c                 |  94 ++++--
 src/box/vy_lsm.h                 |   4 +-
 src/box/vy_mem.c                 |   8 +-
 src/box/vy_quota.h               |  10 +
 src/box/vy_scheduler.c           |  31 +-
 src/box/vy_scheduler.h           |   7 +
 test/box/alter.result            | 501 -----------------------------
 test/box/alter.test.lua          | 159 ---------
 test/engine/ddl.result           | 672 +++++++++++++++++++++++++++++++++++++++
 test/engine/ddl.test.lua         | 221 +++++++++++++
 test/engine/truncate.result      |  87 -----
 test/engine/truncate.test.lua    |  48 ---
 test/unit/vy_mem.c               |   4 +-
 test/vinyl/ddl.result            | 446 +++++++++++++++-----------
 test/vinyl/ddl.test.lua          | 220 ++++++++-----
 test/vinyl/errinj.result         | 256 +++++++++++++++
 test/vinyl/errinj.test.lua       | 115 +++++++
 test/vinyl/errinj_gc.result      |  78 ++++-
 test/vinyl/errinj_gc.test.lua    |  38 ++-
 test/vinyl/errinj_vylog.result   | 216 ++++++++-----
 test/vinyl/errinj_vylog.test.lua | 139 ++++----
 test/vinyl/gh.result             |   2 +-
 25 files changed, 2858 insertions(+), 1375 deletions(-)

-- 
2.11.0

[PATCH v2 1/8] vinyl: allocate key parts in vy_recovery_do_create_lsm

[Vladimir Davydov]

Allocation of vy_lsm_recovery_info::key_parts is a part of the struct
initialization, which is handled by vy_recovery_do_create_lsm().
---
 src/box/vy_log.c | 39 +++++++++++++++++++++------------------
 1 file changed, 21 insertions(+), 18 deletions(-)

diff --git a/src/box/vy_log.c b/src/box/vy_log.c
index 4d561354..bdf9eee3 100644
--- a/src/box/vy_log.c
+++ b/src/box/vy_log.c
@@ -1254,19 +1254,35 @@ vy_recovery_lookup_slice(struct vy_recovery *recovery, int64_t slice_id)
  */
 static struct vy_lsm_recovery_info *
 vy_recovery_do_create_lsm(struct vy_recovery *recovery, int64_t id,
-			  uint32_t space_id, uint32_t index_id)
+			  uint32_t space_id, uint32_t index_id,
+			  const struct key_part_def *key_parts,
+			  uint32_t key_part_count)
 {
+	if (key_parts == NULL) {
+		diag_set(ClientError, ER_INVALID_VYLOG_FILE,
+			 tt_sprintf("Missing key definition for LSM tree %lld",
+				    (long long)id));
+		return NULL;
+	}
 	struct vy_lsm_recovery_info *lsm = malloc(sizeof(*lsm));
 	if (lsm == NULL) {
 		diag_set(OutOfMemory, sizeof(*lsm),
 			 "malloc", "struct vy_lsm_recovery_info");
 		return NULL;
 	}
+	lsm->key_parts = malloc(sizeof(*key_parts) * key_part_count);
+	if (lsm->key_parts == NULL) {
+		diag_set(OutOfMemory, sizeof(*key_parts) * key_part_count,
+			 "malloc", "struct key_part_def");
+		free(lsm);
+		return NULL;
+	}
 	struct mh_i64ptr_t *h = recovery->lsm_hash;
 	struct mh_i64ptr_node_t node = { id, lsm };
 	struct mh_i64ptr_node_t *old_node = NULL;
 	if (mh_i64ptr_put(h, &node, &old_node, NULL) == mh_end(h)) {
 		diag_set(OutOfMemory, 0, "mh_i64ptr_put", "mh_i64ptr_node_t");
+		free(lsm->key_parts);
 		free(lsm);
 		return NULL;
 	}
@@ -1274,8 +1290,8 @@ vy_recovery_do_create_lsm(struct vy_recovery *recovery, int64_t id,
 	lsm->id = id;
 	lsm->space_id = space_id;
 	lsm->index_id = index_id;
-	lsm->key_parts = NULL;
-	lsm->key_part_count = 0;
+	memcpy(lsm->key_parts, key_parts, sizeof(*key_parts) * key_part_count);
+	lsm->key_part_count = key_part_count;
 	lsm->create_lsn = -1;
 	lsm->modify_lsn = -1;
 	lsm->drop_lsn = -1;
@@ -1306,12 +1322,6 @@ vy_recovery_create_lsm(struct vy_recovery *recovery, int64_t id,
 		       uint32_t key_part_count, int64_t create_lsn,
 		       int64_t modify_lsn, int64_t dump_lsn)
 {
-	if (key_parts == NULL) {
-		diag_set(ClientError, ER_INVALID_VYLOG_FILE,
-			 tt_sprintf("Missing key definition for LSM tree %lld",
-				    (long long)id));
-		return -1;
-	}
 	if (vy_recovery_lookup_lsm(recovery, id) != NULL) {
 		diag_set(ClientError, ER_INVALID_VYLOG_FILE,
 			 tt_sprintf("Duplicate LSM tree id %lld",
@@ -1326,18 +1336,11 @@ vy_recovery_create_lsm(struct vy_recovery *recovery, int64_t id,
 				    (unsigned)space_id, (unsigned)index_id));
 		return -1;
 	}
-	lsm = vy_recovery_do_create_lsm(recovery, id, space_id, index_id);
+	lsm = vy_recovery_do_create_lsm(recovery, id, space_id, index_id,
+					key_parts, key_part_count);
 	if (lsm == NULL)
 		return -1;
 
-	lsm->key_parts = malloc(sizeof(*key_parts) * key_part_count);
-	if (lsm->key_parts == NULL) {
-		diag_set(OutOfMemory, sizeof(*key_parts) * key_part_count,
-			 "malloc", "struct key_part_def");
-		return -1;
-	}
-	memcpy(lsm->key_parts, key_parts, sizeof(*key_parts) * key_part_count);
-	lsm->key_part_count = key_part_count;
 	lsm->create_lsn = create_lsn;
 	lsm->modify_lsn = modify_lsn;
 	lsm->dump_lsn = dump_lsn;
-- 
2.11.0

[PATCH v2 2/8] vinyl: update recovery context with records written during recovery

[Vladimir Davydov]

During recovery, we may write VY_LOG_CREATE_LSM and VY_LOG_DROP_LSM
records we failed to write before restart (because those records are
written after WAL and hence may not make it to vylog). Right after
recovery we invoke garbage collection to drop incomplete runs. Once
VY_LOG_PREPARE_LSM record is introduced, we will also collect incomplete
LSM trees there (those we failed to build). However, there may be LSM
trees we managed to build but failed to write VY_LOG_CREATE_LSM for.
This is OK as we will retry vylog write, but currenntly it isn't
reflected in the recovery context used for garbage collection. To avoid
purging such LSM trees, let's update the recovery context with records
written during recovery.

Needed for #1653
---
 src/box/vy_log.c | 12 ++++++++++++
 1 file changed, 12 insertions(+)

diff --git a/src/box/vy_log.c b/src/box/vy_log.c
index bdf9eee3..069cd528 100644
--- a/src/box/vy_log.c
+++ b/src/box/vy_log.c
@@ -177,6 +177,10 @@ static struct vy_log vy_log;
 static struct vy_recovery *
 vy_recovery_new_locked(int64_t signature, bool only_checkpoint);
 
+static int
+vy_recovery_process_record(struct vy_recovery *recovery,
+			   const struct vy_log_record *record);
+
 /**
  * Return the name of the vylog file that has the given signature.
  */
@@ -890,6 +894,14 @@ vy_log_end_recovery(void)
 {
 	assert(vy_log.recovery != NULL);
 
+	/*
+	 * Update the recovery context with records written during
+	 * recovery - we will need them for garbage collection.
+	 */
+	struct vy_log_record *record;
+	stailq_foreach_entry(record, &vy_log.tx, in_tx)
+		vy_recovery_process_record(vy_log.recovery, record);
+
 	/* Flush all pending records. */
 	if (vy_log_flush() < 0) {
 		diag_log();
-- 
2.11.0

[PATCH v2 3/8] vinyl: log new index before WAL write on DDL

[Vladimir Davydov]

Currently, we write new indexes to vylog only after successful WAL
write (see vinyl_index_commit_create). This is incompatible with space
ALTER - the problem is during ALTER vinyl may need to create new run
files, which we need to track in order not to leave garbage if ALTER
fails or tarantool exits before ALTER is complete.

So this patch splits index creation in two stages, prepare and commit.
The 'commit' stage is represented by existing VY_LOG_CREATE_LSM record,
which is written from index_vtab::commit_create callback, just like
before. For the 'prepare' stage we introduce a new record type,
VY_LOG_REPARE_LSM, written from index_vtab::add_primary_key and
index_vtab::build_index callbacks, i.e. before WAL write. For now, we
don't write anything to prepared, but uncommitted indexes (this will be
done later), but we do add prepared indexes to the scheduler so that
they can be dumped and compacted. If ALTER fails, we drop prepared
indexes in index_vtab::abort_create callback. Prepared but uncommitted
indexes are ignored by backup and replication and cleaned up from vylog
on restart.

Note, we have to rework vinyl/errinj_vylog test in this patch, because
index creation (and hence space truncation) commands now fail on vylog
error, i.e. a situation when the same index is dropped and recreated
multiple times in xlog without having corresponding records in vylog is
now impossible.

Also, space truncation is not linearizable for vinyl anymore as it may
yield before WAL write, while trying to prepare an index in vylog. This
is OK - we never promised it is. Remove the corresponding test case.

Needed for #1653
---
 src/box/vinyl.c                  |  75 +++++++++++------
 src/box/vy_log.c                 | 171 +++++++++++++++++++++++++++++++++++----
 src/box/vy_log.h                 |  51 ++++++++++--
 src/box/vy_lsm.c                 |  80 ++++++++++++------
 src/box/vy_lsm.h                 |   4 +-
 test/engine/truncate.result      |  87 --------------------
 test/engine/truncate.test.lua    |  48 -----------
 test/vinyl/errinj_vylog.result   | 169 ++++++++++++++++++--------------------
 test/vinyl/errinj_vylog.test.lua | 112 ++++++++++++-------------
 9 files changed, 438 insertions(+), 359 deletions(-)

diff --git a/src/box/vinyl.c b/src/box/vinyl.c
index f0d26874..4d670328 100644
--- a/src/box/vinyl.c
+++ b/src/box/vinyl.c
@@ -803,6 +803,8 @@ vinyl_index_open(struct index *index)
 	default:
 		unreachable();
 	}
+	if (rc == 0)
+		vy_scheduler_add_lsm(&env->scheduler, lsm);
 	return rc;
 }
 
@@ -824,10 +826,8 @@ vinyl_index_commit_create(struct index *index, int64_t lsn)
 		 * the index isn't in the recovery context and we
 		 * need to retry to log it now.
 		 */
-		if (lsm->commit_lsn >= 0) {
-			vy_scheduler_add_lsm(&env->scheduler, lsm);
+		if (lsm->commit_lsn >= 0)
 			return;
-		}
 	}
 
 	if (env->status == VINYL_INITIAL_RECOVERY_REMOTE) {
@@ -852,9 +852,6 @@ vinyl_index_commit_create(struct index *index, int64_t lsn)
 	assert(lsm->commit_lsn < 0);
 	lsm->commit_lsn = lsn;
 
-	assert(lsm->range_count == 1);
-	struct vy_range *range = vy_range_tree_first(lsm->tree);
-
 	/*
 	 * Since it's too late to fail now, in case of vylog write
 	 * failure we leave the records we attempted to write in
@@ -864,15 +861,36 @@ vinyl_index_commit_create(struct index *index, int64_t lsn)
 	 * recovery.
 	 */
 	vy_log_tx_begin();
-	vy_log_create_lsm(lsm->id, lsm->space_id, lsm->index_id,
-			  lsm->key_def, lsn);
-	vy_log_insert_range(lsm->id, range->id, NULL, NULL);
+	vy_log_create_lsm(lsm->id, lsn);
+	vy_log_tx_try_commit();
+}
+
+static void
+vinyl_index_abort_create(struct index *index)
+{
+	struct vy_env *env = vy_env(index->engine);
+	struct vy_lsm *lsm = vy_lsm(index);
+
+	if (env->status != VINYL_ONLINE) {
+		/* Failure during recovery. Nothing to do. */
+		return;
+	}
+	if (lsm->id < 0) {
+		/*
+		 * ALTER failed before we wrote information about
+		 * the new LSM tree to vylog, see vy_lsm_create().
+		 * Nothing to do.
+		 */
+		return;
+	}
+
+	vy_scheduler_remove_lsm(&env->scheduler, lsm);
+
+	lsm->is_dropped = true;
+
+	vy_log_tx_begin();
+	vy_log_drop_lsm(lsm->id, 0);
 	vy_log_tx_try_commit();
-	/*
-	 * After we committed the index in the log, we can schedule
-	 * a task for it.
-	 */
-	vy_scheduler_add_lsm(&env->scheduler, lsm);
 }
 
 static void
@@ -3111,8 +3129,10 @@ vy_send_lsm(struct vy_join_ctx *ctx, struct vy_lsm_recovery_info *lsm_info)
 {
 	int rc = -1;
 
-	if (lsm_info->drop_lsn >= 0)
+	if (lsm_info->drop_lsn >= 0 || lsm_info->create_lsn < 0) {
+		/* Dropped or not yet built LSM tree. */
 		return 0;
+	}
 
 	/*
 	 * We are only interested in the primary index LSM tree.
@@ -3326,16 +3346,21 @@ vy_gc_run(struct vy_env *env,
 }
 
 /**
- * Given a dropped LSM tree, delete all its ranges and slices and
- * mark all its runs as dropped. Forget the LSM tree if it has no
- * associated objects.
+ * Given a dropped or not fully built LSM tree, delete all its
+ * ranges and slices and mark all its runs as dropped. Forget
+ * the LSM tree if it has no associated objects.
  */
 static void
 vy_gc_lsm(struct vy_lsm_recovery_info *lsm_info)
 {
-	assert(lsm_info->drop_lsn >= 0);
+	assert(lsm_info->drop_lsn >= 0 ||
+	       lsm_info->create_lsn < 0);
 
 	vy_log_tx_begin();
+	if (lsm_info->drop_lsn < 0) {
+		lsm_info->drop_lsn = 0;
+		vy_log_drop_lsm(lsm_info->id, 0);
+	}
 	struct vy_range_recovery_info *range_info;
 	rlist_foreach_entry(range_info, &lsm_info->ranges, in_lsm) {
 		struct vy_slice_recovery_info *slice_info;
@@ -3371,8 +3396,10 @@ vy_gc(struct vy_env *env, struct vy_recovery *recovery,
 	int loops = 0;
 	struct vy_lsm_recovery_info *lsm_info;
 	rlist_foreach_entry(lsm_info, &recovery->lsms, in_recovery) {
-		if ((gc_mask & VY_GC_DROPPED) != 0 &&
-		    lsm_info->drop_lsn >= 0)
+		if ((lsm_info->drop_lsn >= 0 &&
+		     (gc_mask & VY_GC_DROPPED) != 0) ||
+		    (lsm_info->create_lsn < 0 &&
+		     (gc_mask & VY_GC_INCOMPLETE) != 0))
 			vy_gc_lsm(lsm_info);
 
 		struct vy_run_recovery_info *run_info;
@@ -3438,8 +3465,10 @@ vinyl_engine_backup(struct engine *engine, struct vclock *vclock,
 	int loops = 0;
 	struct vy_lsm_recovery_info *lsm_info;
 	rlist_foreach_entry(lsm_info, &recovery->lsms, in_recovery) {
-		if (lsm_info->drop_lsn >= 0)
+		if (lsm_info->drop_lsn >= 0 || lsm_info->create_lsn < 0) {
+			/* Dropped or not yet built LSM tree. */
 			continue;
+		}
 		struct vy_run_recovery_info *run_info;
 		rlist_foreach_entry(run_info, &lsm_info->runs, in_lsm) {
 			if (run_info->is_dropped || run_info->is_incomplete)
@@ -4059,7 +4088,7 @@ static const struct space_vtab vinyl_space_vtab = {
 static const struct index_vtab vinyl_index_vtab = {
 	/* .destroy = */ vinyl_index_destroy,
 	/* .commit_create = */ vinyl_index_commit_create,
-	/* .abort_create = */ generic_index_abort_create,
+	/* .abort_create = */ vinyl_index_abort_create,
 	/* .commit_modify = */ vinyl_index_commit_modify,
 	/* .commit_drop = */ vinyl_index_commit_drop,
 	/* .update_def = */ generic_index_update_def,
diff --git a/src/box/vy_log.c b/src/box/vy_log.c
index 069cd528..e98b95e5 100644
--- a/src/box/vy_log.c
+++ b/src/box/vy_log.c
@@ -121,6 +121,7 @@ static const char *vy_log_type_name[] = {
 	[VY_LOG_TRUNCATE_LSM]		= "truncate_lsm",
 	[VY_LOG_MODIFY_LSM]		= "modify_lsm",
 	[VY_LOG_FORGET_LSM]		= "forget_lsm",
+	[VY_LOG_PREPARE_LSM]		= "prepare_lsm",
 };
 
 /** Metadata log object. */
@@ -1308,6 +1309,7 @@ vy_recovery_do_create_lsm(struct vy_recovery *recovery, int64_t id,
 	lsm->modify_lsn = -1;
 	lsm->drop_lsn = -1;
 	lsm->dump_lsn = -1;
+	lsm->prepared = NULL;
 	rlist_create(&lsm->ranges);
 	rlist_create(&lsm->runs);
 	/*
@@ -1322,17 +1324,27 @@ vy_recovery_do_create_lsm(struct vy_recovery *recovery, int64_t id,
 }
 
 /**
- * Handle a VY_LOG_CREATE_LSM log record.
- * This function allocates a new vinyl LSM tree with ID @id
- * and inserts it to the hash.
- * Return 0 on success, -1 on failure (ID collision or OOM).
+ * Handle a VY_LOG_PREPARE_LSM log record.
+ *
+ * This function allocates a new LSM tree with the given ID and
+ * either associates it with the existing LSM tree hashed under
+ * the same space_id/index_id or inserts it into the hash if
+ * there's none.
+ *
+ * Note, we link incomplete LSM trees to index_id_hash (either
+ * directly or indirectly via vy_lsm_recovery_info::prepared),
+ * because an LSM tree may have been fully built and logged in
+ * WAL, but not committed to vylog. We need to be able to identify
+ * such LSM trees during local recovery so that instead of
+ * rebuilding them we can simply retry vylog write.
+ *
+ * Returns 0 on success, -1 on error.
  */
 static int
-vy_recovery_create_lsm(struct vy_recovery *recovery, int64_t id,
-		       uint32_t space_id, uint32_t index_id,
-		       const struct key_part_def *key_parts,
-		       uint32_t key_part_count, int64_t create_lsn,
-		       int64_t modify_lsn, int64_t dump_lsn)
+vy_recovery_prepare_lsm(struct vy_recovery *recovery, int64_t id,
+			uint32_t space_id, uint32_t index_id,
+			const struct key_part_def *key_parts,
+			uint32_t key_part_count)
 {
 	if (vy_recovery_lookup_lsm(recovery, id) != NULL) {
 		diag_set(ClientError, ER_INVALID_VYLOG_FILE,
@@ -1340,23 +1352,113 @@ vy_recovery_create_lsm(struct vy_recovery *recovery, int64_t id,
 				    (long long)id));
 		return -1;
 	}
+	struct vy_lsm_recovery_info *new_lsm;
+	new_lsm = vy_recovery_do_create_lsm(recovery, id, space_id, index_id,
+					    key_parts, key_part_count);
+	if (new_lsm == NULL)
+		return -1;
+
 	struct vy_lsm_recovery_info *lsm;
 	lsm = vy_recovery_lsm_by_index_id(recovery, space_id, index_id);
-	if (lsm != NULL && lsm->drop_lsn < 0) {
+	if (lsm == NULL) {
+		/*
+		 * There's no LSM tree for these space_id/index_id
+		 * in the recovery context. Insert the new LSM tree
+		 * into the index_id_hash.
+		 */
+		return vy_recovery_hash_index_id(recovery, new_lsm);
+	}
+
+	/*
+	 * If there's an LSM tree for the given space_id/index_id,
+	 * it can't be incomplete (i.e. it must be committed albeit
+	 * it may be dropped), neither can it have a prepared LSM
+	 * tree associated with it.
+	 */
+	if (lsm->create_lsn < 0 || lsm->prepared != NULL) {
 		diag_set(ClientError, ER_INVALID_VYLOG_FILE,
-			 tt_sprintf("LSM tree %u/%u created twice",
+			 tt_sprintf("LSM tree %u/%u prepared twice",
 				    (unsigned)space_id, (unsigned)index_id));
 		return -1;
 	}
-	lsm = vy_recovery_do_create_lsm(recovery, id, space_id, index_id,
-					key_parts, key_part_count);
-	if (lsm == NULL)
-		return -1;
 
+	/* Link the new LSM tree to the existing one. */
+	lsm->prepared = new_lsm;
+	return 0;
+}
+
+/**
+ * Handle a VY_LOG_CREATE_LSM log record.
+ *
+ * Depending on whether the LSM tree was previously prepared,
+ * this function either commits it or allocates a new one and
+ * inserts it into the recovery hash.
+ *
+ * Returns 0 on success, -1 on error.
+ */
+static int
+vy_recovery_create_lsm(struct vy_recovery *recovery, int64_t id,
+		       uint32_t space_id, uint32_t index_id,
+		       const struct key_part_def *key_parts,
+		       uint32_t key_part_count, int64_t create_lsn,
+		       int64_t modify_lsn, int64_t dump_lsn)
+{
+	struct vy_lsm_recovery_info *lsm;
+	lsm = vy_recovery_lookup_lsm(recovery, id);
+	if (lsm != NULL) {
+		/*
+		 * If the LSM tree already exists, it must be in
+		 * the prepared state (i.e. not committed or dropped).
+		 */
+		if (lsm->create_lsn >= 0 || lsm->drop_lsn >= 0) {
+			diag_set(ClientError, ER_INVALID_VYLOG_FILE,
+				 tt_sprintf("Duplicate LSM tree id %lld",
+					    (long long)id));
+			return -1;
+		}
+	} else {
+		lsm = vy_recovery_do_create_lsm(recovery, id, space_id, index_id,
+						key_parts, key_part_count);
+		if (lsm == NULL)
+			return -1;
+		lsm->dump_lsn = dump_lsn;
+	}
+
+	/* Mark the LSM tree committed by assigning LSN. */
 	lsm->create_lsn = create_lsn;
 	lsm->modify_lsn = modify_lsn;
-	lsm->dump_lsn = dump_lsn;
 
+	/*
+	 * Hash the new LSM tree under the given space_id/index_id.
+	 * First, look up the LSM tree that is presently in the hash.
+	 */
+	struct vy_lsm_recovery_info *old_lsm;
+	old_lsm = vy_recovery_lsm_by_index_id(recovery, space_id, index_id);
+	if (old_lsm == lsm) {
+		/*
+		 * The new LSM tree is already hashed, nothing to do
+		 * (it was hashed on prepare).
+		 */
+		return 0;
+	}
+
+	/* Unlink the new LSM tree from the old one, if any. */
+	if (old_lsm != NULL) {
+		assert(old_lsm->create_lsn >= 0);
+		if (old_lsm->drop_lsn < 0) {
+			diag_set(ClientError, ER_INVALID_VYLOG_FILE,
+				 tt_sprintf("LSM tree %u/%u created twice",
+					    (unsigned)space_id,
+					    (unsigned)index_id));
+			return -1;
+		}
+		if (old_lsm->prepared != NULL) {
+			assert(old_lsm->prepared == lsm);
+			old_lsm->prepared = NULL;
+		}
+	}
+
+	/* Update the hash with the new LSM tree. */
 	return vy_recovery_hash_index_id(recovery, lsm);
 }
 
@@ -1421,6 +1523,35 @@ vy_recovery_drop_lsm(struct vy_recovery *recovery, int64_t id, int64_t drop_lsn)
 	}
 	assert(drop_lsn >= 0);
 	lsm->drop_lsn = drop_lsn;
+
+	if (lsm->create_lsn >= 0)
+		return 0;
+	/*
+	 * If the dropped LSM tree has never been committed,
+	 * it means that ALTER for the corresponding index was
+	 * aborted, hence we don't need to keep it in the
+	 * index_id_hash, because the LSM tree is pure garbage
+	 * and will never be recovered. Unlink it now.
+	 */
+	struct vy_lsm_recovery_info *hashed_lsm;
+	hashed_lsm = vy_recovery_lsm_by_index_id(recovery,
+				lsm->space_id, lsm->index_id);
+	if (hashed_lsm == lsm) {
+		/*
+		 * The LSM tree is linked to the index_id_hash
+		 * directly. Remove the corresponding hash entry.
+		 */
+		vy_recovery_unhash_index_id(recovery,
+				lsm->space_id, lsm->index_id);
+	} else {
+		/*
+		 * The LSM tree was linked to an existing LSM
+		 * tree via vy_lsm_recovery_info::prepared.
+		 * Clear the reference.
+		 */
+		assert(hashed_lsm->prepared == lsm);
+		hashed_lsm->prepared = NULL;
+	}
 	return 0;
 }
 
@@ -1873,6 +2004,11 @@ vy_recovery_process_record(struct vy_recovery *recovery,
 {
 	int rc;
 	switch (record->type) {
+	case VY_LOG_PREPARE_LSM:
+		rc = vy_recovery_prepare_lsm(recovery, record->lsm_id,
+				record->space_id, record->index_id,
+				record->key_parts, record->key_part_count);
+		break;
 	case VY_LOG_CREATE_LSM:
 		rc = vy_recovery_create_lsm(recovery, record->lsm_id,
 				record->space_id, record->index_id,
@@ -2130,7 +2266,8 @@ vy_log_append_lsm(struct xlog *xlog, struct vy_lsm_recovery_info *lsm)
 	struct vy_log_record record;
 
 	vy_log_record_init(&record);
-	record.type = VY_LOG_CREATE_LSM;
+	record.type = lsm->create_lsn < 0 ?
+		VY_LOG_PREPARE_LSM : VY_LOG_CREATE_LSM;
 	record.lsm_id = lsm->id;
 	record.index_id = lsm->index_id;
 	record.space_id = lsm->space_id;
diff --git a/src/box/vy_log.h b/src/box/vy_log.h
index d77cb298..7672b8f9 100644
--- a/src/box/vy_log.h
+++ b/src/box/vy_log.h
@@ -65,8 +65,9 @@ struct mh_i64ptr_t;
 enum vy_log_record_type {
 	/**
 	 * Create a new LSM tree.
-	 * Requires vy_log_record::lsm_id, index_id, space_id,
-	 * key_def, create_lsn.
+	 * Requires vy_log_record::lsm_id, create_lsn.
+	 * After rotation, it also stores space_id, index_id, key_def,
+	 * create_lsn, modify_lsn, dump_lsn.
 	 */
 	VY_LOG_CREATE_LSM		= 0,
 	/**
@@ -179,6 +180,21 @@ enum vy_log_record_type {
 	 * from vylog on the next rotation.
 	 */
 	VY_LOG_FORGET_LSM		= 14,
+	/**
+	 * Prepare a new LSM tree for building.
+	 * Requires vy_log_record::lsm_id, index_id, space_id.
+	 *
+	 * Index ALTER operation consists of two stages. First, we
+	 * build a new LSM tree, checking constraints if necessary.
+	 * This is done before writing the operation to WAL. Then,
+	 * provided the first stage succeeded, we commit the LSM
+	 * tree to the metadata log.
+	 *
+	 * The following record is used to prepare a new LSM tree
+	 * for building. Once the index has been built, we write
+	 * a VY_LOG_CREATE_LSM record to commit it.
+	 */
+	VY_LOG_PREPARE_LSM		= 15,
 
 	vy_log_record_type_MAX
 };
@@ -273,7 +289,12 @@ struct vy_lsm_recovery_info {
 	struct key_part_def *key_parts;
 	/** Number of key parts. */
 	uint32_t key_part_count;
-	/** LSN of the WAL row that created the LSM tree. */
+	/**
+	 * LSN of the WAL row that created the LSM tree,
+	 * or -1 if the LSM tree was not committed to WAL
+	 * (that is there was an VY_LOG_PREPARE_LSM record
+	 * but no VY_LOG_CREATE_LSM).
+	 */
 	int64_t create_lsn;
 	/** LSN of the WAL row that last modified the LSM tree. */
 	int64_t modify_lsn;
@@ -295,6 +316,11 @@ struct vy_lsm_recovery_info {
 	 * vy_run_recovery_info::in_lsm.
 	 */
 	struct rlist runs;
+	/**
+	 * Pointer to an LSM tree that is going to replace
+	 * this one after successful ALTER.
+	 */
+	struct vy_lsm_recovery_info *prepared;
 };
 
 /** Vinyl range info stored in a recovery context. */
@@ -527,18 +553,29 @@ vy_log_record_init(struct vy_log_record *record)
 	memset(record, 0, sizeof(*record));
 }
 
-/** Helper to log a vinyl LSM tree creation. */
+/** Helper to log a vinyl LSM tree preparation. */
 static inline void
-vy_log_create_lsm(int64_t id, uint32_t space_id, uint32_t index_id,
-		  const struct key_def *key_def, int64_t create_lsn)
+vy_log_prepare_lsm(int64_t id, uint32_t space_id, uint32_t index_id,
+		   const struct key_def *key_def)
 {
 	struct vy_log_record record;
 	vy_log_record_init(&record);
-	record.type = VY_LOG_CREATE_LSM;
+	record.type = VY_LOG_PREPARE_LSM;
 	record.lsm_id = id;
 	record.space_id = space_id;
 	record.index_id = index_id;
 	record.key_def = key_def;
+	vy_log_write(&record);
+}
+
+/** Helper to log a vinyl LSM tree creation. */
+static inline void
+vy_log_create_lsm(int64_t id, int64_t create_lsn)
+{
+	struct vy_log_record record;
+	vy_log_record_init(&record);
+	record.type = VY_LOG_CREATE_LSM;
+	record.lsm_id = id;
 	record.create_lsn = create_lsn;
 	vy_log_write(&record);
 }
diff --git a/src/box/vy_lsm.c b/src/box/vy_lsm.c
index 289d5c40..fd352c8a 100644
--- a/src/box/vy_lsm.c
+++ b/src/box/vy_lsm.c
@@ -279,21 +279,6 @@ vy_lsm_delete(struct vy_lsm *lsm)
 	free(lsm);
 }
 
-/** Initialize the range tree of a new LSM tree. */
-static int
-vy_lsm_init_range_tree(struct vy_lsm *lsm)
-{
-	struct vy_range *range = vy_range_new(vy_log_next_id(), NULL, NULL,
-					      lsm->cmp_def);
-	if (range == NULL)
-		return -1;
-
-	assert(lsm->range_count == 0);
-	vy_lsm_add_range(lsm, range);
-	vy_lsm_acct_range(lsm, range);
-	return 0;
-}
-
 int
 vy_lsm_create(struct vy_lsm *lsm)
 {
@@ -327,12 +312,34 @@ vy_lsm_create(struct vy_lsm *lsm)
 		return -1;
 	}
 
-	/* Assign unique id. */
-	assert(lsm->id < 0);
-	lsm->id = vy_log_next_id();
+	/*
+	 * Allocate a unique id for the new LSM tree, but don't assign
+	 * it until information about the new LSM tree is successfully
+	 * written to vylog as vinyl_index_abort_create() uses id to
+	 * decide whether it needs to clean up.
+	 */
+	int64_t id = vy_log_next_id();
+
+	/* Create the initial range. */
+	struct vy_range *range = vy_range_new(vy_log_next_id(), NULL, NULL,
+					      lsm->cmp_def);
+	if (range == NULL)
+		return -1;
+	assert(lsm->range_count == 0);
+	vy_lsm_add_range(lsm, range);
+	vy_lsm_acct_range(lsm, range);
+
+	/* Write the new LSM tree record to vylog. */
+	vy_log_tx_begin();
+	vy_log_prepare_lsm(id, lsm->space_id, lsm->index_id, lsm->key_def);
+	vy_log_insert_range(id, range->id, NULL, NULL);
+	if (vy_log_tx_commit() < 0)
+		return -1;
 
-	/* Allocate initial range. */
-	return vy_lsm_init_range_tree(lsm);
+	/* Assign the id. */
+	assert(lsm->id < 0);
+	lsm->id = id;
+	return 0;
 }
 
 static struct vy_run *
@@ -505,7 +512,7 @@ vy_lsm_recover(struct vy_lsm *lsm, struct vy_recovery *recovery,
 	lsm_info = vy_recovery_lsm_by_index_id(recovery,
 			lsm->space_id, lsm->index_id);
 	if (is_checkpoint_recovery) {
-		if (lsm_info == NULL) {
+		if (lsm_info == NULL || lsm_info->create_lsn < 0) {
 			/*
 			 * All LSM trees created from snapshot rows must
 			 * be present in vylog, because snapshot can
@@ -528,16 +535,33 @@ vy_lsm_recover(struct vy_lsm *lsm, struct vy_recovery *recovery,
 		}
 	}
 
-	if (lsm_info == NULL || lsn > lsm_info->create_lsn) {
+	if (lsm_info == NULL || (lsm_info->prepared == NULL &&
+				 lsm_info->create_lsn >= 0 &&
+				 lsn > lsm_info->create_lsn)) {
 		/*
 		 * If we failed to log LSM tree creation before restart,
 		 * we won't find it in the log on recovery. This is OK as
 		 * the LSM tree doesn't have any runs in this case. We will
 		 * retry to log LSM tree in vinyl_index_commit_create().
 		 * For now, just create the initial range and assign id.
+		 *
+		 * Note, this is needed only for backward compatibility
+		 * since now we write VY_LOG_PREPARE_LSM before WAL write
+		 * and hence if the index was committed to WAL, it must be
+		 * present in vylog as well.
 		 */
-		lsm->id = vy_log_next_id();
-		return vy_lsm_init_range_tree(lsm);
+		return vy_lsm_create(lsm);
+	}
+
+	if (lsm_info->create_lsn >= 0 && lsn > lsm_info->create_lsn) {
+		/*
+		 * The index we are recovering was prepared, successfully
+		 * built, and committed to WAL, but it was not marked as
+		 * created in vylog. Recover the prepared LSM tree. We will
+		 * retry vylog write in vinyl_index_commit_create().
+		 */
+		lsm_info = lsm_info->prepared;
+		assert(lsm_info != NULL);
 	}
 
 	lsm->id = lsm_info->id;
@@ -554,7 +578,13 @@ vy_lsm_recover(struct vy_lsm *lsm, struct vy_recovery *recovery,
 		 * We need range tree initialized for all LSM trees,
 		 * even for dropped ones.
 		 */
-		return vy_lsm_init_range_tree(lsm);
+		struct vy_range *range = vy_range_new(vy_log_next_id(),
+						      NULL, NULL, lsm->cmp_def);
+		if (range == NULL)
+			return -1;
+		vy_lsm_add_range(lsm, range);
+		vy_lsm_acct_range(lsm, range);
+		return 0;
 	}
 
 	/*
diff --git a/src/box/vy_lsm.h b/src/box/vy_lsm.h
index 3f820fac..2e99e4d0 100644
--- a/src/box/vy_lsm.h
+++ b/src/box/vy_lsm.h
@@ -355,8 +355,8 @@ vy_lsm_update_pk(struct vy_lsm *lsm, struct vy_lsm *pk)
  *
  * This function is called when an LSM tree is created
  * after recovery is complete or during remote recovery.
- * It initializes the range tree and makes the LSM tree
- * directory.
+ * It initializes the range tree, makes the LSM tree
+ * directory, and writes the LSM tree record to vylog.
  */
 int
 vy_lsm_create(struct vy_lsm *lsm);
diff --git a/test/engine/truncate.result b/test/engine/truncate.result
index 3ad400e2..b4de787f 100644
--- a/test/engine/truncate.result
+++ b/test/engine/truncate.result
@@ -195,93 +195,6 @@ s:drop()
 ---
 ...
 --
--- Check that space truncation is linearizable.
---
--- Create a space with several indexes and start three fibers:
--- 1st and 3rd update the space, 2nd truncates it. Then wait
--- until all fibers are done. The space should contain data
--- inserted by the 3rd fiber.
---
--- Note, this is guaranteed to be true only if space updates
--- don't yield, which is always true for memtx and is true
--- for vinyl in case there's no data on disk, as in this case.
---
-s = box.schema.create_space('test', {engine = engine})
----
-...
-_ = s:create_index('i1', {parts = {1, 'unsigned'}})
----
-...
-_ = s:create_index('i2', {parts = {2, 'unsigned'}})
----
-...
-_ = s:create_index('i3', {parts = {3, 'string'}})
----
-...
-_ = s:insert{1, 1, 'a'}
----
-...
-_ = s:insert{2, 2, 'b'}
----
-...
-_ = s:insert{3, 3, 'c'}
----
-...
-c = fiber.channel(3)
----
-...
-test_run:cmd("setopt delimiter ';'")
----
-- true
-...
-fiber.create(function()
-    box.begin()
-    s:replace{1, 10, 'aa'}
-    s:replace{2, 20, 'bb'}
-    s:replace{3, 30, 'cc'}
-    box.commit()
-    c:put(true)
-end)
-fiber.create(function()
-    s:truncate()
-    c:put(true)
-end)
-fiber.create(function()
-    box.begin()
-    s:replace{1, 100, 'aaa'}
-    s:replace{2, 200, 'bbb'}
-    s:replace{3, 300, 'ccc'}
-    box.commit()
-    c:put(true)
-end)
-test_run:cmd("setopt delimiter ''");
----
-...
-for i = 1, 3 do c:get() end
----
-...
-s.index.i1:select()
----
-- - [1, 100, 'aaa']
-  - [2, 200, 'bbb']
-  - [3, 300, 'ccc']
-...
-s.index.i2:select()
----
-- - [1, 100, 'aaa']
-  - [2, 200, 'bbb']
-  - [3, 300, 'ccc']
-...
-s.index.i3:select()
----
-- - [1, 100, 'aaa']
-  - [2, 200, 'bbb']
-  - [3, 300, 'ccc']
-...
-s:drop()
----
-...
---
 -- Calling space.truncate concurrently.
 --
 s = box.schema.create_space('test', {engine = engine})
diff --git a/test/engine/truncate.test.lua b/test/engine/truncate.test.lua
index df2797a1..74fdd52b 100644
--- a/test/engine/truncate.test.lua
+++ b/test/engine/truncate.test.lua
@@ -82,54 +82,6 @@ s.index.i3:select()
 s:drop()
 
 --
--- Check that space truncation is linearizable.
---
--- Create a space with several indexes and start three fibers:
--- 1st and 3rd update the space, 2nd truncates it. Then wait
--- until all fibers are done. The space should contain data
--- inserted by the 3rd fiber.
---
--- Note, this is guaranteed to be true only if space updates
--- don't yield, which is always true for memtx and is true
--- for vinyl in case there's no data on disk, as in this case.
---
-s = box.schema.create_space('test', {engine = engine})
-_ = s:create_index('i1', {parts = {1, 'unsigned'}})
-_ = s:create_index('i2', {parts = {2, 'unsigned'}})
-_ = s:create_index('i3', {parts = {3, 'string'}})
-_ = s:insert{1, 1, 'a'}
-_ = s:insert{2, 2, 'b'}
-_ = s:insert{3, 3, 'c'}
-c = fiber.channel(3)
-test_run:cmd("setopt delimiter ';'")
-fiber.create(function()
-    box.begin()
-    s:replace{1, 10, 'aa'}
-    s:replace{2, 20, 'bb'}
-    s:replace{3, 30, 'cc'}
-    box.commit()
-    c:put(true)
-end)
-fiber.create(function()
-    s:truncate()
-    c:put(true)
-end)
-fiber.create(function()
-    box.begin()
-    s:replace{1, 100, 'aaa'}
-    s:replace{2, 200, 'bbb'}
-    s:replace{3, 300, 'ccc'}
-    box.commit()
-    c:put(true)
-end)
-test_run:cmd("setopt delimiter ''");
-for i = 1, 3 do c:get() end
-s.index.i1:select()
-s.index.i2:select()
-s.index.i3:select()
-s:drop()
-
---
 -- Calling space.truncate concurrently.
 --
 s = box.schema.create_space('test', {engine = engine})
diff --git a/test/vinyl/errinj_vylog.result b/test/vinyl/errinj_vylog.result
index f78201c9..ca23cb45 100644
--- a/test/vinyl/errinj_vylog.result
+++ b/test/vinyl/errinj_vylog.result
@@ -67,7 +67,7 @@ s:drop()
 ---
 ...
 --
--- Check that an index drop/truncate/create record we failed to
+-- Check that an index drop/create record we failed to
 -- write to vylog is flushed along with the next record.
 --
 fiber = require 'fiber'
@@ -76,62 +76,57 @@ fiber = require 'fiber'
 s1 = box.schema.space.create('test1', {engine = 'vinyl'})
 ---
 ...
-_ = s1:create_index('pk')
----
-...
-_ = s1:insert{1, 'a'}
----
-...
 s2 = box.schema.space.create('test2', {engine = 'vinyl'})
 ---
 ...
 _ = s2:create_index('pk')
 ---
 ...
-_ = s2:insert{2, 'b'}
+_ = s2:insert{1, 'a'}
 ---
 ...
 box.snapshot()
 ---
 - ok
 ...
-_ = s1:insert{3, 'c'}
----
-...
-_ = s2:insert{4, 'd'}
+_ = s2:insert{2, 'b'}
 ---
 ...
-SCHED_TIMEOUT = 0.01
+box.error.injection.set('ERRINJ_WAL_DELAY', true)
 ---
+- ok
 ...
-box.error.injection.set('ERRINJ_VY_SCHED_TIMEOUT', SCHED_TIMEOUT)
+-- VY_LOG_PREPARE_LSM written, but VY_LOG_CREATE_LSM missing
+ch = fiber.channel(1)
 ---
-- ok
 ...
-box.error.injection.set('ERRINJ_VY_LOG_FLUSH', true);
+_ = fiber.create(function() s1:create_index('pk') ch:put(true) end)
 ---
-- ok
 ...
-s1:drop()
+fiber.sleep(0.001)
 ---
 ...
-s2:truncate()
+box.error.injection.set('ERRINJ_VY_LOG_FLUSH', true)
 ---
+- ok
 ...
-_ = s2:insert{5, 'e'}
+box.error.injection.set('ERRINJ_WAL_DELAY', false)
 ---
+- ok
 ...
-s3 = box.schema.space.create('test3', {engine = 'vinyl'})
+ch:get()
 ---
+- true
 ...
-_ = s3:create_index('pk')
+_ = s1:insert{3, 'c'}
 ---
 ...
-_ = s3:insert{6, 'f'}
+-- VY_LOG_DROP_LSM missing
+s2.index.pk:drop()
 ---
 ...
 -- pending records must not be rolled back on error
-box.snapshot()
+s2:create_index('pk') -- error
 ---
 - error: Error injection 'vinyl log flush'
 ...
@@ -139,65 +134,46 @@ box.error.injection.set('ERRINJ_VY_LOG_FLUSH', false);
 ---
 - ok
 ...
-fiber.sleep(2 * SCHED_TIMEOUT) -- wait for scheduler to unthrottle
+_ = s1:insert{4, 'd'}
 ---
 ...
-box.error.injection.set('ERRINJ_VY_SCHED_TIMEOUT', 0)
----
-- ok
-...
-box.snapshot()
----
-- ok
-...
-_ = s2:insert{7, 'g'}
+_ = s2:create_index('pk')
 ---
 ...
-_ = s3:insert{8, 'h'}
+_ = s2:insert{5, 'e'}
 ---
 ...
 test_run:cmd('restart server default')
 s1 = box.space.test1
 ---
 ...
-s1 == nil
+s2 = box.space.test2
 ---
-- true
 ...
-s2 = box.space.test2
+s1:select()
 ---
+- - [3, 'c']
+  - [4, 'd']
 ...
 s2:select()
 ---
 - - [5, 'e']
-  - [7, 'g']
-...
-s2:drop()
----
-...
-s3 = box.space.test3
----
 ...
-s3:select()
+s1:drop()
 ---
-- - [6, 'f']
-  - [8, 'h']
 ...
-s3:drop()
+s2:drop()
 ---
 ...
 --
--- Check that if a buffered index drop/truncate/create record
--- does not make it to the vylog before restart, it will be
--- replayed on recovery.
+-- Check that if a buffered index drop/create record does not
+-- make it to the vylog before restart, it will be replayed on
+-- recovery.
 --
-s1 = box.schema.space.create('test1', {engine = 'vinyl'})
----
-...
-_ = s1:create_index('pk')
+fiber = require 'fiber'
 ---
 ...
-_ = s1:insert{111, 'aaa'}
+s1 = box.schema.space.create('test1', {engine = 'vinyl'})
 ---
 ...
 s2 = box.schema.space.create('test2', {engine = 'vinyl'})
@@ -206,82 +182,97 @@ s2 = box.schema.space.create('test2', {engine = 'vinyl'})
 _ = s2:create_index('pk')
 ---
 ...
-_ = s2:insert{222, 'bbb'}
+_ = s2:insert{111, 'aaa'}
 ---
 ...
 box.snapshot()
 ---
 - ok
 ...
-_ = s1:insert{333, 'ccc'}
----
-...
-_ = s2:insert{444, 'ddd'}
+_ = s2:insert{222, 'bbb'}
 ---
 ...
-box.error.injection.set('ERRINJ_VY_LOG_FLUSH', true);
+box.error.injection.set('ERRINJ_WAL_DELAY', true)
 ---
 - ok
 ...
-s1:drop()
+-- VY_LOG_PREPARE_LSM written, but VY_LOG_CREATE_LSM missing
+ch = fiber.channel(1)
 ---
 ...
-s2:truncate()
+_ = fiber.create(function() s1:create_index('pk') ch:put(true) end)
 ---
 ...
-_ = s2:insert{555, 'eee'}
+fiber.sleep(0.001)
 ---
 ...
-s3 = box.schema.space.create('test3', {engine = 'vinyl'})
+box.error.injection.set('ERRINJ_VY_LOG_FLUSH', true)
 ---
+- ok
 ...
-_ = s3:create_index('pk')
+box.error.injection.set('ERRINJ_WAL_DELAY', false)
 ---
+- ok
 ...
-_ = s3:insert{666, 'fff'}
+ch:get()
 ---
+- true
 ...
--- gh-2532: replaying create/drop from xlog crashes tarantool
-test_run:cmd("setopt delimiter ';'")
+_ = s1:insert{333, 'ccc'}
 ---
-- true
 ...
-for i = 1, 10 do
-    s = box.schema.space.create('test', {engine = 'vinyl'})
-    s:create_index('primary')
-    s:create_index('secondary', {unique = false, parts = {2, 'string'}})
-    s:insert{i, 'test' .. i}
-    s:truncate()
-    s:drop()
-end
-test_run:cmd("setopt delimiter ''");
+-- VY_LOG_DROP_LSM missing
+s2.index.pk:drop()
 ---
 ...
 test_run:cmd('restart server default')
 s1 = box.space.test1
 ---
 ...
-s1 == nil
+s2 = box.space.test2
 ---
-- true
 ...
-s2 = box.space.test2
+_ = s1:insert{444, 'ddd'}
+---
+...
+_ = s2:create_index('pk')
 ---
 ...
+_ = s2:insert{555, 'eee'}
+---
+...
+s1:select()
+---
+- - [333, 'ccc']
+  - [444, 'ddd']
+...
 s2:select()
 ---
 - - [555, 'eee']
 ...
-s2:drop()
+box.snapshot()
 ---
+- ok
 ...
-s3 = box.space.test3
+test_run:cmd('restart server default')
+s1 = box.space.test1
 ---
 ...
-s3:select()
+s2 = box.space.test2
 ---
-- - [666, 'fff']
 ...
-s3:drop()
+s1:select()
+---
+- - [333, 'ccc']
+  - [444, 'ddd']
+...
+s2:select()
+---
+- - [555, 'eee']
+...
+s1:drop()
+---
+...
+s2:drop()
 ---
 ...
diff --git a/test/vinyl/errinj_vylog.test.lua b/test/vinyl/errinj_vylog.test.lua
index 36b3659d..3d90755d 100644
--- a/test/vinyl/errinj_vylog.test.lua
+++ b/test/vinyl/errinj_vylog.test.lua
@@ -32,111 +32,101 @@ s:select()
 s:drop()
 
 --
--- Check that an index drop/truncate/create record we failed to
+-- Check that an index drop/create record we failed to
 -- write to vylog is flushed along with the next record.
 --
 fiber = require 'fiber'
 
 s1 = box.schema.space.create('test1', {engine = 'vinyl'})
-_ = s1:create_index('pk')
-_ = s1:insert{1, 'a'}
-
 s2 = box.schema.space.create('test2', {engine = 'vinyl'})
 _ = s2:create_index('pk')
+_ = s2:insert{1, 'a'}
+box.snapshot()
 _ = s2:insert{2, 'b'}
 
-box.snapshot()
+box.error.injection.set('ERRINJ_WAL_DELAY', true)
 
-_ = s1:insert{3, 'c'}
-_ = s2:insert{4, 'd'}
+-- VY_LOG_PREPARE_LSM written, but VY_LOG_CREATE_LSM missing
+ch = fiber.channel(1)
+_ = fiber.create(function() s1:create_index('pk') ch:put(true) end)
+fiber.sleep(0.001)
 
-SCHED_TIMEOUT = 0.01
-box.error.injection.set('ERRINJ_VY_SCHED_TIMEOUT', SCHED_TIMEOUT)
-box.error.injection.set('ERRINJ_VY_LOG_FLUSH', true);
+box.error.injection.set('ERRINJ_VY_LOG_FLUSH', true)
+box.error.injection.set('ERRINJ_WAL_DELAY', false)
 
-s1:drop()
-s2:truncate()
-_ = s2:insert{5, 'e'}
+ch:get()
+_ = s1:insert{3, 'c'}
 
-s3 = box.schema.space.create('test3', {engine = 'vinyl'})
-_ = s3:create_index('pk')
-_ = s3:insert{6, 'f'}
+-- VY_LOG_DROP_LSM missing
+s2.index.pk:drop()
 
 -- pending records must not be rolled back on error
-box.snapshot()
+s2:create_index('pk') -- error
 
 box.error.injection.set('ERRINJ_VY_LOG_FLUSH', false);
-fiber.sleep(2 * SCHED_TIMEOUT) -- wait for scheduler to unthrottle
-box.error.injection.set('ERRINJ_VY_SCHED_TIMEOUT', 0)
-
-box.snapshot()
 
-_ = s2:insert{7, 'g'}
-_ = s3:insert{8, 'h'}
+_ = s1:insert{4, 'd'}
+_ = s2:create_index('pk')
+_ = s2:insert{5, 'e'}
 
 test_run:cmd('restart server default')
 
 s1 = box.space.test1
-s1 == nil
-
 s2 = box.space.test2
+s1:select()
 s2:select()
+s1:drop()
 s2:drop()
 
-s3 = box.space.test3
-s3:select()
-s3:drop()
-
 --
--- Check that if a buffered index drop/truncate/create record
--- does not make it to the vylog before restart, it will be
--- replayed on recovery.
+-- Check that if a buffered index drop/create record does not
+-- make it to the vylog before restart, it will be replayed on
+-- recovery.
 --
+fiber = require 'fiber'
 
 s1 = box.schema.space.create('test1', {engine = 'vinyl'})
-_ = s1:create_index('pk')
-_ = s1:insert{111, 'aaa'}
-
 s2 = box.schema.space.create('test2', {engine = 'vinyl'})
 _ = s2:create_index('pk')
+_ = s2:insert{111, 'aaa'}
+box.snapshot()
 _ = s2:insert{222, 'bbb'}
 
-box.snapshot()
+box.error.injection.set('ERRINJ_WAL_DELAY', true)
+
+-- VY_LOG_PREPARE_LSM written, but VY_LOG_CREATE_LSM missing
+ch = fiber.channel(1)
+_ = fiber.create(function() s1:create_index('pk') ch:put(true) end)
+fiber.sleep(0.001)
+
+box.error.injection.set('ERRINJ_VY_LOG_FLUSH', true)
+box.error.injection.set('ERRINJ_WAL_DELAY', false)
 
+ch:get()
 _ = s1:insert{333, 'ccc'}
-_ = s2:insert{444, 'ddd'}
 
-box.error.injection.set('ERRINJ_VY_LOG_FLUSH', true);
+-- VY_LOG_DROP_LSM missing
+s2.index.pk:drop()
 
-s1:drop()
-s2:truncate()
+test_run:cmd('restart server default')
+
+s1 = box.space.test1
+s2 = box.space.test2
+
+_ = s1:insert{444, 'ddd'}
+_ = s2:create_index('pk')
 _ = s2:insert{555, 'eee'}
 
-s3 = box.schema.space.create('test3', {engine = 'vinyl'})
-_ = s3:create_index('pk')
-_ = s3:insert{666, 'fff'}
-
--- gh-2532: replaying create/drop from xlog crashes tarantool
-test_run:cmd("setopt delimiter ';'")
-for i = 1, 10 do
-    s = box.schema.space.create('test', {engine = 'vinyl'})
-    s:create_index('primary')
-    s:create_index('secondary', {unique = false, parts = {2, 'string'}})
-    s:insert{i, 'test' .. i}
-    s:truncate()
-    s:drop()
-end
-test_run:cmd("setopt delimiter ''");
+s1:select()
+s2:select()
+
+box.snapshot()
 
 test_run:cmd('restart server default')
 
 s1 = box.space.test1
-s1 == nil
-
 s2 = box.space.test2
+s1:select()
 s2:select()
+s1:drop()
 s2:drop()
-
-s3 = box.space.test3
-s3:select()
-s3:drop()
-- 
2.11.0

[PATCH v2 4/8] vinyl: bump mem version after committing statement

[Vladimir Davydov]

Since commit 1e1c1fdbeddb ("vinyl: make read iterator always return
newest tuple version") vinyl read iterator guarantees that any tuple it
returns is the newest version in the iterator read view. However, if we
don't bump mem version after assigning LSN to a mem statement, a read
iterator using committed_read_view might not see it and return a stale
tuple. Currently, there's no code that relies on this iterator feature,
but we will need it for building new indexes. Without this patch, build
(introduced later in the series) might return inconsistent results.

Needed for #1653
---
 src/box/vy_mem.c   | 1 +
 test/unit/vy_mem.c | 4 ++--
 2 files changed, 3 insertions(+), 2 deletions(-)

diff --git a/src/box/vy_mem.c b/src/box/vy_mem.c
index 65e31ea5..faa2c06f 100644
--- a/src/box/vy_mem.c
+++ b/src/box/vy_mem.c
@@ -250,6 +250,7 @@ vy_mem_commit_stmt(struct vy_mem *mem, const struct tuple *stmt)
 	assert(mem->min_lsn <= lsn);
 	if (mem->max_lsn < lsn)
 		mem->max_lsn = lsn;
+	mem->version++;
 }
 
 void
diff --git a/test/unit/vy_mem.c b/test/unit/vy_mem.c
index 6641dc0c..6666f94c 100644
--- a/test/unit/vy_mem.c
+++ b/test/unit/vy_mem.c
@@ -51,9 +51,9 @@ test_basic(void)
 
 	/* Check version  */
 	stmt = vy_mem_insert_template(mem, &stmts[4]);
-	is(mem->version, 6, "vy_mem->version")
+	is(mem->version, 8, "vy_mem->version")
 	vy_mem_commit_stmt(mem, stmt);
-	is(mem->version, 6, "vy_mem->version")
+	is(mem->version, 9, "vy_mem->version")
 
 	/* Clean up */
 	vy_mem_delete(mem);
-- 
2.11.0

[PATCH v2 5/8] vinyl: allow to commit statements to mem in arbitrary order

[Vladimir Davydov]

vy_mem_commit_stmt() expects statements to be committed in the order
of increasing LSN. Although this condition holds now, it won't once
we start using this function for building indexes. So let's remove
this limitation.

Needed for #1653
---
 src/box/vy_mem.c | 7 ++-----
 1 file changed, 2 insertions(+), 5 deletions(-)

diff --git a/src/box/vy_mem.c b/src/box/vy_mem.c
index faa2c06f..3954bf3a 100644
--- a/src/box/vy_mem.c
+++ b/src/box/vy_mem.c
@@ -245,11 +245,8 @@ vy_mem_commit_stmt(struct vy_mem *mem, const struct tuple *stmt)
 	/* The statement must be from a lsregion. */
 	assert(!vy_stmt_is_refable(stmt));
 	int64_t lsn = vy_stmt_lsn(stmt);
-	if (mem->min_lsn == INT64_MAX)
-		mem->min_lsn = lsn;
-	assert(mem->min_lsn <= lsn);
-	if (mem->max_lsn < lsn)
-		mem->max_lsn = lsn;
+	mem->min_lsn = MIN(mem->min_lsn, lsn);
+	mem->max_lsn = MAX(mem->max_lsn, lsn);
 	mem->version++;
 }
 
-- 
2.11.0

[PATCH v2 6/8] vinyl: relax limitation imposed on run min/max lsn

[Vladimir Davydov]

Currently, we assume that no two runs of the same range intersect by
LSN. This holds, because LSNs grow strictly monotonically, and no
transaction may be split between two runs (as we pin each affected
vy_mem until the transaction is complete). We ensure this with an
assertion in vy_task_dump_complete.

However, when building a new index we can't increment tx_manager->lsn so
there may be multiple statements with the same LSN. This is OK as for
each particular key, two statements will still have different LSNs, but
this may break the assertion in vy_task_dump_complete in case dump
occurs while build is in progress.

To avoid that, let's relax the condition under the assertion and assume
that a run's max_lsn may be equal min_lsn of the newer run.

Needed for #1653
---
 src/box/vy_scheduler.c | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/box/vy_scheduler.c b/src/box/vy_scheduler.c
index d1545e76..eff3a814 100644
--- a/src/box/vy_scheduler.c
+++ b/src/box/vy_scheduler.c
@@ -731,7 +731,7 @@ vy_task_dump_complete(struct vy_scheduler *scheduler, struct vy_task *task)
 		goto delete_mems;
 	}
 
-	assert(new_run->info.min_lsn > lsm->dump_lsn);
+	assert(new_run->info.min_lsn >= lsm->dump_lsn);
 	assert(new_run->info.max_lsn <= dump_lsn);
 
 	/*
-- 
2.11.0

[PATCH v2 7/8] vinyl: factor out vy_check_is_unique_secondary

[Vladimir Davydov]

We need to check unique constraint when building a new index. So let's
factor out this helper function and pass space_name, index_name, and
read view to it explicitly (because index_name_by_id isn't going to work
for an index that is under construction and there's no tx when we are
building a new index). Suggested by @Gerold103.

Needed for #1653
---
 src/box/vinyl.c | 81 +++++++++++++++++++++++++++++++++++++++------------------
 1 file changed, 56 insertions(+), 25 deletions(-)

diff --git a/src/box/vinyl.c b/src/box/vinyl.c
index 4d670328..41ffa42c 100644
--- a/src/box/vinyl.c
+++ b/src/box/vinyl.c
@@ -1353,7 +1353,9 @@ vy_lsm_get(struct vy_lsm *lsm, struct vy_tx *tx,
  * a duplicate key error in the diagnostics area.
  * @param env        Vinyl environment.
  * @param tx         Current transaction.
- * @param space      Target space.
+ * @param rv         Read view.
+ * @param space_name Space name.
+ * @param index_name Index name.
  * @param lsm        LSM tree in which to search.
  * @param key        Key statement.
  *
@@ -1361,7 +1363,9 @@ vy_lsm_get(struct vy_lsm *lsm, struct vy_tx *tx,
  * @retval -1 Memory error or the key is found.
  */
 static inline int
-vy_check_is_unique(struct vy_env *env, struct vy_tx *tx, struct space *space,
+vy_check_is_unique(struct vy_env *env, struct vy_tx *tx,
+		   const struct vy_read_view **rv,
+		   const char *space_name, const char *index_name,
 		   struct vy_lsm *lsm, struct tuple *key)
 {
 	struct tuple *found;
@@ -1371,20 +1375,57 @@ vy_check_is_unique(struct vy_env *env, struct vy_tx *tx, struct space *space,
 	 */
 	if (env->status != VINYL_ONLINE)
 		return 0;
-	if (vy_lsm_get(lsm, tx, vy_tx_read_view(tx), key, &found))
+	if (vy_lsm_get(lsm, tx, rv, key, &found))
 		return -1;
 
 	if (found) {
 		tuple_unref(found);
 		diag_set(ClientError, ER_TUPLE_FOUND,
-			 index_name_by_id(space, lsm->index_id),
-			 space_name(space));
+			 index_name, space_name);
 		return -1;
 	}
 	return 0;
 }
 
 /**
+ * Check if insertion of a new tuple violates unique constraint
+ * of a secondary index.
+ * @param env        Vinyl environment.
+ * @param tx         Current transaction.
+ * @param rv         Read view.
+ * @param space_name Space name.
+ * @param index_name Index name.
+ * @param lsm        LSM tree corresponding to the index.
+ * @param stmt       New tuple.
+ *
+ * @retval  0 Success, unique constraint is satisfied.
+ * @retval -1 Duplicate is found or read error occurred.
+ */
+static int
+vy_check_is_unique_secondary(struct vy_env *env, struct vy_tx *tx,
+			     const struct vy_read_view **rv,
+			     const char *space_name, const char *index_name,
+			     struct vy_lsm *lsm, const struct tuple *stmt)
+{
+	assert(lsm->index_id > 0);
+	struct key_def *def = lsm->key_def;
+	if (lsm->check_is_unique &&
+	    !key_update_can_be_skipped(def->column_mask,
+				       vy_stmt_column_mask(stmt)) &&
+	    (!def->is_nullable || !vy_tuple_key_contains_null(stmt, def))) {
+		struct tuple *key = vy_stmt_extract_key(stmt, def,
+							lsm->env->key_format);
+		if (key == NULL)
+			return -1;
+		int rc = vy_check_is_unique(env, tx, rv, space_name,
+					    index_name, lsm, key);
+		tuple_unref(key);
+		return rc;
+	}
+	return 0;
+}
+
+/**
  * Insert a tuple in a primary index LSM tree.
  * @param env   Vinyl environment.
  * @param tx    Current transaction.
@@ -1407,7 +1448,9 @@ vy_insert_primary(struct vy_env *env, struct vy_tx *tx, struct space *space,
 	 * conflict with existing tuples.
 	 */
 	if (pk->check_is_unique &&
-	    vy_check_is_unique(env, tx, space, pk, stmt) != 0)
+	    vy_check_is_unique(env, tx, vy_tx_read_view(tx), space_name(space),
+			       index_name_by_id(space, pk->index_id),
+			       pk, stmt) != 0)
 		return -1;
 	return vy_tx_set(tx, pk, stmt);
 }
@@ -1431,25 +1474,13 @@ vy_insert_secondary(struct vy_env *env, struct vy_tx *tx, struct space *space,
 	       vy_stmt_type(stmt) == IPROTO_REPLACE);
 	assert(tx != NULL && tx->state == VINYL_TX_READY);
 	assert(lsm->index_id > 0);
-	/*
-	 * If the index is unique then the new tuple must not
-	 * conflict with existing tuples. If the index is not
-	 * unique a conflict is impossible.
-	 */
-	if (lsm->check_is_unique &&
-	    !key_update_can_be_skipped(lsm->key_def->column_mask,
-				       vy_stmt_column_mask(stmt)) &&
-	    (!lsm->key_def->is_nullable ||
-	     !vy_tuple_key_contains_null(stmt, lsm->key_def))) {
-		struct tuple *key = vy_stmt_extract_key(stmt, lsm->key_def,
-							lsm->env->key_format);
-		if (key == NULL)
-			return -1;
-		int rc = vy_check_is_unique(env, tx, space, lsm, key);
-		tuple_unref(key);
-		if (rc != 0)
-			return -1;
-	}
+
+	if (vy_check_is_unique_secondary(env, tx, vy_tx_read_view(tx),
+					 space_name(space),
+					 index_name_by_id(space, lsm->index_id),
+					 lsm, stmt) != 0)
+		return -1;
+
 	/*
 	 * We must always append the statement to transaction write set
 	 * of each LSM tree, even if operation itself does not update
-- 
2.11.0

[PATCH v2 8/8] vinyl: allow to build secondary index for non-empty space

[Vladimir Davydov]

This patch implements space_vtab::build_index callback for vinyl spaces.
Now instead of returning an error in case the space is not empty, the
callback will actually try to build a new index. The build procedure
consists of four steps:

 1. Prepare the LSM tree for building. This implies writing a special
    record to vylog, VY_LOG_PREPARE_LSM, and adding the new index to
    the vinyl scheduler so that it can be dumped during build. We need
    to log the new LSM tree so that we can keep track of run files
    created for it during build and remove them if build procedure
    fails.

 2. Inserting tuples stored in the space into the new LSM tree. Since
    there may concurrent DML requests, we install a trigger to forward
    them to the new index.

 3. Dumping the index to disk so that we don't have to rebuild it after
    recovery.

 4. Committing the new LSM tree in vylog (VY_LOG_CREATE_LSM).

Steps 1-3 are done from the space_vtab::build_index callback while
step 4 is done after WAL write, from index_vtab::commit_create.

While step 3 is being performed, new DML requests may be executed for
the altered space. Those requests will be reflected in the new index
thanks to the on_replace trigger, however they won't be recovered during
WAL recovery as they will appear in WAL before the ALTER record that
created the index. To recover them, we replay all statements stored in
the primary key's memory level when replaying the ALTER record during
WAL recovery.

Closes #1653
---
 src/box/vinyl.c                  | 454 +++++++++++++++++++++++---
 src/box/vy_lsm.c                 |  14 +-
 src/box/vy_quota.h               |  10 +
 src/box/vy_scheduler.c           |  29 ++
 src/box/vy_scheduler.h           |   7 +
 test/box/alter.result            | 501 -----------------------------
 test/box/alter.test.lua          | 159 ---------
 test/engine/ddl.result           | 672 +++++++++++++++++++++++++++++++++++++++
 test/engine/ddl.test.lua         | 221 +++++++++++++
 test/vinyl/ddl.result            | 446 +++++++++++++++-----------
 test/vinyl/ddl.test.lua          | 220 ++++++++-----
 test/vinyl/errinj.result         | 256 +++++++++++++++
 test/vinyl/errinj.test.lua       | 115 +++++++
 test/vinyl/errinj_gc.result      |  78 ++++-
 test/vinyl/errinj_gc.test.lua    |  38 ++-
 test/vinyl/errinj_vylog.result   |  57 ++++
 test/vinyl/errinj_vylog.test.lua |  29 ++
 test/vinyl/gh.result             |   2 +-
 18 files changed, 2334 insertions(+), 974 deletions(-)

diff --git a/src/box/vinyl.c b/src/box/vinyl.c
index 41ffa42c..4c7ed77b 100644
--- a/src/box/vinyl.c
+++ b/src/box/vinyl.c
@@ -1164,52 +1164,6 @@ vinyl_space_drop_primary_key(struct space *space)
 	(void)space;
 }
 
-static int
-vinyl_space_build_index(struct space *src_space, struct index *new_index,
-			struct tuple_format *new_format)
-{
-	(void)new_format;
-
-	struct vy_env *env = vy_env(src_space->engine);
-	struct vy_lsm *pk = vy_lsm(src_space->index[0]);
-
-	/*
-	 * During local recovery we are loading existing indexes
-	 * from disk, not building new ones.
-	 */
-	if (env->status != VINYL_INITIAL_RECOVERY_LOCAL &&
-	    env->status != VINYL_FINAL_RECOVERY_LOCAL) {
-		if (pk->stat.disk.count.rows != 0 ||
-		    pk->stat.memory.count.rows != 0) {
-			diag_set(ClientError, ER_UNSUPPORTED, "Vinyl",
-				 "building an index for a non-empty space");
-			return -1;
-		}
-	}
-
-	/*
-	 * Unlike Memtx, Vinyl does not need building of a secondary index.
-	 * This is true because of two things:
-	 * 1) Vinyl does not support alter of non-empty spaces
-	 * 2) During recovery a Vinyl index already has all needed data on disk.
-	 * And there are 3 cases:
-	 * I. The secondary index is added in snapshot. Then Vinyl was
-	 * snapshotted too and all necessary for that moment data is on disk.
-	 * II. The secondary index is added in WAL. That means that vinyl
-	 * space had no data at that point and had nothing to build. The
-	 * index actually could contain recovered data, but it will handle it
-	 * by itself during WAL recovery.
-	 * III. Vinyl is online. The space is definitely empty and there's
-	 * nothing to build.
-	 */
-	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), pk);
-	return 0;
-}
-
 static size_t
 vinyl_space_bsize(struct space *space)
 {
@@ -4072,6 +4026,414 @@ vinyl_index_get(struct index *index, const char *key,
 
 /*** }}} Cursor */
 
+/* {{{ Index build */
+
+/** Argument passed to vy_build_on_replace(). */
+struct vy_build_ctx {
+	/** Vinyl environment. */
+	struct vy_env *env;
+	/** LSM tree under construction. */
+	struct vy_lsm *lsm;
+	/** Format to check new tuples against. */
+	struct tuple_format *format;
+	/**
+	 * Names of the altered space and the new index.
+	 * Used for error reporting.
+	 */
+	const char *space_name;
+	const char *index_name;
+	/** Set in case a build error occurred. */
+	bool is_failed;
+	/** Container for storing errors. */
+	struct diag diag;
+};
+
+/**
+ * This is an on_replace trigger callback that forwards DML requests
+ * to the index that is currently being built.
+ */
+static void
+vy_build_on_replace(struct trigger *trigger, void *event)
+{
+	struct txn *txn = event;
+	struct txn_stmt *stmt = txn_current_stmt(txn);
+	struct vy_build_ctx *ctx = trigger->data;
+	struct vy_tx *tx = txn->engine_tx;
+	struct tuple_format *format = ctx->format;
+	struct vy_lsm *lsm = ctx->lsm;
+
+	if (ctx->is_failed)
+		return; /* already failed, nothing to do */
+
+	/* Check new tuples for conformity to the new format. */
+	if (stmt->new_tuple != NULL &&
+	    tuple_validate(format, stmt->new_tuple) != 0)
+		goto err;
+
+	/* Check key uniqueness if necessary. */
+	if (stmt->new_tuple != NULL &&
+	    vy_check_is_unique_secondary(ctx->env, tx, vy_tx_read_view(tx),
+					 ctx->space_name, ctx->index_name,
+					 lsm, stmt->new_tuple) != 0)
+		goto err;
+
+	/* Forward the statement to the new LSM tree. */
+	if (stmt->old_tuple != NULL) {
+		struct tuple *delete = vy_stmt_new_surrogate_delete(format,
+							stmt->old_tuple);
+		if (delete == NULL)
+			goto err;
+		int rc = vy_tx_set(tx, lsm, delete);
+		tuple_unref(delete);
+		if (rc != 0)
+			goto err;
+	}
+	if (stmt->new_tuple != NULL) {
+		uint32_t data_len;
+		const char *data = tuple_data_range(stmt->new_tuple, &data_len);
+		struct tuple *insert = vy_stmt_new_insert(format, data,
+							  data + data_len);
+		if (insert == NULL)
+			goto err;
+		int rc = vy_tx_set(tx, lsm, insert);
+		tuple_unref(insert);
+		if (rc != 0)
+			goto err;
+	}
+	return;
+err:
+	ctx->is_failed = true;
+	diag_move(diag_get(), &ctx->diag);
+}
+
+/**
+ * Insert a single statement into the LSM tree that is currently
+ * being built.
+ */
+static int
+vy_build_insert_stmt(struct vy_lsm *lsm, struct vy_mem *mem,
+		     const struct tuple *stmt, int64_t lsn)
+{
+	const struct tuple *region_stmt = vy_stmt_dup_lsregion(stmt,
+				&mem->env->allocator, mem->generation);
+	if (region_stmt == NULL)
+		return -1;
+	vy_stmt_set_lsn((struct tuple *)region_stmt, lsn);
+	if (vy_mem_insert(mem, region_stmt) != 0)
+		return -1;
+	vy_mem_commit_stmt(mem, region_stmt);
+	vy_stmt_counter_acct_tuple(&lsm->stat.memory.count, region_stmt);
+	return 0;
+}
+
+/**
+ * Insert a tuple fetched from the space into the LSM tree that
+ * is currently being built.
+ */
+static int
+vy_build_insert_tuple(struct vy_env *env, struct vy_lsm *lsm,
+		      const char *space_name, const char *index_name,
+		      struct tuple_format *new_format, struct tuple *tuple)
+{
+	int rc;
+	/*
+	 * Use the last LSN to the time, because read iterator
+	 * guarantees that this is the newest tuple version.
+	 */
+	int64_t lsn = env->xm->lsn;
+	struct vy_mem *mem = lsm->mem;
+
+	/* Check the tuple against the new space format. */
+	if (tuple_validate(new_format, tuple) != 0)
+		return -1;
+
+	/*
+	 * Check unique constraint if necessary.
+	 *
+	 * Note, this operation may yield, which opens a time
+	 * window for a concurrent fiber to insert a newer tuple
+	 * version. It's OK - we won't overwrite it, because the
+	 * LSN we use is less. However, we do need to make sure
+	 * we insert the tuple into the in-memory index that was
+	 * active before the yield, otherwise we might break the
+	 * invariant according to which newer in-memory indexes
+	 * store statements with greater LSNs. So we pin the
+	 * in-memory index that is active now and insert the tuple
+	 * into it after the yield.
+	 */
+	vy_mem_pin(mem);
+	rc = vy_check_is_unique_secondary(env, NULL,
+			&env->xm->p_committed_read_view,
+			space_name, index_name, lsm, tuple);
+	vy_mem_unpin(mem);
+	if (rc != 0)
+		return -1;
+
+	/* Reallocate the new tuple using the new space format. */
+	uint32_t data_len;
+	const char *data = tuple_data_range(tuple, &data_len);
+	struct tuple *stmt = vy_stmt_new_replace(new_format, data,
+						 data + data_len);
+	if (stmt == NULL)
+		return -1;
+
+	/* Insert the new tuple into the in-memory index. */
+	size_t mem_used_before = lsregion_used(&env->mem_env.allocator);
+	rc = vy_build_insert_stmt(lsm, mem, stmt, lsn);
+	tuple_unref(stmt);
+
+	/* Consume memory quota. Throttle if it is exceeded. */
+	size_t mem_used_after = lsregion_used(&env->mem_env.allocator);
+	assert(mem_used_after >= mem_used_before);
+	vy_quota_force_use(&env->quota, mem_used_after - mem_used_before);
+	vy_quota_wait(&env->quota);
+	return rc;
+}
+
+/**
+ * Recover a single statement that was inserted into the space
+ * while the newly built index was dumped to disk.
+ */
+static int
+vy_build_recover_stmt(struct vy_lsm *lsm, struct vy_lsm *pk,
+		      const struct tuple *mem_stmt)
+{
+	int64_t lsn = vy_stmt_lsn(mem_stmt);
+	if (lsn <= lsm->dump_lsn)
+		return 0; /* statement was dumped, nothing to do */
+
+	/* Lookup the tuple that was affected by this statement. */
+	const struct vy_read_view rv = { .vlsn = lsn - 1 };
+	const struct vy_read_view *p_rv = &rv;
+	struct tuple *old_tuple;
+	if (vy_point_lookup(pk, NULL, &p_rv, (struct tuple *)mem_stmt,
+			    &old_tuple) != 0)
+		return -1;
+	/*
+	 * Create DELETE + INSERT statements corresponding to
+	 * the given statement in the secondary index.
+	 */
+	struct tuple *delete = NULL;
+	struct tuple *insert = NULL;
+	if (old_tuple != NULL) {
+		delete = vy_stmt_new_surrogate_delete(lsm->mem_format,
+						      old_tuple);
+		if (delete == NULL)
+			return -1;
+	}
+	enum iproto_type type = vy_stmt_type(mem_stmt);
+	if (type == IPROTO_REPLACE || type == IPROTO_INSERT) {
+		uint32_t data_len;
+		const char *data = tuple_data_range(mem_stmt, &data_len);
+		insert = vy_stmt_new_insert(lsm->mem_format,
+					    data, data + data_len);
+		if (insert == NULL)
+			return -1;
+	} else if (type == IPROTO_UPSERT) {
+		struct tuple *new_tuple = vy_apply_upsert(mem_stmt, old_tuple,
+					pk->cmp_def, pk->mem_format, true);
+		if (new_tuple == NULL)
+			return -1;
+		uint32_t data_len;
+		const char *data = tuple_data_range(new_tuple, &data_len);
+		insert = vy_stmt_new_insert(lsm->mem_format,
+					    data, data + data_len);
+		tuple_unref(new_tuple);
+		if (insert == NULL)
+			return -1;
+	}
+
+	/* Insert DELETE + INSERT into the LSM tree. */
+	if (delete != NULL) {
+		int rc = vy_build_insert_stmt(lsm, lsm->mem, delete, lsn);
+		tuple_unref(delete);
+		if (rc != 0)
+			return -1;
+	}
+	if (insert != NULL) {
+		int rc = vy_build_insert_stmt(lsm, lsm->mem, insert, lsn);
+		tuple_unref(insert);
+		if (rc != 0)
+			return -1;
+	}
+	return 0;
+}
+
+/**
+ * Recover all statements stored in the given in-memory index
+ * that were inserted into the space while the newly built index
+ * was dumped to disk.
+ */
+static int
+vy_build_recover_mem(struct vy_lsm *lsm, struct vy_lsm *pk, struct vy_mem *mem)
+{
+	/*
+	 * Recover statements starting from the oldest one.
+	 * Key order doesn't matter so we simply iterate over
+	 * the in-memory index in reverse order.
+	 */
+	struct vy_mem_tree_iterator itr;
+	itr = vy_mem_tree_iterator_last(&mem->tree);
+	while (!vy_mem_tree_iterator_is_invalid(&itr)) {
+		const struct tuple *mem_stmt;
+		mem_stmt = *vy_mem_tree_iterator_get_elem(&mem->tree, &itr);
+		if (vy_build_recover_stmt(lsm, pk, mem_stmt) != 0)
+			return -1;
+		vy_mem_tree_iterator_prev(&mem->tree, &itr);
+	}
+	return 0;
+}
+
+/**
+ * Recover the memory level of a newly built index.
+ *
+ * During the final dump of a newly built index, new statements may
+ * be inserted into the space. If those statements are not dumped to
+ * disk before restart, they won't be recovered from WAL, because at
+ * the time they were generated the new index didn't exist. In order
+ * to recover them, we replay all statements stored in the memory
+ * level of the primary index.
+ */
+static int
+vy_build_recover(struct vy_env *env, struct vy_lsm *lsm, struct vy_lsm *pk)
+{
+	int rc = 0;
+	struct vy_mem *mem;
+	size_t mem_used_before, mem_used_after;
+
+	mem_used_before = lsregion_used(&env->mem_env.allocator);
+	rlist_foreach_entry_reverse(mem, &pk->sealed, in_sealed) {
+		rc = vy_build_recover_mem(lsm, pk, mem);
+		if (rc != 0)
+			break;
+	}
+	if (rc == 0)
+		rc = vy_build_recover_mem(lsm, pk, pk->mem);
+
+	mem_used_after = lsregion_used(&env->mem_env.allocator);
+	assert(mem_used_after >= mem_used_before);
+	vy_quota_force_use(&env->quota, mem_used_after - mem_used_before);
+	return rc;
+}
+
+static int
+vinyl_space_build_index(struct space *src_space, struct index *new_index,
+			struct tuple_format *new_format)
+{
+	struct vy_env *env = vy_env(src_space->engine);
+	struct vy_lsm *pk = vy_lsm(src_space->index[0]);
+	bool is_empty = (pk->stat.disk.count.rows == 0 &&
+			 pk->stat.memory.count.rows == 0);
+
+	if (new_index->def->iid == 0 && !is_empty) {
+		diag_set(ClientError, ER_UNSUPPORTED, "Vinyl",
+			 "rebuilding the primary index of a non-empty space");
+		return -1;
+	}
+
+	if (vinyl_index_open(new_index) != 0)
+		return -1;
+
+	/* Set pointer to the primary key for the new index. */
+	struct vy_lsm *new_lsm = vy_lsm(new_index);
+	vy_lsm_update_pk(new_lsm, pk);
+
+	if (env->status == VINYL_INITIAL_RECOVERY_LOCAL ||
+	    env->status == VINYL_FINAL_RECOVERY_LOCAL)
+		return vy_build_recover(env, new_lsm, pk);
+
+	if (is_empty)
+		return 0;
+
+	/*
+	 * Iterate over all tuples stored in the space and insert
+	 * each of them into the new LSM tree. Since read iterator
+	 * may yield, we install an on_replace trigger to forward
+	 * DML requests issued during the build.
+	 */
+	struct tuple *key = vy_stmt_new_select(pk->env->key_format, NULL, 0);
+	if (key == NULL)
+		return -1;
+
+	struct trigger on_replace;
+	struct vy_build_ctx ctx;
+	ctx.env = env;
+	ctx.lsm = new_lsm;
+	ctx.format = new_format;
+	ctx.space_name = space_name(src_space);
+	ctx.index_name = new_index->def->name;
+	ctx.is_failed = false;
+	diag_create(&ctx.diag);
+	trigger_create(&on_replace, vy_build_on_replace, &ctx, NULL);
+	trigger_add(&src_space->on_replace, &on_replace);
+
+	struct vy_read_iterator itr;
+	vy_read_iterator_open(&itr, pk, NULL, ITER_ALL, key,
+			      &env->xm->p_committed_read_view);
+	int rc;
+	int loops = 0;
+	struct tuple *tuple;
+	int64_t build_lsn = env->xm->lsn;
+	while ((rc = vy_read_iterator_next(&itr, &tuple)) == 0) {
+		if (tuple == NULL)
+			break;
+		/*
+		 * Insert the tuple into the new index unless it
+		 * was inserted into the space after we started
+		 * building the new index - in the latter case
+		 * the new tuple has already been inserted by the
+		 * on_replace trigger.
+		 *
+		 * Note, yield is not allowed between reading the
+		 * tuple from the primary index and inserting it
+		 * into the new index. If we yielded, the tuple
+		 * could be overwritten by a concurrent transaction,
+		 * in which case we would insert an outdated tuple.
+		 */
+		if (vy_stmt_lsn(tuple) <= build_lsn) {
+			rc = vy_build_insert_tuple(env, new_lsm,
+						   space_name(src_space),
+						   new_index->def->name,
+						   new_format, tuple);
+			if (rc != 0)
+				break;
+		}
+		/*
+		 * Read iterator yields only when it reads runs.
+		 * Yield periodically in order not to stall the
+		 * tx thread in case there are a lot of tuples in
+		 * mems or cache.
+		 */
+		if (++loops % VY_YIELD_LOOPS == 0)
+			fiber_sleep(0);
+		if (ctx.is_failed) {
+			diag_move(&ctx.diag, diag_get());
+			rc = -1;
+			break;
+		}
+	}
+	vy_read_iterator_close(&itr);
+	tuple_unref(key);
+
+	/*
+	 * Dump the new index upon build completion so that we don't
+	 * have to rebuild it on recovery.
+	 */
+	if (rc == 0)
+		rc = vy_scheduler_dump(&env->scheduler);
+
+	if (rc == 0 && ctx.is_failed) {
+		diag_move(&ctx.diag, diag_get());
+		rc = -1;
+	}
+
+	diag_destroy(&ctx.diag);
+	trigger_clear(&on_replace);
+	return rc;
+}
+
+/* }}} Index build */
+
 static const struct engine_vtab vinyl_engine_vtab = {
 	/* .shutdown = */ vinyl_engine_shutdown,
 	/* .create_space = */ vinyl_engine_create_space,
diff --git a/src/box/vy_lsm.c b/src/box/vy_lsm.c
index fd352c8a..5ac5548b 100644
--- a/src/box/vy_lsm.c
+++ b/src/box/vy_lsm.c
@@ -773,11 +773,20 @@ int
 vy_lsm_set(struct vy_lsm *lsm, struct vy_mem *mem,
 	   const struct tuple *stmt, const struct tuple **region_stmt)
 {
+	uint32_t format_id = stmt->format_id;
+
 	assert(vy_stmt_is_refable(stmt));
 	assert(*region_stmt == NULL || !vy_stmt_is_refable(*region_stmt));
 
-	/* Allocate region_stmt on demand. */
-	if (*region_stmt == NULL) {
+	/*
+	 * Allocate region_stmt on demand.
+	 *
+	 * Also, reallocate region_stmt if it uses a different tuple
+	 * format. This may happen during ALTER, when the LSM tree
+	 * that is currently being built uses the new space format
+	 * while other LSM trees still use the old space format.
+	 */
+	if (*region_stmt == NULL || (*region_stmt)->format_id != format_id) {
 		*region_stmt = vy_stmt_dup_lsregion(stmt, &mem->env->allocator,
 						    mem->generation);
 		if (*region_stmt == NULL)
@@ -788,7 +797,6 @@ vy_lsm_set(struct vy_lsm *lsm, struct vy_mem *mem,
 	lsm->stat.memory.count.bytes += tuple_size(stmt);
 
 	/* Abort transaction if format was changed by DDL */
-	uint32_t format_id = stmt->format_id;
 	if (format_id != tuple_format_id(mem->format_with_colmask) &&
 	    format_id != tuple_format_id(mem->format)) {
 		diag_set(ClientError, ER_TRANSACTION_CONFLICT);
diff --git a/src/box/vy_quota.h b/src/box/vy_quota.h
index 89f88bdc..73fc6a1b 100644
--- a/src/box/vy_quota.h
+++ b/src/box/vy_quota.h
@@ -175,6 +175,16 @@ vy_quota_use(struct vy_quota *q, size_t size, double timeout)
 	return 0;
 }
 
+/**
+ * Block the caller until the quota is not exceeded.
+ */
+static inline void
+vy_quota_wait(struct vy_quota *q)
+{
+	while (q->used > q->limit)
+		fiber_cond_wait(&q->cond);
+}
+
 #if defined(__cplusplus)
 } /* extern "C" */
 #endif /* defined(__cplusplus) */
diff --git a/src/box/vy_scheduler.c b/src/box/vy_scheduler.c
index eff3a814..51eb3c1a 100644
--- a/src/box/vy_scheduler.c
+++ b/src/box/vy_scheduler.c
@@ -458,6 +458,35 @@ vy_scheduler_trigger_dump(struct vy_scheduler *scheduler)
 	fiber_cond_signal(&scheduler->scheduler_cond);
 }
 
+int
+vy_scheduler_dump(struct vy_scheduler *scheduler)
+{
+	/*
+	 * We must not start dump if checkpoint is in progress
+	 * so first wait for checkpoint to complete.
+	 */
+	while (scheduler->checkpoint_in_progress)
+		fiber_cond_wait(&scheduler->dump_cond);
+
+	/* Trigger dump. */
+	if (scheduler->generation == scheduler->dump_generation)
+		scheduler->dump_start = ev_monotonic_now(loop());
+	int64_t generation = ++scheduler->generation;
+	fiber_cond_signal(&scheduler->scheduler_cond);
+
+	/* Wait for dump to complete. */
+	while (scheduler->dump_generation < generation) {
+		if (scheduler->is_throttled) {
+			/* Dump error occurred. */
+			struct error *e = diag_last_error(&scheduler->diag);
+			diag_add_error(diag_get(), e);
+			return -1;
+		}
+		fiber_cond_wait(&scheduler->dump_cond);
+	}
+	return 0;
+}
+
 void
 vy_scheduler_force_compaction(struct vy_scheduler *scheduler,
 			      struct vy_lsm *lsm)
diff --git a/src/box/vy_scheduler.h b/src/box/vy_scheduler.h
index d3bc4c03..777756c0 100644
--- a/src/box/vy_scheduler.h
+++ b/src/box/vy_scheduler.h
@@ -202,6 +202,13 @@ void
 vy_scheduler_trigger_dump(struct vy_scheduler *scheduler);
 
 /**
+ * Trigger dump of all currently existing in-memory trees
+ * and wait until it is complete. Returns 0 on success.
+ */
+int
+vy_scheduler_dump(struct vy_scheduler *scheduler);
+
+/**
  * Force major compaction of an LSM tree.
  */
 void
diff --git a/test/box/alter.result b/test/box/alter.result
index b3a14db6..eb7014d8 100644
--- a/test/box/alter.result
+++ b/test/box/alter.result
@@ -820,507 +820,6 @@ ts:drop()
 ---
 ...
 --
--- gh-1557: NULL in indexes.
---
-NULL = require('msgpack').NULL
----
-...
-format = {}
----
-...
-format[1] = { name = 'field1', type = 'unsigned', is_nullable = true }
----
-...
-format[2] = { name = 'field2', type = 'unsigned', is_nullable = true }
----
-...
-s = box.schema.space.create('test', { format = format })
----
-...
-s:create_index('primary', { parts = { 'field1' } })
----
-- error: Primary index of the space 'test' can not contain nullable parts
-...
-s:create_index('primary', { parts = {{'field1', is_nullable = false}} })
----
-- error: Field 1 is nullable in space format, but not nullable in index parts
-...
-format[1].is_nullable = false
----
-...
-s:format(format)
----
-...
-s:create_index('primary', { parts = {{'field1', is_nullable = true}} })
----
-- error: Primary index of the space 'test' can not contain nullable parts
-...
-s:create_index('primary', { parts = {'field1'} })
----
-- unique: true
-  parts:
-  - type: unsigned
-    is_nullable: false
-    fieldno: 1
-  id: 0
-  space_id: 733
-  name: primary
-  type: TREE
-...
--- Check that is_nullable can't be set to false on non-empty space
-s:insert({1, NULL})
----
-- [1, null]
-...
-format[1].is_nullable = true
----
-...
-s:format(format)
----
-- error: Field 1 is nullable in space format, but not nullable in index parts
-...
-format[1].is_nullable = false
----
-...
-format[2].is_nullable = false
----
-...
-s:format(format)
----
-- error: 'Tuple field 2 type does not match one required by operation: expected unsigned'
-...
-s:delete(1)
----
-- [1, null]
-...
--- Disable is_nullable on empty space
-s:format(format)
----
-...
--- Disable is_nullable on a non-empty space.
-format[2].is_nullable = true
----
-...
-s:format(format)
----
-...
-s:replace{1, 1}
----
-- [1, 1]
-...
-format[2].is_nullable = false
----
-...
-s:format(format)
----
-...
--- Enable is_nullable on a non-empty space.
-format[2].is_nullable = true
----
-...
-s:format(format)
----
-...
-s:replace{1, box.NULL}
----
-- [1, null]
-...
-s:delete{1}
----
-- [1, null]
-...
-s:format({})
----
-...
-s:create_index('secondary', { parts = {{2, 'string', is_nullable = true}} })
----
-- unique: true
-  parts:
-  - type: string
-    is_nullable: true
-    fieldno: 2
-  id: 1
-  space_id: 733
-  name: secondary
-  type: TREE
-...
-s:insert({1, NULL})
----
-- [1, null]
-...
-s.index.secondary:alter({ parts = {{2, 'string', is_nullable = false} }})
----
-- error: 'Tuple field 2 type does not match one required by operation: expected string'
-...
-s:delete({1})
----
-- [1, null]
-...
-s.index.secondary:alter({ parts = {{2, 'string', is_nullable = false} }})
----
-...
-s:insert({1, NULL})
----
-- error: 'Tuple field 2 type does not match one required by operation: expected string'
-...
-s:insert({2, 'xxx'})
----
-- [2, 'xxx']
-...
-s.index.secondary:alter({ parts = {{2, 'string', is_nullable = true} }})
----
-...
-s:insert({1, NULL})
----
-- [1, null]
-...
-s:drop()
----
-...
-s = box.schema.create_space('test')
----
-...
-test_run:cmd("setopt delimiter ';'")
----
-- true
-...
-s:format({
-    [1] = { name = 'id1', type = 'unsigned'},
-    [2] = { name = 'id2', type = 'unsigned'},
-    [3] = { name = 'id3', type = 'string'},
-    [4] = { name = 'id4', type = 'string'},
-    [5] = { name = 'id5', type = 'string'},
-    [6] = { name = 'id6', type = 'string'},
-});
----
-...
-test_run:cmd("setopt delimiter ''");
----
-- true
-...
-s:format()
----
-- [{'name': 'id1', 'type': 'unsigned'}, {'name': 'id2', 'type': 'unsigned'}, {'name': 'id3',
-    'type': 'string'}, {'name': 'id4', 'type': 'string'}, {'name': 'id5', 'type': 'string'},
-  {'name': 'id6', 'type': 'string'}]
-...
-_ = s:create_index('primary')
----
-...
-s:insert({1, 1, 'a', 'b', 'c', 'd'})
----
-- [1, 1, 'a', 'b', 'c', 'd']
-...
-s:drop()
----
-...
-s = box.schema.create_space('test')
----
-...
-idx = s:create_index('idx')
----
-...
-box.space.test == s
----
-- true
-...
-s:drop()
----
-...
---
--- gh-3000: index modifying must change key_def parts and
--- comparators. They can be changed, if there was compatible index
--- parts change. For example, a part type was changed from
--- unsigned to number. In such a case comparators must be reset
--- and part types updated.
---
-s = box.schema.create_space('test')
----
-...
-pk = s:create_index('pk')
----
-...
-s:replace{1}
----
-- [1]
-...
-pk:alter{parts = {{1, 'integer'}}}
----
-...
-s:replace{-2}
----
-- [-2]
-...
-s:select{}
----
-- - [-2]
-  - [1]
-...
-s:drop()
----
-...
---
--- Allow to change is_nullable in index definition on non-empty
--- space.
---
-s = box.schema.create_space('test')
----
-...
-pk = s:create_index('pk')
----
-...
-sk1 = s:create_index('sk1', {parts = {{2, 'unsigned', is_nullable = true}}})
----
-...
-sk2 = s:create_index('sk2', {parts = {{3, 'unsigned', is_nullable = false}}})
----
-...
-s:replace{1, box.NULL, 1}
----
-- [1, null, 1]
-...
-sk1:alter({parts = {{2, 'unsigned', is_nullable = false}}})
----
-- error: 'Tuple field 2 type does not match one required by operation: expected unsigned'
-...
-s:replace{1, 1, 1}
----
-- [1, 1, 1]
-...
-sk1:alter({parts = {{2, 'unsigned', is_nullable = false}}})
----
-...
-s:replace{1, 1, box.NULL}
----
-- error: 'Tuple field 3 type does not match one required by operation: expected unsigned'
-...
-sk2:alter({parts = {{3, 'unsigned', is_nullable = true}}})
----
-...
-s:replace{1, 1, box.NULL}
----
-- [1, 1, null]
-...
-s:replace{2, 10, 100}
----
-- [2, 10, 100]
-...
-s:replace{3, 0, 20}
----
-- [3, 0, 20]
-...
-s:replace{4, 15, 150}
----
-- [4, 15, 150]
-...
-s:replace{5, 9, box.NULL}
----
-- [5, 9, null]
-...
-sk1:select{}
----
-- - [3, 0, 20]
-  - [1, 1, null]
-  - [5, 9, null]
-  - [2, 10, 100]
-  - [4, 15, 150]
-...
-sk2:select{}
----
-- - [1, 1, null]
-  - [5, 9, null]
-  - [3, 0, 20]
-  - [2, 10, 100]
-  - [4, 15, 150]
-...
-s:drop()
----
-...
---
--- gh-3008: allow multiple types on the same field.
---
-format = {}
----
-...
-format[1] = {name = 'field1', type = 'unsigned'}
----
-...
-format[2] = {name = 'field2', type = 'scalar'}
----
-...
-format[3] = {name = 'field3', type = 'integer'}
----
-...
-s = box.schema.create_space('test', {format = format})
----
-...
-pk = s:create_index('pk')
----
-...
-sk1 = s:create_index('sk1', {parts = {{2, 'number'}}})
----
-...
-sk2 = s:create_index('sk2', {parts = {{2, 'integer'}}})
----
-...
-sk3 = s:create_index('sk3', {parts = {{2, 'unsigned'}}})
----
-...
-sk4 = s:create_index('sk4', {parts = {{3, 'number'}}})
----
-...
-s:format()
----
-- [{'name': 'field1', 'type': 'unsigned'}, {'name': 'field2', 'type': 'scalar'}, {
-    'name': 'field3', 'type': 'integer'}]
-...
-s:replace{1, '100', -20.2}
----
-- error: 'Tuple field 2 type does not match one required by operation: expected unsigned'
-...
-s:replace{1, 100, -20.2}
----
-- error: 'Tuple field 3 type does not match one required by operation: expected integer'
-...
-s:replace{1, 100, -20}
----
-- [1, 100, -20]
-...
-s:replace{2, 50, 0}
----
-- [2, 50, 0]
-...
-s:replace{3, 150, -60}
----
-- [3, 150, -60]
-...
-s:replace{4, 0, 120}
----
-- [4, 0, 120]
-...
-pk:select{}
----
-- - [1, 100, -20]
-  - [2, 50, 0]
-  - [3, 150, -60]
-  - [4, 0, 120]
-...
-sk1:select{}
----
-- - [4, 0, 120]
-  - [2, 50, 0]
-  - [1, 100, -20]
-  - [3, 150, -60]
-...
-sk2:select{}
----
-- - [4, 0, 120]
-  - [2, 50, 0]
-  - [1, 100, -20]
-  - [3, 150, -60]
-...
-sk3:select{}
----
-- - [4, 0, 120]
-  - [2, 50, 0]
-  - [1, 100, -20]
-  - [3, 150, -60]
-...
-sk4:select{}
----
-- - [3, 150, -60]
-  - [1, 100, -20]
-  - [2, 50, 0]
-  - [4, 0, 120]
-...
-sk1:alter{parts = {{2, 'unsigned'}}}
----
-...
-sk2:alter{parts = {{2, 'unsigned'}}}
----
-...
-sk4:alter{parts = {{3, 'integer'}}}
----
-...
-s:replace{1, 50.5, 1.5}
----
-- error: 'Tuple field 2 type does not match one required by operation: expected unsigned'
-...
-s:replace{1, 50, 1.5}
----
-- error: 'Tuple field 3 type does not match one required by operation: expected integer'
-...
-s:replace{5, 5, 5}
----
-- [5, 5, 5]
-...
-sk1:select{}
----
-- - [4, 0, 120]
-  - [5, 5, 5]
-  - [2, 50, 0]
-  - [1, 100, -20]
-  - [3, 150, -60]
-...
-sk2:select{}
----
-- - [4, 0, 120]
-  - [5, 5, 5]
-  - [2, 50, 0]
-  - [1, 100, -20]
-  - [3, 150, -60]
-...
-sk3:select{}
----
-- - [4, 0, 120]
-  - [5, 5, 5]
-  - [2, 50, 0]
-  - [1, 100, -20]
-  - [3, 150, -60]
-...
-sk4:select{}
----
-- - [3, 150, -60]
-  - [1, 100, -20]
-  - [2, 50, 0]
-  - [5, 5, 5]
-  - [4, 0, 120]
-...
-sk1:drop()
----
-...
-sk2:drop()
----
-...
-sk3:drop()
----
-...
--- Remove 'unsigned' constraints from indexes, and 'scalar' now
--- can be inserted in the second field.
-s:replace{1, true, 100}
----
-- [1, true, 100]
-...
-s:select{}
----
-- - [1, true, 100]
-  - [2, 50, 0]
-  - [3, 150, -60]
-  - [4, 0, 120]
-  - [5, 5, 5]
-...
-sk4:select{}
----
-- - [3, 150, -60]
-  - [2, 50, 0]
-  - [5, 5, 5]
-  - [1, true, 100]
-  - [4, 0, 120]
-...
-s:drop()
----
-...
---
 -- gh-2914: Allow any space name which consists of printable characters
 --
 identifier = require("identifier")
diff --git a/test/box/alter.test.lua b/test/box/alter.test.lua
index 11b8125f..b2b25390 100644
--- a/test/box/alter.test.lua
+++ b/test/box/alter.test.lua
@@ -316,165 +316,6 @@ n
 ts:drop()
 
 --
--- gh-1557: NULL in indexes.
---
-
-NULL = require('msgpack').NULL
-
-format = {}
-format[1] = { name = 'field1', type = 'unsigned', is_nullable = true }
-format[2] = { name = 'field2', type = 'unsigned', is_nullable = true }
-s = box.schema.space.create('test', { format = format })
-s:create_index('primary', { parts = { 'field1' } })
-s:create_index('primary', { parts = {{'field1', is_nullable = false}} })
-format[1].is_nullable = false
-s:format(format)
-s:create_index('primary', { parts = {{'field1', is_nullable = true}} })
-
-s:create_index('primary', { parts = {'field1'} })
-
--- Check that is_nullable can't be set to false on non-empty space
-s:insert({1, NULL})
-format[1].is_nullable = true
-s:format(format)
-format[1].is_nullable = false
-format[2].is_nullable = false
-s:format(format)
-s:delete(1)
--- Disable is_nullable on empty space
-s:format(format)
--- Disable is_nullable on a non-empty space.
-format[2].is_nullable = true
-s:format(format)
-s:replace{1, 1}
-format[2].is_nullable = false
-s:format(format)
--- Enable is_nullable on a non-empty space.
-format[2].is_nullable = true
-s:format(format)
-s:replace{1, box.NULL}
-s:delete{1}
-s:format({})
-
-s:create_index('secondary', { parts = {{2, 'string', is_nullable = true}} })
-s:insert({1, NULL})
-s.index.secondary:alter({ parts = {{2, 'string', is_nullable = false} }})
-s:delete({1})
-s.index.secondary:alter({ parts = {{2, 'string', is_nullable = false} }})
-s:insert({1, NULL})
-s:insert({2, 'xxx'})
-s.index.secondary:alter({ parts = {{2, 'string', is_nullable = true} }})
-s:insert({1, NULL})
-
-s:drop()
-
-s = box.schema.create_space('test')
-test_run:cmd("setopt delimiter ';'")
-s:format({
-    [1] = { name = 'id1', type = 'unsigned'},
-    [2] = { name = 'id2', type = 'unsigned'},
-    [3] = { name = 'id3', type = 'string'},
-    [4] = { name = 'id4', type = 'string'},
-    [5] = { name = 'id5', type = 'string'},
-    [6] = { name = 'id6', type = 'string'},
-});
-test_run:cmd("setopt delimiter ''");
-s:format()
-_ = s:create_index('primary')
-s:insert({1, 1, 'a', 'b', 'c', 'd'})
-s:drop()
-
-s = box.schema.create_space('test')
-idx = s:create_index('idx')
-box.space.test == s
-s:drop()
-
---
--- gh-3000: index modifying must change key_def parts and
--- comparators. They can be changed, if there was compatible index
--- parts change. For example, a part type was changed from
--- unsigned to number. In such a case comparators must be reset
--- and part types updated.
---
-s = box.schema.create_space('test')
-pk = s:create_index('pk')
-s:replace{1}
-pk:alter{parts = {{1, 'integer'}}}
-s:replace{-2}
-s:select{}
-s:drop()
-
---
--- Allow to change is_nullable in index definition on non-empty
--- space.
---
-s = box.schema.create_space('test')
-pk = s:create_index('pk')
-sk1 = s:create_index('sk1', {parts = {{2, 'unsigned', is_nullable = true}}})
-sk2 = s:create_index('sk2', {parts = {{3, 'unsigned', is_nullable = false}}})
-s:replace{1, box.NULL, 1}
-sk1:alter({parts = {{2, 'unsigned', is_nullable = false}}})
-s:replace{1, 1, 1}
-sk1:alter({parts = {{2, 'unsigned', is_nullable = false}}})
-s:replace{1, 1, box.NULL}
-sk2:alter({parts = {{3, 'unsigned', is_nullable = true}}})
-s:replace{1, 1, box.NULL}
-s:replace{2, 10, 100}
-s:replace{3, 0, 20}
-s:replace{4, 15, 150}
-s:replace{5, 9, box.NULL}
-sk1:select{}
-sk2:select{}
-s:drop()
-
---
--- gh-3008: allow multiple types on the same field.
---
-format = {}
-format[1] = {name = 'field1', type = 'unsigned'}
-format[2] = {name = 'field2', type = 'scalar'}
-format[3] = {name = 'field3', type = 'integer'}
-s = box.schema.create_space('test', {format = format})
-pk = s:create_index('pk')
-sk1 = s:create_index('sk1', {parts = {{2, 'number'}}})
-sk2 = s:create_index('sk2', {parts = {{2, 'integer'}}})
-sk3 = s:create_index('sk3', {parts = {{2, 'unsigned'}}})
-sk4 = s:create_index('sk4', {parts = {{3, 'number'}}})
-s:format()
-s:replace{1, '100', -20.2}
-s:replace{1, 100, -20.2}
-s:replace{1, 100, -20}
-s:replace{2, 50, 0}
-s:replace{3, 150, -60}
-s:replace{4, 0, 120}
-pk:select{}
-sk1:select{}
-sk2:select{}
-sk3:select{}
-sk4:select{}
-
-sk1:alter{parts = {{2, 'unsigned'}}}
-sk2:alter{parts = {{2, 'unsigned'}}}
-sk4:alter{parts = {{3, 'integer'}}}
-s:replace{1, 50.5, 1.5}
-s:replace{1, 50, 1.5}
-s:replace{5, 5, 5}
-sk1:select{}
-sk2:select{}
-sk3:select{}
-sk4:select{}
-
-sk1:drop()
-sk2:drop()
-sk3:drop()
--- Remove 'unsigned' constraints from indexes, and 'scalar' now
--- can be inserted in the second field.
-s:replace{1, true, 100}
-s:select{}
-sk4:select{}
-s:drop()
-
---
 -- gh-2914: Allow any space name which consists of printable characters
 --
 identifier = require("identifier")
diff --git a/test/engine/ddl.result b/test/engine/ddl.result
index 04062ac1..6c915879 100644
--- a/test/engine/ddl.result
+++ b/test/engine/ddl.result
@@ -1352,3 +1352,675 @@ s:select()
 s:drop()
 ---
 ...
+--
+-- gh-1557: NULL in indexes.
+--
+NULL = require('msgpack').NULL
+---
+...
+format = {}
+---
+...
+format[1] = { name = 'field1', type = 'unsigned', is_nullable = true }
+---
+...
+format[2] = { name = 'field2', type = 'unsigned', is_nullable = true }
+---
+...
+s = box.schema.space.create('test', {engine = engine, format = format})
+---
+...
+s:create_index('primary', { parts = { 'field1' } })
+---
+- error: Primary index of the space 'test' can not contain nullable parts
+...
+s:create_index('primary', { parts = {{'field1', is_nullable = false}} })
+---
+- error: Field 1 is nullable in space format, but not nullable in index parts
+...
+format[1].is_nullable = false
+---
+...
+s:format(format)
+---
+...
+s:create_index('primary', { parts = {{'field1', is_nullable = true}} })
+---
+- error: Primary index of the space 'test' can not contain nullable parts
+...
+i = s:create_index('primary', { parts = {'field1'} })
+---
+...
+i.parts
+---
+- - type: unsigned
+    is_nullable: false
+    fieldno: 1
+...
+-- Check that is_nullable can't be set to false on non-empty space
+s:insert({1, NULL})
+---
+- [1, null]
+...
+format[1].is_nullable = true
+---
+...
+s:format(format)
+---
+- error: Field 1 is nullable in space format, but not nullable in index parts
+...
+format[1].is_nullable = false
+---
+...
+format[2].is_nullable = false
+---
+...
+s:format(format)
+---
+- error: 'Tuple field 2 type does not match one required by operation: expected unsigned'
+...
+_ = s:delete(1)
+---
+...
+-- Disable is_nullable on empty space
+s:format(format)
+---
+...
+-- Disable is_nullable on a non-empty space.
+format[2].is_nullable = true
+---
+...
+s:format(format)
+---
+...
+s:replace{1, 1}
+---
+- [1, 1]
+...
+format[2].is_nullable = false
+---
+...
+s:format(format)
+---
+...
+-- Enable is_nullable on a non-empty space.
+format[2].is_nullable = true
+---
+...
+s:format(format)
+---
+...
+s:replace{1, box.NULL}
+---
+- [1, null]
+...
+_ = s:delete{1}
+---
+...
+s:format({})
+---
+...
+i = s:create_index('secondary', { parts = {{2, 'string', is_nullable = true}} })
+---
+...
+i.parts
+---
+- - type: string
+    is_nullable: true
+    fieldno: 2
+...
+s:insert({1, NULL})
+---
+- [1, null]
+...
+s.index.secondary:alter({ parts = {{2, 'string', is_nullable = false} }})
+---
+- error: 'Tuple field 2 type does not match one required by operation: expected string'
+...
+_ = s:delete({1})
+---
+...
+s.index.secondary:alter({ parts = {{2, 'string', is_nullable = false} }})
+---
+...
+s:insert({1, NULL})
+---
+- error: 'Tuple field 2 type does not match one required by operation: expected string'
+...
+s:insert({2, 'xxx'})
+---
+- [2, 'xxx']
+...
+s.index.secondary:alter({ parts = {{2, 'string', is_nullable = true} }})
+---
+...
+s:insert({1, NULL})
+---
+- [1, null]
+...
+s:drop()
+---
+...
+s = box.schema.space.create('test', {engine = engine})
+---
+...
+inspector:cmd("setopt delimiter ';'")
+---
+- true
+...
+s:format({
+    [1] = { name = 'id1', type = 'unsigned'},
+    [2] = { name = 'id2', type = 'unsigned'},
+    [3] = { name = 'id3', type = 'string'},
+    [4] = { name = 'id4', type = 'string'},
+    [5] = { name = 'id5', type = 'string'},
+    [6] = { name = 'id6', type = 'string'},
+});
+---
+...
+inspector:cmd("setopt delimiter ''");
+---
+- true
+...
+s:format()
+---
+- [{'name': 'id1', 'type': 'unsigned'}, {'name': 'id2', 'type': 'unsigned'}, {'name': 'id3',
+    'type': 'string'}, {'name': 'id4', 'type': 'string'}, {'name': 'id5', 'type': 'string'},
+  {'name': 'id6', 'type': 'string'}]
+...
+_ = s:create_index('primary')
+---
+...
+s:insert({1, 1, 'a', 'b', 'c', 'd'})
+---
+- [1, 1, 'a', 'b', 'c', 'd']
+...
+s:drop()
+---
+...
+s = box.schema.space.create('test', {engine = engine})
+---
+...
+idx = s:create_index('idx')
+---
+...
+box.space.test == s
+---
+- true
+...
+s:drop()
+---
+...
+--
+-- gh-3000: index modifying must change key_def parts and
+-- comparators. They can be changed, if there was compatible index
+-- parts change. For example, a part type was changed from
+-- unsigned to number. In such a case comparators must be reset
+-- and part types updated.
+--
+s = box.schema.space.create('test', {engine = engine})
+---
+...
+pk = s:create_index('pk')
+---
+...
+s:replace{1}
+---
+- [1]
+...
+pk:alter{parts = {{1, 'integer'}}}
+---
+...
+s:replace{-2}
+---
+- [-2]
+...
+s:select{}
+---
+- - [-2]
+  - [1]
+...
+s:drop()
+---
+...
+--
+-- Allow to change is_nullable in index definition on non-empty
+-- space.
+--
+s = box.schema.space.create('test', {engine = engine})
+---
+...
+pk = s:create_index('pk')
+---
+...
+sk1 = s:create_index('sk1', {parts = {{2, 'unsigned', is_nullable = true}}})
+---
+...
+sk2 = s:create_index('sk2', {parts = {{3, 'unsigned', is_nullable = false}}})
+---
+...
+s:replace{1, box.NULL, 1}
+---
+- [1, null, 1]
+...
+sk1:alter({parts = {{2, 'unsigned', is_nullable = false}}})
+---
+- error: 'Tuple field 2 type does not match one required by operation: expected unsigned'
+...
+s:replace{1, 1, 1}
+---
+- [1, 1, 1]
+...
+sk1:alter({parts = {{2, 'unsigned', is_nullable = false}}})
+---
+...
+s:replace{1, 1, box.NULL}
+---
+- error: 'Tuple field 3 type does not match one required by operation: expected unsigned'
+...
+sk2:alter({parts = {{3, 'unsigned', is_nullable = true}}})
+---
+...
+s:replace{1, 1, box.NULL}
+---
+- [1, 1, null]
+...
+s:replace{2, 10, 100}
+---
+- [2, 10, 100]
+...
+s:replace{3, 0, 20}
+---
+- [3, 0, 20]
+...
+s:replace{4, 15, 150}
+---
+- [4, 15, 150]
+...
+s:replace{5, 9, box.NULL}
+---
+- [5, 9, null]
+...
+sk1:select{}
+---
+- - [3, 0, 20]
+  - [1, 1, null]
+  - [5, 9, null]
+  - [2, 10, 100]
+  - [4, 15, 150]
+...
+sk2:select{}
+---
+- - [1, 1, null]
+  - [5, 9, null]
+  - [3, 0, 20]
+  - [2, 10, 100]
+  - [4, 15, 150]
+...
+s:drop()
+---
+...
+--
+-- gh-3008: allow multiple types on the same field.
+--
+format = {}
+---
+...
+format[1] = {name = 'field1', type = 'unsigned'}
+---
+...
+format[2] = {name = 'field2', type = 'scalar'}
+---
+...
+format[3] = {name = 'field3', type = 'integer'}
+---
+...
+s = box.schema.space.create('test', {engine = engine, format = format})
+---
+...
+pk = s:create_index('pk')
+---
+...
+sk1 = s:create_index('sk1', {parts = {{2, 'number'}}})
+---
+...
+sk2 = s:create_index('sk2', {parts = {{2, 'integer'}}})
+---
+...
+sk3 = s:create_index('sk3', {parts = {{2, 'unsigned'}}})
+---
+...
+sk4 = s:create_index('sk4', {parts = {{3, 'number'}}})
+---
+...
+s:format()
+---
+- [{'name': 'field1', 'type': 'unsigned'}, {'name': 'field2', 'type': 'scalar'}, {
+    'name': 'field3', 'type': 'integer'}]
+...
+s:replace{1, '100', -20.2}
+---
+- error: 'Tuple field 2 type does not match one required by operation: expected unsigned'
+...
+s:replace{1, 100, -20.2}
+---
+- error: 'Tuple field 3 type does not match one required by operation: expected integer'
+...
+s:replace{1, 100, -20}
+---
+- [1, 100, -20]
+...
+s:replace{2, 50, 0}
+---
+- [2, 50, 0]
+...
+s:replace{3, 150, -60}
+---
+- [3, 150, -60]
+...
+s:replace{4, 0, 120}
+---
+- [4, 0, 120]
+...
+pk:select{}
+---
+- - [1, 100, -20]
+  - [2, 50, 0]
+  - [3, 150, -60]
+  - [4, 0, 120]
+...
+sk1:select{}
+---
+- - [4, 0, 120]
+  - [2, 50, 0]
+  - [1, 100, -20]
+  - [3, 150, -60]
+...
+sk2:select{}
+---
+- - [4, 0, 120]
+  - [2, 50, 0]
+  - [1, 100, -20]
+  - [3, 150, -60]
+...
+sk3:select{}
+---
+- - [4, 0, 120]
+  - [2, 50, 0]
+  - [1, 100, -20]
+  - [3, 150, -60]
+...
+sk4:select{}
+---
+- - [3, 150, -60]
+  - [1, 100, -20]
+  - [2, 50, 0]
+  - [4, 0, 120]
+...
+sk1:alter{parts = {{2, 'unsigned'}}}
+---
+...
+sk2:alter{parts = {{2, 'unsigned'}}}
+---
+...
+sk4:alter{parts = {{3, 'integer'}}}
+---
+...
+s:replace{1, 50.5, 1.5}
+---
+- error: 'Tuple field 2 type does not match one required by operation: expected unsigned'
+...
+s:replace{1, 50, 1.5}
+---
+- error: 'Tuple field 3 type does not match one required by operation: expected integer'
+...
+s:replace{5, 5, 5}
+---
+- [5, 5, 5]
+...
+sk1:select{}
+---
+- - [4, 0, 120]
+  - [5, 5, 5]
+  - [2, 50, 0]
+  - [1, 100, -20]
+  - [3, 150, -60]
+...
+sk2:select{}
+---
+- - [4, 0, 120]
+  - [5, 5, 5]
+  - [2, 50, 0]
+  - [1, 100, -20]
+  - [3, 150, -60]
+...
+sk3:select{}
+---
+- - [4, 0, 120]
+  - [5, 5, 5]
+  - [2, 50, 0]
+  - [1, 100, -20]
+  - [3, 150, -60]
+...
+sk4:select{}
+---
+- - [3, 150, -60]
+  - [1, 100, -20]
+  - [2, 50, 0]
+  - [5, 5, 5]
+  - [4, 0, 120]
+...
+sk1:drop()
+---
+...
+sk2:drop()
+---
+...
+sk3:drop()
+---
+...
+-- Remove 'unsigned' constraints from indexes, and 'scalar' now
+-- can be inserted in the second field.
+s:replace{1, true, 100}
+---
+- [1, true, 100]
+...
+s:select{}
+---
+- - [1, true, 100]
+  - [2, 50, 0]
+  - [3, 150, -60]
+  - [4, 0, 120]
+  - [5, 5, 5]
+...
+sk4:select{}
+---
+- - [3, 150, -60]
+  - [2, 50, 0]
+  - [5, 5, 5]
+  - [1, true, 100]
+  - [4, 0, 120]
+...
+s:drop()
+---
+...
+--
+-- Creating/altering a secondary index of a non-empty space.
+--
+s = box.schema.space.create('test', {engine = engine})
+---
+...
+_ = s:create_index('pk')
+---
+...
+_ = s:insert{1, 'zzz', 'aaa', 999}
+---
+...
+_ = s:insert{2, 'yyy', 'bbb', 888}
+---
+...
+_ = s:insert{3, 'xxx', 'ccc', 777}
+---
+...
+box.snapshot()
+---
+- ok
+...
+_ = s:update(1, {{'!', -1, 'eee'}})
+---
+...
+_ = s:upsert({2, '2', '2', -2}, {{'=', 4, -888}})
+---
+...
+_ = s:replace(s:get(3):update{{'=', 3, box.NULL}})
+---
+...
+_ = s:upsert({4, 'zzz', 'ddd', -666}, {{'!', -1, 'abc'}})
+---
+...
+box.snapshot()
+---
+- ok
+...
+_ = s:update(1, {{'=', 5, 'fff'}})
+---
+...
+_ = s:upsert({3, '3', '3', -3}, {{'=', 5, 'ggg'}})
+---
+...
+_ = s:insert{5, 'xxx', 'eee', 555, 'hhh'}
+---
+...
+_ = s:replace{6, 'yyy', box.NULL, -444}
+---
+...
+s:select()
+---
+- - [1, 'zzz', 'aaa', 999, 'fff']
+  - [2, 'yyy', 'bbb', -888]
+  - [3, 'xxx', null, 777, 'ggg']
+  - [4, 'zzz', 'ddd', -666]
+  - [5, 'xxx', 'eee', 555, 'hhh']
+  - [6, 'yyy', null, -444]
+...
+s:create_index('sk', {parts = {2, 'string'}}) -- error: unique constraint
+---
+- error: Duplicate key exists in unique index 'sk' in space 'test'
+...
+s:create_index('sk', {parts = {3, 'string'}}) -- error: nullability constraint
+---
+- error: 'Tuple field 3 type does not match one required by operation: expected string'
+...
+s:create_index('sk', {parts = {4, 'unsigned'}}) -- error: field type
+---
+- error: 'Tuple field 4 type does not match one required by operation: expected unsigned'
+...
+s:create_index('sk', {parts = {4, 'integer', 5, 'string'}}) -- error: field missing
+---
+- error: Tuple field count 4 is less than required by space format or defined indexes
+    (expected at least 5)
+...
+i1 = s:create_index('i1', {parts = {2, 'string'}, unique = false})
+---
+...
+i2 = s:create_index('i2', {parts = {{3, 'string', is_nullable = true}}})
+---
+...
+i3 = s:create_index('i3', {parts = {4, 'integer'}})
+---
+...
+i1:select()
+---
+- - [3, 'xxx', null, 777, 'ggg']
+  - [5, 'xxx', 'eee', 555, 'hhh']
+  - [2, 'yyy', 'bbb', -888]
+  - [6, 'yyy', null, -444]
+  - [1, 'zzz', 'aaa', 999, 'fff']
+  - [4, 'zzz', 'ddd', -666]
+...
+i2:select()
+---
+- - [3, 'xxx', null, 777, 'ggg']
+  - [6, 'yyy', null, -444]
+  - [1, 'zzz', 'aaa', 999, 'fff']
+  - [2, 'yyy', 'bbb', -888]
+  - [4, 'zzz', 'ddd', -666]
+  - [5, 'xxx', 'eee', 555, 'hhh']
+...
+i3:select()
+---
+- - [2, 'yyy', 'bbb', -888]
+  - [4, 'zzz', 'ddd', -666]
+  - [6, 'yyy', null, -444]
+  - [5, 'xxx', 'eee', 555, 'hhh']
+  - [3, 'xxx', null, 777, 'ggg']
+  - [1, 'zzz', 'aaa', 999, 'fff']
+...
+i1:alter{unique = true} -- error: unique contraint
+---
+- error: Duplicate key exists in unique index 'i1' in space 'test'
+...
+i2:alter{parts = {3, 'string'}} -- error: nullability contraint
+---
+- error: 'Tuple field 3 type does not match one required by operation: expected string'
+...
+i3:alter{parts = {4, 'unsigned'}} -- error: field type
+---
+- error: 'Tuple field 4 type does not match one required by operation: expected unsigned'
+...
+i3:alter{parts = {4, 'integer', 5, 'string'}} -- error: field missing
+---
+- error: Tuple field count 4 is less than required by space format or defined indexes
+    (expected at least 5)
+...
+i3:alter{parts = {2, 'string', 4, 'integer'}} -- ok
+---
+...
+i3:select()
+---
+- - [5, 'xxx', 'eee', 555, 'hhh']
+  - [3, 'xxx', null, 777, 'ggg']
+  - [2, 'yyy', 'bbb', -888]
+  - [6, 'yyy', null, -444]
+  - [4, 'zzz', 'ddd', -666]
+  - [1, 'zzz', 'aaa', 999, 'fff']
+...
+-- Check that recovery works.
+inspector:cmd("restart server default")
+s = box.space.test
+---
+...
+s.index.i1:select()
+---
+- - [3, 'xxx', null, 777, 'ggg']
+  - [5, 'xxx', 'eee', 555, 'hhh']
+  - [2, 'yyy', 'bbb', -888]
+  - [6, 'yyy', null, -444]
+  - [1, 'zzz', 'aaa', 999, 'fff']
+  - [4, 'zzz', 'ddd', -666]
+...
+s.index.i2:select()
+---
+- - [3, 'xxx', null, 777, 'ggg']
+  - [6, 'yyy', null, -444]
+  - [1, 'zzz', 'aaa', 999, 'fff']
+  - [2, 'yyy', 'bbb', -888]
+  - [4, 'zzz', 'ddd', -666]
+  - [5, 'xxx', 'eee', 555, 'hhh']
+...
+s.index.i3:select()
+---
+- - [5, 'xxx', 'eee', 555, 'hhh']
+  - [3, 'xxx', null, 777, 'ggg']
+  - [2, 'yyy', 'bbb', -888]
+  - [6, 'yyy', null, -444]
+  - [4, 'zzz', 'ddd', -666]
+  - [1, 'zzz', 'aaa', 999, 'fff']
+...
+box.snapshot()
+---
+- ok
+...
+s:drop()
+---
+...
diff --git a/test/engine/ddl.test.lua b/test/engine/ddl.test.lua
index f3d68e1b..2593535b 100644
--- a/test/engine/ddl.test.lua
+++ b/test/engine/ddl.test.lua
@@ -498,3 +498,224 @@ s:format(format)
 s:format()
 s:select()
 s:drop()
+
+--
+-- gh-1557: NULL in indexes.
+--
+
+NULL = require('msgpack').NULL
+
+format = {}
+format[1] = { name = 'field1', type = 'unsigned', is_nullable = true }
+format[2] = { name = 'field2', type = 'unsigned', is_nullable = true }
+s = box.schema.space.create('test', {engine = engine, format = format})
+s:create_index('primary', { parts = { 'field1' } })
+s:create_index('primary', { parts = {{'field1', is_nullable = false}} })
+format[1].is_nullable = false
+s:format(format)
+s:create_index('primary', { parts = {{'field1', is_nullable = true}} })
+
+i = s:create_index('primary', { parts = {'field1'} })
+i.parts
+
+-- Check that is_nullable can't be set to false on non-empty space
+s:insert({1, NULL})
+format[1].is_nullable = true
+s:format(format)
+format[1].is_nullable = false
+format[2].is_nullable = false
+s:format(format)
+_ = s:delete(1)
+-- Disable is_nullable on empty space
+s:format(format)
+-- Disable is_nullable on a non-empty space.
+format[2].is_nullable = true
+s:format(format)
+s:replace{1, 1}
+format[2].is_nullable = false
+s:format(format)
+-- Enable is_nullable on a non-empty space.
+format[2].is_nullable = true
+s:format(format)
+s:replace{1, box.NULL}
+_ = s:delete{1}
+s:format({})
+
+i = s:create_index('secondary', { parts = {{2, 'string', is_nullable = true}} })
+i.parts
+
+s:insert({1, NULL})
+s.index.secondary:alter({ parts = {{2, 'string', is_nullable = false} }})
+_ = s:delete({1})
+s.index.secondary:alter({ parts = {{2, 'string', is_nullable = false} }})
+s:insert({1, NULL})
+s:insert({2, 'xxx'})
+s.index.secondary:alter({ parts = {{2, 'string', is_nullable = true} }})
+s:insert({1, NULL})
+
+s:drop()
+
+s = box.schema.space.create('test', {engine = engine})
+inspector:cmd("setopt delimiter ';'")
+s:format({
+    [1] = { name = 'id1', type = 'unsigned'},
+    [2] = { name = 'id2', type = 'unsigned'},
+    [3] = { name = 'id3', type = 'string'},
+    [4] = { name = 'id4', type = 'string'},
+    [5] = { name = 'id5', type = 'string'},
+    [6] = { name = 'id6', type = 'string'},
+});
+inspector:cmd("setopt delimiter ''");
+s:format()
+_ = s:create_index('primary')
+s:insert({1, 1, 'a', 'b', 'c', 'd'})
+s:drop()
+
+s = box.schema.space.create('test', {engine = engine})
+idx = s:create_index('idx')
+box.space.test == s
+s:drop()
+
+--
+-- gh-3000: index modifying must change key_def parts and
+-- comparators. They can be changed, if there was compatible index
+-- parts change. For example, a part type was changed from
+-- unsigned to number. In such a case comparators must be reset
+-- and part types updated.
+--
+s = box.schema.space.create('test', {engine = engine})
+pk = s:create_index('pk')
+s:replace{1}
+pk:alter{parts = {{1, 'integer'}}}
+s:replace{-2}
+s:select{}
+s:drop()
+
+--
+-- Allow to change is_nullable in index definition on non-empty
+-- space.
+--
+s = box.schema.space.create('test', {engine = engine})
+pk = s:create_index('pk')
+sk1 = s:create_index('sk1', {parts = {{2, 'unsigned', is_nullable = true}}})
+sk2 = s:create_index('sk2', {parts = {{3, 'unsigned', is_nullable = false}}})
+s:replace{1, box.NULL, 1}
+sk1:alter({parts = {{2, 'unsigned', is_nullable = false}}})
+s:replace{1, 1, 1}
+sk1:alter({parts = {{2, 'unsigned', is_nullable = false}}})
+s:replace{1, 1, box.NULL}
+sk2:alter({parts = {{3, 'unsigned', is_nullable = true}}})
+s:replace{1, 1, box.NULL}
+s:replace{2, 10, 100}
+s:replace{3, 0, 20}
+s:replace{4, 15, 150}
+s:replace{5, 9, box.NULL}
+sk1:select{}
+sk2:select{}
+s:drop()
+
+--
+-- gh-3008: allow multiple types on the same field.
+--
+format = {}
+format[1] = {name = 'field1', type = 'unsigned'}
+format[2] = {name = 'field2', type = 'scalar'}
+format[3] = {name = 'field3', type = 'integer'}
+s = box.schema.space.create('test', {engine = engine, format = format})
+pk = s:create_index('pk')
+sk1 = s:create_index('sk1', {parts = {{2, 'number'}}})
+sk2 = s:create_index('sk2', {parts = {{2, 'integer'}}})
+sk3 = s:create_index('sk3', {parts = {{2, 'unsigned'}}})
+sk4 = s:create_index('sk4', {parts = {{3, 'number'}}})
+s:format()
+s:replace{1, '100', -20.2}
+s:replace{1, 100, -20.2}
+s:replace{1, 100, -20}
+s:replace{2, 50, 0}
+s:replace{3, 150, -60}
+s:replace{4, 0, 120}
+pk:select{}
+sk1:select{}
+sk2:select{}
+sk3:select{}
+sk4:select{}
+
+sk1:alter{parts = {{2, 'unsigned'}}}
+sk2:alter{parts = {{2, 'unsigned'}}}
+sk4:alter{parts = {{3, 'integer'}}}
+s:replace{1, 50.5, 1.5}
+s:replace{1, 50, 1.5}
+s:replace{5, 5, 5}
+sk1:select{}
+sk2:select{}
+sk3:select{}
+sk4:select{}
+
+sk1:drop()
+sk2:drop()
+sk3:drop()
+-- Remove 'unsigned' constraints from indexes, and 'scalar' now
+-- can be inserted in the second field.
+s:replace{1, true, 100}
+s:select{}
+sk4:select{}
+s:drop()
+
+--
+-- Creating/altering a secondary index of a non-empty space.
+--
+s = box.schema.space.create('test', {engine = engine})
+_ = s:create_index('pk')
+
+_ = s:insert{1, 'zzz', 'aaa', 999}
+_ = s:insert{2, 'yyy', 'bbb', 888}
+_ = s:insert{3, 'xxx', 'ccc', 777}
+
+box.snapshot()
+
+_ = s:update(1, {{'!', -1, 'eee'}})
+_ = s:upsert({2, '2', '2', -2}, {{'=', 4, -888}})
+_ = s:replace(s:get(3):update{{'=', 3, box.NULL}})
+_ = s:upsert({4, 'zzz', 'ddd', -666}, {{'!', -1, 'abc'}})
+
+box.snapshot()
+
+_ = s:update(1, {{'=', 5, 'fff'}})
+_ = s:upsert({3, '3', '3', -3}, {{'=', 5, 'ggg'}})
+_ = s:insert{5, 'xxx', 'eee', 555, 'hhh'}
+_ = s:replace{6, 'yyy', box.NULL, -444}
+
+s:select()
+
+s:create_index('sk', {parts = {2, 'string'}}) -- error: unique constraint
+s:create_index('sk', {parts = {3, 'string'}}) -- error: nullability constraint
+s:create_index('sk', {parts = {4, 'unsigned'}}) -- error: field type
+s:create_index('sk', {parts = {4, 'integer', 5, 'string'}}) -- error: field missing
+
+i1 = s:create_index('i1', {parts = {2, 'string'}, unique = false})
+i2 = s:create_index('i2', {parts = {{3, 'string', is_nullable = true}}})
+i3 = s:create_index('i3', {parts = {4, 'integer'}})
+
+i1:select()
+i2:select()
+i3:select()
+
+i1:alter{unique = true} -- error: unique contraint
+i2:alter{parts = {3, 'string'}} -- error: nullability contraint
+i3:alter{parts = {4, 'unsigned'}} -- error: field type
+i3:alter{parts = {4, 'integer', 5, 'string'}} -- error: field missing
+
+i3:alter{parts = {2, 'string', 4, 'integer'}} -- ok
+i3:select()
+
+-- Check that recovery works.
+inspector:cmd("restart server default")
+
+s = box.space.test
+s.index.i1:select()
+s.index.i2:select()
+s.index.i3:select()
+
+box.snapshot()
+
+s:drop()
diff --git a/test/vinyl/ddl.result b/test/vinyl/ddl.result
index 4607a44e..378a7071 100644
--- a/test/vinyl/ddl.result
+++ b/test/vinyl/ddl.result
@@ -68,202 +68,61 @@ index = space:create_index('primary', {type = 'hash'})
 space:drop()
 ---
 ...
--- creation of a new index and altering the definition of an existing
--- index are unsupported for non-empty spaces
+-- rebuild of the primary index is not supported
 space = box.schema.space.create('test', { engine = 'vinyl' })
 ---
 ...
-index = space:create_index('primary')
+pk = space:create_index('pk', {run_count_per_level = 1, run_size_ratio = 10})
 ---
 ...
-space:insert({1})
+space:replace{1, 1}
 ---
-- [1]
-...
--- fail because of wrong tuple format {1}, but need {1, ...}
-index2 = space:create_index('secondary', { parts = {2, 'unsigned'} })
----
-- error: Vinyl does not support building an index for a non-empty space
-...
-space.index.primary:alter({parts = {1, 'unsigned', 2, 'unsigned'}})
----
-- error: Vinyl does not support building an index for a non-empty space
-...
-#box.space._index:select({space.id})
----
-- 1
-...
-box.space._index:get{space.id, 0}[6]
----
-- [[0, 'unsigned']]
-...
-space:drop()
----
-...
-space = box.schema.space.create('test', { engine = 'vinyl' })
----
-...
-index = space:create_index('primary')
----
-...
-space:insert({1, 2})
----
-- [1, 2]
-...
-index2 = space:create_index('secondary', { parts = {2, 'unsigned'} })
----
-- error: Vinyl does not support building an index for a non-empty space
-...
-space.index.primary:alter({parts = {1, 'unsigned', 2, 'unsigned'}})
----
-- error: Vinyl does not support building an index for a non-empty space
-...
-#box.space._index:select({space.id})
----
-- 1
-...
-box.space._index:get{space.id, 0}[6]
----
-- [[0, 'unsigned']]
-...
-space:drop()
----
-...
-space = box.schema.space.create('test', { engine = 'vinyl' })
----
-...
-index = space:create_index('primary')
----
-...
-space:insert({1, 2})
----
-- [1, 2]
-...
-index2 = space:create_index('secondary', { parts = {2, 'unsigned'} })
----
-- error: Vinyl does not support building an index for a non-empty space
-...
-space.index.primary:alter({parts = {1, 'unsigned', 2, 'unsigned'}})
----
-- error: Vinyl does not support building an index for a non-empty space
-...
-#box.space._index:select({space.id})
----
-- 1
-...
-box.space._index:get{space.id, 0}[6]
----
-- [[0, 'unsigned']]
-...
-space:delete({1})
----
-...
--- must fail because vy_mems have data
-index2 = space:create_index('secondary', { parts = {2, 'unsigned'} })
----
-- error: Vinyl does not support building an index for a non-empty space
+- [1, 1]
 ...
-space.index.primary:alter({parts = {1, 'unsigned', 2, 'unsigned'}})
+pk:alter{parts = {2, 'unsigned'}} -- error: mem not empty
 ---
-- error: Vinyl does not support building an index for a non-empty space
+- error: Vinyl does not support rebuilding the primary index of a non-empty space
 ...
 box.snapshot()
 ---
 - ok
 ...
-while space.index.primary:info().rows ~= 0 do fiber.sleep(0.01) end
----
-...
--- after a dump REPLACE + DELETE = nothing, so the space is empty now and
--- can be altered.
-index2 = space:create_index('secondary', { parts = {2, 'unsigned'} })
----
-...
-space.index.primary:alter({parts = {1, 'unsigned', 2, 'unsigned'}})
----
-...
-#box.space._index:select({space.id})
----
-- 2
-...
-box.space._index:get{space.id, 0}[6]
+pk:alter{parts = {2, 'unsigned'}} -- error: run not empty
 ---
-- [[0, 'unsigned'], [1, 'unsigned']]
+- error: Vinyl does not support rebuilding the primary index of a non-empty space
 ...
-space:insert({1, 2})
----
-- [1, 2]
-...
-index:select{}
----
-- - [1, 2]
-...
-index2:select{}
----
-- - [1, 2]
-...
-space:drop()
----
-...
-space = box.schema.space.create('test', { engine = 'vinyl' })
----
-...
-index = space:create_index('primary', { run_count_per_level = 2 })
+space:replace{2, 2}
 ---
+- [2, 2]
 ...
-space:insert({1, 2})
+space:delete{1}
 ---
-- [1, 2]
 ...
-box.snapshot()
+space:delete{2}
 ---
-- ok
 ...
-space:delete({1})
+pk:alter{parts = {2, 'unsigned'}} -- error: mem/run not empty
 ---
+- error: Vinyl does not support rebuilding the primary index of a non-empty space
 ...
 box.snapshot()
 ---
 - ok
 ...
-while space.index.primary:info().run_count ~= 2 do fiber.sleep(0.01) end
----
-...
--- must fail because vy_runs have data
-index2 = space:create_index('secondary', { parts = {2, 'unsigned'} })
----
-- error: Vinyl does not support building an index for a non-empty space
-...
-space.index.primary:alter({parts = {1, 'unsigned', 2, 'unsigned'}})
----
-- error: Vinyl does not support building an index for a non-empty space
-...
--- After compaction the REPLACE + DELETE + DELETE = nothing, so
--- the space is now empty and can be altered.
-space:delete({1})
----
-...
--- Make sure the run is big enough to trigger compaction.
-space:replace({2, 3})
----
-- [2, 3]
-...
-space:delete({2})
----
-...
-box.snapshot()
+-- wait for compaction to complete
+while pk:info().disk.compact.count == 0 do fiber.sleep(0.01) end
 ---
-- ok
 ...
--- Wait until the dump is finished.
-while space.index.primary:info().rows ~= 0 do fiber.sleep(0.01) end
+pk:alter{parts = {2, 'unsigned'}} -- success: space is empty now
 ---
 ...
-index2 = space:create_index('secondary', { parts = {2, 'unsigned'} })
+space:replace{1, 2}
 ---
+- [1, 2]
 ...
-space.index.primary:alter({parts = {1, 'unsigned', 2, 'unsigned'}})
+space:get(2)
 ---
+- [1, 2]
 ...
 space:drop()
 ---
@@ -291,7 +150,7 @@ space:auto_increment{3}
 ...
 box.space._index:replace{space.id, 0, 'pk', 'tree', {unique=true}, {{0, 'unsigned'}, {1, 'unsigned'}}}
 ---
-- error: Vinyl does not support building an index for a non-empty space
+- error: Vinyl does not support rebuilding the primary index of a non-empty space
 ...
 space:select{}
 ---
@@ -747,14 +606,6 @@ s.index.secondary.unique
 ---
 - false
 ...
-s.index.secondary:alter{unique = true} -- error
----
-- error: Vinyl does not support building an index for a non-empty space
-...
-s.index.secondary.unique
----
-- false
-...
 s:insert{2, 10}
 ---
 - [2, 10]
@@ -771,27 +622,260 @@ s:drop()
 s = box.schema.space.create('test', {engine = 'vinyl'})
 ---
 ...
-_ = s:create_index('i1')
+pk = s:create_index('pk', {parts = {1, 'unsigned'}})
+---
+...
+s:replace{1}
 ---
+- [1]
 ...
-_ = s:create_index('i2', {parts = {2, 'integer'}})
+-- Extending field type is allowed without rebuild.
+pk:alter{parts = {1, 'integer'}}
 ---
 ...
-_ = s:create_index('i3', {parts = {{3, 'string', is_nullable = true}}})
+-- Should fail as we do not support rebuilding the primary index of a non-empty space.
+pk:alter{parts = {1, 'unsigned'}}
 ---
+- error: Vinyl does not support rebuilding the primary index of a non-empty space
 ...
-_ = s:replace{1, 1, 'test'}
+s:replace{-1}
 ---
+- [-1]
 ...
--- Should fail with 'Vinyl does not support building an index for a non-empty space'.
-s.index.i2:alter{parts = {2, 'unsigned'}}
+s:drop()
 ---
-- error: Vinyl does not support building an index for a non-empty space
 ...
-s.index.i3:alter{parts = {{3, 'string', is_nullable = false}}}
+--
+-- Check that all modifications done to the space during index build
+-- are reflected in the new index.
+--
+math.randomseed(os.time())
 ---
-- error: Vinyl does not support building an index for a non-empty space
 ...
-s:drop()
+s = box.schema.space.create('test', {engine = 'vinyl'})
+---
+...
+_ = s:create_index('pk')
+---
+...
+test_run:cmd("setopt delimiter ';'")
+---
+- true
+...
+box.begin();
+---
+...
+for i = 1, 1000 do
+    if (i % 100 == 0) then
+        box.commit()
+        box.begin()
+    end
+    if i % 300 == 0 then
+        box.snapshot()
+    end
+    box.space.test:replace{i, i, i}
+end;
+---
+...
+box.commit();
+---
+...
+last_val = 1000;
+---
+...
+function gen_load()
+    fiber.sleep(0.001)
+    local s = box.space.test
+    for i = 1, 200 do
+        local op = math.random(4)
+        local key = math.random(1000)
+        local val1 = math.random(1000)
+        local val2 = last_val + 1
+        last_val = val2
+        if op == 1 then
+            pcall(s.insert, s, {key, val1, val2})
+        elseif op == 2 then
+            pcall(s.replace, s, {key, val1, val2})
+        elseif op == 3 then
+            pcall(s.delete, s, {key})
+        elseif op == 4 then
+            pcall(s.upsert, s, {key, val1, val2}, {{'=', 2, val1}, {'=', 3, val2}})
+        end
+    end
+end;
 ---
 ...
+test_run:cmd("setopt delimiter ''");
+---
+- true
+...
+ch = fiber.channel(1)
+---
+...
+_ = fiber.create(function() gen_load() ch:put(true) end)
+---
+...
+_ = box.space.test:create_index('sk', {unique = false, parts = {2, 'unsigned'}})
+---
+...
+ch:get()
+---
+- true
+...
+_ = fiber.create(function() gen_load() ch:put(true) end)
+---
+...
+_ = box.space.test:create_index('tk', {unique = true, parts = {3, 'unsigned'}})
+---
+...
+ch:get()
+---
+- true
+...
+box.space.test.index.pk:count() == box.space.test.index.sk:count()
+---
+- true
+...
+box.space.test.index.pk:count() == box.space.test.index.tk:count()
+---
+- true
+...
+test_run:cmd("restart server default")
+box.space.test.index.pk:count() == box.space.test.index.sk:count()
+---
+- true
+...
+box.space.test.index.pk:count() == box.space.test.index.tk:count()
+---
+- true
+...
+box.snapshot()
+---
+- ok
+...
+box.space.test.index.pk:count() == box.space.test.index.sk:count()
+---
+- true
+...
+box.space.test.index.pk:count() == box.space.test.index.tk:count()
+---
+- true
+...
+box.space.test:drop()
+---
+...
+--
+-- Check that creation of a secondary index triggers dump
+-- if memory quota is exceeded.
+--
+test_run:cmd("create server test with script='vinyl/low_quota.lua'")
+---
+- true
+...
+test_run:cmd("start server test with args='1048576'")
+---
+- true
+...
+test_run:cmd("switch test")
+---
+- true
+...
+_ = box.schema.space.create('test', {engine = 'vinyl'})
+---
+...
+_ = box.space.test:create_index('pk')
+---
+...
+pad = string.rep('x', 1000)
+---
+...
+test_run:cmd("setopt delimiter ';'")
+---
+- true
+...
+box.begin();
+---
+...
+for i = 1, 1000 do
+    if (i % 100 == 0) then
+        box.commit()
+        box.begin()
+    end
+    box.space.test:replace{i, i, pad}
+end;
+---
+...
+box.commit();
+---
+...
+test_run:cmd("setopt delimiter ''");
+---
+- true
+...
+_ = box.space.test:create_index('sk', {parts = {2, 'unsigned', 3, 'string'}})
+---
+...
+box.space.test.index.sk:info().disk.dump.count > 1
+---
+- true
+...
+box.space.test.index.sk:count()
+---
+- 1000
+...
+test_run:cmd("restart server test with args='1048576'")
+box.space.test.index.sk:count()
+---
+- 1000
+...
+box.space.test.index.sk:drop()
+---
+...
+box.snapshot()
+---
+- ok
+...
+--
+-- Check that run files left from an index we failed to build
+-- are removed by garbage collection.
+--
+fio = require('fio')
+---
+...
+box.cfg{checkpoint_count = 1}
+---
+...
+_ = box.space.test:replace{1001, 1, string.rep('x', 1000)}
+---
+...
+box.space.test:create_index('sk', {parts = {2, 'unsigned', 3, 'string'}})
+---
+- error: Duplicate key exists in unique index 'sk' in space 'test'
+...
+#fio.listdir(fio.pathjoin(box.cfg.vinyl_dir, box.space.test.id, 1)) > 0
+---
+- true
+...
+box.snapshot()
+---
+- ok
+...
+#fio.listdir(fio.pathjoin(box.cfg.vinyl_dir, box.space.test.id, 1)) == 0
+---
+- true
+...
+box.space.test:drop()
+---
+...
+test_run:cmd("switch default")
+---
+- true
+...
+test_run:cmd("stop server test")
+---
+- true
+...
+test_run:cmd("cleanup server test")
+---
+- true
+...
diff --git a/test/vinyl/ddl.test.lua b/test/vinyl/ddl.test.lua
index 637a331d..da40a12e 100644
--- a/test/vinyl/ddl.test.lua
+++ b/test/vinyl/ddl.test.lua
@@ -23,76 +23,23 @@ space = box.schema.space.create('test', { engine = 'vinyl' })
 index = space:create_index('primary', {type = 'hash'})
 space:drop()
 
--- creation of a new index and altering the definition of an existing
--- index are unsupported for non-empty spaces
+-- rebuild of the primary index is not supported
 space = box.schema.space.create('test', { engine = 'vinyl' })
-index = space:create_index('primary')
-space:insert({1})
--- fail because of wrong tuple format {1}, but need {1, ...}
-index2 = space:create_index('secondary', { parts = {2, 'unsigned'} })
-space.index.primary:alter({parts = {1, 'unsigned', 2, 'unsigned'}})
-#box.space._index:select({space.id})
-box.space._index:get{space.id, 0}[6]
-space:drop()
-
-space = box.schema.space.create('test', { engine = 'vinyl' })
-index = space:create_index('primary')
-space:insert({1, 2})
-index2 = space:create_index('secondary', { parts = {2, 'unsigned'} })
-space.index.primary:alter({parts = {1, 'unsigned', 2, 'unsigned'}})
-#box.space._index:select({space.id})
-box.space._index:get{space.id, 0}[6]
-space:drop()
-
-space = box.schema.space.create('test', { engine = 'vinyl' })
-index = space:create_index('primary')
-space:insert({1, 2})
-index2 = space:create_index('secondary', { parts = {2, 'unsigned'} })
-space.index.primary:alter({parts = {1, 'unsigned', 2, 'unsigned'}})
-#box.space._index:select({space.id})
-box.space._index:get{space.id, 0}[6]
-space:delete({1})
-
--- must fail because vy_mems have data
-index2 = space:create_index('secondary', { parts = {2, 'unsigned'} })
-space.index.primary:alter({parts = {1, 'unsigned', 2, 'unsigned'}})
-box.snapshot()
-while space.index.primary:info().rows ~= 0 do fiber.sleep(0.01) end
-
--- after a dump REPLACE + DELETE = nothing, so the space is empty now and
--- can be altered.
-index2 = space:create_index('secondary', { parts = {2, 'unsigned'} })
-space.index.primary:alter({parts = {1, 'unsigned', 2, 'unsigned'}})
-#box.space._index:select({space.id})
-box.space._index:get{space.id, 0}[6]
-space:insert({1, 2})
-index:select{}
-index2:select{}
-space:drop()
-
-space = box.schema.space.create('test', { engine = 'vinyl' })
-index = space:create_index('primary', { run_count_per_level = 2 })
-space:insert({1, 2})
-box.snapshot()
-space:delete({1})
+pk = space:create_index('pk', {run_count_per_level = 1, run_size_ratio = 10})
+space:replace{1, 1}
+pk:alter{parts = {2, 'unsigned'}} -- error: mem not empty
 box.snapshot()
-while space.index.primary:info().run_count ~= 2 do fiber.sleep(0.01) end
--- must fail because vy_runs have data
-index2 = space:create_index('secondary', { parts = {2, 'unsigned'} })
-space.index.primary:alter({parts = {1, 'unsigned', 2, 'unsigned'}})
-
--- After compaction the REPLACE + DELETE + DELETE = nothing, so
--- the space is now empty and can be altered.
-space:delete({1})
--- Make sure the run is big enough to trigger compaction.
-space:replace({2, 3})
-space:delete({2})
+pk:alter{parts = {2, 'unsigned'}} -- error: run not empty
+space:replace{2, 2}
+space:delete{1}
+space:delete{2}
+pk:alter{parts = {2, 'unsigned'}} -- error: mem/run not empty
 box.snapshot()
--- Wait until the dump is finished.
-while space.index.primary:info().rows ~= 0 do fiber.sleep(0.01) end
-index2 = space:create_index('secondary', { parts = {2, 'unsigned'} })
-space.index.primary:alter({parts = {1, 'unsigned', 2, 'unsigned'}})
-
+-- wait for compaction to complete
+while pk:info().disk.compact.count == 0 do fiber.sleep(0.01) end
+pk:alter{parts = {2, 'unsigned'}} -- success: space is empty now
+space:replace{1, 2}
+space:get(2)
 space:drop()
 
 --
@@ -273,19 +220,140 @@ _ = s:create_index('secondary', {unique = true, parts = {2, 'unsigned'}})
 s:insert{1, 10}
 s.index.secondary:alter{unique = false} -- ok
 s.index.secondary.unique
-s.index.secondary:alter{unique = true} -- error
-s.index.secondary.unique
 s:insert{2, 10}
 s.index.secondary:select(10)
 s:drop()
 
 -- Narrowing indexed field type entails index rebuild.
 s = box.schema.space.create('test', {engine = 'vinyl'})
-_ = s:create_index('i1')
-_ = s:create_index('i2', {parts = {2, 'integer'}})
-_ = s:create_index('i3', {parts = {{3, 'string', is_nullable = true}}})
-_ = s:replace{1, 1, 'test'}
--- Should fail with 'Vinyl does not support building an index for a non-empty space'.
-s.index.i2:alter{parts = {2, 'unsigned'}}
-s.index.i3:alter{parts = {{3, 'string', is_nullable = false}}}
+pk = s:create_index('pk', {parts = {1, 'unsigned'}})
+s:replace{1}
+-- Extending field type is allowed without rebuild.
+pk:alter{parts = {1, 'integer'}}
+-- Should fail as we do not support rebuilding the primary index of a non-empty space.
+pk:alter{parts = {1, 'unsigned'}}
+s:replace{-1}
 s:drop()
+
+--
+-- Check that all modifications done to the space during index build
+-- are reflected in the new index.
+--
+math.randomseed(os.time())
+
+s = box.schema.space.create('test', {engine = 'vinyl'})
+_ = s:create_index('pk')
+
+test_run:cmd("setopt delimiter ';'")
+
+box.begin();
+for i = 1, 1000 do
+    if (i % 100 == 0) then
+        box.commit()
+        box.begin()
+    end
+    if i % 300 == 0 then
+        box.snapshot()
+    end
+    box.space.test:replace{i, i, i}
+end;
+box.commit();
+
+last_val = 1000;
+
+function gen_load()
+    fiber.sleep(0.001)
+    local s = box.space.test
+    for i = 1, 200 do
+        local op = math.random(4)
+        local key = math.random(1000)
+        local val1 = math.random(1000)
+        local val2 = last_val + 1
+        last_val = val2
+        if op == 1 then
+            pcall(s.insert, s, {key, val1, val2})
+        elseif op == 2 then
+            pcall(s.replace, s, {key, val1, val2})
+        elseif op == 3 then
+            pcall(s.delete, s, {key})
+        elseif op == 4 then
+            pcall(s.upsert, s, {key, val1, val2}, {{'=', 2, val1}, {'=', 3, val2}})
+        end
+    end
+end;
+
+test_run:cmd("setopt delimiter ''");
+
+ch = fiber.channel(1)
+
+_ = fiber.create(function() gen_load() ch:put(true) end)
+_ = box.space.test:create_index('sk', {unique = false, parts = {2, 'unsigned'}})
+ch:get()
+
+_ = fiber.create(function() gen_load() ch:put(true) end)
+_ = box.space.test:create_index('tk', {unique = true, parts = {3, 'unsigned'}})
+ch:get()
+
+box.space.test.index.pk:count() == box.space.test.index.sk:count()
+box.space.test.index.pk:count() == box.space.test.index.tk:count()
+
+test_run:cmd("restart server default")
+
+box.space.test.index.pk:count() == box.space.test.index.sk:count()
+box.space.test.index.pk:count() == box.space.test.index.tk:count()
+box.snapshot()
+box.space.test.index.pk:count() == box.space.test.index.sk:count()
+box.space.test.index.pk:count() == box.space.test.index.tk:count()
+box.space.test:drop()
+
+--
+-- Check that creation of a secondary index triggers dump
+-- if memory quota is exceeded.
+--
+test_run:cmd("create server test with script='vinyl/low_quota.lua'")
+test_run:cmd("start server test with args='1048576'")
+test_run:cmd("switch test")
+
+_ = box.schema.space.create('test', {engine = 'vinyl'})
+_ = box.space.test:create_index('pk')
+
+pad = string.rep('x', 1000)
+
+test_run:cmd("setopt delimiter ';'")
+box.begin();
+for i = 1, 1000 do
+    if (i % 100 == 0) then
+        box.commit()
+        box.begin()
+    end
+    box.space.test:replace{i, i, pad}
+end;
+box.commit();
+test_run:cmd("setopt delimiter ''");
+
+_ = box.space.test:create_index('sk', {parts = {2, 'unsigned', 3, 'string'}})
+box.space.test.index.sk:info().disk.dump.count > 1
+box.space.test.index.sk:count()
+
+test_run:cmd("restart server test with args='1048576'")
+
+box.space.test.index.sk:count()
+box.space.test.index.sk:drop()
+box.snapshot()
+
+--
+-- Check that run files left from an index we failed to build
+-- are removed by garbage collection.
+--
+fio = require('fio')
+box.cfg{checkpoint_count = 1}
+_ = box.space.test:replace{1001, 1, string.rep('x', 1000)}
+box.space.test:create_index('sk', {parts = {2, 'unsigned', 3, 'string'}})
+#fio.listdir(fio.pathjoin(box.cfg.vinyl_dir, box.space.test.id, 1)) > 0
+box.snapshot()
+#fio.listdir(fio.pathjoin(box.cfg.vinyl_dir, box.space.test.id, 1)) == 0
+box.space.test:drop()
+
+test_run:cmd("switch default")
+test_run:cmd("stop server test")
+test_run:cmd("cleanup server test")
diff --git a/test/vinyl/errinj.result b/test/vinyl/errinj.result
index fd21f7bb..fb090109 100644
--- a/test/vinyl/errinj.result
+++ b/test/vinyl/errinj.result
@@ -1395,3 +1395,259 @@ s:count() -- 200
 s:drop()
 ---
 ...
+--
+-- Check that ALTER is aborted if a tuple inserted during index build
+-- doesn't conform to the new format.
+--
+s = box.schema.space.create('test', {engine = 'vinyl'})
+---
+...
+_ = s:create_index('pk', {page_size = 16})
+---
+...
+pad = string.rep('x', 16)
+---
+...
+for i = 101, 200 do s:replace{i, i, pad} end
+---
+...
+box.snapshot()
+---
+- ok
+...
+ch = fiber.channel(1)
+---
+...
+test_run:cmd("setopt delimiter ';'")
+---
+- true
+...
+_ = fiber.create(function()
+    fiber.sleep(0.01)
+    for i = 1, 100 do
+        s:replace{i}
+    end
+    ch:put(true)
+end);
+---
+...
+test_run:cmd("setopt delimiter ''");
+---
+- true
+...
+errinj.set("ERRINJ_VY_READ_PAGE_TIMEOUT", 0.001)
+---
+- ok
+...
+s:create_index('sk', {parts = {2, 'unsigned'}}) -- must fail
+---
+- error: Tuple field count 1 is less than required by space format or defined indexes
+    (expected at least 2)
+...
+errinj.set("ERRINJ_VY_READ_PAGE_TIMEOUT", 0)
+---
+- ok
+...
+ch:get()
+---
+- true
+...
+s:count() -- 200
+---
+- 200
+...
+s:drop()
+---
+...
+--
+-- Check that ALTER is aborted if a tuple inserted during index build
+-- violates unique constraint.
+--
+s = box.schema.space.create('test', {engine = 'vinyl'})
+---
+...
+_ = s:create_index('pk', {page_size = 16})
+---
+...
+pad = string.rep('x', 16)
+---
+...
+for i = 101, 200 do s:replace{i, i, pad} end
+---
+...
+box.snapshot()
+---
+- ok
+...
+ch = fiber.channel(1)
+---
+...
+test_run:cmd("setopt delimiter ';'")
+---
+- true
+...
+_ = fiber.create(function()
+    fiber.sleep(0.01)
+    for i = 1, 100 do
+        s:replace{i, i + 1}
+    end
+    ch:put(true)
+end);
+---
+...
+test_run:cmd("setopt delimiter ''");
+---
+- true
+...
+errinj.set("ERRINJ_VY_READ_PAGE_TIMEOUT", 0.001)
+---
+- ok
+...
+s:create_index('sk', {parts = {2, 'unsigned'}}) -- must fail
+---
+- error: Duplicate key exists in unique index 'sk' in space 'test'
+...
+errinj.set("ERRINJ_VY_READ_PAGE_TIMEOUT", 0)
+---
+- ok
+...
+ch:get()
+---
+- true
+...
+s:count() -- 200
+---
+- 200
+...
+s:drop()
+---
+...
+--
+-- Check that modifications done to the space during the final dump
+-- of a newly built index are recovered properly.
+--
+s = box.schema.space.create('test', {engine = 'vinyl'})
+---
+...
+_ = s:create_index('pk')
+---
+...
+for i = 1, 5 do s:replace{i, i} end
+---
+...
+errinj.set("ERRINJ_VY_RUN_WRITE_TIMEOUT", 0.1)
+---
+- ok
+...
+ch = fiber.channel(1)
+---
+...
+_ = fiber.create(function() s:create_index('sk', {parts = {2, 'integer'}}) ch:put(true) end)
+---
+...
+errinj.set("ERRINJ_VY_RUN_WRITE_TIMEOUT", 0)
+---
+- ok
+...
+fiber.sleep(0.01)
+---
+...
+_ = s:delete{1}
+---
+...
+_ = s:replace{2, -2}
+---
+...
+_ = s:delete{2}
+---
+...
+_ = s:replace{3, -3}
+---
+...
+_ = s:replace{3, -2}
+---
+...
+_ = s:replace{3, -1}
+---
+...
+_ = s:delete{3}
+---
+...
+_ = s:upsert({3, 3}, {{'=', 2, 1}})
+---
+...
+_ = s:upsert({3, 3}, {{'=', 2, 2}})
+---
+...
+_ = s:delete{3}
+---
+...
+_ = s:replace{4, -1}
+---
+...
+_ = s:replace{4, -2}
+---
+...
+_ = s:replace{4, -4}
+---
+...
+_ = s:upsert({5, 1}, {{'=', 2, 1}})
+---
+...
+_ = s:upsert({5, 2}, {{'=', 2, -5}})
+---
+...
+_ = s:replace{6, -6}
+---
+...
+_ = s:upsert({7, -7}, {{'=', 2, -7}})
+---
+...
+ch:get()
+---
+- true
+...
+s.index.sk:select()
+---
+- - [7, -7]
+  - [6, -6]
+  - [5, -5]
+  - [4, -4]
+...
+s.index.sk:info().memory.rows
+---
+- 27
+...
+test_run:cmd('restart server default')
+s = box.space.test
+---
+...
+s.index.sk:select()
+---
+- - [7, -7]
+  - [6, -6]
+  - [5, -5]
+  - [4, -4]
+...
+s.index.sk:info().memory.rows
+---
+- 27
+...
+box.snapshot()
+---
+- ok
+...
+s.index.sk:select()
+---
+- - [7, -7]
+  - [6, -6]
+  - [5, -5]
+  - [4, -4]
+...
+s.index.sk:info().memory.rows
+---
+- 0
+...
+s:drop()
+---
+...
diff --git a/test/vinyl/errinj.test.lua b/test/vinyl/errinj.test.lua
index 64d04c62..e434b098 100644
--- a/test/vinyl/errinj.test.lua
+++ b/test/vinyl/errinj.test.lua
@@ -548,3 +548,118 @@ ch:get()
 
 s:count() -- 200
 s:drop()
+
+--
+-- Check that ALTER is aborted if a tuple inserted during index build
+-- doesn't conform to the new format.
+--
+s = box.schema.space.create('test', {engine = 'vinyl'})
+_ = s:create_index('pk', {page_size = 16})
+
+pad = string.rep('x', 16)
+for i = 101, 200 do s:replace{i, i, pad} end
+box.snapshot()
+
+ch = fiber.channel(1)
+test_run:cmd("setopt delimiter ';'")
+_ = fiber.create(function()
+    fiber.sleep(0.01)
+    for i = 1, 100 do
+        s:replace{i}
+    end
+    ch:put(true)
+end);
+test_run:cmd("setopt delimiter ''");
+
+errinj.set("ERRINJ_VY_READ_PAGE_TIMEOUT", 0.001)
+s:create_index('sk', {parts = {2, 'unsigned'}}) -- must fail
+errinj.set("ERRINJ_VY_READ_PAGE_TIMEOUT", 0)
+
+ch:get()
+
+s:count() -- 200
+s:drop()
+
+--
+-- Check that ALTER is aborted if a tuple inserted during index build
+-- violates unique constraint.
+--
+s = box.schema.space.create('test', {engine = 'vinyl'})
+_ = s:create_index('pk', {page_size = 16})
+
+pad = string.rep('x', 16)
+for i = 101, 200 do s:replace{i, i, pad} end
+box.snapshot()
+
+ch = fiber.channel(1)
+test_run:cmd("setopt delimiter ';'")
+_ = fiber.create(function()
+    fiber.sleep(0.01)
+    for i = 1, 100 do
+        s:replace{i, i + 1}
+    end
+    ch:put(true)
+end);
+test_run:cmd("setopt delimiter ''");
+
+errinj.set("ERRINJ_VY_READ_PAGE_TIMEOUT", 0.001)
+s:create_index('sk', {parts = {2, 'unsigned'}}) -- must fail
+errinj.set("ERRINJ_VY_READ_PAGE_TIMEOUT", 0)
+
+ch:get()
+
+s:count() -- 200
+s:drop()
+
+--
+-- Check that modifications done to the space during the final dump
+-- of a newly built index are recovered properly.
+--
+s = box.schema.space.create('test', {engine = 'vinyl'})
+_ = s:create_index('pk')
+
+for i = 1, 5 do s:replace{i, i} end
+
+errinj.set("ERRINJ_VY_RUN_WRITE_TIMEOUT", 0.1)
+ch = fiber.channel(1)
+_ = fiber.create(function() s:create_index('sk', {parts = {2, 'integer'}}) ch:put(true) end)
+errinj.set("ERRINJ_VY_RUN_WRITE_TIMEOUT", 0)
+
+fiber.sleep(0.01)
+
+_ = s:delete{1}
+_ = s:replace{2, -2}
+_ = s:delete{2}
+_ = s:replace{3, -3}
+_ = s:replace{3, -2}
+_ = s:replace{3, -1}
+_ = s:delete{3}
+_ = s:upsert({3, 3}, {{'=', 2, 1}})
+_ = s:upsert({3, 3}, {{'=', 2, 2}})
+_ = s:delete{3}
+_ = s:replace{4, -1}
+_ = s:replace{4, -2}
+_ = s:replace{4, -4}
+_ = s:upsert({5, 1}, {{'=', 2, 1}})
+_ = s:upsert({5, 2}, {{'=', 2, -5}})
+_ = s:replace{6, -6}
+_ = s:upsert({7, -7}, {{'=', 2, -7}})
+
+ch:get()
+
+s.index.sk:select()
+s.index.sk:info().memory.rows
+
+test_run:cmd('restart server default')
+
+s = box.space.test
+
+s.index.sk:select()
+s.index.sk:info().memory.rows
+
+box.snapshot()
+
+s.index.sk:select()
+s.index.sk:info().memory.rows
+
+s:drop()
diff --git a/test/vinyl/errinj_gc.result b/test/vinyl/errinj_gc.result
index 704cae6e..62204b93 100644
--- a/test/vinyl/errinj_gc.result
+++ b/test/vinyl/errinj_gc.result
@@ -180,9 +180,6 @@ s:select()
 - - [100, '12345']
   - [200, '67890']
 ...
---
--- Cleanup.
---
 s:drop()
 ---
 ...
@@ -199,3 +196,78 @@ temp:drop()
 box.cfg{checkpoint_count = default_checkpoint_count}
 ---
 ...
+--
+-- Check that if we failed to clean up an incomplete index before restart,
+-- we will clean it up after recovery.
+--
+fio = require('fio')
+---
+...
+fiber = require('fiber')
+---
+...
+s = box.schema.space.create('test', {engine = 'vinyl'})
+---
+...
+_ = s:create_index('pk')
+---
+...
+_ = s:insert{1, 1}
+---
+...
+box.error.injection.set('ERRINJ_WAL_DELAY', true)
+---
+- ok
+...
+ch = fiber.channel(1)
+---
+...
+_ = fiber.create(function() pcall(s.create_index, s, 'sk', {parts = {2, 'unsigned'}}) ch:put(true) end)
+---
+...
+-- wait for ALTER to stall on WAL after preparing the new index
+while s.index.pk:info().disk.dump.count == 0 do fiber.sleep(0.001) end
+---
+...
+box.error.injection.set('ERRINJ_VY_GC', true)
+---
+- ok
+...
+box.error.injection.set('ERRINJ_VY_LOG_FLUSH', true);
+---
+- ok
+...
+box.error.injection.set("ERRINJ_WAL_WRITE", true)
+---
+- ok
+...
+box.error.injection.set('ERRINJ_WAL_DELAY', false)
+---
+- ok
+...
+ch:get()
+---
+- true
+...
+#fio.listdir(fio.pathjoin(box.cfg.vinyl_dir, s.id, 1)) > 0
+---
+- true
+...
+test_run:cmd('restart server default')
+box.snapshot()
+---
+- ok
+...
+fio = require('fio')
+---
+...
+s = box.space.test
+---
+...
+#fio.listdir(fio.pathjoin(box.cfg.vinyl_dir, s.id, 1)) == 0
+---
+- true
+...
+s:drop()
+---
+...
diff --git a/test/vinyl/errinj_gc.test.lua b/test/vinyl/errinj_gc.test.lua
index 44291091..191e33d1 100644
--- a/test/vinyl/errinj_gc.test.lua
+++ b/test/vinyl/errinj_gc.test.lua
@@ -85,10 +85,6 @@ file_count()
 
 s:select()
 
---
--- Cleanup.
---
-
 s:drop()
 gc()
 file_count()
@@ -96,3 +92,37 @@ file_count()
 temp:drop()
 
 box.cfg{checkpoint_count = default_checkpoint_count}
+
+--
+-- Check that if we failed to clean up an incomplete index before restart,
+-- we will clean it up after recovery.
+--
+fio = require('fio')
+fiber = require('fiber')
+
+s = box.schema.space.create('test', {engine = 'vinyl'})
+_ = s:create_index('pk')
+_ = s:insert{1, 1}
+
+box.error.injection.set('ERRINJ_WAL_DELAY', true)
+ch = fiber.channel(1)
+_ = fiber.create(function() pcall(s.create_index, s, 'sk', {parts = {2, 'unsigned'}}) ch:put(true) end)
+
+-- wait for ALTER to stall on WAL after preparing the new index
+while s.index.pk:info().disk.dump.count == 0 do fiber.sleep(0.001) end
+
+box.error.injection.set('ERRINJ_VY_GC', true)
+box.error.injection.set('ERRINJ_VY_LOG_FLUSH', true);
+box.error.injection.set("ERRINJ_WAL_WRITE", true)
+box.error.injection.set('ERRINJ_WAL_DELAY', false)
+ch:get()
+
+#fio.listdir(fio.pathjoin(box.cfg.vinyl_dir, s.id, 1)) > 0
+
+test_run:cmd('restart server default')
+box.snapshot()
+
+fio = require('fio')
+s = box.space.test
+#fio.listdir(fio.pathjoin(box.cfg.vinyl_dir, s.id, 1)) == 0
+s:drop()
diff --git a/test/vinyl/errinj_vylog.result b/test/vinyl/errinj_vylog.result
index ca23cb45..89428d71 100644
--- a/test/vinyl/errinj_vylog.result
+++ b/test/vinyl/errinj_vylog.result
@@ -276,3 +276,60 @@ s1:drop()
 s2:drop()
 ---
 ...
+--
+-- Check that an index that was prepared, but not committed,
+-- is recovered properly.
+--
+fiber = require('fiber')
+---
+...
+s = box.schema.space.create('test', {engine = 'vinyl'})
+---
+...
+_ = s:create_index('pk')
+---
+...
+_ = s:insert{1, 1}
+---
+...
+box.error.injection.set('ERRINJ_WAL_DELAY', true)
+---
+- ok
+...
+ch = fiber.channel(1)
+---
+...
+_ = fiber.create(function() s:create_index('sk', {parts = {2, 'unsigned'}}) ch:put(true) end)
+---
+...
+-- wait for ALTER to stall on WAL after preparing the new index
+while s.index.pk:info().disk.dump.count == 0 do fiber.sleep(0.001) end
+---
+...
+box.error.injection.set('ERRINJ_VY_LOG_FLUSH', true);
+---
+- ok
+...
+box.error.injection.set('ERRINJ_WAL_DELAY', false)
+---
+- ok
+...
+ch:get()
+---
+- true
+...
+test_run:cmd('restart server default')
+s = box.space.test
+---
+...
+s.index.pk:select()
+---
+- - [1, 1]
+...
+s.index.sk:select()
+---
+- - [1, 1]
+...
+s:drop()
+---
+...
diff --git a/test/vinyl/errinj_vylog.test.lua b/test/vinyl/errinj_vylog.test.lua
index 3d90755d..514313cf 100644
--- a/test/vinyl/errinj_vylog.test.lua
+++ b/test/vinyl/errinj_vylog.test.lua
@@ -130,3 +130,32 @@ s1:select()
 s2:select()
 s1:drop()
 s2:drop()
+
+--
+-- Check that an index that was prepared, but not committed,
+-- is recovered properly.
+--
+fiber = require('fiber')
+
+s = box.schema.space.create('test', {engine = 'vinyl'})
+_ = s:create_index('pk')
+_ = s:insert{1, 1}
+
+box.error.injection.set('ERRINJ_WAL_DELAY', true)
+ch = fiber.channel(1)
+_ = fiber.create(function() s:create_index('sk', {parts = {2, 'unsigned'}}) ch:put(true) end)
+
+-- wait for ALTER to stall on WAL after preparing the new index
+while s.index.pk:info().disk.dump.count == 0 do fiber.sleep(0.001) end
+
+box.error.injection.set('ERRINJ_VY_LOG_FLUSH', true);
+box.error.injection.set('ERRINJ_WAL_DELAY', false)
+ch:get()
+
+test_run:cmd('restart server default')
+
+s = box.space.test
+s.index.pk:select()
+s.index.sk:select()
+
+s:drop()
diff --git a/test/vinyl/gh.result b/test/vinyl/gh.result
index 47695acb..76beab09 100644
--- a/test/vinyl/gh.result
+++ b/test/vinyl/gh.result
@@ -144,7 +144,7 @@ s:insert{5, 5}
 ...
 s.index.primary:alter({parts={2,'unsigned'}})
 ---
-- error: Vinyl does not support building an index for a non-empty space
+- error: Vinyl does not support rebuilding the primary index of a non-empty space
 ...
 s:drop()
 ---
-- 
2.11.0

Re: [PATCH v2 1/8] vinyl: allocate key parts in vy_recovery_do_create_lsm

[Konstantin Osipov]

* Vladimir Davydov <vdavydov.dev@gmail.com> [18/05/28 07:49]:
> Allocation of vy_lsm_recovery_info::key_parts is a part of the struct
> initialization, which is handled by vy_recovery_do_create_lsm().

Pushed.


-- 
Konstantin Osipov, Moscow, Russia, +7 903 626 22 32
http://tarantool.io - www.twitter.com/kostja_osipov

Re: [PATCH v2 2/8] vinyl: update recovery context with records written during recovery

[Konstantin Osipov]

* Vladimir Davydov <vdavydov.dev@gmail.com> [18/05/28 07:49]:

Pushed.


-- 
Konstantin Osipov, Moscow, Russia, +7 903 626 22 32
http://tarantool.io - www.twitter.com/kostja_osipov

Re: [PATCH v2 3/8] vinyl: log new index before WAL write on DDL

[Konstantin Osipov]

* Vladimir Davydov <vdavydov.dev@gmail.com> [18/05/28 07:49]:

Please add a test case for vinyl_index_abort_create() during
recovery, with force_recovery set to true and error which leads to
abort.

I pushed this patch.

-- 
Konstantin Osipov, Moscow, Russia, +7 903 626 22 32
http://tarantool.io - www.twitter.com/kostja_osipov

Re: [PATCH v2 4/8] vinyl: bump mem version after committing statement

[Konstantin Osipov]

* Vladimir Davydov <vdavydov.dev@gmail.com> [18/05/28 07:49]:
> Since commit 1e1c1fdbeddb ("vinyl: make read iterator always return
> newest tuple version") vinyl read iterator guarantees that any tuple it
> returns is the newest version in the iterator read view. However, if we
> don't bump mem version after assigning LSN to a mem statement, a read
> iterator using committed_read_view might not see it and return a stale
> tuple. Currently, there's no code that relies on this iterator feature,
> but we will need it for building new indexes. Without this patch, build
> (introduced later in the series) might return inconsistent results.

Pushed.

-- 
Konstantin Osipov, Moscow, Russia, +7 903 626 22 32
http://tarantool.io - www.twitter.com/kostja_osipov

Re: [PATCH v2 5/8] vinyl: allow to commit statements to mem in arbitrary order

[Konstantin Osipov]

* Vladimir Davydov <vdavydov.dev@gmail.com> [18/05/28 07:49]:
> vy_mem_commit_stmt() expects statements to be committed in the order
> of increasing LSN. Although this condition holds now, it won't once
> we start using this function for building indexes. So let's remove
> this limitation.
> 
> Needed for #1653

Pushed.

-- 
Konstantin Osipov, Moscow, Russia, +7 903 626 22 32
http://tarantool.io - www.twitter.com/kostja_osipov