Re: [Tarantool-patches] [PATCH v4 2/6] sql: refactor create_table_def and parse

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

Hi!

> On Sep 16, 2020, at 16:27, Nikita Pettik <korablev@tarantool.org> wrote:
> 
> On 12 Sep 00:51, Roman Khabibov wrote:
>> Move ck, fk constraint lists and autoincrement info from
>> struct create_table_def to struct Parse to make the code more
>> reusable when implementing <ALTER TABLE ADD COLUMN>.
>> 
>> Needed for #3075
>> ---
> 
> Does ANSI allow to include constraints in ADD COLUMN statement?
> Why did you decide to add autoincrement to this list? Was there any
> discussion on this subj?
The standard says that the column descriptions in <CREATE> and <ALTER>
are exactly the same. All consrtants, <DEFAULT> and <COLLATE> are allowed.
But <AUTOINCREMENT> is not described in the standard. I suggested (Vlad
didn't mind) to continue this logic and support <AUTOINCREMENT> inside
<ALTER>, because there are no significant restrictions on this. We can
add a sequence to an already existing table.

>> src/box/sql/build.c     | 182 ++++++++++++++++++++++------------------
>> src/box/sql/parse.y     |   2 +
>> src/box/sql/parse_def.h |  53 ++++++------
>> src/box/sql/prepare.c   |   3 +-
>> src/box/sql/sqlInt.h    |  12 +++
>> 5 files changed, 142 insertions(+), 110 deletions(-)
>> 

commit 19e3d75b0c2702fd931244cc4c0a032a5cc67c58
Author: Roman Khabibov <roman.habibov@tarantool.org>
Date:   Sat Aug 1 00:27:52 2020 +0300

    sql: refactor create_table_def and parse
    
    Move ck, fk constraint lists and autoincrement info from
    struct create_table_def to struct Parse to make the code more
    reusable when implementing <ALTER TABLE ADD COLUMN>.
    
    Needed for #3075

diff --git a/src/box/sql/build.c b/src/box/sql/build.c
index d55c1cd71..6cc76bb77 100644
--- a/src/box/sql/build.c
+++ b/src/box/sql/build.c
@@ -590,7 +590,7 @@ sql_create_check_contraint(struct Parse *parser)
 		}
 	} else {
 		assert(! is_alter);
-		uint32_t ck_idx = ++parser->create_table_def.check_count;
+		uint32_t ck_idx = ++parser->checks_def.count;
 		name = tt_sprintf("ck_unnamed_%s_%d", space->def->name, ck_idx);
 	}
 	size_t name_len = strlen(name);
@@ -652,8 +652,7 @@ sql_create_check_contraint(struct Parse *parser)
 		sqlVdbeCountChanges(v);
 		sqlVdbeChangeP5(v, OPFLAG_NCHANGE);
 	} else {
-		rlist_add_entry(&parser->create_table_def.new_check, ck_parse,
-				link);
+		rlist_add_entry(&parser->checks_def.checks, ck_parse, link);
 	}
 }
 
