[Tarantool-patches] [PATCH 2/2] sql: make constraint operations transactional

[Roman Khabibov <roman.habibov@tarantool.org>]

Put constraint names into the space's hash table and drop them on
commit, replace and drop (in the alter.cc triggers).

Closes #3503
---
 src/box/alter.cc                 | 254 +++++++++++++++++++++++++++++--
 test/sql-tap/constraint.test.lua | 244 +++++++++++++++++++++++++++++
 2 files changed, 488 insertions(+), 10 deletions(-)
 create mode 100755 test/sql-tap/constraint.test.lua

diff --git a/src/box/alter.cc b/src/box/alter.cc
index 272d7cd32..8255f2f83 100644
--- a/src/box/alter.cc
+++ b/src/box/alter.cc
@@ -1690,6 +1690,38 @@ MoveCkConstraints::rollback(struct alter_space *alter)
 	space_swap_ck_constraint(alter->new_space, alter->old_space);
 }
 
+/**
+ * Move constraint names hash table from old space to the new one.
+ */
+class MoveConstraints: public AlterSpaceOp
+{
+	void space_swap_constraint(struct space *old_space,
+				   struct space *new_space);
+public:
+	MoveConstraints(struct alter_space *alter) : AlterSpaceOp(alter) {}
+	virtual void alter(struct alter_space *alter);
+	virtual void rollback(struct alter_space *alter);
+};
+
+void
+MoveConstraints::space_swap_constraint(struct space *old_space,
+				       struct space *new_space)
+{
+	SWAP(new_space->constraint_names, old_space->constraint_names);
+}
+
+void
+MoveConstraints::alter(struct alter_space *alter)
+{
+	space_swap_constraint(alter->old_space, alter->new_space);
+}
+
+void
+MoveConstraints::rollback(struct alter_space *alter)
+{
+	space_swap_constraint(alter->new_space, alter->old_space);
+}
+
 /* }}} */
 
 /**
@@ -2224,6 +2256,7 @@ on_replace_dd_space(struct trigger * /* trigger */, void *event)
 		def_guard.is_active = false;
 		/* Create MoveIndex ops for all space indexes. */
 		alter_space_move_indexes(alter, 0, old_space->index_id_max + 1);
+		(void) new MoveConstraints(alter);
 		/* Remember to update schema_version. */
 		(void) new UpdateSchemaVersion(alter);
 		try {
@@ -2258,6 +2291,73 @@ index_is_used_by_fk_constraint(struct rlist *fk_list, uint32_t iid)
 	return false;
 }
 
+/**
+ * Bring back the name of a unique index to the hash table of a
+ * space.
+ *
+ * Rollback DROP index.
+ *
+ * @param trigger Trigger.
+ * @param event Event.
+ */
+static int
+on_drop_index_rollback(struct trigger *trigger, void * /* event */)
+{
+	struct index_def *def = (struct index_def *)trigger->data;
+	struct space *space = space_by_id(def->space_id);
+	if (space_put_constraint_name(space, def->name) != 0) {
+		panic("Can't recover after index drop rollback (out of "
+		      "memory)");
+	}
+	index_def_delete(def);
+	return 0;
+}
+
+/**
+ * Drop the name of a unique index from the hash table of a space.
+ *
+ * Rollback CREATE index.
+ *
+ * @param trigger Trigger.
+ * @param event Event.
+ */
+static int
+on_create_index_rollback(struct trigger *trigger, void * /* event */)
+{
+	struct index_def *def = (struct index_def *)trigger->data;
+	struct space *space = space_by_id(def->space_id);
+	space_drop_constraint_name(space, def->name);
+	index_def_delete(def);
+	return 0;
+}
+
+/**
+ * Bring back new name of a unique index to the hash table of a
+ * space and drop old name respectively.
+ *
+ * Rollback REPLACE index.
+ *
+ * @param trigger Trigger.
+ * @param event Event.
+ */
+static int
+on_replace_index_rollback(struct trigger *trigger, void * /* event */)
+{
+	struct index_def **defs = (struct index_def **)trigger->data;
+	struct index_def *old_def = defs[0];
+	struct index_def *new_def = defs[1];
+
+	struct space *space = space_by_id(old_def->space_id);
+	space_drop_constraint_name(space, new_def->name);
+	if (space_put_constraint_name(space, old_def->name) != 0)
+		panic("Can't recover after index replace rollback (out of "
+		      "memory)");
+	index_def_delete(old_def);
+	index_def_delete(new_def);
+	free(defs);
+	return 0;
+}
+
 /**
  * Just like with _space, 3 major cases:
  *
@@ -2393,17 +2493,46 @@ on_replace_dd_index(struct trigger * /* trigger */, void *event)
 		}
 		alter_space_move_indexes(alter, 0, iid);
 		(void) new DropIndex(alter, old_index);
