rust/tantivy
1
0
Fork 0
mirror of https://github.com/quickwit-oss/tantivy.git synced 2026-10-24 20:52:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

issue/230 Add an optional commit message. (#231)

Browse Source 
Closes #230
This commit is contained in:
Paul Masurel
2017-12-27 12:27:02 +09:00
committed by GitHub
parent 442bc9a1b8
commit f7b0392bd5
7 changed files with 335 additions and 126 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/core/index.rs
+6 -8
View File
@@ -18,6 +18,7 @@ use core::SegmentMeta;
use super::pool::LeasedItem;
use std::path::Path;
use core::IndexMeta;
use indexer::DirectoryLock;
use IndexWriter;
use directory::ManagedDirectory;
use core::META_FILEPATH;
@@ -113,12 +114,8 @@ impl Index {
Index::create_from_metas(directory, &metas)
}
/// Returns the index opstamp.
///
/// The opstamp is the number of documents that have been added
/// from the beginning of time, and until the moment of the last commit.
pub fn opstamp(&self) -> u64 {
load_metas(self.directory()).unwrap().opstamp
pub(crate) fn load_metas(&self) -> Result<IndexMeta> {
load_metas(self.directory())
}
/// Open a new index writer. Attempts to acquire a lockfile.
@@ -141,7 +138,8 @@ impl Index {
num_threads: usize,
heap_size_in_bytes: usize,
) -> Result<IndexWriter> {
open_index_writer(self, num_threads, heap_size_in_bytes)
let directory_lock = DirectoryLock::lock(self.directory().box_clone())?;
open_index_writer(self, num_threads, heap_size_in_bytes, directory_lock)
}
/// Creates a multithreaded writer
@@ -194,7 +192,7 @@ impl Index {
/// Reads the meta.json and returns the list of
/// `SegmentMeta` from the last commit.
pub fn searchable_segment_metas(&self) -> Result<Vec<SegmentMeta>> {
Ok(load_metas(self.directory())?.segments)
Ok(self.load_metas()?.segments)
}
/// Returns the list of segment ids that are searchable.
src/core/index_meta.rs
+30
View File
@@ -14,6 +14,8 @@ pub struct IndexMeta {
pub segments: Vec<SegmentMeta>,
pub schema: Schema,
pub opstamp: u64,
#[serde(skip_serializing_if = "Option::is_none")]
pub payload: Option<String>,
}
impl IndexMeta {
@@ -22,6 +24,34 @@ impl IndexMeta {
segments: vec![],
schema,
opstamp: 0u64,
payload: None,
}
}
}
#[cfg(test)]
mod tests {
use serde_json;
use super::IndexMeta;
use schema::{TEXT, SchemaBuilder};
#[test]
fn test_serialize_metas() {
let schema = {
let mut schema_builder = SchemaBuilder::new();
schema_builder.add_text_field("text", TEXT);
schema_builder.build()
};
let index_metas = IndexMeta {
segments: Vec::new(),
schema: schema,
opstamp: 0u64,
payload: None
};
let json = serde_json::ser::to_string(&index_metas).expect("serialization failed");
assert_eq!(json, r#"{"segments":[],"schema":[{"name":"text","type":"text","options":{"indexing":{"record":"position","tokenizer":"default"},"stored":false}}],"opstamp":0}"#);
}
}
src/indexer/index_writer.rs
+200 -71
View File
@@ -10,7 +10,6 @@ use indexer::stamper::Stamper;
use datastruct::stacker::Heap;
use directory::FileProtection;
use error::{Error, ErrorKind, Result, ResultExt};
use Directory;
use fastfield::write_delete_bitset;
use indexer::delete_queue::{DeleteCursor, DeleteQueue};
use futures::Canceled;
@@ -29,9 +28,10 @@ use schema::Term;
use std::mem;
use std::mem::swap;
use std::thread::JoinHandle;
use super::directory_lock::DirectoryLock;
use indexer::DirectoryLock;
use super::operation::AddOperation;
use super::segment_updater::SegmentUpdater;
use super::PreparedCommit;
use std::thread;
// Size of the margin for the heap. A segment is closed when the remaining memory
@@ -57,7 +57,7 @@ type DocumentReceiver = chan::Receiver<AddOperation>;
pub struct IndexWriter {
// the lock is just used to bind the
// lifetime of the lock with that of the IndexWriter.
_directory_lock: DirectoryLock,
_directory_lock: Option<DirectoryLock>,
index: Index,
@@ -104,6 +104,7 @@ pub fn open_index_writer(
index: &Index,
num_threads: usize,
heap_size_in_bytes_per_thread: usize,
directory_lock: DirectoryLock,
) -> Result<IndexWriter> {
if heap_size_in_bytes_per_thread <= HEAP_SIZE_LIMIT as usize {
panic!(format!(
@@ -111,15 +112,12 @@ pub fn open_index_writer(
HEAP_SIZE_LIMIT
));
}
let directory_lock = DirectoryLock::lock(index.directory().box_clone())?;
let (document_sender, document_receiver): (DocumentSender, DocumentReceiver) =
chan::sync(PIPELINE_MAX_SIZE_IN_DOCS);
let delete_queue = DeleteQueue::new();
let current_opstamp = index.opstamp();
let current_opstamp = index.load_metas()?.opstamp;
let stamper = Stamper::new(current_opstamp);
@@ -127,7 +125,7 @@ pub fn open_index_writer(
SegmentUpdater::new(index.clone(), stamper.clone(), &delete_queue.cursor())?;
let mut index_writer = IndexWriter {
_directory_lock: directory_lock,
_directory_lock: Some(directory_lock),
heap_size_in_bytes_per_thread,
index: index.clone(),
@@ -175,8 +173,10 @@ pub fn compute_deleted_bitset(
// that may be affected by the delete operation.
let limit_doc = doc_opstamps.compute_doc_limit(delete_op.opstamp);
let inverted_index = segment_reader.inverted_index(delete_op.term.field());
if let Some(mut docset) =
inverted_index.read_postings(&delete_op.term, IndexRecordOption::Basic)
if let Some(mut docset) = inverted_index.read_postings(
&delete_op.term,
IndexRecordOption::Basic,
)
{
while docset.advance() {
let deleted_doc = docset.doc();
@@ -286,6 +286,11 @@ fn index_documents(
break;
}
}
if !segment_updater.is_alive() {
return Ok(false);
}
let num_docs = segment_writer.max_doc();
// this is ensured by the call to peek before starting
@@ -333,13 +338,15 @@ impl IndexWriter {
join_handle
.join()
.expect("Indexing Worker thread panicked")
.chain_err(|| ErrorKind::ErrorInThread("Error in indexing worker thread.".into()))?;
.chain_err(|| {
ErrorKind::ErrorInThread("Error in indexing worker thread.".into())
})?;
}
drop(self.workers_join_handle);
let result = self.segment_updater
.wait_merging_thread()
.chain_err(|| ErrorKind::ErrorInThread("Failed to join merging thread.".into()));
let result = self.segment_updater.wait_merging_thread().chain_err(|| {
ErrorKind::ErrorInThread("Failed to join merging thread.".into())
});
if let Err(ref e) = result {
error!("Some merging thread failed {:?}", e);
@@ -352,8 +359,10 @@ impl IndexWriter {
pub fn add_segment(&mut self, segment_meta: SegmentMeta) {
let delete_cursor = self.delete_queue.cursor();
let segment_entry = SegmentEntry::new(segment_meta, delete_cursor, None);
self.segment_updater
.add_segment(self.generation, segment_entry);
self.segment_updater.add_segment(
self.generation,
segment_entry,
);
}
#[doc(hidden)]
@@ -379,7 +388,8 @@ impl IndexWriter {
let join_handle: JoinHandle<Result<()>> = thread::Builder::new()
.name(format!(
"indexing thread {} for gen {}",
self.worker_id, generation
self.worker_id,
generation
))
.spawn(move || {
loop {
@@ -459,10 +469,9 @@ impl IndexWriter {
///
/// Returns the former segment_ready channel.
fn recreate_document_channel(&mut self) -> DocumentReceiver {
let (mut document_sender, mut document_receiver): (
DocumentSender,
DocumentReceiver,
) = chan::sync(PIPELINE_MAX_SIZE_IN_DOCS);
let (mut document_sender, mut document_receiver): (DocumentSender,
DocumentReceiver) =
chan::sync(PIPELINE_MAX_SIZE_IN_DOCS);
swap(&mut self.document_sender, &mut document_sender);
swap(&mut self.document_receiver, &mut document_receiver);
document_receiver
@@ -476,23 +485,102 @@ impl IndexWriter {
/// state as it was after the last commit.
///
/// The opstamp at the last commit is returned.
pub fn rollback(mut self) -> Result<IndexWriter> {
pub fn rollback(&mut self) -> Result<()> {
info!("Rolling back to opstamp {}", self.committed_opstamp);
// marks the segment updater as killed. From now on, all
// segment updates will be ignored.
self.segment_updater.kill();
let document_receiver = self.document_receiver.clone();
// take the directory lock to create a new index_writer.
let directory_lock = self._directory_lock.take().expect(
"The IndexWriter does not have any lock. This is a bug, please report.",
);
let new_index_writer: IndexWriter = open_index_writer(
&self.index,
self.num_threads,
self.heap_size_in_bytes_per_thread,
directory_lock,
)?;
// the current `self` is dropped right away because of this call.
//
// This will drop the document queue, and the thread
// should terminate.
mem::replace(self, new_index_writer);
// Drains the document receiver pipeline :
// Workers don't need to index the pending documents.
let receiver_clone = self.document_receiver.clone();
let index = self.index.clone();
let num_threads = self.num_threads;
let heap_size_in_bytes_per_thread = self.heap_size_in_bytes_per_thread;
drop(self);
for _ in receiver_clone {}
//
// This will reach an end as the only document_sender
// was dropped with the index_writer.
for _ in document_receiver.clone() {}
let index_writer = open_index_writer(&index, num_threads, heap_size_in_bytes_per_thread)?;
Ok(())
}
Ok(index_writer)
/// Prepares a commit.
///
/// Calling `prepare_commit()` will cut the indexing
/// queue. All pending documents will be sent to the
/// indexing workers. They will then terminate, regardless
/// of the size of their current segment and flush their
/// work on disk.
///
/// Once a commit is "prepared", you can either
/// call
/// * `.commit()`: to accept this commit
/// * `.abort()`: to cancel this commit.
///
/// In the current implementation, `PreparedCommit` borrows
/// the `IndexWriter` mutably so we are guaranteed that no new
/// document can be added as long as it is committed or is
/// dropped.
///
/// 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) -> Result<PreparedCommit> {
// Here, because we join all of the worker threads,
// all of the segment update for this commit have been
// sent.
//
// No document belonging to the next generation have been
// pushed too, because add_document can only happen
// on this thread.
// This will move uncommitted segments to the state of
// committed segments.
info!("committing {}", self.committed_opstamp);
// this will drop the current document channel
// and recreate a new one channels.
self.recreate_document_channel();
let mut former_workers_join_handle = Vec::new();
swap(
&mut former_workers_join_handle,
&mut self.workers_join_handle,
);
for worker_handle in former_workers_join_handle {
let indexing_worker_result = worker_handle.join().map_err(|e| {
Error::from_kind(ErrorKind::ErrorInThread(format!("{:?}", e)))
})?;
indexing_worker_result?;
// add a new worker for the next generation.
self.add_indexing_worker()?;
}
let commit_opstamp = self.stamper.stamp();
let prepared_commit = PreparedCommit::new(self, commit_opstamp);
info!("Preparing commit {}", commit_opstamp);
Ok(prepared_commit)
}
/// Commits all of the pending changes
@@ -510,43 +598,11 @@ impl IndexWriter {
/// that made it in the commit.
///
pub fn commit(&mut self) -> Result<u64> {
// here, because we join all of the worker threads,
// all of the segment update for this commit have been
// sent.
//
// No document belonging to the next generation have been
// pushed too, because add_document can only happen
// on this thread.
self.prepare_commit()?.commit()
}
// This will move uncommitted segments to the state of
// committed segments.
self.committed_opstamp = self.stamper.stamp();
info!("committing {}", self.committed_opstamp);
// this will drop the current document channel
// and recreate a new one channels.
self.recreate_document_channel();
let mut former_workers_join_handle = Vec::new();
swap(
&mut former_workers_join_handle,
&mut self.workers_join_handle,
);
for worker_handle in former_workers_join_handle {
let indexing_worker_result = worker_handle
.join()
.map_err(|e| Error::from_kind(ErrorKind::ErrorInThread(format!("{:?}", e))))?;
indexing_worker_result?;
// add a new worker for the next generation.
self.add_indexing_worker()?;
}
// wait for the segment update thread to have processed the info
self.segment_updater.commit(self.committed_opstamp)?;
Ok(self.committed_opstamp)
pub(crate) fn segment_updater(&self) -> &SegmentUpdater {
&self.segment_updater
}
/// Delete all documents containing a given term.
@@ -664,8 +720,7 @@ mod tests {
doc.add_text(text_field, "a");
index_writer.add_document(doc);
}
index_writer = index_writer.rollback().unwrap();
index_writer.rollback().unwrap();
assert_eq!(index_writer.commit_opstamp(), 0u64);
assert_eq!(num_docs_containing("a"), 0);
@@ -718,9 +773,9 @@ mod tests {
}
// this should create 8 segments and trigger a merge.
index_writer.commit().expect("commit failed");
index_writer
.wait_merging_threads()
.expect("waiting merging thread failed");
index_writer.wait_merging_threads().expect(
"waiting merging thread failed",
);
index.load_searchers().unwrap();
assert_eq!(num_docs_containing("a"), 200);
@@ -728,4 +783,78 @@ mod tests {
}
}
#[test]
fn test_prepare_with_commit_message() {
let _ = env_logger::init();
let mut schema_builder = schema::SchemaBuilder::default();
let text_field = schema_builder.add_text_field("text", schema::TEXT);
let index = Index::create_in_ram(schema_builder.build());
{
// writing the segment
let mut index_writer = index.writer_with_num_threads(4, 4 * 30_000_000).unwrap();
// create 8 segments with 100 tiny docs
for _doc in 0..100 {
index_writer.add_document(doc!(text_field => "a"));
}
{
let mut prepared_commit = index_writer.prepare_commit().expect("commit failed");
prepared_commit.set_payload("first commit");
assert_eq!(prepared_commit.opstamp(), 100);
prepared_commit.commit().expect("commit failed");
}
{
let metas = index.load_metas().unwrap();
assert_eq!(metas.payload.unwrap(), "first commit");
}
for _doc in 0..100 {
index_writer.add_document(doc!(text_field => "a"));
}
index_writer.commit().unwrap();
{
let metas = index.load_metas().unwrap();
assert!(metas.payload.is_none());
}
}
}
#[test]
fn test_prepare_but_rollback() {
let _ = env_logger::init();
let mut schema_builder = schema::SchemaBuilder::default();
let text_field = schema_builder.add_text_field("text", schema::TEXT);
let index = Index::create_in_ram(schema_builder.build());
{
// writing the segment
let mut index_writer = index.writer_with_num_threads(4, 4 * 30_000_000).unwrap();
// create 8 segments with 100 tiny docs
for _doc in 0..100 {
index_writer.add_document(doc!(text_field => "a"));
}
{
let mut prepared_commit = index_writer.prepare_commit().expect("commit failed");
prepared_commit.set_payload("first commit");
assert_eq!(prepared_commit.opstamp(), 100);
prepared_commit.abort().expect("commit failed");
}
{
let metas = index.load_metas().unwrap();
assert!(metas.payload.is_none());
}
for _doc in 0..100 {
index_writer.add_document(doc!(text_field => "b"));
}
index_writer.commit().unwrap();
}
index.load_searchers().unwrap();
let num_docs_containing = |s: &str| {
let searcher = index.searcher();
let term_a = Term::from_field_text(text_field, s);
searcher.doc_freq(&term_a)
};
assert_eq!(num_docs_containing("a"), 0);
assert_eq!(num_docs_containing("b"), 100);
}
}
src/indexer/mod.rs
+3
View File
@@ -13,7 +13,9 @@ mod segment_entry;
mod doc_opstamp_mapping;
pub mod operation;
mod stamper;
mod prepared_commit;
pub use self::prepared_commit::PreparedCommit;
pub use self::segment_entry::{SegmentEntry, SegmentState};
pub use self::segment_serializer::SegmentSerializer;
pub use self::segment_writer::SegmentWriter;
@@ -21,6 +23,7 @@ pub use self::index_writer::IndexWriter;
pub use self::log_merge_policy::LogMergePolicy;
pub use self::merge_policy::{MergeCandidate, MergePolicy, NoMergePolicy};
pub use self::segment_manager::SegmentManager;
pub(crate) use self::directory_lock::DirectoryLock;
/// Alias for the default merge policy, which is the `LogMergePolicy`.
pub type DefaultMergePolicy = LogMergePolicy;
src/indexer/prepared_commit.rs
+40
View File
@@ -0,0 +1,40 @@
use Result;
use super::IndexWriter;
/// A prepared commit
pub struct PreparedCommit<'a> {
index_writer: &'a mut IndexWriter,
payload: Option<String>,
opstamp: u64,
}
impl<'a> PreparedCommit<'a> {
pub(crate) fn new(index_writer: &'a mut IndexWriter, opstamp: u64) -> PreparedCommit {
PreparedCommit {
index_writer: index_writer,
payload: None,
opstamp: opstamp,
}
}
pub fn opstamp(&self) -> u64 {
self.opstamp
}
pub fn set_payload(&mut self, payload: &str) {
self.payload = Some(payload.to_string())
}
pub fn abort(self) -> Result<()> {
self.index_writer.rollback()
}
pub fn commit(self) -> Result<u64> {
info!("committing {}", self.opstamp);
self.index_writer.segment_updater().commit(
self.opstamp,
self.payload,
)?;
Ok(self.opstamp)
}
}
src/indexer/segment_updater.rs
+53 -45
View File
@@ -46,7 +46,7 @@ use super::segment_manager::{get_mergeable_segments, SegmentManager};
///
/// This method is not part of tantivy's public API
pub fn save_new_metas(schema: Schema, opstamp: u64, directory: &mut Directory) -> Result<()> {
save_metas(vec![], schema, opstamp, directory)
save_metas(vec![], schema, opstamp, None, directory)
}
/// Save the index meta file.
@@ -62,12 +62,14 @@ pub fn save_metas(
segment_metas: Vec<SegmentMeta>,
schema: Schema,
opstamp: u64,
payload: Option<String>,
directory: &mut Directory,
) -> Result<()> {
let metas = IndexMeta {
segments: segment_metas,
schema,
opstamp,
payload: payload.clone(),
};
let mut buffer = serde_json::to_vec_pretty(&metas)?;
write!(&mut buffer, "\n")?;
@@ -100,7 +102,8 @@ fn perform_merge(
let mut file_protections: Vec<FileProtection> = vec![];
for segment_id in segment_ids {
if let Some(mut segment_entry) = segment_updater.0.segment_manager.segment_entry(segment_id)
if let Some(mut segment_entry) =
segment_updater.0.segment_manager.segment_entry(segment_id)
{
let segment = index.segment(segment_entry.meta().clone());
if let Some(file_protection) =
@@ -132,12 +135,13 @@ fn perform_merge(
// ... we just serialize this index merger in our new segment
// to merge the two segments.
let segment_serializer = SegmentSerializer::for_segment(&mut merged_segment)
.expect("Creating index serializer failed");
let segment_serializer = SegmentSerializer::for_segment(&mut merged_segment).expect(
"Creating index serializer failed",
);
let num_docs = merger
.write(segment_serializer)
.expect("Serializing merged index failed");
let num_docs = merger.write(segment_serializer).expect(
"Serializing merged index failed",
);
let mut segment_meta = SegmentMeta::new(merged_segment.id());
segment_meta.set_max_doc(num_docs);
@@ -222,7 +226,7 @@ impl SegmentUpdater {
self.0.killed.store(true, Ordering::Release);
}
fn is_alive(&self) -> bool {
pub fn is_alive(&self) -> bool {
!self.0.killed.load(Ordering::Acquire)
}
@@ -239,7 +243,7 @@ impl SegmentUpdater {
Ok(segment_entries)
}
pub fn save_metas(&self, opstamp: u64) {
pub fn save_metas(&self, opstamp: u64, commit_message: Option<String>) {
if self.is_alive() {
let index = &self.0.index;
let directory = index.directory();
@@ -247,6 +251,7 @@ impl SegmentUpdater {
self.0.segment_manager.committed_segment_metas(),
index.schema(),
opstamp,
commit_message,
directory.box_clone().borrow_mut(),
).expect("Could not save metas.");
}
@@ -261,22 +266,20 @@ impl SegmentUpdater {
fn garbage_collect_files_exec(&self) {
info!("Running garbage collection");
let mut index = self.0.index.clone();
index
.directory_mut()
.garbage_collect(|| self.0.segment_manager.list_files());
index.directory_mut().garbage_collect(
|| self.0.segment_manager.list_files(),
);
}
pub fn commit(&self, opstamp: u64) -> Result<()> {
self.run_async(move |segment_updater| {
if segment_updater.is_alive() {
let segment_entries = segment_updater
.purge_deletes(opstamp)
.expect("Failed purge deletes");
segment_updater.0.segment_manager.commit(segment_entries);
segment_updater.save_metas(opstamp);
segment_updater.garbage_collect_files_exec();
segment_updater.consider_merge_options();
}
pub fn commit(&self, opstamp: u64, payload: Option<String>) -> Result<()> {
self.run_async(move |segment_updater| if segment_updater.is_alive() {
let segment_entries = segment_updater.purge_deletes(opstamp).expect(
"Failed purge deletes",
);
segment_updater.0.segment_manager.commit(segment_entries);
segment_updater.save_metas(opstamp, payload);
segment_updater.garbage_collect_files_exec();
segment_updater.consider_merge_options();
}).wait()
}
@@ -340,11 +343,10 @@ impl SegmentUpdater {
.remove(&merging_thread_id);
Ok(())
});
self.0
.merging_threads
.write()
.unwrap()
.insert(merging_thread_id, merging_join_handle);
self.0.merging_threads.write().unwrap().insert(
merging_thread_id,
merging_join_handle,
);
merging_future_recv
}
@@ -367,9 +369,10 @@ impl SegmentUpdater {
before_merge_segment_ids: &[SegmentId],
after_merge_segment_entry: SegmentId,
) {
self.0
.segment_manager
.cancel_merge(before_merge_segment_ids, after_merge_segment_entry);
self.0.segment_manager.cancel_merge(
before_merge_segment_ids,
after_merge_segment_entry,
);
}
fn end_merge(
@@ -382,7 +385,11 @@ impl SegmentUpdater {
let mut delete_cursor = after_merge_segment_entry.delete_cursor().clone();
let mut _file_protection_opt = None;
if let Some(delete_operation) = delete_cursor.get() {
let committed_opstamp = segment_updater.0.index.opstamp();
let committed_opstamp = segment_updater.0
.index.load_metas()
.expect("Failed to read opstamp")
.opstamp;
if delete_operation.opstamp < committed_opstamp {
let index = &segment_updater.0.index;
let segment = index.segment(after_merge_segment_entry.meta().clone());
@@ -397,7 +404,8 @@ impl SegmentUpdater {
Err(e) => {
error!(
"Merge of {:?} was cancelled (advancing deletes failed): {:?}",
before_merge_segment_ids, e
before_merge_segment_ids,
e
);
// ... cancel merge
if cfg!(test) {
@@ -412,13 +420,14 @@ impl SegmentUpdater {
}
}
}
segment_updater
.0
.segment_manager
.end_merge(&before_merge_segment_ids, after_merge_segment_entry);
segment_updater.0.segment_manager.end_merge(
&before_merge_segment_ids,
after_merge_segment_entry,
);
segment_updater.consider_merge_options();
info!("save metas");
segment_updater.save_metas(segment_updater.0.index.opstamp());
let previous_metas = segment_updater.0.index.load_metas().unwrap();
segment_updater.save_metas(previous_metas.opstamp, previous_metas.payload);
segment_updater.garbage_collect_files_exec();
}).wait()
}
@@ -450,10 +459,9 @@ impl SegmentUpdater {
}
debug!("wait merging thread {}", new_merging_threads.len());
for (_, merging_thread_handle) in new_merging_threads {
merging_thread_handle
.join()
.map(|_| ())
.map_err(|_| ErrorKind::ErrorInThread("Merging thread failed.".into()))?;
merging_thread_handle.join().map(|_| ()).map_err(|_| {
ErrorKind::ErrorInThread("Merging thread failed.".into())
})?;
}
// Our merging thread may have queued their completed
self.run_async(move |_| {}).wait()?;
@@ -519,9 +527,9 @@ mod tests {
assert_eq!(index.searcher().num_docs(), 302);
{
index_writer
.wait_merging_threads()
.expect("waiting for merging threads");
index_writer.wait_merging_threads().expect(
"waiting for merging threads",
);
}
index.load_searchers().unwrap();
src/lib.rs
+3 -2
View File
@@ -447,7 +447,7 @@ mod tests {
{
index_writer.delete_term(Term::from_field_text(text_field, "c"));
}
index_writer = index_writer.rollback().unwrap();
index_writer.rollback().unwrap();
index_writer.delete_term(Term::from_field_text(text_field, "a"));
index_writer.commit().unwrap();
}
@@ -715,7 +715,8 @@ mod tests {
let mut schema_builder = SchemaBuilder::default();
let text_field = schema_builder.add_text_field("text", TEXT);
let other_text_field = schema_builder.add_text_field("text2", TEXT);
let document = doc!(text_field => "tantivy",
let document =
doc!(text_field => "tantivy",
text_field => "some other value",
other_text_field => "short");
assert_eq!(document.len(), 3);