[PATCH v2 0/3] Support savepoints in DDL transactions

[PATCH v2 0/3] Support savepoints in DDL transactions

[Vladimir Davydov]

DDL statements install commit/rollback triggers to complete/revert
changes done to the schema. If a DDL statement is rolled back by a
savepoint, we must remove commit triggers and run rollback triggers,
otherwise we can get an inconsistent schema state. This patch set
addresses this issue.

The first two patches are not really related to the issue at hand -
they just do some trivial cleanup in txn code. Both bugs are fixed
by the third and the final patch of the series.

https://github.com/tarantool/tarantool/issues/4364
https://github.com/tarantool/tarantool/issues/4365
https://github.com/tarantool/tarantool/commits/dv/gh-4364-4365-fix-ddl-savepoint

Changes in v2:
 - Append DDL commit/rollback triggers to txn statements instead
   of maintaining them in txn. This greatly simplifies rollback.
 - Merge approved patches and rebase.

Vladimir Davydov (3):
  txn: move stmt list/savepoint manipulation out of txn_stmt_new
  txn: kill txn_last_stmt helper
  txn: undo commit/rollback triggers when reverting to savepoint

 src/box/alter.cc              | 90 +++++++++++++++++------------------
 src/box/txn.c                 | 71 +++++++++++++++++++--------
 src/box/txn.h                 | 41 +++++++++++++---
 test/box/transaction.result   | 39 +++++++++++++++
 test/box/transaction.test.lua | 23 +++++++++
 5 files changed, 192 insertions(+), 72 deletions(-)

-- 
2.20.1

[PATCH v2 1/3] txn: move stmt list/savepoint manipulation out of txn_stmt_new

[Vladimir Davydov]

txn_stmt_new is supposed to simply allocate an initialize a new txn_stmt
struct. Adding the new statement to the txn's statement list and setting
up a savepoint for rollback looks confusing. Move it to txn_begin_stmt.
---
 src/box/txn.c | 18 +++++++++---------
 1 file changed, 9 insertions(+), 9 deletions(-)

diff --git a/src/box/txn.c b/src/box/txn.c
index 05ec6f14..27ae43ac 100644
--- a/src/box/txn.c
+++ b/src/box/txn.c
@@ -84,10 +84,10 @@ txn_add_redo(struct txn *txn, struct txn_stmt *stmt, struct request *request)
 
 /** Initialize a new stmt object within txn. */
 static struct txn_stmt *
-txn_stmt_new(struct txn *txn)
+txn_stmt_new(struct region *region)
 {
 	struct txn_stmt *stmt;
-	stmt = region_alloc_object(&txn->region, struct txn_stmt);
+	stmt = region_alloc_object(region, struct txn_stmt);
 	if (stmt == NULL) {
 		diag_set(OutOfMemory, sizeof(*stmt),
 			 "region", "struct txn_stmt");
@@ -100,12 +100,6 @@ txn_stmt_new(struct txn *txn)
 	stmt->new_tuple = NULL;
 	stmt->engine_savepoint = NULL;
 	stmt->row = NULL;
-
-	/* Set the savepoint for statement rollback. */
-	txn->sub_stmt_begin[txn->in_sub_stmt] = stailq_last(&txn->stmts);
-	txn->in_sub_stmt++;
-
-	stailq_add_tail_entry(&txn->stmts, stmt, next);
 	return stmt;
 }
 
@@ -246,9 +240,15 @@ txn_begin_stmt(struct txn *txn, struct space *space)
 		diag_set(ClientError, ER_SUB_STMT_MAX);
 		return -1;
 	}
-	struct txn_stmt *stmt = txn_stmt_new(txn);
+	struct txn_stmt *stmt = txn_stmt_new(&txn->region);
 	if (stmt == NULL)
 		return -1;
+
+	/* Set the savepoint for statement rollback. */
+	txn->sub_stmt_begin[txn->in_sub_stmt] = stailq_last(&txn->stmts);
+	txn->in_sub_stmt++;
+	stailq_add_tail_entry(&txn->stmts, stmt, next);
+
 	if (space == NULL)
 		return 0;
 
-- 
2.20.1

[PATCH v2 2/3] txn: kill txn_last_stmt helper

[Vladimir Davydov]

It's not used anywhere anymore.
---
 src/box/txn.h | 7 -------
 1 file changed, 7 deletions(-)

diff --git a/src/box/txn.h b/src/box/txn.h
index 0a5c5169..9a6c8d63 100644
--- a/src/box/txn.h
+++ b/src/box/txn.h
@@ -416,13 +416,6 @@ txn_current_stmt(struct txn *txn)
 	return stailq_entry(stmt, struct txn_stmt, next);
 }
 
