[tarantool-patches] Re: [PATCH 1/2] sql: introduce sort order to key_part/key_part_def

[Kirill Yukhin <kyukhin@tarantool.org>]

Hello Vlad,
Thanks for review. Updated patch in the bottom.

On 08 мая 19:02, Vladislav Shpilevoy wrote:
> Hello. Thanks for contributing! See 2 comments below.
> 
> 1. I still can grep SQLITE_SO_ASC/DESC in build.c
Fixed.
> 
> On 08/05/2018 10:56, Kirill Yukhin wrote:
> > Legacy SQL DD structs contained sort_order, defined per
> > index column. During integration, those structs are to be
> > vanished. So, introduce new field to part entity of Tarantool.
> > This field states for sorting order of given part in give index.
> > This field is ignored by Tarantool everywhere excpept for
> > some of nested queries in SQL.
> > 
> > Patch also replaces usages of SQL's stored sort order w/ this new
> > field.
> > 
> > Part of #3235
> > ---
> >   src/box/key_def.cc      | 28 ++++++++++++++++++++++------
> >   src/box/key_def.h       | 16 +++++++++++++++-
> >   src/box/schema.cc       | 30 ++++++++++++++++++++----------
> >   src/box/sql.c           | 11 +++++++++--
> >   src/box/sql/build.c     | 47 ++++++++++++++++++++++++++++++++++++-----------
> >   src/box/sql/expr.c      |  9 ++++-----
> >   src/box/sql/insert.c    |  3 ++-
> >   src/box/sql/parse.y     | 10 +++++-----
> >   src/box/sql/pragma.c    |  7 ++++---
> >   src/box/sql/select.c    |  2 +-
> >   src/box/sql/sqliteInt.h | 20 ++++++++++++--------
> >   src/box/sql/vdbe.h      |  1 +
> >   src/box/sql/where.c     | 11 ++++++-----
> >   src/box/sql/wherecode.c | 10 ++++++----
> >   14 files changed, 143 insertions(+), 62 deletions(-)
> > 
> > diff --git a/src/box/sql/sqliteInt.h b/src/box/sql/sqliteInt.h
> > index 8bb45c9..a811932 100644
> > --- a/src/box/sql/sqliteInt.h
> > +++ b/src/box/sql/sqliteInt.h
> > @@ -3540,6 +3534,16 @@ sql_default_coll();
> >   bool
> >   space_is_view(Table *);
> > +/**
> > + * Return name of given column collation from index.
> 
> 2. Irrelevant comment.
Fixed.

--
Regards, Kirill Yukhin

diff --git a/src/box/key_def.cc b/src/box/key_def.cc
index 98719c2..9f1a373 100644
--- a/src/box/key_def.cc
+++ b/src/box/key_def.cc
@@ -36,12 +36,15 @@
 #include "schema_def.h"
 #include "coll_cache.h"
 
+const char *sort_order_strs[] = { "asc", "desc", "undef" };
+
 static const struct key_part_def key_part_def_default = {
 	0,
 	field_type_MAX,
 	COLL_NONE,
 	false,
-	ON_CONFLICT_ACTION_ABORT
+	ON_CONFLICT_ACTION_ABORT,
+	SORT_ORDER_ASC
 };
 
 static int64_t
@@ -55,6 +58,7 @@ part_type_by_name_wrapper(const char *str, uint32_t len)
 #define PART_OPT_COLLATION	 "collation"
 #define PART_OPT_NULLABILITY	 "is_nullable"
 #define PART_OPT_NULLABLE_ACTION "nullable_action"
+#define PART_OPT_SORT_ORDER	 "sort_order"
 
 const struct opt_def part_def_reg[] = {
 	OPT_DEF_ENUM(PART_OPT_TYPE, field_type, struct key_part_def, type,
@@ -65,6 +69,8 @@ const struct opt_def part_def_reg[] = {
 		is_nullable),
 	OPT_DEF_ENUM(PART_OPT_NULLABLE_ACTION, on_conflict_action,
 		     struct key_part_def, nullable_action, NULL),
+	OPT_DEF_ENUM(PART_OPT_SORT_ORDER, sort_order, struct key_part_def,
+		     sort_order, NULL),
 	OPT_END,
 };
 
@@ -169,7 +175,7 @@ key_def_new_with_parts(struct key_part_def *parts, uint32_t part_count)
 			}
 		}
 		key_def_set_part(def, i, part->fieldno, part->type,
-				 part->nullable_action, coll);
+				 part->nullable_action, coll, part->sort_order);
 	}
 	return def;
 }
@@ -199,7 +205,8 @@ box_key_def_new(uint32_t *fields, uint32_t *types, uint32_t part_count)
 	for (uint32_t item = 0; item < part_count; ++item) {
 		key_def_set_part(key_def, item, fields[item],
 				 (enum field_type)types[item],
-				 key_part_def_default.nullable_action, NULL);
+				 key_part_def_default.nullable_action,
+				 NULL, SORT_ORDER_ASC);
 	}
 	return key_def;
 }
@@ -252,7 +259,7 @@ key_part_cmp(const struct key_part *parts1, uint32_t part_count1,
 void
 key_def_set_part(struct key_def *def, uint32_t part_no, uint32_t fieldno,
 		 enum field_type type, enum on_conflict_action nullable_action,
-		 struct coll *coll)
+		 struct coll *coll, enum sort_order sort_order)
 {
 	assert(part_no < def->part_count);
 	assert(type < field_type_MAX);
@@ -261,6 +268,7 @@ key_def_set_part(struct key_def *def, uint32_t part_no, uint32_t fieldno,
 	def->parts[part_no].fieldno = fieldno;
 	def->parts[part_no].type = type;
 	def->parts[part_no].coll = coll;
+	def->parts[part_no].sort_order = sort_order;
 	column_mask_set_fieldno(&def->column_mask, fieldno);
 	/**
 	 * When all parts are set, initialize the tuple
@@ -510,6 +518,12 @@ key_def_decode_parts(struct key_part_def *parts, uint32_t part_count,
 				 "nullable action properties");
 			return -1;
 		}
+		if (part->sort_order == sort_order_MAX) {
+			diag_set(ClientError, ER_WRONG_INDEX_OPTIONS,
+				 i + TUPLE_INDEX_BASE,
+				 "index part: unknown sort order");
+			return -1;
+		}
 	}
 	return 0;
 }
@@ -572,7 +586,8 @@ key_def_merge(const struct key_def *first, const struct key_def *second)
 	end = part + first->part_count;
 	for (; part != end; part++) {
 		key_def_set_part(new_def, pos++, part->fieldno, part->type,
-				 part->nullable_action, part->coll);
+				 part->nullable_action, part->coll,
+				 part->sort_order);
 	}
 
 	/* Set-append second key def's part to the new key def. */
