Merge pull request #1427 from quickwit-oss/empty_segments_crash

handle empty segments for merge

Author: PSeitz

src/indexer/index_writer.rs

@@ -510,10 +510,12 @@ impl IndexWriter {
         Ok(self.committed_opstamp)
     }
 
-    /// Merges a given list of segments
+    /// Merges a given list of segments.
+    ///
+    /// If all segments are empty no new segment will be created.
     ///
     /// `segment_ids` is required to be non-empty.
-    pub fn merge(&mut self, segment_ids: &[SegmentId]) -> FutureResult<SegmentMeta> {
+    pub fn merge(&mut self, segment_ids: &[SegmentId]) -> FutureResult<Option<SegmentMeta>> {
         let merge_operation = self.segment_updater.make_merge_operation(segment_ids);
         let segment_updater = self.segment_updater.clone();
         segment_updater.start_merge(merge_operation)
@@ -1011,6 +1013,92 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn test_merge_on_empty_segments_single_segment() -> crate::Result<()> {
+        let mut schema_builder = schema::Schema::builder();
+        let text_field = schema_builder.add_text_field("text", schema::TEXT);
+        let index = Index::create_in_ram(schema_builder.build());
+        let reader = index
+            .reader_builder()
+            .reload_policy(ReloadPolicy::Manual)
+            .try_into()?;
+        let num_docs_containing = |s: &str| {
+            let term_a = Term::from_field_text(text_field, s);
+            reader.searcher().doc_freq(&term_a).unwrap()
+        };
+        // writing the segment
+        let mut index_writer = index.writer(12_000_000).unwrap();
+        index_writer.add_document(doc!(text_field=>"a"))?;
+        index_writer.commit()?;
+        //  this should create 1 segment
+
+        let segments = index.searchable_segment_ids().unwrap();
+        assert_eq!(segments.len(), 1);
+
+        reader.reload().unwrap();
+        assert_eq!(num_docs_containing("a"), 1);
+
+        index_writer.delete_term(Term::from_field_text(text_field, "a"));
+        index_writer.commit()?;
+
+        reader.reload().unwrap();
+        assert_eq!(num_docs_containing("a"), 0);
+
+        index_writer.merge(&segments);
+        index_writer.wait_merging_threads().unwrap();
+
+        let segments = index.searchable_segment_ids().unwrap();
+        assert_eq!(segments.len(), 0);
+
+        Ok(())
+    }
+
+    #[test]
+    fn test_merge_on_empty_segments() -> crate::Result<()> {
+        let mut schema_builder = schema::Schema::builder();
+        let text_field = schema_builder.add_text_field("text", schema::TEXT);
+        let index = Index::create_in_ram(schema_builder.build());
+        let reader = index
+            .reader_builder()
+            .reload_policy(ReloadPolicy::Manual)
+            .try_into()?;
+        let num_docs_containing = |s: &str| {
+            let term_a = Term::from_field_text(text_field, s);
+            reader.searcher().doc_freq(&term_a).unwrap()
+        };
+        // writing the segment
+        let mut index_writer = index.writer(12_000_000).unwrap();
+        index_writer.add_document(doc!(text_field=>"a"))?;
+        index_writer.commit()?;
+        index_writer.add_document(doc!(text_field=>"a"))?;
+        index_writer.commit()?;
+        index_writer.add_document(doc!(text_field=>"a"))?;
+        index_writer.commit()?;
+        index_writer.add_document(doc!(text_field=>"a"))?;
+        index_writer.commit()?;
+        //  this should create 4 segments
+
+        let segments = index.searchable_segment_ids().unwrap();
+        assert_eq!(segments.len(), 4);
+
+        reader.reload().unwrap();
+        assert_eq!(num_docs_containing("a"), 4);
+
+        index_writer.delete_term(Term::from_field_text(text_field, "a"));
+        index_writer.commit()?;
+
+        reader.reload().unwrap();
+        assert_eq!(num_docs_containing("a"), 0);
+
+        index_writer.merge(&segments);
+        index_writer.wait_merging_threads().unwrap();
+
+        let segments = index.searchable_segment_ids().unwrap();
+        assert_eq!(segments.len(), 0);
+
+        Ok(())
+    }
+
     #[test]
     fn test_with_merges() -> crate::Result<()> {
         let mut schema_builder = schema::Schema::builder();

src/indexer/merger.rs

@@ -200,6 +200,7 @@ impl IndexMerger {
                 readers.push(reader);
             }
         }
+
         let max_doc = readers.iter().map(|reader| reader.num_docs()).sum();
         if let Some(sort_by_field) = index_settings.sort_by_field.as_ref() {
             readers = Self::sort_readers_by_min_sort_field(readers, sort_by_field)?;

src/indexer/segment_manager.rs

@@ -173,6 +173,7 @@ impl SegmentManager {
                 .to_string();
             return Err(TantivyError::InvalidArgument(error_msg));
         }
+
         Ok(segment_entries)
     }
 
@@ -186,7 +187,7 @@ impl SegmentManager {
     pub(crate) fn end_merge(
         &self,
         before_merge_segment_ids: &[SegmentId],
-        after_merge_segment_entry: SegmentEntry,
+        after_merge_segment_entry: Option<SegmentEntry>,
     ) -> crate::Result<SegmentsStatus> {
         let mut registers_lock = self.write();
         let segments_status = registers_lock
@@ -207,7 +208,9 @@ impl SegmentManager {
         for segment_id in before_merge_segment_ids {
             target_register.remove_segment(segment_id);
         }
-        target_register.add_segment_entry(after_merge_segment_entry);
+        if let Some(entry) = after_merge_segment_entry {
+            target_register.add_segment_entry(entry);
+        }
         Ok(segments_status)
     }
 

src/indexer/segment_updater.rs

@@ -91,7 +91,15 @@ fn merge(
     index: &Index,
     mut segment_entries: Vec<SegmentEntry>,
     target_opstamp: Opstamp,
-) -> crate::Result<SegmentEntry> {
+) -> crate::Result<Option<SegmentEntry>> {
+    let num_docs = segment_entries
+        .iter()
+        .map(|segment| segment.meta().num_docs() as u64)
+        .sum::<u64>();
+    if num_docs == 0 {
+        return Ok(None);
+    }
+
     // first we need to apply deletes to our segment.
     let merged_segment = index.new_segment();
 
@@ -120,7 +128,7 @@ fn merge(
     let merged_segment_id = merged_segment.id();
 
     let segment_meta = index.new_segment_meta(merged_segment_id, num_docs);
-    Ok(SegmentEntry::new(segment_meta, delete_cursor, None))
+    Ok(Some(SegmentEntry::new(segment_meta, delete_cursor, None)))
 }
 
 /// Advanced: Merges a list of segments from different indices in a new index.
@@ -475,7 +483,10 @@ impl SegmentUpdater {
     // suggested and the moment when it ended up being executed.)
     //
     // `segment_ids` is required to be non-empty.
-    pub fn start_merge(&self, merge_operation: MergeOperation) -> FutureResult<SegmentMeta> {
+    pub fn start_merge(
+        &self,
+        merge_operation: MergeOperation,
+    ) -> FutureResult<Option<SegmentMeta>> {
         assert!(
             !merge_operation.segment_ids().is_empty(),
             "Segment_ids cannot be empty."
@@ -512,18 +523,19 @@ impl SegmentUpdater {
                 merge_operation.target_opstamp(),
             ) {
                 Ok(after_merge_segment_entry) => {
-                    let segment_meta_res =
-                        segment_updater.end_merge(merge_operation, after_merge_segment_entry);
-                    let _send_result = merging_future_send.send(segment_meta_res);
+                    let res = segment_updater.end_merge(merge_operation, after_merge_segment_entry);
+                    let _send_result = merging_future_send.send(res);
                 }
                 Err(merge_error) => {
                     warn!(
                         "Merge of {:?} was cancelled: {:?}",
                         merge_operation.segment_ids().to_vec(),
                         merge_error
                     );
+                    if cfg!(test) {
+                        panic!("{:?}", merge_error);
+                    }
                     let _send_result = merging_future_send.send(Err(merge_error));
-                    assert!(!cfg!(test), "Merge failed.");
                 }
             }
         });
@@ -573,35 +585,42 @@ impl SegmentUpdater {
     fn end_merge(
         &self,
         merge_operation: MergeOperation,
-        mut after_merge_segment_entry: SegmentEntry,
-    ) -> crate::Result<SegmentMeta> {
+        mut after_merge_segment_entry: Option<SegmentEntry>,
+    ) -> crate::Result<Option<SegmentMeta>> {
         let segment_updater = self.clone();
-        let after_merge_segment_meta = after_merge_segment_entry.meta().clone();
+        let after_merge_segment_meta = after_merge_segment_entry
+            .as_ref()
+            .map(|after_merge_segment_entry| after_merge_segment_entry.meta().clone());
         self.schedule_task(move || {
-            info!("End merge {:?}", after_merge_segment_entry.meta());
+            info!(
+                "End merge {:?}",
+                after_merge_segment_entry.as_ref().map(|entry| entry.meta())
+            );
             {
-                let mut delete_cursor = after_merge_segment_entry.delete_cursor().clone();
-                if let Some(delete_operation) = delete_cursor.get() {
-                    let committed_opstamp = segment_updater.load_meta().opstamp;
-                    if delete_operation.opstamp < committed_opstamp {
-                        let index = &segment_updater.index;
-                        let segment = index.segment(after_merge_segment_entry.meta().clone());
-                        if let Err(advance_deletes_err) = advance_deletes(
-                            segment,
-                            &mut after_merge_segment_entry,
-                            committed_opstamp,
-                        ) {
-                            error!(
-                                "Merge of {:?} was cancelled (advancing deletes failed): {:?}",
-                                merge_operation.segment_ids(),
-                                advance_deletes_err
-                            );
-                            assert!(!cfg!(test), "Merge failed.");
-
-                            // ... cancel merge
-                            // `merge_operations` are tracked. As it is dropped, the
-                            // the segment_ids will be available again for merge.
-                            return Err(advance_deletes_err);
+                if let Some(after_merge_segment_entry) = after_merge_segment_entry.as_mut() {
+                    let mut delete_cursor = after_merge_segment_entry.delete_cursor().clone();
+                    if let Some(delete_operation) = delete_cursor.get() {
+                        let committed_opstamp = segment_updater.load_meta().opstamp;
+                        if delete_operation.opstamp < committed_opstamp {
+                            let index = &segment_updater.index;
+                            let segment = index.segment(after_merge_segment_entry.meta().clone());
+                            if let Err(advance_deletes_err) = advance_deletes(
+                                segment,
+                                after_merge_segment_entry,
+                                committed_opstamp,
+                            ) {
+                                error!(
+                                    "Merge of {:?} was cancelled (advancing deletes failed): {:?}",
+                                    merge_operation.segment_ids(),
+                                    advance_deletes_err
+                                );
+                                assert!(!cfg!(test), "Merge failed.");
+
+                                // ... cancel merge
+                                // `merge_operations` are tracked. As it is dropped, the
+                                // the segment_ids will be available again for merge.
+                                return Err(advance_deletes_err);
+                            }
                         }
                     }
                 }