[PATCH 09/13] vinyl: forward tuple comparison hints to memory tree

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

Instead of computing a statement comparison hint in a vy_mem method,
forward it from the upper level, which already has it computed.

Apart from eliminating extra calls to vy_stmt_hint, this is also a
prerequisite 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                 | 10 ++++++----
 src/box/vy_lsm.c                | 24 +++++++++++++-----------
 src/box/vy_lsm.h                | 10 +++++++---
 src/box/vy_mem.c                | 12 ++++--------
 src/box/vy_mem.h                | 11 +++++++----
 src/box/vy_tx.c                 | 16 ++++++++++------
 test/unit/vy_iterators_helper.c |  5 +++--
 test/unit/vy_mem.c              | 13 ++++++++-----
 8 files changed, 58 insertions(+), 43 deletions(-)

diff --git a/src/box/vinyl.c b/src/box/vinyl.c
index d8e22eb8..3e96f5f5 100644
--- a/src/box/vinyl.c
+++ b/src/box/vinyl.c
@@ -3583,7 +3583,7 @@ vy_squash_process(struct vy_squash *squash)
 	 */
 	size_t mem_used_before = lsregion_used(&env->mem_env.allocator);
 	const struct tuple *region_stmt = NULL;
-	int rc = vy_lsm_set(lsm, mem, result, &region_stmt);
+	int rc = vy_lsm_set(lsm, mem, result, hint, &region_stmt);
 	tuple_unref(result);
 	size_t mem_used_after = lsregion_used(&env->mem_env.allocator);
 	assert(mem_used_after >= mem_used_before);
@@ -4006,12 +4006,13 @@ static int
 vy_build_insert_stmt(struct vy_lsm *lsm, struct vy_mem *mem,
 		     const struct tuple *stmt, int64_t lsn)
 {
+	hint_t hint = vy_stmt_hint(stmt, lsm->cmp_def);
 	const struct tuple *region_stmt = vy_stmt_dup_lsregion(stmt,
 				&mem->env->allocator, mem->generation);
 	if (region_stmt == NULL)
 		return -1;
 	vy_stmt_set_lsn((struct tuple *)region_stmt, lsn);
-	if (vy_mem_insert(mem, region_stmt) != 0)
+	if (vy_mem_insert(mem, region_stmt, hint) != 0)
 		return -1;
 	vy_mem_commit_stmt(mem, region_stmt);
 	vy_stmt_counter_acct_tuple(&lsm->stat.memory.count, region_stmt);
@@ -4470,10 +4471,11 @@ vy_deferred_delete_on_replace(struct trigger *trigger, void *event)
 				break;
 			mem = lsm->mem;
 		}
-		rc = vy_lsm_set(lsm, mem, delete, &region_stmt);
+		hint_t hint = vy_stmt_hint(delete, lsm->cmp_def);
+		rc = vy_lsm_set(lsm, mem, delete, hint, &region_stmt);
 		if (rc != 0)
 			break;
-		vy_lsm_commit_stmt(lsm, mem, region_stmt);
+		vy_lsm_commit_stmt(lsm, mem, region_stmt, hint);
 
 		if (!is_first_statement)
 			continue;