@@ -582,7 +597,8 @@ key_def_merge(const struct key_def *first, const struct key_def *second)
 		if (key_def_find(first, part->fieldno))
 			continue;
 		key_def_set_part(new_def, pos++, part->fieldno, part->type,
-				 part->nullable_action, part->coll);
+				 part->nullable_action, part->coll,
+				 part->sort_order);
 	}
 	return new_def;
 }
diff --git a/src/box/key_def.h b/src/box/key_def.h
index a02bfa1..4b18175 100644
--- a/src/box/key_def.h
+++ b/src/box/key_def.h
@@ -45,6 +45,16 @@ extern "C" {
 /* MsgPack type names */
 extern const char *mp_type_strs[];
 
+/* Sorting order of a part. */
+extern const char *sort_order_strs[];
+
+enum sort_order {
+	SORT_ORDER_ASC = 0,
+	SORT_ORDER_DESC,
+	SORT_ORDER_UNDEF,
+	sort_order_MAX
+};
+
 struct key_part_def {
 	/** Tuple field index for this part. */
 	uint32_t fieldno;
@@ -56,6 +66,8 @@ struct key_part_def {
 	bool is_nullable;
 	/** Action to perform if NULL constraint failed. */
 	enum on_conflict_action nullable_action;
+	/** Part sort order. */
+	enum sort_order sort_order;
 };
 
 /**
@@ -74,6 +86,8 @@ struct key_part {
 	struct coll *coll;
 	/** Action to perform if NULL constraint failed. */
 	enum on_conflict_action nullable_action;
+	/** Part sort order. */
+	enum sort_order sort_order;
 };
 
 struct key_def;
@@ -264,7 +278,7 @@ key_def_dump_parts(const struct key_def *def, struct key_part_def *parts);
 void
 key_def_set_part(struct key_def *def, uint32_t part_no, uint32_t fieldno,
 		 enum field_type type, enum on_conflict_action nullable_action,
-		 struct coll *coll);
+		 struct coll *coll, enum sort_order sort_order);
 
 /**
  * Update 'has_optional_parts' of @a key_def with correspondence
diff --git a/src/box/schema.cc b/src/box/schema.cc
index 2e14ec2..c3a8f93 100644
--- a/src/box/schema.cc
+++ b/src/box/schema.cc
@@ -279,13 +279,15 @@ schema_init()
 	auto key_def_guard = make_scoped_guard([&] { key_def_delete(key_def); });
 
 	key_def_set_part(key_def, 0 /* part no */, 0 /* field no */,
