vinyl: rework replication

Currently, on initial join we send the current vinyl state. To do that,
we open a read iterator over a space's primary index and send statements
returned by it. Such an approach has a number of inherent problems:

 - An open read iterator blocks compaction, which is unacceptable for
   such a long operation as join. To avoid blocking compaction, we open
   the iterator in the dirty mode, i.e. it skims over the tops. This,
   however, introduces a different kind of problem: this makes the
   threshold between initial and final join phases hazy - statements
   sent on final join may or may not be among those sent during the
   initial join, and there's no efficient way to differentiate between
   them w/o sending extra information.

 - The replica expects LSNs to be growing monotonically. This constraint
   is imposed by the lsregion allocator used for storing statements in
   memory, but read iterator returns statements ordered by key, not by
   LSN. Currently, replica simply crashes if statements happen to be
   sent in an order different from chronological, which renders vinyl
   replication unusable. In the scope of the current model, we can't fix
   this by assigning fake LSNs to statements received on initial join,
   because there's no strict LSN threshold between initial and final
   join phases (see the previous paragraph).

 - In the initial join phase, replica is only aware of  spaces that were
   created before the last snapshot, while vinyl sends statements from
   spaces that exist now. As a result, if a space was created after the
   most recent snapshot, the replica won't be able to receive its tuples
   and fail.

To address the above-mentioned problems, we make vinyl initial join send
the latest snapshot, just like in case of memtx. We implement this by
loading the vinyl state from the last snapshot of the metadata log and
sending statements of all runs from the snapshot as is (including
deletes and updates), to be applied by the replica. To make lsregion at
the receiving end happy, we assign fake monotonically growing LSNs to
statements received on initial join. This is OK, because

  any LSN from final join > max real LSN from initial join
  max real LSN from initial join >= max fake LSN

hence

  any LSN from final join > any fake LSN from initial join

Besides fixing vinyl replication, this patch also enables the
replication test suite for the vinyl engine (except for hot_standby)
and makes engine/replica_join cover the following test cases:
 - secondary indexes
 - delete and update statements
 - keys added in an order different from LSN
 - recreate space after checkpoint

Closes #1911
Closes #2001

Author: locker

src/box/box.cc

