[tarantool-patches] [PATCH 4/4] sql: replace pgnoRoot with struct space in BtCursor

[Nikita Pettik <korablev@tarantool.org>]

Instead of passing encoded space id and index id to SQL bindings,
pointers to space and index are saved in cursor and passed implicitly.
Space and index lookups appear during execution of
OP_OpenRead/OP_OpenWrite once. Moreover, having struct space it has
become possible to remove several wrapper-function calls on insertions
and deletions by invoking sql_execute_dml().

Closes #3122
---
 src/box/sql.c              | 87 +++++++++++++++++++++++-----------------------
 src/box/sql/cursor.h       |  3 +-
 src/box/sql/tarantoolInt.h |  2 +-
 src/box/sql/vdbe.c         | 24 ++++++-------
 src/box/sql/vdbeInt.h      |  1 -
 5 files changed, 58 insertions(+), 59 deletions(-)

diff --git a/src/box/sql.c b/src/box/sql.c
index 4db253ff4..59b41186e 100644
--- a/src/box/sql.c
+++ b/src/box/sql.c
@@ -388,9 +388,7 @@ int tarantoolSqlite3Count(BtCursor *pCur, i64 *pnEntry)
 {
 	assert(pCur->curFlags & BTCF_TaCursor);
 
-	uint32_t space_id = SQLITE_PAGENO_TO_SPACEID(pCur->pgnoRoot);
-	uint32_t index_id = SQLITE_PAGENO_TO_INDEXID(pCur->pgnoRoot);
-	*pnEntry = box_index_len(space_id, index_id);
+	*pnEntry = index_size(pCur->index);
 	return SQLITE_OK;
 }
 
@@ -451,6 +449,7 @@ int tarantoolSqlite3EphemeralCreate(BtCursor *pCur, uint32_t field_count,
 		return SQLITE_NOMEM;
 	}
 	pCur->space = ephemer_new_space;
+	pCur->index = *ephemer_new_space->index;
 
 	int unused;
 	return tarantoolSqlite3First(pCur, &unused);
@@ -495,17 +494,19 @@ static int insertOrReplace(BtCursor *pCur, int operationType)
 	assert(operationType == TARANTOOL_INDEX_INSERT ||
 	       operationType == TARANTOOL_INDEX_REPLACE);
 
-	int space_id = SQLITE_PAGENO_TO_SPACEID(pCur->pgnoRoot);
 	int rc;
+	struct request request;
+	memset(&request, 0, sizeof(request));
+	request.tuple = pCur->key;
+	request.tuple_end = pCur->key + pCur->nKey;
+	request.space_id = pCur->space->def->id;
+	mp_tuple_assert(request.tuple, request.tuple_end);
 	if (operationType == TARANTOOL_INDEX_INSERT) {
-		rc = box_insert(space_id, pCur->key,
-				(const char *)pCur->key + pCur->nKey,
-				NULL /* result */);
+		request.type = IPROTO_INSERT;
 	} else {
-		rc = box_replace(space_id, pCur->key,
-				 (const char *)pCur->key + pCur->nKey,
-				 NULL /* result */);
+		request.type = IPROTO_REPLACE;
 	}
+	rc = sql_execute_dml(&request, pCur->space);
 
 	return rc == 0 ? SQLITE_OK : SQL_TARANTOOL_INSERT_FAIL;;
 }
@@ -558,20 +559,23 @@ int tarantoolSqlite3Delete(BtCursor *pCur, u8 flags)
 	assert(pCur->iter != NULL);
 	assert(pCur->last_tuple != NULL);
 
-	uint32_t space_id, index_id;
 	char *key;
 	uint32_t key_size;
 	int rc;
 
-	space_id = SQLITE_PAGENO_TO_SPACEID(pCur->pgnoRoot);
-	index_id = SQLITE_PAGENO_TO_INDEXID(pCur->pgnoRoot);
 	key = tuple_extract_key(pCur->last_tuple,
 				box_iterator_key_def(pCur->iter),
 				&key_size);
 	if (key == NULL)
 		return SQL_TARANTOOL_DELETE_FAIL;
 