-			 FIELD_TYPE_STRING, ON_CONFLICT_ACTION_ABORT, NULL);
+			 FIELD_TYPE_STRING, ON_CONFLICT_ACTION_ABORT, NULL,
+			 SORT_ORDER_ASC);
 	sc_space_new(BOX_SCHEMA_ID, "_schema", key_def, &on_replace_schema,
 		     NULL);
 
 	/* _space - home for all spaces. */
 	key_def_set_part(key_def, 0 /* part no */, 0 /* field no */,
-			 FIELD_TYPE_UNSIGNED, ON_CONFLICT_ACTION_ABORT, NULL);
+			 FIELD_TYPE_UNSIGNED, ON_CONFLICT_ACTION_ABORT, NULL,
+			 SORT_ORDER_ASC);
 
 	/* _collation - collation description. */
 	sc_space_new(BOX_COLLATION_ID, "_collation", key_def,
@@ -329,7 +331,8 @@ schema_init()
 
 	/* _trigger - all existing SQL triggers. */
 	key_def_set_part(key_def, 0 /* part no */, 0 /* field no */,
-			 FIELD_TYPE_STRING, ON_CONFLICT_ACTION_ABORT, NULL);
+			 FIELD_TYPE_STRING, ON_CONFLICT_ACTION_ABORT, NULL,
+			 SORT_ORDER_ASC);
 	sc_space_new(BOX_TRIGGER_ID, "_trigger", key_def, &on_replace_trigger, NULL);
 
 	key_def_delete(key_def);
@@ -338,19 +341,23 @@ schema_init()
 		diag_raise();
 	/* space no */
 	key_def_set_part(key_def, 0 /* part no */, 0 /* field no */,
-			 FIELD_TYPE_UNSIGNED, ON_CONFLICT_ACTION_ABORT, NULL);
+			 FIELD_TYPE_UNSIGNED, ON_CONFLICT_ACTION_ABORT, NULL,
+			 SORT_ORDER_ASC);
 	/* index no */
 	key_def_set_part(key_def, 1 /* part no */, 1 /* field no */,
-			 FIELD_TYPE_UNSIGNED, ON_CONFLICT_ACTION_ABORT, NULL);
+			 FIELD_TYPE_UNSIGNED, ON_CONFLICT_ACTION_ABORT, NULL,
+			 SORT_ORDER_ASC);
 	sc_space_new(BOX_INDEX_ID, "_index", key_def,
 		     &alter_space_on_replace_index, &on_stmt_begin_index);
 
 	/* space name */
 	key_def_set_part(key_def, 0 /* part no */, 0 /* field no */,
-			 FIELD_TYPE_STRING, ON_CONFLICT_ACTION_ABORT, NULL);
+			 FIELD_TYPE_STRING, ON_CONFLICT_ACTION_ABORT, NULL,
+			 SORT_ORDER_ASC);
 	/* index name */
 	key_def_set_part(key_def, 1 /* part no */, 1 /* field no */,
-			 FIELD_TYPE_STRING, ON_CONFLICT_ACTION_ABORT, NULL);
+			 FIELD_TYPE_STRING, ON_CONFLICT_ACTION_ABORT, NULL,
+			 SORT_ORDER_ASC);
 	/* _sql_stat1 - a simpler statistics on space, seen in SQL. */
 	sc_space_new(BOX_SQL_STAT1_ID, "_sql_stat1", key_def, NULL, NULL);
 
