diff --git a/src/core/index_meta.rs b/src/core/index_meta.rs index c42a63bd8..2dbb3d4b1 100644 --- a/src/core/index_meta.rs +++ b/src/core/index_meta.rs @@ -110,7 +110,6 @@ impl SegmentMeta { path.push_str(&*match component { SegmentComponent::Postings => ".idx".to_string(), SegmentComponent::Positions => ".pos".to_string(), - SegmentComponent::PositionsSkip => ".posidx".to_string(), SegmentComponent::Terms => ".term".to_string(), SegmentComponent::Store => ".store".to_string(), SegmentComponent::FastFields => ".fast".to_string(), diff --git a/src/core/inverted_index_reader.rs b/src/core/inverted_index_reader.rs index c7e5db710..2f9515ca8 100644 --- a/src/core/inverted_index_reader.rs +++ b/src/core/inverted_index_reader.rs @@ -26,7 +26,6 @@ pub struct InvertedIndexReader { termdict: TermDictionary, postings_file_slice: FileSlice, positions_file_slice: FileSlice, - positions_idx_file_slice: FileSlice, record_option: IndexRecordOption, total_num_tokens: u64, } @@ -37,7 +36,6 @@ impl InvertedIndexReader { termdict: TermDictionary, postings_file_slice: FileSlice, positions_file_slice: FileSlice, - positions_idx_file_slice: FileSlice, record_option: IndexRecordOption, ) -> io::Result { let (total_num_tokens_slice, postings_body) = postings_file_slice.split(8); @@ -46,7 +44,6 @@ impl InvertedIndexReader { termdict, postings_file_slice: postings_body, positions_file_slice, - positions_idx_file_slice, record_option, total_num_tokens, }) @@ -59,7 +56,6 @@ impl InvertedIndexReader { termdict: TermDictionary::empty(), postings_file_slice: FileSlice::empty(), positions_file_slice: FileSlice::empty(), - positions_idx_file_slice: FileSlice::empty(), record_option, total_num_tokens: 0u64, } @@ -141,12 +137,12 @@ impl InvertedIndexReader { option: IndexRecordOption, ) -> io::Result { let block_postings = self.read_block_postings_from_terminfo(term_info, option)?; - let position_stream = { + let position_reader = { if option.has_positions() { - let position_reader = self.positions_file_slice.clone(); - let skip_reader = self.positions_idx_file_slice.clone(); - let position_reader = - PositionReader::new(position_reader, skip_reader, term_info.positions_idx)?; + let positions_data = self + .positions_file_slice + .read_bytes_slice(term_info.positions_range.clone())?; + let position_reader = PositionReader::new(positions_data)?; Some(position_reader) } else { None @@ -154,7 +150,7 @@ impl InvertedIndexReader { }; Ok(SegmentPostings::from_block_postings( block_postings, - position_stream, + position_reader, )) } diff --git a/src/core/segment_component.rs b/src/core/segment_component.rs index 2a81ff863..e8e1a45b5 100644 --- a/src/core/segment_component.rs +++ b/src/core/segment_component.rs @@ -10,8 +10,6 @@ pub enum SegmentComponent { Postings, /// Positions of terms in each document. Positions, - /// Index to seek within the position file - PositionsSkip, /// Column-oriented random-access storage of fields. FastFields, /// Stores the sum of the length (in terms) of each field for each document. @@ -31,10 +29,9 @@ pub enum SegmentComponent { impl SegmentComponent { /// Iterates through the components. pub fn iterator() -> slice::Iter<'static, SegmentComponent> { - static SEGMENT_COMPONENTS: [SegmentComponent; 8] = [ + static SEGMENT_COMPONENTS: [SegmentComponent; 7] = [ SegmentComponent::Postings, SegmentComponent::Positions, - SegmentComponent::PositionsSkip, SegmentComponent::FastFields, SegmentComponent::FieldNorms, SegmentComponent::Terms, diff --git a/src/core/segment_reader.rs b/src/core/segment_reader.rs index c7d0b76d7..0ade28f05 100644 --- a/src/core/segment_reader.rs +++ b/src/core/segment_reader.rs @@ -46,7 +46,6 @@ pub struct SegmentReader { termdict_composite: CompositeFile, postings_composite: CompositeFile, positions_composite: CompositeFile, - positions_idx_composite: CompositeFile, fast_fields_readers: Arc, fieldnorm_readers: FieldNormReaders, @@ -169,14 +168,6 @@ impl SegmentReader { } }; - let positions_idx_composite = { - if let Ok(positions_skip_file) = segment.open_read(SegmentComponent::PositionsSkip) { - CompositeFile::open(&positions_skip_file)? - } else { - CompositeFile::empty() - } - }; - let schema = segment.schema(); let fast_fields_data = segment.open_read(SegmentComponent::FastFields)?; @@ -207,7 +198,6 @@ impl SegmentReader { store_file, delete_bitset_opt, positions_composite, - positions_idx_composite, schema, }) } @@ -263,18 +253,15 @@ impl SegmentReader { let positions_file = self .positions_composite .open_read(field) - .expect("Index corrupted. Failed to open field positions in composite file."); - - let positions_idx_file = self - .positions_idx_composite - .open_read(field) - .expect("Index corrupted. Failed to open field positions in composite file."); + .ok_or_else(|| { + let error_msg = format!("Failed to open field {:?}'s positions in the composite file. Has the schema been modified?", field_entry.name()); + DataCorruption::comment_only(error_msg) + })?; let inv_idx_reader = Arc::new(InvertedIndexReader::new( TermDictionary::open(termdict_file)?, postings_file, positions_file, - positions_idx_file, record_option, )?); @@ -319,7 +306,6 @@ impl SegmentReader { self.termdict_composite.space_usage(), self.postings_composite.space_usage(), self.positions_composite.space_usage(), - self.positions_idx_composite.space_usage(), self.fast_fields_readers.space_usage(), self.fieldnorm_readers.space_usage(), self.get_store_reader()?.space_usage(), diff --git a/src/positions/mod.rs b/src/positions/mod.rs index 4d64bca7f..2c370a868 100644 --- a/src/positions/mod.rs +++ b/src/positions/mod.rs @@ -31,38 +31,34 @@ pub use self::serializer::PositionSerializer; use bitpacking::{BitPacker, BitPacker4x}; const COMPRESSION_BLOCK_SIZE: usize = BitPacker4x::BLOCK_LEN; -const LONG_SKIP_IN_BLOCKS: usize = 1_024; -const LONG_SKIP_INTERVAL: u64 = (LONG_SKIP_IN_BLOCKS * COMPRESSION_BLOCK_SIZE) as u64; #[cfg(test)] pub mod tests { use super::PositionSerializer; + use crate::directory::OwnedBytes; use crate::positions::reader::PositionReader; - use crate::{common::HasLen, directory::FileSlice}; use std::iter; - fn create_stream_buffer(vals: &[u32]) -> (FileSlice, FileSlice) { - let mut skip_buffer = vec![]; - let mut stream_buffer = vec![]; + fn create_positions_data(vals: &[u32]) -> OwnedBytes { + let mut positions_buffer = vec![]; { - let mut serializer = PositionSerializer::new(&mut stream_buffer, &mut skip_buffer); - for (i, &val) in vals.iter().enumerate() { - assert_eq!(serializer.positions_idx(), i as u64); + let mut serializer = PositionSerializer::new(&mut positions_buffer); + for &val in vals { serializer.write_all(&[val]).unwrap(); } + serializer.close_term().unwrap(); serializer.close().unwrap(); } - (FileSlice::from(stream_buffer), FileSlice::from(skip_buffer)) + OwnedBytes::new(positions_buffer) } #[test] fn test_position_read() { let v: Vec = (0..1000).collect(); - let (stream, skip) = create_stream_buffer(&v[..]); - assert_eq!(skip.len(), 12); - assert_eq!(stream.len(), 1168); - let mut position_reader = PositionReader::new(stream, skip, 0u64).unwrap(); + let positions_data = create_positions_data(&v[..]); + assert_eq!(positions_data.len(), 1224); + let mut position_reader = PositionReader::new(positions_data).unwrap(); for &n in &[1, 10, 127, 128, 130, 312] { let mut v = vec![0u32; n]; position_reader.read(0, &mut v[..]); @@ -75,10 +71,9 @@ pub mod tests { #[test] fn test_position_read_with_offset() { let v: Vec = (0..1000).collect(); - let (stream, skip) = create_stream_buffer(&v[..]); - assert_eq!(skip.len(), 12); - assert_eq!(stream.len(), 1168); - let mut position_reader = PositionReader::new(stream, skip, 0u64).unwrap(); + let positions_data = create_positions_data(&v[..]); + assert_eq!(positions_data.len(), 1224); + let mut position_reader = PositionReader::new(positions_data).unwrap(); for &offset in &[1u64, 10u64, 127u64, 128u64, 130u64, 312u64] { for &len in &[1, 10, 130, 500] { let mut v = vec![0u32; len]; @@ -93,11 +88,10 @@ pub mod tests { #[test] fn test_position_read_after_skip() { let v: Vec = (0..1_000).collect(); - let (stream, skip) = create_stream_buffer(&v[..]); - assert_eq!(skip.len(), 12); - assert_eq!(stream.len(), 1168); + let positions_data = create_positions_data(&v[..]); + assert_eq!(positions_data.len(), 1224); - let mut position_reader = PositionReader::new(stream, skip, 0u64).unwrap(); + let mut position_reader = PositionReader::new(positions_data).unwrap(); let mut buf = [0u32; 7]; let mut c = 0; @@ -115,11 +109,10 @@ pub mod tests { #[test] fn test_position_reread_anchor_different_than_block() { - let v: Vec = (0..2_000_000).collect(); - let (stream, skip) = create_stream_buffer(&v[..]); - assert_eq!(skip.len(), 15_749); - assert_eq!(stream.len(), 4_987_872); - let mut position_reader = PositionReader::new(stream.clone(), skip.clone(), 0).unwrap(); + let positions_delta: Vec = (0..2_000_000).collect(); + let positions_data = create_positions_data(&positions_delta[..]); + assert_eq!(positions_data.len(), 5003499); + let mut position_reader = PositionReader::new(positions_data.clone()).unwrap(); let mut buf = [0u32; 256]; position_reader.read(128, &mut buf); for i in 0..256 { @@ -134,57 +127,53 @@ pub mod tests { #[test] #[should_panic(expected = "offset arguments should be increasing.")] fn test_position_panic_if_called_previous_anchor() { - let v: Vec = (0..2_000_000).collect(); - let (stream, skip) = create_stream_buffer(&v[..]); - assert_eq!(skip.len(), 15_749); - assert_eq!(stream.len(), 4_987_872); + let positions_delta: Vec = (0..2_000_000).collect(); + let positions_data = create_positions_data(&positions_delta[..]); + assert_eq!(positions_data.len(), 5_003_499); let mut buf = [0u32; 1]; - let mut position_reader = - PositionReader::new(stream.clone(), skip.clone(), 200_000).unwrap(); + let mut position_reader = PositionReader::new(positions_data).unwrap(); position_reader.read(230, &mut buf); position_reader.read(9, &mut buf); } #[test] fn test_positions_bug() { - let mut v: Vec = vec![]; + let mut positions_delta: Vec = vec![]; for i in 1..200 { for j in 0..i { - v.push(j); + positions_delta.push(j); } } - let (stream, skip) = create_stream_buffer(&v[..]); + let positions_data = create_positions_data(&positions_delta[..]); let mut buf = Vec::new(); - let mut position_reader = PositionReader::new(stream.clone(), skip.clone(), 0).unwrap(); + let mut position_reader = PositionReader::new(positions_data).unwrap(); let mut offset = 0; for i in 1..24 { buf.resize(i, 0); - position_reader.read(offset, &mut buf[..]); offset += i as u64; - let r: Vec = (0..i).map(|el| el as u32).collect(); - assert_eq!(buf, &r[..]); + position_reader.read(offset, &mut buf[..]); + let expected_positions_delta: Vec = (0..i as u32).collect(); + assert_eq!(buf, &expected_positions_delta[..], "Failed for offset={},i={}", offset, i); } } #[test] - fn test_position_long_skip_const() { + fn test_position() { const CONST_VAL: u32 = 9u32; - let v: Vec = iter::repeat(CONST_VAL).take(2_000_000).collect(); - let (stream, skip) = create_stream_buffer(&v[..]); - assert_eq!(skip.len(), 15_749); - assert_eq!(stream.len(), 1_000_000); - let mut position_reader = PositionReader::new(stream, skip, 128 * 1024).unwrap(); + let positions_delta: Vec = iter::repeat(CONST_VAL).take(2_000_000).collect(); + let positions_data = create_positions_data(&positions_delta[..]); + assert_eq!(positions_data.len(), 1_015_627); + let mut position_reader = PositionReader::new(positions_data).unwrap(); let mut buf = [0u32; 1]; position_reader.read(0, &mut buf); assert_eq!(buf[0], CONST_VAL); } #[test] - fn test_position_long_skip_2() { - let v: Vec = (0..2_000_000).collect(); - let (stream, skip) = create_stream_buffer(&v[..]); - assert_eq!(skip.len(), 15_749); - assert_eq!(stream.len(), 4_987_872); + fn test_position_advance() { + let positions_delta: Vec = (0..2_000_000).collect(); + let positions_data = create_positions_data(&positions_delta[..]); + assert_eq!(positions_data.len(), 5_003_499); for &offset in &[ 10, 128 * 1024, @@ -192,10 +181,9 @@ pub mod tests { 128 * 1024 + 7, 128 * 10 * 1024 + 10, ] { - let mut position_reader = - PositionReader::new(stream.clone(), skip.clone(), offset).unwrap(); + let mut position_reader = PositionReader::new(positions_data.clone()).unwrap(); let mut buf = [0u32; 1]; - position_reader.read(0, &mut buf); + position_reader.read(offset, &mut buf); assert_eq!(buf[0], offset as u32); } } diff --git a/src/positions/reader.rs b/src/positions/reader.rs index eded0c613..08aee6626 100644 --- a/src/positions/reader.rs +++ b/src/positions/reader.rs @@ -1,132 +1,87 @@ use std::io; -use crate::common::{BinarySerializable, FixedSize}; -use crate::directory::FileSlice; +use crate::common::{BinarySerializable, VInt}; use crate::directory::OwnedBytes; use crate::positions::COMPRESSION_BLOCK_SIZE; -use crate::positions::LONG_SKIP_INTERVAL; -use crate::positions::LONG_SKIP_IN_BLOCKS; -use bitpacking::{BitPacker, BitPacker4x}; +use crate::postings::compression::{BlockDecoder, VIntDecoder}; -/// Positions works as a long sequence of compressed block. -/// All terms are chained one after the other. -/// /// When accessing the position of a term, we get a positions_idx from the `Terminfo`. /// This means we need to skip to the `nth` positions efficiently. /// -/// This is done thanks to two levels of skiping that we refer to in the code -/// as `long_skip` and `short_skip`. -/// -/// The `long_skip` makes it possible to skip every 1_024 compression blocks (= 131_072 positions). -/// Skipping offset are simply stored one after as an offset stored over 8 bytes. -/// -/// We find the number of long skips, as `n / long_skip`. -/// -/// Blocks are compressed using bitpacking, so `skip_read` contains the number of bytes -/// (values can go from 0bit to 32 bits) required to decompressed every block. +/// Blocks are compressed using bitpacking, so `skip_read` contains the number of bits +/// (values can go from 0bit to 32 bits) required to decompress every block. /// /// A given block obviously takes `(128 x num_bit_for_the_block / num_bits_in_a_byte)`, /// so skipping a block without decompressing it is just a matter of advancing that many /// bytes. -struct Positions { - bit_packer: BitPacker4x, - skip_file: FileSlice, - position_file: FileSlice, - long_skip_data: OwnedBytes, -} - -impl Positions { - pub fn new(position_file: FileSlice, skip_file: FileSlice) -> io::Result { - let (body, footer) = skip_file.split_from_end(u32::SIZE_IN_BYTES); - let footer_data = footer.read_bytes()?; - let num_long_skips = u32::deserialize(&mut footer_data.as_slice())?; - let (skip_file, long_skip_file) = - body.split_from_end(u64::SIZE_IN_BYTES * (num_long_skips as usize)); - let long_skip_data = long_skip_file.read_bytes()?; - Ok(Positions { - bit_packer: BitPacker4x::new(), - skip_file, - long_skip_data, - position_file, - }) - } - - /// Returns the offset of the block associated to the given `long_skip_id`. - /// - /// One `long_skip_id` means `LONG_SKIP_IN_BLOCKS` blocks. - fn long_skip(&self, long_skip_id: usize) -> u64 { - if long_skip_id == 0 { - return 0; - } - let long_skip_slice = self.long_skip_data.as_slice(); - let mut long_skip_blocks: &[u8] = &long_skip_slice[(long_skip_id - 1) * 8..][..8]; - u64::deserialize(&mut long_skip_blocks).expect("Index corrupted") - } - - fn reader(&self, offset: u64) -> io::Result { - let long_skip_id = (offset / LONG_SKIP_INTERVAL) as usize; - let offset_num_bytes: u64 = self.long_skip(long_skip_id); - let position_read = self - .position_file - .slice_from(offset_num_bytes as usize) - .read_bytes()?; - let skip_read = self - .skip_file - .slice_from(long_skip_id * LONG_SKIP_IN_BLOCKS) - .read_bytes()?; - Ok(PositionReader { - bit_packer: self.bit_packer, - skip_read, - position_read, - buffer: Box::new([0u32; 128]), - block_offset: std::i64::MAX as u64, - anchor_offset: (long_skip_id as u64) * LONG_SKIP_INTERVAL, - abs_offset: offset, - }) - } -} - #[derive(Clone)] pub struct PositionReader { - skip_read: OwnedBytes, - position_read: OwnedBytes, - bit_packer: BitPacker4x, - buffer: Box<[u32; COMPRESSION_BLOCK_SIZE]>, + bit_widths: OwnedBytes, + positions: OwnedBytes, + block_decoder: BlockDecoder, + + // offset, expressed in positions, for the first position of the block currently loaded + // block_offset is a multiple of COMPRESSION_BLOCK_SIZE. block_offset: u64, + // offset, expressed in positions, for the position of the first block encoded + // in the `self.positions` bytes, and if bitpacked, compressed using the bitwidth in + // `self.bit_widths`. + // + // As we advance, anchor increases simultaneously with bit_widths and positions get consumed. anchor_offset: u64, - - abs_offset: u64, } impl PositionReader { - pub fn new( - position_file: FileSlice, - skip_file: FileSlice, - offset: u64, - ) -> io::Result { - let positions = Positions::new(position_file, skip_file)?; - positions.reader(offset) + pub fn new(mut positions_data: OwnedBytes) -> io::Result { + let num_positions_bitpacked_blocks = VInt::deserialize(&mut positions_data)?.0 as usize; + let (bit_widths, positions) = positions_data.split(num_positions_bitpacked_blocks); + Ok(PositionReader { + bit_widths, + positions, + block_decoder: BlockDecoder::default(), + block_offset: std::i64::MAX as u64, + anchor_offset: 0u64, + }) } fn advance_num_blocks(&mut self, num_blocks: usize) { - let num_bits: usize = self.skip_read.as_ref()[..num_blocks] + let num_bits: usize = self.bit_widths.as_ref()[..num_blocks] .iter() .cloned() .map(|num_bits| num_bits as usize) .sum(); let num_bytes_to_skip = num_bits * COMPRESSION_BLOCK_SIZE / 8; - self.skip_read.advance(num_blocks as usize); - self.position_read.advance(num_bytes_to_skip); + self.bit_widths.advance(num_blocks as usize); + self.positions.advance(num_bytes_to_skip); + self.anchor_offset += (num_blocks * COMPRESSION_BLOCK_SIZE) as u64; } + /// block_rel_id is counted relatively to the anchor. + /// block_rel_id = 0 means the anchor block. + /// block_rel_id = i means the ith block after the anchor block. + fn load_block(&mut self, block_rel_id: usize) { + let bit_widths = self.bit_widths.as_slice(); + let byte_offset: usize = bit_widths[0..block_rel_id].iter().map(|&b| b as usize).sum::() * COMPRESSION_BLOCK_SIZE / 8; + let compressed_data = &self.positions.as_slice()[byte_offset..]; + if bit_widths.len() > block_rel_id { + // that block is bitpacked. + let bit_width = bit_widths[block_rel_id]; + self.block_decoder.uncompress_block_unsorted(compressed_data, bit_width); + } else { + // that block is vint encoded. + unimplemented!(); + // self.block_decoder.uncompress_vint_unsorted(compressed_data); + } + self.block_offset = self.anchor_offset + (block_rel_id * COMPRESSION_BLOCK_SIZE) as u64; + } + /// Fills a buffer with the positions `[offset..offset+output.len())` integers. /// /// `offset` is required to have a value >= to the offsets given in previous calls /// for the given `PositionReaderAbsolute` instance. pub fn read(&mut self, mut offset: u64, mut output: &mut [u32]) { - offset += self.abs_offset; assert!( offset >= self.anchor_offset, "offset arguments should be increasing." @@ -134,41 +89,36 @@ impl PositionReader { let delta_to_block_offset = offset as i64 - self.block_offset as i64; if !(0..128).contains(&delta_to_block_offset) { // The first position is not within the first block. - // We need to decompress the first block. + // (Note that it could be before or after) + // We need to possibly skip a few blocks, and decompress the first relevant block. let delta_to_anchor_offset = offset - self.anchor_offset; let num_blocks_to_skip = (delta_to_anchor_offset / (COMPRESSION_BLOCK_SIZE as u64)) as usize; self.advance_num_blocks(num_blocks_to_skip); - self.anchor_offset = offset - (offset % COMPRESSION_BLOCK_SIZE as u64); - self.block_offset = self.anchor_offset; - let num_bits = self.skip_read.as_slice()[0]; - self.bit_packer - .decompress(self.position_read.as_ref(), self.buffer.as_mut(), num_bits); + self.load_block(0); } else { + // The request offset is within the loaded block. + // We still need to advance anchor_offset to our current block. let num_blocks_to_skip = ((self.block_offset - self.anchor_offset) / COMPRESSION_BLOCK_SIZE as u64) as usize; self.advance_num_blocks(num_blocks_to_skip); - self.anchor_offset = self.block_offset; } - let mut num_bits = self.skip_read.as_slice()[0]; - let mut position_data = self.position_read.as_ref(); - + // At this point, the block containing offset is loaded, and anchor has + // been updated to point to it as well. for i in 1.. { + // we copy the part from block i - 1 that is relevant. let offset_in_block = (offset as usize) % COMPRESSION_BLOCK_SIZE; let remaining_in_block = COMPRESSION_BLOCK_SIZE - offset_in_block; if remaining_in_block >= output.len() { - output.copy_from_slice(&self.buffer[offset_in_block..][..output.len()]); + output.copy_from_slice(&self.block_decoder.output_array()[offset_in_block..][..output.len()]); break; } - output[..remaining_in_block].copy_from_slice(&self.buffer[offset_in_block..]); + output[..remaining_in_block].copy_from_slice(&self.block_decoder.output_array()[offset_in_block..]); output = &mut output[remaining_in_block..]; + // we load block #i if necessary. offset += remaining_in_block as u64; - position_data = &position_data[(num_bits as usize * COMPRESSION_BLOCK_SIZE / 8)..]; - num_bits = self.skip_read.as_slice()[i]; - self.bit_packer - .decompress(position_data, self.buffer.as_mut(), num_bits); - self.block_offset += COMPRESSION_BLOCK_SIZE as u64; + self.load_block(i); } } } diff --git a/src/positions/serializer.rs b/src/positions/serializer.rs index 72eb652b4..80a120772 100644 --- a/src/positions/serializer.rs +++ b/src/positions/serializer.rs @@ -1,48 +1,42 @@ -use crate::common::BinarySerializable; -use crate::common::CountingWriter; -use crate::positions::{COMPRESSION_BLOCK_SIZE, LONG_SKIP_INTERVAL}; -use bitpacking::BitPacker; -use bitpacking::BitPacker4x; +use crate::common::{BinarySerializable, CountingWriter, VInt}; +use crate::positions::COMPRESSION_BLOCK_SIZE; +use crate::postings::compression::BlockEncoder; +use crate::postings::compression::VIntEncoder; use std::io::{self, Write}; pub struct PositionSerializer { - bit_packer: BitPacker4x, - write_stream: CountingWriter, - write_skip_index: W, + block_encoder: BlockEncoder, + positions_wrt: CountingWriter, + positions_buffer: Vec, block: Vec, - buffer: Vec, - num_ints: u64, - long_skips: Vec, + bit_widths: Vec, } impl PositionSerializer { - pub fn new(write_stream: W, write_skip_index: W) -> PositionSerializer { + pub fn new(positions_wrt: W) -> PositionSerializer { PositionSerializer { - bit_packer: BitPacker4x::new(), - write_stream: CountingWriter::wrap(write_stream), - write_skip_index, + block_encoder: BlockEncoder::new(), + positions_wrt: CountingWriter::wrap(positions_wrt), + positions_buffer: Vec::with_capacity(128_000), block: Vec::with_capacity(128), - buffer: vec![0u8; 128 * 4], - num_ints: 0u64, - long_skips: Vec::new(), + bit_widths: Vec::new(), } } - pub fn positions_idx(&self) -> u64 { - self.num_ints + pub fn offset(&self) -> u64 { + self.positions_wrt.written_bytes() } fn remaining_block_len(&self) -> usize { COMPRESSION_BLOCK_SIZE - self.block.len() } - pub fn write_all(&mut self, mut vals: &[u32]) -> io::Result<()> { - while !vals.is_empty() { + pub fn write_all(&mut self, mut positions_delta: &[u32]) -> io::Result<()> { + while !positions_delta.is_empty() { let remaining_block_len = self.remaining_block_len(); - let num_to_write = remaining_block_len.min(vals.len()); - self.block.extend(&vals[..num_to_write]); - self.num_ints += num_to_write as u64; - vals = &vals[num_to_write..]; + let num_to_write = remaining_block_len.min(positions_delta.len()); + self.block.extend(&positions_delta[..num_to_write]); + positions_delta = &positions_delta[num_to_write..]; if self.remaining_block_len() == 0 { self.flush_block()?; } @@ -51,30 +45,34 @@ impl PositionSerializer { } fn flush_block(&mut self) -> io::Result<()> { - let num_bits = self.bit_packer.num_bits(&self.block[..]); - self.write_skip_index.write_all(&[num_bits])?; - let written_len = self - .bit_packer - .compress(&self.block[..], &mut self.buffer, num_bits); - self.write_stream.write_all(&self.buffer[..written_len])?; - self.block.clear(); - if (self.num_ints % LONG_SKIP_INTERVAL) == 0u64 { - self.long_skips.push(self.write_stream.written_bytes()); + // encode the positions in the block + if self.block.is_empty() { + return Ok(()); + } else if self.block.len() == COMPRESSION_BLOCK_SIZE { + let (bit_width, block_encoded): (u8, &[u8]) = + self.block_encoder.compress_block_unsorted(&self.block[..]); + self.bit_widths.push(bit_width); + self.positions_buffer.write_all(block_encoded)?; + } else { + debug_assert!(self.block.len() < COMPRESSION_BLOCK_SIZE); + let block_vint_encoded = self.block_encoder.compress_vint_unsorted(&self.block[..]); + self.positions_buffer.write_all(block_vint_encoded)?; } + self.block.clear(); + Ok(()) + } + + pub fn close_term(&mut self) -> io::Result<()> { + self.flush_block()?; + VInt(self.bit_widths.len() as u64).serialize(&mut self.positions_wrt)?; + self.positions_wrt.write_all(&self.bit_widths[..])?; + self.positions_wrt.write_all(&self.positions_buffer)?; + self.bit_widths.clear(); + self.positions_buffer.clear(); Ok(()) } pub fn close(mut self) -> io::Result<()> { - if !self.block.is_empty() { - self.block.resize(COMPRESSION_BLOCK_SIZE, 0u32); - self.flush_block()?; - } - for &long_skip in &self.long_skips { - long_skip.serialize(&mut self.write_skip_index)?; - } - (self.long_skips.len() as u32).serialize(&mut self.write_skip_index)?; - self.write_skip_index.flush()?; - self.write_stream.flush()?; - Ok(()) + self.positions_wrt.flush() } } diff --git a/src/postings/mod.rs b/src/postings/mod.rs index 3e5998e1d..88b650ccb 100644 --- a/src/postings/mod.rs +++ b/src/postings/mod.rs @@ -74,7 +74,7 @@ pub mod tests { mem::drop(field_serializer); posting_serializer.close()?; let read = segment.open_read(SegmentComponent::Positions)?; - assert!(read.len() <= 140); + assert_eq!(read.len(), 207); Ok(()) } diff --git a/src/postings/serializer.rs b/src/postings/serializer.rs index d57b07425..5a63e7c42 100644 --- a/src/postings/serializer.rs +++ b/src/postings/serializer.rs @@ -50,19 +50,17 @@ pub struct InvertedIndexSerializer { terms_write: CompositeWrite, postings_write: CompositeWrite, positions_write: CompositeWrite, - positionsidx_write: CompositeWrite, schema: Schema, } impl InvertedIndexSerializer { /// Open a new `PostingsSerializer` for the given segment pub fn open(segment: &mut Segment) -> crate::Result { - use crate::SegmentComponent::{Positions, PositionsSkip, Postings, Terms}; + use crate::SegmentComponent::{Positions, Postings, Terms}; let inv_index_serializer = InvertedIndexSerializer { terms_write: CompositeWrite::wrap(segment.open_write(Terms)?), postings_write: CompositeWrite::wrap(segment.open_write(Postings)?), positions_write: CompositeWrite::wrap(segment.open_write(Positions)?), - positionsidx_write: CompositeWrite::wrap(segment.open_write(PositionsSkip)?), schema: segment.schema(), }; Ok(inv_index_serializer) @@ -82,7 +80,6 @@ impl InvertedIndexSerializer { let term_dictionary_write = self.terms_write.for_field(field); let postings_write = self.postings_write.for_field(field); let positions_write = self.positions_write.for_field(field); - let positionsidx_write = self.positionsidx_write.for_field(field); let field_type: FieldType = (*field_entry.field_type()).clone(); FieldSerializer::create( &field_type, @@ -90,7 +87,6 @@ impl InvertedIndexSerializer { term_dictionary_write, postings_write, positions_write, - positionsidx_write, fieldnorm_reader, ) } @@ -100,7 +96,6 @@ impl InvertedIndexSerializer { self.terms_write.close()?; self.postings_write.close()?; self.positions_write.close()?; - self.positionsidx_write.close()?; Ok(()) } } @@ -123,7 +118,6 @@ impl<'a> FieldSerializer<'a> { term_dictionary_write: &'a mut CountingWriter, postings_write: &'a mut CountingWriter, positions_write: &'a mut CountingWriter, - positionsidx_write: &'a mut CountingWriter, fieldnorm_reader: Option, ) -> io::Result> { total_num_tokens.serialize(postings_write)?; @@ -145,7 +139,7 @@ impl<'a> FieldSerializer<'a> { let postings_serializer = PostingsSerializer::new(postings_write, average_fieldnorm, mode, fieldnorm_reader); let positions_serializer_opt = if mode.has_positions() { - Some(PositionSerializer::new(positions_write, positionsidx_write)) + Some(PositionSerializer::new(positions_write)) } else { None }; @@ -161,17 +155,17 @@ impl<'a> FieldSerializer<'a> { } fn current_term_info(&self) -> TermInfo { - let positions_idx = + let positions_start = if let Some(positions_serializer) = self.positions_serializer_opt.as_ref() { - positions_serializer.positions_idx() + positions_serializer.offset() } else { 0u64 - }; + } as usize; let addr = self.postings_serializer.addr() as usize; TermInfo { doc_freq: 0, postings_range: addr..addr, - positions_idx, + positions_range: positions_start..positions_start, } } @@ -226,7 +220,11 @@ impl<'a> FieldSerializer<'a> { if self.term_open { self.postings_serializer .close_term(self.current_term_info.doc_freq)?; + if let Some(positions_serializer) = self.positions_serializer_opt.as_mut() { + positions_serializer.close_term()?; + } self.current_term_info.postings_range.end = self.postings_serializer.addr() as usize; + self.current_term_info.positions_range.end = self.postings_serializer.addr() as usize; self.term_dictionary_builder .insert_value(&self.current_term_info)?; self.term_open = false; diff --git a/src/postings/term_info.rs b/src/postings/term_info.rs index 2aad4e45d..b86e56b2f 100644 --- a/src/postings/term_info.rs +++ b/src/postings/term_info.rs @@ -11,8 +11,8 @@ pub struct TermInfo { pub doc_freq: u32, /// Byte range of the posting list within the postings (`.idx`) file. pub postings_range: Range, - /// Start offset of the first block within the position (`.pos`) file. - pub positions_idx: u64, + /// Byte range of the positions of this terms in the positions (`.pos`) file. + pub positions_range: Range, } impl TermInfo { @@ -21,6 +21,12 @@ impl TermInfo { assert!(num_bytes <= std::u32::MAX as usize); num_bytes as u32 } + + pub(crate) fn positions_num_bytes(&self) -> u32 { + let num_bytes = self.positions_range.len(); + assert!(num_bytes <= std::u32::MAX as usize); + num_bytes as u32 + } } impl FixedSize for TermInfo { @@ -28,7 +34,7 @@ impl FixedSize for TermInfo { /// This is large, but in practise, `TermInfo` are encoded in blocks and /// only the first `TermInfo` of a block is serialized uncompressed. /// The subsequent `TermInfo` are delta encoded and bitpacked. - const SIZE_IN_BYTES: usize = 2 * u32::SIZE_IN_BYTES + 2 * u64::SIZE_IN_BYTES; + const SIZE_IN_BYTES: usize = 3 * u32::SIZE_IN_BYTES + 2 * u64::SIZE_IN_BYTES; } impl BinarySerializable for TermInfo { @@ -36,20 +42,23 @@ impl BinarySerializable for TermInfo { self.doc_freq.serialize(writer)?; (self.postings_range.start as u64).serialize(writer)?; self.posting_num_bytes().serialize(writer)?; - self.positions_idx.serialize(writer)?; + (self.positions_range.start as u64).serialize(writer)?; + self.positions_num_bytes().serialize(writer)?; Ok(()) } fn deserialize(reader: &mut R) -> io::Result { let doc_freq = u32::deserialize(reader)?; let postings_start_offset = u64::deserialize(reader)? as usize; - let postings_num_bytes = u32::deserialize(reader)?; - let postings_end_offset = postings_start_offset + u64::from(postings_num_bytes) as usize; - let positions_idx = u64::deserialize(reader)?; + let postings_num_bytes = u32::deserialize(reader)? as usize; + let postings_end_offset = postings_start_offset + postings_num_bytes; + let positions_start_offset = u64::deserialize(reader)? as usize; + let positions_num_bytes = u32::deserialize(reader)? as usize; + let positions_end_offset = positions_start_offset + positions_num_bytes; Ok(TermInfo { doc_freq, postings_range: postings_start_offset..postings_end_offset, - positions_idx, + positions_range: positions_start_offset..positions_end_offset, }) } } @@ -60,6 +69,8 @@ mod tests { use super::TermInfo; use crate::common::test::fixed_size_test; + // TODO add serialize/deserialize test for terminfo + #[test] fn test_fixed_size() { fixed_size_test::(); diff --git a/src/space_usage/mod.rs b/src/space_usage/mod.rs index 4c8374f41..886b9bde3 100644 --- a/src/space_usage/mod.rs +++ b/src/space_usage/mod.rs @@ -69,7 +69,6 @@ pub struct SegmentSpaceUsage { termdict: PerFieldSpaceUsage, postings: PerFieldSpaceUsage, positions: PerFieldSpaceUsage, - positions_idx: PerFieldSpaceUsage, fast_fields: PerFieldSpaceUsage, fieldnorms: PerFieldSpaceUsage, @@ -87,7 +86,6 @@ impl SegmentSpaceUsage { termdict: PerFieldSpaceUsage, postings: PerFieldSpaceUsage, positions: PerFieldSpaceUsage, - positions_idx: PerFieldSpaceUsage, fast_fields: PerFieldSpaceUsage, fieldnorms: PerFieldSpaceUsage, store: StoreSpaceUsage, @@ -105,7 +103,6 @@ impl SegmentSpaceUsage { termdict, postings, positions, - positions_idx, fast_fields, fieldnorms, store, @@ -124,7 +121,6 @@ impl SegmentSpaceUsage { match component { Postings => PerField(self.postings().clone()), Positions => PerField(self.positions().clone()), - PositionsSkip => PerField(self.positions_skip_idx().clone()), FastFields => PerField(self.fast_fields().clone()), FieldNorms => PerField(self.fieldnorms().clone()), Terms => PerField(self.termdict().clone()), @@ -153,11 +149,6 @@ impl SegmentSpaceUsage { &self.positions } - /// Space usage for positions skip idx - pub fn positions_skip_idx(&self) -> &PerFieldSpaceUsage { - &self.positions_idx - } - /// Space usage for fast fields pub fn fast_fields(&self) -> &PerFieldSpaceUsage { &self.fast_fields @@ -358,7 +349,6 @@ mod test { expect_single_field(segment.termdict(), &name, 1, 512); expect_single_field(segment.postings(), &name, 1, 512); assert_eq!(0, segment.positions().total()); - assert_eq!(0, segment.positions_skip_idx().total()); expect_single_field(segment.fast_fields(), &name, 1, 512); expect_single_field(segment.fieldnorms(), &name, 1, 512); // TODO: understand why the following fails @@ -398,7 +388,6 @@ mod test { expect_single_field(segment.termdict(), &name, 1, 512); expect_single_field(segment.postings(), &name, 1, 512); expect_single_field(segment.positions(), &name, 1, 512); - expect_single_field(segment.positions_skip_idx(), &name, 1, 512); assert_eq!(0, segment.fast_fields().total()); expect_single_field(segment.fieldnorms(), &name, 1, 512); // TODO: understand why the following fails @@ -437,7 +426,6 @@ mod test { assert_eq!(0, segment.termdict().total()); assert_eq!(0, segment.postings().total()); assert_eq!(0, segment.positions().total()); - assert_eq!(0, segment.positions_skip_idx().total()); assert_eq!(0, segment.fast_fields().total()); assert_eq!(0, segment.fieldnorms().total()); assert!(segment.store().total() > 0); @@ -483,7 +471,6 @@ mod test { expect_single_field(segment_space_usage.termdict(), &name, 1, 512); expect_single_field(segment_space_usage.postings(), &name, 1, 512); assert_eq!(0, segment_space_usage.positions().total()); - assert_eq!(0, segment_space_usage.positions_skip_idx().total()); assert_eq!(0, segment_space_usage.fast_fields().total()); expect_single_field(segment_space_usage.fieldnorms(), &name, 1, 512); assert!(segment_space_usage.deletes() > 0); diff --git a/src/termdict/fst_termdict/term_info_store.rs b/src/termdict/fst_termdict/term_info_store.rs index 73b69e7e4..8e85a6641 100644 --- a/src/termdict/fst_termdict/term_info_store.rs +++ b/src/termdict/fst_termdict/term_info_store.rs @@ -15,7 +15,7 @@ struct TermInfoBlockMeta { ref_term_info: TermInfo, doc_freq_nbits: u8, postings_offset_nbits: u8, - positions_idx_nbits: u8, + positions_offset_nbits: u8, } impl BinarySerializable for TermInfoBlockMeta { @@ -25,7 +25,7 @@ impl BinarySerializable for TermInfoBlockMeta { write.write_all(&[ self.doc_freq_nbits, self.postings_offset_nbits, - self.positions_idx_nbits, + self.positions_offset_nbits, ])?; Ok(()) } @@ -40,7 +40,7 @@ impl BinarySerializable for TermInfoBlockMeta { ref_term_info, doc_freq_nbits: buffer[0], postings_offset_nbits: buffer[1], - positions_idx_nbits: buffer[2], + positions_offset_nbits: buffer[2], }) } } @@ -52,7 +52,7 @@ impl FixedSize for TermInfoBlockMeta { impl TermInfoBlockMeta { fn num_bits(&self) -> u8 { - self.doc_freq_nbits + self.postings_offset_nbits + self.positions_idx_nbits + self.doc_freq_nbits + self.postings_offset_nbits + self.positions_offset_nbits } // Here inner_offset is the offset within the block, WITHOUT the first term_info. @@ -63,23 +63,30 @@ impl TermInfoBlockMeta { let num_bits = self.num_bits() as usize; let posting_start_addr = num_bits * inner_offset; - // the stop offset is the start offset of the next term info. - let posting_stop_addr = posting_start_addr + num_bits; - let doc_freq_addr = posting_start_addr + self.postings_offset_nbits as usize; - let positions_idx_addr = doc_freq_addr + self.doc_freq_nbits as usize; + // the posting_start is the posting_start of the next term info. + let posting_end_addr = posting_start_addr + num_bits; + let positions_start_addr = posting_start_addr + self.postings_offset_nbits as usize; + // the position_end is the positions_start of the next term info. + let positions_end_addr = positions_start_addr + num_bits as usize; + + let doc_freq_addr = positions_start_addr + self.positions_offset_nbits as usize; let postings_start_offset = self.ref_term_info.postings_range.start + extract_bits(data, posting_start_addr, self.postings_offset_nbits) as usize; let postings_end_offset = self.ref_term_info.postings_range.start - + extract_bits(data, posting_stop_addr, self.postings_offset_nbits) as usize; + + extract_bits(data, posting_end_addr, self.postings_offset_nbits) as usize; + + let positions_start_offset = self.ref_term_info.positions_range.start + + extract_bits(data, positions_start_addr, self.positions_offset_nbits) as usize; + let positions_end_offset = self.ref_term_info.positions_range.start + + extract_bits(data, positions_end_addr, self.positions_offset_nbits) as usize; + let doc_freq = extract_bits(data, doc_freq_addr, self.doc_freq_nbits) as u32; - let positions_idx = self.ref_term_info.positions_idx - + extract_bits(data, positions_idx_addr, self.positions_idx_nbits); TermInfo { doc_freq, postings_range: postings_start_offset..postings_end_offset, - positions_idx, + positions_range: positions_start_offset..positions_end_offset, } } } @@ -167,14 +174,13 @@ fn bitpack_serialize( write, )?; bit_packer.write( - u64::from(term_info.doc_freq), - term_info_block_meta.doc_freq_nbits, + term_info.positions_range.start as u64, + term_info_block_meta.positions_offset_nbits, write, )?; - bit_packer.write( - term_info.positions_idx, - term_info_block_meta.positions_idx_nbits, + u64::from(term_info.doc_freq), + term_info_block_meta.doc_freq_nbits, write, )?; Ok(()) @@ -201,29 +207,29 @@ impl TermInfoStoreWriter { }; let postings_end_offset = last_term_info.postings_range.end - ref_term_info.postings_range.start; + let positions_end_offset = + last_term_info.positions_range.end - ref_term_info.positions_range.start; for term_info in &mut self.term_infos[1..] { term_info.postings_range.start -= ref_term_info.postings_range.start; - term_info.positions_idx -= ref_term_info.positions_idx; + term_info.positions_range.start -= ref_term_info.positions_range.start; } let mut max_doc_freq: u32 = 0u32; - let max_postings_offset: usize = postings_end_offset; - let max_positions_idx: u64 = last_term_info.positions_idx; for term_info in &self.term_infos[1..] { max_doc_freq = cmp::max(max_doc_freq, term_info.doc_freq); } let max_doc_freq_nbits: u8 = compute_num_bits(u64::from(max_doc_freq)); - let max_postings_offset_nbits = compute_num_bits(max_postings_offset as u64); - let max_positions_idx_nbits = compute_num_bits(max_positions_idx); + let max_postings_offset_nbits = compute_num_bits(postings_end_offset as u64); + let max_positions_offset_nbits = compute_num_bits(positions_end_offset as u64); let term_info_block_meta = TermInfoBlockMeta { offset: self.buffer_term_infos.len() as u64, ref_term_info, doc_freq_nbits: max_doc_freq_nbits, postings_offset_nbits: max_postings_offset_nbits, - positions_idx_nbits: max_positions_idx_nbits, + positions_offset_nbits: max_positions_offset_nbits, }; term_info_block_meta.serialize(&mut self.buffer_block_metas)?; @@ -236,11 +242,17 @@ impl TermInfoStoreWriter { )?; } + // We still need to serialize the end offset for postings & positions. bit_packer.write( postings_end_offset as u64, term_info_block_meta.postings_offset_nbits, &mut self.buffer_term_infos, )?; + bit_packer.write( + positions_end_offset as u64, + term_info_block_meta.positions_offset_nbits, + &mut self.buffer_term_infos, + )?; // Block need end up at the end of a byte. bit_packer.flush(&mut self.buffer_term_infos)?; @@ -313,11 +325,11 @@ mod tests { ref_term_info: TermInfo { doc_freq: 512, postings_range: 51..57, - positions_idx: 3584, + positions_range: 110..134, }, doc_freq_nbits: 10, postings_offset_nbits: 5, - positions_idx_nbits: 11, + positions_offset_nbits: 8, }; let mut buffer: Vec = Vec::new(); term_info_block_meta.serialize(&mut buffer).unwrap(); @@ -335,7 +347,7 @@ mod tests { let term_info = TermInfo { doc_freq: i as u32, postings_range: offset(i)..offset(i + 1), - positions_idx: (i * 7) as u64, + positions_range: offset(i) * 3..offset(i + 1) * 3 }; store_writer.write_term_info(&term_info)?; term_infos.push(term_info); diff --git a/src/termdict/tests.rs b/src/termdict/tests.rs index 6672e50dc..54b3e6a90 100644 --- a/src/termdict/tests.rs +++ b/src/termdict/tests.rs @@ -13,7 +13,7 @@ fn make_term_info(term_ord: u64) -> TermInfo { TermInfo { doc_freq: term_ord as u32, postings_range: offset(term_ord)..offset(term_ord + 1), - positions_idx: offset(term_ord) as u64 * 2u64, + positions_range: offset(term_ord) * 2 ..offset(term_ord + 1) * 2, } }