diff --git a/src/box/vy_lsm.c b/src/box/vy_lsm.c
index 52fd429d..4065f711 100644
--- a/src/box/vy_lsm.c
+++ b/src/box/vy_lsm.c
@@ -885,7 +885,8 @@ vy_lsm_delete_mem(struct vy_lsm *lsm, struct vy_mem *mem)
 
 int
 vy_lsm_set(struct vy_lsm *lsm, struct vy_mem *mem,
-	   const struct tuple *stmt, const struct tuple **region_stmt)
+	   const struct tuple *stmt, hint_t hint,
+	   const struct tuple **region_stmt)
 {
 	uint32_t format_id = stmt->format_id;
 
@@ -917,9 +918,9 @@ vy_lsm_set(struct vy_lsm *lsm, struct vy_mem *mem,
 		return -1;
 	}
 	if (vy_stmt_type(*region_stmt) != IPROTO_UPSERT)
-		return vy_mem_insert(mem, *region_stmt);
+		return vy_mem_insert(mem, *region_stmt, hint);
 	else
-		return vy_mem_insert_upsert(mem, *region_stmt);
+		return vy_mem_insert_upsert(mem, *region_stmt, hint);
 }
 
 /**
@@ -931,10 +932,11 @@ vy_lsm_set(struct vy_lsm *lsm, struct vy_mem *mem,
  * @param lsm   LSM tree the statement was committed to.
  * @param mem   In-memory tree where the statement was saved.
  * @param stmt  UPSERT statement to squash.
+ * @param hint  Statement comparison hint.
  */
 static void
 vy_lsm_commit_upsert(struct vy_lsm *lsm, struct vy_mem *mem,
-		       const struct tuple *stmt)
+		     const struct tuple *stmt, hint_t hint)
 {
 	assert(vy_stmt_type(stmt) == IPROTO_UPSERT);
 	assert(vy_stmt_lsn(stmt) < MAX_LSN);
@@ -971,7 +973,7 @@ vy_lsm_commit_upsert(struct vy_lsm *lsm, struct vy_mem *mem,
 		 * one-key continous UPSERTs sequence.
 		 */
 #ifndef NDEBUG
-		older = vy_mem_older_lsn(mem, stmt);
+		older = vy_mem_older_lsn(mem, stmt, hint);
 		assert(older != NULL && vy_stmt_type(older) == IPROTO_UPSERT &&
 		       vy_stmt_n_upserts(older) == VY_UPSERT_THRESHOLD - 1);
 #endif
@@ -1000,7 +1002,7 @@ vy_lsm_commit_upsert(struct vy_lsm *lsm, struct vy_mem *mem,
 	if (n_upserts == 0 &&
 	    lsm->stat.memory.count.rows == lsm->mem->count.rows &&
 	    lsm->run_count == 0) {
-		older = vy_mem_older_lsn(mem, stmt);
+		older = vy_mem_older_lsn(mem, stmt, hint);
 		assert(older == NULL || vy_stmt_type(older) != IPROTO_UPSERT);
 		struct tuple *upserted =
 			vy_apply_upsert(stmt, older, lsm->cmp_def, false);
@@ -1037,7 +1039,7 @@ vy_lsm_commit_upsert(struct vy_lsm *lsm, struct vy_mem *mem,
 			return;
 		}
 
-		int rc = vy_lsm_set(lsm, mem, upserted, &region_stmt);
+		int rc = vy_lsm_set(lsm, mem, upserted, hint, &region_stmt);
 		/**
 		 * Since we have already allocated mem statement and
 		 * now we replacing one statement with another, the
@@ -1052,14 +1054,14 @@ vy_lsm_commit_upsert(struct vy_lsm *lsm, struct vy_mem *mem,
 
 void
 vy_lsm_commit_stmt(struct vy_lsm *lsm, struct vy_mem *mem,
-		     const struct tuple *stmt)
+		     const struct tuple *stmt, hint_t hint)
 {
 	vy_mem_commit_stmt(mem, stmt);
 
 	lsm->stat.memory.count.rows++;
 
 	if (vy_stmt_type(stmt) == IPROTO_UPSERT)
-		vy_lsm_commit_upsert(lsm, mem, stmt);
+		vy_lsm_commit_upsert(lsm, mem, stmt, hint);
 
 	vy_stmt_counter_acct_tuple(&lsm->stat.put, stmt);
 
@@ -1069,9 +1071,9 @@ vy_lsm_commit_stmt(struct vy_lsm *lsm, struct vy_mem *mem,
 
 void
 vy_lsm_rollback_stmt(struct vy_lsm *lsm, struct vy_mem *mem,
-		       const struct tuple *stmt)
+		       const struct tuple *stmt, hint_t hint)
 {
-	vy_mem_rollback_stmt(mem, stmt);
+	vy_mem_rollback_stmt(mem, stmt, hint);
 
 	/* Invalidate cache element. */
 	vy_cache_on_write(&lsm->cache, stmt, NULL);
diff --git a/src/box/vy_lsm.h b/src/box/vy_lsm.h
index c85bd2b3..541c49f8 100644
--- a/src/box/vy_lsm.h
+++ b/src/box/vy_lsm.h
@@ -590,6 +590,7 @@ vy_lsm_force_compaction(struct vy_lsm *lsm);
  * @param lsm         LSM tree the statement is for.
  * @param mem         In-memory tree to insert the statement into.
  * @param stmt        Statement, allocated on malloc().
+ * @param hint        Statement comparison hint.
  * @param region_stmt NULL or the same statement, allocated on
  *                    lsregion.
  *
@@ -598,7 +599,8 @@ vy_lsm_force_compaction(struct vy_lsm *lsm);
  */
 int
 vy_lsm_set(struct vy_lsm *lsm, struct vy_mem *mem,
-	   const struct tuple *stmt, const struct tuple **region_stmt);
+	   const struct tuple *stmt, hint_t hint,
+	   const struct tuple **region_stmt);
 
 /**
  * Confirm that the statement stays in the in-memory index of
@@ -607,10 +609,11 @@ vy_lsm_set(struct vy_lsm *lsm, struct vy_mem *mem,
  * @param lsm   LSM tree the statement is for.
  * @param mem   In-memory tree where the statement was saved.
  * @param stmt  Statement allocated from lsregion.
+ * @param hint  Statement comparison hint.
  */
 void
 vy_lsm_commit_stmt(struct vy_lsm *lsm, struct vy_mem *mem,
-		   const struct tuple *stmt);
+		   const struct tuple *stmt, hint_t hint);
 
 /**
  * Erase a statement from the in-memory index of an LSM tree.
@@ -618,10 +621,11 @@ vy_lsm_commit_stmt(struct vy_lsm *lsm, struct vy_mem *mem,
  * @param lsm   LSM tree to erase from.
  * @param mem   In-memory tree where the statement was saved.
  * @param stmt  Statement allocated from lsregion.
+ * @param hint  Statement comparison hint.
  */
 void
 vy_lsm_rollback_stmt(struct vy_lsm *lsm, struct vy_mem *mem,
-		     const struct tuple *stmt);
+		     const struct tuple *stmt, hint_t hint);
 
 #if defined(__cplusplus)
 } /* extern "C" */