@@ -361,13 +368,16 @@ schema_init()
 
 	/* space name */
 	key_def_set_part(key_def, 0 /* part no */, 0 /* field no */,
-			 FIELD_TYPE_STRING, ON_CONFLICT_ACTION_ABORT, NULL);
+			 FIELD_TYPE_STRING, ON_CONFLICT_ACTION_ABORT, NULL,
+			 SORT_ORDER_ASC);
 	/* index name */
 	key_def_set_part(key_def, 1 /* part no */, 1 /* field no */,
-			 FIELD_TYPE_STRING, ON_CONFLICT_ACTION_ABORT, NULL);
+			 FIELD_TYPE_STRING, ON_CONFLICT_ACTION_ABORT, NULL,
+			 SORT_ORDER_ASC);
 	/* sample */
 	key_def_set_part(key_def, 2 /* part no */, 5 /* field no */,
-			 FIELD_TYPE_SCALAR, ON_CONFLICT_ACTION_ABORT, NULL);
+			 FIELD_TYPE_SCALAR, ON_CONFLICT_ACTION_ABORT, NULL,
+			 SORT_ORDER_ASC);
 	/* _sql_stat4 - extensive statistics on space, seen in SQL. */
 	sc_space_new(BOX_SQL_STAT4_ID, "_sql_stat4", key_def, NULL, NULL);
 }
diff --git a/src/box/sql.c b/src/box/sql.c
index 166bb71..838fcf6 100644
--- a/src/box/sql.c
+++ b/src/box/sql.c
@@ -384,7 +384,8 @@ int tarantoolSqlite3EphemeralCreate(BtCursor *pCur, uint32_t field_count,
 				 part /* filed no */,
 				 FIELD_TYPE_SCALAR,
 				 ON_CONFLICT_ACTION_NONE /* nullable_action */,
-				 aColl /* coll */);
+				 aColl /* coll */,
+				 SORT_ORDER_ASC);
 	}
 
 	struct index_def *ephemer_index_def =
@@ -1555,7 +1556,7 @@ int tarantoolSqlite3MakeIdxParts(SqliteIndex *pIndex, void *buf)
 		else
 			t = convertSqliteAffinity(aCol[col].affinity, aCol[col].notNull == 0);
 		/* do not decode default collation */
-		p = enc->encode_map(p, pIndex->coll_array[i] == NULL ? 4 : 5);
+		p = enc->encode_map(p, pIndex->coll_array[i] == NULL ? 5 : 6);
 		p = enc->encode_str(p, "type", sizeof("type")-1);
 		p = enc->encode_str(p, t, strlen(t));
 		p = enc->encode_str(p, "field", sizeof("field")-1);
@@ -1569,6 +1570,12 @@ int tarantoolSqlite3MakeIdxParts(SqliteIndex *pIndex, void *buf)
 		p = enc->encode_str(p, "nullable_action", 15);
 		const char *action_str = on_conflict_action_strs[aCol[col].notNull];
 		p = enc->encode_str(p, action_str, strlen(action_str));
+
+		p = enc->encode_str(p, "sort_order", 10);
+		enum sort_order sort_order = pIndex->sort_order[i];
+		assert(sort_order < sort_order_MAX);
+		const char *sort_order_str = sort_order_strs[sort_order];
+		p = enc->encode_str(p, sort_order_str, strlen(sort_order_str));
 	}
 	return (int)(p - base);
 }
