Skip to content

Commit d37e61e

Browse files
authored
perf: lazily open positions file (#54)
Not all queries use positions and it's okay if we (from the perspective of `pg_search`, anyways) defer opening/loading them until they're first needed. This is probably completely wrong for a mmap-based Directory, but we (again, `pg_search`) decided long ago that we don't care about that use case. This saves a lot of disk I/O when an index has lots of segments and the query doesn't need positions. As a drive by, make sure a random Vec has enough space before pushing items to it. This showed up in the profiler, believe it or not.
1 parent 3ffaeb8 commit d37e61e

4 files changed

Lines changed: 62 additions & 15 deletions

File tree

common/src/file_slice.rs

Lines changed: 22 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
use std::fs::File;
22
use std::ops::{Deref, Range, RangeBounds};
3-
use std::sync::Arc;
3+
use std::sync::{Arc, OnceLock};
44
use std::{fmt, io};
55

66
use async_trait::async_trait;
@@ -332,6 +332,27 @@ impl FileHandle for OwnedBytes {
332332
}
333333
}
334334

335+
pub struct DeferredFileSlice {
336+
opener: Arc<dyn Fn() -> io::Result<FileSlice> + Send + Sync + 'static>,
337+
file_slice: OnceLock<std::io::Result<FileSlice>>,
338+
}
339+
340+
impl DeferredFileSlice {
341+
pub fn new(opener: impl Fn() -> io::Result<FileSlice> + Send + Sync + 'static) -> Self {
342+
DeferredFileSlice {
343+
opener: Arc::new(opener),
344+
file_slice: OnceLock::default(),
345+
}
346+
}
347+
348+
pub fn open(&self) -> io::Result<&FileSlice> {
349+
match self.file_slice.get_or_init(|| (self.opener)()) {
350+
Ok(file_slice) => Ok(file_slice),
351+
Err(e) => Err(io::Error::new(io::ErrorKind::Other, e.to_string())),
352+
}
353+
}
354+
}
355+
335356
#[cfg(test)]
336357
mod tests {
337358
use std::io;

src/index/inverted_index_reader.rs

Lines changed: 8 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
use std::io;
22

3+
use common::file_slice::DeferredFileSlice;
34
use common::json_path_writer::JSON_END_OF_PATH;
45
use common::BinarySerializable;
56
use fnv::FnvHashSet;
@@ -31,7 +32,7 @@ use crate::termdict::TermDictionary;
3132
pub struct InvertedIndexReader {
3233
termdict: TermDictionary,
3334
postings_file_slice: FileSlice,
34-
positions_file_slice: FileSlice,
35+
positions_file_slice: DeferredFileSlice,
3536
record_option: IndexRecordOption,
3637
total_num_tokens: u64,
3738
}
@@ -40,7 +41,7 @@ impl InvertedIndexReader {
4041
pub(crate) fn new(
4142
termdict: TermDictionary,
4243
postings_file_slice: FileSlice,
43-
positions_file_slice: FileSlice,
44+
positions_file_slice: DeferredFileSlice,
4445
record_option: IndexRecordOption,
4546
) -> io::Result<InvertedIndexReader> {
4647
let (total_num_tokens_slice, postings_body) = postings_file_slice.split(8);
@@ -60,7 +61,7 @@ impl InvertedIndexReader {
6061
InvertedIndexReader {
6162
termdict: TermDictionary::empty(),
6263
postings_file_slice: FileSlice::empty(),
63-
positions_file_slice: FileSlice::empty(),
64+
positions_file_slice: DeferredFileSlice::new(|| Ok(FileSlice::empty())),
6465
record_option,
6566
total_num_tokens: 0u64,
6667
}
@@ -171,6 +172,7 @@ impl InvertedIndexReader {
171172
if option.has_positions() {
172173
let positions_data = self
173174
.positions_file_slice
175+
.open()?
174176
.read_bytes_slice(term_info.positions_range.clone())?;
175177
let position_reader = PositionReader::open(positions_data)?;
176178
Some(position_reader)
@@ -280,6 +282,7 @@ impl InvertedIndexReader {
280282
if with_positions {
281283
let positions = self
282284
.positions_file_slice
285+
.open()?
283286
.read_bytes_slice_async(term_info.positions_range.clone());
284287
futures_util::future::try_join(postings, positions).await?;
285288
} else {
@@ -322,6 +325,7 @@ impl InvertedIndexReader {
322325
if with_positions {
323326
let positions = self
324327
.positions_file_slice
328+
.open()?
325329
.read_bytes_slice_async(positions_range);
326330
futures_util::future::try_join(postings, positions).await?;
327331
} else {
@@ -416,7 +420,7 @@ impl InvertedIndexReader {
416420
pub async fn warm_postings_full(&self, with_positions: bool) -> io::Result<()> {
417421
self.postings_file_slice.read_bytes_async().await?;
418422
if with_positions {
419-
self.positions_file_slice.read_bytes_async().await?;
423+
self.positions_file_slice.open()?.read_bytes_async().await?;
420424
}
421425
Ok(())
422426
}

src/index/segment_reader.rs

Lines changed: 29 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@ use std::path::PathBuf;
44
use std::sync::{Arc, OnceLock, RwLock};
55
use std::{fmt, io};
66

7+
use common::file_slice::DeferredFileSlice;
78
use fnv::FnvHashMap;
89
use itertools::Itertools;
910

@@ -236,19 +237,38 @@ impl SegmentReader {
236237
))
237238
})?;
238239

239-
let positions_file = self.positions_composite().open_read(field).ok_or_else(|| {
240-
let error_msg = format!(
241-
"Failed to open field {:?}'s positions in the composite file. Has the schema been \
242-
modified?",
243-
field_entry.name()
244-
);
245-
DataCorruption::comment_only(error_msg)
246-
})?;
240+
// not all queries require positions.
241+
// we can defer opening the file until needed
242+
let positions_file_opener = {
243+
let path = self.relative_path(SegmentComponent::Positions);
244+
let directory = self.index.directory().clone();
245+
let field_entry = field_entry.clone();
246+
move || {
247+
let composite_file = if let Ok(positions_file) = &directory.open_read(&path) {
248+
CompositeFile::open(&positions_file)
249+
.expect("should be able to open positions composite component")
250+
} else {
251+
CompositeFile::empty()
252+
};
253+
254+
composite_file.open_read(field).ok_or_else(|| {
255+
let error_msg = format!(
256+
"Failed to open field {:?}'s positions in the composite file. Has the \
257+
schema been modified?",
258+
field_entry.name()
259+
);
260+
io::Error::new(
261+
io::ErrorKind::InvalidData,
262+
format!("{:?}", DataCorruption::comment_only(error_msg)),
263+
)
264+
})
265+
}
266+
};
247267

248268
let inv_idx_reader = Arc::new(InvertedIndexReader::new(
249269
TermDictionary::open(termdict_file)?,
250270
postings_file,
251-
positions_file,
271+
DeferredFileSlice::new(positions_file_opener),
252272
record_option,
253273
)?);
254274

src/termdict/sstable_termdict/mod.rs

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -158,10 +158,12 @@ impl ValueReader for TermInfoValueReader {
158158

159159
fn load(&mut self, mut data: &[u8]) -> io::Result<usize> {
160160
let len_before = data.len();
161-
self.term_infos.clear();
162161
let num_els = VInt::deserialize_u64(&mut data)?;
163162
let mut postings_start = VInt::deserialize_u64(&mut data)? as usize;
164163
let mut positions_start = VInt::deserialize_u64(&mut data)? as usize;
164+
165+
self.term_infos.clear();
166+
self.term_infos.reserve_exact(num_els as usize);
165167
for _ in 0..num_els {
166168
let doc_freq = VInt::deserialize_u64(&mut data)? as u32;
167169
let postings_num_bytes = VInt::deserialize_u64(&mut data)?;

0 commit comments

Comments
 (0)