-/** The last statement of the transaction. */
-static inline struct txn_stmt *
-txn_last_stmt(struct txn *txn)
-{
-	return stailq_last_entry(&txn->stmts, struct txn_stmt, next);
-}
-
 /**
  * FFI bindings: do not throw exceptions, do not accept extra
  * arguments
-- 
2.20.1

[PATCH v2 3/3] txn: undo commit/rollback triggers when reverting to savepoint

[Vladimir Davydov]

When reverting to a savepoint inside a DDL transaction, apart from
undoing changes done by the DDL statements to the system spaces, we also
have to

 - Run rollback triggers installed after the savepoint was set, because
   otherwise changes done to the schema by DDL won't be undone.
 - Remove commit triggers installed after the savepoint, because they
   are not relevant anymore, apparently.

To achieve that let's append DDL triggers right to txn statements.
This allows us to easily discard commit triggers and run rollback
triggers when a statement is rolled back.

Note, txn commit/rollback triggers are not removed, because they are
still used by applier and Lua box.on_commit/on_rollback functions.

Closes #4364
Closes #4365
---
 src/box/alter.cc              | 90 +++++++++++++++++------------------
 src/box/txn.c                 | 53 ++++++++++++++++-----
 src/box/txn.h                 | 34 +++++++++++++
 test/box/transaction.result   | 39 +++++++++++++++
 test/box/transaction.test.lua | 23 +++++++++
 5 files changed, 183 insertions(+), 56 deletions(-)

diff --git a/src/box/alter.cc b/src/box/alter.cc
index 3caeac3b..d18a0f1f 100644
--- a/src/box/alter.cc
+++ b/src/box/alter.cc
@@ -895,7 +895,7 @@ alter_space_rollback(struct trigger *trigger, void * /* event */)
  *   On rollback, the new space is deleted.
  */
 static void
-alter_space_do(struct txn *txn, struct alter_space *alter)
+alter_space_do(struct txn_stmt *stmt, struct alter_space *alter)
 {
 	/**
 	 * AlterSpaceOp::prepare() may perform a potentially long
@@ -977,8 +977,8 @@ alter_space_do(struct txn *txn, struct alter_space *alter)
 	 * finish or rollback the DDL depending on the results of
 	 * writing to WAL.
 	 */
-	txn_on_commit(txn, on_commit);
-	txn_on_rollback(txn, on_rollback);
+	txn_stmt_on_commit(stmt, on_commit);
+	txn_stmt_on_rollback(stmt, on_rollback);
 }
 
 /* }}}  */
@@ -1882,7 +1882,7 @@ on_replace_dd_space(struct trigger * /* trigger */, void *event)
 		 */
 		struct trigger *on_rollback =
 			txn_alter_trigger_new(on_create_space_rollback, space);
-		txn_on_rollback(txn, on_rollback);
+		txn_stmt_on_rollback(stmt, on_rollback);
 		if (def->opts.is_view) {
 			struct Select *select = sql_view_compile(sql_get(),
 								 def->opts.sql);
@@ -1906,11 +1906,11 @@ on_replace_dd_space(struct trigger * /* trigger */, void *event)
 			struct trigger *on_commit_view =
 				txn_alter_trigger_new(on_create_view_commit,
 						      select);
-			txn_on_commit(txn, on_commit_view);
+			txn_stmt_on_commit(stmt, on_commit_view);
 			struct trigger *on_rollback_view =
 				txn_alter_trigger_new(on_create_view_rollback,
 						      select);
-			txn_on_rollback(txn, on_rollback_view);
+			txn_stmt_on_rollback(stmt, on_rollback_view);
 			select_guard.is_active = false;
 		}
 	} else if (new_tuple == NULL) { /* DELETE */
@@ -1967,10 +1967,10 @@ on_replace_dd_space(struct trigger * /* trigger */, void *event)
 		++schema_version;
 		struct trigger *on_commit =
 			txn_alter_trigger_new(on_drop_space_commit, old_space);
-		txn_on_commit(txn, on_commit);
+		txn_stmt_on_commit(stmt, on_commit);
 		struct trigger *on_rollback =
 			txn_alter_trigger_new(on_drop_space_rollback, old_space);
-		txn_on_rollback(txn, on_rollback);
+		txn_stmt_on_rollback(stmt, on_rollback);
 		if (old_space->def->opts.is_view) {
 			struct Select *select =
 				sql_view_compile(sql_get(),
@@ -1983,11 +1983,11 @@ on_replace_dd_space(struct trigger * /* trigger */, void *event)
 			struct trigger *on_commit_view =
 				txn_alter_trigger_new(on_drop_view_commit,
 						      select);
-			txn_on_commit(txn, on_commit_view);
+			txn_stmt_on_commit(stmt, on_commit_view);
 			struct trigger *on_rollback_view =
 				txn_alter_trigger_new(on_drop_view_rollback,
 						      select);
-			txn_on_rollback(txn, on_rollback_view);
+			txn_stmt_on_rollback(stmt, on_rollback_view);
 			update_view_references(select, -1, true, NULL);
 			select_guard.is_active = false;
 		}
@@ -2057,7 +2057,7 @@ on_replace_dd_space(struct trigger * /* trigger */, void *event)
 		alter_space_move_indexes(alter, 0, old_space->index_id_max + 1);
 		/* Remember to update schema_version. */
 		(void) new UpdateSchemaVersion(alter);
-		alter_space_do(txn, alter);
+		alter_space_do(stmt, alter);
 		alter_guard.is_active = false;
 	}
 }
