diff --git a/Cargo.toml b/Cargo.toml index 28417a40e..9d568271f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -53,6 +53,7 @@ rayon = "1.5" lru = "0.6.5" fastdivide = "0.3" itertools = "0.10.0" +measure_time = "0.7.0" [target.'cfg(windows)'.dependencies] winapi = "0.3.9" diff --git a/fastfield_codecs/src/lib.rs b/fastfield_codecs/src/lib.rs index a79ebc371..a8d907ed6 100644 --- a/fastfield_codecs/src/lib.rs +++ b/fastfield_codecs/src/lib.rs @@ -51,14 +51,14 @@ pub trait FastFieldCodecSerializer { /// FastFieldDataAccess is the trait to access fast field data during serialization and estimation. pub trait FastFieldDataAccess { - /// Return the value associated to the given document. + /// Return the value associated to the given position. /// /// Whenever possible use the Iterator passed to the fastfield creation instead, for performance reasons. /// /// # Panics /// - /// May panic if `doc` is greater than the segment - fn get(&self, doc: u32) -> u64; + /// May panic if `position` is greater than the index. + fn get_val(&self, position: u64) -> u64; } #[derive(Debug, Clone)] @@ -69,20 +69,14 @@ pub struct FastFieldStats { } impl<'a> FastFieldDataAccess for &'a [u64] { - fn get(&self, doc: u32) -> u64 { - self[doc as usize] - } -} - -impl<'a> FastFieldDataAccess for &'a Vec { - fn get(&self, doc: u32) -> u64 { - self[doc as usize] + fn get_val(&self, position: u64) -> u64 { + self[position as usize] } } impl FastFieldDataAccess for Vec { - fn get(&self, doc: u32) -> u64 { - self[doc as usize] + fn get_val(&self, position: u64) -> u64 { + self[position as usize] } } diff --git a/fastfield_codecs/src/linearinterpol.rs b/fastfield_codecs/src/linearinterpol.rs index a9acaa3f7..3a80656a1 100644 --- a/fastfield_codecs/src/linearinterpol.rs +++ b/fastfield_codecs/src/linearinterpol.rs @@ -123,8 +123,8 @@ impl FastFieldCodecSerializer for LinearInterpolFastFieldSerializer { ) -> io::Result<()> { assert!(stats.min_value <= stats.max_value); - let first_val = fastfield_accessor.get(0); - let last_val = fastfield_accessor.get(stats.num_vals as u32 - 1); + let first_val = fastfield_accessor.get_val(0); + let last_val = fastfield_accessor.get_val(stats.num_vals as u64 - 1); let slope = get_slope(first_val, last_val, stats.num_vals); // calculate offset to ensure all values are positive let mut offset = 0; @@ -191,8 +191,8 @@ impl FastFieldCodecSerializer for LinearInterpolFastFieldSerializer { /// where the local maxima for the deviation of the calculated value are and /// the offset to shift all values to >=0 is also unknown. fn estimate(fastfield_accessor: &impl FastFieldDataAccess, stats: FastFieldStats) -> f32 { - let first_val = fastfield_accessor.get(0); - let last_val = fastfield_accessor.get(stats.num_vals as u32 - 1); + let first_val = fastfield_accessor.get_val(0); + let last_val = fastfield_accessor.get_val(stats.num_vals as u64 - 1); let slope = get_slope(first_val, last_val, stats.num_vals); // let's sample at 0%, 5%, 10% .. 95%, 100% @@ -205,7 +205,7 @@ impl FastFieldCodecSerializer for LinearInterpolFastFieldSerializer { .iter() .map(|pos| { let calculated_value = get_calculated_value(first_val, *pos as u64, slope); - let actual_value = fastfield_accessor.get(*pos as u32); + let actual_value = fastfield_accessor.get_val(*pos as u64); distance(calculated_value, actual_value) }) .max() diff --git a/fastfield_codecs/src/multilinearinterpol.rs b/fastfield_codecs/src/multilinearinterpol.rs index 811b5fbdf..45dfde5c2 100644 --- a/fastfield_codecs/src/multilinearinterpol.rs +++ b/fastfield_codecs/src/multilinearinterpol.rs @@ -196,8 +196,8 @@ impl FastFieldCodecSerializer for MultiLinearInterpolFastFieldSerializer { ) -> io::Result<()> { assert!(stats.min_value <= stats.max_value); - let first_val = fastfield_accessor.get(0); - let last_val = fastfield_accessor.get(stats.num_vals as u32 - 1); + let first_val = fastfield_accessor.get_val(0); + let last_val = fastfield_accessor.get_val(stats.num_vals as u64 - 1); let mut first_function = Function { end_pos: stats.num_vals, @@ -309,9 +309,10 @@ impl FastFieldCodecSerializer for MultiLinearInterpolFastFieldSerializer { /// where the local maxima are for the deviation of the calculated value and /// the offset is also unknown. fn estimate(fastfield_accessor: &impl FastFieldDataAccess, stats: FastFieldStats) -> f32 { - let first_val_in_first_block = fastfield_accessor.get(0); + let first_val_in_first_block = fastfield_accessor.get_val(0); let last_elem_in_first_chunk = CHUNK_SIZE.min(stats.num_vals); - let last_val_in_first_block = fastfield_accessor.get(last_elem_in_first_chunk as u32 - 1); + let last_val_in_first_block = + fastfield_accessor.get_val(last_elem_in_first_chunk as u64 - 1); let slope = get_slope( first_val_in_first_block, last_val_in_first_block, @@ -328,7 +329,7 @@ impl FastFieldCodecSerializer for MultiLinearInterpolFastFieldSerializer { .map(|pos| { let calculated_value = get_calculated_value(first_val_in_first_block, *pos as u64, slope); - let actual_value = fastfield_accessor.get(*pos as u32); + let actual_value = fastfield_accessor.get_val(*pos as u64); distance(calculated_value, actual_value) }) .max() diff --git a/src/core/segment_reader.rs b/src/core/segment_reader.rs index 0ade28f05..015d8c1ee 100644 --- a/src/core/segment_reader.rs +++ b/src/core/segment_reader.rs @@ -1,4 +1,3 @@ -use crate::common::HasLen; use crate::core::InvertedIndexReader; use crate::core::Segment; use crate::core::SegmentComponent; @@ -63,11 +62,9 @@ impl SegmentReader { self.max_doc } - /// Returns the number of documents. + /// Returns the number of alive documents. /// Deleted documents are not counted. /// - /// Today, `tantivy` does not handle deletes so max doc and - /// num_docs are the same. pub fn num_docs(&self) -> DocId { self.num_docs } @@ -81,7 +78,7 @@ impl SegmentReader { /// deleted in the segment. pub fn num_deleted_docs(&self) -> DocId { self.delete_bitset() - .map(|delete_set| delete_set.len() as DocId) + .map(|delete_set| delete_set.num_deleted() as DocId) .unwrap_or(0u32) } @@ -329,6 +326,32 @@ mod test { use crate::schema::{Schema, Term, STORED, TEXT}; use crate::DocId; + #[test] + fn test_num_alive() -> crate::Result<()> { + let mut schema_builder = Schema::builder(); + schema_builder.add_text_field("name", TEXT | STORED); + let schema = schema_builder.build(); + let index = Index::create_in_ram(schema.clone()); + let name = schema.get_field("name").unwrap(); + + { + let mut index_writer = index.writer_for_tests()?; + index_writer.add_document(doc!(name => "tantivy")); + index_writer.add_document(doc!(name => "horse")); + index_writer.add_document(doc!(name => "jockey")); + index_writer.add_document(doc!(name => "cap")); + // we should now have one segment with two docs + index_writer.delete_term(Term::from_field_text(name, "horse")); + index_writer.delete_term(Term::from_field_text(name, "cap")); + + // ok, now we should have a deleted doc + index_writer.commit()?; + } + let searcher = index.reader()?.searcher(); + assert_eq!(2, searcher.segment_reader(0).num_docs()); + assert_eq!(4, searcher.segment_reader(0).max_doc()); + Ok(()) + } #[test] fn test_alive_docs_iterator() -> crate::Result<()> { let mut schema_builder = Schema::builder(); diff --git a/src/fastfield/delete.rs b/src/fastfield/delete.rs index de23effac..a2f14aa7c 100644 --- a/src/fastfield/delete.rs +++ b/src/fastfield/delete.rs @@ -91,6 +91,10 @@ impl DeleteBitSet { b & (1u8 << shift) != 0 } + /// The number of deleted docs + pub fn num_deleted(&self) -> usize { + self.num_deleted + } /// Summarize total space usage of this bitset. pub fn space_usage(&self) -> ByteCount { self.data.len() diff --git a/src/fastfield/multivalued/reader.rs b/src/fastfield/multivalued/reader.rs index ad6c99da6..60691a76f 100644 --- a/src/fastfield/multivalued/reader.rs +++ b/src/fastfield/multivalued/reader.rs @@ -1,8 +1,6 @@ use std::ops::Range; -use crate::fastfield::{ - BitpackedFastFieldReader, DynamicFastFieldReader, FastFieldReader, FastValue, MultiValueLength, -}; +use crate::fastfield::{DynamicFastFieldReader, FastFieldReader, FastValue, MultiValueLength}; use crate::DocId; /// Reader for a multivalued `u64` fast field. @@ -16,13 +14,13 @@ use crate::DocId; #[derive(Clone)] pub struct MultiValuedFastFieldReader { idx_reader: DynamicFastFieldReader, - vals_reader: BitpackedFastFieldReader, + vals_reader: DynamicFastFieldReader, } impl MultiValuedFastFieldReader { pub(crate) fn open( idx_reader: DynamicFastFieldReader, - vals_reader: BitpackedFastFieldReader, + vals_reader: DynamicFastFieldReader, ) -> MultiValuedFastFieldReader { MultiValuedFastFieldReader { idx_reader, @@ -32,6 +30,7 @@ impl MultiValuedFastFieldReader { /// Returns `(start, stop)`, such that the values associated /// to the given document are `start..stop`. + #[inline] fn range(&self, doc: DocId) -> Range { let start = self.idx_reader.get(doc); let stop = self.idx_reader.get(doc + 1); @@ -39,11 +38,12 @@ impl MultiValuedFastFieldReader { } /// Returns the array of values associated to the given `doc`. + #[inline] pub fn get_vals(&self, doc: DocId, vals: &mut Vec) { let range = self.range(doc); let len = (range.end - range.start) as usize; vals.resize(len, Item::make_zero()); - self.vals_reader.get_range_u64(range.start, &mut vals[..]); + self.vals_reader.get_range(range.start, &mut vals[..]); } /// Returns the minimum value for this fast field. @@ -65,12 +65,14 @@ impl MultiValuedFastFieldReader { } /// Returns the number of values associated with the document `DocId`. + #[inline] pub fn num_vals(&self, doc: DocId) -> usize { let range = self.range(doc); (range.end - range.start) as usize } /// Returns the overall number of values in this field . + #[inline] pub fn total_num_vals(&self) -> u64 { self.idx_reader.max_value() } diff --git a/src/fastfield/reader.rs b/src/fastfield/reader.rs index 30ae0aba8..47adcddfb 100644 --- a/src/fastfield/reader.rs +++ b/src/fastfield/reader.rs @@ -44,7 +44,7 @@ pub trait FastFieldReader: Clone { /// /// May panic if `start + output.len()` is greater than /// the segment's `maxdoc`. - fn get_range(&self, start: DocId, output: &mut [Item]); + fn get_range(&self, start: u64, output: &mut [Item]); /// Returns the minimum value for this fast field. /// @@ -120,7 +120,7 @@ impl FastFieldReader for DynamicFastFieldReader { Self::MultiLinearInterpol(reader) => reader.get(doc), } } - fn get_range(&self, start: DocId, output: &mut [Item]) { + fn get_range(&self, start: u64, output: &mut [Item]) { match self { Self::Bitpacked(reader) => reader.get_range(start, output), Self::LinearInterpol(reader) => reader.get_range(start, output), @@ -226,8 +226,8 @@ impl FastFieldReader /// /// May panic if `start + output.len()` is greater than /// the segment's `maxdoc`. - fn get_range(&self, start: DocId, output: &mut [Item]) { - self.get_range_u64(u64::from(start), output); + fn get_range(&self, start: u64, output: &mut [Item]) { + self.get_range_u64(start, output); } /// Returns the minimum value for this fast field. diff --git a/src/fastfield/readers.rs b/src/fastfield/readers.rs index ccd91b61f..f202bbb35 100644 --- a/src/fastfield/readers.rs +++ b/src/fastfield/readers.rs @@ -99,12 +99,19 @@ impl FastFieldReaders { Ok(()) } + pub(crate) fn typed_fast_field_reader_with_idx( + &self, + field: Field, + index: usize, + ) -> crate::Result> { + let fast_field_slice = self.fast_field_data(field, index)?; + DynamicFastFieldReader::open(fast_field_slice) + } pub(crate) fn typed_fast_field_reader( &self, field: Field, ) -> crate::Result> { - let fast_field_slice = self.fast_field_data(field, 0)?; - DynamicFastFieldReader::open(fast_field_slice) + self.typed_fast_field_reader_with_idx(field, 0) } pub(crate) fn typed_fast_field_multi_reader( @@ -112,9 +119,7 @@ impl FastFieldReaders { field: Field, ) -> crate::Result> { let idx_reader = self.typed_fast_field_reader(field)?; - let fast_field_slice_vals = self.fast_field_data(field, 1)?; - let vals_reader: BitpackedFastFieldReader = - BitpackedFastFieldReader::open(fast_field_slice_vals)?; + let vals_reader = self.typed_fast_field_reader_with_idx(field, 1)?; Ok(MultiValuedFastFieldReader::open(idx_reader, vals_reader)) } diff --git a/src/fastfield/serializer/mod.rs b/src/fastfield/serializer/mod.rs index d297c65ff..a4c9fc090 100644 --- a/src/fastfield/serializer/mod.rs +++ b/src/fastfield/serializer/mod.rs @@ -71,7 +71,26 @@ impl CompositeFastFieldSerializer { data_iter_1: impl Iterator, data_iter_2: impl Iterator, ) -> io::Result<()> { - let field_write = self.composite_write.for_field_with_idx(field, 0); + self.create_auto_detect_u64_fast_field_with_idx( + field, + stats, + fastfield_accessor, + data_iter_1, + data_iter_2, + 0, + ) + } + /// Serialize data into a new u64 fast field. The best compression codec will be chosen automatically. + pub fn create_auto_detect_u64_fast_field_with_idx( + &mut self, + field: Field, + stats: FastFieldStats, + fastfield_accessor: impl FastFieldDataAccess, + data_iter_1: impl Iterator, + data_iter_2: impl Iterator, + idx: usize, + ) -> io::Result<()> { + let field_write = self.composite_write.for_field_with_idx(field, idx); let mut estimations = vec![]; diff --git a/src/fastfield/writer.rs b/src/fastfield/writer.rs index a3b3bdb79..9e9893454 100644 --- a/src/fastfield/writer.rs +++ b/src/fastfield/writer.rs @@ -7,7 +7,6 @@ use crate::indexer::doc_id_mapping::DocIdMapping; use crate::postings::UnorderedTermId; use crate::schema::{Cardinality, Document, Field, FieldEntry, FieldType, Schema}; use crate::termdict::TermOrdinal; -use crate::DocId; use fnv::FnvHashMap; use std::collections::HashMap; use std::io; @@ -323,16 +322,17 @@ struct WriterFastFieldAccessProvider<'map, 'bitp> { vals: &'bitp BlockedBitpacker, } impl<'map, 'bitp> FastFieldDataAccess for WriterFastFieldAccessProvider<'map, 'bitp> { - /// Return the value associated to the given document. + /// Return the value associated to the given doc. /// - /// This accessor should return as fast as possible. + /// Whenever possible use the Iterator passed to the fastfield creation instead, for performance reasons. /// /// # Panics /// - /// May panic if `doc` is greater than the segment - fn get(&self, doc: DocId) -> u64 { + /// May panic if `doc` is greater than the index. + fn get_val(&self, doc: u64) -> u64 { if let Some(doc_id_map) = self.doc_id_map { - self.vals.get(doc_id_map.get_old_doc_id(doc) as usize) // consider extra FastFieldReader wrapper for non doc_id_map + self.vals + .get(doc_id_map.get_old_doc_id(doc as u32) as usize) // consider extra FastFieldReader wrapper for non doc_id_map } else { self.vals.get(doc as usize) } diff --git a/src/indexer/doc_id_mapping.rs b/src/indexer/doc_id_mapping.rs index 9b1a3e6a4..155211e55 100644 --- a/src/indexer/doc_id_mapping.rs +++ b/src/indexer/doc_id_mapping.rs @@ -2,13 +2,61 @@ //! to get mappings from old doc_id to new doc_id and vice versa, after sorting //! -use super::SegmentWriter; +use super::{merger::SegmentReaderWithOrdinal, SegmentWriter}; use crate::{ schema::{Field, Schema}, DocId, IndexSortByField, Order, TantivyError, }; -use std::cmp::Reverse; -/// Struct to provide mapping from old doc_id to new doc_id and vice versa +use std::{cmp::Reverse, ops::Index}; + +/// Struct to provide mapping from new doc_id to old doc_id and segment. +#[derive(Clone)] +pub(crate) struct SegmentDocidMapping<'a> { + new_doc_id_to_old_and_segment: Vec<(DocId, SegmentReaderWithOrdinal<'a>)>, + is_trivial: bool, +} + +impl<'a> SegmentDocidMapping<'a> { + pub(crate) fn new( + new_doc_id_to_old_and_segment: Vec<(DocId, SegmentReaderWithOrdinal<'a>)>, + is_trivial: bool, + ) -> Self { + Self { + new_doc_id_to_old_and_segment, + is_trivial, + } + } + pub(crate) fn iter(&self) -> impl Iterator { + self.new_doc_id_to_old_and_segment.iter() + } + pub(crate) fn len(&self) -> usize { + self.new_doc_id_to_old_and_segment.len() + } + /// This flags means the segments are simply stacked in the order of their ordinal. + /// e.g. [(0, 1), .. (n, 1), (0, 2)..., (m, 2)] + /// + /// This allows for some optimization. + pub(crate) fn is_trivial(&self) -> bool { + self.is_trivial + } +} +impl<'a> Index for SegmentDocidMapping<'a> { + type Output = (DocId, SegmentReaderWithOrdinal<'a>); + + fn index(&self, idx: usize) -> &Self::Output { + &self.new_doc_id_to_old_and_segment[idx] + } +} +impl<'a> IntoIterator for SegmentDocidMapping<'a> { + type Item = (DocId, SegmentReaderWithOrdinal<'a>); + type IntoIter = std::vec::IntoIter; + + fn into_iter(self) -> Self::IntoIter { + self.new_doc_id_to_old_and_segment.into_iter() + } +} + +/// Struct to provide mapping from old doc_id to new doc_id and vice versa within a segment. pub struct DocIdMapping { new_doc_id_to_old: Vec, old_doc_id_to_new: Vec, diff --git a/src/indexer/merger.rs b/src/indexer/merger.rs index 5810a6e55..5b667cf0c 100644 --- a/src/indexer/merger.rs +++ b/src/indexer/merger.rs @@ -9,6 +9,7 @@ use crate::fastfield::MultiValuedFastFieldReader; use crate::fieldnorm::FieldNormsSerializer; use crate::fieldnorm::FieldNormsWriter; use crate::fieldnorm::{FieldNormReader, FieldNormReaders}; +use crate::indexer::doc_id_mapping::SegmentDocidMapping; use crate::indexer::SegmentSerializer; use crate::postings::Postings; use crate::postings::{InvertedIndexSerializer, SegmentPostings}; @@ -29,6 +30,7 @@ use crate::{ }; use crate::{DocId, InvertedIndexReader, SegmentComponent}; use itertools::Itertools; +use measure_time::debug_time; use std::cmp; use std::collections::HashMap; use std::sync::Arc; @@ -232,33 +234,24 @@ impl IndexMerger { fn write_fieldnorms( &self, mut fieldnorms_serializer: FieldNormsSerializer, - doc_id_mapping: &Option>, + doc_id_mapping: &SegmentDocidMapping, ) -> crate::Result<()> { let fields = FieldNormsWriter::fields_with_fieldnorm(&self.schema); let mut fieldnorms_data = Vec::with_capacity(self.max_doc as usize); for field in fields { fieldnorms_data.clear(); - if let Some(doc_id_mapping) = doc_id_mapping { - let fieldnorms_readers: Vec = self - .readers - .iter() - .map(|reader| reader.get_fieldnorms_reader(field)) - .collect::>()?; - for (doc_id, reader_with_ordinal) in doc_id_mapping { - let fieldnorms_reader = - &fieldnorms_readers[reader_with_ordinal.ordinal as usize]; - let fieldnorm_id = fieldnorms_reader.fieldnorm_id(*doc_id); - fieldnorms_data.push(fieldnorm_id); - } - } else { - for reader in &self.readers { - let fieldnorms_reader = reader.get_fieldnorms_reader(field)?; - for doc_id in reader.doc_ids_alive() { - let fieldnorm_id = fieldnorms_reader.fieldnorm_id(doc_id); - fieldnorms_data.push(fieldnorm_id); - } - } + + let fieldnorms_readers: Vec = self + .readers + .iter() + .map(|reader| reader.get_fieldnorms_reader(field)) + .collect::>()?; + for (doc_id, reader_with_ordinal) in doc_id_mapping.iter() { + let fieldnorms_reader = &fieldnorms_readers[reader_with_ordinal.ordinal as usize]; + let fieldnorm_id = fieldnorms_reader.fieldnorm_id(*doc_id); + fieldnorms_data.push(fieldnorm_id); } + fieldnorms_serializer.serialize_field(field, &fieldnorms_data[..])?; } fieldnorms_serializer.close()?; @@ -269,8 +262,10 @@ impl IndexMerger { &self, fast_field_serializer: &mut CompositeFastFieldSerializer, mut term_ord_mappings: HashMap, - doc_id_mapping: &Option>, + doc_id_mapping: &SegmentDocidMapping, ) -> crate::Result<()> { + debug_time!("write_fast_fields"); + for (field, field_entry) in self.schema.fields() { let field_type = field_entry.field_type(); match field_type { @@ -318,7 +313,7 @@ impl IndexMerger { &self, field: Field, fast_field_serializer: &mut CompositeFastFieldSerializer, - doc_id_mapping: &Option>, + doc_id_mapping: &SegmentDocidMapping, ) -> crate::Result<()> { let (min_value, max_value) = self.readers.iter().map(|reader|{ let u64_reader: DynamicFastFieldReader = reader @@ -343,68 +338,44 @@ impl IndexMerger { u64_reader }) .collect::>(); - if let Some(doc_id_mapping) = doc_id_mapping { - #[derive(Clone)] - struct SortedDocidFieldAccessProvider<'a> { - doc_id_mapping: &'a Vec<(DocId, SegmentReaderWithOrdinal<'a>)>, - fast_field_readers: &'a Vec>, - } - impl<'a> FastFieldDataAccess for SortedDocidFieldAccessProvider<'a> { - fn get(&self, doc: DocId) -> u64 { - let (doc_id, reader_with_ordinal) = self.doc_id_mapping[doc as usize]; - self.fast_field_readers[reader_with_ordinal.ordinal as usize].get(doc_id) - } - } - let stats = FastFieldStats { - min_value, - max_value, - num_vals: doc_id_mapping.len() as u64, - }; - let fastfield_accessor = SortedDocidFieldAccessProvider { - doc_id_mapping, - fast_field_readers: &fast_field_readers, - }; - let iter = doc_id_mapping.iter().map(|(doc_id, reader_with_ordinal)| { - let fast_field_reader = &fast_field_readers[reader_with_ordinal.ordinal as usize]; - fast_field_reader.get(*doc_id) - }); - fast_field_serializer.create_auto_detect_u64_fast_field( - field, - stats, - fastfield_accessor, - iter.clone(), - iter, - )?; - Ok(()) - } else { - let u64_readers = self.readers.iter() - .filter(|reader|reader.max_doc() != reader.delete_bitset().map(|bit_set|bit_set.len() as u32).unwrap_or(0)) - .map(|reader|{ - let u64_reader: DynamicFastFieldReader = reader - .fast_fields() - .typed_fast_field_reader(field) - .expect("Failed to find a reader for single fast field. This is a tantivy bug and it should never happen."); - (reader.max_doc(), u64_reader, reader.delete_bitset()) - }).collect::>(); - - let mut fast_single_field_serializer = - fast_field_serializer.new_u64_fast_field(field, min_value, max_value)?; - for (max_doc, u64_reader, delete_bitset_opt) in u64_readers { - for doc_id in 0u32..max_doc { - let is_deleted = delete_bitset_opt - .map(|delete_bitset| delete_bitset.is_deleted(doc_id)) - .unwrap_or(false); - if !is_deleted { - let val = u64_reader.get(doc_id); - fast_single_field_serializer.add_val(val)?; - } - } - } - - fast_single_field_serializer.close_field()?; - Ok(()) + let stats = FastFieldStats { + min_value, + max_value, + num_vals: doc_id_mapping.len() as u64, + }; + #[derive(Clone)] + struct SortedDocidFieldAccessProvider<'a> { + doc_id_mapping: &'a SegmentDocidMapping<'a>, + fast_field_readers: &'a Vec>, } + impl<'a> FastFieldDataAccess for SortedDocidFieldAccessProvider<'a> { + fn get_val(&self, doc: u64) -> u64 { + let (doc_id, reader_with_ordinal) = self.doc_id_mapping[doc as usize]; + self.fast_field_readers[reader_with_ordinal.ordinal as usize].get(doc_id) + } + } + let fastfield_accessor = SortedDocidFieldAccessProvider { + doc_id_mapping, + fast_field_readers: &fast_field_readers, + }; + let iter1 = doc_id_mapping.iter().map(|(doc_id, reader_with_ordinal)| { + let fast_field_reader = &fast_field_readers[reader_with_ordinal.ordinal as usize]; + fast_field_reader.get(*doc_id) + }); + let iter2 = doc_id_mapping.iter().map(|(doc_id, reader_with_ordinal)| { + let fast_field_reader = &fast_field_readers[reader_with_ordinal.ordinal as usize]; + fast_field_reader.get(*doc_id) + }); + fast_field_serializer.create_auto_detect_u64_fast_field( + field, + stats, + fastfield_accessor, + iter1, + iter2, + )?; + + Ok(()) } /// Checks if the readers are disjunct for their sort property and in the correct order to be @@ -468,7 +439,7 @@ impl IndexMerger { pub(crate) fn generate_doc_id_mapping( &self, sort_by_field: &IndexSortByField, - ) -> crate::Result> { + ) -> crate::Result { let reader_and_field_accessors = self.get_reader_with_sort_field_accessor(sort_by_field)?; // Loading the field accessor on demand causes a 15x regression @@ -504,7 +475,7 @@ impl IndexMerger { }) .map(|(doc_id, reader_with_id, _)| (doc_id, reader_with_id)) .collect::>(); - Ok(sorted_doc_ids) + Ok(SegmentDocidMapping::new(sorted_doc_ids, false)) } // Creating the index file to point into the data, generic over `BytesFastFieldReader` and @@ -516,9 +487,9 @@ impl IndexMerger { fn write_1_n_fast_field_idx_generic( field: Field, fast_field_serializer: &mut CompositeFastFieldSerializer, - doc_id_mapping: &Option>, + doc_id_mapping: &SegmentDocidMapping, reader_and_field_accessors: &[(&SegmentReader, T)], - ) -> crate::Result<()> { + ) -> crate::Result> { let mut total_num_vals = 0u64; // In the first pass, we compute the total number of vals. // @@ -547,56 +518,37 @@ impl IndexMerger { }; // We can now create our `idx` serializer, and in a second pass, // can effectively push the different indexes. - if let Some(doc_id_mapping) = doc_id_mapping { - // copying into a temp vec is not ideal, but the fast field codec api requires random - // access, which is used in the estimation. It's possible to 1. calculate random - // acccess on the fly or 2. change the codec api to make random access optional, but - // they both have also major drawbacks. - let mut offsets = vec![]; - let mut offset = 0; - for (doc_id, reader) in doc_id_mapping { - let reader = &reader_and_field_accessors[reader.ordinal as usize].1; - offsets.push(offset); - offset += reader.get_len(*doc_id) as u64; - } + // copying into a temp vec is not ideal, but the fast field codec api requires random + // access, which is used in the estimation. It's possible to 1. calculate random + // acccess on the fly or 2. change the codec api to make random access optional, but + // they both have also major drawbacks. + + let mut offsets = vec![]; + let mut offset = 0; + for (doc_id, reader) in doc_id_mapping.iter() { + let reader = &reader_and_field_accessors[reader.ordinal as usize].1; offsets.push(offset); - - fast_field_serializer.create_auto_detect_u64_fast_field( - field, - stats, - &offsets, - offsets.iter().cloned(), - offsets.iter().cloned(), - )?; - } else { - let mut offsets = vec![]; - let mut offset = 0; - for (segment_reader, u64s_reader) in reader_and_field_accessors.iter() { - for doc in segment_reader.doc_ids_alive() { - offsets.push(offset); - offset += u64s_reader.get_len(doc) as u64; - } - } - offsets.push(offset); - - fast_field_serializer.create_auto_detect_u64_fast_field( - field, - stats, - &offsets, - offsets.iter().cloned(), - offsets.iter().cloned(), - )?; + offset += reader.get_len(*doc_id) as u64; } + offsets.push(offset); - Ok(()) + fast_field_serializer.create_auto_detect_u64_fast_field( + field, + stats, + &offsets[..], + offsets.iter().cloned(), + offsets.iter().cloned(), + )?; + Ok(offsets) } + /// Returns the fastfield index (index for the data, not the data). fn write_multi_value_fast_field_idx( &self, field: Field, fast_field_serializer: &mut CompositeFastFieldSerializer, - doc_id_mapping: &Option>, - ) -> crate::Result<()> { + doc_id_mapping: &SegmentDocidMapping, + ) -> crate::Result> { let reader_and_field_accessors = self.readers.iter().map(|reader|{ let u64s_reader: MultiValuedFastFieldReader = reader.fast_fields() .typed_fast_field_multi_reader(field) @@ -617,9 +569,11 @@ impl IndexMerger { field: Field, term_ordinal_mappings: &TermOrdinalMapping, fast_field_serializer: &mut CompositeFastFieldSerializer, - doc_id_mapping: &Option>, + doc_id_mapping: &SegmentDocidMapping, ) -> crate::Result<()> { - // Multifastfield consists in 2 fastfields. + debug_time!("write_hierarchical_facet_field"); + + // Multifastfield consists of 2 fastfields. // The first serves as an index into the second one and is stricly increasing. // The second contains the actual values. @@ -644,53 +598,61 @@ impl IndexMerger { let mut serialize_vals = fast_field_serializer.new_u64_fast_field_with_idx(field, 0u64, max_term_ord, 1)?; let mut vals = Vec::with_capacity(100); - if let Some(doc_id_mapping) = doc_id_mapping { - for (old_doc_id, reader_with_ordinal) in doc_id_mapping { - let term_ordinal_mapping: &[TermOrdinal] = - term_ordinal_mappings.get_segment(reader_with_ordinal.ordinal as usize); - let ff_reader = &fast_field_reader[reader_with_ordinal.ordinal as usize]; - ff_reader.get_vals(*old_doc_id, &mut vals); - for &prev_term_ord in &vals { - let new_term_ord = term_ordinal_mapping[prev_term_ord as usize]; - serialize_vals.add_val(new_term_ord)?; - } - } - } else { - for (segment_ord, segment_reader) in self.readers.iter().enumerate() { - let term_ordinal_mapping: &[TermOrdinal] = - term_ordinal_mappings.get_segment(segment_ord); - let ff_reader = &fast_field_reader[segment_ord as usize]; - // TODO optimize if no deletes - for doc in segment_reader.doc_ids_alive() { - ff_reader.get_vals(doc, &mut vals); - for &prev_term_ord in &vals { - let new_term_ord = term_ordinal_mapping[prev_term_ord as usize]; - serialize_vals.add_val(new_term_ord)?; - } - } + for (old_doc_id, reader_with_ordinal) in doc_id_mapping.iter() { + let term_ordinal_mapping: &[TermOrdinal] = + term_ordinal_mappings.get_segment(reader_with_ordinal.ordinal as usize); + + let ff_reader = &fast_field_reader[reader_with_ordinal.ordinal as usize]; + ff_reader.get_vals(*old_doc_id, &mut vals); + for &prev_term_ord in &vals { + let new_term_ord = term_ordinal_mapping[prev_term_ord as usize]; + serialize_vals.add_val(new_term_ord)?; } } + serialize_vals.close_field()?; } Ok(()) } + /// Creates a mapping if the segments are stacked. this is helpful to merge codelines between index + /// sorting and the others + pub(crate) fn get_doc_id_from_concatenated_data(&self) -> crate::Result { + let mapping: Vec<_> = self + .readers + .iter() + .enumerate() + .map(|(ordinal, reader)| { + let reader_with_ordinal = SegmentReaderWithOrdinal { + ordinal: ordinal as u32, + reader, + }; + reader + .doc_ids_alive() + .map(move |doc_id| (doc_id, reader_with_ordinal)) + }) + .flatten() + .collect(); + Ok(SegmentDocidMapping::new(mapping, true)) + } fn write_multi_fast_field( &self, field: Field, fast_field_serializer: &mut CompositeFastFieldSerializer, - doc_id_mapping: &Option>, + doc_id_mapping: &SegmentDocidMapping, ) -> crate::Result<()> { // Multifastfield consists in 2 fastfields. // The first serves as an index into the second one and is stricly increasing. // The second contains the actual values. // First we merge the idx fast field. - self.write_multi_value_fast_field_idx(field, fast_field_serializer, doc_id_mapping)?; + let offsets = + self.write_multi_value_fast_field_idx(field, fast_field_serializer, doc_id_mapping)?; let mut min_value = u64::max_value(); let mut max_value = u64::min_value(); + let mut num_vals = 0; let mut vals = Vec::with_capacity(100); @@ -716,6 +678,7 @@ impl IndexMerger { min_value = cmp::min(val, min_value); max_value = cmp::max(val, max_value); } + num_vals += vals.len(); } ff_readers.push(ff_reader); // TODO optimize when no deletes @@ -726,40 +689,76 @@ impl IndexMerger { max_value = 0; } - let fast_field_reader = self - .readers - .iter() - .map(|reader| { - let ff_reader : MultiValuedFastFieldReader = reader.fast_fields() - .typed_fast_field_multi_reader(field) - .expect("Failed to find index for multivalued field. This is a bug in tantivy, please report."); - ff_reader - }) - .collect::>(); - // We can now initialize our serializer, and push it the different values - let mut serialize_vals = - fast_field_serializer.new_u64_fast_field_with_idx(field, min_value, max_value, 1)?; - if let Some(doc_id_mapping) = doc_id_mapping { - for (doc_id, reader_with_ordinal) in doc_id_mapping { - let ff_reader = &fast_field_reader[reader_with_ordinal.ordinal as usize]; - ff_reader.get_vals(*doc_id, &mut vals); - for &val in &vals { - serialize_vals.add_val(val)?; - } - } - } else { - for (reader, ff_reader) in self.readers.iter().zip(ff_readers) { - // TODO optimize if no deletes - for doc in reader.doc_ids_alive() { - ff_reader.get_vals(doc, &mut vals); - for &val in &vals { - serialize_vals.add_val(val)?; - } - } + let stats = FastFieldStats { + max_value, + num_vals: num_vals as u64, + min_value, + }; + + struct SortedDocidMultiValueAccessProvider<'a> { + doc_id_mapping: &'a SegmentDocidMapping<'a>, + fast_field_readers: &'a Vec>, + offsets: Vec, + } + impl<'a> FastFieldDataAccess for SortedDocidMultiValueAccessProvider<'a> { + fn get_val(&self, pos: u64) -> u64 { + // use the offsets index to find the doc_id which will contain the position. + // the offsets are stricly increasing so we can do a simple search on it. + let new_docid = self + .offsets + .iter() + .position(|&offset| offset > pos) + .expect("pos is out of bounds") + - 1; + + // now we need to find the position of `pos` in the multivalued bucket + let num_pos_covered_until_now = self.offsets[new_docid]; + let pos_in_values = pos - num_pos_covered_until_now; + + let (old_doc_id, reader_with_ordinal) = self.doc_id_mapping[new_docid as usize]; + let num_vals = self.fast_field_readers[reader_with_ordinal.ordinal as usize] + .get_len(old_doc_id); + assert!(num_vals >= pos_in_values); + let mut vals = vec![]; + self.fast_field_readers[reader_with_ordinal.ordinal as usize] + .get_vals(old_doc_id, &mut vals); + + vals[pos_in_values as usize] } } - serialize_vals.close_field()?; + let fastfield_accessor = SortedDocidMultiValueAccessProvider { + doc_id_mapping, + fast_field_readers: &ff_readers, + offsets, + }; + let iter1 = doc_id_mapping + .iter() + .map(|(doc_id, reader_with_ordinal)| { + let ff_reader = &ff_readers[reader_with_ordinal.ordinal as usize]; + let mut vals = vec![]; + ff_reader.get_vals(*doc_id, &mut vals); + vals.into_iter() + }) + .flatten(); + let iter2 = doc_id_mapping + .iter() + .map(|(doc_id, reader_with_ordinal)| { + let ff_reader = &ff_readers[reader_with_ordinal.ordinal as usize]; + let mut vals = vec![]; + ff_reader.get_vals(*doc_id, &mut vals); + vals.into_iter() + }) + .flatten(); + fast_field_serializer.create_auto_detect_u64_fast_field_with_idx( + field, + stats, + fastfield_accessor, + iter1, + iter2, + 1, + )?; + Ok(()) } @@ -767,7 +766,7 @@ impl IndexMerger { &self, field: Field, fast_field_serializer: &mut CompositeFastFieldSerializer, - doc_id_mapping: &Option>, + doc_id_mapping: &SegmentDocidMapping, ) -> crate::Result<()> { let reader_and_field_accessors = self .readers @@ -786,24 +785,13 @@ impl IndexMerger { &reader_and_field_accessors, )?; let mut serialize_vals = fast_field_serializer.new_bytes_fast_field_with_idx(field, 1); - if let Some(doc_id_mapping) = doc_id_mapping { - for (doc_id, reader_with_ordinal) in doc_id_mapping { - let bytes_reader = - &reader_and_field_accessors[reader_with_ordinal.ordinal as usize].1; - let val = bytes_reader.get_bytes(*doc_id); - serialize_vals.write_all(val)?; - } - } else { - for segment_reader in &self.readers { - let bytes_reader = segment_reader.fast_fields().bytes(field) - .expect("Failed to find bytes field in fast field reader. This is a bug in tantivy. Please report."); - // TODO: optimize if no deletes - for doc in segment_reader.doc_ids_alive() { - let val = bytes_reader.get_bytes(doc); - serialize_vals.write_all(val)?; - } - } + + for (doc_id, reader_with_ordinal) in doc_id_mapping.iter() { + let bytes_reader = &reader_and_field_accessors[reader_with_ordinal.ordinal as usize].1; + let val = bytes_reader.get_bytes(*doc_id); + serialize_vals.write_all(val)?; } + serialize_vals.flush()?; Ok(()) } @@ -814,8 +802,9 @@ impl IndexMerger { field_type: &FieldType, serializer: &mut InvertedIndexSerializer, fieldnorm_reader: Option, - doc_id_mapping: &Option>, + doc_id_mapping: &SegmentDocidMapping, ) -> crate::Result> { + debug_time!("write_postings_for_field"); let mut positions_buffer: Vec = Vec::with_capacity(1_000); let mut delta_computer = DeltaComputer::new(); @@ -840,40 +829,23 @@ impl IndexMerger { }; let mut merged_terms = TermMerger::new(field_term_streams); - let mut max_doc = 0; // map from segment doc ids to the resulting merged segment doc id. - let mut merged_doc_id_map: Vec>> = Vec::with_capacity(self.readers.len()); - if let Some(doc_id_mapping) = doc_id_mapping { - merged_doc_id_map = self - .readers - .iter() - .map(|reader| { - let mut segment_local_map = vec![]; - segment_local_map.resize(reader.max_doc() as usize, None); - segment_local_map - }) - .collect(); - for (new_doc_id, (old_doc_id, segment_and_ordinal)) in doc_id_mapping.iter().enumerate() - { - let segment_map = &mut merged_doc_id_map[segment_and_ordinal.ordinal as usize]; - segment_map[*old_doc_id as usize] = Some(new_doc_id as DocId); - } - } else { - for reader in &self.readers { - let mut segment_local_map = Vec::with_capacity(reader.max_doc() as usize); - for doc_id in 0..reader.max_doc() { - if reader.is_deleted(doc_id) { - segment_local_map.push(None); - } else { - segment_local_map.push(Some(max_doc)); - max_doc += 1u32; - } - } - merged_doc_id_map.push(segment_local_map); - } + let mut merged_doc_id_map: Vec>> = self + .readers + .iter() + .map(|reader| { + let mut segment_local_map = vec![]; + segment_local_map.resize(reader.max_doc() as usize, None); + segment_local_map + }) + .collect(); + for (new_doc_id, (old_doc_id, segment_and_ordinal)) in doc_id_mapping.iter().enumerate() { + let segment_map = &mut merged_doc_id_map[segment_and_ordinal.ordinal as usize]; + segment_map[*old_doc_id as usize] = Some(new_doc_id as DocId); } + // The total number of tokens will only be exact when there has been no deletes. // // Otherwise, we approximate by removing deleted documents proportionally. @@ -969,7 +941,7 @@ impl IndexMerger { // I think this is not strictly necessary, it would be possible to // avoid the loading into a vec via some form of kmerge, but then the merge // logic would deviate much more from the stacking case (unsorted index) - if doc_id_mapping.is_some() { + if !doc_id_mapping.is_trivial() { doc_id_and_positions.push(( remapped_doc_id, term_freq, @@ -984,7 +956,7 @@ impl IndexMerger { doc = segment_postings.advance(); } } - if doc_id_mapping.is_some() { + if !doc_id_mapping.is_trivial() { doc_id_and_positions.sort_unstable_by_key(|&(doc_id, _, _)| doc_id); for (doc_id, term_freq, positions) in &doc_id_and_positions { field_serializer.write_doc(*doc_id, *term_freq, positions); @@ -1003,7 +975,7 @@ impl IndexMerger { &self, serializer: &mut InvertedIndexSerializer, fieldnorm_readers: FieldNormReaders, - doc_id_mapping: &Option>, + doc_id_mapping: &SegmentDocidMapping, ) -> crate::Result> { let mut term_ordinal_mappings = HashMap::new(); for (field, field_entry) in self.schema.fields() { @@ -1026,8 +998,10 @@ impl IndexMerger { fn write_storable_fields( &self, store_writer: &mut StoreWriter, - doc_id_mapping: &Option>, + doc_id_mapping: &SegmentDocidMapping, ) -> crate::Result<()> { + debug_time!("write_storable_fields"); + let store_readers: Vec<_> = self .readers .iter() @@ -1038,8 +1012,8 @@ impl IndexMerger { .enumerate() .map(|(i, store)| store.iter_raw(self.readers[i].delete_bitset())) .collect(); - if let Some(doc_id_mapping) = doc_id_mapping { - for (old_doc_id, reader_with_ordinal) in doc_id_mapping { + if !doc_id_mapping.is_trivial() { + for (old_doc_id, reader_with_ordinal) in doc_id_mapping.iter() { let doc_bytes_it = &mut document_iterators[reader_with_ordinal.ordinal as usize]; if let Some(doc_bytes_res) = doc_bytes_it.next() { let doc_bytes = doc_bytes_res?; @@ -1095,12 +1069,12 @@ impl IndexMerger { // If the documents are already sorted and stackable, we ignore the mapping and execute // it as if there was no sorting if self.is_disjunct_and_sorted_on_sort_property(sort_by_field)? { - None + self.get_doc_id_from_concatenated_data()? } else { - Some(self.generate_doc_id_mapping(sort_by_field)?) + self.generate_doc_id_mapping(sort_by_field)? } } else { - None + self.get_doc_id_from_concatenated_data()? }; if let Some(fieldnorms_serializer) = serializer.extract_fieldnorms_serializer() { diff --git a/src/lib.rs b/src/lib.rs index fbc875a2e..865d41cb7 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -100,7 +100,6 @@ #[cfg_attr(test, macro_use)] extern crate serde_json; - #[macro_use] extern crate log;