@@ -930,7 +929,7 @@ emitNewSysSequenceRecord(Parse *pParse, int reg_seq_id, const char *seq_name)
 static int
 emitNewSysSpaceSequenceRecord(Parse *pParse, int reg_space_id, int reg_seq_id)
 {
-	uint32_t fieldno = pParse->create_table_def.autoinc_fieldno;
+	uint32_t fieldno = pParse->autoinc_fieldno;
 
 	Vdbe *v = sqlGetVdbe(pParse);
 	int first_col = pParse->nMem + 1;
@@ -1147,6 +1146,88 @@ resolve_link(struct Parse *parse_context, const struct space_def *def,
 	return -1;
 }
 
+/**
+ * Emit code to create sequences, indexes, check and foreign key
+ * constraints appeared in <CREATE TABLE>.
+ */
+static void
+sql_vdbe_create_constraints(struct Parse *parse, int reg_space_id)
+{
+	assert(reg_space_id != 0);
+	struct space *space = parse->create_table_def.new_space;
+	assert(space != NULL);
+	uint32_t i = 0;
+	for (; i < space->index_count; ++i) {
+		struct index *idx = space->index[i];
+		vdbe_emit_create_index(parse, space->def, idx->def,
+				       reg_space_id, idx->def->iid);
+	}
+
+	/*
+	 * Check to see if we need to create an _sequence table
+	 * for keeping track of autoincrement keys.
+	 */
+	if (parse->has_autoinc) {
+		/* Do an insertion into _sequence. */
+		int reg_seq_id = ++parse->nMem;
+		struct Vdbe *v = sqlGetVdbe(parse);
+		assert(v != NULL);
+		sqlVdbeAddOp2(v, OP_NextSequenceId, 0, reg_seq_id);
+		int reg_seq_rec = emitNewSysSequenceRecord(parse, reg_seq_id,
+							   space->def->name);
+		sqlVdbeAddOp2(v, OP_SInsert, BOX_SEQUENCE_ID, reg_seq_rec);
+		/* Do an insertion into _space_sequence. */
+		int reg_space_seq_record =
+			emitNewSysSpaceSequenceRecord(parse, reg_space_id,
+						      reg_seq_id);
+		sqlVdbeAddOp2(v, OP_SInsert, BOX_SPACE_SEQUENCE_ID,
+			      reg_space_seq_record);
+	}
+
+	/* Code creation of FK constraints, if any. */
+	struct fk_constraint_parse *fk_parse;
+	rlist_foreach_entry(fk_parse, &parse->fkeys_def.fkeys, link) {
+		struct fk_constraint_def *fk_def = fk_parse->fk_def;
+		if (fk_parse->selfref_cols != NULL) {
+			struct ExprList *cols = fk_parse->selfref_cols;
+			for (uint32_t i = 0; i < fk_def->field_count; ++i) {
+				if (resolve_link(parse, space->def,
+						 cols->a[i].zName,
+						 &fk_def->links[i].parent_field,
+						 fk_def->name) != 0)
+					return;
+			}
+			fk_def->parent_id = reg_space_id;
+		} else if (fk_parse->is_self_referenced) {
+			struct key_def *pk_key_def =
+				sql_space_primary_key(space)->def->key_def;
+			if (pk_key_def->part_count != fk_def->field_count) {
+				diag_set(ClientError, ER_CREATE_FK_CONSTRAINT,
+					 fk_def->name, "number of columns in "\
+					 "foreign key does not match the "\
+					 "number of columns in the primary "\
+					 "index of referenced table");
+				parse->is_aborted = true;
+				return;
+			}
+			for (uint32_t i = 0; i < fk_def->field_count; ++i) {
+				fk_def->links[i].parent_field =
+					pk_key_def->parts[i].fieldno;
+			}
+			fk_def->parent_id = reg_space_id;
+		}
+		fk_def->child_id = reg_space_id;
+		vdbe_emit_fk_constraint_create(parse, fk_def, space->def->name);
+	}
+
+	/* Code creation of CK constraints, if any. */
+	struct ck_constraint_parse *ck_parse;
+	rlist_foreach_entry(ck_parse, &parse->checks_def.checks, link) {
+		vdbe_emit_ck_constraint_create(parse, ck_parse->ck_def,
+					       reg_space_id, space->def->name);
+	}
+}
+
 /*
  * This routine is called to report the final ")" that terminates
  * a CREATE TABLE statement.
@@ -1213,73 +1294,7 @@ sqlEndTable(struct Parse *pParse)
 
 	int reg_space_id = getNewSpaceId(pParse);
 	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,
-				       reg_space_id, idx->def->iid);
-	}
-
-	/*
-	 * Check to see if we need to create an _sequence table
-	 * for keeping track of autoincrement keys.
-	 */
-	if (pParse->create_table_def.has_autoinc) {
-		assert(reg_space_id != 0);
-		/* Do an insertion into _sequence. */
-		int reg_seq_id = ++pParse->nMem;
-		sqlVdbeAddOp2(v, OP_NextSequenceId, 0, reg_seq_id);
-		int reg_seq_record =
-			emitNewSysSequenceRecord(pParse, reg_seq_id,
-						 new_space->def->name);
-		sqlVdbeAddOp2(v, OP_SInsert, BOX_SEQUENCE_ID, reg_seq_record);
-		/* Do an insertion into _space_sequence. */
-		int reg_space_seq_record =
-			emitNewSysSpaceSequenceRecord(pParse, reg_space_id,
-						      reg_seq_id);
-		sqlVdbeAddOp2(v, OP_SInsert, BOX_SPACE_SEQUENCE_ID,
-			      reg_space_seq_record);
-	}
-	/* Code creation of FK constraints, if any. */
-	struct fk_constraint_parse *fk_parse;
-	rlist_foreach_entry(fk_parse, &pParse->create_table_def.new_fkey,
-			    link) {
-		struct fk_constraint_def *fk_def = fk_parse->fk_def;
-		if (fk_parse->selfref_cols != NULL) {
-			struct ExprList *cols = fk_parse->selfref_cols;
-			for (uint32_t i = 0; i < fk_def->field_count; ++i) {
-				if (resolve_link(pParse, new_space->def,
-						 cols->a[i].zName,
-						 &fk_def->links[i].parent_field,
-						 fk_def->name) != 0)
-					return;
-			}
-			fk_def->parent_id = reg_space_id;
-		} else if (fk_parse->is_self_referenced) {
-			struct index *pk = sql_space_primary_key(new_space);
-			if (pk->def->key_def->part_count != fk_def->field_count) {
-				diag_set(ClientError, ER_CREATE_FK_CONSTRAINT,
-					 fk_def->name, "number of columns in "\
-					 "foreign key does not match the "\
-					 "number of columns in the primary "\
-					 "index of referenced table");
-				pParse->is_aborted = true;
-				return;
-			}
-			for (uint32_t i = 0; i < fk_def->field_count; ++i) {
-				fk_def->links[i].parent_field =
-					pk->def->key_def->parts[i].fieldno;
-			}
-			fk_def->parent_id = reg_space_id;
-		}
-		fk_def->child_id = reg_space_id;
-		vdbe_emit_fk_constraint_create(pParse, fk_def, space_name_copy);
-	}
-	struct ck_constraint_parse *ck_parse;
-	rlist_foreach_entry(ck_parse, &pParse->create_table_def.new_check,
-			    link) {
-		vdbe_emit_ck_constraint_create(pParse, ck_parse->ck_def,
-					       reg_space_id, space_name_copy);
-	}
+	sql_vdbe_create_constraints(pParse, reg_space_id);
 }
 
 void
