[RFC PATCH 11/23] xrow: allow to store flags in DML requests

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

In the scope of #2129 we need to mark REPLACE statements for which we
generated DELETE in secondary indexes so that we don't generate DELETE
again on compaction. We also need to mark DELETE statements that were
generated on compaction so that we can skip them on SELECT.

Let's add flags field to struct vy_stmt. Flags are stored both in memory
and on disk so to encode/decode them we also need to add a new iproto
key (IPROTO_FLAGS) and the corresponding field to struct request.

Needed for #2129
---
 src/box/iproto_constants.c |  4 ++--
 src/box/iproto_constants.h |  3 ++-
 src/box/vy_stmt.c          |  4 ++++
 src/box/vy_stmt.h          | 15 +++++++++++++++
 src/box/xrow.c             |  8 ++++++++
 src/box/xrow.h             |  2 ++
 6 files changed, 33 insertions(+), 3 deletions(-)

diff --git a/src/box/iproto_constants.c b/src/box/iproto_constants.c
index 3adb7cd4..651e07b3 100644
--- a/src/box/iproto_constants.c
+++ b/src/box/iproto_constants.c
@@ -61,10 +61,10 @@ const unsigned char iproto_key_type[IPROTO_KEY_MAX] =
 		/* 0x13 */	MP_UINT, /* IPROTO_OFFSET */
 		/* 0x14 */	MP_UINT, /* IPROTO_ITERATOR */
 		/* 0x15 */	MP_UINT, /* IPROTO_INDEX_BASE */
+		/* 0x16 */	MP_UINT, /* IPROTO_FLAGS */
 	/* }}} */
 
 	/* {{{ unused */
-		/* 0x16 */	MP_UINT,
 		/* 0x17 */	MP_UINT,
 		/* 0x18 */	MP_UINT,
 		/* 0x19 */	MP_UINT,
@@ -148,7 +148,7 @@ const char *iproto_key_strs[IPROTO_KEY_MAX] = {
 	"offset",           /* 0x13 */
 	"iterator",         /* 0x14 */
 	"index base",       /* 0x15 */
-	NULL,               /* 0x16 */
+	"flags",            /* 0x16 */
 	NULL,               /* 0x17 */
 	NULL,               /* 0x18 */
 	NULL,               /* 0x19 */
diff --git a/src/box/iproto_constants.h b/src/box/iproto_constants.h
index d1320de7..f11c7fa9 100644
--- a/src/box/iproto_constants.h
+++ b/src/box/iproto_constants.h
@@ -65,6 +65,7 @@ enum iproto_key {
 	IPROTO_OFFSET = 0x13,
 	IPROTO_ITERATOR = 0x14,
 	IPROTO_INDEX_BASE = 0x15,
+	IPROTO_FLAGS = 0x16,
 	/* Leave a gap between integer values and other keys */
 	IPROTO_KEY = 0x20,
 	IPROTO_TUPLE = 0x21,
@@ -89,7 +90,7 @@ enum iproto_key {
 			  bit(LSN) | bit(SCHEMA_VERSION))
 #define IPROTO_DML_BODY_BMAP (bit(SPACE_ID) | bit(INDEX_ID) | bit(LIMIT) |\
 			      bit(OFFSET) | bit(ITERATOR) | bit(INDEX_BASE) |\
-			      bit(KEY) | bit(TUPLE) | bit(OPS))
+			      bit(KEY) | bit(TUPLE) | bit(OPS) | bit(FLAGS))
 
 static inline bool
 xrow_header_has_key(const char *pos, const char *end)
diff --git a/src/box/vy_stmt.c b/src/box/vy_stmt.c
index a4b7975b..09daa7f4 100644
--- a/src/box/vy_stmt.c
+++ b/src/box/vy_stmt.c
@@ -112,6 +112,7 @@ vy_stmt_alloc(struct tuple_format *format, uint32_t bsize)
 	tuple->data_offset = sizeof(struct vy_stmt) + meta_size;;
 	vy_stmt_set_lsn(tuple, 0);
 	vy_stmt_set_type(tuple, 0);
