[tarantool-patches] [PATCH 1/2] alter: do not access space in drop ck commit trigger

[Vladislav Shpilevoy <v.shpilevoy@tarantool.org>]

When transactional DDL is introduced, 'DROP TABLE' will remove a
space and its cks in one transaction. At the moment of that
transaction commit the space is already removed from _space and
from space cache, and can't be accessed from other triggers.

This patch makes all space-related actions in on_replace.

Part of #4086
---
 src/box/alter.cc | 49 +++++++-----------------------------------------
 src/box/space.c  | 32 +++++++++++++++++++++++++++++++
 src/box/space.h  | 17 +++++++++++++++++
 3 files changed, 56 insertions(+), 42 deletions(-)

diff --git a/src/box/alter.cc b/src/box/alter.cc
index e8721fd16..b3244094e 100644
--- a/src/box/alter.cc
+++ b/src/box/alter.cc
@@ -4526,17 +4526,10 @@ on_create_ck_constraint_rollback(struct trigger *trigger, void * /* event */)
 	assert(ck != NULL);
 	struct space *space = space_by_id(ck->def->space_id);
 	assert(space != NULL);
-	struct trigger *ck_trigger = space->ck_constraint_trigger;
-	assert(ck_trigger != NULL);
 	assert(space_ck_constraint_by_name(space, ck->def->name,
 					   strlen(ck->def->name)) != NULL);
-	rlist_del_entry(ck, link);
+	space_remove_ck_constraint(space, ck);
 	ck_constraint_delete(ck);
-	if (rlist_empty(&space->ck_constraint)) {
-		trigger_clear(ck_trigger);
-		ck_trigger->destroy(ck_trigger);
-		space->ck_constraint_trigger = NULL;
-	}
 	trigger_run_xc(&on_alter_space, space);
 }
 
@@ -4546,15 +4539,7 @@ on_drop_ck_constraint_commit(struct trigger *trigger, void * /* event */)
 {
 	struct ck_constraint *ck = (struct ck_constraint *)trigger->data;
 	assert(ck != NULL);
-	struct space *space = space_by_id(ck->def->space_id);
-	assert(space != NULL);
-	struct trigger *ck_trigger = space->ck_constraint_trigger;
-	assert(ck_trigger != NULL);
-	if (rlist_empty(&space->ck_constraint)) {
-		ck_trigger->destroy(ck_trigger);
-		space->ck_constraint_trigger = NULL;
-		ck_constraint_delete(ck);
-	}
+	ck_constraint_delete(ck);
 }
 
 /** Rollback DELETE check constraint. */
@@ -4565,13 +4550,10 @@ on_drop_ck_constraint_rollback(struct trigger *trigger, void * /* event */)
 	assert(ck != NULL);
 	struct space *space = space_by_id(ck->def->space_id);
 	assert(space != NULL);
-	struct trigger *ck_trigger = space->ck_constraint_trigger;
-	assert(ck_trigger != NULL);
 	assert(space_ck_constraint_by_name(space, ck->def->name,
 					   strlen(ck->def->name)) == NULL);
-	rlist_add_entry(&space->ck_constraint, ck, link);
-	if (rlist_empty(&ck_trigger->link))
-		trigger_add(&space->on_replace, ck_trigger);
+	if (space_add_ck_constraint(space, ck) != 0)
+		panic("Can't recover after CK constraint drop rollback");
 	trigger_run_xc(&on_alter_space, space);
 }
 
@@ -4613,8 +4595,6 @@ on_replace_dd_ck_constraint(struct trigger * /* trigger*/, void *event)
 		tuple_field_u32_xc(old_tuple != NULL ? old_tuple : new_tuple,
 				   BOX_CK_CONSTRAINT_FIELD_SPACE_ID);
 	struct space *space = space_cache_find_xc(space_id);
-	struct trigger *ck_trigger = space->ck_constraint_trigger;
-	assert(ck_trigger == NULL || !rlist_empty(&ck_trigger->link));
 	struct trigger *on_rollback = txn_alter_trigger_new(NULL, NULL);
 	struct trigger *on_commit = txn_alter_trigger_new(NULL, NULL);
 
@@ -4655,20 +4635,8 @@ on_replace_dd_ck_constraint(struct trigger * /* trigger*/, void *event)
 			space_ck_constraint_by_name(space, name, strlen(name));
 		if (old_ck_constraint != NULL)
 			rlist_del_entry(old_ck_constraint, link);
