[PATCH] sql: do not store key_def in VDBE op

[Vladimir Davydov <vdavydov.dev@gmail.com>]

The parser must not deal with internal Tarantool objects, such as space,
index, or key_def, directly, because it (a) violates encapsulation,
turning the code into a convoluted mess, and (b) makes it impossible to
run the parser and VDBE on different instances, which might come in
handy for cluster SQL implementation. Instead, it should store plain
names in the generated VDBE code. It may use objects the sole purpose
of which is to represent object definitions, such as key_part_def or
field_def, though.

This patch does a tiny step in this direction. It replaces key_def with
sql_key_info in VDBE arguments. The new structure is a trivial wrapper
around an array of key_part_def. It is ref-countable, just like its
predecessor KeyInfo, so as to avoid unnecessary memory duplication.
Since key_def spread roots deeply into the parser implementation, the
new structure has two extra methods:

  sql_key_info_new_from_key_def
  sql_key_info_to_key_def

so that it can be converted to/from a key definition. Note, the latter
caches the result so as not to create a new key definition on subsequent
function calls.

This partially undoes the work done by commit 501c6e28b095 ("sql:
replace KeyInfo with key_def").

The reason why I'm doing this now is that I want to dispose of the
key_def_set_part function, which is used extensively by the parser,
because the latter stores key_def directly in VDBE op. This function
has a vague semantic and rather obscures construction of a key
definition. It will get especially nasty once JSON path indexes are
introduced in Tarantool core. The new struct, sql_key_info, allows us
to get rid of most of those calls.
---
https://github.com/tarantool/tarantool/tree/merge-2.0

 src/box/sql.c              |  11 +-
 src/box/sql/delete.c       |  17 ++-
 src/box/sql/expr.c         |  39 ++----
 src/box/sql/select.c       | 300 ++++++++++++++++++++++++++-------------------
 src/box/sql/sqliteInt.h    |  59 ++++++++-
 src/box/sql/tarantoolInt.h |   2 +-
 src/box/sql/vdbe.c         |  14 ++-
 src/box/sql/vdbe.h         |   6 +-
 src/box/sql/vdbeaux.c      |  22 ++--
 9 files changed, 280 insertions(+), 190 deletions(-)

diff --git a/src/box/sql.c b/src/box/sql.c
index 0a2c2b23d47e..e97423582ecf 100644
--- a/src/box/sql.c
+++ b/src/box/sql.c
@@ -364,17 +364,24 @@ int tarantoolSqlite3Count(BtCursor *pCur, i64 *pnEntry)
  *
  * @param pCur Cursor which will point to the new ephemeral space.
  * @param field_count Number of fields in ephemeral space.
- * @param def Keys description for new ephemeral space.
+ * @param key_info Keys description for new ephemeral space.
  *
  * @retval SQLITE_OK on success, SQLITE_TARANTOOL_ERROR otherwise.
  */
 int
 tarantoolSqlite3EphemeralCreate(BtCursor *pCur, uint32_t field_count,
-				struct key_def *def)
+				struct sql_key_info *key_info)
 {
 	assert(pCur);
 	assert(pCur->curFlags & BTCF_TEphemCursor);
 
+	struct key_def *def = NULL;
+	if (key_info != NULL) {
+		def = sql_key_info_to_key_def(key_info);
+		if (def == NULL)
+			return SQL_TARANTOOL_ERROR;
+	}
+
 	struct key_def *ephemer_key_def = key_def_new(field_count);
 	if (ephemer_key_def == NULL)
 		return SQL_TARANTOOL_ERROR;
diff --git a/src/box/sql/delete.c b/src/box/sql/delete.c
index 08469946ada9..9aa705847630 100644
--- a/src/box/sql/delete.c
+++ b/src/box/sql/delete.c
@@ -240,7 +240,7 @@ sql_table_delete_from(struct Parse *parse, struct SrcList *tab_list,
 		 * is held in ephemeral table, there is no PK for
 		 * it, so columns should be loaded manually.
 		 */
-		struct key_def *pk_def = NULL;
+		struct sql_key_info *pk_info = NULL;
 		int reg_pk = parse->nMem + 1;
 		int pk_len;
 		int eph_cursor = parse->nTab++;
@@ -252,16 +252,15 @@ sql_table_delete_from(struct Parse *parse, struct SrcList *tab_list,
 					  eph_cursor, pk_len);
 		} else {
                         assert(space->index_count > 0);
-                        pk_def = key_def_dup(space->index[0]->def->key_def);
-                        if(pk_def == NULL) {
-                                sqlite3OomFault(parse->db);
+                        pk_info = sql_key_info_new_from_key_def(db,
+					space->index[0]->def->key_def);
+                        if (pk_info == NULL)
                                 goto delete_from_cleanup;
-                        }
-                        pk_len = pk_def->part_count;
+                        pk_len = pk_info->part_count;
                         parse->nMem += pk_len;
                         sqlite3VdbeAddOp4(v, OP_OpenTEphemeral, eph_cursor,
                                           pk_len, 0,
-                                          (char *)pk_def, P4_KEYDEF);
+                                          (char *)pk_info, P4_KEYINFO);
 		}
 
 		/* Construct a query to find the primary key for
@@ -299,7 +298,7 @@ sql_table_delete_from(struct Parse *parse, struct SrcList *tab_list,
 
 		/* Extract the primary key for the current row */
 		if (!is_view) {
-			struct key_part *part = pk_def->parts;
+			struct key_part_def *part = pk_info->parts;
 			for (int i = 0; i < pk_len; i++, part++) {
 				struct space_def *def = space->def;
 				sqlite3ExprCodeGetColumnOfTable(v, def,
@@ -385,7 +384,7 @@ sql_table_delete_from(struct Parse *parse, struct SrcList *tab_list,
 		if (one_pass != ONEPASS_OFF) {
 			/* OP_Found will use an unpacked key. */
 			assert(key_len == pk_len);
-			assert(pk_def != NULL || table->def->opts.is_view);
+			assert(pk_info != NULL || table->def->opts.is_view);
 			sqlite3VdbeAddOp4Int(v, OP_NotFound, tab_cursor,
 					     addr_bypass, reg_key, key_len);
 
diff --git a/src/box/sql/expr.c b/src/box/sql/expr.c
index 8063a60591d0..a13de4f0e0b8 100644
--- a/src/box/sql/expr.c
+++ b/src/box/sql/expr.c
@@ -2716,11 +2716,9 @@ sqlite3CodeSubselect(Parse * pParse,	/* Parsing context */
 			pExpr->is_ephemeral = 1;
 			addr = sqlite3VdbeAddOp2(v, OP_OpenTEphemeral,
 						 pExpr->iTable, nVal);
-			struct key_def *key_def = key_def_new(nVal);
-			if (key_def == NULL) {
-				sqlite3OomFault(pParse->db);
+			struct sql_key_info *key_info = sql_key_info_new(pParse->db, nVal);
+			if (key_info == NULL)
 				return 0;
-			}
 
 			if (ExprHasProperty(pExpr, EP_xIsSelect)) {
 				/* Case 1:     expr IN (SELECT ...)
@@ -2750,7 +2748,7 @@ sqlite3CodeSubselect(Parse * pParse,	/* Parsing context */
 					    (pParse, pSelect, &dest)) {
 						sqlite3DbFree(pParse->db,
 							      dest.zAffSdst);
-						key_def_delete(key_def);
+						sql_key_info_unref(key_info);
 						return 0;
 					}
 					sqlite3DbFree(pParse->db,
@@ -2761,19 +2759,9 @@ sqlite3CodeSubselect(Parse * pParse,	/* Parsing context */
 						Expr *p =
 						    sqlite3VectorFieldSubexpr
 						    (pLeft, i);
-
-						uint32_t id;
-						struct coll *coll =
-							sql_binary_compare_coll_seq(
-								pParse, p,
-								pEList->a[i].pExpr,
-								&id);
-
-						key_def_set_part(key_def, i, i,
-								 FIELD_TYPE_SCALAR,
-								 ON_CONFLICT_ACTION_ABORT,
-								 coll, id,
-								 SORT_ORDER_ASC);
+						sql_binary_compare_coll_seq(pParse, p,
+							pEList->a[i].pExpr,
+							&key_info->parts[i].coll_id);
 					}
 				}
 			} else if (ALWAYS(pExpr->x.pList != 0)) {
@@ -2795,15 +2783,8 @@ sqlite3CodeSubselect(Parse * pParse,	/* Parsing context */
 					affinity = AFFINITY_BLOB;
 				}
 				bool unused;
-				uint32_t id;
-				struct coll *coll =
-					sql_expr_coll(pParse, pExpr->pLeft,
-						      &unused, &id);
-
-				key_def_set_part(key_def, 0, 0,
-						 FIELD_TYPE_SCALAR,
-						 ON_CONFLICT_ACTION_ABORT, coll,
-						 id, SORT_ORDER_ASC);
+				sql_expr_coll(pParse, pExpr->pLeft,
+					      &unused, &key_info->parts[0].coll_id);
 
 				/* Loop through each expression in <exprlist>. */
 				r1 = sqlite3GetTempReg(pParse);
@@ -2833,8 +2814,8 @@ sqlite3CodeSubselect(Parse * pParse,	/* Parsing context */
 				sqlite3ReleaseTempReg(pParse, r1);
 				sqlite3ReleaseTempReg(pParse, r2);
 			}
-			sqlite3VdbeChangeP4(v, addr, (void *)key_def,
-					    P4_KEYDEF);
+			sqlite3VdbeChangeP4(v, addr, (void *)key_info,
+					    P4_KEYINFO);
 			break;
 		}
 
diff --git a/src/box/sql/select.c b/src/box/sql/select.c
index d6b4f82753a4..6685ee6d6f4e 100644
--- a/src/box/sql/select.c
+++ b/src/box/sql/select.c
@@ -642,12 +642,12 @@ sqliteProcessJoin(Parse * pParse, Select * p)
 }
 
 /**
- * Given an expression list, generate a key_def structure that
+ * Given an expression list, generate a key_info structure that
  * records the collating sequence for each expression in that
  * expression list.
  *
  * If the ExprList is an ORDER BY or GROUP BY clause then the
- * resulting key_def structure is appropriate for initializing
+ * resulting key_info structure is appropriate for initializing
  * a virtual index to implement that clause.  If the ExprList is
  * the result set of a SELECT then the key_info structure is
  * appropriate for initializing a virtual index to implement a
@@ -661,10 +661,10 @@ sqliteProcessJoin(Parse * pParse, Select * p)
  * @param list Expression list.
  * @param start No of leading parts to skip.
  *
- * @retval Allocated key_def, NULL in case of OOM.
+ * @retval Allocated key_info, NULL in case of OOM.
  */
-static struct key_def *
-sql_expr_list_to_key_def(struct Parse *parse, struct ExprList *list, int start);
+static struct sql_key_info *
+sql_expr_list_to_key_info(struct Parse *parse, struct ExprList *list, int start);
 
 
 /*
@@ -740,17 +740,13 @@ pushOntoSorter(Parse * pParse,		/* Parser context */
 		if (pParse->db->mallocFailed)
 			return;
 		pOp->p2 = nKey + nData;
-		struct key_def *def = key_def_dup(pOp->p4.key_def);
-		if (def == NULL) {
-			sqlite3OomFault(pParse->db);
-			return;
-		}
-		for (uint32_t i = 0; i < def->part_count; ++i)
-			pOp->p4.key_def->parts[i].sort_order = SORT_ORDER_ASC;
-		sqlite3VdbeChangeP4(v, -1, (char *)def, P4_KEYDEF);
-		pOp->p4.key_def = sql_expr_list_to_key_def(pParse,
-							   pSort->pOrderBy,
-							   nOBSat);
+		struct sql_key_info *key_info = pOp->p4.key_info;
+		for (uint32_t i = 0; i < key_info->part_count; i++)
+			key_info->parts[i].sort_order = SORT_ORDER_ASC;
+		sqlite3VdbeChangeP4(v, -1, (char *)key_info, P4_KEYINFO);
+		pOp->p4.key_info = sql_expr_list_to_key_info(pParse,
+							     pSort->pOrderBy,
+							     nOBSat);
 		addrJmp = sqlite3VdbeCurrentAddr(v);
 		sqlite3VdbeAddOp3(v, OP_Jump, addrJmp + 1, 0, addrJmp + 1);
 		VdbeCoverage(v);
@@ -1292,26 +1288,103 @@ selectInnerLoop(Parse * pParse,		/* The parser context */
 	}
 }
 
-static struct key_def *
-sql_expr_list_to_key_def(struct Parse *parse, struct ExprList *list, int start)
+static inline size_t
+sql_key_info_sizeof(uint32_t part_count)
 {
-	int expr_count = list->nExpr;
-	struct key_def *def = key_def_new(expr_count);
-	if (def == NULL) {
-		sqlite3OomFault(parse->db);
+	return sizeof(struct sql_key_info) +
+		part_count * sizeof(struct key_part_def);
+}
+
+struct sql_key_info *
+sql_key_info_new(sqlite3 *db, uint32_t part_count)
+{
+	struct sql_key_info *key_info = sqlite3DbMallocRawNN(db,
+				sql_key_info_sizeof(part_count));
+	if (key_info == NULL) {
+		sqlite3OomFault(db);
+		return NULL;
+	}
+	key_info->db = db;
+	key_info->key_def = NULL;
+	key_info->refs = 1;
+	key_info->part_count = part_count;
+	for (uint32_t i = 0; i < part_count; i++) {
+		struct key_part_def *part = &key_info->parts[i];
+		part->fieldno = i;
+		part->type = FIELD_TYPE_SCALAR;
+		part->coll_id = COLL_NONE;
+		part->is_nullable = false;
+		part->nullable_action = ON_CONFLICT_ACTION_ABORT;
+		part->sort_order = SORT_ORDER_ASC;
+	}
+	return key_info;
+}
+
+struct sql_key_info *
+sql_key_info_new_from_key_def(sqlite3 *db, const struct key_def *key_def)
+{
+	struct sql_key_info *key_info = sqlite3DbMallocRawNN(db,
+				sql_key_info_sizeof(key_def->part_count));
+	if (key_info == NULL) {
+		sqlite3OomFault(db);
 		return NULL;
 	}
+	key_info->db = db;
+	key_info->key_def = NULL;
+	key_info->refs = 1;
+	key_info->part_count = key_def->part_count;
+	key_def_dump_parts(key_def, key_info->parts);
+	return key_info;
+}
+
+struct sql_key_info *
+sql_key_info_ref(struct sql_key_info *key_info)
+{
+	assert(key_info->refs > 0);
+	key_info->refs++;
+	return key_info;
+}
+
+void
+sql_key_info_unref(struct sql_key_info *key_info)
+{
+	if (key_info == NULL)
+		return;
+	assert(key_info->refs > 0);
+	if (--key_info->refs == 0) {
+		if (key_info->key_def != NULL)
+			key_def_delete(key_info->key_def);
+		sqlite3DbFree(key_info->db, key_info);
+	}
+}
+
+struct key_def *
+sql_key_info_to_key_def(struct sql_key_info *key_info)
+{
+	if (key_info->key_def == NULL) {
+		key_info->key_def = key_def_new_with_parts(key_info->parts,
+							   key_info->part_count);
+	}
+	return key_info->key_def;
+}
+
+static struct sql_key_info *
+sql_expr_list_to_key_info(struct Parse *parse, struct ExprList *list, int start)
+{
+	int expr_count = list->nExpr;
+	struct sql_key_info *key_info = sql_key_info_new(parse->db, expr_count);
+	if (key_info == NULL)
+		return NULL;
 	struct ExprList_item *item = list->a + start;
 	for (int i = start; i < expr_count; ++i, ++item) {
+		struct key_part_def *part = &key_info->parts[i - start];
 		bool unused;
 		uint32_t id;
-		struct coll *coll =
-			sql_expr_coll(parse, item->pExpr, &unused, &id);
-		key_def_set_part(def, i-start, i-start, FIELD_TYPE_SCALAR,
-				 ON_CONFLICT_ACTION_ABORT, coll, id,
-				 item->sort_order);
+		sql_expr_coll(parse, item->pExpr, &unused, &id);
+		part->coll_id = id;
+		part->sort_order = item->sort_order;
 	}
-	return def;
+	return key_info;
 }
 
 /*
@@ -2196,10 +2269,10 @@ multi_select_coll_seq(Parse *parser, Select *p, int n, uint32_t *coll_id)
 /**
  * The select statement passed as the second parameter is a
  * compound SELECT with an ORDER BY clause. This function
- * allocates and returns a key_def structure suitable for
+ * allocates and returns a sql_key_info structure suitable for
  * implementing the ORDER BY.
  *
- * Space to hold the key_def structure is obtained from malloc.
+ * Space to hold the sql_key_info structure is obtained from malloc.
  * The calling function is responsible for ensuring that this
  * structure is eventually freed.
  *
@@ -2207,33 +2280,33 @@ multi_select_coll_seq(Parse *parser, Select *p, int n, uint32_t *coll_id)
  * @param s Select struct to analyze.
  * @param extra No of extra slots to allocate.
  *
- * @retval Allocated key_def, NULL in case of OOM.
+ * @retval Allocated key_info, NULL in case of OOM.
  */
-static struct key_def *
-sql_multiselect_orderby_to_key_def(struct Parse *parse, struct Select *s,
+static struct sql_key_info *
+sql_multiselect_orderby_to_key_info(struct Parse *parse, struct Select *s,
 				   int extra)
 {
 	int ob_count = s->pOrderBy->nExpr;
-	struct key_def *key_def = key_def_new(ob_count + extra);
-	if (key_def == NULL) {
+	struct sql_key_info *key_info = sql_key_info_new(parse->db,
+							 ob_count + extra);
+	if (key_info == NULL) {
 		sqlite3OomFault(parse->db);
 		return NULL;
 	}
 
 	ExprList *order_by = s->pOrderBy;
 	for (int i = 0; i < ob_count; i++) {
+		struct key_part_def *part = &key_info->parts[i];
 		struct ExprList_item *item = &order_by->a[i];
 		struct Expr *term = item->pExpr;
-		struct coll *coll;
 		uint32_t id;
 		if ((term->flags & EP_Collate) != 0) {
 			bool is_found = false;
-			coll = sql_expr_coll(parse, term, &is_found, &id);
+			sql_expr_coll(parse, term, &is_found, &id);
 		} else {
-			coll = multi_select_coll_seq(parse, s,
-						     item->u.x.iOrderByCol - 1,
-						     &id);
-			if (coll != NULL) {
+			multi_select_coll_seq(parse, s,
+					      item->u.x.iOrderByCol - 1, &id);
+			if (id != COLL_NONE) {
 				const char *name = coll_by_id(id)->name;
 				order_by->a[i].pExpr =
 					sqlite3ExprAddCollateString(parse, term,
@@ -2244,12 +2317,11 @@ sql_multiselect_orderby_to_key_def(struct Parse *parse, struct Select *s,
 								    "BINARY");
 			}
 		}
-		key_def_set_part(key_def, i, i, FIELD_TYPE_SCALAR,
-				 ON_CONFLICT_ACTION_ABORT, coll, id,
-				 order_by->a[i].sort_order);
+		part->coll_id = id;
+		part->sort_order = order_by->a[i].sort_order;
 	}
 
-	return key_def;
+	return key_info;
 }
 
 #ifndef SQLITE_OMIT_CTE
@@ -2349,11 +2421,11 @@ generateWithRecursiveQuery(Parse * pParse,	/* Parsing context */
 	regCurrent = ++pParse->nMem;
 	sqlite3VdbeAddOp3(v, OP_OpenPseudo, iCurrent, regCurrent, nCol);
 	if (pOrderBy) {
-		struct key_def *def = sql_multiselect_orderby_to_key_def(pParse,
-									 p, 1);
+		struct sql_key_info *key_info =
+			sql_multiselect_orderby_to_key_info(pParse, p, 1);
 		sqlite3VdbeAddOp4(v, OP_OpenTEphemeral, iQueue,
-				  pOrderBy->nExpr + 2, 0, (char *)def,
-				  P4_KEYDEF);
+				  pOrderBy->nExpr + 2, 0, (char *)key_info,
+				  P4_KEYINFO);
 		VdbeComment((v, "Orderby table"));
 		destQueue.pOrderBy = pOrderBy;
 	} else {
@@ -2866,7 +2938,7 @@ multiSelect(Parse * pParse,	/* Parsing context */
 
 	/* Compute collating sequences used by
 	 * temporary tables needed to implement the compound select.
-	 * Attach the key_def structure to all temporary tables.
+	 * Attach the key_info structure to all temporary tables.
 	 *
 	 * This section is run by the right-most SELECT statement only.
 	 * SELECT statements to the left always skip this part.  The right-most
@@ -2876,19 +2948,12 @@ multiSelect(Parse * pParse,	/* Parsing context */
 	if (p->selFlags & SF_UsesEphemeral) {
 		assert(p->pNext == NULL);
 		int nCol = p->pEList->nExpr;
-		struct key_def *key_def = key_def_new(nCol);
-		if (key_def == NULL) {
-			sqlite3OomFault(db);
+		struct sql_key_info *key_info = sql_key_info_new(db, nCol);
+		if (key_info == NULL)
 			goto multi_select_end;
-		}
 		for (int i = 0; i < nCol; i++) {
-			uint32_t id;
-			struct coll *coll = multi_select_coll_seq(pParse, p, i,
-								  &id);
-			key_def_set_part(key_def, i, i,
-					 FIELD_TYPE_SCALAR,
-					 ON_CONFLICT_ACTION_ABORT, coll, id,
-					 SORT_ORDER_ASC);
+			multi_select_coll_seq(pParse, p, i,
+					      &key_info->parts[i].coll_id);
 		}
 
 		for (struct Select *pLoop = p; pLoop; pLoop = pLoop->pPrior) {
@@ -2902,19 +2967,13 @@ multiSelect(Parse * pParse,	/* Parsing context */
 					break;
 				}
 				sqlite3VdbeChangeP2(v, addr, nCol);
-				struct key_def *dup_def = key_def_dup(key_def);
-				if (dup_def == NULL) {
-					free(key_def);
-					sqlite3OomFault(db);
-					goto multi_select_end;
-				}
-
-				sqlite3VdbeChangeP4(v, addr, (char *)dup_def,
-						    P4_KEYDEF);
+				sqlite3VdbeChangeP4(v, addr,
+						    (char *)sql_key_info_ref(key_info),
+						    P4_KEYINFO);
 				pLoop->addrOpenEphm[i] = -1;
 			}
 		}
-		free(key_def);
+		sql_key_info_unref(key_info);
 	}
 
  multi_select_end:
@@ -2952,7 +3011,7 @@ sqlite3SelectWrongNumTermsError(struct Parse *parse, struct Select * p)
  * If regPrev>0 then it is the first register in a vector that
  * records the previous output.  mem[regPrev] is a flag that is
  * false if there has been no previous output.  If regPrev>0 then
- * code is generated to suppress duplicates.  def is used for
+ * code is generated to suppress duplicates. key_info is used for
  * comparing keys.
  *
  * If the LIMIT found in p->iLimit is reached, jump immediately to
@@ -2964,7 +3023,7 @@ sqlite3SelectWrongNumTermsError(struct Parse *parse, struct Select * p)
  * @param dest Where to send the data.
  * @param reg_ret The return address register.
  * @param reg_prev Previous result register.  No uniqueness if 0.
- * @param def For comparing with previous entry.
+ * @param key_info For comparing with previous entry.
  * @param break_addr Jump here if we hit the LIMIT.
  *
  * @retval Address of generated routine.
@@ -2972,7 +3031,8 @@ sqlite3SelectWrongNumTermsError(struct Parse *parse, struct Select * p)
 static int
 generateOutputSubroutine(struct Parse *parse, struct Select *p,
 			 struct SelectDest *in, struct SelectDest *dest,
-			 int reg_ret, int reg_prev, const struct key_def *def,
+			 int reg_ret, int reg_prev,
+			 struct sql_key_info *key_info,
 			 int break_addr)
 {
 	Vdbe *v = parse->pVdbe;
@@ -2988,16 +3048,11 @@ generateOutputSubroutine(struct Parse *parse, struct Select *p,
 		int addr1, addr2;
 		addr1 = sqlite3VdbeAddOp1(v, OP_IfNot, reg_prev);
 		VdbeCoverage(v);
-		struct key_def *dup_def = key_def_dup(def);
-		if (dup_def == NULL) {
-			sqlite3OomFault(parse->db);
-			return 0;
-		}
 		addr2 =
 		    sqlite3VdbeAddOp4(v, OP_Compare, in->iSdst, reg_prev + 1,
 				      in->nSdst,
-				      (char *)dup_def,
-				      P4_KEYDEF);
+				      (char *)sql_key_info_ref(key_info),
+				      P4_KEYINFO);
 		sqlite3VdbeAddOp3(v, OP_Jump, addr2 + 2, iContinue, addr2 + 2);
 		VdbeCoverage(v);
 		sqlite3VdbeJumpHere(v, addr1);
@@ -3229,9 +3284,9 @@ multiSelectOrderBy(Parse * pParse,	/* Parsing context */
 	int addr1;		/* Jump instructions that get retargetted */
 	int op;			/* One of TK_ALL, TK_UNION, TK_EXCEPT, TK_INTERSECT */
 	/* Comparison information for duplicate removal */
-	struct key_def *def_dup = NULL;
+	struct sql_key_info *key_info_dup = NULL;
 	/* Comparison information for merging rows */
-	struct key_def *def_merge;
+	struct sql_key_info *key_info_merge;
 	sqlite3 *db;		/* Database connection */
 	ExprList *pOrderBy;	/* The ORDER BY clause */
 	int nOrderBy;		/* Number of terms in the ORDER BY clause */
@@ -3283,7 +3338,7 @@ multiSelectOrderBy(Parse * pParse,	/* Parsing context */
 		}
 	}
 
-	/* Compute the comparison permutation and key_def that is used with
+	/* Compute the comparison permutation and key_info that is used with
 	 * the permutation used to determine if the next
 	 * row of results comes from selectA or selectB.  Also add explicit
 	 * collations to the ORDER BY clause terms so that when the subqueries
@@ -3299,9 +3354,10 @@ multiSelectOrderBy(Parse * pParse,	/* Parsing context */
 			assert(pItem->u.x.iOrderByCol <= p->pEList->nExpr);
 			aPermute[i] = pItem->u.x.iOrderByCol - 1;
 		}
-		def_merge = sql_multiselect_orderby_to_key_def(pParse, p, 1);
+		key_info_merge = sql_multiselect_orderby_to_key_info(pParse,
+								     p, 1);
 	} else {
-		def_merge = NULL;
+		key_info_merge = NULL;
 	}
 
 	/* Reattach the ORDER BY clause to the query.
@@ -3309,7 +3365,7 @@ multiSelectOrderBy(Parse * pParse,	/* Parsing context */
 	p->pOrderBy = pOrderBy;
 	pPrior->pOrderBy = sql_expr_list_dup(pParse->db, pOrderBy, 0);
 
-	/* Allocate a range of temporary registers and the key_def needed
+	/* Allocate a range of temporary registers and the key_info needed
 	 * for the logic that removes duplicate result rows when the
 	 * operator is UNION, EXCEPT, or INTERSECT (but not UNION ALL).
 	 */
@@ -3321,20 +3377,12 @@ multiSelectOrderBy(Parse * pParse,	/* Parsing context */
 		regPrev = pParse->nMem + 1;
 		pParse->nMem += expr_count + 1;
 		sqlite3VdbeAddOp2(v, OP_Integer, 0, regPrev);
-		def_dup = key_def_new(expr_count);
-		if (def_dup != NULL) {
+		key_info_dup = sql_key_info_new(db, expr_count);
+		if (key_info_dup != NULL) {
 			for (int i = 0; i < expr_count; i++) {
-				uint32_t id;
-				struct coll *coll =
-					multi_select_coll_seq(pParse, p, i,
-							      &id);
-				key_def_set_part(def_dup, i, i,
-						 FIELD_TYPE_SCALAR,
-						 ON_CONFLICT_ACTION_ABORT, coll,
-						 id, SORT_ORDER_ASC);
+				multi_select_coll_seq(pParse, p, i,
+					&key_info_dup->parts[i].coll_id);
 			}
-		} else {
-			sqlite3OomFault(db);
 		}
 	}
 
@@ -3408,7 +3456,7 @@ multiSelectOrderBy(Parse * pParse,	/* Parsing context */
 	VdbeNoopComment((v, "Output routine for A"));
 	addrOutA = generateOutputSubroutine(pParse,
 					    p, &destA, pDest, regOutA,
-					    regPrev, def_dup, labelEnd);
+					    regPrev, key_info_dup, labelEnd);
 
 	/* Generate a subroutine that outputs the current row of the B
 	 * select as the next output row of the compound select.
@@ -3417,10 +3465,11 @@ multiSelectOrderBy(Parse * pParse,	/* Parsing context */
 		VdbeNoopComment((v, "Output routine for B"));
 		addrOutB = generateOutputSubroutine(pParse,
 						    p, &destB, pDest, regOutB,
-						    regPrev, def_dup, labelEnd);
+						    regPrev, key_info_dup,
+						    labelEnd);
 	}
 
-	key_def_delete(def_dup);
+	sql_key_info_unref(key_info_dup);
 
 	/* Generate a subroutine to run when the results from select A
 	 * are exhausted and only data in select B remains.
@@ -3500,7 +3549,7 @@ multiSelectOrderBy(Parse * pParse,	/* Parsing context */
 	sqlite3VdbeAddOp4(v, OP_Permutation, 0, 0, 0, (char *)aPermute,
 			  P4_INTARRAY);
 	sqlite3VdbeAddOp4(v, OP_Compare, destA.iSdst, destB.iSdst, nOrderBy,
-			  (char *)def_merge, P4_KEYDEF);
+			  (char *)key_info_merge, P4_KEYINFO);
 	sqlite3VdbeChangeP5(v, OPFLAG_PERMUTE);
 	sqlite3VdbeAddOp3(v, OP_Jump, addrAltB, addrAeqB, addrAgtB);
 	VdbeCoverage(v);
@@ -5233,13 +5282,13 @@ resetAccumulator(Parse * pParse, AggInfo * pAggInfo)
 						"argument");
 				pFunc->iDistinct = -1;
 			} else {
-				struct key_def *def =
-					sql_expr_list_to_key_def(pParse,
-								 pE->x.pList,
-								 0);
+				struct sql_key_info *key_info =
+					sql_expr_list_to_key_info(pParse,
+								  pE->x.pList,
+								  0);
 				sqlite3VdbeAddOp4(v, OP_OpenTEphemeral,
 						  pFunc->iDistinct, 1, 0,
-						  (char *)def, P4_KEYDEF);
+						  (char *)key_info, P4_KEYINFO);
 			}
 		}
 	}
@@ -5745,22 +5794,22 @@ sqlite3Select(Parse * pParse,		/* The parser context */
 	 * that change.
 	 */
 	if (sSort.pOrderBy) {
-		struct key_def *def =
-			sql_expr_list_to_key_def(pParse, sSort.pOrderBy, 0);
+		struct sql_key_info *key_info =
+			sql_expr_list_to_key_info(pParse, sSort.pOrderBy, 0);
 		sSort.iECursor = pParse->nTab++;
 		/* Number of columns in transient table equals to number of columns in
 		 * SELECT statement plus number of columns in ORDER BY statement
 		 * and plus one column for ID.
 		 */
 		int nCols = pEList->nExpr + sSort.pOrderBy->nExpr + 1;
-		if (def->parts[0].sort_order == SORT_ORDER_DESC) {
+		if (key_info->parts[0].sort_order == SORT_ORDER_DESC) {
 			sSort.sortFlags |= SORTFLAG_DESC;
 		}
 		sSort.addrSortIndex =
 		    sqlite3VdbeAddOp4(v, OP_OpenTEphemeral,
 				      sSort.iECursor,
 				      nCols,
-				      0, (char *)def, P4_KEYDEF);
+				      0, (char *)key_info, P4_KEYINFO);
 		VdbeComment((v, "Sort table"));
 	} else {
 		sSort.addrSortIndex = -1;
@@ -5791,12 +5840,13 @@ sqlite3Select(Parse * pParse,		/* The parser context */
 	 */
 	if (p->selFlags & SF_Distinct) {
 		sDistinct.tabTnct = pParse->nTab++;
-		struct key_def *def = sql_expr_list_to_key_def(pParse, p->pEList, 0);
+		struct sql_key_info *key_info =
+			sql_expr_list_to_key_info(pParse, p->pEList, 0);
 		sDistinct.addrTnct = sqlite3VdbeAddOp4(v, OP_OpenTEphemeral,
 						       sDistinct.tabTnct,
-						       def->part_count,
-						       0, (char *)def,
-						       P4_KEYDEF);
+						       key_info->part_count,
+						       0, (char *)key_info,
+						       P4_KEYINFO);
 		VdbeComment((v, "Distinct table"));
 		sDistinct.eTnctType = WHERE_DISTINCT_UNORDERED;
 	} else {
@@ -5952,13 +6002,13 @@ sqlite3Select(Parse * pParse,		/* The parser context */
 			 * will be converted into a Noop.
 			 */
 			sAggInfo.sortingIdx = pParse->nTab++;
-			struct key_def *def =
-				sql_expr_list_to_key_def(pParse, pGroupBy, 0);
+			struct sql_key_info *key_info =
+				sql_expr_list_to_key_info(pParse, pGroupBy, 0);
 			addrSortingIdx =
 			    sqlite3VdbeAddOp4(v, OP_SorterOpen,
 					      sAggInfo.sortingIdx,
 					      sAggInfo.nSortingColumn, 0,
-					      (char *)def, P4_KEYDEF);
+					      (char *)key_info, P4_KEYINFO);
 
 			/* Initialize memory locations used by GROUP BY aggregate processing
 			 */
@@ -5996,7 +6046,7 @@ sqlite3Select(Parse * pParse,		/* The parser context */
 			if (sqlite3WhereIsOrdered(pWInfo) == pGroupBy->nExpr) {
 				/* The optimizer is able to deliver rows in group by order so
 				 * we do not have to sort.  The OP_OpenEphemeral table will be
-				 * cancelled later because we still need to use the key_def
+				 * cancelled later because we still need to use the key_info
 				 */
 				groupBySort = 0;
 			} else {
@@ -6106,14 +6156,10 @@ sqlite3Select(Parse * pParse,		/* The parser context */
 							iBMem + j);
 				}
 			}
-			struct key_def *dup_def = key_def_dup(def);
-			if (dup_def == NULL) {
-				sqlite3OomFault(db);
-				goto select_end;
-			}
 			sqlite3VdbeAddOp4(v, OP_Compare, iAMem, iBMem,
-					  pGroupBy->nExpr, (char*)dup_def,
-					  P4_KEYDEF);
+					  pGroupBy->nExpr,
+					  (char*)sql_key_info_ref(key_info),
+					  P4_KEYINFO);
 			addr1 = sqlite3VdbeCurrentAddr(v);
 			sqlite3VdbeAddOp3(v, OP_Jump, addr1 + 1, 0, addr1 + 1);
 			VdbeCoverage(v);
diff --git a/src/box/sql/sqliteInt.h b/src/box/sql/sqliteInt.h
index f56090dc1fac..53188e74d161 100644
--- a/src/box/sql/sqliteInt.h
+++ b/src/box/sql/sqliteInt.h
@@ -2443,12 +2443,12 @@ struct NameContext {
  *
  * addrOpenEphm[] entries contain the address of OP_OpenEphemeral opcodes.
  * These addresses must be stored so that we can go back and fill in
- * the P4_KEYDEF and P2 parameters later.  Neither the key_def nor
+ * the P4_KEYINFO and P2 parameters later.  Neither the key_info nor
  * the number of columns in P2 can be computed at the same time
  * as the OP_OpenEphm instruction is coded because not
  * enough information about the compound query is known at that point.
- * The key_def for addrOpenTran[0] and [1] contains collating sequences
- * for the result set.  The key_def for addrOpenEphm[2] contains collating
+ * The key_info for addrOpenTran[0] and [1] contains collating sequences
+ * for the result set. The key_info for addrOpenEphm[2] contains collating
  * sequences for the ORDER BY clause.
  */
 struct Select {
@@ -4459,6 +4459,59 @@ sql_index_tuple_size(struct space *space, struct index *idx);
 int
 sql_analysis_load(struct sqlite3 *db);
 
+/**
+ * An instance of the following structure controls how keys
+ * are compared by VDBE, see P4_KEYINFO.
+ */
+struct sql_key_info {
+	sqlite3 *db;
+	/**
+	 * Key definition created from this object,
+	 * see sql_key_info_to_key_def().
+	 */
+	struct key_def *key_def;
+	/** Reference counter. */
+	uint32_t refs;
+	/** Number of parts in the key. */
+	uint32_t part_count;
+	/** Definition of the key parts. */
+	struct key_part_def parts[];
+};
+
+/**
+ * Allocate a key_info object sufficient for an index with
+ * the given number of key columns.
+ */
+struct sql_key_info *
+sql_key_info_new(sqlite3 *db, uint32_t part_count);
+
+/**
+ * Allocate a key_info object from the given key definition.
+ */
+struct sql_key_info *
+sql_key_info_new_from_key_def(sqlite3 *db, const struct key_def *key_def);
+
+/**
+ * Increment the reference counter of a key_info object.
+ */
+struct sql_key_info *
+sql_key_info_ref(struct sql_key_info *key_info);
+
+/**
+ * Decrement the reference counter of a key_info object and
+ * free memory if the object isn't referenced anymore.
+ */
+void
+sql_key_info_unref(struct sql_key_info *key_info);
+
+/**
+ * Create a key definition from a key_info object.
+ * The new key definition is cached in key_info struct
+ * so that subsequent calls to this function are free.
+ */
+struct key_def *
+sql_key_info_to_key_def(struct sql_key_info *key_info);
+
 void sqlite3RegisterLikeFunctions(sqlite3 *, int);
 int sqlite3IsLikeFunction(sqlite3 *, Expr *, int *, char *);
 int sqlite3CreateFunc(sqlite3 *, const char *, int, int, void *,
diff --git a/src/box/sql/tarantoolInt.h b/src/box/sql/tarantoolInt.h
index 1f78cbbc6ac0..8017742b59f7 100644
--- a/src/box/sql/tarantoolInt.h
+++ b/src/box/sql/tarantoolInt.h
@@ -99,7 +99,7 @@ int tarantoolSqlite3RenameTrigger(const char *zTriggerName,
 
 /* Interface for ephemeral tables. */
 int tarantoolSqlite3EphemeralCreate(BtCursor * pCur, uint32_t filed_count,
-				    struct key_def *def);
+				    struct sql_key_info *key_info);
 /**
  * Insert tuple into ephemeral space.
  * In contrast to ordinary spaces, there is no need to create and
diff --git a/src/box/sql/vdbe.c b/src/box/sql/vdbe.c
index 0efc4dd79ec1..00ffb0c5def7 100644
--- a/src/box/sql/vdbe.c
+++ b/src/box/sql/vdbe.c
@@ -2259,12 +2259,14 @@ case OP_Compare: {
 
 	int n = pOp->p3;
 
-	assert(pOp->p4type == P4_KEYDEF);
+	assert(pOp->p4type == P4_KEYINFO);
 	assert(n>0);
 	p1 = pOp->p1;
 	p2 = pOp->p2;
 
-	struct key_def *def = pOp->p4.key_def;
+	struct key_def *def = sql_key_info_to_key_def(pOp->p4.key_info);
+	if (def == NULL)
+		goto no_mem;
 #if SQLITE_DEBUG
 	if (aPermute) {
 		int mx = 0;
@@ -3150,7 +3152,7 @@ case OP_OpenTEphemeral: {
 	BtCursor *pBtCur;
 	assert(pOp->p1 >= 0);
 	assert(pOp->p2 > 0);
-	assert(pOp->p4type != P4_KEYDEF || pOp->p4.key_def != NULL);
+	assert(pOp->p4type != P4_KEYINFO || pOp->p4.key_info != NULL);
 
 	pCx = allocateCursor(p, pOp->p1, pOp->p2, CURTYPE_TARANTOOL);
 	if (pCx == 0) goto no_mem;
@@ -3162,7 +3164,7 @@ case OP_OpenTEphemeral: {
 	pBtCur->curFlags = BTCF_TEphemCursor;
 
 	rc = tarantoolSqlite3EphemeralCreate(pCx->uc.pCursor, pOp->p2,
-					     pOp->p4.key_def);
+					     pOp->p4.key_info);
 	pCx->key_def = pCx->uc.pCursor->index->def->key_def;
 	if (rc) goto abort_due_to_error;
 	break;
@@ -3183,9 +3185,11 @@ case OP_SorterOpen: {
 
 	assert(pOp->p1>=0);
 	assert(pOp->p2>=0);
+	struct key_def *def = sql_key_info_to_key_def(pOp->p4.key_info);
+	if (def == NULL) goto no_mem;
 	pCx = allocateCursor(p, pOp->p1, pOp->p2, CURTYPE_SORTER);
 	if (pCx==0) goto no_mem;
-	pCx->key_def = pOp->p4.key_def;
+	pCx->key_def = def;
 	rc = sqlite3VdbeSorterInit(db, pCx);
 	if (rc) goto abort_due_to_error;
 	break;
diff --git a/src/box/sql/vdbe.h b/src/box/sql/vdbe.h
index a39ddfd52e41..d5da5d5713b1 100644
--- a/src/box/sql/vdbe.h
+++ b/src/box/sql/vdbe.h
@@ -80,8 +80,8 @@ struct VdbeOp {
 		int *ai;	/* Used when p4type is P4_INTARRAY */
 		SubProgram *pProgram;	/* Used when p4type is P4_SUBPROGRAM */
 		int (*xAdvance) (BtCursor *, int *);
-		/** Used when p4type is P4_KEYDEF. */
-		struct key_def *key_def;
+		/** Used when p4type is P4_KEYINFO. */
+		struct sql_key_info *key_info;
 		/** Used when p4type is P4_SPACEPTR. */
 		struct space *space;
 	} p4;
@@ -129,7 +129,7 @@ struct SubProgram {
 #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. */
+#define P4_KEYINFO  (-19)       /* P4 is a pointer to sql_key_info structure. */
 #define P4_SPACEPTR (-20)       /* P4 is a space pointer */
 
 /* Error message codes for OP_Halt */
diff --git a/src/box/sql/vdbeaux.c b/src/box/sql/vdbeaux.c
index c843bc786f71..54edf1b039b6 100644
--- a/src/box/sql/vdbeaux.c
+++ b/src/box/sql/vdbeaux.c
@@ -867,8 +867,8 @@ freeP4(sqlite3 * db, int p4type, void *p4)
 			sqlite3DbFree(db, p4);
 			break;
 		}
-	case P4_KEYDEF:
-		key_def_delete(p4);
+	case P4_KEYINFO:
+		sql_key_info_unref(p4);
 		break;
 	case P4_FUNCDEF:{
 			freeEphemeralFunction(db, (FuncDef *) p4);
@@ -1054,11 +1054,10 @@ sql_vdbe_set_p4_key_def(struct Parse *parse, struct key_def *key_def)
 	struct Vdbe *v = parse->pVdbe;
 	assert(v != NULL);
 	assert(key_def != NULL);
-	key_def = key_def_dup(key_def);
-	if (key_def == NULL)
-		sqlite3OomFault(parse->db);
-	else
-		sqlite3VdbeAppendP4(v, key_def, P4_KEYDEF);
+	struct sql_key_info *key_info =
+		sql_key_info_new_from_key_def(parse->db, key_def);
+	if (key_info != NULL)
+		sqlite3VdbeAppendP4(v, key_info, P4_KEYINFO);
 }
 
 #ifdef SQLITE_ENABLE_EXPLAIN_COMMENTS
@@ -1290,12 +1289,13 @@ displayP4(Op * pOp, char *zTemp, int nTemp)
 	assert(nTemp >= 20);
 	sqlite3StrAccumInit(&x, 0, zTemp, nTemp, 0);
 	switch (pOp->p4type) {
-	case P4_KEYDEF:{
-
-			if (pOp->p4.key_def == NULL) {
+	case P4_KEYINFO:{
+			struct key_def *def = NULL;
+			if (pOp->p4.key_info != NULL)
+				def = sql_key_info_to_key_def(pOp->p4.key_info);
+			if (def == NULL) {
 				sqlite3XPrintf(&x, "k[NULL]");
 			} else {
-				struct key_def *def = pOp->p4.key_def;
 				sqlite3XPrintf(&x, "k(%d", def->part_count);
 				for (int j = 0; j < (int)def->part_count; j++) {
 					struct coll *coll = def->parts[j].coll;
-- 
2.11.0