[PATCH 2/3] vinyl: account disk statements of each type

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

This patch adds a new entry to per index statistics reported by
index.stat():

  disk.statement
    inserts
    replaces
    deletes
    upserts

It shows the number of statements of each type stored in run files.
The new statistics are persisted in index files. We will need this
information so that we can force major compaction when there are too
many DELETE statements accumulated in run files.

Needed for #3225
---
 src/box/iproto_constants.c |   1 +
 src/box/iproto_constants.h |   2 +
 src/box/vinyl.c            |   6 +++
 src/box/vy_lsm.c           |   2 +
 src/box/vy_run.c           |  68 ++++++++++++++++++++++++++-
 src/box/vy_run.h           |   2 +
 src/box/vy_stat.h          |  59 +++++++++++++++++++++++
 test/vinyl/info.result     | 113 ++++++++++++++++++++++++++++++++++++++++++---
 test/vinyl/info.test.lua   |  36 +++++++++++++++
 test/vinyl/layout.result   |  12 +++--
 10 files changed, 289 insertions(+), 12 deletions(-)

diff --git a/src/box/iproto_constants.c b/src/box/iproto_constants.c
index 4f3510e4..7fd29577 100644
--- a/src/box/iproto_constants.c
+++ b/src/box/iproto_constants.c
@@ -216,6 +216,7 @@ const char *vy_run_info_key_strs[VY_RUN_INFO_KEY_MAX] = {
 	"page count",
 	"bloom filter legacy",
 	"bloom filter",
+	"stmt stat",
 };
 
 const char *vy_row_index_key_strs[VY_ROW_INDEX_KEY_MAX] = {
diff --git a/src/box/iproto_constants.h b/src/box/iproto_constants.h
index f571375e..ab7db1f0 100644
--- a/src/box/iproto_constants.h
+++ b/src/box/iproto_constants.h
@@ -354,6 +354,8 @@ enum vy_run_info_key {
 	VY_RUN_INFO_BLOOM_LEGACY = 6,
 	/** Bloom filter for keys. */
 	VY_RUN_INFO_BLOOM = 7,
+	/** Number of statements of each type (map). */
+	VY_RUN_INFO_STMT_STAT = 8,
 	/** The last key in this enum + 1 */
 	VY_RUN_INFO_KEY_MAX
 };
diff --git a/src/box/vinyl.c b/src/box/vinyl.c
index 09b66559..e9df7463 100644
--- a/src/box/vinyl.c
+++ b/src/box/vinyl.c
@@ -397,6 +397,12 @@ vinyl_index_stat(struct index *index, struct info_handler *h)
 
 	info_table_begin(h, "disk");
 	vy_info_append_disk_stmt_counter(h, NULL, &stat->disk.count);
+	info_table_begin(h, "statement");
+	info_append_int(h, "inserts", stat->disk.stmt.inserts);
+	info_append_int(h, "replaces", stat->disk.stmt.replaces);
+	info_append_int(h, "deletes", stat->disk.stmt.deletes);
+	info_append_int(h, "upserts", stat->disk.stmt.upserts);
+	info_table_end(h); /* statement */
 	info_table_begin(h, "iterator");
 	info_append_int(h, "lookup", stat->disk.iterator.lookup);
 	vy_info_append_stmt_counter(h, "get", &stat->disk.iterator.get);
diff --git a/src/box/vy_lsm.c b/src/box/vy_lsm.c
index de5ad314..7e6bc3c2 100644
--- a/src/box/vy_lsm.c
+++ b/src/box/vy_lsm.c
@@ -695,6 +695,7 @@ vy_lsm_add_run(struct vy_lsm *lsm, struct vy_run *run)
 	rlist_add_entry(&lsm->runs, run, in_lsm);
 	lsm->run_count++;
 	vy_disk_stmt_counter_add(&lsm->stat.disk.count, &run->count);
+	vy_stmt_stat_add(&lsm->stat.disk.stmt, &run->info.stmt_stat);
 
 	lsm->bloom_size += bloom_size;
 	lsm->page_index_size += page_index_size;
@@ -723,6 +724,7 @@ vy_lsm_remove_run(struct vy_lsm *lsm, struct vy_run *run)
 	rlist_del_entry(run, in_lsm);
 	lsm->run_count--;
 	vy_disk_stmt_counter_sub(&lsm->stat.disk.count, &run->count);
+	vy_stmt_stat_sub(&lsm->stat.disk.stmt, &run->info.stmt_stat);
 
 	lsm->bloom_size -= bloom_size;
 	lsm->page_index_size -= page_index_size;
diff --git a/src/box/vy_run.c b/src/box/vy_run.c
index bb5baf2c..5f2b980b 100644
--- a/src/box/vy_run.c
+++ b/src/box/vy_run.c
@@ -542,6 +542,33 @@ vy_page_info_decode(struct vy_page_info *page, const struct xrow_header *xrow,
 	return 0;
 }
 
+/** Decode statement statistics from @data and advance @data. */
+static void
+vy_stmt_stat_decode(struct vy_stmt_stat *stat, const char **data)
+{
+	uint32_t size = mp_decode_map(data);
+	for (uint32_t i = 0; i < size; i++) {
+		uint64_t key = mp_decode_uint(data);
+		uint64_t value = mp_decode_uint(data);
+		switch (key) {
+		case IPROTO_INSERT:
+			stat->inserts = value;
+			break;
+		case IPROTO_REPLACE:
+			stat->replaces = value;
+			break;
+		case IPROTO_DELETE:
+			stat->deletes = value;
+			break;
+		case IPROTO_UPSERT:
+			stat->upserts = value;
+			break;
+		default:
+			break;
+		}
+	}
+}
+
 /**
  * Decode the run metadata from xrow.
  *
@@ -603,6 +630,9 @@ vy_run_info_decode(struct vy_run_info *run_info,
 			if (run_info->bloom == NULL)
 				return -1;
 			break;
+		case VY_RUN_INFO_STMT_STAT:
+			vy_stmt_stat_decode(&run_info->stmt_stat, &pos);
+			break;
 		default:
 			diag_set(ClientError, ER_INVALID_INDEX_FILE, filename,
 				"Can't decode run info: unknown key %u",
@@ -1876,6 +1906,37 @@ vy_page_info_encode(const struct vy_page_info *page_info,
 
 /** {{{ vy_run_info */
 
+/** Return the size of encoded statement statistics. */
+static size_t
+vy_stmt_stat_sizeof(const struct vy_stmt_stat *stat)
+{
+	return mp_sizeof_map(4) +
+		mp_sizeof_uint(IPROTO_INSERT) +
+		mp_sizeof_uint(IPROTO_REPLACE) +
+		mp_sizeof_uint(IPROTO_DELETE) +
+		mp_sizeof_uint(IPROTO_UPSERT) +
+		mp_sizeof_uint(stat->inserts) +
+		mp_sizeof_uint(stat->replaces) +
+		mp_sizeof_uint(stat->deletes) +
+		mp_sizeof_uint(stat->upserts);
+}
+
+/** Encode statement statistics to @buf and return advanced @buf. */
+static char *
+vy_stmt_stat_encode(const struct vy_stmt_stat *stat, char *buf)
+{
+	buf = mp_encode_map(buf, 4);
+	buf = mp_encode_uint(buf, IPROTO_INSERT);
+	buf = mp_encode_uint(buf, stat->inserts);
+	buf = mp_encode_uint(buf, IPROTO_REPLACE);
+	buf = mp_encode_uint(buf, stat->replaces);
+	buf = mp_encode_uint(buf, IPROTO_DELETE);
+	buf = mp_encode_uint(buf, stat->deletes);
+	buf = mp_encode_uint(buf, IPROTO_UPSERT);
+	buf = mp_encode_uint(buf, stat->upserts);
+	return buf;
+}
+
 /**
  * Encode vy_run_info as xrow
  * Allocates using region alloc
@@ -1898,7 +1959,7 @@ vy_run_info_encode(const struct vy_run_info *run_info,
 	mp_next(&tmp);
 	size_t max_key_size = tmp - run_info->max_key;
 
-	uint32_t key_count = 5;
+	uint32_t key_count = 6;
 	if (run_info->bloom != NULL)
 		key_count++;
 
@@ -1914,6 +1975,8 @@ vy_run_info_encode(const struct vy_run_info *run_info,
 	if (run_info->bloom != NULL)
 		size += mp_sizeof_uint(VY_RUN_INFO_BLOOM) +
 			tuple_bloom_size(run_info->bloom);
+	size += mp_sizeof_uint(VY_RUN_INFO_STMT_STAT) +
+		vy_stmt_stat_sizeof(&run_info->stmt_stat);
 
 	char *pos = region_alloc(&fiber()->gc, size);
 	if (pos == NULL) {
@@ -1940,6 +2003,8 @@ vy_run_info_encode(const struct vy_run_info *run_info,
 		pos = mp_encode_uint(pos, VY_RUN_INFO_BLOOM);
 		pos = tuple_bloom_encode(run_info->bloom, pos);
 	}
+	pos = mp_encode_uint(pos, VY_RUN_INFO_STMT_STAT);
+	pos = vy_stmt_stat_encode(&run_info->stmt_stat, pos);
 	xrow->body->iov_len = (void *)pos - xrow->body->iov_base;
 	xrow->bodycnt = 1;
 	xrow->type = VY_INDEX_RUN_INFO;
@@ -2136,6 +2201,7 @@ vy_run_writer_write_to_page(struct vy_run_writer *writer, struct tuple *stmt)
 	int64_t lsn = vy_stmt_lsn(stmt);
 	run->info.min_lsn = MIN(run->info.min_lsn, lsn);
 	run->info.max_lsn = MAX(run->info.max_lsn, lsn);
+	vy_stmt_stat_acct(&run->info.stmt_stat, vy_stmt_type(stmt));
 	return 0;
 }
 
diff --git a/src/box/vy_run.h b/src/box/vy_run.h
index 4b919a8f..990daffa 100644
--- a/src/box/vy_run.h
+++ b/src/box/vy_run.h
@@ -87,6 +87,8 @@ struct vy_run_info {
 	uint32_t page_count;
 	/** Bloom filter of all tuples in run */
 	struct tuple_bloom *bloom;
+	/** Statement statistics. */
+	struct vy_stmt_stat stmt_stat;
 };
 
 /**
diff --git a/src/box/vy_stat.h b/src/box/vy_stat.h
index 1545115a..9bb09174 100644
--- a/src/box/vy_stat.h
+++ b/src/box/vy_stat.h
@@ -36,11 +36,20 @@
 
 #include "latency.h"
 #include "tuple.h"
+#include "iproto_constants.h"
 
 #if defined(__cplusplus)
 extern "C" {
 #endif /* defined(__cplusplus) */
 
