Hotfix #681

Fixing the construction of the DeleteBitset.
Cherry picked from master.

Changing version to 0.10.3

CHANGELOG.md

@@ -1,7 +1,12 @@
-Tantivy 0.11.0
+Tantivy 0.10.3
+==========================
+
+- Fix crash when committing multiple times with deleted documents. #681 (@brainlock)
+
+  Tantivy 0.10.2
 =====================
 
-- Added f64 field. Internally reuse u64 code the same way i64 does (@fdb-hiroshima)
+- Closes #656. Solving memory leak.
 
 Tantivy 0.10.1
 =====================

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "tantivy"
-version = "0.10.1"
+version = "0.10.3"
 authors = ["Paul Masurel <paul.masurel@gmail.com>"]
 license = "MIT"
 categories = ["database-implementations", "data-structures"]
@@ -86,6 +86,7 @@ travis-ci = { repository = "tantivy-search/tantivy" }
 
 [dev-dependencies.fail]
 features = ["failpoints"]
+version = "0.3"
 
 
 # Following the "fail" crate best practises, we isolate
@@ -98,4 +99,4 @@ features = ["failpoints"]
 [[test]]
 name = "failpoints"
 path = "tests/failpoints/mod.rs"
-required-features = ["fail/failpoints"]
+required-features = ["fail/failpoints"]

src/collector/top_score_collector.rs

@@ -13,7 +13,6 @@ use crate::Result;
 use crate::Score;
 use crate::SegmentLocalId;
 use crate::SegmentReader;
-use std::fmt;
 
 /// The Top Score Collector keeps track of the K documents
 /// sorted by their score.
@@ -69,12 +68,6 @@ use std::fmt;
 /// ```
 pub struct TopDocs(TopCollector<Score>);
 
