[tarantool-patches] [PATCH v4 4/7] sql: start using collations and is_nullable from space_def

[Kirill Shcherbatov <kshcherbatov@tarantool.org>]

Part of #3272.
---
 src/box/sql.c           | 36 +++++++++++++++++++++++++-----------
 src/box/sql/alter.c     | 11 ++++++++---
 src/box/sql/build.c     | 20 +++++++++++++-------
 src/box/sql/fkey.c      |  7 ++-----
 src/box/sql/select.c    |  6 ++++--
 src/box/sql/sqliteInt.h |  7 -------
 src/box/sql/vdbeaux.c   |  5 ++++-
 7 files changed, 56 insertions(+), 36 deletions(-)

diff --git a/src/box/sql.c b/src/box/sql.c
index 2893d70..ef11eb9 100644
--- a/src/box/sql.c
+++ b/src/box/sql.c
@@ -1449,7 +1449,9 @@ int tarantoolSqlite3MakeTableFormat(Table *pTable, void *buf)
 
 	for (i = 0; i < n; i++) {
 		const char *t;
-		struct coll *coll = aCol[i].coll;
+		struct coll *coll =
+			coll_by_id(pTable->def->fields[i].coll_id);
+
 		struct field_def *field = &def->fields[i];
 		const char *zToken = field->default_value;
 		int base_len = 4;
@@ -1461,19 +1463,25 @@ int tarantoolSqlite3MakeTableFormat(Table *pTable, void *buf)
 		p = enc->encode_str(p, "name", 4);
 		p = enc->encode_str(p, field->name, strlen(field->name));
 		p = enc->encode_str(p, "type", 4);
+
+		assert(def->fields[i].is_nullable ==
+		       (def->fields[i].nullable_action ==
+			ON_CONFLICT_ACTION_NONE));
+
 		if (i == pk_forced_int) {
 			t = "integer";
 		} else {
 			t = aCol[i].affinity == SQLITE_AFF_BLOB ? "scalar" :
-				convertSqliteAffinity(aCol[i].affinity, aCol[i].notNull == 0);
+				convertSqliteAffinity(aCol[i].affinity,
+						      def->fields[i].is_nullable);
 		}
 		p = enc->encode_str(p, t, strlen(t));
 		p = enc->encode_str(p, "is_nullable", 11);
-		p = enc->encode_bool(p, aCol[i].notNull ==
-				     ON_CONFLICT_ACTION_NONE);
+		p = enc->encode_bool(p, def->fields[i].is_nullable);
 		p = enc->encode_str(p, "nullable_action", 15);
-		assert(aCol[i].notNull < on_conflict_action_MAX);
-		const char *action = on_conflict_action_strs[aCol[i].notNull];
+		assert(def->fields[i].nullable_action < on_conflict_action_MAX);
+		const char *action =
+			on_conflict_action_strs[def->fields[i].nullable_action];
 		p = enc->encode_str(p, action, strlen(action));
 		if (coll != NULL) {
 			p = enc->encode_str(p, "collation", strlen("collation"));
@@ -1552,11 +1560,16 @@ int tarantoolSqlite3MakeIdxParts(SqliteIndex *pIndex, void *buf)
 	p = enc->encode_array(base, n);
 	for (i = 0; i < n; i++) {
 		int col = pIndex->aiColumn[i];
+		assert(def->fields[col].is_nullable ==
+		       (def->fields[col].nullable_action ==
+			ON_CONFLICT_ACTION_NONE));
 		const char *t;
-		if (pk_forced_int == col)
+		if (pk_forced_int == col) {
 			t = "integer";
-		else
-			t = convertSqliteAffinity(aCol[col].affinity, aCol[col].notNull == 0);
+		} else {
+			t = convertSqliteAffinity(aCol[col].affinity,
+						  def->fields[col].is_nullable);
+		}
 		/* do not decode default collation */
 		p = enc->encode_map(p, pIndex->coll_array[i] == NULL ? 4 : 5);
 		p = enc->encode_str(p, "type", sizeof("type")-1);
@@ -1568,9 +1581,10 @@ int tarantoolSqlite3MakeIdxParts(SqliteIndex *pIndex, void *buf)
 			p = enc->encode_uint(p, pIndex->coll_array[i]->id);
 		}
 		p = enc->encode_str(p, "is_nullable", 11);
-		p = enc->encode_bool(p, aCol[col].notNull == ON_CONFLICT_ACTION_NONE);
+		p = enc->encode_bool(p, def->fields[col].is_nullable);
 		p = enc->encode_str(p, "nullable_action", 15);
-		const char *action_str = on_conflict_action_strs[aCol[col].notNull];
+		const char *action_str =
+			on_conflict_action_strs[def->fields[col].nullable_action];
 		p = enc->encode_str(p, action_str, strlen(action_str));
 	}
 	return (int)(p - base);
diff --git a/src/box/sql/alter.c b/src/box/sql/alter.c
index bedf602..f830a15 100644
--- a/src/box/sql/alter.c
+++ b/src/box/sql/alter.c
@@ -198,7 +198,12 @@ sqlite3AlterFinishAddColumn(Parse * pParse, Token * pColDef)
 				"Cannot add a REFERENCES column with non-NULL default value");
 		return;
 	}
