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

First stab

Browse Source
This commit is contained in:
Paul Masurel
2021-04-23 12:12:02 +09:00
parent 6fb3622abb
commit aead5d4068
13 changed files with 226 additions and 304 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/core/index_meta.rs
-1
View File
@@ -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(),
src/core/inverted_index_reader.rs
+6 -10
View File
@@ -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<InvertedIndexReader> {
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<SegmentPostings> {
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,
))
}
src/core/segment_component.rs
+1 -4
View File
@@ -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,
src/core/segment_reader.rs
+4 -18
View File
@@ -46,7 +46,6 @@ pub struct SegmentReader {
termdict_composite: CompositeFile,
postings_composite: CompositeFile,
positions_composite: CompositeFile,
positions_idx_composite: CompositeFile,
fast_fields_readers: Arc<FastFieldReaders>,
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(),
src/positions/mod.rs
+42 -54
View File
@@ -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<u32> = (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<u32> = (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<u32> = (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<u32> = (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<u32> = (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<u32> = (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<u32> = (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<u32> = vec![];
let mut positions_delta: Vec<u32> = 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<u32> = (0..i).map(|el| el as u32).collect();
assert_eq!(buf, &r[..]);
position_reader.read(offset, &mut buf[..]);
let expected_positions_delta: Vec<u32> = (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<u32> = 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<u32> = 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<u32> = (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<u32> = (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);
}
}
src/positions/reader.rs
+60 -110
View File
@@ -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<Positions> {
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<PositionReader> {
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<PositionReader> {
let positions = Positions::new(position_file, skip_file)?;
positions.reader(offset)
pub fn new(mut positions_data: OwnedBytes) -> io::Result<PositionReader> {
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::<usize>() * 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);
}
}
}
src/positions/serializer.rs
+44 -46
View File
@@ -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<W: io::Write> {
bit_packer: BitPacker4x,
write_stream: CountingWriter<W>,
write_skip_index: W,
block_encoder: BlockEncoder,
positions_wrt: CountingWriter<W>,
positions_buffer: Vec<u8>,
block: Vec<u32>,
buffer: Vec<u8>,
num_ints: u64,
long_skips: Vec<u64>,
bit_widths: Vec<u8>,
}
impl<W: io::Write> PositionSerializer<W> {
pub fn new(write_stream: W, write_skip_index: W) -> PositionSerializer<W> {
pub fn new(positions_wrt: W) -> PositionSerializer<W> {
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<W: io::Write> PositionSerializer<W> {
}
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()
}
}
src/postings/mod.rs
+1 -1
View File
@@ -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(())
}
src/postings/serializer.rs
+10 -12
View File
@@ -50,19 +50,17 @@ pub struct InvertedIndexSerializer {
terms_write: CompositeWrite<WritePtr>,
postings_write: CompositeWrite<WritePtr>,
positions_write: CompositeWrite<WritePtr>,
positionsidx_write: CompositeWrite<WritePtr>,
schema: Schema,
}
impl InvertedIndexSerializer {
/// Open a new `PostingsSerializer` for the given segment
pub fn open(segment: &mut Segment) -> crate::Result<InvertedIndexSerializer> {
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<WritePtr>,
postings_write: &'a mut CountingWriter<WritePtr>,
positions_write: &'a mut CountingWriter<WritePtr>,
positionsidx_write: &'a mut CountingWriter<WritePtr>,
fieldnorm_reader: Option<FieldNormReader>,
) -> io::Result<FieldSerializer<'a>> {
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;
src/postings/term_info.rs
+19 -8
View File
@@ -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<usize>,
/// 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<usize>,
}
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<R: io::Read>(reader: &mut R) -> io::Result<Self> {
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::<TermInfo>();
src/space_usage/mod.rs
-13
View File
@@ -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);
src/termdict/fst_termdict/term_info_store.rs
+38 -26
View File
@@ -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<W: Write>(
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<u8> = 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);
src/termdict/tests.rs
+1 -1
View File
@@ -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,
}
}