[Tarantool-patches] [PATCH] sql: fix fk-constraint error on update of a self-referenced table

[Chris Sosnin <k.sosnin@tarantool.org>]

Currently, update operation of a self-referenced table is not handled
correctly. Without this fix, we will succeed only if the row refers to
itself, otherwise, the nFkConstraint will be incremented. As it is
mentioned in comments, we shouldn't do the lookup of the table on deletion,
as long as the tuple will be modified, but it is still necessary to inspect
it on insertion.

Closes #4661
---
branch:https://github.com/tarantool/tarantool/tree/ksosnin/gh-4661-fk-self-ref-table
issue:https://github.com/tarantool/tarantool/issues/4661

 src/box/sql/fk_constraint.c    | 51 ++++++++++++++++++++++++++++------
 test/sql/foreign-keys.result   | 44 +++++++++++++++++++++++++++++
 test/sql/foreign-keys.test.lua | 12 ++++++++
 3 files changed, 98 insertions(+), 9 deletions(-)

diff --git a/src/box/sql/fk_constraint.c b/src/box/sql/fk_constraint.c
index 482220a95..8a329a00d 100644
--- a/src/box/sql/fk_constraint.c
+++ b/src/box/sql/fk_constraint.c
@@ -199,6 +199,7 @@ fk_constraint_lookup_parent(struct Parse *parse_context, struct space *parent,
 	struct Vdbe *v = sqlGetVdbe(parse_context);
 	int cursor = parse_context->nTab - 1;
 	int ok_label = sqlVdbeMakeLabel(v);
+	bool is_self_referenced = fk_constraint_is_self_referenced(fk_def);
 	/*
 	 * If incr_count is less than zero, then check at runtime
 	 * if there are any outstanding constraints to resolve.
@@ -225,7 +226,8 @@ fk_constraint_lookup_parent(struct Parse *parse_context, struct space *parent,
 	 * we are about to increment the constraint-counter (i.e.
 	 * this is an INSERT operation), then check if the row
 	 * being inserted matches itself. If so, do not increment
-	 * the constraint-counter.
+	 * the constraint-counter. Otherwise, if this is an insertion
+	 * during update, we must inspect the table.
 	 *
 	 * If any of the parent-key values are NULL, then the row
 	 * cannot match itself. So set JUMPIFNULL to make sure we
@@ -233,7 +235,7 @@ fk_constraint_lookup_parent(struct Parse *parse_context, struct space *parent,
 	 * NULL (at this point it is known that none of the child
 	 * key values are).
 	 */
-	if (fk_constraint_is_self_referenced(fk_def) && incr_count == 1) {
+	if (is_self_referenced && incr_count == 1) {
 		int jump = sqlVdbeCurrentAddr(v) + field_count + 1;
 		link = fk_def->links;
 		for (uint32_t i = 0; i < field_count; ++i, ++link) {
@@ -244,14 +246,16 @@ fk_constraint_lookup_parent(struct Parse *parse_context, struct space *parent,
 		}
 		sqlVdbeGoto(v, ok_label);
 	}
-	/**
+	/* This case is handled explicitly. */
+	bool self_insert_on_update = incr_count == 1 && is_update &&
+			       is_self_referenced;
+	/*
 	 * Inspect a parent table with OP_Found.
-	 * We mustn't make it for a self-referenced table since
-	 * it's tuple will be modified by the update operation.
-	 * And since the foreign key has already detected a
-	 * conflict, fk counter must be increased.
+	 * We mustn't make it for a self-referenced
+	 * table on deletion during an update since
+	 * it's tuple will be modified later.
 	 */
-	if (!(fk_constraint_is_self_referenced(fk_def) && is_update)) {
+	if (!(is_self_referenced && is_update) || self_insert_on_update) {
 		int temp_regs = sqlGetTempRange(parse_context, field_count);
 		int rec_reg = sqlGetTempReg(parse_context);
 		vdbe_emit_open_cursor(parse_context, cursor, referenced_idx,
@@ -269,7 +273,36 @@ fk_constraint_lookup_parent(struct Parse *parse_context, struct space *parent,
 				  (char *) sql_index_type_str(parse_context->db,
 							      idx->def),
 				  P4_DYNAMIC);
-		sqlVdbeAddOp4Int(v, OP_Found, cursor, ok_label, rec_reg, 0);
+		/*
+		 * In case of an update on a self-referenced table, we check,
+		 * whether we found the same tuple we're updating. It is
+		 * necessary, since if we're changing the value of the parent,
+		 * which was equal to the child, we are still going to succeed
+		 * and find the old tuple. However, such update obviously
+		 * violates fk constraint. And as long as we've already checked,
+		 * that new values child-parent are different, this check is
+		 * correct.
+		 */
+		int jump = sqlVdbeCurrentAddr(v) + 1;
+		sqlVdbeAddOp4Int(v, OP_Found, cursor,
+				 self_insert_on_update ? jump + 1 : ok_label,
+				 rec_reg, 0);
+		if (self_insert_on_update) {
+			/* We found nothing, proceed to counter incrementing. */
+			sqlVdbeGoto(v, jump + 4);
+			/*
+			 * Otherwise, cursor is pointing on the found entry,
+			 * compare its index to the index of the tuple being
+			 * updated.
+			 */
+			int idx_reg = sqlGetTempReg(parse_context);
+			sqlVdbeAddOp3(v, OP_Column, cursor, referenced_idx,
+				      idx_reg);
+			sqlVdbeAddOp3(v, OP_Eq, reg_data + referenced_idx + 1,
+				      jump + 4, idx_reg);
+			sqlReleaseTempReg(parse_context, idx_reg);
+			sqlVdbeGoto(v, ok_label);
+		}
 		sqlReleaseTempReg(parse_context, rec_reg);
 		sqlReleaseTempRange(parse_context, temp_regs, field_count);
 	}
diff --git a/test/sql/foreign-keys.result b/test/sql/foreign-keys.result
index f1d973443..a9b10ee38 100644
--- a/test/sql/foreign-keys.result
+++ b/test/sql/foreign-keys.result
@@ -429,6 +429,50 @@ box.execute("CREATE TABLE t4 (id INT PRIMARY KEY REFERENCES t4);")
 box.space.T4:drop()
 ---
 ...
+-- gh-4661: UPDATE must work correctly on self-referenced spaces.
+--
+box.execute('CREATE TABLE t5(a INT UNIQUE, b INT REFERENCES t5(a), id INT PRIMARY KEY AUTOINCREMENT)')
+---
+- row_count: 1
+...
+box.execute('INSERT INTO t5(a,b) VALUES(1, null)')
+---
+- autoincrement_ids:
+  - 1
+  row_count: 1
+...
+box.execute('INSERT INTO t5(a,b) VALUES(2, null)')
+---
+- autoincrement_ids:
+  - 2
+  row_count: 1
+...
+box.execute('UPDATE t5 SET b = 2 WHERE a = 1')
+---
+- row_count: 1
+...
+box.execute('UPDATE t5 SET b = 1 WHERE a = 2')
+---
+- row_count: 1
+...
+box.execute('INSERT INTO t5(a,b) VALUES(3, 2)')
+---
+- autoincrement_ids:
+  - 3
+  row_count: 1
+...
+box.execute('UPDATE t5 SET a = 4 WHERE a = 3')
+---
+- row_count: 1
+...
+box.execute('UPDATE t5 SET b = 4 WHERE a = 1')
+---
+- row_count: 1
+...
+box.execute('DROP TABLE t5')
+---
+- row_count: 1
+...
 -- Make sure that child space can feature no PK.
 --
 t1 = box.schema.create_space("T1", {format = {'ID'}})
diff --git a/test/sql/foreign-keys.test.lua b/test/sql/foreign-keys.test.lua
index d2dd88d28..4fb651787 100644
--- a/test/sql/foreign-keys.test.lua
+++ b/test/sql/foreign-keys.test.lua
@@ -179,6 +179,18 @@ box.space.T1:drop()
 box.execute("CREATE TABLE t4 (id INT PRIMARY KEY REFERENCES t4);")
 box.space.T4:drop()
 
+-- gh-4661: UPDATE must work correctly on self-referenced spaces.
+--
+box.execute('CREATE TABLE t5(a INT UNIQUE, b INT REFERENCES t5(a), id INT PRIMARY KEY AUTOINCREMENT)')
+box.execute('INSERT INTO t5(a,b) VALUES(1, null)')
+box.execute('INSERT INTO t5(a,b) VALUES(2, null)')
+box.execute('UPDATE t5 SET b = 2 WHERE a = 1')
+box.execute('UPDATE t5 SET b = 1 WHERE a = 2')
+box.execute('INSERT INTO t5(a,b) VALUES(3, 2)')
+box.execute('UPDATE t5 SET a = 4 WHERE a = 3')
+box.execute('UPDATE t5 SET b = 4 WHERE a = 1')
+box.execute('DROP TABLE t5')
+
 -- Make sure that child space can feature no PK.
 --
 t1 = box.schema.create_space("T1", {format = {'ID'}})
-- 
2.21.1 (Apple Git-122.3)