diff --git a/src/box/vy_mem.c b/src/box/vy_mem.c
index 87ad0661..4947df2a 100644
--- a/src/box/vy_mem.c
+++ b/src/box/vy_mem.c
@@ -149,9 +149,8 @@ vy_mem_delete(struct vy_mem *index)
 }
 
 const struct tuple *
-vy_mem_older_lsn(struct vy_mem *mem, const struct tuple *stmt)
+vy_mem_older_lsn(struct vy_mem *mem, const struct tuple *stmt, hint_t hint)
 {
-	hint_t hint = vy_stmt_hint(stmt, mem->cmp_def);
 	struct vy_mem_tree_key tree_key = vy_mem_tree_key(stmt, hint,
 						vy_stmt_lsn(stmt) - 1);
 	bool exact = false;
@@ -170,14 +169,13 @@ vy_mem_older_lsn(struct vy_mem *mem, const struct tuple *stmt)
 }
 
 int
-vy_mem_insert_upsert(struct vy_mem *mem, const struct tuple *stmt)
+vy_mem_insert_upsert(struct vy_mem *mem, const struct tuple *stmt, hint_t hint)
 {
 	assert(vy_stmt_type(stmt) == IPROTO_UPSERT);
 	/* Check if the statement can be inserted in the vy_mem. */
 	assert(stmt->format_id == tuple_format_id(mem->format));
 	/* The statement must be from a lsregion. */
 	assert(!vy_stmt_is_refable(stmt));
-	hint_t hint = vy_stmt_hint(stmt, mem->cmp_def);
 	size_t size = tuple_size(stmt);
 	struct vy_mem_tree_elem new = vy_mem_tree_elem(stmt, hint);
 	struct vy_mem_tree_elem replaced = vy_mem_tree_elem_invalid();
@@ -233,7 +231,7 @@ vy_mem_insert_upsert(struct vy_mem *mem, const struct tuple *stmt)
 }
 
 int