-	rc = box_delete(space_id, index_id, key, key + key_size, NULL);
+	struct request request;
+	memset(&request, 0, sizeof(request));
+	request.type = IPROTO_DELETE;
+	request.key = key;
+	request.key_end = key + key_size;
+	request.space_id = pCur->space->def->id;
+	rc = sql_execute_dml(&request, pCur->space);
 
 	return rc == 0 ? SQLITE_OK : SQL_TARANTOOL_DELETE_FAIL;
 }
@@ -1060,6 +1064,7 @@ int tarantoolSqlite3IncrementMaxid(BtCursor *pCur)
 		(char)0xa6, /* MsgPack string(6) */
 		'm', 'a', 'x', '_', 'i', 'd'
 	};
+
 	/* [["+", 1, 1]]*/
 	static const char ops[] = {
 		(char)0x91, /* MsgPack array(1) */
@@ -1070,17 +1075,26 @@ int tarantoolSqlite3IncrementMaxid(BtCursor *pCur)
 		1           /* MsgPack int(1) */
 	};
 
-	uint32_t space_id = SQLITE_PAGENO_TO_SPACEID(pCur->pgnoRoot);
-	uint32_t index_id = SQLITE_PAGENO_TO_INDEXID(pCur->pgnoRoot);
-	box_tuple_t *res;
+	struct tuple *res;
 	int rc;
 
-	rc = box_update(space_id, index_id,
-		key, key + sizeof(key),
-		ops, ops + sizeof(ops),
-		0,
-		&res);
-	if (rc != 0 || res == NULL) {
+	struct iterator *it = index_create_iterator(pCur->index, ITER_ALL,
+						    &key[1], sizeof(key) - 1);
+	if (iterator_next(it, &res) != 0 || res == NULL) {
+		iterator_delete(it);
+		return SQL_TARANTOOL_ERROR;
+	}
+	struct request request;
+	memset(&request, 0, sizeof(request));
+	request.tuple = key;
+	request.tuple_end = key + sizeof(key);
+	request.ops = ops;
+	request.ops_end = ops + sizeof(ops);
+	request.type = IPROTO_UPSERT;
+	request.space_id = pCur->space->def->id;
+	rc = sql_execute_dml(&request, pCur->space);
+	if (rc != 0) {
+		iterator_delete(it);
 		return SQL_TARANTOOL_ERROR;
 	}
 	if (pCur->last_tuple != NULL) {
@@ -1089,6 +1103,7 @@ int tarantoolSqlite3IncrementMaxid(BtCursor *pCur)
 	box_tuple_ref(res);
 	pCur->last_tuple = res;
 	pCur->eState = CURSOR_VALID;
+	iterator_delete(it);
 	return SQLITE_OK;
 }
 
@@ -1129,30 +1144,15 @@ cursor_seek(BtCursor *pCur, int *pRes)
 		box_iterator_free(pCur->iter);
 		pCur->iter = NULL;
 	}
-
-	struct space *space;
-	struct index *index;
-	uint32_t space_id = SQLITE_PAGENO_TO_SPACEID(pCur->pgnoRoot);
-	if (space_id != 0) {
-		space = space_cache_find(space_id);
-		if (space == NULL)
-			return SQL_TARANTOOL_ERROR;
-		uint32_t index_id = SQLITE_PAGENO_TO_INDEXID(pCur->pgnoRoot);
-		index = index_find(space, index_id);
-	} else {
-		space = pCur->space;
-		index = *space->index;
-	}
-
 	const char *key = (const char *)pCur->key;
 	uint32_t part_count = mp_decode_array(&key);