+/** Number of statements of each type. */
+struct vy_stmt_stat {
+	int64_t inserts;
+	int64_t replaces;
+	int64_t deletes;
+	int64_t upserts;
+};
+
 /** Used for accounting statements stored in memory. */
 struct vy_stmt_counter {
 	/** Number of statements. */
@@ -130,6 +139,8 @@ struct vy_lsm_stat {
 	struct {
 		/** Number of statements stored on disk. */
 		struct vy_disk_stmt_counter count;
+		/** Statement statistics. */
+		struct vy_stmt_stat stmt;
 		/** Run iterator statistics. */
 		struct vy_run_iterator_stat iterator;
 		/** Dump statistics. */
@@ -298,6 +309,54 @@ vy_disk_stmt_counter_sub(struct vy_disk_stmt_counter *c1,
 	c1->pages -= c2->pages;
 }
 
+/**
+ * Account a single statement of the given type in @stat.
+ */
+static inline void
+vy_stmt_stat_acct(struct vy_stmt_stat *stat, enum iproto_type type)
+{
+	switch (type) {
+	case IPROTO_INSERT:
+		stat->inserts++;
+		break;
+	case IPROTO_REPLACE:
+		stat->replaces++;
+		break;
+	case IPROTO_DELETE:
+		stat->deletes++;
+		break;
+	case IPROTO_UPSERT:
+		stat->upserts++;
+		break;
+	default:
+		break;
+	}
+}
+
+/**
+ * Add statistics accumulated in @s2 to @s1.
+ */
+static inline void
+vy_stmt_stat_add(struct vy_stmt_stat *s1, const struct vy_stmt_stat *s2)
+{
+	s1->inserts += s2->inserts;
+	s1->replaces += s2->replaces;
+	s1->deletes += s2->deletes;
+	s1->upserts += s2->upserts;
+}
+
+/**
+ * Subtract statistics accumulated in @s2 from @s1.
+ */
+static inline void
+vy_stmt_stat_sub(struct vy_stmt_stat *s1, const struct vy_stmt_stat *s2)
+{
+	s1->inserts -= s2->inserts;
+	s1->replaces -= s2->replaces;
+	s1->deletes -= s2->deletes;
+	s1->upserts -= s2->upserts;
+}
+
 #if defined(__cplusplus)
 } /* extern "C" */
 #endif /* defined(__cplusplus) */
diff --git a/test/vinyl/info.result b/test/vinyl/info.result
index a47c9a1b..49c2037a 100644
--- a/test/vinyl/info.result
+++ b/test/vinyl/info.result
@@ -151,7 +151,11 @@ istat()
   disk:
     index_size: 0
     rows: 0
-    bytes: 0
+    statement:
+      inserts: 0
+      replaces: 0
+      upserts: 0
+      deletes: 0
     dump:
       in:
         rows: 0
@@ -192,9 +196,10 @@ istat()
       get:
         rows: 0
         bytes: 0
+    bloom_size: 0
     pages: 0
     bytes_compressed: 0
-    bloom_size: 0
+    bytes: 0
   txw:
     bytes: 0
     rows: 0
@@ -286,10 +291,12 @@ stat_diff(istat(), st)
         rows: 25
     index_size: 294
     rows: 25
-    bloom_size: 70
-    pages: 7
     bytes: 26049
     bytes_compressed: <bytes_compressed>
+    bloom_size: 70
+    statement:
+      replaces: 25
+    pages: 7
   bytes: 26049
   put:
     rows: 25
@@ -324,9 +331,11 @@ stat_diff(istat(), st)
         rows: 50
     index_size: 252
     rows: 25
+    bytes: 26042
     bytes_compressed: <bytes_compressed>
     pages: 6
-    bytes: 26042
+    statement:
+      replaces: 25
     compact:
       in:
         bytes: 78140
@@ -992,7 +1001,11 @@ istat()
   disk:
     index_size: 1050
     rows: 100
-    bytes: 104300
+    statement:
+      inserts: 0
+      replaces: 100
+      upserts: 0
+      deletes: 0
     dump:
       in:
         rows: 0
@@ -1033,9 +1046,10 @@ istat()
       get:
         rows: 0
         bytes: 0
+    bloom_size: 140
     pages: 25
     bytes_compressed: <bytes_compressed>
-    bloom_size: 140
+    bytes: 104300
   txw:
     bytes: 0
     rows: 0
@@ -1392,6 +1406,91 @@ gst.disk.index == 0
 ---
 - true
 ...
+--
+-- Statement statistics.
+--
+s = box.schema.space.create('test', {engine = 'vinyl'})
+---
+...
+i = s:create_index('primary', {run_count_per_level = 10})
+---
+...
+i:stat().disk.statement
+---
+- inserts: 0
+  replaces: 0
+  upserts: 0
+  deletes: 0
+...
+s:insert{1, 1}
+---
+- [1, 1]
+...
+s:replace{2, 2}
+---
+- [2, 2]
+...
+box.snapshot()
+---
+- ok
+...
+i:stat().disk.statement
+---
+- inserts: 1
+  replaces: 1
+  upserts: 0
+  deletes: 0
+...
+s:upsert({1, 1}, {{'+', 2, 1}})
+---
+...
+s:delete{2}
+---
+...
+box.snapshot()
+---
+- ok
+...
+i:stat().disk.statement
+---
+- inserts: 1
+  replaces: 1
+  upserts: 1
+  deletes: 1
+...
+test_run:cmd('restart server test')
+fiber = require('fiber')
+---
+...
+s = box.space.test
+---
+...
+i = s.index.primary
+---
+...
+i:stat().disk.statement
+---
+- inserts: 1
+  replaces: 1
+  upserts: 1
+  deletes: 1
+...
+i:compact()
+---
+...
+while i:stat().disk.compact.count == 0 do fiber.sleep(0.01) end
+---
+...
+i:stat().disk.statement
+---
+- inserts: 1
+  replaces: 0
+  upserts: 0
+  deletes: 0
+...
+s:drop()
+---
+...
 test_run:cmd('switch default')
 ---
 - true
diff --git a/test/vinyl/info.test.lua b/test/vinyl/info.test.lua
index e5794a23..b11fc044 100644
--- a/test/vinyl/info.test.lua
+++ b/test/vinyl/info.test.lua
@@ -411,6 +411,42 @@ gst.memory.bloom_filter == 0
 gst.disk.data == 0
 gst.disk.index == 0
 
+--
+-- Statement statistics.
+--
+s = box.schema.space.create('test', {engine = 'vinyl'})
+i = s:create_index('primary', {run_count_per_level = 10})
+
+i:stat().disk.statement
+
+s:insert{1, 1}
+s:replace{2, 2}
+box.snapshot()
+
+i:stat().disk.statement
+
+s:upsert({1, 1}, {{'+', 2, 1}})
+s:delete{2}
+box.snapshot()
+
+i:stat().disk.statement
+
+test_run:cmd('restart server test')
+
+fiber = require('fiber')
+
+s = box.space.test
+i = s.index.primary
+
+i:stat().disk.statement
+
+i:compact()
+while i:stat().disk.compact.count == 0 do fiber.sleep(0.01) end
+
+i:stat().disk.statement
+
+s:drop()
+
 test_run:cmd('switch default')
 test_run:cmd('stop server test')
 test_run:cmd('cleanup server test')
diff --git a/test/vinyl/layout.result b/test/vinyl/layout.result
index ee14cd51..989ddf64 100644
--- a/test/vinyl/layout.result
+++ b/test/vinyl/layout.result
@@ -236,9 +236,10 @@ result
           type: RUNINFO
         BODY:
           min_lsn: 8
+          bloom_filter: <bloom_filter>
           max_key: ['ЭЭЭ']
           page_count: 1
-          bloom_filter: <bloom_filter>
+          stmt_stat: {9: 0, 2: 0, 5: 0, 3: 3}
           max_lsn: 10
           min_key: ['ёёё']
       - HEADER:
@@ -275,9 +276,10 @@ result
           type: RUNINFO
         BODY:
           min_lsn: 11
+          bloom_filter: <bloom_filter>
           max_key: ['ЮЮЮ']
           page_count: 1
-          bloom_filter: <bloom_filter>
+          stmt_stat: {9: 0, 2: 0, 5: 0, 3: 3}
           max_lsn: 13
           min_key: ['ёёё']
       - HEADER:
@@ -317,9 +319,10 @@ result
           type: RUNINFO
         BODY:
           min_lsn: 8
+          bloom_filter: <bloom_filter>
           max_key: [null, 'ЭЭЭ']
           page_count: 1
-          bloom_filter: <bloom_filter>
+          stmt_stat: {9: 0, 2: 0, 5: 0, 3: 3}
           max_lsn: 10
           min_key: [null, 'ёёё']
       - HEADER:
@@ -356,9 +359,10 @@ result
           type: RUNINFO
         BODY:
           min_lsn: 11
+          bloom_filter: <bloom_filter>
           max_key: [789, 'ююю']
           page_count: 1
-          bloom_filter: <bloom_filter>
+          stmt_stat: {9: 0, 2: 0, 5: 0, 3: 3}
           max_lsn: 13
           min_key: [123, 'ёёё']
       - HEADER:
-- 
2.11.0