forward delete bitsets on merge, add tests

src/core/segment_reader.rs

@@ -5,6 +5,7 @@ use crate::core::SegmentId;
 use crate::directory::CompositeFile;
 use crate::directory::FileSlice;
 use crate::error::DataCorruption;
+use crate::fastfield::merge_delete_bitset;
 use crate::fastfield::DeleteBitSet;
 use crate::fastfield::FacetReader;
 use crate::fastfield::FastFieldReaders;
@@ -69,6 +70,17 @@ impl SegmentReader {
         &self.schema
     }
 
+    /// Merges the passed bitset with the existing one.
+    pub fn apply_delete_bitset(&mut self, delete_bitset: DeleteBitSet) {
+        if let Some(existing_bitset) = self.delete_bitset_opt.as_mut() {
+            let merged_bitset = merge_delete_bitset(&delete_bitset, existing_bitset);
+            self.delete_bitset_opt = Some(merged_bitset);
+        } else {
+            self.delete_bitset_opt = Some(delete_bitset);
+        }
+        self.num_docs = self.max_doc - self.num_deleted_docs();
+    }
+
     /// Return the number of documents that have been
     /// deleted in the segment.
     pub fn num_deleted_docs(&self) -> DocId {

src/fastfield/delete.rs

@@ -69,7 +69,7 @@ pub fn merge_delete_bitset(left: &DeleteBitSet, right: &DeleteBitSet) -> DeleteB
 /// Set of deleted `DocId`s.
 #[derive(Clone)]
 pub struct DeleteBitSet {
-    pub(crate) data: OwnedBytes,
+    data: OwnedBytes,
     num_deleted: usize,
 }
 

src/indexer/merger.rs

@@ -1,5 +1,4 @@
 use crate::error::DataCorruption;
-use crate::fastfield::merge_delete_bitset;
 use crate::fastfield::CompositeFastFieldSerializer;
 use crate::fastfield::DeleteBitSet;
 use crate::fastfield::DynamicFastFieldReader;
@@ -93,7 +92,7 @@ impl<'a> From<(usize, &'a SegmentReader)> for SegmentReaderWithOrdinal<'a> {
 pub struct IndexMerger {
     index_settings: IndexSettings,
     schema: Schema,
-    readers: Vec<SegmentReader>,
+    pub(crate) readers: Vec<SegmentReader>,
     max_doc: u32,
 }
 
@@ -208,20 +207,15 @@ impl IndexMerger {
         for segment in segments {
             if segment.meta().num_docs() > 0 {
                 let reader = SegmentReader::open(segment)?;
-                max_doc += reader.num_docs();
                 readers.push(reader);
             }
         }
         for (reader, new_delete_bitset_opt) in readers.iter_mut().zip(delete_bitset_opt.into_iter())
         {
             if let Some(new_delete_bitset) = new_delete_bitset_opt {
-                if let Some(existing_bitset) = reader.delete_bitset_opt.as_mut() {
-                    let merged_bitset = merge_delete_bitset(&new_delete_bitset, existing_bitset);
-                    reader.delete_bitset_opt = Some(merged_bitset);
-                } else {
-                    reader.delete_bitset_opt = Some(new_delete_bitset);
-                }
+                reader.apply_delete_bitset(new_delete_bitset);
             }
+            max_doc += reader.num_docs();
         }
         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/mod.rs

@@ -26,7 +26,7 @@ pub use self::prepared_commit::PreparedCommit;
 pub use self::segment_entry::SegmentEntry;
 pub use self::segment_manager::SegmentManager;
 pub use self::segment_serializer::SegmentSerializer;
-pub use self::segment_updater::merge_segments;
+pub use self::segment_updater::merge_indices;
 pub use self::segment_writer::SegmentWriter;
 
 /// Alias for the default merge policy, which is the `LogMergePolicy`.