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

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

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