[PATCH v3 1/3] vinyl: make tx_manager_abort_writers_for_ddl more thorough

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

We need to abort all transactions writing to an altered space when
a new index is built. Currently, we use the write set to look up such
transactions, but it isn't quite correct, because a transaction could
yield on disk read before inserting a statement into the write set.
To address this problem, this patch adds vy_tx->last_stmt_space, which
points to the space affected by the last prepared transaction. Now,
tx_manager_abort_writers_for_ddl will look not only at the write set,
but also at this variable to check if it needs to abort a transaction.

Needed for #3420
---
 src/box/vinyl.c | 12 ++++++------
 src/box/vy_tx.c | 14 +++++++++++---
 src/box/vy_tx.h | 15 ++++++++++++---
 3 files changed, 29 insertions(+), 12 deletions(-)

diff --git a/src/box/vinyl.c b/src/box/vinyl.c
index 3ef43e18..a6a5f187 100644
--- a/src/box/vinyl.c
+++ b/src/box/vinyl.c
@@ -1034,14 +1034,14 @@ vinyl_space_prepare_alter(struct space *old_space, struct space *new_space)
 /**
  * This function is called after installing on_replace trigger
  * used for propagating changes done during DDL. It aborts all
- * rw transactions affecting the given LSM tree that began
+ * rw transactions affecting the given space that began
  * before the trigger was installed so that DDL doesn't miss
  * their working set.
  */
 static void