@@ -1893,7 +1908,8 @@ sql_create_foreign_key(struct Parse *parse_context)
 			goto tnt_error;
 		}
 		memset(fk_parse, 0, sizeof(*fk_parse));
-		rlist_add_entry(&table_def->new_fkey, fk_parse, link);
+		rlist_add_entry(&parse_context->fkeys_def.fkeys, fk_parse,
+				link);
 	}
 	struct Token *parent = create_fk_def->parent_name;
 	assert(parent != NULL);
@@ -1910,8 +1926,9 @@ sql_create_foreign_key(struct Parse *parse_context)
 	struct space *parent_space = space_by_name(parent_name);
 	if (parent_space == NULL) {
 		if (is_self_referenced) {
+			struct rlist *fkeys = &parse_context->fkeys_def.fkeys;
 			struct fk_constraint_parse *fk =
-				rlist_first_entry(&table_def->new_fkey,
+				rlist_first_entry(fkeys,
 						  struct fk_constraint_parse,
 						  link);
 			fk->selfref_cols = parent_cols;
@@ -1926,7 +1943,7 @@ sql_create_foreign_key(struct Parse *parse_context)
 			constraint_name =
 				sqlMPrintf(db, "fk_unnamed_%s_%d",
 					   space->def->name,
-					   ++table_def->fkey_count);
+					   ++parse_context->fkeys_def.count);
 		} else {
 			constraint_name =
 				sql_name_from_token(db, &create_def->name);
@@ -2042,7 +2059,7 @@ sql_create_foreign_key(struct Parse *parse_context)
 	 */
 	if (!is_alter) {
 		struct fk_constraint_parse *fk_parse =
-			rlist_first_entry(&table_def->new_fkey,
+			rlist_first_entry(&parse_context->fkeys_def.fkeys,
 					  struct fk_constraint_parse, link);
 		fk_parse->fk_def = fk_def;
 	} else {
@@ -2065,12 +2082,11 @@ tnt_error:
 void
 fk_constraint_change_defer_mode(struct Parse *parse_context, bool is_deferred)
 {
-	if (parse_context->db->init.busy ||
-	    rlist_empty(&parse_context->create_table_def.new_fkey))
+	struct rlist *fkeys = &parse_context->fkeys_def.fkeys;
+	if (parse_context->db->init.busy || rlist_empty(fkeys))
 		return;
-	rlist_first_entry(&parse_context->create_table_def.new_fkey,
-			  struct fk_constraint_parse, link)->fk_def->is_deferred =
-		is_deferred;
+	rlist_first_entry(fkeys, struct fk_constraint_parse,
+			  link)->fk_def->is_deferred = is_deferred;
 }
 
 /**
@@ -3306,15 +3322,15 @@ vdbe_emit_halt_with_presence_test(struct Parse *parser, int space_id,
 int
 sql_add_autoincrement(struct Parse *parse_context, uint32_t fieldno)
 {
-	if (parse_context->create_table_def.has_autoinc) {
+	if (parse_context->has_autoinc) {
 		diag_set(ClientError, ER_SQL_SYNTAX_WITH_POS,
 			 parse_context->line_count, parse_context->line_pos,
 			 "table must feature at most one AUTOINCREMENT field");
 		parse_context->is_aborted = true;
 		return -1;
 	}
-	parse_context->create_table_def.has_autoinc = true;
-	parse_context->create_table_def.autoinc_fieldno = fieldno;
+	parse_context->has_autoinc = true;
+	parse_context->autoinc_fieldno = fieldno;
 	return 0;
 }
 
diff --git a/src/box/sql/parse.y b/src/box/sql/parse.y
index 995875566..d81045d50 100644
--- a/src/box/sql/parse.y
+++ b/src/box/sql/parse.y
@@ -181,6 +181,8 @@ cmd ::= ROLLBACK TO savepoint_opt nm(X). {
 cmd ::= create_table create_table_args with_opts create_table_end.
 create_table ::= createkw TABLE ifnotexists(E) nm(Y). {
   create_table_def_init(&pParse->create_table_def, &Y, E);
+  create_checks_def_init(&pParse->checks_def);
+  create_fkeys_def_init(&pParse->fkeys_def);
   pParse->create_table_def.new_space = sqlStartTable(pParse, &Y);
   pParse->initiateTTrans = true;
 }
diff --git a/src/box/sql/parse_def.h b/src/box/sql/parse_def.h
index cb0ecd2fc..c44e3dbbe 100644
--- a/src/box/sql/parse_def.h
+++ b/src/box/sql/parse_def.h
@@ -205,26 +205,16 @@ struct create_entity_def {
 struct create_table_def {
 	struct create_entity_def base;
 	struct space *new_space;
-	/**
-	 * Number of FK constraints declared within
-	 * CREATE TABLE statement.
-	 */
-	uint32_t fkey_count;
-	/**
-	 * Foreign key constraint appeared in CREATE TABLE stmt.
-	 */
-	struct rlist new_fkey;
-	/**
-	 * Number of CK constraints declared within
-	 * CREATE TABLE statement.
-	 */
-	uint32_t check_count;
-	/** Check constraint appeared in CREATE TABLE stmt. */
-	struct rlist new_check;
-	/** True, if table to be created has AUTOINCREMENT PK. */
-	bool has_autoinc;
-	/** Id of field with AUTOINCREMENT. */
-	uint32_t autoinc_fieldno;
+};
+
+struct create_checks_def {
+	struct rlist checks;
+	uint32_t count;
+};
+
+struct create_fkeys_def {
+	struct rlist fkeys;
+	uint32_t count;
 };
 
 struct create_view_def {
@@ -482,9 +472,20 @@ create_table_def_init(struct create_table_def *table_def, struct Token *name,
 {
 	create_entity_def_init(&table_def->base, ENTITY_TYPE_TABLE, NULL, name,
 			       if_not_exists);
-	rlist_create(&table_def->new_fkey);
-	rlist_create(&table_def->new_check);
-	table_def->autoinc_fieldno = 0;
+}
+
+static inline void
+create_checks_def_init(struct create_checks_def *checks_def)
+{
+	rlist_create(&checks_def->checks);
+	checks_def->count = 0;
+}
+
+static inline void
+create_fkeys_def_init(struct create_fkeys_def *fkeys_def)
+{
+	rlist_create(&fkeys_def->fkeys);
+	fkeys_def->count = 0;
 }
 
 static inline void
@@ -500,12 +501,12 @@ create_view_def_init(struct create_view_def *view_def, struct Token *name,
 }
 
 static inline void
-create_table_def_destroy(struct create_table_def *table_def)
+fkeys_def_destroy(struct create_fkeys_def *fkeys_def)
 {
-	if (table_def->new_space == NULL)
+	if (fkeys_def->count == 0)
 		return;
 	struct fk_constraint_parse *fk;
-	rlist_foreach_entry(fk, &table_def->new_fkey, link)
+	rlist_foreach_entry(fk, &fkeys_def->fkeys, link)
 		sql_expr_list_delete(sql_get(), fk->selfref_cols);
 }
 
diff --git a/src/box/sql/prepare.c b/src/box/sql/prepare.c
index a5a258805..2eab12a0b 100644
--- a/src/box/sql/prepare.c
+++ b/src/box/sql/prepare.c
@@ -200,6 +200,7 @@ sql_parser_create(struct Parse *parser, struct sql *db, uint32_t sql_flags)
 	parser->sql_flags = sql_flags;
 	parser->line_count = 1;
 	parser->line_pos = 1;
+	parser->has_autoinc = false;
 	region_create(&parser->region, &cord()->slabc);
 }
 
@@ -211,7 +212,7 @@ sql_parser_destroy(Parse *parser)
 	sql *db = parser->db;
 	sqlDbFree(db, parser->aLabel);
 	sql_expr_list_delete(db, parser->pConstExpr);
-	create_table_def_destroy(&parser->create_table_def);
+	fkeys_def_destroy(&parser->fkeys_def);
 	if (db != NULL) {
 		assert(db->lookaside.bDisable >=
 		       parser->disableLookaside);
diff --git a/src/box/sql/sqlInt.h b/src/box/sql/sqlInt.h
index 5913d7614..7057c20b6 100644
--- a/src/box/sql/sqlInt.h
+++ b/src/box/sql/sqlInt.h
@@ -2257,6 +2257,18 @@ struct Parse {
 	 * sqlEndTable() function).
 	 */
 	struct create_table_def create_table_def;
+	/*
+	 * FK and CK constraints appeared in a <CREATE TABLE>.
+	 */
+	struct create_fkeys_def fkeys_def;
+	struct create_checks_def checks_def;
+	/*
+	 * True, if column within a <CREATE TABLE> statement to be
+	 * created has <AUTOINCREMENT>.
+	 */
+	bool has_autoinc;
+	/* Id of field with <AUTOINCREMENT>. */
+	uint32_t autoinc_fieldno;
 	bool initiateTTrans;	/* Initiate Tarantool transaction */
 	/** If set - do not emit byte code at all, just parse.  */
 	bool parse_only;