[PATCH 3/4] vinyl: store lsn of index drop record in vylog

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

This is required to rework garbage collection in vinyl.
---
 src/box/vinyl.c  |  7 +++----
 src/box/vy_log.c | 43 ++++++++++++++++++++++++++++++++-----------
 src/box/vy_log.h | 17 +++++++++++++----
 src/box/vy_lsm.c |  2 +-
 4 files changed, 49 insertions(+), 20 deletions(-)

diff --git a/src/box/vinyl.c b/src/box/vinyl.c
index a423e95b..63f1c5ff 100644
--- a/src/box/vinyl.c
+++ b/src/box/vinyl.c
@@ -931,7 +931,6 @@ vy_log_lsm_prune(struct vy_lsm *lsm, int64_t gc_lsn)
 static void
 vinyl_index_commit_drop(struct index *index, int64_t lsn)
 {
-	(void)lsn;
 	struct vy_env *env = vy_env(index->engine);
 	struct vy_lsm *lsm = vy_lsm(index);
 
@@ -952,7 +951,7 @@ vinyl_index_commit_drop(struct index *index, int64_t lsn)
 
 	vy_log_tx_begin();
 	vy_log_lsm_prune(lsm, checkpoint_last(NULL));
-	vy_log_drop_lsm(lsm->id);
+	vy_log_drop_lsm(lsm->id, lsn);
 	vy_log_tx_try_commit();
 }
 