@@ -2295,7 +2295,7 @@ on_replace_dd_index(struct trigger * /* trigger */, void *event)
 	(void) new MoveCkConstraints(alter);
 	/* Add an op to update schema_version on commit. */
 	(void) new UpdateSchemaVersion(alter);
-	alter_space_do(txn, alter);
+	alter_space_do(stmt, alter);
 	scoped_guard.is_active = false;
 }
 
@@ -2371,7 +2371,7 @@ on_replace_dd_truncate(struct trigger * /* trigger */, void *event)
 	}
 
 	(void) new MoveCkConstraints(alter);
-	alter_space_do(txn, alter);
+	alter_space_do(stmt, alter);
 	scoped_guard.is_active = false;
 }
 
@@ -2559,7 +2559,7 @@ on_replace_dd_user(struct trigger * /* trigger */, void *event)
 		def_guard.is_active = false;
 		struct trigger *on_rollback =
 			txn_alter_trigger_new(user_cache_remove_user, new_tuple);
-		txn_on_rollback(txn, on_rollback);
+		txn_stmt_on_rollback(stmt, on_rollback);
 	} else if (new_tuple == NULL) { /* DELETE */
 		access_check_ddl(old_user->def->name, old_user->def->uid,
 				 old_user->def->owner, old_user->def->type,
@@ -2581,7 +2581,7 @@ on_replace_dd_user(struct trigger * /* trigger */, void *event)
 		user_cache_delete(uid);
 		struct trigger *on_rollback =
 			txn_alter_trigger_new(user_cache_alter_user, old_tuple);
-		txn_on_rollback(txn, on_rollback);
+		txn_stmt_on_rollback(stmt, on_rollback);
 	} else { /* UPDATE, REPLACE */
 		assert(old_user != NULL && new_tuple != NULL);
 		/*
@@ -2597,7 +2597,7 @@ on_replace_dd_user(struct trigger * /* trigger */, void *event)
 		def_guard.is_active = false;
 		struct trigger *on_rollback =
 			txn_alter_trigger_new(user_cache_alter_user, old_tuple);
-		txn_on_rollback(txn, on_rollback);
+		txn_stmt_on_rollback(stmt, on_rollback);
 	}
 }
 
@@ -2848,7 +2848,7 @@ on_replace_dd_func(struct trigger * /* trigger */, void *event)
 		def_guard.is_active = false;
 		func_cache_insert(func);
 		on_rollback->data = func;
-		txn_on_rollback(txn, on_rollback);
+		txn_stmt_on_rollback(stmt, on_rollback);
 		trigger_run_xc(&on_alter_func, func);
 	} else if (new_tuple == NULL) {         /* DELETE */
 		uint32_t uid;
@@ -2876,8 +2876,8 @@ on_replace_dd_func(struct trigger * /* trigger */, void *event)
 		struct trigger *on_rollback =
 			txn_alter_trigger_new(on_drop_func_rollback, old_func);
 		func_cache_delete(old_func->def->fid);
-		txn_on_commit(txn, on_commit);
-		txn_on_rollback(txn, on_rollback);
+		txn_stmt_on_commit(stmt, on_commit);
+		txn_stmt_on_rollback(stmt, on_rollback);
 		trigger_run_xc(&on_alter_func, old_func);
 	} else {                                /* UPDATE, REPLACE */
 		assert(new_tuple != NULL && old_tuple != NULL);
@@ -3062,8 +3062,8 @@ on_replace_dd_collation(struct trigger * /* trigger */, void *event)
 		coll_id_cache_delete(old_coll_id);
 		on_rollback->data = old_coll_id;
 		on_commit->data = old_coll_id;
-		txn_on_rollback(txn, on_rollback);
-		txn_on_commit(txn, on_commit);
+		txn_stmt_on_rollback(stmt, on_rollback);
+		txn_stmt_on_commit(stmt, on_commit);
 	} else if (new_tuple != NULL && old_tuple == NULL) {
 		/* INSERT */
 		struct trigger *on_rollback =
@@ -3082,7 +3082,7 @@ on_replace_dd_collation(struct trigger * /* trigger */, void *event)
 		}
 		assert(replaced_id == NULL);
 		on_rollback->data = new_coll_id;
