[tarantool-patches] [PATCH v1 4/4] sql: get rid of Column structure

[Kirill Shcherbatov <kshcherbatov@tarantool.org>]

Get rid of is_primkey in Column structure as it become
redundant. Moved the last member coll with collation pointer
to field_def structure. Finally, dropped Column.
---
 src/box/field_def.c     |  1 +
 src/box/field_def.h     |  2 ++
 src/box/sql/alter.c     | 16 +++-------
 src/box/sql/build.c     | 79 +++++++++++++++++--------------------------------
 src/box/sql/resolve.c   | 11 +++----
 src/box/sql/select.c    | 43 ++++++++++-----------------
 src/box/sql/sqliteInt.h | 23 +++++++-------
 7 files changed, 65 insertions(+), 110 deletions(-)

diff --git a/src/box/field_def.c b/src/box/field_def.c
index 8dbead6..12cc3b2 100644
--- a/src/box/field_def.c
+++ b/src/box/field_def.c
@@ -106,6 +106,7 @@ const struct field_def field_def_default = {
 	.is_nullable = false,
 	.nullable_action = ON_CONFLICT_ACTION_DEFAULT,
 	.coll_id = COLL_NONE,
+	.coll = NULL,
 	.default_value = NULL,
 	.default_value_expr = NULL
 };
diff --git a/src/box/field_def.h b/src/box/field_def.h
index 05f80d4..c8f158c 100644
--- a/src/box/field_def.h
+++ b/src/box/field_def.h
@@ -123,6 +123,8 @@ struct field_def {
 	enum on_conflict_action nullable_action;
 	/** Collation ID for string comparison. */
 	uint32_t coll_id;
+	/** Collating sequence for string comparison. */
+	struct coll *coll;
 	/** 0-terminated SQL expression for DEFAULT value. */
 	char *default_value;
 	/** AST for parsed default value. */
diff --git a/src/box/sql/alter.c b/src/box/sql/alter.c
index fe54e55..4f50988 100644
--- a/src/box/sql/alter.c
+++ b/src/box/sql/alter.c
@@ -146,7 +146,6 @@ sqlite3AlterFinishAddColumn(Parse * pParse, Token * pColDef)
 	Table *pNew;		/* Copy of pParse->pNewTable */
 	Table *pTab;		/* Table being altered */
 	const char *zTab;	/* Table name */
-	Column *pCol;		/* The new column */
 	Expr *pDflt;		/* Default value for the new column */
 	sqlite3 *db;		/* The database connection; */
 	Vdbe *v = pParse->pVdbe;	/* The prepared statement under construction */
@@ -161,7 +160,6 @@ sqlite3AlterFinishAddColumn(Parse * pParse, Token * pColDef)
 
 	/* Skip the "sqlite_altertab_" prefix on the name. */
 	zTab = &pNew->def->name[16];
-	pCol = &pNew->aCol[pNew->def->field_count - 1];
 	assert(pNew->def != NULL);
 	pDflt = space_column_default_expr(SQLITE_PAGENO_TO_SPACEID(pNew->tnum),
 					  pNew->def->field_count - 1);
@@ -181,7 +179,9 @@ sqlite3AlterFinishAddColumn(Parse * pParse, Token * pColDef)
 	 * If there is a NOT NULL constraint, then the default value for the
 	 * column must not be NULL.
 	 */
-	if (pCol->is_primkey) {
+	struct Index *pk = sqlite3PrimaryKeyIndex(pTab);
+	if (index_has_column(pk->aiColumn, pk->nColumn,
+			     pNew->def->field_count - 1)) {
 		sqlite3ErrorMsg(pParse, "Cannot add a PRIMARY KEY column");
 		return;
 	}
@@ -299,19 +299,11 @@ sqlite3AlterBeginAddColumn(Parse * pParse, SrcList * pSrc)
 	nAlloc = (((pNew->def->field_count - 1) / 8) * 8) + 8;
 	assert((uint32_t)nAlloc >= pNew->def->field_count && nAlloc % 8 == 0 &&
 	       nAlloc - pNew->def->field_count < 8);
-	pNew->aCol =
-	    (Column *) sqlite3DbMallocZero(db, sizeof(Column) * nAlloc);
 	/* FIXME: pNew->zName = sqlite3MPrintf(db, "sqlite_altertab_%s", pTab->zName); */
 	/* FIXME: if (!pNew->aCol || !pNew->zName) { */
-	if (!pNew->aCol) {
-		assert(db->mallocFailed);
-		goto exit_begin_add_column;
-	}
-	memcpy(pNew->aCol, pTab->aCol, sizeof(Column) * pNew->def->field_count);
 	for (i = 0; i < pNew->def->field_count; i++) {
-		Column *pCol = &pNew->aCol[i];
 		/* FIXME: pNew->def->name = sqlite3DbStrDup(db, pCol->zName); */
-		pCol->coll = NULL;
+		pNew->def->fields[i].coll = NULL;
 		pNew->def->fields[i].coll_id = COLL_NONE;
 	}
 	pNew->pSchema = db->pSchema;
diff --git a/src/box/sql/build.c b/src/box/sql/build.c
index ec0bc4b..a6f3559 100644
--- a/src/box/sql/build.c
+++ b/src/box/sql/build.c
@@ -114,6 +114,16 @@ sql_finish_coding(struct Parse *parse_context)
 	}
 }
 
