From b974e7ce344d057be65aeef0ec96e7f81d5f0afd Mon Sep 17 00:00:00 2001
From: Paul Masurel <paul.masurel@gmail.com>
Date: Fri, 22 May 2020 16:24:07 +0900
Subject: [PATCH] Closes #828. (#829)

There was a bug in the LogMergePolicy that was surfacing when there were
segments, but all of the segments were larger than the max limit.

After filtering, the list of segments candidate for merge was 0, and
the code was indexing the first element of an empty Vec.
---
 src/indexer/log_merge_policy.rs | 52 +++++++++++++++++++++------------
 1 file changed, 34 insertions(+), 18 deletions(-)

diff --git a/src/indexer/log_merge_policy.rs b/src/indexer/log_merge_policy.rs
index 591d5e840..caf50f6f8 100644
--- a/src/indexer/log_merge_policy.rs
+++ b/src/indexer/log_merge_policy.rs
@@ -54,10 +54,6 @@ impl LogMergePolicy {
 
 impl MergePolicy for LogMergePolicy {
     fn compute_merge_candidates(&self, segments: &[SegmentMeta]) -> Vec<MergeCandidate> {
-        if segments.is_empty() {
-            return Vec::new();
-        }
-
         let mut size_sorted_tuples = segments
             .iter()
             .map(SegmentMeta::num_docs)
@@ -67,27 +63,35 @@ impl MergePolicy for LogMergePolicy {
 
         size_sorted_tuples.sort_by(|x, y| y.1.cmp(&(x.1)));
 
+        if size_sorted_tuples.len() <= 1 {
+            return Vec::new();
+        }
+
         let size_sorted_log_tuples: Vec<_> = size_sorted_tuples
             .into_iter()
             .map(|(ind, num_docs)| (ind, f64::from(self.clip_min_size(num_docs)).log2()))
             .collect();
 
-        let (first_ind, first_score) = size_sorted_log_tuples[0];
-        let mut current_max_log_size = first_score;
-        let mut levels = vec![vec![first_ind]];
-        for &(ind, score) in (&size_sorted_log_tuples).iter().skip(1) {
-            if score < (current_max_log_size - self.level_log_size) {
-                current_max_log_size = score;
-                levels.push(Vec::new());
+        if let Some(&(first_ind, first_score)) = size_sorted_log_tuples.first() {
+            let mut current_max_log_size = first_score;
+            let mut levels = vec![vec![first_ind]];
+            for &(ind, score) in (&size_sorted_log_tuples).iter().skip(1) {
+                if score < (current_max_log_size - self.level_log_size) {
+                    current_max_log_size = score;
+                    levels.push(Vec::new());
+                }
+                levels.last_mut().unwrap().push(ind);
             }
-            levels.last_mut().unwrap().push(ind);
+            levels
+                .iter()
+                .filter(|level| level.len() >= self.min_merge_size)
+                .map(|ind_vec| {
+                    MergeCandidate(ind_vec.iter().map(|&ind| segments[ind].id()).collect())
+                })
+                .collect()
+        } else {
+            return vec![];
         }
-
-        levels
-            .iter()
-            .filter(|level| level.len() >= self.min_merge_size)
-            .map(|ind_vec| MergeCandidate(ind_vec.iter().map(|&ind| segments[ind].id()).collect()))
-            .collect()
     }
 }
 
@@ -179,6 +183,7 @@ mod tests {
         let result_list = test_merge_policy().compute_merge_candidates(&test_input);
         assert_eq!(result_list.len(), 2);
     }
+
     #[test]
     fn test_log_merge_policy_small_segments() {
         // segments under min_layer_size are merged together
@@ -194,6 +199,17 @@ mod tests {
         assert_eq!(result_list.len(), 1);
     }
 
+    #[test]
+    fn test_log_merge_policy_all_segments_too_large_to_merge() {
+        let eight_large_segments: Vec<SegmentMeta> =
+            std::iter::repeat_with(|| create_random_segment_meta(100_001))
+                .take(8)
+                .collect();
+        assert!(test_merge_policy()
+            .compute_merge_candidates(&eight_large_segments)
+            .is_empty());
+    }
+
     #[test]
     fn test_large_merge_segments() {
         let test_input = vec![