+	vy_stmt_set_flags(tuple, 0);
 	return tuple;
 }
 
@@ -498,6 +499,7 @@ vy_stmt_encode_primary(const struct tuple *value,
 	struct request request;
 	memset(&request, 0, sizeof(request));
 	request.type = type;
+	request.flags = vy_stmt_flags(value);
 	request.space_id = space_id;
 	uint32_t size;
 	const char *extracted = NULL;
@@ -544,6 +546,7 @@ vy_stmt_encode_secondary(const struct tuple *value,
 	struct request request;
 	memset(&request, 0, sizeof(request));
 	request.type = type;
+	request.flags = vy_stmt_flags(value);
 	uint32_t size;
 	const char *extracted = tuple_extract_key(value, cmp_def, &size);
 	if (extracted == NULL)
@@ -614,6 +617,7 @@ vy_stmt_decode(struct xrow_header *xrow, const struct key_def *key_def,
 		return NULL; /* OOM */
 
 	vy_stmt_set_lsn(stmt, xrow->lsn);
+	vy_stmt_set_flags(stmt, request.flags);
 	return stmt;
 }
 
diff --git a/src/box/vy_stmt.h b/src/box/vy_stmt.h
index e53f98ce..bcf855dd 100644
--- a/src/box/vy_stmt.h
+++ b/src/box/vy_stmt.h
@@ -103,6 +103,7 @@ struct vy_stmt {
 	struct tuple base;
 	int64_t lsn;
 	uint8_t  type; /* IPROTO_SELECT/REPLACE/UPSERT/DELETE */
+	uint8_t flags;
 	/**
 	 * Offsets array concatenated with MessagePack fields
 	 * array.
@@ -138,6 +139,20 @@ vy_stmt_set_type(struct tuple *stmt, enum iproto_type type)
 	((struct vy_stmt *) stmt)->type = type;
 }
 
+/** Get flags of the vinyl statement. */
+static inline uint8_t
+vy_stmt_flags(const struct tuple *stmt)
+{
+	return ((const struct vy_stmt *)stmt)->flags;
+}
+
+/** Set flags of the vinyl statement. */
+static inline void
+vy_stmt_set_flags(struct tuple *stmt, uint8_t flags)
+{
+	((struct vy_stmt *)stmt)->flags = flags;
+}
+
 /**
  * Get upserts count of the vinyl statement.
  * Only for UPSERT statements allocated on lsregion.
diff --git a/src/box/xrow.c b/src/box/xrow.c
index 48fbff27..b74925e0 100644
--- a/src/box/xrow.c
+++ b/src/box/xrow.c
@@ -454,6 +454,9 @@ error:
 		case IPROTO_INDEX_BASE:
 			request->index_base = mp_decode_uint(&value);
 			break;
+		case IPROTO_FLAGS:
+			request->flags = mp_decode_uint(&value);
+			break;
 		case IPROTO_LIMIT:
 			request->limit = mp_decode_uint(&value);
 			break;
@@ -547,6 +550,11 @@ xrow_encode_dml(const struct request *request, struct iovec *iov)
 		pos = mp_encode_uint(pos, request->index_base);
 		map_size++;
 	}
+	if (request->flags) {
+		pos = mp_encode_uint(pos, IPROTO_FLAGS);
+		pos = mp_encode_uint(pos, request->flags);
+		map_size++;
+	}
 	if (request->key) {
 		pos = mp_encode_uint(pos, IPROTO_KEY);
 		memcpy(pos, request->key, key_len);
diff --git a/src/box/xrow.h b/src/box/xrow.h
index 1bb5f103..c69e450c 100644
--- a/src/box/xrow.h
+++ b/src/box/xrow.h
@@ -128,6 +128,8 @@ struct request {
 	const char *ops_end;
 	/** Base field offset for UPDATE/UPSERT, e.g. 0 for C and 1 for Lua. */
 	int index_base;
+	/** Engine-specific statement flags. */
+	uint8_t flags;
 };
 
 /**
-- 
2.11.0