[PATCH 11/13] vinyl: forward tuple comparison hints to read iterator

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

Pass a comparison hint along with a search key in the arguments instead
of computing it in the read iterator. This will allow to avoid computing
a hint for the second time in case the calling code already has it.

Besides, it is needed for multikey indexes, which will reuse hints to
store offsets of indexed array entries and thus make hints impossible to
be computed in an arbitrary place in code.
---
 src/box/vinyl.c             | 40 +++++++++++++++++++++++-----------------
 src/box/vy_lsm.c            |  2 +-
 src/box/vy_lsm.h            |  3 ++-
 src/box/vy_point_lookup.c   |  9 +++------
 src/box/vy_point_lookup.h   |  5 +++--
 src/box/vy_read_iterator.c  |  5 +++--
 src/box/vy_read_iterator.h  |  4 +++-
 src/box/vy_tx.c             |  3 ++-
 test/unit/vy_point_lookup.c |  5 +++--
 9 files changed, 43 insertions(+), 33 deletions(-)

diff --git a/src/box/vinyl.c b/src/box/vinyl.c
index 8832db2c..edf8b8b3 100644
--- a/src/box/vinyl.c
+++ b/src/box/vinyl.c
@@ -1139,7 +1139,7 @@ vinyl_space_check_format(struct space *space, struct tuple_format *format)
 
 	struct vy_read_iterator itr;
 	vy_read_iterator_open(&itr, pk, NULL, ITER_ALL, pk->env->empty_key,
-			      &env->xm->p_committed_read_view);
+			      HINT_NONE, &env->xm->p_committed_read_view);
 	int rc;
 	int loops = 0;
 	hint_t hint;
@@ -1346,9 +1346,9 @@ vy_get_by_secondary_tuple(struct vy_lsm *lsm, struct vy_tx *tx,
 		tuple_ref(key);
 	}
 