-		txn_on_rollback(txn, on_rollback);
+		txn_stmt_on_rollback(stmt, on_rollback);
 	} else {
 		/* UPDATE */
 		assert(new_tuple != NULL && old_tuple != NULL);
@@ -3335,7 +3335,7 @@ on_replace_dd_priv(struct trigger * /* trigger */, void *event)
 		grant_or_revoke(&priv);
 		struct trigger *on_rollback =
 			txn_alter_trigger_new(revoke_priv, new_tuple);
-		txn_on_rollback(txn, on_rollback);
+		txn_stmt_on_rollback(stmt, on_rollback);
 	} else if (new_tuple == NULL) {                /* revoke */
 		assert(old_tuple);
 		priv_def_create_from_tuple(&priv, old_tuple);
@@ -3344,14 +3344,14 @@ on_replace_dd_priv(struct trigger * /* trigger */, void *event)
 		grant_or_revoke(&priv);
 		struct trigger *on_rollback =
 			txn_alter_trigger_new(modify_priv, old_tuple);
-		txn_on_rollback(txn, on_rollback);
+		txn_stmt_on_rollback(stmt, on_rollback);
 	} else {                                       /* modify */
 		priv_def_create_from_tuple(&priv, new_tuple);
 		priv_def_check(&priv, PRIV_GRANT);
 		grant_or_revoke(&priv);
 		struct trigger *on_rollback =
 			txn_alter_trigger_new(modify_priv, old_tuple);
-		txn_on_rollback(txn, on_rollback);
+		txn_stmt_on_rollback(stmt, on_rollback);
 	}
 }
 
@@ -3481,7 +3481,7 @@ on_replace_dd_cluster(struct trigger *trigger, void *event)
 			struct trigger *on_commit;
 			on_commit = txn_alter_trigger_new(register_replica,
 							  new_tuple);
-			txn_on_commit(txn, on_commit);
+			txn_stmt_on_commit(stmt, on_commit);
 		}
 	} else {
 		/*
@@ -3496,7 +3496,7 @@ on_replace_dd_cluster(struct trigger *trigger, void *event)
 		struct trigger *on_commit;
 		on_commit = txn_alter_trigger_new(unregister_replica,
 						  old_tuple);
-		txn_on_commit(txn, on_commit);
+		txn_stmt_on_commit(stmt, on_commit);
 	}
 }
 
@@ -3619,7 +3619,7 @@ on_replace_dd_sequence(struct trigger * /* trigger */, void *event)
 		seq = sequence_new_xc(new_def);
 		sequence_cache_insert(seq);
 		on_rollback->data = seq;
-		txn_on_rollback(txn, on_rollback);
+		txn_stmt_on_rollback(stmt, on_rollback);
 	} else if (old_tuple != NULL && new_tuple == NULL) {	/* DELETE */
 		uint32_t id = tuple_field_u32_xc(old_tuple,
 						 BOX_SEQUENCE_DATA_FIELD_ID);
@@ -3641,8 +3641,8 @@ on_replace_dd_sequence(struct trigger * /* trigger */, void *event)
 		struct trigger *on_rollback =
 			txn_alter_trigger_new(on_drop_sequence_rollback, seq);
 		sequence_cache_delete(seq->def->id);
-		txn_on_commit(txn, on_commit);
-		txn_on_rollback(txn, on_rollback);
+		txn_stmt_on_commit(stmt, on_commit);
+		txn_stmt_on_rollback(stmt, on_rollback);
 	} else {						/* UPDATE */
 		new_def = sequence_def_new_from_tuple(new_tuple,
 						      ER_ALTER_SEQUENCE);
@@ -3655,8 +3655,8 @@ on_replace_dd_sequence(struct trigger * /* trigger */, void *event)
 		struct trigger *on_rollback =
 			txn_alter_trigger_new(on_alter_sequence_rollback, seq->def);
 		seq->def = new_def;
-		txn_on_commit(txn, on_commit);
-		txn_on_rollback(txn, on_rollback);
+		txn_stmt_on_commit(stmt, on_commit);
+		txn_stmt_on_rollback(stmt, on_rollback);
 	}
 
 	def_guard.is_active = false;