-	if (key_validate(index->def, pCur->iter_type, key, part_count)) {
+	if (key_validate(pCur->index->def, pCur->iter_type, key, part_count)) {
 		diag_log();
 		return SQL_TARANTOOL_ITERATOR_FAIL;
 	}
 
-	struct iterator *it = index_create_iterator(index, pCur->iter_type, key,
-						    part_count);
+	struct iterator *it = index_create_iterator(pCur->index, pCur->iter_type,
+						    key, part_count);
 	if (it == NULL) {
 		pCur->eState = CURSOR_INVALID;
 		return SQL_TARANTOOL_ITERATOR_FAIL;
@@ -1690,13 +1690,14 @@ int tarantoolSqlite3EphemeralGetMaxId(BtCursor *pCur, uint32_t fieldno,
  * If index is empty - return 0 in max_id and success status
  */
 int
-tarantoolSqlGetMaxId(uint32_t space_id, uint32_t index_id, uint32_t fieldno,
+tarantoolSqlGetMaxId(BtCursor *cur, uint32_t fieldno,
 		     uint64_t *max_id)
 {
 	char key[16];
 	struct tuple *tuple;
 	char *key_end = mp_encode_array(key, 0);
-	if (box_index_max(space_id, index_id, key, key_end, &tuple) != 0)
+	if (box_index_max(cur->space->def->id, cur->index->def->iid,
+			  key, key_end, &tuple) != 0)
 		return -1;
 
 	/* Index is empty  */
diff --git a/src/box/sql/cursor.h b/src/box/sql/cursor.h
index 6b23ebf4a..5b8e7c29d 100644
--- a/src/box/sql/cursor.h
+++ b/src/box/sql/cursor.h
@@ -32,7 +32,6 @@
 #ifndef SQLITE_CURSOR_H
 #define SQLITE_CURSOR_H
 
-typedef u32 Pgno;
 typedef struct BtCursor BtCursor;
 
 /*
@@ -59,11 +58,11 @@ typedef struct BtCursor BtCursor;
  */
 struct BtCursor {
 	i64 nKey;		/* Size of pKey, or last integer key */
-	Pgno pgnoRoot;		/* Contains both space_id and index_id */
 	u8 curFlags;		/* zero or more BTCF_* flags defined below */
 	u8 eState;		/* One of the CURSOR_XXX constants (see below) */
 	u8 hints;		/* As configured by CursorSetHints() */
 	struct space *space;
+	struct index *index;
 	struct iterator *iter;
 	enum iterator_type iter_type;
 	struct tuple *last_tuple;
diff --git a/src/box/sql/tarantoolInt.h b/src/box/sql/tarantoolInt.h
index c98871cd5..6f1ba3784 100644
--- a/src/box/sql/tarantoolInt.h
+++ b/src/box/sql/tarantoolInt.h
@@ -150,5 +150,5 @@ int tarantoolSqlite3MakeIdxOpts(Index * index, const char *zSql, void *buf);
  * Fetch maximum value from ineger column number `fieldno` of space_id/index_id
  * Return 0 on success, -1 otherwise
  */
-int tarantoolSqlGetMaxId(uint32_t space_id, uint32_t index_id, uint32_t fieldno,
+int tarantoolSqlGetMaxId(BtCursor *cur, uint32_t fieldno,
 			 uint64_t * max_id);
diff --git a/src/box/sql/vdbe.c b/src/box/sql/vdbe.c
index cd3321c4f..1ee8c4d7b 100644
--- a/src/box/sql/vdbe.c
+++ b/src/box/sql/vdbe.c
@@ -3137,10 +3137,15 @@ case OP_ReopenIdx: {
 
 	assert(pOp->p5==0 || pOp->p5==OPFLAG_SEEKEQ);
 	assert(pOp->p4type==P4_KEYINFO);
-	pCur = p->apCsr[pOp->p1];
+	/*
+	 * FIXME: optimization temporary disabled until the whole
+	 * OP_Open* is reworked, i.e. pointer to space is passed
+	 * to this opcode instead of space id.
+	 */
+	/* pCur = p->apCsr[pOp->p1];
 	if (pCur && pCur->pgnoRoot==(u32)pOp->p2) {
 		goto open_cursor_set_hints;
-	}
+	} */
 	/* If the cursor is not currently open or is open on a different
 	 * index, then fall through into OP_OpenRead to force a reopen
 	 */
@@ -3209,16 +3214,17 @@ case OP_OpenWrite:
 	pCur = allocateCursor(p, pOp->p1, nField, CURTYPE_TARANTOOL);
 	if (pCur==0) goto no_mem;
 	pCur->nullRow = 1;
-	pCur->pgnoRoot = p2;
 
 	assert(p2 >= 1);
 	pBtCur = pCur->uc.pCursor;
-	pBtCur->pgnoRoot = p2;
+	pBtCur->space = space_by_id(SQLITE_PAGENO_TO_SPACEID(p2));
+	pBtCur->index = space_index(pBtCur->space, SQLITE_PAGENO_TO_INDEXID(p2));
+	assert(pBtCur->space != NULL && pBtCur->index != NULL);
 	pBtCur->eState = CURSOR_INVALID;
 	pBtCur->curFlags |= BTCF_TaCursor;
 	pCur->pKeyInfo = pKeyInfo;
 
-	open_cursor_set_hints:
+//	open_cursor_set_hints:
 	assert(OPFLAG_BULKCSR==BTREE_BULKLOAD);
 	assert(OPFLAG_SEEKEQ==BTREE_SEEK_EQ);
 	testcase( pOp->p5 & OPFLAG_BULKCSR);
@@ -3251,7 +3257,6 @@ case OP_OpenTEphemeral: {
 	pCx->pKeyInfo  = pOp->p4.pKeyInfo;
 	pBtCur = pCx->uc.pCursor;
 	/* Ephemeral spaces don't have space_id */
-	pBtCur->pgnoRoot = 0;
 	pBtCur->eState = CURSOR_INVALID;
 	pBtCur->curFlags = BTCF_TEphemCursor;
 
@@ -3791,7 +3796,6 @@ case OP_Sequence: {           /* out2 */
 case OP_NextId: {     /* out3 */
 	VdbeCursor *pC;    /* The VDBE cursor */
 	int p2;            /* Column number, which stores the id */
-	int pgno;          /* Page number of the cursor */
 	pC = p->apCsr[pOp->p1];
 	p2 = pOp->p2;
 	pOut = &aMem[pOp->p3];
@@ -3799,11 +3803,7 @@ case OP_NextId: {     /* out3 */
 	/* This opcode is Tarantool specific.  */
 	assert(pC->uc.pCursor->curFlags & BTCF_TaCursor);
 
-	pgno = pC->pgnoRoot;
-
-	tarantoolSqlGetMaxId(SQLITE_PAGENO_TO_SPACEID(pgno),
-			     SQLITE_PAGENO_TO_INDEXID(pgno),
-			     p2,
+	tarantoolSqlGetMaxId(pC->uc.pCursor, p2,
 			     (uint64_t *) &pOut->u.i);
 
 	pOut->u.i += 1;
diff --git a/src/box/sql/vdbeInt.h b/src/box/sql/vdbeInt.h
index 1836673a4..8b622de5b 100644
--- a/src/box/sql/vdbeInt.h
+++ b/src/box/sql/vdbeInt.h
@@ -122,7 +122,6 @@ struct VdbeCursor {
 		VdbeSorter *pSorter;	/* CURTYPE_SORTER. Sorter object */
 	} uc;
 	KeyInfo *pKeyInfo;	/* Info about index keys needed by index cursors */
-	Pgno pgnoRoot;		/* Root page of the open cursor */
 	i16 nField;		/* Number of fields in the header */
 	u16 nHdrParsed;		/* Number of header fields parsed so far */
 	const u8 *aRow;		/* Data for the current row, if all on one page */
-- 
2.15.1