[tarantool-patches] [PATCH v1 1/1] sql: move space existence check to VDBE

[imeevma@tarantool.org]

Before this patch, the existence of space was checked for the
CREATE TABLE or CREATE VIEW statements during the parsing. If the
space already exists, an error has been set and cleanup is
performed. But, if the statement contained 'IF NOT EXIST', the
cleanup was performed, but the error was not set. This causes an
error when creating a foreign key during space creation.

This patch moves this check to VDBE. Thus, the parsing should now
be properly closed before performing this check.

Closes #4196
---
https://github.com/tarantool/tarantool/issues/4196
https://github.com/tarantool/tarantool/tree/imeevma/gh-4196-move-space-existence-check-to-vdbe

 src/box/sql/build.c         | 49 +++++++++++++++++++++++++++++++++------------
 src/box/sql/parse.y         |  2 +-
 src/box/sql/sqlInt.h        |  2 +-
 test/sql-tap/table.test.lua | 30 +++++++++++++++++++++++++--
 4 files changed, 66 insertions(+), 17 deletions(-)

diff --git a/src/box/sql/build.c b/src/box/sql/build.c
index 6051a25..a25479e 100644
--- a/src/box/sql/build.c
+++ b/src/box/sql/build.c
@@ -300,10 +300,9 @@ sql_space_primary_key(const struct space *space)
  *
  * @param pParse Parser context.
  * @param pName1 First part of the name of the table or view.
- * @param noErr Do nothing if table already exists.
  */
 struct space *
-sqlStartTable(Parse *pParse, Token *pName, int noErr)
+sqlStartTable(Parse *pParse, Token *pName)
 {
 	char *zName = 0;	/* The name of the new table */
 	sql *db = pParse->db;
@@ -322,15 +321,6 @@ sqlStartTable(Parse *pParse, Token *pName, int noErr)
 	if (sqlCheckIdentifierName(pParse, zName) != 0)
 		goto cleanup;
 
-	struct space *space = space_by_name(zName);
-	if (space != NULL) {
-		if (!noErr) {
-			diag_set(ClientError, ER_SPACE_EXISTS, zName);
-			pParse->is_aborted = true;
-		}
-		goto cleanup;
-	}
-
 	new_space = sql_ephemeral_space_new(pParse, zName);
 	if (new_space == NULL)
 		goto cleanup;
@@ -1143,6 +1133,23 @@ sqlEndTable(struct Parse *pParse)
 	struct Vdbe *v = sqlGetVdbe(pParse);
 	if (NEVER(v == 0))
 		return;
+	const char *space_name =
+		sql_name_from_token(db, &pParse->create_table_def.base.name);
+	char *name_copy = sqlDbStrDup(db, space_name);
+	if (name_copy == NULL)
+		goto cleanup;
+	int name_reg = ++pParse->nMem;
+	sqlVdbeAddOp4(pParse->pVdbe, OP_String8, 0, name_reg, 0, name_copy,
+		      P4_DYNAMIC);
+	const char *error_msg =
+		tt_sprintf(tnt_errcode_desc(ER_SPACE_EXISTS), space_name);
+	bool no_err = pParse->create_table_def.base.if_not_exist;
+	if (vdbe_emit_halt_with_presence_test(pParse, BOX_SPACE_ID, 2,
+					      name_reg, 1, ER_SPACE_EXISTS,
+					      error_msg, (no_err != 0),
+					      OP_NoConflict) != 0)
+		goto cleanup;
+
 	int reg_space_id = getNewSpaceId(pParse);
 	vdbe_emit_space_create(pParse, reg_space_id, new_space);
 	for (uint32_t i = 0; i < new_space->index_count; ++i) {
@@ -1234,8 +1241,7 @@ sql_create_view(struct Parse *parse_context)
 		goto create_view_fail;
 	}
 	struct space *space = sqlStartTable(parse_context,
-					    &create_entity_def->name,
-					    create_entity_def->if_not_exist);
+					    &create_entity_def->name);
 	if (space == NULL || parse_context->is_aborted)
 		goto create_view_fail;
 	struct space *select_res_space =
@@ -1285,6 +1291,23 @@ sql_create_view(struct Parse *parse_context)
 		parse_context->is_aborted = true;
 		goto create_view_fail;
 	}
