[tarantool-patches] [PATCH 4/5] sql: display error on FK creation and drop failure

[Nikita Pettik <korablev@tarantool.org>]

Before insertion to _fk_constraint we must be sure that there in no
entry with given <name, child id>. Otherwise, insertion will fail and
'duplicate key' will be shown. Such error message doesn't seem to be
informative enough, so lets manually iterate through whole space looking
for appropriate record.
The same is for dropping constraint, but here vice versa: we test
that _fk_contraint contains entry with given name and child id.

It is worth mentioning that during CREATE TABLE processing schema id
changes and check in OP_OpenRead opcode fails (which in turn shows that
pointer to space may expire). On the other hand, _fk_constraint space
itself remains immutable, so as a temporary workaround lets use flag
indicating pointer to system space passed to OP_OpenRead. It makes
possible to use pointer to space, even if schema has changed.

Closes #3271
---
 src/box/errcode.h            |  2 ++
 src/box/sql/build.c          | 43 +++++++++++++++++++++++++++++++------------
 src/box/sql/sqliteInt.h      | 10 +++++++---
 src/box/sql/trigger.c        | 24 ++++++++++++++++--------
 src/box/sql/vdbe.c           |  3 ++-
 test/box/misc.result         |  2 ++
 test/sql-tap/alter2.test.lua | 25 ++++++++++++++++++++++++-
 7 files changed, 84 insertions(+), 25 deletions(-)

diff --git a/src/box/errcode.h b/src/box/errcode.h
index 1558cfae8..a3c27d6b8 100644
--- a/src/box/errcode.h
+++ b/src/box/errcode.h
@@ -217,6 +217,8 @@ struct errcode_record {
 	/*162 */_(ER_FOREIGN_KEY_CONSTRAINT,	"Can not commit transaction: deferred foreign keys violations are not resolved") \
 	/*163 */_(ER_CREATE_FK_CONSTRAINT,	"Failed to create foreign key constraint '%s': %s") \
 	/*164 */_(ER_DROP_FK_CONSTRAINT,	"Failed to drop foreign key constraint '%s': %s") \
+	/*165 */_(ER_NO_SUCH_CONSTRAINT,	"Constraint %s does not exist") \
+	/*165 */_(ER_CONSTRAINT_EXISTS,		"Constraint %s already exists") \
 
 
 
diff --git a/src/box/sql/build.c b/src/box/sql/build.c
index c2d3cd035..20ace09e4 100644
--- a/src/box/sql/build.c
+++ b/src/box/sql/build.c
@@ -1784,6 +1784,20 @@ vdbe_fkey_code_creation(struct Parse *parse_context, const struct fkey_def *fk)
 		sqlite3VdbeAddOp2(vdbe, OP_Integer, fk->parent_id,
 				  constr_tuple_reg + 2);
 	}
