[PATCH 02/13] vinyl: store tuple comparison hints in cache tree

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

This patch incorporates tuple comparison hints into vy_cache_tree,
similarly to how it was done in case of memtx_tree.

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_cache.c      | 119 +++++++++++++++++++++++++++++++-----------------
 src/box/vy_cache.h      |  32 +++++++++++--
 test/vinyl/cache.result |   6 +--
 test/vinyl/stat.result  |   4 +-
 4 files changed, 109 insertions(+), 52 deletions(-)

diff --git a/src/box/vy_cache.c b/src/box/vy_cache.c
index e526ab1e..8fafd648 100644
--- a/src/box/vy_cache.c
+++ b/src/box/vy_cache.c
@@ -84,7 +84,7 @@ vy_cache_entry_size(const struct vy_cache_entry *entry)
 
 static struct vy_cache_entry *
 vy_cache_entry_new(struct vy_cache_env *env, struct vy_cache *cache,
-		   struct tuple *stmt)
+		   struct tuple *stmt, hint_t hint)
 {
 	struct vy_cache_entry *entry = (struct vy_cache_entry *)
 		mempool_alloc(&env->cache_entry_mempool);
@@ -93,6 +93,7 @@ vy_cache_entry_new(struct vy_cache_env *env, struct vy_cache *cache,
 	tuple_ref(stmt);
 	entry->cache = cache;
 	entry->stmt = stmt;
+	entry->hint = hint;
 	entry->flags = 0;
 	entry->left_boundary_level = cache->cmp_def->part_count;
 	entry->right_boundary_level = cache->cmp_def->part_count;
@@ -173,10 +174,11 @@ vy_cache_gc_step(struct vy_cache_env *env)
 	struct vy_cache_tree *tree = &cache->cache_tree;
 	if (entry->flags & (VY_CACHE_LEFT_LINKED |
 			    VY_CACHE_RIGHT_LINKED)) {
+		struct vy_cache_tree_key tree_key;
+		tree_key = vy_cache_tree_key(entry->stmt, entry->hint);
 		bool exact;
 		struct vy_cache_tree_iterator itr =
-			vy_cache_tree_lower_bound(tree, entry->stmt,
-						  &exact);
+			vy_cache_tree_lower_bound(tree, &tree_key, &exact);
 		assert(exact);
 		if (entry->flags & VY_CACHE_LEFT_LINKED) {
 			struct vy_cache_tree_iterator prev = itr;
@@ -257,6 +259,11 @@ vy_cache_add(struct vy_cache *cache, struct tuple *stmt,
 		return;
 	}
 
+	hint_t hint = stmt == NULL ? HINT_NONE :
+		      vy_stmt_hint(stmt, cache->cmp_def);
+	hint_t prev_hint = prev_stmt == NULL ? HINT_NONE :
+			   vy_stmt_hint(prev_stmt, cache->cmp_def);
+
 	int direction = iterator_direction(order);
 	/**
 	 * Let's determine boundary_level (left/right) of the new record
@@ -291,6 +298,7 @@ vy_cache_add(struct vy_cache *cache, struct tuple *stmt,
 		 */
 		direction = -direction;
 		stmt = prev_stmt;
+		hint = prev_hint;
 		prev_stmt = NULL;
 	}
 	TRASH(&order);
@@ -304,7 +312,7 @@ vy_cache_add(struct vy_cache *cache, struct tuple *stmt,
 
 	/* Insert/replace new entry to the tree */
 	struct vy_cache_entry *entry =
-		vy_cache_entry_new(cache->env, cache, stmt);
+		vy_cache_entry_new(cache->env, cache, stmt, hint);
 	if (entry == NULL) {
 		/* memory error, let's live without a cache */
 		return;
@@ -370,8 +378,11 @@ vy_cache_add(struct vy_cache *cache, struct tuple *stmt,
 							&inserted);
 		assert(*prev_check_entry != NULL);
 		struct tuple *prev_check_stmt = (*prev_check_entry)->stmt;
-		int cmp = vy_stmt_compare(prev_stmt, prev_check_stmt,
-					  cache->cmp_def);
+		hint_t prev_check_hint = (*prev_check_entry)->hint;
+		int cmp = vy_stmt_compare_hinted(prev_stmt, prev_hint,
+						 prev_check_stmt,
+						 prev_check_hint,
+						 cache->cmp_def);
 
 		if (entry->flags & flag) {
 			/* The found entry must be exactly prev_stmt. (2) */
@@ -394,7 +405,7 @@ vy_cache_add(struct vy_cache *cache, struct tuple *stmt,
 
 	/* Insert/replace entry with previous statement */
 	struct vy_cache_entry *prev_entry =
-		vy_cache_entry_new(cache->env, cache, prev_stmt);
+		vy_cache_entry_new(cache->env, cache, prev_stmt, prev_hint);
 	if (prev_entry == NULL) {
 		/* memory error, let's live without a chain */
 		return;
@@ -422,8 +433,11 @@ vy_cache_add(struct vy_cache *cache, struct tuple *stmt,
 struct tuple *
 vy_cache_get(struct vy_cache *cache, const struct tuple *key)
 {
+	hint_t hint = vy_stmt_hint(key, cache->cmp_def);
+	struct vy_cache_tree_key tree_key;
+	tree_key = vy_cache_tree_key(key, hint);
 	struct vy_cache_entry **entry =
-		vy_cache_tree_find(&cache->cache_tree, key);
+		vy_cache_tree_find(&cache->cache_tree, &tree_key);
 	if (entry == NULL)
 		return NULL;
 	return (*entry)->stmt;
@@ -435,8 +449,11 @@ vy_cache_on_write(struct vy_cache *cache, const struct tuple *stmt,
 {
 	vy_cache_gc(cache->env);
 	bool exact = false;
+	hint_t hint = vy_stmt_hint(stmt, cache->cmp_def);
+	struct vy_cache_tree_key tree_key;
+	tree_key = vy_cache_tree_key(stmt, hint);
 	struct vy_cache_tree_iterator itr;
-	itr = vy_cache_tree_lower_bound(&cache->cache_tree, stmt, &exact);
+	itr = vy_cache_tree_lower_bound(&cache->cache_tree, &tree_key, &exact);
 	struct vy_cache_entry **entry =
 		vy_cache_tree_iterator_get_elem(&cache->cache_tree, &itr);
 	assert(!exact || entry != NULL);
@@ -507,18 +524,6 @@ vy_cache_on_write(struct vy_cache *cache, const struct tuple *stmt,
 }
 
 /**
- * Get a stmt by current position
- */
-static struct tuple *
-vy_cache_iterator_curr_stmt(struct vy_cache_iterator *itr)
-{
-	struct vy_cache_tree *tree = &itr->cache->cache_tree;
-	struct vy_cache_entry **entry =
-		vy_cache_tree_iterator_get_elem(tree, &itr->curr_pos);
-	return entry ? (*entry)->stmt : NULL;
-}
-
-/**
  * Determine whether the merge iterator must be stopped or not.
  * That is made by examining flags of a cache record.
  *
@@ -596,6 +601,7 @@ vy_cache_iterator_step(struct vy_cache_iterator *itr)
 	if (itr->curr_stmt != NULL) {
 		tuple_unref(itr->curr_stmt);
 		itr->curr_stmt = NULL;
+		itr->curr_hint = HINT_NONE;
 	}
 	struct vy_cache_tree *tree = &itr->cache->cache_tree;
 	struct vy_cache_entry *prev_entry =
@@ -610,10 +616,12 @@ vy_cache_iterator_step(struct vy_cache_iterator *itr)
 		*vy_cache_tree_iterator_get_elem(tree, &itr->curr_pos);
 
 	if (itr->iterator_type == ITER_EQ &&
-	    vy_stmt_compare(itr->key, entry->stmt, itr->cache->cmp_def)) {
+	    vy_stmt_compare_hinted(itr->key, itr->hint, entry->stmt,
+				   entry->hint, itr->cache->cmp_def)) {
 		return vy_cache_iterator_is_end_stop(itr, prev_entry);
 	}
 	itr->curr_stmt = entry->stmt;
+	itr->curr_hint = entry->hint;
 	tuple_ref(itr->curr_stmt);
 	return vy_cache_iterator_is_stop(itr, entry);
 }
@@ -647,31 +655,33 @@ vy_cache_iterator_skip_to_read_view(struct vy_cache_iterator *itr, bool *stop)
  */
 static bool
 vy_cache_iterator_seek(struct vy_cache_iterator *itr,
-		       const struct tuple *last_key)
+		       const struct tuple *last_key, hint_t last_hint)
 {
 	struct vy_cache_tree *tree = &itr->cache->cache_tree;
 
 	if (itr->curr_stmt != NULL) {
 		tuple_unref(itr->curr_stmt);
 		itr->curr_stmt = NULL;
+		itr->curr_hint = HINT_NONE;
 	}
 	itr->cache->stat.lookup++;
 
-	const struct tuple *key = itr->key;
+	struct vy_cache_tree_key tree_key;
+	tree_key = vy_cache_tree_key(itr->key, itr->hint);
 	enum iterator_type iterator_type = itr->iterator_type;
 	if (last_key != NULL) {
-		key = last_key;
+		tree_key = vy_cache_tree_key(last_key, last_hint);
 		iterator_type = iterator_direction(itr->iterator_type) > 0 ?
 				ITER_GT : ITER_LT;
 	}
 
 	bool exact;
-	if (!vy_stmt_is_empty_key(key)) {
+	if (!vy_stmt_is_empty_key(tree_key.stmt)) {
 		itr->curr_pos = iterator_type == ITER_EQ ||
 				iterator_type == ITER_GE ||
 				iterator_type == ITER_LT ?
-				vy_cache_tree_lower_bound(tree, key, &exact) :
-				vy_cache_tree_upper_bound(tree, key, &exact);
+				vy_cache_tree_lower_bound(tree, &tree_key, &exact) :
+				vy_cache_tree_upper_bound(tree, &tree_key, &exact);
 	} else if (iterator_type == ITER_LE) {
 		itr->curr_pos = vy_cache_tree_invalid_iterator();
 	} else {
@@ -689,11 +699,13 @@ vy_cache_iterator_seek(struct vy_cache_iterator *itr,
 
 	if (itr->iterator_type == ITER_EQ &&
 	    ((last_key == NULL && !exact) ||
-	     (last_key != NULL && vy_stmt_compare(itr->key, entry->stmt,
-						  itr->cache->cmp_def) != 0)))
+	     (last_key != NULL &&
+	      vy_stmt_compare_hinted(itr->key, itr->hint, entry->stmt,
+				     entry->hint, itr->cache->cmp_def) != 0)))
 		return false;
 
 	itr->curr_stmt = entry->stmt;
+	itr->curr_hint = entry->hint;
 	tuple_ref(itr->curr_stmt);
 	return vy_cache_iterator_is_stop(itr, entry);
 }
@@ -708,7 +720,7 @@ vy_cache_iterator_next(struct vy_cache_iterator *itr,
 		assert(itr->curr_stmt == NULL);
 		itr->search_started = true;
 		itr->version = itr->cache->version;
-		*stop = vy_cache_iterator_seek(itr, NULL);
+		*stop = vy_cache_iterator_seek(itr, NULL, HINT_NONE);
 	} else {
 		assert(itr->version == itr->cache->version);
 		if (itr->curr_stmt == NULL)
@@ -732,6 +744,8 @@ vy_cache_iterator_skip(struct vy_cache_iterator *itr,
 {
 	assert(!itr->search_started || itr->version == itr->cache->version);
 
+	hint_t last_hint = last_stmt == NULL ? HINT_NONE :
+			   vy_stmt_hint(last_stmt, itr->cache->cmp_def);
 	/*
 	 * Check if the iterator is already positioned
 	 * at the statement following last_stmt.
@@ -739,15 +753,16 @@ vy_cache_iterator_skip(struct vy_cache_iterator *itr,
 	if (itr->search_started &&
 	    (itr->curr_stmt == NULL || last_stmt == NULL ||
 	     iterator_direction(itr->iterator_type) *
-	     vy_stmt_compare(itr->curr_stmt, last_stmt,
-			     itr->cache->cmp_def) > 0))
+	     vy_stmt_compare_hinted(itr->curr_stmt, itr->curr_hint,
+				    last_stmt, last_hint,
+				    itr->cache->cmp_def) > 0))
 		return 0;
 
 	vy_history_cleanup(history);
 
 	itr->search_started = true;
 	itr->version = itr->cache->version;
-	*stop = vy_cache_iterator_seek(itr, last_stmt);
+	*stop = vy_cache_iterator_seek(itr, last_stmt, last_hint);
 	vy_cache_iterator_skip_to_read_view(itr, stop);
 
 	if (itr->curr_stmt != NULL) {
@@ -766,17 +781,33 @@ vy_cache_iterator_restore(struct vy_cache_iterator *itr,
 	if (!itr->search_started || itr->version == itr->cache->version)
 		return 0;
 
+	hint_t last_hint = last_stmt == NULL ? HINT_NONE :
+			   vy_stmt_hint(last_stmt, itr->cache->cmp_def);
+
+	/* Check if the iterator position is still valid. */
+	bool pos_invalid = false;
+	struct vy_cache_tree *tree = &itr->cache->cache_tree;
+	if (itr->curr_stmt != NULL) {
+		struct vy_cache_entry **entry;
+		entry = vy_cache_tree_iterator_get_elem(tree, &itr->curr_pos);
+		if (entry == NULL || (*entry)->stmt != itr->curr_stmt ||
+				     (*entry)->hint != itr->curr_hint)
+			pos_invalid = true;
+	} else {
+		if (itr->iterator_type != ITER_EQ)
+			itr->curr_pos = vy_cache_tree_invalid_iterator();
+		else
+			pos_invalid = true;
+	}
 	bool pos_changed = false;
 	itr->version = itr->cache->version;
-	if ((itr->curr_stmt == NULL && itr->iterator_type == ITER_EQ) ||
-	    (itr->curr_stmt != NULL &&
-	     itr->curr_stmt != vy_cache_iterator_curr_stmt(itr))) {
+	if (pos_invalid) {
 		/*
 		 * EQ search ended or the iterator was invalidated.
 		 * In either case the best we can do is restart the
 		 * search.
 		 */
-		*stop = vy_cache_iterator_seek(itr, last_stmt);
+		*stop = vy_cache_iterator_seek(itr, last_stmt, last_hint);
 		vy_cache_iterator_skip_to_read_view(itr, stop);
 		pos_changed = true;
 	} else {
@@ -788,18 +819,17 @@ vy_cache_iterator_restore(struct vy_cache_iterator *itr,
 		 */
 		bool key_belongs = false;
 		const struct tuple *key = last_stmt;
+		hint_t hint = last_hint;
 		if (key == NULL) {
 			key = itr->key;
+			hint = itr->hint;
 			key_belongs = (itr->iterator_type == ITER_EQ ||
 				       itr->iterator_type == ITER_GE ||
 				       itr->iterator_type == ITER_LE);
 		}
 		int dir = iterator_direction(itr->iterator_type);
 		struct key_def *def = itr->cache->cmp_def;
-		struct vy_cache_tree *tree = &itr->cache->cache_tree;
 		struct vy_cache_tree_iterator pos = itr->curr_pos;
-		if (itr->curr_stmt == NULL)
-			pos = vy_cache_tree_invalid_iterator();
 		while (true) {
 			if (dir > 0)
 				vy_cache_tree_iterator_prev(tree, &pos);
@@ -809,7 +839,9 @@ vy_cache_iterator_restore(struct vy_cache_iterator *itr,
 				break;
 			struct vy_cache_entry *entry =
 				*vy_cache_tree_iterator_get_elem(tree, &pos);
-			int cmp = dir * vy_stmt_compare(entry->stmt, key, def);
+			int cmp = dir * vy_stmt_compare_hinted(entry->stmt,
+							       entry->hint,
+							       key, hint, def);
 			if (cmp < 0 || (cmp == 0 && !key_belongs))
 				break;
 			if (vy_stmt_lsn(entry->stmt) <= (**itr->read_view).vlsn) {
@@ -817,6 +849,7 @@ vy_cache_iterator_restore(struct vy_cache_iterator *itr,
 				if (itr->curr_stmt != NULL)
 					tuple_unref(itr->curr_stmt);
 				itr->curr_stmt = entry->stmt;
+				itr->curr_hint = entry->hint;
 				tuple_ref(itr->curr_stmt);
 				*stop = vy_cache_iterator_is_stop(itr, entry);
 				pos_changed = true;
@@ -856,9 +889,11 @@ vy_cache_iterator_open(struct vy_cache_iterator *itr, struct vy_cache *cache,
 	itr->cache = cache;
 	itr->iterator_type = iterator_type;
 	itr->key = key;
+	itr->hint = vy_stmt_hint(key, cache->cmp_def);
 	itr->read_view = rv;
 
 	itr->curr_stmt = NULL;
+	itr->curr_hint = HINT_NONE;
 	itr->curr_pos = vy_cache_tree_invalid_iterator();
 
 	itr->version = 0;
diff --git a/src/box/vy_cache.h b/src/box/vy_cache.h
index a846622e..4fcdb747 100644
--- a/src/box/vy_cache.h
+++ b/src/box/vy_cache.h
@@ -56,6 +56,8 @@ struct vy_cache_entry {
 	struct vy_cache *cache;
 	/* Statement in cache */
 	struct tuple *stmt;
+	/* Comparison hint of the cached statement. */
+	hint_t hint;
 	/* Link in LRU list */
 	struct rlist in_lru;
 	/* VY_CACHE_LEFT_LINKED and/or VY_CACHE_RIGHT_LINKED, see
@@ -67,6 +69,20 @@ struct vy_cache_entry {
 	uint8_t right_boundary_level;
 };
 
+struct vy_cache_tree_key {
+	const struct tuple *stmt;
+	hint_t hint;
+};
+
+static inline struct vy_cache_tree_key
+vy_cache_tree_key(const struct tuple *stmt, hint_t hint)
+{
+	struct vy_cache_tree_key key;
+	key.stmt = stmt;
+	key.hint = hint;
+	return key;
+}
+
 /**
  * Internal comparator (1) for BPS tree.
  */
@@ -74,17 +90,19 @@ static inline int
 vy_cache_tree_cmp(struct vy_cache_entry *a,
 		  struct vy_cache_entry *b, struct key_def *cmp_def)
 {
-	return vy_stmt_compare(a->stmt, b->stmt, cmp_def);
+	return vy_stmt_compare_hinted(a->stmt, a->hint,
+				      b->stmt, b->hint, cmp_def);
 }
 
 /**
  * Internal comparator (2) for BPS tree.
  */
 static inline int
-vy_cache_tree_key_cmp(struct vy_cache_entry *a,
-		      const struct tuple *b, struct key_def *cmp_def)
+vy_cache_tree_key_cmp(struct vy_cache_entry *a, struct vy_cache_tree_key *b,
+		      struct key_def *cmp_def)
 {
-	return vy_stmt_compare(a->stmt, b, cmp_def);
+	return vy_stmt_compare_hinted(a->stmt, a->hint,
+				      b->stmt, b->hint, cmp_def);
 }
 
 #define VY_CACHE_TREE_EXTENT_SIZE (16 * 1024)
@@ -95,7 +113,7 @@ vy_cache_tree_key_cmp(struct vy_cache_entry *a,
 #define BPS_TREE_COMPARE(a, b, cmp_def) vy_cache_tree_cmp(a, b, cmp_def)
 #define BPS_TREE_COMPARE_KEY(a, b, cmp_def) vy_cache_tree_key_cmp(a, b, cmp_def)
 #define bps_tree_elem_t struct vy_cache_entry *
-#define bps_tree_key_t const struct tuple *
+#define bps_tree_key_t struct vy_cache_tree_key *
 #define bps_tree_arg_t struct key_def *
 #define BPS_TREE_NO_DEBUG
 
@@ -239,6 +257,8 @@ struct vy_cache_iterator {
 	enum iterator_type iterator_type;
 	/* Search key data in terms of vinyl, vy_stmt_compare argument */
 	const struct tuple *key;
+	/** Comparison hint of the search key. */
+	hint_t hint;
 	/* LSN visibility, iterator shows values with lsn <= vlsn */
 	const struct vy_read_view **read_view;
 
@@ -247,6 +267,8 @@ struct vy_cache_iterator {
 	struct vy_cache_tree_iterator curr_pos;
 	/* stmt in current position in tree */
 	struct tuple *curr_stmt;
+	/* Comparison hint of the current statement. */
+	hint_t curr_hint;
 
 	/* Last version of cache */
 	uint32_t version;
diff --git a/test/vinyl/cache.result b/test/vinyl/cache.result
index 85741604..49d2bcc7 100644
--- a/test/vinyl/cache.result
+++ b/test/vinyl/cache.result
@@ -1033,14 +1033,14 @@ for i = 1, 100 do s:get{i} end
 ...
 box.stat.vinyl().memory.tuple_cache
 ---
-- 107700
+- 108500
 ...
 box.cfg{vinyl_cache = 50 * 1000}
 ---
 ...
 box.stat.vinyl().memory.tuple_cache
 ---
-- 49542
+- 49910
 ...
 box.cfg{vinyl_cache = 0}
 ---
@@ -1116,7 +1116,7 @@ s.index.i2:count()
 ...
 box.stat.vinyl().memory.tuple_cache -- should be about 200 KB
 ---
-- 216800
+- 219200
 ...
 s:drop()
 ---
diff --git a/test/vinyl/stat.result b/test/vinyl/stat.result
index ff73d42a..934e0807 100644
--- a/test/vinyl/stat.result
+++ b/test/vinyl/stat.result
@@ -769,7 +769,7 @@ _ = s:get(1)
 ...
 stat_diff(gstat(), st, 'memory.tuple_cache')
 ---
-- 1101
+- 1109
 ...
 s:delete(1)
 ---
@@ -1122,7 +1122,7 @@ gstat()
     gap_locks: 0
     read_views: 0
   memory:
-    tuple_cache: 14313
+    tuple_cache: 14417
     tx: 0
     level0: 262583
     page_index: 1050
-- 
2.11.0