diff --git a/src/box/sql/build.c b/src/box/sql/build.c
index a2b712a..029c71e 100644
--- a/src/box/sql/build.c
+++ b/src/box/sql/build.c
@@ -930,7 +930,7 @@ sqlite3AddPrimaryKey(Parse * pParse,	/* Parsing context */
 		     ExprList * pList,	/* List of field names to be indexed */
 		     int onError,	/* What to do with a uniqueness conflict */
 		     int autoInc,	/* True if the AUTOINCREMENT keyword is present */
-		     int sortOrder	/* SQLITE_SO_ASC or SQLITE_SO_DESC */
+		     enum sort_order sortOrder
     )
 {
 	Table *pTab = pParse->pNewTable;
@@ -976,7 +976,7 @@ sqlite3AddPrimaryKey(Parse * pParse,	/* Parsing context */
 	if (nTerm == 1
 	    && pCol
 	    && (sqlite3ColumnType(pCol) == FIELD_TYPE_INTEGER)
-	    && sortOrder != SQLITE_SO_DESC) {
+	    && sortOrder != SORT_ORDER_DESC) {
 		assert(autoInc == 0 || autoInc == 1);
 		pTab->iPKey = iCol;
 		pTab->keyConf = (u8) onError;
@@ -1125,6 +1125,30 @@ sql_index_collation(Index *idx, uint32_t column)
 	return index->def->key_def->parts[column].coll;
 }
 
+enum sort_order
+sql_index_column_sort_order(Index *idx, uint32_t column)
+{
+	assert(idx != NULL);
+	uint32_t space_id = SQLITE_PAGENO_TO_SPACEID(idx->pTable->tnum);
+	struct space *space = space_by_id(space_id);
+
+	assert(column < idx->nColumn);
+	/*
+	 * If space is still under construction, or it is
+	 * an ephemeral space, then fetch collation from
+	 * SQL internal structure.
+	 */
+	if (space == NULL) {
+		assert(column < idx->nColumn);
+		return idx->sort_order[column];
+	}
+
+	uint32_t index_id = SQLITE_PAGENO_TO_INDEXID(idx->tnum);
+	struct index *index = space_index(space, index_id);
+	assert(index != NULL && index->def->key_def->part_count >= column);
+	return index->def->key_def->parts[column].sort_order;
+}
+
 /**
  * Return true if space which corresponds to
  * given table has view option.
@@ -2694,11 +2718,11 @@ sqlite3AllocateIndexObject(sqlite3 * db,	/* Database connection */
 	Index *p;		/* Allocated index object */
 	int nByte;		/* Bytes of space for Index object + arrays */
 
-	nByte = ROUND8(sizeof(Index)) +	/* Index structure  */
-	    ROUND8(sizeof(char *) * nCol) +	/* Index.azColl     */
-	    ROUND8(sizeof(LogEst) * (nCol + 1) +	/* Index.aiRowLogEst   */
-		   sizeof(i16) * nCol +	/* Index.aiColumn   */
-		   sizeof(u8) * nCol);	/* Index.aSortOrder */
+	nByte = ROUND8(sizeof(Index)) +		    /* Index structure  */
+	    ROUND8(sizeof(char *) * nCol) +	    /* Index.azColl     */
+	    ROUND8(sizeof(LogEst) * (nCol + 1) +    /* Index.aiRowLogEst   */
+		   sizeof(i16) * nCol +		    /* Index.aiColumn   */
+		   sizeof(enum sort_order) * nCol); /* Index.sort_order */
 	p = sqlite3DbMallocZero(db, nByte + nExtra);
 	if (p) {
 		char *pExtra = ((char *)p) + ROUND8(sizeof(Index));
@@ -2708,7 +2732,7 @@ sqlite3AllocateIndexObject(sqlite3 * db,	/* Database connection */
 		pExtra += sizeof(LogEst) * (nCol + 1);
 		p->aiColumn = (i16 *) pExtra;
 		pExtra += sizeof(i16) * nCol;
-		p->aSortOrder = (u8 *) pExtra;
+		p->sort_order = (enum sort_order *) pExtra;
 		p->nColumn = nCol;
 		*ppExtra = ((char *)p) + nByte;
 	}
@@ -3037,7 +3061,7 @@ sqlite3CreateIndex(Parse * pParse,	/* All information about this parse */
 	 */
 	for (i = 0, pListItem = pList->a; i < pList->nExpr; i++, pListItem++) {
 		Expr *pCExpr;	/* The i-th index expression */
-		int requestedSortOrder;	/* ASC or DESC on the i-th expression */
+		enum sort_order requested_so;	/* ASC or DESC on the i-th expression */
 		sqlite3ResolveSelfReference(pParse, pTab, NC_IdxExpr,
 					    pListItem->pExpr, 0);
 		if (pParse->nErr)
@@ -3075,8 +3099,8 @@ sqlite3CreateIndex(Parse * pParse,	/* All information about this parse */
 		/* Tarantool: DESC indexes are not supported so far.
 		 * See gh-3016.
 		 */
-		requestedSortOrder = pListItem->sortOrder & 0;
-		pIndex->aSortOrder[i] = (u8) requestedSortOrder;
+		requested_so = pListItem->sortOrder & 0;
+		pIndex->sort_order[i] = requested_so;
 	}
 
 	sqlite3DefaultRowEst(pIndex);
@@ -4199,7 +4223,8 @@ sqlite3KeyInfoOfIndex(Parse * pParse, sqlite3 * db, Index * pIdx)
 		assert(sqlite3KeyInfoIsWriteable(pKey));
 		for (i = 0; i < nCol; i++) {
 			pKey->aColl[i] = sql_index_collation(pIdx, i);
-			pKey->aSortOrder[i] = pIdx->aSortOrder[i];
+			pKey->aSortOrder[i] = sql_index_column_sort_order(pIdx,
+									  i);
 		}
 		if (pParse && pParse->nErr) {
 			sqlite3KeyInfoUnref(pKey);
diff --git a/src/box/sql/expr.c b/src/box/sql/expr.c
index 0c86761..cc969ca 100644
--- a/src/box/sql/expr.c
+++ b/src/box/sql/expr.c
@@ -1764,11 +1764,9 @@ sqlite3ExprListSetSortOrder(ExprList * p, int iSortOrder)
 {
 	if (p == 0)
 		return;
-	assert(SQLITE_SO_UNDEFINED < 0 && SQLITE_SO_ASC >= 0
-	       && SQLITE_SO_DESC > 0);
 	assert(p->nExpr > 0);
-	if (iSortOrder < 0) {
-		assert(p->a[p->nExpr - 1].sortOrder == SQLITE_SO_ASC);
+	if (iSortOrder == SORT_ORDER_UNDEF) {
+		assert(p->a[p->nExpr - 1].sortOrder == SORT_ORDER_ASC);
 		return;
 	}
 	p->a[p->nExpr - 1].sortOrder = (u8) iSortOrder;
@@ -2529,7 +2527,8 @@ sqlite3FindInIndex(Parse * pParse,	/* Parsing context */
 					assert(IN_INDEX_INDEX_DESC ==
 					       IN_INDEX_INDEX_ASC + 1);
 					eType = IN_INDEX_INDEX_ASC +
-						pIdx->aSortOrder[0];
+						sql_index_column_sort_order(pIdx,
+									    0);
 
 					if (prRhsHasNull) {
 #ifdef SQLITE_ENABLE_COLUMN_USED_MASK
diff --git a/src/box/sql/insert.c b/src/box/sql/insert.c
index 939b5e3..1a34f71 100644
--- a/src/box/sql/insert.c
+++ b/src/box/sql/insert.c
@@ -1653,7 +1653,8 @@ xferCompatibleIndex(Index * pDest, Index * pSrc)
 				return 0;	/* Different expressions in the index */
 			}
 		}
-		if (pSrc->aSortOrder[i] != pDest->aSortOrder[i]) {
+		if (sql_index_column_sort_order(pSrc, i) !=
+		    sql_index_column_sort_order(pDest, i)) {
 			return 0;	/* Different sort orders */
 		}
 		if (sql_index_collation(pSrc, i) !=
diff --git a/src/box/sql/parse.y b/src/box/sql/parse.y
index b078e20..249d20d 100644
--- a/src/box/sql/parse.y
+++ b/src/box/sql/parse.y
@@ -684,9 +684,9 @@ sortlist(A) ::= expr(Y) sortorder(Z). {
 
 %type sortorder {int}
 
-sortorder(A) ::= ASC.           {A = SQLITE_SO_ASC;}
-sortorder(A) ::= DESC.          {A = SQLITE_SO_DESC;}
-sortorder(A) ::= .              {A = SQLITE_SO_UNDEFINED;}
+sortorder(A) ::= ASC.           {A = SORT_ORDER_ASC;}
+sortorder(A) ::= DESC.          {A = SORT_ORDER_DESC;}
+sortorder(A) ::= .              {A = SORT_ORDER_UNDEF;}
 
 %type groupby_opt {ExprList*}
 %destructor groupby_opt {sqlite3ExprListDelete(pParse->db, $$);}
@@ -1244,7 +1244,7 @@ cmd ::= createkw(S) uniqueflag(U) INDEX ifnotexists(NE) nm(X)
         ON nm(Y) LP sortlist(Z) RP where_opt(W). {
   sqlite3CreateIndex(pParse, &X, 
                      sqlite3SrcListAppend(pParse->db,0,&Y), Z, U,
-                      &S, W, SQLITE_SO_ASC, NE, SQLITE_IDXTYPE_APPDEF);
+                      &S, W, SORT_ORDER_ASC, NE, SQLITE_IDXTYPE_APPDEF);
 }
 
 %type uniqueflag {int}
@@ -1279,7 +1279,7 @@ uniqueflag(A) ::= .        {A = ON_CONFLICT_ACTION_NONE;}
     int sortOrder
   ){
     ExprList *p = sqlite3ExprListAppend(pParse, pPrior, 0);
-    if( (hasCollate || sortOrder!=SQLITE_SO_UNDEFINED)
+    if( (hasCollate || sortOrder != SORT_ORDER_UNDEF)
         && pParse->db->init.busy==0
     ){
       sqlite3ErrorMsg(pParse, "syntax error after column name \"%.*s\"",
diff --git a/src/box/sql/pragma.c b/src/box/sql/pragma.c
index e41f69b..738c254 100644
--- a/src/box/sql/pragma.c
+++ b/src/box/sql/pragma.c
@@ -463,12 +463,13 @@ sqlite3Pragma(Parse * pParse, Token * pId,	/* First part of [schema.]id field */
 								c_n = coll->name;
 							else
 								c_n = "BINARY";
+							enum sort_order sort_order;
+							sort_order = sql_index_column_sort_order(pIdx,
+												 i);
 							sqlite3VdbeMultiLoad(v,
 									     4,
 									     "isi",
-									     pIdx->
-									     aSortOrder
-									     [i],
+									     sort_order,
 									     c_n,
 									     i <
 									     mx);
diff --git a/src/box/sql/select.c b/src/box/sql/select.c
index 5a50413..aff534d3 100644
--- a/src/box/sql/select.c
+++ b/src/box/sql/select.c
@@ -5620,7 +5620,7 @@ sqlite3Select(Parse * pParse,		/* The parser context */
 		 * and plus one column for ID.
 		 */
 		int nCols = pEList->nExpr + sSort.pOrderBy->nExpr + 1;
-		if (pKeyInfo->aSortOrder[0] == SQLITE_SO_DESC) {
+		if (pKeyInfo->aSortOrder[0] == SORT_ORDER_DESC) {
 			sSort.sortFlags |= SORTFLAG_DESC;
 		}
 		sSort.addrSortIndex =
diff --git a/src/box/sql/sqliteInt.h b/src/box/sql/sqliteInt.h
index 8bb45c9..c2c1090 100644
--- a/src/box/sql/sqliteInt.h
+++ b/src/box/sql/sqliteInt.h
@@ -1881,13 +1881,6 @@ struct Column {
 	u8 is_primkey;		/* Boolean propertie for being PK */
 };
 
-/*
- * A sort order can be either ASC or DESC.
- */
-#define SQLITE_SO_ASC       0	/* Sort in ascending order */
-#define SQLITE_SO_DESC      1	/* Sort in ascending order */
-#define SQLITE_SO_UNDEFINED -1	/* No sort order specified */
-
 /*
  * Column affinity types.
  *
@@ -2143,7 +2136,8 @@ struct Index {
 	char *zColAff;		/* String defining the affinity of each column */
 	Index *pNext;		/* The next index associated with the same table */
 	Schema *pSchema;	/* Schema containing this index */
-	u8 *aSortOrder;		/* for each column: True==DESC, False==ASC */
+	/** Sorting order for each column. */
+	enum sort_order *sort_order;
 	/** Array of collation sequences for index. */
 	struct coll **coll_array;
 	Expr *pPartIdxWhere;	/* WHERE clause for partial indices */
@@ -3522,7 +3516,7 @@ Index *sqlite3PrimaryKeyIndex(Table *);
 void sqlite3StartTable(Parse *, Token *, int);
 void sqlite3AddColumn(Parse *, Token *, Token *);
 void sqlite3AddNotNull(Parse *, int);
-void sqlite3AddPrimaryKey(Parse *, ExprList *, int, int, int);
+void sqlite3AddPrimaryKey(Parse *, ExprList *, int, int, enum sort_order);
 void sqlite3AddCheckConstraint(Parse *, Expr *);
 void sqlite3AddDefaultValue(Parse *, ExprSpan *);
 void sqlite3AddCollateType(Parse *, Token *);
@@ -3540,6 +3534,16 @@ sql_default_coll();
 bool
 space_is_view(Table *);
 
+/**
+ * Return sort order of given column from index.
+ *
+ * @param idx Index which is used to fetch column.
+ * @param column Number of column.
+ * @retval Sort order of requested column.
+ */
+enum sort_order
+sql_index_column_sort_order(Index *idx, uint32_t column);
+
 void sqlite3EndTable(Parse *, Token *, Token *, Select *);
 int
 emit_open_cursor(Parse *, int, int);
diff --git a/src/box/sql/vdbe.h b/src/box/sql/vdbe.h
index 340ddc7..e244606 100644
--- a/src/box/sql/vdbe.h
+++ b/src/box/sql/vdbe.h
@@ -145,6 +145,7 @@ typedef struct VdbeOpList VdbeOpList;
 #define P4_FUNCCTX  (-16)	/* P4 is a pointer to an sqlite3_context object */
 #define P4_BOOL     (-17)	/* P4 is a bool value */
 #define P4_PTR      (-18)	/* P4 is a generic pointer */
+#define P4_KEYDEF   (-19)       /* P4 is a pointer to key_def structure. */
 
 
 /* Error message codes for OP_Halt */
diff --git a/src/box/sql/where.c b/src/box/sql/where.c
index 7a7103c..bad964a 100644
--- a/src/box/sql/where.c
+++ b/src/box/sql/where.c
@@ -1336,8 +1336,8 @@ whereRangeScanEst(Parse * pParse,	/* Parsing & code generating context */
 			       || (pLower->eOperator & (WO_GT | WO_GE)) != 0);
 			assert(pUpper == 0
 			       || (pUpper->eOperator & (WO_LT | WO_LE)) != 0);
-			assert(p->aSortOrder != 0);
-			if (p->aSortOrder[nEq]) {
+			if (sql_index_column_sort_order(p, nEq) !=
+			    SORT_ORDER_ASC) {
 				/* The roles of pLower and pUpper are swapped for a DESC index */
 				SWAP(pLower, pUpper);
 				SWAP(nBtm, nTop);
@@ -2229,8 +2229,8 @@ whereRangeVectorLen(Parse * pParse,	/* Parsing context */
 		if (pLhs->op != TK_COLUMN
 		    || pLhs->iTable != iCur
 		    || pLhs->iColumn != pIdx->aiColumn[i + nEq]
-		    || pIdx->aSortOrder[i + nEq] != pIdx->aSortOrder[nEq]
-		    ) {
+		    || sql_index_column_sort_order(pIdx, i + nEq) !=
+		       sql_index_column_sort_order(pIdx, nEq)) {
 			break;
 		}
 
@@ -3316,7 +3316,8 @@ wherePathSatisfiesOrderBy(WhereInfo * pWInfo,	/* The WHERE clause */
 				 */
 				if (pIndex) {
 					iColumn = pIndex->aiColumn[j];
-					revIdx = pIndex->aSortOrder[j];
+					revIdx = sql_index_column_sort_order(pIndex,
+									     j);
 					if (iColumn == pIndex->pTable->iPKey)
 						iColumn = -1;
 				} else {
diff --git a/src/box/sql/wherecode.c b/src/box/sql/wherecode.c
index f1112f2..231d690 100644
--- a/src/box/sql/wherecode.c
+++ b/src/box/sql/wherecode.c
@@ -462,7 +462,8 @@ codeEqualityTerm(Parse * pParse,	/* The parsing context */
 		int nEq = 0;
 		int *aiMap = 0;
 
-		if (pLoop->pIndex != 0 && pLoop->pIndex->aSortOrder[iEq]) {
+		if (pLoop->pIndex != 0 &&
+		    sql_index_column_sort_order(pLoop->pIndex, iEq)) {
 			testcase(iEq == 0);
 			testcase(bRev);
 			bRev = !bRev;
@@ -1296,12 +1297,12 @@ sqlite3WhereCodeOneLoopStart(WhereInfo * pWInfo,	/* Complete information about t
 				 */
 				testcase(bRev);
 				testcase(pIdx->aSortOrder[nEq] ==
-					 SQLITE_SO_DESC);
+					 SORT_ORDER_DESC);
 				assert((bRev & ~1) == 0);
 				pLevel->iLikeRepCntr <<= 1;
 				pLevel->iLikeRepCntr |=
-				    bRev ^ (pIdx->aSortOrder[nEq] ==
-					    SQLITE_SO_DESC);
+					bRev ^ (sql_index_column_sort_order(pIdx, nEq) ==
+						SORT_ORDER_DESC);
 			}
 #endif
 			if (pRangeStart == 0) {
@@ -1320,7 +1321,8 @@ sqlite3WhereCodeOneLoopStart(WhereInfo * pWInfo,	/* Complete information about t
 		 * start and end terms (pRangeStart and pRangeEnd).
 		 */
 		if ((nEq < nIdxCol &&
-		     bRev == (pIdx->aSortOrder[nEq] == SQLITE_SO_ASC)) ||
+		     bRev == (sql_index_column_sort_order(pIdx, nEq) ==
+			      SORT_ORDER_ASC)) ||
 		    (bRev && nIdxCol == nEq)) {
 			SWAP(pRangeEnd, pRangeStart);
 			SWAP(bSeekPastNull, bStopAtNull);