+		if (old_index->def->opts.is_unique &&
+		    !space_is_system(old_space)) {
+			space_drop_constraint_name(old_space,
+						   old_index->def->name);
+			struct trigger *on_rollback =
+				txn_alter_trigger_new(NULL, NULL);
+			if (on_rollback == NULL)
+				return -1;
+			on_rollback->data = index_def_dup(old_index->def);
+			on_rollback->run = on_drop_index_rollback;
+			txn_stmt_on_rollback(stmt, on_rollback);
+		}
 	}
 	/* Case 2: create an index, if it is simply created. */
 	if (old_index == NULL && new_tuple != NULL) {
+		struct index_def *def = NULL;
+		def = index_def_new_from_tuple(new_tuple, old_space);
+		if (def == NULL)
+			return -1;
+		if (def->opts.is_unique && !space_is_system(old_space)) {
+			if (space_has_constraint(old_space, def->name)) {
+				diag_set(ClientError, ER_CONSTRAINT_EXISTS,
+					 def->name);
+				return -1;
+			}
+			if (space_put_constraint_name(old_space,
+						      def->name) != 0)
+				return -1;
+			struct trigger *on_rollback =
+				txn_alter_trigger_new(NULL, NULL);
+			if (on_rollback == NULL)
+				return -1;
+			on_rollback->data = index_def_dup(def);
+			on_rollback->run = on_create_index_rollback;
+			txn_stmt_on_rollback(stmt, on_rollback);
+		}
 		alter_space_move_indexes(alter, 0, iid);
 		CreateIndex *create_index = new CreateIndex(alter);
-		create_index->new_index_def =
-			index_def_new_from_tuple(new_tuple, old_space);
-		if (create_index->new_index_def == NULL)
-			return -1;
-		index_def_update_optionality(create_index->new_index_def,
-					     alter->new_min_field_count);
+		create_index->new_index_def = def;
+		index_def_update_optionality(def, alter->new_min_field_count);
 	}
 	/* Case 3 and 4: check if we need to rebuild index data. */
 	if (old_index != NULL && new_tuple != NULL) {
@@ -2411,6 +2540,82 @@ on_replace_dd_index(struct trigger * /* trigger */, void *event)
 		index_def = index_def_new_from_tuple(new_tuple, old_space);
 		if (index_def == NULL)
 			return -1;
+		struct index_def *old_def = old_index->def;
+		if (!space_is_system(old_space)) {
+			/**
+			 * We put a new name either the an index
+			 * is  becoming unique (i.e. constraint),
+			 * or when an unique index's name is under
+			 * change.
+			 */
+			if (!old_def->opts.is_unique &&
+			    index_def->opts.is_unique) {
+				if (space_has_constraint(old_space,
+					index_def->name) != 0) {
+					diag_set(ClientError,
+						 ER_CONSTRAINT_EXISTS,
+						 index_def->name);
+					return -1;
+				}
+				if (space_put_constraint_name(old_space,
+							      index_def->name)
+				    != 0)
+					return -1;
+				struct trigger *on_rollback =
+					txn_alter_trigger_new(NULL, NULL);
+				if (on_rollback == NULL)
+					return -1;
+				on_rollback->data = index_def_dup(index_def);
+				on_rollback->run = on_create_index_rollback;
+				txn_stmt_on_rollback(stmt, on_rollback);
+			}
+			if (old_def->opts.is_unique && index_def->opts.is_unique
+			    && strcmp(old_def->name, index_def->name) != 0) {
+				if (space_has_constraint(old_space,
+							 index_def->name)
+				    != 0) {
+					diag_set(ClientError,
+						 ER_CONSTRAINT_EXISTS,
+						 index_def->name);
+					return -1;
+				}
+				if (space_put_constraint_name(old_space,
+							      index_def->name)
+				    != 0)
+					return -1;
+				space_drop_constraint_name(old_space,
+							   old_def->name);
+				/**
+				 * Array with the pair of 2
+				 * index_def structures: old and
+				 * new.
+				 */
+				struct index_def **defs = NULL;
+				defs = (struct index_def **)
+				       malloc(2 * sizeof(struct index_def *));
+				defs[0] = index_def_dup(old_def);
+				defs[1] = index_def_dup(index_def);
+				struct trigger *on_rollback =
+					txn_alter_trigger_new(NULL, NULL);
+				if (on_rollback == NULL)
+					return -1;
+				on_rollback->data = defs;
+				on_rollback->run = on_replace_index_rollback;
+				txn_stmt_on_rollback(stmt, on_rollback);
+			}
+			if (old_def->opts.is_unique &&
+			    !index_def->opts.is_unique) {
+				space_drop_constraint_name(old_space,
+							   old_def->name);
+				struct trigger *on_rollback =
+					txn_alter_trigger_new(NULL, NULL);
+				if (on_rollback == NULL)
+					return -1;
+				on_rollback->data = index_def_dup(old_def);
+				on_rollback->run = on_drop_index_rollback;
+				txn_stmt_on_rollback(stmt, on_rollback);
+			}
+		}
 		auto index_def_guard =
 			make_scoped_guard([=] { index_def_delete(index_def); });
 		/*
@@ -2487,6 +2692,7 @@ on_replace_dd_index(struct trigger * /* trigger */, void *event)
 	 */
 	alter_space_move_indexes(alter, iid + 1, old_space->index_id_max + 1);
 	(void) new MoveCkConstraints(alter);
