[PATCH 03/13] vinyl: store tuple comparison hints in tx write set

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

This patch incorporates tuple comparison hints into the transaction
write set. Now, beside a statement, each txv also stores its hint,
which is used in all comparison operations.

Apart from speeding up lookups, this is also needed for multikey index
support, because multikey indexes will reuse hints to store offsets of
indexed array entries.
---
 src/box/vy_point_lookup.c | 10 ++++---
 src/box/vy_tx.c           | 70 ++++++++++++++++++++++++++++++-----------------
 src/box/vy_tx.h           |  9 ++++--
 3 files changed, 58 insertions(+), 31 deletions(-)

diff --git a/src/box/vy_point_lookup.c b/src/box/vy_point_lookup.c
index 9e1f3ca7..5f1be274 100644
--- a/src/box/vy_point_lookup.c
+++ b/src/box/vy_point_lookup.c
@@ -53,13 +53,13 @@
  */
 static int
 vy_point_lookup_scan_txw(struct vy_lsm *lsm, struct vy_tx *tx,
-			 struct tuple *key, struct vy_history *history)
+			 struct tuple *key, hint_t hint,
+			 struct vy_history *history)
 {
 	if (tx == NULL)
 		return 0;
 	lsm->stat.txw.iterator.lookup++;
-	struct txv *txv =
-		write_set_search_key(&tx->write_set, lsm, key);
+	struct txv *txv = write_set_search_key(&tx->write_set, lsm, key, hint);
 	assert(txv == NULL || txv->lsm == lsm);
 	if (txv == NULL)
 		return 0;
@@ -212,7 +212,9 @@ vy_point_lookup(struct vy_lsm *lsm, struct vy_tx *tx,
 	vy_history_create(&mem_history, &lsm->env->history_node_pool);
 	vy_history_create(&disk_history, &lsm->env->history_node_pool);
 
-	rc = vy_point_lookup_scan_txw(lsm, tx, key, &history);
+	hint_t hint = vy_stmt_hint(key, lsm->cmp_def);
+
+	rc = vy_point_lookup_scan_txw(lsm, tx, key, hint, &history);
 	if (rc != 0 || vy_history_is_terminal(&history))
 		goto done;
 
diff --git a/src/box/vy_tx.c b/src/box/vy_tx.c
index 05f757d8..6a464a1e 100644
--- a/src/box/vy_tx.c
+++ b/src/box/vy_tx.c
@@ -68,7 +68,9 @@ write_set_cmp(struct txv *a, struct txv *b)
 {
 	int rc = a->lsm < b->lsm ? -1 : a->lsm > b->lsm;
 	if (rc == 0)
-		return vy_stmt_compare(a->stmt, b->stmt, a->lsm->cmp_def);
+		return vy_stmt_compare_hinted(a->stmt, a->hint,
+					      b->stmt, b->hint,
+					      a->lsm->cmp_def);
 	return rc;
 }
 
@@ -77,7 +79,9 @@ write_set_key_cmp(struct write_set_key *a, struct txv *b)
 {
 	int rc = a->lsm < b->lsm ? -1 : a->lsm > b->lsm;
 	if (rc == 0)
-		return vy_stmt_compare(a->stmt, b->stmt, a->lsm->cmp_def);
+		return vy_stmt_compare_hinted(a->stmt, a->hint,
+					      b->stmt, b->hint,
+					      a->lsm->cmp_def);
 	return rc;
 }
 
@@ -213,7 +217,7 @@ tx_manager_destroy_read_view(struct tx_manager *xm,
 
 static struct txv *
 txv_new(struct vy_tx *tx, struct vy_lsm *lsm,
-	struct tuple *stmt, uint64_t column_mask)
+	struct tuple *stmt, hint_t hint, uint64_t column_mask)
 {
 	struct tx_manager *xm = tx->xm;
 	struct txv *v = mempool_alloc(&xm->txv_mempool);
@@ -225,6 +229,7 @@ txv_new(struct vy_tx *tx, struct vy_lsm *lsm,
 	vy_lsm_ref(v->lsm);
 	v->mem = NULL;
 	v->stmt = stmt;
+	v->hint = hint;
 	tuple_ref(stmt);
 	v->region_stmt = NULL;
 	v->column_mask = column_mask;
@@ -626,8 +631,9 @@ vy_tx_handle_deferred_delete(struct vy_tx *tx, struct txv *v)
 	int rc = 0;
 	for (uint32_t i = 1; i < space->index_count; i++) {
 		struct vy_lsm *lsm = vy_lsm(space->index[i]);
+		hint_t hint = vy_stmt_hint(delete_stmt, lsm->cmp_def);
 		struct txv *delete_txv = txv_new(tx, lsm, delete_stmt,
-						 UINT64_MAX);
+						 hint, UINT64_MAX);
 		if (delete_txv == NULL) {
 			rc = -1;
 			break;
@@ -1007,7 +1013,8 @@ vy_tx_track_point(struct vy_tx *tx, struct vy_lsm *lsm, struct tuple *stmt)
 		return 0;
 	}
 
-	struct txv *v = write_set_search_key(&tx->write_set, lsm, stmt);
+	hint_t hint = vy_stmt_hint(stmt, lsm->cmp_def);
+	struct txv *v = write_set_search_key(&tx->write_set, lsm, stmt, hint);
 	if (v != NULL && vy_stmt_type(v->stmt) != IPROTO_UPSERT) {
 		/* Reading from own write set is serializable. */
 		return 0;
@@ -1028,7 +1035,8 @@ vy_tx_set_with_colmask(struct vy_tx *tx, struct vy_lsm *lsm,
 	vy_stmt_set_lsn(stmt, INT64_MAX);
 	struct tuple *applied = NULL;
 
-	struct txv *old = write_set_search_key(&tx->write_set, lsm, stmt);
+	hint_t hint = vy_stmt_hint(stmt, lsm->cmp_def);
+	struct txv *old = write_set_search_key(&tx->write_set, lsm, stmt, hint);
 	/* Found a match of the previous action of this transaction */
 	if (old != NULL && vy_stmt_type(stmt) == IPROTO_UPSERT) {
 		assert(lsm->index_id == 0);
@@ -1049,7 +1057,7 @@ vy_tx_set_with_colmask(struct vy_tx *tx, struct vy_lsm *lsm,
 	}
 
 	/* Allocate a MVCC container. */
-	struct txv *v = txv_new(tx, lsm, stmt, column_mask);
+	struct txv *v = txv_new(tx, lsm, stmt, hint, column_mask);
 	if (applied != NULL)
 		tuple_unref(applied);
 	if (v == NULL)
@@ -1122,7 +1130,7 @@ tx_manager_abort_writers_for_ddl(struct tx_manager *xm, struct space *space)
 			continue;
 		if (tx->last_stmt_space == space ||
 		    write_set_search_key(&tx->write_set, lsm,
-					 lsm->env->empty_key) != NULL)
+					 lsm->env->empty_key, HINT_NONE) != NULL)
 			vy_tx_abort(tx);
 	}
 }
@@ -1150,6 +1158,7 @@ vy_txw_iterator_open(struct vy_txw_iterator *itr,
 	itr->lsm = lsm;
 	itr->iterator_type = iterator_type;
 	itr->key = key;
+	itr->hint = vy_stmt_hint(key, lsm->cmp_def);
 	itr->version = UINT32_MAX;
 	itr->curr_txv = NULL;
 	itr->search_started = false;
@@ -1161,24 +1170,25 @@ vy_txw_iterator_open(struct vy_txw_iterator *itr,
  * given key (pass NULL to start iteration).
  */
 static void
-vy_txw_iterator_seek(struct vy_txw_iterator *itr, const struct tuple *last_key)
+vy_txw_iterator_seek(struct vy_txw_iterator *itr, const struct tuple *last_key,
+		     hint_t last_hint)
 {
 	itr->stat->lookup++;
 	itr->version = itr->tx->write_set_version;
 	itr->curr_txv = NULL;
 
-	const struct tuple *key = itr->key;
+	struct vy_lsm *lsm = itr->lsm;
+	struct write_set_key k = { lsm, itr->key, itr->hint };
 	enum iterator_type iterator_type = itr->iterator_type;
 	if (last_key != NULL) {
-		key = last_key;
+		k.stmt = last_key;
+		k.hint = last_hint;
 		iterator_type = iterator_direction(iterator_type) > 0 ?
 				ITER_GT : ITER_LT;
 	}
 
-	struct vy_lsm *lsm = itr->lsm;
-	struct write_set_key k = { lsm, key };
 	struct txv *txv;
-	if (!vy_stmt_is_empty_key(key)) {
+	if (!vy_stmt_is_empty_key(k.stmt)) {
 		if (iterator_type == ITER_EQ)
 			txv = write_set_search(&itr->tx->write_set, &k);
 		else if (iterator_type == ITER_GE || iterator_type == ITER_GT)
@@ -1187,7 +1197,8 @@ vy_txw_iterator_seek(struct vy_txw_iterator *itr, const struct tuple *last_key)
 			txv = write_set_psearch(&itr->tx->write_set, &k);
 		if (txv == NULL || txv->lsm != lsm)
 			return;
-		if (vy_stmt_compare(key, txv->stmt, lsm->cmp_def) == 0) {
+		if (vy_stmt_compare_hinted(k.stmt, k.hint, txv->stmt, txv->hint,
+					   lsm->cmp_def) == 0) {
 			while (true) {
 				struct txv *next;
 				if (iterator_type == ITER_LE ||
@@ -1197,8 +1208,9 @@ vy_txw_iterator_seek(struct vy_txw_iterator *itr, const struct tuple *last_key)
 					next = write_set_prev(&itr->tx->write_set, txv);
 				if (next == NULL || next->lsm != lsm)
 					break;
-				if (vy_stmt_compare(key, next->stmt,
-						    lsm->cmp_def) != 0)
+				if (vy_stmt_compare_hinted(k.stmt, k.hint,
+							   next->stmt, next->hint,
+							   lsm->cmp_def) != 0)
 					break;
 				txv = next;
 			}
@@ -1216,7 +1228,8 @@ vy_txw_iterator_seek(struct vy_txw_iterator *itr, const struct tuple *last_key)
 	if (txv == NULL || txv->lsm != lsm)
 		return;
 	if (itr->iterator_type == ITER_EQ && last_key != NULL &&
-	    vy_stmt_compare(itr->key, txv->stmt, lsm->cmp_def) != 0)
+	    vy_stmt_compare_hinted(itr->key, itr->hint, txv->stmt, txv->hint,
+				   lsm->cmp_def) != 0)
 		return;
 	itr->curr_txv = txv;
 }
@@ -1228,7 +1241,7 @@ vy_txw_iterator_next(struct vy_txw_iterator *itr,
 	vy_history_cleanup(history);
 	if (!itr->search_started) {
 		itr->search_started = true;
-		vy_txw_iterator_seek(itr, NULL);
+		vy_txw_iterator_seek(itr, NULL, HINT_NONE);
 		goto out;
 	}
 	assert(itr->version == itr->tx->write_set_version);
@@ -1241,8 +1254,8 @@ vy_txw_iterator_next(struct vy_txw_iterator *itr,
 	if (itr->curr_txv != NULL && itr->curr_txv->lsm != itr->lsm)
 		itr->curr_txv = NULL;
 	if (itr->curr_txv != NULL && itr->iterator_type == ITER_EQ &&
-	    vy_stmt_compare(itr->key, itr->curr_txv->stmt,
-			    itr->lsm->cmp_def) != 0)
+	    vy_stmt_compare_hinted(itr->key, itr->hint, itr->curr_txv->stmt,
+				   itr->curr_txv->hint, itr->lsm->cmp_def) != 0)
 		itr->curr_txv = NULL;
 out:
 	if (itr->curr_txv != NULL) {
@@ -1261,6 +1274,8 @@ vy_txw_iterator_skip(struct vy_txw_iterator *itr,
 	assert(!itr->search_started ||
 	       itr->version == itr->tx->write_set_version);
 
+	hint_t last_hint = last_stmt == NULL ? HINT_NONE :
+			   vy_stmt_hint(last_stmt, itr->lsm->cmp_def);
 	/*
 	 * Check if the iterator is already positioned
 	 * at the statement following last_stmt.
@@ -1268,14 +1283,16 @@ vy_txw_iterator_skip(struct vy_txw_iterator *itr,
 	if (itr->search_started &&
 	    (itr->curr_txv == NULL || last_stmt == NULL ||
 	     iterator_direction(itr->iterator_type) *
-	     vy_stmt_compare(itr->curr_txv->stmt, last_stmt,
-			     itr->lsm->cmp_def) > 0))
+	     vy_stmt_compare_hinted(itr->curr_txv->stmt,
+				    itr->curr_txv->hint,
+				    last_stmt, last_hint,
+				    itr->lsm->cmp_def) > 0))
 		return 0;
 
 	vy_history_cleanup(history);
 
 	itr->search_started = true;
-	vy_txw_iterator_seek(itr, last_stmt);
+	vy_txw_iterator_seek(itr, last_stmt, last_hint);
 
 	if (itr->curr_txv != NULL) {
 		vy_stmt_counter_acct_tuple(&itr->stat->get,
@@ -1293,7 +1310,10 @@ vy_txw_iterator_restore(struct vy_txw_iterator *itr,
 	if (!itr->search_started || itr->version == itr->tx->write_set_version)
 		return 0;
 
-	vy_txw_iterator_seek(itr, last_stmt);
+	hint_t last_hint = last_stmt == NULL ? HINT_NONE :
+			   vy_stmt_hint(last_stmt, itr->lsm->cmp_def);
+
+	vy_txw_iterator_seek(itr, last_stmt, last_hint);
 
 	vy_history_cleanup(history);
 	if (itr->curr_txv != NULL) {
diff --git a/src/box/vy_tx.h b/src/box/vy_tx.h
index 93e3a8cd..907957ae 100644
--- a/src/box/vy_tx.h
+++ b/src/box/vy_tx.h
@@ -84,6 +84,8 @@ struct txv {
 	struct vy_mem *mem;
 	/** Statement of this operation. */
 	struct tuple *stmt;
+	/** Statement comparison hint. */
+	hint_t hint;
 	/** Statement allocated on vy_mem->allocator. */
 	const struct tuple *region_stmt;
 	/** Mask of columns modified by this operation. */
@@ -115,6 +117,7 @@ struct txv {
 struct write_set_key {
 	struct vy_lsm *lsm;
 	const struct tuple *stmt;
+	hint_t hint;
 };
 
 int
@@ -128,9 +131,9 @@ rb_gen_ext_key(MAYBE_UNUSED static inline, write_set_, write_set_t, struct txv,
 
 static inline struct txv *
 write_set_search_key(write_set_t *tree, struct vy_lsm *lsm,
-		     const struct tuple *stmt)
+		     const struct tuple *stmt, hint_t hint)
 {
-	struct write_set_key key = { .lsm = lsm, .stmt = stmt };
+	struct write_set_key key = { .lsm = lsm, .stmt = stmt, .hint = hint };
 	return write_set_search(tree, &key);
 }
 
@@ -421,6 +424,8 @@ struct vy_txw_iterator {
 	enum iterator_type iterator_type;
 	/** Search key. */
 	const struct tuple *key;
+	/** Comparison hint of the search key. */
+	hint_t hint;
 	/* Last seen value of the write set version. */
 	uint32_t version;
 	/* Current position in the write set. */
-- 
2.11.0