[PATCH 6/7] vinyl: use multikey hints while writing runs

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

Currently, we completely ignore vy_entry.hint while writing a run file,
because they only contain auxiliary information for tuple comparison.
However, soon we will use hints to store multikey offsets, which is
mandatory for extracting keys and hence writing secondary run files.
So this patch propagates vy_entry.hint as multikey offset to tuple_bloom
and tuple_extract_key in vy_run implementation.
---
 src/box/vy_run.c  | 34 ++++++++++++++++++++--------------
 src/box/vy_stmt.c | 21 +++++++++++++--------
 src/box/vy_stmt.h | 24 +++++++++++++++++++-----
 3 files changed, 52 insertions(+), 27 deletions(-)

diff --git a/src/box/vy_run.c b/src/box/vy_run.c
index 409c3d96..0b9950f6 100644
--- a/src/box/vy_run.c
+++ b/src/box/vy_run.c
@@ -1333,8 +1333,7 @@ vy_run_iterator_seek(struct vy_run_iterator *itr, struct vy_entry last,
 	/* Check the bloom filter on the first iteration. */
 	bool check_bloom = (itr->iterator_type == ITER_EQ &&
 			    itr->curr.stmt == NULL && bloom != NULL);
-	if (check_bloom && !vy_stmt_bloom_maybe_has(bloom, itr->key.stmt,
-						    itr->key_def)) {
+	if (check_bloom && !vy_bloom_maybe_has(bloom, itr->key, itr->key_def)) {
 		vy_run_iterator_stop(itr);
 		itr->stat->bloom_hit++;
 		return 0;
@@ -1773,7 +1772,9 @@ vy_run_dump_stmt(struct vy_entry entry, struct xlog *data_xlog,
 	struct xrow_header xrow;
 	int rc = (is_primary ?
 		  vy_stmt_encode_primary(entry.stmt, key_def, 0, &xrow) :
-		  vy_stmt_encode_secondary(entry.stmt, key_def, &xrow));
+		  vy_stmt_encode_secondary(entry.stmt, key_def,
+					   vy_entry_multikey_idx(entry, key_def),
+					   &xrow));
 	if (rc != 0)
 		return -1;
 
@@ -2166,8 +2167,10 @@ vy_run_writer_start_page(struct vy_run_writer *writer,
 		return -1;
 	const char *key = vy_stmt_is_key(first_entry.stmt) ?
 			  tuple_data(first_entry.stmt) :
-			  tuple_extract_key(first_entry.stmt,
-					    writer->cmp_def, -1, NULL);
+			  tuple_extract_key(first_entry.stmt, writer->cmp_def,
+					    vy_entry_multikey_idx(first_entry,
+								  writer->cmp_def),
+					    NULL);
 	if (key == NULL)
 		return -1;
 	if (run->info.page_count == 0) {
@@ -2196,8 +2199,7 @@ static int
 vy_run_writer_write_to_page(struct vy_run_writer *writer, struct vy_entry entry)
 {
 	if (writer->bloom != NULL &&
-	    vy_stmt_bloom_builder_add(writer->bloom, entry.stmt,
-				      writer->key_def) != 0)
+	    vy_bloom_builder_add(writer->bloom, entry, writer->key_def) != 0)
 		return -1;
 	if (writer->last.stmt != NULL)
 		vy_stmt_unref_if_possible(writer->last.stmt);
@@ -2320,8 +2322,10 @@ vy_run_writer_commit(struct vy_run_writer *writer)
 	assert(writer->last.stmt != NULL);
 	const char *key = vy_stmt_is_key(writer->last.stmt) ?
 		          tuple_data(writer->last.stmt) :
-			  tuple_extract_key(writer->last.stmt,
-					    writer->cmp_def, -1, NULL);
+			  tuple_extract_key(writer->last.stmt, writer->cmp_def,
+					    vy_entry_multikey_idx(writer->last,
+								  writer->cmp_def),
+					    NULL);
 	if (key == NULL)
 		goto out;
 
@@ -2424,11 +2428,13 @@ vy_run_rebuild_index(struct vy_run *run, const char *dir,
 			struct tuple *tuple = vy_stmt_decode(&xrow, format);
 			if (tuple == NULL)
 				goto close_err;
-			if (bloom_builder != NULL &&
-			    vy_stmt_bloom_builder_add(bloom_builder, tuple,
-						      key_def) != 0) {
-				tuple_unref(tuple);
-				goto close_err;
+			if (bloom_builder != NULL) {
+				struct vy_entry entry = {tuple, HINT_NONE};
+				if (vy_bloom_builder_add(bloom_builder, entry,
+							 key_def) != 0) {
+					tuple_unref(tuple);
+					goto close_err;
+				}
 			}
 			key = vy_stmt_is_key(tuple) ? tuple_data(tuple) :
 			      tuple_extract_key(tuple, cmp_def, -1, NULL);
diff --git a/src/box/vy_stmt.c b/src/box/vy_stmt.c
index 170c258a..8ce785a5 100644
--- a/src/box/vy_stmt.c
+++ b/src/box/vy_stmt.c
@@ -527,30 +527,34 @@ vy_stmt_extract_key_raw(const char *data, const char *data_end,
 }
 
 int
-vy_stmt_bloom_builder_add(struct tuple_bloom_builder *builder,
-			  struct tuple *stmt, struct key_def *key_def)
+vy_bloom_builder_add(struct tuple_bloom_builder *builder,
+		     struct vy_entry entry, struct key_def *key_def)
 {
+	struct tuple *stmt = entry.stmt;
 	if (vy_stmt_is_key(stmt)) {
 		const char *data = tuple_data(stmt);
 		uint32_t part_count = mp_decode_array(&data);
 		return tuple_bloom_builder_add_key(builder, data,
 						   part_count, key_def);
 	} else {
-		return tuple_bloom_builder_add(builder, stmt, key_def, -1);
+		return tuple_bloom_builder_add(builder, stmt, key_def,
+				vy_entry_multikey_idx(entry, key_def));
 	}
 }
 
 bool
-vy_stmt_bloom_maybe_has(const struct tuple_bloom *bloom,
-			struct tuple *stmt, struct key_def *key_def)
+vy_bloom_maybe_has(const struct tuple_bloom *bloom,
+		   struct vy_entry entry, struct key_def *key_def)
 {
+	struct tuple *stmt = entry.stmt;
 	if (vy_stmt_is_key(stmt)) {
 		const char *data = tuple_data(stmt);
 		uint32_t part_count = mp_decode_array(&data);
 		return tuple_bloom_maybe_has_key(bloom, data,
 						 part_count, key_def);
 	} else {
-		return tuple_bloom_maybe_has(bloom, stmt, key_def, -1);
+		return tuple_bloom_maybe_has(bloom, stmt, key_def,
+				vy_entry_multikey_idx(entry, key_def));
 	}
 }
 
@@ -656,7 +660,7 @@ vy_stmt_encode_primary(struct tuple *value, struct key_def *key_def,
 
 int
 vy_stmt_encode_secondary(struct tuple *value, struct key_def *cmp_def,
-			 struct xrow_header *xrow)
+			 int multikey_idx, struct xrow_header *xrow)
 {
 	memset(xrow, 0, sizeof(*xrow));
 	enum iproto_type type = vy_stmt_type(value);
@@ -669,7 +673,8 @@ vy_stmt_encode_secondary(struct tuple *value, struct key_def *cmp_def,
 	uint32_t size;
 	const char *extracted = vy_stmt_is_key(value) ?
 				tuple_data_range(value, &size) :
-				tuple_extract_key(value, cmp_def, -1, &size);
+				tuple_extract_key(value, cmp_def,
+						  multikey_idx, &size);
 	if (extracted == NULL)
 		return -1;
 	if (type == IPROTO_REPLACE || type == IPROTO_INSERT) {
diff --git a/src/box/vy_stmt.h b/src/box/vy_stmt.h
index 929e537a..b0731d3d 100644
--- a/src/box/vy_stmt.h
+++ b/src/box/vy_stmt.h
@@ -623,16 +623,16 @@ vy_stmt_extract_key_raw(const char *data, const char *data_end,
  * See tuple_bloom_builder_add() for more details.
  */
 int
-vy_stmt_bloom_builder_add(struct tuple_bloom_builder *builder,
-			  struct tuple *stmt, struct key_def *key_def);
+vy_bloom_builder_add(struct tuple_bloom_builder *builder,
+		     struct vy_entry entry, struct key_def *key_def);
 
 /**
  * Check if a statement hash is present in a bloom filter.
  * See tuple_bloom_maybe_has() for more details.
  */
 bool
-vy_stmt_bloom_maybe_has(const struct tuple_bloom *bloom,
-			struct tuple *stmt, struct key_def *key_def);
+vy_bloom_maybe_has(const struct tuple_bloom *bloom,
+		   struct vy_entry entry, struct key_def *key_def);
 
 /**
  * Encode vy_stmt for a primary key as xrow_header
@@ -655,6 +655,7 @@ vy_stmt_encode_primary(struct tuple *value, struct key_def *key_def,
  *
  * @param value statement to encode
  * @param key_def key definition
+ * @param multikey_idx multikey index hint
  * @param xrow[out] xrow to fill
  *
  * @retval 0 if OK
@@ -662,7 +663,7 @@ vy_stmt_encode_primary(struct tuple *value, struct key_def *key_def,
  */
 int
 vy_stmt_encode_secondary(struct tuple *value, struct key_def *cmp_def,
-			 struct xrow_header *xrow);
+			 int multikey_idx, struct xrow_header *xrow);
 
 /**
  * Reconstruct vinyl tuple info and data from xrow
@@ -689,6 +690,19 @@ const char *
 vy_stmt_str(struct tuple *stmt);
 
 /**
+ * Extract a multikey index hint from a statement entry.
+ * Returns -1 if the key definition isn't multikey.
+ */
+static inline int
+vy_entry_multikey_idx(struct vy_entry entry, struct key_def *key_def)
+{
+	if (!key_def_is_multikey(key_def) || vy_stmt_is_key(entry.stmt))
+		return -1;
+	assert(entry.hint != HINT_NONE);
+	return (int)entry.hint;
+}
+
+/**
  * Create a key entry from a MessagePack array without a header.
  */
 static inline struct vy_entry
-- 
2.11.0