-vy_mem_insert(struct vy_mem *mem, const struct tuple *stmt)
+vy_mem_insert(struct vy_mem *mem, const struct tuple *stmt, hint_t hint)
 {
 	assert(vy_stmt_type(stmt) != IPROTO_UPSERT);
 	/* Check if the statement can be inserted in the vy_mem. */
@@ -241,7 +239,6 @@ vy_mem_insert(struct vy_mem *mem, const struct tuple *stmt)
 	       stmt->format_id == tuple_format_id(mem->format));
 	/* The statement must be from a lsregion. */
 	assert(!vy_stmt_is_refable(stmt));
-	hint_t hint = vy_stmt_hint(stmt, mem->cmp_def);
 	size_t size = tuple_size(stmt);
 	struct vy_mem_tree_elem new = vy_mem_tree_elem(stmt, hint);
 	struct vy_mem_tree_elem replaced = vy_mem_tree_elem_invalid();
@@ -281,11 +278,10 @@ vy_mem_commit_stmt(struct vy_mem *mem, const struct tuple *stmt)
 }
 
 void
-vy_mem_rollback_stmt(struct vy_mem *mem, const struct tuple *stmt)
+vy_mem_rollback_stmt(struct vy_mem *mem, const struct tuple *stmt, hint_t hint)
 {
 	/* This is the statement we've inserted before. */
 	assert(!vy_stmt_is_refable(stmt));
-	hint_t hint = vy_stmt_hint(stmt, mem->cmp_def);
 	struct vy_mem_tree_elem elem = vy_mem_tree_elem(stmt, hint);
 	int rc = vy_mem_tree_delete(&mem->tree, elem);
 	assert(rc == 0);
diff --git a/src/box/vy_mem.h b/src/box/vy_mem.h
index f31deaed..360e1d9c 100644
--- a/src/box/vy_mem.h
+++ b/src/box/vy_mem.h
@@ -309,30 +309,32 @@ vy_mem_delete(struct vy_mem *index);
  * Return the older statement for the given one.
  */
 const struct tuple *
-vy_mem_older_lsn(struct vy_mem *mem, const struct tuple *stmt);
+vy_mem_older_lsn(struct vy_mem *mem, const struct tuple *stmt, hint_t hint);
 
 /**
  * Insert a statement into the in-memory level.
  * @param mem        vy_mem.
  * @param stmt       Vinyl statement.
+ * @param hint       Statement comparison hint.
  *
  * @retval  0 Success.
  * @retval -1 Memory error.
  */
 int
-vy_mem_insert(struct vy_mem *mem, const struct tuple *stmt);
+vy_mem_insert(struct vy_mem *mem, const struct tuple *stmt, hint_t hint);
 
 /**
  * Insert an upsert statement into the mem.
  *
  * @param mem Mem to insert to.
  * @param stmt Upsert statement to insert.
+ * @param hint Statement comparison hint.
  *
  * @retval  0 Success.
  * @retval -1 Memory error.
  */
 int
-vy_mem_insert_upsert(struct vy_mem *mem, const struct tuple *stmt);
+vy_mem_insert_upsert(struct vy_mem *mem, const struct tuple *stmt, hint_t hint);
 
 /**
  * Confirm insertion of a statement into the in-memory level.
@@ -346,9 +348,10 @@ vy_mem_commit_stmt(struct vy_mem *mem, const struct tuple *stmt);
  * Remove a statement from the in-memory level.
  * @param mem        vy_mem.
  * @param stmt       Vinyl statement.
+ * @param hint       Statement comparison hint.
  */
 void
-vy_mem_rollback_stmt(struct vy_mem *mem, const struct tuple *stmt);
+vy_mem_rollback_stmt(struct vy_mem *mem, const struct tuple *stmt, hint_t hint);
 
 /**
  * Iterator for in-memory level.
diff --git a/src/box/vy_tx.c b/src/box/vy_tx.c
index d8d7c905..fd8822a1 100644
--- a/src/box/vy_tx.c
+++ b/src/box/vy_tx.c
@@ -490,6 +490,7 @@ vy_tx_write_prepare(struct txv *v)
  * @param lsm         LSM tree to write to.
  * @param mem         In-memory tree to write to.
  * @param stmt        Statement allocated with malloc().
+ * @param hint        Statement comparison hint.
  * @param region_stmt NULL or the same statement as stmt,
  *                    but allocated on lsregion.
  *
@@ -498,7 +499,8 @@ vy_tx_write_prepare(struct txv *v)
  */
 static int
 vy_tx_write(struct vy_lsm *lsm, struct vy_mem *mem,
-	    struct tuple *stmt, const struct tuple **region_stmt)
+	    struct tuple *stmt, hint_t hint,
+	    const struct tuple **region_stmt)
 {
 	assert(vy_stmt_is_refable(stmt));
 	assert(*region_stmt == NULL || !vy_stmt_is_refable(*region_stmt));
@@ -520,7 +522,7 @@ vy_tx_write(struct vy_lsm *lsm, struct vy_mem *mem,
 			tuple_unref(deleted);
 			if (applied != NULL) {
 				assert(vy_stmt_type(applied) == IPROTO_REPLACE);
-				int rc = vy_lsm_set(lsm, mem, applied,
+				int rc = vy_lsm_set(lsm, mem, applied, hint,
 						    region_stmt);
 				tuple_unref(applied);
 				return rc;
@@ -534,7 +536,7 @@ vy_tx_write(struct vy_lsm *lsm, struct vy_mem *mem,
 		/* Invalidate cache element. */
 		vy_cache_on_write(&lsm->cache, stmt, NULL);
 	}
-	return vy_lsm_set(lsm, mem, stmt, region_stmt);
+	return vy_lsm_set(lsm, mem, stmt, hint, region_stmt);
 }
 
 /**
@@ -775,7 +777,8 @@ vy_tx_prepare(struct vy_tx *tx)
 		vy_stmt_set_lsn(v->stmt, MAX_LSN + tx->psn);
 		const struct tuple **region_stmt =
 			(type == IPROTO_DELETE) ? &delete : &repsert;
-		if (vy_tx_write(lsm, v->mem, v->stmt, region_stmt) != 0)
+		if (vy_tx_write(lsm, v->mem, v->stmt, v->hint,
+				region_stmt) != 0)
 			return -1;
 		v->region_stmt = *region_stmt;
 	}
@@ -805,7 +808,8 @@ vy_tx_commit(struct vy_tx *tx, int64_t lsn)
 	stailq_foreach_entry(v, &tx->log, next_in_log) {
 		if (v->region_stmt != NULL) {
 			vy_stmt_set_lsn((struct tuple *)v->region_stmt, lsn);
-			vy_lsm_commit_stmt(v->lsm, v->mem, v->region_stmt);
+			vy_lsm_commit_stmt(v->lsm, v->mem,
+					   v->region_stmt, v->hint);
 		}
 		if (v->mem != NULL)
 			vy_mem_unpin(v->mem);
@@ -851,7 +855,7 @@ vy_tx_rollback_after_prepare(struct vy_tx *tx)
 	stailq_foreach_entry(v, &tx->log, next_in_log) {
 		if (v->region_stmt != NULL)
 			vy_lsm_rollback_stmt(v->lsm, v->mem,
-					     v->region_stmt);
+					     v->region_stmt, v->hint);
 		if (v->mem != NULL)
 			vy_mem_unpin(v->mem);
 	}
diff --git a/test/unit/vy_iterators_helper.c b/test/unit/vy_iterators_helper.c
index 173d58db..d2ac21a0 100644
--- a/test/unit/vy_iterators_helper.c
+++ b/test/unit/vy_iterators_helper.c
@@ -136,12 +136,13 @@ vy_mem_insert_template(struct vy_mem *mem, const struct vy_stmt_template *templ)
 	struct tuple *stmt = vy_new_simple_stmt(mem->format, templ);
 	struct tuple *region_stmt = vy_stmt_dup_lsregion(stmt,
 			&mem->env->allocator, mem->generation);
+	hint_t hint = vy_stmt_hint(stmt, mem->cmp_def);
 	assert(region_stmt != NULL);
 	tuple_unref(stmt);
 	if (templ->type == IPROTO_UPSERT)
-		vy_mem_insert_upsert(mem, region_stmt);
+		vy_mem_insert_upsert(mem, region_stmt, hint);
 	else
-		vy_mem_insert(mem, region_stmt);
+		vy_mem_insert(mem, region_stmt, hint);
 	return region_stmt;
 }
 
diff --git a/test/unit/vy_mem.c b/test/unit/vy_mem.c
index 4453e076..0355653a 100644
--- a/test/unit/vy_mem.c
+++ b/test/unit/vy_mem.c
@@ -27,6 +27,7 @@ test_basic(void)
 
 	/* Check dump lsn */
 	const struct tuple *stmt = vy_mem_insert_template(mem, &stmts[0]);
+	hint_t hint = vy_stmt_hint(stmt, key_def);
 	is(mem->dump_lsn, -1, "mem->dump_lsn after prepare");
 	vy_mem_commit_stmt(mem, stmt);
 	is(mem->dump_lsn, 100, "mem->dump_lsn after commit");
@@ -34,17 +35,19 @@ test_basic(void)
 	/* Check vy_mem_older_lsn */
 	const struct tuple *older = stmt;
 	stmt = vy_mem_insert_template(mem, &stmts[1]);
-	is(vy_mem_older_lsn(mem, stmt), older, "vy_mem_older_lsn 1");
-	is(vy_mem_older_lsn(mem, older), NULL, "vy_mem_older_lsn 2");
+	hint = vy_stmt_hint(stmt, key_def);
+	is(vy_mem_older_lsn(mem, stmt, hint), older, "vy_mem_older_lsn 1");
+	is(vy_mem_older_lsn(mem, older, hint), NULL, "vy_mem_older_lsn 2");
 	vy_mem_commit_stmt(mem, stmt);
 
 	/* Check rollback  */
 	const struct tuple *olderolder = stmt;
 	older = vy_mem_insert_template(mem, &stmts[2]);
 	stmt = vy_mem_insert_template(mem, &stmts[3]);
-	is(vy_mem_older_lsn(mem, stmt), older, "vy_mem_rollback 1");
-	vy_mem_rollback_stmt(mem, older);
-	is(vy_mem_older_lsn(mem, stmt), olderolder, "vy_mem_rollback 2");
+	hint = vy_stmt_hint(stmt, key_def);
+	is(vy_mem_older_lsn(mem, stmt, hint), older, "vy_mem_rollback 1");
+	vy_mem_rollback_stmt(mem, older, hint);
+	is(vy_mem_older_lsn(mem, stmt, hint), olderolder, "vy_mem_rollback 2");
 
 	/* Check version  */
 	stmt = vy_mem_insert_template(mem, &stmts[4]);
-- 
2.11.0