-impl fmt::Debug for TopDocs {
-    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        write!(f, "TopDocs({})", self.0.limit())
-    }
-}
-
 impl TopDocs {
     /// Creates a top score collector, with a number of documents equal to "limit".
     ///

src/core/index_meta.rs

@@ -151,6 +151,21 @@ impl SegmentMeta {
         self.num_deleted_docs() > 0
     }
 
+    /// Updates the max_doc value from the `SegmentMeta`.
+    ///
+    /// This method is only used when updating `max_doc` from 0
+    /// as we finalize a fresh new segment.
+    pub(crate) fn with_max_doc(self, max_doc: u32) -> SegmentMeta {
+        assert_eq!(self.tracked.max_doc, 0);
+        assert!(self.tracked.deletes.is_none());
+        let tracked = self.tracked.map(move |inner_meta| InnerSegmentMeta {
+            segment_id: inner_meta.segment_id,
+            max_doc,
+            deletes: None,
+        });
+        SegmentMeta { tracked }
+    }
+
     #[doc(hidden)]
     pub fn with_delete_meta(self, num_deleted_docs: u32, opstamp: Opstamp) -> SegmentMeta {
         let delete_meta = DeleteMeta {

src/core/segment.rs

@@ -50,6 +50,17 @@ impl Segment {
         &self.meta
     }
 
+    /// Updates the max_doc value from the `SegmentMeta`.
+    ///
+    /// This method is only used when updating `max_doc` from 0
+    /// as we finalize a fresh new segment.
+    pub(crate) fn with_max_doc(self, max_doc: u32) -> Segment {
+        Segment {
+            index: self.index,
+            meta: self.meta.with_max_doc(max_doc),
+        }
+    }
+
     #[doc(hidden)]
     pub fn with_delete_meta(self, num_deleted_docs: u32, opstamp: Opstamp) -> Segment {
         Segment {

src/directory/mmap_directory.rs

@@ -141,42 +141,28 @@ impl MmapCache {
     }
 }
 
-struct InnerWatcherWrapper {
-    _watcher: Mutex<notify::RecommendedWatcher>,
-    watcher_router: WatchCallbackList,
-}
-
-impl InnerWatcherWrapper {
-    pub fn new(path: &Path) -> Result<(Self, Receiver<notify::RawEvent>), notify::Error> {
-        let (tx, watcher_recv): (Sender<RawEvent>, Receiver<RawEvent>) = channel();
-        // We need to initialize the
-        let mut watcher = notify::raw_watcher(tx)?;
-        watcher.watch(path, RecursiveMode::Recursive)?;
-        let inner = InnerWatcherWrapper {
-            _watcher: Mutex::new(watcher),
-            watcher_router: Default::default(),
-        };
-        Ok((inner, watcher_recv))
-    }
-}
-
-#[derive(Clone)]
 struct WatcherWrapper {
-    inner: Arc<InnerWatcherWrapper>,
+    _watcher: Mutex<notify::RecommendedWatcher>,
+    watcher_router: Arc<WatchCallbackList>,
 }
 
 impl WatcherWrapper {
     pub fn new(path: &Path) -> Result<Self, OpenDirectoryError> {
-        let (inner, watcher_recv) = InnerWatcherWrapper::new(path).map_err(|err| match err {
-            notify::Error::PathNotFound => OpenDirectoryError::DoesNotExist(path.to_owned()),
-            _ => {
-                panic!("Unknown error while starting watching directory {:?}", path);
-            }
-        })?;
-        let watcher_wrapper = WatcherWrapper {
-            inner: Arc::new(inner),
-        };
-        let watcher_wrapper_clone = watcher_wrapper.clone();
+        let (tx, watcher_recv): (Sender<RawEvent>, Receiver<RawEvent>) = channel();
+        // We need to initialize the
+        let watcher = notify::raw_watcher(tx)
+            .and_then(|mut watcher| {
+                watcher.watch(path, RecursiveMode::Recursive)?;
+                Ok(watcher)
+            })
+            .map_err(|err| match err {
+                notify::Error::PathNotFound => OpenDirectoryError::DoesNotExist(path.to_owned()),
+                _ => {
+                    panic!("Unknown error while starting watching directory {:?}", path);
+                }
+            })?;
+        let watcher_router: Arc<WatchCallbackList> = Default::default();
+        let watcher_router_clone = watcher_router.clone();
         thread::Builder::new()
             .name("meta-file-watch-thread".to_string())
             .spawn(move || {
@@ -187,7 +173,7 @@ impl WatcherWrapper {
                             // We might want to be more accurate than this at one point.
                             if let Some(filename) = changed_path.file_name() {
                                 if filename == *META_FILEPATH {
-                                    watcher_wrapper_clone.inner.watcher_router.broadcast();
+                                    watcher_router_clone.broadcast();
                                 }
                             }
                         }
@@ -200,13 +186,15 @@ impl WatcherWrapper {
                         }
                     }
                 }
-            })
-            .expect("Failed to spawn thread to watch meta.json");
-        Ok(watcher_wrapper)
+            })?;
+        Ok(WatcherWrapper {
+            _watcher: Mutex::new(watcher),
+            watcher_router,
+        })
     }
 
     pub fn watch(&mut self, watch_callback: WatchCallback) -> WatchHandle {
-        self.inner.watcher_router.subscribe(watch_callback)
+        self.watcher_router.subscribe(watch_callback)
     }
 }
 

src/fastfield/delete.rs

@@ -10,11 +10,14 @@ use std::io::Write;
 /// Write a delete `BitSet`
 ///
 /// where `delete_bitset` is the set of deleted `DocId`.