+	(void) new MoveConstraints(alter);
 	/* Add an op to update schema_version on commit. */
 	(void) new UpdateSchemaVersion(alter);
 	try {
@@ -2576,6 +2782,7 @@ on_replace_dd_truncate(struct trigger * /* trigger */, void *event)
 	}
 
 	(void) new MoveCkConstraints(alter);
+	(void) new MoveConstraints(alter);
 	try {
 		alter_space_do(stmt, alter);
 	} catch (Exception *e) {
@@ -4777,6 +4984,8 @@ on_create_fk_constraint_rollback(struct trigger *trigger, void *event)
 	struct fk_constraint *fk = (struct fk_constraint *)trigger->data;
 	rlist_del_entry(fk, in_parent_space);
 	rlist_del_entry(fk, in_child_space);
+	space_drop_constraint_name(space_by_id(fk->def->child_id),
+				   fk->def->name);
 	space_reset_fk_constraint_mask(space_by_id(fk->def->parent_id));
 	space_reset_fk_constraint_mask(space_by_id(fk->def->child_id));
 	fk_constraint_delete(fk);
@@ -4812,6 +5021,10 @@ on_drop_fk_constraint_rollback(struct trigger *trigger, void *event)
 	struct space *child = space_by_id(old_fk->def->child_id);
 	rlist_add_entry(&child->child_fk_constraint, old_fk, in_child_space);
 	rlist_add_entry(&parent->parent_fk_constraint, old_fk, in_parent_space);
+	if (space_put_constraint_name(child, old_fk->def->name) != 0) {
+		panic("Can't recover after FK constraint drop rollback (out of "
+		      "memory)");
+	}
 	fk_constraint_set_mask(old_fk, &child->fk_constraint_mask,
 			       FIELD_LINK_CHILD);
 	fk_constraint_set_mask(old_fk, &parent->fk_constraint_mask,
@@ -4987,6 +5200,10 @@ on_replace_dd_fk_constraint(struct trigger * /* trigger*/, void *event)
 		fk->def = fk_def;
 		fk->index_id = fk_index->def->iid;
 		if (old_tuple == NULL) {
+			if (space_has_constraint(child_space, fk_def->name)) {
+				diag_set(ClientError, ER_CONSTRAINT_EXISTS, fk_def->name);
+				return -1;
+			}
 			rlist_add_entry(&child_space->child_fk_constraint,
 					fk, in_child_space);
 			rlist_add_entry(&parent_space->parent_fk_constraint,
@@ -4996,6 +5213,9 @@ on_replace_dd_fk_constraint(struct trigger * /* trigger*/, void *event)
 						      fk);
 			if (on_rollback == NULL)
 				return -1;
+			if (space_put_constraint_name(child_space, fk_def->name)
+			    != 0)
+				return -1;
 			txn_stmt_on_rollback(stmt, on_rollback);
 			fk_constraint_set_mask(fk,
 					       &parent_space->fk_constraint_mask,
@@ -5041,6 +5261,7 @@ on_replace_dd_fk_constraint(struct trigger * /* trigger*/, void *event)
 		struct fk_constraint *old_fk=
 			fk_constraint_remove(&child_space->child_fk_constraint,
 					     fk_def->name);
+		space_drop_constraint_name(child_space, old_fk->def->name);
 		struct trigger *on_commit =
 			txn_alter_trigger_new(on_drop_or_replace_fk_constraint_commit,
 					      old_fk);
@@ -5106,9 +5327,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);
-	assert(space_ck_constraint_by_name(space, ck->def->name,
-					   strlen(ck->def->name)) != NULL);
+	const char *name = ck->def->name;
+	assert(space_ck_constraint_by_name(space, name, strlen(name)) != NULL);
 	space_remove_ck_constraint(space, ck);
+	space_drop_constraint_name(space, name);
 	ck_constraint_delete(ck);
 	if (trigger_run(&on_alter_space, space) != 0)
 		return -1;
@@ -5121,6 +5343,9 @@ 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);
+	if (space != NULL)
+		space_drop_constraint_name(space, ck->def->name);
 	ck_constraint_delete(ck);
 	return 0;
 }
@@ -5135,7 +5360,8 @@ on_drop_ck_constraint_rollback(struct trigger *trigger, void * /* event */)
 	assert(space != NULL);
 	assert(space_ck_constraint_by_name(space, ck->def->name,
 					   strlen(ck->def->name)) == NULL);
-	if (space_add_ck_constraint(space, ck) != 0)
+	if (space_add_ck_constraint(space, ck) != 0 ||
+	    space_put_constraint_name(space, ck->def->name) != 0)
 		panic("Can't recover after CK constraint drop rollback");
 	if (trigger_run(&on_alter_space, space) != 0)
 		return -1;
@@ -5250,8 +5476,14 @@ on_replace_dd_ck_constraint(struct trigger * /* trigger*/, void *event)
 		auto ck_guard = make_scoped_guard([=] {
 			ck_constraint_delete(new_ck_constraint);
 		});
-		if (old_ck_constraint != NULL)
+		if (old_ck_constraint != NULL) {
 			rlist_del_entry(old_ck_constraint, link);
+		} else if (space_has_constraint(space, name)) {
+			diag_set(ClientError, ER_CONSTRAINT_EXISTS, name);
+			return -1;
+		} else if (space_put_constraint_name(space, name) != 0) {
+			return -1;
+		}
 		if (space_add_ck_constraint(space, new_ck_constraint) != 0)
 			return -1;
 		ck_guard.is_active = false;
@@ -5274,6 +5506,7 @@ on_replace_dd_ck_constraint(struct trigger * /* trigger*/, void *event)
 			return -1;
 		struct ck_constraint *old_ck_constraint =
 			space_ck_constraint_by_name(space, name, name_len);
+		space_drop_constraint_name(space, old_ck_constraint->def->name);
 		assert(old_ck_constraint != NULL);
 		space_remove_ck_constraint(space, old_ck_constraint);
 		on_commit->data = old_ck_constraint;
@@ -5372,6 +5605,7 @@ on_replace_dd_func_index(struct trigger *trigger, void *event)
 	alter_space_move_indexes(alter, index->def->iid + 1,
 				 space->index_id_max + 1);
 	(void) new MoveCkConstraints(alter);
+	(void) new MoveConstraints(alter);
 	(void) new UpdateSchemaVersion(alter);
 	try {
 		alter_space_do(stmt, alter);
diff --git a/test/sql-tap/constraint.test.lua b/test/sql-tap/constraint.test.lua
new file mode 100755
index 000000000..6444c06e9
--- /dev/null
+++ b/test/sql-tap/constraint.test.lua
@@ -0,0 +1,244 @@
+#!/usr/bin/env tarantool
+test = require("sqltester")
+test:plan(20)
+
+--!./tcltestrunner.lua
+-- 2001 September 15
+--
+-- The author disclaims copyright to this source code.  In place
+-- of a legal notice, here is a blessing:
+--
+--    May you do good and not evil.
+--    May you find forgiveness for yourself and forgive others.
+--    May you share freely, never taking more than you give.
+--
+------------------------------------------------------------------
+-- This file implements regression tests for sql library. The
+-- focus of this file is testing the constraint creation, alter
+-- and drop.
+--
+
+--
+-- gh-3503 Check constraint name for duplicate.
+--
+test:do_execsql_test(
+    "constraint-1.1",
+    [[
+        CREATE TABLE t1 (i INT PRIMARY KEY);
+        DROP TABLE IF EXISTS t2;
+    ]], {
+        -- <constraint-1.1>
+        -- </constraint-1.1>
+    })
+
+test:do_catchsql_test(
+    "constraint-1.2",
+    [[
+        CREATE TABLE t2 (i INT, CONSTRAINT c CHECK (i > 0),
+                         CONSTRAINT c PRIMARY KEY (i));
+    ]], {
+        -- <constraint-1.2>
+        1,"Constraint C already exists"
+        -- </constraint-1.2>
+    })
+
+test:do_catchsql_test(
+    "constraint-1.3",
+    [[
+        CREATE TABLE t2 (i INT,
+                         CONSTRAINT c FOREIGN KEY(i) REFERENCES t1(i),
+                         CONSTRAINT c PRIMARY KEY (i));
+    ]], {
+        -- <constraint-1.3>
+        1,"Constraint C already exists"
+        -- </constraint-1.3>
+    })
+
+test:do_catchsql_test(
+    "constraint-1.4",
+    [[
+        CREATE TABLE t2 (i INT PRIMARY KEY,
+                         CONSTRAINT c CHECK (i > 0),
+                         CONSTRAINT c UNIQUE (i));
+    ]], {
+        -- <constraint-1.4>
+        1,"Constraint C already exists"
+        -- </constraint-1.4>
+    })
+
+test:do_catchsql_test(
+    "constraint-1.5",
+    [[
+        CREATE TABLE t2 (i INT PRIMARY KEY,
+                         CONSTRAINT c FOREIGN KEY(i) REFERENCES t1(i),
+                         CONSTRAINT c UNIQUE (i));
+    ]], {
+        -- <constraint-1.5>
+        1,"Constraint C already exists"
+        -- </constraint-1.5>
+    })
+
+test:do_catchsql_test(
+    "constraint-1.6",
+    [[
+        CREATE TABLE t2 (i INT CONSTRAINT c PRIMARY KEY,
+                         CONSTRAINT c CHECK (i > 0));
+    ]], {
+        -- <constraint-1.6>
+        1,"Constraint C already exists"
+        -- </constraint-1.6>
+    })
+
+test:do_catchsql_test(
+    "constraint-1.7",
+    [[
+        CREATE TABLE t2 (i INT PRIMARY KEY,
+                         CONSTRAINT c UNIQUE (i),
+                         CONSTRAINT c CHECK (i > 0));
+    ]], {
+        -- <constraint-1.7>
+        1,"Constraint C already exists"
+        -- </constraint-1.7>
+    })
+
+test:do_catchsql_test(
+    "constraint-1.8",
+    [[
+        CREATE TABLE t2 (i INT PRIMARY KEY,
+                         CONSTRAINT c CHECK (i > 0),
+                         CONSTRAINT c CHECK (i < 0));
+    ]], {
+        -- <constraint-1.8>
+        1,"Constraint C already exists"
+        -- </constraint-1.8>
+    })
+
+test:do_catchsql_test(
+    "constraint-1.9",
+    [[
+        CREATE TABLE t2 (i INT PRIMARY KEY,
+                         CONSTRAINT c FOREIGN KEY(i) REFERENCES t1(i),
+                         CONSTRAINT c CHECK (i > 0));
+    ]], {
+        -- <constraint-1.9>
+        1,"Constraint C already exists"
+        -- </constraint-1.9>
+    })
+
+test:do_catchsql_test(
+    "constraint-1.10",
+    [[
+        CREATE TABLE t2 (i INT CONSTRAINT c PRIMARY KEY,
+                         CONSTRAINT c FOREIGN KEY(i) REFERENCES t1(i));
+    ]], {
+        -- <constraint-1.10>
+        1,"Constraint C already exists"
+        -- </constraint-1.10>
+    })
+
+test:do_catchsql_test(
+    "constraint-1.11",
+    [[
+        CREATE TABLE t2 (i INT PRIMARY KEY,
+                         CONSTRAINT c UNIQUE (i),
+                         CONSTRAINT c FOREIGN KEY(i) REFERENCES t1(i));
+    ]], {
+        -- <constraint-1.11>
+        1,"Constraint C already exists"
+        -- </constraint-1.11>
+    })
+
+test:do_catchsql_test(
+    "constraint-1.12",
+    [[
+        CREATE TABLE t2 (i INT PRIMARY KEY,
+                         CONSTRAINT c CHECK (i > 0),
+                         CONSTRAINT c FOREIGN KEY(i) REFERENCES t1(i));
+    ]], {
+        -- <constraint-1.12>
+        1,"Constraint C already exists"
+        -- </constraint-1.12>
+    })
+
+test:do_catchsql_test(
+    "constraint-1.13",
+    [[
+        CREATE TABLE t2 (i INT PRIMARY KEY CONSTRAINT c REFERENCES t1(i),
+                         CONSTRAINT c FOREIGN KEY(i) REFERENCES t1(i));
+    ]], {
+        -- <constraint-1.13>
+        1,"Constraint C already exists"
+        -- </constraint-1.13>
+    })
+
+--
+-- gh-3503 Make sure that _constraint works.
+--
+test:do_execsql_test(
+    "constraint-2.1",
+    [[
+        DROP TABLE IF EXISTS t2;
+        CREATE TABLE t2 (i INT CONSTRAINT c PRIMARY KEY);
+    ]], {
+        -- <constraint-2.1>
+        -- </constraint-2.1>
+    })
+
+test:do_catchsql_test(
+    "constraint-2.2",
+    [[
+        ALTER TABLE t2 ADD CONSTRAINT c CHECK(i > 0);
+    ]], {
+        -- <constraint-2.2>
+        1, "Constraint C already exists"
+        -- </constraint-2.2>
+    })
+
+test:do_catchsql_test(
+    "constraint-2.3",
+    [[
+        ALTER TABLE t2 ADD CONSTRAINT c UNIQUE(i);
+    ]], {
+        -- <constraint-2.3>
+        1, "Index 'C' already exists in space 'T2'"
+        -- </constraint-2.3>
+    })
+
+test:do_catchsql_test(
+    "constraint-2.4",
+    [[
+        ALTER TABLE t2 ADD CONSTRAINT c FOREIGN KEY(i) REFERENCES t1(i);
+    ]], {
+        -- <constraint-2.4>
+        1, "Constraint C already exists"
+        -- </constraint-2.4>
+    })
+
+test:do_execsql_test(
+    "constraint-2.5",
+    [[
+        ALTER TABLE t2 ADD CONSTRAINT f FOREIGN KEY(i) REFERENCES t1(i);
+    ]], {
+        -- <constraint-2.5>
+        -- </constraint-2.5>
+    })
+
+test:do_execsql_test(
+    "constraint-2.6",
+    [[
+        ALTER TABLE t2 DROP CONSTRAINT f;
+    ]], {
+        -- <constraint-2.6>
+        -- </constraint-2.6>
+    })
+
+test:do_execsql_test(
+    "constraint-2.7",
+    [[
+        ALTER TABLE t2 ADD CONSTRAINT f FOREIGN KEY(i) REFERENCES t1(i);
+    ]], {
+        -- <constraint-2.7>
+        -- </constraint-2.7>
+    })
+
+test:finish_test()
\ No newline at end of file
-- 
2.21.0 (Apple Git-122)