@@ -3138,7 +3137,7 @@ vy_send_lsm(struct vy_join_ctx *ctx, struct vy_lsm_recovery_info *lsm_info)
 {
 	int rc = -1;
 
-	if (lsm_info->is_dropped)
+	if (lsm_info->drop_lsn >= 0)
 		return 0;
 
 	/*
@@ -3429,7 +3428,7 @@ vinyl_engine_backup(struct engine *engine, struct vclock *vclock,
 	int loops = 0;
 	struct vy_lsm_recovery_info *lsm_info;
 	rlist_foreach_entry(lsm_info, &recovery->lsms, in_recovery) {
-		if (lsm_info->is_dropped)
+		if (lsm_info->drop_lsn >= 0)
 			continue;
 		struct vy_run_recovery_info *run_info;
 		rlist_foreach_entry(run_info, &lsm_info->runs, in_lsm) {
diff --git a/src/box/vy_log.c b/src/box/vy_log.c
index d3c2bf6c..4e459459 100644
--- a/src/box/vy_log.c
+++ b/src/box/vy_log.c
@@ -82,6 +82,7 @@ enum vy_log_key {
 	VY_LOG_KEY_TRUNCATE_COUNT	= 11,
 	VY_LOG_KEY_CREATE_LSN		= 12,
 	VY_LOG_KEY_MODIFY_LSN		= 13,
+	VY_LOG_KEY_DROP_LSN		= 14,
 };
 
 /** vy_log_key -> human readable name. */
@@ -100,6 +101,7 @@ static const char *vy_log_key_name[] = {
 	[VY_LOG_KEY_TRUNCATE_COUNT]	= "truncate_count",
 	[VY_LOG_KEY_CREATE_LSN]		= "create_lsn",
 	[VY_LOG_KEY_MODIFY_LSN]		= "modify_lsn",
+	[VY_LOG_KEY_DROP_LSN]		= "drop_lsn",
 };
 
 /** vy_log_type -> human readable name. */
@@ -258,6 +260,10 @@ vy_log_record_snprint(char *buf, int size, const struct vy_log_record *record)
 		SNPRINT(total, snprintf, buf, size, "%s=%"PRIi64", ",
 			vy_log_key_name[VY_LOG_KEY_MODIFY_LSN],
 			record->modify_lsn);
+	if (record->drop_lsn > 0)
+		SNPRINT(total, snprintf, buf, size, "%s=%"PRIi64", ",
+			vy_log_key_name[VY_LOG_KEY_DROP_LSN],
+			record->drop_lsn);
 	if (record->dump_lsn > 0)
 		SNPRINT(total, snprintf, buf, size, "%s=%"PRIi64", ",
 			vy_log_key_name[VY_LOG_KEY_DUMP_LSN],
@@ -372,6 +378,11 @@ vy_log_record_encode(const struct vy_log_record *record,
 		size += mp_sizeof_uint(record->modify_lsn);
 		n_keys++;
 	}
+	if (record->drop_lsn > 0) {
+		size += mp_sizeof_uint(VY_LOG_KEY_DROP_LSN);
+		size += mp_sizeof_uint(record->drop_lsn);
+		n_keys++;
+	}
 	if (record->dump_lsn > 0) {
 		size += mp_sizeof_uint(VY_LOG_KEY_DUMP_LSN);
 		size += mp_sizeof_uint(record->dump_lsn);
@@ -448,6 +459,10 @@ vy_log_record_encode(const struct vy_log_record *record,
 		pos = mp_encode_uint(pos, VY_LOG_KEY_MODIFY_LSN);
 		pos = mp_encode_uint(pos, record->modify_lsn);
 	}
+	if (record->drop_lsn > 0) {
+		pos = mp_encode_uint(pos, VY_LOG_KEY_DROP_LSN);
+		pos = mp_encode_uint(pos, record->drop_lsn);
+	}
 	if (record->dump_lsn > 0) {
 		pos = mp_encode_uint(pos, VY_LOG_KEY_DUMP_LSN);
 		pos = mp_encode_uint(pos, record->dump_lsn);
@@ -571,6 +586,9 @@ vy_log_record_decode(struct vy_log_record *record,
 		case VY_LOG_KEY_MODIFY_LSN:
 			record->modify_lsn = mp_decode_uint(&pos);
 			break;
+		case VY_LOG_KEY_DROP_LSN:
+			record->drop_lsn = mp_decode_uint(&pos);
+			break;
 		case VY_LOG_KEY_DUMP_LSN:
 			record->dump_lsn = mp_decode_uint(&pos);
 			break;
@@ -1242,9 +1260,9 @@ vy_recovery_do_create_lsm(struct vy_recovery *recovery, int64_t id,
 	lsm->index_id = index_id;
 	lsm->key_parts = NULL;
 	lsm->key_part_count = 0;
-	lsm->is_dropped = false;
 	lsm->create_lsn = -1;
 	lsm->modify_lsn = -1;
+	lsm->drop_lsn = -1;
 	lsm->dump_lsn = -1;
 	rlist_create(&lsm->ranges);
 	rlist_create(&lsm->runs);
@@ -1286,7 +1304,7 @@ vy_recovery_create_lsm(struct vy_recovery *recovery, int64_t id,
 	}
 	struct vy_lsm_recovery_info *lsm;
 	lsm = vy_recovery_lsm_by_index_id(recovery, space_id, index_id);
-	if (lsm != NULL && !lsm->is_dropped) {
+	if (lsm != NULL && lsm->drop_lsn < 0) {
 		diag_set(ClientError, ER_INVALID_VYLOG_FILE,
 			 tt_sprintf("LSM tree %u/%u created twice",
 				    (unsigned)space_id, (unsigned)index_id));
@@ -1329,7 +1347,7 @@ vy_recovery_modify_lsm(struct vy_recovery *recovery, int64_t id,
 				    (long long)id));
 		return -1;
 	}
-	if (lsm->is_dropped) {
+	if (lsm->drop_lsn >= 0) {
 		diag_set(ClientError, ER_INVALID_VYLOG_FILE,
 			 tt_sprintf("Update of deleted LSM tree %lld",
 				    (long long)id));
@@ -1355,7 +1373,7 @@ vy_recovery_modify_lsm(struct vy_recovery *recovery, int64_t id,
  * Returns 0 on success, -1 if ID not found or LSM tree is already marked.
  */
 static int
-vy_recovery_drop_lsm(struct vy_recovery *recovery, int64_t id)
+vy_recovery_drop_lsm(struct vy_recovery *recovery, int64_t id, int64_t drop_lsn)
 {
 	struct vy_lsm_recovery_info *lsm;
 	lsm = vy_recovery_lookup_lsm(recovery, id);
@@ -1365,7 +1383,7 @@ vy_recovery_drop_lsm(struct vy_recovery *recovery, int64_t id)
 				    (long long)id));
 		return -1;
 	}
-	if (lsm->is_dropped) {
+	if (lsm->drop_lsn >= 0) {
 		diag_set(ClientError, ER_INVALID_VYLOG_FILE,
 			 tt_sprintf("LSM tree %lld deleted twice",
 				    (long long)id));
@@ -1386,7 +1404,8 @@ vy_recovery_drop_lsm(struct vy_recovery *recovery, int64_t id)
 			return -1;
 		}
 	}
-	lsm->is_dropped = true;
+	assert(drop_lsn >= 0);
+	lsm->drop_lsn = drop_lsn;
 	return 0;
 }
 
@@ -1408,7 +1427,7 @@ vy_recovery_dump_lsm(struct vy_recovery *recovery,
 				    (long long)id));
 		return -1;
 	}
-	if (lsm->is_dropped) {
+	if (lsm->drop_lsn >= 0) {
 		diag_set(ClientError, ER_INVALID_VYLOG_FILE,
 			 tt_sprintf("Dump of deleted LSM tree %lld",
 				    (long long)id));
@@ -1508,7 +1527,7 @@ vy_recovery_create_run(struct vy_recovery *recovery, int64_t lsm_id,
 				    (long long)lsm_id));
 		return -1;
 	}
-	if (lsm->is_dropped) {
+	if (lsm->drop_lsn >= 0) {
 		diag_set(ClientError, ER_INVALID_VYLOG_FILE,
 			 tt_sprintf("Run %lld created for deleted "
 				    "LSM tree %lld", (long long)run_id,
@@ -1829,7 +1848,8 @@ vy_recovery_process_record(struct vy_recovery *recovery,
 				record->modify_lsn);
 		break;
 	case VY_LOG_DROP_LSM:
-		rc = vy_recovery_drop_lsm(recovery, record->lsm_id);
+		rc = vy_recovery_drop_lsm(recovery, record->lsm_id,
+					  record->drop_lsn);
 		break;
 	case VY_LOG_INSERT_RANGE:
 		rc = vy_recovery_insert_range(recovery, record->lsm_id,
@@ -2132,10 +2152,11 @@ vy_log_append_lsm(struct xlog *xlog, struct vy_lsm_recovery_info *lsm)
 		}
 	}
 
-	if (lsm->is_dropped) {
+	if (lsm->drop_lsn >= 0) {
 		vy_log_record_init(&record);
 		record.type = VY_LOG_DROP_LSM;
 		record.lsm_id = lsm->id;
+		record.drop_lsn = lsm->drop_lsn;
 		if (vy_log_append_record(xlog, &record) != 0)
 			return -1;
 	}
@@ -2162,7 +2183,7 @@ vy_log_create(const struct vclock *vclock, struct vy_recovery *recovery)
 		 * (and thus not needed for garbage collection) from the
 		 * log on rotation.
 		 */
-		if (lsm->is_dropped && rlist_empty(&lsm->runs))
+		if (lsm->drop_lsn >= 0 && rlist_empty(&lsm->runs))
 			continue;
 
 		/* Create the log file on the first write. */
diff --git a/src/box/vy_log.h b/src/box/vy_log.h
index 1b2b419f..22c31825 100644
--- a/src/box/vy_log.h
+++ b/src/box/vy_log.h
@@ -71,7 +71,7 @@ enum vy_log_record_type {
 	VY_LOG_CREATE_LSM		= 0,
 	/**
 	 * Drop an LSM tree.
-	 * Requires vy_log_record::lsm_id.
+	 * Requires vy_log_record::lsm_id, drop_lsn.
 	 */
 	VY_LOG_DROP_LSM			= 1,
 	/**
@@ -209,6 +209,11 @@ struct vy_log_record {
 	int64_t create_lsn;
 	/** LSN of the WAL row that last modified the LSM tree. */
 	int64_t modify_lsn;
+	/**
+	 * LSN of the WAL row that dropped the LSM tree or -1
+	 * if the tree is still active.
+	 */
+	int64_t drop_lsn;
 	/** Max LSN stored on disk. */
 	int64_t dump_lsn;
 	/**
@@ -258,12 +263,15 @@ struct vy_lsm_recovery_info {
 	struct key_part_def *key_parts;
 	/** Number of key parts. */
 	uint32_t key_part_count;
-	/** True if the LSM tree was dropped. */
-	bool is_dropped;
 	/** LSN of the WAL row that created the LSM tree. */
 	int64_t create_lsn;
 	/** LSN of the WAL row that last modified the LSM tree. */
 	int64_t modify_lsn;
+	/**
+	 * LSN of the WAL row that dropped the LSM tree or -1
+	 * if the tree is still active.
+	 */
+	int64_t drop_lsn;
 	/** LSN of the last LSM tree dump. */
 	int64_t dump_lsn;
 	/**
@@ -540,12 +548,13 @@ vy_log_modify_lsm(int64_t id, const struct key_def *key_def, int64_t modify_lsn)
 
 /** Helper to log a vinyl LSM tree drop. */
 static inline void
-vy_log_drop_lsm(int64_t id)
+vy_log_drop_lsm(int64_t id, int64_t drop_lsn)
 {
 	struct vy_log_record record;
 	vy_log_record_init(&record);
 	record.type = VY_LOG_DROP_LSM;
 	record.lsm_id = id;
+	record.drop_lsn = drop_lsn;
 	vy_log_write(&record);
 }
 
diff --git a/src/box/vy_lsm.c b/src/box/vy_lsm.c
index edc3b1a4..289d5c40 100644
--- a/src/box/vy_lsm.c
+++ b/src/box/vy_lsm.c
@@ -543,7 +543,7 @@ vy_lsm_recover(struct vy_lsm *lsm, struct vy_recovery *recovery,
 	lsm->id = lsm_info->id;
 	lsm->commit_lsn = lsm_info->modify_lsn;
 
-	if (lsn < lsm_info->create_lsn || lsm_info->is_dropped) {
+	if (lsn < lsm_info->create_lsn || lsm_info->drop_lsn >= 0) {
 		/*
 		 * Loading a past incarnation of the LSM tree, i.e.
 		 * the LSM tree is going to dropped during final
-- 
2.11.0