-pub fn write_delete_bitset(delete_bitset: &BitSet, writer: &mut WritePtr) -> io::Result<()> {
-    let max_doc = delete_bitset.capacity();
+pub fn write_delete_bitset(
+    delete_bitset: &BitSet,
+    max_doc: u32,
+    writer: &mut WritePtr,
+) -> io::Result<()> {
     let mut byte = 0u8;
     let mut shift = 0u8;
-    for doc in 0..max_doc {
+    for doc in 0..(max_doc as usize) {
         if delete_bitset.contains(doc) {
             byte |= 1 << shift;
         }
@@ -86,18 +89,17 @@ mod tests {
     use bit_set::BitSet;
     use std::path::PathBuf;
 
-    fn test_delete_bitset_helper(bitset: &BitSet) {
+    fn test_delete_bitset_helper(bitset: &BitSet, max_doc: u32) {
         let test_path = PathBuf::from("test");
         let mut directory = RAMDirectory::create();
         {
             let mut writer = directory.open_write(&*test_path).unwrap();
-            write_delete_bitset(bitset, &mut writer).unwrap();
+            write_delete_bitset(bitset, max_doc, &mut writer).unwrap();
         }
         {
             let source = directory.open_read(&test_path).unwrap();
             let delete_bitset = DeleteBitSet::open(source);
-            let n = bitset.capacity();
-            for doc in 0..n {
+            for doc in 0..max_doc as usize {
                 assert_eq!(bitset.contains(doc), delete_bitset.is_deleted(doc as DocId));
             }
             assert_eq!(delete_bitset.len(), bitset.len());
@@ -110,7 +112,7 @@ mod tests {
             let mut bitset = BitSet::with_capacity(10);
             bitset.insert(1);
             bitset.insert(9);
-            test_delete_bitset_helper(&bitset);
+            test_delete_bitset_helper(&bitset, 10);
         }
         {
             let mut bitset = BitSet::with_capacity(8);
@@ -119,7 +121,7 @@ mod tests {
             bitset.insert(3);
             bitset.insert(5);
             bitset.insert(7);
-            test_delete_bitset_helper(&bitset);
+            test_delete_bitset_helper(&bitset, 8);
         }
     }
 }

src/indexer/index_writer.rs

@@ -147,7 +147,6 @@ pub(crate) fn advance_deletes(
         };
 
         let delete_cursor = segment_entry.delete_cursor();
-
         compute_deleted_bitset(
             &mut delete_bitset,
             &segment_reader,
@@ -167,8 +166,8 @@ pub(crate) fn advance_deletes(
         if num_deleted_docs > 0 {
             segment = segment.with_delete_meta(num_deleted_docs as u32, target_opstamp);
             let mut delete_file = segment.open_write(SegmentComponent::DELETE)?;
-            write_delete_bitset(&delete_bitset, &mut delete_file)?;
-        }
+            write_delete_bitset(&delete_bitset, max_doc, &mut delete_file)?;
+}
     }
     segment_entry.set_meta(segment.meta().clone());
     Ok(())
@@ -176,13 +175,13 @@ pub(crate) fn advance_deletes(
 
 fn index_documents(
     memory_budget: usize,
-    segment: &Segment,
+    segment: Segment,
     grouped_document_iterator: &mut dyn Iterator<Item = OperationGroup>,
     segment_updater: &mut SegmentUpdater,
     mut delete_cursor: DeleteCursor,
 ) -> Result<bool> {
     let schema = segment.schema();
-    let segment_id = segment.id();
+
     let mut segment_writer = SegmentWriter::for_segment(memory_budget, segment.clone(), &schema)?;
     for document_group in grouped_document_iterator {
         for doc in document_group {
@@ -202,24 +201,28 @@ fn index_documents(
         return Ok(false);
     }
 
-    let num_docs = segment_writer.max_doc();
+    let max_doc = segment_writer.max_doc();
 
     // this is ensured by the call to peek before starting
     // the worker thread.
-    assert!(num_docs > 0);
+    assert!(max_doc > 0);
 
     let doc_opstamps: Vec<Opstamp> = segment_writer.finalize()?;
-    let segment_meta = segment
-        .index()
-        .inventory()
-        .new_segment_meta(segment_id, num_docs);
-
+    let segment_with_max_doc = segment.with_max_doc(max_doc);
     let last_docstamp: Opstamp = *(doc_opstamps.last().unwrap());
 
-    let delete_bitset_opt =
-        apply_deletes(&segment, &mut delete_cursor, &doc_opstamps, last_docstamp)?;
+    let delete_bitset_opt = apply_deletes(
+        &segment_with_max_doc,
+        &mut delete_cursor,
+        &doc_opstamps,
+        last_docstamp,
+    )?;
 
-    let segment_entry = SegmentEntry::new(segment_meta, delete_cursor, delete_bitset_opt);
+    let segment_entry = SegmentEntry::new(
+        segment_with_max_doc.meta().clone(),
+        delete_cursor,
+        delete_bitset_opt,
+    );
     Ok(segment_updater.add_segment(segment_entry))
 }
 
@@ -236,7 +239,9 @@ fn apply_deletes(
     }
     let segment_reader = SegmentReader::open(segment)?;
     let doc_to_opstamps = DocToOpstampMapping::from(doc_opstamps);
-    let mut deleted_bitset = BitSet::with_capacity(segment_reader.max_doc() as usize);
+
+    let max_doc = segment.meta().max_doc();
+    let mut deleted_bitset = BitSet::with_capacity(max_doc as usize);
     let may_have_deletes = compute_deleted_bitset(
         &mut deleted_bitset,
         &segment_reader,
@@ -408,7 +413,7 @@ impl IndexWriter {
                     let segment = index.new_segment();
                     index_documents(
                         mem_budget,
-                        &segment,
+                        segment,
                         &mut document_iterator,
                         &mut segment_updater,
                         delete_cursor.clone(),

src/indexer/mod.rs

@@ -28,3 +28,25 @@ pub use self::segment_writer::SegmentWriter;
 
 /// Alias for the default merge policy, which is the `LogMergePolicy`.
 pub type DefaultMergePolicy = LogMergePolicy;
+
+#[cfg(test)]
+mod tests {
+    use crate::schema::{self, Schema};
+    use crate::{Index, Term};
+    #[test]
+    fn test_advance_delete_bug() {
+        let mut schema_builder = Schema::builder();
+        let text_field = schema_builder.add_text_field("text", schema::TEXT);
+        let index = Index::create_from_tempdir(schema_builder.build()).unwrap();
+        let mut index_writer = index.writer_with_num_threads(1, 3_000_000).unwrap();
+        // there must be one deleted document in the segment
+        index_writer.add_document(doc!(text_field=>"b"));
+        index_writer.delete_term(Term::from_field_text(text_field, "b"));
+        // we need enough data to trigger the bug (at least 32 documents)
+        for _ in 0..32 {
+            index_writer.add_document(doc!(text_field=>"c"));
+        }
+        index_writer.commit().unwrap();
+        index_writer.commit().unwrap();
+    }
+}

src/lib.rs

@@ -171,16 +171,16 @@ pub use self::snippet::{Snippet, SnippetGenerator};
 mod docset;
 pub use self::docset::{DocSet, SkipResult};
 
-pub use crate::common::{f64_to_u64, i64_to_u64, u64_to_f64, u64_to_i64};
 pub use crate::core::SegmentComponent;
 pub use crate::core::{Index, IndexMeta, Searcher, Segment, SegmentId, SegmentMeta};
 pub use crate::core::{InvertedIndexReader, SegmentReader};
 pub use crate::directory::Directory;
 pub use crate::indexer::IndexWriter;
 pub use crate::postings::Postings;
-pub use crate::reader::LeasedItem;
 pub use crate::schema::{Document, Term};
 
+pub use crate::common::{i64_to_u64, u64_to_i64, f64_to_u64, u64_to_f64};
+
 /// Expose the current version of tantivy, as well
 /// whether it was compiled with the simd compression.
 pub fn version() -> &'static str {
@@ -849,8 +849,7 @@ mod tests {
         let index = Index::create_in_ram(schema);
         let mut index_writer = index.writer_with_num_threads(1, 50_000_000).unwrap();
         {
-            let document =
-                doc!(fast_field_unsigned => 4u64, fast_field_signed=>4i64, fast_field_float=>4f64);
+            let document = doc!(fast_field_unsigned => 4u64, fast_field_signed=>4i64, fast_field_float=>4f64);
             index_writer.add_document(document);
             index_writer.commit().unwrap();
         }

src/query/query_parser/query_parser.rs

@@ -18,56 +18,42 @@ use crate::schema::{FieldType, Term};
 use crate::tokenizer::TokenizerManager;
 use combine::Parser;
 use std::borrow::Cow;
-use std::num::{ParseFloatError, ParseIntError};
+use std::num::{ParseIntError, ParseFloatError};
 use std::ops::Bound;
 use std::str::FromStr;
 
 /// Possible error that may happen when parsing a query.
-#[derive(Debug, PartialEq, Eq, Fail)]
+#[derive(Debug, PartialEq, Eq)]
 pub enum QueryParserError {
     /// Error in the query syntax
-    #[fail(display = "Syntax Error")]
     SyntaxError,
     /// `FieldDoesNotExist(field_name: String)`
     /// The query references a field that is not in the schema
-    #[fail(display = "File does not exists: '{:?}'", _0)]
     FieldDoesNotExist(String),
     /// The query contains a term for a `u64` or `i64`-field, but the value
     /// is neither.
-    #[fail(display = "Expected a valid integer: '{:?}'", _0)]
     ExpectedInt(ParseIntError),
     /// The query contains a term for a `f64`-field, but the value
     /// is not a f64.
-    #[fail(display = "Invalid query: Only excluding terms given")]
     ExpectedFloat(ParseFloatError),
     /// It is forbidden queries that are only "excluding". (e.g. -title:pop)
-    #[fail(display = "Invalid query: Only excluding terms given")]
     AllButQueryForbidden,
     /// If no default field is declared, running a query without any
     /// field specified is forbbidden.
-    #[fail(display = "No default field declared and no field specified in query")]
     NoDefaultFieldDeclared,
     /// The field searched for is not declared
     /// as indexed in the schema.
-    #[fail(display = "The field '{:?}' is not declared as indexed", _0)]
     FieldNotIndexed(String),
     /// A phrase query was requested for a field that does not
     /// have any positions indexed.
-    #[fail(display = "The field '{:?}' does not have positions indexed", _0)]
     FieldDoesNotHavePositionsIndexed(String),
     /// The tokenizer for the given field is unknown
     /// The two argument strings are the name of the field, the name of the tokenizer
-    #[fail(
-        display = "The tokenizer '{:?}' for the field '{:?}' is unknown",
-        _0, _1
-    )]
     UnknownTokenizer(String, String),
     /// The query contains a range query with a phrase as one of the bounds.
     /// Only terms can be used as bounds.
-    #[fail(display = "A range query cannot have a phrase as one of the bounds")]
     RangeMustNotHavePhrase,
     /// The format for the date field is not RFC 3339 compliant.
-    #[fail(display = "The date field has an invalid format")]
     DateFormatError(chrono::ParseError),
 }
 

src/reader/mod.rs

@@ -1,7 +1,6 @@
 mod pool;
 
-pub use self::pool::LeasedItem;
-use self::pool::Pool;
+use self::pool::{LeasedItem, Pool};
 use crate::core::Segment;
 use crate::directory::Directory;
 use crate::directory::WatchHandle;

src/reader/pool.rs

@@ -123,10 +123,6 @@ impl<T> Pool<T> {
     }
 }
 
-/// A LeasedItem holds an object borrowed from a Pool.
-///
-/// Upon drop, the object is automatically returned
-/// into the pool.
 pub struct LeasedItem<T> {
     gen_item: Option<GenerationItem<T>>,
     recycle_queue: Arc<Queue<GenerationItem<T>>>,

src/schema/field.rs

@@ -12,6 +12,19 @@ use std::io::Write;
 #[derive(Copy, Clone, Debug, PartialEq, PartialOrd, Eq, Ord, Hash, Serialize, Deserialize)]
 pub struct Field(pub u32);
 
+impl Field {
+    /// Create a new field object for the given FieldId.
+    pub fn from_field_id(field_id: u32) -> Field {
+        Field(field_id)
+    }
+
+    /// Returns a u32 identifying uniquely a field within a schema.
+    #[allow(clippy::trivially_copy_pass_by_ref)]
+    pub fn field_id(&self) -> u32 {
+        self.0
+    }
+}
+
 impl BinarySerializable for Field {
     fn serialize<W: Write>(&self, writer: &mut W) -> io::Result<()> {
         self.0.serialize(writer)

src/schema/schema.rs

@@ -261,24 +261,6 @@ impl Schema {
         NamedFieldDocument(field_map)
     }
 
-    /// Converts a named doc into a document.
-    pub fn from_named_doc(
-        &self,
-        named_doc: NamedFieldDocument,
-    ) -> Result<Document, DocParsingError> {
-        let mut doc = Document::default();
-        for (field_name, field_values) in named_doc.0 {
-            if let Some(field) = self.get_field(&field_name) {
-                for field_value in field_values {
-                    doc.add(FieldValue::new(field, field_value));
-                }
-            } else {
-                return Err(DocParsingError::NoSuchFieldInSchema(field_name.clone()));
-            }
-        }
-        Ok(doc)
-    }
-
     /// Encode the schema in JSON.
     ///
     /// Encoding a document cannot fail.
@@ -297,6 +279,7 @@ impl Schema {
                 };
                 DocParsingError::NotJSON(doc_json_sample)
             })?;
+
         let mut doc = Document::default();
         for (field_name, json_value) in json_obj.iter() {
             match self.get_field(field_name) {
@@ -377,16 +360,13 @@ impl<'de> Deserialize<'de> for Schema {
 
 /// Error that may happen when deserializing
 /// a document from JSON.
-#[derive(Debug, Fail)]
+#[derive(Debug)]
 pub enum DocParsingError {
     /// The payload given is not valid JSON.
-    #[fail(display = "The provided string is not valid JSON")]
     NotJSON(String),
     /// One of the value node could not be parsed.
-    #[fail(display = "The field '{:?}' could not be parsed: {:?}", _0, _1)]
     ValueError(String, ValueParsingError),
     /// The json-document contains a field that is not declared in the schema.
-    #[fail(display = "The json-document contains an unknown field: {:?}", _0)]
     NoSuchFieldInSchema(String),
 }