-	if (pCol->notNull && !pDflt) {
+	assert(pNew->def->fields[pNew->def->field_count - 1].is_nullable ==
+	       (pNew->def->fields[pNew->def->field_count - 1].nullable_action ==
+		ON_CONFLICT_ACTION_NONE));
+
+	if (pNew->def->fields[pNew->def->field_count - 1].nullable_action
+	    && !pDflt) {
 		sqlite3ErrorMsg(pParse,
 				"Cannot add a NOT NULL column with default value NULL");
 		return;
@@ -307,9 +312,9 @@ sqlite3AlterBeginAddColumn(Parse * pParse, SrcList * pSrc)
 	}
 	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: Column *pCol = &pNew->aCol[i]; */
 		/* FIXME: pCol->zName = sqlite3DbStrDup(db, pCol->zName); */
-		pCol->coll = NULL;
+		/* FIXME: pCol->coll = NULL; */
 	}
 	pNew->pSchema = db->pSchema;
 	pNew->addColOffset = pTab->addColOffset;
diff --git a/src/box/sql/build.c b/src/box/sql/build.c
index 99059b3..e9c0686 100644
--- a/src/box/sql/build.c
+++ b/src/box/sql/build.c
@@ -694,6 +694,8 @@ sqlite3AddColumn(Parse * pParse, Token * pName, Token * pType)
 	memset(pCol, 0, sizeof(p->aCol[0]));
 	struct field_def *column_def = &p->def->fields[p->def->field_count];
 	column_def->name = z;