+bool
+index_has_column(const i16 *parts, int parts_count, int column_idx)
+{
+	while (parts_count-- > 0) {
+		if (column_idx == *(parts++))
+			return 1;
+	}
+	return 0;
+}
+
 /**
  * This is a function which should be called during execution
  * of sqlite3EndTable. It ensures that only PRIMARY KEY
@@ -131,10 +141,10 @@ check_on_conflict_replace_entries(struct Table *table)
 	for (int i = 0; i < (int)table->def->field_count; i++) {
 		enum on_conflict_action on_error =
 			table->def->fields[i].nullable_action;
+		struct Index *pk = sqlite3PrimaryKeyIndex(table);
 		if (on_error == ON_CONFLICT_ACTION_REPLACE &&
-		    table->aCol[i].is_primkey == false) {
+		    !index_has_column(pk->aiColumn, pk->nColumn, i))
 			return true;
-		}
 	}
 	/* Check all UNIQUE constraints. */
 	for (struct Index *idx = table->pIndex; idx; idx = idx->pNext) {
@@ -337,7 +347,6 @@ deleteTable(sqlite3 * db, Table * pTable)
 	/* Delete the Table structure itself.
 	 */
 	sqlite3HashClear(&pTable->idxHash);
-	sqlite3DbFree(db, pTable->aCol);
 	sqlite3DbFree(db, pTable->zColAff);
 	assert(pTable->def != NULL);
 	/* Do not delete pTable->def allocated on region. */
@@ -601,7 +610,6 @@ sqlite3AddColumn(Parse * pParse, Token * pName, Token * pType)
 	int i;
 	char *z;
 	char *zType;
-	Column *pCol;
 	sqlite3 *db = pParse->db;
 	if ((p = pParse->pNewTable) == 0)
 		return;
@@ -639,17 +647,6 @@ sqlite3AddColumn(Parse * pParse, Token * pName, Token * pType)
 			return;
 		}
 	}
-	if ((p->def->field_count & 0x7) == 0) {
-		Column *aNew =
-			sqlite3DbRealloc(db, p->aCol,
-					 (p->def->field_count + 8) *
-					 sizeof(p->aCol[0]));
-		if (aNew == NULL)
-			return;
-		p->aCol = aNew;
-	}
-	pCol = &p->aCol[p->def->field_count];
-	memset(pCol, 0, sizeof(p->aCol[0]));
 	struct field_def *column_def = &p->def->fields[p->def->field_count];
 	memcpy(column_def, &field_def_default, sizeof(field_def_default));
 	column_def->name = z;