-vy_abort_writers_for_ddl(struct vy_env *env, struct vy_lsm *lsm)
+vy_abort_writers_for_ddl(struct vy_env *env, struct space *space)
 {
-	tx_manager_abort_writers_for_ddl(env->xm, lsm);
+	tx_manager_abort_writers_for_ddl(env->xm, space);
 	/*
 	 * Wait for prepared transactions to complete
 	 * (we can't abort them as they reached WAL).
@@ -1115,7 +1115,7 @@ vinyl_space_check_format(struct space *space, struct tuple_format *format)
 	trigger_create(&on_replace, vy_check_format_on_replace, &ctx, NULL);
 	trigger_add(&space->on_replace, &on_replace);
 
-	vy_abort_writers_for_ddl(env, pk);
+	vy_abort_writers_for_ddl(env, space);
 
 	struct vy_read_iterator itr;
 	vy_read_iterator_open(&itr, pk, NULL, ITER_ALL, pk->env->empty_key,
@@ -2434,7 +2434,7 @@ vinyl_engine_begin_statement(struct engine *engine, struct txn *txn)
 	struct vy_tx *tx = txn->engine_tx;
 	struct txn_stmt *stmt = txn_current_stmt(txn);
 	assert(tx != NULL);
-	return vy_tx_begin_statement(tx, &stmt->engine_savepoint);
+	return vy_tx_begin_statement(tx, stmt->space, &stmt->engine_savepoint);
 }
 
 static void
@@ -4229,7 +4229,7 @@ vinyl_space_build_index(struct space *src_space, struct index *new_index,
 	trigger_create(&on_replace, vy_build_on_replace, &ctx, NULL);
 	trigger_add(&src_space->on_replace, &on_replace);
 
-	vy_abort_writers_for_ddl(env, pk);
+	vy_abort_writers_for_ddl(env, src_space);
 
 	struct vy_read_iterator itr;
 	vy_read_iterator_open(&itr, pk, NULL, ITER_ALL, pk->env->empty_key,
diff --git a/src/box/vy_tx.c b/src/box/vy_tx.c
index 1b8224f4..9c416c0b 100644
--- a/src/box/vy_tx.c
+++ b/src/box/vy_tx.c
@@ -322,6 +322,7 @@ vy_tx_read_set_free_cb(vy_tx_read_set_t *read_set,
 void
 vy_tx_create(struct tx_manager *xm, struct vy_tx *tx)
 {
+	tx->last_stmt_space = NULL;
 	stailq_create(&tx->log);
 	write_set_new(&tx->write_set);
 	tx->write_set_version = 0;
@@ -869,13 +870,14 @@ vy_tx_rollback(struct vy_tx *tx)
 }
 
 int
-vy_tx_begin_statement(struct vy_tx *tx, void **savepoint)
+vy_tx_begin_statement(struct vy_tx *tx, struct space *space, void **savepoint)
 {
 	if (tx->state == VINYL_TX_ABORT) {
 		diag_set(ClientError, ER_TRANSACTION_CONFLICT);
 		return -1;
 	}
 	assert(tx->state == VINYL_TX_READY);
+	tx->last_stmt_space = space;
 	if (stailq_empty(&tx->log))
 		rlist_add_entry(&tx->xm->writers, tx, in_writers);
 	*savepoint = stailq_last(&tx->log);
@@ -907,6 +909,7 @@ vy_tx_rollback_statement(struct vy_tx *tx, void *svp)
 	}
 	if (stailq_empty(&tx->log))
 		rlist_del_entry(tx, in_writers);
+	tx->last_stmt_space = NULL;
 }
 
 int
@@ -1108,11 +1111,16 @@ vy_tx_set_with_colmask(struct vy_tx *tx, struct vy_lsm *lsm,
 }
 
 void
-tx_manager_abort_writers_for_ddl(struct tx_manager *xm, struct vy_lsm *lsm)
+tx_manager_abort_writers_for_ddl(struct tx_manager *xm, struct space *space)
 {
+	if (space->index_count == 0)
+		return; /* no indexes, no conflicts */
+	struct vy_lsm *lsm = vy_lsm(space->index[0]);
 	struct vy_tx *tx;
 	rlist_foreach_entry(tx, &xm->writers, in_writers) {
-		if (tx->state == VINYL_TX_READY &&
+		if (tx->state != VINYL_TX_READY)
+			continue;
+		if (tx->last_stmt_space == space ||
 		    write_set_search_key(&tx->write_set, lsm,
 					 lsm->env->empty_key) != NULL)
 			vy_tx_abort(tx);
diff --git a/src/box/vy_tx.h b/src/box/vy_tx.h
index aaa31bee..93e3a8cd 100644
--- a/src/box/vy_tx.h
+++ b/src/box/vy_tx.h
@@ -51,6 +51,7 @@
 extern "C" {
 #endif /* defined(__cplusplus) */
 
+struct space;
 struct tuple;
 struct tx_manager;
 struct vy_mem;
@@ -140,6 +141,14 @@ struct vy_tx {
 	/** Transaction manager. */
 	struct tx_manager *xm;
 	/**
+	 * Pointer to the space affected by the last prepared statement.
+	 * We need it so that we can abort a transaction on DDL even
+	 * if it hasn't inserted anything into the write set yet (e.g.
+	 * yielded on unique check) and therefore would otherwise be
+	 * ignored by tx_manager_abort_writers_for_ddl().
+	 */
+	struct space *last_stmt_space;
+	/**
 	 * In memory transaction log. Contains both reads
 	 * and writes.
 	 */
@@ -277,12 +286,12 @@ size_t
 tx_manager_mem_used(struct tx_manager *xm);
 
 /**
- * Abort all rw transactions that affect the given LSM tree
+ * Abort all rw transactions that affect the given space
  * and haven't reached WAL yet. Called before executing a DDL
  * operation.
  */
 void
-tx_manager_abort_writers_for_ddl(struct tx_manager *xm, struct vy_lsm *lsm);
+tx_manager_abort_writers_for_ddl(struct tx_manager *xm, struct space *space);
 
 /**
  * Abort all local rw transactions that haven't reached WAL yet.
@@ -327,7 +336,7 @@ vy_tx_rollback(struct vy_tx *tx);
  * to a save point with vy_tx_rollback_statement().
  */
 int
-vy_tx_begin_statement(struct vy_tx *tx, void **savepoint);
+vy_tx_begin_statement(struct vy_tx *tx, struct space *space, void **savepoint);
 
 /**
  * Rollback a transaction statement.
-- 
2.11.0