@@ -3709,7 +3709,7 @@ on_replace_dd_sequence_data(struct trigger * /* trigger */, void *event)
 		 */
 		struct trigger *on_rollback = txn_alter_trigger_new(
 				on_drop_sequence_data_rollback, old_tuple);
-		txn_on_rollback(txn, on_rollback);
+		txn_stmt_on_rollback(stmt, on_rollback);
 		sequence_reset(seq);
 	}
 }
@@ -3864,7 +3864,7 @@ on_replace_dd_space_sequence(struct trigger * /* trigger */, void *event)
 		free(space->sequence_path);
 		space->sequence_path = sequence_path;
 		sequence_path_guard.is_active = false;
-		txn_on_rollback(txn, on_rollback);
+		txn_stmt_on_rollback(stmt, on_rollback);
 	} else {					/* DELETE */
 		struct trigger *on_rollback;
 		on_rollback = txn_alter_trigger_new(set_space_sequence,
@@ -3875,7 +3875,7 @@ on_replace_dd_space_sequence(struct trigger * /* trigger */, void *event)
 		space->sequence_fieldno = 0;
 		free(space->sequence_path);
 		space->sequence_path = NULL;
-		txn_on_rollback(txn, on_rollback);
+		txn_stmt_on_rollback(stmt, on_rollback);
 	}
 	trigger_run_xc(&on_alter_space, space);
 }
@@ -4036,8 +4036,8 @@ on_replace_dd_trigger(struct trigger * /* trigger */, void *event)
 		new_trigger_guard.is_active = false;
 	}
 
-	txn_on_rollback(txn, on_rollback);
-	txn_on_commit(txn, on_commit);
+	txn_stmt_on_rollback(stmt, on_rollback);
+	txn_stmt_on_commit(stmt, on_commit);
 }
 
 /**
@@ -4441,7 +4441,7 @@ on_replace_dd_fk_constraint(struct trigger * /* trigger*/, void *event)
 			struct trigger *on_rollback =
 				txn_alter_trigger_new(on_create_fk_constraint_rollback,
 						      fk);
-			txn_on_rollback(txn, on_rollback);
+			txn_stmt_on_rollback(stmt, on_rollback);
 			fk_constraint_set_mask(fk,
 					       &parent_space->fk_constraint_mask,
 					       FIELD_LINK_PARENT);
@@ -4459,11 +4459,11 @@ on_replace_dd_fk_constraint(struct trigger * /* trigger*/, void *event)
 			struct trigger *on_rollback =
 				txn_alter_trigger_new(on_replace_fk_constraint_rollback,
 						      old_fk);
-			txn_on_rollback(txn, on_rollback);
+			txn_stmt_on_rollback(stmt, on_rollback);
 			struct trigger *on_commit =
 				txn_alter_trigger_new(on_drop_or_replace_fk_constraint_commit,
 						      old_fk);
-			txn_on_commit(txn, on_commit);
+			txn_stmt_on_commit(stmt, on_commit);
 			space_reset_fk_constraint_mask(child_space);
 			space_reset_fk_constraint_mask(parent_space);
 		}
@@ -4485,11 +4485,11 @@ on_replace_dd_fk_constraint(struct trigger * /* trigger*/, void *event)
 		struct trigger *on_commit =
 			txn_alter_trigger_new(on_drop_or_replace_fk_constraint_commit,
 					      old_fk);
-		txn_on_commit(txn, on_commit);
+		txn_stmt_on_commit(stmt, on_commit);
 		struct trigger *on_rollback =
 			txn_alter_trigger_new(on_drop_fk_constraint_rollback,
 					      old_fk);
-		txn_on_rollback(txn, on_rollback);
+		txn_stmt_on_rollback(stmt, on_rollback);
 		space_reset_fk_constraint_mask(child_space);
 		space_reset_fk_constraint_mask(parent_space);
 	}
@@ -4713,8 +4713,8 @@ on_replace_dd_ck_constraint(struct trigger * /* trigger*/, void *event)
 		on_rollback->run = on_drop_ck_constraint_rollback;
 	}
 
-	txn_on_rollback(txn, on_rollback);
-	txn_on_commit(txn, on_commit);
+	txn_stmt_on_rollback(stmt, on_rollback);
+	txn_stmt_on_commit(stmt, on_commit);
 
 	trigger_run_xc(&on_alter_space, space);
 }
