Added soft commits

CHANGELOG.md

@@ -1,3 +1,9 @@
+Tantivy 0.12.0
+==========================
+- By default IndexReader are in `Manual` mode.
+
+
+
 Tantivy 0.11.3
 =======================
 - Fixed DateTime as a fast field (#735)

src/common/composite_file.rs

@@ -186,7 +186,7 @@ mod test {
     use super::{CompositeFile, CompositeWrite};
     use crate::common::BinarySerializable;
     use crate::common::VInt;
-    use crate::directory::{Directory, RAMDirectory, ReadOnlyDirectory};
+    use crate::directory::{Directory, RAMDirectory};
     use crate::schema::Field;
     use std::io::Write;
     use std::path::Path;

src/core/index.rs

@@ -1,4 +1,3 @@
-use super::segment::create_segment;
 use super::segment::Segment;
 use crate::core::Executor;
 use crate::core::IndexMeta;
@@ -331,19 +330,25 @@ impl Index {
             .collect())
     }
 
-    #[doc(hidden)]
-    pub fn segment(&self, segment_meta: SegmentMeta) -> Segment {
-        create_segment(self.clone(), segment_meta)
+    pub(crate) fn segment(&self, segment_meta: SegmentMeta) -> Segment {
+        Segment::for_index(self.clone(), segment_meta)
     }
 
     /// Creates a new segment.
     pub fn new_segment(&self) -> Segment {
-        let mut segment_meta = self
+        let segment_meta = self
             .inventory
             .new_segment_meta(SegmentId::generate_random(), 0);
         self.segment(segment_meta)
     }
 
+    /// Creates a new segment.
+    pub(crate) fn new_segment_unpersisted(&self) -> Segment {
+        let meta = self
+            .inventory
+            .new_segment_meta(SegmentId::generate_random(), 0);
+        Segment::new_volatile(meta, self.schema())
+    }
     /// Return a reference to the index directory.
     pub fn directory(&self) -> &ManagedDirectory {
         &self.directory
@@ -466,7 +471,7 @@ mod tests {
             .try_into()
             .unwrap();
         assert_eq!(reader.searcher().num_docs(), 0);
-        test_index_on_commit_reload_policy_aux(field, &index, &reader);
+        test_index_on_commit_reload_policy_aux(field, index.clone(), &index, &reader);
     }
 
     #[cfg(feature = "mmap")]
@@ -490,7 +495,7 @@ mod tests {
                 .try_into()
                 .unwrap();
             assert_eq!(reader.searcher().num_docs(), 0);
-            test_index_on_commit_reload_policy_aux(field, &index, &reader);
+            test_index_on_commit_reload_policy_aux(field, index.clone(), &index, &reader);
         }
 
         #[test]
@@ -532,12 +537,11 @@ mod tests {
                 .try_into()
                 .unwrap();
             assert_eq!(reader.searcher().num_docs(), 0);
-            test_index_on_commit_reload_policy_aux(field, &write_index, &reader);
+            test_index_on_commit_reload_policy_aux(field, read_index, &write_index, &reader);
         }
     }
 
-    fn test_index_on_commit_reload_policy_aux(field: Field, index: &Index, reader: &IndexReader) {
-        let mut reader_index = reader.index();
+    fn test_index_on_commit_reload_policy_aux(field: Field, mut reader_index: Index, index: &Index, reader: &IndexReader) {
         let (sender, receiver) = crossbeam::channel::unbounded();
         let _watch_handle = reader_index.directory_mut().watch(Box::new(move || {
             let _ = sender.send(());

src/core/index_meta.rs

@@ -35,7 +35,6 @@ impl SegmentMetaInventory {
             segment_id,
             max_doc,
             deletes: None,
-            bundled: false,
         };
         SegmentMeta::from(self.inventory.track(inner))
     }
@@ -82,19 +81,6 @@ impl SegmentMeta {
         self.tracked.segment_id
     }
 
-    pub fn with_bundled(self) -> SegmentMeta {
-        SegmentMeta::from(self.tracked.map(|inner| InnerSegmentMeta {
-            segment_id: inner.segment_id,
-            max_doc: inner.max_doc,
-            deletes: inner.deletes.clone(),
-            bundled: true,
-        }))
-    }
-
-    pub fn is_bundled(&self) -> bool {
-        self.tracked.bundled
-    }
-
     /// Returns the number of deleted documents.
     pub fn num_deleted_docs(&self) -> u32 {
         self.tracked
@@ -121,12 +107,8 @@ impl SegmentMeta {
     /// It just joins the segment id with the extension
     /// associated to a segment component.
     pub fn relative_path(&self, component: SegmentComponent) -> PathBuf {
-        let suffix = self.suffix(component);
-        self.relative_path_from_suffix(&suffix)
-    }
-
-    fn suffix(&self, component: SegmentComponent) -> String {
-        match component {
+        let mut path = self.id().uuid_string();
+        path.push_str(&*match component {
             SegmentComponent::POSTINGS => ".idx".to_string(),
             SegmentComponent::POSITIONS => ".pos".to_string(),
             SegmentComponent::POSITIONSSKIP => ".posidx".to_string(),
@@ -135,17 +117,7 @@ impl SegmentMeta {
             SegmentComponent::FASTFIELDS => ".fast".to_string(),
             SegmentComponent::FIELDNORMS => ".fieldnorm".to_string(),
             SegmentComponent::DELETE => format!(".{}.del", self.delete_opstamp().unwrap_or(0)),
-        }
-    }
-
-    /// Returns the relative path of a component of our segment.
-    ///
-    /// It just joins the segment id with the extension
-    /// associated to a segment component.
-    pub fn relative_path_from_suffix(&self, suffix: &str) -> PathBuf {
-        let mut path = self.id().uuid_string();
-        path.push_str(".");
-        path.push_str(&suffix);
+        });
         PathBuf::from(path)
     }
 
@@ -189,7 +161,6 @@ impl SegmentMeta {
             segment_id: inner_meta.segment_id,
             max_doc,
             deletes: None,
-            bundled: inner_meta.bundled,
         });
         SegmentMeta { tracked }
     }
@@ -204,7 +175,6 @@ impl SegmentMeta {
             segment_id: inner_meta.segment_id,
             max_doc: inner_meta.max_doc,
             deletes: Some(delete_meta),
-            bundled: inner_meta.bundled,
         });
         SegmentMeta { tracked }
     }
@@ -215,7 +185,6 @@ struct InnerSegmentMeta {
     segment_id: SegmentId,
     max_doc: u32,
     deletes: Option<DeleteMeta>,
-    bundled: bool,
 }
 
 impl InnerSegmentMeta {

src/core/segment.rs

@@ -3,19 +3,62 @@ use crate::core::Index;
 use crate::core::SegmentId;
 use crate::core::SegmentMeta;
 use crate::directory::error::{OpenReadError, OpenWriteError};
-use crate::directory::Directory;
-use crate::directory::{ReadOnlyDirectory, ReadOnlySource, WritePtr};
+use crate::directory::{Directory, ManagedDirectory, RAMDirectory};
+use crate::directory::{ReadOnlySource, WritePtr};
 use crate::indexer::segment_serializer::SegmentSerializer;
 use crate::schema::Schema;
 use crate::Opstamp;
+use crate::Result;
+use failure::_core::ops::DerefMut;
 use std::fmt;
+use std::ops::Deref;
 use std::path::PathBuf;
+use std::result;
+
+#[derive(Clone)]
+pub(crate) enum SegmentDirectory {
+    Persisted(ManagedDirectory),
+    Volatile(RAMDirectory),
+}
+
+impl SegmentDirectory {
+    pub fn new_volatile() -> SegmentDirectory {
+        SegmentDirectory::Volatile(RAMDirectory::default())
+    }
+}
+
+impl From<ManagedDirectory> for SegmentDirectory {
+    fn from(directory: ManagedDirectory) -> Self {
+        SegmentDirectory::Persisted(directory)
+    }
+}
+
+impl Deref for SegmentDirectory {
+    type Target = dyn Directory;
+
+    fn deref(&self) -> &Self::Target {
+        match self {
+            SegmentDirectory::Volatile(dir) => dir,
+            SegmentDirectory::Persisted(dir) => dir,
+        }
+    }
+}
+
+impl DerefMut for SegmentDirectory {
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        match self {
+            SegmentDirectory::Volatile(dir) => dir,
+            SegmentDirectory::Persisted(dir) => dir,
+        }
+    }
+}
 
 /// A segment is a piece of the index.
 #[derive(Clone)]
 pub struct Segment {
-    index: Index,
+    schema: Schema,
     meta: SegmentMeta,
+    directory: SegmentDirectory,
 }
 
 impl fmt::Debug for Segment {
@@ -24,23 +67,56 @@ impl fmt::Debug for Segment {
     }
 }
 
-/// Creates a new segment given an `Index` and a `SegmentId`
-///
-/// The function is here to make it private outside `tantivy`.
-/// #[doc(hidden)]
-pub fn create_segment(index: Index, meta: SegmentMeta) -> Segment {
-    Segment { index, meta }
-}
-
 impl Segment {
-    /// Returns the index the segment belongs to.
-    pub fn index(&self) -> &Index {
-        &self.index
+    /// Returns our index's schema.
+    // TODO return a ref.
+    pub fn schema(&self) -> Schema {
+        self.schema.clone()
     }
 
-    /// Returns our index's schema.
-    pub fn schema(&self) -> Schema {
-        self.index.schema()
+    pub(crate) fn new_persisted(
+        meta: SegmentMeta,
+        directory: ManagedDirectory,
+        schema: Schema,
+    ) -> Segment {
+        Segment {
+            meta,
+            schema,
+            directory: SegmentDirectory::from(directory),
+        }
+    }
+
+    /// Creates a new segment that embeds its own `RAMDirectory`.
+    ///
+    /// That segment is entirely dissociated from the index directory.
+    /// It will be persisted by a background thread in charge of IO.
+    pub fn new_volatile(meta: SegmentMeta, schema: Schema) -> Segment {
+        Segment {
+            schema,
+            meta,
+            directory: SegmentDirectory::new_volatile(),
+        }
+    }
+
+    /// Creates a new segment given an `Index` and a `SegmentId`
+    pub(crate) fn for_index(index: Index, meta: SegmentMeta) -> Segment {
+        Segment {
+            directory: SegmentDirectory::Persisted(index.directory().clone()),
+            schema: index.schema(),
+            meta,
+        }
+    }
+
+    pub fn persist(&mut self, mut dest_directory: ManagedDirectory) -> crate::Result<()> {
+        if let SegmentDirectory::Persisted(_) = self.directory {
+            // this segment is already persisted.
+            return Ok(());
+        }
+        if let SegmentDirectory::Volatile(ram_directory) = &self.directory {
+            ram_directory.persist(&mut dest_directory)?;
+        }
+        self.directory = SegmentDirectory::Persisted(dest_directory);
+        Ok(())
     }
 
     /// Returns the segment meta-information
@@ -54,7 +130,8 @@ impl Segment {
     /// as we finalize a fresh new segment.
     pub(crate) fn with_max_doc(self, max_doc: u32) -> Segment {
         Segment {
-            index: self.index,
+            directory: self.directory,
+            schema: self.schema,
             meta: self.meta.with_max_doc(max_doc),
         }
     }
@@ -62,7 +139,8 @@ impl Segment {
     #[doc(hidden)]
     pub fn with_delete_meta(self, num_deleted_docs: u32, opstamp: Opstamp) -> Segment {
         Segment {
-            index: self.index,
+            directory: self.directory,
+            schema: self.schema,
             meta: self.meta.with_delete_meta(num_deleted_docs, opstamp),
         }
     }
@@ -81,30 +159,23 @@ impl Segment {
     }
 
     /// Open one of the component file for a *regular* read.
-    pub fn open_read(&self, component: SegmentComponent) -> Result<ReadOnlySource, OpenReadError> {
+    pub fn open_read(
+        &self,
+        component: SegmentComponent,
+    ) -> result::Result<ReadOnlySource, OpenReadError> {
         let path = self.relative_path(component);
-        let source = self.index.directory().open_read(&path)?;
+        let source = self.directory.open_read(&path)?;
         Ok(source)
     }
 
     /// Open one of the component file for *regular* write.
-    pub fn open_write(&mut self, component: SegmentComponent) -> Result<WritePtr, OpenWriteError> {
-        let path = self.relative_path(component);
-        self.index.directory_mut().open_write(&path)
-    }
-
-    pub fn open_bundle_writer(&mut self) -> Result<WritePtr, OpenWriteError> {
-        let path = self.meta.relative_path_from_suffix("bundle");
-        self.index.directory_mut().open_write(&path)
-    }
-
-    pub(crate) fn open_write_in_directory(
+    pub fn open_write(
         &mut self,
         component: SegmentComponent,
-        directory: &mut dyn Directory,
-    ) -> Result<WritePtr, OpenWriteError> {
+    ) -> result::Result<WritePtr, OpenWriteError> {
         let path = self.relative_path(component);
-        directory.open_write(&path)
+        let write = self.directory.open_write(&path)?;
+        Ok(write)
     }
 }
 
@@ -114,5 +185,5 @@ pub trait SerializableSegment {
     ///
     /// # Returns
     /// The number of documents in the segment.
-    fn write(&self, serializer: SegmentSerializer) -> crate::Result<u32>;
+    fn write(&self, serializer: SegmentSerializer) -> Result<u32>;
 }

src/core/segment_reader.rs

@@ -57,6 +57,68 @@ pub struct SegmentReader {
 }
 
 impl SegmentReader {
+    /// Open a new segment for reading.
+    pub fn open(segment: &Segment) -> Result<SegmentReader> {
+        let termdict_source = segment.open_read(SegmentComponent::TERMS)?;
+        let termdict_composite = CompositeFile::open(&termdict_source)?;
+
+        let store_source = segment.open_read(SegmentComponent::STORE)?;
+
+        fail_point!("SegmentReader::open#middle");
+
+        let postings_source = segment.open_read(SegmentComponent::POSTINGS)?;
+        let postings_composite = CompositeFile::open(&postings_source)?;
+
+        let positions_composite = {
+            if let Ok(source) = segment.open_read(SegmentComponent::POSITIONS) {
+                CompositeFile::open(&source)?
+            } else {
+                CompositeFile::empty()
+            }
+        };
+
+        let positions_idx_composite = {
+            if let Ok(source) = segment.open_read(SegmentComponent::POSITIONSSKIP) {
+                CompositeFile::open(&source)?
+            } else {
+                CompositeFile::empty()
+            }
+        };
+
+        let schema = segment.schema();
+
+        let fast_fields_data = segment.open_read(SegmentComponent::FASTFIELDS)?;
+        let fast_fields_composite = CompositeFile::open(&fast_fields_data)?;
+        let fast_field_readers =
+            Arc::new(FastFieldReaders::load_all(&schema, &fast_fields_composite)?);
+
+        let fieldnorms_data = segment.open_read(SegmentComponent::FIELDNORMS)?;
+        let fieldnorms_composite = CompositeFile::open(&fieldnorms_data)?;
+
+        let delete_bitset_opt = if segment.meta().has_deletes() {
+            let delete_data = segment.open_read(SegmentComponent::DELETE)?;
+            Some(DeleteBitSet::open(delete_data))
+        } else {
+            None
+        };
+
+        Ok(SegmentReader {
+            inv_idx_reader_cache: Arc::new(RwLock::new(HashMap::new())),
+            max_doc: segment.meta().max_doc(),
+            num_docs: segment.meta().num_docs(),
+            termdict_composite,
+            postings_composite,
+            fast_fields_readers: fast_field_readers,
+            fieldnorms_composite,
+            segment_id: segment.id(),
+            store_source,
+            delete_bitset_opt,
+            positions_composite,
+            positions_idx_composite,
+            schema,
+        })
+    }
+
     /// Returns the highest document id ever attributed in
     /// this segment + 1.
     /// Today, `tantivy` does not handle deletes, so it happens
@@ -144,68 +206,6 @@ impl SegmentReader {
         StoreReader::from_source(self.store_source.clone())
     }
 
-    /// Open a new segment for reading.
-    pub fn open(segment: &Segment) -> Result<SegmentReader> {
-        let termdict_source = segment.open_read(SegmentComponent::TERMS)?;
-        let termdict_composite = CompositeFile::open(&termdict_source)?;
-
-        let store_source = segment.open_read(SegmentComponent::STORE)?;
-
-        fail_point!("SegmentReader::open#middle");
-
-        let postings_source = segment.open_read(SegmentComponent::POSTINGS)?;
-        let postings_composite = CompositeFile::open(&postings_source)?;
-
-        let positions_composite = {
-            if let Ok(source) = segment.open_read(SegmentComponent::POSITIONS) {
-                CompositeFile::open(&source)?
-            } else {
-                CompositeFile::empty()
-            }
-        };
-
-        let positions_idx_composite = {
-            if let Ok(source) = segment.open_read(SegmentComponent::POSITIONSSKIP) {
-                CompositeFile::open(&source)?
-            } else {
-                CompositeFile::empty()
-            }
-        };
-
-        let schema = segment.schema();
-
-        let fast_fields_data = segment.open_read(SegmentComponent::FASTFIELDS)?;
-        let fast_fields_composite = CompositeFile::open(&fast_fields_data)?;
-        let fast_field_readers =
-            Arc::new(FastFieldReaders::load_all(&schema, &fast_fields_composite)?);
-
-        let fieldnorms_data = segment.open_read(SegmentComponent::FIELDNORMS)?;
-        let fieldnorms_composite = CompositeFile::open(&fieldnorms_data)?;
-
-        let delete_bitset_opt = if segment.meta().has_deletes() {
-            let delete_data = segment.open_read(SegmentComponent::DELETE)?;
-            Some(DeleteBitSet::open(delete_data))
-        } else {
-            None
-        };
-
-        Ok(SegmentReader {
-            inv_idx_reader_cache: Arc::new(RwLock::new(HashMap::new())),
-            max_doc: segment.meta().max_doc(),
-            num_docs: segment.meta().num_docs(),
-            termdict_composite,
-            postings_composite,
-            fast_fields_readers: fast_field_readers,
-            fieldnorms_composite,
-            segment_id: segment.id(),
-            store_source,
-            delete_bitset_opt,
-            positions_composite,
-            positions_idx_composite,
-            schema,
-        })
-    }
-
     /// Returns a field reader associated to the field given in argument.
     /// If the field was not present in the index during indexing time,
     /// the InvertedIndexReader is empty.

src/directory/bundle_directory.rs

@@ -1,97 +0,0 @@
-use crate::directory::directory::ReadOnlyDirectory;
-use crate::directory::error::OpenReadError;
-use crate::directory::ReadOnlySource;
-use crate::error::DataCorruption;
-use std::collections::HashMap;
-use std::path::{Path, PathBuf};
-use std::sync::Arc;
-
-#[derive(Clone)]
-struct BundleDirectory {
-    source_map: Arc<HashMap<PathBuf, ReadOnlySource>>,
-}
-
-impl BundleDirectory {
-    pub fn from_source(source: ReadOnlySource) -> Result<BundleDirectory, DataCorruption> {
-        let mut index_offset_buf = [0u8; 8];
-        let (body_idx, footer_offset) = source.split_from_end(8);
-        index_offset_buf.copy_from_slice(footer_offset.as_slice());
-        let offset = u64::from_le_bytes(index_offset_buf);
-        let (body_source, idx_source) = body_idx.split(offset as usize);
-        let idx: HashMap<PathBuf, (u64, u64)> = serde_json::from_slice(idx_source.as_slice())
-            .map_err(|err| {
-                let msg = format!("Failed to read index from bundle. {:?}", err);
-                DataCorruption::comment_only(msg)
-            })?;
-        let source_map: HashMap<PathBuf, ReadOnlySource> = idx
-            .into_iter()
-            .map(|(path, (start, stop))| {
-                let source = body_source.slice(start as usize, stop as usize);
-                (path, source)
-            })
-            .collect();
-        Ok(BundleDirectory {
-            source_map: Arc::new(source_map),
-        })
-    }
-}
-
-impl ReadOnlyDirectory for BundleDirectory {
-    fn open_read(&self, path: &Path) -> Result<ReadOnlySource, OpenReadError> {
-        self.source_map
-            .get(path)
-            .cloned()
-            .ok_or_else(|| OpenReadError::FileDoesNotExist(path.to_path_buf()))
-    }
-
-    fn exists(&self, path: &Path) -> bool {
-        self.source_map.contains_key(path)
-    }
-
-    fn atomic_read(&self, path: &Path) -> Result<Vec<u8>, OpenReadError> {
-        let source = self
-            .source_map
-            .get(path)
-            .ok_or_else(|| OpenReadError::FileDoesNotExist(path.to_path_buf()))?;
-        Ok(source.as_slice().to_vec())
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::BundleDirectory;
-    use crate::directory::{RAMDirectory, ReadOnlyDirectory, TerminatingWrite};
-    use crate::Directory;
-    use std::io::Write;
-    use std::path::Path;
-
-    #[test]
-    fn test_bundle_directory() {
-        let mut ram_directory = RAMDirectory::default();
-        let test_path_atomic = Path::new("testpath_atomic");
-        let test_path_wrt = Path::new("testpath_wrt");
-        assert!(ram_directory
-            .atomic_write(test_path_atomic, b"titi")
-            .is_ok());
-        {
-            let mut test_wrt = ram_directory.open_write(test_path_wrt).unwrap();
-            assert!(test_wrt.write_all(b"toto").is_ok());
-            assert!(test_wrt.terminate().is_ok());
-        }
-        let mut dest_directory = RAMDirectory::default();
-        let bundle_path = Path::new("bundle");
-        let mut wrt = dest_directory.open_write(bundle_path).unwrap();
-        assert!(ram_directory.serialize_bundle(&mut wrt).is_ok());
-        assert!(wrt.terminate().is_ok());
-        let source = dest_directory.open_read(bundle_path).unwrap();
-        let bundle_directory = BundleDirectory::from_source(source).unwrap();
-        assert_eq!(
-            &bundle_directory.atomic_read(test_path_atomic).unwrap()[..],
-            b"titi"
-        );
-        assert_eq!(
-            &bundle_directory.open_read(test_path_wrt).unwrap()[..],
-            b"toto"
-        );
-    }
-}

src/directory/directory.rs

@@ -100,7 +100,17 @@ fn retry_policy(is_blocking: bool) -> RetryPolicy {
     }
 }
 
-pub trait ReadOnlyDirectory {
+/// Write-once read many (WORM) abstraction for where
+/// tantivy's data should be stored.
+///
+/// There are currently two implementations of `Directory`
+///
+/// - The [`MMapDirectory`](struct.MmapDirectory.html), this
+/// should be your default choice.
+/// - The [`RAMDirectory`](struct.RAMDirectory.html), which
+/// should be used mostly for tests.
+///
+pub trait Directory: DirectoryClone + fmt::Debug + Send + Sync + 'static {
     /// Opens a virtual file for read.
     ///
     /// Once a virtual file is open, its data may not
@@ -112,31 +122,6 @@ pub trait ReadOnlyDirectory {
     /// You should only use this to read files create with [Directory::open_write].
     fn open_read(&self, path: &Path) -> result::Result<ReadOnlySource, OpenReadError>;
 
-    /// Returns true iff the file exists
-    fn exists(&self, path: &Path) -> bool;
-
-    /// Reads the full content file that has been written using
-    /// atomic_write.
-    ///
-    /// This should only be used for small files.
-    ///
-    /// You should only use this to read files create with [Directory::atomic_write].
-    fn atomic_read(&self, path: &Path) -> Result<Vec<u8>, OpenReadError>;
-}
-
-/// Write-once read many (WORM) abstraction for where
-/// tantivy's data should be stored.
-///
-/// There are currently two implementations of `Directory`
-///
-/// - The [`MMapDirectory`](struct.MmapDirectory.html), this
-/// should be your default choice.
-/// - The [`RAMDirectory`](struct.RAMDirectory.html), which
-/// should be used mostly for tests.
-///
-pub trait Directory:
-    DirectoryClone + ReadOnlyDirectory + fmt::Debug + Send + Sync + 'static
-{
     /// Removes a file
     ///
     /// Removing a file will not affect an eventual
@@ -146,6 +131,9 @@ pub trait Directory:
     /// `DeleteError::DoesNotExist`.
     fn delete(&self, path: &Path) -> result::Result<(), DeleteError>;
 
+    /// Returns true iff the file exists
+    fn exists(&self, path: &Path) -> bool;
+
     /// Opens a writer for the *virtual file* associated with
     /// a Path.
     ///
@@ -167,6 +155,14 @@ pub trait Directory:
     /// The file may not previously exist.
     fn open_write(&mut self, path: &Path) -> Result<WritePtr, OpenWriteError>;
 
+    /// Reads the full content file that has been written using
+    /// atomic_write.
+    ///
+    /// This should only be used for small files.
+    ///
+    /// You should only use this to read files create with [Directory::atomic_write].
+    fn atomic_read(&self, path: &Path) -> Result<Vec<u8>, OpenReadError>;
+
     /// Atomically replace the content of a file with data.
     ///
     /// This calls ensure that reads can never *observe*

src/directory/managed_directory.rs

@@ -10,7 +10,6 @@ use crate::directory::{WatchCallback, WatchHandle};
 use crate::error::DataCorruption;
 use crate::Directory;
 
-use crate::directory::directory::ReadOnlyDirectory;
 use crc32fast::Hasher;
 use serde_json;
 use std::collections::HashSet;
@@ -265,6 +264,14 @@ impl ManagedDirectory {
 }
 
 impl Directory for ManagedDirectory {
+    fn open_read(&self, path: &Path) -> result::Result<ReadOnlySource, OpenReadError> {
+        let read_only_source = self.directory.open_read(path)?;
+        let (footer, reader) = Footer::extract_footer(read_only_source)
+            .map_err(|err| IOError::with_path(path.to_path_buf(), err))?;
+        footer.is_compatible()?;
+        Ok(reader)
+    }
+
     fn open_write(&mut self, path: &Path) -> result::Result<WritePtr, OpenWriteError> {
         self.register_file_as_managed(path)
             .map_err(|e| IOError::with_path(path.to_owned(), e))?;
@@ -282,10 +289,18 @@ impl Directory for ManagedDirectory {
         self.directory.atomic_write(path, data)
     }
 
+    fn atomic_read(&self, path: &Path) -> result::Result<Vec<u8>, OpenReadError> {
+        self.directory.atomic_read(path)
+    }
+
     fn delete(&self, path: &Path) -> result::Result<(), DeleteError> {
         self.directory.delete(path)
     }
 
+    fn exists(&self, path: &Path) -> bool {
+        self.directory.exists(path)
+    }
+
     fn acquire_lock(&self, lock: &Lock) -> result::Result<DirectoryLock, LockError> {
         self.directory.acquire_lock(lock)
     }
@@ -295,24 +310,6 @@ impl Directory for ManagedDirectory {
     }
 }
 
-impl ReadOnlyDirectory for ManagedDirectory {
-    fn open_read(&self, path: &Path) -> result::Result<ReadOnlySource, OpenReadError> {
-        let read_only_source = self.directory.open_read(path)?;
-        let (footer, reader) = Footer::extract_footer(read_only_source)
-            .map_err(|err| IOError::with_path(path.to_path_buf(), err))?;
-        footer.is_compatible()?;
-        Ok(reader)
-    }
-
-    fn exists(&self, path: &Path) -> bool {
-        self.directory.exists(path)
-    }
-
-    fn atomic_read(&self, path: &Path) -> result::Result<Vec<u8>, OpenReadError> {
-        self.directory.atomic_read(path)
-    }
-}
-
 impl Clone for ManagedDirectory {
     fn clone(&self) -> ManagedDirectory {
         ManagedDirectory {
@@ -326,9 +323,7 @@ impl Clone for ManagedDirectory {
 #[cfg(test)]
 mod tests_mmap_specific {
 
-    use crate::directory::{
-        Directory, ManagedDirectory, MmapDirectory, ReadOnlyDirectory, TerminatingWrite,
-    };
+    use crate::directory::{Directory, ManagedDirectory, MmapDirectory, TerminatingWrite};
     use std::collections::HashSet;
     use std::fs::OpenOptions;
     use std::io::Write;

src/directory/mmap_directory.rs

@@ -6,7 +6,6 @@ use self::notify::RawEvent;
 use self::notify::RecursiveMode;
 use self::notify::Watcher;
 use crate::core::META_FILEPATH;
-use crate::directory::directory::ReadOnlyDirectory;
 use crate::directory::error::LockError;
 use crate::directory::error::{
     DeleteError, IOError, OpenDirectoryError, OpenReadError, OpenWriteError,
@@ -408,6 +407,24 @@ impl TerminatingWrite for SafeFileWriter {
 }
 
 impl Directory for MmapDirectory {
+    fn open_read(&self, path: &Path) -> result::Result<ReadOnlySource, OpenReadError> {
+        debug!("Open Read {:?}", path);
+        let full_path = self.resolve_path(path);
+
+        let mut mmap_cache = self.inner.mmap_cache.write().map_err(|_| {
+            let msg = format!(
+                "Failed to acquired write lock \
+                 on mmap cache while reading {:?}",
+                path
+            );
+            IOError::with_path(path.to_owned(), make_io_err(msg))
+        })?;
+        Ok(mmap_cache
+            .get_mmap(&full_path)?
+            .map(ReadOnlySource::from)
+            .unwrap_or_else(ReadOnlySource::empty))
+    }
+
     /// Any entry associated to the path in the mmap will be
     /// removed before the file is deleted.
     fn delete(&self, path: &Path) -> result::Result<(), DeleteError> {
@@ -426,6 +443,11 @@ impl Directory for MmapDirectory {
         }
     }
 
+    fn exists(&self, path: &Path) -> bool {
+        let full_path = self.resolve_path(path);
+        full_path.exists()
+    }
+
     fn open_write(&mut self, path: &Path) -> Result<WritePtr, OpenWriteError> {
         debug!("Open Write {:?}", path);
         let full_path = self.resolve_path(path);
@@ -456,6 +478,25 @@ impl Directory for MmapDirectory {
         Ok(BufWriter::new(Box::new(writer)))
     }
 
+    fn atomic_read(&self, path: &Path) -> Result<Vec<u8>, OpenReadError> {
+        let full_path = self.resolve_path(path);
+        let mut buffer = Vec::new();
+        match File::open(&full_path) {
+            Ok(mut file) => {
+                file.read_to_end(&mut buffer)
+                    .map_err(|e| IOError::with_path(path.to_owned(), e))?;
+                Ok(buffer)
+            }
+            Err(e) => {
+                if e.kind() == io::ErrorKind::NotFound {
+                    Err(OpenReadError::FileDoesNotExist(path.to_owned()))
+                } else {
+                    Err(IOError::with_path(path.to_owned(), e).into())
+                }
+            }
+        }
+    }
+
     fn atomic_write(&mut self, path: &Path, data: &[u8]) -> io::Result<()> {
         debug!("Atomic Write {:?}", path);
         let full_path = self.resolve_path(path);
@@ -489,50 +530,6 @@ impl Directory for MmapDirectory {
     }
 }
 
-impl ReadOnlyDirectory for MmapDirectory {
-    fn open_read(&self, path: &Path) -> result::Result<ReadOnlySource, OpenReadError> {
-        debug!("Open Read {:?}", path);
-        let full_path = self.resolve_path(path);
-
-        let mut mmap_cache = self.inner.mmap_cache.write().map_err(|_| {
-            let msg = format!(
-                "Failed to acquired write lock \
-                 on mmap cache while reading {:?}",
-                path
-            );
-            IOError::with_path(path.to_owned(), make_io_err(msg))
-        })?;
-        Ok(mmap_cache
-            .get_mmap(&full_path)?
-            .map(ReadOnlySource::from)
-            .unwrap_or_else(ReadOnlySource::empty))
-    }
-
-    fn exists(&self, path: &Path) -> bool {
-        let full_path = self.resolve_path(path);
-        full_path.exists()
-    }
-
-    fn atomic_read(&self, path: &Path) -> Result<Vec<u8>, OpenReadError> {
-        let full_path = self.resolve_path(path);
-        let mut buffer = Vec::new();
-        match File::open(&full_path) {
-            Ok(mut file) => {
-                file.read_to_end(&mut buffer)
-                    .map_err(|e| IOError::with_path(path.to_owned(), e))?;
-                Ok(buffer)
-            }
-            Err(e) => {
-                if e.kind() == io::ErrorKind::NotFound {
-                    Err(OpenReadError::FileDoesNotExist(path.to_owned()))
-                } else {
-                    Err(IOError::with_path(path.to_owned(), e).into())
-                }
-            }
-        }
-    }
-}
-
 #[cfg(test)]
 mod tests {
 

src/directory/mod.rs

@@ -7,7 +7,6 @@ WORM directory abstraction.
 #[cfg(feature = "mmap")]
 mod mmap_directory;
 
-mod bundle_directory;
 mod directory;
 mod directory_lock;
 mod footer;
@@ -20,7 +19,7 @@ mod watch_event_router;
 pub mod error;
 
 pub use self::directory::DirectoryLock;
-pub use self::directory::{Directory, DirectoryClone, ReadOnlyDirectory};
+pub use self::directory::{Directory, DirectoryClone};
 pub use self::directory_lock::{Lock, INDEX_WRITER_LOCK, META_LOCK};
 pub use self::ram_directory::RAMDirectory;
 pub use self::read_only_source::ReadOnlySource;

src/directory/ram_directory.rs

@@ -1,6 +1,4 @@
-use crate::common::CountingWriter;
 use crate::core::META_FILEPATH;
-use crate::directory::directory::ReadOnlyDirectory;
 use crate::directory::error::{DeleteError, OpenReadError, OpenWriteError};
 use crate::directory::AntiCallToken;
 use crate::directory::WatchCallbackList;
@@ -117,22 +115,6 @@ impl InnerDirectory {
     fn total_mem_usage(&self) -> usize {
         self.fs.values().map(|f| f.len()).sum()
     }
-
-    fn serialize_bundle(&self, wrt: &mut WritePtr) -> io::Result<()> {
-        let mut counting_writer = CountingWriter::wrap(wrt);
-        let mut file_index: HashMap<PathBuf, (u64, u64)> = HashMap::default();
-        for (path, source) in &self.fs {
-            let start = counting_writer.written_bytes();
-            counting_writer.write_all(source.as_slice())?;
-            let stop = counting_writer.written_bytes();
-            file_index.insert(path.to_path_buf(), (start, stop));
-        }
-        let index_offset = counting_writer.written_bytes();
-        serde_json::to_writer(&mut counting_writer, &file_index)?;
-        let index_offset_buffer = index_offset.to_le_bytes();
-        counting_writer.write_all(&index_offset_buffer[..])?;
-        Ok(())
-    }
 }
 
 impl fmt::Debug for RAMDirectory {
@@ -163,17 +145,22 @@ impl RAMDirectory {
         self.fs.read().unwrap().total_mem_usage()
     }
 
-    /// Serialize the RAMDirectory into a bundle.
-    ///
-    /// This method will fail, write nothing, and return an error if a
-    /// clone of this repository exists.
-    pub fn serialize_bundle(self, wrt: &mut WritePtr) -> io::Result<()> {
-        let inner_directory_rlock = self.fs.read().unwrap();
-        inner_directory_rlock.serialize_bundle(wrt)
+    pub fn persist(&self, dest: &mut dyn Directory) -> crate::Result<()> {
+        let wlock = self.fs.write().unwrap();
+        for (path, source) in wlock.fs.iter() {
+            let mut dest_wrt = dest.open_write(path)?;
+            dest_wrt.write_all(source.as_slice())?;
+            dest_wrt.terminate()?;
+        }
+        Ok(())
     }
 }
 
 impl Directory for RAMDirectory {
+    fn open_read(&self, path: &Path) -> result::Result<ReadOnlySource, OpenReadError> {
+        self.fs.read().unwrap().open_read(path)
+    }
+
     fn delete(&self, path: &Path) -> result::Result<(), DeleteError> {
         fail_point!("RAMDirectory::delete", |_| {
             use crate::directory::error::IOError;
@@ -183,6 +170,10 @@ impl Directory for RAMDirectory {
         self.fs.write().unwrap().delete(path)
     }
 
+    fn exists(&self, path: &Path) -> bool {
+        self.fs.read().unwrap().exists(path)
+    }
+
     fn open_write(&mut self, path: &Path) -> Result<WritePtr, OpenWriteError> {
         let mut fs = self.fs.write().unwrap();
         let path_buf = PathBuf::from(path);
@@ -196,6 +187,10 @@ impl Directory for RAMDirectory {
         }
     }
 
+    fn atomic_read(&self, path: &Path) -> Result<Vec<u8>, OpenReadError> {
+        Ok(self.open_read(path)?.as_slice().to_owned())
+    }
+
     fn atomic_write(&mut self, path: &Path, data: &[u8]) -> io::Result<()> {
         fail_point!("RAMDirectory::atomic_write", |msg| Err(io::Error::new(
             io::ErrorKind::Other,
@@ -219,17 +214,3 @@ impl Directory for RAMDirectory {
         Ok(self.fs.write().unwrap().watch(watch_callback))
     }
 }
-
-impl ReadOnlyDirectory for RAMDirectory {
-    fn open_read(&self, path: &Path) -> result::Result<ReadOnlySource, OpenReadError> {
-        self.fs.read().unwrap().open_read(path)
-    }
-
-    fn exists(&self, path: &Path) -> bool {
-        self.fs.read().unwrap().exists(path)
-    }
-
-    fn atomic_read(&self, path: &Path) -> Result<Vec<u8>, OpenReadError> {
-        Ok(self.open_read(path)?.as_slice().to_owned())
-    }
-}

src/error.rs

@@ -25,10 +25,10 @@ impl DataCorruption {
         }
     }
 
-    pub fn comment_only<TS: ToString>(comment: TS) -> DataCorruption {
+    pub fn comment_only(comment: String) -> DataCorruption {
         DataCorruption {
             filepath: None,
-            comment: comment.to_string(),
+            comment,
         }
     }
 }

src/fastfield/mod.rs

@@ -179,7 +179,7 @@ mod tests {
 
     use super::*;
     use crate::common::CompositeFile;
-    use crate::directory::{Directory, RAMDirectory, ReadOnlyDirectory, WritePtr};
+    use crate::directory::{Directory, RAMDirectory, WritePtr};
     use crate::fastfield::FastFieldReader;
     use crate::merge_policy::NoMergePolicy;
     use crate::schema::Field;

src/fastfield/reader.rs

@@ -4,7 +4,7 @@ use crate::common::compute_num_bits;
 use crate::common::BinarySerializable;
 use crate::common::CompositeFile;
 use crate::directory::ReadOnlySource;
-use crate::directory::{Directory, RAMDirectory, ReadOnlyDirectory, WritePtr};
+use crate::directory::{Directory, RAMDirectory, WritePtr};
 use crate::fastfield::{FastFieldSerializer, FastFieldsWriter};
 use crate::schema::Schema;
 use crate::schema::FAST;

src/indexer/index_writer.rs

@@ -8,30 +8,33 @@ use crate::core::SegmentComponent;
 use crate::core::SegmentId;
 use crate::core::SegmentMeta;
 use crate::core::SegmentReader;
-use crate::directory::TerminatingWrite;
 use crate::directory::{DirectoryLock, GarbageCollectionResult};
+use crate::directory::{TerminatingWrite, WatchCallbackList};
 use crate::docset::DocSet;
 use crate::error::TantivyError;
 use crate::fastfield::write_delete_bitset;
 use crate::indexer::delete_queue::{DeleteCursor, DeleteQueue};
 use crate::indexer::doc_opstamp_mapping::DocToOpstampMapping;
 use crate::indexer::operation::DeleteOperation;
+use crate::indexer::segment_manager::SegmentRegisters;
+use crate::indexer::segment_register::SegmentRegister;
 use crate::indexer::stamper::Stamper;
 use crate::indexer::MergePolicy;
 use crate::indexer::SegmentEntry;
 use crate::indexer::SegmentWriter;
+use crate::reader::NRTReader;
 use crate::schema::Document;
 use crate::schema::IndexRecordOption;
 use crate::schema::Term;
-use crate::Opstamp;
+use crate::tokenizer::TokenizerManager;
+use crate::{IndexReader, Opstamp};
 use crossbeam::channel;
 use futures::executor::block_on;
 use futures::future::Future;
-use smallvec::smallvec;
-use smallvec::SmallVec;
+use smallvec::{smallvec, SmallVec};
 use std::mem;
 use std::ops::Range;
-use std::sync::Arc;
+use std::sync::{Arc, RwLock};
 use std::thread;
 use std::thread::JoinHandle;
 
@@ -68,6 +71,8 @@ pub struct IndexWriter {
     // lifetime of the lock with that of the IndexWriter.
     _directory_lock: Option<DirectoryLock>,
 
+    segment_registers: Arc<RwLock<SegmentRegisters>>,
+
     index: Index,
 
     heap_size_in_bytes_per_thread: usize,
@@ -87,6 +92,8 @@ pub struct IndexWriter {
 
     stamper: Stamper,
     committed_opstamp: Opstamp,
+
+    on_commit: WatchCallbackList,
 }
 
 fn compute_deleted_bitset(
@@ -133,7 +140,6 @@ fn compute_deleted_bitset(
 /// For instance, there was no delete operation between the state of the `segment_entry` and
 /// the `target_opstamp`, `segment_entry` is not updated.
 pub(crate) fn advance_deletes(
-    mut segment: Segment,
     segment_entry: &mut SegmentEntry,
     target_opstamp: Opstamp,
 ) -> crate::Result<()> {
@@ -142,28 +148,38 @@ pub(crate) fn advance_deletes(
         return Ok(());
     }
 
-    if segment_entry.delete_bitset().is_none() && segment_entry.delete_cursor().get().is_none() {
+    let delete_bitset_opt = segment_entry.take_delete_bitset();
+
+    // We avoid directly advancing the `SegmentEntry` delete cursor, because
+    // we do not want to end up in an invalid state if the delete bitset
+    // serialization fails.
+    let mut delete_cursor = segment_entry.delete_cursor();
+
+    if delete_bitset_opt.is_none() && delete_cursor.get().is_none() {
         // There has been no `DeleteOperation` between the segment status and `target_opstamp`.
         return Ok(());
     }
 
+    // We open our current serialized segment to compute the new deleted bitset.
+    let segment = segment_entry.segment().clone();
     let segment_reader = SegmentReader::open(&segment)?;
 
     let max_doc = segment_reader.max_doc();
-    let mut delete_bitset: BitSet = match segment_entry.delete_bitset() {
-        Some(previous_delete_bitset) => (*previous_delete_bitset).clone(),
-        None => BitSet::with_max_value(max_doc),
-    };
+    let mut delete_bitset: BitSet =
+        delete_bitset_opt.unwrap_or_else(|| BitSet::with_max_value(max_doc));
+
+    let num_deleted_docs_before = segment.meta().num_deleted_docs();
 
     compute_deleted_bitset(
         &mut delete_bitset,
         &segment_reader,
-        segment_entry.delete_cursor(),
+        &mut delete_cursor,
         &DocToOpstampMapping::None,
         target_opstamp,
     )?;
 
-    // TODO optimize
+    // TODO optimize... We are simply manipulating bitsets here.
+    // We should be able to compute the union much faster.
     if let Some(seg_delete_bitset) = segment_reader.delete_bitset() {
         for doc in 0u32..max_doc {
             if seg_delete_bitset.is_deleted(doc) {
@@ -172,15 +188,23 @@ pub(crate) fn advance_deletes(
         }
     }
 
-    let num_deleted_docs = delete_bitset.len();
-    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)?;
+    let num_deleted_docs = delete_bitset.len() as u32;
+
+    if num_deleted_docs > num_deleted_docs_before {
+        // We need to write a new delete file.
+        let mut delete_file = segment
+            .with_delete_meta(num_deleted_docs, target_opstamp)
+            .open_write(SegmentComponent::DELETE)?;
         write_delete_bitset(&delete_bitset, max_doc, &mut delete_file)?;
         delete_file.terminate()?;
+        segment_entry.reset_delete_meta(num_deleted_docs as u32, target_opstamp);
     }
 
-    segment_entry.set_meta(segment.meta().clone());
+    // Regardless of whether we did end up having to write a new file or not
+    // we advance the `delete_cursor`. This is an optimisation. We want to ensure we do not
+    // check that a given deleted term does not match any of our docs more than once.
+    segment_entry.set_delete_cursor(delete_cursor);
+
     Ok(())
 }
 
@@ -189,11 +213,12 @@ fn index_documents(
     segment: Segment,
     grouped_document_iterator: &mut dyn Iterator<Item = OperationGroup>,
     segment_updater: &mut SegmentUpdater,
+    tokenizers: &TokenizerManager,
     mut delete_cursor: DeleteCursor,
 ) -> crate::Result<bool> {
+    let mut segment_writer =
+        SegmentWriter::for_segment(memory_budget, segment.clone(), tokenizers)?;
     let schema = segment.schema();
-
-    let mut segment_writer = SegmentWriter::for_segment(memory_budget, segment.clone(), &schema)?;
     for document_group in grouped_document_iterator {
         for doc in document_group {
             segment_writer.add_document(doc, &schema)?;
@@ -231,11 +256,7 @@ fn index_documents(
         last_docstamp,
     )?;
 
-    let segment_entry = SegmentEntry::new(
-        segment_with_max_doc.meta().clone(),
-        delete_cursor,
-        delete_bitset_opt,
-    );
+    let segment_entry = SegmentEntry::new(segment_with_max_doc, delete_cursor, delete_bitset_opt);
     block_on(segment_updater.schedule_add_segment(segment_entry))?;
     Ok(true)
 }
@@ -307,16 +328,24 @@ impl IndexWriter {
 
         let delete_queue = DeleteQueue::new();
 
-        let current_opstamp = index.load_metas()?.opstamp;
+        let meta = index.load_metas()?;
 
-        let stamper = Stamper::new(current_opstamp);
+        let stamper = Stamper::new(meta.opstamp);
+
+        let commited_segments = SegmentRegister::new(
+            index.directory(),
+            &index.schema(),
+            meta.segments,
+            &delete_queue.cursor(),
+        );
+        let segment_registers = Arc::new(RwLock::new(SegmentRegisters::new(commited_segments)));
 
         let segment_updater =
-            SegmentUpdater::create(index.clone(), stamper.clone(), &delete_queue.cursor())?;
+            SegmentUpdater::create(segment_registers.clone(), index.clone(), stamper.clone())?;
 
         let mut index_writer = IndexWriter {
             _directory_lock: Some(directory_lock),
-
+            segment_registers,
             heap_size_in_bytes_per_thread,
             index: index.clone(),
 
@@ -330,10 +359,12 @@ impl IndexWriter {
 
             delete_queue,
 
-            committed_opstamp: current_opstamp,
+            committed_opstamp: meta.opstamp,
             stamper,
 
             worker_id: 0,
+
+            on_commit: Default::default(),
         };
         index_writer.start_workers()?;
         Ok(index_writer)
@@ -373,13 +404,6 @@ impl IndexWriter {
         result
     }
 
-    #[doc(hidden)]
-    pub fn add_segment(&self, segment_meta: SegmentMeta) -> crate::Result<()> {
-        let delete_cursor = self.delete_queue.cursor();
-        let segment_entry = SegmentEntry::new(segment_meta, delete_cursor, None);
-        block_on(self.segment_updater.schedule_add_segment(segment_entry))
-    }
-
     /// Creates a new segment.
     ///
     /// This method is useful only for users trying to do complex
@@ -428,12 +452,13 @@ impl IndexWriter {
                         // was dropped.
                         return Ok(());
                     }
-                    let segment = index.new_segment();
+                    let segment = index.new_segment_unpersisted();
                     index_documents(
                         mem_budget,
                         segment,
                         &mut document_iterator,
                         &mut segment_updater,
+                        index.tokenizers(),
                         delete_cursor.clone(),
                     )?;
                 }
@@ -460,6 +485,21 @@ impl IndexWriter {
         Ok(())
     }
 
+    // TODO move me
+    pub(crate) fn trigger_commit(&self) -> impl Future<Output = ()> {
+        self.on_commit.broadcast()
+    }
+
+    pub fn reader(&self, num_searchers: usize) -> crate::Result<IndexReader> {
+        let nrt_reader = NRTReader::create(
+            num_searchers,
+            self.index.clone(),
+            self.segment_registers.clone(),
+            &self.on_commit,
+        )?;
+        Ok(IndexReader::NRT(nrt_reader))
+    }
+
     /// Detects and removes the files that are not used by the index anymore.
     pub fn garbage_collect_files(
         &self,
@@ -603,7 +643,7 @@ impl IndexWriter {
     /// It is also possible to add a payload to the `commit`
     /// using this API.
     /// See [`PreparedCommit::set_payload()`](PreparedCommit.html)
-    pub fn prepare_commit(&mut self) -> crate::Result<PreparedCommit> {
+    pub fn prepare_commit(&mut self, soft_commit: bool) -> crate::Result<PreparedCommit> {
         // Here, because we join all of the worker threads,
         // all of the segment update for this commit have been
         // sent.
@@ -631,7 +671,7 @@ impl IndexWriter {
         }
 
         let commit_opstamp = self.stamper.stamp();
-        let prepared_commit = PreparedCommit::new(self, commit_opstamp);
+        let prepared_commit = PreparedCommit::new(self, commit_opstamp, soft_commit);
         info!("Prepared commit {}", commit_opstamp);
         Ok(prepared_commit)
     }
@@ -651,7 +691,11 @@ impl IndexWriter {
     /// that made it in the commit.
     ///
     pub fn commit(&mut self) -> crate::Result<Opstamp> {
-        self.prepare_commit()?.commit()
+        self.prepare_commit(false)?.commit()
+    }
+
+    pub fn soft_commit(&mut self) -> crate::Result<Opstamp> {
+        self.prepare_commit(true)?.commit()
     }
 
     pub(crate) fn segment_updater(&self) -> &SegmentUpdater {
@@ -776,7 +820,6 @@ impl Drop for IndexWriter {
 
 #[cfg(test)]
 mod tests {
-
     use super::super::operation::UserOperation;
     use crate::collector::TopDocs;
     use crate::directory::error::LockError;
@@ -1009,7 +1052,8 @@ mod tests {
                 index_writer.add_document(doc!(text_field => "a"));
             }
             {
-                let mut prepared_commit = index_writer.prepare_commit().expect("commit failed");
+                let mut prepared_commit =
+                    index_writer.prepare_commit(false).expect("commit failed");
                 prepared_commit.set_payload("first commit");
                 prepared_commit.commit().expect("commit failed");
             }
@@ -1042,7 +1086,8 @@ mod tests {
                 index_writer.add_document(doc!(text_field => "a"));
             }
             {
-                let mut prepared_commit = index_writer.prepare_commit().expect("commit failed");
+                let mut prepared_commit =
+                    index_writer.prepare_commit(false).expect("commit failed");
                 prepared_commit.set_payload("first commit");
                 prepared_commit.abort().expect("commit failed");
             }
@@ -1217,7 +1262,43 @@ mod tests {
         let index = Index::create_in_ram(schema_builder.build());
         let mut index_writer = index.writer_with_num_threads(1, 3_000_000).unwrap();
         index_writer.add_document(doc!(idfield=>"myid"));
-        let commit = index_writer.commit();
-        assert!(commit.is_ok());
+        assert!(index_writer.commit().is_ok());
+    }
+
+    #[test]
+    fn test_index_writer_reader() {
+        let mut schema_builder = schema::Schema::builder();
+        let idfield = schema_builder.add_text_field("id", STRING);
+        schema_builder.add_text_field("optfield", STRING);
+        let index = Index::create_in_ram(schema_builder.build());
+        let mut index_writer = index.writer_with_num_threads(1, 3_000_000).unwrap();
+        index_writer.add_document(doc!(idfield=>"myid"));
+        assert!(index_writer.commit().is_ok());
+        let reader = index_writer.reader(2).unwrap();
+        let searcher = reader.searcher();
+        assert_eq!(searcher.num_docs(), 1u64);
+        index_writer.add_document(doc!(idfield=>"myid"));
+        assert!(index_writer.commit().is_ok());
+        assert_eq!(reader.searcher().num_docs(), 2u64);
+        assert_eq!(searcher.num_docs(), 1u64);
+    }
+
+    #[test]
+    fn test_index_writer_reader_soft_commit() {
+        let mut schema_builder = schema::Schema::builder();
+        let idfield = schema_builder.add_text_field("id", STRING);
+        schema_builder.add_text_field("optfield", STRING);
+        let index = Index::create_in_ram(schema_builder.build());
+        let mut index_writer = index.writer_with_num_threads(1, 3_000_000).unwrap();
+        index_writer.add_document(doc!(idfield=>"myid"));
+        assert!(index_writer.soft_commit().is_ok());
+        let nrt_reader = index_writer.reader(2).unwrap();
+        let normal_reader = index.reader_builder().try_into().unwrap();
+        assert_eq!(nrt_reader.searcher().num_docs(), 1u64);
+        assert_eq!(normal_reader.searcher().num_docs(), 0u64);
+        assert!(index_writer.commit().is_ok());
+        assert!(normal_reader.reload().is_ok());
+        assert_eq!(nrt_reader.searcher().num_docs(), 1u64);
+        assert_eq!(normal_reader.searcher().num_docs(), 1u64);
     }
 }

src/indexer/mod.rs

@@ -23,6 +23,7 @@ pub use self::merge_policy::{MergeCandidate, MergePolicy, NoMergePolicy};
 pub use self::prepared_commit::PreparedCommit;
 pub use self::segment_entry::SegmentEntry;
 pub use self::segment_manager::SegmentManager;
+pub(crate) use self::segment_manager::SegmentRegisters;
 pub use self::segment_serializer::SegmentSerializer;
 pub use self::segment_writer::SegmentWriter;
 

src/indexer/prepared_commit.rs

@@ -8,14 +8,20 @@ pub struct PreparedCommit<'a> {
     index_writer: &'a mut IndexWriter,
     payload: Option<String>,
     opstamp: Opstamp,
+    soft_commit: bool,
 }
 
 impl<'a> PreparedCommit<'a> {
-    pub(crate) fn new(index_writer: &'a mut IndexWriter, opstamp: Opstamp) -> PreparedCommit<'_> {
+    pub(crate) fn new(
+        index_writer: &'a mut IndexWriter,
+        opstamp: Opstamp,
+        soft_commit: bool,
+    ) -> PreparedCommit {
         PreparedCommit {
             index_writer,
             payload: None,
             opstamp,
+            soft_commit,
         }
     }
 
@@ -33,11 +39,12 @@ impl<'a> PreparedCommit<'a> {
 
     pub fn commit(self) -> Result<Opstamp> {
         info!("committing {}", self.opstamp);
-        let _ = block_on(
-            self.index_writer
-                .segment_updater()
-                .schedule_commit(self.opstamp, self.payload),
-        );
+        block_on(self.index_writer.segment_updater().schedule_commit(
+            self.opstamp,
+            self.payload,
+            self.soft_commit,
+        ))?;
+        block_on(self.index_writer.trigger_commit());
         Ok(self.opstamp)
     }
 }

src/indexer/segment_entry.rs

@@ -1,7 +1,9 @@
 use crate::common::BitSet;
 use crate::core::SegmentId;
 use crate::core::SegmentMeta;
+use crate::directory::ManagedDirectory;
 use crate::indexer::delete_queue::DeleteCursor;
+use crate::{Opstamp, Segment};
 use std::fmt;
 
 /// A segment entry describes the state of
@@ -19,55 +21,81 @@ use std::fmt;
 /// in the .del file or in the `delete_bitset`.
 #[derive(Clone)]
 pub struct SegmentEntry {
-    meta: SegmentMeta,
+    segment: Segment,
     delete_bitset: Option<BitSet>,
     delete_cursor: DeleteCursor,
 }
 
 impl SegmentEntry {
     /// Create a new `SegmentEntry`
-    pub fn new(
-        segment_meta: SegmentMeta,
+    pub(crate) fn new(
+        segment: Segment,
         delete_cursor: DeleteCursor,
         delete_bitset: Option<BitSet>,
     ) -> SegmentEntry {
         SegmentEntry {
-            meta: segment_meta,
+            segment,
             delete_bitset,
             delete_cursor,
         }
     }
 
-    /// Return a reference to the segment entry deleted bitset.
+    pub fn persist(&mut self, dest_directory: ManagedDirectory) -> crate::Result<()> {
+        // TODO take in account delete bitset?
+        self.segment.persist(dest_directory)?;
+        Ok(())
+    }
+
+    pub fn segment(&self) -> &Segment {
+        &self.segment
+    }
+
+    /// `Takes` (as in Option::take) the delete bitset of
+    /// a segment entry.
     ///
     /// `DocId` in this bitset are flagged as deleted.
-    pub fn delete_bitset(&self) -> Option<&BitSet> {
-        self.delete_bitset.as_ref()
+    pub fn take_delete_bitset(&mut self) -> Option<BitSet> {
+        self.delete_bitset.take()
     }
 
-    /// Set the `SegmentMeta` for this segment.
-    pub fn set_meta(&mut self, segment_meta: SegmentMeta) {
-        self.meta = segment_meta;
+    /// Reset the delete informmationo in this segment.
+    ///
+    /// The `SegmentEntry` segment's `SegmentMeta` gets updated, and
+    /// any delete bitset is drop and set to None.
+    pub fn reset_delete_meta(&mut self, num_deleted_docs: u32, target_opstamp: Opstamp) {
+        self.segment = self
+            .segment
+            .clone()
+            .with_delete_meta(num_deleted_docs, target_opstamp);
+        self.delete_bitset = None;
     }
 
+    pub fn set_delete_cursor(&mut self, delete_cursor: DeleteCursor) {
+        self.delete_cursor = delete_cursor;
+    }
     /// Return a reference to the segment_entry's delete cursor
-    pub fn delete_cursor(&mut self) -> &mut DeleteCursor {
-        &mut self.delete_cursor
+    pub fn delete_cursor(&mut self) -> DeleteCursor {
+        self.delete_cursor.clone()
     }
 
     /// Returns the segment id.
     pub fn segment_id(&self) -> SegmentId {
-        self.meta.id()
+        self.segment.id()
     }
 
     /// Accessor to the `SegmentMeta`
     pub fn meta(&self) -> &SegmentMeta {
-        &self.meta
+        self.segment.meta()
     }
 }
 
 impl fmt::Debug for SegmentEntry {
     fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
-        write!(formatter, "SegmentEntry({:?})", self.meta)
+        let num_deletes = self.delete_bitset.as_ref().map(|bitset| bitset.len());
+        write!(
+            formatter,
+            "SegmentEntry(seg={:?}, ndel={:?})",
+            self.segment, num_deletes
+        )
     }
 }

src/indexer/segment_manager.rs

@@ -1,17 +1,14 @@
 use super::segment_register::SegmentRegister;
 use crate::core::SegmentId;
 use crate::core::SegmentMeta;
-use crate::error::TantivyError;
-use crate::indexer::delete_queue::DeleteCursor;
 use crate::indexer::SegmentEntry;
-use crate::Result as TantivyResult;
+use crate::Segment;
 use std::collections::hash_set::HashSet;
-use std::fmt::{self, Debug, Formatter};
-use std::sync::RwLock;
+use std::sync::{Arc, RwLock};
 use std::sync::{RwLockReadGuard, RwLockWriteGuard};
 
 #[derive(Default)]
-struct SegmentRegisters {
+pub(crate) struct SegmentRegisters {
     uncommitted: SegmentRegister,
     committed: SegmentRegister,
 }
@@ -23,6 +20,17 @@ pub(crate) enum SegmentsStatus {
 }
 
 impl SegmentRegisters {
+    pub fn new(committed: SegmentRegister) -> SegmentRegisters {
+        SegmentRegisters {
+            uncommitted: Default::default(),
+            committed,
+        }
+    }
+
+    pub fn committed_segment(&self) -> Vec<Segment> {
+        self.committed.segments()
+    }
+
     /// Check if all the segments are committed or uncommited.
     ///
     /// If some segment is missing or segments are in a different state (this should not happen
@@ -45,18 +53,7 @@ impl SegmentRegisters {
 /// changes (merges especially)
 #[derive(Default)]
 pub struct SegmentManager {
-    registers: RwLock<SegmentRegisters>,
-}
-
-impl Debug for SegmentManager {
-    fn fmt(&self, f: &mut Formatter<'_>) -> Result<(), fmt::Error> {
-        let lock = self.read();
-        write!(
-            f,
-            "{{ uncommitted: {:?}, committed: {:?} }}",
-            lock.uncommitted, lock.committed
-        )
-    }
+    registers: Arc<RwLock<SegmentRegisters>>,
 }
 
 pub fn get_mergeable_segments(
@@ -75,16 +72,8 @@ pub fn get_mergeable_segments(
 }
 
 impl SegmentManager {
-    pub fn from_segments(
-        segment_metas: Vec<SegmentMeta>,
-        delete_cursor: &DeleteCursor,
-    ) -> SegmentManager {
-        SegmentManager {
-            registers: RwLock::new(SegmentRegisters {
-                uncommitted: SegmentRegister::default(),
-                committed: SegmentRegister::new(segment_metas, delete_cursor),
-            }),
-        }
+    pub(crate) fn new(registers: Arc<RwLock<SegmentRegisters>>) -> SegmentManager {
+        SegmentManager { registers }
     }
 
     /// Returns all of the segment entries (committed or uncommitted)
@@ -145,7 +134,7 @@ impl SegmentManager {
     /// Returns an error if some segments are missing, or if
     /// the `segment_ids` are not either all committed or all
     /// uncommitted.
-    pub fn start_merge(&self, segment_ids: &[SegmentId]) -> TantivyResult<Vec<SegmentEntry>> {
+    pub fn start_merge(&self, segment_ids: &[SegmentId]) -> crate::Result<Vec<SegmentEntry>> {
         let registers_lock = self.read();
         let mut segment_entries = vec![];
         if registers_lock.uncommitted.contains_all(segment_ids) {
@@ -166,7 +155,7 @@ impl SegmentManager {
             let error_msg = "Merge operation sent for segments that are not \
                              all uncommited or commited."
                 .to_string();
-            return Err(TantivyError::InvalidArgument(error_msg));
+            return Err(crate::Error::InvalidArgument(error_msg));
         }
         Ok(segment_entries)
     }

src/indexer/segment_register.rs

@@ -1,7 +1,10 @@
 use crate::core::SegmentId;
 use crate::core::SegmentMeta;
+use crate::directory::ManagedDirectory;
 use crate::indexer::delete_queue::DeleteCursor;
 use crate::indexer::segment_entry::SegmentEntry;
+use crate::schema::Schema;
+use crate::Segment;
 use std::collections::HashMap;
 use std::collections::HashSet;
 use std::fmt::{self, Debug, Formatter};
@@ -46,6 +49,13 @@ impl SegmentRegister {
             .collect()
     }
 
+    pub fn segments(&self) -> Vec<Segment> {
+        self.segment_states
+            .values()
+            .map(|segment_entry| segment_entry.segment().clone())
+            .collect()
+    }
+
     pub fn segment_entries(&self) -> Vec<SegmentEntry> {
         self.segment_states.values().cloned().collect()
     }
@@ -79,11 +89,17 @@ impl SegmentRegister {
         self.segment_states.get(segment_id).cloned()
     }
 
-    pub fn new(segment_metas: Vec<SegmentMeta>, delete_cursor: &DeleteCursor) -> SegmentRegister {
+    pub fn new(
+        directory: &ManagedDirectory,
+        schema: &Schema,
+        segment_metas: Vec<SegmentMeta>,
+        delete_cursor: &DeleteCursor,
+    ) -> SegmentRegister {
         let mut segment_states = HashMap::new();
-        for segment_meta in segment_metas {
-            let segment_id = segment_meta.id();
-            let segment_entry = SegmentEntry::new(segment_meta, delete_cursor.clone(), None);
+        for meta in segment_metas {
+            let segment_id = meta.id();
+            let segment = Segment::new_persisted(meta, directory.clone(), schema.clone());
+            let segment_entry = SegmentEntry::new(segment, delete_cursor.clone(), None);
             segment_states.insert(segment_id, segment_entry);
         }
         SegmentRegister { segment_states }
@@ -95,6 +111,7 @@ mod tests {
     use super::*;
     use crate::core::{SegmentId, SegmentMetaInventory};
     use crate::indexer::delete_queue::*;
+    use crate::schema::Schema;
 
     fn segment_ids(segment_register: &SegmentRegister) -> Vec<SegmentId> {
         segment_register
@@ -108,6 +125,7 @@ mod tests {
     fn test_segment_register() {
         let inventory = SegmentMetaInventory::default();
         let delete_queue = DeleteQueue::new();
+        let schema = Schema::builder().build();
 
         let mut segment_register = SegmentRegister::default();
         let segment_id_a = SegmentId::generate_random();
@@ -115,21 +133,24 @@ mod tests {
         let segment_id_merged = SegmentId::generate_random();
 
         {
-            let segment_meta = inventory.new_segment_meta(segment_id_a, 0u32);
-            let segment_entry = SegmentEntry::new(segment_meta, delete_queue.cursor(), None);
+            let meta = inventory.new_segment_meta(segment_id_a, 0u32);
+            let segment = Segment::new_volatile(meta, schema.clone());
+            let segment_entry = SegmentEntry::new(segment, delete_queue.cursor(), None);
             segment_register.add_segment_entry(segment_entry);
         }
         assert_eq!(segment_ids(&segment_register), vec![segment_id_a]);
         {
             let segment_meta = inventory.new_segment_meta(segment_id_b, 0u32);
-            let segment_entry = SegmentEntry::new(segment_meta, delete_queue.cursor(), None);
+            let segment = Segment::new_volatile(segment_meta, schema.clone());
+            let segment_entry = SegmentEntry::new(segment, delete_queue.cursor(), None);
             segment_register.add_segment_entry(segment_entry);
         }
         segment_register.remove_segment(&segment_id_a);
         segment_register.remove_segment(&segment_id_b);
         {
             let segment_meta_merged = inventory.new_segment_meta(segment_id_merged, 0u32);
-            let segment_entry = SegmentEntry::new(segment_meta_merged, delete_queue.cursor(), None);
+            let segment = Segment::new_volatile(segment_meta_merged, schema);
+            let segment_entry = SegmentEntry::new(segment, delete_queue.cursor(), None);
             segment_register.add_segment_entry(segment_entry);
         }
         assert_eq!(segment_ids(&segment_register), vec![segment_id_merged]);

src/indexer/segment_serializer.rs

@@ -1,13 +1,10 @@
-use crate::Directory;
+use crate::Result;
 
 use crate::core::Segment;
 use crate::core::SegmentComponent;
-use crate::directory::error::OpenWriteError;
-use crate::directory::{DirectoryClone, RAMDirectory, TerminatingWrite, WritePtr};
 use crate::fastfield::FastFieldSerializer;
 use crate::fieldnorm::FieldNormsSerializer;
 use crate::postings::InvertedIndexSerializer;
-use crate::schema::Schema;
 use crate::store::StoreWriter;
 
 /// Segment serializer is in charge of laying out on disk
@@ -17,50 +14,25 @@ pub struct SegmentSerializer {
     fast_field_serializer: FastFieldSerializer,
     fieldnorms_serializer: FieldNormsSerializer,
     postings_serializer: InvertedIndexSerializer,
-    bundle_writer: Option<(RAMDirectory, WritePtr)>,
-}
-
-pub(crate) struct SegmentSerializerWriters {
-    postings_wrt: WritePtr,
-    positions_skip_wrt: WritePtr,
-    positions_wrt: WritePtr,
-    terms_wrt: WritePtr,
-    fast_field_wrt: WritePtr,
-    fieldnorms_wrt: WritePtr,
-    store_wrt: WritePtr,
-}
-
-impl SegmentSerializerWriters {
-    pub(crate) fn for_segment(segment: &mut Segment) -> Result<Self, OpenWriteError> {
-        Ok(SegmentSerializerWriters {
-            postings_wrt: segment.open_write(SegmentComponent::POSTINGS)?,
-            positions_skip_wrt: segment.open_write(SegmentComponent::POSITIONS)?,
-            positions_wrt: segment.open_write(SegmentComponent::POSITIONSSKIP)?,
-            terms_wrt: segment.open_write(SegmentComponent::TERMS)?,
-            fast_field_wrt: segment.open_write(SegmentComponent::FASTFIELDS)?,
-            fieldnorms_wrt: segment.open_write(SegmentComponent::FIELDNORMS)?,
-            store_wrt: segment.open_write(SegmentComponent::STORE)?,
-        })
-    }
 }
 
 impl SegmentSerializer {
-    pub(crate) fn new(schema: Schema, writers: SegmentSerializerWriters) -> crate::Result<Self> {
-        let fast_field_serializer = FastFieldSerializer::from_write(writers.fast_field_wrt)?;
-        let fieldnorms_serializer = FieldNormsSerializer::from_write(writers.fieldnorms_wrt)?;
-        let postings_serializer = InvertedIndexSerializer::open(
-            schema,
-            writers.terms_wrt,
-            writers.postings_wrt,
-            writers.positions_wrt,
-            writers.positions_skip_wrt,
-        );
+    /// Creates a new `SegmentSerializer`.
+    pub fn for_segment(segment: &mut Segment) -> Result<SegmentSerializer> {
+        let store_write = segment.open_write(SegmentComponent::STORE)?;
+
+        let fast_field_write = segment.open_write(SegmentComponent::FASTFIELDS)?;
+        let fast_field_serializer = FastFieldSerializer::from_write(fast_field_write)?;
+
+        let fieldnorms_write = segment.open_write(SegmentComponent::FIELDNORMS)?;
+        let fieldnorms_serializer = FieldNormsSerializer::from_write(fieldnorms_write)?;
+
+        let postings_serializer = InvertedIndexSerializer::open(segment)?;
         Ok(SegmentSerializer {
-            store_writer: StoreWriter::new(writers.store_wrt),
+            store_writer: StoreWriter::new(store_write),
             fast_field_serializer,
             fieldnorms_serializer,
             postings_serializer,
-            bundle_writer: None,
         })
     }
 
@@ -85,15 +57,11 @@ impl SegmentSerializer {
     }
 
     /// Finalize the segment serialization.
-    pub fn close(mut self) -> crate::Result<()> {
+    pub fn close(self) -> Result<()> {
         self.fast_field_serializer.close()?;
         self.postings_serializer.close()?;
         self.store_writer.close()?;
         self.fieldnorms_serializer.close()?;
-        if let Some((ram_directory, mut bundle_wrt)) = self.bundle_writer.take() {
-            ram_directory.serialize_bundle(&mut bundle_wrt)?;
-            bundle_wrt.terminate()?;
-        }
         Ok(())
     }
 }

src/indexer/segment_updater.rs

@@ -7,12 +7,10 @@ use crate::core::SegmentMeta;
 use crate::core::SerializableSegment;
 use crate::core::META_FILEPATH;
 use crate::directory::{Directory, DirectoryClone, GarbageCollectionResult};
-use crate::indexer::delete_queue::DeleteCursor;
 use crate::indexer::index_writer::advance_deletes;
 use crate::indexer::merge_operation::MergeOperationInventory;
 use crate::indexer::merger::IndexMerger;
-use crate::indexer::segment_manager::SegmentsStatus;
-use crate::indexer::segment_serializer::SegmentSerializerWriters;
+use crate::indexer::segment_manager::{SegmentRegisters, SegmentsStatus};
 use crate::indexer::stamper::Stamper;
 use crate::indexer::SegmentEntry;
 use crate::indexer::SegmentSerializer;
@@ -118,8 +116,7 @@ fn merge(
 
     // First we apply all of the delet to the merged segment, up to the target opstamp.
     for segment_entry in &mut segment_entries {
-        let segment = index.segment(segment_entry.meta().clone());
-        advance_deletes(segment, segment_entry, target_opstamp)?;
+        advance_deletes(segment_entry, target_opstamp)?;
     }
 
     let delete_cursor = segment_entries[0].delete_cursor().clone();
@@ -133,15 +130,15 @@ fn merge(
     let merger: IndexMerger = IndexMerger::open(index.schema(), &segments[..])?;
 
     // ... we just serialize this index merger in our new segment to merge the two segments.
-    let segment_serializer_wrts = SegmentSerializerWriters::for_segment(&mut merged_segment)?;
-    let segment_serializer =
-        SegmentSerializer::new(merged_segment.schema(), segment_serializer_wrts)?;
+    let segment_serializer = SegmentSerializer::for_segment(&mut merged_segment)?;
 
-    let num_docs = merger.write(segment_serializer)?;
+    let max_doc = merger.write(segment_serializer)?;
 
-    let segment_meta = index.new_segment_meta(merged_segment.id(), num_docs);
-
-    Ok(SegmentEntry::new(segment_meta, delete_cursor, None))
+    Ok(SegmentEntry::new(
+        merged_segment.with_max_doc(max_doc),
+        delete_cursor,
+        None,
+    ))
 }
 
 pub(crate) struct InnerSegmentUpdater {
@@ -165,12 +162,11 @@ pub(crate) struct InnerSegmentUpdater {
 
 impl SegmentUpdater {
     pub fn create(
+        segment_registers: Arc<RwLock<SegmentRegisters>>,
         index: Index,
         stamper: Stamper,
-        delete_cursor: &DeleteCursor,
     ) -> crate::Result<SegmentUpdater> {
-        let segments = index.searchable_segment_metas()?;
-        let segment_manager = SegmentManager::from_segments(segments, delete_cursor);
+        let segment_manager = SegmentManager::new(segment_registers);
         let pool = ThreadPoolBuilder::new()
             .name_prefix("segment_updater")
             .pool_size(1)
@@ -233,6 +229,7 @@ impl SegmentUpdater {
         &self,
         segment_entry: SegmentEntry,
     ) -> impl Future<Output = crate::Result<()>> {
+        // TODO temporary: serializing the segment at this point.
         let segment_updater = self.clone();
         self.schedule_future(async move {
             segment_updater.segment_manager.add_segment(segment_entry);
@@ -261,8 +258,7 @@ impl SegmentUpdater {
     fn purge_deletes(&self, target_opstamp: Opstamp) -> crate::Result<Vec<SegmentEntry>> {
         let mut segment_entries = self.segment_manager.segment_entries();
         for segment_entry in &mut segment_entries {
-            let segment = self.index.segment(segment_entry.meta().clone());
-            advance_deletes(segment, segment_entry, target_opstamp)?;
+            advance_deletes(segment_entry, target_opstamp)?;
         }
         Ok(segment_entries)
     }
@@ -330,12 +326,21 @@ impl SegmentUpdater {
         &self,
         opstamp: Opstamp,
         payload: Option<String>,
+        soft_commit: bool,
     ) -> impl Future<Output = crate::Result<()>> {
         let segment_updater: SegmentUpdater = self.clone();
+        let directory = self.index.directory().clone();
         self.schedule_future(async move {
-            let segment_entries = segment_updater.purge_deletes(opstamp)?;
+            let mut segment_entries = segment_updater.purge_deletes(opstamp)?;
+            if !soft_commit {
+                for segment_entry in &mut segment_entries {
+                    segment_entry.persist(directory.clone())?;
+                }
+            }
             segment_updater.segment_manager.commit(segment_entries);
-            segment_updater.save_metas(opstamp, payload)?;
+            if !soft_commit {
+                segment_updater.save_metas(opstamp, payload)?;
+            }
             let _ = garbage_collect_files(segment_updater.clone()).await;
             segment_updater.consider_merge_options().await;
             Ok(())
@@ -473,17 +478,14 @@ impl SegmentUpdater {
         let end_merge_future = self.schedule_future(async move {
             info!("End merge {:?}", after_merge_segment_entry.meta());
             {
-                let mut delete_cursor = after_merge_segment_entry.delete_cursor().clone();
+                let mut delete_cursor = after_merge_segment_entry.delete_cursor();
                 if let Some(delete_operation) = delete_cursor.get() {
                     let committed_opstamp = segment_updater.load_metas().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(e) = advance_deletes(
-                            segment,
-                            &mut after_merge_segment_entry,
-                            committed_opstamp,
-                        ) {
+                        let _index = &segment_updater.index;
+                        if let Err(e) =
+                            advance_deletes(&mut after_merge_segment_entry, committed_opstamp)
+                        {
                             error!(
                                 "Merge of {:?} was cancelled (advancing deletes failed): {:?}",
                                 merge_operation.segment_ids(),

src/indexer/segment_writer.rs

@@ -3,7 +3,7 @@ use crate::core::Segment;
 use crate::core::SerializableSegment;
 use crate::fastfield::FastFieldsWriter;
 use crate::fieldnorm::FieldNormsWriter;
-use crate::indexer::segment_serializer::{SegmentSerializer, SegmentSerializerWriters};
+use crate::indexer::segment_serializer::SegmentSerializer;
 use crate::postings::compute_table_size;
 use crate::postings::MultiFieldPostingsWriter;
 use crate::schema::FieldType;
@@ -11,9 +11,9 @@ use crate::schema::Schema;
 use crate::schema::Term;
 use crate::schema::Value;
 use crate::schema::{Field, FieldEntry};
-use crate::tokenizer::BoxedTokenizer;
 use crate::tokenizer::FacetTokenizer;
 use crate::tokenizer::PreTokenizedStream;
+use crate::tokenizer::{BoxedTokenizer, TokenizerManager};
 use crate::tokenizer::{TokenStream, TokenStreamChain, Tokenizer};
 use crate::DocId;
 use crate::Opstamp;
@@ -66,12 +66,12 @@ impl SegmentWriter {
     pub fn for_segment(
         memory_budget: usize,
         mut segment: Segment,
-        schema: &Schema,
+        tokenizers: &TokenizerManager,
     ) -> Result<SegmentWriter> {
+        let schema = segment.schema();
         let table_num_bits = initial_table_size(memory_budget)?;
-        let segment_serializer_wrts = SegmentSerializerWriters::for_segment(&mut segment)?;
-        let segment_serializer = SegmentSerializer::new(segment.schema(), segment_serializer_wrts)?;
-        let multifield_postings = MultiFieldPostingsWriter::new(schema, table_num_bits);
+        let segment_serializer = SegmentSerializer::for_segment(&mut segment)?;
+        let multifield_postings = MultiFieldPostingsWriter::new(&schema, table_num_bits);
         let tokenizers = schema
             .fields()
             .map(
@@ -80,7 +80,7 @@ impl SegmentWriter {
                         .get_indexing_options()
                         .and_then(|text_index_option| {
                             let tokenizer_name = &text_index_option.tokenizer();
-                            segment.index().tokenizers().get(tokenizer_name)
+                            tokenizers.get(tokenizer_name)
                         }),
                     _ => None,
                 },
@@ -89,9 +89,9 @@ impl SegmentWriter {
         Ok(SegmentWriter {
             max_doc: 0,
             multifield_postings,
-            fieldnorms_writer: FieldNormsWriter::for_schema(schema),
+            fieldnorms_writer: FieldNormsWriter::for_schema(&schema),
             segment_serializer,
-            fast_field_writers: FastFieldsWriter::from_schema(schema),
+            fast_field_writers: FastFieldsWriter::from_schema(&schema),
             doc_opstamps: Vec::with_capacity(1_000),
             tokenizers,
         })

src/postings/mod.rs

@@ -75,7 +75,7 @@ pub mod tests {
         let schema = schema_builder.build();
         let index = Index::create_in_ram(schema);
         let mut segment = index.new_segment();
-        let mut posting_serializer = InvertedIndexSerializer::for_segment(&mut segment).unwrap();
+        let mut posting_serializer = InvertedIndexSerializer::open(&mut segment).unwrap();
         {
             let mut field_serializer = posting_serializer.new_field(text_field, 120 * 4).unwrap();
             field_serializer.new_term("abc".as_bytes()).unwrap();
@@ -220,7 +220,7 @@ pub mod tests {
 
         {
             let mut segment_writer =
-                SegmentWriter::for_segment(3_000_000, segment.clone(), &schema).unwrap();
+                SegmentWriter::for_segment(3_000_000, segment.clone(), index.tokenizers()).unwrap();
             {
                 let mut doc = Document::default();
                 // checking that position works if the field has two values

src/postings/serializer.rs

@@ -10,8 +10,8 @@ use crate::postings::USE_SKIP_INFO_LIMIT;
 use crate::schema::Schema;
 use crate::schema::{Field, FieldEntry, FieldType};
 use crate::termdict::{TermDictionaryBuilder, TermOrdinal};
+use crate::DocId;
 use crate::Result;
-use crate::{Directory, DocId};
 use std::io::{self, Write};
 
 /// `InvertedIndexSerializer` is in charge of serializing
@@ -54,36 +54,33 @@ pub struct InvertedIndexSerializer {
 }
 
 impl InvertedIndexSerializer {
-    pub(crate) fn for_segment(segment: &mut Segment) -> crate::Result<Self> {
-        let schema = segment.schema();
-        use crate::core::SegmentComponent;
-        let terms_wrt = segment.open_write(SegmentComponent::TERMS)?;
-        let postings_wrt = segment.open_write(SegmentComponent::POSTINGS)?;
-        let positions_wrt = segment.open_write(SegmentComponent::POSITIONS)?;
-        let positions_idx_wrt = segment.open_write(SegmentComponent::POSITIONSSKIP)?;
-        Ok(Self::open(
-            schema,
-            terms_wrt,
-            postings_wrt,
-            positions_wrt,
-            positions_idx_wrt,
-        ))
-    }
-    /// Open a new `PostingsSerializer` for the given segment
-    pub(crate) fn open(
+    /// Open a new `InvertedIndexSerializer` for the given segment
+    fn create(
+        terms_write: CompositeWrite<WritePtr>,
+        postings_write: CompositeWrite<WritePtr>,
+        positions_write: CompositeWrite<WritePtr>,
+        positionsidx_write: CompositeWrite<WritePtr>,
         schema: Schema,
-        terms_wrt: WritePtr,
-        postings_wrt: WritePtr,
-        positions_wrt: WritePtr,
-        positions_idx_wrt: WritePtr,
-    ) -> InvertedIndexSerializer {
-        InvertedIndexSerializer {
-            terms_write: CompositeWrite::wrap(terms_wrt),
-            postings_write: CompositeWrite::wrap(postings_wrt),
-            positions_write: CompositeWrite::wrap(positions_wrt),
-            positionsidx_write: CompositeWrite::wrap(positions_idx_wrt),
+    ) -> Result<InvertedIndexSerializer> {
+        Ok(InvertedIndexSerializer {
+            terms_write,
+            postings_write,
+            positions_write,
+            positionsidx_write,
             schema,
-        }
+        })
+    }
+
+    /// Open a new `PostingsSerializer` for the given segment
+    pub fn open(segment: &mut Segment) -> Result<InvertedIndexSerializer> {
+        use crate::SegmentComponent::{POSITIONS, POSITIONSSKIP, POSTINGS, TERMS};
+        InvertedIndexSerializer::create(
+            CompositeWrite::wrap(segment.open_write(TERMS)?),
+            CompositeWrite::wrap(segment.open_write(POSTINGS)?),
+            CompositeWrite::wrap(segment.open_write(POSITIONS)?),
+            CompositeWrite::wrap(segment.open_write(POSITIONSSKIP)?),
+            segment.schema(),
+        )
     }
 
     /// Must be called before starting pushing terms of

src/query/term_query/mod.rs

@@ -112,6 +112,7 @@ mod tests {
         let term_a = Term::from_field_text(text_field, "a");
         let term_query = TermQuery::new(term_a, IndexRecordOption::Basic);
         let reader = index.reader().unwrap();
+        assert_eq!(reader.searcher().segment_readers().len(), 1);
         assert_eq!(term_query.count(&*reader.searcher()).unwrap(), 1);
     }
 

src/reader/index_writer_reader.rs

@@ -0,0 +1,83 @@
+use crate::directory::{WatchCallbackList, WatchHandle};
+use crate::indexer::SegmentRegisters;
+use crate::reader::pool::Pool;
+use crate::{Index, LeasedItem, Searcher, Segment, SegmentReader};
+use std::iter::repeat_with;
+use std::sync::{Arc, RwLock, Weak};
+
+struct InnerNRTReader {
+    num_searchers: usize,
+    index: Index,
+    searcher_pool: Pool<Searcher>,
+    segment_registers: Arc<RwLock<SegmentRegisters>>,
+}
+
+impl InnerNRTReader {
+    fn load_segment_readers(&self) -> crate::Result<Vec<SegmentReader>> {
+        let segments: Vec<Segment> = {
+            let rlock = self.segment_registers.read().unwrap();
+            rlock.committed_segment()
+        };
+        segments
+            .iter()
+            .map(SegmentReader::open)
+            .collect::<crate::Result<Vec<SegmentReader>>>()
+    }
+
+    pub fn reload(&self) -> crate::Result<()> {
+        let segment_readers: Vec<SegmentReader> = self.load_segment_readers()?;
+        let schema = self.index.schema();
+        let searchers = repeat_with(|| {
+            Searcher::new(schema.clone(), self.index.clone(), segment_readers.clone())
+        })
+        .take(self.num_searchers)
+        .collect();
+        self.searcher_pool.publish_new_generation(searchers);
+        Ok(())
+    }
+
+    pub fn searcher(&self) -> LeasedItem<Searcher> {
+        self.searcher_pool.acquire()
+    }
+}
+
+#[derive(Clone)]
+pub struct NRTReader {
+    inner: Arc<InnerNRTReader>,
+    watch_handle: WatchHandle,
+}
+
+impl NRTReader {
+    pub fn reload(&self) -> crate::Result<()> {
+        self.inner.reload()
+    }
+
+    pub fn searcher(&self) -> LeasedItem<Searcher> {
+        self.inner.searcher()
+    }
+
+    pub(crate) fn create(
+        num_searchers: usize,
+        index: Index,
+        segment_registers: Arc<RwLock<SegmentRegisters>>,
+        watch_callback_list: &WatchCallbackList,
+    ) -> crate::Result<Self> {
+        let inner_reader: Arc<InnerNRTReader> = Arc::new(InnerNRTReader {
+            num_searchers,
+            index,
+            searcher_pool: Pool::new(),
+            segment_registers,
+        });
+        let inner_reader_weak: Weak<InnerNRTReader> = Arc::downgrade(&inner_reader);
+        let watch_handle = watch_callback_list.subscribe(Box::new(move || {
+            if let Some(nrt_reader_arc) = inner_reader_weak.upgrade() {
+                let _ = nrt_reader_arc.reload();
+            }
+        }));
+        inner_reader.reload()?;
+        Ok(NRTReader {
+            inner: inner_reader,
+            watch_handle,
+        })
+    }
+}

src/reader/meta_file_reader.rs

@@ -0,0 +1,180 @@
+use super::pool::Pool;
+use crate::core::Segment;
+use crate::directory::Directory;
+use crate::directory::WatchHandle;
+use crate::directory::META_LOCK;
+use crate::Searcher;
+use crate::SegmentReader;
+use crate::{Index, LeasedItem};
+use crate::{IndexReader, Result};
+use std::iter::repeat_with;
+use std::sync::Arc;
+
+/// Defines when a new version of the index should be reloaded.
+///
+/// Regardless of whether you search and index in the same process, tantivy does not necessarily
+/// reflects the change that are commited to your index. `ReloadPolicy` precisely helps you define
+/// when you want your index to be reloaded.
+#[derive(Clone, Copy)]
+pub enum ReloadPolicy {
+    /// The index is entirely reloaded manually.
+    /// All updates of the index should be manual.
+    ///
+    /// No change is reflected automatically. You are required to call `.load_seacher()` manually.
+    Manual,
+    /// The index is reloaded within milliseconds after a new commit is available.
+    /// This is made possible by watching changes in the `meta.json` file.
+    OnCommit, // TODO add NEAR_REAL_TIME(target_ms)
+}
+
+/// `IndexReader` builder
+///
+/// It makes it possible to set the following values.
+///
+/// - `num_searchers` (by default, the number of detected CPU threads):
+///
+///   When `num_searchers` queries are requested at the same time, the `num_searchers` will block
+///   until the one of the searcher in-use gets released.
+/// - `reload_policy` (by default `ReloadPolicy::OnCommit`):
+///
+///   See [`ReloadPolicy`](./enum.ReloadPolicy.html) for more details.
+#[derive(Clone)]
+pub struct IndexReaderBuilder {
+    num_searchers: usize,
+    reload_policy: ReloadPolicy,
+    index: Index,
+}
+
+impl IndexReaderBuilder {
+    pub(crate) fn new(index: Index) -> IndexReaderBuilder {
+        IndexReaderBuilder {
+            num_searchers: num_cpus::get(),
+            reload_policy: ReloadPolicy::Manual,
+            index,
+        }
+    }
+
+    /// Builds the reader.
+    ///
+    /// Building the reader is a non-trivial operation that requires
+    /// to open different segment readers. It may take hundreds of milliseconds
+    /// of time and it may return an error.
+    /// TODO(pmasurel) Use the `TryInto` trait once it is available in stable.
+    pub fn try_into(self) -> Result<IndexReader> {
+        let inner_reader = MetaFileIndexReaderInner {
+            index: self.index,
+            num_searchers: self.num_searchers,
+            searcher_pool: Pool::new(),
+        };
+        inner_reader.reload()?;
+        let inner_reader_arc = Arc::new(inner_reader);
+        let watch_handle_opt: Option<WatchHandle>;
+        match self.reload_policy {
+            ReloadPolicy::Manual => {
+                // No need to set anything...
+                watch_handle_opt = None;
+            }
+            ReloadPolicy::OnCommit => {
+                let inner_reader_arc_clone = inner_reader_arc.clone();
+                let callback = move || {
+                    if let Err(err) = inner_reader_arc_clone.reload() {
+                        error!(
+                            "Error while loading searcher after commit was detected. {:?}",
+                            err
+                        );
+                    }
+                };
+                let watch_handle = inner_reader_arc
+                    .index
+                    .directory()
+                    .watch(Box::new(callback))?;
+                watch_handle_opt = Some(watch_handle);
+            }
+        }
+        Ok(IndexReader::from(MetaFileIndexReader {
+            inner: inner_reader_arc,
+            watch_handle_opt,
+        }))
+    }
+
+    /// Sets the reload_policy.
+    ///
+    /// See [`ReloadPolicy`](./enum.ReloadPolicy.html) for more details.
+    pub fn reload_policy(mut self, reload_policy: ReloadPolicy) -> IndexReaderBuilder {
+        self.reload_policy = reload_policy;
+        self
+    }
+
+    /// Sets the number of `Searcher` in the searcher pool.
+    pub fn num_searchers(mut self, num_searchers: usize) -> IndexReaderBuilder {
+        self.num_searchers = num_searchers;
+        self
+    }
+}
+
+struct MetaFileIndexReaderInner {
+    num_searchers: usize,
+    searcher_pool: Pool<Searcher>,
+    index: Index,
+}
+
+impl MetaFileIndexReaderInner {
+    fn load_segment_readers(&self) -> crate::Result<Vec<SegmentReader>> {
+        // We keep the lock until we have effectively finished opening the
+        // the `SegmentReader` because it prevents a diffferent process
+        // to garbage collect these file while we open them.
+        //
+        // Once opened, on linux & mac, the mmap will remain valid after
+        // the file has been deleted
+        // On windows, the file deletion will fail.
+        let _meta_lock = self.index.directory().acquire_lock(&META_LOCK)?;
+        let searchable_segments = self.searchable_segments()?;
+        searchable_segments
+            .iter()
+            .map(SegmentReader::open)
+            .collect::<Result<_>>()
+    }
+
+    fn reload(&self) -> crate::Result<()> {
+        let segment_readers: Vec<SegmentReader> = self.load_segment_readers()?;
+        let schema = self.index.schema();
+        let searchers = repeat_with(|| {
+            Searcher::new(schema.clone(), self.index.clone(), segment_readers.clone())
+        })
+        .take(self.num_searchers)
+        .collect();
+        self.searcher_pool.publish_new_generation(searchers);
+        Ok(())
+    }
+
+    /// Returns the list of segments that are searchable
+    fn searchable_segments(&self) -> crate::Result<Vec<Segment>> {
+        self.index.searchable_segments()
+    }
+
+    fn searcher(&self) -> LeasedItem<Searcher> {
+        self.searcher_pool.acquire()
+    }
+}
+
+/// `IndexReader` is your entry point to read and search the index.
+///
+/// It controls when a new version of the index should be loaded and lends
+/// you instances of `Searcher` for the last loaded version.
+///
+/// `Clone` does not clone the different pool of searcher. `IndexReader`
+/// just wraps and `Arc`.
+#[derive(Clone)]
+pub struct MetaFileIndexReader {
+    inner: Arc<MetaFileIndexReaderInner>,
+    watch_handle_opt: Option<WatchHandle>,
+}
+
+impl MetaFileIndexReader {
+    pub fn reload(&self) -> crate::Result<()> {
+        self.inner.reload()
+    }
+    pub fn searcher(&self) -> LeasedItem<Searcher> {
+        self.inner.searcher()
+    }
+}

src/reader/mod.rs

@@ -1,172 +1,20 @@
+mod index_writer_reader;
+mod meta_file_reader;
 mod pool;
 
+use self::meta_file_reader::MetaFileIndexReader;
+pub use self::meta_file_reader::{IndexReaderBuilder, ReloadPolicy};
 pub use self::pool::LeasedItem;
-use self::pool::Pool;
-use crate::core::Segment;
-use crate::directory::Directory;
-use crate::directory::WatchHandle;
-use crate::directory::META_LOCK;
-use crate::Index;
-use crate::Result;
+pub(crate) use crate::reader::index_writer_reader::{NRTReader};
 use crate::Searcher;
-use crate::SegmentReader;
-use std::sync::Arc;
 
-/// Defines when a new version of the index should be reloaded.
-///
-/// Regardless of whether you search and index in the same process, tantivy does not necessarily
-/// reflects the change that are commited to your index. `ReloadPolicy` precisely helps you define
-/// when you want your index to be reloaded.
-#[derive(Clone, Copy)]
-pub enum ReloadPolicy {
-    /// The index is entirely reloaded manually.
-    /// All updates of the index should be manual.
-    ///
-    /// No change is reflected automatically. You are required to call `.load_seacher()` manually.
-    Manual,
-    /// The index is reloaded within milliseconds after a new commit is available.
-    /// This is made possible by watching changes in the `meta.json` file.
-    OnCommit, // TODO add NEAR_REAL_TIME(target_ms)
-}
-
-/// `IndexReader` builder
-///
-/// It makes it possible to set the following values.
-///
-/// - `num_searchers` (by default, the number of detected CPU threads):
-///
-///   When `num_searchers` queries are requested at the same time, the `num_searchers` will block
-///   until the one of the searcher in-use gets released.
-/// - `reload_policy` (by default `ReloadPolicy::OnCommit`):
-///
-///   See [`ReloadPolicy`](./enum.ReloadPolicy.html) for more details.
 #[derive(Clone)]
-pub struct IndexReaderBuilder {
-    num_searchers: usize,
-    reload_policy: ReloadPolicy,
-    index: Index,
-}
-
-impl IndexReaderBuilder {
-    pub(crate) fn new(index: Index) -> IndexReaderBuilder {
-        IndexReaderBuilder {
-            num_searchers: num_cpus::get(),
-            reload_policy: ReloadPolicy::OnCommit,
-            index,
-        }
-    }
-
-    /// Builds the reader.
-    ///
-    /// Building the reader is a non-trivial operation that requires
-    /// to open different segment readers. It may take hundreds of milliseconds
-    /// of time and it may return an error.
-    /// TODO(pmasurel) Use the `TryInto` trait once it is available in stable.
-    pub fn try_into(self) -> Result<IndexReader> {
-        let inner_reader = InnerIndexReader {
-            index: self.index,
-            num_searchers: self.num_searchers,
-            searcher_pool: Pool::new(),
-        };
-        inner_reader.reload()?;
-        let inner_reader_arc = Arc::new(inner_reader);
-        let watch_handle_opt: Option<WatchHandle>;
-        match self.reload_policy {
-            ReloadPolicy::Manual => {
-                // No need to set anything...
-                watch_handle_opt = None;
-            }
-            ReloadPolicy::OnCommit => {
-                let inner_reader_arc_clone = inner_reader_arc.clone();
-                let callback = move || {
-                    if let Err(err) = inner_reader_arc_clone.reload() {
-                        error!(
-                            "Error while loading searcher after commit was detected. {:?}",
-                            err
-                        );
-                    }
-                };
-                let watch_handle = inner_reader_arc
-                    .index
-                    .directory()
-                    .watch(Box::new(callback))?;
-                watch_handle_opt = Some(watch_handle);
-            }
-        }
-        Ok(IndexReader {
-            inner: inner_reader_arc,
-            watch_handle_opt,
-        })
-    }
-
-    /// Sets the reload_policy.
-    ///
-    /// See [`ReloadPolicy`](./enum.ReloadPolicy.html) for more details.
-    pub fn reload_policy(mut self, reload_policy: ReloadPolicy) -> IndexReaderBuilder {
-        self.reload_policy = reload_policy;
-        self
-    }
-
-    /// Sets the number of `Searcher` in the searcher pool.
-    pub fn num_searchers(mut self, num_searchers: usize) -> IndexReaderBuilder {
-        self.num_searchers = num_searchers;
-        self
-    }
-}
-
-struct InnerIndexReader {
-    num_searchers: usize,
-    searcher_pool: Pool<Searcher>,
-    index: Index,
-}
-
-impl InnerIndexReader {
-    fn reload(&self) -> Result<()> {
-        let segment_readers: Vec<SegmentReader> = {
-            let _meta_lock = self.index.directory().acquire_lock(&META_LOCK)?;
-            let searchable_segments = self.searchable_segments()?;
-            searchable_segments
-                .iter()
-                .map(SegmentReader::open)
-                .collect::<Result<_>>()?
-        };
-        let schema = self.index.schema();
-        let searchers = (0..self.num_searchers)
-            .map(|_| Searcher::new(schema.clone(), self.index.clone(), segment_readers.clone()))
-            .collect();
-        self.searcher_pool.publish_new_generation(searchers);
-        Ok(())
-    }
-
-    /// Returns the list of segments that are searchable
-    fn searchable_segments(&self) -> Result<Vec<Segment>> {
-        self.index.searchable_segments()
-    }
-
-    fn searcher(&self) -> LeasedItem<Searcher> {
-        self.searcher_pool.acquire()
-    }
-}
-
-/// `IndexReader` is your entry point to read and search the index.
-///
-/// It controls when a new version of the index should be loaded and lends
-/// you instances of `Searcher` for the last loaded version.
-///
-/// `Clone` does not clone the different pool of searcher. `IndexReader`
-/// just wraps and `Arc`.
-#[derive(Clone)]
-pub struct IndexReader {
-    inner: Arc<InnerIndexReader>,
-    watch_handle_opt: Option<WatchHandle>,
+pub enum IndexReader {
+    FromMetaFile(MetaFileIndexReader),
+    NRT(NRTReader),
 }
 
 impl IndexReader {
-    #[cfg(test)]
-    pub(crate) fn index(&self) -> Index {
-        self.inner.index.clone()
-    }
-
     /// Update searchers so that they reflect the state of the last
     /// `.commit()`.
     ///
@@ -176,8 +24,11 @@ impl IndexReader {
     ///
     /// This automatic reload can take 10s of milliseconds to kick in however, and in unit tests
     /// it can be nice to deterministically force the reload of searchers.
-    pub fn reload(&self) -> Result<()> {
-        self.inner.reload()
+    pub fn reload(&self) -> crate::Result<()> {
+        match self {
+            IndexReader::FromMetaFile(meta_file_reader) => meta_file_reader.reload(),
+            IndexReader::NRT(nrt_reader) => nrt_reader.reload(),
+        }
     }
 
     /// Returns a searcher
@@ -191,6 +42,21 @@ impl IndexReader {
     /// The same searcher must be used for a given query, as it ensures
     /// the use of a consistent segment set.
     pub fn searcher(&self) -> LeasedItem<Searcher> {
-        self.inner.searcher()
+        match self {
+            IndexReader::FromMetaFile(meta_file_reader) => meta_file_reader.searcher(),
+            IndexReader::NRT(nrt_reader) => nrt_reader.searcher(),
+        }
+    }
+}
+
+impl From<MetaFileIndexReader> for IndexReader {
+    fn from(meta_file_reader: MetaFileIndexReader) -> Self {
+        IndexReader::FromMetaFile(meta_file_reader)
+    }
+}
+
+impl From<NRTReader> for IndexReader {
+    fn from(nrt_reader: NRTReader) -> Self {
+        IndexReader::NRT(nrt_reader)
     }
 }

src/store/mod.rs

@@ -57,7 +57,7 @@ use self::compression_snap::{compress, decompress};
 pub mod tests {
 
     use super::*;
-    use crate::directory::{Directory, RAMDirectory, ReadOnlyDirectory, WritePtr};
+    use crate::directory::{Directory, RAMDirectory, WritePtr};
     use crate::schema::Document;
     use crate::schema::FieldValue;
     use crate::schema::Schema;

src/termdict/mod.rs

@@ -36,7 +36,7 @@ pub use self::termdict::{TermDictionary, TermDictionaryBuilder};
 mod tests {
     use super::{TermDictionary, TermDictionaryBuilder, TermStreamer};
     use crate::core::Index;
-    use crate::directory::{Directory, RAMDirectory, ReadOnlyDirectory, ReadOnlySource};
+    use crate::directory::{Directory, RAMDirectory, ReadOnlySource};
     use crate::postings::TermInfo;
     use crate::schema::{Document, FieldType, Schema, TEXT};
     use std::path::PathBuf;

tests/failpoints/mod.rs

@@ -1,8 +1,6 @@
 use fail;
 use std::path::Path;
-use tantivy::directory::{
-    Directory, ManagedDirectory, RAMDirectory, ReadOnlyDirectory, TerminatingWrite,
-};
+use tantivy::directory::{Directory, ManagedDirectory, RAMDirectory, TerminatingWrite};
 use tantivy::doc;
 use tantivy::schema::{Schema, TEXT};
 use tantivy::{Index, Term};