-	hint_t primary_hint;
-	if (vy_point_lookup(lsm->pk, tx, rv, key, result,
-			    &primary_hint) != 0) {
+	hint_t primary_hint = vy_stmt_hint(key, lsm->pk->cmp_def);
+	if (vy_point_lookup(lsm->pk, tx, rv, key, primary_hint,
+			    result, &primary_hint) != 0) {
 		rc = -1;
 		goto out;
 	}
@@ -1426,13 +1426,14 @@ vy_get(struct vy_lsm *lsm, struct vy_tx *tx,
 	hint_t hint;
 	struct tuple *tuple;
 
+	hint = vy_stmt_hint(key, lsm->cmp_def);
 	if (vy_stmt_is_full_key(key, lsm->cmp_def)) {
 		/*
 		 * Use point lookup for a full key.
 		 */
 		if (tx != NULL && vy_tx_track_point(tx, lsm, key) != 0)
 			return -1;
-		if (vy_point_lookup(lsm, tx, rv, key, &tuple, &hint) != 0)
+		if (vy_point_lookup(lsm, tx, rv, key, hint, &tuple, &hint) != 0)
 			return -1;
 		if (lsm->index_id > 0 && tuple != NULL) {
 			rc = vy_get_by_secondary_tuple(lsm, tx, rv,
@@ -1450,7 +1451,7 @@ vy_get(struct vy_lsm *lsm, struct vy_tx *tx,
 	}
 
 	struct vy_read_iterator itr;
-	vy_read_iterator_open(&itr, lsm, tx, ITER_EQ, key, rv);
+	vy_read_iterator_open(&itr, lsm, tx, ITER_EQ, key, hint, rv);
 	while ((rc = vy_read_iterator_next(&itr, &tuple, &hint)) == 0) {
 		if (lsm->index_id == 0 || tuple == NULL) {
 			*result = tuple;
@@ -2545,7 +2546,7 @@ vy_squash_queue_new(void);
 static void
 vy_squash_queue_delete(struct vy_squash_queue *q);
 static void
-vy_squash_schedule(struct vy_lsm *lsm, struct tuple *stmt,
+vy_squash_schedule(struct vy_lsm *lsm, struct tuple *stmt, hint_t hint,
 		   void /* struct vy_env */ *arg);
 
 static struct vy_env *
@@ -3485,6 +3486,8 @@ struct vy_squash {
 	struct vy_lsm *lsm;
 	/** Key to squash upserts for. */
 	struct tuple *stmt;
+	/** Statement comparison hint. */
+	hint_t hint;
 };
 
 struct vy_squash_queue {
@@ -3500,7 +3503,7 @@ struct vy_squash_queue {
 
 static struct vy_squash *
 vy_squash_new(struct mempool *pool, struct vy_env *env,
-	      struct vy_lsm *lsm, struct tuple *stmt)
+	      struct vy_lsm *lsm, struct tuple *stmt, hint_t hint)
 {
 	struct vy_squash *squash;
 	squash = mempool_alloc(pool);
@@ -3511,6 +3514,7 @@ vy_squash_new(struct mempool *pool, struct vy_env *env,
 	squash->lsm = lsm;
 	tuple_ref(stmt);
 	squash->stmt = stmt;
+	squash->hint = hint;
 	return squash;
 }
 
@@ -3542,7 +3546,7 @@ vy_squash_process(struct vy_squash *squash)
 	hint_t hint;
 	struct tuple *result;
 	if (vy_point_lookup(lsm, NULL, &env->xm->p_committed_read_view,
-			    squash->stmt, &result, &hint) != 0)
+			    squash->stmt, squash->hint, &result, &hint) != 0)
 		return -1;
 	if (result == NULL)
 		return 0;
@@ -3661,7 +3665,8 @@ vy_squash_queue_f(va_list va)
  * statement after it. Done in a background fiber.
  */
 static void
-vy_squash_schedule(struct vy_lsm *lsm, struct tuple *stmt, void *arg)
+vy_squash_schedule(struct vy_lsm *lsm, struct tuple *stmt, hint_t hint,
+		   void *arg)
 {
 	struct vy_env *env = arg;
 	struct vy_squash_queue *sq = env->squash_queue;
@@ -3677,7 +3682,8 @@ vy_squash_schedule(struct vy_lsm *lsm, struct tuple *stmt, void *arg)
 		fiber_start(sq->fiber, sq);
 	}
 
-	struct vy_squash *squash = vy_squash_new(&sq->pool, env, lsm, stmt);
+	struct vy_squash *squash = vy_squash_new(&sq->pool, env, lsm,
+						 stmt, hint);
 	if (squash == NULL)
 		goto fail;
 
@@ -3895,7 +3901,8 @@ vinyl_index_create_iterator(struct index *base, enum iterator_type type,
 	}
 	it->tx = tx;
 
-	vy_read_iterator_open(&it->iterator, lsm, tx, type, it->key,
+	hint_t hint = vy_stmt_hint(it->key, lsm->cmp_def);
+	vy_read_iterator_open(&it->iterator, lsm, tx, type, it->key, hint,
 			      (const struct vy_read_view **)&tx->read_view);
 	return (struct iterator *)it;
 }
@@ -4103,7 +4110,7 @@ vy_build_insert_tuple(struct vy_env *env, struct vy_lsm *lsm,
  */
 static int
 vy_build_recover_stmt(struct vy_lsm *lsm, struct vy_lsm *pk,
-		      const struct tuple *mem_stmt)
+		      const struct tuple *mem_stmt, hint_t hint)
 {
 	int64_t lsn = vy_stmt_lsn(mem_stmt);
 	if (lsn <= lsm->dump_lsn)
@@ -4112,9 +4119,8 @@ vy_build_recover_stmt(struct vy_lsm *lsm, struct vy_lsm *pk,
 	/* Lookup the tuple that was affected by this statement. */
 	const struct vy_read_view rv = { .vlsn = lsn - 1 };
 	const struct vy_read_view *p_rv = &rv;
-	hint_t hint;
 	struct tuple *old_tuple;
-	if (vy_point_lookup(pk, NULL, &p_rv, (struct tuple *)mem_stmt,
+	if (vy_point_lookup(pk, NULL, &p_rv, (struct tuple *)mem_stmt, hint,
 			    &old_tuple, &hint) != 0)
 		return -1;
 	/*
@@ -4186,7 +4192,7 @@ vy_build_recover_mem(struct vy_lsm *lsm, struct vy_lsm *pk, struct vy_mem *mem)
 	while (!vy_mem_tree_iterator_is_invalid(&itr)) {
 		struct vy_mem_tree_elem elem;
 		elem = *vy_mem_tree_iterator_get_elem(&mem->tree, &itr);
-		if (vy_build_recover_stmt(lsm, pk, elem.stmt) != 0)
+		if (vy_build_recover_stmt(lsm, pk, elem.stmt, elem.hint) != 0)
 			return -1;
 		vy_mem_tree_iterator_prev(&mem->tree, &itr);
 	}
@@ -4271,7 +4277,7 @@ vinyl_space_build_index(struct space *src_space, struct index *new_index,
 
 	struct vy_read_iterator itr;
 	vy_read_iterator_open(&itr, pk, NULL, ITER_ALL, pk->env->empty_key,
-			      &env->xm->p_committed_read_view);
+			      HINT_NONE, &env->xm->p_committed_read_view);
 	int rc;
 	int loops = 0;
 	hint_t hint;
diff --git a/src/box/vy_lsm.c b/src/box/vy_lsm.c
index 8230410d..f2f9fdaf 100644
--- a/src/box/vy_lsm.c
+++ b/src/box/vy_lsm.c
@@ -984,7 +984,7 @@ vy_lsm_commit_upsert(struct vy_lsm *lsm, struct vy_mem *mem,
 
 		struct tuple *dup = vy_stmt_dup(stmt);
 		if (dup != NULL) {
-			lsm->env->upsert_thresh_cb(lsm, dup,
+			lsm->env->upsert_thresh_cb(lsm, dup, hint,
 					lsm->env->upsert_thresh_arg);
 			tuple_unref(dup);
 		}
diff --git a/src/box/vy_lsm.h b/src/box/vy_lsm.h
index 541c49f8..fdc93dc5 100644
--- a/src/box/vy_lsm.h
+++ b/src/box/vy_lsm.h
@@ -61,7 +61,8 @@ struct vy_run;
 struct vy_run_env;
 
 typedef void
-(*vy_upsert_thresh_cb)(struct vy_lsm *lsm, struct tuple *stmt, void *arg);
+(*vy_upsert_thresh_cb)(struct vy_lsm *lsm, struct tuple *stmt, hint_t hint,
+		       void *arg);
 
 /** Common LSM tree environment. */
 struct vy_lsm_env {
diff --git a/src/box/vy_point_lookup.c b/src/box/vy_point_lookup.c
index bac6db2b..8e44cbce 100644
--- a/src/box/vy_point_lookup.c
+++ b/src/box/vy_point_lookup.c
@@ -197,7 +197,8 @@ vy_point_lookup_scan_slices(struct vy_lsm *lsm, const struct vy_read_view **rv,
 
 int
 vy_point_lookup(struct vy_lsm *lsm, struct vy_tx *tx,
-		const struct vy_read_view **rv, struct tuple *key,
+		const struct vy_read_view **rv,
+		struct tuple *key, hint_t hint,
 		struct tuple **ret, hint_t *ret_hint)
 {
 	/* All key parts must be set for a point lookup. */
@@ -217,8 +218,6 @@ 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);
 
-	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;
@@ -319,7 +318,7 @@ done:
 
 int
 vy_point_lookup_mem(struct vy_lsm *lsm, const struct vy_read_view **rv,
-		    struct tuple *key, struct tuple **ret)
+		    struct tuple *key, hint_t hint, struct tuple **ret)
 {
 	assert(vy_stmt_is_full_key(key, lsm->cmp_def));
 
@@ -327,8 +326,6 @@ vy_point_lookup_mem(struct vy_lsm *lsm, const struct vy_read_view **rv,
 	struct vy_history history;
 	vy_history_create(&history, &lsm->env->history_node_pool);
 
-	hint_t hint = vy_stmt_hint(key, lsm->cmp_def);
-
 	rc = vy_point_lookup_scan_cache(lsm, rv, key, hint, &history);
 	if (rc != 0 || vy_history_is_terminal(&history))
 		goto done;
diff --git a/src/box/vy_point_lookup.h b/src/box/vy_point_lookup.h
index a0e23ecb..2046744a 100644
--- a/src/box/vy_point_lookup.h
+++ b/src/box/vy_point_lookup.h
@@ -69,7 +69,8 @@ struct tuple;
  */
 int
 vy_point_lookup(struct vy_lsm *lsm, struct vy_tx *tx,
-		const struct vy_read_view **rv, struct tuple *key,
+		const struct vy_read_view **rv,
+		struct tuple *key, hint_t hint,
 		struct tuple **ret, hint_t *ret_hint);
 
 /**
@@ -88,7 +89,7 @@ vy_point_lookup(struct vy_lsm *lsm, struct vy_tx *tx,
  */
 int
 vy_point_lookup_mem(struct vy_lsm *lsm, const struct vy_read_view **rv,
-		    struct tuple *key, struct tuple **ret);
+		    struct tuple *key, hint_t hint, struct tuple **ret);
 
 #if defined(__cplusplus)
 } /* extern "C" */
diff --git a/src/box/vy_read_iterator.c b/src/box/vy_read_iterator.c
index d0b25610..69fea0cf 100644
--- a/src/box/vy_read_iterator.c
+++ b/src/box/vy_read_iterator.c
@@ -704,7 +704,8 @@ vy_read_iterator_cleanup(struct vy_read_iterator *itr)
 void
 vy_read_iterator_open(struct vy_read_iterator *itr, struct vy_lsm *lsm,
 		      struct vy_tx *tx, enum iterator_type iterator_type,
-		      struct tuple *key, const struct vy_read_view **rv)
+		      struct tuple *key, hint_t hint,
+		      const struct vy_read_view **rv)
 {
 	memset(itr, 0, sizeof(*itr));
 
@@ -712,7 +713,7 @@ vy_read_iterator_open(struct vy_read_iterator *itr, struct vy_lsm *lsm,
 	itr->tx = tx;
 	itr->iterator_type = iterator_type;
 	itr->key = key;
-	itr->hint = vy_stmt_hint(key, lsm->cmp_def);
+	itr->hint = hint;
 	itr->read_view = rv;
 	itr->last_hint = HINT_NONE;
 	itr->last_cached_hint = HINT_NONE;
diff --git a/src/box/vy_read_iterator.h b/src/box/vy_read_iterator.h
index fd876fcc..10a91ca7 100644
--- a/src/box/vy_read_iterator.h
+++ b/src/box/vy_read_iterator.h
@@ -136,12 +136,14 @@ struct vy_read_iterator {
  * @param iterator_type Type of the iterator that determines order
  *                      of the iteration.
  * @param key           Key for the iteration.
+ * @param hint          Comparison hint of the search key.
  * @param rv            Read view.
  */
 void
 vy_read_iterator_open(struct vy_read_iterator *itr, struct vy_lsm *lsm,
 		      struct vy_tx *tx, enum iterator_type iterator_type,
-		      struct tuple *key, const struct vy_read_view **rv);
+		      struct tuple *key, hint_t hint,
+		      const struct vy_read_view **rv);
 
 /**
  * Get the next statement with another key, or start the iterator,
diff --git a/src/box/vy_tx.c b/src/box/vy_tx.c
index 5ded9637..621898c9 100644
--- a/src/box/vy_tx.c
+++ b/src/box/vy_tx.c
@@ -563,6 +563,7 @@ vy_tx_handle_deferred_delete(struct vy_tx *tx, struct txv *v)
 {
 	struct vy_lsm *pk = v->lsm;
 	struct tuple *stmt = v->stmt;
+	hint_t hint = v->hint;
 	uint8_t flags = vy_stmt_flags(stmt);
 
 	assert(pk->index_id == 0);
@@ -580,7 +581,7 @@ vy_tx_handle_deferred_delete(struct vy_tx *tx, struct txv *v)
 	/* Look up the tuple overwritten by this statement. */
 	struct tuple *tuple;
 	if (vy_point_lookup_mem(pk, &tx->xm->p_global_read_view,
-				stmt, &tuple) != 0)
+				stmt, hint, &tuple) != 0)
 		return -1;
 
 	if (tuple == NULL) {
diff --git a/test/unit/vy_point_lookup.c b/test/unit/vy_point_lookup.c
index 93841c0a..52c2affd 100644
--- a/test/unit/vy_point_lookup.c
+++ b/test/unit/vy_point_lookup.c
@@ -275,9 +275,10 @@ test_basic()
 				STMT_TEMPLATE(0, SELECT, i);
 			struct tuple *key = vy_new_simple_stmt(format,
 							       &tmpl_key);
-			hint_t hint;
+			hint_t hint = vy_stmt_hint(key, key_def);
 			struct tuple *res;
-			rc = vy_point_lookup(pk, NULL, &prv, key, &res, &hint);
+			rc = vy_point_lookup(pk, NULL, &prv, key, hint,
+					     &res, &hint);
 			tuple_unref(key);
 			if (rc != 0) {
 				has_errors = true;
-- 
2.11.0