+	/*
+	 * Lets check that constraint with this name hasn't
+	 * been created before.
+	 */
+	const char *error_msg =
+		tt_sprintf(tnt_errcode_desc(ER_CONSTRAINT_EXISTS), name_copy);
+	if (vdbe_emit_halt_with_presence_test(parse_context,
+					      BOX_FK_CONSTRAINT_ID, 0,
+					      constr_tuple_reg, 2,
+					      ER_CONSTRAINT_EXISTS, error_msg,
+					      false, OP_NoConflict) != 0) {
+		free((void *) name_copy);
+		return;
+	}
 	sqlite3VdbeAddOp2(vdbe, OP_Bool, 0, constr_tuple_reg + 3);
 	sqlite3VdbeChangeP4(vdbe, -1, (char*)&fk->is_deferred, P4_BOOL);
 	sqlite3VdbeAddOp4(vdbe, OP_String8, 0, constr_tuple_reg + 4, 0,
@@ -2224,6 +2238,17 @@ vdbe_fkey_code_drop(struct Parse *parse_context, const char *constraint_name,
 	sqlite3VdbeAddOp4(vdbe, OP_String8, 0, key_reg, 0, constraint_name,
 			  P4_DYNAMIC);
 	sqlite3VdbeAddOp2(vdbe, OP_Integer, child_id,  key_reg + 1);
+	const char *error_msg =
+		tt_sprintf(tnt_errcode_desc(ER_NO_SUCH_CONSTRAINT),
+			   constraint_name);
+	if (vdbe_emit_halt_with_presence_test(parse_context,
+					      BOX_FK_CONSTRAINT_ID, 0,
+					      key_reg, 2, ER_NO_SUCH_CONSTRAINT,
+					      error_msg, false,
+					      OP_Found) != 0) {
+		free((void *) constraint_name);
+		return;
+	}
 	sqlite3VdbeAddOp3(vdbe, OP_MakeRecord, key_reg, 2, key_reg + 2);
 	sqlite3VdbeAddOp2(vdbe, OP_SDelete, BOX_FK_CONSTRAINT_ID, key_reg + 2);
 	sqlite3VdbeChangeP5(vdbe, OPFLAG_NCHANGE);
@@ -4247,7 +4272,7 @@ sqlite3WithDelete(sqlite3 * db, With * pWith)
 
 int
 vdbe_emit_halt_with_presence_test(struct Parse *parser, int space_id,
-				  int index_id, const char *name_src,
+				  int index_id, int key_reg, uint32_t key_len,
 				  int tarantool_error_code,
 				  const char *error_src, bool no_error,
 				  int cond_opcode)
@@ -4257,22 +4282,16 @@ vdbe_emit_halt_with_presence_test(struct Parse *parser, int space_id,
 	assert(v != NULL);
 
 	struct sqlite3 *db = parser->db;
-	char *name = sqlite3DbStrDup(db, name_src);
-	if (name == NULL)
-		return -1;
 	char *error = sqlite3DbStrDup(db, error_src);
-	if (error == NULL) {
-		sqlite3DbFree(db, name);
+	if (error == NULL)
 		return -1;
-	}
 
 	int cursor = parser->nTab++;
 	vdbe_emit_open_cursor(parser, cursor, index_id, space_by_id(space_id));
-
-	int name_reg = parser->nMem++;
-	int label = sqlite3VdbeAddOp4(v, OP_String8, 0, name_reg, 0, name,
-				      P4_DYNAMIC);
-	sqlite3VdbeAddOp4Int(v, cond_opcode, cursor, label + 3, name_reg, 1);
+	sqlite3VdbeChangeP5(v, OPFLAG_SYSTEMSP);
+	int label = sqlite3VdbeCurrentAddr(v);
+	sqlite3VdbeAddOp4Int(v, cond_opcode, cursor, label + 3, key_reg,
+			     key_len);
 	if (no_error) {
 		sqlite3VdbeAddOp0(v, OP_Halt);
 	} else {
diff --git a/src/box/sql/sqliteInt.h b/src/box/sql/sqliteInt.h
index 2489b31b2..f882d747d 100644
--- a/src/box/sql/sqliteInt.h
+++ b/src/box/sql/sqliteInt.h
@@ -3012,6 +3012,9 @@ struct Parse {
 					 * is fresh, even in case schema
 					 * changes previously.
 					 */
+#define OPFLAG_SYSTEMSP      0x40	/* OP_Open**: set if space pointer
+					 * points to system space.
+					 */
 
 /*
  * Each trigger present in the database schema is stored as an
@@ -4870,7 +4873,7 @@ table_column_nullable_action(struct Table *tab, uint32_t column);
 
 /**
  * Generate VDBE code to halt execution with correct error if
- * the object with specified name is already present (or doesn't
+ * the object with specified key is already present (or doesn't
  * present - configure with cond_opcodeq) in specified space.
  * The function allocates error and name resources for VDBE
  * itself.
@@ -4878,7 +4881,8 @@ table_column_nullable_action(struct Table *tab, uint32_t column);
  * @param parser Parsing context.
  * @param space_id Space to lookup identifier.
  * @param index_id Index identifier of key.
- * @param name_src Name of object to test on existence.
+ * @param key_reg Register where key to be found is held.
+ * @param key_len Lenght of key (number of resiters).
  * @param tarantool_error_code to set on halt.
  * @param error_src Error message to display on VDBE halt.
  * @param no_error Do not raise error flag.
@@ -4890,7 +4894,7 @@ table_column_nullable_action(struct Table *tab, uint32_t column);
  */
 int
 vdbe_emit_halt_with_presence_test(struct Parse *parser, int space_id,
-				  int index_id, const char *name_src,
+				  int index_id, int key_reg, uint32_t key_len,
 				  int tarantool_error_code,
 				  const char *error_src, bool no_error,
 				  int cond_opcode);
diff --git a/src/box/sql/trigger.c b/src/box/sql/trigger.c
index 801013b5a..c24235128 100644
--- a/src/box/sql/trigger.c
+++ b/src/box/sql/trigger.c
@@ -125,8 +125,14 @@ sql_trigger_begin(struct Parse *parse, struct Token *name, int tr_tm,
 		const char *error_msg =
 			tt_sprintf(tnt_errcode_desc(ER_TRIGGER_EXISTS),
 				   trigger_name);
+		char *name_copy = sqlite3DbStrDup(db, trigger_name);
+		if (name_copy == NULL)
+			goto trigger_cleanup;
+		int name_reg = ++parse->nMem;
+		sqlite3VdbeAddOp4(parse->pVdbe, OP_String8, 0, name_reg, 0,
+				  name_copy, P4_DYNAMIC);
 		if (vdbe_emit_halt_with_presence_test(parse, BOX_TRIGGER_ID, 0,
-						      trigger_name,
+						      name_reg, 1,
 						      ER_TRIGGER_EXISTS,
 						      error_msg, (no_err != 0),
 						      OP_NoConflict) != 0)
@@ -472,26 +478,28 @@ sql_drop_trigger(struct Parse *parser, struct SrcList *name, bool no_err)
 	if (db->mallocFailed)
 		goto drop_trigger_cleanup;
 	assert(db->pSchema != NULL);
-
+	struct Vdbe *v = sqlite3GetVdbe(parser);
 	/* Do not account nested operations: the count of such
 	 * operations depends on Tarantool data dictionary internals,
 	 * such as data layout in system spaces. Activate the counter
 	 * here to account DROP TRIGGER IF EXISTS case if the trigger
 	 * actually does not exist.
 	 */
-	if (!parser->nested) {
-		Vdbe *v = sqlite3GetVdbe(parser);
-		if (v != NULL)
-			sqlite3VdbeCountChanges(v);
-	}
+	if (!parser->nested && v != NULL)
+		sqlite3VdbeCountChanges(v);
 
 	assert(name->nSrc == 1);
 	const char *trigger_name = name->a[0].zName;
 	const char *error_msg =
 		tt_sprintf(tnt_errcode_desc(ER_NO_SUCH_TRIGGER),
 			   trigger_name);
+	char *name_copy = sqlite3DbStrDup(db, trigger_name);
+	if (name_copy == NULL)
+		goto drop_trigger_cleanup;
+	int name_reg = ++parser->nMem;
+	sqlite3VdbeAddOp4(v, OP_String8, 0, name_reg, 0, name_copy, P4_DYNAMIC);
 	if (vdbe_emit_halt_with_presence_test(parser, BOX_TRIGGER_ID, 0,
-					      trigger_name, ER_NO_SUCH_TRIGGER,
+					      name_reg, 1, ER_NO_SUCH_TRIGGER,
 					      error_msg, no_err, OP_Found) != 0)
 		goto drop_trigger_cleanup;
 
diff --git a/src/box/sql/vdbe.c b/src/box/sql/vdbe.c
index b9723e2e7..0f227e637 100644
--- a/src/box/sql/vdbe.c
+++ b/src/box/sql/vdbe.c
@@ -3172,7 +3172,8 @@ case OP_OpenWrite:
 	 * during runtime.
 	 */
 	if (box_schema_version() != p->schema_ver &&
-	    (pOp->p5 & OPFLAG_FRESH_PTR) == 0) {
+	    (pOp->p5 & OPFLAG_FRESH_PTR) == 0 &&
+	    (pOp->p5 & OPFLAG_SYSTEMSP) == 0) {
 		p->expired = 1;
 		rc = SQLITE_ERROR;
 		sqlite3VdbeError(p, "schema version has changed: " \
diff --git a/test/box/misc.result b/test/box/misc.result
index 315499f3e..fb7c5311c 100644
--- a/test/box/misc.result
+++ b/test/box/misc.result
@@ -489,6 +489,8 @@ t;
   162: box.error.FOREIGN_KEY_CONSTRAINT
   163: box.error.CREATE_FK_CONSTRAINT
   164: box.error.DROP_FK_CONSTRAINT
+  165: box.error.NO_SUCH_CONSTRAINT
+  166: box.error.CONSTRAINT_EXISTS
 ...
 test_run:cmd("setopt delimiter ''");
 ---
diff --git a/test/sql-tap/alter2.test.lua b/test/sql-tap/alter2.test.lua
index e4470ecbb..be83c225f 100755
--- a/test/sql-tap/alter2.test.lua
+++ b/test/sql-tap/alter2.test.lua
@@ -1,6 +1,6 @@
 #!/usr/bin/env tarantool
 test = require("sqltester")
-test:plan(15)
+test:plan(17)
 
 -- This suite is aimed to test ALTER TABLE ADD CONSTRAINT statement.
 --
@@ -193,4 +193,27 @@ test:do_execsql_test(
         -- </alter2-3.2>
     })
 
+test:do_catchsql_test(
+    "alter2-4.1",
+    [[
+        DROP TABLE child;
+        CREATE TABLE child (id PRIMARY KEY, a UNIQUE);
+        ALTER TABLE child ADD CONSTRAINT fk FOREIGN KEY (id) REFERENCES child;
+        ALTER TABLE child ADD CONSTRAINT fk FOREIGN KEY (a) REFERENCES child;
+    ]], {
+        -- <alter2-4.1>
+        1, "Constraint FK already exists"
+        -- </alter2-4.1>
+    })
+
+test:do_catchsql_test(
+    "alter2-4.2",
+    [[
+        ALTER TABLE child DROP CONSTRAINT fake;
+    ]], {
+        -- <alter2-4.2>
+        1, "Constraint FAKE does not exist"
+        -- </alter2-4.2>
+    })
+
 test:finish_test()
-- 
2.15.1