@@ -4772,7 +4772,7 @@ on_replace_dd_func_index(struct trigger *trigger, void *event)
 				 space->index_id_max + 1);
 	(void) new MoveCkConstraints(alter);
 	(void) new UpdateSchemaVersion(alter);
-	alter_space_do(txn, alter);
+	alter_space_do(stmt, alter);
 
 	scoped_guard.is_active = false;
 }
diff --git a/src/box/txn.c b/src/box/txn.c
index 27ae43ac..9599284b 100644
--- a/src/box/txn.c
+++ b/src/box/txn.c
@@ -46,6 +46,9 @@ txn_on_stop(struct trigger *trigger, void *event);
 static void
 txn_on_yield(struct trigger *trigger, void *event);
 
+static void
+txn_run_rollback_triggers(struct txn *txn, struct rlist *triggers);
+
 static inline void
 fiber_set_txn(struct fiber *fiber, struct txn *txn)
 {
@@ -100,6 +103,7 @@ txn_stmt_new(struct region *region)
 	stmt->new_tuple = NULL;
 	stmt->engine_savepoint = NULL;
 	stmt->row = NULL;
+	stmt->has_triggers = false;
 	return stmt;
 }
 
@@ -115,11 +119,25 @@ txn_stmt_unref_tuples(struct txn_stmt *stmt)
 static void
 txn_rollback_to_svp(struct txn *txn, struct stailq_entry *svp)
 {
+	/*
+	 * Undo changes done to the database by the rolled back
+	 * statements and collect corresponding rollback triggers.
+	 * Don't release the tuples as rollback triggers might
+	 * still need them.
+	 */
 	struct txn_stmt *stmt;
 	struct stailq rollback;
+	RLIST_HEAD(on_rollback);
 	stailq_cut_tail(&txn->stmts, svp, &rollback);
 	stailq_reverse(&rollback);
 	stailq_foreach_entry(stmt, &rollback, next) {
+		/*
+		 * Note the statement list is reversed hence
+		 * we must append triggers to the tail so as
+		 * to preserve the rollback order.
+		 */
+		if (stmt->has_triggers)
+			rlist_splice_tail(&on_rollback, &stmt->on_rollback);
 		if (txn->engine != NULL && stmt->space != NULL)
 			engine_rollback_statement(txn->engine, txn, stmt);
 		if (stmt->row != NULL && stmt->row->replica_id == 0) {
@@ -132,10 +150,15 @@ txn_rollback_to_svp(struct txn *txn, struct stailq_entry *svp)
 			assert(txn->n_applier_rows > 0);
 			txn->n_applier_rows--;
 		}
-		txn_stmt_unref_tuples(stmt);
 		stmt->space = NULL;
 		stmt->row = NULL;
 	}
+	/* Run rollback triggers installed after the savepoint. */
+	txn_run_rollback_triggers(txn, &on_rollback);
+
+	/* Release the tuples. */
+	stailq_foreach_entry(stmt, &rollback, next)
+		txn_stmt_unref_tuples(stmt);
 }
 
 /*
@@ -350,14 +373,14 @@ fail:
  * A helper function to process on_commit triggers.
  */
 static void