-		rlist_add_entry(&space->ck_constraint, new_ck_constraint, link);
-		if (ck_trigger == NULL) {
-			ck_trigger =
-				(struct trigger *)malloc(sizeof(*ck_trigger));
-			if (ck_trigger == NULL) {
-				tnt_raise(OutOfMemory, sizeof(*ck_trigger),
-					  "malloc", "ck_trigger");
-			}
-			trigger_create(ck_trigger,
-				       ck_constraint_on_replace_trigger, NULL,
-				       (trigger_f0) free);
-			trigger_add(&space->on_replace, ck_trigger);
-			space->ck_constraint_trigger = ck_trigger;
-		}
+		if (space_add_ck_constraint(space, new_ck_constraint) != 0)
+			diag_raise();
 		ck_guard.is_active = false;
 		if (old_tuple != NULL) {
 			on_rollback->data = old_ck_constraint;
@@ -4680,7 +4648,6 @@ on_replace_dd_ck_constraint(struct trigger * /* trigger*/, void *event)
 		on_commit->data = old_ck_constraint;
 		on_commit->run = on_replace_ck_constraint_commit;
 	} else {
-		assert(ck_trigger != NULL);
 		assert(new_tuple == NULL && old_tuple != NULL);
 		/* Drop check constraint. */
 		uint32_t name_len;
@@ -4691,9 +4658,7 @@ on_replace_dd_ck_constraint(struct trigger * /* trigger*/, void *event)
 		struct ck_constraint *old_ck_constraint =
 			space_ck_constraint_by_name(space, name, name_len);
 		assert(old_ck_constraint != NULL);
-		rlist_del_entry(old_ck_constraint, link);
-		if (rlist_empty(&space->ck_constraint))
-			trigger_clear(ck_trigger);
+		space_remove_ck_constraint(space, old_ck_constraint);
 		on_commit->data = old_ck_constraint;
 		on_commit->run = on_drop_ck_constraint_commit;
 		on_rollback->data = old_ck_constraint;
diff --git a/src/box/space.c b/src/box/space.c
index e7babb2ae..d81948ff3 100644
--- a/src/box/space.c
+++ b/src/box/space.c
@@ -545,6 +545,38 @@ space_execute_dml(struct space *space, struct txn *txn,
 	return 0;
 }
 
+int
+space_add_ck_constraint(struct space *space, struct ck_constraint *ck)
+{
+	rlist_add_entry(&space->ck_constraint, ck, link);
+	if (space->ck_constraint_trigger == NULL) {
+		struct trigger *ck_trigger =
+			(struct trigger *) malloc(sizeof(*ck_trigger));
+		if (ck_trigger == NULL) {
+			diag_set(OutOfMemory, sizeof(*ck_trigger), "malloc",
+				 "ck_trigger");
+			return -1;
+		}
+		trigger_create(ck_trigger, ck_constraint_on_replace_trigger,
+			       NULL, (trigger_f0) free);
+		trigger_add(&space->on_replace, ck_trigger);
+		space->ck_constraint_trigger = ck_trigger;
+	}
+	return 0;
+}
+
+void
+space_remove_ck_constraint(struct space *space, struct ck_constraint *ck)
+{
+	rlist_del_entry(ck, link);
+	if (rlist_empty(&space->ck_constraint)) {
+		struct trigger *ck_trigger = space->ck_constraint_trigger;
+		trigger_clear(ck_trigger);
+		ck_trigger->destroy(ck_trigger);
+		space->ck_constraint_trigger = NULL;
+	}
+}
+
 /* {{{ Virtual method stubs */
 
 size_t
diff --git a/src/box/space.h b/src/box/space.h
index 88db4ec52..708584968 100644
--- a/src/box/space.h
+++ b/src/box/space.h
@@ -50,6 +50,7 @@ struct request;
 struct port;
 struct tuple;
 struct tuple_format;
+struct ck_constraint;
 
 struct space_vtab {
 	/** Free a space instance. */
@@ -475,6 +476,22 @@ space_dump_def(const struct space *space, struct rlist *key_list);
 void
 space_fill_index_map(struct space *space);
 
+/**
+ * Add a new ck constraint to the space. A ck constraint check
+ * trigger is created, if this is a first ck in this space. The
+ * space takes ownership of this object.
+ */
+int
+space_add_ck_constraint(struct space *space, struct ck_constraint *ck);
+
+/**
+ * Remove a ck constraint from the space. A ck constraint check
+ * trigger is deleted, if this is a last ck in this space. This
+ * object may be deleted manually after the call.
+ */
+void
+space_remove_ck_constraint(struct space *space, struct ck_constraint *ck);
+
 /*
  * Virtual method stubs.
  */
-- 
2.20.1 (Apple Git-117)