@@ -284,11 +284,6 @@ wal_stream_create(struct wal_stream *ctx, size_t wal_max_rows)
 static void
 apply_initial_join_row(struct xstream *stream, struct xrow_header *row)
 {
-	if (row->type != IPROTO_INSERT) {
-		tnt_raise(ClientError, ER_UNKNOWN_REQUEST_TYPE,
-				(uint32_t) row->type);
-	}
-
 	(void) stream;
 	struct request *request;
 	request = region_alloc_object_xc(&fiber()->gc, struct request);

src/box/memtx_space.cc

@@ -249,7 +249,10 @@ dup_replace_mode(uint32_t op)
 void
 MemtxSpace::applyInitialJoinRow(struct space *space, struct request *request)
 {
-	assert(request->type == IPROTO_INSERT);
+	if (request->type != IPROTO_INSERT) {
+		tnt_raise(ClientError, ER_UNKNOWN_REQUEST_TYPE,
+				(uint32_t) request->type);
+	}
 	request->header->replica_id = 0;
 	struct txn *txn = txn_begin_stmt(space);
 	try {

src/box/vinyl.c

@@ -7493,6 +7493,18 @@ vy_page_delete(struct vy_page *page)
 	free(page);
 }
 
+static int
+vy_page_xrow(struct vy_page *page, uint32_t stmt_no,
+	     struct xrow_header *xrow)
+{
+	assert(stmt_no < page->count);
+	const char *data = page->data + page->row_index[stmt_no];
+	const char *data_end = stmt_no + 1 < page->count ?
+		page->data + page->row_index[stmt_no + 1] :
+		page->data + page->unpacked_size;
+	return xrow_header_decode(xrow, &data, data_end);
+}
+
 /**
  * Read raw stmt data from the page
  * @param page          Page.
@@ -7509,13 +7521,8 @@ vy_page_stmt(struct vy_page *page, uint32_t stmt_no,
 	     struct tuple_format *format, struct tuple_format *upsert_format,
 	     struct key_def *key_def)
 {
-	assert(stmt_no < page->count);
-	const char *data = page->data + page->row_index[stmt_no];
-	const char *data_end = stmt_no + 1 < page->count ?
-		page->data + page->row_index[stmt_no + 1] :
-		page->data + page->unpacked_size;
 	struct xrow_header xrow;
-	if (xrow_header_decode(&xrow, &data, data_end) != 0)
+	if (vy_page_xrow(page, stmt_no, &xrow) != 0)
 		return NULL;
 	struct tuple_format *format_to_use = (xrow.type == IPROTO_UPSERT)
 		? upsert_format : format;
@@ -10019,31 +10026,119 @@ vy_read_iterator_close(struct vy_read_iterator *itr)
 
 /** {{{ Replication */
 
-int
-vy_index_send(struct vy_index *index, vy_send_row_f sendrow, void *ctx)
+/** Argument passed to vy_join_cb(). */
+struct vy_join_arg {
+	/** Vinyl environment. */
+	struct vy_env *env;
+	/** Function for sending a statement. */
+	vy_send_row_f send_row;
+	/** Argument passed to send_row(). */
+	void *send_row_arg;
+	/** ID of the space currently being relayed. */
+	uint32_t space_id;
+	/** Ordinal number of the index. */
+	uint32_t index_id;
+	/** Path to the index directory. */
+	char *index_path;
+	/**
+	 * LSN to assign to the next statement.
+	 *
+	 * We can't use original statements' LSNs, because we
+	 * send statements not in the chronological order while
+	 * the receiving end expects LSNs to grow monotonically
+	 * due to the design of the lsregion allocator, which is
+	 * used for storing statements in memory.
+	 */
+	int64_t lsn;
+};
+
+/** Relay callback, passed to xctl_relay(). */
+static int
+vy_join_cb(const struct xctl_record *record, void *cb_arg)
 {
-	struct vy_env *env = index->env;
-	static const int64_t vlsn = INT64_MAX;
+	struct vy_join_arg *arg = cb_arg;
 	int rc = 0;
 
-	struct vy_read_iterator ri;
-	struct tuple *stmt;
-	struct tuple *key = vy_stmt_new_select(env->key_format, NULL, 0);
-	if (key == NULL)
+	if (record->type == XCTL_CREATE_VY_INDEX) {
+		arg->space_id = record->space_id;
+		arg->index_id = record->iid;
+		vy_index_snprint_path(arg->index_path, PATH_MAX,
+				      arg->env->conf->path,
+				      arg->space_id, arg->index_id);
+	}
+
+	/*
+	 * We are only interested in the primary index.
+	 * Secondary keys will be rebuilt on the destination.
+	 */
+	if (arg->index_id != 0)
+		goto out;
+
+	/*
+	 * We only send statements, not metadata, because the
+	 * latter is a replica's private business.
+	 */
+	if (record->type != XCTL_INSERT_VY_RUN)
+		goto out;
+
+	rc = -1;
+
+	/* Load the run. */
+	struct vy_run *run = vy_run_new(record->vy_run_id);
+	if (run == NULL)
+		goto out;
+	if (vy_run_recover(run, arg->index_path) != 0)
+		goto out_free_run;
+
+	ZSTD_DStream *zdctx = vy_env_get_zdctx(arg->env);
+	if (zdctx == NULL)
+		goto out_free_run;
+
+	/* Send the run's statements to the replica. */
+	for (uint32_t page_no = 0; page_no < run->info.count; page_no++) {
+		struct vy_page_info *pi = vy_run_page_info(run, page_no);
+		struct vy_page *page = vy_page_new(pi);
+		if (page == NULL)
+			goto out_free_run;
+		if (vy_page_read(page, pi, run->fd, zdctx) != 0)
+			goto out_free_page;
+		for (uint32_t stmt_no = 0; stmt_no < pi->count; stmt_no++) {
+			struct xrow_header xrow;
+			if (vy_page_xrow(page, stmt_no, &xrow) != 0)
+				goto out_free_page;
+			xrow.lsn = arg->lsn++;
+			if (arg->send_row(&xrow, arg->send_row_arg) != 0)
+				goto out_free_page;
+		}
+		vy_page_delete(page);
+		continue;
+out_free_page:
+		vy_page_delete(page);
+		goto out_free_run;
+	}
+	rc = 0; /* success */
+
+out_free_run:
+	vy_run_delete(run);
+out:
+	return rc;
+}
+
+int
+vy_join(struct vy_env *env, vy_send_row_f send_row, void *send_row_arg)
+{
+	struct vy_join_arg arg = {
+		.env = env,
+		.send_row = send_row,
+		.send_row_arg = send_row_arg,
+	};
+	arg.index_path = malloc(PATH_MAX);
+	if (arg.index_path == NULL) {
+		diag_set(OutOfMemory, PATH_MAX, "malloc", "path");
 		return -1;
-	vy_read_iterator_open(&ri, index, NULL, ITER_GT, key, &vlsn, true);
-	rc = vy_read_iterator_next(&ri, &stmt);
-	for (; rc == 0 && stmt; rc = vy_read_iterator_next(&ri, &stmt)) {
-		uint32_t mp_size;
-		const char *mp_data;
-		mp_data = tuple_data_range(stmt, &mp_size);
-		int64_t lsn = vy_stmt_lsn(stmt);
-		rc = sendrow(ctx, mp_data, mp_size, lsn);
-		if (rc != 0)
-			break;
 	}
-	vy_read_iterator_close(&ri);
-	tuple_unref(key);
+	int rc = xctl_relay(vy_join_cb, &arg);
+	free(arg.index_path);
 	return rc;
 }
 

src/box/vinyl.h

@@ -50,6 +50,7 @@ struct vclock;
 struct request;
 struct space;
 struct txn_stmt;
+struct xrow_header;
 enum iterator_type;
 
 /*
@@ -297,9 +298,10 @@ vy_cursor_next(struct vy_cursor *cursor, struct tuple **result);
  */
 
 typedef int
-(*vy_send_row_f)(void *, const char *tuple, uint32_t tuple_size, int64_t lsn);
+(*vy_send_row_f)(struct xrow_header *row, void *arg);
+
 int
-vy_index_send(struct vy_index *index, vy_send_row_f sendrow, void *ctx);
+vy_join(struct vy_env *env, vy_send_row_f send_row, void *send_row_arg);
 
 #ifdef __cplusplus
 }

src/box/vinyl_engine.cc

@@ -156,74 +156,23 @@ VinylEngine::buildSecondaryKey(struct space *old_space,
 	 */
 }
 
-struct vinyl_send_row_arg {
-	struct xstream *stream;
-	uint32_t space_id;
-};
-
 static int
-vinyl_send_row(void *arg, const char *tuple, uint32_t tuple_size, int64_t lsn)
+vinyl_send_row(struct xrow_header *row, void *arg)
 {
-	struct xstream *stream = ((struct vinyl_send_row_arg *) arg)->stream;
-	uint32_t space_id = ((struct vinyl_send_row_arg *) arg)->space_id;
-
-	struct request_replace_body body;
-	memset(&body, 0, sizeof(body));
-	body.m_body = 0x82; /* map of two elements. */
-	body.k_space_id = IPROTO_SPACE_ID;
-	body.m_space_id = 0xce; /* uint32 */
-	body.v_space_id = mp_bswap_u32(space_id);
-	body.k_tuple = IPROTO_TUPLE;
-	struct xrow_header row;
-	memset(&row, 0, sizeof(row));
-	row.type = IPROTO_INSERT;
-	row.replica_id = 0;
-	row.lsn = lsn;
-	row.bodycnt = 2;
-	row.body[0].iov_base = &body;
-	row.body[0].iov_len = sizeof(body);
-	row.body[1].iov_base = (char *) tuple;
-	row.body[1].iov_len = tuple_size;
+	struct xstream *stream = (struct xstream *) arg;
 	try {
-		xstream_write(stream, &row);
+		xstream_write(stream, row);
 	} catch (Exception *e) {
 		 return -1;
 	}
 	return 0;
 }
 
-struct join_send_space_arg {
-	struct vy_env *env;
-	struct xstream *stream;
-};
-
-static void
-join_send_space(struct space *sp, void *data)
-{
-	struct xstream *stream = ((struct join_send_space_arg *) data)->stream;
-	if (space_is_temporary(sp))
-		return;
-	if (!space_is_vinyl(sp))
-		return;
-	VinylIndex *pk = (VinylIndex *) space_index(sp, 0);
-	if (!pk)
-		return;
-
-	/* send database */
-	struct vinyl_send_row_arg arg = { stream, sp->def.id };
-	if (vy_index_send(pk->db, vinyl_send_row, &arg) != 0)
-		diag_raise();
-}
-
-/**
- * Relay all data currently stored in Vinyl engine
- * to the replica.
- */
 void
 VinylEngine::join(struct xstream *stream)
 {
-	struct join_send_space_arg arg = { env, stream };
-	space_foreach(join_send_space, &arg);
+	if (vy_join(env, vinyl_send_row, stream) != 0)
+		diag_raise();
 }
 
 void

src/box/vinyl_space.cc

@@ -46,23 +46,41 @@ VinylSpace::VinylSpace(Engine *e)
 void
 VinylSpace::applyInitialJoinRow(struct space *space, struct request *request)
 {
-	assert(request->type == IPROTO_INSERT);
 	assert(request->header != NULL);
 	struct vy_env *env = ((VinylEngine *)space->handler->engine)->env;
 
-	/* Check the tuple fields. */
-	if (tuple_validate_raw(space->format, request->tuple))
-		diag_raise();
-
 	struct vy_tx *tx = vy_begin(env);
 	if (tx == NULL)
 		diag_raise();
 
 	int64_t signature = request->header->lsn;
 
-	if (vy_replace(tx, NULL, space, request))
+	struct txn_stmt stmt;
+	memset(&stmt, 0, sizeof(stmt));
+
+	int rc;
+	switch (request->type) {
+	case IPROTO_REPLACE:
+		rc = vy_replace(tx, &stmt, space, request);
+		break;
+	case IPROTO_UPSERT:
+		rc = vy_upsert(tx, &stmt, space, request);
+		break;
+	case IPROTO_DELETE:
+		rc = vy_delete(tx, &stmt, space, request);
+		break;
+	default:
+		tnt_raise(ClientError, ER_UNKNOWN_REQUEST_TYPE,
+			  (uint32_t) request->type);
+	}
+	if (rc != 0)
 		diag_raise();
 
+	if (stmt.old_tuple)
+		tuple_unref(stmt.old_tuple);
+	if (stmt.new_tuple)
+		tuple_unref(stmt.new_tuple);
+
 	if (vy_prepare(env, tx)) {
 		vy_rollback(env, tx);
 		diag_raise();