From f7b0392bd55d43d394848873fa3442db2e3bf250 Mon Sep 17 00:00:00 2001 From: Paul Masurel Date: Wed, 27 Dec 2017 12:27:02 +0900 Subject: [PATCH] issue/230 Add an optional commit message. (#231) Closes #230 --- src/core/index.rs | 14 +- src/core/index_meta.rs | 30 ++++ src/indexer/index_writer.rs | 271 ++++++++++++++++++++++++--------- src/indexer/mod.rs | 3 + src/indexer/prepared_commit.rs | 40 +++++ src/indexer/segment_updater.rs | 98 ++++++------ src/lib.rs | 5 +- 7 files changed, 335 insertions(+), 126 deletions(-) create mode 100644 src/indexer/prepared_commit.rs diff --git a/src/core/index.rs b/src/core/index.rs index 159924552..2ec390e45 100644 --- a/src/core/index.rs +++ b/src/core/index.rs @@ -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 { + 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 { - 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> { - Ok(load_metas(self.directory())?.segments) + Ok(self.load_metas()?.segments) } /// Returns the list of segment ids that are searchable. diff --git a/src/core/index_meta.rs b/src/core/index_meta.rs index d40198f92..636117166 100644 --- a/src/core/index_meta.rs +++ b/src/core/index_meta.rs @@ -14,6 +14,8 @@ pub struct IndexMeta { pub segments: Vec, pub schema: Schema, pub opstamp: u64, + #[serde(skip_serializing_if = "Option::is_none")] + pub payload: Option, } 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}"#); + } +} \ No newline at end of file diff --git a/src/indexer/index_writer.rs b/src/indexer/index_writer.rs index 1c0609c1c..23b2dee4f 100644 --- a/src/indexer/index_writer.rs +++ b/src/indexer/index_writer.rs @@ -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; 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, 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 { 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> = 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 { + 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 { + // 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 { - // 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); + } + } diff --git a/src/indexer/mod.rs b/src/indexer/mod.rs index b567186b8..e10c01151 100644 --- a/src/indexer/mod.rs +++ b/src/indexer/mod.rs @@ -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; diff --git a/src/indexer/prepared_commit.rs b/src/indexer/prepared_commit.rs new file mode 100644 index 000000000..7853d2130 --- /dev/null +++ b/src/indexer/prepared_commit.rs @@ -0,0 +1,40 @@ +use Result; +use super::IndexWriter; + +/// A prepared commit +pub struct PreparedCommit<'a> { + index_writer: &'a mut IndexWriter, + payload: Option, + 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 { + info!("committing {}", self.opstamp); + self.index_writer.segment_updater().commit( + self.opstamp, + self.payload, + )?; + Ok(self.opstamp) + } +} diff --git a/src/indexer/segment_updater.rs b/src/indexer/segment_updater.rs index ef5ac6ff1..bf4a9cc08 100644 --- a/src/indexer/segment_updater.rs +++ b/src/indexer/segment_updater.rs @@ -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, schema: Schema, opstamp: u64, + payload: Option, 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 = 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) { 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) -> 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(); diff --git a/src/lib.rs b/src/lib.rs index eca6eb692..af0c19e16 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -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);