@@ -887,7 +884,6 @@ sqlite3AddPrimaryKey(Parse * pParse,	/* Parsing context */
     )
 {
 	Table *pTab = pParse->pNewTable;
-	Column *pCol = 0;
 	int iCol = -1, i;
 	int nTerm;
 	if (pTab == 0)
@@ -901,8 +897,6 @@ sqlite3AddPrimaryKey(Parse * pParse,	/* Parsing context */
 	pTab->tabFlags |= TF_HasPrimaryKey;
 	if (pList == 0) {
 		iCol = pTab->def->field_count - 1;
-		pCol = &pTab->aCol[iCol];
-		pCol->is_primkey = 1;
 		nTerm = 1;
 	} else {
 		nTerm = pList->nExpr;
@@ -915,21 +909,19 @@ sqlite3AddPrimaryKey(Parse * pParse,	/* Parsing context */
 				goto primary_key_exit;
 			}
 			const char *zCName = pCExpr->u.zToken;
-			for (iCol = 0;
-				iCol < (int)pTab->def->field_count; iCol++) {
-				if (strcmp
-				    (zCName,
-				     pTab->def->fields[iCol].name) == 0) {
-					pCol = &pTab->aCol[iCol];
-					pCol->is_primkey = 1;
+			for (uint32_t collumn_id = 0;
+			     collumn_id < pTab->def->field_count; collumn_id++) {
+				if (strcmp(zCName,
+					   pTab->def->fields[collumn_id].name)
+				    == 0) {
+					iCol = collumn_id;
 					break;
 				}
 			}
 		}
 	}
-	assert(pCol == NULL || pCol == &pTab->aCol[iCol]);
-	if (nTerm == 1 && pCol != NULL &&
-	    (pTab->def->fields[iCol].type == FIELD_TYPE_INTEGER) &&
+	if (nTerm == 1 && iCol != -1 &&
+	    pTab->def->fields[iCol].type == FIELD_TYPE_INTEGER &&
 	    sortOrder != SORT_ORDER_DESC) {
 		assert(autoInc == 0 || autoInc == 1);
 		pTab->iPKey = iCol;
@@ -998,8 +990,8 @@ sqlite3AddCollateType(Parse * pParse, Token * pToken)
 	if (!zColl)
 		return;
 	uint32_t *id = &p->def->fields[i].coll_id;
-	p->aCol[i].coll = sql_get_coll_seq(pParse, zColl, id);
-	if (p->aCol[i].coll != NULL) {
+	p->def->fields[i].coll = sql_get_coll_seq(pParse, zColl, id);
+	if (p->def->fields[i].coll != NULL) {
 		Index *pIdx;
 		/* If the column is declared as "<name> PRIMARY KEY COLLATE <type>",
 		 * then an index may have been created on this column before the
@@ -1220,14 +1212,11 @@ identPut(char *z, int *pIdx, char *zSignedIdent)
 static char *
 createTableStmt(sqlite3 * db, Table * p)
 {
-	int i, k, n;
 	char *zStmt;
 	char *zSep, *zSep2, *zEnd;
-	Column *pCol;
-	n = 0;
-	for (pCol = p->aCol, i = 0; i < (int)p->def->field_count; i++, pCol++) {
+	int n = 0;
+	for (uint32_t i = 0; i < p->def->field_count; i++)
 		n += identLength(p->def->fields[i].name) + 5;
-	}
 	n += identLength(p->def->name);
 	if (n < 50) {
 		zSep = "";
@@ -1245,10 +1234,10 @@ createTableStmt(sqlite3 * db, Table * p)
 		return 0;
 	}
 	sqlite3_snprintf(n, zStmt, "CREATE TABLE ");
-	k = sqlite3Strlen30(zStmt);
+	int k = sqlite3Strlen30(zStmt);
 	identPut(zStmt, &k, p->def->name);
 	zStmt[k++] = '(';
-	for (pCol = p->aCol, i = 0; i < (int)p->def->field_count; i++, pCol++) {
+	for (uint32_t i = 0; i < p->def->field_count; i++) {
 		static const char *const azType[] = {
 			/* AFFINITY_BLOB    */ "",
 			/* AFFINITY_TEXT    */ " TEXT",
@@ -1284,17 +1273,6 @@ createTableStmt(sqlite3 * db, Table * p)
 	return zStmt;
 }
 
-/* Return true if value x is found any of the first nCol entries of aiCol[]
- */
-static int
-hasColumn(const i16 * aiCol, int nCol, int x)
-{
-	while (nCol-- > 0)
-		if (x == *(aiCol++))
-			return 1;
-	return 0;
-}
-
 static int
 field_def_create_for_pk(struct Parse *parser, struct field_def *field,
 			const char *space_name)
@@ -1369,7 +1347,7 @@ convertToWithoutRowidTable(Parse * pParse, Table * pTab)
 		u16 pk_columns = pPk->nColumn;
 		for (i = j = 0; i < pk_columns; i++) {
 			uint32_t part_no = pPk->aiColumn[i];
-			if (hasColumn(pPk->aiColumn, j, part_no)) {
+			if (index_has_column(pPk->aiColumn, j, part_no)) {
 				pPk->nColumn--;
 			} else {
 				pPk->aiColumn[j++] = part_no;
@@ -1961,12 +1939,9 @@ sql_create_view(struct Parse *parse_context, struct Token *begin,
 		sqlite3SelectAddColumnTypeAndCollation(parse_context, p,
 						       select);
 	} else {
-		assert(p->aCol == NULL);
 		assert(sel_tab->def->opts.is_temporary);
 		p->def->fields = sel_tab->def->fields;
 		p->def->field_count = sel_tab->def->field_count;
-		p->aCol = sel_tab->aCol;
-		sel_tab->aCol = NULL;
 		sel_tab->def->fields = NULL;
 		sel_tab->def->field_count = 0;
 	}
diff --git a/src/box/sql/resolve.c b/src/box/sql/resolve.c
index a185473..2c49e2c 100644
--- a/src/box/sql/resolve.c
+++ b/src/box/sql/resolve.c
@@ -219,7 +219,6 @@ lookupName(Parse * pParse,	/* The parsing context */
 	NameContext *pTopNC = pNC;	/* First namecontext in the list */
 	int isTrigger = 0;	/* True if resolved to a trigger column */
 	Table *pTab = 0;	/* Table hold the row */
-	Column *pCol;		/* A column of pTab */
 
 	assert(pNC);		/* the name context cannot be NULL. */
 	assert(zCol);		/* The Z in X.Y.Z cannot be NULL */
@@ -272,9 +271,8 @@ lookupName(Parse * pParse,	/* The parsing context */
 				if (0 == (cntTab++)) {
 					pMatch = pItem;
 				}
-				for (j = 0, pCol = pTab->aCol;
-					j < (int)pTab->def->field_count;
-					j++, pCol++) {
+				for (j = 0; j < (int)pTab->def->field_count;
+				     j++) {
 					if (strcmp(pTab->def->fields[j].name,
 						   zCol) == 0) {
 						/* If there has been exactly one prior match and this match
@@ -331,9 +329,8 @@ lookupName(Parse * pParse,	/* The parsing context */
 			if (pTab) {
 				int iCol;
 				cntTab++;
-				for (iCol = 0, pCol = pTab->aCol;
-				     iCol < (int)pTab->def->field_count;
-				     iCol++, pCol++) {
+				for (iCol = 0; iCol <
+				     (int)pTab->def->field_count; iCol++) {
 					if (strcmp(pTab->def->fields[iCol].name,
 						   zCol) == 0) {
 						if (iCol == pTab->iPKey) {
diff --git a/src/box/sql/select.c b/src/box/sql/select.c
index ceb7e34..745ae2f 100644
--- a/src/box/sql/select.c
+++ b/src/box/sql/select.c
@@ -1811,25 +1811,15 @@ sqlite3ColumnsFromExprList(Parse * parse, ExprList * expr_list, Table *table)
 {
 	/* Database connection */
 	sqlite3 *db = parse->db;
-	int i, j;		/* Loop counters */
 	u32 cnt;		/* Index added to make the name unique */
-	Column *aCol, *pCol;	/* For looping over result columns */
-	int nCol;		/* Number of columns in the result set */
 	Expr *p;		/* Expression for a single result column */
 	char *zName;		/* Column name */
 	int nName;		/* Size of name in zName[] */
 	Hash ht;		/* Hash table of column names */
 
 	sqlite3HashInit(&ht);
-	if (expr_list) {
-		nCol = expr_list->nExpr;
-		aCol = sqlite3DbMallocZero(db, sizeof(aCol[0]) * nCol);
-		testcase(aCol == 0);
-	} else {
-		nCol = 0;
-		aCol = NULL;
-	}
-	assert(nCol == (i16) nCol);
+	uint32_t column_count =
+		(expr_list != NULL) ? (uint32_t)expr_list->nExpr : 0;
 	/*
 	 * This should be a table without resolved columns.
 	 * sqlite3ViewGetColumnNames could use it to resolve
@@ -1838,21 +1828,21 @@ sqlite3ColumnsFromExprList(Parse * parse, ExprList * expr_list, Table *table)
 	assert(table->def->fields == NULL);
 	struct region *region = &parse->region;
 	table->def->fields =
-		region_alloc(region, nCol * sizeof(table->def->fields[0]));
+		region_alloc(region,
+			     column_count * sizeof(table->def->fields[0]));
 	if (table->def->fields == NULL) {
 		sqlite3OomFault(db);
 		goto cleanup;
 	}
-	for (int i = 0; i < nCol; i++) {
+	for (uint32_t i = 0; i < column_count; i++) {
 		memcpy(&table->def->fields[i], &field_def_default,
 		       sizeof(field_def_default));
 		table->def->fields[i].nullable_action = ON_CONFLICT_ACTION_NONE;
 		table->def->fields[i].is_nullable = true;
 	}
-	table->def->field_count = (uint32_t)nCol;
-	table->aCol = aCol;
+	table->def->field_count = column_count;
 
-	for (i = 0, pCol = aCol; i < nCol; i++, pCol++) {
+	for (uint32_t i = 0; i < column_count; i++) {
 		/* Get an appropriate name for the column
 		 */
 		p = sqlite3ExprSkipCollate(expr_list->a[i].pExpr);
@@ -1889,9 +1879,9 @@ sqlite3ColumnsFromExprList(Parse * parse, ExprList * expr_list, Table *table)
 		while (zName && sqlite3HashFind(&ht, zName) != 0) {
 			nName = sqlite3Strlen30(zName);
 			if (nName > 0) {
+				int j;
 				for (j = nName - 1;
-				     j > 0 && sqlite3Isdigit(zName[j]); j--) {
-				}
+				     j > 0 && sqlite3Isdigit(zName[j]); j--);
 				if (zName[j] == ':')
 					nName = j;
 			}
@@ -1901,7 +1891,9 @@ sqlite3ColumnsFromExprList(Parse * parse, ExprList * expr_list, Table *table)
 				sqlite3_randomness(sizeof(cnt), &cnt);
 		}
 		size_t name_len = strlen(zName);
-		if (zName != NULL && sqlite3HashInsert(&ht, zName, pCol) == pCol)
+		void *field = &table->def->fields[i];
+		if (zName != NULL &&
+		    sqlite3HashInsert(&ht, zName, field) == field)
 			sqlite3OomFault(db);
 		table->def->fields[i].name =
 			region_alloc(region, name_len + 1);
@@ -1921,10 +1913,8 @@ cleanup:
 		 * pTable->def could be not temporal in
 		 * sqlite3ViewGetColumnNames so we need clean-up.
 		 */
-		sqlite3DbFree(db, aCol);
 		table->def->fields = NULL;
 		table->def->field_count = 0;
-		table->aCol = NULL;
 		rc = SQLITE_NOMEM_BKPT;
 	}
 	return rc;
@@ -1949,8 +1939,6 @@ sqlite3SelectAddColumnTypeAndCollation(Parse * pParse,		/* Parsing contexts */
 {
 	sqlite3 *db = pParse->db;
 	NameContext sNC;
-	Column *pCol;
-	int i;
 	Expr *p;
 	struct ExprList_item *a;
 
@@ -1963,8 +1951,7 @@ sqlite3SelectAddColumnTypeAndCollation(Parse * pParse,		/* Parsing contexts */
 	memset(&sNC, 0, sizeof(sNC));
 	sNC.pSrcList = pSelect->pSrc;
 	a = pSelect->pEList->a;
-	for (i = 0, pCol = pTab->aCol;
-		i < (int)pTab->def->field_count; i++, pCol++) {
+	for (uint32_t i = 0; i < pTab->def->field_count; i++) {
 		enum field_type type;
 		p = a[i].pExpr;
 		type = columnType(&sNC, p, 0, 0);
@@ -1977,8 +1964,8 @@ sqlite3SelectAddColumnTypeAndCollation(Parse * pParse,		/* Parsing contexts */
 		bool unused;
 		uint32_t id;
 		struct coll *coll = sql_expr_coll(pParse, p, &unused, &id);
-		if (coll != NULL && pCol->coll == NULL) {
-			pCol->coll = coll;
+		if (coll != NULL && pTab->def->fields[i].coll == NULL) {
+			pTab->def->fields[i].coll = coll;
 			pTab->def->fields[i].coll_id = id;
 		}
 	}
diff --git a/src/box/sql/sqliteInt.h b/src/box/sql/sqliteInt.h
index c89d256..a67627f 100644
--- a/src/box/sql/sqliteInt.h
+++ b/src/box/sql/sqliteInt.h
@@ -1872,16 +1872,6 @@ struct Savepoint {
 #define SAVEPOINT_RELEASE    1
 #define SAVEPOINT_ROLLBACK   2
 
-/*
- * information about each column of an SQL table is held in an instance
- * of this structure.
- */
-struct Column {
-	/** Collating sequence. */
-	struct coll *coll;
-	u8 is_primkey;		/* Boolean propertie for being PK */
-};
-
 #define sqlite3IsNumericAffinity(X)  ((X)>=AFFINITY_NUMERIC)
 
 /*
@@ -1910,7 +1900,6 @@ struct Column {
  * by an instance of the following structure.
  */
 struct Table {
-	Column *aCol;		/* Information about each column */
 	Index *pIndex;		/* List of SQL indexes on this table. */
 	FKey *pFKey;		/* Linked list of all foreign keys in this table */
 	char *zColAff;		/* String defining the affinity of each column */
@@ -4418,6 +4407,18 @@ void sqlite3FileSuffix3(const char *, char *);
 #endif
 u8 sqlite3GetBoolean(const char *z, u8);
 
+/**
+ * Test if @column_idx is a part of first @parts_count of index.
+ *
+ * @param parts pointer to array with columns representing index.
+ * @param parts_count count of index parts.
+ * @column_idx index of column to test.
+ * @retval true if index contains column_idx.
+ * @retval false else.
+ */
+bool
+index_has_column(const i16 *parts, int parts_count, int column_idx);
+
 const void *sqlite3ValueText(sqlite3_value *);
 int sqlite3ValueBytes(sqlite3_value *);
 void sqlite3ValueSetStr(sqlite3_value *, int, const void *,
-- 
2.7.4