+	column_def->is_nullable = true;
+	column_def->nullable_action = ON_CONFLICT_ACTION_NONE;
 	if (pType->n == 0) {
 		/* If there is no type specified, columns have the default affinity
 		 * 'BLOB' and type SCALAR.
@@ -740,7 +742,9 @@ sqlite3AddNotNull(Parse * pParse, int onError)
 	p = pParse->pNewTable;
 	if (p == 0 || NEVER(p->def->field_count < 1))
 		return;
-	p->aCol[p->def->field_count - 1].notNull = (u8) onError;
+	p->def->fields[p->def->field_count - 1].nullable_action = (u8)onError;
+	p->def->fields[p->def->field_count - 1].is_nullable =
+		(onError == ON_CONFLICT_ACTION_NONE);
 }
 
 /*
@@ -1018,10 +1022,10 @@ sqlite3AddCollateType(Parse * pParse, Token * pToken)
 	if (!zColl)
 		return;
 
-	struct coll *coll = sqlite3LocateCollSeq(pParse, db, zColl);
+	struct coll *coll = coll_by_name(zColl, strlen(zColl));
 	if (coll != NULL) {
 		Index *pIdx;
-		p->aCol[i].coll = coll;
+		p->def->fields[i].coll_id = coll->id;
 
 		/* If the column is declared as "<name> PRIMARY KEY COLLATE <type>",
 		 * then an index may have been created on this column before the
@@ -1032,8 +1036,6 @@ sqlite3AddCollateType(Parse * pParse, Token * pToken)
 			if (pIdx->aiColumn[0] == i)
 				pIdx->coll_array[0] = sql_column_collation(p, i);
 		}
-	} else {
-		sqlite3DbFree(db, zColl);
 	}
 }
 
@@ -1064,7 +1066,9 @@ sql_column_collation(Table *table, uint32_t column)
 	 */
 	if (space == NULL || space_index(space, 0) == NULL) {
 		assert(column < (uint32_t)table->def->field_count);
-		return table->aCol[column].coll;
+		struct coll *coll =
+			coll_by_id(table->def->fields[column].coll_id);
+		return coll;
 	}
 
 	return space->format->fields[column].coll;
@@ -1411,7 +1415,9 @@ convertToWithoutRowidTable(Parse * pParse, Table * pTab)
 	if (!db->init.imposterTable) {
 		for (i = 0; i < (int)pTab->def->field_count; i++) {
 			if (pTab->aCol[i].is_primkey) {
-				pTab->aCol[i].notNull = ON_CONFLICT_ACTION_ABORT;
+				pTab->def->fields[i].nullable_action
+					= ON_CONFLICT_ACTION_ABORT;
+				pTab->def->fields[i].is_nullable = false;
 			}
 		}
 	}
diff --git a/src/box/sql/fkey.c b/src/box/sql/fkey.c
index c15ad8c..8c015c9 100644
--- a/src/box/sql/fkey.c
+++ b/src/box/sql/fkey.c
@@ -34,6 +34,7 @@
  * support to compiled SQL statements.
  */
 #include <box/coll.h>
+#include "box/coll_cache.h"
 #include "sqliteInt.h"
 #include "box/session.h"
 #include "tarantoolInt.h"
@@ -535,11 +536,7 @@ exprTableRegister(Parse * pParse,	/* Parsing and code generating context */
 			pCol = &pTab->aCol[iCol];
 			pExpr->iTable = regBase + iCol + 1;
 			pExpr->affinity = pCol->affinity;
-			const char *coll_name;
-			if (pCol->coll == NULL && pCol->coll != NULL)
-				coll_name = pCol->coll->name;
-			else
-				coll_name = "binary";
+			const char *coll_name = "binary";
 			pExpr = sqlite3ExprAddCollateString(pParse, pExpr,
 							    coll_name);
 		} else {
diff --git a/src/box/sql/select.c b/src/box/sql/select.c
index 03bfcf9..fa1de9b 100644
--- a/src/box/sql/select.c
+++ b/src/box/sql/select.c
@@ -1836,6 +1836,8 @@ sqlite3ColumnsFromExprList(Parse * pParse,	/* Parsing context */
 	pTable->def->fields =
 		region_alloc(region, nCol*sizeof(pTable->def->fields[0]));
 	memset(pTable->def->fields, 0, nCol*sizeof(pTable->def->fields[0]));
+	for (int i = 0; i < nCol; i++)
+		pTable->def->fields[i].is_nullable = true;
 	pTable->def->field_count = (uint32_t)nCol;
 	pTable->aCol = aCol;
 
@@ -1961,8 +1963,8 @@ sqlite3SelectAddColumnTypeAndCollation(Parse * pParse,		/* Parsing contexts */
 			pCol->affinity = SQLITE_AFF_BLOB;
 		bool unused;
 		struct coll *coll = sql_expr_coll(pParse, p, &unused);
-		if (coll != NULL && pCol->coll == NULL)
-			pCol->coll = coll;
+		if (coll != NULL && pTab->def->fields[i].coll_id == COLL_NONE)
+			pTab->def->fields[i].coll_id = coll->id;
 	}
 	pTab->szTabRow = sqlite3LogEst(szAll * 4);
 }
diff --git a/src/box/sql/sqliteInt.h b/src/box/sql/sqliteInt.h
index 520b74c..a045286 100644
--- a/src/box/sql/sqliteInt.h
+++ b/src/box/sql/sqliteInt.h
@@ -1867,13 +1867,6 @@ struct Savepoint {
  * of this structure.
  */
 struct Column {
-	/** Collating sequence. */
-	struct coll *coll;
-	/**
-	 * An ON_CONFLICT_ACTION code for handling a NOT NULL
-	 * constraint.
-	 */
-	enum on_conflict_action notNull;
 	char affinity;		/* One of the SQLITE_AFF_... values */
 	u8 szEst;		/* Estimated size of value in this column. sizeof(INT)==1 */
 	u8 is_primkey;		/* Boolean propertie for being PK */
diff --git a/src/box/sql/vdbeaux.c b/src/box/sql/vdbeaux.c
index b3998ea..f76c689 100644
--- a/src/box/sql/vdbeaux.c
+++ b/src/box/sql/vdbeaux.c
@@ -4743,6 +4743,9 @@ table_column_is_nullable(struct Table *tab, uint32_t column)
 			ON_CONFLICT_ACTION_NONE;
 	} else {
 		/* tab is ephemeral (in SQLite sense).  */
-		return tab->aCol[column].notNull == 0;
+		assert(tab->def->fields[column].is_nullable ==
+		       (tab->def->fields[column].nullable_action ==
+			ON_CONFLICT_ACTION_NONE));
+		return tab->def->fields[column].is_nullable;
 	}
 }
-- 
2.7.4