-txn_run_commit_triggers(struct txn *txn)
+txn_run_commit_triggers(struct txn *txn, struct rlist *triggers)
 {
 	/*
 	 * Commit triggers must be run in the same order they
 	 * were added so that a trigger sees the changes done
 	 * by previous triggers (this is vital for DDL).
 	 */
-	if (trigger_run_reverse(&txn->on_commit, txn) != 0) {
+	if (trigger_run_reverse(triggers, txn) != 0) {
 		/*
 		 * As transaction couldn't handle a trigger error so
 		 * there is no option except panic.
@@ -372,9 +395,9 @@ txn_run_commit_triggers(struct txn *txn)
  * A helper function to process on_rollback triggers.
  */
 static void
-txn_run_rollback_triggers(struct txn *txn)
+txn_run_rollback_triggers(struct txn *txn, struct rlist *triggers)
 {
-	if (trigger_run(&txn->on_rollback, txn) != 0) {
+	if (trigger_run(triggers, txn) != 0) {
 		/*
 		 * As transaction couldn't handle a trigger error so
 		 * there is no option except panic.
@@ -400,13 +423,13 @@ txn_complete(struct txn *txn)
 		if (txn->engine)
 			engine_rollback(txn->engine, txn);
 		if (txn_has_flag(txn, TXN_HAS_TRIGGERS))
-			txn_run_rollback_triggers(txn);
+			txn_run_rollback_triggers(txn, &txn->on_rollback);
 	} else {
 		/* Commit the transaction. */
 		if (txn->engine != NULL)
 			engine_commit(txn->engine, txn);
 		if (txn_has_flag(txn, TXN_HAS_TRIGGERS))
-			txn_run_commit_triggers(txn);
+			txn_run_commit_triggers(txn, &txn->on_commit);
 
 		double stop_tm = ev_monotonic_now(loop());
 		if (stop_tm - txn->start_tm > too_long_threshold) {
@@ -468,6 +491,11 @@ txn_write_to_wal(struct txn *txn)
 	struct xrow_header **remote_row = req->rows;
 	struct xrow_header **local_row = req->rows + txn->n_applier_rows;
 	stailq_foreach_entry(stmt, &txn->stmts, next) {
+		if (stmt->has_triggers) {
+			txn_init_triggers(txn);
+			rlist_splice(&txn->on_commit, &stmt->on_commit);
+			rlist_splice(&txn->on_rollback, &stmt->on_rollback);
+		}
 		if (stmt->row == NULL)
 			continue; /* A read (e.g. select) request */
 		if (stmt->row->replica_id == 0)
@@ -591,6 +619,13 @@ txn_rollback(struct txn *txn)
 	trigger_clear(&txn->fiber_on_stop);
 	if (!txn_has_flag(txn, TXN_CAN_YIELD))
 		trigger_clear(&txn->fiber_on_yield);
+	struct txn_stmt *stmt;
+	stailq_foreach_entry(stmt, &txn->stmts, next) {
+		if (stmt->has_triggers) {
+			txn_init_triggers(txn);
+			rlist_splice(&txn->on_rollback, &stmt->on_rollback);
+		}
+	}
 	txn->signature = -1;
 	txn_complete(txn);
 	fiber_set_txn(fiber(), NULL);
@@ -789,9 +824,5 @@ txn_on_yield(struct trigger *trigger, void *event)
 	assert(txn != NULL);
 	assert(!txn_has_flag(txn, TXN_CAN_YIELD));
 	txn_rollback_to_svp(txn, NULL);
-	if (txn_has_flag(txn, TXN_HAS_TRIGGERS)) {
-		txn_run_rollback_triggers(txn);
-		txn_clear_flag(txn, TXN_HAS_TRIGGERS);
-	}
 	txn_set_flag(txn, TXN_IS_ABORTED_BY_YIELD);
 }
diff --git a/src/box/txn.h b/src/box/txn.h
index 9a6c8d63..0c6996f8 100644
--- a/src/box/txn.h
+++ b/src/box/txn.h
@@ -93,6 +93,11 @@ struct txn_stmt {
 	void *engine_savepoint;
 	/** Redo info: the binary log row */
 	struct xrow_header *row;
+	/** on_commit and/or on_rollback list is not empty. */
+	bool has_triggers;
+	/** Commit/rollback triggers associated with this statement. */
+	struct rlist on_commit;
+	struct rlist on_rollback;
 };
 
 /**
@@ -304,6 +309,35 @@ txn_on_rollback(struct txn *txn, struct trigger *trigger)
 	trigger_add(&txn->on_rollback, trigger);
 }
 
+/**
+ * Most statements don't have triggers, and txn objects
+ * are created on every access to data, so statements
+ * are partially initialized.
+ */
+static inline void
+txn_stmt_init_triggers(struct txn_stmt *stmt)
+{
+	if (!stmt->has_triggers) {
+		rlist_create(&stmt->on_commit);
+		rlist_create(&stmt->on_rollback);
+		stmt->has_triggers = true;
+	}
+}
+
+static inline void
+txn_stmt_on_commit(struct txn_stmt *stmt, struct trigger *trigger)
+{
+	txn_stmt_init_triggers(stmt);
+	trigger_add(&stmt->on_commit, trigger);
+}
+
+static inline void
+txn_stmt_on_rollback(struct txn_stmt *stmt, struct trigger *trigger)
+{
+	txn_stmt_init_triggers(stmt);
+	trigger_add(&stmt->on_rollback, trigger);
+}
+
 /**
  * Start a new statement.
  */
diff --git a/test/box/transaction.result b/test/box/transaction.result
index 9a48f2f7..1ed649d2 100644
--- a/test/box/transaction.result
+++ b/test/box/transaction.result
@@ -778,3 +778,42 @@ box.begin() drop() box.rollback()
 box.begin() drop() box.commit()
 ---
 ...
+--
+-- gh-4364: DDL doesn't care about savepoints.
+--
+test_run:cmd("setopt delimiter ';'")
+---
+- true
+...
+box.begin()
+s1 = box.schema.create_space('test1')
+save = box.savepoint()
+s2 = box.schema.create_space('test2')
+check1 = box.space.test1 ~= nil
+check2 = box.space.test2 ~= nil
+box.rollback_to_savepoint(save)
+check3 = box.space.test1 ~= nil
+check4 = box.space.test2 ~= nil
+box.commit();
+---
+...
+test_run:cmd("setopt delimiter ''");
+---
+- true
+...
+check1, check2, check3, check4
+---
+- true
+- true
+- true
+- false
+...
+--
+-- gh-4365: DDL reverted by yield triggers crash.
+--
+box.begin() box.execute([[CREATE TABLE test(id INTEGER PRIMARY KEY AUTOINCREMENT)]]) fiber.yield()
+---
+...
+box.rollback()
+---
+...
diff --git a/test/box/transaction.test.lua b/test/box/transaction.test.lua
index 0792cc3c..a0014c9f 100644
--- a/test/box/transaction.test.lua
+++ b/test/box/transaction.test.lua
@@ -415,3 +415,26 @@ box.begin() create() box.rollback()
 box.begin() create() box.commit()
 box.begin() drop() box.rollback()
 box.begin() drop() box.commit()
+
+--
+-- gh-4364: DDL doesn't care about savepoints.
+--
+test_run:cmd("setopt delimiter ';'")
+box.begin()
+s1 = box.schema.create_space('test1')
+save = box.savepoint()
+s2 = box.schema.create_space('test2')
+check1 = box.space.test1 ~= nil
+check2 = box.space.test2 ~= nil
+box.rollback_to_savepoint(save)
+check3 = box.space.test1 ~= nil
+check4 = box.space.test2 ~= nil
+box.commit();
+test_run:cmd("setopt delimiter ''");
+check1, check2, check3, check4
+
+--
+-- gh-4365: DDL reverted by yield triggers crash.
+--
+box.begin() box.execute([[CREATE TABLE test(id INTEGER PRIMARY KEY AUTOINCREMENT)]]) fiber.yield()
+box.rollback()
-- 
2.20.1

[tarantool-patches] Re: [PATCH v2 1/3] txn: move stmt list/savepoint manipulation out of txn_stmt_new

[Konstantin Osipov]

* Vladimir Davydov <vdavydov.dev@gmail.com> [19/07/30 15:04]:
> txn_stmt_new is supposed to simply allocate an initialize a new txn_stmt
> struct. Adding the new statement to the txn's statement list and setting
> up a savepoint for rollback looks confusing. Move it to txn_begin_stmt.
> ---

lgtm


-- 
Konstantin Osipov, Moscow, Russia

[tarantool-patches] Re: [PATCH v2 2/3] txn: kill txn_last_stmt helper

[Konstantin Osipov]

* Vladimir Davydov <vdavydov.dev@gmail.com> [19/07/30 15:04]:
> It's not used anywhere anymore.

lgtm


-- 
Konstantin Osipov, Moscow, Russia

[tarantool-patches] Re: [PATCH v2 3/3] txn: undo commit/rollback triggers when reverting to savepoint

[Konstantin Osipov]

* Vladimir Davydov <vdavydov.dev@gmail.com> [19/07/30 15:04]:
> When reverting to a savepoint inside a DDL transaction, apart from
> undoing changes done by the DDL statements to the system spaces, we also
> have to
> 
>  - Run rollback triggers installed after the savepoint was set, because
>    otherwise changes done to the schema by DDL won't be undone.
>  - Remove commit triggers installed after the savepoint, because they
>    are not relevant anymore, apparently.
> 
> To achieve that let's append DDL triggers right to txn statements.
> This allows us to easily discard commit triggers and run rollback
> triggers when a statement is rolled back.
> 
> Note, txn commit/rollback triggers are not removed, because they are
> still used by applier and Lua box.on_commit/on_rollback functions.
> 
> Closes #4364
> Closes #4365

lgtm


-- 
Konstantin Osipov, Moscow, Russia

Re: [PATCH v2 0/3] Support savepoints in DDL transactions

[Vladimir Davydov]

On Tue, Jul 30, 2019 at 01:49:12PM +0300, Vladimir Davydov wrote:
> Vladimir Davydov (3):
>   txn: move stmt list/savepoint manipulation out of txn_stmt_new
>   txn: kill txn_last_stmt helper
>   txn: undo commit/rollback triggers when reverting to savepoint

Pushed to master.