rust/tantivy
1
0
Fork 0
mirror of https://github.com/quickwit-oss/tantivy.git synced 2026-08-18 12:08:22 +00:00
Code Issues Packages Projects Releases Wiki Activity

introducing a field serializer

Browse Source
This commit is contained in:
Paul Masurel
2017-08-11 18:11:32 +09:00
parent 1e89f86267
commit 8f377b92d0
5 changed files with 298 additions and 238 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/indexer/merger.rs
+99 -88
View File
@@ -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<Field> = 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<u32> {
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)
}
}
src/postings/mod.rs
+10 -7
View File
@@ -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);
src/postings/postings_writer.rs
+8 -11
View File
@@ -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(())
}
src/postings/recorder.rs
+7 -7
View File
@@ -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(())
}
src/postings/serializer.rs
+174 -125
View File
@@ -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<InvertedIndexSerializer> {
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<InvertedIndexSerializer> {
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<WritePtr, TermInfo>,
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<WritePtr, TermInfo>,
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<WritePtr>,
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<InvertedIndexSerializer> {
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<InvertedIndexSerializer> {
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(())
}
}