[PATCH 3/3] vinyl: implement support of replica local spaces

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

Store replication group id in vylog and ignore spaces whose group_id
equals GROUP_LOCAL when relaying initial join rows.

Folow-up #3443
---
 src/box/vinyl.c                        | 13 ++++++------
 src/box/vy_log.c                       | 38 +++++++++++++++++++++++++++-------
 src/box/vy_log.h                       | 14 +++++++++----
 src/box/vy_lsm.c                       |  8 ++++---
 src/box/vy_lsm.h                       |  6 ++++--
 test/replication/local_spaces.result   |  9 +++++---
 test/replication/local_spaces.test.lua |  7 ++++---
 test/replication/suite.cfg             |  1 -
 test/unit/vy_point_lookup.c            |  2 +-
 test/vinyl/ddl.result                  |  5 -----
 test/vinyl/ddl.test.lua                |  3 ---
 11 files changed, 68 insertions(+), 38 deletions(-)

diff --git a/src/box/vinyl.c b/src/box/vinyl.c
index b3c97196..1c5192ff 100644
--- a/src/box/vinyl.c
+++ b/src/box/vinyl.c
@@ -71,6 +71,7 @@
 #include "trigger.h"
 #include "checkpoint.h"
 #include "session.h"
+#include "replication.h" /* GROUP_LOCAL */
 #include "wal.h" /* wal_mode() */
 
 /**
@@ -595,11 +596,6 @@ vinyl_engine_check_space_def(struct space_def *def)
 			 def->name, "engine does not support temporary flag");
 		return -1;
 	}
-	if (def->opts.group_id != 0) {
-		diag_set(ClientError, ER_ALTER_SPACE, def->name,
-			 "engine does not support replication groups");
-		return -1;
-	}
 	return 0;
 }
 
@@ -724,7 +720,8 @@ vinyl_space_create_index(struct space *space, struct index_def *index_def)
 	}
 	struct vy_lsm *lsm = vy_lsm_new(&env->lsm_env, &env->cache_env,
 					&env->mem_env, index_def,
-					space->format, pk);
+					space->format, pk,
+					space_group_id(space));
 	if (lsm == NULL) {
 		free(index);
 		return NULL;
@@ -3135,6 +3132,10 @@ vy_send_lsm(struct vy_join_ctx *ctx, struct vy_lsm_recovery_info *lsm_info)
 		/* Dropped or not yet built LSM tree. */
 		return 0;
 	}
+	if (lsm_info->group_id == GROUP_LOCAL) {
+		/* Replica local space. */
+		return 0;
+	}
 
 	/*
 	 * We are only interested in the primary index LSM tree.
diff --git a/src/box/vy_log.c b/src/box/vy_log.c
index 760a253d..fbbc7a7f 100644
--- a/src/box/vy_log.c
+++ b/src/box/vy_log.c
@@ -83,6 +83,7 @@ enum vy_log_key {
 	VY_LOG_KEY_CREATE_LSN		= 12,
 	VY_LOG_KEY_MODIFY_LSN		= 13,
 	VY_LOG_KEY_DROP_LSN		= 14,
+	VY_LOG_KEY_GROUP_ID		= 15,
 };
 
 /** vy_log_key -> human readable name. */
@@ -102,6 +103,7 @@ static const char *vy_log_key_name[] = {
 	[VY_LOG_KEY_CREATE_LSN]		= "create_lsn",
 	[VY_LOG_KEY_MODIFY_LSN]		= "modify_lsn",
 	[VY_LOG_KEY_DROP_LSN]		= "drop_lsn",
+	[VY_LOG_KEY_GROUP_ID]		= "group_id",
 };
 
 /** vy_log_type -> human readable name. */
@@ -247,6 +249,9 @@ vy_log_record_snprint(char *buf, int size, const struct vy_log_record *record)
 	if (record->space_id > 0)
 		SNPRINT(total, snprintf, buf, size, "%s=%"PRIu32", ",
 			vy_log_key_name[VY_LOG_KEY_SPACE_ID], record->space_id);
