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

[Mergen Imeev <imeevma@tarantool.org>]

Hi! Thank you for review! My answers and new patch below.
I have included the full patch, as there will be quite a
few fixes.

On Sat, May 11, 2019 at 02:56:21PM +0300, n.pettik wrote:
> 
> 
> > On 8 May 2019, at 20:02, imeevma@tarantool.org wrote:
> > 
> > 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.
> 
> -> Meanwhile, create_foreign_key() assumes that if
> create_table_def->new_space is NULL, then we are
> dealing with ALTER TABLE statement. This in turn false,
> since ctd->new_space is nullified also in case of already
> existing table.
> 
Thanks, fixed.

> > This causes an
> > error when creating a foreign key during space creation.
> 
> Not error, but segmentation fault.
> 
Fixed. Also added thфе it could cause an assertion.

> > This patch moves this check to VDBE. Thus, the parsing should now
> > be properly closed before performing this check.
> 
> -> parsing now is always processed till the end as in case space
> doesn’t exist.
> 
Added.

> > 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
> > 
> > @@ -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);
> 
> Why not using new_space->def->name ?
> 
Fixed.

> > +	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);
> 
> In vdbe_emit_space_create() name of space is duplicated as well,
> so to avoid processing copy twice we can pass register holding the
> name of space to vdbe_emut_space_create().
> 
> Consider this diff:
> 
> diff --git a/src/box/sql/build.c b/src/box/sql/build.c
> index a25479edc..61f4f4127 100644
> --- a/src/box/sql/build.c
> +++ b/src/box/sql/build.c
> @@ -833,7 +833,7 @@ error:
>   */
>  static void
>  vdbe_emit_space_create(struct Parse *pParse, int space_id_reg,
> -                      struct space *space)
> +                      int space_name_reg, struct space *space)
>  {
>         Vdbe *v = sqlGetVdbe(pParse);
>         int iFirstCol = ++pParse->nMem;
> @@ -862,9 +862,7 @@ vdbe_emit_space_create(struct Parse *pParse, int space_id_reg,
>         sqlVdbeAddOp2(v, OP_SCopy, space_id_reg, iFirstCol /* spaceId */ );
>         sqlVdbeAddOp2(v, OP_Integer, effective_user()->uid,
>                           iFirstCol + 1 /* owner */ );
> -       sqlVdbeAddOp4(v, OP_String8, 0, iFirstCol + 2 /* name */ , 0,
> -                         sqlDbStrDup(pParse->db, space->def->name),
> -                         P4_DYNAMIC);
> +       sqlVdbeAddOp2(v, OP_SCopy, space_name_reg, iFirstCol + 2);
>         sqlVdbeAddOp4(v, OP_String8, 0, iFirstCol + 3 /* engine */ , 0,
>                           sqlDbStrDup(pParse->db, space->def->engine_name),
>                           P4_DYNAMIC);
> @@ -1133,16 +1131,20 @@ 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)
> +       /*
> +        * Firstly, check if space with given name already exists.
> +        * In case IF NOT EXISTS clause is specified and table
> +        * exists, we will silently halt VDBE execution.
> +        */
> +       char *space_name_copy =
> +               sqlDbStrDup(db, new_space->def->name);
> +       if (space_name_copy == NULL)
>                 goto cleanup;
>         int name_reg = ++pParse->nMem;
> -       sqlVdbeAddOp4(pParse->pVdbe, OP_String8, 0, name_reg, 0, name_copy,
> +       sqlVdbeAddOp4(pParse->pVdbe, OP_String8, 0, name_reg, 0, space_name_copy,
>                       P4_DYNAMIC);
>         const char *error_msg =
> -               tt_sprintf(tnt_errcode_desc(ER_SPACE_EXISTS), space_name);
> +               tt_sprintf(tnt_errcode_desc(ER_SPACE_EXISTS), space_name_copy);
>         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,
> @@ -1151,7 +1153,7 @@ sqlEndTable(struct Parse *pParse)
>                 goto cleanup;
>  
>         int reg_space_id = getNewSpaceId(pParse);
> -       vdbe_emit_space_create(pParse, reg_space_id, new_space);
> +       vdbe_emit_space_create(pParse, reg_space_id, name_reg, new_space);
>         for (uint32_t i = 0; i < new_space->index_count; ++i) {
>                 struct index *idx = new_space->index[i];
>                 vdbe_emit_create_index(pParse, new_space->def, idx->def,
> @@ -1309,7 +1311,7 @@ sql_create_view(struct Parse *parse_context)
>                 goto create_view_fail;
>  
>         vdbe_emit_space_create(parse_context, getNewSpaceId(parse_context),
> -                              space);
> +                              name_reg, space);
> 
Thanks, fixed.

> > 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
> > 
> > @@ -1442,4 +1442,30 @@ test:do_execsql_test(
> > 
> >         -- </check-23.cleanup>
> >     })
> > +
> > +--
> > +-- gh-4196: Error creating table twice.
> 
> -> Segmentation fault happened when IF NOT EXISTS
> clause was specified during creation of space featuring 
> FK constraints.
> 
Fixed.

> > +--
> > +test:do_execsql_test(
> 
> I’d make this test be catchsql to indicate that it should
> test that no error is raised.
> 
Fixed.

> > +	"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>
> > +	})
> > +
>


New patch:

From 46729f50aa4a76330201dbaf77d4d17798032d83 Mon Sep 17 00:00:00 2001
Date: Wed, 8 May 2019 18:07:15 +0300
Subject: [PATCH] sql: move space existence check to VDBE
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

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. Meanwhile,
create_foreign_key() assumes that if create_table_def->new_space
is NULL, then we are dealing with ALTER TABLE statement. This in
turn false, since ctd->new_space is nullified also in case of
already existing table. This causes an assertion or a segmentation
fault when creating a foreign key during space creation.

This patch moves this check to VDBE. Parsing now is always
processed till the end as in case space doesn’t exist.

Closes #4196

diff --git a/src/box/sql/build.c b/src/box/sql/build.c
index 6051a25..343dcb1 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;
@@ -843,7 +833,7 @@ error:
  */
 static void
 vdbe_emit_space_create(struct Parse *pParse, int space_id_reg,
-		       struct space *space)
+		       int space_name_reg, struct space *space)
 {
 	Vdbe *v = sqlGetVdbe(pParse);
 	int iFirstCol = ++pParse->nMem;
@@ -872,9 +862,7 @@ vdbe_emit_space_create(struct Parse *pParse, int space_id_reg,
 	sqlVdbeAddOp2(v, OP_SCopy, space_id_reg, iFirstCol /* spaceId */ );
 	sqlVdbeAddOp2(v, OP_Integer, effective_user()->uid,
 			  iFirstCol + 1 /* owner */ );
-	sqlVdbeAddOp4(v, OP_String8, 0, iFirstCol + 2 /* name */ , 0,
-			  sqlDbStrDup(pParse->db, space->def->name),
-			  P4_DYNAMIC);
+	sqlVdbeAddOp2(v, OP_SCopy, space_name_reg, iFirstCol + 2);
 	sqlVdbeAddOp4(v, OP_String8, 0, iFirstCol + 3 /* engine */ , 0,
 			  sqlDbStrDup(pParse->db, space->def->engine_name),
 			  P4_DYNAMIC);
@@ -1143,8 +1131,29 @@ sqlEndTable(struct Parse *pParse)
 	struct Vdbe *v = sqlGetVdbe(pParse);
 	if (NEVER(v == 0))
 		return;
+
+	/*
+	 * Firstly, check if space with given name already exists.
+	 * In case IF NOT EXISTS clause is specified and table
+	 * exists, we will silently halt VDBE execution.
+	 */
+	char *space_name_copy = sqlDbStrDup(db, new_space->def->name);
+	if (space_name_copy == NULL)
+		goto cleanup;
+	int name_reg = ++pParse->nMem;
+	sqlVdbeAddOp4(pParse->pVdbe, OP_String8, 0, name_reg, 0,
+		      space_name_copy, P4_DYNAMIC);
+	const char *error_msg =
+		tt_sprintf(tnt_errcode_desc(ER_SPACE_EXISTS), space_name_copy);
+	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);
+	vdbe_emit_space_create(pParse, reg_space_id, name_reg, new_space);
 	for (uint32_t i = 0; i < new_space->index_count; ++i) {
 		struct index *idx = new_space->index[i];
 		vdbe_emit_create_index(pParse, new_space->def, idx->def,
@@ -1234,8 +1243,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,8 +1293,25 @@ 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);
+			       name_reg, space);
 
  create_view_fail:
 	sql_expr_list_delete(db, view_def->aliases);
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..b59bc23 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,34 @@ test:do_execsql_test(
 
         -- </check-23.cleanup>
     })
+
+--
+-- gh-4196: Segmentation fault or assertion happened when IF NOT
+-- EXISTS clause was specified during creation of space featuring
+-- FK constraints.
+--
+test:do_catchsql_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>
+		0
+		-- </table-24.1>
+	})
+
+test:do_catchsql_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>
+		0
+		-- </table-24.2>
+	})
+
 test:finish_test()