[PATCH 11/12] vinyl: don't use IPROTO_SELECT type for key statements

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

To differentiate between key and tuple statements in comparators, we set
IPROTO_SELECT type for key statements. As a result, we can't use key
statements in the run iterator directly although secondary index runs do
store statements in key format. Instead we create surrogate tuples
filling missing fields with NULLs. This won't play nicely with multikey
indexes so we need to teach iterators to deal with statements in key
format. The first step in this direction is dropping IPROTO_SELECT and
starting to identify key statements by format instead.
---
 src/box/vy_run.c   |  2 +-
 src/box/vy_stmt.c  |  6 +++++-
 src/box/vy_stmt.h  | 38 +++++++++++++++++++++++++-------------
 test/unit/vy_mem.c |  2 +-
 4 files changed, 32 insertions(+), 16 deletions(-)

diff --git a/src/box/vy_run.c b/src/box/vy_run.c
index 06ef299e..e3470c09 100644
--- a/src/box/vy_run.c
+++ b/src/box/vy_run.c
@@ -1250,7 +1250,7 @@ vy_run_iterator_do_seek(struct vy_run_iterator *itr,
 	struct key_def *key_def = itr->key_def;
 	if (iterator_type == ITER_EQ && bloom != NULL) {
 		bool need_lookup;
-		if (vy_stmt_type(key) == IPROTO_SELECT) {
+		if (vy_stmt_is_key(key)) {
 			const char *data = tuple_data(key);
 			uint32_t part_count = mp_decode_array(&data);
 			need_lookup = tuple_bloom_maybe_has_key(bloom, data,
diff --git a/src/box/vy_stmt.c b/src/box/vy_stmt.c
index 5ac7d833..5a76bfe4 100644
--- a/src/box/vy_stmt.c
+++ b/src/box/vy_stmt.c
@@ -240,6 +240,7 @@ struct tuple *
 vy_stmt_new_key(struct tuple_format *format, const char *key,
 		uint32_t part_count)
 {
+	assert(vy_stmt_is_key_format(format));
 	assert(part_count == 0 || key != NULL);
 	/* Key don't have field map */
 	assert(format->field_map_size == 0);
@@ -260,7 +261,6 @@ vy_stmt_new_key(struct tuple_format *format, const char *key,
 	char *data = mp_encode_array(raw, part_count);
 	memcpy(data, key, key_size);
 	assert(data + key_size == raw + bsize);
-	vy_stmt_set_type(stmt, IPROTO_SELECT);
 	return stmt;
 }
 
@@ -839,6 +839,10 @@ vy_stmt_snprint(char *buf, int size, const struct tuple *stmt)
 		SNPRINT(total, snprintf, buf, size, "<NULL>");
 		return total;
 	}
+	if (vy_stmt_type(stmt) == 0) {
+		SNPRINT(total, mp_snprint, buf, size, tuple_data(stmt));
+		return total;
+	}
 	SNPRINT(total, snprintf, buf, size, "%s(",
 		iproto_type_name(vy_stmt_type(stmt)));
 		SNPRINT(total, mp_snprint, buf, size, tuple_data(stmt));
diff --git a/src/box/vy_stmt.h b/src/box/vy_stmt.h
index 6c834aee..88d74de7 100644
--- a/src/box/vy_stmt.h
+++ b/src/box/vy_stmt.h
@@ -133,12 +133,9 @@ enum {
 };
 
 /**
- * There are two groups of statements:
+ * A vinyl statement can have either key or tuple format.
  *
- *  - SELECT is "key" statement.
- *  - DELETE, UPSERT and REPLACE are "tuple" statements.
- *
- * REPLACE/UPSERT/DELETE statements structure:
+ * Tuple statement structure:
  *                               data_offset
  *                                    ^
  * +----------------------------------+
@@ -154,7 +151,7 @@ enum {
  * indexed then before MessagePack data are stored offsets only for field 3 and
  * field 5.
  *
- * SELECT statements structure.
+ * Key statement structure:
  * +--------------+-----------------+
  * | array header | part1 ... partN |  -  MessagePack data
  * +--------------+-----------------+
@@ -164,7 +161,7 @@ enum {
 struct vy_stmt {
 	struct tuple base;
 	int64_t lsn;
-	uint8_t  type; /* IPROTO_SELECT/REPLACE/UPSERT/DELETE */
+	uint8_t  type; /* IPROTO_INSERT/REPLACE/UPSERT/DELETE */
 	uint8_t flags;
 	/**
 	 * Offsets array concatenated with MessagePack fields
@@ -239,6 +236,21 @@ vy_stmt_set_n_upserts(struct tuple *stmt, uint8_t n)
 	*((uint8_t *)stmt - 1) = n;
 }
 
+/** Return true if the given format is a key format. */
+static inline bool
+vy_stmt_is_key_format(const struct tuple_format *format)
+{
+	struct vy_stmt_env *env = format->engine;
+	return env->key_format == format;
+}
+
+/** Return true if the vinyl statement has key format. */
+static inline bool
+vy_stmt_is_key(const struct tuple *stmt)
+{
+	return vy_stmt_is_key_format(tuple_format(stmt));
+}
+
 /**
  * Return the number of key parts defined in the given vinyl
  * statement.
@@ -249,7 +261,7 @@ vy_stmt_set_n_upserts(struct tuple *stmt, uint8_t n)
 static inline uint32_t
 vy_stmt_key_part_count(const struct tuple *stmt, struct key_def *key_def)
 {
-	if (vy_stmt_type(stmt) == IPROTO_SELECT) {
+	if (vy_stmt_is_key(stmt)) {
 		uint32_t part_count = tuple_field_count(stmt);
 		assert(part_count <= key_def->part_count);
 		return part_count;
@@ -348,8 +360,8 @@ static inline int
 vy_stmt_compare(const struct tuple *a, const struct tuple *b,
 		struct key_def *key_def)
 {
-	bool a_is_tuple = vy_stmt_type(a) != IPROTO_SELECT;
-	bool b_is_tuple = vy_stmt_type(b) != IPROTO_SELECT;
+	bool a_is_tuple = !vy_stmt_is_key(a);
+	bool b_is_tuple = !vy_stmt_is_key(b);
 	if (a_is_tuple && b_is_tuple) {
 		return tuple_compare(a, b, key_def);
 	} else if (a_is_tuple && !b_is_tuple) {
@@ -374,7 +386,7 @@ static inline int
 vy_stmt_compare_with_raw_key(const struct tuple *stmt, const char *key,
 			     struct key_def *key_def)
 {
-	if (vy_stmt_type(stmt) != IPROTO_SELECT) {
+	if (!vy_stmt_is_key(stmt)) {
 		uint32_t part_count = mp_decode_array(&key);
 		return tuple_compare_with_key(stmt, key, part_count, key_def);
 	}
@@ -557,7 +569,7 @@ vy_stmt_new_key_from_array(struct tuple_format *format, const char *key)
 
 /**
  * Extract the key from a tuple by the given key definition
- * and store the result in a SELECT statement allocated with
+ * and store the result in a key statement allocated with
  * malloc().
  */
 struct tuple *
@@ -566,7 +578,7 @@ vy_stmt_extract_key(const struct tuple *stmt, struct key_def *key_def,
 
 /**
  * Extract the key from msgpack by the given key definition
- * and store the result in a SELECT statement allocated with
+ * and store the result in a key statement allocated with
  * malloc().
  */
 struct tuple *
diff --git a/test/unit/vy_mem.c b/test/unit/vy_mem.c
index da4117c5..b672ee91 100644
--- a/test/unit/vy_mem.c
+++ b/test/unit/vy_mem.c
@@ -86,7 +86,7 @@ test_iterator_restore_after_insertion()
 	struct slab_cache *slab_cache = cord_slab_cache();
 	lsregion_create(&lsregion, slab_cache->arena);
 
-	struct tuple *select_key = vy_stmt_new_key(format, NULL, 0);
+	struct tuple *select_key = vy_stmt_new_key(stmt_env.key_format, NULL, 0);
 
 	struct mempool history_node_pool;
 	mempool_create(&history_node_pool, cord_slab_cache(),
-- 
2.11.0