From 8f377b92d009df66c2e9e954f8628f7cc5689454 Mon Sep 17 00:00:00 2001 From: Paul Masurel Date: Fri, 11 Aug 2017 18:11:32 +0900 Subject: [PATCH] introducing a field serializer --- src/indexer/merger.rs | 187 ++++++++++---------- src/postings/mod.rs | 17 +- src/postings/postings_writer.rs | 19 +- src/postings/recorder.rs | 14 +- src/postings/serializer.rs | 299 +++++++++++++++++++------------- 5 files changed, 298 insertions(+), 238 deletions(-) diff --git a/src/indexer/merger.rs b/src/indexer/merger.rs index 6318a17c9..468d867e7 100644 --- a/src/indexer/merger.rs +++ b/src/indexer/merger.rs @@ -19,7 +19,6 @@ use store::StoreWriter; use std::cmp::{min, max}; use schema::Term; use termdict::TermStreamer; -use postings::SegmentPostingsOption; pub struct IndexMerger { schema: Schema, @@ -215,103 +214,115 @@ impl IndexMerger { merged_doc_id_map.push(segment_local_map); } - let mut last_field: Option = None; + // Create the total list of doc ids + // by stacking the doc ids from the different segment. + // + // In the new segments, the doc id from the different + // segment are stacked so that : + // - Segment 0's doc ids become doc id [0, seg.max_doc] + // - Segment 1's doc ids become [seg0.max_doc, seg0.max_doc + seg.max_doc] + // - Segment 2's doc ids become [seg0.max_doc + seg1.max_doc, + // seg0.max_doc + seg1.max_doc + seg2.max_doc] + // ... + if !merged_terms.advance() { + return Ok(()); + } - let mut segment_postings_option = SegmentPostingsOption::FreqAndPositions; + let mut current_field = Term::wrap(merged_terms.key()).field(); - while merged_terms.advance() { + loop { + // this loop processes all fields. + let mut field_serializer = serializer.new_field(current_field); - // Create the total list of doc ids - // by stacking the doc ids from the different segment. - // - // In the new segments, the doc id from the different - // segment are stacked so that : - // - Segment 0's doc ids become doc id [0, seg.max_doc] - // - Segment 1's doc ids become [seg0.max_doc, seg0.max_doc + seg.max_doc] - // - Segment 2's doc ids become [seg0.max_doc + seg1.max_doc, - // seg0.max_doc + seg1.max_doc + seg2.max_doc] - // ... - let term = Term::wrap(merged_terms.key()); - let current_field = term.field(); - - if last_field != Some(current_field) { - // we reached a new field. - let field_entry = self.schema.get_field_entry(current_field); - // ... set segment postings option the new field. - segment_postings_option = field_entry - .field_type() - .get_segment_postings_option() - .expect("Encountered a field that is not supposed to be + // we reached a new field. + let field_entry = self.schema.get_field_entry(current_field); + // ... set segment postings option the new field. + let segment_postings_option = field_entry + .field_type() + .get_segment_postings_option() + .expect("Encountered a field that is not supposed to be indexed. Have you modified the schema?"); - last_field = Some(current_field); + loop { + // this loops processes a field. + { + let term = Term::wrap(merged_terms.key()); - // it is perfectly safe to call `.new_field` - // even if there is no postings associated. - serializer.new_field(current_field); - } + // Let's compute the list of non-empty posting lists + let segment_postings: Vec<_> = merged_terms + .current_kvs() + .iter() + .flat_map(|heap_item| { + let segment_ord = heap_item.segment_ord; + let term_info = heap_item.streamer.value(); + let segment_reader = &self.readers[heap_item.segment_ord]; + let mut segment_postings = + segment_reader + .read_postings_from_terminfo(term_info, segment_postings_option); + if segment_postings.advance() { + Some((segment_ord, segment_postings)) + } else { + None + } + }) + .collect(); - // Let's compute the list of non-empty posting lists - let segment_postings: Vec<_> = merged_terms - .current_kvs() - .iter() - .flat_map(|heap_item| { - let segment_ord = heap_item.segment_ord; - let term_info = heap_item.streamer.value(); - let segment_reader = &self.readers[heap_item.segment_ord]; - let mut segment_postings = - segment_reader - .read_postings_from_terminfo(term_info, segment_postings_option); - if segment_postings.advance() { - Some((segment_ord, segment_postings)) - } else { - None + // At this point, `segment_postings` contains the posting list + // of all of the segments containing the given term. + // + // These segments are non-empty and advance has already been called. + + if !segment_postings.is_empty() { + // If not, the `term` will be entirely removed. + + // We know that there is at least one document containing + // the term, so we add it. + field_serializer.new_term(term.as_ref())?; + + // We can now serialize this postings, by pushing each document to the + // postings serializer. + + for (segment_ord, mut segment_postings) in segment_postings { + let old_to_new_doc_id = &merged_doc_id_map[segment_ord]; + loop { + // `.advance()` has been called once before the loop. + // Hence we cannot use a `while segment_postings.advance()` loop. + if let Some(remapped_doc_id) = + old_to_new_doc_id[segment_postings.doc() as usize] { + // we make sure to only write the term iff + // there is at least one document. + let positions: &[u32] = segment_postings.positions(); + let term_freq = segment_postings.term_freq(); + let delta_positions = delta_computer.compute_delta(positions); + field_serializer + .write_doc(remapped_doc_id, term_freq, delta_positions)?; + } + if !segment_postings.advance() { + break; + } + } + } + + // closing the term. + field_serializer.close_term()?; } - }) - .collect(); - // At this point, `segment_postings` contains the posting list - // of all of the segments containing the given term. - // - // These segments are non-empty and advance has already been called. + } - if segment_postings.is_empty() { - // by continuing here, the `term` will be entirely removed. - continue; - } - // We know that there is at least one document containing - // the term, so we add it. - serializer.new_term(term.as_ref())?; + if !merged_terms.advance() { + return Ok(()) + } - // We can now serialize this postings, by pushing each document to the - // postings serializer. - - for (segment_ord, mut segment_postings) in segment_postings { - let old_to_new_doc_id = &merged_doc_id_map[segment_ord]; - loop { - // `.advance()` has been called once before the loop. - // Hence we cannot use a `while segment_postings.advance()` loop. - if let Some(remapped_doc_id) = - old_to_new_doc_id[segment_postings.doc() as usize] { - // we make sure to only write the term iff - // there is at least one document. - let positions: &[u32] = segment_postings.positions(); - let term_freq = segment_postings.term_freq(); - let delta_positions = delta_computer.compute_delta(positions); - serializer - .write_doc(remapped_doc_id, term_freq, delta_positions)?; - } - if !segment_postings.advance() { + { + let next_term_field = Term::wrap(merged_terms.key()).field(); + if next_term_field != current_field { + current_field = next_term_field; break; } } } - - // closing the term. - serializer.close_term()?; } - Ok(()) } fn write_storable_fields(&self, store_writer: &mut StoreWriter) -> Result<()> { @@ -319,9 +330,9 @@ impl IndexMerger { let store_reader = reader.get_store_reader(); for doc_id in 0..reader.max_doc() { if !reader.is_deleted(doc_id) { - let doc = try!(store_reader.get(doc_id)); + let doc = store_reader.get(doc_id)?; let field_values: Vec<&FieldValue> = doc.field_values().iter().collect(); - try!(store_writer.store(&field_values)); + store_writer.store(&field_values)?; } } } @@ -331,11 +342,11 @@ impl IndexMerger { impl SerializableSegment for IndexMerger { fn write(&self, mut serializer: SegmentSerializer) -> Result { - try!(self.write_postings(serializer.get_postings_serializer())); - try!(self.write_fieldnorms(serializer.get_fieldnorms_serializer())); - try!(self.write_fast_fields(serializer.get_fast_field_serializer())); - try!(self.write_storable_fields(serializer.get_store_writer())); - try!(serializer.close()); + self.write_postings(serializer.get_postings_serializer())?; + self.write_fieldnorms(serializer.get_fieldnorms_serializer())?; + self.write_fast_fields(serializer.get_fast_field_serializer())?; + self.write_storable_fields(serializer.get_store_writer())?; + serializer.close()?; Ok(self.max_doc) } } diff --git a/src/postings/mod.rs b/src/postings/mod.rs index 06e893646..fd78cbded 100644 --- a/src/postings/mod.rs +++ b/src/postings/mod.rs @@ -17,8 +17,9 @@ mod segment_postings_option; pub use self::docset::{SkipResult, DocSet}; use self::recorder::{Recorder, NothingRecorder, TermFrequencyRecorder, TFAndPositionRecorder}; -pub use self::serializer::InvertedIndexSerializer; +pub use self::serializer::{InvertedIndexSerializer, FieldSerializer}; pub(crate) use self::postings_writer::MultiFieldPostingsWriter; + pub use self::term_info::TermInfo; pub use self::postings::Postings; @@ -59,13 +60,15 @@ mod tests { let index = Index::create_in_ram(schema); let mut segment = index.new_segment(); let mut posting_serializer = InvertedIndexSerializer::open(&mut segment).unwrap(); - posting_serializer.new_field(text_field); - posting_serializer.new_term("abc".as_bytes()).unwrap(); - for doc_id in 0u32..120u32 { - let delta_positions = vec![1, 2, 3, 2]; - posting_serializer.write_doc(doc_id, 2, &delta_positions).unwrap(); + { + let mut field_serializer = posting_serializer.new_field(text_field); + field_serializer.new_term("abc".as_bytes()).unwrap(); + for doc_id in 0u32..120u32 { + let delta_positions = vec![1, 2, 3, 2]; + field_serializer.write_doc(doc_id, 2, &delta_positions).unwrap(); + } + field_serializer.close_term().unwrap(); } - posting_serializer.close_term().unwrap(); posting_serializer.close().unwrap(); let read = segment.open_read(SegmentComponent::POSITIONS).unwrap(); assert!(read.len() <= 140); diff --git a/src/postings/postings_writer.rs b/src/postings/postings_writer.rs index 0a995a889..813073b4c 100644 --- a/src/postings/postings_writer.rs +++ b/src/postings/postings_writer.rs @@ -1,7 +1,7 @@ use DocId; use schema::Term; use schema::FieldValue; -use postings::InvertedIndexSerializer; +use postings::{InvertedIndexSerializer, FieldSerializer}; use std::io; use postings::Recorder; use analyzer::SimpleTokenizer; @@ -101,8 +101,8 @@ impl<'a> MultiFieldPostingsWriter<'a> { let (field, start) = offsets[i]; let (_, stop) = offsets[i + 1]; let postings_writer = &self.per_field_postings_writers[field.0 as usize]; - postings_writer - .serialize(field, &term_offsets[start..stop], serializer, self.heap)?; + let field_serializer = serializer.new_field(field); + postings_writer.serialize(&term_offsets[start..stop], field_serializer, self.heap)?; } Ok(()) } @@ -136,9 +136,8 @@ pub trait PostingsWriter { /// Serializes the postings on disk. /// The actual serialization format is handled by the `PostingsSerializer`. fn serialize(&self, - field: Field, term_addrs: &[(&[u8], u32)], - serializer: &mut InvertedIndexSerializer, + serializer: FieldSerializer, heap: &Heap) -> io::Result<()>; @@ -214,17 +213,15 @@ impl<'a, Rec: Recorder + 'static> PostingsWriter for SpecializedPostingsWriter<' } fn serialize(&self, - field: Field, term_addrs: &[(&[u8], u32)], - serializer: &mut InvertedIndexSerializer, + mut serializer: FieldSerializer, heap: &Heap) -> io::Result<()> { - serializer.new_field(field); for &(term_bytes, addr) in term_addrs { let recorder: &mut Rec = self.heap.get_mut_ref(addr); - try!(serializer.new_term(term_bytes)); - try!(recorder.serialize(addr, serializer, heap)); - try!(serializer.close_term()); + serializer.new_term(term_bytes)?; + recorder.serialize(addr, &mut serializer, heap)?; + serializer.close_term()?; } Ok(()) } diff --git a/src/postings/recorder.rs b/src/postings/recorder.rs index d7f91d35c..dde85d66c 100644 --- a/src/postings/recorder.rs +++ b/src/postings/recorder.rs @@ -1,6 +1,6 @@ use DocId; use std::io; -use postings::InvertedIndexSerializer; +use postings::FieldSerializer; use datastruct::stacker::{ExpUnrolledLinkedList, Heap, HeapAllocable}; const EMPTY_ARRAY: [u32; 0] = [0u32; 0]; @@ -29,7 +29,7 @@ pub trait Recorder: HeapAllocable { /// Pushes the postings information to the serializer. fn serialize(&self, self_addr: u32, - serializer: &mut InvertedIndexSerializer, + serializer: &mut FieldSerializer, heap: &Heap) -> io::Result<()>; } @@ -66,11 +66,11 @@ impl Recorder for NothingRecorder { fn serialize(&self, self_addr: u32, - serializer: &mut InvertedIndexSerializer, + serializer: &mut FieldSerializer, heap: &Heap) -> io::Result<()> { for doc in self.stack.iter(self_addr, heap) { - try!(serializer.write_doc(doc, 0u32, &EMPTY_ARRAY)); + serializer.write_doc(doc, 0u32, &EMPTY_ARRAY)?; } Ok(()) } @@ -118,7 +118,7 @@ impl Recorder for TermFrequencyRecorder { fn serialize(&self, self_addr: u32, - serializer: &mut InvertedIndexSerializer, + serializer: &mut FieldSerializer, heap: &Heap) -> io::Result<()> { // the last document has not been closed... @@ -173,7 +173,7 @@ impl Recorder for TFAndPositionRecorder { fn serialize(&self, self_addr: u32, - serializer: &mut InvertedIndexSerializer, + serializer: &mut FieldSerializer, heap: &Heap) -> io::Result<()> { let mut doc_positions = Vec::with_capacity(100); @@ -189,7 +189,7 @@ impl Recorder for TFAndPositionRecorder { prev_position = position; } } - try!(serializer.write_doc(doc, doc_positions.len() as u32, &doc_positions)); + serializer.write_doc(doc, doc_positions.len() as u32, &doc_positions)?; } Ok(()) } diff --git a/src/postings/serializer.rs b/src/postings/serializer.rs index c4b9c1146..c3f5f101e 100644 --- a/src/postings/serializer.rs +++ b/src/postings/serializer.rs @@ -52,14 +52,182 @@ pub struct InvertedIndexSerializer { postings_serializer: PostingsSerializer, positions_serializer: PositionSerializer, schema: Schema, - - term_open: bool, - text_indexing_options: TextIndexingOptions, - - current_term_info: TermInfo, - } + +impl InvertedIndexSerializer { + /// Open a new `PostingsSerializer` for the given segment + fn new(terms_write: WritePtr, + postings_write: WritePtr, + positions_write: WritePtr, + schema: Schema) + -> Result { + let terms_fst_builder = TermDictionaryBuilderImpl::new(terms_write)?; + Ok(InvertedIndexSerializer { + terms_fst_builder: terms_fst_builder, + positions_serializer: PositionSerializer::new(positions_write), + postings_serializer: PostingsSerializer::new(postings_write), + schema: schema, + }) + } + + + /// Open a new `PostingsSerializer` for the given segment + pub fn open(segment: &mut Segment) -> Result { + use SegmentComponent::{TERMS, POSTINGS, POSITIONS}; + InvertedIndexSerializer::new(segment.open_write(TERMS)?, + segment.open_write(POSTINGS)?, + segment.open_write(POSITIONS)?, + segment.schema()) + } + + /// Must be called before starting pushing terms of + /// a given field. + /// + /// Loads the indexing options for the given field. + pub fn new_field(&mut self, field: Field) -> FieldSerializer { + let field_entry: &FieldEntry = self.schema.get_field_entry(field); + let text_indexing_options = match *field_entry.field_type() { + FieldType::Str(ref text_options) => text_options.get_indexing_options(), + FieldType::U64(ref int_options) | + FieldType::I64(ref int_options) => { + if int_options.is_indexed() { + TextIndexingOptions::Unindexed + } else { + TextIndexingOptions::Untokenized + } + } + }; + FieldSerializer::new( + text_indexing_options, + &mut self.terms_fst_builder, + &mut self.postings_serializer, + &mut self.positions_serializer, + ) + } + + /// Closes the serializer. + pub fn close(self) -> io::Result<()> { + self.terms_fst_builder.finish()?; + self.postings_serializer.close()?; + self.positions_serializer.close()?; + Ok(()) + } +} + + +/* +let field_entry: &FieldEntry = self.schema.get_field_entry(field); +self.text_indexing_options = match *field_entry.field_type() { + FieldType::Str(ref text_options) => text_options.get_indexing_options(), + FieldType::U64(ref int_options) | + FieldType::I64(ref int_options) => { + if int_options.is_indexed() { + TextIndexingOptions::Unindexed + } else { + TextIndexingOptions::Untokenized + } + } +}; +self.postings_serializer.set_termfreq_enabled(self.text_indexing_options.is_termfreq_enabled()); + + */ + +pub struct FieldSerializer<'a> { + text_indexing_options: TextIndexingOptions, + terms_fst_builder: &'a mut TermDictionaryBuilderImpl, + postings_serializer: &'a mut PostingsSerializer, + positions_serializer: &'a mut PositionSerializer, + current_term_info: TermInfo, + term_open: bool, +} + + +impl<'a> FieldSerializer<'a> { + + fn new( + text_indexing_options: TextIndexingOptions, + terms_fst_builder: &'a mut TermDictionaryBuilderImpl, + postings_serializer: &'a mut PostingsSerializer, + positions_serializer: &'a mut PositionSerializer + ) -> FieldSerializer<'a> { + + postings_serializer.set_termfreq_enabled(text_indexing_options.is_termfreq_enabled()); + + FieldSerializer { + text_indexing_options: text_indexing_options, + terms_fst_builder: terms_fst_builder, + postings_serializer: postings_serializer, + positions_serializer: positions_serializer, + current_term_info: TermInfo::default(), + term_open: false, + } + } + + fn current_term_info(&self) -> TermInfo { + let (filepos, offset) = self.positions_serializer.addr(); + TermInfo { + doc_freq: 0, + postings_offset: self.postings_serializer.addr(), + positions_offset: filepos, + positions_inner_offset: offset, + } + } + + /// Starts the postings for a new term. + /// * term - the term. It needs to come after the previous term according + /// to the lexicographical order. + /// * doc_freq - return the number of document containing the term. + pub fn new_term(&mut self, term: &[u8]) -> io::Result<()> { + if self.term_open { + panic!("Called new_term, while the previous term was not closed."); + } + self.term_open = true; + self.postings_serializer.clear(); + self.current_term_info = self.current_term_info(); + self.terms_fst_builder.insert_key(term) + } + + /// Serialize the information that a document contains the current term, + /// its term frequency, and the position deltas. + /// + /// At this point, the positions are already `delta-encoded`. + /// For instance, if the positions are `2, 3, 17`, + /// `position_deltas` is `2, 1, 14` + /// + /// Term frequencies and positions may be ignored by the serializer depending + /// on the configuration of the field in the `Schema`. + pub fn write_doc(&mut self, + doc_id: DocId, + term_freq: u32, + position_deltas: &[u32]) + -> io::Result<()> { + self.current_term_info.doc_freq += 1; + self.postings_serializer.write_doc(doc_id, term_freq)?; + if self.text_indexing_options.is_position_enabled() { + self.positions_serializer.write(position_deltas)?; + } + Ok(()) + } + + /// Finish the serialization for this term postings. + /// + /// If the current block is incomplete, it need to be encoded + /// using `VInt` encoding. + pub fn close_term(&mut self) -> io::Result<()> { + if self.term_open { + self.terms_fst_builder.insert_value(&self.current_term_info)?; + self.postings_serializer.close_term()?; + self.term_open = false; + } + Ok(()) + } +} + +// TODO is the last term always closed? + + + struct PostingsSerializer { postings_write: CountingWriter, last_doc_id_encoded: u32, @@ -204,122 +372,3 @@ impl PositionSerializer { } } -impl InvertedIndexSerializer { - /// Open a new `PostingsSerializer` for the given segment - pub fn new(terms_write: WritePtr, - postings_write: WritePtr, - positions_write: WritePtr, - schema: Schema) - -> Result { - let terms_fst_builder = TermDictionaryBuilderImpl::new(terms_write)?; - Ok(InvertedIndexSerializer { - terms_fst_builder: terms_fst_builder, - positions_serializer: PositionSerializer::new(positions_write), - postings_serializer: PostingsSerializer::new(postings_write), - schema: schema, - term_open: false, - current_term_info: TermInfo::default(), - text_indexing_options: TextIndexingOptions::Untokenized, - }) - } - - - /// Open a new `PostingsSerializer` for the given segment - pub fn open(segment: &mut Segment) -> Result { - use SegmentComponent::{TERMS, POSTINGS, POSITIONS}; - InvertedIndexSerializer::new(segment.open_write(TERMS)?, - segment.open_write(POSTINGS)?, - segment.open_write(POSITIONS)?, - segment.schema()) - } - - /// Must be called before starting pushing terms of - /// a given field. - /// - /// Loads the indexing options for the given field. - pub fn new_field(&mut self, field: Field) { - let field_entry: &FieldEntry = self.schema.get_field_entry(field); - self.text_indexing_options = match *field_entry.field_type() { - FieldType::Str(ref text_options) => text_options.get_indexing_options(), - FieldType::U64(ref int_options) | - FieldType::I64(ref int_options) => { - if int_options.is_indexed() { - TextIndexingOptions::Unindexed - } else { - TextIndexingOptions::Untokenized - } - } - }; - self.postings_serializer.set_termfreq_enabled(self.text_indexing_options.is_termfreq_enabled()); - } - - fn current_term_info(&self) -> TermInfo { - let (filepos, offset) = self.positions_serializer.addr(); - TermInfo { - doc_freq: 0, - postings_offset: self.postings_serializer.addr(), - positions_offset: filepos, - positions_inner_offset: offset, - } - } - - /// Starts the postings for a new term. - /// * term - the term. It needs to come after the previous term according - /// to the lexicographical order. - /// * doc_freq - return the number of document containing the term. - pub fn new_term(&mut self, term: &[u8]) -> io::Result<()> { - if self.term_open { - panic!("Called new_term, while the previous term was not closed."); - } - self.term_open = true; - self.postings_serializer.clear(); - self.current_term_info = self.current_term_info(); - self.terms_fst_builder.insert_key(term) - } - - /// Finish the serialization for this term postings. - /// - /// If the current block is incomplete, it need to be encoded - /// using `VInt` encoding. - pub fn close_term(&mut self) -> io::Result<()> { - if self.term_open { - self.terms_fst_builder.insert_value(&self.current_term_info)?; - self.postings_serializer.close_term()?; - self.term_open = false; - } - Ok(()) - } - - - /// Serialize the information that a document contains the current term, - /// its term frequency, and the position deltas. - /// - /// At this point, the positions are already `delta-encoded`. - /// For instance, if the positions are `2, 3, 17`, - /// `position_deltas` is `2, 1, 14` - /// - /// Term frequencies and positions may be ignored by the serializer depending - /// on the configuration of the field in the `Schema`. - pub fn write_doc(&mut self, - doc_id: DocId, - term_freq: u32, - position_deltas: &[u32]) - -> io::Result<()> { - self.current_term_info.doc_freq += 1; - self.postings_serializer.write_doc(doc_id, term_freq)?; - if self.text_indexing_options.is_position_enabled() { - self.positions_serializer.write(position_deltas)?; - } - - Ok(()) - } - - /// Closes the serializer. - pub fn close(mut self) -> io::Result<()> { - self.close_term()?; - self.terms_fst_builder.finish()?; - self.postings_serializer.close()?; - self.positions_serializer.close()?; - Ok(()) - } -}