[PATCH] vinyl: factor out procedure looking up LSM tree range intersection

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

It's an independent piece of code that is definitely worth moving
from vy_scheduler to vy_lsm internals anyway. Besides, having it
wrapped up in a separate function will make it easier to patch.
---
 src/box/vy_lsm.c       | 24 ++++++++++++++++++++++++
 src/box/vy_lsm.h       | 16 ++++++++++++++++
 src/box/vy_scheduler.c | 25 +++----------------------
 3 files changed, 43 insertions(+), 22 deletions(-)

diff --git a/src/box/vy_lsm.c b/src/box/vy_lsm.c
index 727a5397..1e08c0e1 100644
--- a/src/box/vy_lsm.c
+++ b/src/box/vy_lsm.c
@@ -1065,6 +1065,30 @@ vy_lsm_rollback_stmt(struct vy_lsm *lsm, struct vy_mem *mem,
 	vy_cache_on_write(&lsm->cache, stmt, NULL);
 }
 
+int
+vy_lsm_find_range_intersection(struct vy_lsm *lsm,
+		const char *min_key, const char *max_key,
+		struct vy_range **begin, struct vy_range **end)
+{
+	struct tuple_format *key_format = lsm->env->key_format;
+	struct tuple *stmt;
+
+	stmt = vy_key_from_msgpack(key_format, min_key);
+	if (stmt == NULL)
+		return -1;
+	*begin = vy_range_tree_psearch(&lsm->range_tree, stmt);
+	tuple_unref(stmt);
+
+	stmt = vy_key_from_msgpack(key_format, max_key);
+	if (stmt == NULL)
+		return -1;
+	*end = vy_range_tree_psearch(&lsm->range_tree, stmt);
+	*end = vy_range_tree_next(&lsm->range_tree, *end);
+	tuple_unref(stmt);
+
+	return 0;
+}
+
 bool
 vy_lsm_split_range(struct vy_lsm *lsm, struct vy_range *range)
 {
diff --git a/src/box/vy_lsm.h b/src/box/vy_lsm.h
index e969e45c..c85bd2b3 100644
--- a/src/box/vy_lsm.h
+++ b/src/box/vy_lsm.h
@@ -536,6 +536,22 @@ void
 vy_lsm_delete_mem(struct vy_lsm *lsm, struct vy_mem *mem);
 
 /**
+ * Lookup ranges intersecting [min_key, max_key] interval in
+ * the given LSM tree.
+ *
+ * On success returns 0 and sets @begin to the first range
+ * and @end to the one following the last range instersecting
+ * the given interval (NULL if max_key lays in the rightmost
+ * range).
+ *
+ * On memory allocation error returns -1 and sets diag.
+ */
+int
+vy_lsm_find_range_intersection(struct vy_lsm *lsm,
+		const char *min_key, const char *max_key,
+		struct vy_range **begin, struct vy_range **end);
+
+/**
  * Split a range if it has grown too big, return true if the range
  * was split. Splitting is done by making slices of the runs used
  * by the original range, adding them to new ranges, and reflecting
diff --git a/src/box/vy_scheduler.c b/src/box/vy_scheduler.c
index 6191bcd9..aaae1a89 100644
--- a/src/box/vy_scheduler.c
+++ b/src/box/vy_scheduler.c
@@ -1103,11 +1103,9 @@ vy_task_dump_complete(struct vy_task *task)
 	double dump_time = ev_monotonic_now(loop()) - task->start_time;
 	struct vy_disk_stmt_counter dump_output = new_run->count;
 	struct vy_stmt_counter dump_input;
-	struct tuple_format *key_format = lsm->env->key_format;
 	struct vy_mem *mem, *next_mem;
 	struct vy_slice **new_slices, *slice;
 	struct vy_range *range, *begin_range, *end_range;
-	struct tuple *min_key, *max_key;
 	int i;
 
 	assert(lsm->is_dumping);
@@ -1131,28 +1129,11 @@ vy_task_dump_complete(struct vy_task *task)
 
 	/*
 	 * Figure out which ranges intersect the new run.
-	 * @begin_range is the first range intersecting the run.
-	 * @end_range is the range following the last range
-	 * intersecting the run or NULL if the run itersects all
-	 * ranges.
 	 */
-	min_key = vy_key_from_msgpack(key_format, new_run->info.min_key);
-	if (min_key == NULL)
+	if (vy_lsm_find_range_intersection(lsm, new_run->info.min_key,
+					   new_run->info.max_key,
+					   &begin_range, &end_range) != 0)
 		goto fail;
-	max_key = vy_key_from_msgpack(key_format, new_run->info.max_key);
-	if (max_key == NULL) {
-		tuple_unref(min_key);
-		goto fail;
-	}
-	begin_range = vy_range_tree_psearch(&lsm->range_tree, min_key);
-	end_range = vy_range_tree_psearch(&lsm->range_tree, max_key);
-	/*
-	 * If min_key == max_key, the slice has to span over at
-	 * least one range.
-	 */
-	end_range = vy_range_tree_next(&lsm->range_tree, end_range);
-	tuple_unref(min_key);
-	tuple_unref(max_key);
 
 	/*
 	 * For each intersected range allocate a slice of the new run.
-- 
2.11.0