+	if (record->group_id > 0)
+		SNPRINT(total, snprintf, buf, size, "%s=%"PRIu32", ",
+			vy_log_key_name[VY_LOG_KEY_GROUP_ID], record->group_id);
 	if (record->key_parts != NULL) {
 		SNPRINT(total, snprintf, buf, size, "%s=",
 			vy_log_key_name[VY_LOG_KEY_DEF]);
@@ -362,6 +367,11 @@ vy_log_record_encode(const struct vy_log_record *record,
 		size += mp_sizeof_uint(record->space_id);
 		n_keys++;
 	}
+	if (record->group_id > 0) {
+		size += mp_sizeof_uint(VY_LOG_KEY_GROUP_ID);
+		size += mp_sizeof_uint(record->group_id);
+		n_keys++;
+	}
 	if (record->key_parts != NULL) {
 		size += mp_sizeof_uint(VY_LOG_KEY_DEF);
 		size += mp_sizeof_array(record->key_part_count);
@@ -447,6 +457,10 @@ vy_log_record_encode(const struct vy_log_record *record,
 		pos = mp_encode_uint(pos, VY_LOG_KEY_SPACE_ID);
 		pos = mp_encode_uint(pos, record->space_id);
 	}
+	if (record->group_id > 0) {
+		pos = mp_encode_uint(pos, VY_LOG_KEY_GROUP_ID);
+		pos = mp_encode_uint(pos, record->group_id);
+	}
 	if (record->key_parts != NULL) {
 		pos = mp_encode_uint(pos, VY_LOG_KEY_DEF);
 		pos = mp_encode_array(pos, record->key_part_count);
@@ -561,6 +575,9 @@ vy_log_record_decode(struct vy_log_record *record,
 		case VY_LOG_KEY_SPACE_ID:
 			record->space_id = mp_decode_uint(&pos);
 			break;
+		case VY_LOG_KEY_GROUP_ID:
+			record->group_id = mp_decode_uint(&pos);
+			break;
 		case VY_LOG_KEY_DEF: {
 			uint32_t part_count = mp_decode_array(&pos);
 			struct key_part_def *parts = region_alloc(&fiber()->gc,
@@ -1238,6 +1255,7 @@ vy_recovery_lookup_slice(struct vy_recovery *recovery, int64_t slice_id)
 static struct vy_lsm_recovery_info *
 vy_recovery_do_create_lsm(struct vy_recovery *recovery, int64_t id,
 			  uint32_t space_id, uint32_t index_id,
+			  uint32_t group_id,
 			  const struct key_part_def *key_parts,
 			  uint32_t key_part_count)
 {
@@ -1273,6 +1291,7 @@ vy_recovery_do_create_lsm(struct vy_recovery *recovery, int64_t id,
 	lsm->id = id;
 	lsm->space_id = space_id;
 	lsm->index_id = index_id;
+	lsm->group_id = group_id;
 	memcpy(lsm->key_parts, key_parts, sizeof(*key_parts) * key_part_count);
 	lsm->key_part_count = key_part_count;
 	lsm->create_lsn = -1;
@@ -1313,6 +1332,7 @@ vy_recovery_do_create_lsm(struct vy_recovery *recovery, int64_t id,
 static int
 vy_recovery_prepare_lsm(struct vy_recovery *recovery, int64_t id,
 			uint32_t space_id, uint32_t index_id,
+			uint32_t group_id,
 			const struct key_part_def *key_parts,
 			uint32_t key_part_count)
 {
@@ -1323,7 +1343,8 @@ vy_recovery_prepare_lsm(struct vy_recovery *recovery, int64_t id,
 		return -1;
 	}
 	if (vy_recovery_do_create_lsm(recovery, id, space_id, index_id,
-				      key_parts, key_part_count) == NULL)
+				      group_id, key_parts,
+				      key_part_count) == NULL)
 		return -1;
 	return 0;
 }
@@ -1339,7 +1360,7 @@ vy_recovery_prepare_lsm(struct vy_recovery *recovery, int64_t id,
  */
 static int
 vy_recovery_create_lsm(struct vy_recovery *recovery, int64_t id,
-		       uint32_t space_id, uint32_t index_id,
+		       uint32_t space_id, uint32_t index_id, uint32_t group_id,
 		       const struct key_part_def *key_parts,
 		       uint32_t key_part_count, int64_t create_lsn,
 		       int64_t modify_lsn, int64_t dump_lsn)
@@ -1358,7 +1379,8 @@ vy_recovery_create_lsm(struct vy_recovery *recovery, int64_t id,
 			return -1;
 		}
 	} else {
-		lsm = vy_recovery_do_create_lsm(recovery, id, space_id, index_id,
+		lsm = vy_recovery_do_create_lsm(recovery, id, space_id,
+						index_id, group_id,
 						key_parts, key_part_count);
 		if (lsm == NULL)
 			return -1;
@@ -1883,14 +1905,15 @@ vy_recovery_process_record(struct vy_recovery *recovery,
 	case VY_LOG_PREPARE_LSM:
 		rc = vy_recovery_prepare_lsm(recovery, record->lsm_id,
 				record->space_id, record->index_id,
-				record->key_parts, record->key_part_count);
+				record->group_id, record->key_parts,
+				record->key_part_count);
 		break;
 	case VY_LOG_CREATE_LSM:
 		rc = vy_recovery_create_lsm(recovery, record->lsm_id,
 				record->space_id, record->index_id,
-				record->key_parts, record->key_part_count,
-				record->create_lsn, record->modify_lsn,
-				record->dump_lsn);
+				record->group_id, record->key_parts,
+				record->key_part_count, record->create_lsn,
+				record->modify_lsn, record->dump_lsn);
 		break;
 	case VY_LOG_MODIFY_LSM:
 		rc = vy_recovery_modify_lsm(recovery, record->lsm_id,
@@ -2218,6 +2241,7 @@ vy_log_append_lsm(struct xlog *xlog, struct vy_lsm_recovery_info *lsm)
 	record.lsm_id = lsm->id;
 	record.index_id = lsm->index_id;
 	record.space_id = lsm->space_id;
+	record.group_id = lsm->group_id;
 	record.key_parts = lsm->key_parts;
 	record.key_part_count = lsm->key_part_count;
 	record.create_lsn = lsm->create_lsn;
diff --git a/src/box/vy_log.h b/src/box/vy_log.h
index 0a216de8..da0745b2 100644
--- a/src/box/vy_log.h
+++ b/src/box/vy_log.h
@@ -66,8 +66,8 @@ enum vy_log_record_type {
 	/**
 	 * Create a new LSM tree.
 	 * Requires vy_log_record::lsm_id, create_lsn.
-	 * After rotation, it also stores space_id, index_id, key_def,
-	 * create_lsn, modify_lsn, dump_lsn.
+	 * After rotation, it also stores space_id, index_id, group_id,
+	 * key_def, create_lsn, modify_lsn, dump_lsn.
 	 */
 	VY_LOG_CREATE_LSM		= 0,
 	/**
@@ -182,7 +182,8 @@ enum vy_log_record_type {
 	VY_LOG_FORGET_LSM		= 14,
 	/**
 	 * Prepare a new LSM tree for building.
-	 * Requires vy_log_record::lsm_id, index_id, space_id.
+	 * Requires vy_log_record::lsm_id, index_id, space_id, group_id,
+	 * key_def.
 	 *
 	 * Index ALTER operation consists of two stages. First, we
 	 * build a new LSM tree, checking constraints if necessary.
@@ -225,6 +226,8 @@ struct vy_log_record {
 	uint32_t index_id;
 	/** Space ID. */
 	uint32_t space_id;
+	/** Replication group ID. */
+	uint32_t group_id;
 	/** Index key definition, as defined by the user. */
 	const struct key_def *key_def;
 	/** Array of key part definitions. */
@@ -285,6 +288,8 @@ struct vy_lsm_recovery_info {
 	uint32_t index_id;
 	/** Space ID. */
 	uint32_t space_id;
+	/** Replication group ID. */
+	uint32_t group_id;
 	/** Array of key part definitions. */
 	struct key_part_def *key_parts;
 	/** Number of key parts. */
@@ -562,7 +567,7 @@ vy_log_record_init(struct vy_log_record *record)
 /** Helper to log a vinyl LSM tree preparation. */
 static inline void
 vy_log_prepare_lsm(int64_t id, uint32_t space_id, uint32_t index_id,
-		   const struct key_def *key_def)
+		   uint32_t group_id, const struct key_def *key_def)
 {
 	struct vy_log_record record;
 	vy_log_record_init(&record);
@@ -570,6 +575,7 @@ vy_log_prepare_lsm(int64_t id, uint32_t space_id, uint32_t index_id,
 	record.lsm_id = id;
 	record.space_id = space_id;
 	record.index_id = index_id;
+	record.group_id = group_id;
 	record.key_def = key_def;
 	vy_log_write(&record);
 }
diff --git a/src/box/vy_lsm.c b/src/box/vy_lsm.c
index f2400657..cb3c436f 100644
--- a/src/box/vy_lsm.c
+++ b/src/box/vy_lsm.c
@@ -110,8 +110,8 @@ vy_lsm_mem_tree_size(struct vy_lsm *lsm)
 
 struct vy_lsm *
 vy_lsm_new(struct vy_lsm_env *lsm_env, struct vy_cache_env *cache_env,
-	     struct vy_mem_env *mem_env, struct index_def *index_def,
-	     struct tuple_format *format, struct vy_lsm *pk)
+	   struct vy_mem_env *mem_env, struct index_def *index_def,
+	   struct tuple_format *format, struct vy_lsm *pk, uint32_t group_id)
 {
 	static int64_t run_buckets[] = {
 		0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 15, 20, 25, 50, 100,
@@ -202,6 +202,7 @@ vy_lsm_new(struct vy_lsm_env *lsm_env, struct vy_cache_env *cache_env,
 	lsm->in_compact.pos = UINT32_MAX;
 	lsm->space_id = index_def->space_id;
 	lsm->index_id = index_def->iid;
+	lsm->group_id = group_id;
 	lsm->opts = index_def->opts;
 	lsm->check_is_unique = lsm->opts.is_unique;
 	vy_lsm_read_set_new(&lsm->read_set);
@@ -331,7 +332,8 @@ vy_lsm_create(struct vy_lsm *lsm)
 
 	/* Write the new LSM tree record to vylog. */
 	vy_log_tx_begin();
-	vy_log_prepare_lsm(id, lsm->space_id, lsm->index_id, lsm->key_def);
+	vy_log_prepare_lsm(id, lsm->space_id, lsm->index_id,
+			   lsm->group_id, lsm->key_def);
 	vy_log_insert_range(id, range->id, NULL, NULL);
 	if (vy_log_tx_commit() < 0)
 		return -1;
diff --git a/src/box/vy_lsm.h b/src/box/vy_lsm.h
index 90ccb534..f0b7ec9c 100644
--- a/src/box/vy_lsm.h
+++ b/src/box/vy_lsm.h
@@ -158,6 +158,8 @@ struct vy_lsm {
 	uint32_t index_id;
 	/** ID of the space this LSM tree is for. */
 	uint32_t space_id;
+	/** Replication group ID. */
+	uint32_t group_id;
 	/** Index options. */
 	struct index_opts opts;
 	/** Key definition used to compare tuples. */
@@ -301,8 +303,8 @@ vy_lsm_mem_tree_size(struct vy_lsm *lsm);
 /** Allocate a new LSM tree object. */
 struct vy_lsm *
 vy_lsm_new(struct vy_lsm_env *lsm_env, struct vy_cache_env *cache_env,
-	     struct vy_mem_env *mem_env, struct index_def *index_def,
-	     struct tuple_format *format, struct vy_lsm *pk);
+	   struct vy_mem_env *mem_env, struct index_def *index_def,
+	   struct tuple_format *format, struct vy_lsm *pk, uint32_t group_id);
 
 /** Free an LSM tree object. */
 void
diff --git a/test/replication/local_spaces.result b/test/replication/local_spaces.result
index 06782788..15173553 100644
--- a/test/replication/local_spaces.result
+++ b/test/replication/local_spaces.result
@@ -4,17 +4,20 @@ env = require('test_run')
 test_run = env.new()
 ---
 ...
+engine = test_run:get_cfg('engine')
+---
+...
 --
 -- gh-3443: Check that changes done to spaces marked as local
 -- are not replicated, but vclock is still promoted.
 --
-s1 = box.schema.space.create('test1')
+s1 = box.schema.space.create('test1', {engine = engine})
 ---
 ...
 _ = s1:create_index('pk')
 ---
 ...
-s2 = box.schema.space.create('test2', {is_local = true})
+s2 = box.schema.space.create('test2', {engine = engine, is_local = true})
 ---
 ...
 _ = s2:create_index('pk')
@@ -48,7 +51,7 @@ box.space._space:update(s2.id, {{'=', 6, {group_id = 0}}}) -- error
 ...
 -- Currently, there are only two replication groups:
 -- 0 (global) and 1 (local)
-box.space._space:insert{9000, 1, 'test', 'memtx', 0, {group_id = 2}, {}} -- error
+box.space._space:insert{9000, 1, 'test', engine, 0, {group_id = 2}, {}} -- error
 ---
 - error: Replication group '2' does not exist
 ...
diff --git a/test/replication/local_spaces.test.lua b/test/replication/local_spaces.test.lua
index ec676a18..06e2b0bd 100644
--- a/test/replication/local_spaces.test.lua
+++ b/test/replication/local_spaces.test.lua
@@ -1,14 +1,15 @@
 env = require('test_run')
 test_run = env.new()
+engine = test_run:get_cfg('engine')
 
 --
 -- gh-3443: Check that changes done to spaces marked as local
 -- are not replicated, but vclock is still promoted.
 --
 
-s1 = box.schema.space.create('test1')
+s1 = box.schema.space.create('test1', {engine = engine})
 _ = s1:create_index('pk')
-s2 = box.schema.space.create('test2', {is_local = true})
+s2 = box.schema.space.create('test2', {engine = engine, is_local = true})
 _ = s2:create_index('pk')
 s1.is_local
 s2.is_local
@@ -23,7 +24,7 @@ box.space._space:update(s2.id, {{'=', 6, {group_id = 0}}}) -- error
 
 -- Currently, there are only two replication groups:
 -- 0 (global) and 1 (local)
-box.space._space:insert{9000, 1, 'test', 'memtx', 0, {group_id = 2}, {}} -- error
+box.space._space:insert{9000, 1, 'test', engine, 0, {group_id = 2}, {}} -- error
 
 -- Temporary local spaces should behave in the same fashion as
 -- plain temporary spaces, i.e. neither replicated nor persisted.
diff --git a/test/replication/suite.cfg b/test/replication/suite.cfg
index 283edcad..95e94e5a 100644
--- a/test/replication/suite.cfg
+++ b/test/replication/suite.cfg
@@ -6,7 +6,6 @@
     "wal_off.test.lua": {},
     "hot_standby.test.lua": {},
     "rebootstrap.test.lua": {},
-    "local_spaces.test.lua": {},
     "*": {
         "memtx": {"engine": "memtx"},
         "vinyl": {"engine": "vinyl"}
diff --git a/test/unit/vy_point_lookup.c b/test/unit/vy_point_lookup.c
index ebbb87d5..b9b7d6ff 100644
--- a/test/unit/vy_point_lookup.c
+++ b/test/unit/vy_point_lookup.c
@@ -93,7 +93,7 @@ test_basic()
 			      &index_opts, key_def, NULL);
 
 	struct vy_lsm *pk = vy_lsm_new(&lsm_env, &cache_env, &mem_env,
-				       index_def, format, NULL);
+				       index_def, format, NULL, 0);
 	isnt(pk, NULL, "lsm is not NULL")
 
 	struct vy_range *range = vy_range_new(1, NULL, NULL, pk->cmp_def);
diff --git a/test/vinyl/ddl.result b/test/vinyl/ddl.result
index 5b49f51f..3e65e232 100644
--- a/test/vinyl/ddl.result
+++ b/test/vinyl/ddl.result
@@ -44,11 +44,6 @@ space:create_index('pk', {bloom_fpr = 1.1})
 space:drop()
 ---
 ...
--- vinyl does not support replica local spaces
-space = box.schema.space.create('test', {engine = 'vinyl', is_local = true})
----
-- error: 'Can''t modify space ''test'': engine does not support replication groups'
-...
 -- space secondary index create
 space = box.schema.space.create('test', { engine = 'vinyl' })
 ---
diff --git a/test/vinyl/ddl.test.lua b/test/vinyl/ddl.test.lua
index 44cfa2ac..45c5cf8e 100644
--- a/test/vinyl/ddl.test.lua
+++ b/test/vinyl/ddl.test.lua
@@ -12,9 +12,6 @@ space:create_index('pk', {bloom_fpr = 0})
 space:create_index('pk', {bloom_fpr = 1.1})
 space:drop()
 
--- vinyl does not support replica local spaces
-space = box.schema.space.create('test', {engine = 'vinyl', is_local = true})
-
 -- space secondary index create
 space = box.schema.space.create('test', { engine = 'vinyl' })
 index1 = space:create_index('primary')
-- 
2.11.0