+	const char *space_name =
+		sql_name_from_token(db, &create_entity_def->name);
+	char *name_copy = sqlDbStrDup(db, space_name);
+	if (name_copy == NULL)
+		goto create_view_fail;
+	int name_reg = ++parse_context->nMem;
+	sqlVdbeAddOp4(parse_context->pVdbe, OP_String8, 0, name_reg, 0, name_copy,
+		      P4_DYNAMIC);
+	const char *error_msg =
+		tt_sprintf(tnt_errcode_desc(ER_SPACE_EXISTS), space_name);
+	bool no_err = create_entity_def->if_not_exist;
+	if (vdbe_emit_halt_with_presence_test(parse_context, BOX_SPACE_ID, 2,
+					      name_reg, 1, ER_SPACE_EXISTS,
+					      error_msg, (no_err != 0),
+					      OP_NoConflict) != 0)
+		goto create_view_fail;
+
 	vdbe_emit_space_create(parse_context, getNewSpaceId(parse_context),
 			       space);
 
diff --git a/src/box/sql/parse.y b/src/box/sql/parse.y
index 3a443a0..d5d18c6 100644
--- a/src/box/sql/parse.y
+++ b/src/box/sql/parse.y
@@ -173,7 +173,7 @@ cmd ::= ROLLBACK TO savepoint_opt nm(X). {
 cmd ::= create_table create_table_args.
 create_table ::= createkw TABLE ifnotexists(E) nm(Y). {
   create_table_def_init(&pParse->create_table_def, &Y, E);
-  pParse->create_table_def.new_space = sqlStartTable(pParse, &Y, E);
+  pParse->create_table_def.new_space = sqlStartTable(pParse, &Y);
 }
 createkw(A) ::= CREATE(A).  {disableLookaside(pParse);}
 
diff --git a/src/box/sql/sqlInt.h b/src/box/sql/sqlInt.h
index 997a465..7a3d66d 100644
--- a/src/box/sql/sqlInt.h
+++ b/src/box/sql/sqlInt.h
@@ -3320,7 +3320,7 @@ sqlSelectAddColumnTypeAndCollation(Parse *, struct space_def *, Select *);
 struct space *sqlResultSetOfSelect(Parse *, Select *);
 
 struct space *
-sqlStartTable(Parse *, Token *, int);
+sqlStartTable(Parse *, Token *);
 void sqlAddColumn(Parse *, Token *, struct type_def *);
 
 /**
diff --git a/test/sql-tap/table.test.lua b/test/sql-tap/table.test.lua
index 5b793c0..7881cff 100755
--- a/test/sql-tap/table.test.lua
+++ b/test/sql-tap/table.test.lua
@@ -1,6 +1,6 @@
 #!/usr/bin/env tarantool
 test = require("sqltester")
-test:plan(77)
+test:plan(79)
 
 --!./tcltestrunner.lua
 -- 2001 September 15
@@ -128,7 +128,7 @@ test:do_test(
     "table-2.1",
     function()
         test:execsql "CREATE TABLE TEST2(one text primary key)"
-        return test:catchsql "CREATE TABLE test2(id primary key, two text default 'hi')"
+        return test:catchsql "CREATE TABLE test2(id int primary key, two text default 'hi')"
     end, {
         -- <table-2.1>
         1, "Space 'TEST2' already exists"
@@ -1442,4 +1442,30 @@ test:do_execsql_test(
 
         -- </check-23.cleanup>
     })
+
+--
+-- gh-4196: Error creating table twice.
+--
+test:do_execsql_test(
+	"table-24.1",
+	[[
+		CREATE TABLE IF NOT EXISTS a1 (i INT PRIMARY KEY);
+		CREATE TABLE IF NOT EXISTS b1 (i INT PRIMARY KEY, j INT, CONSTRAINT aa FOREIGN KEY(j) REFERENCES a1(i));
+		CREATE TABLE IF NOT EXISTS b1 (i INT PRIMARY KEY, j INT, CONSTRAINT aa FOREIGN KEY(j) REFERENCES a1(i));
+	]], {
+		-- <table-24.1>
+		-- </table-24.1>
+	})
+
+test:do_execsql_test(
+	"table-24.2",
+	[[
+		CREATE TABLE IF NOT EXISTS a2 (i INT PRIMARY KEY);
+		CREATE TABLE IF NOT EXISTS b2 (i INT PRIMARY KEY, j INT REFERENCES a2(i));
+		CREATE TABLE IF NOT EXISTS b2 (i INT PRIMARY KEY, j INT REFERENCES a2(i));
+	]], {
+		-- <table-24.2>
+		-- </table-24.2>
+	})
+
 test:finish_test()
-- 
2.7.4